架构（二）：pi-agent-core 的运行循环

上一篇 架构（一）pi-ai 讲的是循环下游那个流。这一篇往上一层看：驱动这个流的循环本身在哪、长什么样、边界在哪。

它是 agent 的心脏。整篇专栏里所有关于「Pi 怎么跑起来」的问题，最后都落在这六个文件上。而且它小到你可以在一个下午读完——agent-loop.ts 949 行，agent.ts 613 行，types.ts 539 行，proxy.ts 393 行，stream-fn.ts 20 行，index.ts 5 行。

这一篇的主线是一个结构判断：

同一个包里住着两个角色。一个是无状态的纯循环，一个是有状态的类。前者你直接调，后者把前者包起来加上队列、事件和中止。

先说为什么这个区分重要，再说不做什么。

18.1 两个文件，两种角色

依赖面小得惊人

packages/agent/package.json 的全部运行时依赖是两个：

"dependencies": {
  "@earendil-works/pi-ai": "1.3.0",
  "typebox": "1.3.27"
}

typebox 只用于 types.ts 里的 TSchema 类型标注。除此之外没有任何第三方依赖。

更有说服力的是三条 grep 结果：

这个包不读一个文件，不 import 一个供应商。 它唯一的网络访问是 proxy.ts 里那个可选的、面向浏览器应用的代理转发。这不是代码风格，是可用性证明：一个能被你 head -100 读完的 agent 循环，和一个必须读完才能信任的 agent 循环，是两种不同的东西。

无状态的那一半

runAgentLoop()（agent-loop.ts:102-126）的签名把设计意图写在脸上：

export async function runAgentLoop(
	prompts: AgentMessage[],
	context: AgentContext,          // ← 所有状态都在这里
	config: AgentLoopConfig,        // ← 所有策略都在这里
	emit: AgentEventSink,           // ← 所有输出走这里
	signal: AbortSignal | undefined,
	streamFn: StreamFn,             // ← 传输从这里进
): Promise<AgentMessage[]> {
	const initialMessages = declareToolChanges(context, prompts);
	const newMessages: AgentMessage[] = [...initialMessages];
	const currentContext: AgentContext = {
		...context,
		messages: [...context.messages, ...initialMessages],
	};
	...

五个参数，没有 this。AgentContext 只有两个字段——messages 和 tools——而 messages 这个数组它在函数内部原地 mutate：

// types.ts:501-507
export interface AgentContext {
	/** Transcript visible to the model. */
	messages: AgentMessage[];
	/** Tools available for execution in this run. */
	tools?: AgentTool<any>[];
}

注意顶部那句文件注释：

Agent loop that works with AgentMessage throughout. Transforms to Message[] only at the LLM call boundary.

AgentMessage 可以是任何东西（自定义消息类型那一节会讲声明合并），而 Message[] 是供应商能接受的东西。转换只发生一次，就在流式请求前的那一行。 这是整个包最重要的一条内部纪律：循环的其余部分不知道 LLM 协议长什么样。

配套的还有两个薄包装：

• agentLoop() — 起一个新循环，把 prompts 加进 context，然后返回 EventStream<AgentEvent, AgentMessage[]>；
• agentLoopContinue() — 从已有 context 继续，不加新消息。

agentLoopContinue() 有两条前置检查值得注意：context 为空直接抛错，最后一条是 assistant 也直接抛错。注释解释了原因：

Important: The last message in context must convert to a user or toolResult message via convertToLlm. If it doesn't, the LLM provider will reject the request. This cannot be validated here since convertToLlm is only called once per turn.

翻译：循环无法预检这件事，因为 convertToLlm 每个 turn 只调一次，而且是宿主给的函数。所以它在最便宜的位置拒绝，而不是等到供应商返回 400。

有状态的那一半

Agent 类（agent.ts:188-613）的文档注释三行讲完了它多出来的职责：

Stateful wrapper around the low-level agent loop. Agent owns the current transcript, emits lifecycle events, executes tools, and exposes queueing APIs for steering and follow-up messages.

对照一下它比纯循环多出来的四样东西：

它做的事情也很朴素：prompt() → runPromptMessages() → runWithLifecycle() → 调 runAgentLoop()，把 emit 接到自己的 processEvents() 上。

// agent.ts:432-446
private async runPromptMessages(messages, options = {}): Promise<void> {
	await this.runWithLifecycle(async (signal) => {
		await runAgentLoop(
			messages,
			this.createContextSnapshot(),
			this.createLoopConfig(options),
			(event) => this.processEvents(event),
			signal,
			this.streamFunction,
		);
	});
}

中间那一层 processEvents() 就是这个包最重要的一段代码，因为它同时是两件事：

1. 把循环事件归约成 AgentState（streamingMessage、pendingToolCalls、messages 追加、errorMessage）；
2. 然后 await 所有监听器。

// agent.ts:600-611
case "agent_end":
	this._state.streamingMessage = undefined;
	break;
}

const signal = this.activeRun?.abortController.signal;
if (!signal) {
	throw new Error("Agent listener invoked outside active run");
}
for (const listener of this.listeners) {
	await listener(event, signal);      // ← 逐个 await，按注册顺序
}

