第22章调度台——pi-agent-core

本章导读： 源码路线第 ④ 站（核心层）。仍在 pi-agent-core，但深入 harness/ 目录——读 agent-harness.ts、session/jsonl-store.ts 和 session/fork.ts。这一站要拆掉小a以为的"fork 就是 Git 分支"的错觉，看清会话到底靠什么活过一次进程。

小a照着 19 章，把 loop 里的状态变化理清楚了。但他盯着代码里 fork 这个函数名，忽然冒出个想法：“这不就是 Git 分支吗？我 fork 一个会话，再让两个 Agent 各改各的，互不干扰？”

“停。”老z说，“**会话 fork 不是 Git 分支，也不会创建或隔离 worktree。**你这句话里有两个概念被焊在一起了。我们先把‘会话活过一次进程’这件事，按 Harness 的装配层讲清楚。”

低层 loop 能把一次流和工具结果推进完，却不知道会话应保存到哪里、何时压缩、哪些 Skills 可用。AgentHarness 是把这些宿主决策接到核心 loop 的装配层。

本章依据 pi-mono 提交 583f153d502aa8e958eefdb9af0fbd3344e68f95、@earendil-works/pi-agent-core 与 @earendil-works/pi-storage-sqlite-node 0.83.0。跟读一次 AgentHarness.prompt、session repository／JSONL、fork、compaction、skills 和 Node SQLite 边界。不覆盖 coding-agent 如何选工作目录，也不证明文件系统或数据库在断电下的耐久性。

Harness 先定义运行期所有权

小a：“那 Harness 到底是什么？一个更大的循环？”

“不是。它是装配层——把宿主决策接到核心 loop。”老z说，“AgentHarness 构造时接收一个 Session、Models、model、可选 resources、tools、prompt 生成器、stream options 和 retry policy。它维护 phase（初值 idle）、active abort controller、pending session writes、三个消息队列、工具表和事件 handlers。”

比喻：调度台
loop 是引擎，Harness 是调度台：它决定会话写哪、何时压缩、哪些技能可用——但它自己不造引擎。

“注意，这不表示每个功能都会启用。源码只说明 Harness 可以组合它们；例如 resources 默认为空对象，system prompt 可以是字符串、函数或未提供，skills 是否加载由宿主传入。”

成员／依赖	本次 turn 中的职责	主要边界
`session`	build context、append entry、取得 branch	具体后端由 repository／store 决定
`models`	调用 `streamSimple`	认证与 Provider 仍在 pi-ai
`tools`／active names	构成 agent context 的可执行工具	名称重复／缺失会拒绝配置
`resources`	skills、prompt templates	文本资源不等于权限
`phase`＋controller	串行 turn、abort、shutdown	不撤销已发生的 tool 副作用
pending writes	run 时延后会话 mutation	写失败仍会向上报告

一次 `prompt` 的调用和写入序列

小a：“那一次 prompt 从进来到会话写盘，怎么走？”

“prompt 只在 phase === "idle" 时可开始，否则抛 AgentHarnessError("busy")。”老z打开代码：

// packages/agent/src/harness/agent-harness.ts，prompt（583f153d，节选）
async prompt(text: string, options?): Promise<AssistantMessage> {
	this.assertNotShutDown();
	if (this.phase !== "idle") throw new AgentHarnessError("busy", "AgentHarness is busy");
	this.phase = "turn";
	const operation = this.startOperation();
	try {
		const turnState = await this.createTurnState();
		return await this.executeTurn(turnState, text, operation.signal, options);
	} catch (error) {
		this.phase = "idle";
		throw normalizeHarnessError(error, "unknown");
	} finally { operation.finish(); }
}

“createTurnState 是首次读 session 的位置：session.buildContext() 给 messages，session.getMetadata() 给 sessionId，随后解析 tool context、收集 active tools、并生成 system prompt。它返回的是 turn snapshot，避免中途读到一半新配置。”

“executeTurn 首先创建 user message；若 nextTurnQueue 有内容则排在用户消息前，随后运行 before_agent_start hook。最终它调用第21章双层循环——pi-agent-core中的 runAgentLoop，但把自己的 handleAgentEvent 和 createStreamFn 注入进去。”

text

prompt(text)
  → phase: idle → turn；创建 AbortSignal
  → createTurnState: session.buildContext / metadata / tools / prompt
  → executeTurn: before_agent_start，组装初始 messages
  → runAgentLoop
      → createStreamFn → models.streamSimple
      → handleAgentEvent(message_end) → session.appendMessage
      → handleAgentEvent(turn_end) → flush pending writes → save_point
      → handleAgentEvent(agent_end) → flush → phase idle → settled
  → 从 newMessages 反向找到最后一条 assistant 并返回
  → finally flushPendingSessionWrites / finish operation

