Harness 与 Durable Agent

一、这篇要追踪什么

前几章的 Pi CLI 已经具备模型调用、工具执行、会话保存与压缩能力，广义上就是一个 Agent Harness： 让模型能够在真实环境中完成任务的运行系统。

本章研究仓库中另一套独立实现 pi-durable。它要解决的核心问题是：进程退出后，怎样知道任务停在哪里， 并按明确规则接着处理？ “Durable Agent”指这种持久化执行能力，不是一种新的模型或推理方式。

先确定源码范围，避免把两套运行时混起来：

当前标准 CLI：coding-agent → AgentSession + pi-agent-core → pi-ai
本章运行时：  宿主应用 → pi-durable 的 Harness → pi-ai

pi-durable 自己实现调度、工具执行和上下文管理，不依赖 coding-agent 或 pi-agent-core。 当前 CLI 主流程尚未切换到它。packages/agent/src/harness 中另有 AgentHarness，也不是本章的 Harness。 以下依据 8ce69e9d2 版本的 packages/durable；该包目前标为 experimental。

全篇只跟踪一个例子：用户要求修正配置，模型先调用 read，再调用 edit，最后回答用户。 这是用于解释的一种模型行为，不是源码强制三轮。

宿主提交输入
  → 创建第一轮模型任务 G1
  → 调度器运行 G1，模型提出 read
  → G1 创建工具任务 T1，然后等待
  → 调度器运行 T1，保存读取结果
  → 调度器唤醒 G1，G1 创建下一轮 G2 并结束自己
  → G2 提出 edit，以相同方式完成工具任务 T2
  → G3 根据修改结果给出回答，原输入完成

先在第二至七节走通这条正常路径，再在第八节让 T2 中途崩溃。排队、取消和压缩放到最后， 因为它们都是在这条主路径上增加分支。文中的代码是源码缩写或公开 API 用法，省略 import 和无关分支，不能直接拼成可运行程序。

二、宿主打开运行时，再提交一项输入

pi-durable 是库，需要宿主应用启动它。宿主先提供三个运行条件：模型接口 models、 安装工具实现的 registry、访问文件和进程的执行环境 env，再打开存储。

// 公开 API 用法；省略 models、env、context 的构造与错误清理。
const registry = createRegistry();
registry.install(CodingTools); // 安装 read/write/edit/bash 实现。
const storage = await openNodeSqliteStorage("./session.sqlite");
const harness = await Harness.open(storage, { models, registry, env }, context);
const conversation = await harness.root(context, {
  agent: { model: { provider: "openai", modelId }, cwd },
});
const submission = await conversation.submit({
  type: "input", content: "读取并修正配置文件", requestId: "fix-config-1",
}, context);
const result = await submission.wait(context);
await harness.close(context);

这里三个返回对象分别负责不同事情：

• Harness 持有打开的存储、调度器及相关服务，管理整个运行时。
• Conversation 是一段对话的操作句柄；本例把这段对话记为 C。
• Submission 是这一次输入的操作句柄；本例记为 S，后面几轮模型和工具都在为 S 工作。

submit() 返回表示输入已被接纳；wait() 返回才表示 S 已结算，还需要查看结果是 done 还是 unanswered。 context 是 Chord 的取消上下文，不是发给模型的 messages。

这些能力怎样接起来？Conversation.submit() 本身很薄，它把工作交给 Submissions 服务：

// ConversationImpl.submit()
return this.#host.submissions.submit(this.id, submission, context);

// Submissions.submit()：省略时间与队列设置参数。
this.#resume(); // 启用调度器。
const id = await this.#session.commitWith(
  tx => admitSubmission(tx, conversationId, draft, now, queueModes), context,
);
return new SubmissionHandle(id, this);

下一步不是立刻调用模型，而是 admitSubmission() 先把“已接纳这项工作”写进存储。

三、接纳输入：在同一次提交里建立运行起点

先看最简单的情况：C 当前空闲，没有排队输入。接纳函数做三件事：保存用户消息、保存 S、启动一次 run。 run 是从用户输入到最终回答的这一段工作；它可以包含多个模型轮次。

// admitSubmission() 的空闲 input 分支；省略去重和队列处理。
const entry = await tx.appendEntry(UserEntry, conversationId, { model: [message] });
const { id } = await tx.createSubmission({
  conversationId, type: "input", status: "placed", entry: entry.id,
});
await startRun(tx, conversationId, live, [id]);

