Agent 崩溃恢复机制：从 Checkpoint 到副作用一致性

monsion

266人浏览 · 2026-07-21 19:44:11

monsion · 2026-07-21 19:44:11 发布

Agent 崩溃恢复机制：从 Checkpoint 到副作用一致性

Agent 可以连续工作数十分钟甚至数小时，也可能同时调用模型、Shell、浏览器、数据库、消息系统和多个子 Agent。一旦进程在中途退出，问题便不再是传统意义上的「重新启动服务」：对话也许已经保存，文件也许已经修改，邮件也许已经发出，但系统并不知道最后一步究竟完成到了哪里。

这正是 Agent 崩溃恢复最难的地方。它同时涉及分布式系统、工作流引擎、事件日志、任务调度、外部副作用、沙箱生命周期和人机协作。单独增加一张 checkpoint 表，或者提供一个 resume 命令，都不足以解决完整问题。

本文从故障语义出发，分析 LangGraph、DeerFlow、Open Managed Agents、AgentScope、OpenCode、Codex 和 Pi 的具体实现，再给出一套可落地的生产级架构、恢复算法和验证方法。源码分析基于 2026-07-21 的固定提交；GitHub 源码链接固定到对应提交，官方文档链接则是滚动更新页面，访问日期为 2026-07-21。

一、先给结论：崩溃恢复不是一个开关

Agent 的恢复能力可以分成四级：

等级	能力	崩溃后的表现
L1：会话可恢复	保存消息、上下文、分支和摘要	重启后可以重新打开会话
L2：状态可收敛	识别悬空流、工具调用和运行状态	不会永久停在 `running`，而是变为 `interrupted`、`error` 或待确认
L3：执行可续跑	从安全边界重新调度未完成步骤	已完成步骤不重跑，未完成步骤继续执行
L4：副作用可恢复	识别外部操作是否已经生效	不重复付款、发消息、创建工单或执行破坏性命令

很多 Coding Agent 已经达到 L1，部分达到 L2。LangGraph 在图执行边界上具备较强的 L3 能力。L4 则没有任何通用框架能够默认保证，因为外部系统通常不与 Agent 的 checkpoint 共用一个数据库事务。

可以用一句话概括：

Checkpoint 只能证明「系统记得执行到哪里」，不能证明「外部世界只被修改了一次」。

本文对几个代表性项目的分析重点如下：

LangGraph：checkpoint、超级步（superstep，见 6.2）和 pending writes。
Open Managed Agents：冷启动修复和真实 SIGKILL 测试。
DeerFlow：run lease、heartbeat、孤儿任务扫描、SSE 回放和沙箱所有权。
AgentScope：分布式 session lock、持久状态、事件 projection 和读时修复。
OpenCode、Codex、Pi：本地会话日志、历史重建和人工继续体验。

如何阅读本文：本文较长，可按需取用——只想要结论，读第一、二十、三十一节；想落地实现，重点读第五、九、二十二节；关注具体框架，直接看第十三到十九节。第 9.3 节的工具包装器和第二十二节的伪代码属于实现细节，首次阅读可以先跳过主干之外的分支，理解主流程后再回看。

二、为什么 Agent 比普通后台任务更难恢复

传统任务队列通常处理一个相对明确的函数：输入固定、执行边界清晰、结果结构确定。Agent 则在运行中动态决定下一步动作。一次模型返回可能创建多个工具调用，工具结果又会改变后续计划，子 Agent 还可能继续创建新任务。

一个长时间运行的 Agent 至少同时维护以下状态：

对话消息和模型上下文；
当前计划、待办项和完成条件；
工作流节点和路由位置；
模型流式响应的部分内容；
工具调用及其参数、结果和副作用；
workspace 中的文件状态；
Shell、PTY、浏览器和远端 job 等运行中资源；
父子 Agent 的依赖关系；
客户端已经看到的事件游标；
当前 worker 对 run 的所有权。

这些状态不在同一个介质中。消息可能在 PostgreSQL，流事件可能在 Redis，workspace 可能在持久卷，浏览器进程可能只在内存，付款结果则存在外部支付系统。恢复的本质，是重新判断这些事实之间是否一致。

2.1 进程恢复、状态恢复和业务恢复是三件事

进程恢复只需要 Kubernetes 重启 Pod。状态恢复要求新进程能读取原 session。业务恢复还要回答：

上一个工具调用是否已经执行；
工具成功后是否已经保存结果；
客户端是否收到成功事件；
旧 worker 是否仍可能恢复运行；
父 Agent 是否已经收到子 Agent 结果；
重试是否会再次产生外部副作用。

因此，restartPolicy: Always 只能解决进程存活，不能解决执行正确性。

2.2 Agent 的正确性不是「最后成功」这么简单

如果一个邮件 Agent 崩溃后把同一封邮件发送两次，即使第二次 run 最终显示成功，也不能算正确恢复。恢复系统必须维护一组 invariant：

同一时刻，一个 run 最多只有一个有效 owner；
terminal 状态不能倒退成 running；
已成功的 step 不能因恢复而重复提交；
同一个 tool operation 只能产生一个业务结果；
本文参考设计中的事件序号从 1 开始连续递增，客户端可以检测并补齐漏掉的事件；
在相同外部事实下，recovery 可以重复执行且不产生重复持久写入；外部事实变化时，恢复状态只能单调收敛；
旧恢复任务不能覆盖崩溃后创建的新 turn。

这些 invariant 比「服务重启成功」更接近真实的验收标准。

三、建立统一故障模型

3.1 八层恢复模型

层面	必须保存或协调的内容	典型故障
请求接入	tenant-scoped request ID、request hash、durable inbox、受理结果	请求已受理但尚未派发，客户端因响应丢失而重试，或相同 ID 被误用于不同请求
会话历史	用户消息、模型消息、压缩摘要、分支	重启后上下文丢失或顺序错误
工作流状态	当前节点、状态值、已完成节点写入	整个任务从头重跑
Run 所有权	owner、lease、heartbeat、fencing token	lease 交接时旧、新 worker 物理执行重叠，旧 token 仍尝试提交
工具副作用	operation ID、幂等键、远端请求 ID、结果	重复支付、重复发送、重复写入
沙箱状态	workspace、容器 ID、持久卷、进程句柄	文件存在，但进程或浏览器状态消失
事件流	连续 seq、游标、保留窗口、结束标记	客户端漏事件、重复事件或一直加载
多 Agent 依赖	parent、child、等待条件、终态事件	父 Agent 永久等待或重复创建 child

3.2 六个最重要的崩溃窗口

窗口	崩溃后的事实	推荐机制
A：接收后、保存前	系统是否接受请求未知	durable inbox；保存后再返回「已受理」
B：保存后、派发前	工作尚未开始	durable queue 或数据库扫描重新派发
C：派发后、工具执行前	可以安全重派	step attempt 状态和稳定 operation ID
D：工具生效后、结果保存前	最危险，结果未知	外部幂等键、远端查询、对账或人工确认
E：结果保存后、checkpoint 前	不应再次执行工具	持久结果缓存；恢复时直接推进图状态
F：checkpoint 后、客户端收到前	计算完成，展示缺失	outbox + 有序事件日志 + cursor replay

窗口 D 是所有设计的分水岭。如果工具系统无法接收幂等键，也无法查询历史请求，那么就不存在普遍安全的自动重试。

3.3 恢复点不等于事务点

假设一个节点执行以下代码：

def charge_card(state):
    result = payment_api.charge(state["amount"])
    return {"payment_id": result.id}

checkpoint 通常发生在函数返回之后。如果 charge() 已成功，但进程在 return 之前退出，checkpoint 看不到 payment ID。恢复后重新运行节点，就可能再次扣款。

可靠实现需要把一次逻辑操作拆成可识别的 operation：

def charge_card(state):
    operation_id = stable_id(
        state["run_id"],
        state["node_id"],
        state["logical_attempt"],
        state["tool_call_id"],
    )
    return payment_api.charge(
        amount=state["amount"],
        idempotency_key=operation_id,
    )

稳定 ID 的意义不是避免底层重试，而是让每次重试仍代表同一个业务操作。其中 tool_call_id 必须来自已经持久化的模型响应、确定性调用序号或业务操作 ID，不能在进程重启时重新随机生成；logical_attempt 在同一次逻辑操作的进程重试中也不能变化。

四、恢复语义：At-most-once、At-least-once 与 Effectively-once

4.1 At-most-once

任务最多执行一次。一种常见实现是在执行前确认或移除消息，因此 worker 随后崩溃时可能丢失工作。它避免重复，却牺牲完成率，不适合长时间 Agent。

4.2 At-least-once

任务至少会被执行一次。只有成功提交后才确认消息，因此 worker 崩溃时任务会重新投递。这是大多数可靠队列的基础语义，但可能重复执行。

4.3 Exactly-once 为什么困难

如果 Agent 数据库和外部邮件服务不共享事务，就无法原子完成以下两件事：

邮件服务接受发送请求；
Agent 数据库保存发送结果。

两者之间总会存在崩溃窗口。两阶段提交理论上可以连接多个事务资源，但公开 API、Shell 和浏览器通常不参与分布式事务。

4.4 Effectively-once 更现实

生产系统通常采用 effectively-once：

队列允许重复投递；
每个逻辑工具操作有稳定 operation ID；
外部系统使用相同 idempotency key；
本地结果表有唯一约束；
不确定结果通过远端请求 ID 查询；
无法查询的非幂等操作转入人工确认。

这样即使底层执行多次，对业务观察者仍只产生一个结果。

五、生产级参考架构

5.1 总体架构

这一架构把职责拆开：

inbox 证明请求已被接受；
queue 保证任务至少一次派发；
lease 决定谁可以执行；
checkpoint 保存图状态；
tool operation store 保存外部操作事实；
outbox 保证状态提交后事件最终可见；
event log 支持断线回放；
reconciler 负责发现和处理孤儿状态；
sandbox manager 恢复 workspace 与远端资源引用。

5.2 推荐数据模型

下面使用 PostgreSQL 风格的逻辑模型。字段类型可以按具体数据库调整；但要提醒的是，下文多处依赖「某几张表位于同一事务边界」这一前提，它必须由具体实现真正保证，不能仅凭方法名想当然，后文会指出哪些表必须同事务提交：

CREATE TABLE request_inbox (
  tenant_id         TEXT NOT NULL,
  request_id        TEXT NOT NULL,
  thread_id         TEXT NOT NULL,
  request_json      JSON NOT NULL,
  request_hash      TEXT NOT NULL,
  accepted_at       TIMESTAMPTZ NOT NULL,
  PRIMARY KEY (tenant_id, request_id)
);

CREATE TABLE run (
  id                TEXT PRIMARY KEY,
  tenant_id         TEXT NOT NULL,
  thread_id         TEXT NOT NULL,
  request_id        TEXT NOT NULL,
  request_hash      TEXT NOT NULL,
  status            TEXT NOT NULL,
  owner_id          TEXT,
  lease_until       TIMESTAMPTZ,
  fencing_token     BIGINT NOT NULL DEFAULT 0,
  recovery_owner_id     TEXT,
  recovery_lease_until  TIMESTAMPTZ,
  recovery_fencing_token BIGINT NOT NULL DEFAULT 0,
  next_event_seq    BIGINT NOT NULL DEFAULT 1,
  checkpoint_id     TEXT,
  created_at        TIMESTAMPTZ NOT NULL,
  updated_at        TIMESTAMPTZ NOT NULL,
  UNIQUE (tenant_id, request_id)
);

CREATE TABLE run_checkpoint (
  id                TEXT PRIMARY KEY,
  run_id            TEXT NOT NULL,
  parent_id         TEXT,
  state_json        JSON NOT NULL,
  created_at        TIMESTAMPTZ NOT NULL
);

CREATE TABLE step_attempt (
  id                TEXT PRIMARY KEY,
  run_id            TEXT NOT NULL,
  node_id           TEXT NOT NULL,
  logical_attempt   INTEGER NOT NULL,
  status            TEXT NOT NULL,
  input_hash        TEXT NOT NULL,
  created_under_fencing_token BIGINT NOT NULL,
  started_at        TIMESTAMPTZ,
  finished_at       TIMESTAMPTZ,
  UNIQUE (run_id, node_id, logical_attempt)
);

CREATE TABLE tool_operation (
  id                TEXT PRIMARY KEY,
  step_attempt_id   TEXT NOT NULL,
  tool_call_id      TEXT NOT NULL,
  call_index        INTEGER,
  tool_name         TEXT NOT NULL,
  effect_class      TEXT NOT NULL,
  idempotency_guarantee TEXT NOT NULL
                      CHECK (idempotency_guarantee IN ('none', 'strong')),
  status            TEXT NOT NULL,
  dispatch_generation BIGINT NOT NULL DEFAULT 0,
  idempotency_key   TEXT NOT NULL UNIQUE,
  remote_request_id TEXT,
  request_json      JSON NOT NULL,
  request_hash      TEXT NOT NULL,
  result_json       JSON,
  error_json        JSON,
  applied_checkpoint_id TEXT,
  started_at        TIMESTAMPTZ,
  finished_at       TIMESTAMPTZ,
  UNIQUE (step_attempt_id, tool_call_id)
);

CREATE TABLE run_event (
  run_id            TEXT NOT NULL,
  seq               BIGINT NOT NULL,
  event_type        TEXT NOT NULL,
  payload_json      JSON NOT NULL,
  created_at        TIMESTAMPTZ NOT NULL,
  PRIMARY KEY (run_id, seq)
);

CREATE TABLE outbox (
  id                TEXT PRIMARY KEY,
  topic             TEXT NOT NULL,
  event_key         TEXT NOT NULL,
  run_id            TEXT NOT NULL,
  event_seq         BIGINT,
  payload_json      JSON NOT NULL,
  created_at        TIMESTAMPTZ NOT NULL,
  published_at      TIMESTAMPTZ,
  UNIQUE (topic, event_key),
  UNIQUE (run_id, event_seq)
);

