accTitle: 第三部的工具执行流水线
accDescr: 模型输出 tool-call blocks 后，调度器按 executionMode 分组，经过注册过滤、参数校验、并行池、审批、沙箱、子进程执行，结果经 spill 截断后回到模型；run_code 内部有镜像调度器。
flowchart TD
    MODEL["模型输出 tool-call blocks"] --> SCHED["executeToolCalls 调度器"]
    SCHED --> GROUP["按 executionMode 分组<br/>连续 parallel = 一组<br/>exclusive = 全屏障"]
    GROUP --> VIEW["view() 作用域解析<br/>ScopedLayers 过滤<br/>run_code 懒加载追加"]
    VIEW --> VALID["参数校验<br/>外层硬校验 throw<br/>内层软校验 undefined"]
    VALID --> POOL["bounded pool<br/>maxParallelToolCalls=10<br/>fillPool 重新 classify"]
    POOL --> APPROVAL{"审批路径"}
    APPROVAL -->|Path A: hooks/plugins| ASK["pre-execute 问人"]
    APPROVAL -->|Path B: bash/fs| ESCALATE["deny→retry with sandbox_permissions<br/>严格 widening"]
    ASK --> DISPATCH["dispatch body 并发<br/>pre/post 串行"]
    ESCALATE --> DISPATCH
    DISPATCH --> SANDBOX{"danger-full-access?"}
    SANDBOX -->|否| CONFINE["SandboxProvider.confine()<br/>bwrap/seatbelt/windows-acl"]
    SANDBOX -->|是| SPAWN["直接 spawn"]
    CONFINE --> SPAWN
    SPAWN --> SUBPROC["SubprocessRuntime<br/>scrubbedParentEnv<br/>SIGTERM→grace→SIGKILL tree"]
    SUBPROC --> PTY["PTY 由 LocalBashExecutor 管理"]
    PTY --> RESULT["工具结果"]
    RESULT --> SPILL{"spill-policy 检查大小"}
    SPILL -->|> maxInlineBytes| SAVE["SpillStore.saveText<br/>head+tail 50/50 split"]
    SPILL -->|≤ cap| BACK
    SAVE --> BACK["按模型顺序 commitReady<br/>concludesTurn 标记结束"]
    BACK --> MODEL2["结果回到模型"]
    
    subgraph CODE_MODE["run_code 内部镜像调度器"]
        direction LR
        SUB["子调用队列"] --> SUB_CLASSIFY["executionMode 重新分类"]
        SUB_CLASSIFY --> SUB_POOL["maxParallelSubCaps 并行池"]
        SUB_POOL --> SUB_BARRIER["exclusive 屏障含 post-execute"]
        SUB_BARRIER --> SUB_COMMIT["head-of-line 保序提交"]
    end
    
    DISPATCH -.-> CODE_MODE

repo="${DSH_SOURCE_DIR:?set DSH_SOURCE_DIR to the official fixed checkout}"
test "$(git -C "$repo" rev-parse HEAD)" = \
  47f943859bef60e4160492346772ded9b24f765a
grep -n "executeToolCalls" "$repo/packages/core/agent-loop/src/tool-calls.ts" | head -5
grep -n "class ToolRuntime" "$repo/packages/core/tools/src/index.ts" | head -3
grep -n "approveEscalation" "$repo/packages/sandbox/sandbox/src/escalation.ts" | head -3
grep -n "apply(ctx" "$repo/packages/spill/spill-policy/src/index.ts" | head -3

工具调用不是一行 await

从产品界面看，工具调用像三步：模型决定调工具，工具执行，结果返回。但代码里不能只靠一个 switch (toolName) 加一句 await tool.execute(args) 撑住这件事。

Harness 把工具调用做成了一个完整的运行时子系统，它同时承担三个角色：

