From 7c19c3c06854d720eb7835b3faf977267d32f24d Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 16 Jul 2026 00:41:30 -0700 Subject: [PATCH] refactor(agents): share Responses tool-call tracking (#108721) --- packages/ai/src/internal/openai.ts | 1 + .../src/providers/openai-responses-shared.ts | 171 +++--------------- .../openai-responses-tool-call-tracker.ts | 148 +++++++++++++++ src/agents/openai-responses-transport.ts | 166 +++-------------- 4 files changed, 207 insertions(+), 279 deletions(-) create mode 100644 packages/ai/src/providers/openai-responses-tool-call-tracker.ts diff --git a/packages/ai/src/internal/openai.ts b/packages/ai/src/internal/openai.ts index c9c8b9372e94..3461d7df5c10 100644 --- a/packages/ai/src/internal/openai.ts +++ b/packages/ai/src/internal/openai.ts @@ -7,6 +7,7 @@ export * from "../providers/openai-prompt-cache.js"; export * from "../providers/openai-reasoning-effort.js"; export * from "../providers/openai-responses.js"; export * from "../providers/openai-responses-stream-compat.js"; +export * from "../providers/openai-responses-tool-call-tracker.js"; export * from "../providers/openai-stop-reason.js"; export * from "../providers/openai-tool-projection.js"; export * from "../providers/openai-tool-schema.js"; diff --git a/packages/ai/src/providers/openai-responses-shared.ts b/packages/ai/src/providers/openai-responses-shared.ts index 58641d1565c8..4ecbb0cfe5f2 100644 --- a/packages/ai/src/providers/openai-responses-shared.ts +++ b/packages/ai/src/providers/openai-responses-shared.ts @@ -57,6 +57,11 @@ import { isResponsesTextContentPartType, resolveResponsesMessageSnapshotCollapse, } from "./openai-responses-stream-compat.js"; +import { + createResponsesToolCallTracker, + readResponsesToolCallItemIdentity, + type ResponsesToolCallState, +} from "./openai-responses-tool-call-tracker.js"; import { convertResponsesToolPayload } from "./openai-responses-tools.js"; import { describeToolResultMediaPlaceholder, extractToolResultText } from "./tool-result-text.js"; import { transformMessages } from "./transform-messages.js"; @@ -692,14 +697,11 @@ export async function processResponsesStream( let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null = null; type StreamingToolCallBlock = ToolCall & { partialJson: string }; - type StreamingToolCallIdentity = { itemId?: string; callId?: string }; - type StreamingToolCallState = StreamingToolCallIdentity & { + type StreamingToolCallState = ResponsesToolCallState & { block: StreamingToolCallBlock; contentIndex: number; - argumentStreamReliable: boolean; }; - const toolCallsByOutputIndex = new Map(); - const unindexedToolCalls = new Set(); + const streamingToolCalls = createResponsesToolCallTracker(); let lastTextBlock: { block: TextContent; index: number; @@ -711,119 +713,10 @@ export async function processResponsesStream( let pendingMessageText: string | null = null; const blocks = output.content; const blockIndex = () => blocks.length - 1; - const readOutputIndex = (event: { output_index?: unknown }): number | undefined => - typeof event.output_index === "number" && - Number.isInteger(event.output_index) && - event.output_index >= 0 - ? event.output_index - : undefined; const readIdentityValue = (value: unknown): string | undefined => { const identity = typeof value === "string" ? value.trim() : ""; return identity || undefined; }; - const readEventToolCallIdentity = (event: { item_id?: unknown }): StreamingToolCallIdentity => ({ - itemId: readIdentityValue(event.item_id), - }); - const readItemToolCallIdentity = (item: { - id?: unknown; - call_id?: unknown; - }): StreamingToolCallIdentity => ({ - itemId: readIdentityValue(item.id), - callId: readIdentityValue(item.call_id), - }); - const identitiesConflict = ( - state: StreamingToolCallState, - identity: StreamingToolCallIdentity, - ): boolean => - Boolean( - (state.itemId && identity.itemId && state.itemId !== identity.itemId) || - (state.callId && identity.callId && state.callId !== identity.callId), - ); - const sharesIdentity = ( - state: StreamingToolCallState, - identity: StreamingToolCallIdentity, - ): boolean => - Boolean( - (state.itemId && identity.itemId && state.itemId === identity.itemId) || - (state.callId && identity.callId && state.callId === identity.callId), - ); - const adoptToolCallIdentity = ( - state: StreamingToolCallState, - identity: StreamingToolCallIdentity, - ): StreamingToolCallState => { - state.itemId ??= identity.itemId; - state.callId ??= identity.callId; - return state; - }; - const resolveCompatibleToolCall = ( - candidates: Iterable, - identity: StreamingToolCallIdentity, - ): StreamingToolCallState | undefined => { - const uniqueCandidates = [...new Set(candidates)]; - if (!identity.itemId && !identity.callId) { - return uniqueCandidates.length === 1 ? uniqueCandidates.at(0) : undefined; - } - const compatible = uniqueCandidates.filter((state) => !identitiesConflict(state, identity)); - const matches = compatible.filter((state) => sharesIdentity(state, identity)); - const matched = matches.length === 1 ? matches.at(0) : undefined; - if (matched) { - return adoptToolCallIdentity(matched, identity); - } - // Only a sole active call may adopt an identity it did not already know. - // Parallel calls require a positive match so missing indices stay fail-closed. - const soleCompatible = - uniqueCandidates.length === 1 && compatible.length === 1 && matches.length === 0 - ? compatible.at(0) - : undefined; - return soleCompatible ? adoptToolCallIdentity(soleCompatible, identity) : undefined; - }; - const resolveStreamingToolCall = ( - event: { output_index?: unknown; item_id?: unknown }, - identity: StreamingToolCallIdentity = readEventToolCallIdentity(event), - ): StreamingToolCallState | undefined => { - const outputIndex = readOutputIndex(event); - if (outputIndex !== undefined) { - const indexed = toolCallsByOutputIndex.get(outputIndex); - if (indexed) { - if (indexed.callId && identity.callId && indexed.callId !== identity.callId) { - return undefined; - } - // output_index owns routing once registered, but call_id stays stable; - // compatible providers may rotate item_id for the same output item. - return adoptToolCallIdentity(indexed, identity); - } - // A compatibility stream may add calls without indices, then start - // including them. Bind only the one identity-matched (or sole) candidate. - const unindexed = resolveCompatibleToolCall(unindexedToolCalls, identity); - if (unindexed) { - unindexedToolCalls.delete(unindexed); - toolCallsByOutputIndex.set(outputIndex, unindexed); - } - return unindexed; - } - - return resolveCompatibleToolCall( - [...toolCallsByOutputIndex.values(), ...unindexedToolCalls], - identity, - ); - }; - const forgetStreamingToolCall = (toolCall: StreamingToolCallState) => { - for (const [trackedIndex, tracked] of toolCallsByOutputIndex) { - if (tracked === toolCall) { - toolCallsByOutputIndex.delete(trackedIndex); - } - } - unindexedToolCalls.delete(toolCall); - }; - const markActiveToolCallArgumentsUnreliable = () => { - // An unrouteable argument event may belong to any active call. Only an - // authoritative full argument snapshot can recover that call. - for (const toolCall of new Set([...toolCallsByOutputIndex.values(), ...unindexedToolCalls])) { - toolCall.argumentStreamReliable = false; - } - }; - const hasActiveStreamingToolCall = () => - toolCallsByOutputIndex.size > 0 || unindexedToolCalls.size > 0; // Opening fragments may carry the only function name. A conflicting // completion must never retarget an already-started call. const resolveCompletedToolCallName = ( @@ -910,31 +803,24 @@ export async function processResponsesStream( stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); } } else if (item.type === "function_call") { - const outputIndex = readOutputIndex(event); - if (outputIndex !== undefined && toolCallsByOutputIndex.has(outputIndex)) { - throw new Error(`Responses stream reused active tool-call output index ${outputIndex}`); - } - currentItem = item; - currentBlock = { + const toolCallBlock: StreamingToolCallBlock = { type: "toolCall", id: resolveResponsesToolCallId(item), name: readIdentityValue(item.name) ?? "", arguments: {}, partialJson: item.arguments || "", }; - output.content.push(currentBlock); - const contentIndex = blockIndex(); - const toolCallState = { - block: currentBlock, + const contentIndex = output.content.length; + const toolCallState: StreamingToolCallState = { + block: toolCallBlock, contentIndex, argumentStreamReliable: true, - ...readItemToolCallIdentity(item), + ...readResponsesToolCallItemIdentity(item), }; - if (outputIndex !== undefined) { - toolCallsByOutputIndex.set(outputIndex, toolCallState); - } else { - unindexedToolCalls.add(toolCallState); - } + streamingToolCalls.register(event, toolCallState); + currentItem = item; + currentBlock = toolCallBlock; + output.content.push(toolCallBlock); stream.push({ type: "toolcall_start", contentIndex, partial: output }); } } else if (event.type === "response.reasoning_summary_part.added") { @@ -1057,7 +943,7 @@ export async function processResponsesStream( } } } else if (event.type === "response.function_call_arguments.delta") { - const toolCall = resolveStreamingToolCall(event); + const toolCall = streamingToolCalls.resolve(event); if (toolCall) { toolCall.block.partialJson += event.delta; toolCall.block.arguments = parseStreamingJson(toolCall.block.partialJson); @@ -1067,11 +953,11 @@ export async function processResponsesStream( delta: event.delta, partial: output, }); - } else if (hasActiveStreamingToolCall()) { - markActiveToolCallArgumentsUnreliable(); + } else if (streamingToolCalls.hasActive()) { + streamingToolCalls.markArgumentsUnreliable(); } } else if (event.type === "response.function_call_arguments.done") { - const toolCall = resolveStreamingToolCall(event); + const toolCall = streamingToolCalls.resolve(event); if (toolCall) { const previousPartialJson = toolCall.block.partialJson; const doneArguments = typeof event.arguments === "string" ? event.arguments : undefined; @@ -1096,8 +982,8 @@ export async function processResponsesStream( }); } } - } else if (hasActiveStreamingToolCall()) { - markActiveToolCallArgumentsUnreliable(); + } else if (streamingToolCalls.hasActive()) { + streamingToolCalls.markArgumentsUnreliable(); } } else if (event.type === "response.output_item.done") { const item = event.item; @@ -1176,10 +1062,13 @@ export async function processResponsesStream( } currentBlock = null; } else if (item.type === "function_call") { - const streamingToolCall = resolveStreamingToolCall(event, readItemToolCallIdentity(item)); + const streamingToolCall = streamingToolCalls.resolve( + event, + readResponsesToolCallItemIdentity(item), + ); // Do not turn an unresolved completion into a second public call while // an indexed call is still open. Its identity or index must match. - if (!streamingToolCall && hasActiveStreamingToolCall()) { + if (!streamingToolCall && streamingToolCalls.hasActive()) { continue; } const completedName = resolveCompletedToolCallName(streamingToolCall, item.name); @@ -1224,7 +1113,7 @@ export async function processResponsesStream( } if (streamingToolCall) { - forgetStreamingToolCall(streamingToolCall); + streamingToolCalls.forget(streamingToolCall); } if (currentBlock === toolCall) { currentBlock = null; @@ -1238,7 +1127,7 @@ export async function processResponsesStream( }); } } else if (event.type === "response.completed") { - if (hasActiveStreamingToolCall()) { + if (streamingToolCalls.hasActive()) { throw new Error("Responses stream completed with unresolved tool calls"); } const response = event.response; @@ -1289,7 +1178,7 @@ export async function processResponsesStream( throw new Error(msg); } } - if (hasActiveStreamingToolCall()) { + if (streamingToolCalls.hasActive()) { throw new Error("Responses stream ended with unresolved tool calls"); } } diff --git a/packages/ai/src/providers/openai-responses-tool-call-tracker.ts b/packages/ai/src/providers/openai-responses-tool-call-tracker.ts new file mode 100644 index 000000000000..87a2564c7e00 --- /dev/null +++ b/packages/ai/src/providers/openai-responses-tool-call-tracker.ts @@ -0,0 +1,148 @@ +export type ResponsesToolCallIdentity = { itemId?: string; callId?: string }; + +export type ResponsesToolCallState = ResponsesToolCallIdentity & { + argumentStreamReliable: boolean; +}; + +type ResponsesToolCallEvent = { + output_index?: unknown; + item_id?: unknown; +}; + +function readIdentityValue(value: unknown): string | undefined { + const identity = typeof value === "string" ? value.trim() : ""; + return identity || undefined; +} + +function readOutputIndex(event: ResponsesToolCallEvent): number | undefined { + return typeof event.output_index === "number" && + Number.isInteger(event.output_index) && + event.output_index >= 0 + ? event.output_index + : undefined; +} + +function readEventIdentity(event: ResponsesToolCallEvent): ResponsesToolCallIdentity { + return { itemId: readIdentityValue(event.item_id) }; +} + +export function readResponsesToolCallItemIdentity(item: { + id?: unknown; + call_id?: unknown; +}): ResponsesToolCallIdentity { + return { + itemId: readIdentityValue(item.id), + callId: readIdentityValue(item.call_id), + }; +} + +export function createResponsesToolCallTracker() { + const indexedCalls = new Map(); + const unindexedCalls = new Set(); + + const identitiesConflict = (state: TState, identity: ResponsesToolCallIdentity): boolean => + Boolean( + (state.itemId && identity.itemId && state.itemId !== identity.itemId) || + (state.callId && identity.callId && state.callId !== identity.callId), + ); + + const sharesIdentity = (state: TState, identity: ResponsesToolCallIdentity): boolean => + Boolean( + (state.itemId && identity.itemId && state.itemId === identity.itemId) || + (state.callId && identity.callId && state.callId === identity.callId), + ); + + const adoptIdentity = (state: TState, identity: ResponsesToolCallIdentity): TState => { + state.itemId ??= identity.itemId; + state.callId ??= identity.callId; + return state; + }; + + const resolveCompatible = ( + candidates: Iterable, + identity: ResponsesToolCallIdentity, + ): TState | undefined => { + const uniqueCandidates = [...new Set(candidates)]; + if (!identity.itemId && !identity.callId) { + return uniqueCandidates.length === 1 ? uniqueCandidates.at(0) : undefined; + } + const compatible = uniqueCandidates.filter((state) => !identitiesConflict(state, identity)); + const matches = compatible.filter((state) => sharesIdentity(state, identity)); + const matched = matches.length === 1 ? matches.at(0) : undefined; + if (matched) { + return adoptIdentity(matched, identity); + } + + // Only a sole active call may adopt an identity it did not already know. + // Parallel calls require a positive match so missing indices stay fail-closed. + const soleCompatible = + uniqueCandidates.length === 1 && compatible.length === 1 && matches.length === 0 + ? compatible.at(0) + : undefined; + return soleCompatible ? adoptIdentity(soleCompatible, identity) : undefined; + }; + + return { + register(event: ResponsesToolCallEvent, state: TState): void { + const outputIndex = readOutputIndex(event); + if (outputIndex === undefined) { + unindexedCalls.add(state); + return; + } + if (indexedCalls.has(outputIndex)) { + throw new Error(`Responses stream reused active tool-call output index ${outputIndex}`); + } + indexedCalls.set(outputIndex, state); + }, + + resolve( + event: ResponsesToolCallEvent, + identity: ResponsesToolCallIdentity = readEventIdentity(event), + ): TState | undefined { + const outputIndex = readOutputIndex(event); + if (outputIndex !== undefined) { + const indexed = indexedCalls.get(outputIndex); + if (indexed) { + if (indexed.callId && identity.callId && indexed.callId !== identity.callId) { + return undefined; + } + // output_index owns routing once registered, but call_id stays stable; + // compatible providers may rotate item_id for the same output item. + return adoptIdentity(indexed, identity); + } + + // A compatibility stream may add calls without indices, then start + // including them. Bind only the one identity-matched (or sole) candidate. + const unindexed = resolveCompatible(unindexedCalls, identity); + if (unindexed) { + unindexedCalls.delete(unindexed); + indexedCalls.set(outputIndex, unindexed); + } + return unindexed; + } + + return resolveCompatible([...indexedCalls.values(), ...unindexedCalls], identity); + }, + + forget(toolCall: TState): void { + for (const [outputIndex, tracked] of indexedCalls) { + if (tracked === toolCall) { + indexedCalls.delete(outputIndex); + } + } + unindexedCalls.delete(toolCall); + }, + + markArgumentsUnreliable(): void { + // An unrouteable argument event may belong to any active call. Only an + // authoritative full argument snapshot can recover that call. + for (const toolCall of new Set([...indexedCalls.values(), ...unindexedCalls])) { + toolCall.argumentStreamReliable = false; + } + }, + + hasActive(): boolean { + return indexedCalls.size > 0 || unindexedCalls.size > 0; + }, + }; +} diff --git a/src/agents/openai-responses-transport.ts b/src/agents/openai-responses-transport.ts index 90482711693a..87f1aff8b896 100644 --- a/src/agents/openai-responses-transport.ts +++ b/src/agents/openai-responses-transport.ts @@ -3,12 +3,14 @@ */ import { randomUUID } from "node:crypto"; import { + createResponsesToolCallTracker, isOpenAICompatibleAzureResponsesBaseUrl, isResponsesTextContentPartType, isResponsesTextDeltaEventType, normalizeOpenAIReasoningEffort, normalizeOpenAIStrictToolParameters, projectOpenAITools, + readResponsesToolCallItemIdentity, reconcileOpenAIResponsesToolChoice, resolveAzureDeploymentNameFromMap, resolveOpenAIReasoningEffortForModel, @@ -16,6 +18,7 @@ import { type OpenAIApiReasoningEffort, type OpenAIReasoningEffort, type OpenAIToolProjection, + type ResponsesToolCallState, } from "@openclaw/ai/internal/openai"; import { calculateCost, @@ -1226,14 +1229,11 @@ async function processResponsesStream( }; let currentItem: Record | null = null; let currentBlock: Record | null = null; - type StreamingToolCallIdentity = { itemId?: string; callId?: string }; - type StreamingToolCallState = StreamingToolCallIdentity & { + type StreamingToolCallState = ResponsesToolCallState & { block: Record; contentIndex: number; - argumentStreamReliable: boolean; }; - const toolCallsByOutputIndex = new Map(); - const unindexedToolCalls = new Set(); + const streamingToolCalls = createResponsesToolCallTracker(); let lastTextBlock: { block: Record; index: number; @@ -1248,116 +1248,10 @@ async function processResponsesStream( const eventTypes = new Map(); const sseDebugMode = resolveModelSseDebugMode(); const blockIndex = () => output.content.length - 1; - const readOutputIndex = (event: Record): number | undefined => - typeof event.output_index === "number" && - Number.isInteger(event.output_index) && - event.output_index >= 0 - ? event.output_index - : undefined; const readIdentityValue = (value: unknown): string | undefined => { const identity = typeof value === "string" ? value.trim() : ""; return identity || undefined; }; - const readEventToolCallIdentity = ( - event: Record, - ): StreamingToolCallIdentity => ({ itemId: readIdentityValue(event.item_id) }); - const readItemToolCallIdentity = (item: Record): StreamingToolCallIdentity => ({ - itemId: readIdentityValue(item.id), - callId: readIdentityValue(item.call_id), - }); - const identitiesConflict = ( - state: StreamingToolCallState, - identity: StreamingToolCallIdentity, - ): boolean => - Boolean( - (state.itemId && identity.itemId && state.itemId !== identity.itemId) || - (state.callId && identity.callId && state.callId !== identity.callId), - ); - const sharesIdentity = ( - state: StreamingToolCallState, - identity: StreamingToolCallIdentity, - ): boolean => - Boolean( - (state.itemId && identity.itemId && state.itemId === identity.itemId) || - (state.callId && identity.callId && state.callId === identity.callId), - ); - const adoptToolCallIdentity = ( - state: StreamingToolCallState, - identity: StreamingToolCallIdentity, - ): StreamingToolCallState => { - state.itemId ??= identity.itemId; - state.callId ??= identity.callId; - return state; - }; - const resolveCompatibleToolCall = ( - candidates: Iterable, - identity: StreamingToolCallIdentity, - ): StreamingToolCallState | undefined => { - const uniqueCandidates = [...new Set(candidates)]; - if (!identity.itemId && !identity.callId) { - return uniqueCandidates.length === 1 ? uniqueCandidates[0] : undefined; - } - const compatible = uniqueCandidates.filter((state) => !identitiesConflict(state, identity)); - const matches = compatible.filter((state) => sharesIdentity(state, identity)); - if (matches.length === 1) { - const match = matches.at(0); - return match ? adoptToolCallIdentity(match, identity) : undefined; - } - // Only a sole active call may adopt an identity it did not already know. - // Parallel calls require a positive match so missing indices stay fail-closed. - if (uniqueCandidates.length !== 1 || compatible.length !== 1 || matches.length !== 0) { - return undefined; - } - const candidate = compatible.at(0); - return candidate ? adoptToolCallIdentity(candidate, identity) : undefined; - }; - const resolveStreamingToolCall = ( - event: Record, - identity: StreamingToolCallIdentity = readEventToolCallIdentity(event), - ): StreamingToolCallState | undefined => { - const outputIndex = readOutputIndex(event); - if (outputIndex !== undefined) { - const indexed = toolCallsByOutputIndex.get(outputIndex); - if (indexed) { - if (indexed.callId && identity.callId && indexed.callId !== identity.callId) { - return undefined; - } - // output_index owns routing once registered, but call_id stays stable; - // compatible providers may rotate item_id for the same output item. - return adoptToolCallIdentity(indexed, identity); - } - // A compatibility stream may add calls without indices, then start - // including them. Bind only the one identity-matched (or sole) candidate. - const unindexed = resolveCompatibleToolCall(unindexedToolCalls, identity); - if (unindexed) { - unindexedToolCalls.delete(unindexed); - toolCallsByOutputIndex.set(outputIndex, unindexed); - } - return unindexed; - } - - return resolveCompatibleToolCall( - [...toolCallsByOutputIndex.values(), ...unindexedToolCalls], - identity, - ); - }; - const forgetStreamingToolCall = (toolCall: StreamingToolCallState) => { - for (const [trackedIndex, tracked] of toolCallsByOutputIndex) { - if (tracked === toolCall) { - toolCallsByOutputIndex.delete(trackedIndex); - } - } - unindexedToolCalls.delete(toolCall); - }; - const markActiveToolCallArgumentsUnreliable = () => { - // An unrouteable argument event may belong to any active call. Only an - // authoritative full argument snapshot can recover that call. - for (const toolCall of new Set([...toolCallsByOutputIndex.values(), ...unindexedToolCalls])) { - toolCall.argumentStreamReliable = false; - } - }; - const hasActiveStreamingToolCall = () => - toolCallsByOutputIndex.size > 0 || unindexedToolCalls.size > 0; // Opening fragments may carry the only function name. A conflicting // completion must never retarget an already-started call. const resolveCompletedToolCallName = ( @@ -1553,31 +1447,24 @@ async function processResponsesStream( stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); } } else if (item.type === "function_call") { - const outputIndex = readOutputIndex(event); - if (outputIndex !== undefined && toolCallsByOutputIndex.has(outputIndex)) { - throw new Error(`Responses stream reused active tool-call output index ${outputIndex}`); - } - currentItem = item; - currentBlock = { + const toolCallBlock: Record = { type: "toolCall", id: resolveToolCallId(item), name: readIdentityValue(item.name) ?? "", arguments: {}, partialJson: stringifyJsonLike(item.arguments), }; - output.content.push(currentBlock); - const contentIndex = blockIndex(); - const toolCallState = { - block: currentBlock, + const contentIndex = output.content.length; + const toolCallState: StreamingToolCallState = { + block: toolCallBlock, contentIndex, argumentStreamReliable: true, - ...readItemToolCallIdentity(item), + ...readResponsesToolCallItemIdentity(item), }; - if (outputIndex !== undefined) { - toolCallsByOutputIndex.set(outputIndex, toolCallState); - } else { - unindexedToolCalls.add(toolCallState); - } + streamingToolCalls.register(event, toolCallState); + currentItem = item; + currentBlock = toolCallBlock; + output.content.push(toolCallBlock); stream.push({ type: "toolcall_start", contentIndex, partial: output }); } } else if (type === "response.reasoning_summary_text.delta") { @@ -1604,7 +1491,7 @@ async function processResponsesStream( } } } else if (type === "response.function_call_arguments.delta") { - const toolCall = resolveStreamingToolCall(event); + const toolCall = streamingToolCalls.resolve(event); if (toolCall) { toolCall.block.partialJson = `${stringifyJsonLike(toolCall.block.partialJson)}${stringifyJsonLike(event.delta)}`; toolCall.block.arguments = parseStreamingJson( @@ -1616,11 +1503,11 @@ async function processResponsesStream( delta: stringifyJsonLike(event.delta), partial: output, }); - } else if (hasActiveStreamingToolCall()) { - markActiveToolCallArgumentsUnreliable(); + } else if (streamingToolCalls.hasActive()) { + streamingToolCalls.markArgumentsUnreliable(); } } else if (type === "response.function_call_arguments.done") { - const toolCall = resolveStreamingToolCall(event); + const toolCall = streamingToolCalls.resolve(event); if (toolCall) { const previousPartialJson = stringifyJsonLike(toolCall.block.partialJson); const doneArguments = typeof event.arguments === "string" ? event.arguments : undefined; @@ -1643,8 +1530,8 @@ async function processResponsesStream( }); } } - } else if (hasActiveStreamingToolCall()) { - markActiveToolCallArgumentsUnreliable(); + } else if (streamingToolCalls.hasActive()) { + streamingToolCalls.markArgumentsUnreliable(); } } else if (type === "response.output_item.done") { const item = event.item as Record; @@ -1745,10 +1632,13 @@ async function processResponsesStream( } currentBlock = null; } else if (item.type === "function_call") { - const streamingToolCall = resolveStreamingToolCall(event, readItemToolCallIdentity(item)); + const streamingToolCall = streamingToolCalls.resolve( + event, + readResponsesToolCallItemIdentity(item), + ); // Do not turn an unresolved completion into a second public call while // an indexed call is still open. Its identity or index must match. - if (!streamingToolCall && hasActiveStreamingToolCall()) { + if (!streamingToolCall && streamingToolCalls.hasActive()) { await cooperativeScheduler.afterEvent(); continue; } @@ -1802,7 +1692,7 @@ async function processResponsesStream( partial: output, }); if (streamingToolCall) { - forgetStreamingToolCall(streamingToolCall); + streamingToolCalls.forget(streamingToolCall); } if (currentBlock === toolCallBlock) { currentBlock = null; @@ -1810,7 +1700,7 @@ async function processResponsesStream( } } } else if (type === "response.completed") { - if (hasActiveStreamingToolCall()) { + if (streamingToolCalls.hasActive()) { throw new Error("Responses stream completed with unresolved tool calls"); } const response = event.response as Record | undefined; @@ -1879,7 +1769,7 @@ async function processResponsesStream( } await cooperativeScheduler.afterEvent(); } - if (hasActiveStreamingToolCall()) { + if (streamingToolCalls.hasActive()) { throw new Error("Responses stream ended with unresolved tool calls"); } const eventTypeSummary = [...eventTypes.entries()]