// startRun()：建立第一轮 GenerationTask，并记录它负责哪些输入。
live.run = { taskId: await createGeneration(tx, conversationId), inputs };

// createGeneration()
return tx.createTask(GenerationTask, {}, {
  ownership: { kind: "conversation" }, conversationId,
});

本次提交成功后，存储中有四个互相关联的事实：

用户消息 U1：内容为“读取并修正配置文件”
输入 S：placed，entry=U1
任务 G1：kind=pi.generation，state={ status: pending, checkpoint: { phase: prepare, attempt: 1 } }
C 的 pi.live.run：{ taskId: G1, inputs: [S] }

其中 GenerationTask 就是一种 Task：Task 是可调度任务的通用记录，GenerationTask 定义怎样调用模型并处理响应。 后面的 ToolTask 定义怎样执行一个工具调用。G1、T1 是任务实例 ID，不是两套独立调度系统。

3.1 这些记录保存在哪里？

这里的记录都属于 pi-durable 的 Session 存储层。Session 负责组织提交，Storage 负责实际保存； 它们不是 CLI SessionManager 的对象，也不是待修改配置文件里的内容。

逻辑对象	本例作用	SQLite 表
Conversation	C 的身份与所有权	conversations
Submission	S 的输入身份与结算状态	submissions
Entry	U1，以及后续 Assistant 和工具结果；不可变历史	entries
Task	G1、T1 等执行任务，含调度状态和 checkpoint	tasks
Document	C 的可更新状态，例如 pi.live	documents 保存身份，document_revisions 保存内容

pi.agent、pi.live、pi.inbox、pi.usage 都是 Document 类型，不是各自一张表：

• pi.agent 保存这段对话的模型、扩展与工具选择、cwd 等配置。
• pi.live 保存当前 run、模型生成进度和工具槽位。
• pi.inbox 保存等待处理的输入或写入。
• pi.usage 保存消耗统计。

这些内建文档关联 C；通用 Document 机制也支持 session、task 范围。 在 SQLite 中，文档类型由 documents.kind 区分，内容通过 document_id 关联到 document_revisions。 后者保存完整内容 base 或增量 delta，部分 current-only 文档写入新 base 时会清理旧版本。 Task 的 checkpoint 则在 tasks.record 的 JSON 内；经常用于查询的 kind、status 等也有独立 SQL 列。

-- 实际 schema 的字段缩写；省略约束、索引及其他列。
CREATE TABLE tasks (id INTEGER PRIMARY KEY, conversation_id INTEGER,
                    kind TEXT, status TEXT, record TEXT);
CREATE TABLE documents (id INTEGER PRIMARY KEY, kind TEXT,
                        scope_kind TEXT, owner_id INTEGER, record TEXT);
CREATE TABLE document_revisions (document_id INTEGER, seq INTEGER,
                                 kind TEXT, version INTEGER, content TEXT);

3.2 为什么要放在同一次提交里？

如果先保存用户消息、稍后才创建 G1，进程可能在两次写入之间退出，留下“用户输入存在，但没有任务负责执行”的状态。 因此 Session 将这组变化收集为一次 commit，再交给 Storage 原子保存。

// Session 的提交顺序缩写。
const result = await change(tx);
const writes = await tx.settleSuccess();
const seq = await storage.commit(writes, withoutAbortSignal(context));
publish(seq, writes, documents, context); // 存储成功后再通知观察者。

// SQLite commit 内部：把相关表的变化放在一个数据库事务中。
await this.db.transaction(async transaction => {
  for (const write of writes) {
    await this.applyTableWrite(transaction, write, committedSeq);
  }
  await this.applyDocumentActions(transaction, documentActions, committedSeq);
  // 省略 ID 校验与提交序号更新。
});

现在 G1 与它所需的数据都已保存。下一节要解决的是：谁发现 G1，怎样把一条数据库记录变成实际执行？

四、调度器读取任务记录，调用对应的阶段函数

调度器的基本工作是：检查任务清单，找出现在能执行的任务，调用它的阶段函数。 这份清单不是靠定时器反复扫描数据库得到的；Scheduler 订阅 Session 的提交， 把未结束任务维护在内存索引 #live 中，把当前正在执行的任务维护在 #invocations 中。

