From 3e47eecad1cf9aa84fbcca61b3cb304af6109bea Mon Sep 17 00:00:00 2001 From: Dennis Jeong <3719829+w0nche0l@users.noreply.github.com> Date: Tue, 25 Aug 2026 01:55:30 -0400 Subject: [PATCH 1/2] fix(agent): broadcast tool.result per call as it settles, not after the round (DEV-1067) Co-authored-by: Cursor --- .changeset/tool-result-per-call-settle.md | 5 + packages/agent/src/lib/model-result.ts | 369 ++++++++++-------- .../unit/tool-round-settle-broadcast.test.ts | 312 +++++++++++++++ 3 files changed, 526 insertions(+), 160 deletions(-) create mode 100644 .changeset/tool-result-per-call-settle.md create mode 100644 packages/agent/tests/unit/tool-round-settle-broadcast.test.ts diff --git a/.changeset/tool-result-per-call-settle.md b/.changeset/tool-result-per-call-settle.md new file mode 100644 index 00000000..78e04ddf --- /dev/null +++ b/.changeset/tool-result-per-call-settle.md @@ -0,0 +1,5 @@ +--- +'@openrouter/agent': patch +--- + +Broadcast each tool call's `tool.result` / `tool.call_output` (and the legacy `tool_result`) the moment that call settles, instead of holding every broadcast until the whole round's `Promise.allSettled` resolves. With serially executing calls (e.g. `maxConcurrency: 1`), stream consumers now see each call complete live rather than all at once when the round ends. Model-facing `function_call_output` assembly is unchanged: outputs are still collected in call order, so follow-up request input and prompt-cache behavior are identical. (DEV-1067) diff --git a/packages/agent/src/lib/model-result.ts b/packages/agent/src/lib/model-result.ts index 6c1ce1f7..ce70b13d 100644 --- a/packages/agent/src/lib/model-result.ts +++ b/packages/agent/src/lib/model-result.ts @@ -219,6 +219,66 @@ interface RunBinding { task: () => ToolTask | null; } +/** + * Tagged outcome of one tool call in a round, produced by + * `executeSingleToolCall` and consumed by the settle-time finalizers in + * `executeToolRound`. `null` means the call is manual (client-executed) and + * yields no output this round. + */ +type SingleToolCallOutcome = + | null + | { + type: 'parse_error'; + output: models.FunctionCallOutputItem; + } + | { + type: 'hook_blocked'; + output: models.FunctionCallOutputItem; + } + | { + type: 'paused'; + toolCall: ParsedToolCall; + } + | { + type: 'async'; + toolCall: ParsedToolCall; + tool: Tool; + invocation: AsyncToolInvocation; + controller: AbortController; + timeoutMs: number | undefined; + runBinding: RunBinding; + preliminaryResultsForCall: InferToolEventsUnion[]; + } + | { + type: 'execution'; + toolCall: ParsedToolCall; + tool: Tool; + result: { + result: unknown; + error?: Error; + }; + preliminaryResultsForCall: InferToolEventsUnion[]; + }; + +/** + * A round call after settle-time finalization: broadcasts already emitted, + * model-facing output (if any) computed. `executeToolRound`'s ordered + * assembly loop only sorts these into the round's result arrays. + */ +type FinalizedRoundCall = + | { + kind: 'skipped'; + } + | { + kind: 'paused'; + toolCall: ParsedToolCall; + } + | { + kind: 'output'; + output: models.FunctionCallOutputItem; + deferredTask?: PendingAsyncTool; + }; + /** * Extract the call identity echoed on a server-tool output item, for * doom-loop fingerprinting. Server tools never pass through the client @@ -3631,41 +3691,7 @@ export class ModelResult< private async executeSingleToolCall( toolCall: ParsedToolCall, turnContext: TurnContext, - ): Promise< - | null - | { - type: 'parse_error'; - output: models.FunctionCallOutputItem; - } - | { - type: 'hook_blocked'; - output: models.FunctionCallOutputItem; - } - | { - type: 'paused'; - toolCall: ParsedToolCall; - } - | { - type: 'async'; - toolCall: ParsedToolCall; - tool: Tool; - invocation: AsyncToolInvocation; - controller: AbortController; - timeoutMs: number | undefined; - runBinding: RunBinding; - preliminaryResultsForCall: InferToolEventsUnion[]; - } - | { - type: 'execution'; - toolCall: ParsedToolCall; - tool: Tool; - result: { - result: unknown; - error?: Error; - }; - preliminaryResultsForCall: InferToolEventsUnion[]; - } - > { + ): Promise> { // Universal task-tool dispatch: ONE static tool ("task") handles every // check/steer/result/cancel interaction with running tasks, addressed // by taskId — the wire surface stays constant regardless of how many @@ -3944,153 +3970,176 @@ export class ModelResult< pausedCalls: ParsedToolCall[]; deferredTasks: PendingAsyncTool[]; }> { - const toolCallPromises = toolCalls.map((toolCall) => - this.executeSingleToolCall(toolCall, turnContext), + // Each call is finalized the moment ITS promise settles (DEV-1067): the + // tool.result / tool_result broadcasts and the tool.call_output event + // fire per call, so stream consumers see an early call complete while + // later calls in the round are still running. This also starts async + // invocations (grace-window race, registry tracking) at settle time + // rather than after the whole round, so N background calls never + // serialize into (N-1)×graceMs of stagger. + const finalizedPromises = toolCalls.map((toolCall) => + this.executeSingleToolCall(toolCall, turnContext).then( + (value) => this.finalizeSettledRoundCall(value), + (reason: unknown) => this.finalizeRejectedRoundCall(toolCall, reason), + ), ); + const finalizedCalls = await Promise.all(finalizedPromises); - const settledResults = await Promise.allSettled(toolCallPromises); + // Ordered assembly: broadcasts already happened per call above, but the + // MODEL-facing outputs are collected in call order so the follow-up + // request's input is deterministic (prompt-cache stability). const toolResults: models.FunctionCallOutputItem[] = []; const pausedCalls: ParsedToolCall[] = []; const deferredTasks: PendingAsyncTool[] = []; - - // Start ALL async invocations before consuming any outcome: the work - // (and its grace window) begins in handleAsyncInvocation, so awaiting - // it inside the ordered loop below would serialize N background calls - // into (N-1)×graceMs of stagger. Kicked off here in parallel; the - // ordered loop awaits the per-call promise, so OUTPUT order stays call - // order (prompt-cache stability). - const asyncOutcomes = new Map< - number, - Promise<{ - output: models.FunctionCallOutputItem; - deferredTask?: PendingAsyncTool; - }> - >(); - for (let i = 0; i < settledResults.length; i++) { - const settled = settledResults[i]; - if (settled?.status === 'fulfilled' && settled.value?.type === 'async') { - asyncOutcomes.set(i, this.handleAsyncInvocation(settled.value)); + for (const finalized of finalizedCalls) { + switch (finalized.kind) { + case 'skipped': + break; + case 'paused': + // HITL tool returned null — record the pause so the caller can + // break out of the outer loop before attempting a follow-up + // request with an incomplete set of outputs. The call will be + // surfaced via state (pendingToolCalls + status='awaiting_hitl') + // for manual resume. + pausedCalls.push(finalized.toolCall); + break; + case 'output': + toolResults.push(finalized.output); + if (finalized.deferredTask) { + deferredTasks.push(finalized.deferredTask); + } + break; + default: + finalized satisfies never; } } - for (let i = 0; i < settledResults.length; i++) { - const settled = settledResults[i]; - const originalToolCall = toolCalls[i]; - if (!settled || !originalToolCall) { - continue; - } - - if (settled.status === 'rejected') { - const errorMessage = - settled.reason instanceof Error ? settled.reason.message : String(settled.reason); + return { + toolResults, + pausedCalls, + deferredTasks, + }; + } - // `runToolWithHooks` is the single point of emission for PostToolUseFailure. - this.broadcastToolResult( - originalToolCall.id, - String(originalToolCall.name), - this.toolSourceByName(String(originalToolCall.name)), - { - error: errorMessage, - } as InferToolOutputsUnion, - ); + /** + * Finalize one settled tool call of a round, at the moment it settles: + * emit its tool.result / tool.call_output events and compute the + * model-facing output. Runs concurrently across the round's calls; + * `executeToolRound` re-orders the returned outputs into call order. + */ + private async finalizeSettledRoundCall( + value: SingleToolCallOutcome, + ): Promise { + // Manual (client-executed) tool — no output this round. + if (!value) { + return { + kind: 'skipped', + }; + } - const rejectedOutput: models.FunctionCallOutputItem = { - type: 'function_call_output' as const, - id: `output_${originalToolCall.id}`, - callId: originalToolCall.id, - output: JSON.stringify({ - error: errorMessage, - }), - }; - toolResults.push(rejectedOutput); - this.turnBroadcaster?.push({ - type: 'tool.call_output' as const, - output: rejectedOutput, - timestamp: Date.now(), - } satisfies ToolCallOutputEvent); - continue; - } + if (value.type === 'parse_error' || value.type === 'hook_blocked') { + this.pushToolCallOutputEvent(value.output); + return { + kind: 'output', + output: value.output, + }; + } - const value = settled.value; - if (!value) { - continue; - } + if (value.type === 'paused') { + return { + kind: 'paused', + toolCall: value.toolCall, + }; + } - if (value.type === 'parse_error' || value.type === 'hook_blocked') { - toolResults.push(value.output); - this.turnBroadcaster?.push({ - type: 'tool.call_output' as const, - output: value.output, - timestamp: Date.now(), - } satisfies ToolCallOutputEvent); - continue; - } + if (value.type === 'async') { + // Background / deferred: the call escapes the round (grace-window + // race, registry tracking, placeholder synthesis). + const asyncOutcome = await this.handleAsyncInvocation(value); + this.pushToolCallOutputEvent(asyncOutcome.output); + return { + kind: 'output', + output: asyncOutcome.output, + ...(asyncOutcome.deferredTask && { + deferredTask: asyncOutcome.deferredTask, + }), + }; + } - if (value.type === 'paused') { - // HITL tool returned null — record the pause so the caller can break - // out of the outer loop before attempting a follow-up request with an - // incomplete set of outputs. The call will be surfaced via state - // (pendingToolCalls + status='awaiting_hitl') for manual resume. - pausedCalls.push(value.toolCall); - continue; - } + const toolResult = ( + value.result.error + ? { + error: value.result.error.message, + } + : value.result.result + ) as InferToolOutputsUnion; + this.broadcastToolResult( + value.toolCall.id, + String(value.toolCall.name), + isMcpTool(value.tool) ? 'mcp' : 'client', + toolResult, + value.preliminaryResultsForCall.length > 0 ? value.preliminaryResultsForCall : undefined, + ); - if (value.type === 'async') { - // Background / deferred: the call escapes the round. Its handling - // (grace-window race, registry tracking, placeholder synthesis) - // was started above in parallel with the round's other async - // calls; awaiting here keeps outputs in call order. - const asyncOutcome = await (asyncOutcomes.get(i) ?? this.handleAsyncInvocation(value)); - toolResults.push(asyncOutcome.output); - this.turnBroadcaster?.push({ - type: 'tool.call_output' as const, - output: asyncOutcome.output, - timestamp: Date.now(), - } satisfies ToolCallOutputEvent); - if (asyncOutcome.deferredTask) { - deferredTasks.push(asyncOutcome.deferredTask); - } - continue; - } + const outputForModel = await this.computeToolOutputForModel(value); - const toolResult = ( - value.result.error - ? { - error: value.result.error.message, - } - : value.result.result - ) as InferToolOutputsUnion; - this.broadcastToolResult( - value.toolCall.id, - String(value.toolCall.name), - isMcpTool(value.tool) ? 'mcp' : 'client', - toolResult, - value.preliminaryResultsForCall.length > 0 ? value.preliminaryResultsForCall : undefined, - ); + const executedOutput: models.FunctionCallOutputItem = { + type: 'function_call_output' as const, + id: `output_${value.toolCall.id}`, + callId: value.toolCall.id, + output: outputForModel, + }; + this.pushToolCallOutputEvent(executedOutput); + return { + kind: 'output', + output: executedOutput, + }; + } - const outputForModel = await this.computeToolOutputForModel(value); + /** + * Finalize one REJECTED tool call of a round: synthesize the error output + * and emit its events at the moment the rejection settles. + */ + private finalizeRejectedRoundCall( + toolCall: ParsedToolCall, + reason: unknown, + ): FinalizedRoundCall { + const errorMessage = reason instanceof Error ? reason.message : String(reason); - const executedOutput: models.FunctionCallOutputItem = { - type: 'function_call_output' as const, - id: `output_${value.toolCall.id}`, - callId: value.toolCall.id, - output: outputForModel, - }; - toolResults.push(executedOutput); - this.turnBroadcaster?.push({ - type: 'tool.call_output' as const, - output: executedOutput, - timestamp: Date.now(), - } satisfies ToolCallOutputEvent); - } + // `runToolWithHooks` is the single point of emission for PostToolUseFailure. + this.broadcastToolResult( + toolCall.id, + String(toolCall.name), + this.toolSourceByName(String(toolCall.name)), + { + error: errorMessage, + } as InferToolOutputsUnion, + ); + const rejectedOutput: models.FunctionCallOutputItem = { + type: 'function_call_output' as const, + id: `output_${toolCall.id}`, + callId: toolCall.id, + output: JSON.stringify({ + error: errorMessage, + }), + }; + this.pushToolCallOutputEvent(rejectedOutput); return { - toolResults, - pausedCalls, - deferredTasks, + kind: 'output', + output: rejectedOutput, }; } + /** Push a `tool.call_output` event to the unified turn broadcaster. */ + private pushToolCallOutputEvent(output: models.FunctionCallOutputItem): void { + this.turnBroadcaster?.push({ + type: 'tool.call_output' as const, + output, + timestamp: Date.now(), + } satisfies ToolCallOutputEvent); + } + /** * Handle one async invocation from a round. * diff --git a/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts b/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts new file mode 100644 index 00000000..f828418b --- /dev/null +++ b/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts @@ -0,0 +1,312 @@ +import type { OpenRouterCore } from '@openrouter/sdk/core'; +import type * as models from '@openrouter/sdk/models'; +import { describe, expect, it, vi } from 'vitest'; +import { z } from 'zod/v4'; +import type { GetResponseOptions } from '../../src/lib/model-result.js'; +import { ModelResult } from '../../src/lib/model-result.js'; +import { tool } from '../../src/lib/tool.js'; +import type { Tool } from '../../src/lib/tool-types.js'; +import { isToolResultEvent } from '../../src/lib/tool-types.js'; + +type Internal = { + currentState: { + id: string; + messages: models.BaseInputsUnion[]; + status: 'in_progress'; + createdAt: number; + updatedAt: number; + } | null; + initPromise: Promise | null; + getInitialResponse: () => Promise; + makeFollowupRequest: ( + currentResponse: models.OpenResponsesResult, + toolResults: models.FunctionCallOutputItem[], + turnNumber: number, + ) => Promise; + shouldStopExecution: () => Promise; + executeToolsIfNeeded: () => Promise; + ensureTurnBroadcaster: () => { + createConsumer: () => AsyncIterableIterator; + push: (event: unknown) => void; + complete: () => void; + }; +}; + +function makeResponseWithToolCalls( + calls: Array<{ + id: string; + name: string; + arguments: string; + }>, +): models.OpenResponsesResult { + return { + id: 'resp_test', + object: 'response', + createdAt: 0, + model: 'test-model', + status: 'completed', + output: calls.map((c) => ({ + type: 'function_call' as const, + id: c.id, + callId: c.id, + name: c.name, + arguments: c.arguments, + status: 'completed' as const, + })), + usage: { + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + }, + } as unknown as models.OpenResponsesResult; +} + +function makeFinalResponse(): models.OpenResponsesResult { + return { + id: 'resp_final', + object: 'response', + createdAt: 0, + model: 'test-model', + status: 'completed', + output: [ + { + type: 'message', + id: 'msg_1', + role: 'assistant', + status: 'completed', + content: [ + { + type: 'output_text', + text: 'done', + }, + ], + }, + ], + usage: { + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + }, + } as unknown as models.OpenResponsesResult; +} + +function buildModelResult(tools: readonly Tool[]): Internal { + const config: GetResponseOptions = { + request: { + model: 'test-model', + input: 'hello', + }, + client: {} as OpenRouterCore, + tools, + }; + const internal = new ModelResult(config) as unknown as Internal; + internal.currentState = { + id: 'conv', + messages: [], + status: 'in_progress', + createdAt: 0, + updatedAt: 0, + }; + internal.initPromise = Promise.resolve(); + internal.shouldStopExecution = async () => false; + return internal; +} + +async function drainInto( + consumer: AsyncIterableIterator, + events: unknown[], +): Promise { + for await (const event of consumer) { + events.push(event); + } +} + +function isToolCallOutputEvent(event: unknown): event is { + type: 'tool.call_output'; + output: models.FunctionCallOutputItem; +} { + return ( + typeof event === 'object' && + event !== null && + 'type' in event && + event.type === 'tool.call_output' + ); +} + +/** + * DEV-1067 regression: when a round has multiple tool calls, each call's + * tool.result / tool.call_output must be broadcast the moment THAT call + * settles — not held until the whole round's Promise.allSettled resolves — + * while the model-facing outputs stay in call order. + */ +describe('per-call broadcast on settlement (DEV-1067)', () => { + it('broadcasts a fast call before a slow sibling in the same round finishes', async () => { + let releaseSlow: () => void = () => undefined; + const slowGate = new Promise((resolve) => { + releaseSlow = resolve; + }); + + const slow = tool({ + name: 'slow', + inputSchema: z.object({}), + outputSchema: z.object({ + done: z.boolean(), + }), + execute: async () => { + await slowGate; + return { + done: true, + }; + }, + }); + const fast = tool({ + name: 'fast', + inputSchema: z.object({}), + outputSchema: z.object({ + done: z.boolean(), + }), + execute: async () => ({ + done: true, + }), + }); + + const internal = buildModelResult([ + slow, + fast, + ]); + // Call order: slow FIRST, fast second — so the fast call settling early + // is only observable if broadcasts are per-call, and the ordered + // model-facing assembly is only correct if it re-sorts to call order. + internal.getInitialResponse = async () => + makeResponseWithToolCalls([ + { + id: 'call_slow', + name: 'slow', + arguments: '{}', + }, + { + id: 'call_fast', + name: 'fast', + arguments: '{}', + }, + ]); + + let followupToolResults: models.FunctionCallOutputItem[] | null = null; + internal.makeFollowupRequest = async (_currentResponse, toolResults) => { + followupToolResults = toolResults; + return makeFinalResponse(); + }; + + const broadcaster = internal.ensureTurnBroadcaster(); + const events: unknown[] = []; + const consumer = broadcaster.createConsumer(); + const drainPromise = drainInto(consumer, events); + + const executionPromise = internal.executeToolsIfNeeded(); + + // The fast call's completion must appear on the wire while the slow + // call is still running (its gate is unreleased). + await vi.waitFor(() => { + const toolResults = events.filter(isToolResultEvent); + expect(toolResults.some((e) => e.toolCallId === 'call_fast')).toBe(true); + }); + expect(events.filter(isToolResultEvent).some((e) => e.toolCallId === 'call_slow')).toBe(false); + const fastOutputs = events.filter(isToolCallOutputEvent); + expect(fastOutputs.some((e) => e.output.callId === 'call_fast')).toBe(true); + expect(fastOutputs.some((e) => e.output.callId === 'call_slow')).toBe(false); + + releaseSlow(); + await executionPromise; + broadcaster.complete(); + await drainPromise; + + // Both calls completed on the wire, fast before slow (settlement order). + const toolResultIds = events.filter(isToolResultEvent).map((e) => e.toolCallId); + expect(toolResultIds).toEqual([ + 'call_fast', + 'call_slow', + ]); + + // Model-facing outputs stay in CALL order regardless of settlement + // order (prompt-cache stability). + expect(followupToolResults).not.toBeNull(); + expect(followupToolResults?.map((o) => o.callId)).toEqual([ + 'call_slow', + 'call_fast', + ]); + }); + + it('broadcasts a rejected call as it settles without waiting for the round', async () => { + let releaseSlow: () => void = () => undefined; + const slowGate = new Promise((resolve) => { + releaseSlow = resolve; + }); + + const slow = tool({ + name: 'slow', + inputSchema: z.object({}), + outputSchema: z.object({ + done: z.boolean(), + }), + execute: async () => { + await slowGate; + return { + done: true, + }; + }, + }); + const boom = tool({ + name: 'boom', + inputSchema: z.object({}), + outputSchema: z.object({ + ok: z.boolean(), + }), + execute: async () => { + throw new Error('explode'); + }, + }); + + const internal = buildModelResult([ + slow, + boom, + ]); + internal.getInitialResponse = async () => + makeResponseWithToolCalls([ + { + id: 'call_slow', + name: 'slow', + arguments: '{}', + }, + { + id: 'call_boom', + name: 'boom', + arguments: '{}', + }, + ]); + internal.makeFollowupRequest = async () => makeFinalResponse(); + + const broadcaster = internal.ensureTurnBroadcaster(); + const events: unknown[] = []; + const consumer = broadcaster.createConsumer(); + const drainPromise = drainInto(consumer, events); + + const executionPromise = internal.executeToolsIfNeeded(); + + await vi.waitFor(() => { + const toolResults = events.filter(isToolResultEvent); + expect(toolResults.some((e) => e.toolCallId === 'call_boom')).toBe(true); + }); + const boomResult = events.filter(isToolResultEvent).find((e) => e.toolCallId === 'call_boom'); + expect(boomResult).toMatchObject({ + result: { + error: 'explode', + }, + }); + expect(events.filter(isToolResultEvent).some((e) => e.toolCallId === 'call_slow')).toBe(false); + + releaseSlow(); + await executionPromise; + broadcaster.complete(); + await drainPromise; + }); +}); From cc5477173ac146fa6a41a7a2596f9e506ed88413 Mon Sep 17 00:00:00 2001 From: Luke Parke <5702154+LukasParke@users.noreply.github.com> Date: Wed, 21 Oct 2026 17:58:47 +0000 Subject: [PATCH 2/2] fix(agent): settle all round calls before rejecting; order defer claims by call - executeToolRound awaits Promise.allSettled over the per-call finalizers and rethrows the first rejection in call order. Promise.all rejected the round as soon as one call's toModelOutput threw, orphaning still-running siblings (late broadcasts dropped, async calls registering after the run failed). - ctx.defer() registration waits on earlier calls' claim gates so rejectDuplicateTaskId picks the earlier call deterministically, as it did before per-call finalization. Only deferred calls wait. - Regression tests for both; changeset calls out completion-order events. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/tool-result-per-call-settle.md | 6 +- packages/agent/src/lib/model-result.ts | 62 ++++++- .../unit/tool-round-settle-broadcast.test.ts | 162 ++++++++++++++++++ 3 files changed, 222 insertions(+), 8 deletions(-) diff --git a/.changeset/tool-result-per-call-settle.md b/.changeset/tool-result-per-call-settle.md index 78e04ddf..27f6f77a 100644 --- a/.changeset/tool-result-per-call-settle.md +++ b/.changeset/tool-result-per-call-settle.md @@ -2,4 +2,8 @@ '@openrouter/agent': patch --- -Broadcast each tool call's `tool.result` / `tool.call_output` (and the legacy `tool_result`) the moment that call settles, instead of holding every broadcast until the whole round's `Promise.allSettled` resolves. With serially executing calls (e.g. `maxConcurrency: 1`), stream consumers now see each call complete live rather than all at once when the round ends. Model-facing `function_call_output` assembly is unchanged: outputs are still collected in call order, so follow-up request input and prompt-cache behavior are identical. (DEV-1067) +Broadcast each tool call's `tool.result` / `tool.call_output` (and the legacy `tool_result`) the moment that call settles, instead of holding every broadcast until the whole round's `Promise.allSettled` resolves. With serially executing calls (e.g. `maxConcurrency: 1`), stream consumers now see each call complete live rather than all at once when the round ends. (DEV-1067) + +**Behaviour change for stream consumers:** within a round, `tool.call_output`, `tool.result` and legacy `tool_result` events now arrive in **completion order**, not call order. Correlate them by `callId` / `toolCallId` rather than by position. Model-facing `function_call_output` items are still assembled in **call order**, so follow-up request input and prompt-cache behaviour don't change. + +If a call's `toModelOutput` throws, the round still waits for every sibling call to settle before it rejects with the first error in call order, so no broadcasts or async registrations land after the run has failed. When two `ctx.defer()` calls in the same round claim the same taskId, the earlier call in call order still wins. To make that deterministic, a deferred call's registration and placeholder broadcast wait for earlier calls in the round to settle. diff --git a/packages/agent/src/lib/model-result.ts b/packages/agent/src/lib/model-result.ts index 678dad41..ed0e6dbc 100644 --- a/packages/agent/src/lib/model-result.ts +++ b/packages/agent/src/lib/model-result.ts @@ -3963,13 +3963,57 @@ export class ModelResult< // invocations (grace-window race, registry tracking) at settle time // rather than after the whole round, so N background calls never // serialize into (N-1)×graceMs of stagger. - const finalizedPromises = toolCalls.map((toolCall) => - this.executeSingleToolCall(toolCall, turnContext).then( - (value) => this.finalizeSettledRoundCall(value), - (reason: unknown) => this.finalizeRejectedRoundCall(toolCall, reason), - ), - ); - const finalizedCalls = await Promise.all(finalizedPromises); + // + // Exception: `ctx.defer()` registration is ordered by CALL index. A + // deferred call's taskId is caller-supplied, and when two calls in the + // same round claim the same id, `rejectDuplicateTaskId` must pick the + // same winner (the earlier call) regardless of which settled first. So + // each call opens a "claim gate" that releases once the call has either + // settled as a non-deferred outcome or finished registering its + // deferred task, and a deferred call waits for the gates of every + // earlier call before registering. Only deferred calls ever wait; plain + // and background calls still finalize the moment they settle. + const claimGates: Promise[] = []; + const finalizedPromises = toolCalls.map((toolCall) => { + const earlierClaims = claimGates.slice(); + let releaseClaim!: () => void; + claimGates.push( + new Promise((resolve) => { + releaseClaim = resolve; + }), + ); + return this.executeSingleToolCall(toolCall, turnContext).then( + async (value) => { + try { + if (value?.type === 'async' && value.invocation.asyncMode === 'defer') { + await Promise.all(earlierClaims); + } else { + releaseClaim(); + } + return await this.finalizeSettledRoundCall(value); + } finally { + releaseClaim(); + } + }, + (reason: unknown) => { + releaseClaim(); + return this.finalizeRejectedRoundCall(toolCall, reason); + }, + ); + }); + // allSettled, not all: if one call's finalizer throws (e.g. its + // `toModelOutput` rejects), the round must still wait for its siblings + // before failing. Rejecting early would orphan them — their broadcasts + // would land after the run failed, and async calls could register + // background tasks on a run that has already errored. The first + // rejection in CALL order is rethrown so the error is deterministic. + const settledCalls = await Promise.allSettled(finalizedPromises); + const finalizedCalls = settledCalls.map((settled) => { + if (settled.status === 'rejected') { + throw settled.reason; + } + return settled.value; + }); // Ordered assembly: broadcasts already happened per call above, but the // MODEL-facing outputs are collected in call order so the follow-up @@ -4329,6 +4373,10 @@ export class ModelResult< * conversation would wait forever. Caller-supplied ids * (`ctx.defer(ticket.id)`) carry no uniqueness guarantee; reject the * collision at start time, when the tool can still see it. + * + * Within one round, `executeToolRound` orders deferred registration by + * call index, so when two calls claim the same id the EARLIER call wins + * regardless of which settled first. * @returns the error output for the colliding call, or null when unique. */ private rejectDuplicateTaskId( diff --git a/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts b/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts index f828418b..54b3c848 100644 --- a/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts +++ b/packages/agent/tests/unit/tool-round-settle-broadcast.test.ts @@ -309,4 +309,166 @@ describe('per-call broadcast on settlement (DEV-1067)', () => { broadcaster.complete(); await drainPromise; }); + + it("waits for siblings before rejecting when one call's toModelOutput throws", async () => { + let releaseSlow: () => void = () => undefined; + const slowGate = new Promise((resolve) => { + releaseSlow = resolve; + }); + let slowFinished = false; + + const slow = tool({ + name: 'slow', + inputSchema: z.object({}), + outputSchema: z.object({ + done: z.boolean(), + }), + execute: async () => { + await slowGate; + slowFinished = true; + return { + done: true, + }; + }, + }); + const fast = tool({ + name: 'fast', + inputSchema: z.object({}), + outputSchema: z.object({ + done: z.boolean(), + }), + execute: async () => ({ + done: true, + }), + toModelOutput: () => { + throw new Error('toModelOutput exploded'); + }, + }); + + const internal = buildModelResult([ + slow, + fast, + ]); + internal.getInitialResponse = async () => + makeResponseWithToolCalls([ + { + id: 'call_slow', + name: 'slow', + arguments: '{}', + }, + { + id: 'call_fast', + name: 'fast', + arguments: '{}', + }, + ]); + const followup = vi.fn(async () => makeFinalResponse()); + internal.makeFollowupRequest = followup; + + const broadcaster = internal.ensureTurnBroadcaster(); + const events: unknown[] = []; + const drainPromise = drainInto(broadcaster.createConsumer(), events); + + let settled = false; + let slowFinishedAtSettle = false; + const executionPromise = internal.executeToolsIfNeeded().finally(() => { + settled = true; + slowFinishedAtSettle = slowFinished; + }); + // Attach a handler now so the eventual rejection is never unhandled. + const outcome = executionPromise.then( + () => null, + (error: unknown) => error, + ); + + // The fast call settles (and its toModelOutput throws) while the slow + // call is still gated. + await vi.waitFor(() => { + expect(events.filter(isToolResultEvent).some((e) => e.toolCallId === 'call_fast')).toBe(true); + }); + // Give an early rejection (the Promise.all regression) time to surface. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(settled).toBe(false); + + releaseSlow(); + const error = await outcome; + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain('toModelOutput exploded'); + // The round rejected only after the slow sibling finished, and the slow + // call's broadcast landed before the failure. + expect(slowFinishedAtSettle).toBe(true); + expect(events.filter(isToolResultEvent).some((e) => e.toolCallId === 'call_slow')).toBe(true); + expect(followup).not.toHaveBeenCalled(); + + broadcaster.complete(); + await drainPromise; + }); + + it('resolves a duplicate ctx.defer() taskId in call order, not settlement order', async () => { + let releaseFirst: () => void = () => undefined; + const firstGate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const deferred = tool({ + name: 'deferred', + lifecycle: 'deferred', + inputSchema: z.object({ + gated: z.boolean(), + }), + outputSchema: z.object({ + ok: z.boolean(), + }), + run: async ({ gated }, ctx) => { + if (gated) { + await firstGate; + } + if (!ctx) { + throw new Error('run context missing'); + } + return ctx.defer('ticket_dup'); + }, + }); + + const internal = buildModelResult([ + deferred, + ]); + // call_first is FIRST in call order but settles LAST. + internal.getInitialResponse = async () => + makeResponseWithToolCalls([ + { + id: 'call_first', + name: 'deferred', + arguments: '{"gated":true}', + }, + { + id: 'call_second', + name: 'deferred', + arguments: '{"gated":false}', + }, + ]); + internal.makeFollowupRequest = async () => makeFinalResponse(); + + const broadcaster = internal.ensureTurnBroadcaster(); + const events: unknown[] = []; + const drainPromise = drainInto(broadcaster.createConsumer(), events); + + const executionPromise = internal.executeToolsIfNeeded(); + // Let call_second settle well before call_first is released. + await new Promise((resolve) => setTimeout(resolve, 20)); + releaseFirst(); + await executionPromise; + broadcaster.complete(); + await drainPromise; + + const outputs = new Map( + events.filter(isToolCallOutputEvent).map((e) => [ + e.output.callId, + String(e.output.output), + ]), + ); + // The earlier call wins the taskId; the later one is rejected. + expect(outputs.get('call_first')).toContain('"status":"pending"'); + expect(outputs.get('call_second')).toContain('already in use'); + }); });