BPOS-TMP-REPO/spike/mastra-reopen/FINDINGS.md

回边 reopen 在 Mastra workflow 里怎么表达 —— 隔离验证结论

这是一个 spike(隔离验证原型),不属于主工程。所有结论都有运行输出支撑;没跑过的一律写在最后「没验证到的部分」。

• 环境:@mastra/core@1.53.0、zod@3.25.76、Node v24.14.0、pnpm 10.33.0
• 装:cd spike/mastra-reopen && pnpm install --ignore-workspace
• 跑:pnpm route-a / pnpm route-b / pnpm probe:smoke / probe:a-snapshot / probe:a-pitfalls / probe:a-loop-detail / probe:b-pitfalls / probe:misresume / neg:runtime / typecheck

0. 结论先行

推荐路线 B(回跳 = 结束本 run,由外部胶合层用同一份 workflow 定义新起一个 run)。

一句话理由:我们要的「新实例号 + 新任务 ID + 旧实例留痕不动」在路线 B 里是 Mastra 自带的 —— 一个 run 就是一个实例,run 快照天然不可变、可枚举、可回查;路线 A 里第二轮 A4 在 Mastra 眼里就是同一个 step 的重跑,快照被就地覆盖、第一轮痕迹被引擎抹掉(实测),留痕只能靠我们自己另建表。而且:

路线 A 里所有节点都藏在同一个顶层 step(dowhile 循环体)之下,任何一次打错 step 的 resume(迟到回调、重复回调)都会穿透外层校验、在循环体内部炸开,把整条 run 打成 failed 且不可恢复;路线 B 的顶层 step 就是节点本身,外层校验直接拒绝,run 毫发无伤。

1. 验证的拓扑(对齐种子 unified-quote)

start → A4(定价,活动点) → A5(审批点)
                            ├─ E11  verdict="pass"   → A6(议价,活动点)
                            └─ E10  verdict="reject" → A4   (reopen)
                                              A6 ├─ E13 outcome="won"    → end
                                                 └─ E12 outcome="revise" → A4   (reopen)

活动点/审批点在运行时都是「起一个异步任务 → suspend()」,终态由另一条调用链里的 run.resume() 送达。分支用 .branch() + XOR 收口断言。src/ledger.ts 是 sys.node_instance 的替身。

2. 可跑证据

2.1 路线 A:.dowhile() 子图

routeA = createWorkflow({...})
  .then(start)
  .dowhile(roundBody, async ({ inputData, iterationCount }) => inputData.reopen === true)
  .then(end)

roundBody = createWorkflow({ id:'round-body', inputSchema: RoundState, outputSchema: RoundState })
  .then(a4).then(a5)
  .branch([[E11 pass?, a6Sub], [E10 reject?, e10Reopen]])
  .then(pick('pick-A5-out'))

场景 ①(一次过):

#01 start.execute              起单 Q-2026-0001
#03 A4.execute                 起任务 → suspend {"instanceNo":1,"taskId":"T-A4-001"}
#04 [driver]  getSuspendedStep() {"stepId":"round-body","path":["round-body","A4"],...}
#16 cond(E11 pass?)  evaluated {"verdict":"pass","上一步输出可见":"pass","requestContext.tenant":"acme"}
#30 dowhile.condition evaluated {"iterationCount":1,"reopen":false,"round":1,"继续循环?":false}
最终 status = success    result = { rounds: 1, finalOutcome: 'won' }

场景 ②(A5 驳回后再通过):

#18 E10.execute     回边命中,轮次 1 → 2
#20 dowhile.condition evaluated {"iterationCount":1,"reopen":true,"round":2,"继续循环?":true}
#22 A4.execute      起任务 → suspend {"instanceNo":2,"taskId":"T-A4-003"}
最终 status = success   result = { rounds: 2, finalOutcome: 'won' }

── 台账 ② ──
节点  实例  任务ID      轮次   重开自  状态    结论        runId
A4    #1   T-A4-001    it=1   -      closed  price=100   c9bbcaf3
A5    #1   T-A5-002    it=1   -      closed  reject      c9bbcaf3
A4    #2   T-A4-003    it=2   E10    closed  price=90    c9bbcaf3
A5    #2   T-A5-004    it=2   -      closed  pass        c9bbcaf3
A6    #1   T-A6-005    it=2   -      closed  won         c9bbcaf3

三场景全通。suspend/resume 在 dowhile 循环体内(以及循环体里再嵌一层子 workflow 里)完全正常,最深挂起路径 ["round-body","A6-sub","A6"]。