1. 分层注册域——工具不是丢进一个 flat map 就完事，而是按作用域组织成 global / ancestor / own 三层，带可见性过滤和影子覆盖。
2. 四阶段执行管线——每个调用经过 pre-execute（策略门）→ around-dispatch（包装层）→ post-execute（结果策略）→ finalize（内容终结化）四个瀑布。
3. 并发调度器——一次助手回复可能包含多个工具调用，调度器按 parallel/exclusive 两种模式编排它们的并发关系。

这三者合在一起，就是 ToolRuntime 这个 Service。

ToolRuntime 的三合一身份

先看它的声明：

export class ToolRuntime extends Service {
  static inject = ['systemPrompt']
  // ...
  readonly [TOOL_RUNTIME_SCHEDULER]: ToolRuntimeScheduler = { /* ... */ }
  private readonly layers = new ScopedLayers(/* ... */)
}

layers 是分层注册域的存储；TOOL_RUNTIME_SCHEDULER 是暴露给 agent-loop 的调度接口——这个 Symbol-keyed 属性故意不出现在公开 API 上，只有 dsh-agent-loop 直接消费它。

第一幕：分层注册域

三层结构

工具注册不是全局唯一的。ScopedLayers 把工具组织成三种来源：

• Global layer：进程级别，所有 agent 都能看到。
• Ancestor layers：从最远祖先到直接父级的链式贡献。一个 agent preset 在它自己的 scope 注册的工具，对挂在它下面的所有子 agent 可见。
• Own layer：agent 自己的 scope 注册的工具。own-layer 的注册不受 restriction 过滤——这是有意设计，因为一个 delegation runtime 注册在子 agent 自己层上的汇报工具不应被父级的过滤条件误杀。

视图解析发生在 view() 方法里：

private view(scope?: ScopeKey): ToolView {
  const layers = this.layers.chainLayers(scope)
  const own = this.layers.peek(scope)
  const inherited = new Map<string, ToolDefinition>(this.layers.global.tools.entries())
  for (const layer of layers) {
    if (layer === own) continue
    for (const [name, definition] of layer.tools.entries()) inherited.set(name, definition)
  }
  // ... apply restrictions, add own-layer, add code transport
}

关键细节：restrictions 是对 inherited 表面的交集过滤。own-layer 的注册永远穿透。

Restrictions: allow/deny 交集

restrict(filter: ToolRestriction): () => void {
  // 必须在 scoped context 上调用
  // allow/deny 只能命名已知 global 工具
  // 编译为 CompiledToolRestriction 存入 layer.restrictions
}

多个 restriction 之间做 交集：链上任何一个 layer 拒绝了某个名字，那个工具就不可见。这意味着越深层的 agent 只能看到越少的东西——这就是最小权限。

ToolPresentationMode

注册域还负责一件事：决定模型看到工具的方式。三种模式：

• native：每个可见工具直接暴露给模型（经典 function calling）。
• code：只暴露 run_code，所有其他工具通过生成的 SDK prompt 在程序内部调用。
• both：两种方式并存。

模式按 scope chain 继承——一个 preset 声明 code，它下面所有 agent 都走 code 模式，除非自己覆盖。

第二幕：并发调度器

入口：executeToolCalls

当 agent-loop 收到助手回复中的 toolCalls[]，它调用 executeToolCalls：

export async function executeToolCalls(
  ctx: Context,
  turn: number, step: number,
  toolCalls: ToolCallBlock[],
  signal: AbortSignal,
  acceptContext: (context: UserMessage) => void,
): Promise<{ concluded: boolean }>

逻辑清晰得令人意外：一个 while 循环从头到尾扫描 planned 数组。每到一个位置，先问注册表”这个调用是什么模式”：

• 如果是 parallel：贪心地把从当前位置到末尾的所有剩余调用作为一组交给 runGroup（组内再做滚动池限流）。
• 如果是 exclusive：只取当前这一个调用作为一组。