这里采用通用 outbox：派发事件可以让 event_seq 为空，客户端可见的 run 事件则必须携带事务内分配的序号。参考实现假设 run、run_checkpoint、run_event 和 outbox 位于同一个事务数据库中。若 checkpointer 使用独立物理存储，就不能声称 checkpoint 与 run 终态处于同一事务，必须采用后文所述的修复语义。

时间字段使用 TIMESTAMPTZ，由数据库统一比较绝对时间。lease_until 尤其不能依赖各 worker 的本地时区解释；应用层展示时再转换为用户时区。

step_attempt 表示逻辑节点尝试，接管后仍复用同一个 (run_id, node_id, logical_attempt)。created_under_fencing_token 只记录该行首次创建时的执行 token，用于审计，不代表当前 owner，也不能用于后续写入授权。所有状态提交都以事务中锁定并重新校验的 run.fencing_token 为准。若需要记录每次物理执行，可以另建 step_execution_attempt(step_attempt_id, owner_id, fencing_token, started_at, finished_at)，每次接管新增一行，而不改变逻辑 step attempt 的身份。

5.3 Run 状态机

uncertain 必须是一等状态。把未知结果直接写成 failed，会诱导上层自动重试；直接写成 succeeded，则可能伪造业务事实。

工具操作恢复为成功，只能说明该工具结果可以复用，不能说明整个 run 已完成。只要 checkpoint 仍有后续节点，run 就应回到 queued；只有终态 checkpoint 能支持 uncertain -> succeeded。

运行中取消是一条需要验证的状态转换，不是把 status 直接改为 cancelled。worker 收到取消意图后先停止发起新工具操作，再处理中断中的任务，并检查所有 started/unknown operation。只有不存在未确认副作用，且取消 checkpoint、run 终态和 terminal event 能按当前 fencing token 提交时，才允许 running -> cancelled；存在无法确认的外部操作时必须进入 uncertain。

六、核心机制一：Checkpoint 与确定性重放

6.1 Checkpoint 保存什么

一个完整 checkpoint 至少包含：

当前状态值；
下一批待执行节点；
checkpoint ID 和 parent ID；
每个 channel 的版本；
已完成 task 的 pending writes；
interrupt 信息；
子图状态引用。

只保存对话文本并不足够。恢复器还要知道「接下来执行哪个节点」以及「同一并行步骤中哪些节点已经完成」。

6.2 超级步为什么重要

以并行图为例：

A 与 B 属于同一超级步。假设 A 已完成、B 失败：

只保存完整 checkpoint 的引擎会回到超级步开始，A 和 B 都重跑；
保存 task-level pending writes 的引擎可以复用 A 的结果，只重跑 B；
如果 A 有外部副作用，而结果未保存，则仍可能重复执行。

因此，pending writes 降低了重复计算范围，但不能替代工具幂等。

6.3 Durability 模式的取舍

LangGraph 当前定义三种 durability：

模式	持久化时机	崩溃时最大回退范围	代价
`sync`	下一超级步开始前等待写入完成	最近一个超级步	写入延迟进入关键流程
`async`	checkpoint 写入与下一步并行	可能回退一个尚未完成持久化的边界	吞吐和安全性折中
`exit`	图退出时保存	整个运行过程	中途恢复能力最低

官方文档说明 checkpoint 位于超级步边界，节点级 task writes 则支持同一超级步内的 pending writes 恢复。LangGraph Persistence

6.4 确定性重放

恢复执行通常不是恢复 Python 调用栈，而是重新执行工作流函数，再复用已经保存的 task 结果。因此工作流代码需要满足：

相同历史应选择相同分支；
随机数、当前时间和外部 API 调用应封装为可持久化 task；
已完成 task 的结果需要可序列化；
interrupt 前的逻辑可能再次运行，必须幂等。

官方 Functional API 文档也明确建议：API 调用放入 task，并为可能的重新执行设计幂等键。LangGraph Functional API

6.5 Checkpoint 的边界

Checkpoint 不会自动保存：

运行中的操作系统进程；
打开的 TCP 连接；
浏览器页面的 JavaScript 堆；
未提交数据库事务；
外部服务已经执行但未返回的请求；
仅存在于 worker 内存中的取消标记。

这些状态必须通过稳定标识重新 attach、查询或重建。

七、核心机制二：事件日志、快照与重放

7.1 Event Sourcing 与 Snapshot 的分工

事件日志记录「发生过什么」，snapshot 记录「现在是什么状态」。

方式	优点	缺点
只用 snapshot	读取快，结构直观	难以解释状态如何形成，崩溃窗口不易修复
只用 event log	可审计、可重放	历史很长时恢复慢，事件演进复杂
event log + snapshot	兼顾恢复速度和审计	需要维护 snapshot 版本与事件兼容性

一种更准确的组合方式是：

外部操作执行前保存 operation intent，执行后保存 result、failure 或 unknown；
内部状态变化与 outbox event 在同一事务中提交；
周期性生成 snapshot；
恢复时从最近 snapshot 开始重放后续事件；
snapshot 是读取优化，event log 记录系统已知的因果事实，但不能替代对外部系统的权威查询。

7.2 不完整模型流如何恢复

模型流可能已经产生 200 个 token，但最终 assistant message 尚未保存。常见策略有三种：

丢弃 partial stream：历史整洁，但用户已经看到的内容消失；
保留 partial message 并标记 interrupted：可解释性最好；
自动重新调用模型：可能得到完全不同的答案，也会增加费用。

Open Managed Agents 采用第二种策略：保存 stream chunk，冷启动时合并为 partial message，并将流状态改为 interrupted。这种策略不假装原模型调用可以继续，也不抹掉用户已经看到的内容。

7.3 事件回放不是无限保留

Redis Stream 很适合短期重连，但通常配置最大长度或 TTL。完整会话历史仍应保存到数据库或对象存储。可靠系统需要明确两种游标：

传输游标：用于几分钟或几小时内的 SSE/WebSocket 重连；
业务序号：长期保存在事件表，用于审计和重新构建状态。

如果只保留 Redis Stream，超过 TTL 的客户端无法证明自己漏了哪些事件。

7.4 事件写入顺序

一次状态变化通常会同时影响数据库和事件流。先推送、后保存可能让客户端看到最终无法恢复的事件；先保存、后推送则可能在崩溃时造成客户端漏事件。

Transactional outbox 的写入顺序更可靠：

连续分配序号不等于按序发布。若两个 Relay 同时读取同一 run，仍可能先发送 N+1。本文采用以下发布协议：

run.events 按 run_id 分区，同一 run 只由一个逻辑消费者串行处理；
Relay 在数据库中获取 per-run advisory lock 或等价发布锁，并持有到 published_at 提交；若使用 TTL 锁，还要按 token 续租；
每次只读取该 run 最小的未发布 event_seq，N 未标记为 published 前不读取 N+1；
Event Log 以 (run_id, seq) 幂等保存，确认持久化后 Relay 才更新 published_at；
Relay 崩溃后锁释放或过期，新 Relay 从仍未发布的 N 开始，而不是跳到 N+1。

async def relay_run_events(run_id):
    async with publish_lock(run_id):
        while True:
            event = await outbox.lowest_unpublished(
                topic="run.events",
                run_id=run_id,
            )
            if event is None:
                return

            await event_log.append_once(
                run_id=run_id,
                seq=event.event_seq,
                payload=event.payload_json,
            )
            await outbox.mark_published(event.id, db.now())

客户端仍要把序号缺口视为未完成读取。若最后确认的游标是 N-1，却先观察到 N+1，则暂停推进游标并主动补读 N。目标 Event Log 即使因网络或消费层重排收到乱序写入，也应按 (run_id, seq) 存储和读取。

八、核心机制三：Lease、Heartbeat 与 Fencing Token

8.1 为什么互斥锁不够

分布式 worker 可能经历：

进程退出；
网络分区；
event loop 长时间暂停；
stop-the-world GC；
容器被挂起后恢复；
数据库连接暂时失败。

旧 worker 失去锁后不一定立即停止。它可能在网络恢复后继续提交结果。因此，只有「锁过期后允许新 worker 接管」还不够。

8.2 Lease 的基本流程

8.3 Fencing Token 的作用

每次所有权变更时，数据库递增 token。后续所有状态写入必须带 token：

UPDATE run
   SET checkpoint_id = :checkpoint_id,
       updated_at = NOW()
 WHERE id = :run_id
   AND fencing_token = :token
   AND owner_id = :owner_id;

如果受影响行数为 0，worker 必须停止。token 解决的是「旧 owner 恢复后继续写」问题，这是单纯 TTL 锁无法彻底解决的。

8.4 续租与释放必须校验 owner

以 Redis 为例，以下做法不安全：

EXPIRE lock_key 30
GET lock_key
DEL lock_key

如果旧 lease 已过期，新 worker 已获取同一个 key，旧 worker 的 EXPIRE 可能延长新锁，GET 与 DEL 之间也可能发生所有权变化。更安全的方式是使用 Lua，把 token 比较和 PEXPIRE／DEL 放在一个原子操作中。

即便锁操作完全正确，业务数据库写入仍建议使用 fencing token。锁保护的是协调过程，token 保护的是最终状态。

8.5 Lease 参数如何确定

lease 太短会因为偶发停顿产生误接管，太长会增加恢复时间。通常需要结合：

worker 的 P99 event-loop stall；
数据库 P99 写入延迟；
网络短暂故障时间；
可接受的恢复时间目标；
任务是否允许并发重复。

常见起点是：

lease TTL：30～120 秒；
heartbeat interval：TTL 的 1/3；
takeover grace：额外 5～30 秒；
每次 heartbeat 记录续租失败原因和延迟。

这不是固定配置。浏览器自动化、GPU 推理和普通 API Agent 的暂停特征并不相同。

九、核心机制四：工具副作用与不确定状态

9.1 工具操作状态机

这里的 unknown 是 operation 级状态，只描述单个外部操作结果未知；它与 5.3 中 run 级的 uncertain 不在同一层：前者是「这一个工具调用结果不明」，后者是「整个 run 因此无法安全收敛」。本文刻意用两个不同的词区分这两层，读者不必把它们当作同义词。

9.2 工具分类

类型	示例	恢复策略
纯读	搜索、读取文件、GET API	通常可重试，仍记录 operation ID
确定性覆盖写	写入固定内容、PUT 同一资源	比较目标版本或 hash 后重试
支持去重的写	支付、工单、消息 API 支持幂等键	使用原 idempotency key 重投
可查询的异步任务	CI、云构建、远端 batch job	保存 remote job ID，恢复时 attach/poll
不可去重写	任意 Shell、无请求 ID 的邮件、数据库自增	不盲目重试，进入查询或人工确认

本文只使用一套持久化的幂等能力定义：idempotency_guarantee 取值为 none 或 strong。标记为 strong，要求外部系统在完整重试窗口内对同一 key 原子去重，即：

同一 key、相同请求返回同一业务结果；
同一 key、不同请求返回冲突；
并发请求也不会产生两个业务效果。

仅接受一个幂等参数、却没有上述保证，不能标记为 strong。工具包装器读取 tool.idempotency_guarantee，operation 创建时保存同一值，Reconciler 再从 operation 记录读取，避免运行时与恢复器采用不同标准。

9.3 推荐工具包装器

以下为机制示意：

async def execute_effectful_tool(
    run,
    step,
    tool_call,
    tool,
    args,
    *,
    worker_id,
    fencing_token,
):
    if not await lease_store.is_valid(
        run.id, worker_id, fencing_token
    ):
        raise LeaseLost(run.id, worker_id, fencing_token)

    call_identity = (
        tool_call.business_operation_id
        or tool_call.stable_tool_call_id
    )
    if call_identity is None:
        if tool_call.stable_index is None:
            raise MissingStableToolCallIdentity(tool.name)
        call_identity = f"call-index:{tool_call.stable_index}"

    operation_id = stable_operation_id(
        run.id,
        step.node_id,
        step.logical_attempt,
        tool.name,
        call_identity,
    )
    request_hash = canonical_hash(args)

    op = await operation_store.create_if_absent_if_owner(
        id=operation_id,
        step_attempt_id=step.id,
        tool_name=tool.name,
        effect_class=tool.effect_class,
        idempotency_guarantee=tool.idempotency_guarantee,
        status="prepared",
        idempotency_key=operation_id,
        request_json=args,
        request_hash=request_hash,
        tool_call_id=call_identity,
        call_index=tool_call.stable_index,
        run_id=run.id,
        owner_id=worker_id,
        fencing_token=fencing_token,
    )

    if (
        op.step_attempt_id != step.id
        or op.tool_name != tool.name
        or op.effect_class != tool.effect_class
        or op.idempotency_guarantee != tool.idempotency_guarantee
        or op.idempotency_key != operation_id
        or op.request_hash != request_hash
    ):
        raise OperationIdentityConflict(operation_id)

    if op.status == "succeeded":
        return op.result

    if op.status == "failed":
        raise StoredToolFailure(op.error_json)

    if op.status == "manual_review":
        raise ManualReviewRequired(operation_id)

    if op.status in {"started", "unknown"}:
        reconciled = await tool.lookup(op.remote_request_id, operation_id)
        if reconciled.status == "succeeded":
            finished = await operation_store.finish_once_if_owner(
                operation_id,
                result=reconciled,
                expected_status=op.status,
                expected_dispatch_generation=op.dispatch_generation,
                run_id=run.id,
                owner_id=worker_id,
                fencing_token=fencing_token,
            )
            return finished.result
        if reconciled.status == "failed":
            failed = await operation_store.mark_failed_once_if_owner(
                operation_id,
                error=reconciled.error,
                expected_status=op.status,
                expected_dispatch_generation=op.dispatch_generation,
                run_id=run.id,
                owner_id=worker_id,
                fencing_token=fencing_token,
            )
            raise StoredToolFailure(failed.error_json)
        if reconciled.status == "manual_review":
            await operation_store.mark_manual_review_if_owner(
                operation_id,
                run.id,
                worker_id,
                fencing_token,
                expected_status=op.status,
                expected_dispatch_generation=op.dispatch_generation,
            )
            raise ManualReviewRequired(operation_id)

        if not reconciled.old_executor_cannot_send:
            raise OperationDispatchInProgress(operation_id)

        if reconciled.status != "not_executed":
            await operation_store.mark_manual_review_if_owner(
                operation_id,
                run.id,
                worker_id,
                fencing_token,
                expected_status=op.status,
                expected_dispatch_generation=op.dispatch_generation,
            )
            raise ManualReviewRequired(operation_id)

        reliable_no_effect = (
            reconciled.authoritative
            or reconciled.consistency_window_elapsed
        )
        retry_is_safe = (
            reliable_no_effect
            and reconciled.old_executor_cannot_send
            and (
                tool.idempotency_guarantee == "strong"
                or reconciled.retry_approved
            )
        )
        if not retry_is_safe:
            await operation_store.mark_manual_review_if_owner(
                operation_id,
                run.id,
                worker_id,
                fencing_token,
                expected_status=op.status,
                expected_dispatch_generation=op.dispatch_generation,
            )
            raise ManualReviewRequired(operation_id)

        op = await operation_store.authorize_retry_if_owner(
            operation_id=operation_id,
            run_id=run.id,
            owner_id=worker_id,
            fencing_token=fencing_token,
            expected_status=op.status,
            expected_dispatch_generation=op.dispatch_generation,
            next_status="prepared",
        )
        if op is None:
            if not await lease_store.is_valid(
                run.id, worker_id, fencing_token
            ):
                raise LeaseLost(run.id, worker_id, fencing_token)
            raise OperationRetryAuthorizationLost(operation_id)

    dispatch = await operation_store.claim_dispatch_if_owner(
        operation_id=operation_id,
        run_id=run.id,
        owner_id=worker_id,
        fencing_token=fencing_token,
        expected_dispatch_generation=op.dispatch_generation,
    )
    if dispatch is None:
        if not await lease_store.is_valid(
            run.id, worker_id, fencing_token
        ):
            raise LeaseLost(run.id, worker_id, fencing_token)
        raise OperationDispatchClaimLost(operation_id)

    try:
        result = await tool.execute(args, idempotency_key=operation_id)
    except DefiniteToolFailure as exc:
        await operation_store.mark_failed_if_owner(
            operation_id,
            error=serialize_error(exc),
            expected_status="started",
            expected_dispatch_generation=dispatch.generation,
            run_id=run.id,
            owner_id=worker_id,
            fencing_token=fencing_token,
        )
        raise
    except TransportLost:
        await operation_store.mark_unknown_if_owner(
            operation_id,
            expected_status="started",
            expected_dispatch_generation=dispatch.generation,
            run_id=run.id,
            owner_id=worker_id,
            fencing_token=fencing_token,
        )
        raise

    finished = await operation_store.finish_once_if_owner(
        operation_id,
        result=result,
        expected_status="started",
        expected_dispatch_generation=dispatch.generation,
        run_id=run.id,
        owner_id=worker_id,
        fencing_token=fencing_token,
    )
    return finished.result

