Pi Agent 源码解析（一）：从调试环境到双层主循环

async function fetchNvidiaNimModelIds(): Promise<Map<string, string>> {
	try {
		console.log("Fetching models from NVIDIA NIM API...");
		const response = await fetch(`${NVIDIA_BASE_URL}/models`);
		if (!response.ok) throw new Error(`NVIDIA NIM API returned ${response.status}`);
		// ...
	} catch (error) {
		console.error("Failed to fetch NVIDIA NIM models:", error);
		if (generatorOptions.strict) throw error;
		return new Map();
	}
}

async function fetchOpenRouterModels(): Promise<Model<any>[]> {
	try {
		console.log("Fetching models from OpenRouter API...");
		const response = await fetch("https://openrouter.ai/api/v1/models");
		if (!response.ok) throw new Error(`OpenRouter API returned ${response.status}`);
		// ...
	}
}

async function loadModelsDevData(): Promise<Model<any>[]> {
	try {
		console.log("Fetching models from models.dev API...");
		const response = await fetch("https://models.dev/api.json");
		if (!response.ok) throw new Error(`models.dev API returned ${response.status}`);
		// ...
	}
}

const providersDir = join(packageRoot, "src/providers");
const dataDir = join(providersDir, "data");
const stagingRoot = mkdtempSync(join(providersDir, ".model-generation-"));
const stagedDataDir = join(stagingRoot, "data");
const previousDataDir = join(stagingRoot, "previous-data");

// 先向 stagedDataDir 写入各 provider 的 JSON 和 manifest
validateModelDataDirectory(modelDataStructure, stagedDataDir);

const hadPreviousData = existsSync(dataDir);
if (hadPreviousData) renameSync(dataDir, previousDataDir);
try {
	renameSync(stagedDataDir, dataDir);
	validateGeneratedModelData(packageRoot);
} catch (error) {
	rmSync(dataDir, { recursive: true, force: true });
	if (hadPreviousData && existsSync(previousDataDir)) {
		renameSync(previousDataDir, dataDir);
	}
	throw error;
}

try {
	validateGeneratedModelData(packageRoot);
	console.log("Generated model data is valid.");
} catch (error) {
	console.error(error instanceof Error ? error.message : String(error));
	console.error(
		"\nModel data is missing or stale. " +
		"Run `npm run hydrate:model-data` from the repository root.",
	);
	process.exitCode = 1;
}

const authPath = options.agentDir ? join(agentDir, "auth.json") : undefined;
const modelsPath = options.agentDir ? join(agentDir, "models.json") : undefined;
const modelRuntime =
	options.modelRuntime ?? (await ModelRuntime.create({ authPath, modelsPath }));

const modelsPath =
	options.modelsPath === null
		? undefined
		: (options.modelsPath ?? join(getAgentDir(), "models.json"));
const config = await ModelConfig.load(modelsPath);

#!/usr/bin/env node
import { setupCli } from "./cli/setup.ts";
import { main } from "./main.ts";

setupCli();
main(process.argv.slice(2));

if (initialMessage) {
	await session.prompt(initialMessage, { images: initialImages });
}

for (const message of messages) {
	await session.prompt(message);
}

import {
	Agent,
	type AgentMessage,
	setDefaultStreamFn,
	type ThinkingLevel,
} from "@earendil-works/pi-agent-core";
import {
	clampThinkingLevel,
	type Message,
	type Model,
	streamSimple,
} from "@earendil-works/pi-ai/compat";

// 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);

import type { StreamFn } from "./types.ts";

let defaultStreamFn: StreamFn | undefined;

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;
}