4.1 谁通知调度器检查任务清单？

第三节最后的 publish() 就是衔接点。Scheduler 在打开时注册提交监听器：

// Scheduler.open() 中注册。
this.#session.subscribeCommits(publication => this.#observe(publication));

// #observe() 的流程缩写：
// 根据本次提交更新 #live：新增、更新任务，或移除 terminal 任务。
// 若本次提交有 Task 变化：
this.#kick();

因此任务不需要直接调用“运行下一个任务”。它只需提交自己的变化， Session 保存成功后通知 Scheduler，Scheduler 再判断有没有任务可以运行。

任务代码提交状态
  → Session 保存成功并发布变更
  → Scheduler 的 #observe() 更新任务清单
  → #kick() 通知需要检查
  → #drain() 发起调度扫描
  → #reserve() 找可运行任务并提交占用状态
  → #start() 启动对应函数

主流程中有三个触发时机：

触发者	时机	为什么需要检查
resume()	启用调度；submit/wait 内部也会启用	已存在的任务可能需要开始或恢复
提交监听器 #observe()	收到包含 Task 变化的提交	可能新增了任务，或某个等待条件已经解除
#start() 的 finally	一次 invocation 退出，清理执行占用之后	原来的占用已释放，可能有任务可以继续

这是事件驱动的调度。不是每行代码或每个 await 完成都重新调度，也不是每次通知都必然启动一个任务。 通知的含义只是“现在可能有任务能运行了，请再检查”。此外，Registry 变化也会触发检查， 让此前因实现不可用而无法运行的任务有机会被重新解析。

#kick() 会合并通知：已经有调度扫描在进行时，只标记还需要检查；否则安排一个微任务开始扫描。