create_if_absent_if_owner() 必须填充数据模型中的全部必填字段，并在插入 prepared 前校验当前 run owner、fencing token 和 lease。单独使用 INSERT ... SELECT FROM run 不够，因为读取 run 不会持有 run 行锁。该方法必须在同一事务中先锁定并校验 run，再插入或读取 operation。校验失败时回滚并抛出 LeaseLost，不能由旧 worker 插入过期 operation。

prepared -> started 不仅要校验 run owner，还要取得本次 operation 的唯一发送权。dispatch_generation 是 operation 级 CAS 版本：调用方必须提交自己读取到的精确状态和 generation，成功更新者把 generation 加一并获得发送权。started/unknown 必须先经对账和 CAS 回到 prepared。发送权更新的示意 SQL 如下：

BEGIN;

-- 必须先取得 run 行锁；没有返回行时回滚并报告 LeaseLost。
SELECT id
  FROM run
 WHERE id = :run_id
   AND status = 'running'
   AND owner_id = :worker_id
   AND fencing_token = :fencing_token
   AND lease_until > NOW()
 FOR UPDATE;

UPDATE tool_operation AS o
   SET status = 'started',
       dispatch_generation = o.dispatch_generation + 1,
       started_at = COALESCE(o.started_at, NOW())
 WHERE o.id = :operation_id
   AND o.status = 'prepared'
   AND o.dispatch_generation = :expected_dispatch_generation
   AND EXISTS (
       SELECT 1
         FROM step_attempt AS s
        WHERE s.id = o.step_attempt_id
          AND s.run_id = :run_id
   )
RETURNING o.id, o.dispatch_generation;

COMMIT;

claim_dispatch_if_owner() 只允许从 prepared 取得发送权。started/unknown operation 必须先完成外部对账，再由 authorize_retry_if_owner() 按精确状态和 generation CAS 回 prepared；不能直接再次取得发送权。两个协程即使同时读到 prepared/generation=0，也只有一个能提交为 started/generation=1；另一个得到 OperationDispatchClaimLost，不得发送外部请求。run lease 限制 worker，operation generation 限制同一 worker 内的并发协程，两者不能互相替代。

所有 worker 侧 operation 转换都遵循同一锁顺序：在一个事务中先 SELECT run ... FOR UPDATE 并重新校验 owner/token/lease，再创建或更新 operation。适用范围包括：

创建 prepared；
授权重试；
取得发送权；
保存 succeeded/failed/unknown/manual_review；
把结果标记为已应用。

Reconciler fencing 也要先更新并锁住同一 run 行，这样两类事务会在 run 行上串行化，不能用旧语句快照绕过新 token。任何代码都不能先锁 operation、再锁 run，否则不同路径之间可能死锁。

结果更新还要同时校验预期状态和 dispatch generation：成功与明确失败只接受调用方读取到的精确 started/unknown 状态；mark_unknown 只接受 started；mark_manual_review 只接受 started/unknown。succeeded 和 failed 是终态，普通转换不能覆盖。CAS 没有更新行时重新读取 operation：已经成功则复用结果，已经失败则返回已保存错误，处于 manual_review 则暂停，其他状态按协调信号重试；不能用无条件更新覆盖新状态。

协调信号不是工具失败

OperationDispatchInProgress、OperationDispatchClaimLost、OperationRetryAuthorizationLost 和结果更新产生的 OperationStateChanged 都是非终态协调信号。execute_step_with_operations() 在 operation 级 singleflight 内处理这些信号，不生成工具失败结果，也不把 run 改成 terminal failure：

COORDINATION_SIGNALS = (
    OperationDispatchInProgress,
    OperationDispatchClaimLost,
    OperationRetryAuthorizationLost,
    OperationStateChanged,
)


async def execute_with_operation_coordination(
    operation_id,
    invoke_once,
):
    async def run_as_local_leader():
        while True:
            try:
                return await invoke_once()
            except OperationDispatchInProgress:
                await operation_store.wait_for_change(
                    operation_id,
                    timeout_seconds=2,
                )
            except (
                OperationDispatchClaimLost,
                OperationRetryAuthorizationLost,
                OperationStateChanged,
            ):
                pass

            op = await operation_store.get(operation_id)
            if op.status == "succeeded":
                return op.result
            if op.status == "failed":
                raise StoredToolFailure(op.error_json)
            if op.status == "manual_review":
                raise ManualReviewRequired(operation_id)

            await jittered_backoff()

    return await operation_singleflight.run(
        operation_id,
        run_as_local_leader,
    )

同一进程中已有本地执行时，后来的协程等待 singleflight leader；当前 dispatch 仍在其他执行器中运行时，leader 等待状态变化并退避轮询；CAS 竞争失败时重新读取 operation 再决定下一步。等待期间继续续租 run，收到取消或 LeaseLost 时按控制面中断退出。即使协调信号因关机传播到执行循环，也必须按「可重试／待处理」分类，不能进入确定性工具失败分支。

关键点有五个：

operation ID 来自逻辑身份，不来自本次进程或物理 attempt；
同一节点内的两次同名工具调用由业务操作 ID、稳定 tool_call_id 或稳定调用序号区分；三者都不存在时拒绝执行；
logical_attempt 在进程重试时保持不变，只有业务明确开始一次新操作时才递增；
参数摘要只用于检测同一身份是否绑定了不同请求，不能区分「相同参数的两次合法调用」；
按当前 owner/token 创建 prepared，再通过精确状态和 dispatch generation CAS 把发送意图持久化为 started，最后发送请求；因此 started 表示请求可能尚未发送，也可能已经发送。确定性工具错误写为 failed，连接丢失则写为 unknown。

包装器读取到 failed 时只返回已保存错误，读取到 manual_review 时只保持暂停，不会再次发送。只有独立的人工审核命令可以把 manual_review 改回 prepared；该命令必须记录「确认未生效并批准重试」，后续仍复用原 operation ID。若人工决定不再重试，则转入 failed。

owner/token 条件更新只能证明发送前一刻 lease 有效，不能把数据库事务与外部请求合并成原子操作。worker 仍可能在写入 started 后暂停、失去 lease，再恢复发送。因此，非幂等操作即使远端暂时返回 not found，也必须证明旧执行器已终止、其执行代次已被下游 fencing，或其他条件足以保证它不可能随后发送，并取得明确的重试批准，才允许重试。

更一般地说，外部查询返回 not found 不一定表示操作从未发生。只有查询结果来自权威事实源，或者已经越过外部系统声明的一致性窗口，并且能够排除旧请求仍在途中，才允许把它解释为「未执行」。无法排除时应保留 manual_review/unknown；若外部系统提供原子幂等键，则重投也必须沿用同一 operation ID。

9.4 不确定副作用的恢复时序

9.5 Shell 为什么最难

Shell 命令可能同时：

修改 workspace；
调用外部网络服务；
向数据库写数据；
创建后台进程；
修改系统级配置。

仅对 workspace 做 snapshot 无法回滚其他副作用。安全策略通常是：

将明显的纯读命令标为可重试；
将文件写入包装为 compare-and-swap；
给远端 CLI 保存 job/request ID；
对无法证明幂等的命令，恢复后展示「结果未知」，由人工检查；
高风险命令在执行前持久化审批和 operation 记录。

十、核心机制五：沙箱与运行中资源

10.1 Workspace 持久化不等于执行环境恢复

持久卷可以保存：

源码修改；
构建产物；
下载文件；
工具生成的中间文件。

它通常不能保存：

内存中的 Shell 进程；
PTY 缓冲区；
浏览器页面的 JavaScript 堆；
临时监听端口；
内存数据库；
未写入磁盘的编辑器缓冲区。

10.2 资源恢复需要稳定标识

长运行工具应返回可重新连接的标识：

资源	应保存的标识	恢复动作
云端构建	job ID	查询状态并继续拉取日志
Kubernetes Job	namespace + name + UID	重新 watch
浏览器服务	remote session ID	attach；不支持时明确中断
远端 Shell	execution ID	查询 exit code 和日志
沙箱容器	sandbox ID + workspace volume	重新获取 client，验证容器健康

如果工具只返回一个进程内对象句柄，worker 崩溃后就没有恢复入口。

10.3 沙箱所有权和 Run 所有权应分离

一个 sandbox 可能跨多个 turn 复用，而一个 run 只对应一次执行。销毁 sandbox 的 lease 与执行 run 的 lease 具有不同生命周期：

run owner 决定谁可以推进本次工作；
sandbox owner 决定谁负责续租、回收或销毁容器；
sandbox 的使用权可能在 turn 之间转移；
sandbox 销毁期间必须阻止新的 acquire。

Reaper 不能只靠定时扫描后直接销毁。扫描到过期的 del: 记录后，它需要通过 CAS 写入独立的 deletion_owner_id、deletion_fencing_token 和 deletion_lease_until；只有获取成功的 reaper 才能继续，并在销毁期间按 owner/token 条件续租。销毁操作本身也应按 sandbox ID 幂等。该 deletion lease 与 run 执行 lease 相互独立。

DeerFlow 源码已经明确区分「谁负责回收容器」与「谁可以使用容器」，这是多实例沙箱管理中很重要的设计细节。

10.4 沙箱恢复架构

恢复流程应先确认 sandbox 是否仍健康，再判断 workspace 是否完整，最后单独处理运行中进程。容器存在不代表进程仍属于原 run。

十一、核心机制六：多 Agent 恢复

多 Agent 系统不能只保存子进程 PID。父子关系必须成为 durable state：

11.1 创建 child 的原子性

以下两个事实必须同时成立：

child run 已经创建；
parent checkpoint 记录自己正在等待这个 child。

如果先创建 child、后保存 parent，崩溃后 parent 可能重复创建 child。如果先保存 parent、后创建 child，parent 可能永远等待不存在的任务。

可选方案包括：

在同一数据库事务中写 parent dependency 与 child run；
写一条 child_requested event，再由幂等 dispatcher 创建 child；
child run ID 由 parent step 确定性生成，重复创建命中唯一约束。

11.2 子 Agent 结果投递

child 完成后，不应直接调用父进程的内存 callback。更可靠的流程是：

child 形成 terminal checkpoint 和结构化结果；
若 child checkpoint、run 状态和 outbox 共用事务数据库，则在同一事务写入 child 终态与 parent inbox/outbox event；
若 checkpointer 是独立物理存储，则先保存 child terminal checkpoint，再由幂等 relay 或恢复器补写 parent event；
dispatcher 至少一次投递；
parent 按 child run ID 去重；
parent 重新检查全部依赖，再决定是否继续。

11.3 Parent 崩溃后的策略

需要明确三种策略之一：

child 独立完成，新 parent owner 随后收集结果；
child 随 parent 取消；
child 进入 orphan 状态，由调度器决定是否领养。

不能把 parent 进程死亡直接等同于 child 业务任务失效。

11.4 多 Agent 恢复时序

十二、核心机制七：Human-in-the-loop

等待人工批准是一种正常持久状态，不是异常。

恢复时节点可能从开头重新执行，因此 interrupt 之前的副作用必须放入可缓存 task，或者具备幂等性。LangGraph 的 interrupt() 文档明确说明恢复会重新执行节点逻辑，而不是恢复函数栈。LangGraph Interrupts

十三、开源实现一：LangGraph

