Code Agent 崩溃恢复：对话历史如何持久化

当一个 Agent 运行的时候，所有的对话历史都保存在 HistoryManager 的内存列表里。一旦进程崩溃——网络超时、OOM、Ctrl+C——这些历史记录全部丢失。重启之后，用户从头开始；几十步的探索结果彻底消失。

工具调用中断问题更加棘手：如果 Agent 在 Edit 工具修改文件的过程中崩溃，文件可能是改了一半，也可能是毫发无损，进程日志里什么都没有。重启后，Agent 看不到这次 Edit 调用的结果，不知道文件到底有没有被改过。

MyCodeAgent 用 Transcript 解决了这两个问题：持续将所有关键事实写入一个只追加的 JSONL 文件，崩溃后从该文件重建运行时状态。

核心设计：事件流而非快照

常见的持久化方案有两种：

快照：定期将整个状态序列化保存（如数据库备份）

事件流：每次操作完成后追加一条记录（如数据库 WAL）

Transcript 采用事件流。原因：快照需要原子性保证（不能写到一半），重启后还需要选择从哪个快照恢复。事件流只追加，每行独立天然原子，重放所有事件即可恢复状态。

memory/transcripts/transcript-{session_id}.jsonl

{"event_id":"evt-abc","timestamp":"...","session_id":"s-123","run_id":"run-1","step":0,"event_type":"message","payload":{"role":"user","content":"Help me refactor the auth module"}}
{"event_id":"evt-def","timestamp":"...","session_id":"s-123","run_id":"run-1","step":1,"event_type":"state_transition","payload":{"reason":"model_returned_tool_calls",...}}
{"event_id":"evt-ghi","timestamp":"...","session_id":"s-123","run_id":"run-1","step":1,"event_type":"tool_lifecycle","payload":{"tool_name":"Read","tool_call_id":"call-xyz","status":"requested"}}
{"event_id":"evt-jkl","timestamp":"...","session_id":"s-123","run_id":"run-1","step":1,"event_type":"tool_lifecycle","payload":{"tool_name":"Read","tool_call_id":"call-xyz","status":"completed","result":"..."}}
...
{"event_id":"evt-zzz","timestamp":"...","session_id":"s-123","run-1","step":28,"event_type":"terminal","payload":{"reason":"completed"}}

每条事件记录捕获循环中的一个具体事实：

RuntimeRunner 调用 self._emit(event_type, payload, step=step)
    ↓
RuntimeEventSink.emit(RuntimeEvent)
    ↓  [CompositeRuntimeEventSink]
    ├─ TraceRuntimeEventSink → trace_logger（JSONL trace 文件，用于调试）
    └─ TranscriptRuntimeEventSink → TranscriptRecorder → TranscriptStore.append_event()

Transcript 和 Trace 是两个独立的下游，它们订阅同一个事件流。Trace 是详细的调试日志；Transcript 是用于恢复的事实日志。两者内容有重叠，但用途不同。

TranscriptStore：只追加写入

# runtime/transcript.py
class TranscriptStore:
    def append_event(self, event: TranscriptEvent) -> TranscriptEvent:
        line = json.dumps(event.to_dict(), ensure_ascii=False)
        with self._lock:                      # 文件锁，防止并发写入损坏
            self._repair_trailing_record()    # 修复末尾不完整记录（崩溃残留）
            with self.path.open("a") as f:
                f.write(line)
                f.write("\n")
                f.flush()                     # 立即刷盘，不依赖 OS 缓冲区

_repair_trailing_record() 每次写入前检查文件末尾内容：如果存在不完整的行（没有换行，或 JSON 解析失败），说明上次写入在半途中断——截断该行。这保证了文件中每行都是完整、有效的 JSON。

事件如何流入 Transcript

写入不直接调用 TranscriptStore，中间有两层抽象：

循环的 self._emit("message", {...}, step=step)
    ↓
RuntimeRunner._emit_runtime_event(run_id, step, event_type, payload)
    ↓
host.runtime_event_sink.emit(RuntimeEvent)
    ↓  CompositeRuntimeEventSink 同时转发到两个 sink
    ├─ TraceRuntimeEventSink.emit()      → extensions/tracing/logger.py（调试 trace）
    └─ TranscriptRuntimeEventSink.emit() → TranscriptRecorder.record_*()
                                               ↓
                                           TranscriptStore.append_*()
                                               ↓
                                           JSONL 文件追加一行

TranscriptRecorder 是 TranscriptStore 的 Facade：它将高级操作——'写入消息'、'写入 state_transition'、'写入 tool_lifecycle'——封装为 record_message()、record_state_transition()、record_tool_lifecycle()，隐藏了底层 JSON 序列化细节。它还持有一个 on_recorded 回调，每次写入后调用 SessionMemoryManager.ingest_event() 来增量更新 Session Memory。

恢复：从事件流重建运行时状态

用户重启后，调用 agent.resume_transcript() 或使用 CLI 的 --resume flag，会走到 ResumeLoader.load_session()：

# runtime/transcript.py — ResumeLoader._load_events()（简化版）
def _load_events(self, events, *, run_id):
    history_messages = []    # 重建历史消息列表
    checkpoint = None        # 上一次压缩 checkpoint
    terminal = None          # terminal 事件
    tool_events = {}         # 每个工具调用的完整生命周期

    for event in events:
        if event.event_type is MESSAGE:
            history_messages.append({role, content, metadata})

        elif event.event_type is CHECKPOINT:
            checkpoint = event.payload   # 保留最后一个 checkpoint

        elif event.event_type is TERMINAL:
            terminal = event.payload

        elif event.event_type is TOOL_LIFECYCLE:
            # 聚合同一个 tool_call_id 的所有状态
            tool_events[(run_id, tool_call_id)]["statuses"].append(status)

