工具并发与结果顺序
区分工具的实际完成顺序与持久化的消息顺序，说明 executionMode 逐工具覆盖、beforeToolCall 拦截和 terminate 整批一致规则。
工具并发与结果顺序
模型一轮里要读三个文件、跑两条命令，是完全正常的。Pi 默认就并发跑它们。
但并发带来一个必须讲清楚的问题：工具完成的顺序，和结果写进会话的顺序，不是同一个东西。 界面上的进度条按完成顺序跳动，会话文件里的工具结果消息却按助手消息里的原始顺序排列。搞混这两者，界面就会显示"第三个工具先完成"，而模型收到的结果顺序是另一个样子。
这篇的核心目标很具体：你能说清每种模式下事件与消息各自的顺序，并写出一个能验证并发确实发生了的测试。
读它需要会用 Vitest 或同类测试框架，并且对 Agent 循环有一个基本印象——模型发起调用，程序校验参数后执行，再把结果回填。读完你能解释为什么界面进度和会话记录可以采用不同的顺序视图，以及如何给一个工具单独关掉并发。
Pi 迭代很快，接口可能变化。动手前先记录 pi --version，并按你安装的版本复核参数。
最终要跑通什么
- 写一个测试，证明并发模式下两个工具确实重叠执行。
- 断言
tool_execution_end的顺序是完成顺序。 - 断言持久化的工具结果消息顺序仍是助手消息的源顺序。
- 确认批里有一个
executionMode: "sequential"的工具时，整批退回串行。
两种执行模式
执行模式有两个取值，定义在 Agent Core 类型定义：
// 执行模式只有两个取值，默认是 parallel
export type ToolExecutionMode = "sequential" | "parallel";默认值是 "parallel"。Agent 类在构造时读它：
// 不传就用默认的并行模式
this.toolExecution = runtimeOptions.toolExecution ?? "parallel";SDK 层可以直接改：
// SDK 层改全局执行模式：改的是 agent 上的字段，不是会话 API
session.agent.toolExecution = "sequential";两种模式的差别在源码里非常清楚：
parallel 模式下两个顺序不一致，正是本篇要讲清楚的事。
并行模式的三段式
源码里 executeToolCallsParallel() 的结构可以拆成三段。
第一段：顺序预检（preflight）。 遍历助手消息里的工具调用，逐个发出 tool_execution_start，然后逐个调用 prepareToolCall()：
// 预检阶段是顺序的：start 事件按源序发出，beforeToolCall 也按源序跑
for (const toolCall of toolCalls) {
await emit({ type: "tool_execution_start", toolCallId: toolCall.id, toolName: toolCall.name, args: toolCall.arguments });
const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal);
if (preparation.kind === "immediate") {
// 校验失败 / 工具不存在 / 被拦截：不执行，直接终结
const finalized = { toolCall, result: preparation.result, isError: preparation.isError };
await emitToolExecutionEnd(finalized, emit);
finalizedCalls.push(finalized);
continue;
}
// 可以执行的调用，登记成待并发执行的闭包
finalizedCalls.push(async () => { /* ... */ });
}预检是顺序的，执行是并发的。 这个区分很重要：预检阶段会跑 beforeToolCall 钩子，所以钩子里看到的顺序是确定的、可预测的源顺序，不会因为并发而乱。
第二段：并发执行。 所有登记好的闭包一起跑：
// Promise.all 保证返回数组与输入同序，但每个闭包内部的 end 事件是完成时立刻发的
const orderedFinalizedCalls = await Promise.all(
finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
);Promise.all 保证返回数组的顺序与输入一致——但每个闭包内部的 emitToolExecutionEnd() 是在它自己完成时立刻触发的。所以 tool_execution_end 事件的发出顺序是完成顺序，而 orderedFinalizedCalls 数组的顺序仍是源顺序。
第三段：按源顺序持久化。 遍历 orderedFinalizedCalls，逐个构造结果消息并发事件：
// 按源顺序构造并发出结果消息：会话文件和 turn_end.toolResults 都走这条路径
const messages: ToolResultMessage[] = [];
for (const finalized of orderedFinalizedCalls) {
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit);
messages.push(toolResultMessage);
}这就是"完成顺序"与"消息顺序"的分离点。 同一批工具，三个顺序各自独立：
串行模式为什么存在
不是所有工具都能并发。源码里的判定逻辑只有一句，但影响很大：
// 批里只要有一个工具声明了 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(...);只要批里有一个工具声明了 executionMode: "sequential"，整批都退回串行。
不是"那个工具串行、其它并发"，而是整批串行。这是保守但正确的设计：一旦有工具依赖共享的可变状态，把其它工具也降级串行才不会产生难以复现的竞态。
AgentTool 上的字段定义：
export interface AgentTool<TParameters extends TSchema = TSchema, TDetails = any> extends Tool<TParameters> {
/**
* Per-tool execution mode override.
* - "sequential": this tool must execute one at a time with other tool calls.
* - "parallel": this tool can execute concurrently with other tool calls.
*
* If omitted, the default execution mode applies.
*/
executionMode?: ToolExecutionMode;
}两种选型：
官方扩展文档的一句话总结是："Use sequential execution when tools share mutable in-memory state."
注意：executionMode 是 AgentTool 上的字段。自定义工具用 defineTool() 定义时，把 executionMode 写在定义对象里即可——内置工具注册时会通过 wrapToolDefinition() 从定义复制到运行时工具上。
两个钩子的时机
beforeToolCall 和 afterToolCall 在并发模式下的位置很讲究。
beforeToolCall 在参数校验之后运行。 源码顺序：
// beforeToolCall 的源码顺序：先校验参数，再跑钩子
const preparedToolCall = prepareToolCallArguments(tool, toolCall);
const validatedArgs = validateToolArguments(tool, preparedToolCall); // 1. schema 校验
if (config.beforeToolCall) {
const beforeResult = await config.beforeToolCall({ assistantMessage, toolCall, args: validatedArgs, context }, signal);
// 2. 钩子拿到的是已校验的参数
if (beforeResult?.block) {
const result = createErrorToolResult(beforeResult.reason || "Tool execution was blocked");
if (beforeResult.terminate === true) result.terminate = true;
// 被拦下的调用不执行，但仍会发出 end 事件和结果消息
return { kind: "immediate", result, isError: true };
}
}拦截的返回值形态：
export interface BeforeToolCallResult {
// block 为 true 时这次调用不会执行
block?: boolean;
// reason 会成为给模型看的错误文本，要给出可照做的修正方向
reason?: string;
// terminate 提示这一批之后不要再发下一次模型请求
terminate?: boolean;
}被拦截的调用不执行，但会产生 tool_execution_end（isError: true）和一条工具结果消息——和失败路径完全一致。
afterToolCall 在执行完成后、终态事件发出前运行。 源码里它在 finalizeExecutedToolCall() 中：
// afterToolCall 在执行完成后、终态事件发出前运行
if (config.afterToolCall) {
// 钩子拿到的是已执行完的结果和 isError，可以改写但看不到同批其它工具
const afterResult = await config.afterToolCall({ assistantMessage, toolCall, args, result, isError, context }, signal);
if (afterResult) {
// 逐字段覆盖：给了就整体替换 content / details，没有深合并
result = {
...result,
content: afterResult.content ?? result.content,
details: afterResult.details ?? result.details,
usage: afterResult.usage ?? result.usage,
terminate: afterResult.terminate ?? result.terminate,
};
isError = afterResult.isError ?? isError;
}
}字段是逐字段覆盖语义：content 一给就整体替换 content 数组，details 一给就整体替换，没有深合并。
两个钩子的时机汇总：
在并发模式下，beforeToolCall 全部在预检阶段跑（顺序），afterToolCall 在各自工具完成时跑（并发）。
afterToolCall 抛错的话，整个结果会被替换成错误结果：
// afterToolCall 抛错，整个结果被替换成错误结果
} catch (error) {
result = createErrorToolResult(error instanceof Error ? error.message : String(error));
isError = true;
}terminate 的整批一致
terminate: true 提示 Agent "这一批跑完了，不用再自动发下一次请求"。判定规则只有一条：
// 整批一致：批里每个最终结果都必须带 terminate: true，有一个不带就整批不终止
function shouldTerminateToolBatch(finalizedCalls: FinalizedToolCallOutcome[]): boolean {
return finalizedCalls.length > 0 && finalizedCalls.every((finalized) => finalized.result.terminate === true);
}批里每一个最终结果都必须带 terminate: true，有一个不带就整批不终止。
三种产生 terminate 的途径：
三个必须知道的边界：
第一，失败天然不终止。 异常路径生成的错误结果没有 terminate 字段。所以一批工具里有一个抛错，整批就不终止——这通常正是你想要的。
第二，terminate 只是运行时提示。 官方说明写得很清楚："The hint is runtime-only; emitted toolResult transcript messages remain standard LLM tool results." 也就是说 terminate 不会写进会话文件里的工具结果消息。
第三，并行模式下判定用的是源顺序数组。 shouldTerminateToolBatch(orderedFinalizedCalls) 传的是 Promise.all 的结果数组，元素顺序是源顺序——但 every() 与顺序无关，所以这一点不影响判定结果，只影响可读性。
写一个验证并发顺序的测试
这个测试是本篇最有价值的产物。它不依赖真实模型，直接断言三个顺序。
import { describe, expect, it } from "vitest";
import { agentLoop } from "@earendil-works/pi-agent-core";
import { createAssistantMessageEventStream } from "@earendil-works/pi-ai";
import { Type } from "typebox";
import type { AgentContext, AgentEvent, AgentLoopConfig, AgentMessage, AgentTool } from "@earendil-works/pi-agent-core";
const echoSchema = Type.Object({ value: Type.String() });
/** 让第一个调用等第二个完成，制造出确定的"完成顺序 ≠ 源顺序"。 */
function createEchoTool() {
let releaseFirst: (() => void) | undefined;
const firstDone = new Promise<void>((resolve) => {
releaseFirst = resolve;
});
let firstResolved = false;
let parallelObserved = false;
const tool: AgentTool<typeof echoSchema> = {
name: "echo",
label: "Echo",
description: "回显工具",
parameters: echoSchema,
async execute(_toolCallId, params) {
if (params.value === "first") {
await firstDone;
firstResolved = true;
}
// 第二个调用在第一个还没完成时就跑完了 —— 这就是并发的证据
if (params.value === "second" && !firstResolved) {
parallelObserved = true;
}
return {
content: [{ type: "text", text: `echoed: ${params.value}` }],
details: { value: params.value },
};
},
};
return { tool, release: () => releaseFirst?.(), wasParallel: () => parallelObserved };
}
describe("工具并发与结果顺序", () => {
it("完成顺序按完成时刻，持久化消息按源顺序", async () => {
const { tool, release, wasParallel } = createEchoTool();
const context: AgentContext = { messages: [], tools: [tool] };
const config: AgentLoopConfig = {
model: {
id: "mock", name: "mock", api: "openai-responses", provider: "openai",
baseUrl: "https://example.invalid", reasoning: false, input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: 8192, maxTokens: 2048,
},
convertToLlm: (messages) => messages as never,
toolExecution: "parallel",
};
const userMessage: AgentMessage = { role: "user", content: "跑两个", timestamp: Date.now() };
let callIndex = 0;
const stream = agentLoop([userMessage], context, config, undefined, () => {
const mock = createAssistantMessageEventStream();
queueMicrotask(() => {
if (callIndex === 0) {
mock.push({
type: "done",
reason: "toolUse",
message: {
role: "assistant",
content: [
{ type: "toolCall", id: "tool-1", name: "echo", arguments: { value: "first" } },
{ type: "toolCall", id: "tool-2", name: "echo", arguments: { value: "second" } },
],
api: "openai-responses", provider: "openai", model: "mock",
usage: {
input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "toolUse",
timestamp: Date.now(),
},
});
// 20ms 后放行第一个调用，制造确定的完成顺序
setTimeout(() => release(), 20);
} else {
mock.push({
type: "done",
reason: "stop",
message: {
role: "assistant", content: [{ type: "text", text: "done" }],
api: "openai-responses", provider: "openai", model: "mock",
usage: {
input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.now(),
},
});
}
callIndex++;
});
return mock;
});
const events: AgentEvent[] = [];
for await (const event of stream) {
events.push(event);
}
// 1) 并发确实发生了：第二个调用在第一个完成前就跑完了
expect(wasParallel()).toBe(true);
// 2) tool_execution_end 按完成顺序
const endIds = events.flatMap((e) =>
e.type === "tool_execution_end" ? [e.toolCallId] : [],
);
expect(endIds).toEqual(["tool-2", "tool-1"]);
// 3) 持久化的 toolResult 消息按源顺序
const resultIds = events.flatMap((e) =>
e.type === "message_end" && e.message.role === "toolResult" ? [e.message.toolCallId] : [],
);
expect(resultIds).toEqual(["tool-1", "tool-2"]);
// 4) turn_end.toolResults 同样是源顺序
const turnIds = events.flatMap((e) =>
e.type === "turn_end" ? e.toolResults.map((r) => r.toolCallId) : [],
);
expect(turnIds).toEqual(["tool-1", "tool-2"]);
});
});第 1 条断言是整组测试的核心。光断言顺序不够——如果实现退化成串行，"tool-2 先完成"根本不会发生，顺序断言会因为数据不对而失败，但失败原因不是你以为的那个。加一条"确实重叠了"的断言，才能真正证明并发路径被走到。
验证整批串行的降级
第二个测试确认 executionMode 的逐工具覆盖会让整批退回串行：
it("批里有一个 sequential 工具，整批退回串行", async () => {
// started 记录每个工具的启动顺序，用来判断有没有真正重叠
const started: string[] = [];
let releaseFirst: (() => void) | undefined;
const firstDone = new Promise<void>((resolve) => {
releaseFirst = resolve;
});
// 声明了 executionMode: "sequential" 的那个工具就是让整批降级的元凶
const slow: AgentTool<typeof echoSchema> = {
name: "slow",
label: "Slow",
description: "慢工具",
parameters: echoSchema,
executionMode: "sequential",
async execute(_toolCallId, params) {
started.push(`slow:${params.value}`);
if (params.value === "first") await firstDone;
return { content: [{ type: "text", text: "ok" }], details: {} };
},
};
// fast 没声明 executionMode，本来可以并发，但整批串行把它也拖住了
const fast: AgentTool<typeof echoSchema> = {
name: "fast",
label: "Fast",
description: "快工具，未声明 executionMode",
parameters: echoSchema,
async execute(_toolCallId, params) {
started.push(`fast:${params.value}`);
return { content: [{ type: "text", text: "ok" }], details: {} };
},
};
const context: AgentContext = { messages: [], tools: [slow, fast] };
const config: AgentLoopConfig = {
model: {
id: "mock", name: "mock", api: "openai-responses", provider: "openai",
baseUrl: "https://example.invalid", reasoning: false, input: ["text"],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: 8192, maxTokens: 2048,
},
convertToLlm: (messages) => messages as never,
toolExecution: "parallel", // 全局是并行
};
const userMessage: AgentMessage = { role: "user", content: "跑", timestamp: Date.now() };
let callIndex = 0;
const stream = agentLoop([userMessage], context, config, undefined, () => {
const mock = createAssistantMessageEventStream();
queueMicrotask(() => {
if (callIndex === 0) {
mock.push({
type: "done",
reason: "toolUse",
message: {
role: "assistant",
content: [
{ type: "toolCall", id: "tool-1", name: "slow", arguments: { value: "first" } },
{ type: "toolCall", id: "tool-2", name: "fast", arguments: { value: "second" } },
],
api: "openai-responses", provider: "openai", model: "mock",
usage: {
input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "toolUse",
timestamp: Date.now(),
},
});
setTimeout(() => releaseFirst?.(), 20);
} else {
mock.push({
type: "done",
reason: "stop",
message: {
role: "assistant", content: [{ type: "text", text: "done" }],
api: "openai-responses", provider: "openai", model: "mock",
usage: {
input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.now(),
},
});
}
callIndex++;
});
return mock;
});
for await (const _event of stream) {
// 消费事件流
}
// 慢工具先开始且未完成时，快工具不能启动 —— 说明整批串行
expect(started).toEqual(["slow:first", "fast:second"]);
});在真实会话里观察并发
上面的测试用的是低层 agentLoop()。想在真实会话里看到同样的三个顺序，用 SDK 层就够了：
import { createAgentSession, SessionManager } from "@earendil-works/pi-coding-agent";
const { session } = await createAgentSession({
sessionManager: SessionManager.inMemory(),
});
try {
// 三条数组分别记录三个顺序，可以直接对照着看差异
const starts: string[] = [];
const ends: string[] = [];
const results: string[] = [];
const unsubscribe = session.subscribe((event) => {
if (event.type === "tool_execution_start") starts.push(event.toolCallId);
if (event.type === "tool_execution_end") ends.push(event.toolCallId);
if (event.type === "message_end" && event.message.role === "toolResult") {
results.push(event.message.toolCallId);
}
});
await session.prompt("并行读三个文件");
console.log("start 顺序（源序）:", starts);
console.log("end 顺序（完成序）:", ends);
console.log("结果消息顺序（源序）:", results);
// 退订只移除这一个监听器
unsubscribe();
} finally {
// 释放：dispose 会中止在跑的工作、清空全部监听器、断开 agent 连接
session.dispose();
}三层顺序在真实会话里同样成立：start 按源序、end 按完成序、结果消息按源序。
要真正验证并发是否发生，光看顺序还不够——加一条"两个工具确实重叠执行"的断言（像上面的测试那样用 executionMode 和一个共享的完成标记），才能排除"实现退化成串行但顺序恰好相同"这种假通过。
在扩展层做拦截
SDK 层的 session.agent.beforeToolCall 可以直接赋值。但在 Pi 的扩展系统里，同样的能力由 tool_call 事件提供：
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
export default function (pi: ExtensionAPI) {
pi.on("tool_call", async (event) => {
// 危险命令直接拦下。tool_call 事件按 toolName 区分类型，
// 窄化到 bash 之后 event.input 就是 BashToolInput。
if (event.toolName === "bash" && /rm\s+-rf/.test(event.input.command)) {
return { block: true, reason: "禁止执行 rm -rf，请改用更精确的删除命令" };
}
return undefined;
});
pi.on("tool_result", async (event) => {
// 结果后处理：给失败的命令结果补一句上下文
if (event.isError && event.toolName === "bash") {
return { content: [...event.content, { type: "text", text: "提示：先跑 ls 确认路径再重试" }] };
}
return undefined;
});
}扩展层的钩子会由 AgentSession 装到 Agent 实例上，所以两套 API 最终落到同一个位置。
扩展层必须遵守并发约束。 官方扩展文档写得很明确：
Tool calls from one assistant message can run in parallel. Do not assume a sibling call or result exists when another tool event runs.
也就是说，扩展处理器里不能假设同批的另一个工具已经跑完。需要顺序保证时，用 executionMode: "sequential" 让整批串行，而不是在处理器里自己加锁。
谁负责什么
最后一行值得展开：界面和持久化记录可以采用不同的顺序视图，但要让用户知道这是两种东西。界面上把三个工具的进度条按完成顺序点亮是自然的，只要别暗示"模型收到的结果也是这个顺序"。
三种场景的可观察结果
你现在能做什么
- 把上面两个测试直接跑起来，确认你安装的版本里"完成序 ≠ 源序"这件事成立——这是并发行为最容易被误解的地方。
- 跑一次
session.agent.toolExecution = "sequential"，观察tool_execution_end的顺序是否变成源序，验证全局开关生效。 - 给你那个会改共享内存状态的自定义工具加上
executionMode: "sequential"，然后写一个断言证明整批变成了串行。 - 在扩展的
tool_result处理器里故意读同批另一个工具的结果，观察它是否拿到——如果拿不到，就说明你依赖了并发下不成立的假设。