13.1 持久化模型

LangGraph 的 checkpointer 在每个超级步保存 StateSnapshot。核心字段包括：

values：当前 channel 值；
next：下一步要执行的节点；
config：thread_id、checkpoint_ns、checkpoint_id；
metadata：步骤编号和节点写入；
tasks：待执行任务、错误和 interrupt；
parent_config：上一个 checkpoint。

同一超级步中，每个节点完成后会通过 put_writes 保存 task-level writes。另一个节点失败时，成功节点的 writes 作为 pending writes 被复用。Checkpoint README

13.2 写入顺序

Pregel loop 的实现会等待产生 delta 的 put_writes，再让引用这些 writes 的 checkpoint 变成 durable，避免读取到「checkpoint 已存在、增量却尚未保存」的状态。Pregel loop source

sync 模式还会在下一 tick 前等待 checkpoint future。async 模式允许下一步与写入并行，换取吞吐。Pregel main source

13.3 恢复行为

假设 A、B 并行，A 成功、B 失败：

A 的 writes 已保存；
B 的错误也可作为 task write 保存；
graph 保持在当前超级步；
使用同一 thread_id 重新调用；
A 结果被复用，B 重新执行；
超级步完成后生成完整 checkpoint。

13.4 Interrupt 的恢复

interrupt() 首次执行时抛出可恢复中断，checkpointer 保存当前状态。调用方随后以 Command(resume=...) 重新执行。节点从函数开头运行，interrupt 根据稳定顺序消费之前保存的 resume value。

这意味着：

def approval_node(state):
    send_email("approval requested")   # 可能重复
    approved = interrupt("approve?")
    return {"approved": approved}

更可靠的写法是：

@task
def notify_once(operation_id):
    return mailer.send(
        template="approval requested",
        idempotency_key=operation_id,
    )

def approval_node(state):
    notify_once(stable_id(state["run_id"], "approval_notice")).result()
    approved = interrupt("approve?")
    return {"approved": approved}

task 结果被保存后，resume 重放可以复用它；若 task 在完成外部发送后、保存结果前退出，邮件服务仍需要支持幂等键。

13.5 优点

工作流状态语义明确；
checkpoint history 支持 time travel 和 fork；
pending writes 缩小失败后的重算范围；
interrupt 与持久化天然结合；
多种 checkpointer 可替换；
task 机制能够隔离非确定性操作。

13.6 边界

核心库不负责 worker lease、孤儿 run 扫描和自动调度；
replay 会重跑 checkpoint 之后的节点，包括 LLM、API 和 interrupt；
节点内部副作用仍需幂等；
InMemorySaver 只适合测试，进程退出后无法恢复；
exit durability 不适合中途恢复；
stream 事件不等同于长期 durable event log。

因此，LangGraph 更像可靠计算状态机，而不是完整的 Agent 托管平台。

十四、开源实现二：DeerFlow

DeerFlow 在 LangGraph 之上增加了 run、线程、事件流、沙箱和 Gateway 层，是研究「图引擎如何进入产品」的典型样本。

14.1 Run 持久化

RunRecord 保存：

run_id、thread_id；
pending/running/success/error/timeout/interrupted 状态；
owner_worker_id；
lease_expires_at；
disconnect 和 multitask 策略；
error、stop reason 和运行进度。

创建 run 时先写入持久 store。如果保存失败，会移除内存记录，避免出现只在当前进程存在的幽灵 run。

14.2 Heartbeat 与孤儿任务扫描

多 worker 模式启用 heartbeat：

lease 更新周期约为 lease_seconds / 3；
heartbeat loop 每个 lease 周期扫描一次过期 owner；
Gateway lifespan 在启动时另行执行一次扫描；
本地 worker 若续租失败，会设置 abort 并取消本地 task；
store 中租约已过期的 pending/running run 被标记为 error。

对应实现位于 RunManager。

14.3 启动恢复

Gateway 启动时：

初始化持久化后端和 RunManager；
扫描过期或无 lease 的 in-flight run；
将其标为 error；
为已恢复 run 的 stream 发布结束标记；
更新相关 thread 的错误状态；
启动后台 heartbeat。

参见 Gateway dependencies。

14.4 运行关闭

正常关闭时，DeerFlow 先停止 heartbeat，再取消并有限等待 in-flight run。目的是让 LangGraph 在 checkpointer 连接池关闭前写完最终 checkpoint。

这个顺序避免了：

run 仍在执行；
数据库连接池先关闭；
LangGraph 后台 checkpoint task 写入失败；
asyncio shutdown 报告未处理异常。

如果 run 在 drain 期间正常完成，则保留真实终态；只有仍未完成或被取消的 run 才会写为 interrupted。整个过程受 shutdown timeout 限制。

14.5 SSE 回放

Redis stream bridge 支持：

跨 worker 发布事件；
按 Last-Event-ID 回放；
rolling TTL；
启动恢复时为仍保留 stream 的孤儿 run 发布结束标记，并延迟清理。

周期 orphan scan 只调用 RunManager，将孤儿 run 标为 error。该路径没有 Stream Bridge 引用，因此不会发布 END sentinel，也不会执行 Gateway 启动路径中的最新 thread 状态修复。启动恢复发布的只是 END sentinel，不是显式的 SSE error 事件；错误事实保存在 run 状态中，启动路径还会条件更新相关的最新 thread 状态。

但 README 明确说明，stream bridge 本身不使取消、请求去重和 IM state 自动具备跨 worker 一致性。Gateway 因此默认使用单 worker，增加 worker 前需要显式 sticky routing 或所有权设计。DeerFlow README

14.6 沙箱恢复

DeerFlow 的远端 sandbox 具有独立 ownership store：

take() 用于 turn 到达不同实例时转移负责者；
claim() 用于 adopt 或 destroy；
lease 状态区分 own: 和 del:；
destroy 标记必须在整个销毁期间续租；
owner 死亡后 del: 最终随 TTL 过期。

这说明 sandbox recovery 不只是「容器还在不在」，还包括「哪个实例有权复用或销毁」。

14.7 完整崩溃时序

14.8 优点与边界

优点：

图 checkpoint、run lease、SSE replay、thread state 和 sandbox ownership 已经分层；
启动扫描与周期扫描同时存在，但只有启动恢复负责发布保留 stream 的 END sentinel；
正常关闭会先 drain run，再关闭 checkpointer 连接池；
run 状态更新带条件保护，减少终态被覆盖。

边界：

孤儿 run 当前主要被标为 error，不会自动重新调度原图；
Gateway 内存仍拥有 active task，stream bridge 不能替代 task ownership；
browser session 仍是进程本地资源；
外部工具的窗口 D 需要业务层幂等和对账。

DeerFlow 的价值在于展示完整平台骨架，同时也清楚暴露了「状态收敛」和「自动续跑」之间的距离。

十五、开源实现三：Open Managed Agents

Open Managed Agents 采用 event log、stream row 和 turn marker 处理崩溃。它的恢复目标很克制：让历史重新变得一致，让下一次请求可以继续，而不是猜测原工具是否应该重试。

15.1 冷启动发现

开始 turn 时，session 行写入：

status='running'；
turn_id；
turn_started_at。

正常结束使用 WHERE turn_id = ? 清理运行标记。旧 recovery 即使与新 turn 竞争，也无法覆盖新 turn。

冷启动的 onWake() 会查询仍为 running 的 turn，并过滤当前实例真正拥有的 active turn。Session state machine

15.2 两类恢复

recoverInterruptedState() 处理：

模型流中断：读取 streaming row，将已保存 chunk 合并成 partial assistant message，再把 stream 标为 interrupted。
工具协议悬空：查找没有 result 的 agent.tool_use 或 agent.mcp_tool_use，追加一个明确的中断 result。

对于 agent.custom_tool_use，服务端不伪造 user.custom_tool_result，只发 warning，等待调用方确认。Recovery source

15.3 为什么要补占位 Tool Result

部分模型 API 要求每个 tool use 后必须存在对应 tool result。若崩溃留下单独 tool use，下一次请求会直接被模型服务拒绝。

占位 result 的作用是修复协议，而不是声明工具没有执行。固定提交中的两条路径并不完全相同：

普通 agent.tool_result 携带原 tool_use_id 和 interrupted 文本，但没有设置 is_error；
agent.mcp_tool_result 携带原 mcp_tool_use_id 和 interrupted 文本，并设置 is_error=true；
recovery warning 通过返回值交给实时订阅者，不写入 event log。

15.4 幂等恢复

从状态机语义看，partial stream、普通 tool use 和 MCP tool use 在首次恢复提交后再次运行恢复函数时：

stream 已不是 streaming；
tool use 已存在 result；
session 已不是 running。

因此不应再次追加占位结果。固定提交中的单元级二次 recovery 测试覆盖 stuck stream 和普通 tool use，没有单独覆盖 MCP tool use；进程级 double-restart 去重也只覆盖普通 tool use。

agent.custom_tool_use 是例外：服务端既不注入 result，也不把 warning 写入历史，所以只要对应的 user.custom_tool_result 仍未出现，每次扫描都可能再次返回 warning。不能把这一分支描述为严格的二次 no-op。

15.5 真实 SIGKILL 测试

项目的进程级测试会启动真实 Node.js 服务，对子进程发送 SIGKILL，再使用同一 SQLite 数据目录重启。各用例分别验证一次 partial stream、普通 tool use、MCP tool use 和 custom tool use 恢复，以及 running session 收敛为 idle。double-restart 用例只验证普通 tool use 不会在第二次启动时重复注入。

单个进程级用例大致执行：

启动真实 Node.js 服务；
使用 SQLite 构造运行中 session、partial stream 或悬空 tool use；
对子进程发送 SIGKILL；
使用同一数据目录重启；
查询数据库验证恢复结果；
在普通 tool use 的 double-restart 用例中再次重启，验证不会重复注入。

参见 crash-recovery.test.ts 和 recovery.test.ts。

15.6 优点与边界

优点：

event log、stream row 和 turn marker 共同构成恢复事实来源；
已持久化的 stream chunk 可以恢复；
tool use/result 协议可自动修复；
turn ID 条件更新防止旧恢复覆盖新执行；
真实进程死亡测试比只 mock exception 更可信。

边界：

恢复不会自动重试工具；
占位 result 无法说明外部副作用是否已经发生；
Node.js SQL adapter 的同步 append() 内部为 fire-and-forget，源码注明存在很小的 lost-write window；需要强持久保证的流程应等待 appendAsync()；
不同存储适配器仍需验证相同的事务和排序语义。

15.7 与 LangGraph 的差异

LangGraph 试图恢复计算进度，Open Managed Agents 试图恢复事件历史的一致性。两者可以组合：

LangGraph 决定从哪个节点继续；
event log 修复 partial message 和工具协议；
run lease 决定哪个 worker 有权继续；
operation store 决定外部工具能否重试。

十六、开源实现四：AgentScope

AgentScope 的 app 层提供持久 AgentState、分布式会话锁、事件回放、跨 session projection 和读时修复。

16.1 Session 状态来源

Session 状态由两个事实合成：

Redis session lock 是否存在；
持久化 AgentState.context 的末尾是否等待 permission 或 external result。

只要集群中有 worker 持有 lock，状态就是 RUNNING。没有 lock 时，才根据持久 context 计算 AWAITING_PERMISSION、AWAITING_EXTERNAL_RESULT 或 IDLE。Session service

这避免了用运行期间的旧 snapshot 错判状态。worker 崩溃后 lock TTL 到期，session 会自动离开 RUNNING。

16.2 Chat 持久化顺序

ChatService 在分布式 session lock 内运行：

event 先同步应用到内存 reply；
再发布到 replay log 和 live channel；
run 结束时统一保存 reply 和 AgentState；
最终保存放在独立 coroutine 中；
使用 asyncio.shield() 抵抗 cancellation；
保存完成后才释放 session lock。

参见 ChatService。

这种顺序可以避免正常取消发生在 await 边界时丢掉刚产生的 event，也避免另一个 worker 获取锁后读到旧 snapshot。

16.3 Replay 与 Projection

Redis Stream 同时支持：

追加事件；
cursor 读取；
live pub/sub；
运行结束后 trim。

跨 session 的 HITL UI 卡片则保存在无 TTL registry。projection 不是权威事实；真正状态在拥有者 session 的 context。旧 projection 会在 SSE replay 时 reconcile-on-read。Session projection

16.4 硬崩溃边界

asyncio.shield() 只能抵抗任务 cancellation，不能抵抗 SIGKILL、OOM 或机器断电。硬崩溃后：

lock 最终过期；
已成功保存的 snapshot 可恢复；
未执行完的 finally 不会运行；
正在进行的 LLM 和工具不会自动继续；
replay log 中可能有比 snapshot 更新的事件，需要定义权威关系。

16.5 锁实现值得验证的窗口

当前 Redis heartbeat 使用普通 EXPIRE，释放过程是 token 检查后的 GET 与 DEL。续租没有在同一原子操作里验证 token，释放也不是原子 compare-and-delete。

当旧 worker 暂停超过 TTL、新 worker 已重新获取锁、旧 worker 又恢复时，理论上存在旧 heartbeat 或释放动作干扰新 owner 的窗口。更稳妥的实现应：

使用 Lua 原子比较 token 后续租；
使用 Lua 原子比较 token 后删除；
数据库写入携带 fencing token；
故障测试主动暂停 worker 超过 TTL，再恢复旧进程。

这是源码审阅得到的风险点，不等同于已经观察到的生产故障。

十七、开源实现五：OpenCode

OpenCode 以数据库中的 message 和 part 持续保存 Agent 过程。它不是分布式工作流引擎，但对悬空工具协议做了实用处理。

17.1 增量持久化

运行过程中会更新：

assistant message；
text delta；
reasoning part；
tool part；
step-start、step-finish；
workspace snapshot 和 patch；
token、cost 和完成时间。

正常取消时，processor cleanup 会结束文本和 reasoning，把仍在运行的工具标为 Tool execution aborted，写入 metadata.interrupted=true，再完成 assistant message。Session processor

17.2 SIGKILL 后的协议投影

