Mastra 化阶段1:抽 FlowRuntime 端口 —— 编排层执行面成为可替换切面,解耦完全成形

目标(用户定):完全实现编排层与活动解耦。解耦线(TaskGateway 起活动 / 终态回报)
本就存在;缺的是 CONTRACT M5-1 那句「Mastra 编译器作可替换 adapter、usecase 不动」
的端口 —— 勘察坐实它此前不存在,状态机直接写在 usecase 里揉着五端口副作用编排。

- port.ts 立 FlowRuntime 端口:begin / onTaskTerminal / applyFailure /
  onApprovalVerdict / refreshState / terminate —— 执行面的全部
- 状态机整体迁入 internal/runtime/self-hosted-runtime.ts(逐字搬迁,行为零变化):
  enter(调度+组轮值分派+实例落 assignee)、advanceFrom(条件出边选路+死路诊断+
  分支不连坐)、失败走向四选一、审批裁决(边优先、声明兜底)、三态聚合投影、终止级联
- process-service 573→248 行薄壳:定义生命周期(保存/激活/校验)+ 起单前置
  (取生效版/入参校验/工作组快照/建单留痕)+ 五方法委托;
  ProcessServiceDeps.runtime 可注入 —— Mastra 编译产物从这里换入,路由/接线零动
- index 出口 FlowRuntime 型;CONTRACT.md 记切面与已核实的语义冲突清单
  (reopen 无原语待 spike / parallel 失败连坐须 step 内自吞 / abnormal 三态 /
  node_instance 投影必留);debt M5-1 → 切面已立

等价性护栏:process 49 单测 + 八例端到端真库,提取前后原样全绿 —— 行为零变化实证。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m

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<string, unknown>,
-  ): Promise<void> {
-    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<string, { staffId: string; name: string; openTasks: number }>();
-    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<string, unknown>,
-    quiet = false,
-  ): Promise<boolean> {
-    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<string, Record<string, unknown>>) {
-      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<string, unknown> = 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<WorkflowState> {
-      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);
     },
   };
 }