forked from ChaiHao/BPOS-MVP
Mastra 编译器落地:流程包 → workflow(路线 B)—— 回边成出口,变体各带入口
按 spike 实测结论(spike/mastra-reopen/FINDINGS.md)落地 M5-1 的编译器:
- **回边不连回去**,落成回边出口步骤:本 run 到此为止,把
`{reopen, reopenedBy, reopenTarget, payload}` 写进 result,胶合层起新 run。
1 run = 1 实例,与「重开产生新实例、旧实例留痕不动」一一对应
- **每个回边目标各编译一份「以它为入口」的变体**(`<包id>@<节点id>`)——
新 run 只能从入口起步;种子 unified-quote 编出 `@start` 与 `@A4` 两份
- 条件出边 → `.branch()` + **XOR 收口步**:Mastra 把未命中的分支键也放进输出
(值 undefined),收口步过滤并断言恰一条 —— 「恰取一条」运行时照旧成立;
条件谓词排掉「前面已命中」的情形(Mastra 求值非短路),无条件边作兜底,
连兜底都没有时留一条「全不中 → 死路诊断」的出路
- 活动点/审批点 → 「起任务 → suspend()」,副作用写在 `if (!resumeData)` 里
(实测 resume 时 execute 从头重跑,不这么写会起两次任务、旧的成孤儿)
- 编译期全验:悬空边、正向边成环(未标 reopen)当场报 —— 编译只走到得了的边,
悬空边会落在递归碰不到的角落(实测:多默认边情形下 E9 压根没被编译)
- Mastra 泛型桥接收在两个薄工厂里,不把类型逃逸撒满文件
测试:编译器四例(变体/无回边/成环/悬空)+ app 层一例(种子真图编得出来);
五闸全过。胶合层(FlowRuntime 的 Mastra 实现)与 CONTRACT/debt 对账随后。
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KeDBx2PHLavbURXB8LJJ7m
This commit is contained in:
@@ -3,6 +3,7 @@
|
||||
* 审批点两个去向都由图表达、图上每条语义边都画成真边(卡点画法已废)。
|
||||
*/
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { compilePackage } from '@modules/process';
|
||||
import { MVP_PACKAGE } from '../seed-mvp-flow.ts';
|
||||
|
||||
describe('MVP_PACKAGE', () => {
|
||||
@@ -101,3 +102,24 @@ describe('MVP_PACKAGE', () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
/*
|
||||
* 真图能不能编译成 Mastra 图 —— 放在 app 层测:模块测试不引 app(§11.3),
|
||||
* 而这一例要的正是「app 层的种子 × process 模块的编译器」。
|
||||
*/
|
||||
describe('MVP_PACKAGE → Mastra 编译', () => {
|
||||
const hooks = {
|
||||
startTask: async () => ({ taskId: 'T-1', instance: 1 }),
|
||||
raiseApproval: async () => ({ approvalId: 'A-1', instance: 1 }),
|
||||
assembleEdge: async ({ payload }: { payload: Record<string, unknown> }) => payload,
|
||||
evaluate: async () => true,
|
||||
finish: async () => undefined,
|
||||
deadEnd: async () => undefined,
|
||||
};
|
||||
|
||||
it('主入口 + A4 变体(两条回边都指向 A4)', () => {
|
||||
const flows = compilePackage(MVP_PACKAGE, hooks);
|
||||
expect(flows.map((f) => f.key).sort()).toEqual(['unified-quote@A4', 'unified-quote@start']);
|
||||
for (const f of flows) expect(f.workflow).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -94,8 +94,8 @@ describe('装配:功能工具', () => {
|
||||
|
||||
it('工具调用失败如实回报,不抛掉整拍', async () => {
|
||||
const h = harness();
|
||||
h.ports.invoke = async () => ({ ok: false, error: { message: '端点不可达' } });
|
||||
const tools = buildMastraTools([READ_TOOL], h.ports, { taskId: TASK, ownerId: OWNER, humanCallCap: 9, model: 'm' }, h.trace);
|
||||
const failing: ToolBusPorts = { ...h.ports, invoke: async () => ({ ok: false, error: { message: '端点不可达' } }) };
|
||||
const tools = buildMastraTools([READ_TOOL], failing, { taskId: TASK, ownerId: OWNER, humanCallCap: 9, model: 'm' }, h.trace);
|
||||
const out = await run(Object.values(tools)[0], { pn: 'PN-1' });
|
||||
expect(out).toEqual({ ok: false, error: '端点不可达' });
|
||||
});
|
||||
|
||||
@@ -84,7 +84,7 @@
|
||||
- **驳回走向声明式**:审批点的驳回按 `reject` 声明回指定节点重开新实例,不由临场决定
|
||||
- **凡单必有来源**:`source` 必填;终点按输出声明交付、按来源回执
|
||||
|
||||
## M5-1:运行时切面已立(2026-07-28),编译器实现仍欠
|
||||
## M5-1:运行时切面已立 + Mastra 编译器已落(2026-07-28/29)
|
||||
|
||||
「Mastra 编译器作可替换 adapter、usecase 不动」的端口**已经存在**:`port.ts` 的
|
||||
`FlowRuntime`(begin / onTaskTerminal / applyFailure / onApprovalVerdict /
|
||||
@@ -93,11 +93,23 @@ refreshState / terminate)。执行面的语义、留痕与副作用编排整体
|
||||
持久);`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 感知)、
|
||||
**编译器**(`internal/adapter/workflow-compiler.ts`,出口 `compilePackage`)已落,
|
||||
走 spike 定的**路线 B**(证据:`spike/mastra-reopen/FINDINGS.md`,实测):
|
||||
- **回边不连回去**,落成一枚**回边出口步骤** —— 本 run 到此为止,把
|
||||
`{ reopen, reopenedBy, reopenTarget, payload }` 写进 `result`,胶合层据此起**新 run**。
|
||||
理由:①1 run = 1 实例,与「重开产生新实例、旧实例留痕不动」一一对应
|
||||
(`.dowhile` 那条路引擎会**覆盖上一轮记录**);②迟到/重复回调在 `.dowhile` 下
|
||||
把整条 run 打成**不可恢复的 failed**,而活动终态本就来自不受控的外部调用链。
|
||||
- **每个回边目标各编译一份「以它为入口」的变体**(`<包id>@<节点id>`)—— 新 run 只能从入口起步。
|
||||
- 条件出边 → `.branch()` + **XOR 收口步**:Mastra 把未命中的分支键也放进输出
|
||||
(值 undefined),收口步过滤并断言恰一条 —— 「恰取一条」在运行时照旧成立。
|
||||
- 活动点/审批点 → 「起任务 → `suspend()`」,副作用写在 `if (!resumeData)` 里
|
||||
(resume 时 execute **从头重跑**,不这么写会起两次任务)。
|
||||
- 编译期全验:悬空边、正向边成环(未标 reopen)一律当场报 —— 编译只走到得了的边,
|
||||
悬空边可能落在递归碰不到的角落。
|
||||
|
||||
替换路径:`ProcessServiceDeps.runtime` 注入第二实现(胶合层 = `FlowRuntime` 的
|
||||
Mastra 实现,尚欠 —— 见 debt M5-1)。八例端到端是两个实现的**等价性护栏**。
|
||||
仍须绕的语义冲突:`.parallel()` 任一失败即整体失败(step 内自吞,不让 Mastra 感知)、
|
||||
三态聚合投影(abnormal 防假成功,Mastra run 无此态)、`node_instance`
|
||||
投影表必须照写(治理台按行查)。
|
||||
投影表必须照写(治理台按行查、实例发号权仍在主工程)。
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
/**
|
||||
* 流程包 → Mastra workflow 编译器(M5-1)—— 编译期该定死的事。
|
||||
*
|
||||
* 走 spike 定的**路线 B**:回边落成出口步骤(本 run 到此为止,胶合层起新 run),
|
||||
* 编译产物是纯 DAG。这里守三条:
|
||||
* ① 每个回边目标各编译一份「以它为入口」的变体 —— 新 run 只能从入口起步;
|
||||
* ② 正向边成环在**编译期**就报(别等运行时无声打转);
|
||||
* ③ 悬空边编译期全验(编译只走到得了的边,悬空边可能落在编译不到的角落)。
|
||||
* 真图(种子 unified-quote)那一例在 app 层 —— 模块测试不引 app(§11.3)。
|
||||
*/
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { processPackageSchema } from '@contracts';
|
||||
import type { ProcessPackage } from '@contracts';
|
||||
import { compilePackage } from '../internal/adapter/workflow-compiler.ts';
|
||||
import type { CompileHooks } from '../internal/adapter/workflow-compiler.ts';
|
||||
|
||||
const hooks: CompileHooks = {
|
||||
startTask: async () => ({ taskId: 'T-1', instance: 1 }),
|
||||
raiseApproval: async () => ({ approvalId: 'A-1', instance: 1 }),
|
||||
assembleEdge: async ({ payload }) => payload,
|
||||
evaluate: async () => true,
|
||||
finish: async () => undefined,
|
||||
deadEnd: async () => undefined,
|
||||
};
|
||||
|
||||
const pkg = (over: Partial<ProcessPackage> = {}): ProcessPackage =>
|
||||
processPackageSchema.parse({
|
||||
id: 'p1',
|
||||
name: '包',
|
||||
version: 1,
|
||||
state: 'active',
|
||||
callInterface: { input: { type: 'object' }, output: { type: 'object' } },
|
||||
nodes: [
|
||||
{ id: 'start', kind: 'start', label: '起' },
|
||||
{ id: 'A', kind: 'activity', label: '甲', activityId: 'act-a', failure: { kind: 'terminate' } },
|
||||
{ id: 'B', kind: 'approval', label: '审', reject: { kind: 'terminate' } },
|
||||
{ id: 'end', kind: 'end', label: '终' },
|
||||
],
|
||||
edges: [
|
||||
{ id: 'E1', from: 'start', to: 'A', mapping: [] },
|
||||
{ id: 'E2', from: 'A', to: 'B', mapping: [] },
|
||||
{ id: 'E3', from: 'B', to: 'end', condition: 'verdict = "pass"', mapping: [] },
|
||||
{ id: 'E4', from: 'B', to: 'A', condition: 'verdict = "reject"', mapping: [], reopen: true },
|
||||
],
|
||||
variables: [],
|
||||
...over,
|
||||
});
|
||||
|
||||
describe('compilePackage —— 路线 B:回边成出口,变体各带入口', () => {
|
||||
it('主入口一份 + 每个回边目标一份变体', () => {
|
||||
const flows = compilePackage(pkg(), hooks);
|
||||
expect(flows.map((f) => f.key).sort()).toEqual(['p1@A', 'p1@start']);
|
||||
expect(flows.map((f) => f.entryNodeId).sort()).toEqual(['A', 'start']);
|
||||
for (const f of flows) expect(f.workflow).toBeDefined();
|
||||
});
|
||||
|
||||
it('没有回边的包只编一份(主入口)', () => {
|
||||
const linear = pkg({ edges: pkg().edges.filter((e) => e.id !== 'E4').map((e) => (e.id === 'E3' ? { ...e, condition: undefined } : e)) });
|
||||
const flows = compilePackage(linear, hooks);
|
||||
expect(flows).toHaveLength(1);
|
||||
expect(flows[0]!.key).toBe('p1@start');
|
||||
});
|
||||
|
||||
/* 回边标了 reopen 才是回边;没标就是正向成环 —— 运行时会无声打转,编译期必须拦 */
|
||||
it('正向边成环:编译期就报,不放过', () => {
|
||||
const cyclic = pkg({
|
||||
edges: pkg().edges.map((e) => (e.id === 'E4' ? { ...e, reopen: false } : e)),
|
||||
});
|
||||
expect(() => compilePackage(cyclic, hooks)).toThrow(/成环/);
|
||||
});
|
||||
|
||||
it('边指向不存在的节点:编译期就报', () => {
|
||||
const dangling = pkg({
|
||||
edges: [...pkg().edges, { id: 'E9', from: 'A', to: 'ghost', mapping: [], reopen: false } as never],
|
||||
});
|
||||
expect(() => compilePackage(dangling, hooks)).toThrow(/不存在/);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -58,3 +58,7 @@ export {
|
||||
|
||||
export { createProcessStore } from './internal/adapter/process-store.ts';
|
||||
export { nodeInstanceTable, processPackageTable, workflowTable } from './internal/adapter/process-schema.ts';
|
||||
|
||||
/* Mastra 编译器(M5-1,路线 B:回边成出口、胶合层起新 run)—— adapter 出口 */
|
||||
export type { CompiledFlow, CompileHooks, RunResult } from './internal/adapter/workflow-compiler.ts';
|
||||
export { compilePackage } from './internal/adapter/workflow-compiler.ts';
|
||||
|
||||
@@ -0,0 +1,351 @@
|
||||
/**
|
||||
* 流程包 → Mastra workflow 编译器(编码规范 §6.1;M5-1;`@mastra` 只准 adapter/app §6.2)。
|
||||
*
|
||||
* **走 spike 定的路线 B**(`spike/mastra-reopen/FINDINGS.md`):
|
||||
* 回边不连回去,落成一枚**回边出口步骤** —— 本 run 到此为止,把重开意图写进 `result`,
|
||||
* 由胶合层(`mastra-runtime.ts`)起一个**新 run**。理由是实测出来的两条:
|
||||
* ① 1 run = 1 实例,与「重开产生新实例、旧实例留痕不动」一一对应;
|
||||
* `.dowhile` 那条路里引擎会**覆盖上一轮记录**(第 2 轮开始时上一轮的步结果直接消失);
|
||||
* ② 迟到 / 重复回调在 `.dowhile` 下会把整条 run 打成**不可恢复的 failed**,
|
||||
* 而我们的活动终态本来就来自不受控的外部调用链。
|
||||
*
|
||||
* 编译产物是**纯 DAG**:回边成了出口,`.commit()` 不会因环报错。
|
||||
* 每个回边目标各编译一份「以它为入口」的变体(`<包id>@<节点id>`)—— 新 run 只能从入口起步。
|
||||
*
|
||||
* 节点 → step 的形状(三类):
|
||||
* · 活动点 / 审批点:`起任务 → suspend()`;副作用写在 `if (!resumeData)` 里
|
||||
* (resume 时 execute 会**从头重跑**,不这么写就会起两次任务);
|
||||
* · 终点:收束,写 `{ finished: true }`;
|
||||
* · 回边出口:写 `{ reopen: true, reopenedBy, reopenTarget, payload }`。
|
||||
*
|
||||
* 条件出边 → `.branch()` + 一枚 **XOR 收口步**:Mastra 把未命中的分支键也放进输出
|
||||
* (值为 undefined),收口步过滤它们并断言恰好命中一条 —— `selectEdges` 的
|
||||
* 「恰取一条」在运行时照旧成立。
|
||||
*/
|
||||
import { createStep, createWorkflow } from '@mastra/core/workflows';
|
||||
import { z } from 'zod';
|
||||
|
||||
/*
|
||||
* Mastra 的 Step/Workflow 泛型极深,在 `exactOptionalPropertyTypes` 下与我们的
|
||||
* schema 推断对不齐。**桥接收在这一处**:编译器内部一律走这两个薄工厂,
|
||||
* 别把 `as never` 撒得满文件都是(§12.5 禁 any,这里是类型桥不是逃逸)。
|
||||
*/
|
||||
type FlowBuilder = {
|
||||
then: (s: unknown) => FlowBuilder;
|
||||
branch: (pairs: unknown) => FlowBuilder;
|
||||
commit: () => unknown;
|
||||
};
|
||||
const step = (config: Record<string, unknown>): unknown => createStep(config as never);
|
||||
const flow = (config: Record<string, unknown>): FlowBuilder => createWorkflow(config as never) as never;
|
||||
import type { Edge, ProcessNode, ProcessPackage } from '@contracts';
|
||||
|
||||
/** 编译期给运行时的钩子 —— 编译产物不认识任何模块,只认这几个动作 */
|
||||
export type CompileHooks = {
|
||||
/** 起一个活动任务;回 `{ taskId, instance }` 供 suspend 载荷带走 */
|
||||
readonly startTask: (input: {
|
||||
workflowId: string;
|
||||
nodeId: string;
|
||||
node: Extract<ProcessNode, { kind: 'activity' }>;
|
||||
payload: Record<string, unknown>;
|
||||
}) => Promise<{ taskId: string; instance: number }>;
|
||||
/** 卡一个审批到人的工位 */
|
||||
readonly raiseApproval: (input: {
|
||||
workflowId: string;
|
||||
nodeId: string;
|
||||
node: Extract<ProcessNode, { kind: 'approval' }>;
|
||||
payload: Record<string, unknown>;
|
||||
}) => Promise<{ approvalId: string; instance: number }>;
|
||||
/** 边映射装配:`assemble` 的外包 —— 编译产物不认识 sfeel,也不认识作用域怎么拼 */
|
||||
readonly assembleEdge: (input: { workflowId: string; edge: Edge; payload: Record<string, unknown> }) => Promise<Record<string, unknown>>;
|
||||
/** 条件求值:`@sfeel` 的外包(前后端同源那一份) */
|
||||
readonly evaluate: (input: { workflowId: string; condition: string; payload: Record<string, unknown> }) => Promise<boolean>;
|
||||
/** 终点收束 */
|
||||
readonly finish: (input: { workflowId: string; nodeId: string; payload: Record<string, unknown> }) => Promise<void>;
|
||||
/** 死路:一条边都没中而该节点确有出边 —— 诊断照旧发事件 */
|
||||
readonly deadEnd: (input: { workflowId: string; nodeId: string; payload: Record<string, unknown> }) => Promise<void>;
|
||||
};
|
||||
|
||||
/** run 里流动的东西:一份载荷 + 这一单是谁 */
|
||||
const FLOW_STATE = z.object({
|
||||
workflowId: z.string(),
|
||||
nodeId: z.string(),
|
||||
payload: z.record(z.string(), z.unknown()),
|
||||
});
|
||||
type FlowState = z.infer<typeof FLOW_STATE>;
|
||||
|
||||
/** run 的终局:走到终点 / 要重开 / 死路停住 —— 胶合层读它决定下一步 */
|
||||
const RUN_RESULT = z.object({
|
||||
kind: z.enum(['finished', 'reopen', 'dead-end']),
|
||||
nodeId: z.string(),
|
||||
reopenedBy: z.string().optional(),
|
||||
reopenTarget: z.string().optional(),
|
||||
payload: z.record(z.string(), z.unknown()),
|
||||
});
|
||||
export type RunResult = z.infer<typeof RUN_RESULT>;
|
||||
|
||||
/** 编译产物 */
|
||||
export type CompiledFlow = {
|
||||
/** 变体键:`<包id>@<入口节点>` */
|
||||
readonly key: string;
|
||||
readonly entryNodeId: string;
|
||||
/** Mastra workflow(已 commit);类型体操收在编译器内部,外面按 unknown 传给 mastra 实例 */
|
||||
readonly workflow: unknown;
|
||||
};
|
||||
|
||||
/** Mastra 的 id 不收点号等怪字符 —— 与工具总线同一条规矩 */
|
||||
const idOf = (raw: string): string => raw.replace(/[^a-zA-Z0-9_-]/g, '_');
|
||||
|
||||
/**
|
||||
* 编译一份流程包。
|
||||
*
|
||||
* 返回**多份**产物:主入口(start)一份,外加每个回边目标各一份变体 ——
|
||||
* 新 run 只能从入口起步,回边目标是中间节点时就得有一份「以它为入口」的图。
|
||||
*/
|
||||
export function compilePackage(pkg: ProcessPackage, hooks: CompileHooks): CompiledFlow[] {
|
||||
/*
|
||||
* 先验全图,再编译:编译只走**到得了的边**,悬空边可能落在编译不到的角落
|
||||
* (实测:多出边里那条没被选成默认的悬空边,递归压根不会碰它)。
|
||||
* 一条指向虚空的边就是坏数据,编译期一次全验掉,别留到运行时才炸。
|
||||
*/
|
||||
const ids = new Set(pkg.nodes.map((n) => n.id as string));
|
||||
for (const edge of pkg.edges) {
|
||||
if (!ids.has(edge.from)) throw new Error(`编译失败:边 ${edge.id} 的起点不存在(${edge.from})`);
|
||||
if (!ids.has(edge.to)) throw new Error(`编译失败:边 ${edge.id} 的终点不存在(${edge.to})`);
|
||||
}
|
||||
|
||||
const entries = new Set<string>();
|
||||
const start = pkg.nodes.find((n) => n.kind === 'start');
|
||||
if (start !== undefined) entries.add(start.id);
|
||||
for (const edge of pkg.edges) if (edge.reopen === true) entries.add(edge.to);
|
||||
|
||||
return [...entries].map((entryNodeId) => ({
|
||||
key: `${pkg.id}@${entryNodeId}`,
|
||||
entryNodeId,
|
||||
workflow: compileFrom(pkg, entryNodeId, hooks),
|
||||
}));
|
||||
}
|
||||
|
||||
function compileFrom(pkg: ProcessPackage, entryNodeId: string, hooks: CompileHooks): unknown {
|
||||
const byId = new Map(pkg.nodes.map((n) => [n.id as string, n]));
|
||||
const outgoing = (nodeId: string) => pkg.edges.filter((e) => e.from === nodeId);
|
||||
|
||||
/**
|
||||
* 节点 → step。**去重复用**:同一个节点在图里可能被多条路径指到,
|
||||
* Mastra 要求同一 step 实例只出现一次,不然 id 撞车。
|
||||
*/
|
||||
const built = new Map<string, unknown>();
|
||||
|
||||
const stepOf = (nodeId: string): unknown => {
|
||||
const cached = built.get(nodeId);
|
||||
if (cached !== undefined) return cached;
|
||||
const node = byId.get(nodeId);
|
||||
if (node === undefined) throw new Error(`编译失败:边指向不存在的节点 ${nodeId}`);
|
||||
|
||||
const built_ = step({
|
||||
id: idOf(nodeId),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: FLOW_STATE,
|
||||
resumeSchema: z.object({ terminal: z.string().optional(), verdict: z.string().optional(), ports: z.record(z.string(), z.unknown()).optional() }),
|
||||
suspendSchema: z.object({ nodeId: z.string(), instance: z.number(), taskId: z.string().optional(), approvalId: z.string().optional() }),
|
||||
execute: async (inputData: unknown, context: unknown) => {
|
||||
const state = inputData as FlowState;
|
||||
const ctx = context as {
|
||||
workflow?: {
|
||||
suspend: (payload: Record<string, unknown>) => Promise<unknown>;
|
||||
resumeData?: { terminal?: string; verdict?: string; ports?: Record<string, unknown> };
|
||||
};
|
||||
};
|
||||
const resumeData = ctx.workflow?.resumeData;
|
||||
|
||||
/*
|
||||
* **副作用只在第一次进来时做**:resume 时 execute 会从头重跑(实测),
|
||||
* 不这么写会起第二个任务、第二张审批单,旧的成孤儿。
|
||||
*/
|
||||
if (node.kind === 'activity' || node.kind === 'approval') {
|
||||
if (resumeData === undefined) {
|
||||
const raised =
|
||||
node.kind === 'activity'
|
||||
? await hooks.startTask({ workflowId: state.workflowId, nodeId, node, payload: state.payload })
|
||||
: await hooks.raiseApproval({ workflowId: state.workflowId, nodeId, node, payload: state.payload });
|
||||
return ctx.workflow!.suspend({
|
||||
nodeId,
|
||||
instance: raised.instance,
|
||||
...('taskId' in raised ? { taskId: raised.taskId } : { approvalId: raised.approvalId }),
|
||||
}) as never;
|
||||
}
|
||||
// 终态回来了:裁决与各口载荷进判据作用域(审批 verdict 是系统注入的隐性上下文)
|
||||
return {
|
||||
workflowId: state.workflowId,
|
||||
nodeId,
|
||||
payload: {
|
||||
...state.payload,
|
||||
...(resumeData.ports ?? {}),
|
||||
...(resumeData.verdict === undefined ? {} : { verdict: resumeData.verdict }),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// 起点是结构件:不建实例、不挂起,原样过
|
||||
return { workflowId: state.workflowId, nodeId, payload: state.payload };
|
||||
},
|
||||
});
|
||||
built.set(nodeId, built_);
|
||||
return built_;
|
||||
};
|
||||
|
||||
/** 终点:收束,写终局 */
|
||||
const endStep = (nodeId: string) =>
|
||||
step({
|
||||
id: idOf(`${nodeId}__end`),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: RUN_RESULT,
|
||||
execute: async (inputData: unknown) => {
|
||||
const state = inputData as FlowState;
|
||||
await hooks.finish({ workflowId: state.workflowId, nodeId, payload: state.payload });
|
||||
return { kind: 'finished' as const, nodeId, payload: state.payload };
|
||||
},
|
||||
});
|
||||
|
||||
/** 回边出口:本 run 到此为止,把重开意图交给胶合层 */
|
||||
const reopenExit = (edge: Edge) =>
|
||||
step({
|
||||
id: idOf(`${edge.id}__reopen`),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: RUN_RESULT,
|
||||
execute: async (inputData: unknown) => {
|
||||
const state = inputData as FlowState;
|
||||
const payload = await hooks.assembleEdge({ workflowId: state.workflowId, edge, payload: state.payload });
|
||||
return { kind: 'reopen' as const, nodeId: edge.from, reopenedBy: edge.id, reopenTarget: edge.to, payload };
|
||||
},
|
||||
});
|
||||
|
||||
/** 死路:一条边都没中而该节点确有出边 —— 诊断留痕后停住 */
|
||||
const deadEndStep = (nodeId: string) =>
|
||||
step({
|
||||
id: idOf(`${nodeId}__dead-end`),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: RUN_RESULT,
|
||||
execute: async (inputData: unknown) => {
|
||||
const state = inputData as FlowState;
|
||||
await hooks.deadEnd({ workflowId: state.workflowId, nodeId, payload: state.payload });
|
||||
return { kind: 'dead-end' as const, nodeId, payload: state.payload };
|
||||
},
|
||||
});
|
||||
|
||||
/** 边 → 装配步:把上游载荷按边映射装配成下游入参 */
|
||||
const mapStep = (edge: Edge) =>
|
||||
step({
|
||||
id: idOf(`${edge.id}__map`),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: FLOW_STATE,
|
||||
execute: async (inputData: unknown) => {
|
||||
const state = inputData as FlowState;
|
||||
const payload = await hooks.assembleEdge({ workflowId: state.workflowId, edge, payload: state.payload });
|
||||
return { workflowId: state.workflowId, nodeId: edge.to, payload };
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* 从一个节点往下编译成一条链。递归、**带访问集防环**:
|
||||
* 正向边不该成环(reopen 已经被摘成出口),但数据来自库,防一手不吃亏。
|
||||
*/
|
||||
const chainFrom = (nodeId: string, visited: ReadonlySet<string>): unknown => {
|
||||
const node = byId.get(nodeId);
|
||||
if (node === undefined) throw new Error(`编译失败:节点不存在 ${nodeId}`);
|
||||
|
||||
const chain = flow({
|
||||
id: idOf(`${pkg.id}__${entryNodeId}__from__${nodeId}`),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: RUN_RESULT,
|
||||
});
|
||||
|
||||
if (node.kind === 'end') return chain.then(endStep(nodeId)).commit() as never;
|
||||
|
||||
const edges = outgoing(nodeId);
|
||||
if (edges.length === 0) return chain.then(deadEndStep(nodeId)).commit() as never;
|
||||
|
||||
const nextVisited = new Set([...visited, nodeId]);
|
||||
|
||||
/** 一条边 → 「装配 → 下游子链 / 回边出口」 */
|
||||
const branchOf = (edge: Edge): unknown => {
|
||||
const sub = flow({
|
||||
id: idOf(`${pkg.id}__${entryNodeId}__edge__${edge.id}`),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: RUN_RESULT,
|
||||
});
|
||||
if (edge.reopen === true) return sub.then(reopenExit(edge)).commit() as never;
|
||||
if (nextVisited.has(edge.to)) {
|
||||
// 正向边成环:数据坏了,编译期就报 —— 别等运行时无声打转
|
||||
throw new Error(`编译失败:正向边成环(${edge.id}:${edge.from} → ${edge.to});回边须标 reopen`);
|
||||
}
|
||||
return sub.then(mapStep(edge)).then(chainFrom(edge.to, nextVisited)).commit() as never;
|
||||
};
|
||||
|
||||
// 单出边:直连,不必上分支
|
||||
if (edges.length === 1) {
|
||||
const only = edges[0]!;
|
||||
return chain.then(stepOf(nodeId)).then(branchOf(only)).commit() as never;
|
||||
}
|
||||
|
||||
/*
|
||||
* 多出边 = 异或分支:条件命中优先(按声明序),无条件边作默认兜底。
|
||||
* Mastra 的 `.branch()` 把**所有**分支条件求一遍(非短路),所以「恰取一条」
|
||||
* 要自己保证:条件边的谓词里排掉「前面已有条件命中」的情形,默认边的谓词是
|
||||
* 「所有条件都不中」。收口步再断言一次命中数。
|
||||
*/
|
||||
const conditional = edges.filter((e) => e.condition !== undefined && e.condition.trim() !== '');
|
||||
const fallback = edges.find((e) => e.condition === undefined || e.condition.trim() === '');
|
||||
|
||||
const hitAny = async (state: FlowState, upto: number): Promise<boolean> => {
|
||||
for (const e of conditional.slice(0, upto)) {
|
||||
if (await hooks.evaluate({ workflowId: state.workflowId, condition: e.condition!, payload: state.payload })) return true;
|
||||
}
|
||||
return false;
|
||||
};
|
||||
|
||||
const pairs: Array<[(args: { inputData: FlowState }) => Promise<boolean>, unknown]> = [];
|
||||
conditional.forEach((edge, index) => {
|
||||
pairs.push([
|
||||
async ({ inputData }) =>
|
||||
!(await hitAny(inputData, index)) &&
|
||||
(await hooks.evaluate({ workflowId: inputData.workflowId, condition: edge.condition!, payload: inputData.payload })),
|
||||
branchOf(edge),
|
||||
]);
|
||||
});
|
||||
if (fallback !== undefined) {
|
||||
pairs.push([async ({ inputData }) => !(await hitAny(inputData, conditional.length)), branchOf(fallback)]);
|
||||
}
|
||||
// 没有兜底边时也要有一条「全不中」的出路 —— 否则 Mastra 会卡在无分支命中
|
||||
pairs.push([
|
||||
async ({ inputData }) =>
|
||||
!(await hitAny(inputData, conditional.length)) && fallback === undefined,
|
||||
flow({
|
||||
id: idOf(`${pkg.id}__${entryNodeId}__${nodeId}__nohit`),
|
||||
inputSchema: FLOW_STATE,
|
||||
outputSchema: RUN_RESULT,
|
||||
})
|
||||
.then(deadEndStep(nodeId))
|
||||
.commit() as never,
|
||||
]);
|
||||
|
||||
/** XOR 收口:未命中的分支键也在(值 undefined)—— 过滤掉它们,断言恰一条 */
|
||||
const settle = step({
|
||||
id: idOf(`${nodeId}__settle`),
|
||||
inputSchema: z.record(z.string(), RUN_RESULT.optional()),
|
||||
outputSchema: RUN_RESULT,
|
||||
execute: async (inputData: unknown) => {
|
||||
const hits = Object.values((inputData ?? {}) as Record<string, RunResult | undefined>).filter(
|
||||
(v): v is RunResult => v !== undefined,
|
||||
);
|
||||
if (hits.length !== 1) {
|
||||
throw new Error(`异或分支恰取一条被破坏:节点 ${nodeId} 命中 ${hits.length} 条`);
|
||||
}
|
||||
return hits[0]!;
|
||||
},
|
||||
});
|
||||
|
||||
return chain.then(stepOf(nodeId)).branch(pairs).then(settle).commit() as never;
|
||||
};
|
||||
|
||||
return chainFrom(entryNodeId, new Set()) as never;
|
||||
}
|
||||
Reference in New Issue
Block a user