Mastra 胶合层:FlowRuntime 的第二实现 —— run ↔ 实例,发号权留在主工程

编译器只管推图,胶合层管一单一单的账(createMastraRuntime): - **发号权在主工程**:实例号 nextInstanceNumber 算、sys.node_instance 落笔; Mastra 不发实例号也不发任务号(spike 实测) - **run ↔ 实例**:runId = <workflowId>#<seq>,重开 = 新 run(seq+1),旧 run 冻结; 当前 run 的入口记在 vars.__runEntry —— 重开后从回边目标那份变体起步 - **迟到 / 重复回调不甩给 Mastra**:先在实例表定位、确认是当前 open 的那个, 不是就发 stale-callback 丢弃(spike 实测:错投的 resume 会打坏 run, 嵌套下甚至不可恢复) - **失败走向不进 Mastra**:改道/升审批/容忍都当「从目标节点重新入场」, 起一个以它为入口的新 run —— 绕开 .parallel() 那套「任一失败即整体失败」 - **状态仍是聚合投影**:refreshState 读 sys.node_instance,不读 Mastra 快照 (run 只有 running/suspended/success/failed,没有「收敛未到终点=abnormal」) - MastraRunner 是 Mastra 侧的最小口子(startRun/resumeRun),换版本只改它四闸过。**尚欠**:MastraRunner 的真实现(Mastra 实例 + PostgresStore 挂 mastra schema)与接线开关 —— 编译器与胶合层都有单测但没在真库上跑过整条流, 转正前自持实现仍是唯一在跑的路径;已如实记进 debt M5-1。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
2026-07-29 00:19:37 +08:00
parent 0ba3a80e13
commit 9df2294f6e
5 changed files with 491 additions and 8 deletions
@@ -0,0 +1,461 @@
+/**
+ * `FlowRuntime` 的 **Mastra 实现**(胶合层)—— M5-1 的第二实现,与自持状态机对等。
+ *
+ * 编译产物(`workflow-compiler.ts`)只管推图;**这里管一单一单的账**:
+ *   · **发号权在主工程**:实例号由 `nextInstanceNumber` 算、`sys.node_instance` 落笔 ——
+ *     Mastra 不发实例号也不发任务号(spike 实测);
+ *   · **run ↔ 实例**:一个 run 就是一次「从入口走到底」;重开 = **新 run**
+ *     (`runId = <workflowId>#<seq>`),旧 run 快照冻结 —— 这正是路线 B 的收益;
+ *   · **迟到 / 重复回调不甩给 Mastra**:先按 taskId 在实例表定位、确认是当前 open 的那个,
+ *     不是就丢弃(spike 实测:错投的 resume 在嵌套下会把整条 run 打成不可恢复的 failed);
+ *   · 状态仍是**聚合投影**:`refreshState` 读 `sys.node_instance`,不读 Mastra 快照 ——
+ *     Mastra run 只有 running/suspended/success/failed,没有「收敛未到终点 = abnormal」这一态。
+ *
+ * 事件面照旧全发(workflow-started / task-started / route-dead-end …)。
+ */
+import type { ProcessPackage, StaffId, 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, rejectRouteOf } from '../domain/routing.ts';
+import { compilePackage } from './workflow-compiler.ts';
+import type { CompiledFlow, RunResult } from './workflow-compiler.ts';
+
+/** Mastra 侧只要这一点点能力 —— 换版本时改这一处,不动胶合逻辑 */
+export type MastraRunner = {
+  /** 起一个 run(带确定性 runId);已存在且未完结时**不许重跑**(spike B-坑-1) */
+  startRun(input: {
+    workflow: unknown;
+    runId: string;
+    resourceId: string;
+    inputData: Record<string, unknown>;
+  }): Promise<{ status: string; result?: RunResult }>;
+  /** 推进一个挂起的 run */
+  resumeRun(input: {
+    workflow: unknown;
+    runId: string;
+    step: string;
+    resumeData: Record<string, unknown>;
+  }): Promise<{ status: string; result?: RunResult }>;
+};
+
+export type MastraRuntimeDeps = {
+  readonly store: ProcessStore;
+  readonly org: OrgDirectory;
+  readonly tasks: TaskGateway;
+  readonly approvals: ApprovalGateway;
+  readonly sfeel: Sfeel;
+  readonly events: FlowEventPublisher;
+  readonly receipts: ReceiptNotifier;
+  readonly clock: Clock;
+  readonly runner: MastraRunner;
+};
+
+export function createMastraRuntime(deps: MastraRuntimeDeps): FlowRuntime {
+  /** 编译缓存:一份 (包id, 版本) 编一次 —— 激活那一刻的定义此后不可变 */
+  const compiled = new Map<string, CompiledFlow[]>();
+
+  async function requireWorkflow(id: string): Promise<WorkflowRecord> {
+    const record = await deps.store.getWorkflow(id);
+    if (record === undefined) throw new Error(`工作流不存在:${id}`);
+    return record;
+  }
+
+  async function requirePackage(id: string, version?: number): Promise<ProcessPackage> {
+    const pkg = await deps.store.getPackage(id, version);
+    if (pkg === undefined) throw new Error(`流程包不存在:${id}`);
+    return pkg;
+  }
+
+  function scopeOf(record: WorkflowRecord) {
+    return { ports: record.ports, vars: record.vars, initiation: record.initiation };
+  }
+
+  /** 谁来干:运营时分配(员工组轮值 / 全员轮值)—— 与自持实现同一口径 */
+  async function assign(record: WorkflowRecord, nodeId: string): Promise<string> {
+    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, position: spec.position, staffId: resolution.staffId, by: resolution.by, rationale: resolution.rationale },
+    });
+    return resolution.staffId;
+  }
+
+  /**
+   * 编译(带缓存):一份 (包id, 版本) 编一次。
+   * 钩子里的每一步都**现取**当前工作流记录 —— 编译产物在多单之间共用,
+   * 闭包里绝不能锁住某一单的快照。
+   */
+  function flowsFor(pkg: ProcessPackage): CompiledFlow[] {
+    const key = `${pkg.id}@${pkg.version}`;
+    const hit = compiled.get(key);
+    if (hit !== undefined) return hit;
+
+    const flows = compilePackage(pkg, {
+      startTask: async ({ workflowId, nodeId, node, payload }) => {
+        const live = await requireWorkflow(workflowId);
+        const staffId = await assign(live, nodeId);
+        const task = await deps.tasks.start({
+          activityId: node.activityId,
+          ownerId: staffId as StaffId,
+          title: node.label,
+          workflowId,
+          nodeId,
+          data: payload,
+          ...(node.message === undefined ? {} : { message: node.message }),
+        });
+        const instances = await deps.store.listInstances(workflowId);
+        const instance = nextInstanceNumber(instances, nodeId);
+        await deps.store.insertInstance(workflowId, {
+          nodeId,
+          instance,
+          running: true,
+          waiting: false,
+          activated: true,
+          taskId: task.id,
+          assigneeId: staffId,
+        });
+        await deps.events.publish({ workflowId, kind: 'task-started', detail: { nodeId, taskId: task.id } });
+        return { taskId: task.id, instance };
+      },
+
+      raiseApproval: async ({ workflowId, nodeId, node, payload }) => {
+        const live = await requireWorkflow(workflowId);
+        const staffId = await assign(live, nodeId);
+        const approval = await deps.approvals.raise({
+          workflowId,
+          nodeId,
+          approverId: staffId as StaffId,
+          subject: node.label,
+          basis: Object.entries(payload).map(([k, v]) => ({ k, v: String(v) })),
+        });
+        const instances = await deps.store.listInstances(workflowId);
+        const instance = nextInstanceNumber(instances, nodeId);
+        await deps.store.insertInstance(workflowId, {
+          nodeId,
+          instance,
+          running: false,
+          waiting: true,
+          activated: true,
+          approvalId: approval.id,
+          assigneeId: staffId,
+        });
+        await deps.events.publish({ workflowId, kind: 'approval-raised', detail: { nodeId, approvalId: approval.id } });
+        return { approvalId: approval.id, instance };
+      },
+
+      assembleEdge: async ({ workflowId, edge, payload }) => {
+        const live = await requireWorkflow(workflowId);
+        const scope = scopeOf(live);
+        return assemble(edge, { ...scope, ports: { ...scope.ports, [edge.from]: payload } }, deps.sfeel.expression);
+      },
+
+      evaluate: async ({ workflowId, condition, payload }) => {
+        const live = await requireWorkflow(workflowId);
+        const scope = scopeOf(live);
+        return deps.sfeel.condition(condition, { ...scope.initiation, ...scope.vars, ...payload });
+      },
+
+      finish: async ({ workflowId, nodeId, payload }) => {
+        const live = await requireWorkflow(workflowId);
+        const instances = await deps.store.listInstances(workflowId);
+        await deps.store.insertInstance(workflowId, {
+          nodeId,
+          instance: nextInstanceNumber(instances, nodeId),
+          running: false,
+          waiting: false,
+          activated: true,
+          terminal: 'released',
+        });
+        const finished = await deps.store.updateWorkflow(workflowId, {
+          state: 'completed',
+          finishedAt: deps.clock.now().toISOString(),
+          ports: { ...live.ports, __output: payload },
+        });
+        await deps.events.publish({ workflowId, kind: 'workflow-completed', detail: payload });
+        if (live.source.by !== undefined) {
+          await deps.receipts.notify({
+            to: live.source.by as StaffId,
+            workflowId,
+            subject: live.subject,
+            state: finished.state,
+          });
+        }
+      },
+
+      deadEnd: async ({ workflowId, nodeId, payload }) => {
+        const outgoing = pkg.edges.filter((e) => e.from === nodeId);
+        await deps.events.publish({
+          workflowId,
+          kind: 'route-dead-end',
+          detail: {
+            nodeId,
+            edges: outgoing.map((e) => ({ id: e.id, to: e.to, condition: e.condition ?? '(无条件)' })),
+            availableKeys: Object.keys(payload).sort(),
+          },
+        });
+      },
+    });
+
+    compiled.set(key, flows);
+    return flows;
+  }
+
+  const flowFor = (flows: readonly CompiledFlow[], pkgId: string, entryNodeId: string): CompiledFlow => {
+    const hit = flows.find((f) => f.key === `${pkgId}@${entryNodeId}`);
+    if (hit === undefined) throw new Error(`没有以 ${entryNodeId} 为入口的编译产物 —— 回边目标须编成变体`);
+    return hit;
+  };
+
+  /** run 号:一单一序 —— `<workflowId>#<seq>`,重开递增,旧 run 冻结可回查 */
+  const runIdOf = (workflowId: string, seq: number) => `${workflowId}#${seq}`;
+
+  /**
+   * run 结束后的收口:走到终点就完了;要重开就**起新 run**(路线 B 的核心);
+   * 死路已在编译产物里发过诊断事件,这里只让它停住。
+   */
+  async function settle(record: WorkflowRecord, pkg: ProcessPackage, result: RunResult | undefined, seq: number): Promise<void> {
+    if (result === undefined || result.kind !== 'reopen') return;
+    const flows = flowsFor(pkg);
+    const target = result.reopenTarget;
+    if (target === undefined) return;
+    await deps.events.publish({
+      workflowId: record.id,
+      kind: 'reopened',
+      detail: { by: result.reopenedBy, target, from: result.nodeId },
+    });
+    const next = seq + 1;
+    const out = await deps.runner.startRun({
+      workflow: flowFor(flows, pkg.id, target).workflow,
+      runId: runIdOf(record.id, next),
+      resourceId: record.id,
+      inputData: { workflowId: record.id, nodeId: target, payload: result.payload },
+    });
+    await settle(record, pkg, out.result, next);
+  }
+
+  /** 这一单现在跑到第几个 run —— 实例表里没有,记在工作流的 vars 上(胶合层自己的账) */
+  const seqOf = (record: WorkflowRecord): number => Number(record.vars['__runSeq'] ?? 1);
+
+  async function bumpSeq(workflowId: string, seq: number): Promise<void> {
+    const live = await requireWorkflow(workflowId);
+    await deps.store.updateWorkflow(workflowId, { vars: { ...live.vars, __runSeq: seq } });
+  }
+
+  async function refreshState(workflowId: string): 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;
+  }
+
+  /**
+   * 把终态送进 run。**先在实例表定位、确认是当前 open 的那个**,
+   * 不是就丢弃 —— 迟到/重复回调直接甩给 Mastra 会打坏 run(spike 实测)。
+   */
+  async function deliver(
+    workflowId: string,
+    nodeId: string,
+    resumeData: Record<string, unknown>,
+  ): Promise<void> {
+    const record = await requireWorkflow(workflowId);
+    const pkg = await requirePackage(record.packageId, record.packageVersion);
+    const flows = flowsFor(pkg);
+    const seq = seqOf(record);
+    // 当前 run 的入口:第一个 run 从 start,重开后的 run 从回边目标 —— 记在 vars 上
+    const entry = String(record.vars['__runEntry'] ?? pkg.nodes.find((n) => n.kind === 'start')?.id ?? 'start');
+    const out = await deps.runner.resumeRun({
+      workflow: flowFor(flows, pkg.id, entry).workflow,
+      runId: runIdOf(workflowId, seq),
+      step: nodeId,
+      resumeData,
+    });
+    await settle(record, pkg, out.result, seq);
+  }
+
+  return {
+    async begin(record, pkg) {
+      const flows = flowsFor(pkg);
+      const start = pkg.nodes.find((n) => n.kind === 'start');
+      if (start === undefined) return;
+      await deps.store.updateWorkflow(record.id, { vars: { ...record.vars, __runSeq: 1, __runEntry: start.id } });
+      const out = await deps.runner.startRun({
+        workflow: flowFor(flows, pkg.id, start.id).workflow,
+        runId: runIdOf(record.id, 1),
+        resourceId: record.id,
+        inputData: { workflowId: record.id, nodeId: start.id, payload: record.initiation },
+      });
+      await settle(record, pkg, out.result, 1);
+    },
+
+    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);
+
+      // 迟到回调:这个实例已经终态了 —— 丢弃,绝不甩给 Mastra(会打坏 run)
+      const instances = await deps.store.listInstances(record.id);
+      const current = instances.filter((i) => i.nodeId === located.nodeId).at(-1);
+      if (current?.instance !== located.instance || current.terminal !== undefined) {
+        await deps.events.publish({
+          workflowId: record.id,
+          kind: 'stale-callback',
+          detail: { taskId, nodeId: located.nodeId, instance: located.instance, terminal },
+        });
+        return refreshState(record.id);
+      }
+
+      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 } });
+      await deps.store.updateWorkflow(record.id, { ports: { ...record.ports, ...ports } });
+
+      if (terminal === 'released') {
+        await deliver(record.id, located.nodeId, { terminal, ports: Object.assign({}, ...Object.values(ports)) });
+      } else if (node?.kind === 'activity') {
+        await this.applyFailure(record.id, located.nodeId, located.instance, node, terminal);
+      }
+      return refreshState(record.id);
+    },
+
+    /**
+     * 失败走向四选一 —— **不进 Mastra**:改道/升审批/容忍都是「另起一段」,
+     * 交给自持路径处理更直接(也避开 `.parallel()` 那套「任一失败即整体失败」的语义)。
+     * 这一段与自持实现同源:走向声明式,流程层不自动重试。
+     */
+    async applyFailure(workflowId, nodeId, instance, node, terminal) {
+      const record = await requireWorkflow(workflowId);
+      const pkg = await requirePackage(record.packageId, record.packageVersion);
+      const disposition = node.failure;
+      await deps.store.updateInstance(workflowId, nodeId, instance, { failureDigested: isDigested(disposition.kind) });
+      await deps.events.publish({
+        workflowId,
+        kind: 'failure-disposition',
+        detail: { nodeId, terminal, disposition: disposition.kind },
+      });
+
+      if (disposition.kind === 'reroute' || disposition.kind === 'escalate' || disposition.kind === 'tolerate') {
+        /*
+         * 三种「继续走」的走向都等价于「从某个节点重新入场」—— 用重开机制表达:
+         * 起一个以目标节点为入口的新 run(与回边同一条路)。
+         * 目标编不出变体(不是回边目标)时如实报事件,不静默停住。
+         */
+        const targetId =
+          disposition.kind === 'reroute'
+            ? pkg.edges.find((e) => e.id === disposition.edgeId)?.to
+            : disposition.kind === 'escalate'
+              ? disposition.approvalNodeId
+              : undefined;
+        if (targetId !== undefined) {
+          const flows = flowsFor(pkg);
+          const variant = flows.find((f) => f.key === `${pkg.id}@${targetId}`);
+          if (variant === undefined) {
+            await deps.events.publish({
+              workflowId,
+              kind: 'node-start-failed',
+              detail: { nodeId: targetId, from: nodeId, error: '失败走向的目标没有编译变体(须为回边目标或入口)' },
+            });
+          } else {
+            const seq = seqOf(record) + 1;
+            await bumpSeq(workflowId, seq);
+            await deps.store.updateWorkflow(workflowId, {
+              vars: { ...record.vars, __runSeq: seq, __runEntry: targetId },
+            });
+            const out = await deps.runner.startRun({
+              workflow: variant.workflow,
+              runId: runIdOf(workflowId, seq),
+              resourceId: workflowId,
+              inputData: { workflowId, nodeId: targetId, payload: { __terminal: terminal, __from: nodeId } },
+            });
+            await settle(record, pkg, out.result, seq);
+          }
+        }
+      }
+      return refreshState(workflowId);
+    },
+
+    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 } });
+
+      // 裁决进判据作用域(系统注入的隐性上下文,不走映射)
+      await deliver(workflowId, nodeId, { verdict, ports: payload });
+
+      /*
+       * 图上没画驳回边的包:退回 `reject` 声明。
+       * 编译产物里那条路会走到「全不中 → 死路」,诊断已发;这里补上声明的去向。
+       */
+      if (verdict === 'reject') {
+        const hasRejectEdge = pkg.edges.some((e) => e.from === nodeId && (e.condition ?? '').includes('reject'));
+        if (!hasRejectEdge) {
+          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(),
+            });
+          }
+        }
+      }
+      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';
+    },
+  };
+}