2.2 路线 B:外部胶合

workflow 定义里没有循环,回边落成「回边出口」终止步骤,把 reopen 意图写进 result,run 正常 success 结束;胶合层看到 result.reopen 就 createRun 新 run,runId = ${quoteId}-inst-${seq}、resourceId = quoteId。

#16 E10-reopen-exit.execute  回边命中 → 本 run 到此为止,交给外部胶合层重开 A4
#19 [glue] createRun {"runId":"Q-ikden7-inst-2",...,"inputData":{"instanceSeq":2,"reopenedBy":"E10","prevRunId":"Q-ikden7-inst-1"}}

── run 链 ──
seq=1  runId=Q-ikden7-inst-1  prev=-                reopenedBy=-    reopen(E10) → A4
seq=2  runId=Q-ikden7-inst-2  prev=Q-ikden7-inst-1  reopenedBy=E10  end(won)

── 历史 run 是否还查得到(旧实例留痕不动) ──
Q-ikden7-inst-1: status=success A4.taskId=T-A4-001 A4.price=100 result={"reopen":true,...}
Q-ikden7-inst-2: status=success A4.taskId=T-A4-003 A4.price=90  result={"reopen":false,"finalOutcome":"won"}
listWorkflowRuns({resourceId:'Q-ikden7'}) → total=2

3. 逐条事实答复

Q1「A4 的第二次执行」在 Mastra 眼里是什么?

	路线 A(dowhile)	路线 B(新 run)
身份	同一个 step 的第 2 次迭代;两轮 path 都是 ["round-body","A4"]	新 run 里的新 step 实例
第 1 轮 step 记录	被覆盖/清空	完整保留在上一个 run 的快照里
新实例号来源	只能自己在循环状态里带 round 字段	胶合层递增 instanceSeq,还能编进 runId

路线 B 更自然:「一次重开 = 一个新 run」与主工程「一次重开 = 一个新 node_instance」一一对应。(两条路线的发号动作都得我们自己做 —— Mastra 不发实例号/任务号。)

Q2 suspend/resume 在 dowhile 里能用吗?能定位「第几轮」吗?

能用,但没有办法定位「第几轮」:

1. path 不带轮次:第 1、2 轮的 A4 都是 ["round-body","A4"]。
2. forEachIndex 只对 .foreach() 有效,对 dowhile 是哑参:传 forEachIndex: 0 想指第 1 轮,值仍打进当前轮。
3. resumeLabel 只保留当前这次挂起的标签;传不存在的 label 不报错,静默 resume 当前挂起的 step:

当前 resumeLabels = {"A4-round-2":{"stepId":"round-body"}}
resume(label=A4-round-1) → status = suspended    报错? null
...(终态)内层 A4.output = {..., "price":111}   ← 给 "A4-round-1" 的值打到了第 2 轮身上

Q3 快照里循环轮次怎么表示?

只有一个计数器,没有历史:

[第1轮 A4 挂起]            内层快照 context 的键 = ["input","A4"]
[第1轮已推进,A5 挂起]      内层快照 context 的键 = ["input","A4","A5"]
[第2轮 A4 挂起]            内层快照 context 的键 = ["input","A4"]   ← A5 那一格没了
→ 第 1 轮的 A5(verdict=reject)记录还在吗? 不在(嵌套快照按轮重置了)

轮次唯一痕迹:外层 context['round-body'].metadata = { "iterationCount": 2 }。路线 B 对照:每个 run 快照独立不可变,listWorkflowRuns({resourceId}) 一次列全。

Q4 分支条件什么时候求值?拿得到什么?

时机:源 step 拿到终态后、执行被选中分支前,一次性把该 .branch() 的所有条件都求一遍(非短路)。 能拿到:inputData、getStepResult(step)、getInitData()、外部注入的 requestContext。iterationCount 只有 loop 条件有;条件里不能 suspend。

⚠️ 注入 context 必须是 RequestContext 实例,传普通对象炸:TypeError: requestContext.forEach is not a function。

Q5 .branch() 的输出形状

下一步拿到 { [每一条分支的 stepId]: 输出 | undefined } —— 没命中的分支键也在,值是 undefined。收口步骤必须允许 undefined,XOR 断言自己做。

4. 两条路线各自的坑(全部实测)

路线 A

A-坑-1(决定性):误 resume 把整条 run 打死,不可恢复。

B-1:顶层 step 打错  → 抛异常,run 之后 status = suspended,补正确 resume 仍可继续
A-1:路线 A(嵌套)  → 返回 status = failed;再补正确的 resume 抛 "This workflow run was not suspended" ← 救不回来

