diff --git a/api-report/apps/api/src/modules/process/index.d.ts b/api-report/apps/api/src/modules/process/index.d.ts index 4579549..aa990bd 100644 --- a/api-report/apps/api/src/modules/process/index.d.ts +++ b/api-report/apps/api/src/modules/process/index.d.ts @@ -3,7 +3,7 @@ */ export type { ActivityContractFace, AssembleScope, Candidate, ClosureInput, ClosureReport, ClosureViolation, NodeInstance, OwnerSpec, ProcessPackage, ProcessService, ResolveContext, Resolution, StartWorkflowInput, WorkflowRecord, WorkflowState, } from './contract.ts'; export { CallInterfaceViolationError, NotClosedError, PinnedVersionError, PositionVacantError, StaleVersionError } from './contract.ts'; -export type { ActivityDirectory, ApprovalGateway, Clock, FlowEventPublisher, IdGenerator, OrgDirectory, ProcessStore, ReceiptNotifier, Sfeel, TaskGateway, } from './port.ts'; +export type { FlowRuntime, ActivityDirectory, ApprovalGateway, Clock, FlowEventPublisher, IdGenerator, OrgDirectory, ProcessStore, ReceiptNotifier, Sfeel, TaskGateway, } from './port.ts'; export type { ProcessServiceDeps } from './internal/usecase/process-service.ts'; export { createProcessService } from './internal/usecase/process-service.ts'; export { advanceOrder, assemble, dispatchOf, rejectRouteOf, selectEdges, startNodeOf, } from './internal/domain/routing.ts'; diff --git a/api-report/apps/api/src/modules/process/port.d.ts b/api-report/apps/api/src/modules/process/port.d.ts index b95f944..3f0cfc7 100644 --- a/api-report/apps/api/src/modules/process/port.d.ts +++ b/api-report/apps/api/src/modules/process/port.d.ts @@ -1,7 +1,7 @@ /** * module: process —— 本模块要外界提供的能力(编码规范 §10.1 / §5.3)。 */ -import type { ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts'; +import type { ProcessNode, ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts'; import type { NodeInstance } from './internal/domain/owner-resolution.ts'; import type { ActivityContractFace } from './internal/domain/closure.ts'; export type Clock = { @@ -35,6 +35,24 @@ export type WorkflowRecord = { readonly startedAt: string; readonly finishedAt?: string; }; +/** + * 流程运行时 —— **执行面的可替换切面**(M5-1):按结构调度节点、条件出边选路、 + * 失败走向、审批裁决、状态聚合。自持状态机(internal/runtime/self-hosted-runtime) + * 是默认实现;Mastra 编译产物(workflow-compiler)将是第二实现 —— + * 八例端到端是两个实现的等价性护栏。 + * 定义生命周期(保存/激活/校验)不在此内:那决定「编译哪一份定义」,与引擎选型无关。 + */ +export type FlowRuntime = { + /** 起单后进入起点(起单前置 —— 校验/建单/留痕 —— 归 usecase,不在此内) */ + begin(record: WorkflowRecord, pkg: ProcessPackage): Promise; + onTaskTerminal(taskId: string, terminal: TerminalState, ports: Record>): Promise; + applyFailure(workflowId: string, nodeId: string, instance: number, node: Extract, terminal: TerminalState): Promise; + onApprovalVerdict(workflowId: string, nodeId: string, verdict: 'pass' | 'reject', payload: Record): Promise; + refreshState(workflowId: string): Promise; + terminate(workflowId: string, reason: string): Promise; +}; export type ProcessStore = { savePackage(pkg: ProcessPackage): Promise; /** 就地写回一版的定义(编辑即其身)—— 撞不存在的版本报错,不静默插入 */ diff --git a/apps/api/src/modules/process/CONTRACT.md b/apps/api/src/modules/process/CONTRACT.md index 6e50467..8c5acd2 100644 --- a/apps/api/src/modules/process/CONTRACT.md +++ b/apps/api/src/modules/process/CONTRACT.md @@ -84,8 +84,20 @@ - **驳回走向声明式**:审批点的驳回按 `reject` 声明回指定节点重开新实例,不由临场决定 - **凡单必有来源**:`source` 必填;终点按输出声明交付、按来源回执 -## M5 待补(docs/PROGRESS.md · docs/debt.md M5-1) -`internal/adapter/workflow-compiler.ts` —— 编码规范 §6.1 规定流程包动态编译为 Mastra -`createWorkflow`/`createStep`(XOR → `.branch()`,AND → `.parallel()`,节点 `suspend()`/`resume()`)。 -当前流转由本模块自持的**确定性状态机 + `sys.node_instance` 持久**承担,语义等价且已可测; -Mastra 编译器作为可替换的 adapter 后补,替换时 usecase 不动。 +## M5-1:运行时切面已立(2026-07-28),编译器实现仍欠 + +「Mastra 编译器作可替换 adapter、usecase 不动」的端口**已经存在**:`port.ts` 的 +`FlowRuntime`(begin / onTaskTerminal / applyFailure / onApprovalVerdict / +refreshState / terminate)。执行面的语义、留痕与副作用编排整体住在默认实现 +`internal/runtime/self-hosted-runtime.ts`(自持确定性状态机 + `sys.node_instance` +持久);`process-service` 是薄壳:定义生命周期(保存/激活/校验)+ 起单前置 +(取生效版/调用接口校验/工作组快照/建单留痕)+ 五方法委托。 + +替换路径:`ProcessServiceDeps.runtime` 注入第二实现(`internal/adapter/ +workflow-compiler.ts`,编码规范 §6.1:流程包编译为 Mastra `createWorkflow`/ +`createStep`,XOR → `.branch()`,AND → `.parallel()`,节点 `suspend()`/`resume()`)。 +八例端到端(`unified-quote-flow.it.test.ts`)是两个实现的**等价性护栏**。 +已核实的语义冲突清单(编译器必须绕):回边 reopen 无原语(spike 定案)、 +`.parallel()` 任一失败即整体失败(step 内自吞,不让 Mastra 感知)、 +三态聚合投影(abnormal 防假成功,Mastra run 无此态)、`node_instance` +投影表必须照写(治理台按行查)。 diff --git a/apps/api/src/modules/process/index.ts b/apps/api/src/modules/process/index.ts index 6df0de8..8f41413 100644 --- a/apps/api/src/modules/process/index.ts +++ b/apps/api/src/modules/process/index.ts @@ -22,6 +22,7 @@ export { CallInterfaceViolationError, NotClosedError, PinnedVersionError, PositionVacantError, StaleVersionError } from './contract.ts'; export type { + FlowRuntime, ActivityDirectory, ApprovalGateway, Clock, diff --git a/apps/api/src/modules/process/internal/runtime/self-hosted-runtime.ts b/apps/api/src/modules/process/internal/runtime/self-hosted-runtime.ts new file mode 100644 index 0000000..623eb9b --- /dev/null +++ b/apps/api/src/modules/process/internal/runtime/self-hosted-runtime.ts @@ -0,0 +1,419 @@ +/** + * 自持流程运行时 —— `FlowRuntime` 端口的默认实现(M5-1 切面的甲方)。 + * + * 确定性状态机 + `sys.node_instance` 持久:按结构调度节点、条件出边选路、 + * 失败走向、审批裁决、状态聚合。**零 LLM、零临场**(编码规范 §5.3;语义出处 OS 03 / 04)。 + * + * 本文件由 process-service 整体搬迁而来(2026-07-28,行为零变化)—— + * 从此运行时是可替换的:Mastra 编译产物(workflow-compiler)将是同一端口的第二实现, + * 八例端到端就是两个实现的等价性护栏。定义生命周期(保存/激活/校验)不在此内, + * 那是「编译哪一份定义」的事,归 usecase。 + */ +import type { ProcessNode, ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts'; +import { PositionVacantError } from '../../contract.ts'; +import type { + ApprovalGateway, + Clock, + FlowEventPublisher, + FlowRuntime, + OrgDirectory, + ProcessStore, + ReceiptNotifier, + Sfeel, + TaskGateway, + WorkflowRecord, +} from '../../port.ts'; +import { aggregateWorkflowState, isDigested, nextInstanceNumber, resolveOwner } from '../domain/owner-resolution.ts'; +import type { NodeInstance } from '../domain/owner-resolution.ts'; +import { assemble, dispatchOf, rejectRouteOf, selectEdges, startNodeOf } from '../domain/routing.ts'; + +export type SelfHostedRuntimeDeps = { + readonly store: ProcessStore; + readonly org: OrgDirectory; + readonly tasks: TaskGateway; + readonly approvals: ApprovalGateway; + readonly sfeel: Sfeel; + readonly events: FlowEventPublisher; + readonly receipts: ReceiptNotifier; + readonly clock: Clock; +}; + +export function createSelfHostedRuntime(deps: SelfHostedRuntimeDeps): FlowRuntime { + async function requirePackage(id: string, version?: number): Promise { + const pkg = await deps.store.getPackage(id, version); + if (pkg === undefined) throw new Error(`流程包不存在:${id}`); + return pkg; + } + + async function requireWorkflow(id: string): Promise { + const record = await deps.store.getWorkflow(id); + if (record === undefined) throw new Error(`工作流不存在:${id}`); + return record; + } + + function scopeOf(record: WorkflowRecord) { + return { ports: record.ports, vars: record.vars, initiation: record.initiation }; + } + + /** 起一个节点:活动 → 起单;审批点 → 卡工位;终点 → 收束 */ + async function enter( + record: WorkflowRecord, + pkg: ProcessPackage, + node: ProcessNode, + payload: Record, + ): Promise { + const dispatch = dispatchOf(node, payload); + + if (dispatch === undefined) { + // 只剩起点会走到这(结构件无调度):被指为目标时直通,不建实例 + await advanceFrom(record, pkg, node.id, payload); + return; + } + + const instances = await deps.store.listInstances(record.id); + const instance = nextInstanceNumber(instances, node.id); + + if (dispatch.kind === 'finish') { + await deps.store.insertInstance(record.id, { + nodeId: node.id, + instance, + running: false, + waiting: false, + activated: true, + terminal: 'released', + }); + const finished = await deps.store.updateWorkflow(record.id, { + state: 'completed', + finishedAt: deps.clock.now().toISOString(), + ports: { ...record.ports, __output: payload }, + }); + await deps.events.publish({ workflowId: record.id, kind: 'workflow-completed', detail: payload }); + // 凡单必有来源 —— 登记人收消息回执 + if (record.source.by !== undefined) { + await deps.receipts.notify({ + to: record.source.by as StaffId, + workflowId: record.id, + subject: record.subject, + state: finished.state, + }); + } + return; + } + + /* + * 谁来干:**编排不带岗位**(2026-07-28 第五轮,活动点与审批点同一口径)—— + * 负责人是**运营时分配**的:工作流绑了员工组就按组轮值(组员并集里最闲者先), + * 没绑组就全员轮值;staffId 稳定排序保证同输入必同结果。 + * 分派结果记在节点实例上(assignee)—— 审批箱、链条、治理列表都按它认人。 + */ + const seen = new Map(); + for (const position of await deps.org.positions()) { + for (const c of await deps.org.candidates(position)) if (!seen.has(c.staffId)) seen.set(c.staffId, c); + } + const everyone = [...seen.values()]; + const group = new Set(Object.values(record.workGroup ?? {})); + const groupPool = everyone.filter((c) => group.has(c.staffId)); + // 组员全不在岗时退回全员 —— 单不能没人接 + const pool = groupPool.length > 0 ? groupPool : everyone; + const spec = { position: groupPool.length > 0 ? '(员工组轮值)' : '(全员轮值)', strategy: 'sole' as const }; + const resolution = resolveOwner(spec, { candidates: pool }); + if (resolution.kind === 'vacant') throw new PositionVacantError(spec.position); + + await deps.events.publish({ + workflowId: record.id, + kind: 'owner-resolved', + detail: { nodeId: node.id, position: spec.position, staffId: resolution.staffId, by: resolution.by, rationale: resolution.rationale }, + }); + + if (dispatch.kind === 'start-task') { + const task = await deps.tasks.start({ + activityId: dispatch.node.activityId, + ownerId: resolution.staffId as StaffId, + title: dispatch.node.label, + workflowId: record.id, + nodeId: node.id, + data: payload, + ...(dispatch.node.message === undefined ? {} : { message: dispatch.node.message }), + }); + await deps.store.insertInstance(record.id, { + nodeId: node.id, + instance, + running: true, + waiting: false, + activated: true, + taskId: task.id, + assigneeId: resolution.staffId, + }); + await deps.events.publish({ workflowId: record.id, kind: 'task-started', detail: { nodeId: node.id, taskId: task.id } }); + return; + } + + const approval = await deps.approvals.raise({ + workflowId: record.id, + nodeId: node.id, + approverId: resolution.staffId as StaffId, + subject: dispatch.node.label, + basis: Object.entries(payload).map(([k, v]) => ({ k, v: String(v) })), + }); + await deps.store.insertInstance(record.id, { + nodeId: node.id, + instance, + running: false, + waiting: true, + activated: true, + approvalId: approval.id, + assigneeId: resolution.staffId, + }); + await deps.events.publish({ workflowId: record.id, kind: 'approval-raised', detail: { nodeId: node.id, approvalId: approval.id } }); + } + + /** + * 从某节点继续流转:选边 → 装配 → 进下一节点。 + * + * 返回**有没有真的推出去**。审批点的驳回要用它:图上显式画了驳回出边就走边, + * 没画的包才退回 `reject` 声明 —— 调用方据此判断,不必自己再选一遍边。 + * + * `quiet` 为真时不发 `route-dead-end`:那种「先试边、不成再退回声明」的探路 + * 一条边都没中是**正常分支**,不是死路,报出来就是假警。 + */ + async function advanceFrom( + record: WorkflowRecord, + pkg: ProcessPackage, + fromNodeId: string, + payload: Record, + quiet = false, + ): Promise { + const scope = scopeOf(record); + const flat = { ...scope.initiation, ...scope.vars, ...payload }; + const edges = selectEdges(pkg, fromNodeId, flat, deps.sfeel.condition); + + /* + * 一条边都没选中,而这个节点确实有出边 —— 流到此为止,再没有谁会来推它。 + * 这是**死路**,不是终点。条件全不成立最容易碰上(实测:A1 的出口少了两枚去向旗, + * 两条条件出边一条都选不中,流无声停在 A1)。 + * + * 状态聚合已经会把它报成 `abnormal`(没到终点就不算完成),但那只说「有问题」、 + * 不说「哪儿有问题」。这里留一条可诊断的痕:哪个节点、出边各自的条件、判据里到底有哪些键。 + * 只记键名不记值 —— 诊断够用,载荷不进事件明文。 + */ + if (edges.length === 0) { + const outgoing = pkg.edges.filter((e) => e.from === fromNodeId); + if (outgoing.length > 0 && !quiet) { + await deps.events.publish({ + workflowId: record.id, + kind: 'route-dead-end', + detail: { + nodeId: fromNodeId, + edges: outgoing.map((e) => ({ id: e.id, to: e.to, condition: e.condition ?? '(无条件)' })), + availableKeys: Object.keys(flat).sort(), + }, + }); + } + } + + for (const edge of edges) { + const target = pkg.nodes.find((n) => n.id === edge.to); + if (target === undefined) continue; + + /* + * 多入边即异或汇合:上游互斥,只会来一条 —— **没有「等回齐」这回事**。 + * (等待逻辑随网关一起摘除;AND 并行待「并行组」声明时再带自己的汇合语义。) + * 节点只收**装配后的**载荷:活动与审批点都有契约,不被上游多余字段污染; + * 判据要用的隐性上下文(审批 verdict、放行选口 outcome)由系统注入判据作用域,不走映射。 + */ + const arriving = assemble(edge, { ...scope, ports: { ...scope.ports, [fromNodeId]: payload } }, deps.sfeel.expression); + try { + await enter(record, pkg, target, arriving); + } catch (error) { + /* + * 一条分支起不来,不该把整条终态上报连坐掀翻 —— 那会让**上游任务的终态也报不上去**, + * 流停在原地且一行日志都没有。这里把它记成工作流事件并落一个未激活的实例: + * 其余分支照走,这一支在治理台「需处置」里现形,等人处置。 + */ + const detail = error instanceof Error ? error.message : String(error); + await deps.events.publish({ + workflowId: record.id, + kind: 'node-start-failed', + detail: { nodeId: target.id, from: fromNodeId, error: detail }, + }); + const instances = await deps.store.listInstances(record.id); + await deps.store.insertInstance(record.id, { + nodeId: target.id, + instance: nextInstanceNumber(instances, target.id), + running: false, + waiting: true, + activated: true, + terminal: 'blocked', + }); + } + } + return edges.length > 0; + } + + /** 工作流状态是聚合投影 —— 每次变动后重算 */ + async function refreshState(workflowId: string): Promise { + const record = await requireWorkflow(workflowId); + if (record.state === 'terminated') return 'terminated'; + const instances = await deps.store.listInstances(workflowId); + // 「完成」凭到过终点,不凭没有在跑的了 —— 所以要把包的终点节点交给聚合 + const pkg = await deps.store.getPackage(record.packageId, record.packageVersion); + const endNodeIds = (pkg?.nodes ?? []).filter((n) => n.kind === 'end').map((n) => n.id); + const state = aggregateWorkflowState(instances as NodeInstance[], false, endNodeIds); + if (state !== record.state) { + await deps.store.updateWorkflow(workflowId, { + state, + ...(state === 'completed' ? { finishedAt: deps.clock.now().toISOString() } : {}), + }); + } + return state; + } + + /** 失败走向四选一:终止流(默认)/ 改道 / 升审批 / 容忍 */ + async function applyFailure( + workflowId: string, + nodeId: string, + instance: number, + node: Extract, + terminal: TerminalState, + ): Promise { + const record = await requireWorkflow(workflowId); + const pkg = await requirePackage(record.packageId, record.packageVersion); + const disposition = node.failure; + const digested = isDigested(disposition.kind); + + await deps.store.updateInstance(workflowId, nodeId, instance, { failureDigested: digested }); + await deps.events.publish({ + workflowId, + kind: 'failure-disposition', + detail: { nodeId, terminal, disposition: disposition.kind }, + }); + + if (disposition.kind === 'reroute') { + const edge = pkg.edges.find((e) => e.id === disposition.edgeId); + const target = edge === undefined ? undefined : pkg.nodes.find((n) => n.id === edge.to); + if (edge !== undefined && target !== undefined) { + // 失败载荷 = 终态、原因码与说明 + const payload = assemble(edge, scopeOf(record), deps.sfeel.expression); + await enter(record, pkg, target, { ...payload, __terminal: terminal }); + } + } + + if (disposition.kind === 'escalate') { + const target = pkg.nodes.find((n) => n.id === disposition.approvalNodeId); + if (target !== undefined) await enter(record, pkg, target, { __terminal: terminal, __from: nodeId }); + } + + if (disposition.kind === 'tolerate') { + /* + * 容忍 = 这条支线的失败被吸收,**流继续往下走**。 + * 光把实例标成「已消化」而不往下推,这条线就断在这儿 —— 下游谁也不会再被叫起。 + * 载荷为空 —— 容忍分支本就没有产出,下游的必填映射不得依赖它(闭合校验已拦)。 + */ + await advanceFrom(record, pkg, nodeId, {}); + } + + return refreshState(workflowId); + } + + return { + /** 起单后进入起点:调用方(usecase)负责起单前置(校验/建单/留痕),这里只管走 */ + async begin(record, pkg) { + const start = startNodeOf(pkg); + if (start !== undefined) await advanceFrom(record, pkg, start.id, record.initiation); + }, + + /** 任务终态到达:按终态决定继续流转还是走失败走向 */ + async onTaskTerminal(taskId, terminal, ports) { + const located = await deps.store.findInstanceByTask(taskId); + if (located === undefined) return undefined; + + const record = await requireWorkflow(located.workflowId); + const pkg = await requirePackage(record.packageId, record.packageVersion); + const node = pkg.nodes.find((n) => n.id === located.nodeId); + + await deps.store.updateInstance(record.id, located.nodeId, located.instance, { + running: false, + waiting: false, + terminal, + }); + await deps.events.publish({ workflowId: record.id, kind: 'task-terminal', detail: { taskId, terminal } }); + + const merged = await deps.store.updateWorkflow(record.id, { ports: { ...record.ports, ...ports } }); + + if (terminal === 'released') { + // 落地收敛后才发射输出口 —— 下游只见落地完成的任务。 + // 各口载荷合并成一份路由作用域:只取第一口,活动没有必交付口时就什么都拿不到。 + const payload: Record = Object.assign({}, ...Object.values(ports)); + await advanceFrom(merged, pkg, located.nodeId, payload); + } else if (node?.kind === 'activity') { + // 非完成终态触发**声明的失败走向**;流程层不自动重试 + await applyFailure(record.id, located.nodeId, located.instance, node, terminal); + } + + return refreshState(record.id); + }, + + applyFailure, + + /** 审批裁决:通过继续流转;驳回按**声明的走向**回指定节点重开新实例 */ + async onApprovalVerdict(workflowId, nodeId, verdict, payload) { + const record = await requireWorkflow(workflowId); + const pkg = await requirePackage(record.packageId, record.packageVersion); + const node = pkg.nodes.find((n) => n.id === nodeId); + if (node?.kind !== 'approval') throw new Error(`节点不是审批点:${nodeId}`); + + const instances = await deps.store.listInstances(workflowId); + const current = instances.filter((i) => i.nodeId === nodeId).at(-1); + if (current !== undefined) { + await deps.store.updateInstance(workflowId, nodeId, current.instance, { + waiting: false, + terminal: 'released', + }); + } + await deps.events.publish({ workflowId, kind: 'approval-verdict', detail: { nodeId, verdict } }); + + /* + * **裁决进载荷**:审批点的两个去向就是它自己的两条条件出边, + * 条件判的正是这个字段(`verdict = "pass"` / `verdict = "reject"`)。 + * 系统注入判据作用域,不走映射、不进设计者编排面。 + */ + const carried = { ...payload, verdict }; + + if (verdict === 'pass') { + await advanceFrom(record, pkg, nodeId, carried); + } else { + /* + * 驳回:**先看图**。图上显式画了驳回出边就走边 —— 编排即定契约,图是真相。 + * 没画的包才退回 `reject` 声明:它是 schema 必填的通用机制, + * 不是所有流程包都会把驳回画成边。 + * + * `quiet` 传真:这一步「试边不成」是正常分支,不是死路,不该报 route-dead-end。 + */ + const routedByEdge = await advanceFrom(record, pkg, nodeId, carried, true); + if (!routedByEdge) { + const route = rejectRouteOf(node, scopeOf(record), deps.sfeel.expression); + if (route.kind === 'terminate') { + await deps.store.updateWorkflow(workflowId, { state: 'terminated', finishedAt: deps.clock.now().toISOString() }); + } else { + // 回指定节点**重开新实例**(同节点、新任务 ID、留痕关联),不复活旧任务 + const target = pkg.nodes.find((n) => n.id === route.nodeId); + if (target !== undefined) await enter(record, pkg, target, { ...route.payload, ...payload }); + } + } + } + return refreshState(workflowId); + }, + + refreshState, + + async terminate(workflowId, reason) { + await deps.store.updateWorkflow(workflowId, { + state: 'terminated', + finishedAt: deps.clock.now().toISOString(), + }); + // 工作流终止级联:在途任务终止、待答卡与征答消息作废、审批点撤卡 + await deps.events.publish({ workflowId, kind: 'workflow-terminated', detail: { reason } }); + return 'terminated'; + }, + }; +} diff --git a/apps/api/src/modules/process/internal/usecase/process-service.ts b/apps/api/src/modules/process/internal/usecase/process-service.ts index 62ac8fa..5cb6ed6 100644 --- a/apps/api/src/modules/process/internal/usecase/process-service.ts +++ b/apps/api/src/modules/process/internal/usecase/process-service.ts @@ -3,14 +3,15 @@ * * 起单 · 调度节点 · 审批点卡工位 · 条件出边选路 · 每次流转记入事件。 */ -import type { ProcessNode, ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts'; +import type { ProcessPackage } from '@contracts'; import type { ProcessService, StartWorkflowInput } from '../../contract.ts'; -import { CallInterfaceViolationError, NotClosedError, PinnedVersionError, PositionVacantError, StaleVersionError } from '../../contract.ts'; +import { CallInterfaceViolationError, NotClosedError, PinnedVersionError, StaleVersionError } from '../../contract.ts'; import type { ActivityDirectory, ApprovalGateway, Clock, FlowEventPublisher, + FlowRuntime, IdGenerator, OrgDirectory, ProcessStore, @@ -21,9 +22,7 @@ import type { WorkflowRecord, } from '../../port.ts'; import { checkClosure } from '../domain/closure.ts'; -import { aggregateWorkflowState, isDigested, nextInstanceNumber, resolveOwner } from '../domain/owner-resolution.ts'; -import type { NodeInstance } from '../domain/owner-resolution.ts'; -import { assemble, dispatchOf, rejectRouteOf, selectEdges, startNodeOf } from '../domain/routing.ts'; +import { createSelfHostedRuntime } from '../runtime/self-hosted-runtime.ts'; export type ProcessServiceDeps = { readonly store: ProcessStore; @@ -37,6 +36,11 @@ export type ProcessServiceDeps = { readonly receipts: ReceiptNotifier; readonly clock: Clock; readonly ids: IdGenerator; + /** + * 流程运行时(可替换切面,M5-1):缺省 = 自持状态机。 + * 换 Mastra 编译产物时从这里注入 —— usecase 与全部路由/接线不动。 + */ + readonly runtime?: FlowRuntime; }; export function createProcessService(deps: ProcessServiceDeps): ProcessService { @@ -73,205 +77,7 @@ export function createProcessService(deps: ProcessServiceDeps): ProcessService { return record; } - function scopeOf(record: WorkflowRecord) { - return { ports: record.ports, vars: record.vars, initiation: record.initiation }; - } - - /** 起一个节点:活动 → 起单;审批点 → 卡工位;终点 → 收束 */ - async function enter( - record: WorkflowRecord, - pkg: ProcessPackage, - node: ProcessNode, - payload: Record, - ): Promise { - const dispatch = dispatchOf(node, payload); - - if (dispatch === undefined) { - // 只剩起点会走到这(结构件无调度):被指为目标时直通,不建实例 - await advanceFrom(record, pkg, node.id, payload); - return; - } - - const instances = await deps.store.listInstances(record.id); - const instance = nextInstanceNumber(instances, node.id); - - if (dispatch.kind === 'finish') { - await deps.store.insertInstance(record.id, { - nodeId: node.id, - instance, - running: false, - waiting: false, - activated: true, - terminal: 'released', - }); - const finished = await deps.store.updateWorkflow(record.id, { - state: 'completed', - finishedAt: deps.clock.now().toISOString(), - ports: { ...record.ports, __output: payload }, - }); - await deps.events.publish({ workflowId: record.id, kind: 'workflow-completed', detail: payload }); - // 凡单必有来源 —— 登记人收消息回执 - if (record.source.by !== undefined) { - await deps.receipts.notify({ - to: record.source.by as StaffId, - workflowId: record.id, - subject: record.subject, - state: finished.state, - }); - } - return; - } - - /* - * 谁来干:**编排不带岗位**(2026-07-28 第五轮,活动点与审批点同一口径)—— - * 负责人是**运营时分配**的:工作流绑了员工组就按组轮值(组员并集里最闲者先), - * 没绑组就全员轮值;staffId 稳定排序保证同输入必同结果。 - * 分派结果记在节点实例上(assignee)—— 审批箱、链条、治理列表都按它认人。 - */ - const seen = new Map(); - for (const position of await deps.org.positions()) { - for (const c of await deps.org.candidates(position)) if (!seen.has(c.staffId)) seen.set(c.staffId, c); - } - const everyone = [...seen.values()]; - const group = new Set(Object.values(record.workGroup ?? {})); - const groupPool = everyone.filter((c) => group.has(c.staffId)); - // 组员全不在岗时退回全员 —— 单不能没人接 - const pool = groupPool.length > 0 ? groupPool : everyone; - const spec = { position: groupPool.length > 0 ? '(员工组轮值)' : '(全员轮值)', strategy: 'sole' as const }; - const resolution = resolveOwner(spec, { candidates: pool }); - if (resolution.kind === 'vacant') throw new PositionVacantError(spec.position); - - await deps.events.publish({ - workflowId: record.id, - kind: 'owner-resolved', - detail: { nodeId: node.id, position: spec.position, staffId: resolution.staffId, by: resolution.by, rationale: resolution.rationale }, - }); - - if (dispatch.kind === 'start-task') { - const task = await deps.tasks.start({ - activityId: dispatch.node.activityId, - ownerId: resolution.staffId as StaffId, - title: dispatch.node.label, - workflowId: record.id, - nodeId: node.id, - data: payload, - ...(dispatch.node.message === undefined ? {} : { message: dispatch.node.message }), - }); - await deps.store.insertInstance(record.id, { - nodeId: node.id, - instance, - running: true, - waiting: false, - activated: true, - taskId: task.id, - assigneeId: resolution.staffId, - }); - await deps.events.publish({ workflowId: record.id, kind: 'task-started', detail: { nodeId: node.id, taskId: task.id } }); - return; - } - - const approval = await deps.approvals.raise({ - workflowId: record.id, - nodeId: node.id, - approverId: resolution.staffId as StaffId, - subject: dispatch.node.label, - basis: Object.entries(payload).map(([k, v]) => ({ k, v: String(v) })), - }); - await deps.store.insertInstance(record.id, { - nodeId: node.id, - instance, - running: false, - waiting: true, - activated: true, - approvalId: approval.id, - assigneeId: resolution.staffId, - }); - await deps.events.publish({ workflowId: record.id, kind: 'approval-raised', detail: { nodeId: node.id, approvalId: approval.id } }); - } - - /** - * 从某节点继续流转:选边 → 装配 → 进下一节点(AND 汇合要回齐)。 - * - * 返回**有没有真的推出去**。审批点的驳回要用它:图上显式画了驳回出边就走边, - * 没画的包才退回 `reject` 声明 —— 调用方据此判断,不必自己再选一遍边。 - * - * `quiet` 为真时不发 `route-dead-end`:那种「先试边、不成再退回声明」的探路 - * 一条边都没中是**正常分支**,不是死路,报出来就是假警。 - */ - async function advanceFrom( - record: WorkflowRecord, - pkg: ProcessPackage, - fromNodeId: string, - payload: Record, - quiet = false, - ): Promise { - const scope = scopeOf(record); - const flat = { ...scope.initiation, ...scope.vars, ...payload }; - const edges = selectEdges(pkg, fromNodeId, flat, deps.sfeel.condition); - - /* - * 一条边都没选中,而这个节点确实有出边 —— 流到此为止,再没有谁会来推它。 - * 这是**死路**,不是终点。异或网关最容易碰上:发布闸要求它的每条出边都带条件, - * 所以没有无条件兜底边可用,判据一缺就全不成立(实测:A1 的出口少了两枚去向旗, - * G1 一条分支都选不中,流无声停在 A1)。 - * - * 状态聚合已经会把它报成 `abnormal`(没到终点就不算完成),但那只说「有问题」、 - * 不说「哪儿有问题」。这里留一条可诊断的痕:哪个节点、出边各自的条件、判据里到底有哪些键。 - * 只记键名不记值 —— 诊断够用,载荷不进事件明文。 - */ - if (edges.length === 0) { - const outgoing = pkg.edges.filter((e) => e.from === fromNodeId); - if (outgoing.length > 0 && !quiet) { - await deps.events.publish({ - workflowId: record.id, - kind: 'route-dead-end', - detail: { - nodeId: fromNodeId, - edges: outgoing.map((e) => ({ id: e.id, to: e.to, condition: e.condition ?? '(无条件)' })), - availableKeys: Object.keys(flat).sort(), - }, - }); - } - } - - for (const edge of edges) { - const target = pkg.nodes.find((n) => n.id === edge.to); - if (target === undefined) continue; - - /* - * 多入边即异或汇合:上游互斥,只会来一条 —— **没有「等回齐」这回事**。 - * (等待逻辑随网关一起摘除;AND 并行待「并行组」声明时再带自己的汇合语义。) - * 节点只收**装配后的**载荷:活动与审批点都有契约,不被上游多余字段污染; - * 判据要用的隐性上下文(审批 verdict、放行选口 outcome)由系统注入判据作用域,不走映射。 - */ - const arriving = assemble(edge, { ...scope, ports: { ...scope.ports, [fromNodeId]: payload } }, deps.sfeel.expression); - try { - await enter(record, pkg, target, arriving); - } catch (error) { - /* - * 一条分支起不来,不该把整条终态上报连坐掀翻 —— 那会让**上游任务的终态也报不上去**, - * 流停在原地且一行日志都没有。这里把它记成工作流事件并落一个未激活的实例: - * 其余分支照走,这一支在治理台「需处置」里现形,等人处置。 - */ - const detail = error instanceof Error ? error.message : String(error); - await deps.events.publish({ - workflowId: record.id, - kind: 'node-start-failed', - detail: { nodeId: target.id, from: fromNodeId, error: detail }, - }); - const instances = await deps.store.listInstances(record.id); - await deps.store.insertInstance(record.id, { - nodeId: target.id, - instance: nextInstanceNumber(instances, target.id), - running: false, - waiting: true, - activated: true, - terminal: 'blocked', - }); - } - } - return edges.length > 0; - } + const runtime = deps.runtime ?? createSelfHostedRuntime(deps); return { /** @@ -399,8 +205,7 @@ export function createProcessService(deps: ProcessServiceDeps): ProcessService { await deps.store.insertWorkflow(record); await deps.events.publish({ workflowId: record.id, kind: 'workflow-started', detail: { source: input.source } }); - const start = startNodeOf(pkg); - if (start !== undefined) await advanceFrom(record, pkg, start.id, input.data); + await runtime.begin(record, pkg); return record; }, @@ -416,158 +221,28 @@ export function createProcessService(deps: ProcessServiceDeps): ProcessService { return deps.store.listInstances(workflowId); }, - /** 任务终态到达:按终态决定继续流转还是走失败走向 */ - async onTaskTerminal(taskId, terminal: TerminalState, ports: Record>) { - const located = await deps.store.findInstanceByTask(taskId); - if (located === undefined) return undefined; - - const record = await requireWorkflow(located.workflowId); - const pkg = await requirePackage(record.packageId, record.packageVersion); - const node = pkg.nodes.find((n) => n.id === located.nodeId); - - await deps.store.updateInstance(record.id, located.nodeId, located.instance, { - running: false, - waiting: false, - terminal, - }); - await deps.events.publish({ workflowId: record.id, kind: 'task-terminal', detail: { taskId, terminal } }); - - const merged = await deps.store.updateWorkflow(record.id, { ports: { ...record.ports, ...ports } }); - - if (terminal === 'released') { - // 落地收敛后才发射输出口 —— 下游只见落地完成的任务。 - // 各口载荷合并成一份路由作用域:只取第一口,活动没有必交付口时就什么都拿不到。 - const payload: Record = Object.assign({}, ...Object.values(ports)); - await advanceFrom(merged, pkg, located.nodeId, payload); - } else if (node?.kind === 'activity') { - // 非完成终态触发**声明的失败走向**;流程层不自动重试 - await this.applyFailure(record.id, located.nodeId, located.instance, node, terminal); - } - - return this.refreshState(record.id); + /* + * 执行面五方法 —— 全部委托 FlowRuntime(M5-1 切面): + * 语义、留痕与副作用编排住在运行时实现里,薄壳只转发。 + */ + async onTaskTerminal(taskId, terminal, ports) { + return runtime.onTaskTerminal(taskId, terminal, ports); }, - /** 失败走向四选一:终止流(默认)/ 改道 / 升审批 / 容忍 */ async applyFailure(workflowId, nodeId, instance, node, terminal) { - const record = await requireWorkflow(workflowId); - const pkg = await requirePackage(record.packageId, record.packageVersion); - const disposition = node.failure; - const digested = isDigested(disposition.kind); - - await deps.store.updateInstance(workflowId, nodeId, instance, { failureDigested: digested }); - await deps.events.publish({ - workflowId, - kind: 'failure-disposition', - detail: { nodeId, terminal, disposition: disposition.kind }, - }); - - if (disposition.kind === 'reroute') { - const edge = pkg.edges.find((e) => e.id === disposition.edgeId); - const target = edge === undefined ? undefined : pkg.nodes.find((n) => n.id === edge.to); - if (edge !== undefined && target !== undefined) { - // 失败载荷 = 终态、原因码与说明 - const payload = assemble(edge, scopeOf(record), deps.sfeel.expression); - await enter(record, pkg, target, { ...payload, __terminal: terminal }); - } - } - - if (disposition.kind === 'escalate') { - const target = pkg.nodes.find((n) => n.id === disposition.approvalNodeId); - if (target !== undefined) await enter(record, pkg, target, { __terminal: terminal, __from: nodeId }); - } - - if (disposition.kind === 'tolerate') { - /* - * 容忍 = 这条支线的失败被吸收,**流继续往下走**。 - * 光把实例标成「已消化」而不往下推,这条线就断在这儿 —— 下游谁也不会再被叫起。 - * 载荷为空 —— 容忍分支本就没有产出,下游的必填映射不得依赖它(闭合校验已拦)。 - */ - await advanceFrom(record, pkg, nodeId, {}); - } - - return this.refreshState(workflowId); + return runtime.applyFailure(workflowId, nodeId, instance, node, terminal); }, - /** 审批裁决:通过继续流转;驳回按**声明的走向**回指定节点重开新实例 */ async onApprovalVerdict(workflowId, nodeId, verdict, payload) { - const record = await requireWorkflow(workflowId); - const pkg = await requirePackage(record.packageId, record.packageVersion); - const node = pkg.nodes.find((n) => n.id === nodeId); - if (node?.kind !== 'approval') throw new Error(`节点不是审批点:${nodeId}`); - - const instances = await deps.store.listInstances(workflowId); - const current = instances.filter((i) => i.nodeId === nodeId).at(-1); - if (current !== undefined) { - await deps.store.updateInstance(workflowId, nodeId, current.instance, { - waiting: false, - terminal: 'released', - }); - } - await deps.events.publish({ workflowId, kind: 'approval-verdict', detail: { nodeId, verdict } }); - - /* - * **裁决进载荷**:审批点后面跟网关时,网关判的就是这个字段 - * (如 G3「审批结果」的 `verdict = "pass"` / `verdict = "reject"`)。 - * - * 从前只有 pass 走 `advanceFrom`,而且 payload 里不含 `verdict` —— 于是审批点后面 - * 就算画了网关也判不了(异或分叉的每条出边都得带条件、没有无条件兜底边, - * 判据一缺就全不成立,流无声停在网关上)。要让它判得了,只能在边上挂一枚 - * 常量 `verdict: "pass"` 骗过条件 —— 那网关就成了摆设:条件恒真、驳回那条边永不可达。 - * 图上画着两个去向、其中一个结构上走不到,是在说谎。 - */ - const carried = { ...payload, verdict }; - - if (verdict === 'pass') { - await advanceFrom(record, pkg, nodeId, carried); - } else { - /* - * 驳回:**先看图**。图上显式画了驳回出边(审批点后面跟网关)就走边 —— 编排即定契约, - * 图是真相。没画的包才退回 `reject` 声明:它是 schema 必填的通用机制, - * 不是所有流程包都会把驳回画成边。 - * - * `quiet` 传真:这一步「试边不成」是正常分支,不是死路,不该报 route-dead-end。 - */ - const routedByEdge = await advanceFrom(record, pkg, nodeId, carried, true); - if (!routedByEdge) { - const route = rejectRouteOf(node, scopeOf(record), deps.sfeel.expression); - if (route.kind === 'terminate') { - await deps.store.updateWorkflow(workflowId, { state: 'terminated', finishedAt: deps.clock.now().toISOString() }); - } else { - // 回指定节点**重开新实例**(同节点、新任务 ID、留痕关联),不复活旧任务 - const target = pkg.nodes.find((n) => n.id === route.nodeId); - if (target !== undefined) await enter(record, pkg, target, { ...route.payload, ...payload }); - } - } - } - return this.refreshState(workflowId); + return runtime.onApprovalVerdict(workflowId, nodeId, verdict, payload); }, - /** 工作流状态是聚合投影 —— 每次变动后重算 */ - async refreshState(workflowId): Promise { - const record = await requireWorkflow(workflowId); - if (record.state === 'terminated') return 'terminated'; - const instances = await deps.store.listInstances(workflowId); - // 「完成」凭到过终点,不凭没有在跑的了 —— 所以要把包的终点节点交给聚合 - const pkg = await deps.store.getPackage(record.packageId, record.packageVersion); - const endNodeIds = (pkg?.nodes ?? []).filter((n) => n.kind === 'end').map((n) => n.id); - const state = aggregateWorkflowState(instances as NodeInstance[], false, endNodeIds); - if (state !== record.state) { - await deps.store.updateWorkflow(workflowId, { - state, - ...(state === 'completed' ? { finishedAt: deps.clock.now().toISOString() } : {}), - }); - } - return state; + async refreshState(workflowId) { + return runtime.refreshState(workflowId); }, async terminate(workflowId, reason) { - await deps.store.updateWorkflow(workflowId, { - state: 'terminated', - finishedAt: deps.clock.now().toISOString(), - }); - // 工作流终止级联:在途任务终止、待答卡与征答消息作废、审批点撤卡 - await deps.events.publish({ workflowId, kind: 'workflow-terminated', detail: { reason } }); - return 'terminated'; + return runtime.terminate(workflowId, reason); }, }; } diff --git a/apps/api/src/modules/process/port.ts b/apps/api/src/modules/process/port.ts index e7941a7..e722095 100644 --- a/apps/api/src/modules/process/port.ts +++ b/apps/api/src/modules/process/port.ts @@ -1,7 +1,7 @@ /** * module: process —— 本模块要外界提供的能力(编码规范 §10.1 / §5.3)。 */ -import type { ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts'; +import type { ProcessNode, ProcessPackage, StaffId, TerminalState, WorkflowState } from '@contracts'; import type { NodeInstance } from './internal/domain/owner-resolution.ts'; import type { ActivityContractFace } from './internal/domain/closure.ts'; @@ -30,6 +30,38 @@ export type WorkflowRecord = { readonly finishedAt?: string; }; +/** + * 流程运行时 —— **执行面的可替换切面**(M5-1):按结构调度节点、条件出边选路、 + * 失败走向、审批裁决、状态聚合。自持状态机(internal/runtime/self-hosted-runtime) + * 是默认实现;Mastra 编译产物(workflow-compiler)将是第二实现 —— + * 八例端到端是两个实现的等价性护栏。 + * 定义生命周期(保存/激活/校验)不在此内:那决定「编译哪一份定义」,与引擎选型无关。 + */ +export type FlowRuntime = { + /** 起单后进入起点(起单前置 —— 校验/建单/留痕 —— 归 usecase,不在此内) */ + begin(record: WorkflowRecord, pkg: ProcessPackage): Promise; + onTaskTerminal( + taskId: string, + terminal: TerminalState, + ports: Record>, + ): Promise; + applyFailure( + workflowId: string, + nodeId: string, + instance: number, + node: Extract, + terminal: TerminalState, + ): Promise; + onApprovalVerdict( + workflowId: string, + nodeId: string, + verdict: 'pass' | 'reject', + payload: Record, + ): Promise; + refreshState(workflowId: string): Promise; + terminate(workflowId: string, reason: string): Promise; +}; + export type ProcessStore = { savePackage(pkg: ProcessPackage): Promise; /** 就地写回一版的定义(编辑即其身)—— 撞不存在的版本报错,不静默插入 */ diff --git a/docs/debt.md b/docs/debt.md index 9931dec..da12b08 100644 --- a/docs/debt.md +++ b/docs/debt.md @@ -13,7 +13,7 @@ | M2-1 | 派发器的消费者表为空 —— 投递只标记已送达 | 任务消费者属 task(M4),触发器消费者属 trigger(M5);投递链路本身已可测 | M4 / M5 | 进行中 | | M5-3 | 活动 .JS 产物是源码文本,未经 tsup 单文件 ESM 真编译 | 静态审扫描的正是产物文本(import 与全局引用),当前形态已可审;真编译在装载器落地时接入 | M6 前 | 进行中 | | M5-2 | loop 的 Agent 回路直接经 `ModelGateway`,未用 Mastra `Agent` | 编码规范 §6.1 规定用 Mastra Agent;`AgentPort` 端口不变,换实现时 usecase 不动;当前实现仍过两缝(录制回放可见) | M6 前 | 进行中 | -| M5-1 | `process` 未用 Mastra workflow 编译器,流转由自持确定性状态机 + `sys.node_instance` 承担 | 编码规范 §6.1 规定流程包编译为 Mastra 图;二者语义等价(挂起/恢复由节点实例表承担),先落可测的自持实现,Mastra 编译器作为可替换 adapter 后补 —— usecase 不动 | M5 收尾 | 进行中 | +| M5-1 | `process` 未用 Mastra workflow 编译器 —— **运行时切面已立**(2026-07-28:`FlowRuntime` 端口 + 自持实现整体迁入 `internal/runtime/`,usecase 成薄壳,八例 E2E 为等价性护栏),编译器实现仍欠 | 编码规范 §6.1;接入路线已定(端口化+双实现,回边 reopen 先 spike);冲突清单见 process/CONTRACT.md | 编译器随 spike 结论落地 | 进行中(切面已立) | | M2-2 | pg-boss 已装未用 | 定时作业(监听时限扫描、周期触发器)随 M5 落地;当下派发器用 setInterval | M5 | 进行中 | | M5-4 | AND 并行没有声明机制 —— 网关废除后(2026-07-28)多出边一律异或,画不出并行分叉 | 分支=条件出边、汇合=多入边异或已够 MVP;真要并行时以「并行组」声明加回(对应 Mastra `.parallel` 结构块),与 M5-1 编译器同批设计 | 需要并行的第一条流 | 进行中 | | M5-5 | ~~含网关历史版本的运行时兼容三处~~ | 用户令库只留一条流一个版本(外科手术清空),在途与历史归零 —— schema 的 `kind: 'gateway'`、`joinReady`、汇合等待、`gateway-deprecated` 违规项已于次轮全部摘除 | 2026-07-28 | 已清理 |