崩溃之后:Checkpoint、重放与「恰好一次」

系列第 3 篇,共 8 篇。

持久化最容易被误解的一点,是把它当成「定期存盘」。存盘只能恢复数据,不能恢复正在做的事。如果进程死在「支付已经扣款、结果还没写回」之间,下一次启动时你只知道有一个未完成的任务,却不知道那笔钱到底扣了没有。

Pi Durable 的做法是把每个任务做成显式状态机:每一步先保存「我要做什么」,执行完再保存「结果是什么」。进程死在任何位置,下一个进程都能算出唯一合理的下一步。

一、崩溃只可能发生在两个 checkpoint 之间

一个任务的状态由两部分组成:

  • checkpoint:下一步要做什么,例如 { phase: "charge", key: "payment-42" };
  • memo:这一步已经确认过的事实,例如「tick 2 已经打印」。

任务运行时,runtime.commit() 把新的 checkpoint 或最终 outcome 与本次回调里写的其他一切(documents、entries)在同一次提交中落盘。因此崩溃窗口只有一段:

flowchart LR
    C1["checkpoint N 已提交"] --> EXEC["执行副作用"]
    EXEC --> C2["checkpoint N+1 / outcome 已提交"]
    C1 -.->|"崩溃窗口"| C2

如果崩溃发生在 EXEC 之前,重跑是安全的:副作用根本没发生。 如果发生在 EXEC 之后、C2 之前,重跑会再次执行副作用——这就是为什么这一步必须幂等。

任务的生命周期:

stateDiagram-v2
    [*] --> pending
    pending --> running: resume / waitForTask
    running --> waiting: 等待子任务
    waiting --> running: 子任务完成
    running --> completing: outcome 已确定
    completing --> terminal: 拥有的工作全部结束
    running --> terminal: 无子任务,直接完成
    running --> faulted: phase 抛异常
    faulted --> running: 重启后再次调度
    waiting --> aborted: abort
    completing --> aborted: 拥有的工作失败
    terminal --> [*]
    aborted --> [*]

completing 是一个容易被忽略的状态:结果已经确定,但任务还在等它拥有的子任务收尾。只有子任务全部结束,waitForTask() 才会返回。

二、恢复时的决策树

新进程打开同一个存储后,看到的是「未完成的任务 + 最后一个 checkpoint」。接下来怎么走,取决于三件事:崩溃点、工具的重放声明、以及是否存在 memo。

flowchart TD
    A["重启,发现未完成任务"] --> B{"有 checkpoint?"}
    B -- 否 --> C["从 initial 状态开始"]
    B -- 是 --> D{"崩溃发生在副作用之前?"}
    D -- 是 --> E["重跑当前阶段"]
    D -- 否 --> F{"该步骤声明了 replay: safe?"}
    F -- 是 --> E
    F -- 否 --> G["向模型报告 interrupted<br/>附上已提交的输出"]
    E --> H["memo 去重"]
    G --> H
    H --> I["继续下一阶段"]

三个分支对应三种语义:

情况处理方式谁来兜底
副作用未发生直接重跑框架
副作用已发生,且步骤声明可安全重跑重跑工具作者
副作用已发生,且无法安全重跑报告 interrupted,由模型决定模型 / 业务

注意最后一条:框架不会盲目重跑一个没有声明 replay: "safe" 的工具。它会把调用标记为中断,并把已经产生的那部分输出交给模型。模型可以换个方式重试,也可以向用户解释。

三、memo:让重跑不重复产生可见效果

runtime.memo(name, value) 在任务内保存一个小值,规则是先写者胜。它的典型用途是记录「某个可见副作用已经发生」:

tick: async (task, runtime, taskContext) => {
    const n = task.state.checkpoint.n;
    if ((await runtime.memo(`printed-${n}`, taskContext)) === undefined) {
        await runtime.memo(`printed-${n}`, true, taskContext);
        console.log(`tick ${n}`);
    }
    await runtime.commit(
        () =>
            n === task.input.to
                ? { status: "terminal", outcome: { status: "completed", result: `counted to ${n}` } }
                : { status: "running", checkpoint: { phase: "tick", n: n + 1 } },
        taskContext,
    );
}

这段代码的顺序值得注意:先写 memo,再做副作用。如果反过来,崩溃发生在两者之间,重跑时 memo 还不存在,副作用会执行第二次。

13-recovery.ts 完整演示了这个场景:第一次运行打印到 tick 2,在保存下一个 checkpoint 之前关闭 Harness;重新打开后,tick 2 再次进入阶段函数,但 memo 说它已经打印过,于是只继续 tick 3。