这意味着一个 exclusive 调用会在前后形成 barrier——前面的 parallel 组必须全部完成，exclusive 独占执行，然后后面的调用才继续分组。

ExecutionMode 的 fail-closed 分类

executionMode(exec: ToolExecutionInput): ToolExecutionMode {
  const tool = this.resolveExecution(exec.name, exec.agent, exec.parent !== undefined)
  if (!tool?.isConcurrencySafe) return { kind: 'exclusive' }
  try {
    const concurrencySafe: unknown = tool.isConcurrencySafe(exec.arguments)
    return concurrencySafe === true ? { kind: 'parallel' } : { kind: 'exclusive' }
  } catch {
    return { kind: 'exclusive' }
  }
}

注意这里的极端保守策略：

• 工具没声明 isConcurrencySafe？Exclusive。
• isConcurrencySafe 返回了 "yes"（truthy 但不是 boolean）？Exclusive。
• isConcurrencySafe 抛了异常？Exclusive。
• 工具根本不存在（unknown tool）？Exclusive。

只有严格等于 true 才走并行。这是一个 fail-closed 安全边界——任何不确定性都退化为串行。

滚动并发池

进入 parallel 组后，runGroup 使用一个滚动池来限制实际并发数：

const { maxParallelToolCalls } = ctx.agentLoop.config
// ...
while (!aborted && nextToStart < group.length && inFlight.size < maxParallelToolCalls) {
  // 重新分类后续调用
  if (nextToStart > 0 && mode === 'parallel'
    && ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
  await startCall(nextToStart)
  nextToStart++
}

默认 maxParallelToolCalls 是 10：

export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10

池子的工作方式是”滚动”的：不是等全部完成再开下一批，而是有一个完成就立刻尝试填充一个新的。这最大化了吞吐。

动态重分类

这个调度器有个容易忽略的关键点：每次准备启动下一个调用时，都会重新询问注册表它的模式。这意味着：

• 一个 exclusive 调用的 body 可以替换注册表中的工具定义（比如 unregister 旧的、register 新的）。
• 后续原本是 parallel 的调用，重分类后可能变成 exclusive——fillPool 中的 break 立刻停止填充，形成新的 barrier。

测试中有一个经典场景：工具 replace 在执行时替换了工具 x 的注册（从 parallel-safe 变为 exclusive），导致后续的 x 调用从并发变为串行。

it('reclassifies pending calls after an exclusive barrier replaces their tool', async () => {
  // ... replace tool swaps x from parallel to exclusive ...
  // After the barrier, c2 and c3 run one at a time
})

这不是一个 edge case 兼容——这是有意设计的动态能力授予/撤销机制。

第三幕：四阶段执行管线

每个工具调用从”被调度器启动”到”结果提交”之间，经历一个严格的四阶段管线。调度器通过 TOOL_RUNTIME_SCHEDULER 暴露的三个方法来分阶段驱动：

export interface ToolRuntimeScheduler {
  prepare(exec: ToolExecutionInput): Promise<ScheduledToolPreparation>
  dispatch(exec: ToolRunContext): Promise<ScheduledToolDispatch>
  finalize(exec: ToolRunContext, result: ToolExecutionResult): Promise<ToolExecutionResult>
  finish(exec: ToolRunContext, result: ToolExecutionResult): ToolExecutionResult
}

为什么要拆成三个方法而不是一个 execute？因为调度器需要在 prepare 之后、dispatch 之前做并发控制。prepare 是有序的（按模型输出顺序），dispatch 可以并发重叠，finalize/finish 又回到有序提交。

阶段一：prepare（pre-execute + guard）

ToolExecutionInput → createExecution() → pre-execute waterfall → guard chain → ScheduledToolPreparation

createExecution 做三件事：

1. 给调用分配一个 opaque ToolExecutionToken（用于关联身份）。
2. 对 arguments 做 snapshotJsonValue + deepFreeze（不可变化快照）。
3. 检查 code-mode collapse（code 模式下只有 run_code 能被模型直接调用）。

然后进入 tools/pre-execute 瀑布：

'tools/pre-execute'(exec, next): Promise<PreToolDecision>

监听器可以返回三种决策：

• allow：放行。
• deny：带原因拒绝，tool body 永远不会执行。
• ask：转交给 ApprovalService（用户确认），确认后才放行。

瀑布之后还有 monotonic guard——一个纯同步检查链，任何一个 guard 返回 reason 就拒绝，且没有 guard 能撤销另一个 guard 的拒绝：

private guardReason(exec: ToolExecution): string | undefined {
  const globalReason = this.layers.global.guardReason(exec)
  if (globalReason !== undefined) return globalReason
  for (const layer of this.layers.chainLayers(exec.agent)) {
    const reason = layer.guardReason(exec)
    if (reason !== undefined) return reason
  }
  return undefined
}

Guard 的设计哲学：单调递增的拒绝权，任何人可以拒，没人可以强制放行。

阶段二：dispatch（around-execute + body）

prepare 返回 { kind: 'dispatch' } 后，调度器在合适的时机调用 dispatch(exec)。这个阶段是可以重叠的——多个 parallel 调用的 dispatch 并发执行。

'tools/execute'(exec, next): Promise<ToolExecutionResult>

tools/execute 是 around-dispatch 瀑布。一个典型用途是超时策略：包装层替换 exec.signal 为一个带 timeout 的新 signal，然后调 next() 进入实际 body。

但这里有一个重要的安全保证：信号融合（signal fusing）。

function fuseToolSignals(caller: AbortSignal, wrapper: AbortSignal): FusedToolSignal {
  if (caller === wrapper) return { signal: caller, dispose() {} }
  const controller = new AbortController()
  // ... 监听两个 signal, 任一 abort 就 abort fused
}

即使 around-wrapper 替换了 signal，注册表在调用 body 之前会把原始 caller signal 和 wrapper signal 融合。这保证了：wrapper 无法让调用脱离 caller 的取消控制。

阶段三：finalize（post-execute）

dispatch 返回后，调度器按模型顺序调用 finalize：

'tools/post-execute'(exec, result, next): Promise<PostToolDecision>

Post-execute 监听器看到的是已经标准化的结果，它可以：

• accept：保持结果（可选替换 content 或 value）。
• block：把成功结果变成错误（corrective feedback），丢弃 body 产生的 deferContext。

这里有一个微妙的不对称：accept 保留 body 的 deferred context，block 丢弃它。这是因为 block 语义上意味着”这次调用不该发生”，所以它产生的副作用上下文不应该传递给下一轮。

阶段四：finish（content finalization + notify）

最后，finish 做三件事：

1. 调用 ToolDefinition.finalizeContent——工具自己的最后内容变换。
2. materializeFinalResult——对整个结果做 snapshotJsonValue + deepFreeze。
3. notifyResult——发射 tools/result 事件（emit 模式，错误被吞）。

从 finish 出来的结果是完全不可变的冻结对象。任何观察者拿到的都是同一个冻结快照。

第四幕：模型顺序提交

调度器可以并发 dispatch，但结果必须按模型输出顺序提交到 session：

const commitReady = async (): Promise<void> => {
  while (committed < group.length) {
    const slot = slots[committed]
    if (slot === undefined) break
    // finalize or finish, then appendToolResult
    committed++
  }
}

这意味着：即使 call-3 先完成，它也要等 call-1 和 call-2 提交后才能提交。这保证了 session 事件流的顺序确定性——replay 时完全可重现。

每个提交包含两个 session event：

• tool/call：在 startCall 时写入，记录调用开始。
• tool/result：在 commitReady 时写入，通过 sourceEventSeqs 链接回对应的 tool/call。

第五幕：取消语义

取消不是”丢掉结果就完了”。调度器区分两种取消状态：

1. ABORTED_BEFORE_DISPATCH（code: ABORTED_BEFORE_DISPATCH）：tool body 从未被调用。调用被跳过，session 记录一个合成错误结果。
2. ABORTED（code: ABORTED）：tool body 已经启动并被 drain 到 quiescence。已启动的 body 不会被强杀——调度器等它自然结束，然后把成功结果替换为 ABORTED 错误。

function toolAbortedBeforeDispatchResult(): ToolExecutionResult {
  return {
    content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],
    isError: true,
    error: { message: 'tool call aborted before dispatch',
             info: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },
  }
}

