BPOS-TMP-REPO/spike/mastra-reopen/src/probe-02-route-a-pitfalls.ts

/**
 * 路线 A 的坑(每个坑一个全新的 run,避免互相污染):
 *  1. dowhile 循环体的嵌套快照存在哪?循环第 2 轮会不会覆盖第 1 轮?
 *  2. forEachIndex 能不能定位 dowhile 的第 N 轮?
 *  3. resume 打错 step(或任务回调重复到达)会怎样?
 *
 * 运行:SPIKE_NO_AUTORUN=1 tsx src/probe-02-route-a-pitfalls.ts
 */
import { createWorkflowStateReader } from '@mastra/core/workflows';
import { DecisionScript, Ledger, banner } from './ledger.js';
import { mastraA, storeA, _internals } from './route-a-dowhile.js';
import { resetTrace } from './trace.js';

_internals.setLedger(new Ledger());
_internals.setScript(new DecisionScript([]));
resetTrace();

const wf = mastraA.getWorkflow('routeA');
const snapOf = async (workflowName: string, runId: string) =>
  (await (storeA as any).stores.workflows.loadWorkflowSnapshot({ workflowName, runId })) as
    | Record<string, any>
    | null;

/** 开一个 run 并推进到「第 2 轮 A4 挂起」 */
async function toRound2() {
  const run = await wf.createRun();
  await run.start({ inputData: { quoteId: 'Q-pit' } });
  await run.resume({ step: ['round-body', 'A4'], resumeData: { price: 100 } });
  await run.resume({ step: ['round-body', 'A5'], resumeData: { verdict: 'reject' } });
  return run;
}

banner('坑 1:循环体(嵌套 workflow)的快照存在哪?第 2 轮会不会盖掉第 1 轮?');
{
  const run = await toRound2();
  const outer = await snapOf('route-a', run.runId);
  const inner = await snapOf('round-body', run.runId);
  console.log('外层快照 context 键 =', JSON.stringify(Object.keys(outer?.context ?? {})));
  console.log('外层 context["round-body"].metadata =', JSON.stringify(outer?.context?.['round-body']?.metadata));
  console.log('\n内层(workflowName="round-body", 同一个 runId)快照存在吗?', !!inner);
  console.log('内层 context 键 =', JSON.stringify(Object.keys(inner?.context ?? {})));
  console.log('内层 A4.suspendPayload =', JSON.stringify(inner?.context?.['A4']?.suspendPayload));
  console.log('内层 A4.output        =', JSON.stringify(inner?.context?.['A4']?.output));
  console.log('内层 A5.output        =', JSON.stringify(inner?.context?.['A5']?.output));
  console.log(
    '\n判定:第 1 轮的 A4(taskId=T-A4-001 / price=100)还在内层快照里吗?→',
    JSON.stringify(inner?.context?.['A4']).includes('T-A4-001') ? '在' : '不在(已被第 2 轮覆盖)',
  );
  const runs = await wf.listWorkflowRuns();
  console.log('listWorkflowRuns() 里能看到几条?', runs.total, JSON.stringify(runs.runs.map((r: any) => `${r.workflowName}/${r.runId.slice(0, 8)}`)));
}

banner('坑 2:forEachIndex 能不能定位 dowhile 的第 N 轮?(此刻第 2 轮 A4 正挂起)');
{
  const run = await toRound2();
  for (const idx of [0, 1]) {
    const fresh = await wf.createRun({ runId: `${run.runId}-x${idx}` });
    void fresh; // 只是为了不复用同一个 run
  }
  try {
    const r = await run.resume({
      step: ['round-body', 'A4'],
      resumeData: { price: 999 },
      forEachIndex: 0,
    });
    console.log('resume(step=[round-body,A4], forEachIndex=0) → status =', r.status);
    console.log('  error =', JSON.stringify((r as any).error ?? null));
    const inner = await snapOf('round-body', run.runId);
    console.log('  内层 A4.output.price =', JSON.stringify(inner?.context?.['A4']?.output?.price));
    console.log('  → forEachIndex 被忽略了(打进的是「当前正挂起的那一轮」)');
  } catch (e) {
    console.log('抛异常:', (e as Error).message);
  }
}

banner('坑 3:resume 打错 step(= 旧任务的回调迟到)会怎样?');
{
  const run = await toRound2();
  const st = await wf.getWorkflowRunById(run.runId);
  console.log('resume 之前 status =', st?.status, '挂起的是',
    JSON.stringify(createWorkflowStateReader(st!).getSuspendedStep()?.path));
  let r;
  try {
    r = await run.resume({ step: ['round-body', 'A5'], resumeData: { verdict: 'pass' } });
    console.log('resume(错的 step=A5) → status =', r.status);
    console.log('  error =', JSON.stringify((r as any).error ?? null));
  } catch (e) {
    console.log('resume(错的 step=A5) 抛异常:', (e as Error).message);
  }
  const after = await wf.getWorkflowRunById(run.runId);
  console.log('  之后整个 run 的 status =', after?.status, '  ← 注意:run 被打成 failed 了');
  try {
    const r2 = await run.resume({ step: ['round-body', 'A4'], resumeData: { price: 1 } });
    console.log('  再补一次正确的 resume → status =', r2.status);
  } catch (e) {
    console.log('  再补一次正确的 resume 抛异常:', (e as Error).message, ' ← 这一单救不回来了');
  }
}

banner('坑 4:同一个任务回调重复到达(重复 resume 同一个 step)');
{
  const run = await toRound2();
  const r1 = await run.resume({ step: ['round-body', 'A4'], resumeData: { price: 50 } });
  console.log('第一次 resume A4 → status =', r1.status, '(此时应轮到 A5 挂起)');
  try {
    const r2 = await run.resume({ step: ['round-body', 'A4'], resumeData: { price: 50 } });
    console.log('第二次 resume A4 → status =', r2.status, 'error =', JSON.stringify((r2 as any).error ?? null));
  } catch (e) {
    console.log('第二次 resume A4 抛异常:', (e as Error).message);
  }
  const after = await wf.getWorkflowRunById(run.runId);
  console.log('之后 run 的 status =', after?.status);
}

banner('坑 5:dowhile 的循环体输入/输出必须同构吗?');
console.log('见 route-a-dowhile.ts:循环体 round-body 的 inputSchema 与 outputSchema 都是 RoundState。');
console.log('把 outputSchema 改成别的形状时,TS 报错(dowhile 的 TStepInputSchema extends TPrevSchema 约束)。');