// 第一次运行:打印到 tick 2 后关闭
const firstRun = await open();
const tickerId = await (await firstRun.root(context)).commit(
    (tx) => tx.createTask(Ticker, { to: 5 }, { ownership: { kind: "conversation" } }),
    context,
);
await tickTwo;
await firstRun.close(context);

// 第二次运行:同一个任务 ID,从最后一个 checkpoint 继续
const secondRun = await open();
const counted = await secondRun.waitForTask(tickerId, context);
console.log("after reopen:", counted.state.outcome);

整个恢复过程没有「手动重启任务」这一步。打开存储、等待任务,调度器就会接着跑。

四、requestId:提交入口的恰好一次

任务内部的 memo 解决「步骤重复执行」,但客户端重试还需要另一层保证:同一条用户输入不能被处理两次。

Pi Durable 把幂等键放在提交入口上,而不是模型调用上:

const submission = await root.submit({
    type: "input",
    content: "Hello",
    requestId: "greeting-1",
}, context);

// 重启之后,同一个 requestId 找到同一个 submission
const again = await root.submit({
    type: "input",
    content: "Hello",
    requestId: "greeting-1",
}, context);

// again.id === submission.id

语义如下:

sequenceDiagram
    autonumber
    participant C as 客户端
    participant H as Harness
    participant S as Storage

    C->>H: submit(requestId = "job-42")
    H->>S: 查找 requestId
    alt 已存在
        S-->>H: 返回既有 submission
        H-->>C: 同一个 id,不重复执行
    else 不存在
        H->>S: commit(pi.user + submission)
        S-->>H: 已落盘
        H-->>C: 新 submission
    end

这条规则让「客户端超时后重试」变得安全:重试不会产生第二条用户消息,也不会触发第二次模型调用。harness.submission(id) 则允许重启后的进程按 ID 重新取得一个 submission,继续等待它。

五、错误边界:三种错误,三种后果

不是所有错误都应该被同样对待。Pi Durable 用三条规则划分边界:

flowchart TD
    E["发生错误"] --> T{"错误来自哪里?"}
    T -- 存储层 --> S["终止整个 Harness<br/>所有等待以 SessionFailed reject"]
    T -- 非法读取 --> R["只让这次读取失败<br/>StorageRequestError"]
    T -- 业务代码 --> B["只让当前单元失败<br/>task faulted / tool 错误结果 / hook 跳过"]
    S --> S2["重开 Harness<br/>用同一 requestId 重提"]

具体规则:

  1. 存储错误会终止 Harness。 撞上它的调用以该错误 reject;之后的一切、以及每个 pending 的 wait,都以 SessionFailed reject;watches 与事件流以 session_failed 结束;运行中的任务收到 abort 信号;onReport 收到一次该错误。然后 Harness 自行关闭。不重试任何东西——自定义存储自己重试自己的瞬态失败。恢复方式是重开:提交是原子的,所以存储是一致的;用同一个 requestId 重提,会找到已被接收的 submission。
  2. 非法读取只让那次调用失败。 坏 cursor、未知对话、document 未保留的历史,都会抛 StorageRequestError。这个豁免仅限读取:commit() 或 mintId() 的任何错误都按规则 1 处理。
  3. 业务代码只让自己的单元失败。 task phase 抛异常 → 该 task faulted;tool 抛异常 → 模型收到错误结果;hook 抛异常 → 上报并跳过(beforeTool 里则阻断该调用);commit 回调抛异常 → 回滚这次提交。

宿主可以监听 Harness 的关闭:

harness.closed.then((end) => {
    if (end.reason === "failed") reopen(end.error);
});

这里的设计判断很明确:存储不可信时,不要假装还能继续。 继续运行只会产生更多无法解释的状态。业务代码的错误则必须被隔离在当前单元内,否则一个工具的 bug 会拖垮整个对话。

六、恢复语义速查

被中断的对象恢复行为
模型请求重发;partial 回答留在 transcript 里,标记为 aborted
工具调用声明 replay: "safe" 则重跑;否则模型收到 interrupted 与已有输出
子智能体它跑在自己的对话里,从自己的 checkpoint 继续
排队消息仍在 inbox 里,按原策略放置
提交同一 requestId 返回既有 submission,不重复执行
Code Mode 脚本用 journal 确定性重放,不重跑已完成的嵌套调用

小结

崩溃恢复不是「重启时多读一个文件」,而是一组互相咬合的语义:

  • checkpoint 把「要做什么」持久化;
  • memo 让重复执行不产生重复的可见效果;
  • replay: "safe" 把「能否重跑」的判断交给工具作者;
  • requestId 把「恰好一次」放在提交入口;
  • 三条错误规则决定哪些失败可以隔离、哪些必须终止。

下一篇换一个视角:从一次用户输入出发,拆开 Turn、Run、Submission、Inbox 与压缩,看看这条流水线在正常路径上如何运转。