34 KiB
逐行精读：agent-loop.ts —— 核心循环（Pi 的心脏）
源文件：
pi-main/packages/agent/src/agent-loop.ts（792 行）前置阅读：01-types-and-protocol.md（types.ts 逐行精读）。 循环只是在操作那些类型——不懂类型就读不懂循环。
文件结构总览
agent-loop.ts (792 行)
├── 公共入口（L1-151）
│ ├── agentLoop() —— 带 prompt 启动
│ ├── agentLoopContinue() —— 不加 prompt 续跑（用于重试）
│ └── createAgentStream() —— EventStream 工厂
├── 核心循环 runLoop()（L155-275）⭐⭐⭐
│ ├── 外层 while：follow-up 循环
│ └── 内层 while：steering + 工具循环
├── streamAssistantResponse()（L281-374）⭐⭐ 边界转换
├── 工具执行（L383-755）
│ ├── failToolCallsFromTruncatedMessage() —— 截断时全部失败
│ ├── executeToolCalls() —— 模式选择
│ ├── executeToolCallsSequential() —— 顺序执行
│ ├── executeToolCallsParallel() —— 并发执行
│ ├── prepareToolCall() —— 验证 + before hook
│ ├── executePreparedToolCall() —— 执行 + 流式更新
│ └── finalizeExecutedToolCall() —— after hook
└── 辅助函数（L757-792）
├── createErrorToolResult()
├── emitToolExecutionEnd()
├── createToolResultMessage()
└── emitToolResultMessage()
一、公共入口（L1-151）
agentLoop —— 带 prompt 启动（L34-118）
// L34-101（签名）
export function agentLoop(
prompts: AgentMessage[],
context: AgentContext,
config: AgentLoopConfig,
signal?: AbortSignal,
streamFn?: StreamFn,
): EventStream<AgentEvent, AgentMessage[]> {
const stream = createAgentStream();
void runAgentLoop(prompts, context, config, async (event) => stream.push(event), signal, streamFn)
.then((messages) => stream.end(messages));
return stream;
}
关键模式：立即返回 EventStream，后台异步推事件。void runAgentLoop(...).then(...) 启动一个不等待的 promise，事件通过 stream.push 推给消费者。这让调用方能边迭代事件边让循环跑——生产者/消费者解耦。
// L103-117（runAgentLoop 主体）
async function runAgentLoop(...): Promise<AgentMessage[]> {
const newMessages: AgentMessage[] = [...prompts]; // 本次 run 产生的所有新消息
const currentContext: AgentContext = {
...context,
messages: [...context.messages, ...prompts], // 上下文 = 原有 + 新 prompt
};
await emit({ type: "agent_start" }); // L109 生命周期开始
await emit({ type: "turn_start" }); // L110
for (const prompt of prompts) { // L111-114 为每个 prompt 发 message 事件
await emit({ type: "message_start", message: prompt });
await emit({ type: "message_end", message: prompt });
}
await runLoop(currentContext, newMessages, config, signal, emit, streamFn); // L116 ← 核心循环
return newMessages; // L117 返回本次产生的新消息
}
newMessages 是只追加的列表——本次 run 产生的所有新消息（prompt + assistant + toolResult + steering + follow-up）。它和 currentContext.messages（完整 transcript）是两个不同的东西。
agentLoopContinue —— 续跑（L120-143）
// L127-133 两个前置校验
if (context.messages.length === 0) throw new Error("Cannot continue: no messages in context");
if (context.messages[last].role === "assistant") throw new Error("Cannot continue from assistant");
为什么最后一条不能是 assistant？ 因为继续循环意味着要调 LLM，而 LLM 调用要求最后一条是 user 或 toolResult（Anthropic/OpenAI 的协议约束）。最后一条是 assistant 时"继续"无意义——你应该用 prompt 加新消息。
createAgentStream（L145-150）
function createAgentStream(): EventStream<AgentEvent, AgentMessage[]> {
return new EventStream<AgentEvent, AgentMessage[]>(
(event) => event.type === "agent_end", // 完成检测器
(event) => event.type === "agent_end" ? event.messages : [], // 结果提取器
);
}
EventStream<T, R> 的两个回调：① 判断哪个事件是"完成"事件；② 从完成事件提取最终结果 R。这里完成事件是 agent_end，结果是它的 messages。
二、核心循环 runLoop（L155-275）⭐⭐⭐
这是整个文件最重要的一段。双层循环 + 四个停止条件。
函数签名与初始状态（L155-167）
async function runLoop(
initialContext: AgentContext,
newMessages: AgentMessage[],
initialConfig: AgentLoopConfig,
signal: AbortSignal | undefined,
emit: AgentEventSink,
streamFn?: StreamFn,
): Promise<void> {
let currentContext = initialContext; // 可变：prepareNextTurn 会替换
let config = initialConfig; // 可变：prepareNextTurn 会替换 model/reasoning
let firstTurn = true;
// L167 启动时就检查 steering（用户可能在等待时已输入）
let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || [];
currentContext 和 config 都是 let——prepareNextTurn 可以整体替换它们。这是"每轮重建上下文"的实现基础。
L167 的 getSteeringMessages 在循环开始就调用——用户可能在 agentLoop 返回 EventStream 后、循环真正开始前就输入了 steering 消息。这是为了不丢消息。
双层循环骨架（L170-272）
// L170 外层 while：follow-up 循环
while (true) {
let hasMoreToolCalls = true;
// L174 内层 while：steering + 工具循环
while (hasMoreToolCalls || pendingMessages.length > 0) {
为什么是双层？
- 内层：处理"一个连贯的工作流"——有工具调用就继续，有 steering 就注入。
- 外层：当内层要退出（没工具调用也没 steering 了），再检查 follow-up——"任务做完了，还有别的吗？"有就重启内层。
这种分层让 steering（紧急插话）和 follow-up（后续任务）的语义清晰分离。
内层循环体（L175-260）
① Turn 标记与消息注入（L175-190）
if (!firstTurn) {
await emit({ type: "turn_start" }); // L176 非首轮才发 turn_start
} else {
firstTurn = false; // 首轮的 turn_start 已在 runAgentLoop 发过
}
// L182-190 注入 pendingMessages（steering 或 follow-up）
if (pendingMessages.length > 0) {
for (const message of pendingMessages) {
await emit({ type: "message_start", message });
await emit({ type: "message_end", message });
currentContext.messages.push(message); // 加入 transcript
newMessages.push(message); // 加入本次 run 的新消息
}
pendingMessages = []; // 清空
}
steering/follow-up 消息在调 LLM 之前注入——模型会在下一轮看到它们。
② 流式接收 assistant 响应（L193-200）
const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFn);
newMessages.push(message);
// L196-200 错误/中止：直接结束
if (message.stopReason === "error" || message.stopReason === "aborted") {
await emit({ type: "turn_end", message, toolResults: [] });
await emit({ type: "agent_end", messages: newMessages });
return;
}
这是停止条件①：provider 返回 error/aborted。注意仍然发 turn_end 和 agent_end——保证事件序列完整。
③ 提取并执行工具调用（L202-222）
const toolCalls = message.content.filter((c) => c.type === "toolCall");
const toolResults: ToolResultMessage[] = [];
hasMoreToolCalls = false; // L206 默认假设不继续
if (toolCalls.length > 0) {
// L208-214 关键分支：截断时全部失败
const executedToolBatch =
message.stopReason === "length"
? await failToolCallsFromTruncatedMessage(toolCalls, emit)
: await executeToolCalls(currentContext, message, config, signal, emit);
toolResults.push(...executedToolBatch.messages);
hasMoreToolCalls = !executedToolBatch.terminate; // L216 terminate 决定是否继续
for (const result of toolResults) {
currentContext.messages.push(result); // 工具结果入 transcript
newMessages.push(result);
}
}
length 停止原因的特殊处理（L208-214）——这是停止条件的一个微妙点：
为什么截断时要全部失败？因为流式 tool call 的参数是用增量 JSON 解析（parseStreamingJson）拼出来的。如果输出被 token 限制截断，拼出来的 JSON 可能恰好能解析、能通过 schema 验证，但内容是残缺的——执行它会有不可预测的后果。所以宁可全部报错让模型重新发。
hasMoreToolCalls = !executedToolBatch.terminate（L216）——只有批内所有结果都 terminate: true 才停止（见 shouldTerminateToolBatch，L584-586）。
④ prepareNextTurn + shouldStopAfterTurn（L224-259）
await emit({ type: "turn_end", message, toolResults });
// L226-245 prepareNextTurn：每轮后重建上下文
const nextTurnContext = { message, toolResults, context: currentContext, newMessages };
const nextTurnSnapshot = await config.prepareNextTurn?.(nextTurnContext);
if (nextTurnSnapshot) {
currentContext = nextTurnSnapshot.context ?? currentContext; // 替换上下文
config = {
...config,
model: nextTurnSnapshot.model ?? config.model, // 替换模型
reasoning: nextTurnSnapshot.thinkingLevel === undefined // 替换思考级别
? config.reasoning
: nextTurnSnapshot.thinkingLevel === "off" ? undefined : nextTurnSnapshot.thinkingLevel,
};
}
// L247-257 停止条件②：shouldStopAfterTurn
if (await config.shouldStopAfterTurn?.({ message, toolResults, context: currentContext, newMessages })) {
await emit({ type: "agent_end", messages: newMessages });
return;
}
pendingMessages = (await config.getSteeringMessages?.()) || []; // L259 再 poll steering
} // 内层 while 结束
prepareNextTurn 是 harness 层动态上下文的钩子（L226-245）。harness 在这里：从 session 重载消息、重算系统提示（可含 cwd/git status）、重建工具列表。模型切换、上下文压缩、工具动态加载都通过这个生效。
L247-257 是停止条件②：shouldStopAfterTurn 返回 true。注意它在 steering poll 之前——如果要求停止，就不 poll steering/follow-up 了（L255 直接 return）。用途：上下文快满了提前停。
L259 再次 poll steering——内层循环每次 turn 结束都检查。这就是 steering 的"紧急插话"语义。
外层循环：follow-up（L262-271）
// 内层退出，agent 本来要停了
const followUpMessages = (await config.getFollowUpMessages?.()) || []; // L263
if (followUpMessages.length > 0) {
pendingMessages = followUpMessages; // L266 设为 pending，内层会注入
continue; // L267 重启内层
}
break; // L271 真的没东西了，退出
}
await emit({ type: "agent_end", messages: newMessages }); // L274 最终结束
}
follow-up 只在内层"本来要退出"时才检查——这是它和 steering 的本质区别。steering 是打断（内层每次都 poll），follow-up 是续接（内层要停才问）。
停止条件总结
| # | 条件 | 触发位置 |
|---|---|---|
| ① | provider 返回 error/aborted |
L196 |
| ② | shouldStopAfterTurn 返回 true |
L247 |
| ③ | 没工具调用、没 steering、没 follow-up（自然结束） | 内层退出 + 外层 break |
| ④ | 工具批全部 terminate: true（通过 hasMoreToolCalls=false） |
L216 + 内层退出 |
截断（length）不是停止条件——它只是让工具全部失败，循环继续（模型会看到错误，重发工具调用）。
三、streamAssistantResponse —— 边界转换（L281-374）⭐⭐
这是 AgentMessage↔Message 唯一的翻译点，也是循环和 provider 的接口。
三步转换（L288-314）
async function streamAssistantResponse(...): Promise<AssistantMessage> {
// 第①步：可选的上下文变换（AgentMessage → AgentMessage）
let messages = context.messages;
if (config.transformContext) {
messages = await config.transformContext(messages, signal); // 剪枝/注入
}
// 第②步：转换成 LLM 消息（AgentMessage → Message[]）
const llmMessages = await config.convertToLlm(messages);
// 第③步：构建 LLM 上下文
const llmContext: Context = {
systemPrompt: context.systemPrompt,
messages: llmMessages,
tools: context.tools,
};
const streamFunction = streamFn || streamSimple; // L304 可注入自定义 streamFn
// 动态解析 API key（处理过期 token）
const resolvedApiKey =
(config.getApiKey ? await config.getApiKey(config.model.provider) : undefined) || config.apiKey;
// 调 provider
const response = await streamFunction(config.model, llmContext, { ...config, apiKey: resolvedApiKey, signal });
为什么 transformContext 在 convertToLlm 之前？ 因为 transformContext 操作 AgentMessage[]（应用级，可剪枝自定义消息、注入外部上下文），而 convertToLlm 是格式转换（AgentMessage → Message）。顺序是"先处理应用语义，再转换协议格式"。
流式累加（L316-373）—— 流式消息是"活"的
let partialMessage: AssistantMessage | null = null;
let addedPartial = false; // 标记是否已 push 到 context.messages
for await (const event of response) {
switch (event.type) {
case "start": // L321-326
partialMessage = event.partial;
context.messages.push(partialMessage); // ★ 立即入栈
addedPartial = true;
await emit({ type: "message_start", message: { ...partialMessage } });
break;
case "text_start": case "text_delta": case "text_end":
case "thinking_start": case "thinking_delta": case "thinking_end":
case "toolcall_start": case "toolcall_delta": case "toolcall_end":
// L337-346 所有 delta 事件统一处理
if (partialMessage) {
partialMessage = event.partial;
context.messages[context.messages.length - 1] = partialMessage; // ★ 原地替换最后一个
await emit({ type: "message_update", assistantMessageEvent: event, message: { ...partialMessage } });
}
break;
case "done": case "error": { // L348-361
const finalMessage = await response.result();
if (addedPartial) {
context.messages[last] = finalMessage; // 替换为最终版
} else {
context.messages.push(finalMessage); // 没 push 过就 push
}
if (!addedPartial) {
await emit({ type: "message_start", message: { ...finalMessage } }); // 补发 message_start
}
await emit({ type: "message_end", message: finalMessage });
return finalMessage;
}
}
}
// L365-373 流自然结束（没 done/error 事件）的兜底
const finalMessage = await response.result();
if (addedPartial) context.messages[last] = finalMessage;
else { context.messages.push(finalMessage); await emit({ type: "message_start", message: { ...finalMessage } }); }
await emit({ type: "message_end", message: finalMessage });
return finalMessage;
}
核心设计：流式消息在流式过程中就躺在 transcript 末尾。
start事件 → push 空的 partial 消息- 每个 delta 事件 → 用新 partial 原地替换最后一个元素
done/error→ 用最终消息替换
这让 UI 能实时看到流式输出（通过 message_update 事件），而 transcript 始终保持"当前最新状态"。
addedPartial 标志（L317）处理一个边界情况：如果 provider 没发 start 事件直接发 done（某些 provider 的行为），那 transcript 里还没有 partial，需要补发 message_start。
L365-373 的兜底：如果流自然耗尽（for-await 结束）但没收到 done/error，用 response.result() 取最终消息——保证总有最终消息入栈。这是健壮性设计。
为什么 emit 时用 { ...partialMessage } 浅拷贝？（L325, L343）因为 partial 后续会被替换，emit 出去的应该是那一刻的快照，避免消费者持有的是会被原地修改的引用。
四、工具执行（L383-755）
截断失败（L383-408）
async function failToolCallsFromTruncatedMessage(toolCalls, emit): Promise<ExecutedToolCallBatch> {
const messages: ToolResultMessage[] = [];
for (const toolCall of toolCalls) {
await emit({ type: "tool_execution_start", toolCallId: toolCall.id, toolName: toolCall.name, args: toolCall.arguments });
const finalized: FinalizedToolCallOutcome = {
toolCall,
result: createErrorToolResult(`...arguments may be truncated. Re-issue the tool call...`),
isError: true,
};
await emitToolExecutionEnd(finalized, emit);
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit);
messages.push(toolResultMessage);
}
return { messages, terminate: false }; // 不终止——让模型看到错误重发
}
模拟完整的工具执行事件序列（start → end → message），但结果永远是错误。terminate: false 让循环继续——模型会看到"参数可能截断"的错误，重新发工具调用。
模式选择（L413-428）
async function executeToolCalls(...): Promise<ExecutedToolCallBatch> {
const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall");
// 只要有一个工具声明了 sequential，整个批次就退化为 sequential
const hasSequentialToolCall = toolCalls.some((tc) =>
currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential");
if (config.toolExecution === "sequential" || hasSequentialToolCall) {
return executeToolCallsSequential(...);
}
return executeToolCallsParallel(...);
}
保守策略：批内有一个 sequential 工具就全部顺序执行。因为 sequential 通常意味着有副作用（如写文件），并发可能冲突。
顺序执行（L435-489）
async function executeToolCallsSequential(...): Promise<ExecutedToolCallBatch> {
const finalizedCalls: FinalizedToolCallOutcome[] = [];
const messages: ToolResultMessage[] = [];
for (const toolCall of toolCalls) { // 一个一个来
await emit({ type: "tool_execution_start", ... });
const preparation = await prepareToolCall(...); // 验证 + before hook
let finalized: FinalizedToolCallOutcome;
if (preparation.kind === "immediate") { // 验证失败/blocked/未找到 → 直接出结果
finalized = { toolCall, result: preparation.result, isError: preparation.isError };
} else {
const executed = await executePreparedToolCall(...); // 执行
finalized = await finalizeExecutedToolCall(...); // after hook
}
await emitToolExecutionEnd(finalized, emit); // 立即发 end
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit); // 立即发 toolResult message
finalizedCalls.push(finalized);
messages.push(toolResultMessage);
if (signal?.aborted) break; // 中止则跳出
}
return { messages, terminate: shouldTerminateToolBatch(finalizedCalls) };
}
顺序模式的特点：start→execute→end→message 一个工具完整走完才下一个。中止会 break，已完成的工具结果保留。
并发执行（L491-556）—— 精妙之处
async function executeToolCallsParallel(...): Promise<ExecutedToolCallBatch> {
const finalizedCalls: FinalizedToolCallEntry[] = []; // 注意类型是 Entry（可能是函数）
// 第一阶段：顺序 prepare（含 before hook）
for (const toolCall of toolCalls) {
await emit({ type: "tool_execution_start", ... });
const preparation = await prepareToolCall(...);
if (preparation.kind === "immediate") { // immediate 直接 finalize + 发 end
const finalized = { toolCall, result: preparation.result, isError: preparation.isError };
await emitToolExecutionEnd(finalized, emit);
finalizedCalls.push(finalized); // 直接存结果
if (signal?.aborted) break;
continue;
}
// prepared → 存一个 thunk（延迟执行）
finalizedCalls.push(async () => {
const executed = await executePreparedToolCall(preparation, signal, emit);
const finalized = await finalizeExecutedToolCall(...);
await emitToolExecutionEnd(finalized, emit); // end 按完成顺序发
return finalized;
});
if (signal?.aborted) break;
}
// 第二阶段：并发执行所有 thunk
const orderedFinalizedCalls = await Promise.all(
finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
);
// 第三阶段：toolResult message 按 assistant 源顺序发（不是完成顺序！）
const messages: ToolResultMessage[] = [];
for (const finalized of orderedFinalizedCalls) {
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit);
messages.push(toolResultMessage);
}
return { messages, terminate: shouldTerminateToolBatch(orderedFinalizedCalls) };
}
并发模式的三个阶段：
- 顺序 prepare：所有工具的
prepareToolCall（含 before hook）顺序跑。为什么顺序？因为 before hook 可能 block，且 prepare 可能修改共享状态。 - 并发 execute：prepared 的工具存成 thunk，
Promise.all并发执行。tool_execution_end按完成顺序发（每个 thunk 完成时立即发）。 - 顺序 emit message：
toolResult的message_start/end按源顺序发（orderedFinalizedCalls保持原顺序）。
FinalizedToolCallEntry 是联合类型（L582）：
type FinalizedToolCallEntry = FinalizedToolCallOutcome | (() => Promise<FinalizedToolCallOutcome>);
immediate 的存结果对象，prepared 的存 thunk。第二阶段用 typeof entry === "function" 区分。
为什么 end 按完成顺序、message 按源顺序？ 因为 tool_execution_end 是给 UI 看的（哪个工具先做完先显示），而 toolResult message 是给下一轮 LLM看的（要和 assistant 的 toolCall 顺序对应，否则模型会困惑）。
prepareToolCall —— 验证 + before hook（L602-666）
async function prepareToolCall(...): Promise<PreparedToolCall | ImmediateToolCallOutcome> {
const tool = currentContext.tools?.find((t) => t.name === toolCall.name);
if (!tool) { // L610 工具未找到
return { kind: "immediate", result: createErrorToolResult(`Tool ${toolCall.name} not found`), isError: true };
}
try {
const preparedToolCall = prepareToolCallArguments(tool, toolCall); // L619 prepareArguments shim
const validatedArgs = validateToolArguments(tool, preparedToolCall); // L620 schema 验证
if (config.beforeToolCall) { // L621 before hook
const beforeResult = await config.beforeToolCall(
{ assistantMessage, toolCall, args: validatedArgs, context: currentContext }, signal);
if (signal?.aborted) return { kind: "immediate", result: createErrorToolResult("Operation aborted"), isError: true };
if (beforeResult?.block) { // L638 block → 错误结果
return { kind: "immediate", result: createErrorToolResult(beforeResult.reason || "Tool execution was blocked"), isError: true };
}
}
if (signal?.aborted) return { kind: "immediate", result: createErrorToolResult("Operation aborted"), isError: true };
return { kind: "prepared", toolCall, tool, args: validatedArgs }; // L653 通过 → 执行
} catch (error) { // L659 验证抛错 → 错误结果
return { kind: "immediate", result: createErrorToolResult(error.message), isError: true };
}
}
三道关卡：① 工具存在性 ② prepareArguments（LLM 输出规整）+ validateToolArguments（schema 验证）③ beforeToolCall hook（权限/安全）。
返回 immediate 的情况：工具未找到、参数验证失败、before hook block、已中止。这些都跳过执行，直接产生错误结果——错误被编码成结果，而非抛出。
executePreparedToolCall —— 执行 + 流式更新（L668-709）
async function executePreparedToolCall(prepared, signal, emit): Promise<ExecutedToolCallOutcome> {
const updateEvents: Promise<void>[] = [];
let acceptingUpdates = true; // 更新开关
try {
const result = await prepared.tool.execute(
prepared.toolCall.id,
prepared.args as never,
signal,
(partialResult) => { // onUpdate 回调
if (!acceptingUpdates) return; // settle 后的调用忽略
updateEvents.push(Promise.resolve(emit({ type: "tool_execution_update", ... partialResult })));
},
);
acceptingUpdates = false; // 执行完成，关闭更新
await Promise.all(updateEvents); // 等待所有 update 事件 flush
return { result, isError: false };
} catch (error) {
acceptingUpdates = false;
await Promise.all(updateEvents); // 错误时也 flush 更新
return { result: createErrorToolResult(error.message), isError: true };
} finally {
acceptingUpdates = false; // 确保 finally 也关闭
}
}
onUpdate 回调的作用域：只在 execute() promise 未 settle 时有效。acceptingUpdates 三处置 false（成功/失败/finally），保证 settle 后的调用被忽略。
updateEvents 收集机制：update 回调里不直接 await emit（回调是同步的），而是把 emit 的 promise 推入数组，execute 完成后统一 Promise.all flush。这避免了回调内的异步竞态。
错误处理（L699-705）：工具 execute 抛错 → createErrorToolResult → isError: true。错误编码成结果，绝不抛出。
finalizeExecutedToolCall —— after hook（L711-755）
async function finalizeExecutedToolCall(...): Promise<FinalizedToolCallOutcome> {
let result = executed.result;
let isError = executed.isError;
if (config.afterToolCall) {
try {
const afterResult = await config.afterToolCall(
{ assistantMessage, toolCall: prepared.toolCall, args: prepared.args, result, isError, context: currentContext },
signal);
if (afterResult) {
// 字段级覆盖，无深度合并
result = {
...result,
content: afterResult.content ?? result.content, // 给了就替换
details: afterResult.details ?? result.details,
terminate: afterResult.terminate ?? result.terminate,
};
isError = afterResult.isError ?? isError;
}
} catch (error) { // after hook 抛错 → 错误结果
result = createErrorToolResult(error.message);
isError = true;
}
}
return { toolCall: prepared.toolCall, result, isError };
}
after hook 的字段级覆盖（L736-742）：content/details/terminate 用 ??（给了就替换，没给保持原值）。isError 同理。无深度合并——这是有意的，避免意外的部分覆盖导致状态不一致。
用途：脱敏（改写 content）、强制标记错误（isError: true）、审计日志（读取 details）。
五、辅助函数（L757-792）
// L757-762 错误结果工厂
function createErrorToolResult(message: string): AgentToolResult<any> {
return { content: [{ type: "text", text: message }], details: {} };
}
// L774-787 构造 ToolResultMessage
function createToolResultMessage(finalized: FinalizedToolCallOutcome): ToolResultMessage {
return {
role: "toolResult",
toolCallId: finalized.toolCall.id,
toolName: finalized.toolCall.name,
content: finalized.result.content ?? [], // L781 null 归一化（JS 扩展工具可能无 content）
details: finalized.result.details,
...(finalized.result.addedToolNames?.length ? { addedToolNames: finalized.result.addedToolNames } : {}),
isError: finalized.isError,
timestamp: Date.now(),
};
}
L778-781 的 null 归一化：注释说"untyped tools (JS extensions) can return results without content"。因为 JS 扩展工具不走 Rust 的强类型，可能返回 content: null。这里 ?? [] 保证 null 永不进入 session 历史或 provider payload——防御性编程。
addedToolNames 条件展开（L783）：只在非空时加入字段，避免序列化出空的 addedToolNames: []。
六、设计模式总结
读完 792 行，提炼出 agent-loop.ts 的核心设计模式：
1. 错误编码而非抛出（贯穿全文）
StreamFn契约：不 throw，错误进 streamprepareToolCall：验证失败返回immediate错误结果executePreparedToolCall：工具抛错 →createErrorToolResultfinalizeExecutedToolCall：after hook 抛错 → 错误结果
所有错误都变成 ToolResultMessage(isError: true)，循环永远不会因为工具错误而中断事件序列。
2. 生产者-消费者解耦（L34-49）
agentLoop 立即返回 EventStream，后台 void promise 推事件。消费者 for await 拉事件，生产者跑循环。
3. 流式消息"活性"（L321-361）
partial 消息在流式过程中就躺在 transcript 末尾，delta 原地替换。UI 实时可见，transcript 始终最新。
4. 双层循环分离语义（L170-272）
- 内层：steering（打断）+ 工具调用
- 外层：follow-up（续接） 两种"中途加消息"的语义清晰区分。
5. 边界唯一转换（L288-295）
convertToLlm 是 AgentMessage→Message 的唯一桥梁。transformContext（AgentMessage→AgentMessage）在它之前。
6. 顺序 vs 并发的精巧设计（L435-556）
- sequential：完整流水线一个一个来
- parallel：prepare 顺序、execute 并发、message 按源序
7. 字段级覆盖无深度合并（L736-742）
after hook 的覆盖是"给了就整体替换，没给就保持"，避免部分覆盖导致状态不一致。
8. 防御性归一化（L781）
外部来源（JS 扩展）的 content: null 被 ?? [] 归一化，保证内部不变量。
七、纯 std 实现启示
读完循环逻辑，对 focus 实现的关键启示：
-
循环本身是纯的状态机——
runLoop只是while+ 条件判断 + 函数调用，零 I/O。纯 std 完全可实现。 -
StreamFn可以是 trait：trait StreamFn: Send { fn stream(&self, model: &Model, ctx: &Context, opts: &StreamOptions) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>>; }或者更简单——用同步迭代器/channel。核心循环不关心 stream 怎么来。
-
EventStream 可用
mpsc::Receiver替代：let (tx, rx) = mpsc::channel(); // 后台线程：循环跑，tx.send(event) // 主线程：rx.iter() 消费Rust 的 std::sync::mpsc 完全够用。或者用 crossbeam（但那不是 std）。
-
并发工具执行可用
std::thread+JoinHandle：顺序 prepare，spawn 线程并发 execute，join 收集结果。 -
AgentMessage→Message 转换是纯函数——不依赖任何外部库。
结论：agent-loop.ts 的逻辑可以几乎 1:1 移植到纯 std Rust，唯一的妥协是 async（可以用同步阻塞 + 线程替代）和 JSON（手写极简解析器或认栽用 serde）。
至此，Pi 的核心两层（类型 + 循环）已完整解读。配合 00-pi-architecture-overview.md 的全景，你已具备从零实现一个 agent 框架所需的全部知识。