“createStreamFn 会先触发 before_provider_request，再调用 this.models.streamSimple(model, context, …)，并将 sessionId、headers、timeout、retry 等 stream options 传下去。这里连接了 Harness 与第19章目录与户口——pi-ai中的 Provider 链；它没有绕开 ModelsImpl.applyAuth。”

逐段读 `executeTurn` 的失败归约

小a：“那 loop 抛错了怎么办？会话会不会只剩半条？”

“看 executeTurn 的失败归约。”老z说：

// packages/agent/src/harness/agent-harness.ts，executeTurn（583f153d，节选）
try {
	return await runAgentLoop(/* messages、context、loop config、event handler、signal、stream fn */);
} catch (error) {
	try {
		return await this.emitRunFailure(activeTurnState.model, error, signal.aborted, signal);
	} catch (failureError) {
		throw new AgentHarnessError("unknown", "Agent run failed and failure reporting failed", cause);
	}
}

“外层 run 抛错时，Harness 不直接让 session 只有半条事件：它用 emitRunFailure 构造 assistant failure message，依次交给 message_start、message_end、turn_end、agent_end。createFailureMessage 根据 signal.aborted 设置 stop reason 为 aborted 或 error，usage 为零。若连失败报告也失败，才以 aggregate cause 的 AgentHarnessError 终止。”

“这带来更可审计的正常失败路径，但代价是写入／listener 再失败时仍无法凭空保证会话完整。executeTurn 的 finally 还会 flush pending session writes；flush 自己失败时，错误会传播，不能被描述为‘最终一定保存’。”

路径	触发点	Harness 行为	不覆盖的承诺
正常 turn	loop 返回 messages	返回最后 assistant	assistant 内容一定有用
并发 prompt	phase 非 idle	`busy` error	排队执行第二个 prompt
loop／hook／provider 抛错	`runAgentLoop` reject	生成并 append failure message	后端写入必成功
abort	controller abort	失败消息 reason 为 `aborted`；队列清理	外部工具回滚
shutdown	`requestShutdown`	清队列、abort active operation	删除 durable session

session repository 把 store 变成可用会话

小a：“那 fork 到底在复制什么？我们回开头那个问题。”

“好，看 SessionRepository.fork 的真实代码。”老z打开 repository.ts：

SessionRepository 接收 store、可选 search 和 context build options。create／open 用 store 返回的 reader 创建 session；fork 将外部选择参数转为 SessionForkSelection，交给 store.fork，再创建新 Session。

// packages/agent/src/harness/session/repository.ts，SessionRepository.fork（583f153d，节选）
async fork(source, options): Promise<Session<TMetadata>> {
	const selection = createSessionForkSelection(options);
	const createOptions = { ...options };
	delete createOptions.entryId;
	delete createOptions.position;
	return createSessionFromReader(
		this.store,
		await this.store.fork(source, createOptions, selection),
		this.contextBuildOptions,
	);
}

“这段清楚表明 fork 的输入是 session metadata、创建选项和 entry selection；没有 Git 命令、目录复制或 worktree API。createSessionForkSelection 在无 entryId 时选 all；position: "at" 时到指定条目为止；默认／before 时要求目标是 user message，并取其 parent 的路径。”

“readSessionEntriesForFork 找不到 entry 会抛 SessionError("invalid_fork_target")；如果选择 before-user 但目标不是 user message 也抛相同代码。故 fork 并非任意条目都可以在‘之前’切开。”

fork selection	读取行为	错误边界
无 `entryId`	`reader.readEntries()`	依 reader 的存储错误
`position: "at"`	`readPathToRootOrCompaction(target.id)`	目标不存在
默认／`before`	目标必须是 user message，读其 parent path	非 user 目标或不存在
新 session	store 创建新 metadata／document	不复制磁盘 worktree

小a：“那我想让两个实验各改各的文件，怎么办？”

“用 Git branch／worktree 或其他工作区隔离。”老z说，“会话 fork 的含义是‘从历史上下文另开路径’。从 session fork 推出文件隔离，是没有源码证据的逻辑跳跃。”

JSONL 的单位是 session header 加树条目

小a：“那 JSONL 文件里，每行都是一条消息吗？”

“不是。”老z说，“JSONL store 的 SessionHeader 固定有 type: "session"、version: 3、id、timestamp、cwd，可选 parentSession 和 metadata；后续每一行才是 SessionTreeEntry。因此不能写成‘每行都是用户或助手消息’。”