硬崩溃不会执行 cleanup，数据库可能保留 pending 或 running tool part。生成下一次模型上下文时，message-v2 会把这两种状态转换为 [Tool execution was interrupted] 的 error output，避免形成没有 result 的 tool use。Message projection

17.3 处理位置的差异

Open Managed Agents 在冷启动时向 event log 写入恢复事件。OpenCode 在读取历史并构造模型请求时修复投影。

方式	优点	代价
写入恢复事件	历史自解释，后续读取一致	recovery 会修改 canonical log
读取时投影	不修改原始记录，兼容旧数据	原始状态仍可能显示 running，每次读取都要应用规则

17.4 边界

没有通用的跨 worker lease 和接管；
provider retry 解决短暂 API 故障，不等于进程恢复；
悬空工具被视为中断，不会自动续跑；
工具实际是否已经产生外部效果仍然未知。

十八、开源实现六：Codex

Codex 的恢复核心是 append-only rollout 和 thread history reconstruction。

18.1 Rollout 保存的内容

Rollout 不只保存自然语言消息，还包括：

session metadata；
turn started、complete、aborted；
user message 和 response item；
turn context；
compaction replacement history；
world state；
multi-agent communication；
rollback event。

resume 从 thread store 或 rollout 文件重建历史、上一个 turn 设置、reference context 和 world state baseline。Rollout reconstruction

18.2 中断边界

thread manager 能判断 snapshot 是否结束在 turn 中间。创建 interrupted fork 时，会追加 TurnAborted 以及可选的中断说明，防止不完整 turn 被当成正常完成。Thread manager

18.3 可恢复与不可恢复

可恢复：

已写入 rollout 的历史；
已完成的上下文压缩；
已写入 workspace 的文件；
已保存的多 Agent 元数据。

不可直接恢复：

原 Rust async task；
运行中的 Shell 子进程；
未保存的工具结果；
模型网络连接；
外部 API 是否已经生效。

因此 Codex 的 resume 主要是让新的 Agent turn 根据历史继续工作，而不是恢复原进程的执行栈。

十九、开源实现七：Pi

Pi 使用树形 JSONL 保存 session，并提供人工 continue、resume、fork 和 clone。

19.1 Append-only Session

每个 entry 一行 JSON，使用 id/parentId 形成树。历史包含消息、模型切换、thinking level、label、compaction 和 branch summary。

SessionManager 使用同步 append。读取时逐行解析并跳过 malformed line，因此进程在最后一行写到一半时，前面的完整 entry 仍可恢复。SessionManager

19.2 Orchestrator 重启

Pi orchestrator 启动时不会假设旧 RPC 子进程仍存在，而是把持久记录中的 online 或 starting 改为 stopped，并清理 presence。Orchestrator supervisor

这是状态收敛，不是任务重建。原 session 仍可由人工继续，但 child process 和 in-flight turn 不会自动恢复。

19.3 边界

跳过 malformed tail 可以抵抗普通进程崩溃；
没有显式 fsync 时，不等于抵抗整机掉电后的 page cache 丢失；
session tree 解决历史分支，不解决工具副作用；
orchestrator 将实例置为 stopped，不会自动重新 spawn 并 replay 原命令。

二十、横向比较

20.1 机制比较

项目	权威持久状态	孤儿检测	恢复粒度	所有权	事件回放	默认处理
LangGraph	checkpoint + pending writes	由上层负责	节点／超级步	无平台级 lease	核心 stream 非长期日志	相同 thread 再调用后恢复
DeerFlow	checkpoint + run + thread + run event	启动与周期 lease scan	图状态 + run 收敛	owner + lease + heartbeat	Redis + Last-Event-ID	孤儿 run 标为 error
Open Managed Agents	event log + stream row + turn marker	onWake 查询 running turn	流和工具协议	turn ID 条件更新	历史事件 replay	补中断结果并回 idle
AgentScope	AgentState/message snapshot + Redis state	session lock TTL	最近持久 snapshot	Redis lease lock	Redis Stream cursor	离开 RUNNING，恢复 parked/idle
OpenCode	database message/part	读取时识别悬空 tool	消息 part	主要为进程内	应用事件	投影为 interrupted error
Codex	rollout + thread store	人工 resume	历史／turn 边界	本地 thread manager	rollout replay	新 turn 继续
Pi	JSONL session tree	人工 resume；orchestrator 清理	session entry	本地实例状态	JSONL replay	实例置 stopped

20.2 能力分级

下表依据本文的四级模型做定性评估，用于比较机制覆盖范围，不代表项目官方评级或基准测试结果。

项目	L1 会话恢复	L2 状态收敛	L3 执行可续跑	L4 副作用恢复
LangGraph	强	强	强；限 checkpoint 边界，需上层重新调用和调度	需要业务设计
DeerFlow	强	强	中；底层图具备能力，孤儿 run 默认失败	需要业务设计
Open Managed Agents	强	强	中弱；修复后等待后续输入	保守处理，不自动重试
AgentScope	强	中强	中弱；回到最近 snapshot	需要业务设计
OpenCode	强	中	弱；人工继续	需要业务设计
Codex	强	中	弱；人工继续	需要业务设计
Pi	强	中弱	弱；人工继续	需要业务设计

20.3 三类设计路线

这些项目大致分为三类：

工作流内核：LangGraph 解决状态机的 durable execution。
托管平台：DeerFlow、Open Managed Agents、AgentScope 解决 run、session、stream、sandbox 和多实例协调。
交互式 Coding Agent：Codex、OpenCode、Pi 解决会话保存、历史重建和人工恢复。

评价恢复能力时，应先判断项目属于哪一类。要求本地 CLI 具备跨集群 lease 并不合理；把一个平台的 session resume 当作步骤级 durable execution 同样不准确。

二十一、一次完整的崩溃恢复应该怎样发生

21.1 正常执行时序

这里有三个不可省略的等待点：

API 等待 inbox 和 run 记录提交后才返回 202；
worker 等待 tool result 保存后才推进 checkpoint；
终态 checkpoint、run 终态、terminal event 和 outbox 在同一事务中形成。

该时序展示的是本文参考实现：这些表位于同一个事务数据库。若 LangGraph checkpointer 与 run store 是独立物理存储，只能采用较弱语义：先提交 terminal checkpoint，再更新 run 终态和 outbox。两次提交之间若发生崩溃，run 可能暂时仍是 running，恢复器必须读取 terminal checkpoint，幂等修复 run 终态和缺失事件。

21.2 Worker 崩溃后的接管时序

21.3 恢复模式应由策略决定

不同 run 可以配置不同策略：

策略	适用任务	孤儿 run 的处理
`fail_fast`	低延迟聊天、不可安全重试的工具	标为 interrupted/error，等待新输入
`resume_safe`	纯读 research、确定性构建	从 checkpoint 自动续跑
`reconcile_then_resume`	支持幂等键或远端查询的写操作	先对账，再继续
`manual_on_unknown`	支付、发布、删除等高风险操作	unknown 时暂停并请求人工确认

平台不应对所有工具使用一个统一的「失败后重试」开关。

二十二、参考实现：从请求接入到恢复器

22.1 接入层：请求先持久化

客户端必须在第一次发送请求前生成稳定 request ID。服务端响应后才生成的 ID 无法覆盖「服务端已提交、响应在网络中丢失」这一窗口。request ID 还必须限制在 tenant、project 或 organization 等作用域内，不能作为全局裸 ID 使用。

服务端从认证上下文取得 tenant ID，再对会影响执行语义的字段做规范化并计算 request hash。tenant ID 不能直接信任请求体。hash 至少覆盖 thread、输入内容和执行配置，不应包含每次重试都会变化的时间戳或认证头：

async def accept_request(auth, request):
    if not request.request_id:
        raise ClientRequestIdRequired()

    tenant_id = auth.tenant_id
    request_hash = canonical_hash(
        request.to_idempotency_payload()
    )

    async with db.transaction():
        existing = await runs.find_by_request_key(
            tenant_id=tenant_id,
            request_id=request.request_id,
        )
        if existing:
            if existing.request_hash != request_hash:
                raise IdempotencyConflict(request.request_id)
            return existing

        inserted = await inbox.insert_if_absent(
            tenant_id=tenant_id,
            request_id=request.request_id,
            thread_id=request.thread_id,
            request_json=request.to_json(),
            request_hash=request_hash,
            accepted_at=db.now(),
        )
        if not inserted:
            accepted = await inbox.get(
                tenant_id=tenant_id,
                request_id=request.request_id,
            )
            if accepted.request_hash != request_hash:
                raise IdempotencyConflict(request.request_id)

            existing = await runs.find_by_request_key(
                tenant_id=tenant_id,
                request_id=request.request_id,
            )
            if existing:
                return existing
            raise InconsistentInbox(request.request_id)

        run = await runs.insert(
            tenant_id=tenant_id,
            request_id=request.request_id,
            request_hash=request_hash,
            thread_id=request.thread_id,
            status="queued",
        )
        await outbox.insert(
            topic="run.dispatch",
            event_key=run.id,
            run_id=run.id,
            event_seq=None,
            payload_json={"run_id": run.id},
            created_at=db.now(),
        )
    return run

相同作用域、相同 request ID 和相同 hash 的 HTTP 重试返回原 run；相同 ID 对应不同 hash 时返回幂等冲突，例如 HTTP 409 Conflict，不能静默复用旧任务。outbox 确保数据库提交后，即使 API 进程立即退出，dispatcher 仍会发送任务。

22.2 Lease 获取、续租与重复投递

Lease 获取和续租必须是两条不同的状态转换。worker 只从 queued 获取新 lease；过期的 running run 先由 Reconciler 对账并改回 queued。获取新 lease 时递增 fencing token：

UPDATE run
   SET owner_id = :worker_id,
       lease_until = NOW() + INTERVAL '60 seconds',
       fencing_token = fencing_token + 1,
       status = 'running',
       updated_at = NOW()
 WHERE id = :run_id
   AND status = 'queued'
   AND (
       owner_id IS NULL
       OR lease_until IS NULL
       OR lease_until < NOW()
   )
RETURNING fencing_token, checkpoint_id;

续租只延长当前 lease，不递增 token，并且同时匹配 owner 和 token：

UPDATE run
   SET lease_until = NOW() + INTERVAL '60 seconds',
       updated_at = NOW()
 WHERE id = :run_id
   AND status = 'running'
   AND owner_id = :worker_id
   AND fencing_token = :token
   AND lease_until > NOW()
RETURNING fencing_token, lease_until;

同一 worker 收到重复队列消息时，应先按 run_id 返回已有本地任务。即使两个 handler 同时越过本地检查，数据库获取语句也只允许一个成功；条件中不能加入 owner_id = :worker_id，否则第二次投递会递增 token，把第一个本地任务变成旧 owner。

async def handle_delivery(run_id, worker_id):
    existing = local_tasks.get(run_id)
    if existing is not None:
        return await existing

    lease = await lease_store.acquire_queued(run_id, worker_id)
    if lease is None:
        existing = local_tasks.get(run_id)
        return await existing if existing is not None else None

    task = asyncio.create_task(
        run_once(run_id, worker_id, lease.fencing_token)
    )
    local_tasks[run_id] = task
    try:
        return await task
    finally:
        if local_tasks.get(run_id) is task:
            del local_tasks[run_id]

需要注意：

数据库时间比 worker 本地时间更适合作为租约判断依据；
renew 必须同时匹配 owner 和 fencing token，且不能递增 token；
terminal run 不允许再次 acquire；
获取成功后，worker 才能创建本地执行 task。

22.3 执行循环

以下伪代码中的 checkpoints 绑定当前 db.transaction()，写入前述 run_checkpoint 表。若使用独立 checkpointer，不能直接套用这个事务边界，应改用第 21.1 节说明的终态修复流程。

async def run_once(run_id, worker_id, token):
    checkpoint = await checkpoints.load(run_id)
    machine = rebuild_state_machine(checkpoint)

    while not machine.terminal:
        if not await lease_store.is_valid(run_id, worker_id, token):
            raise LeaseLost(run_id, worker_id, token)

        step = machine.next_step()

        result = await execute_step_with_operations(
            run_id=run_id,
            step=step,
            worker_id=worker_id,
            fencing_token=token,
        )
        next_state = machine.apply(result)
        event = (
            run_terminal_event(next_state)
            if next_state.terminal
            else step_completed_event(step, result)
        )

        async with db.transaction():
            await lease_store.require_valid_for_update(
                run_id, worker_id, token
            )
            seq = await run_events.allocate_next_seq(run_id)
            committed_checkpoint = await checkpoints.commit(
                run_id=run_id,
                previous_id=(checkpoint.id if checkpoint else None),
                next_state=next_state,
                fencing_token=token,
            )
            await operation_store.mark_applied(
                operation_ids=result.operation_ids,
                checkpoint_id=committed_checkpoint.id,
            )
            if next_state.terminal:
                await runs.mark_terminal_if_owner(
                    run_id=run_id,
                    worker_id=worker_id,
                    fencing_token=token,
                    status=next_state.status,
                )
            await run_events.insert(
                run_id=run_id,
                seq=seq,
                event_type=event.type,
                payload_json=event.payload,
                created_at=db.now(),
            )
            await outbox.insert(
                topic="run.events",
                event_key=f"{run_id}:{seq}",
                run_id=run_id,
                event_seq=seq,
                payload_json=event.to_json(),
                created_at=db.now(),
            )

        checkpoint = committed_checkpoint
        machine.accept(next_state)

每次提交都在事务内再次检查 token。只在循环入口检查一次 lease 不够，因为旧 worker 可能在长工具调用期间失去所有权。显式抛出 LeaseLost 也不能替换事务内 fencing；它只是让本地任务尽早停止。长工具调用已经产生的外部效果仍由 operation ID 和对账机制处理。

22.4 Recovery Reconciler

Reconciler 不成为 run 的执行 owner，但必须成为本次恢复工作的 owner。首次认领会 fencing 过期 worker；若前一个 Reconciler 已经把 run 改为 orphaned 后崩溃，新 Reconciler 可以在 recovery lease 过期后再次认领：

