工具参数校验与并行执行池

你可能以为”调用一个工具”就是把参数传进去、等返回就行了。但在 DeepSeek Harness 里，“调用”这个动作被拆成了两个独立的工程问题：参数怎么校验，多个调用怎么并行。这两个机制彼此正交但共同约束了执行行为：参数校验决定一个调用能否执行，并行池决定多个合法调用如何重叠。

本章拆解这两个机制的源码实现。

第一部分：参数校验——从作者 DSL 到运行时验证

1.1 作者 DSL：ValueSchemaSpec 和 ParameterSchemaSpec

当你用 defineTool() 定义一个工具时，你写的 parameters 不是原始 JSON Schema——它是一个叫 ParameterSchemaSpec 的作者 DSL。这个 DSL 的特点是每个 property 自带 required?: true 标注，整个对象是一个”隐式 open object root”。

/** One implicit parameter-root property, optionally required. */
export type ParameterPropertySpec = ValueSchemaSpec & { required?: true }

/**
 * Tool parameter schema. The map itself is an implicit open object root;
 * requiredness remains a per-property `required: true` annotation.
 */
export type ParameterSchemaSpec = {
  [key: string]: ParameterPropertySpec
  [key: symbol]: never
}

为什么不直接用 JSON Schema？因为 Harness 只支持 JSON Schema 的一个受控子集——不支持 allOf、anyOf、$ref、pattern 等等。这个子集的边界由 assertSupportedJsonSchema 守住。超出子集的 schema 在定义时就会被拒绝，而不是在运行时偷偷忽略某个关键词。

1.2 编译：parameterSchemaSpecToJsonSchema

defineTool() 在注册时就把作者 DSL 编译成了标准 JSON Schema（受控子集内的）：

export function parameterSchemaSpecToJsonSchema(spec: ParameterSchemaSpec): ParameterJsonSchema {
  const compiled = compilePropertyMap(spec, 'parameters')
  const schema: ParameterJsonSchema = {
    type: 'object',
    properties: compiled.properties,
    ...(compiled.required === undefined ? {} : { required: compiled.required }),
  }
  assertSupportedJsonSchema(schema)
  return schema
}

编译过程本身是非递归的。runSchemaCompiler 用一个显式任务栈来避免深层嵌套 schema 导致栈溢出：

function runSchemaCompiler(initial: CompileTask): void {
  const seen = new Set<object>()
  const tasks: CompileTask[] = [initial]
  for (let task = tasks.pop(); task !== undefined; task = tasks.pop()) {
    // ... 逐项处理，循环引用用 seen 集合检测
  }
}

seen 集合保证了循环引用的检测——如果你的 schema 对象引用了自己，编译阶段就会报 is circular 错误。

1.3 运行时校验：validateJsonSchemaValue

编译后的 JSON Schema 被 defineTool 缓存在闭包里。每次模型调用工具时，execute 包装函数在调用真正的 body 之前先做一次完整的 schema 验证：

async execute(args: unknown, exec: ToolRunContext): Promise<JsonValue> {
  const violations = validate(args)
  if (violations.length > 0) throw new ToolArgsError(violations)
  return userExecute(args as InferArgs<S>, exec) as Promise<JsonValue>
},

这里 validate 就是 validateJsonSchemaValue(parameters, args, '')。注意它返回的是所有违规项的数组，而不是遇到第一个就停。这是为了给模型提供足够的修复信息。

validateJsonSchemaValue 本身也是非递归的——它用显式帧栈（ValueFrame）来遍历嵌套结构：

function checkValue(schema: JsonSchemaNode, value: unknown, path: string): string[] {
  const frames: ValueFrame[] = [valueFrame(schema, value, path)]
  let rootResult: string[] | undefined

  // ...
  while (frames.length > 0) {
    const frame = frames.at(-1)
    // ...处理 children phase、oneOf 分支匹配计数等
  }
  return rootResult ?? losslessValueViolation(path)
}

关键点：oneOf 的处理是计数匹配分支数——只有 matches === 1 才通过。这意味着如果你的值碰巧同时匹配了两个分支，也会报错。

1.4 ToolArgsError：一个 HarnessError

校验失败不是普通的 Error，而是 ToolArgsError extends HarnessError，它的 code 是 'INVALID_ARGS'。这使得下游的 policy 和 replay 代码可以精确区分”参数不对”和”工具内部报错”——两者的重试策略完全不同。

export class ToolArgsError extends HarnessError {
  readonly violations: string[]
  constructor(violations: string[]) {
    super(`invalid arguments: ${violations.join('; ')}`, 'INVALID_ARGS')
    this.name = 'ToolArgsError'
    this.violations = violations
  }
}