逐个 await，串行，按注册顺序。 这不是实现细节，是一个可观察的语义：你的慢监听器真的会拖慢整个循环。

两层 API 的差别大到什么程度

README 里有一段专门讲这件事，值得抄下来：

These low-level streams are observational. They preserve event order, but they do not wait for your async event handling to settle before later producer phases continue. If you need message processing to act as a barrier before tool preflight, use the Agent class instead of raw agentLoop().

差别用具体场景说清楚。假设你要在每条 message_end 到达时把消息写进数据库：

• 用 agentLoop()：你收到 message_end，开始一个异步写库操作。循环不等你，立刻进入工具预检阶段去执行工具。结果是数据库写入和工具执行并发，顺序不受你控制。
• 用 Agent：你的监听器是 processEvents() 里 await 的一环。Agent.processEvents() 在 message_end 分支里先把消息推进 _state.messages，然后才 await 你的监听器。监听器 settle 之后循环才继续。 这时 beforeToolCall 看到的状态已经包含那条 assistant 消息，数据库也已经落盘。

这就是为什么 Pi CLI 走 Agent 而不是裸循环。 差别不在功能，在于你的副作用能不能当屏障用。

18.2 运行循环

外面一层，里面一层

runLoop()（agent-loop.ts:163-321）是全包最该逐行读的一段。它的形状是双层循环：

// Outer loop: continues when queued follow-up messages arrive after agent would stop
while (true) {
	let hasMoreToolCalls = true;

	// Inner loop: process tool calls and steering messages
	while (hasMoreToolCalls || pendingMessages.length > 0) {
		...一个 turn...
	}

	// Agent would stop here. Check for follow-up messages.
	const followUpMessages = (await config.getFollowUpMessages?.()) || [];
	if (followUpMessages.length > 0) { pendingMessages = followUpMessages; continue; }
	if (explicitContinuation) { explicitContinuation = false; continue; }
	break;
}

内层循环回答的是「这次请求之后还有工具要跑吗、有 steering 消息要插吗」，外层循环回答的是「Agent 本来要停了，现在还有 follow-up 吗」。

这个拆分不是为了好看，它对应两种完全不同的用户意图。steering 是「你正在做的事我改主意了」，所以要赶在下一轮请求之前进去；follow-up 是「你手上的事做完再说」，所以要等到 Agent 自然停下来才进去。外层循环存在的唯一理由就是给 follow-up 一个不会被工具调用无限推迟的入口。

内层循环的退出条件是 hasMoreToolCalls === false && pendingMessages.length === 0。注意它不是「消息类型」判断，是「有没有下一件事」判断——这意味着一次纯文本回复（没有工具调用）也会退出内层，落到外层去查 follow-up。

一个 turn 的完整路径

把内层循环一次迭代拆开，按 runLoop() 的行序：

一整轮的事件序列，无工具调用时（README 的原图）：

prompt("Hello")
├─ agent_start
├─ turn_start
├─ message_start   { message: userMessage }      // 你的 prompt
├─ message_end     { message: userMessage }
├─ message_start   { message: assistantMessage } // LLM 开始响应
├─ message_update  { message: partial... }       // 流式增量
├─ message_end     { message: assistantMessage } // 定稿
├─ turn_end        { message, toolResults: [] }
└─ agent_end       { messages: [...] }

有工具调用时，多出来的部分是 tool_execution_start → _update → _end → 一对 message_start/message_end（工具结果消息），然后再开一个 turn_start。

AgentEvent 一共 10 种（types.ts:516-539）：

message_update 只对 assistant 消息发出，且它带一个嵌套的 assistantMessageEvent——也就是 pi-ai 的那 12 个流式事件原样透传。两层事件不合并、不翻译，这是 pi-agent-core 和 pi-ai 之间那条最干净的边界。

它带的 message 字段是顶层浅拷贝：

// agent-loop.ts:435-440
await emit({
	type: "message_update",
	assistantMessageEvent: event,
	message: { ...partialMessage },     // ← 只复制了顶层字段
});

[17 篇文章](/docs/pi-agent/17-architecture-pi-ai) 警告过 partial 是同一个 AssistantMessage 被反复 mutate 的共享引用，留着当历史状态会拿到 N 份同一引用。这里多做的 { ...partialMessage } 确实挡了一层——但只挡了顶层：content 数组仍然是同一份，里面的内容块仍然会被后续事件改写。

所以结论不变，只是原因不同：要在事件里当场读，不要把 message 存下来等流结束。 顶层浅拷贝让你 message.role、message.stopReason 这些会变的字段安全了一点，content 则完全没变。

declareToolChanges：可执行工具集怎么告诉模型

这是上一节没展开的一个设计，值得单独讲，因为它是纯循环和无状态模型的又一次交锋。

AgentState.tools 是运行时能执行的工具；transcript 里的 system 消息是模型被告知能调的工具。这两个集合可以不一致——exposure: "hidden" 的工具能执行但模型看不到，codemode 的工具不声明但 codemode 脚本能调。

declareToolChanges()（agent-loop.ts:333-363）负责在每次请求前把这个差集变成一条 system 消息：

const changes = getToolStateChanges(
	getCurrentTools([...context.messages, ...baseline]),        // transcript 声明的
	(context.tools ?? []).map(toToolDeclaration),               // 运行时可执行的
);