UPDATE run
   SET status = 'orphaned',
       owner_id = NULL,
       lease_until = NULL,
       fencing_token = fencing_token
           + CASE WHEN status = 'running' THEN 1 ELSE 0 END,
       recovery_owner_id = :reconciler_id,
       recovery_lease_until = NOW() + INTERVAL '60 seconds',
       recovery_fencing_token = recovery_fencing_token + 1,
       updated_at = NOW()
 WHERE id = :run_id
   AND (
       (
           status = 'running'
           AND (lease_until IS NULL OR lease_until < NOW())
       )
       OR
       (
           status = 'orphaned'
           AND (
               recovery_lease_until IS NULL
               OR recovery_lease_until < NOW()
           )
       )
   )
RETURNING fencing_token AS execution_fence,
          recovery_fencing_token AS recovery_token,
          checkpoint_id;

class RecoveryAction(Enum):
    RETRY = "retry"
    REUSE = "reuse"
    FAILED = "failed"
    MANUAL = "manual"


def decide_from_external_fact(op, fact):
    if fact.kind == "succeeded":
        return RecoveryDecision.reuse_remote_result(
            op,
            operation_update=OperationUpdate(
                status="succeeded",
                result=fact.result,
                remote_request_id=fact.remote_request_id,
            ),
        )

    if fact.kind == "failed":
        return RecoveryDecision.use_remote_failure(
            op,
            operation_update=OperationUpdate(
                status="failed",
                error=fact.error,
                remote_request_id=fact.remote_request_id,
            ),
        )

    reliable_no_effect = (
        fact.kind == "not_executed"
        and (fact.authoritative or fact.consistency_window_elapsed)
    )
    non_idempotent_retry_is_safe = (
        reliable_no_effect
        and fact.old_executor_cannot_send
        and fact.retry_approved
    )
    if reliable_no_effect and (
        op.idempotency_guarantee == "strong"
        or non_idempotent_retry_is_safe
    ):
        return RecoveryDecision.retry_after_reconciliation(
            op,
            operation_update=OperationUpdate(status="prepared"),
        )

    # unknown、defer 和未来新增的 fact kind 全部 fail-closed。
    return RecoveryDecision.require_manual_review(
        op,
        operation_update=OperationUpdate(
            status="manual_review",
            error={"reason": "external fact is not conclusive"},
        ),
    )


async def reconcile_orphan(run, reconciler_id):
    recovery = await runs.claim_recovery(
        run_id=run.id,
        reconciler_id=reconciler_id,
    )
    if recovery is None:
        return

    async with recovery_lease_guard(
        run_id=run.id,
        owner_id=reconciler_id,
        token=recovery.token,
        renew_every_seconds=20,
    ):
        await reconcile_owned(run, reconciler_id, recovery)


async def reconcile_owned(run, reconciler_id, recovery):
    checkpoint = await checkpoints.load(run.id)
    checkpoint_id = checkpoint.id if checkpoint is not None else None
    operations = await operation_store.list_unapplied_operations(
        run_id=run.id,
        checkpoint_id=checkpoint_id,
    )

    decisions = []
    for op in operations:
        if op.status == "prepared":
            decisions.append(RecoveryDecision.retry_existing(op))
            continue

        if op.status == "succeeded":
            decisions.append(RecoveryDecision.reuse_existing(op))
            continue

        if op.status == "failed":
            decisions.append(RecoveryDecision.use_stored_failure(op))
            continue

        if op.status == "manual_review":
            decisions.append(RecoveryDecision.keep_manual(op))
            continue

        if op.status in {"started", "unknown"}:
            # 只查询外部事实，不在这里写数据库。
            fact = await query_external_operation(op)
            decisions.append(decide_from_external_fact(op, fact))
            continue

        # 新增或无法识别的状态默认转人工，不允许自动续跑。
        decisions.append(RecoveryDecision.manual_for_unknown_status(op))

    retry_safe = {
        RecoveryAction.RETRY,
        RecoveryAction.REUSE,
        RecoveryAction.FAILED,
    }
    requires_manual = any(
        decision.action not in retry_safe
        for decision in decisions
    )

    async with db.transaction():
        await recovery_lease.require_valid_for_update(
            run_id=run.id,
            owner_id=reconciler_id,
            token=recovery.token,
        )

        for decision in decisions:
            await operation_store.apply_recovery_decision_if_owner(
                run_id=run.id,
                operation_id=decision.operation_id,
                expected_status=decision.expected_status,
                expected_dispatch_generation=(
                    decision.expected_dispatch_generation
                ),
                update=decision.operation_update,
                recovery_owner_id=reconciler_id,
                recovery_token=recovery.token,
            )

        if requires_manual:
            await runs.mark_uncertain_if_recovery_owner(
                run_id=run.id,
                recovery_owner_id=reconciler_id,
                recovery_token=recovery.token,
            )
            event = manual_review_event(decisions)
        else:
            await runs.mark_queued_if_recovery_owner(
                run_id=run.id,
                recovery_owner_id=reconciler_id,
                recovery_token=recovery.token,
            )
            event = recovery_queued_event(decisions)
            await outbox.insert(
                topic="run.dispatch",
                event_key=f"{run.id}:recovery:{recovery.token}",
                run_id=run.id,
                event_seq=None,
                payload_json={
                    "run_id": run.id,
                    "resume_from": checkpoint_id,
                },
                created_at=db.now(),
            )

        seq = await run_events.allocate_next_seq(run.id)
        await run_events.insert(
            run_id=run.id,
            seq=seq,
            event_type=event.type,
            payload_json=event.payload,
            created_at=db.now(),
        )
        await outbox.insert(
            topic="run.events",
            event_key=f"{run.id}:{seq}",
            run_id=run.id,
            event_seq=seq,
            payload_json=event.to_json(),
            created_at=db.now(),
        )

claim_recovery() 对过期 running run 递增执行 fencing token，以阻止旧 worker 写入；对过期 orphaned run 只递增 recovery token，避免重复改变执行代次。recovery_lease_guard 按 recovery owner/token 条件续租，续租失败时抛出 RecoveryLeaseLost。

query_external_operation() 是只读查询，只返回外部事实。decide_from_external_fact() 将明确成功映射为 REUSE，将明确失败映射为 FAILED；强幂等工具可以映射为 RETRY。非幂等工具只有在「未执行」事实足够可靠、旧执行器不可能再发送并且已批准重试时才能映射为 RETRY。其他结果，包括 unknown、defer 和未来新增但尚未处理的枚举值，一律映射为 MANUAL。

apply_recovery_decision_if_owner() 根据 decision 保存 operation 的 succeeded/result、failed/error、prepared 或 manual_review，同时校验 operation 的预期状态、dispatch generation，以及 run 的 recovery owner/token。即使 decision 不需要修改 operation，该调用也要验证查询后 operation 没有被其他事务改变；CAS 失败时整个事务回滚并重新对账。

这里的 recovery_lease.require_valid_for_update() 必须实际执行 SELECT run ... FOR UPDATE，不能只是普通条件查询。事务取得 run 行锁并验证 recovery owner/token 后，才能调用 apply_recovery_decision_if_owner() 锁定 operation。worker、Reconciler 和人工审核命令都遵守 run 行优先的全局锁顺序。

operation 状态与结果、run 的 queued/uncertain、恢复事件及 outbox 必须在同一个数据库事务中提交。mark_queued_if_recovery_owner() 和 mark_uncertain_if_recovery_owner() 同时匹配 status='orphaned'、recovery_owner_id、recovery_fencing_token 和尚未过期的 recovery_lease_until，成功后清空 recovery owner 和 lease。旧 Reconciler 即使在外部查询完成后恢复运行，也无法提交过期决定。

Reconciler 不把任何 lease token 放进队列。对账完成后，queued 状态与 dispatch outbox 在同一事务中提交；真正收到消息的 worker 再获取新的执行 lease 和 execution fencing token。

list_unapplied_operations() 不等同于「查询运行中操作」。它返回 prepared/started/unknown/manual_review/failed，也返回已经保存成功结果、但 applied_checkpoint_id 为空或不在当前 checkpoint 祖先链中的操作。因此，Reconciler 保存 manual_review 后即使再次崩溃，下次扫描仍会把 run 保持为 uncertain；succeeded -> reuse 分支则覆盖「工具结果已保存、checkpoint 尚未推进」窗口。正常 checkpoint 提交会在同一事务中设置 applied_checkpoint_id。

Recovery lease 与 run 执行 lease 是两组独立字段和状态机。前者只保护「谁有权做对账并提交恢复决定」，后者只保护「谁有权推进 Agent 执行」。

22.5 事件序号

本文选择每个 run 连续且严格递增的事件序号。序号由 run.next_event_seq 在写入 event 和 outbox 的同一事务内分配：

UPDATE run
   SET next_event_seq = next_event_seq + 1
 WHERE id = :run_id
RETURNING next_event_seq - 1 AS allocated_seq;

行锁会串行化同一 run 的并发事件写入；事务回滚时计数器更新也回滚，因此已提交事件从 1 开始连续。这里不能改用独立 PostgreSQL sequence，因为 sequence 在事务回滚、缓存和并发调用时允许出现缺口。若系统只要求严格递增而允许缺口，也可以使用 sequence，但相应的连续性审计必须删除。

客户端需要保存最后成功处理的 seq，而不是最后接收到的 seq。处理失败时不推进 cursor，重连后才能再次读取。

这一分配算法只保证数据库中的序号连续，不自动保证 Relay 的网络发布顺序。发布端必须同时遵守第 7.4 节的 per-run 串行协议。

22.6 Checkpoint 并发控制

checkpoint 提交至少要校验：

run ID；
previous checkpoint ID；
fencing token；
step attempt ID；
当前状态仍允许推进。

这相当于图状态的 compare-and-swap。若 previous checkpoint 已变化，说明另一个 worker 已推进，当前提交必须失败。

二十三、如何验证：不要只测试抛异常

23.1 为什么普通单元测试不够

try/finally、defer、context manager 和 graceful shutdown 都可能在普通测试中运行。真实硬崩溃不会执行这些逻辑。

以下故障需要分别测试：

故障类型	模拟方法	能发现的问题
协作取消	cancel task、发送 `SIGTERM`	cleanup、shield、drain
进程硬退出	`SIGKILL`、容器强制停止	未执行 finally、内存完全丢失
OOM	限制容器内存并触发 kill	写入中断、日志截断、进程树残留
机器掉电	VM 强制关闭、磁盘缓存策略	`fsync` 和 WAL 语义
网络分区	Toxiproxy、iptables、延迟注入	lease 误判、重复 owner
数据库故障	kill connection、只读切换、主从故障	事务边界、重试和未知提交
Redis 故障	主从切换、删除 key、阻塞连接	lock、stream cursor 和 TTL
外部 API 响应丢失	服务执行成功但丢弃响应	不确定副作用处理

23.2 故障注入测试架构

测试控制器不应只查看 HTTP 最终响应。Invariant Oracle 需要直接检查：

run 表；
checkpoint 和 task writes；
tool operation 表；
event seq；
外部 fake API 的实际副作用次数；
workspace 内容；
worker 日志中的 token 和 owner。

23.3 Failpoint 设计

在每个不可逆动作前后提供只在测试环境启用的 failpoint：

await failpoint("after_inbox_commit")
await failpoint("after_lease_acquire")
await failpoint("before_tool_request")
await failpoint("after_tool_effect_before_result_save")
await failpoint("after_result_save_before_checkpoint")
await failpoint("after_checkpoint_write_before_tx_commit")
await failpoint("after_state_commit_before_outbox_relay")
await failpoint("after_event_publish_before_ack")

Failpoint 动作可以是：

阻塞，等待测试控制器发送 SIGKILL；
立即 os._exit(137)；
丢弃下一次网络响应；
让数据库提交成功但客户端收到连接断开；
暂停 event loop 超过 lease TTL。

23.4 核心测试矩阵