1.5 双层校验：硬 throw vs 软 undefined

除了 execute 包装里的硬校验，defineTool 还为 presentCall 和 presentResult 生成了软校验包装：

if (userPresentCall) {
  tool.presentCall = (args: unknown): ToolCallView | undefined => {
    if (validate(args).length > 0) return undefined
    return userPresentCall(args as InferArgs<S>)
  }
}

为什么用软校验？因为 presenter 可能在 replay 阶段被调用——此时 session log 里记录的参数可能来自旧版 schema。硬 throw 会让 replay 崩溃；返回 undefined 会让 UI 优雅降级到 generic 渲染。

同样的软校验也应用于 isConcurrencySafe：

if (userIsConcurrencySafe) {
  tool.isConcurrencySafe = (args: unknown): boolean => {
    if (validate(args).length > 0) return false
    return userIsConcurrencySafe(args as InferArgs<S>)
  }
}

参数不合法时，isConcurrencySafe 返回 false——即”不知道能不能并行，就当不能”。这就是 fail-closed 的设计哲学。

1.6 JSON Schema 子集的边界

Harness 的 JSON Schema 子集支持：

• 标量类型：string、number、integer、boolean、null
• 容器类型：object（带 properties/required/additionalProperties）、array（带 items）
• 字面约束：enum（非空同类数组）、const
• 联合类型：oneOf（至少两个分支，不能与 type 共存）
• 注解：description、title、default、examples
• 一个无约束占位：省略 type 和 oneOf = 接受任何无损 JSON

不支持：allOf、anyOf、not、$ref、if/then/else、pattern、minimum/maximum、format、类型数组 type: ["string", "number"]。

这个子集约束同时生效于输入 schema（参数）和输出 schema（output.schema）。输出 schema 在 body 返回后、render 前做校验——如果你的工具返回了不符合自己声明的值，会触发 ToolOutputError。

第二部分：并行执行池——不是 Promise.all 那么简单

2.1 executionMode：parallel 还是 exclusive

当模型一次返回多个 tool_call 时，调度器需要决定：这些调用能并行吗？

答案取决于每个工具的 isConcurrencySafe 分类器。ToolRuntime.executionMode 是分类入口：

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

注意三个 fail-closed 路径：

1. 工具没声明 isConcurrencySafe → exclusive
2. 分类器返回非 true 值（包括 false、undefined、1）→ exclusive
3. 分类器抛异常 → exclusive（被吃掉，不传播）

这意味着只有显式的 return true 才能让一个调用进入并行组。

2.2 分类器可以是参数感知的

isConcurrencySafe 接收的是解析后的参数，所以同一个工具的不同调用可以有不同的 executionMode：

isConcurrencySafe: args => args.mode === 'read'  // 只读并行，写入排他

这在测试中有明确覆盖：read 调用分类为 parallel，write 调用分类为 exclusive。

2.3 executeToolCalls：调度入口

executeToolCalls 是 agent-loop 层面的调度入口。它把一个 step 的所有 tool calls 按 executionMode 分组执行：

export async function executeToolCalls(
  ctx: Context, turn: number, step: number,
  toolCalls: ToolCallBlock[], signal: AbortSignal,
  acceptContext: (context: UserMessage) => void,
): Promise<{ concluded: boolean }> {
  // ...
  let next = 0
  let concluded = false
  while (next < planned.length) {
    const first = planned[next]!
    const mode = ctx.tools.executionMode(first.exec).kind
    const group = mode === 'parallel' ? planned.slice(next) : [first]
    const outcome = await runGroup(ctx, turn, step, group, mode, signal, acceptContext)
    next += outcome.consumed
    concluded ||= outcome.concluded
    if (outcome.aborted) {
      for (const call of planned.slice(next)) appendSkippedToolCall(session, turn, step, call.block)
      return { concluded }
    }
  }
  return { concluded }
}

关键逻辑：

• 从当前位置的第一个 call 读 mode
• 如果是 parallel，把当前位置往后的所有剩余 calls 作为候选组传入 runGroup（但 runGroup 内部会重新分类截断）
• 如果是 exclusive，只传入这一个 call
• outcome.consumed 告诉主循环这组实际执行了多少

2.4 runGroup：滚动窗口并行池

runGroup 是并行调度的核心。它管理一个滚动窗口，窗口大小是 maxParallelToolCalls：