// packages/agent/src/harness/session/jsonl-store.ts，appendEntry（583f153d，节选）
if (entryIds.has(entry.id)) throw new SessionError("invalid_entry", `Entry ${entry.id} already exists`);
getFileSystemResultOrThrow(
	await this.fs.appendFile(metadata.path, `${JSON.stringify(entry)}\n`),
	`Failed to append session entry ${entry.id}`,
);
entryIds.add(entry.id);
this.entriesByPath.get(metadata.path)!.push(entry);

“append 前检查 file 存在、加载／维护每个 path 的 entry ID 集，再串行入队写入。store 用 KeyedOperationQueue 以 document operation key 排队，默认最多四个跨 key 操作；这能支持‘同一会话条目写入有队列控制’，不能证明跨进程追加或崩溃耐久性。”

“加载时，第一行必须是 version 3 session header；后续每行会验证 JSON、entry type、id、parentId、timestamp，leaf 还校验 targetId。重复 entry ID 也会让 session 无效。正常 JSONL 的可恢复性来自重新读取 header 与条目，不来自‘append 所以永不丢数据’的承诺。”

JSONL／session 状态	例子	意义
header	session、id、cwd、parentSession	文档身份与来源
`message` entry	user、assistant、toolResult 等 AgentMessage	对话／运行记录的一类
`model_change`／`thinking_level_change`	配置变更	不是对话消息
`compaction`／`branch_summary`	摘要与树导航记录	影响后续 context
`leaf`／`label`／`session_info`	树位置、标注、会话名	会话元数据／导航

“JSONL store fork 会先从源文档读取 readSessionEntriesForFork 的选中条目，再创建新文件，header 的 parentSession 默认指源路径。它复制的是选中 session entries 和 metadata，不是 Git 工作区。”

压缩不是无损截断

小a：“那上下文满了，Harness 怎么压缩？”

“Harness 的 compact() 只允许 idle phase：读 session.getBranch()，运行 prepareCompaction，可让 session_before_compact hook cancel 或提供摘要；否则调用 compact(...)，成功才 append compaction entry。”老z说，“prepareCompaction 的职责是从树条目选择保留边界、估算 token 和准备 summary 任务。compact 再调用模型。generateSummaryWithUsage 明确将模型 response 的 stopReason === "aborted" 转为 CompactionError("aborted", …)，"error" 转为 CompactionError("summary_failed", …)；Harness 将其归约为 AgentHarnessError("compaction", …)。”

“这是一条独立模型请求，不是静默删消息；completeSimpleWithRetries 为 summary 设置 cacheRetention: "none" 与新的 sessionId。收益是控制上下文长度且保留摘要和 recent tail；代价是摘要会丢失信息、生成要花 token，且 abort／error 都可能阻止写入。”

Skills 加载文本和诊断，不授予执行权限

小a：“那 Skills 呢？加载了就能用？”

“加载和授权是两回事。”老z打开 skills.ts：

loadSkills(env, dirs) 接受一个或多个目录。不存在的根目录会跳过；其他 fileInfo、list、read、frontmatter parse、metadata 问题变成 diagnostics。它递归查找 SKILL.md，还允许根目录直接 .md，尊重 .gitignore、.ignore、.fdignore。

// packages/agent/src/harness/skills.ts，loadSkills（583f153d，节选）
export async function loadSkills(env, dirs): Promise<{ skills: Skill[]; diagnostics: SkillDiagnostic[] }> {
	const skills: Skill[] = [];
	const diagnostics: SkillDiagnostic[] = [];
	for (const dir of Array.isArray(dirs) ? dirs : [dirs]) {
		const rootInfoResult = await env.fileInfo(dir);
		if (!rootInfoResult.ok) { /* not_found 跳过，其余记 diagnostic */ continue; }
		const result = await loadSkillsFromDirInternal(env, rootInfo.path, true, ignore(), rootInfo.path);
		skills.push(...result.skills);
		diagnostics.push(...result.diagnostics);
	}
	return { skills, diagnostics };
}

“一个有效 skill 仍需要 description；name 有小写、长度、目录名一致等校验。加载结果是 name、description、content、filePath 等文本资源。它本身不执行 shell、不启用 tool、也不强制路径权限；把 SKILL.md 的建议当成安全沙箱，是错误前提。”

`pi-storage-sqlite-node` 的明确边界

小a：“那 SQLite 包呢？它是不是 JSONL 的升级版？”