agent = new Agent({
	// initialState 只是 Agent Core 的初始快照。
	// systemPrompt 和 tools 暂时为空，AgentSession 随后会根据资源加载结果填充。
	initialState: {
		systemPrompt: "",
		model,
		thinkingLevel,
		tools: [],
	},

	// Coding Agent 中除了 user/assistant/toolResult，还有扩展产生的 custom message。
	// 这里负责把产品层消息转成 pi-ai 接受的 Message，并按设置屏蔽图片。
	convertToLlm: convertToLlmWithBlockImages,

	// 这是本文最关键的注入点。
	// Agent Core 只保存此函数的引用，不知道认证文件、供应商 SDK 和重试配置。
	streamFn: async (model, context, options) => {
		// 每次请求时动态读取设置，所以运行中修改配置也能影响后续 turn。
		const providerRetrySettings = settingsManager.getProviderRetrySettings();
		const httpIdleTimeoutMs = settingsManager.getHttpIdleTimeoutMs();

		// SDKs treat timeout=0 as 0ms (immediate timeout), not "no timeout".
		// Use max int32 to effectively disable the timeout.
		const effectiveTimeoutMs =
			httpIdleTimeoutMs === 0 ? 2147483647 : httpIdleTimeoutMs;

		// 优先级：单次调用参数 > provider 重试配置 > 全局 HTTP 空闲超时。
		const timeoutMs =
			options?.timeoutMs ??
			providerRetrySettings.timeoutMs ??
			effectiveTimeoutMs;

		// WebSocket 连接超时独立于普通 HTTP 空闲超时。
		const websocketConnectTimeoutMs =
			options?.websocketConnectTimeoutMs ??
			settingsManager.getWebSocketConnectTimeoutMs();

		// extensionRunnerRef 在 Agent 创建后由 AgentSession 绑定。
		// 用 ref 是为了让闭包始终拿到当前 runner，而不是初始化时的 undefined。
		const headerRunner = extensionRunnerRef.current;

		// 真正进入模型运行时。ModelRuntime 会解析 provider、认证与协议实现。
		return modelRuntime.streamSimple(model, context, {
			// 先保留 Agent Core 传来的 signal、apiKey、thinking 等请求选项。
			...options,
			timeoutMs,
			websocketConnectTimeoutMs,

			// 单次请求可覆盖默认重试策略；未传入才读取 SettingsManager。
			maxRetries:
				options?.maxRetries ?? providerRetrySettings.maxRetries,
			maxRetryDelayMs:
				options?.maxRetryDelayMs ??
				providerRetrySettings.maxRetryDelayMs,

			// provider 即将发送请求前执行。先补 Pi 的来源/session header，
			// 再让扩展做最后修改，扩展因此拥有最终 header 的控制权。
			transformHeaders: async (requestHeaders) => {
				const headers = mergeProviderAttributionHeaders(
					model,
					settingsManager,
					options?.sessionId,
					requestHeaders,
				);
				return headerRunner?.hasHandlers("before_provider_headers")
					? headerRunner.emitBeforeProviderHeaders(headers ?? {})
					: (headers ?? {});
			},
		});
	},

	// onPayload 处理已经转换成供应商协议、但尚未真正发出的请求体。
	// 没有扩展监听时直接原样返回，避免无意义复制。
	onPayload: async (payload, _model) => {
		const runner = extensionRunnerRef.current;
		if (!runner?.hasHandlers("before_provider_request")) {
			return payload;
		}
		return runner.emitBeforeProviderRequest(payload);
	},

	// onResponse 暴露 HTTP 状态与响应头，而正文仍由流事件处理。
	onResponse: async (response, _model) => {
		const runner = extensionRunnerRef.current;
		if (!runner?.hasHandlers("after_provider_response")) {
			return;
		}
		await runner.emit({
			type: "after_provider_response",
			status: response.status,
			headers: response.headers,
		});
	},

	// sessionId 会一路传到 provider，可供支持会话亲和或缓存的后端使用。
	sessionId: sessionManager.getSessionId(),

	// 在 convertToLlm 之前运行扩展的上下文 hook。
	// 它操作的仍然是 AgentMessage[]，适合裁剪、插入或改写上下文。
	transformContext: async (messages) => {
		const runner = extensionRunnerRef.current;
		if (!runner) return messages;
		return runner.emitContext(messages);
	},

	// steering 选择何时向进行中的任务插话；followUp 决定后续任务如何出队。
	steeringMode: settingsManager.getSteeringMode(),
	followUpMode: settingsManager.getFollowUpMode(),

	// 这些值不会在 sdk.ts 中消费，而是继续透传给 Agent Loop 和 provider。
	transport: settingsManager.getTransport(),
	thinkingBudgets: settingsManager.getThinkingBudgets(),
	maxRetryDelayMs: settingsManager.getProviderRetrySettings().maxRetryDelayMs,
});

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