然后分析 tool_events，对每个工具调用进行分类：

completed   → 已完成，无需重放
failed      → 已失败，无需重放
requested but not started → 未执行，处于 pending（可以重新规划）
started but no completed/failed → 状态不确定（中断了，状态未知）

UncertainAction 是恢复中最微妙的概念：工具已经开始执行，但结果写回之前 Agent 就崩溃了。结果可能是成功、失败或任何中间状态。

uncertain_actions.append(UncertainAction(
    tool_name=tool_name,
    tool_call_id=tool_call_id,
    replay_allowed=tool_name not in {"Edit", "Bash", "Task"},
    # Read/Grep/Glob：幂等，可以重放
    # Edit/Bash/Task：有副作用，不能盲目重放；需要用户判断
))

CLI 恢复过程中，uncertain actions 会打印出来，让用户知道"这些工具可能执行了也可能没执行——请自行核实"。

重建完成后 apply_to_host

ResumeState.apply_to_host(host) 将重建的状态注入运行中的 Agent：

# runtime/transcript.py — ResumeState.apply_to_host()
def apply_to_host(self, host):
    # 1. 重置 context engine，清除压缩 checkpoint
    host.context_engine.reset()

    # 2. 将重建的历史消息写入 HistoryManager
    host.history_manager.load_messages(self.history_messages)

    # 3. 如果有压缩 checkpoint，重新激活它
    #    （ProjectionBuilder 读取后会折叠旧历史）
    if self.checkpoint:
        host.context_engine.compact_store.set_active(CompactCheckpoint(...))

    # 4. 恢复 Read 工具的乐观锁缓存（mtime 快照，避免 Edit 冲突误报）
    if read_cache := self.runtime_state.get("read_cache"):
        host.tool_registry.import_read_cache(read_cache)

恢复之后，Agent 的行为就好像从来没有崩溃过——历史完整、压缩状态恢复、乐观锁缓存有效。

Session Memory：跨运行的前馈记忆

Transcript 存储完整的事实流；Session Memory 是从事实流派生出的有界摘要。

SessionMemoryDeriver 扫描所有 transcript 事件，提取高层信息：

@dataclass(frozen=True)
class SessionMemory:
    current_goal: SessionMemoryItem | None  # 用户最新目标（最新的 user 消息）
    completed_work: tuple[...]              # 已完成的工作（final-type assistant 消息）
    key_decisions: tuple[...]               # 关键决策（压缩 checkpoint、重要状态转换）
    failed_attempts: tuple[...]             # 失败尝试（model_recovery_failed 等）
    todo_items: tuple[...]                  # 未完成的 TodoWrite 项
    verification_status: tuple[...]         # 验证状态（completion gate 信息）

Session Memory 在 build_model_view() 中作为 system message 注入，置于 system prompt 之后、历史消息之前：

[system prompt]
[Session Memory]  ← "此前你完成了 X，Y 失败了，当前目标是 Z"
[历史消息]

为什么用 Session Memory 而不是直接读 transcript？

Transcript 可能包含数千行事件；将它们全部放入 model view 会超出 token 预算。Session Memory 是一个有界的高层摘要，控制在几百行以内，让模型拥有跨运行的上下文，而无需读取完整事件流。

Session Memory 是增量维护的：TranscriptRecorder 每次写入事件时都会调用 SessionMemoryManager.ingest_event()；SessionMemoryDeriver.update() 增量追加新事件，无需每次全量重建。

实际文件结构

memory/
├── transcripts/
│   ├── transcript-session-abc123.jsonl   ← 主 Agent 会话
│   └── transcript-subagent-child-xyz.jsonl ← 子 Agent 会话（每个 Task 调用独立）
└── traces/
    ├── session-abc123.jsonl              ← 调试 trace（详细）
    └── session-abc123.html               ← 可视化报告（可选）

Transcript 和 Trace 文件并列放置，但用途不同：

• Transcript：崩溃恢复的事实来源；只存储关键事实
• Trace：用于调试分析；记录所有细节（token 使用量、每步耗时、原始模型输出）

设计亮点

事件流而非快照：每个事件在写入时就持久化；不等待 Agent 完成；崩溃后从最后一个完整事件开始恢复。

不确定操作显式标记：工具中断不是静默失败——它们被显式标记为不确定，由用户决定是否重放，而不是由 Agent 猜测。

恢复是确定性的：从同一份 transcript 读取，每次产生的 history_messages 和 loop_state 相同；不依赖任何随机状态。

Session Memory 增量维护：恢复时不整体重建；每次写入事件后增量更新，将重建开销分摊到每次写入操作中。

关于源代码

本系列所有分析均基于开源项目 MyCodeAgent。

源代码在关键位置包含了与系列讲解顺序一致的内联注释——可以边读边对照，也可以 clone 下来，跑一跑，改一改，扩展出你自己的 Agent。

git clone https://github.com/chendongqi/MyCodeAgent
cd MyCodeAgent
cp .env.example .env   # 填写你的 LLM API key
uv sync
uv run python main.py

了解更多可以访问 PrimeSkills —— 经过真实企业级工作流验证的 AI Agent 和 Skills 精选市场，没有废话，只有真正有效的东西。

在我的 Homepage 找到更多有用的知识和有趣的产品。