差集为空就直接放行。差集非空时，两种落法：

• 待处理消息里已经有 system 消息 → 就地替换它的 toolsAdded / toolsRemoved（withToolChanges()）；待处理消息对象本身被保留。
• 没有 → 造一条空的 system 消息，插到第一条非 system 的待处理消息之前。

第二条的插入位置不是随便定的。system 消息必须在它管用的那段历史之前，插在末尾模型就看不到。

源码注释把这个契约写得很清楚：

context.tools is what the runtime can execute; the transcript's system messages declare what the model may call. Before each request the difference becomes toolsAdded and toolsRemoved on a system message.

执行权和知情权是两个东西。 整个 17 篇文章 讲的 resolveTranscriptTools() 在 pi-ai 层做的是同一件事的下一段（决定发什么形状的工具列表给供应商），declareToolChanges() 做的是它前面那段（把差集变成 transcript 里的消息）。两层都存在，不是重复。

finishTurn 与那个显式续跑标志

finishTurn 是宿主唯一能在 turn 边界上决定还要不要再来一轮的钩子。它的返回值是个二选一：

// types.ts:147
export type AgentTurnDecision = { action: "continue" } | { action: "end" };

{ action: "end" } 很直接：turn_end 之后立刻 agent_end 收尾，连队列都不查。{ action: "continue" } 复杂一点，因为循环不会因为它就多跑一轮。

看 runLoop() 里这三行的配合就明白了：

explicitContinuation = decision?.action === "continue";
pendingMessages = (await config.getSteeringMessages?.()) || [];
if (hasMoreToolCalls || pendingMessages.length > 0) {
	explicitContinuation = false;      // ← 已经有事要做了，标志作废
}

然后在外层循环里消费：

// No natural request was selected, so fulfill the continuation decision with one context-only turn.
if (explicitContinuation) {
	explicitContinuation = false;
	continue;
}

README 把这个机制说得很直白：

If tool results, steering, or a follow-up already cause that request, they satisfy the decision and no additional request is made; otherwise the loop makes one context-only request.

也就是说：continue 不是「多跑一轮」，而是「保证至少再有一次请求」。 已经有工具结果或队列消息要跑了，这个决定就被满足了，不额外发请求。这么设计的原因很实在——finishTurn 是宿主写的，它看到的东西和循环看到的一样；如果它每次返回 continue 都硬跑一轮，Pi 里那些基于 finishTurn 做重试的扩展会疯狂重发。

而 context-only 那一次请求，正是「把刚才被删掉的东西再问模型一次」这类操作的实现基础——配合 prepareNextTurn 换掉的 context，它不需要任何新消息就能问出一个新问题。

一个必须自己加的护栏：无条件返回 continue 会死循环。 README 专门警告了这点，而且给了正确写法——先判断：

agent.finishTurn = async ({ message }) => {
	if (message.stopReason === "error" || message.stopReason === "aborted") return;
	if (shouldEndRun(message)) return { action: "end" };
	return needsAnotherResponse(message) ? { action: "continue" } : undefined;
};

error 和 aborted 是硬退出，它们的 finishTurn 决定被忽略。这不是优化，是防止「一次网络失败被当成「再来一轮」」导致的重试风暴。

prepareRequest 与 prepareNextTurn 的分工

这两个钩子很容易混。差别是一个 poll 队列，一个不 poll：

prepareRequest 不能追加消息，是有意的：它跑在「消息已经 emit 完」之后，这时候再插消息就要额外发一对事件，语义会变得含混。要在请求前插内容，用 prepareNextTurn 的 messages。

README 给的 prepareRequest 例子说明了它真正的用途：

agent.prepareRequest = async ({ context }) => ({
	context: { ...context, messages: await session.loadModelContext() },
});

每次请求前都从持久层重载一次。 这是「transcript 才是真相」这条原则的执行方式：内存里的数组可以被任何 hook 改，但落盘的版本才是下一次请求应该用的。

18.3 工具执行：并行、顺序与截断保护

预检：每一次调用的必经之路

不管并行还是顺序，每个工具调用都要先过 prepareToolCall()（agent-loop.ts:708-777）。它返回两种形态之一：

• kind: "immediate" — 不需要执行，直接产出一个结果（未知工具、被阻断、已中止、校验失败）；
• kind: "prepared" — 一切就绪，带上工具和校验过的参数。

五步，顺序固定：

1. 找工具。 找不到 → Tool <name> not found，isError: true。
2. prepareArguments。 工具可选的兼容垫片，在 schema 校验之前跑。返回值和输入是同一个引用时直接短路。
3. validateToolArguments。 按 TypeBox schema 校验。
4. beforeToolCall 钩子。 拿到的是已校验的参数——所以钩子里不用再校验一遍。
5. 查 signal.aborted。 钩子返回后、执行前各查一次。

整个函数被一个 try/catch 包着，任何异常都变成 isError: true 的工具结果。没有一个工具异常能逃出去杀死 agent 循环。

beforeToolCall 的返回结构只有三个字段：