streamSimple(
	model: Model<Api>,
	context: Context,
	options?: ModelsSimpleStreamOptions,
): AssistantMessageEventStream {
	return lazyStream(model, async () => {
		const prepared = await this.prepareRequest(model, options);
		return prepared.provider.streamSimple(
			prepared.model,
			context,
			prepared.options as SimpleStreamOptions,
		);
	});
}

private async _runAgentPrompt(
	messages: AgentMessage | AgentMessage[],
): Promise<void> {
	this._isAgentRunActive = true;
	try {
		await this.agent.prompt(messages);
		while (await this._handlePostAgentRun()) {
			await this.agent.continue();
		}
	} finally {
		this._systemPromptOverride = undefined;
		this._flushPendingBashMessages();
		this._flushPendingCustomMessages();
		await this._emitAgentSettled();
	}
}

async prompt(message: AgentMessage | AgentMessage[]): Promise<void>;
async prompt(input: string, images?: ImageContent[]): Promise<void>;
async prompt(
	input: string | AgentMessage | AgentMessage[],
	images?: ImageContent[],
): Promise<void> {
	// activeRun 同时充当并发保护：一个 Agent 实例不能并行跑两个 prompt。
	if (this.activeRun) {
		throw new Error(
			"Agent is already processing a prompt. Use steer() or followUp() " +
			"to queue messages, or wait for completion.",
		);
	}
	// string、单个 AgentMessage 和 AgentMessage[] 最终统一成数组。
	// 字符串会被包装为 role=user，并附加 timestamp 与可选图片。
	const messages = this.normalizePromptInput(input, images);
	await this.runPromptMessages(messages);
}

private async runPromptMessages(
	messages: AgentMessage[],
	options: { skipInitialSteeringPoll?: boolean } = {},
): Promise<void> {
	// runWithLifecycle 创建 AbortController、设置 isStreaming，
	// 并保证成功、异常或取消后都能清理 activeRun。
	await this.runWithLifecycle(async (signal) => {
		await runAgentLoop(
			// 本次新输入；runAgentLoop 会把它拼到历史上下文末尾。
			messages,
			// 使用数组副本，避免低层循环直接修改 Agent.state 的顶层数组。
			this.createContextSnapshot(),
			// 把模型、hook、队列读取器和执行策略冻结为本次运行配置。
			this.createLoopConfig(options),
			// 主循环只 emit；Agent 通过 processEvents 归并自己的状态。
			(event) => this.processEvents(event),
			// 同一个 AbortSignal 贯穿模型请求、hook 与工具调用。
			signal,
			// 即 sdk.ts 构造 Agent 时显式注入的 streamFn。
			this.streamFunction,
		);
	});
}

