From 3494c1aaaf2cd65e17c92fee25aa487759656ee8 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 1 Oct 2026 09:33:55 +0300 Subject: [PATCH] LOU-V3: run a turn's tool calls in parallel (toolConcurrency) When the model returned several tool calls in one turn, AgentExecutor ran them one after another. They now run concurrently, capped by a new `toolConcurrency?: number | 'unbounded'` option on AgentExecutor.execute() and createAgent() (default 'unbounded'; 1 restores sequential execution; anything else throws a descriptive error). Contract (TSDoc on ExecuteOptions.toolConcurrency, docs/api-overview.md): - Tool results are appended in the model's call order, not completion order, so the next provider request is deterministic. - Calls start in call order; each call's tool-call event, onToolCall, validation, preToolCall hooks and needsApproval check run just before it starts, one call at a time. tool-result events fire on completion. With toolConcurrency 1 the event order is exactly the old interleaved order. - Approval: the first call needing approval halts the batch. Earlier calls run (concurrently) and are recorded, then the run pauses on that call as before. Later calls never start (resume behavior for them is unchanged; LOU-U7). - A failing tool gets its own error result; siblings are unaffected. A propagating error stops new calls from starting, awaits running ones, then rejects. - An abort mid-batch keeps finished results and cancels the rest. - Checkpoints are written whenever the in-order prefix of finished calls grows (per call with concurrency 1, as before), so a recorded call is never re-run on resume. Implementation: runToolCall() is split into a gate (prepareToolCall) and execution (settleToolCall); the new toolBatch.ts schedules a batch with a small limiter. Tool-result message building moved to agentRunState. Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + docs/api-overview.md | 57 ++++ src/createAgent.ts | 15 + src/execution/AgentExecutor.ts | 195 +++++++---- src/execution/agentRunState.ts | 69 +++- src/execution/index.ts | 1 + src/execution/parallelToolCalls.test.ts | 429 ++++++++++++++++++++++++ src/execution/toolBatch.ts | 240 +++++++++++++ src/execution/toolCallExecution.ts | 152 ++++++--- 9 files changed, 1046 insertions(+), 113 deletions(-) create mode 100644 src/execution/parallelToolCalls.test.ts create mode 100644 src/execution/toolBatch.ts diff --git a/README.md b/README.md index dcf664f3f..6fb453251 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ dashboard — any provider, any deploy target, no lock-in. - **Human-in-the-loop** — flag a tool `needsApproval` and pause execution until a human approves or rejects it, then `resumeAfterApproval()` from any process - **Durable execution** — pass a `sessionId` + `checkpointStore` and a crash mid-conversation resumes instead of restarting - **Cancellation** — pass an `AbortSignal` (`agent.send(input, { signal })`) to stop a run; it resolves with `finishReason: 'aborted'` and the transcript so far, and the signal reaches the provider, tools and delegated agents +- **Parallel tool calls** — when the model asks for several tools in one turn they run concurrently (cap it with `toolConcurrency`, or `1` for sequential), and results still reach the transcript in the model's call order - **Multi-agent delegation** — wrap a child agent as a tool with `createDelegateTool()`, with a `maxDepth` guard against delegation loops - **Pre/post hooks** — a `HookRegistry` of `AgentHook`s that can inspect or mutate a tool call/LLM generate step, or throw to abort it, run sandboxed by Agent Forge's hook editor - **Guardrails** — fail-closed, concurrently-run checks (secret scan, diff size, test/lint commands) that gate a fixer agent's patch before it's used diff --git a/docs/api-overview.md b/docs/api-overview.md index 726fe9834..546452cd5 100644 --- a/docs/api-overview.md +++ b/docs/api-overview.md @@ -63,6 +63,63 @@ How it behaves: cancelled, so the conversation stays valid for the provider. - An already-aborted signal returns at once without calling the provider. +### Parallel tool calls + +When the model asks for several tools in one turn, they run concurrently. +`toolConcurrency` (on `createAgent()` and `AgentExecutor.execute()`) caps how +many run at once: a positive integer, or `'unbounded'` (the default). Use `1` +for strictly sequential execution, e.g. when your tools share state that is +not safe to touch concurrently. + +```ts +import { createAgent, createMockProvider, defineTool } from '@loushy/build-ai-agent'; +import { z } from 'zod'; + +const getWeather = defineTool({ + name: 'get_weather', + description: 'Current weather for a city', + input: z.object({ city: z.string() }), + execute: async ({ city }) => ({ city, tempC: 21 }), +}); + +const agent = createAgent({ + prompt: 'You are a travel assistant.', + provider: createMockProvider(), + tools: [getWeather], + toolConcurrency: 4, // at most 4 tool calls of a turn in flight +}); +``` + +Guarantees, whatever the limit: + +- **Transcript order is call order.** Tool results are appended in the order + the model requested the calls, not the order they finish, so the next + provider request is deterministic. +- **Events.** Calls start in call order. A call's `tool-call` event, + `onToolCall`, argument validation, `preToolCall` hooks and `needsApproval` + check run just before it starts, one call at a time. Its `tool-result` + event fires when it finishes, so results arrive in completion order. With + `toolConcurrency: 1` events alternate call/result exactly as before. +- **Approvals.** The first call that needs approval stops the batch: the + calls before it run (concurrently) and their results are recorded, then the + run pauses on that call (`finishReason: 'awaiting-approval'`). Calls after + it never start in this run. **Known gap:** after `resumeAfterApproval()` + those later calls are still not run and get no result (tracked as LOU-U7). + Until that lands, use `toolConcurrency: 1` and keep approval tools out of + multi-call turns if this matters to you. +- **Failures are isolated.** A tool that throws gets its own error result; + its siblings carry on. A propagating error (`PropagatingToolError`, such as + the delegation depth guard, or a throwing hook) stops new calls from + starting, waits for the running ones to settle, then rejects the run. No + tool is left running detached. +- **Cancellation.** An abort while a batch runs resolves with + `finishReason: 'aborted'`. Calls that finished keep their results; the rest + get a "cancelled" result. Running tools see the abort through their + `abortSignal`. +- **Checkpoints.** With `sessionId` + `checkpointStore`, a checkpoint is + written each time the in-order run of finished calls grows (with `1`, after + every call, as before). A resumed run never runs a recorded call again. + ## Declarative specs | Export | Description | diff --git a/src/createAgent.ts b/src/createAgent.ts index c7214fd70..3ab14a45f 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -22,6 +22,7 @@ import { LLMProvider } from './providers/llm'; import { ToolRegistry } from './tools/ToolRegistry'; import { ToolDescriptor } from './types'; import type { DefinedTool } from './tools/defineTool'; +import { ToolConcurrency, assertToolConcurrency } from './execution/toolBatch'; /** * Configuration for createAgent(). Tools are keyed by the name the agent @@ -45,6 +46,18 @@ export interface CreateAgentConfig { name?: string; /** Optional maxSteps passed through to AgentExecutor.execute(). */ maxSteps?: number; + /** + * How many tool calls from one model turn may run at once (LOU-V3). + * Defaults to `'unbounded'`; `1` runs them one at a time. Results always + * reach the transcript in the model's call order. See + * `ExecuteOptions.toolConcurrency` for the full contract. + * + * @example + * ```ts + * const agent = createAgent({ prompt: '...', provider, tools: [sendEmail], toolConcurrency: 1 }); + * ``` + */ + toolConcurrency?: ToolConcurrency; } /** Per-call options for `SimpleAgent.send()`. */ @@ -92,6 +105,7 @@ export function createAgent(config: CreateAgentConfig): SimpleAgent { provider: config.provider, toolRegistry, maxSteps: config.maxSteps, + toolConcurrency: config.toolConcurrency, signal: options.signal, }); }, @@ -109,6 +123,7 @@ function assertCreateAgentConfig(config: CreateAgentConfig): void { "createAgent: 'prompt' is required. Example: createAgent({ prompt: 'You are a helpful assistant', provider: myProvider })" ); } + assertToolConcurrency(config.toolConcurrency, 'createAgent'); } /** diff --git a/src/execution/AgentExecutor.ts b/src/execution/AgentExecutor.ts index 31be10e8d..2ad473cfc 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -13,7 +13,19 @@ import { CheckpointStore } from './checkpoint'; import { TraceExporter, withSpan } from './tracing'; import { HookRegistry } from './hooks'; import { isAbortError } from './errors'; -import { ToolCallOutcome, parseToolArguments, runToolCall } from './toolCallExecution'; +import { + PreparedToolCall, + ToolCallOutcome, + parseToolArguments, + runToolCall, +} from './toolCallExecution'; +import { + StartedToolCall, + ToolBatchResult, + ToolConcurrency, + assertToolConcurrency, + runToolBatch, +} from './toolBatch'; import { buildTools, compactGenerateError, @@ -26,7 +38,8 @@ import { AgentRunState, addUsage, loadRunState, - pushCancelledToolResults, + pushAbortedBatchResults, + pushToolResult, saveStepCheckpoint, toExecutionResult, } from './agentRunState'; @@ -282,6 +295,40 @@ export interface ExecuteOptions { * ``` */ signal?: AbortSignal; + /** + * LOU-V3: how many tool calls from ONE model turn may run at the same + * time. Defaults to `'unbounded'` (every call of the turn runs + * concurrently); `1` restores strictly sequential execution. Must be a + * positive integer or `'unbounded'`. + * + * Guarantees, whatever the limit: + * - Tool-result messages are appended to the transcript in the model's + * call order, never completion order, so the next provider request is + * deterministic. + * - Calls are started in call order. Each call's `tool-call` event, + * `onToolCall`, argument validation, `preToolCall` hooks and + * `needsApproval` check run before it starts, one call at a time; its + * `tool-result` event fires as soon as it completes (completion order). + * With `1`, events alternate call/result exactly as before. + * - Approval: the first call that needs approval stops the batch. Calls + * before it run (concurrently) and are recorded; the run then pauses + * on that call (`finishReason: 'awaiting-approval'`); calls after it + * never start in this run. + * - A failing tool does not affect its siblings - each gets its own error + * result. A propagating error (`PropagatingToolError`, a throwing hook) + * stops new calls from starting, waits for the running ones, then + * rejects `execute()` - no tool is left running detached. + * - An abort (`signal`) mid-batch resolves with `finishReason: 'aborted'`: + * finished calls keep their results, the rest get a "cancelled" result. + * - With checkpointing, results are checkpointed as the in-order prefix + * of finished calls grows, so a resumed run never re-runs a recorded call. + * + * @example + * ```ts + * await AgentExecutor.execute({ agent, input: 'Compare 3 cities', provider, toolConcurrency: 2 }); + * ``` + */ + toolConcurrency?: ToolConcurrency; } /** @@ -526,9 +573,11 @@ export class AgentExecutor { } /** - * Executes the tool calls of one assistant turn in order, appending each - * result to the conversation and checkpointing after it. Resolves to the - * ExecutionResult to return when a tool call requires approval. + * Executes the tool calls of one assistant turn - concurrently up to + * `toolConcurrency` (LOU-V3, see that option for the ordering contract) - + * appending the results to the conversation in call order and + * checkpointing as they are recorded. Resolves to the ExecutionResult to + * return when the run pauses for approval or is aborted. */ private static async runToolCalls( options: ExecuteOptions, @@ -547,53 +596,71 @@ export class AgentExecutor { toolCalls, }); - // Execute tools and add results - for (const [index, toolCall] of toolCalls.entries()) { - if (options.signal?.aborted) { - pushCancelledToolResults(state, toolCalls.slice(index)); - return this.abortRun(options, state); - } - - this.emitEvent(onEvent, { - type: 'tool-call', - timestamp: new Date(), - toolCall, - }); - - const toolResult = await this.runToolCallInSpan(options, state, toolCall, agentSpanId); - - if (toolResult.requiresApproval) { - return this.pauseForApproval(options, state, toolCall, toolResult); - } - - this.emitEvent(onEvent, { - type: 'tool-result', - timestamp: new Date(), - toolResult, - }); + const batch = await runToolBatch( + toolCalls, + options.toolConcurrency ?? 'unbounded', + { + start: (toolCall) => { + this.emitEvent(onEvent, { type: 'tool-call', timestamp: new Date(), toolCall }); + return this.startToolCall(options, state, toolCall, agentSpanId); + }, + onComplete: (toolResult) => + this.emitEvent(onEvent, { type: 'tool-result', timestamp: new Date(), toolResult }), + record: (toolCall, outcome) => pushToolResult(state, toolCall, outcome), + persist: () => saveStepCheckpoint(options, state), + }, + options.signal + ); - // A failed tool carries its message as `{error}` (the same shape - // resume.ts uses) - `result` is null then, so the model would - // otherwise see a bare "null" and never learn the call failed. A - // failure that already has a structured result (argument validation) - // keeps it, so the model gets the per-issue detail. - const failed = toolResult.error !== undefined; - const failurePayload = toolResult.result ?? { error: toolResult.error }; - state.messages.push({ - role: 'tool', - content: JSON.stringify(failed ? failurePayload : toolResult.result), - name: toolCall.function.name, - toolCallId: toolCall.id, - toolName: toolCall.function.name, - ...(failed && { isError: true }), - }); + return this.settleToolBatch(options, state, batch); + } - await saveStepCheckpoint(options, state); + /** + * Turns a finished batch into the run's next move: abort (finished calls + * keep their results, the rest are cancelled), reject with the first + * fatal error, pause for approval, or carry on (`undefined`). + */ + private static settleToolBatch( + options: ExecuteOptions, + state: AgentRunState, + batch: ToolBatchResult + ): Promise | undefined { + if (options.signal?.aborted) { + pushAbortedBatchResults(state, batch.unrecorded); + return this.abortRun(options, state); + } + if (batch.failure) { + throw batch.failure.error; + } + if (batch.approval) { + return this.pauseForApproval(options, state, batch.approval.toolCall, batch.approval.outcome); } - return undefined; } + /** + * Starts one tool call. `gate` settles once it passed its pre-execution + * checks (or rejects with the error that stopped it); `done` settles with + * its outcome. + */ + private static startToolCall( + options: ExecuteOptions, + state: AgentRunState, + toolCall: ToolCall, + agentSpanId: string + ): StartedToolCall { + let openGate!: (prepared: PreparedToolCall) => void; + let failGate!: (error: unknown) => void; + const gate = new Promise((resolve, reject) => { + openGate = resolve; + failGate = reject; + }); + const done = this.runToolCallInSpan(options, state, toolCall, agentSpanId, openGate); + // Rejecting an already-resolved gate is a no-op. + done.catch(failGate); + return { gate, done }; + } + /** * Runs one tool call inside a `tool.call` span parented to the run's * `agent.run` span. @@ -602,7 +669,8 @@ export class AgentExecutor { options: ExecuteOptions, state: AgentRunState, toolCall: ToolCall, - agentSpanId: string + agentSpanId: string, + onPrepared?: (prepared: PreparedToolCall) => void ): Promise { const { agent, @@ -633,7 +701,8 @@ export class AgentExecutor { hooks, sessionId, state.messages, - signal + signal, + onPrepared ); const parsedArgs = executed.args === undefined @@ -791,19 +860,24 @@ export class AgentExecutor { hooks?: HookRegistry, sessionId?: string, messages: Message[] = [], - signal?: AbortSignal + signal?: AbortSignal, + onPrepared?: (prepared: PreparedToolCall) => void ): Promise { - return runToolCall(toolCall, { - agent, - toolRegistry, - onToolCall, - onToolResult, - sandbox, - hooks, - sessionId, - messages, - signal, - }); + return runToolCall( + toolCall, + { + agent, + toolRegistry, + onToolCall, + onToolResult, + sandbox, + hooks, + sessionId, + messages, + signal, + }, + onPrepared + ); } /** @@ -833,6 +907,7 @@ export class AgentExecutor { "Example: AgentExecutor.execute({ agent, input: 'hello', provider })" ); } + assertToolConcurrency(options.toolConcurrency, 'AgentExecutor.execute'); } /** diff --git a/src/execution/agentRunState.ts b/src/execution/agentRunState.ts index 4abf34732..18cad33f5 100644 --- a/src/execution/agentRunState.ts +++ b/src/execution/agentRunState.ts @@ -10,6 +10,8 @@ import { AgentConfig } from '../types'; import { Checkpoint } from './checkpoint'; import { CompactedLLMProviderError } from './errors'; import type { ExecuteOptions, ExecutionResult } from './AgentExecutor'; +import type { ToolCallOutcome } from './toolCallExecution'; +import type { UnrecordedToolCall } from './toolBatch'; type TokenUsage = ExecutionResult['usage']; @@ -156,23 +158,60 @@ export async function saveStepCheckpoint( await checkpointStore.save(sessionId, checkpoint); } +/** Appends one settled tool call's result to the transcript. */ +export function pushToolResult( + state: AgentRunState, + toolCall: ToolCall, + outcome: ToolCallOutcome +): void { + // A failed tool carries its message as `{error}` (the same shape + // resume.ts uses) - `result` is null then, so the model would otherwise + // see a bare "null" and never learn the call failed. A failure that + // already has a structured result (argument validation) keeps it, so the + // model gets the per-issue detail. + const failed = outcome.error !== undefined; + const failurePayload = outcome.result ?? { error: outcome.error }; + state.messages.push({ + role: 'tool', + content: JSON.stringify(failed ? failurePayload : outcome.result), + name: toolCall.function.name, + toolCallId: toolCall.id, + toolName: toolCall.function.name, + ...(failed && { isError: true }), + }); +} + /** - * LOU-V1: gives each tool call the run was aborted before reaching a - * `{error}` tool result, so the transcript stays well-formed (every - * assistant tool call has a matching result) and a checkpointed run can be - * resumed without the provider rejecting an unanswered tool call. + * LOU-V1: gives a tool call the run was aborted before reaching a `{error}` + * tool result, so the transcript stays well-formed (every assistant tool + * call has a matching result) and a checkpointed run can be resumed without + * the provider rejecting an unanswered tool call. */ -export function pushCancelledToolResults(state: AgentRunState, toolCalls: ToolCall[]): void { - for (const toolCall of toolCalls) { - state.messages.push({ - role: 'tool', - content: JSON.stringify({ - error: 'Tool call was cancelled before it ran because the run was aborted', - }), - name: toolCall.function.name, - toolCallId: toolCall.id, - toolName: toolCall.function.name, - }); +function pushCancelledToolResult(state: AgentRunState, toolCall: ToolCall): void { + state.messages.push({ + role: 'tool', + content: JSON.stringify({ + error: 'Tool call was cancelled before it ran because the run was aborted', + }), + name: toolCall.function.name, + toolCallId: toolCall.id, + toolName: toolCall.function.name, + }); +} + +/** + * LOU-V3: closes out a batch cut short by an abort - in call order, each + * call that was not yet in the transcript gets its result if it finished, + * or a cancelled result if it never started, was waiting on an approval, + * or ended in a fatal error. + */ +export function pushAbortedBatchResults(state: AgentRunState, calls: UnrecordedToolCall[]): void { + for (const { toolCall, outcome } of calls) { + if (outcome && !outcome.requiresApproval) { + pushToolResult(state, toolCall, outcome); + } else { + pushCancelledToolResult(state, toolCall); + } } } diff --git a/src/execution/index.ts b/src/execution/index.ts index 0949d1df5..2a8c728e7 100644 --- a/src/execution/index.ts +++ b/src/execution/index.ts @@ -18,3 +18,4 @@ export * from './guardrails'; export * from './hooks'; export { ToolArgumentsValidationError } from './toolArgsValidation'; export type { ToolArgumentIssue } from './toolArgsValidation'; +export type { ToolConcurrency } from './toolBatch'; diff --git a/src/execution/parallelToolCalls.test.ts b/src/execution/parallelToolCalls.test.ts new file mode 100644 index 000000000..1b8034fd9 --- /dev/null +++ b/src/execution/parallelToolCalls.test.ts @@ -0,0 +1,429 @@ +/** + * LOU-V3: the tool calls of one model turn run concurrently (up to + * `toolConcurrency`), while the transcript, approvals, failures, aborts and + * checkpoints stay deterministic. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { z } from 'zod'; +import { AgentExecutor, ExecutionEvent, ExecuteOptions, PropagatingToolError } from './AgentExecutor'; +import type { ApprovalStore, ExecutionSnapshot, PendingApproval } from './ApprovalGate'; +import type { Checkpoint, CheckpointStore } from './checkpoint'; +import { resumeAfterApproval } from './resume'; +import { createAgent } from '../createAgent'; +import { defineTool, DefinedTool } from '../tools/defineTool'; +import { ToolRegistry } from '../tools'; +import { mockModel, MockToolCall } from '../testing'; +import { AgentConfig, AgentType } from '../types'; +import type { Message } from '../providers'; + +interface Deferred { + promise: Promise; + resolve: (value: T) => void; + reject: (error: unknown) => void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +/** A tool whose result the test releases by hand, recording start/end into `log`. */ +function controllableTool(name: string, log: string[]) { + const started = deferred(); + const release = deferred(); + const tool = defineTool({ + name, + description: name, + input: z.object({}), + execute: async () => { + log.push(`start:${name}`); + started.resolve(); + const value = await release.promise; + log.push(`end:${name}`); + return value; + }, + }); + return { tool, started: started.promise, release }; +} + +function instantTool(name: string, log: string[], extra: { needsApproval?: boolean } = {}) { + return defineTool({ + name, + description: name, + input: z.object({}), + ...extra, + execute: async () => { + log.push(`start:${name}`); + log.push(`end:${name}`); + return `${name} done`; + }, + }); +} + +function agentFor(tools: DefinedTool[]): AgentConfig { + const config: AgentConfig['tools'] = {}; + for (const t of tools) config[t.name] = { tool: t.name }; + return { id: 'agent-1', name: 'Agent', agentType: AgentType.SmartAssistant, prompt: 'p', tools: config }; +} + +function registryOf(tools: DefinedTool[]): ToolRegistry { + const registry = new ToolRegistry(); + registry.registerMany(tools); + return registry; +} + +function callsTo(...names: string[]): MockToolCall[] { + return names.map((name) => ({ name, id: `call_${name}` })); +} + +/** Runs one turn calling `names` (in order), then a final text turn. */ +function run(tools: DefinedTool[], names: string[], extra: Partial = {}) { + const events: ExecutionEvent[] = []; + const result = AgentExecutor.execute({ + agent: agentFor(tools), + input: 'go', + provider: mockModel([{ toolCalls: callsTo(...names) }, 'all done']), + toolRegistry: registryOf(tools), + onEvent: (event) => events.push(event), + ...extra, + }); + return { result, events }; +} + +function toolMessageIds(messages: readonly Message[]): Array { + return messages.filter((m) => m.role === 'tool').map((m) => m.toolCallId); +} + +function eventTrace(events: ExecutionEvent[]): string[] { + return events.flatMap((e) => { + if (e.type === 'tool-call') return [`call:${e.toolCall?.function.name}`]; + if (e.type === 'tool-result') return [`result:${e.toolResult?.toolName}`]; + return []; + }); +} + +function settledFlag(promise: Promise): { settled: boolean } { + const flag = { settled: false }; + promise.then( + () => (flag.settled = true), + () => (flag.settled = true) + ); + return flag; +} + +const tick = () => new Promise((resolve) => setTimeout(resolve, 5)); + +describe('parallel tool calls (LOU-V3)', () => { + it('runs the calls of one turn concurrently by default', async () => { + const log: string[] = []; + const a = controllableTool('a', log); + const b = controllableTool('b', log); + const { result } = run([a.tool, b.tool], ['a', 'b']); + + // Both have started while neither has finished. + await Promise.all([a.started, b.started]); + expect(log).toEqual(['start:a', 'start:b']); + + a.release.resolve('A'); + b.release.resolve('B'); + expect((await result).finishReason).toBe('stop'); + }); + + it('records results in call order even when they complete in reverse order', async () => { + const log: string[] = []; + const a = controllableTool('a', log); + const b = controllableTool('b', log); + const { result, events } = run([a.tool, b.tool], ['a', 'b']); + await Promise.all([a.started, b.started]); + + b.release.resolve('B'); + await tick(); + a.release.resolve('A'); + const { messages } = await result; + + expect(log).toEqual(['start:a', 'start:b', 'end:b', 'end:a']); + const toolMessages = messages.filter((m) => m.role === 'tool'); + expect(toolMessages.map((m) => [m.toolCallId, m.content])).toEqual([ + ['call_a', '"A"'], + ['call_b', '"B"'], + ]); + // tool-call events in call order before any result; results in completion order. + expect(eventTrace(events)).toEqual(['call:a', 'call:b', 'result:b', 'result:a']); + }); + + it('sends the next provider request with results in call order', async () => { + const log: string[] = []; + const a = controllableTool('a', log); + const b = controllableTool('b', log); + const provider = mockModel([{ toolCalls: callsTo('a', 'b') }, 'all done']); + const agent = createAgent({ prompt: 'p', provider, tools: [a.tool, b.tool] }); + const sent = agent.send('go'); + await Promise.all([a.started, b.started]); + b.release.resolve('B'); + await tick(); + a.release.resolve('A'); + await sent; + + expect(toolMessageIds(provider.calls[1].messages as Message[])).toEqual(['call_a', 'call_b']); + }); + + it('toolConcurrency: 1 is strictly sequential with interleaved events', async () => { + const log: string[] = []; + const a = controllableTool('a', log); + const tools = [a.tool, instantTool('b', log), instantTool('c', log)]; + const { result, events } = run(tools, ['a', 'b', 'c'], { toolConcurrency: 1 }); + await a.started; + await tick(); + expect(log).toEqual(['start:a']); // `b` waits for `a` + + a.release.resolve('A'); + await result; + expect(log).toEqual(['start:a', 'end:a', 'start:b', 'end:b', 'start:c', 'end:c']); + expect(eventTrace(events)).toEqual([ + 'call:a', 'result:a', 'call:b', 'result:b', 'call:c', 'result:c', + ]); + }); + + it('never has more calls in flight than the limit', async () => { + let inFlight = 0; + let maxInFlight = 0; + const names = ['a', 'b', 'c', 'd']; + const delays: Record = { a: 20, b: 5, c: 15, d: 1 }; + const tools = names.map((name) => + defineTool({ + name, + description: name, + input: z.object({}), + execute: async () => { + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, delays[name])); + inFlight--; + return name; + }, + }) + ); + const { result } = run(tools, names, { toolConcurrency: 2 }); + const { messages } = await result; + + expect(maxInFlight).toBe(2); + expect(toolMessageIds(messages)).toEqual(['call_a', 'call_b', 'call_c', 'call_d']); + }); + + it('a failing tool does not affect its siblings', async () => { + const log: string[] = []; + const broken = defineTool({ + name: 'broken', + description: 'always throws', + input: z.object({}), + execute: async (): Promise => { + throw new Error('kaput'); + }, + }); + const { result } = run([broken, instantTool('ok', log)], ['broken', 'ok']); + const { messages, finishReason } = await result; + + expect(finishReason).toBe('stop'); + const [brokenMsg, okMsg] = messages.filter((m) => m.role === 'tool'); + expect(brokenMsg).toMatchObject({ toolCallId: 'call_broken', isError: true }); + expect(JSON.parse(brokenMsg.content)).toMatchObject({ message: 'kaput' }); + expect(okMsg).toMatchObject({ toolCallId: 'call_ok', content: '"ok done"' }); + expect(okMsg.isError).toBeUndefined(); + }); + + it('a propagating error waits for running siblings, starts no new calls, then rejects', async () => { + const log: string[] = []; + const fatal = controllableTool('fatal', log); + const slow = controllableTool('slow', log); + const neverExecute = vi.fn(async () => 'never'); + const queued = defineTool({ name: 'queued', description: 'q', input: z.object({}), execute: neverExecute }); + const { result, events } = run([fatal.tool, slow.tool, queued], ['fatal', 'slow', 'queued'], { + toolConcurrency: 2, + }); + const settled = settledFlag(result); + await Promise.all([fatal.started, slow.started]); + + fatal.release.reject(new PropagatingToolError('stop everything')); + await tick(); + expect(settled.settled).toBe(false); // still waiting for `slow` + + slow.release.resolve('slow done'); + await expect(result).rejects.toThrow('stop everything'); + expect(log).toContain('end:slow'); + expect(neverExecute).not.toHaveBeenCalled(); + expect(events.some((e) => e.type === 'error')).toBe(true); + }); + + it('an abort mid-batch keeps finished results and cancels the rest', async () => { + const controller = new AbortController(); + const log: string[] = []; + const finisher = controllableTool('finisher', log); + const listenerStarted = deferred(); + const listener = defineTool({ + name: 'listener', + description: 'honors the abort signal', + input: z.object({}), + execute: (_args, { abortSignal }) => + new Promise((_resolve, reject) => { + abortSignal?.addEventListener('abort', () => reject(abortSignal.reason)); + listenerStarted.resolve(); + }), + }); + const neverExecute = vi.fn(async () => 'never'); + const queued = defineTool({ name: 'queued', description: 'q', input: z.object({}), execute: neverExecute }); + const { result } = run([finisher.tool, listener, queued], ['finisher', 'listener', 'queued'], { + toolConcurrency: 2, + signal: controller.signal, + }); + await Promise.all([finisher.started, listenerStarted.promise]); + + controller.abort(); + finisher.release.resolve('finished anyway'); + const { finishReason, messages } = await result; + + expect(finishReason).toBe('aborted'); + expect(neverExecute).not.toHaveBeenCalled(); + const toolMessages = messages.filter((m) => m.role === 'tool'); + expect(toolMessages.map((m) => m.toolCallId)).toEqual(['call_finisher', 'call_listener', 'call_queued']); + expect(toolMessages[0].content).toBe('"finished anyway"'); + expect(toolMessages[1].isError).toBe(true); + expect(JSON.parse(toolMessages[2].content)).toEqual({ error: expect.stringContaining('cancelled') }); + }); + + it('checkpoints only the in-order prefix of finished calls', async () => { + const saved: Array> = []; + const checkpoints = new Map(); + const checkpointStore: CheckpointStore = { + async save(sessionId, checkpoint) { + saved.push(toolMessageIds(checkpoint.messages)); + checkpoints.set(sessionId, checkpoint); + }, + async load(sessionId) { + return checkpoints.get(sessionId) ?? null; + }, + async delete(sessionId) { + checkpoints.delete(sessionId); + }, + }; + const log: string[] = []; + const a = controllableTool('a', log); + const b = controllableTool('b', log); + const { result } = run([a.tool, b.tool], ['a', 'b'], { sessionId: 's1', checkpointStore }); + await Promise.all([a.started, b.started]); + + b.release.resolve('B'); + await tick(); + expect(saved).toEqual([]); // `b` finished, but `a` (before it) has not + + a.release.resolve('A'); + await result; + expect(saved).toEqual([['call_a', 'call_b']]); + }); + + describe('approval inside a batch', () => { + function createApprovalStore(): ApprovalStore & { snapshots: ExecutionSnapshot[] } { + const records = new Map(); + const snapshots: ExecutionSnapshot[] = []; + return { + snapshots, + save: async (pending, snapshot) => { + snapshots.push(snapshot); + records.set(pending.id, { pending, snapshot }); + }, + resolve: async (id) => records.get(id) ?? null, + }; + } + + it.each([0, 1, 2])('pauses at the approval call in position %i; earlier calls run once', async (position) => { + const names = ['t0', 't1', 't2']; + const runs: Record = { t0: 0, t1: 0, t2: 0 }; + const tools = names.map((name, index) => + defineTool({ + name, + description: name, + input: z.object({}), + needsApproval: index === position, + execute: async () => { + runs[name]++; + return `${name} done`; + }, + }) + ); + const approvalStore = createApprovalStore(); + const toolRegistry = registryOf(tools); + const agent = agentFor(tools); + + const paused = await AgentExecutor.execute({ + agent, + input: 'go', + provider: mockModel([{ toolCalls: callsTo(...names) }]), + toolRegistry, + approvalStore, + }); + + expect(paused.finishReason).toBe('awaiting-approval'); + const before = names.slice(0, position); + expect(names.filter((n) => runs[n] === 1)).toEqual(before); + expect(toolMessageIds(paused.messages)).toEqual(before.map((n) => `call_${n}`)); + const [snapshot] = approvalStore.snapshots; + expect(snapshot.pendingToolCall.toolCallId).toBe(`call_${names[position]}`); + expect(toolMessageIds(snapshot.currentMessages)).toEqual(before.map((n) => `call_${n}`)); + + const resumed = await resumeAfterApproval( + { id: paused.approvalId!, approved: true }, + approvalStore, + toolRegistry, + mockModel(['all done']) + ); + + expect(resumed.finishReason).toBe('stop'); + // The approved call ran once; calls before it were not re-executed. + for (const name of [...before, names[position]]) expect(runs[name]).toBe(1); + // Calls after the approval call are not run on resume - today's + // behavior, tracked separately by LOU-U7. + for (const name of names.slice(position + 1)) expect(runs[name]).toBe(0); + }); + }); + + describe('toolConcurrency validation', () => { + it.each([0, -1, 1.5, Number.NaN, 'all'])('rejects %s with a clear error', async (value) => { + const options = { + agent: agentFor([]), + input: 'go', + provider: mockModel(['hi']), + toolConcurrency: value as ExecuteOptions['toolConcurrency'], + }; + await expect(AgentExecutor.execute(options)).rejects.toThrow( + /'toolConcurrency' must be a positive integer or 'unbounded'/ + ); + expect(() => + createAgent({ prompt: 'p', provider: mockModel([]), toolConcurrency: options.toolConcurrency }) + ).toThrow(/createAgent: 'toolConcurrency'/); + }); + + it('createAgent passes toolConcurrency through', async () => { + const log: string[] = []; + const a = controllableTool('a', log); + const agent = createAgent({ + prompt: 'p', + provider: mockModel([{ toolCalls: callsTo('a', 'b') }, 'done']), + tools: [a.tool, instantTool('b', log)], + toolConcurrency: 1, + }); + const sent = agent.send('go'); + await a.started; + await tick(); + expect(log).toEqual(['start:a']); + + a.release.resolve('A'); + await sent; + expect(log).toEqual(['start:a', 'end:a', 'start:b', 'end:b']); + }); + }); +}); diff --git a/src/execution/toolBatch.ts b/src/execution/toolBatch.ts new file mode 100644 index 000000000..b0492ca82 --- /dev/null +++ b/src/execution/toolBatch.ts @@ -0,0 +1,240 @@ +/** + * LOU-V3: runs the tool calls of one assistant turn concurrently, up to a + * limit, while keeping everything the transcript and resumption depend on + * deterministic. See `ExecuteOptions.toolConcurrency` for the public + * contract; this module is the scheduler behind it. + */ + +import type { ToolCall } from '../providers'; +import type { PreparedToolCall, ToolCallOutcome } from './toolCallExecution'; + +/** + * How many tool calls of one model turn may run at the same time: + * a positive integer, or `'unbounded'` (all of them). + */ +export type ToolConcurrency = number | 'unbounded'; + +/** + * Throws a descriptive error unless `value` is a valid {@link ToolConcurrency} + * (`undefined` means "use the default" and is accepted). + */ +export function assertToolConcurrency(value: unknown, caller: string): void { + if (value === undefined || value === 'unbounded') { + return; + } + if (typeof value === 'number' && Number.isInteger(value) && value >= 1) { + return; + } + throw new Error( + `${caller}: 'toolConcurrency' must be a positive integer or 'unbounded', got ${describe(value)}. ` + + "Use 1 to run a turn's tool calls one at a time, e.g. { toolConcurrency: 4 } or { toolConcurrency: 'unbounded' }." + ); +} + +function describe(value: unknown): string { + return typeof value === 'string' ? `'${value}'` : String(value); +} + +/** A tool call that has been started by {@link ToolBatchCallbacks.start}. */ +export interface StartedToolCall { + /** Settles once the call passed (or failed) its gate - see {@link PreparedToolCall}. */ + gate: Promise; + /** Settles with the call's outcome (or its fatal error). */ + done: Promise; +} + +/** How the batch talks back to the executor. */ +export interface ToolBatchCallbacks { + /** Starts one call (emitting its `tool-call` event first). */ + start(toolCall: ToolCall): StartedToolCall; + /** A call finished (completion order); never called for the approval call. */ + onComplete(outcome: ToolCallOutcome): void; + /** Appends a result to the transcript; called in call order. */ + record(toolCall: ToolCall, outcome: ToolCallOutcome): void; + /** Persists the transcript after the recorded prefix grew. */ + persist(): Promise; +} + +/** A call that was not appended to the transcript by the batch. */ +export interface UnrecordedToolCall { + toolCall: ToolCall; + /** Its outcome, when it completed (it may be the approval outcome). */ + outcome?: ToolCallOutcome; +} + +/** What happened to a batch, once no call of it is running any more. */ +export interface ToolBatchResult { + /** The first call (in call order) that needs approval; nothing after it ran. */ + approval?: { toolCall: ToolCall; outcome: ToolCallOutcome }; + /** The first fatal error, in call order (a `PropagatingToolError`, a throwing hook, ...). */ + failure?: { error: unknown }; + /** The calls, in order, whose results are not in the transcript. */ + unrecorded: UnrecordedToolCall[]; +} + +type Slot = + | { status: 'not-started' | 'running' } + | { status: 'completed'; outcome: ToolCallOutcome } + | { status: 'failed'; error: unknown }; + +/** Hands out at most `limit` slots; waiters are served first-come first-served. */ +class Limiter { + private active = 0; + private readonly waiters: Array<() => void> = []; + + constructor(private readonly limit: number) {} + + acquire(): Promise { + if (this.active < this.limit) { + this.active++; + return Promise.resolve(); + } + return new Promise((resolve) => this.waiters.push(resolve)); + } + + release(): void { + const next = this.waiters.shift(); + if (next) { + next(); + } else { + this.active--; + } + } +} + +/** + * Runs one turn's tool calls. Calls are started in call order, each one + * only after the previous one passed its gate (so the first call that needs + * approval is known before anything after it starts) and a concurrency slot + * is free. Results are recorded in call order as the completed prefix grows. + * Resolves only once every started call has settled. + */ +class ToolBatch { + private readonly slots: Slot[]; + private readonly limiter: Limiter; + private readonly running: Array> = []; + private recorded = 0; + private halted = false; + private persisting: Promise = Promise.resolve(); + + constructor( + private readonly toolCalls: ToolCall[], + concurrency: ToolConcurrency, + private readonly callbacks: ToolBatchCallbacks, + private readonly signal: AbortSignal | undefined + ) { + this.slots = toolCalls.map(() => ({ status: 'not-started' })); + this.limiter = new Limiter(concurrency === 'unbounded' ? Infinity : concurrency); + } + + async run(): Promise { + for (const [index, toolCall] of this.toolCalls.entries()) { + if (!(await this.acquireSlot())) { + break; + } + await this.launch(index, toolCall); + } + await Promise.all(this.running); + return this.result(); + } + + private stopped(): boolean { + return this.halted || Boolean(this.signal?.aborted); + } + + /** Waits for a free slot; false (slot returned) when the batch stopped meanwhile. */ + private async acquireSlot(): Promise { + if (this.stopped()) { + return false; + } + await this.limiter.acquire(); + if (this.stopped()) { + this.limiter.release(); + return false; + } + return true; + } + + /** Starts a call and waits for its gate; halts the batch at an approval or a gate error. */ + private async launch(index: number, toolCall: ToolCall): Promise { + this.slots[index] = { status: 'running' }; + const started = this.callbacks.start(toolCall); + this.running.push(this.track(index, started.done)); + try { + const prepared = await started.gate; + if (prepared.requiresApproval) { + this.halted = true; + } + } catch { + // The same error rejects `done`; track() records it. + this.halted = true; + } + } + + /** Records a call's settlement; never rejects. Frees its slot afterwards. */ + private async track(index: number, done: Promise): Promise { + try { + const outcome = await done; + this.slots[index] = { status: 'completed', outcome }; + if (!outcome.requiresApproval) { + this.callbacks.onComplete(outcome); + await this.recordPrefix(); + } + } catch (error) { + this.slots[index] = { status: 'failed', error }; + this.halted = true; + } finally { + this.limiter.release(); + } + } + + /** Appends every newly contiguous completed result, then persists once. */ + private async recordPrefix(): Promise { + const before = this.recorded; + for (let slot = this.slots[this.recorded]; isRecordable(slot); slot = this.slots[this.recorded]) { + this.callbacks.record(this.toolCalls[this.recorded], slot.outcome); + this.recorded++; + } + if (this.recorded === before) { + return; + } + // Checkpoint writes are serialized so an older snapshot can never land + // after a newer one. + const write = this.persisting.then(() => this.callbacks.persist()); + this.persisting = write.catch(() => undefined); + await write; + } + + private result(): ToolBatchResult { + const result: ToolBatchResult = { unrecorded: [] }; + for (const [index, slot] of this.slots.entries()) { + const toolCall = this.toolCalls[index]; + if (slot.status === 'failed' && !result.failure) { + result.failure = { error: slot.error }; + } + if (slot.status === 'completed' && slot.outcome.requiresApproval) { + result.approval = { toolCall, outcome: slot.outcome }; + } + if (index >= this.recorded) { + result.unrecorded.push({ toolCall, ...(slot.status === 'completed' && { outcome: slot.outcome }) }); + } + } + return result; + } +} + +function isRecordable( + slot: Slot | undefined +): slot is { status: 'completed'; outcome: ToolCallOutcome } { + return slot?.status === 'completed' && !slot.outcome.requiresApproval; +} + +/** Runs one turn's tool calls - see {@link ToolBatch}. */ +export function runToolBatch( + toolCalls: ToolCall[], + concurrency: ToolConcurrency, + callbacks: ToolBatchCallbacks, + signal?: AbortSignal +): Promise { + return new ToolBatch(toolCalls, concurrency, callbacks, signal).run(); +} diff --git a/src/execution/toolCallExecution.ts b/src/execution/toolCallExecution.ts index 6973234e8..b0613e965 100644 --- a/src/execution/toolCallExecution.ts +++ b/src/execution/toolCallExecution.ts @@ -75,48 +75,103 @@ function toolHookContext( }; } +/** + * A tool call that has passed its pre-execution gate - onToolCall, argument + * validation, pre-tool hooks and the `needsApproval` check - but has not run. + * LOU-V3: the executor gates the calls of a batch one at a time, in call + * order, so it knows whether a call needs approval before it starts the next. + */ +export interface PreparedToolCall { + toolCall: ToolCall; + /** The validated (and possibly hook-mutated) arguments `execute` will get. */ + args: Record; + /** Set when the call must not execute (invalid args, or `needsApproval` threw). */ + rejection?: ToolCallOutcome; + /** True when the call must wait for a human decision instead of running. */ + requiresApproval: boolean; +} + /** * Execute a tool call, wrapped in the onToolCall/onToolResult callbacks - * and pre/post tool-call hooks. + * and pre/post tool-call hooks. `onPrepared` (LOU-V3) is told the moment + * the call has passed its gate (see {@link PreparedToolCall}), before it runs. */ export async function runToolCall( toolCall: ToolCall, - ctx: ToolCallContext + ctx: ToolCallContext, + onPrepared?: (prepared: PreparedToolCall) => void ): Promise { + const prepared = await prepareToolCall(toolCall, ctx); + onPrepared?.(prepared); + return settleToolCall(prepared, ctx); +} + +/** Runs a tool call's gate: onToolCall, validation, pre-tool hooks, approval check. */ +async function prepareToolCall(toolCall: ToolCall, ctx: ToolCallContext): Promise { if (ctx.onToolCall) { await ctx.onToolCall(toolCall); } // Parse args up front (best-effort) so hooks get a real object to - // inspect/mutate even before doExecuteToolCall() parses them again for - // its own use (needsApproval/execute). A hook mutating this object has - // no effect on the actual call in this fallback case; see the - // `hooks.runPreToolCall` call below for the real, load-bearing parse. - // - // LOU-U4: the args are then validated against the tool's schema FIRST, - // so pre-tool hooks, `needsApproval` and `execute` all see the parsed - // (defaults/transforms applied) value. Invalid args skip the pre-hooks - // and `execute`; the structured error flows through the normal + // inspect/mutate. LOU-U4: the args are then validated against the tool's + // schema FIRST, so pre-tool hooks, `needsApproval` and `execute` all see + // the parsed (defaults/transforms applied) value. Invalid args skip the + // pre-hooks and `execute`; the structured error flows through the normal // error-outcome path (post hook, onToolResult, events, tracing). const checked = await checkToolArguments( toolCall, ctx.toolRegistry, parseToolArguments(toolCall, {}) ); - const hookArgs = checked.args; + if (checked.rejection) { + return { toolCall, args: checked.args, rejection: checked.rejection, requiresApproval: false }; + } + + // A `preToolCall` hook (e.g. redact-pii) may mutate `args` in place; that + // same object is what `needsApproval` and `execute` receive. + if (ctx.hooks) { + await ctx.hooks.runPreToolCall(toolHookContext(toolCall, ctx, checked.args)); + } + + const approval = await checkNeedsApproval(toolCall, ctx.toolRegistry, checked.args); + return { toolCall, args: checked.args, ...approval }; +} - if (ctx.hooks && !checked.rejection) { - await ctx.hooks.runPreToolCall(toolHookContext(toolCall, ctx, hookArgs)); +/** + * Resolves the called tool's `needsApproval`. A throwing `needsApproval` + * becomes the call's error outcome (a `PropagatingToolError` is rethrown). + */ +async function checkNeedsApproval( + toolCall: ToolCall, + toolRegistry: ToolRegistry | undefined, + args: Record +): Promise> { + const toolDesc = toolRegistry && findExecutableTool(toolRegistry, toolCall.function.name); + if (!toolDesc) { + return { requiresApproval: false }; + } + try { + return { requiresApproval: await resolveNeedsApproval(toolDesc, args) }; + } catch (error) { + return { requiresApproval: false, rejection: thrownToolFailure(toolCall, error) }; } +} +/** + * Runs a prepared tool call (or settles it as rejected / awaiting approval), + * followed by the post-tool hooks and onToolResult. + */ +async function settleToolCall( + prepared: PreparedToolCall, + ctx: ToolCallContext +): Promise { + const { toolCall, args: hookArgs } = prepared; const toolStart = Date.now(); let outcome: ToolCallOutcome | undefined; let thrown: unknown; try { - outcome = - checked.rejection ?? - (await doExecuteToolCall(toolCall, ctx.toolRegistry, ctx.sandbox, hookArgs, ctx.signal)); + outcome = await executePrepared(prepared, ctx); if (ctx.hooks) { await ctx.hooks.runPostToolCall(toolHookContext(toolCall, ctx, hookArgs), { result: outcome.result, @@ -136,6 +191,20 @@ export async function runToolCall( } } +/** The outcome of a prepared call: its rejection, its approval pause, or its actual run. */ +async function executePrepared( + prepared: PreparedToolCall, + ctx: ToolCallContext +): Promise { + if (prepared.rejection) { + return prepared.rejection; + } + if (prepared.requiresApproval) { + return approvalOutcome(prepared); + } + return doExecuteToolCall(prepared.toolCall, ctx.toolRegistry, ctx.sandbox, prepared.args, ctx.signal); +} + /** * Validates `rawArgs` against the called tool's schema. Returns the parsed * args, or a `rejection` outcome (structured error as `result`, message as @@ -195,6 +264,32 @@ async function resolveNeedsApproval( : !!toolDesc.needsApproval; } +/** The outcome of a call that is paused for a human approval decision. */ +function approvalOutcome(prepared: PreparedToolCall): ToolCallOutcome { + return { + toolCallId: prepared.toolCall.id, + toolName: prepared.toolCall.function.name, + result: null, + requiresApproval: true, + args: prepared.args, + }; +} + +/** + * The outcome for a thrown tool error. Errors that mark themselves as + * `PropagatingToolError` (e.g. DelegationDepthExceededError) are rethrown + * by toolErrorMessage() so they propagate out of execute() as a rejected + * promise instead of becoming a conversational {error} tool-result. + * LOU-U12: the model gets a structured error (not the string "null"), + * while `outcome.error` keeps the plain message for events/hooks. + */ +function thrownToolFailure(toolCall: ToolCall, error: unknown): ToolCallOutcome { + return { + ...toolFailure(toolCall, toolErrorMessage(error)), + result: toolErrorResult(toolCall.function.name, error), + }; +} + /** * Actual tool-execution logic, split out from runToolCall() so the * onToolCall/onToolResult hooks (LOU-E2) can wrap it uniformly via @@ -218,22 +313,12 @@ async function doExecuteToolCall( } // `overrideArgs` is the (possibly hook-mutated) object built by - // runToolCall() before preToolCall hooks ran - using it here + // prepareToolCall() before preToolCall hooks ran - using it here // instead of re-parsing `toolCall.function.arguments` is what makes a // `preToolCall` hook (e.g. redact-pii) that mutates `ctx.args` // actually affect what the tool is invoked with. const args = overrideArgs ?? JSON.parse(toolCall.function.arguments); - if (await resolveNeedsApproval(toolDesc, args)) { - return { - toolCallId: toolCall.id, - toolName: toolCall.function.name, - result: null, - requiresApproval: true, - args, - }; - } - // The 'ai' SDK tool.execute expects (args, context). Tools flagged // `requiresSandbox` (LOU-F5) are routed through the configured // SandboxAdapter instead of being invoked directly here; a tool @@ -255,15 +340,6 @@ async function doExecuteToolCall( result, }; } catch (error) { - // Errors that mark themselves as `PropagatingToolError` (e.g. - // DelegationDepthExceededError) are rethrown by toolErrorMessage() so - // they propagate out of execute() as a rejected promise instead of - // becoming a conversational {error} tool-result. - // LOU-U12: the model gets a structured error (not the string "null"), - // while `outcome.error` keeps the plain message for events/hooks. - return { - ...toolFailure(toolCall, toolErrorMessage(error)), - result: toolErrorResult(toolCall.function.name, error), - }; + return thrownToolFailure(toolCall, error); } }