编号	崩溃位置	恢复后的期望
1	请求到达、inbox 提交前	客户端在相同认证租户下使用相同 request ID 和内容重试，不产生两个 run
2	inbox 提交后、返回 202 前	相同认证租户、request ID 和 hash 的重试返回原 run，dispatcher 最终派发一次逻辑任务
3	run 创建后、queue publish 前	outbox relay 补发
4	queue deliver 后、lease 获取前	消息重新投递，不推进状态
5	lease 获取后、checkpoint 读取前	lease 过期后由 Reconciler fencing、对账并重新排队
6	checkpoint 读取后、节点开始前	相同节点可安全重跑
7	并行节点 A writes 已保存、B 失败	A 不重跑，B 重跑
8	tool operation 为 prepared	使用原 operation ID 执行
9	外部工具未生效、连接断开	权威查询确认未执行，或越过声明的一致性窗口且排除旧请求仍在途中后，才使用原 operation ID 重投
10	外部工具已生效、响应丢失	查询后保存原结果，不重复副作用
11	外部工具已生效、无法查询	run 进入 uncertain/manual review
12	tool result 已保存、checkpoint 前	复用 result，不再次调用工具
13	checkpoint 提交后、outbox 前	同事务设计下不存在该窗口；否则由扫描补事件
14	event publish 后、标记 published 前	允许重复发送同一 seq，客户端去重
15	SSE 发送一半后 worker 退出	客户端按 cursor 补齐
16	独立物理 checkpointer 已提交 terminal checkpoint、run status 尚未更新	recovery 根据 checkpoint 修复 terminal status；同库事务设计不存在该窗口
17	parent 保存 waiting 前 child 已创建	确定性 child ID 防止重复 child
18	child 完成后 parent 崩溃	parent inbox event 可重放且只消费一次
19	sandbox destroy 中 owner 崩溃	`del:` lease 过期后，reaper 通过 CAS 获取并续租独立 deletion lease，再继续销毁；未获取者退出
20	旧 worker 暂停超过 TTL 后恢复	fencing token 阻止旧 worker 写入
21	Reconciler 持有 recovery lease 时崩溃	recovery lease 过期后由第二个 Reconciler 以新 token 接管；旧 token 恢复后不能提交
22	recovery 与新用户 turn 竞争	turn ID／checkpoint CAS 阻止旧恢复覆盖新状态
23	event log 最后一行或最后一页损坏	前序完整事件可读取，损坏部分可检测
24	checkpoint schema 升级后恢复旧 run	migration 可重入，旧状态能明确转换或拒绝
25	相同 tenant 和 request ID 携带不同 thread、输入或执行配置	返回幂等冲突，不复用旧 run，也不创建第二个 run
26	两个 Outbox Relay 同时处理同一 run 的 N 和 N+1	per-run 发布锁只允许 N 发送；N 未确认前 N+1 保持未发布
27	旧 worker 在 `prepared` 后暂停，lease 过期并被接管后恢复	`prepared -> started` 的 owner/token 条件更新失败，旧 worker 的外部请求次数为 0
28	包装器读取到 `failed` 或 `manual_review`	返回已保存失败或人工确认信号，不取得 dispatch claim，不发送工具请求
29	非幂等 operation 为 `started`，查询暂时返回 `not found`，旧执行器仍可能恢复	保持人工确认，不自动重投
30	`running` run 收到取消，且不存在未确认副作用	取消 checkpoint、`cancelled` 终态和 terminal event 按当前 token 原子提交
31	`running` run 收到取消，但存在 `started/unknown` 非幂等操作	run 进入 `uncertain`，不伪装成安全取消
32	旧 worker 在创建 `prepared` 前失去 lease	`create_if_absent_if_owner` 拒绝插入，不留下过期 operation
33	同一 worker 的两个协程同时读取 `prepared/generation=0`	只有一个 CAS 得到 `generation=1` 的发送权，另一个不调用外部工具
34	外部事实查询返回 `unknown`、`defer` 或未识别 fact kind	operation 进入或保持 `manual_review`，run 进入 `uncertain`，不自动排队
35	Reconciler 查询到成功结果后失去 recovery lease	旧 recovery token 不能保存 operation 结果或 run 决定；新 Reconciler 重新查询并原子提交
36	旧 worker 的 operation 转换与 Reconciler fencing 并发	两者先竞争同一 run 行锁；若 fencing 先提交，旧 token 的转换更新 0 行且不能发送请求
37	同一 dispatch generation 已被另一协程改成 `manual_review/failed/succeeded`	结果更新因 expected status 不匹配而失败，随后读取并采用已有状态，不覆盖终态
38	包装器产生 dispatch in-progress、claim lost 或 retry authorization lost	上层 singleflight 等待、退避或重读 operation，不生成工具失败结果，不把 run 标为失败
39	新 owner 接管并复用同一逻辑 step attempt	`created_under_fencing_token` 保留首次创建 token，当前写入权限只检查锁定后的 `run.fencing_token`

23.5 每个测试都要检查的 Invariant

valid_lease_owner_count(run_id) <= 1
accepted_committer_count(run_id, fencing_token) <= 1
request_key_binds_single_hash(tenant_id, request_id)
operation_identity_is_non_null_and_stable(operation_id)
accepted_dispatch_claim_count(operation_id, dispatch_generation) <= 1
operation_transition_holds_run_lock_before_operation_lock(run_id)
terminal_operation_state_never_regresses(operation_id)
terminal_status_never_regresses(run_id)
successful_result_count(operation_id) <= 1
external_effect_count(idempotency_key) <= 1
event_seq_is_contiguous_from_one(run_id)
published_event_prefix_is_contiguous(run_id)
checkpoint_parent_chain_is_valid(run_id)
recovery_with_same_facts_has_no_duplicate_durable_writes(run_id)
recovery_state_monotonically_converges(run_id)
old_fencing_token_cannot_commit(run_id)
old_recovery_token_cannot_commit(run_id)

这里约束的是「有效所有者和可接受提交者」，不是物理执行进程数。accepted_committer_count 统计同一 token 下被数据库接受写入的不同 worker 身份，不是事务数量。旧 worker 可能在 lease 过期后继续占用 CPU、运行 Shell 或等待外部响应，同时新 worker 已开始恢复；fencing 能保证旧 token 无法提交受保护状态，但不能保证任意时刻只有一个进程在运行。

恢复幂等性以「相同外部事实」为前提：重复扫描不得产生重复事件、重复 dispatch 或重复工具调用。外部操作的查询结果可能稍后才可见，因此第二次恢复可以合法地把 unknown 推进为 succeeded 或 failed；此时要求的是状态单调收敛，而不是两次函数调用返回逐字相同的状态。

23.6 真实外部副作用测试替身

Fake API 应模拟真实的不确定性：

记录每个 idempotency key；
可以执行副作用后主动断开连接；
支持按 request ID 查询；
支持配置查询可见性延迟，用于区分短暂 not found 与权威未执行；
支持配置「不支持幂等」；
统计实际 effect count，而不只是请求次数；
可以延迟成功响应直到 worker 被杀。

只有这样才能验证窗口 D。简单地让 mock 在调用前抛异常，只覆盖了安全重试路径。

二十四、结合开源项目的验证方案

24.1 LangGraph

构造图：

验证步骤：

A 写入确定性结果，B 调用带计数器的 fake API；
分别在 B 调用前、调用后、返回前、put_writes 前杀进程；
使用持久 PostgreSQL checkpointer 和同一 thread_id 恢复；
对比 sync、async、exit；
检查 A 的 pending write 是否复用；
检查 B 的外部 effect count；
给 B 增加 idempotency key 后重复实验。

预期结论：

pending writes 能避免重跑 A；
sync 缩小 checkpoint 回退范围；
B 的工具幂等决定是否重复副作用；
exit 在进程中途退出时缺少可用中间 checkpoint。

24.2 DeerFlow

测试环境：

两个 Gateway worker；
PostgreSQL 作为 run store 和 checkpointer；
Redis stream bridge；
共享 sandbox backend；
一个可控的慢工具。

验证步骤：

run 由 Gateway A 获取；
产生多个 SSE event；
在工具中途 SIGKILL Gateway A；
客户端连接 Gateway B 并携带 Last-Event-ID；
保持 Gateway B 运行，等待 lease 过期和周期 orphan scan；
检查 run 变为 error，并确认该周期路径没有发布 END sentinel；
另建孤儿 run，在 lease 过期后重启 Gateway，验证启动恢复会在 retained stream 存在时发布 END sentinel；
确认 SSE 中只有 END sentinel，错误原因从 run／thread 状态读取；
恢复被暂停而非杀死的 Gateway A，验证它无法覆盖终态；
对 sandbox destroy 和 acquire 做同样的 TTL 竞争测试。

需要特别验证：run store 的条件更新是否足以抵抗旧 worker，而不是只检查 UI 最终显示 error。

24.3 Open Managed Agents

固定提交中的进程级 SIGKILL 用例分别覆盖了一次恢复：

running session 变 idle；
partial stream 合并；
dangling tool use 注入 result；
MCP tool use 注入带 is_error=true 的 result；
custom tool use 不伪造用户结果。

进程级 double-restart 去重只覆盖普通 tool use。单元级二次 recovery 测试覆盖 stuck stream 和普通 tool use，没有单独覆盖 MCP tool use 的二次恢复。因此，现有证据只能说明这些一次恢复路径可工作，以及普通 tool use 的重复恢复不会再次注入结果，不能推广为 partial stream、普通工具和 MCP 工具都已通过进程级重复恢复测试。Crash recovery process tests Recovery unit tests

custom tool 分支只返回 warning，不写入结果或 warning 事件；在用户结果到达前，后续扫描仍可能再次返回 warning。

建议增加：

外部 fake API 已执行成功、tool result 尚未保存；
recovery 执行一半再次 SIGKILL；
event append() fire-and-forget 窗口；
PostgreSQL 与 SQLite 适配器一致性；
recovery 与新 turn 并发；
event ID 命名空间重叠测试；
partial stream 与 MCP tool use 的进程级 double-restart 去重。

24.4 AgentScope

重点不是只杀进程，而是模拟旧 owner 复活：

Worker A 获取 session lock；
使用 SIGSTOP 暂停 A，超过 TTL；
Worker B 获取同一 session lock 并开始 run；
使用 SIGCONT 恢复 A；
检查 A 的 heartbeat 是否会延长 B 的锁；
检查 A 的最终 snapshot 是否可能覆盖 B；
将锁替换为 token-checked Lua 和 fencing token 后重复。

还要测试 event 已写入 replay log、snapshot 尚未保存时的恢复规则，确认 canonical state 的来源。

24.5 OpenCode

验证以下持久状态：

graceful cancel 后 running tool 被标为 interrupted；
SIGKILL 后 running part 仍存在；
下一次消息投影生成对应 error result；
模型请求不包含 dangling tool use；
下一 turn 不会自动重新执行该工具；
workspace snapshot 与实际文件状态一致。

24.6 Codex

验证 rollout 尾部：

turn started 后立即退出；
assistant delta 已记录但 turn complete 缺失；
tool call 已记录、tool result 缺失；
compaction 完成前退出；
multi-agent child 正在运行时 parent 退出。

恢复后检查 history reconstruction 是否保留完整 turn、标记中断 turn，并避免把不完整工具调用当作成功结果。

24.7 Pi

在不同字节位置截断 JSONL 最后一行：

UTF-8 多字节字符中间；
JSON 字符串中间；
entry 结束但换行前；
header 行中间；
compaction entry 中间。

验证前序完整 entry 是否仍能加载，session discovery 是否跳过完全损坏文件，以及 orchestrator 重启后旧 online 实例是否稳定变为 stopped。

二十五、性质测试、模型检查与数据校验

25.1 Property-based Testing

为状态机随机生成动作序列：

accept
dispatch
acquire
renew
start_tool
finish_tool
save_result
save_checkpoint
publish_event
expire_lease
fence_orphan
reconcile_operation
requeue
acquire_new_lease
recover_old_worker
cancel

每一步后检查 invariant。与固定用例相比，性质测试更容易发现：

orphan fencing 与 completion 同时发生；
cancel 与 checkpoint commit 竞争；
recovery 重入；
outbox 重发和 cursor 推进竞争；
parent/child 终态乱序。

25.2 状态机模型检查

对于 lease 和 tool operation，可以用 TLA+、PlusCal、Alloy 或轻量状态遍历验证：

两个 owner 是否可能同时提交；
unknown 是否可能被自动转为 retry；
terminal 状态是否可能回退；
outbox 是否可能永久丢失 committed event；
parent 是否可能永久等待已经完成的 child。

模型不需要还原整个 Agent，只需覆盖最危险的状态转换。

25.3 数据库校验查询

运行期间可周期执行只读审计：

-- 已过期但仍显示运行中的 run
SELECT id, owner_id, lease_until
  FROM run
 WHERE status = 'running'
   AND (lease_until IS NULL OR lease_until < NOW());

-- 有 tool use 但没有 terminal result
SELECT o.id,
       s.run_id,
       o.status,
       o.dispatch_generation,
       o.remote_request_id
  FROM tool_operation AS o
  JOIN step_attempt AS s ON s.id = o.step_attempt_id
 WHERE o.status IN ('started', 'unknown');

-- recovery owner 已过期、可由其他 Reconciler 重新认领的 run
SELECT id,
       recovery_owner_id,
       recovery_lease_until,
       recovery_fencing_token
  FROM run
 WHERE status = 'orphaned'
   AND (
       recovery_lease_until IS NULL
       OR recovery_lease_until < NOW()
   );

-- 连续序号异常；仅适用于从 1 开始且尚未执行 retention 的完整日志
SELECT run_id,
       MIN(seq) AS first_seq,
       MAX(seq) AS last_seq,
       COUNT(*) AS event_count
  FROM run_event
 GROUP BY run_id
HAVING MIN(seq) <> 1
    OR MAX(seq) <> COUNT(*);

-- 已提交但长期未发布的 outbox
SELECT id, topic, event_key, run_id, event_seq, created_at
  FROM outbox
 WHERE published_at IS NULL
   AND created_at < NOW() - INTERVAL '1 minute';

-- 已发布 N+1，但更小的 N 仍未发布
SELECT later.run_id, earlier.event_seq, later.event_seq
  FROM outbox AS later
  JOIN outbox AS earlier
    ON earlier.run_id = later.run_id
   AND earlier.topic = 'run.events'
   AND earlier.event_seq < later.event_seq
 WHERE later.topic = 'run.events'
   AND later.published_at IS NOT NULL
   AND earlier.published_at IS NULL;

审计查询本身不能替代恢复器，但能发现状态长期没有收敛。

二十六、可观测性与恢复 SLO

26.1 必须记录的结构化事件

事件	关键字段
`run.accepted`	tenant_id、request_id、request_hash、run_id、thread_id
`lease.acquired`	run_id、owner_id、token、lease_until
`lease.renewed`	run_id、token、latency
`lease.lost`	run_id、old_token、reason
`checkpoint.committed`	run_id、checkpoint_id、parent_id、step
`tool.prepared`	operation_id、tool、effect_class
`tool.dispatch_claimed`	operation_id、dispatch_generation、run_id、owner_id、fencing_token
`tool.coordination_wait`	operation_id、signal、status、dispatch_generation、wait_ms
`tool.unknown`	operation_id、remote_request_id、error
`recovery.claimed`	run_id、old_owner、recovery_owner_id、execution_fence、recovery_token、recovery_lease_until
`recovery.renewed`	run_id、recovery_owner_id、recovery_token、latency
`recovery.lease_lost`	run_id、recovery_owner_id、recovery_token、reason
`recovery.decision`	run_id、action、reason
`event.replayed`	run_id、from_seq、to_seq、client_id
`run.terminal`	run_id、status、recovery_count

日志必须能用 run ID、step attempt ID、operation ID 和 fencing token 关联，单独记录自然语言错误不够。

26.2 建议指标

agent_runs_inflight；
agent_orphan_runs_total；
agent_recovery_attempts_total{decision}；
agent_recovery_duration_seconds；
agent_recovery_lease_lost_total；
agent_lease_renew_failures_total；
agent_fencing_rejections_total；
agent_tool_unknown_total{tool}；
agent_tool_dispatch_claim_conflicts_total{tool}；
agent_tool_coordination_wait_seconds{signal}；
agent_duplicate_effect_prevented_total{tool}；
agent_outbox_oldest_unpublished_seconds；
agent_event_replay_gap_total；
agent_event_publish_order_violation_total；
agent_manual_review_pending；
agent_checkpoint_commit_seconds{mode}。