export async function runAgentLoop(
	prompts: AgentMessage[],
	context: AgentContext,
	config: AgentLoopConfig,
	emit: AgentEventSink,
	signal: AbortSignal | undefined,
	streamFn: StreamFn,
): Promise<AgentMessage[]> {
	// newMessages 是本次运行的增量结果，初始就包含用户新提交的 prompts。
	const newMessages: AgentMessage[] = [...prompts];

	// currentContext 是本次循环使用的工作副本：保留旧历史，再拼接新输入。
	// 展开 context 只浅复制对象，因此 messages 必须单独创建新数组。
	const currentContext: AgentContext = {
		...context,
		messages: [...context.messages, ...prompts],
	};

	// 事件顺序是协议的一部分：先开始 Agent 和 turn，再完整发布用户消息。
	await emit({ type: "agent_start" });
	await emit({ type: "turn_start" });
	for (const prompt of prompts) {
		await emit({ type: "message_start", message: prompt });
		await emit({ type: "message_end", message: prompt });
	}

	// 正常的 Coding Agent 总会显式传 streamFn；右侧是旧调用方的兼容 fallback。
	await runLoop(
		currentContext,
		newMessages,
		config,
		signal,
		emit,
		streamFn ?? getDefaultStreamFn(),
	);
	// runLoop 内部通过 push 修改 newMessages，这里返回的就是本次运行完整增量。
	return newMessages;
}