const fillPool = async (): Promise<void> => {
  while (!aborted && nextToStart < group.length && inFlight.size < maxParallelToolCalls) {
    const nextCall = group[nextToStart]!
    if (nextToStart > 0 && mode === 'parallel'
      && ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
    await startCall(nextToStart)
    nextToStart++
    throwSchedulerFailure()
    await commitReady()
    throwSchedulerFailure()
    if (signal.aborted) aborted = true
  }
}

这段代码里有四个约束条件：

1. 未 abort
2. 还有未启动的 call
3. inFlight.size < maxParallelToolCalls（并发窗口未满）
4. 下一个 call 的 executionMode 仍然是 parallel（重分类截断）

第四点特别重要：即使调度器一开始把后续所有 calls 都传进来了，但如果某个位置的 call 在启动前被重分类为 exclusive（比如因为 registry 在提交间隙发生了变化），它就会截断当前组。runGroup 返回 consumed: started（只消费已启动的），主循环的下一次迭代会为这个 exclusive call 开一个新的 barrier group。

2.5 maxParallelToolCalls：默认 10

/** Default maximum in-flight parallel-safe calls per agent step. */
export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10

这个值通过 agent-loop config 可配：

maxParallelToolCalls: z.number().step(1).min(1).default(DEFAULT_MAX_PARALLEL_TOOL_CALLS)

maxParallelToolCalls 限制的是 inFlight（已 dispatch 但未 settle 的 promise）的数量，不是 group 的总大小。所以即使模型一次返回 50 个 parallel 调用，最多也只有 10 个同时在跑。

对于 code-mode 的 sub-dispatch，有独立的 maxParallelSubCalls（同样默认 10），但走的是同样的调度机制。

2.6 模型顺序提交保证

并行池最关键的约束是：dispatch 可以乱序完成，但结果必须按模型顺序提交。

这通过 slots 数组 + committed 指针实现：

const slots: (Slot | undefined)[] = group.map(() => undefined)
let committed = 0

const commitReady = async (): Promise<void> => {
  while (committed < group.length) {
    const slot = slots[committed]
    if (slot === undefined) break  // 这个位置还没完成，停住
    // ...提交 tool/result 事件
    committed++
  }
}

slots[i] 被填充的时机是 call i 的 dispatch promise resolve 的时刻——这个时刻可以是任意顺序。但 commitReady 只从 committed 指针开始顺序扫描：如果 slot 2 先完成了但 slot 0 和 1 还没完成，slot 2 的结果就在数组里等着，直到 0 和 1 都完成后才一起被顺序提交。

这保证了 session log 里 tool/result 事件的顺序永远和模型原始输出的 tool_call 顺序一致，无论实际执行顺序如何。这对 replay 和上下文重建至关重要。

2.7 Exclusive 工具形成 Barrier

当 executeToolCalls 遇到一个 exclusive 调用时，它只传入这一个 call 作为 group。这意味着：

1. 前面的 parallel group 必须完全提交后才开始这个 exclusive call
2. 这个 exclusive call 完成后才继续看下一个 call

Exclusive 是一个排序屏障。如果模型返回了 [read, read, write, read, read]，调度器会执行为：

• Group 1: [read, read] 并行
• Group 2: [write] 独占
• Group 3: [read, read] 并行

2.8 并行组内的重分类截断

fillPool 在启动每个 call 之前重新调用 executionMode。这不是多此一举——它处理的是一个微妙的时序问题：

在前面的 call 的 commitReady 过程中，finalize 阶段可能触发 registry 变化（比如一个 post-execute listener 注册了新工具或修改了限制）。如果后续 call 的 mode 因此从 parallel 变成了 exclusive，fillPool 的 break 会立即停止填充，让主循环为它开新的 barrier group。

这就是为什么 fillPool 里有这行：

if (nextToStart > 0 && mode === 'parallel'
  && ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break

nextToStart > 0 的检查确保组的第一个 call 不受这个截断影响（它的 mode 已经在主循环里确认了）。

2.9 Abort 行为

Abort 到达时的处理精确区分了三种状态的 calls：

1. 已启动已完成（在 slots 里）：正常 commitReady 提交
2. 已启动未完成（在 inFlight 里）：等它们 settle，然后提交（可能是 ABORTED 结果）
3. 未启动：记录为 ABORTED_BEFORE_DISPATCH 合成结果

if (aborted) {
  for (const call of group.slice(started)) appendSkippedToolCall(session, turn, step, call.block)
  return { consumed: group.length, aborted: true, concluded }
}

未启动的 calls 也会记录 tool/call + tool/result 事件对——这保证了 session log 里每个模型请求的 tool_call 都有对应的结果记录，replay 不会遇到缺失的事件。

2.10 Scheduler Failure 隔离

如果一个 call 的 dispatch promise reject 了（工具内部基础设施错误，不是业务错误），调度器不会立即 throw：

const promise = ctx.tools[TOOL_RUNTIME_SCHEDULER].dispatch(prepared.exec).then(
  (outcome) => {
    slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
    return index
  },
  (error: unknown) => {
    schedulerFailure ??= { error }
    return index
  },
)

它把错误记录到 schedulerFailure，然后在下一次 throwSchedulerFailure() 检查点抛出。在抛出前，已启动的其他 calls 会被 Promise.allSettled 等待完毕——不会留下悬挂的 promise。但已 settle 的 calls 不会尝试提交结果（不制造合成恢复）。

2.11 三阶段调度器接口

ToolRuntimeScheduler 把一个 call 的执行分为三个可独立调度的阶段：

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
}

• prepare：参数物化、pre-execute waterfall、guard 检查——这些是有序的（在 fillPool 的 await startCall 里顺序执行）
• dispatch：around-dispatch waterfall + tool body——这是可重叠的（通过 inFlight Map 管理）
• finalize/finish：post-execute + content finalization——这是有序的（在 commitReady 里按 committed 指针顺序执行）

所以并行池的并行度精确限于 dispatch 阶段。pre-execute 和 post-execute 仍然是模型顺序的。这让 approval 审批、guard 检查、post-execute policy 都能看到稳定的顺序语义。

2.12 parseArguments：容错入口

在到达 schema 校验之前，模型返回的原始字符串参数会被 parseArguments 处理：

function parseArguments(raw: string): unknown {
  try {
    return raw ? JSON.parse(raw) : {}
  } catch {
    return raw  // JSON 解析失败，保留原始字符串
  }
}

空字符串变成 {}；非法 JSON 保留为原始字符串。保留原始字符串不会让后续校验通过（validateJsonSchemaValue 对字符串输入检查 type: 'object' 会立即报 must be an object），但它确保了 session log 能记录模型到底输出了什么。

第三部分：两个机制的交汇

3.1 isConcurrencySafe 的软校验保护

回顾一下，defineTool 为 isConcurrencySafe 生成的包装是：

tool.isConcurrencySafe = (args: unknown): boolean => {
  if (validate(args).length > 0) return false
  return userIsConcurrencySafe(args as InferArgs<S>)
}

这意味着参数不合法的 call 永远被分类为 exclusive。这有实际意义：一个参数不合法的 call 在 prepare 阶段就会被 ToolArgsError 终止，但在分类阶段（executionMode 被调用时），这个 call 还没到 prepare。如果把它分类为 parallel，它会和其他 call 一起进入 fillPool，在 startCall 的 prepare 里失败，然后 slot 会被填充为错误结果。这本身不会出 bug，但把它分类为 exclusive 可以避免浪费一个 inFlight 槽位——反正它一定会 fast-fail。

3.2 从分类到执行的时间差

executionMode 在 executeToolCalls 主循环里对第一个 call 调用一次，在 fillPool 里对后续 call 调用一次。但 prepare（包含参数校验）发生在 startCall 里，在 executionMode 之后。

这意味着：

1. 分类时参数可能是不合法的（但 soft-validate 会返回 exclusive）
2. 即使分类为 parallel，prepare 阶段仍然会做完整的 pre-execute policy（审批等）
3. 一个 call 被分类为 parallel 不意味着它一定能执行——它可能在 pre-execute 被 deny

3.3 输出校验和并行提交

工具返回值的 output schema 校验发生在 createSuccessResult 里：

private createSuccessResult(exec: ToolExecution, tool: ToolDefinition, candidate: unknown): ToolExecutionSuccess {
  const detached = snapshotToolValue(tool.name, candidate)
  const violations = validateJsonSchemaValue(tool.output.schema, detached, 'value')
  if (violations.length > 0) throw new ToolOutputError(tool.name, violations)
  // ...
}

这发生在 dispatch 阶段，所以多个 parallel calls 的输出校验是并行的——它们各自独立校验各自工具的 output schema。但校验失败转化为的 ToolOutputError 结果仍然按模型顺序提交。

收口

这两套机制的取舍很清楚：

参数校验：

• 编译时验证 schema 合法性（子集边界）
• 运行时在 body 前硬校验参数
• presenter 和分类器用软校验（fail-silent / fail-closed）
• 非递归实现避免栈溢出

并行执行池：

• fail-closed 分类（只有 true 是 parallel）
• 滚动窗口而非全量 Promise.all
• 重分类截断处理 registry 变化
• dispatch 重叠但 pre/post 有序
• 模型顺序提交保证（slot 数组 + committed 指针）

这两个机制共同确保了：无论模型输出多么混乱的 tool calls，系统层面永远能维持一致的校验语义和可预测的执行顺序。