26.3 恢复目标

可以把传统 RPO/RTO 映射到 Agent：

状态 RPO：崩溃后最多回退多少个超级步或 token；
副作用不确定性预算：允许多少个外部操作处于 unknown，以及最长允许持续多久；
恢复 RTO：lease 失效到 run 重新 queued 或进入人工确认的时间；
事件 RTO：客户端重连到补齐事件的时间；
人工确认时限：高风险 unknown operation 多久必须处理。

一个示例目标：

99.9% 的纯读 run 在 worker 崩溃后 90 秒内自动恢复
99.99% 的 terminal event 在 checkpoint 提交后 10 秒内可被客户端读取
任何非幂等 tool operation 不允许在 unknown 状态下自动重试
同一逻辑操作的外部副作用次数必须 <= 1
已经确认成功的写操作，其外部副作用次数必须 = 1

26.4 Trace 的作用

Trace 有助于解释恢复，但不能作为权威状态。采样、异步导出和网络故障都可能造成 trace 缺失。恢复判断应来自 transactional store、checkpoint 和外部系统查询，而不是「trace 中有没有成功 span」。

二十七、关键设计取舍

27.1 Checkpoint 越频繁越好吗

更频繁的 checkpoint：

缩小重算范围；
增加数据库写入；
增加序列化开销；
可能放大大状态的存储成本。

划分节点时应围绕：

外部副作用边界；
费用较高的 LLM 调用；
可并行单元；
人工审批点；
长时间任务。

不要为了减少节点数量，把「查询、规划、发邮件、更新数据库」全部放进一个不可分割函数。

27.2 Snapshot 还是 Event Log

推荐组合，而不是二选一：

event log 负责因果和审计；
checkpoint/snapshot 负责快速恢复；
tool operation store 负责外部事实；
outbox 负责跨系统发布。

把所有内容塞进一张 JSON snapshot，会让去重、查询和并发控制变得困难。

27.3 自动恢复还是标记失败

条件	推荐动作
纯计算或纯读	自动续跑
已有 durable tool result	复用结果后续跑
工具支持幂等键	使用原 key 重投
可按远端 ID 查询	查询后决定
非幂等且结果未知	人工确认
状态 schema 不兼容	标记 blocked，不强行解释

Open Managed Agents 和 DeerFlow 默认偏向保守收敛，这是合理的产品选择，不表示底层没有恢复能力。

27.4 Redis 还是数据库

Redis 适合：

短期 lease；
heartbeat；
live pub/sub；
有限窗口 stream replay。

数据库适合：

run terminal state；
fencing token；
checkpoint metadata；
tool operation；
长期 event log；
outbox 和唯一约束。

如果 Redis 被清空，业务正确性仍应可以从数据库重建。否则 Redis 已经不是缓存或协调设施，而是未声明的权威数据库。

27.5 保存每个 Token 吗

保存每个 token 可以最大化 partial stream 恢复，却带来高写放大。常见折中：

在内存中聚合 50～200 ms；
按 chunk 批量保存；
最终 message 提交后删除或压缩 chunk；
UI 事件与模型 canonical message 分开；
明确 partial message 只用于展示，还是会进入下一次模型上下文。

二十八、常见反模式

28.1 把 `status='running'` 当成真实存活

数据库状态只是上一次写入。必须结合 lease 或 worker liveness。没有 TTL 的 running 标记会永久变成僵尸。

28.2 只有 Heartbeat，没有 Fencing

heartbeat 可以发现 owner 失联，却不能阻止旧 owner 复活后写入。最终状态更新需要 token。

28.3 重启后把所有 Running 都标为 Failed

单 worker 本地应用可以这样做；多 worker 环境会误伤其他仍在运行的 worker。必须先判断 owner 和 lease。

28.4 看到 Tool Use 没有 Result 就自动重试

缺少 result 只说明本地不知道结果，不说明外部操作没有发生。非幂等工具必须进入 unknown。

28.5 事件先推给客户端，状态随后保存

客户端可能看到一个无法恢复的成功。使用 outbox，让事件由已提交状态派生。

28.6 依赖 Graceful Shutdown

正常 drain 很有价值，但 OOM、SIGKILL 和机器故障不会运行 shutdown hook。恢复设计必须在没有 finally 的情况下仍然成立。

28.7 把 Workspace Snapshot 当作所有副作用的回滚

Git diff 只能描述文件变化，不能撤销已经发送的消息、云资源创建或数据库写入。

28.8 用内存 Callback 等待子 Agent

父进程退出后 callback 消失。parent-child dependency 和 child terminal event 必须持久化。

28.9 假设 Recovery 只会运行一次

恢复器自身也会崩溃。每个修复动作都要有唯一约束或条件更新，保证相同外部事实下重复执行不产生重复写入。远端结果稍后变得可见时，后一次恢复可以推进状态，但不能倒退或重放已经确认的副作用。

28.10 只测试异常，不测试进程死亡

异常测试通常执行 cleanup，无法覆盖最危险窗口。至少需要一组真实子进程 SIGKILL 测试。

二十九、分阶段实施建议

阶段 1：会话和状态可恢复

session/message 使用持久数据库或 append-only 文件；
每个 run 有稳定 ID；
保存 terminal status；
修复 dangling tool use/result；
客户端支持 cursor replay；
增加真实 SIGKILL 基线测试。

这一阶段目标是 L1～L2：重启后历史可读，状态不会永久卡住。

阶段 2：工作流可续跑

引入 checkpoint；
明确节点和超级步边界；
保存 pending writes；
使用 durable queue；
增加 lease、heartbeat 和 startup/periodic reconciler；
正常关闭时先 drain run。

这一阶段达到 L3，但只适合纯计算、纯读和已知可重试工具。

阶段 3：副作用安全

建立 tool operation store；
为工具声明 effect class；
使用稳定 operation ID；
接入外部幂等键和 remote request ID；
增加 unknown/manual review；
使用 transactional outbox；
所有状态提交增加 fencing token。

这一阶段开始具备 L4 的 effectively-once 语义。

阶段 4：多 Agent 和沙箱

parent-child dependency 持久化；
child 创建和 parent waiting 原子化；
parent inbox 去重；
sandbox ID 和 ownership lease；
长任务支持 attach/poll；
定义 parent/child orphan policy。

阶段 5：持续验证

failpoint 覆盖所有提交边界；
每个版本运行 SIGKILL 测试；
定期执行网络分区和 Redis/PostgreSQL 故障演练；
监控 unknown operation 和 fencing rejection；
对旧 checkpoint 做 schema migration 测试；
发布前检查相同外部事实下 recovery 不产生重复持久写入，并验证事实变化时状态单调收敛。

三十、上线前检查清单

状态与存储

请求有客户端预先生成、限定 tenant 作用域的稳定 request ID
request hash 在响应前持久化，相同 ID 对应不同 hash 时返回冲突
run、turn、step 和 tool operation 的身份不会因重试变化
逻辑 step attempt 的创建 token 只用于审计；当前权限始终检查 run.fencing_token
checkpoint 使用持久后端，不使用进程内 saver
checkpoint 有 parent/version/CAS 校验
terminal 状态不能被普通 update 覆盖
event log 使用连续 seq；执行 retention 时记录首个保留序号
outbox 与状态变更在同一事务
Outbox Relay 按 run 串行发布，N 确认前不会发送 N+1

所有权

run 有 owner、lease_until 和 fencing token
heartbeat 按 owner/token 条件续租
旧 token 的 checkpoint 和 terminal 写入会被拒绝
启动扫描与周期扫描同时存在
recovery 有独立 owner、lease 和 fencing token，并能接管已过期的 orphaned run
recovery 最终决定按 recovery owner/token 条件提交，旧 Reconciler 无法覆盖新决定
外部对账查询不写状态；operation 结果、run 决定、事件和 outbox 在校验 recovery token 的同一事务中提交
未识别的 recovery action 默认进入人工确认，manual_review operation 会被后续扫描再次读取
在相同外部事实下重复 recovery 不产生重复持久写入；事实变化时状态单调收敛
graceful shutdown 会先停止续租并 drain task

工具

每个工具声明读、幂等写、可查询写或非幂等写
包装器与 Reconciler 使用同一份持久化 idempotency_guarantee 定义
side effect 具有稳定 operation ID，调用身份缺失时拒绝执行
创建 prepared 时填充全部必填字段，并校验 run owner、fencing token 和有效 lease
支持时向外部 API 传 idempotency key
保存 remote request/job ID
started 表示发送意图已保存，不能据此判断请求已经到达外部系统
确定性工具错误写为 failed，传输结果未知时写为 unknown
prepared -> started 与 run owner、fencing token 和有效 lease 在同一条件更新中校验
所有 operation 转换先锁 run、再锁 operation；不存在相反锁顺序
operation 使用 dispatch generation CAS，只有取得当前发送代次的协程可以调用外部工具
结果转换同时校验 expected status 和 dispatch generation，终态不能被覆盖
operation 协调信号由 singleflight、等待和重读处理，不会生成工具失败结果
包装器读取到 failed/manual_review 时不会重新发送请求
非幂等 unknown 会进入人工确认
工具 result 保存后才能推进 checkpoint

沙箱与多 Agent

workspace 与进程状态分别处理
远端任务可以按 ID 重新 attach
sandbox ownership 与 run ownership 分离
sandbox reaper 通过独立 deletion lease 取得销毁权限
parent-child 关系持久化
child 创建不会因 parent 重放而重复
child terminal event 可以至少一次投递并去重

验证

有真实 SIGKILL 测试
有旧 worker 暂停超过 TTL 后恢复的测试
有外部操作成功但响应丢失的测试
有 event publish 与 ACK 间崩溃的测试
有两个 Relay 竞争同一 run 相邻序号的测试
有 recovery 中途再次崩溃的测试
有旧 Reconciler 在 recovery lease 过期并被接管后恢复的测试
有外部查询完成、恢复决定提交前丢失 recovery lease 的测试
有同一 worker 内两个协程竞争同一 operation 发送权的测试
有 operation 转换与 Reconciler fencing 竞争同一 run 行锁的测试
有同 generation、不同 operation 状态之间结果 CAS 竞争的测试
有协调信号不会把 run 标为失败的测试
有运行中取消分别收敛到 cancelled 和 uncertain 的测试
有 checkpoint schema 升级测试
测试断言外部 effect count，而不只断言最终 HTTP 状态

三十一、最终结论

Agent 崩溃恢复可以归纳为四个互相独立的问题：

记住历史：session、event log、rollout；
记住进度：checkpoint、pending writes、snapshot；
决定所有权：lease、heartbeat、fencing token；
确认外部事实：operation ID、幂等键、结果查询和人工确认。

LangGraph 解决了第 2 项的核心语义；DeerFlow 将第 2 项扩展到 run、stream 和 sandbox；Open Managed Agents 对第 1 项的崩溃修复和测试最直接；AgentScope 展示了 session lock、持久 AgentState 和跨 session projection；OpenCode、Codex、Pi 则提供了不同形式的本地历史恢复。

真正的生产级实现不能只选择其中一个项目的一个机制。更完整的组合是：

Tenant-scoped Durable Inbox / Request Hash
  + At-least-once Queue
  + Checkpoint / Pending Writes
  + Lease / Heartbeat / Fencing Token
  + Tool Operation Store
  + Idempotency / Reconciliation
  + Transactional Outbox / Per-run Ordered Relay
  + Sequenced Event Replay
  + SIGKILL Fault Injection

其中最容易被忽略的不是 checkpoint，而是「不确定」。当系统无法证明一个外部操作成功或失败时，正确状态不是强行选择 success 或 failure，而是保留 unknown，并通过查询、对账或人工判断消除不确定性。

一个可靠的 Agent 平台，不是保证进程永不退出，而是保证进程在任意边界退出后，系统仍能回答三个问题：

已经发生了什么；
接下来允许谁做什么；
哪些外部事实仍然未知。

能够稳定回答这三个问题，才算真正具备崩溃恢复能力。

附录 A：源码基线

项目	固定提交	重点文件
LangGraph	`9578140336b0`	checkpointer、Pregel loop、durability、interrupt
DeerFlow	`fa496c0c8df4`	RunManager、Gateway lifespan、stream bridge、sandbox ownership
Open Managed Agents	`870b9e29a973`	recovery、session machine、adapter、SIGKILL tests
AgentScope	`f5f9ca6050b8`	ChatService、SessionService、Redis message bus、projection
OpenCode	`849c2598abc7`	session processor、message-v2
Codex	`1836ae061205`	rollout reconstruction、thread manager
Pi	`bb019731fc23`	SessionManager、orchestrator supervisor

附录 B：延伸阅读

本文优先使用固定提交的源码、测试和官方文档。所有实现结论均以表中列出的版本为准；后续版本可能调整接口或内部结构。

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

AI Agent编排：复杂工作流的自动化实现

在深入探讨Agent编排之前，我们首先需要明确什么是AI Agent。AI Agent是一个能够感知环境、做出决策并采取行动的自主实体。感知模块：接收和处理环境信息推理/决策模块：基于感知信息进行思考和决策行动模块：执行具体的操作（如调用工具、生成文本）记忆模块：存储历史信息和知识通信模块：与其他Agent或系统进行交互Agent编排（Agent Orchestration）是指对多个Agent的

AI Agent技术社区

AI Agent的多轮对话状态管理

在深入探讨之前，让我们首先明确几个核心概念：AI Agent：一种能够感知环境、做出决策并执行行动的智能实体。在对话系统中，AI Agent通常指能够理解用户意图、保持对话上下文并生成有意义响应的系统。多轮对话：不同于单轮问答，多轮对话是指用户与系统之间进行的连续、有上下文关联的交互序列。对话状态管理(Dialogue State Tracking, DST)：是多轮对话系统中的核心组件，负责在对