async function runLoop(
	initialContext: AgentContext,
	newMessages: AgentMessage[],
	initialConfig: AgentLoopConfig,
	signal: AbortSignal | undefined,
	emit: AgentEventSink,
	streamFunction: StreamFn,
): Promise<void> {
	// 这两个变量都可能被 prepareNextTurn 替换，所以不能使用 const。
	let currentContext = initialContext;
	let config = initialConfig;
	// 第一轮之前没有已完成的 turn；从第二轮起它携带上一轮快照。
	let lastCompletedTurn: PrepareNextTurnContext | undefined;

	// 进入循环前先拉取一次 steering。例如 Agent 真正开始执行前，
	// 用户已经排入了一条纠偏消息，它也应进入第一次请求。
	let pendingMessages: AgentMessage[] =
		(await config.getSteeringMessages?.()) || [];

	// 外层：Agent 本来要停下时，如果还有 follow-up，就重新进入内层
	while (true) {
		// 初始为 true，保证没有排队消息时也至少请求一次 LLM。
		// 后续它表示上一条 assistant message 是否留下了待反馈的工具结果。
		let hasMoreToolCalls = true;

		// 内层：只要还有工具结果需要交给模型，或有 steering 消息，就继续
		while (hasMoreToolCalls || pendingMessages.length > 0) {
			// 第一轮会跳过。第二轮起，在每次 LLM 请求前给上层一次机会，
			// 用上一轮完整结果更新上下文、模型或 thinking level。
			if (lastCompletedTurn) {
				const nextTurnSnapshot =
					await config.prepareNextTurn?.(lastCompletedTurn);
				if (nextTurnSnapshot) {
					// hook 可以只返回需要替换的字段，其余部分继续沿用。
					currentContext =
						nextTurnSnapshot.context ?? currentContext;
					config = {
						...config,
						model: nextTurnSnapshot.model ?? config.model,
						reasoning:
							nextTurnSnapshot.thinkingLevel === undefined
								? config.reasoning
								: nextTurnSnapshot.thinkingLevel === "off"
									? undefined
									: nextTurnSnapshot.thinkingLevel,
					};
				}

				// prepareNextTurn 可能在做耗时的上下文压缩。压缩期间到达的
				// steering 也要进入下一轮；已有 pending 时不再拉取，避免
				// one-at-a-time 模式在一个 turn 中取出两条消息。
				if (pendingMessages.length === 0) {
					pendingMessages =
						(await config.getSteeringMessages?.()) || [];
				}
				// turn_start 放在准备结束后，表示下一轮现在才正式开始。
				await emit({ type: "turn_start" });
			}

			// 在下一次 assistant 响应前注入 steering 消息
			if (pendingMessages.length > 0) {
				for (const message of pendingMessages) {
					// 先发标准消息事件供 Session/UI 消费，再同步写入两个数组。
					await emit({ type: "message_start", message });
					await emit({ type: "message_end", message });
					currentContext.messages.push(message);
					newMessages.push(message);
				}
				// 已转移到上下文，必须清空，防止下个 turn 重复注入。
				pendingMessages = [];
			}

			// 一次真正的 LLM 请求与流式响应消费。函数返回时，
			// currentContext 最后一项已经从 partial message 变成最终消息。
			const message = await streamAssistantResponse(
				currentContext,
				config,
				signal,
				emit,
				streamFunction,
			);
			// streamAssistantResponse 已维护完整上下文；这里只记录本次增量。
			newMessages.push(message);

			// provider 将请求失败与取消编码进最终 AssistantMessage。
			// 这两种情况直接正常发出结束事件，不再执行工具或读取队列。
			if (
				message.stopReason === "error" ||
				message.stopReason === "aborted"
			) {
				await emit({ type: "turn_end", message, toolResults: [] });
				await emit({ type: "agent_end", messages: newMessages });
				return;
			}

			const toolCalls = message.content.filter(
				(c) => c.type === "toolCall",
			);
			// turn_end 始终需要统一的数组结构，所以无工具时也创建空数组。
			const toolResults: ToolResultMessage[] = [];
			// 先假设工具链结束。只有确实执行过工具且批次未要求终止，
			// 才会在下面把它恢复为 true。
			hasMoreToolCalls = false;

			if (toolCalls.length > 0) {
				// length 表示 assistant 输出被 token 上限截断。
				// 此时所有 toolCall 参数都不可信，即使其中某个 JSON 可以解析。
				const executedToolBatch =
					message.stopReason === "length"
						? await failToolCallsFromTruncatedMessage(
								toolCalls,
								emit,
							)
						: await executeToolCalls(
								currentContext,
								message,
								config,
								signal,
								emit,
							);

				toolResults.push(...executedToolBatch.messages);
				// 普通工具结果必须再交给 LLM，总结结果或决定下一步。
				// 只有工具批次明确 terminate 时，才不自动继续这一工具链。
				hasMoreToolCalls = !executedToolBatch.terminate;

				for (const result of toolResults) {
					// 完整上下文决定下次发给 LLM 什么；newMessages 只负责
					// 本次 Agent Run 的返回值和 agent_end 事件。
					currentContext.messages.push(result);
					newMessages.push(result);
				}
			}

			// 一个 turn 包含一次 assistant 响应及其引发的整批工具执行，
			// 所以 turn_end 必须等所有 toolResults 准备好后再发送。
			await emit({ type: "turn_end", message, toolResults });

			// 下一轮的 hook 需要同时看到响应、工具结果、完整上下文和运行增量。
			lastCompletedTurn = {
				message,
				toolResults,
				context: currentContext,
				newMessages,
			};

			// “优雅停止”发生在当前 turn 完整结束后，优先于队列轮询。
			// 因而不会中断已开始的工具，但会结束本次 Agent Run。
			if (await config.shouldStopAfterTurn?.(lastCompletedTurn)) {
				await emit({ type: "agent_end", messages: newMessages });
				return;
			}

			// 模型响应与工具执行期间新到达的 steering 在这里读取。
			// 非空时，下一次 while 条件成立并开始一个新 turn。
			pendingMessages =
				(await config.getSteeringMessages?.()) || [];
		}

		// 内层结束，Agent 本来准备停止；再检查延后执行的 follow-up
		const followUpMessages =
			(await config.getFollowUpMessages?.()) || [];
		if (followUpMessages.length > 0) {
			// 复用 pendingMessages 的标准注入路径，使 follow-up 同样产生
			// message_start/message_end，并进入上下文与本次运行增量。
			pendingMessages = followUpMessages;
			// 回到外层会重新令 hasMoreToolCalls=true，所以 follow-up
			// 至少触发一次 LLM 请求，而不只是被追加到历史。
			continue;
		}

		// 工具链、steering 和 follow-up 都耗尽，才退出整个 Agent Run。
		break;
	}

	// agent_end 是本次 run 的最后一个协议事件，只携带本次新增消息。
	await emit({ type: "agent_end", messages: newMessages });
}

while (hasMoreToolCalls || pendingMessages.length > 0)