// types.ts:66-74
export interface BeforeToolCallResult {
	block?: boolean;
	reason?: string;
	/**
	 * Hint that the agent should stop after the current tool batch when this call is blocked.
	 * Early termination only happens when every finalized tool result in the batch sets this to true.
	 */
	terminate?: boolean;
}

block: true 的调用不会执行，而是产生一个错误工具结果，reason 成为给模型看的文本。这是宿主接权限系统的正确入口——但注意措辞：这是「宿主可以实现的机制」，不是「这个包提供的权限系统」。后者不存在，下一节专门说。

并行与顺序的选择

选择发生在 executeToolCalls()（agent-loop.ts:508-523）：

const hasSequentialToolCall = toolCalls.some(
	(tc) => currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential",
);
if (config.toolExecution === "sequential" || hasSequentialToolCall) {
	return executeToolCallsSequential(...);
}
return executeToolCallsParallel(...);

两个条件是「或」，不是「与」。 也就是说：只要批次里有任何一个工具声明了 executionMode: "sequential"，整批都顺序执行，哪怕全局配置是 parallel。

这条规则是保守的，但方向是对的。工具 A 声明顺序执行，意思就是「我在别的工具运行时不能跑」——因为 A 本身可能依赖「此刻没有别的工具在改同一份状态」。批次里并发跑掉，等于让 A 的前提失效。放宽并行会让 A 的作者自己承担风险，所以默认收紧。

并行模式里那个刻意的不对称

这是整篇文章最值得抄进笔记的一段设计。

并行模式的两阶段：第一阶段 for 循环串行地 emit tool_execution_start 并做预检，把真正要执行的包成闭包推进数组；第二阶段 Promise.all 一次性放它们跑。

// agent-loop.ts:596-654（压缩版）
for (const toolCall of toolCalls) {
	await emit({ type: "tool_execution_start", ... });
	const preparation = await prepareToolCall(...);
	if (preparation.kind === "immediate") { /* 直接定稿 */ continue; }
	finalizedCalls.push(async () => {
		const executed = await executePreparedToolCall(preparation, signal, ...);
		const finalized = await finalizeExecutedToolCall(...);
		await emitToolExecutionEnd(finalized, emit);     // ← 在闭包里
		return finalized;
	});
}

const orderedFinalizedCalls = await Promise.all(...);   // ← 按源顺序 await
for (const finalized of orderedFinalizedCalls) {
	await emitToolResultMessage(createToolResultMessage(finalized), emit);   // ← 按源顺序
}

两条事件流，两种顺序：

Promise.all 的存在保证了第二阶段整体等完，但它不改变结果数组的顺序——数组里闭包的顺序就是源顺序，await 它们只是等完成。所以第二阶段天然是源顺序。

README 和 types.ts 用同一句话说了两遍，值得原样引用：

tool completion events follow tool completion order, but persisted toolResult messages still follow assistant source order.（README:114）

tool_execution_end is emitted in tool completion order after each tool is finalized, while tool-result message artifacts are emitted later in assistant source order.（types.ts:44-45）

这个不对称是解决两个矛盾需求，不是可以简化的冗余。 UI 要「完成即报」，transcript 要「顺序稳定」。如果你把它们合并成一种顺序，要么 UI 慢，要么 transcript 乱。

顺带一提 tool_execution_update（流式部分结果）也在闭包里，走完成顺序——它和 tool_execution_end 是同一个消费者。

terminate 的全票通过规则

shouldTerminateToolBatch() 只有一行，但语义是硬的：

// agent-loop.ts:690-692
function shouldTerminateToolBatch(finalizedCalls: FinalizedToolCallOutcome[]): boolean {
	return finalizedCalls.length > 0 && finalizedCalls.every((finalized) => finalized.result.terminate === true);
}

每一个都必须是 true，有一个不是就整体不终止。 空批次返回 false（length > 0 那半边）。

三处可以产生 terminate：工具的 execute() 返回值、beforeToolCall 阻断时返回的结果、afterToolCall 的覆盖。三个来源，最后都汇进同一个 result.terminate。

这个「全票通过」在 CLI 里有一个直观的对应：用户在工具运行中按了中止，但有个工具正在做写文件——中止不能让它跳过中止响应。反过来，如果批里只有一个工具、它说终止，那 every 自然成立。规则简单到不需要想，但正因为简单，写代码时容易用错成 some。

截断保护：这一条不能改

stopReason === "length" 时，这条消息里的工具调用一个都不执行。 不是「参数有问题的那些不执行」，是全部。

代码很短，判断也很短：