“不是‘升级版’，是另一个后端。”老z说，“这个包不是 JSONL 的内部实现替代文字，而是一个 Node 专用 SQLite 后端。其 src/index.ts 从 node:sqlite 导入 DatabaseSync，createNodeSqliteFactory().open(path) 用它创建 database wrapper；随后 re-export SQLite session backend 和 types。”

// packages/storage/sqlite-node/src/index.ts，transaction（583f153d，节选）
async transaction<T>(fn: () => Promise<T>): Promise<T> {
	this.db.exec("BEGIN");
	try {
		const result = await fn();
		this.db.exec("COMMIT");
		return result;
	} catch (error) {
		try { this.db.exec("ROLLBACK"); } catch { /* rethrow original error */ }
		throw error;
	}
}

“它提供的是 SqliteDatabase／factory 抽象的 Node adapter，事务中 fn 抛错时尝试 rollback 并重抛原错误。包元数据要求 Node >=22.19.0，并直接依赖 pi-ai 和 pi-agent-core。因此无法由此文件推出浏览器可用性、SQLite 文件备份策略、跨多个 DatabaseSync 实例的并发语义，或 JSONL 与 SQLite 自动迁移。”

收益、代价与未覆盖范围

基于源码的推断： Harness 把 loop 的事件、会话读写、hook、资源和 provider options 集中，使产品宿主可以换 JSONL／内存／SQLite 后端而保持 Agent 核心的接口。代价是它是有状态装配层：队列、phase、pending writes、hook 和后端错误必须一并测试，不能只验证模型文本。

本章未覆盖 pi-coding-agent 如何实际构造 store 或定位 .pi，没有验证 JSONL 与 SQLite 的性能／崩溃恢复，也没有把 Skills 内容当作可信指令。会话 fork 只描述上下文历史；Git worktree 属于版本控制层。

小结

这一章回答的是：一个回合跑完后，会话凭什么"活"过进程重启。答案是 AgentHarness 这个装配层——prompt 从 session 构建 turn snapshot，向 runAgentLoop 注入 provider stream 和事件处理，并在 message_end、turn_end、agent_end 三个事件点写会话、发出保存语义。repository／JSONL 的 fork 复制的是选中的 session history，不是 Git 工作区；compaction 是可能失败的独立摘要请求，不是静默截断；skills 是带 diagnostics 的文本加载，不授予执行权限。

边界：session fork 不创建 Git 分支或 worktree——要隔离文件副作用，用 Git 或工作区机制，别指望会话 fork。compaction 的 aborted 与 summary_failed 是两个独立的失败 code，各有恢复路径。

记住：会话 fork 不是 Git 分支。 它复制的是上下文历史，不是文件系统。这是这一章最容易踩的坑——把"fork 会话"当成"分支工作区"，就会在文件隔离上做出错误的安全假设。

源码走查

从 packages/agent/src/harness/agent-harness.ts 的 prompt 跟到 createTurnState、executeTurn；记录 sessionId 在哪个函数读取、在哪个 stream options 字段向下传递。
阅读 handleAgentEvent；分别验证 message_end 立即 append、turn_end flush 后发 save_point、agent_end 再 flush 并置 idle 的顺序。
在 repository.ts、fork.ts、jsonl-store.ts 搜索 fork；确认 selection 只读取 session entry，且 JSONL 新 header 使用 parentSession，没有任何 Git 或 worktree 调用。
手工构造三行 JSONL：合法 header、合法 entry、重复 entry ID；对照 loadJsonlSession 的重复检查，预测第三种为何失败。不要把它写入真实会话目录。
从 compact() 跟到 prepareCompaction 与 generateSummaryWithUsage；分别检查 aborted 与 error 被转换成的 CompactionError code。
阅读 loadSkills 和 loadSkillFromFile；验证缺失根目录与解析失败都是 diagnostics 路径，并指出为何这不构成权限校验。
阅读 packages/storage/sqlite-node/src/index.ts 的 transaction 与 createNodeSqliteFactory；确认它依 node:sqlite，并列出该事实无法证明的一个耐久性结论。

第22章 调度台——pi-agent-core ​

Harness 先定义运行期所有权 ​

一次 prompt 的调用和写入序列 ​

逐段读 executeTurn 的失败归约 ​

session repository 把 store 变成可用会话 ​

JSONL 的单位是 session header 加树条目 ​

压缩不是无损截断 ​

Skills 加载文本和诊断，不授予执行权限 ​

pi-storage-sqlite-node 的明确边界 ​

收益、代价与未覆盖范围 ​

小结 ​

源码走查 ​