取消到来时调度器的行为：

1. 停止从池中启动新调用（fillPool 的 !aborted 守卫）。
2. 等已启动的调用全部 settle（await Promise.allSettled(inFlight.values())）。
3. 对所有未启动的调用，按模型顺序追加 appendSkippedToolCall（合成 tool/call + tool/result 对）。

为什么未启动的调用也要写入 session？因为 replay 协议要求：模型输出了 N 个 tool_call，session 就必须有 N 个 tool/call + tool/result 对。缺失会导致重放断裂。

最简场景：单个 exclusive 调用

把上面的机制串起来。模型回复了一个 tool_call Read { file_path: "/foo.ts" }：

1. executeToolCalls 收到 toolCalls = [Read]。
2. executionMode(Read) 返回 { kind: 'exclusive' }（假设 Read 没声明 isConcurrencySafe）。
3. runGroup 拿到只有一个元素的 group。
4. startCall(0)：
   • prepare：创建 execution → tools/pre-execute waterfall（没有监听器，默认 allow）→ guard chain（无拒绝）→ 返回 { kind: 'dispatch' }。
   • dispatch：进入 tools/execute waterfall → dispatchToolBody → fuse signals → tool.execute(args, exec) → 读文件 → 返回文件内容。
   • slot 被填充。