async function streamAssistantResponse(
	context: AgentContext,
	config: AgentLoopConfig,
	signal: AbortSignal | undefined,
	emit: AgentEventSink,
	streamFunction: StreamFn,
): Promise<AssistantMessage> {
	// context.messages 是循环内的权威历史。transformContext 不直接写回它，
	// 而是生成“仅用于本次请求”的视图，可用于压缩、裁剪或扩展上下文。
	// AgentMessage[] → AgentMessage[]，此时仍允许包含产品层 custom message。
	let messages = context.messages;
	if (config.transformContext) {
		messages = await config.transformContext(messages, signal);
	}

	// AgentMessage[] → pi-ai Message[]。UI 通知等模型不认识的消息在这里过滤，
	// custom message 也必须在这里变换成 user/assistant/toolResult 之一。
	const llmMessages = await config.convertToLlm(messages);

	// 把 system prompt、已转换消息和工具 schema 组成一次完整模型输入。
	// 注意这里传的是工具“定义”，真正的 execute 函数仍只保留在 Agent Context。
	const llmContext: Context = {
		systemPrompt: context.systemPrompt,
		messages: llmMessages,
		tools: context.tools,
	};

	// 每次请求都重新取 key，兼容会过期和刷新的 OAuth token。
	// getApiKey 的动态结果优先；没有结果时才回退到 config.apiKey。
	const resolvedApiKey =
		(config.getApiKey
			? await config.getApiKey(config.model.provider)
			: undefined) || config.apiKey;

	// streamFunction 返回可异步迭代的 AssistantMessageEventStream，
	// 调用到这里并不代表完整回答已经到达。
	const response = await streamFunction(config.model, llmContext, {
		// config 继承 SimpleStreamOptions，因此请求 hook、thinking、transport 等
		// 选项先整体透传，再用下面两个字段覆盖成当前调用的最终值。
		...config,
		apiKey: resolvedApiKey,
		signal,
	});

	// partialMessage 始终指向“截至当前事件已经拼装出的完整消息”，
	// 而不是单独的 token/delta。
	let partialMessage: AssistantMessage | null = null;
	// 标记 context.messages 中是否已经为这个回答占了最后一个位置。
	// 它用于兼容没有 start 事件、直接结束的 provider 流。
	let addedPartial = false;

	// Provider 事件按产生顺序异步到达；for await 自然形成背压，
	// 当前事件处理和 emit 没结束前不会消费下一事件。
	for await (const event of response) {
		switch (event.type) {
			case "start":
				// start.partial 通常只有 assistant 消息的骨架和空 content。
				partialMessage = event.partial;
				// 先把骨架放到历史末尾，为后面的 delta 保留固定槽位。
				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":
				// 所有内容事件走同一条归并路径。event.partial 已由 pi-ai
				// 累积好当前文本、thinking 或 toolCall，不需要 Core 自己拼 token。
				if (partialMessage) {
					partialMessage = event.partial;
					// 替换最后一项而非 push：一个回答无论包含多少 delta，
					// 在上下文中始终只占一条 assistant message。
					context.messages[context.messages.length - 1] =
						partialMessage;
					// 原始 provider 事件一并向上传递，TUI 可区分文本、思考和工具参数。
					await emit({
						type: "message_update",
						assistantMessageEvent: event,
						message: { ...partialMessage },
					});
				}
				break;

			case "done":
			case "error": {
				// result() 给出标准化后的最终 AssistantMessage，包含 usage、
				// stopReason、errorMessage 以及最终 content。
				const finalMessage = await response.result();
				if (addedPartial) {
					// 正常流：用最终对象替换之前反复更新的临时槽位。
					context.messages[context.messages.length - 1] =
						finalMessage;
				} else {
					// 防御非常规 provider：没有 start 事件时在结束处补入历史。
					context.messages.push(finalMessage);
				}
				if (!addedPartial) {
					// 保持 Agent 事件协议完整：即使 provider 没 start，
					// 上层仍会按顺序收到 message_start → message_end。
					await emit({
						type: "message_start",
						message: { ...finalMessage },
					});
				}
				// message_end 才表示这条消息可以进入 Agent.state.messages。
				await emit({ type: "message_end", message: finalMessage });
				return finalMessage;
			}
		}
	}

	// 有些实现通过结束 AsyncIterable 表示完成，不额外发 done。
	// 这个兜底分支与上面的 done/error 分支保持完全相同的落库和事件语义。
	const finalMessage = await response.result();
	if (addedPartial) {
		context.messages[context.messages.length - 1] = finalMessage;
	} else {
		context.messages.push(finalMessage);
		await emit({
			type: "message_start",
			message: { ...finalMessage },
		});
	}
	await emit({ type: "message_end", message: finalMessage });
	return finalMessage;
}