机理:Mastra 先校验 path[0] 是不是当前挂起的顶层入口。路线 A 里 path[0] 永远是 round-body(它确实正挂起),校验通过,错误在循环体内部炸,整条 run failed。业务上不是假想:活动点终态由外部调用链送达,迟到/重复/重开后旧任务才回来 —— 在 A 下都是「一次错投 = 报废一单」。

A-坑-2:重复 resume 同一 step 同样打成 failed。 A-坑-3:循环体输入/输出必须同构,TS 拦不住 —— 所有节点被迫共用一个「循环状态对象」schema,与流程包「每条边各自映射」拧着。 A-坑-4:回边只能落在循环体出口 —— 中间节点发出的回边要把「其后所有东西」塞进分支;两条回边目标必须相同才能共用循环体。 A-坑-5:快照不留轮次历史。

路线 B

B-坑-1:确定性 runId 撞车不报错,还会静默重跑 —— 对已挂起的 run 再 start(),Mastra 直接从头重跑、冲掉旧挂起、起第二个任务。胶合层必须先 getWorkflowRunById 判状态。 B-坑-2:重复 resume 抛异常但 run 不受影响(仍 suspended,可继续)。 B-坑-3:已完成的 run 再 resume → 抛错,老 run 保持 success。 B-坑-4:新 run 只能从图的入口起步 —— 目标是中间节点时要编译「以它为入口」的变体;run.timeTravel() 可用但原地改写同一 run 并抹掉原 result,破坏「旧实例留痕不动」,不推荐。 B-坑-5: 每次重开多一次 createRun 写入;run 链靠 prevRunId / resourceId 维持。

共有的坑

• step 的 execute 在 resume 时从头重跑(inputData 不变,多了 resumeData),「起任务」这类副作用必须写在 if (!resumeData) {...} 里。
• requestContext 必须是 RequestContext 实例。
• .branch() 输出含 undefined 键。

5. 推荐:路线 B,理由排序

1. 语义对齐(决定性):1 run = 1 实例,新 runId/新任务 ID/旧快照冻结全是引擎自带;A 里引擎主动覆盖第 1 轮记录,与「旧实例留痕不动」正面冲突。
2. 健壮性(决定性):迟到/重复回调在 A 下打成不可恢复 failed,在 B 下只是抛异常。
3. 编译器复杂度:B 是「一条回边 → 一个终止步骤」,图与流程包 1:1,零图重写。
4. 可观测性:B 每实例一份独立快照。
5. 可迁移性:B 只用 start/suspend/resume+快照三件套。

6. 走路线 B,主工程编译器要额外做什么

1. 回边落成「回边出口步骤」:输出 { reopen:true, reopenedBy:<边号>, reopenTarget:<目标节点>, payload } 写进 result。编译后是 DAG,.commit() 不会因环报错。
2. 条件出边编译成 .branch() + 生成 XOR 收口步骤:无条件边 = 兜底分支;收口步骤过滤未命中键、命中数 ≠ 1 抛错。
3. 活动点/审批点统一编译成「起任务 → suspend」模板:副作用写在 if (!resumeData);suspend(payload) 至少带 { nodeId, instanceNo, taskId };不能依赖 resumeLabel 做校验。
4. 胶合层多做四件事:createRun 前判重;resume 前用 taskId 定位实例并确认是当前 open 的那个(迟到回调丢弃/入死信,绝不甩给 Mastra);run 结束读 result.reopen 起新 run(发号权仍在主工程);refreshState 仍读 sys.node_instance。
5. 回边目标必须是「可作为入口的节点」:非入口就编译「以该节点为入口」的变体(workflowId = <包ID>@<目标节点>)。不要用 timeTravel()。
6. run 与实例的关联落库:加 runId/prev_run_id/reopened_by_edge;resourceId 一律填 workflowId。
7. requestContext 用 RequestContext 实例。

7. 没验证到的部分(不要当结论用)

• 只用了 InMemoryStore,没验证 LibSQL/Postgres 下的快照行为(尤其并发写)。
• 单进程内跑完全部 suspend/resume;跨进程用 createRun({ runId }) 重新取句柄再 resume 这条路没实测。
• 没验证 .dountil();没验证 .foreach() 的 forEachIndex(只验证了它对 dowhile 无效)。
• 没验证多回边指向不同目标节点时路线 A 的可行性(推论,未跑反例)。
• 没验证 restart()/cancel()/事件流与 reopen 的相互作用;性能/写放大未测。