崩溃之后：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["继续下一阶段"]

三个分支对应三种语义：

注意最后一条：框架不会盲目重跑一个没有声明 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 会拖垮整个对话。

六、恢复语义速查

小结

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

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

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