async function executeToolCalls(
	currentContext: AgentContext,
	assistantMessage: AssistantMessage,
	config: AgentLoopConfig,
	signal: AbortSignal | undefined,
	emit: AgentEventSink,
): Promise<ExecutedToolCallBatch> {
	// 再从最终 assistant message 中提取一次，确保执行的是已经完成流式拼装的调用。
	const toolCalls = assistantMessage.content.filter(
		(c) => c.type === "toolCall",
	);
	// 工具自身可声明 sequential。例如修改同一状态的工具不能安全并发。
	// 只要批次中有一个这样的工具，整批都按串行处理，避免部分交错。
	const hasSequentialToolCall = toolCalls.some(
		(tc) =>
			currentContext.tools?.find((t) => t.name === tc.name)
				?.executionMode === "sequential",
	);

	// 全局策略和工具局部声明任一要求串行，就走串行实现。
	if (config.toolExecution === "sequential" || hasSequentialToolCall) {
		return executeToolCallsSequential(
			currentContext,
			assistantMessage,
			toolCalls,
			config,
			signal,
			emit,
		);
	}

	// 默认并行路径仍会先顺序执行参数校验和 beforeToolCall，
	// 只有通过预检的 tool.execute 才会并发。
	return executeToolCallsParallel(
		currentContext,
		assistantMessage,
		toolCalls,
		config,
		signal,
		emit,
	);
}

async function prepareToolCall(
	currentContext: AgentContext,
	assistantMessage: AssistantMessage,
	toolCall: AgentToolCall,
	config: AgentLoopConfig,
	signal: AbortSignal | undefined,
): Promise<PreparedToolCall | ImmediateToolCallOutcome> {
	// 模型只能按名称调用当前上下文公开的工具，不能凭空调用任意函数。
	const tool = currentContext.tools?.find(
		(t) => t.name === toolCall.name,
	);
	if (!tool) {
		// “工具不存在”也规范化成即时结果，让 LLM 有机会改正名称。
		return {
			kind: "immediate",
			result: createErrorToolResult(
				`Tool ${toolCall.name} not found`,
			),
			isError: true,
		};
	}

	try {
		// prepareArguments 可做兼容性归一化；随后必须按工具 schema 校验。
		// validatedArgs 才是最终允许交给 execute 的参数。
		const preparedToolCall = prepareToolCallArguments(tool, toolCall);
		const validatedArgs = validateToolArguments(tool, preparedToolCall);

		if (config.beforeToolCall) {
			// hook 能看到 assistant 原消息、原 toolCall、已校验参数和上下文，
			// 可用于权限确认、策略限制、日志或参数审计。
			const beforeResult = await config.beforeToolCall(
				{
					assistantMessage,
					toolCall,
					args: validatedArgs,
					context: currentContext,
				},
				signal,
			);

			// hook 等待期间用户可能取消，因此返回后立即再次检查 signal。
			if (signal?.aborted) {
				return {
					kind: "immediate",
					result: createErrorToolResult("Operation aborted"),
					isError: true,
				};
			}

			if (beforeResult?.block) {
				// block 不抛异常，而是产生模型可读的错误 toolResult。
				const result = createErrorToolResult(
					beforeResult.reason || "Tool execution was blocked",
				);
				if (beforeResult.terminate === true) {
					// terminate 会参与整个批次的终止判断，而不是只标记 UI。
					result.terminate = true;
				}
				return { kind: "immediate", result, isError: true };
			}
		}

		// prepared 是预检与执行之间的边界：只有这种结果会调用 tool.execute。
		return {
			kind: "prepared",
			toolCall,
			tool,
			args: validatedArgs,
		};
	} catch (error) {
		// 参数预处理、schema 校验和 hook 抛错都降级为即时错误结果，
		// 保证 Agent Loop 仍能把错误放回对话，而不是破坏事件序列。
		return {
			kind: "immediate",
			result: createErrorToolResult(
				error instanceof Error ? error.message : String(error),
			),
			isError: true,
		};
	}
}