5. commitReady：
   • finalize：tools/post-execute（无监听器，默认 accept）→ applyFinalContent → materializeFinalResult → notifyResult。
   • appendToolResult 写入 session。
6. 循环结束，返回 { concluded: false }。

一个调用走完了整个管线。没有并发，没有 barrier，但每个阶段都经过了。

复杂场景：parallel 组遇到动态重分类

模型一次回复了五个调用：[A, B, C, D, E]。其中 A、B、C、D 声明了 isConcurrencySafe: () => true，E 没有。maxParallelToolCalls = 2。

1. 第一个调用 A 分类为 parallel，整个 [A, B, C, D, E] 作为一组进入 runGroup（E 也在组里——它会在启动前被重分类）。
2. fillPool：
   • 启动 A（inFlight = 1）→ 启动 B（inFlight = 2）→ 池满，停。
3. A 先完成。commitReady 提交 A 的结果。fillPool 再检查 C：重分类为 parallel，启动 C。
4. B 完成。提交 B。fillPool 检查 D：重分类为 parallel，启动 D。
5. C 完成。提交 C。fillPool 检查 E：重分类为 exclusive——break！
6. D 完成。提交 D。池空了。
7. runGroup 返回 consumed = 4，因为只消费了到 exclusive barrier 之前的部分。
8. 外层 while 循环回到 E，E 作为 exclusive 独占调用开新组。

结果顺序永远是模型输出顺序 A→B→C→D→E，即使实际执行顺序可能是 A→B→C→D 并发加 E 串行。

失败边界

边界1：调度器内部失败

如果一个 tool body 抛出的异常没被 dispatchToolBody 捕获（实际上不太可能，因为有 try/catch），或者 tools/execute wrapper 本身崩了，那会触发 schedulerFailure：

schedulerFailure ??= { error }

一旦设置，throwSchedulerFailure() 会在下一个检查点抛出。此时：

• 停止所有新的 dispatch。
• await Promise.allSettled(inFlight.values()) 等所有已启动调用结束。
• 不为已记录的 tool/call 制造合成结果——这和 abort 不同。abort 会补结果，scheduler failure 不会。

