diff --git a/README.md b/README.md index a64a0430e..88779f50b 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 - **Skills** — `defineSkill()` / `loadSkills('./skills')`: only each skill's name and description sit in the system prompt; the model loads the full markdown on demand through an auto-registered `load_skill` tool - **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 diff --git a/docs/api-overview.md b/docs/api-overview.md index 21b01b14b..383df6a3b 100644 --- a/docs/api-overview.md +++ b/docs/api-overview.md @@ -70,6 +70,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 2614e1cad..c3cdcb67c 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -23,6 +23,7 @@ import { ToolRegistry } from './tools/ToolRegistry'; import { ToolDescriptor } from './types'; import { modelFromEnv, resolveProviderSpec } from './providers/providerSpec'; import type { DefinedTool } from './tools/defineTool'; +import { ToolConcurrency, assertToolConcurrency } from './execution/toolBatch'; import type { Skill } from './skills/defineSkill'; /** @@ -53,6 +54,18 @@ export interface CreateAgentBase { 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; } /** @@ -151,6 +164,7 @@ export interface SimpleAgent { * const { text } = await agent.send('Hello!'); */ export function createAgent(config: CreateAgentConfig = {}): SimpleAgent { + assertToolConcurrency(config.toolConcurrency, 'createAgent'); const instructions = resolveInstructions(config); const provider = resolveModelSource(config); @@ -174,6 +188,7 @@ export function createAgent(config: CreateAgentConfig = {}): SimpleAgent { toolRegistry, skills: config.skills, maxSteps: config.maxSteps, + toolConcurrency: config.toolConcurrency, signal: options.signal, }); }, diff --git a/src/execution/AgentExecutor.ts b/src/execution/AgentExecutor.ts index bf8a1406b..555d6b0b7 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -15,7 +15,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, @@ -28,7 +40,8 @@ import { AgentRunState, addUsage, loadRunState, - pushCancelledToolResults, + pushAbortedBatchResults, + pushToolResult, saveStepCheckpoint, toExecutionResult, } from './agentRunState'; @@ -294,6 +307,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; } /** @@ -542,9 +589,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, @@ -563,53 +612,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. @@ -618,7 +685,8 @@ export class AgentExecutor { options: ExecuteOptions, state: AgentRunState, toolCall: ToolCall, - agentSpanId: string + agentSpanId: string, + onPrepared?: (prepared: PreparedToolCall) => void ): Promise { const { agent, @@ -649,7 +717,8 @@ export class AgentExecutor { hooks, sessionId, state.messages, - signal + signal, + onPrepared ); const parsedArgs = executed.args === undefined @@ -807,19 +876,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 + ); } /** @@ -849,6 +923,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); } }