function createToolResultMessage(
	finalized: FinalizedToolCallOutcome,
): ToolResultMessage {
	return {
		role: "toolResult",
		toolCallId: finalized.toolCall.id,
		toolName: finalized.toolCall.name,
		content: finalized.result.content ?? [],
		details: finalized.result.details,
		usage: finalized.result.usage,
		...(finalized.result.addedToolNames?.length
			? { addedToolNames: finalized.result.addedToolNames }
			: {}),
		isError: finalized.isError,
		timestamp: Date.now(),
	};
}

private async processEvents(event: AgentEvent): Promise<void> {
	// 先归并 Agent 自己的可观察状态，再通知外部 listener。
	// 因此外部收到事件时，读取 state 已经能看到与该事件一致的新值。
	switch (event.type) {
		case "message_start":
			// streamingMessage 是正在生成或刚开始发布的临时消息，
			// 此时还不能放入正式历史，否则取消或 delta 会留下重复记录。
			this._state.streamingMessage = event.message;
			break;

		case "message_update":
			// 每个 delta 都用新的完整 partial snapshot 覆盖旧值。
			this._state.streamingMessage = event.message;
			break;

		case "message_end":
			// 只有结束事件才把消息追加到 Agent 的权威 transcript。
			// agent-loop 内部的 currentContext 是独立数组，所以这里不是重复 push。
			this._state.streamingMessage = undefined;
			this._state.messages.push(event.message);
			break;

		case "tool_execution_start": {
			// 创建新 Set 而不是原地 add，保证观察 state 引用变化的 UI 能刷新。
			const pendingToolCalls = new Set(
				this._state.pendingToolCalls,
			);
			pendingToolCalls.add(event.toolCallId);
			this._state.pendingToolCalls = pendingToolCalls;
			break;
		}

		case "tool_execution_end": {
			// 与 start 对称；并行模式下多个工具会按实际完成顺序依次移除。
			const pendingToolCalls = new Set(
				this._state.pendingToolCalls,
			);
			pendingToolCalls.delete(event.toolCallId);
			this._state.pendingToolCalls = pendingToolCalls;
			break;
		}

		case "turn_end":
			// 普通工具错误保存在 toolResult.isError；这里只记录 assistant 级错误。
			if (event.message.role === "assistant" && event.message.errorMessage) {
				this._state.errorMessage = event.message.errorMessage;
			}
			break;

		case "agent_end":
			// agent_end 是事件流终点，但 activeRun 要等所有 listener 完成后
			// 才会由 finishRun 清理，因此这里仅清空临时消息。
			this._state.streamingMessage = undefined;
			break;
	}

	// 所有事件都必须发生在一次 activeRun 内，并共享这次运行的取消信号。
	const signal = this.activeRun?.abortController.signal;
	if (!signal) {
		throw new Error("Agent listener invoked outside active run");
	}
	// 串行 await 保证 listener 看到严格事件顺序，同时也会形成背压。
	for (const listener of this.listeners) {
		await listener(event, signal);
	}
}