这个区别存在的理由：abort 是预期内的优雅停止，结果可以合成；scheduler failure 是非预期的系统错误，制造合成结果可能掩盖问题。

边界2：pre-execute 期间取消

prepare 阶段的 tools/pre-execute waterfall 是异步的。如果在 waterfall 执行过程中 signal 被 abort：

• waterfall 仍然完成（注册表不会 abandon promise）。
• 完成后检查 callerCancelled(exec)，如果已取消，返回 post-result + ABORTED_BEFORE_DISPATCH。
• 但如果是 ask 分支且 approval service 报告了 cancelled，结果也走 ABORTED_BEFORE_DISPATCH 但仍然过 post-execute。

边界3：around-dispatch 替换 signal 后不恢复

如果一个 tools/execute wrapper 替换了 exec.signal 但没有正确清理（比如忘了在 finally 里恢复），信号融合保证了至少 caller signal 的取消语义不会丢失。但 wrapper 自己的 timeout signal 可能泄漏——这是 wrapper 的 bug，不是注册表的。

边界4：post-execute 替换 value

post-execute 的 accept 决策可以替换 value（通过返回 { kind: 'accept', value: newValue }）。但它不能同时替换 content 和 value——注册表会抛 TypeError。替换 value 会重新走 output.render，生成新的 content。替换 content 只改展示层，不改语义值。

注册域的参数验证

工具注册时，注册表做以下检查：

1. output 必须是 { schema, render, presentationMeta? } 形状。
2. output.schema 必须通过 assertSupportedJsonSchema（支持的 JSON Schema 子集）。
3. timeoutMs 如果声明，必须是正有限数。
4. 名字不能是 run_code（保留给 Code Mode transport）。
5. 同一 layer 内不能重名。

执行时，工具 body 返回的 value 会被 validateJsonSchemaValue 验证——如果不符合声明的 output schema，直接变成 ToolOutputError。这意味着工具不能撒谎：你声明了返回 { type: 'string' }，就必须返回 string。

isConcurrencySafe 的参数感知分类

isConcurrencySafe 不仅仅是一个静态声明——它接收 parsed arguments：

isConcurrencySafe?(args: unknown): boolean

这允许同一个工具根据参数做不同的并发决策。一个典型例子是文件系统工具：

isConcurrencySafe: (args) => args.mode === 'read'

读操作可以并行，写操作必须独占。调度器在每次分类时传入当前调用的参数，让工具自己决定。

但要注意：defineTool 的 validated 版本会在分类前验证参数。如果参数不合法（比如缺少 required field），分类器会抛异常——按 fail-closed 规则退化为 exclusive。不会有”参数错误的并行调用”这种事。

Scoped 事件分发

管线中的每个瀑布事件（tools/pre-execute、tools/execute、tools/post-execute）都是scope-filtered的：

const carrier = scopeTarget(this, exec.agent)
const gate = await this.ctx.waterfall(carrier, 'tools/pre-execute', exec, ...)

这意味着：如果你在某个 agent 的 scope 上注册了 tools/pre-execute 监听器，它只会收到那个 agent 发起的调用。全局监听器收所有。

唯一的例外是 tools/change——它是 unfiltered emit，因为注册变更可能影响所有 agent。

连接下一章

你现在知道了工具调度器的完整图景：分层注册域决定谁能看到什么，executionMode 决定并发策略，滚动池限流执行，四阶段管线提供策略注入点。

但有一个问题我们故意跳过了：当 ToolPresentationMode 是 code 时，run_code 这个 transport 是怎么把生成的程序里的工具调用桥接回注册表的？那个桥接层——Code Mode Runtime——有自己的调度器（maxParallelSubCalls）和自己的一套事件协议（tool/code-dispatch-start、tool/code-dispatch）。那是下一章的内容。