// agent-loop.ts:263-270
if (toolCalls.length > 0) {
	const executedToolBatch =
		message.stopReason === "length"
			? await failToolCallsFromTruncatedMessage(toolCalls, emit)
			: await executeToolCalls(currentContext, message, config, signal, emit);

但为什么是全部，值得展开三层推理，因为这是全文最容易写反的设计。

第一层：输出被 max_tokens 砍断了。 模型本来想发三个工具调用，第三个写到一半被截断。

第二层：截断出来的参数「看起来是合法的」。 这是关键。pi-ai 的工具参数在流式过程中走的是 parseStreamingJson()——四级降级解析（17 篇文章 讲过），它保证任何时刻都返回一个对象。于是 {"path": "src/ind 被降级解析器读成 {"path": "src/ind"}——语法合法，schema 校验通过，工具照单执行，但路径是错的。

那个「看起来合法但其实不完整」的判断，就是这个包对流式兜底解析器的已知代价做出的补偿。

第三层：为什么不做部分执行。 假设前两个工具调用确实完整、第三个被截断，只跳过第三个行不行？不行，因为你无法可靠地判断哪个完整。流式解析器的兜底路径对所有调用一视同仁，模型写到哪个算哪个，没有「写完了」的标记。唯一能被信任的信号是「这条消息没有因为长度而停」。有 length 这个信号，就意味着这一轮的所有工具调用都不可信。

源码注释把三层合在一段里：

Streamed tool-call arguments are finalized with a best-effort JSON salvage parser, so a truncated message can yield tool calls whose arguments parse and validate but are silently incomplete. None of them are safe to execute; report each as an error so the model can re-issue them.

失败时的处理值得看：它仍然完整地走完一遍事件序列——tool_execution_start → tool_execution_end → 一对 message_start/message_end。也就是说 UI 会看到「有 N 个工具被调用了但都失败了」，而不是「什么都没发生」。

给模型的错误文本也写得很具体：

Tool call "X" was not executed: the response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments.

它给的是一条可执行的指令，不是一个错误码。 模型看到这句话，知道该做什么。这和 pi-ai 里 argumentsError 带上原始文本是同一个思路：错误信息要能被模型用于自我纠正。

terminate 在这条路径上是 false。截断失败不构成终止理由——模型要能重来，一次终止就断了重来机会。

afterToolCall 的逐字段覆盖

finalizeExecutedToolCall()（agent-loop.ts:859-910）处理工具跑完之后的改写。逐字段覆盖，不做深合并：

structuredContent 那一行有额外逻辑，注释给了理由：

// Structured content not replaced along with the content may no longer match it.
const structuredContent =
	afterResult.structuredContent ?? (afterResult.content ? undefined : result.structuredContent);

你改了 content 却不改 structuredContent，结构化内容会被丢掉。 因为声明了 outputSchema 的工具，structuredContent 是给程序化消费者用的；内容都换了，它大概率不再对应。与其留一个可能撒谎的字段，不如删掉。这条约束在 10 扩展开发 里也出现过，是同一个判断。

afterToolCall 自己抛异常时，结果被换成错误结果、isError 变成 true。

runToolCall：给嵌套调用准备的同一条管道

runToolCall()（agent-loop.ts:811-819）是一个独立导出的函数，把单次调用跑完 prepareArguments → 校验 → beforeToolCall → 执行 → afterToolCall，但不发任何事件、不写任何消息。

它的存在是为了让「工具调用工具」走同一条路。注释写得很清楚：

Tools that call other tools use this so the hooks (for example permission checks) apply to those calls too.

这就是「ctx.executeTool() 绕不开你的权限扩展」这条承诺的物理基础。 codemode 写的 JS 调了一个工具，看起来像是脚本自己发起的，实际上它走的是 runToolCall()，beforeToolCall 照样跑。

runToolCall 还有一条更值得注意的契约：

Never rejects for tool failures: unknown tools, validation errors, blocked calls, and thrown errors come back as isError: true.

CLI 那边正是这么接的——agent-session.ts:749 把 beforeToolCall / afterToolCall 原样转发给 runToolCall，只多加一个 parentId。

18.4 状态与两个消息队列

九个字段的状态

AgentState（types.ts:382-421）九个成员：

systemPrompt 那行值得停一下。它的实现是个 getter：

// agent.ts:89-91
get systemPrompt() {
	return getCurrentSystemPrompt(messages);
},

它没有独立存储，messages 变了它就变。 想改提示，追加一条带 content 或 sections 的 system 消息。README 说得更狠：

agent.state.systemPrompt is read-only and replays the transcript.

这不是「顺手做成只读」，是把上一篇文章讲的 TranscriptContext 设计贯彻到底：提示住在对话里，不住在变量里。你不可能让提示和 transcript 脱节，因为它们是同一个东西的两种视图。

tools 和 messages 的 setter 都做 .slice()，注释也解释了：

tools and messages use accessor properties so implementations can copy assigned arrays before storing them.

注意 getter 返回的是内部数组本身，不是副本。 所以 agent.state.messages.push(msg) 是有效的（README 的示例里就是这么写的），而 agent.state.messages = [...] 是替换。

pendingToolCalls 的更新方式是每次新建一个 Set 而不是原地增删：

case "tool_execution_start": {
	const pendingToolCalls = new Set(this._state.pendingToolCalls);
	pendingToolCalls.add(event.toolCallId);
	this._state.pendingToolCalls = pendingToolCalls;
	break;
}

类型声明的是 ReadonlySet<string>，用换引用代替冻结，是保证「读到的那个 Set 不会在你读的时候被人改」。UI 组件订阅 state 时不会读到半改的状态。

两个队列，一份实现

PendingMessageQueue（agent.ts:143-174）三十行，管两个队列（steeringQueue / followUpQueue），区别只有构造时的 mode：

peek(): AgentMessage[] {
	if (this.mode === "all") return this.messages.slice();
	const first = this.messages[0];
	return first ? [first] : [];
}

drain(): AgentMessage[] {
	const drained = this.peek();
	this.messages = this.messages.slice(drained.length);
	return drained;
}

QueueMode 只有两个值——"all" 和 "one-at-a-time"，两个队列各自独立配置，默认都是 one-at-a-time。

这个默认值值得解释，因为它决定了默认交互下的手感：one-at-a-time 意味着用户连按三次 Enter 排三条纠正消息，模型会分三轮分别看到它们，而不是一次性全砸进上下文。后者在纠错场景里经常导致模型只处理最后一条。默认值偏保守是对的。

两个队列在循环里的轮询点完全不同，这是 runLoop() 里最容易读漏的部分：

第 ③ 处的注释解释了为什么需要它：

// Preparation can be long-running (for example, compaction). Pick up steering
// queued while it ran. Only poll again if the earlier poll returned nothing;
// otherwise one-at-a-time mode would deliver two messages in this turn.

压缩可能要跑十几秒。用户在这十几秒里敲了字，那些字必须被捡起来。 但又不能无条件再轮询一次——如果上一次已经取到了一条，one-at-a-time 模式下再取一次就会在这一轮里塞进两条，破坏模式语义。所以条件是「上次轮询返回了空才再轮询」。

这一处的逻辑密度是全包最高的：三个轮询点、一个互斥条件、一个「慢操作期间不丢消息」的目标。读源码时值得对照 03 交互式会话 的用户视角看一遍。

continue() 里那个 skipInitialSteeringPoll

Agent.continue()（agent.ts:384-411）有一个分支专门处理「transcript 最后一条是 assistant」的情况。

逻辑是这样的：正常情况下 continue() 要求最后一条是 user 或 toolResult。如果它是 assistant，说明这个 run 刚结束、尾巴还挂着一条 assistant 消息。这时不能直接继续，只能用队列里的消息当新的输入。代码先试 steering，再试 follow-up，两个都没有才抛错。

然后是那个参数：

const queuedSteering = this.steeringQueue.drain();
if (queuedSteering.length > 0) {
	await this.runPromptMessages(queuedSteering, { skipInitialSteeringPoll: true });
	return;
}

为什么需要 skipInitialSteeringPoll？ 因为 runAgentLoop() 的第一件事就是 runLoop() 入口那次 steering 轮询。continue() 已经把队列 drain 过一次了，消息已经作为 prompts 传进去了。不跳过的话，入口那次轮询会去 drain 一个已经清空的队列——这一次是安全的。真正会出问题的是 continue() drain 完队列、还没调 runAgentLoop 的那个窗口里，用户又敲了一条——那一条会被入口轮询取走，跟同一个 run 里的其它消息混在一起，顺序就乱了。

skipInitialSteeringPoll 在 createLoopConfig() 里被实现成一个自我消费的开关：

getSteeringMessages: async () => {
	if (skipInitialSteeringPoll) {
		skipInitialSteeringPoll = false;   // 用一次就失效
		return [];
	}
	return this.steeringQueue.drain();
},

一次性。 之后的轮询恢复正常。这类「只在第一次生效」的标志是异步控制流里最容易写错的地方——写成永久生效会丢消息，写成永不到效会重复注入。

单run 独占与中止

Agent 有一条硬的独占规则：同一时刻只能有一个 run。 三处检查，三句不同的话：

prompt() 那句错信息本身是说明书：

Agent is already processing a prompt. Use steer() or followUp() to queue messages, or wait for completion.

独占不是限制，是引导。 你不能在 Agent 干活时改它的输入，但你可以往两个队列里投消息。Agent 永远不会「接受输入然后排队」——它要么现在处理，要么明确告诉你去排队。

中止的机制是一个 AbortController 存在 activeRun 里：

type ActiveRun = { promise: Promise<void>; resolve: () => void; abortController: AbortController };

abort() 调它的 abort()，signal getter 读它的 signal。这个 signal 被传给每一个 emit：

const signal = this.activeRun?.abortController.signal;
if (!signal) throw new Error("Agent listener invoked outside active run");
for (const listener of this.listeners) {
	await listener(event, signal);
}

所以你的监听器能拿到本次 run 的中止信号，不需要自己去猜一个。

abort() 本身不 resolve waitForIdle()——它只是 abort。循环会走到自己的收尾路径（工具执行里检查 signal.aborted 产出「Operation aborted」错误结果，模型侧产出 stopReason: "aborted"），最终 finishRun() 才 resolve promise。中止是请求，不是强制中断。

agent_end 不等于空闲

这是最容易误解的一条，README 和源码注释都在强调。

/**
 * Subscribe to agent lifecycle events.
 *
 * Listener promises are awaited in subscription order and are **included in
 * the current run's settlement**. Listeners also receive the active abort
 * signal for the current run.
 *
 * `agent_end` is the final emitted event for a run, but **the agent does not
 * become idle until all awaited listeners for that event have settled.**
 */

三个时间点要分清：

Agent.state.isStreaming 的注释也是同一句话：

This remains true until awaited agent_end listeners settle.

为什么重要：如果你在 agent_end 里做异步的会话落盘，那么「会话已保存」这件事发生在 waitForIdle() 返回之前。反过来，如果你以为 agent_end 一到就空闲、直接开始下一次 run，落盘还没完成。

reset() 保留什么

reset()（agent.ts:355-368）不是清空一切：

const baseline = getCurrentSystemMessage(this._state.messages);
this._state.messages = baseline ? [baseline] : [];

它保留重放出来的基线 system 消息。 系统提示和工具声明活下来了，对话历史没了。

这个取舍是对的：一个没有基线的 Agent 会连自己的身份都没有，而把基线一起删掉会让「清空对话」变成「重建成一个空白的新对话」——后者的构造更麻烦，用户要的往往只是前者。reset() 还会跑一次运行中检查，在处理中调用直接抛。

18.5 传输缝

二十行的 stream-fn.ts

整个文件：

// packages/agent/src/stream-fn.ts（全文 20 行）
import type { StreamFn } from "./types.ts";

let defaultStreamFn: StreamFn | undefined;

/**
 * Configure the fallback used by Agent and low-level loops when callers omit streamFn.
 *
 * Hosts that provide a default model runtime can install its stream function here
 * without making pi-agent-core depend on a provider catalog or compatibility layer.
 */
export function setDefaultStreamFn(streamFn: StreamFn | undefined): void {
	defaultStreamFn = streamFn;
}

export function getDefaultStreamFn(): StreamFn {
	if (!defaultStreamFn) {
		throw new Error("No default stream function configured. Pass streamFn explicitly or call setDefaultStreamFn().");
	}
	return defaultStreamFn;
}

一个模块级变量，两个函数。 这就是全部。

StreamFn 的类型本身（types.ts:33-37）：

export type StreamFn = (
	model: Model<Api>,
	context: TranscriptContext,
	options?: SimpleStreamOptions,
) => AssistantMessageEventStream | Promise<AssistantMessageEventStream>;

注意 context 的类型是 TranscriptContext——17 篇文章 讲过的那个品牌类型。这个签名让 pi-agent-core 和 pi-ai 的形态保证对齐了：循环保证传进来的 context 是归一化过的（streamAssistantResponse() 里那一行 normalizeContext({ messages: llmMessages })），所以 StreamFn 的实现方可以假设系统提示在 context.messages 里，而不需要处理 context.systemPrompt 或 context.tools。

StreamFn 的契约注释里有两条硬约束：

• Must not throw or return a rejected promise for request/model/runtime failures.
• Failures must be encoded in the returned stream via protocol events and a final AssistantMessage with stopReason "error" or "aborted" and errorMessage.

「不许抛」是给宿主实现的硬性要求。 循环不会 try/catch 包住流调用，抛出来的异常会一路冒到 runWithLifecycle()，那就会丢掉正常的事件序列。错误必须编码成流里的 error 事件。

为什么是这个形状

源码注释里有一句设计意图，值得原样引用：

Hosts that provide a default model runtime can install its stream function here without making pi-agent-core depend on a provider catalog or compatibility layer.

翻译一下这个约束在架构上换来了什么：

需要说清楚一件事：这里没有「transport 抽象类」。 StreamFn 是一个函数类型，缝就是一个模块级变量加两个函数。你既不需要继承什么，也不需要实现接口。（pi-ai 里确实有个叫 Transport 的东西，但那是一个字符串偏好——"sse" / "websocket" / "auto——被 StreamFn 原样转发给下游，不是类，也不是这层缝。packages/agent/package.json 的 description 里写着 "transport abstraction"，指的是这个转发能力，不是继承体系。）

默认值在哪里被装上

Pi CLI 的注入点是 src/core/sdk.ts 的模块顶层：

// packages/coding-agent/src/core/sdk.ts:41-46
// Preserve the pre-0.81 fallback for extensions that construct Agent instances
// or invoke low-level agent loops without supplying streamFn. Agent core remains
// provider-agnostic and does not import pi-ai/compat itself.
setDefaultStreamFn(streamSimple);

streamSimple 来自 @earendil-works/pi-ai/compat——17 篇文章 末尾提到的那个「已标了临时」的过渡层。注释写得很清楚这是为了兼容旧扩展而保留的。

这里有一个值得注意的不对称：Agent 类和裸循环对同一个缺省值的处理时机不同。

// agent.ts:236 —— 构造器里立即求值
this.streamFunction = runtimeOptions.streamFn ?? getDefaultStreamFn();

// agent-loop.ts:124 —— 循环开始时才求值
await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn());

后果是：new Agent({}) 在构造的那一刻就抛 No default stream function configured；而直接调 runAgentLoop(..., undefined) 要等到进入 runLoop 才抛。

如果你在写一个不传 streamFn 的库（或者一个测试），构造 Agent 会在你还没跑到任何逻辑时就炸。这个行为不算错（早失败好过晚失败），但值得知道。

proxy.ts：缝的一个现成实现

streamProxy()（proxy.ts:107-252）是这道缝最完整的示范：同样的 StreamFn 签名，换成走自己的后端。

它做的事情里，有两处是任何自建传输都要处理的，值得单独拎出来：

一、服务端剥掉 partial 字段。 ProxyAssistantMessageEvent（proxy.ts:23-46）里每个事件都没有 partial，只有 delta。理由写在文件头：

Proxy event types - server sends these with partial field stripped to reduce bandwidth. We reconstruct the partial message client-side.

processProxyEvent()（proxy.ts:257-393）逐个事件重建那个 AssistantMessage，text_delta 就 content.text += delta。这意味着客户端要完整实现一遍内容块的累积逻辑——省下的带宽要自己用代码换回来。

二、干净的 EOF 不等于成功。 这是 proxy.ts:219-230 里我觉得最值得抄的一段：

if (!sawTerminalEvent) {
	// A clean EOF without a done/error event means the server dropped the
	// response mid-stream. Surface it as an error instead of leaving
	// consumers waiting on a result that never arrives.
	partial.stopReason = "error";
	partial.errorMessage = "Connection closed by proxy server before the response completed";
	stream.push({ type: "error", reason: "error", error: partial });
}

HTTP 流在连接断开时可能返回 200。如果不检查有没有收到 done / error 事件，你的消费者会永远等一个永远不会来的 result()。「流结束了」和「流正常结束了」是两件事。

代理请求体是 { model, context, options } 三个字段，和 pi-messages 协议的形状一样（见 17 篇文章的事实来源表）。

18.6 宿主要自己提供什么

一张否定清单

这一节是本篇最实用的部分。这个包不做的东西，比它做的更能说明它是什么。

这七条不是「功能待补」，是这个包的定位。 一个 agent 循环能做的事情本来就是「拿到消息、请求模型、执行工具、把结果塞回去」。持久化、权限、压缩、子 agent 都不是循环的职责——它们是围绕循环的策略。

关键证据：replay 字段

有一个细节能把这套分工讲透。AgentTool 上有一个字段（types.ts:489-490）：

/** Recovery policy for an effect whose durable intent exists but whose outcome is unknown. */
replay?: "never" | "safe";

语义是「这个工具有副作用，执行意图已落盘但结果未知时，能不能安全重放」——这是崩溃恢复需要的元数据。

它在 packages/agent/src 和 packages/coding-agent/src 里一次都没被读过。全仓库唯一的消费点是：

// packages/durable/src/harness/tool.ts:170、:180、:290
const intent = { phase: "execute", arguments: final, replay: tool.replay ?? "unsafe" } as const;
...
if (replay === "safe" && tool?.replay === "safe") { ... }
const resumes = tool.replay === "safe";

声明在 agent 循环里，被 Durable 树消费。 而 pi-durable 不在 CLI 的依赖路径上（01 项目概览 那张架构图：Durable 是另一棵树）。

这个字段是整套分工的一个完美切片：类型契约住在最小的那个包里，实现住在需要它的那个宿主里。 顺带说明另一件事——packages/agent/src/harness/ 不存在，harness 在 packages/durable/src/harness/。

宿主的七种需求，各走哪个钩子

反过来看，宿主需要自己提供的东西，每一样都有对应的接缝：

Pi CLI 做的全部工作就是把这些钩子接到扩展系统上。 pi-agent-core 不需要知道扩展的存在，扩展系统也不需要知道循环的内部结构。这条界线是两个包能各自保持小的原因。

压缩是最能说明问题的那一格

值得把压缩那一格展开，因为它同时说明了「这个包不做什么」和「宿主能做什么」。

pi-agent-core 里没有任何一处读 contextWindow。grep -rn contextWindow packages/agent/src/ 只有一行命中：agent.ts:66，那个占位用的 DEFAULT_MODEL 对象里的一个 contextWindow: 0。除此之外，它对 Model 类型的全部使用都集中在把它当不透明对象转发——config.model.provider（取 API key 用）、传给 StreamFn、传给 prepareRequest 上下文。

压缩的全过程在宿主侧：

// agent-session.ts:905-907
this.agent.prepareNextTurnWithContext = async (turn, signal) => {
	const context = await this._compactBeforeNextAssistantResponse(turn.context);
	const previousSnapshot = await previousPrepareNextTurnWithContext?.({ ...turn, context }, signal);

而 _compactBeforeNextAssistantResponse() 自己还要先判断阈值：

// agent-session.ts:777-786
const projection = this.sessionManager.buildSessionProjection();
if (!model || isVirtualModel(model) || !this._exceedsCompactionThreshold(model, projection)) {
	return { ...context, messages: projection.messages };
}
await this._runAutoCompaction("threshold", false);

注意它连当前模型是不是虚拟模型都要先排除。 这类判断属于「什么时候压缩」的政策，不是「怎么跑循环」的机制。

这也解释了为什么压缩能被做成 prepareNextTurn 而不是循环的一部分：压缩是慢操作，而循环需要一个能挂长操作、并在长操作期间不丢消息的位置。 18.4 讲的那个「prepareNextTurn 跑完之后补一次 steering 轮询」的逻辑，正是为压缩这种场景准备的。

具体阈值、保留多少 token、怎么写摘要，在 15 上下文与压缩。

什么时候用哪一层

18.7 事实来源

本文所有断言都可以在仓库里复算：

完整事实表（含每条断言的推导命令）见专栏仓库的 research/pi-agent/FACTS.md。

上一篇 架构（一）pi-ai 讲的是循环下游那个流是怎么产生的；下一层是终端 UI，见 架构（三）pi-tui。宿主怎么接上 UI、怎么持久化，见 06 工具与信任边界。