/** * 路线 B 的坑 + 一个额外候选(同 run 内 timeTravel): * 1. 自定义 runId 撞车会怎样?(重开用确定性 runId 时的幂等性) * 2. run 结束后再 resume / 重复 resume 会怎样? * 3. 旧 run 的快照会不会被新 run 影响? * 4. 回边目标如果不是图的入口节点,新 run 怎么从中间开始?—— 试 run.timeTravel() * * 运行:SPIKE_NO_AUTORUN=1 tsx src/probe-04-route-b-pitfalls.ts */ import { createWorkflowStateReader } from '@mastra/core/workflows'; import { DecisionScript, Ledger, banner } from './ledger.js'; import { mastraB, storeB, _internals } from './route-b-external.js'; import { resetTrace } from './trace.js'; _internals.setLedger(new Ledger()); _internals.setScript(new DecisionScript([])); resetTrace(); const wf = mastraB.getWorkflow('routeB'); const snapOf = async (workflowName: string, runId: string) => (await (storeB as any).stores.workflows.loadWorkflowSnapshot({ workflowName, runId })) as | Record | null; const drive = async (run: any, steps: Array<[string[], any]>) => { for (const [path, data] of steps) await run.resume({ step: path, resumeData: data }); }; banner('坑 1:确定性 runId 撞车(重开时用 `${quoteId}-inst-N` 这种可读 ID)'); { const runId = 'dup-run-1'; const r1 = await wf.createRun({ runId }); await r1.start({ inputData: { quoteId: 'Q1', instanceSeq: 1, reopenedBy: null, prevRunId: null } }); console.log('第一次 createRun+start → status =', (await wf.getWorkflowRunById(runId))?.status); const r2 = await wf.createRun({ runId }); console.log('用同一个 runId 再 createRun → 没报错,拿到的是同一个 run?', r2.runId === runId); const st = await wf.getWorkflowRunById(runId); console.log(' 此时 status =', st?.status, ' 挂起的是', JSON.stringify(createWorkflowStateReader(st!).getSuspendedStep()?.path)); try { const rr = await r2.start({ inputData: { quoteId: 'Q1', instanceSeq: 99, reopenedBy: null, prevRunId: null }, }); console.log(' 对已挂起的 run 再 start() → status =', rr.status, 'error=', JSON.stringify((rr as any).error ?? null)); const st2 = await wf.getWorkflowRunById(runId); console.log(' A4.suspendPayload 变了吗?', JSON.stringify((st2?.steps as any)?.A4?.suspendPayload)); } catch (e) { console.log(' 对已挂起的 run 再 start() 抛异常:', (e as Error).message); } } banner('坑 2:run 已经 success 之后再 resume'); { const runId = 'done-run-1'; const run = await wf.createRun({ runId }); await run.start({ inputData: { quoteId: 'Q2', instanceSeq: 1, reopenedBy: null, prevRunId: null } }); await drive(run, [ [['A4'], { price: 100 }], [['A5'], { verdict: 'pass' }], [['A6-sub', 'A6'], { outcome: 'won' }], ]); console.log('status =', (await wf.getWorkflowRunById(runId))?.status); try { const r = await run.resume({ step: ['A6-sub', 'A6'], resumeData: { outcome: 'won' } }); console.log('已完成的 run 再 resume → status =', r.status, JSON.stringify((r as any).error ?? null)); } catch (e) { console.log('已完成的 run 再 resume 抛异常:', (e as Error).message); } console.log('之后 status 还是 =', (await wf.getWorkflowRunById(runId))?.status, ' ← 老 run 没被打坏'); } banner('坑 3:重复 resume 同一个 step(任务回调重复到达)'); { const runId = 'dupresume-1'; const run = await wf.createRun({ runId }); await run.start({ inputData: { quoteId: 'Q3', instanceSeq: 1, reopenedBy: null, prevRunId: null } }); await run.resume({ step: ['A4'], resumeData: { price: 100 } }); try { const r = await run.resume({ step: ['A4'], resumeData: { price: 100 } }); console.log('第二次 resume A4 → status =', r.status, 'error =', JSON.stringify((r as any).error ?? null)); } catch (e) { console.log('第二次 resume A4 抛异常:', (e as Error).message); } console.log('run 状态 =', (await wf.getWorkflowRunById(runId))?.status, ' ← 与路线 A 一样会被打成 failed'); } banner('坑 4:回边目标不是入口节点时,新 run 能不能「从中间开始」?试 run.timeTravel()'); { const runId = 'tt-1'; const run = await wf.createRun({ runId }); await run.start({ inputData: { quoteId: 'Q4', instanceSeq: 1, reopenedBy: null, prevRunId: null } }); await run.resume({ step: ['A4'], resumeData: { price: 100 } }); await run.resume({ step: ['A5'], resumeData: { verdict: 'reject' } }); const st = await wf.getWorkflowRunById(runId); console.log('E10 回边出口跑完,run status =', st?.status, 'result =', JSON.stringify(st?.result)); // 用 timeTravel 在同一个 run 里回到 A5(模拟「回边目标是中间节点」) try { const r = await run.timeTravel({ step: 'A5', inputData: { quoteId: 'Q4', instanceSeq: 2, reopenedBy: 'E10', prevRunId: null, price: 100, verdict: null, finalOutcome: null, }, }); console.log('timeTravel(step=A5) → status =', r.status); const st2 = await wf.getWorkflowRunById(runId); console.log(' A5.suspendPayload =', JSON.stringify((st2?.steps as any)?.A5?.suspendPayload)); console.log(' A4 的记录还在吗? ', JSON.stringify((st2?.steps as any)?.A4?.output)); console.log(' 第一次跑的 result 还在吗?', JSON.stringify(st2?.result)); } catch (e) { console.log('timeTravel 抛异常:', (e as Error).message); } } banner('坑 5:旧 run 的快照会不会被新 run 影响'); { const q = 'Q5'; const r1 = await wf.createRun({ runId: `${q}-inst-1`, resourceId: q }); await r1.start({ inputData: { quoteId: q, instanceSeq: 1, reopenedBy: null, prevRunId: null } }); await r1.resume({ step: ['A4'], resumeData: { price: 100 } }); await r1.resume({ step: ['A5'], resumeData: { verdict: 'reject' } }); const r2 = await wf.createRun({ runId: `${q}-inst-2`, resourceId: q }); await r2.start({ inputData: { quoteId: q, instanceSeq: 2, reopenedBy: 'E10', prevRunId: `${q}-inst-1` } }); await r2.resume({ step: ['A4'], resumeData: { price: 90 } }); await r2.resume({ step: ['A5'], resumeData: { verdict: 'pass' } }); await r2.resume({ step: ['A6-sub', 'A6'], resumeData: { outcome: 'won' } }); for (const id of [`${q}-inst-1`, `${q}-inst-2`]) { const s = await snapOf('route-b', id); console.log( `${id}: status=${s?.status} A4.taskId=${s?.context?.A4?.suspendPayload?.taskId} A4.price=${s?.context?.A4?.output?.price} result=${JSON.stringify(s?.result)}`, ); } const runs = await wf.listWorkflowRuns({ resourceId: q }); console.log('listWorkflowRuns({resourceId}) →', runs.total, JSON.stringify(runs.runs.map((r: any) => r.runId))); _internals.getLedger().dump('probe-04'); }