// 源码缩写。
#kick() {
  this.#dirty = true;
  if (this.#draining || !this.#enabled || this.#closing) return;
  this.#draining = true;
  queueMicrotask(() => void this.#drain());
}

这样既不会在提交监听器里同步嵌套另一次提交，也不会每收到一条通知就开启一个互相竞争的扫描循环。

4.2 调度器怎样决定运行谁？

调度器不需要理解“修正配置”的业务。它主要根据任务记录做三件事：

1. 检查它是否还在等待其他任务，或已经被当前进程的执行过程占用。
2. 按 kind 找已注册的任务定义，将记录标为 running，并创建一次内存执行过程 invocation。
3. 按 checkpoint.phase 找任务定义中的阶段函数，调用它。

// Scheduler 的选取、占用与执行逻辑示意；合并了不同方法，省略取消和版本迁移。
if (hasInvocation(record.id) || waitingOn(record).length > 0) continue;
const definition = registry.task(record.kind); // G1 → GenerationTask。
// 在事务中将 state.status 改为 running，保留 checkpoint；提交后启动执行。
await definition.phases[checkpoint.phase](record, runtime, context);
// G1 的 phase=prepare，因此运行 GenerationTask.phases.prepare。

这里 kind 决定谁来处理，phase 决定处理到哪一步，status 决定现在能否处理。 例如 G1 从 pending/prepare 变为 running/prepare，只表示调度器已经接手，业务阶段还没变。

先用 G1 和 T1 看一次“切换”。第五、六节会展开它们的业务代码，这里只看谁运行、谁等待：

刚发生的变化	调度器检查时看到什么	接下来调用谁
提交创建 G1	G1 没在执行，也没有依赖	调用 G1 的 prepare 阶段
G1 创建 T1，并提交 waiting、on=[T1]	G1 仍要等 T1；T1 可以运行	调用 T1 的 call 阶段
T1 保存结果并提交 terminal	T1 从未结束清单移除；G1 的依赖已结束	重新调用 G1 的 tools 阶段

G1 进入 waiting 后，本次函数执行退出，但任务记录保留。它不靠一个内存 Promise 一直挂着等 T1。 T1 完成后，调度器创建新的 invocation，按 G1 保存的 tools 阶段继续执行。 所谓 Task 之间的切换，就是根据记录结束或启动这些函数执行，不是把一个函数的调用栈搬到另一个函数里。

一次扫描可以选中多个符合条件的任务，并启动它们的异步执行。存在等待关系的任务才需要按依赖先后推进。 #invocations 则避免同一任务在已被占用时再次启动；即使状态提交已触发检查， 旧 invocation 尚未退出也会先跳过它，待 finally 清理后再次触发检查。

4.3 状态与阶段怎样控制后续执行？

Task 的调度状态只有以下几种，不要与模型请求阶段混为一谈：

Task 状态	含义
pending	等待调度，保存 checkpoint
running	已被一次 invocation 占用，按 checkpoint 执行业务代码
waiting	保存 on 依赖任务 ID 和恢复 checkpoint，暂不执行业务代码
completing	outcome 已决定，仍等待其拥有的普通子工作收尾，不再运行阶段函数
terminal	已结算，保存 outcome，不再执行

terminal 不等于成功。outcome 可以是 completed、任务明确报告的 failed、取消 aborted， 也可能是定义或迁移缺失导致的 orphaned 结算，或运行时契约错误 faulted。 不是每个任务都会经过 waiting 或 completing。

阶段函数也必须提交进度。调度器的执行循环在阶段返回后重新读取状态：仍为 running 且 checkpoint 已推进， 就执行下一个阶段；进入 waiting 或 terminal 则结束本次 invocation。 如果阶段既不结束也不改变 checkpoint 就返回，调度器会视为没有持久进度的契约错误。

因此，checkpoint 是下一次执行的入口数据，不是保存了 JavaScript 调用栈或一个未结束的 Promise。 接下来直接看 G1 的 prepare 和 request 怎样推进它。

五、G1 准备上下文并请求模型，然后创建工具任务 T1

5.1 prepare：先固定这次请求的依据

G1 的 prepare 阶段读取 C 的 pi.agent 配置，再通过 Registry 找到选中的工具、提示词片段和 hooks。 Registry 保存的是当前进程安装的实现；pi.agent 保存的是选择。这就是第二节必须先安装 CodingTools 的原因。

prepare 读取对话历史，渲染提示词，并按变化追加 system entries。然后保存 request checkpoint：

// GenerationTask.prepare 的结构缩写；省略阈值压缩和 system entries 的构造。
await runtime.commit(async tx => {
  // 追加需要更新的 system entries，并取得这次请求的历史终点 cutoff。
  return {
    status: "running",
    checkpoint: {
      phase: "request", attempt, model, thinkingLevel,
      streamOptions: settings.stream, cutoff,
    },
  };
}, context);

cutoff 指定这次请求读取到哪条 entry，避免恢复时无条件混入后来新增的所有历史。 这里完成后，G1 仍是 running，但 phase 从 prepare 变成 request。调度器循环据此进入 request 函数。

5.2 request：调用模型，将响应变成下一步任务

request 按保存的模型配置和 cutoff 重建 messages，经 beforeRequest hooks 处理后调用 pi-ai：

// request → streamResponse → classify 的调用关系缩写。
const view = await runtime.context(conversationId, context, cutoff);
// 省略 hooks、options 构造以及流式响应收集。
const stream = runtime.models.streamSimple(model, { messages: [...messages] }, options);
// 收集得到 AssistantMessage 后交给 classify() 判断下一步。

本例模型提出 read("config/dev.json")。classify 进入 startToolRound()， 保存 Assistant 与创建工具任务发生在同一次 commit 中：

// startToolRound() 缩写；本例只有一个调用，省略不可用工具和串行批次分支。
await runtime.commit(async tx => {
  const assistant = await appendAssistant(tx, conversationId, message); // A1。
  const toolId = await createToolTask(tx, runtime, assistant.id, call.id); // T1，owner=G1。
  // 原码同时把 T1 及 callId 写入 pi.live.tools 的工具槽位。
  return {
    status: "waiting",
    checkpoint: { phase: "tools", assistant: assistant.id, tools: [toolId], pending: [] },
    on: [toolId], policy: "allSettled",
  };
}, context);

这次提交后的状态很具体：

S：仍为 placed
G1：waiting / tools，on=[T1]
T1：pending / call，owner=G1，输入指向 A1 中的这条工具调用
历史：增加 A1；pi.live.tools：增加待执行的 read 槽位

G1 的 invocation 随后退出。它没有占着一个内存 Promise 等 T1，而是把等待关系存了下来。 Scheduler 看到 T1 可以运行，接下来执行 ToolTask.phases.call()。

六、T1 执行 read，把结果提交后唤醒 G1

6.1 工具先保存执行意图，再访问外部环境

ToolTask.call 从 A1 读取工具名与参数，解析当前选中的实现，校验参数，执行 beforeTool hooks， 并对 hooks 改写后的参数再校验一次。通过后先提交执行意图，再运行工具：

// ToolTask.phases.call 内部缩写，省略上述校验与工具槽位更新。
await runtime.commit(async tx => ({
  status: "running",
  checkpoint: {
    phase: "execute", arguments: finalArgs, replay: tool.replay ?? "unsafe",
  },
}), context);
await run(runtime, call, tool, finalArgs, context); // 内部实际调用工具 execute()。

此时 checkpoint 已是 execute，但正常路径仍在 call 函数里直接运行工具。 这份 execute checkpoint 是为中途退出准备的恢复入口；第八节再解释 replay。

工具通过宿主提供的 env 读取文件，run 处理结果、afterTool hooks 与输出。 随后 settle() 在同一次提交中追加 tool-result、结算工具槽位，并把 T1 结算为 terminal。

T1：running / call
  → 校验与 hooks
  → 提交 running / execute，保存最终参数和 replay
  → 实际 read 文件
  → 提交结果 R1，T1=terminal，outcome=completed

意图提交和结果提交之间有一个真正的外部操作。SQLite 事务只能保证内部记录一致，不能自动把文件操作也包进同一个事务。

6.2 谁通知 G1，怎样继续？

T1 不直接调用 G1。Session 发布 T1 的结算提交后，Scheduler 把 T1 从 #live 中移除。 再次选取任务时，G1 的等待集合已经没有未结束任务：

// Scheduler.#waitingOn() 的普通等待分支。
return record.state.on.filter(id => this.#live.has(id));
// 对 G1：on=[T1]，但 T1 已 terminal，不在 #live 中，结果为 []。

Scheduler 因而把 G1 从 waiting 改为 running，保留 phase=tools，再调用 GenerationTask.phases.tools()。这次不是重新请求模型，而是先处理上一轮工具的完成情况。

本例只有 T1，且已结束，所以 tools 阶段进入 finishToolRound()。 若采用串行模式且还有未启动调用，则 tools 阶段先创建下一个 ToolTask，再提交 waiting，直到这轮调用全部处理完。 allSettled 表示工具失败也可以解除等待，让父任务拿到失败结果；不要求所有工具成功才继续。

七、G1 创建下一轮 G2，直到 S 得到最终回答

finishToolRound() 读取工具 outcome、处理 afterTools hooks 与控制信息，再决定结束还是继续。 本例没有 terminate、handoff 或 reset，因此创建新的 GenerationTask G2：

// finishToolRound() 的正常继续分支；在 runtime.commit 回调内。
delete live.tools;
handOver(live, runtime.taskId, await createGeneration(tx, conversationId));
return {
  status: "terminal",
  outcome: { status: "completed", result: { entryId: assistant } },
};

// handOver()：只切换当前推进者，保留 run.inputs 中的 S。
if (live.run?.taskId === from) live.run.taskId = to;

一次提交同时完成：G1 结束、G2 建立、pi.live.run.taskId 从 G1 改为 G2。 因此 G1 完成只表示这一轮完成，S 仍然是 placed。Scheduler 随后像运行 G1 那样运行 G2： prepare 读取包含 R1 的历史，request 再次请求模型。

模型据此提出 edit，G2 创建 T2 并等待；T2 修改文件并保存 R2；G2 恢复后创建 G3。 第三轮模型看到修改结果，给出最终回答 A3。这一次响应没有工具调用，走 answer() 分支：

// answer() 的完成路径缩写；位于 runtime.commit 内。
// 省略 onYield 要求继续的分支，以及排队输入和 reset 处理。
const entry = await appendAssistant(tx, conversationId, message); // A3。
endRun(tx, live, runtime.taskId, { status: "done", answer: entry.id });
return {
  status: "terminal",
  outcome: { status: "completed", result: { entryId: entry.id } },
};

endRun() 结算本次 run 的输入，S 变为 done，answer 指向 A3。 Submissions 服务观察到这次提交，通知等待 S 的调用方，于是第二节的 submission.wait() 返回。 若已有后续输入，边界处理还可能启动新的 run，但那是另一项工作的开始。

把整个过程压到一张表里，注意 S 一直是同一个：

节点	S	当前任务与调度动作	保存的数据
接纳输入	placed	创建 G1，pending/prepare	U1、S、pi.live.run
G1 请求模型	placed	running/prepare → running/request	system 变化、请求配置与 cutoff
模型提出 read	placed	G1 waiting/tools；T1 pending/call	A1、工具槽位、等待关系
read 执行	placed	T1 running/call → 保存 execute 意图 → terminal	最终参数、replay、R1
下一轮开始	placed	唤醒 G1 → G1 结束并创建 G2	run.taskId 改为 G2
edit 执行	placed	G2 与 T2 重复上一轮机制，然后创建 G3	A2、R2、run.taskId 改为 G3
最终回答	done	G3 terminal	A3、S.answer，当前 run 结束

到这里，正常路径已经闭合：调度器调用阶段函数，阶段函数提交状态，提交又驱动调度器推进其他任务。 接下来不引入新执行系统，只让这个循环在 T2 中途退出。

八、在 edit 中途退出后，怎样恢复同一项工作？

8.1 先看数据库留下了什么

假设 T2 已把文件改好，但还没提交 R2，进程就退出了。SQLite 中最后一次成功提交仍是：

S：placed
G2：waiting，checkpoint.phase=tools，on=[T2]
T2：running，checkpoint={ phase: execute, arguments: 最终参数, replay: unsafe }
历史中有 A2，但没有 R2

内存中的 invocation、Promise 和连接都没了。不过数据库仍足以判断：T2 停在执行意图之后， G2 等它结算，S 尚未完成。不能判断的是外部文件究竟改没改：意图提交后、实际修改前退出，也会留下同样的数据库状态。

8.2 safe / unsafe 只决定这种情况下能否自动重跑

replay 策略	第一次正常调用	意图已保存、结果未保存时恢复
safe	正常执行	保存的策略与当前工具定义都为 safe，才允许从头再执行
unsafe，也是默认值	正常执行	不自动重跑这次调用，保存 interrupted 错误结果

它不表示工具有没有权限或输入是否安全。safe 是工具作者的承诺：即使此前可能执行过，也允许再执行一次。 例如“查询当前配置”可能允许重复读取；“向文件追加一行”可能因此追加两行，不能随意声明 safe。 这只是业务语义示例，不能据此推断当前内置 read 已声明 safe。

safe 不等于从断点继续，也不等于只执行一次。 运行时无法替任意外部系统保证幂等。 需要严格去重时，工具应与外部服务合作，例如用稳定操作 ID 原子记录执行结果，再次调用时返回旧结果。 这属于工具的实现责任。

8.3 宿主重新启动，调度器按 checkpoint 找恢复函数

宿主重新构造 models、registry 和 env，打开同一数据库，再启用调度：

const storage = await openNodeSqliteStorage("./session.sqlite");
const harness = await Harness.open(storage, { models, registry, env }, context);
const conversation = await harness.root(context); // 同一个 C。
harness.resume();
// 可用 harness.submission(savedId, context) 重新取得 S，并等待结果。

重新安装实现是必要的：数据库保存工具选择、任务 kind 与 JSON 状态，不保存 JavaScript 函数。 Scheduler.open 扫描 pending、running、waiting、completing 任务；其中旧 running 的占用已经失效， 所以改回 pending，但不改变 checkpoint：

// Scheduler.open() 缩写；此时只恢复记录，不派发任务。
if (record.state.status === "running") {
  tx.setTask(withState(record, {
    status: "pending", checkpoint: record.state.checkpoint,
  }));
}
// resume() 启用后，仍使用第四节同一套选取、占用与执行流程。

现在 G2 仍在等 T2，先不能运行。T2 被 Scheduler 占用，根据 kind=pi.tool 找到 ToolTask，再按 phase=execute 调用 ToolTask.phases.execute()。

这里的阶段函数不是工具自己的 execute()。 它是恢复处理函数，先判断能否再次调用实际工具：

// ToolTask.phases.execute() 缩写。
const { arguments: args, replay } = task.state.checkpoint;
const tool = /* 按保存的调用名解析当前选中的工具 */;
if (replay === "safe" && tool?.replay === "safe") {
  // 原码先通过 commit 清除已保存的部分输出。
  return run(runtime, call, tool, args, context); // 用保存的参数从头执行。
}
// 否则 settle：保存 interrupted 工具结果，并以 failed outcome 结算 T2。

本例 unsafe，所以不会再改一次文件，而是生成“调用中断，可能已部分执行”的 R2。 T2 结算后，G2 的等待集合变空，恢复到 tools 阶段；之后与第六、七节相同，通常创建 G3， 把中断结果交给模型。模型可以解释问题或提出进一步检查。

T2 running/execute（旧进程）
 → open：pending/execute
 → resume：running/execute（新 invocation）
 → 恢复处理函数：重跑，或保存 interrupted
 → T2 terminal
 → G2 waiting/tools → running/tools
 → 正常继续分支创建 G3，最后结算 S

这就是恢复：读取持久记录，再运行记录指定的阶段。不是重新提交 U1，也不是重放所有历史； 已经 terminal 的 T1 不会因为恢复而再执行。unsafe 阻止的是这次 ToolTask 自动重放， 不是永久禁止模型提出一个新的 edit；有严格业务约束时，仍需工具自行核验外部状态。

8.4 不同中断位置，对应不同入口

最后保存的位置	恢复后的处理
工具 call，尚未提交执行意图	重新准备、校验、执行 hooks，然后尝试工具
工具 execute，尚未保存结果	按两份 replay 策略重跑或生成 interrupted
工具已经 terminal	使用保存的结果，由等待它的任务继续
模型 request 尚未完成	根据保存的请求依据重新处理请求，普通流式调用可能重发
模型 retry	按保存的时间继续等待，然后重试
模型 poll	使用保存的延迟响应 handle 继续查询
S 已 done	找回已完成结果，不再为同一 requestId 新建输入

恢复是接管未结束的工作，重试只是其中一种决策。普通模型请求重发不保证只计费一次； hooks 也可能再次运行，外部副作用不能只在工具 execute 中考虑。

九、主流程遇到新输入、错误或取消时，怎样分支？

现在已经知道每次执行都围绕 S、当前 generation 和工具任务推进，其他行为可以直接放回这些位置理解。

9.1 运行中又收到输入：先排队，在边界放入

await conversation.submit({ type: "input", content: "改完再检查一次" }, context);
await conversation.submit({
  type: "input", content: "使用 dev 配置", whenBusy: "steer",
}, context);

默认 follow-up 放入 pi.inbox，等当前 run 回答后启动下一次 run；steer 则在当前工具轮结束后放入， 加入正在进行的工作。whenBusy: "reject" 在忙碌时直接拒绝。 所以它们的接入点分别是第七节的最终回答边界和工具轮结束边界，不是在任意 token 中途改写请求。

输入 S 的完整生命周期是 queued → placed → done，空闲时可直接 placed。 被撤回、失败或取消等不能获得答案的输入结算为 unanswered，并带原因。 相同 Conversation 内，相同 requestId 会找回既有 submission；这解决重复接纳输入，不替外部工具请求去重。

9.2 模型失败：Generation 保存下一次行动

Generation 的 checkpoint 除 prepare、request、tools 外，还包含 retry 和 poll：

// GenerationCheckpoint 的额外分支缩写。
{ phase: "retry", attempt, until }
{ phase: "poll", attempt, model, cutoff, handle, pollAt }

可重试错误会保存等待截止时间，retry 阶段等待后提交新的 prepare checkpoint； 支持延迟结果的 Provider 返回 handle 后，poll 阶段用 fetchDeferred() 查询。 等待定时器并不自动意味着 Task.status=waiting；waiting 专门描述持久化任务依赖。 不可继续的模型错误则结算任务，并结束相应输入。工具失败不同：它通常成为工具结果，仍可交给下一轮模型。

9.3 用户取消：取消任务树，不是回滚文件

await conversation.abort(context);

取消标记会落盘，并中断当前执行；普通被拥有的子任务、子 Conversation 也需要收尾。 如果某任务已经决定 outcome，但普通子工作仍存活，就暂处于 completing，等子工作结束才 terminal。 父会话 abort 会撤回排队输入，排队的被动写入保留；调用等到会话 idle 才返回。

这不会撤销已经写入的文件。后台任务是显式的所有权边界，默认不阻塞父会话 idle，也不随父级 abort 取消； 需要包含它们时使用 abort(context, { background: true })。

子 Agent 也沿同一套机制接入：工具通过事务创建由自己拥有的子 Conversation，再向它提交输入。 可重放实现需要按 ownerTaskId 找回原子会话、用稳定 requestId 找回原输入，避免重启时重复创建子工作。 这不是在内存里启动一个 Promise 就能获得的保证。

十、怎样看到进度，怎样让长任务继续？

10.1 观察已经提交的状态

第三节的 commit 最后会 publish，调度器据此推进任务，界面也可以据此更新显示：

const watch = await conversation.watch(context);
render(watch.value);
watch.start(async (value, ops) => { await send(ops); });
await watch.stop(); // 停止观察，不是取消任务。

watch 返回历史与 documents 的视图，pi.live 提供当前生成和工具进度。 部分输出会节流提交，尚未提交的最新片段可能在崩溃时丢失；慢消费者积压超过限制时会收到最新完整视图， 不保证重放所有 token。取消 submission 的等待也只停止等待；要停止工作，应显式 abort。 客户端断开后宿主进程是否继续运行，由宿主决定。

10.2 模型窗口不够时，把压缩也交给任务系统

第五节的 prepare 不总是立刻进入 request。达到压缩阈值时，它可以先创建 CompactionTask， 把自己置为 waiting，等压缩完成后回到 prepare；也可以提前启动后台压缩。 Provider 报上下文过长时，还有压缩后重试一次的路径。

// 手动触发同类压缩任务的公开 API。
const id = await conversation.compact("保留配置修改的依据和检查结果", context);
const { outcome } = (await harness.waitForTask(id, context)).state;
if (outcome.status === "completed" && outcome.result.submissionId !== undefined) {
  const submission = await harness.submission(outcome.result.submissionId, context);
  const placed = await submission!.wait(context);
}

摘要生成与放入会话是两步：会话继续工作时，摘要等到合适边界应用；过时切点的摘要可能以 stale 结算。 pi.compaction 改变模型读取的上下文，旧历史保留。reset 则从新的上下文边界开始，也不删除旧记录。 这些都不代替任务 checkpoint：摘要说明“为什么修改配置”，checkpoint 指明“哪段执行还没结束”。 这里是 durable 的实现，不能套用 CLI AgentSession.compact() 的具体流程。

十一、读到这里应能还原的执行机制

整个运行时的骨架只有一条循环：

外部输入或阶段函数
  → 事务提交：历史 + 文档 + 任务状态
  → 发布提交
  → Scheduler 找可运行任务
  → 按 kind 和 checkpoint.phase 调用实现
  → 实现再次提交进度、等待关系或 outcome

模型请求和工具操作发生在这个循环内部，但处于存储事务之外。 因此 durable 的核心不是多写一份聊天日志，而是把“下一步由谁执行、从哪里执行、等待谁、怎样结束”也保存下来。 重启后重新建立调度器与实现注册，就能继续遵守这些关系；外部操作结果不确定时，则按明确策略处理。

这套保证还依赖运行条件：MemoryStorage 不支持跨进程保留；SQLite 默认 WAL + synchronous=NORMAL， 可承受进程崩溃，但断电或宿主故障可能丢失最近提交；JSONL 可启用 fsync: true。 当前存储由一个进程拥有，没有跨进程锁；宿主还需负责重启进程、恢复模型凭据和扩展代码。 不能据此推导出多 worker 竞争调度、自动故障转移，或任意工具恰好执行一次。

源码回查按本文执行顺序即可：

流程	源码入口（均在 packages/durable/src 下）
打开与提交	harness/harness.ts、harness/submissions.ts
原子保存与物理表	session/session.ts、storage/sqlite/storage.ts、storage/sqlite/migrations.ts
调度与状态定义	harness/scheduler.ts、types.ts
模型轮次与移交	harness/generation.ts：prepare、request、startToolRound、finishToolRound、answer
工具与恢复	harness/tool.ts：call、execute、run、settle
配置、排队、观察、压缩	harness/agent.ts、harness/inbox.ts、harness/view.ts、harness/compaction.ts

公开用法见 pi-durable README。 恢复与边界测试可从 harness-tools-recovery.test.ts、harness-generation-recovery.test.ts、 harness-submissions.test.ts、harness-ownership.test.ts 回查。 本文依据源码及测试内容复核；没有把阅读测试等同于执行测试，也未验证真实 Provider 或崩溃恢复 E2E。