From acb4bc040d34c4206b2b32a88f37c6e4b11f8032 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 28 Aug 2026 22:56:57 -0400 Subject: [PATCH 1/5] fix(ai): reconcile response calls by call id --- packages/ai/src/protocols/open-responses.ts | 17 +++-- .../provider/open-responses-lifecycle.test.ts | 70 +++++++++++++++++++ 2 files changed, 78 insertions(+), 9 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 79fec2d2b8bd..26ddf5ed3ef5 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -1118,12 +1118,8 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( if (state.completedTools.has(callID)) return [state, NO_EVENTS] satisfies StepResult const metadata = item.id !== undefined ? providerMetadata(state, { itemId: item.id }) : undefined const fallback = item.id ?? callID - // Match the pending tool by call id so item events that disagree on - // whether `item.id` is present still resolve the same call. - const registered = - state.tools[fallback] !== undefined - ? fallback - : Object.keys(state.tools).find((key) => state.tools[key]?.id === callID) + // Match by call id before the optional item id, which may change or collide. + const registered = Object.keys(state.tools).find((key) => state.tools[key]?.id === callID) const id = registered ?? fallback const tools = registered !== undefined @@ -1245,9 +1241,12 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( const events: LLMEvent[] = [] if (event.type === "response.completed") { for (const item of event.response?.output ?? []) { - const id = item.id ?? (item.type === "function_call" ? item.call_id : undefined) - if (id === undefined) continue - if (item.type !== "function_call" || !current.tools[id]) continue + if ( + item.type !== "function_call" || + !item.call_id || + !Object.values(current.tools).some((tool) => tool?.id === item.call_id) + ) + continue const [next, emitted] = yield* onOutputItemDone(current, item) current = next events.push(...emitted) diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts index dfd3c8e4a4ab..79214bb08744 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -344,6 +344,76 @@ describe("Open Responses basic-item lifecycles", () => { ]) }), ) + ;[ + { label: "introduced", firstID: undefined, finalID: "fc_1" }, + { label: "omitted", firstID: "fc_1", finalID: undefined }, + { label: "changed", firstID: "fc_old", finalID: "fc_new" }, + ].forEach((scenario) => { + it.effect(`reconciles a terminal call whose item id is ${scenario.label}`, () => + Effect.gen(function* () { + const first = { + type: "function_call", + ...(scenario.firstID === undefined ? {} : { id: scenario.firstID }), + call_id: "call_1", + name: "lookup", + } + const terminal = { + ...first, + ...(scenario.finalID === undefined ? { id: undefined } : { id: scenario.finalID }), + arguments: '{"query":"final"}', + } + const events = yield* collect( + { type: "response.output_item.added", item: first }, + { + type: "response.completed", + response: { id: "resp_1", output: [terminal, terminal] }, + }, + ) + const providerMetadata = + scenario.firstID === undefined ? undefined : { "openai-compatible": { itemId: scenario.firstID } } + expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([ + { type: "tool-input-start", id: "call_1", name: "lookup", providerMetadata }, + { type: "tool-input-end", id: "call_1", name: "lookup", providerMetadata }, + { type: "tool-call", id: "call_1", name: "lookup", input: { query: "final" }, providerMetadata }, + ]) + }), + ) + }) + + it.effect("does not reconcile an unseen terminal call through a colliding item id", () => + Effect.gen(function* () { + const events = yield* collect( + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: "{}" }, + }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [ + { + type: "function_call", + id: "fc_1", + call_id: "call_unseen", + name: "wrong", + arguments: '{"wrong":true}', + }, + ], + }, + }, + ) + expect(events.filter(LLMEvent.is.toolCall)).toEqual([ + { + type: "tool-call", + id: "call_1", + name: "lookup", + input: {}, + providerMetadata: { "openai-compatible": { itemId: "fc_1" } }, + }, + ]) + }), + ) it.effect("preserves call identity and pending order when an item id is reused", () => Effect.gen(function* () { From 169940d5cb5476b24a036f5491e41969a8025c39 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 28 Aug 2026 23:06:29 -0400 Subject: [PATCH 2/5] fix(ai): isolate colliding response calls --- packages/ai/src/protocols/open-responses.ts | 4 +- .../provider/open-responses-lifecycle.test.ts | 44 +++++++++++++++++++ 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 26ddf5ed3ef5..7f34d1b39456 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -1124,7 +1124,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( const tools = registered !== undefined ? state.tools - : ToolStream.start(state.tools, id, { + : ToolStream.start(ToolStream.empty(), id, { id: callID, name: item.name, providerMetadata: metadata, @@ -1149,7 +1149,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( hasFunctionCall: resultEvents.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)) || state.hasFunctionCall, - tools: result.tools, + tools: registered === undefined ? state.tools : result.tools, completedTools: new Set([...state.completedTools, callID]), }, events, diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts index 79214bb08744..87bfaa1c2c8b 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -302,6 +302,50 @@ describe("Open Responses basic-item lifecycles", () => { ) }) + it.effect("isolates a done-only call whose item id collides with a pending call", () => + Effect.gen(function* () { + const first = { type: "function_call", id: "fc_1", call_id: "call_1", name: "first" } + const second = { + type: "function_call", + id: "fc_1", + call_id: "call_2", + name: "second", + arguments: '{"second":true}', + } + const events = yield* collect( + { type: "response.output_item.added", item: first }, + { type: "response.function_call_arguments.delta", item_id: "fc_1", delta: '{"first":"draft"}' }, + { type: "response.output_item.done", item: second }, + { type: "response.output_item.done", item: second }, + { + type: "response.output_item.done", + item: { ...first, arguments: '{"first":"final"}' }, + }, + { + type: "response.output_item.done", + item: { ...first, arguments: '{"first":"duplicate"}' }, + }, + completed, + ) + const providerMetadata = { "openai-compatible": { itemId: "fc_1" } } + expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([ + { type: "tool-input-start", id: "call_1", name: "first", providerMetadata }, + { + type: "tool-input-delta", + id: "call_1", + name: "first", + text: '{"first":"draft"}', + input: { first: "draft" }, + }, + { type: "tool-input-start", id: "call_2", name: "second", providerMetadata }, + { type: "tool-input-end", id: "call_2", name: "second", providerMetadata }, + { type: "tool-call", id: "call_2", name: "second", input: { second: true }, providerMetadata }, + { type: "tool-input-end", id: "call_1", name: "first", providerMetadata }, + { type: "tool-call", id: "call_1", name: "first", input: { first: "final" }, providerMetadata }, + ]) + }), + ) + it.effect("recovers pending calls without reconciling terminal reasoning", () => Effect.gen(function* () { const events = yield* collect( From d877bccc2a2faf952006347f18db8943b2e962dc Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 28 Aug 2026 23:17:17 -0400 Subject: [PATCH 3/5] fix(ai): isolate response call accumulators --- packages/ai/src/protocols/open-responses.ts | 66 ++++++--- .../provider/open-responses-lifecycle.test.ts | 135 ++++++++++++------ 2 files changed, 140 insertions(+), 61 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 7f34d1b39456..5bc799c497c8 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -390,8 +390,13 @@ export interface ParserState { readonly id: string readonly name: string readonly providerMetadataKey: string + // Pending calls use generated keys. Wire call ids, item ids, and output + // indexes are aliases only and can therefore collide without sharing state. readonly tools: ToolStream.State - // Call ids stay independent of item ids, which may be omitted or reused. + readonly toolCalls: Readonly> + readonly toolItems: Readonly>> + readonly toolOutputs: Readonly> + readonly nextTool: number readonly completedTools: ReadonlySet readonly hasFunctionCall: boolean readonly lifecycle: Lifecycle.State @@ -863,6 +868,19 @@ const joinReasoningText = (parts: ReadonlyArray) => { export const outputItemID = (state: ParserState, event: Event) => event.output_index === undefined ? event.item_id : (state.outputItems[event.output_index] ?? event.item_id) +const pendingToolID = (state: ParserState, event: Event) => { + // Output position is the exact streaming identity. Item ids are aliases and + // only resolve when unambiguous; never guess from a colliding call id. + if (event.output_index !== undefined) { + const id = state.toolOutputs[event.output_index] + if (id !== undefined && state.tools[id] !== undefined) return id + } + if (event.item_id === undefined) return undefined + const active = (state.toolItems[event.item_id] ?? []).filter((id) => state.tools[id] !== undefined) + if (active.length > 1) return null + return active[0] +} + const startReasoningSummaryPart = (state: ParserState, itemID: string, index: number): StepResult => { const item = state.reasoningItems[itemID] if (!item?.open || index === 0 || item.summaryParts[index] !== undefined) return [state, NO_EVENTS] @@ -1000,9 +1018,9 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { ] } if (item?.type !== "function_call" || !item.call_id) return [state, NO_EVENTS] - const id = item.id ?? item.call_id - if (Object.values(state.tools).some((tool) => tool?.id === item.call_id) || state.completedTools.has(item.call_id)) - return [state, NO_EVENTS] + if (state.toolCalls[item.call_id] !== undefined || state.completedTools.has(item.call_id)) return [state, NO_EVENTS] + const id = `tool:${state.nextTool}` + const itemID = item.id ?? item.call_id const metadata = item.id !== undefined ? providerMetadata(state, { itemId: item.id }) : undefined const events: LLMEvent[] = [] const lifecycle = Lifecycle.stepStart(state.lifecycle, events) @@ -1016,6 +1034,11 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { input: item.arguments ?? "", providerMetadata: metadata, }), + toolCalls: { ...state.toolCalls, [item.call_id]: id }, + toolItems: { ...state.toolItems, [itemID]: [...(state.toolItems[itemID] ?? []), id] }, + toolOutputs: + event.output_index === undefined ? state.toolOutputs : { ...state.toolOutputs, [event.output_index]: id }, + nextTool: state.nextTool + 1, }, [...events, LLMEvent.toolInputStart({ id: item.call_id, name: item.name ?? "", providerMetadata: metadata })], ] @@ -1053,15 +1076,22 @@ const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgu state: ParserState, event: Event, ) { - if (event.item_id === undefined) return [state, NO_EVENTS] satisfies StepResult - const tool = state.tools[event.item_id] + const id = pendingToolID(state, event) + if (id === null) + return yield* ProviderShared.eventError( + state.id, + `${state.name} tool argument event has an ambiguous item_id without a matching output_index`, + ProviderShared.encodeJson(event), + ) + if (id === undefined) return [state, NO_EVENTS] satisfies StepResult + const tool = state.tools[id] if (!tool) return [state, NO_EVENTS] satisfies StepResult const final = event.type === "response.function_call_arguments.done" ? event.arguments : undefined if (event.type === "response.function_call_arguments.done" && final === undefined) return [state, NO_EVENTS] satisfies StepResult if (final !== undefined && !final.startsWith(tool.input)) return [ - { ...state, tools: ToolStream.start(state.tools, event.item_id, { ...tool, input: final }) }, + { ...state, tools: ToolStream.start(state.tools, id, { ...tool, input: final }) }, NO_EVENTS, ] satisfies StepResult const delta = final === undefined ? event.delta : final.slice(tool.input.length) @@ -1069,7 +1099,7 @@ const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgu const result = ToolStream.appendExisting( state.id, state.tools, - event.item_id, + id, delta, `${state.name} tool argument delta is missing its tool call`, ) @@ -1117,10 +1147,9 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( const callID = item.call_id if (state.completedTools.has(callID)) return [state, NO_EVENTS] satisfies StepResult const metadata = item.id !== undefined ? providerMetadata(state, { itemId: item.id }) : undefined - const fallback = item.id ?? callID - // Match by call id before the optional item id, which may change or collide. - const registered = Object.keys(state.tools).find((key) => state.tools[key]?.id === callID) - const id = registered ?? fallback + const admitted = state.toolCalls[callID] + const registered = admitted !== undefined && state.tools[admitted] !== undefined ? admitted : undefined + const id = registered ?? callID const tools = registered !== undefined ? state.tools @@ -1241,12 +1270,9 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( const events: LLMEvent[] = [] if (event.type === "response.completed") { for (const item of event.response?.output ?? []) { - if ( - item.type !== "function_call" || - !item.call_id || - !Object.values(current.tools).some((tool) => tool?.id === item.call_id) - ) - continue + if (item.type !== "function_call" || !item.call_id) continue + const id = current.toolCalls[item.call_id] + if (id === undefined || current.tools[id] === undefined) continue const [next, emitted] = yield* onOutputItemDone(current, item) current = next events.push(...emitted) @@ -1415,6 +1441,10 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse providerMetadataKey: request.model.route.providerMetadataKey ?? "openresponses", hasFunctionCall: false, tools: ToolStream.empty(), + toolCalls: {}, + toolItems: {}, + toolOutputs: {}, + nextTool: 0, completedTools: new Set(), lifecycle: Lifecycle.initial(), outputItems: {}, diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts index 87bfaa1c2c8b..74bbcfd2c2b1 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -301,50 +301,99 @@ describe("Open Responses basic-item lifecycles", () => { }), ) }) - - it.effect("isolates a done-only call whose item id collides with a pending call", () => - Effect.gen(function* () { - const first = { type: "function_call", id: "fc_1", call_id: "call_1", name: "first" } - const second = { - type: "function_call", - id: "fc_1", - call_id: "call_2", - name: "second", - arguments: '{"second":true}', - } - const events = yield* collect( - { type: "response.output_item.added", item: first }, - { type: "response.function_call_arguments.delta", item_id: "fc_1", delta: '{"first":"draft"}' }, - { type: "response.output_item.done", item: second }, - { type: "response.output_item.done", item: second }, - { - type: "response.output_item.done", - item: { ...first, arguments: '{"first":"final"}' }, - }, - { - type: "response.output_item.done", - item: { ...first, arguments: '{"first":"duplicate"}' }, - }, - completed, - ) - const providerMetadata = { "openai-compatible": { itemId: "fc_1" } } - expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([ - { type: "tool-input-start", id: "call_1", name: "first", providerMetadata }, - { - type: "tool-input-delta", - id: "call_1", + ;[ + { + label: "same item id with direct completion", + firstID: "shared", + firstCallID: "call_1", + secondID: "shared", + completion: "direct", + reverse: false, + }, + { + label: "same item id with reversed response completion", + firstID: "shared", + firstCallID: "call_1", + secondID: "shared", + completion: "response", + reverse: true, + }, + { + label: "call id matching another item id with reversed direct completion", + firstID: "fc_1", + firstCallID: "shared", + secondID: "shared", + completion: "direct", + reverse: true, + }, + { + label: "call id matching another item id with response completion", + firstID: "fc_1", + firstCallID: "shared", + secondID: "shared", + completion: "response", + reverse: false, + }, + ].forEach((fixture) => { + it.effect(`isolates pending calls with ${fixture.label}`, () => + Effect.gen(function* () { + const first = { + type: "function_call", + id: fixture.firstID, + call_id: fixture.firstCallID, name: "first", - text: '{"first":"draft"}', - input: { first: "draft" }, - }, - { type: "tool-input-start", id: "call_2", name: "second", providerMetadata }, - { type: "tool-input-end", id: "call_2", name: "second", providerMetadata }, - { type: "tool-call", id: "call_2", name: "second", input: { second: true }, providerMetadata }, - { type: "tool-input-end", id: "call_1", name: "first", providerMetadata }, - { type: "tool-call", id: "call_1", name: "first", input: { first: "final" }, providerMetadata }, - ]) - }), - ) + } + const second = { type: "function_call", id: fixture.secondID, call_id: "call_2", name: "second" } + const finished = [ + { ...first, arguments: '{"first":"final"}' }, + { ...second, arguments: '{"second":"final"}' }, + ] + const ordered = fixture.reverse ? finished.toReversed() : finished + const terminal: OpenResponses.Event[] = + fixture.completion === "direct" + ? [...ordered, ...ordered].map((item) => ({ type: "response.output_item.done", item })) + : [ + { + type: "response.completed", + response: { id: "resp_1", output: [...ordered, ...ordered] }, + }, + ] + const events = yield* collect( + { type: "response.output_item.added", output_index: 0, item: first }, + { type: "response.output_item.added", output_index: 1, item: second }, + { + type: "response.function_call_arguments.delta", + output_index: 1, + item_id: fixture.firstID, + delta: '{"second":"draft"}', + }, + { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: fixture.secondID, + delta: '{"first":"draft"}', + }, + ...terminal, + ...(fixture.completion === "direct" ? [completed] : []), + ) + ;[ + { item: first, input: { first: "final" } }, + { item: second, input: { second: "final" } }, + ].forEach((expected) => { + expect( + events.filter((event) => "id" in event && event.id === expected.item.call_id).map((event) => event.type), + ).toEqual(["tool-input-start", "tool-input-delta", "tool-input-end", "tool-call"]) + expect(events.filter(LLMEvent.is.toolCall).find((event) => event.id === expected.item.call_id)).toEqual({ + type: "tool-call", + id: expected.item.call_id, + name: expected.item.name, + input: expected.input, + providerMetadata: { "openai-compatible": { itemId: expected.item.id } }, + }) + }) + }), + ) + }) it.effect("recovers pending calls without reconciling terminal reasoning", () => Effect.gen(function* () { From 13277cbcf83f7f1c01bdf2aa51915f82793ff51e Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 28 Aug 2026 23:28:16 -0400 Subject: [PATCH 4/5] fix(ai): tombstone response call identities --- packages/ai/src/protocols/open-responses.ts | 32 ++--- .../provider/open-responses-lifecycle.test.ts | 111 ++++++++++++++++++ 2 files changed, 127 insertions(+), 16 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 5bc799c497c8..0c9f4f8865b1 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -393,9 +393,9 @@ export interface ParserState { // Pending calls use generated keys. Wire call ids, item ids, and output // indexes are aliases only and can therefore collide without sharing state. readonly tools: ToolStream.State - readonly toolCalls: Readonly> - readonly toolItems: Readonly>> - readonly toolOutputs: Readonly> + readonly toolCalls: ReadonlyMap + readonly toolItems: ReadonlyMap> + readonly toolOutputs: ReadonlyMap readonly nextTool: number readonly completedTools: ReadonlySet readonly hasFunctionCall: boolean @@ -871,12 +871,12 @@ export const outputItemID = (state: ParserState, event: Event) => const pendingToolID = (state: ParserState, event: Event) => { // Output position is the exact streaming identity. Item ids are aliases and // only resolve when unambiguous; never guess from a colliding call id. - if (event.output_index !== undefined) { - const id = state.toolOutputs[event.output_index] - if (id !== undefined && state.tools[id] !== undefined) return id + if (event.output_index !== undefined && state.toolOutputs.has(event.output_index)) { + const id = state.toolOutputs.get(event.output_index) + return id !== undefined && state.tools[id] !== undefined ? id : undefined } if (event.item_id === undefined) return undefined - const active = (state.toolItems[event.item_id] ?? []).filter((id) => state.tools[id] !== undefined) + const active = (state.toolItems.get(event.item_id) ?? []).filter((id) => state.tools[id] !== undefined) if (active.length > 1) return null return active[0] } @@ -1018,7 +1018,7 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { ] } if (item?.type !== "function_call" || !item.call_id) return [state, NO_EVENTS] - if (state.toolCalls[item.call_id] !== undefined || state.completedTools.has(item.call_id)) return [state, NO_EVENTS] + if (state.toolCalls.has(item.call_id) || state.completedTools.has(item.call_id)) return [state, NO_EVENTS] const id = `tool:${state.nextTool}` const itemID = item.id ?? item.call_id const metadata = item.id !== undefined ? providerMetadata(state, { itemId: item.id }) : undefined @@ -1034,10 +1034,10 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { input: item.arguments ?? "", providerMetadata: metadata, }), - toolCalls: { ...state.toolCalls, [item.call_id]: id }, - toolItems: { ...state.toolItems, [itemID]: [...(state.toolItems[itemID] ?? []), id] }, + toolCalls: new Map(state.toolCalls).set(item.call_id, id), + toolItems: new Map(state.toolItems).set(itemID, [...(state.toolItems.get(itemID) ?? []), id]), toolOutputs: - event.output_index === undefined ? state.toolOutputs : { ...state.toolOutputs, [event.output_index]: id }, + event.output_index === undefined ? state.toolOutputs : new Map(state.toolOutputs).set(event.output_index, id), nextTool: state.nextTool + 1, }, [...events, LLMEvent.toolInputStart({ id: item.call_id, name: item.name ?? "", providerMetadata: metadata })], @@ -1147,7 +1147,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( const callID = item.call_id if (state.completedTools.has(callID)) return [state, NO_EVENTS] satisfies StepResult const metadata = item.id !== undefined ? providerMetadata(state, { itemId: item.id }) : undefined - const admitted = state.toolCalls[callID] + const admitted = state.toolCalls.get(callID) const registered = admitted !== undefined && state.tools[admitted] !== undefined ? admitted : undefined const id = registered ?? callID const tools = @@ -1271,7 +1271,7 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( if (event.type === "response.completed") { for (const item of event.response?.output ?? []) { if (item.type !== "function_call" || !item.call_id) continue - const id = current.toolCalls[item.call_id] + const id = current.toolCalls.get(item.call_id) if (id === undefined || current.tools[id] === undefined) continue const [next, emitted] = yield* onOutputItemDone(current, item) current = next @@ -1441,9 +1441,9 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse providerMetadataKey: request.model.route.providerMetadataKey ?? "openresponses", hasFunctionCall: false, tools: ToolStream.empty(), - toolCalls: {}, - toolItems: {}, - toolOutputs: {}, + toolCalls: new Map(), + toolItems: new Map(), + toolOutputs: new Map(), nextTool: 0, completedTools: new Set(), lifecycle: Lifecycle.initial(), diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts index 74bbcfd2c2b1..356d451bd070 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -395,6 +395,117 @@ describe("Open Responses basic-item lifecycles", () => { ) }) + it.effect("tombstones a completed output index instead of resolving its late events through a shared item id", () => + Effect.gen(function* () { + const first = { type: "function_call", id: "shared", call_id: "call_1", name: "first" } + const second = { type: "function_call", id: "shared", call_id: "call_2", name: "second" } + const events = yield* collect( + { type: "response.output_item.added", output_index: 0, item: first }, + { type: "response.output_item.added", output_index: 1, item: second }, + { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: "shared", + delta: '{"first":"draft"}', + }, + { + type: "response.function_call_arguments.delta", + output_index: 1, + item_id: "shared", + delta: '{"second":"kept"}', + }, + { + type: "response.output_item.done", + output_index: 0, + item: { ...first, arguments: '{"first":"final"}' }, + }, + { type: "response.output_item.added", output_index: 0, item: first }, + { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: "shared", + delta: '{"wrong":"delta"}', + }, + { + type: "response.function_call_arguments.done", + output_index: 0, + item_id: "shared", + arguments: '{"wrong":"done"}', + }, + { + type: "response.output_item.done", + output_index: 0, + item: { ...first, arguments: '{"wrong":"item"}' }, + }, + { type: "response.output_item.done", output_index: 1, item: second }, + completed, + ) + expect(events.filter(LLMEvent.is.toolCall)).toEqual([ + { + type: "tool-call", + id: "call_1", + name: "first", + input: { first: "final" }, + providerMetadata: { "openai-compatible": { itemId: "shared" } }, + }, + { + type: "tool-call", + id: "call_2", + name: "second", + input: { second: "kept" }, + providerMetadata: { "openai-compatible": { itemId: "shared" } }, + }, + ]) + ;["call_1", "call_2"].forEach((id) => { + expect(events.filter((event) => "id" in event && event.id === id).map((event) => event.type)).toEqual([ + "tool-input-start", + "tool-input-delta", + "tool-input-end", + "tool-call", + ]) + }) + }), + ) + + it.effect("treats prototype property names as ordinary call and item ids", () => + Effect.gen(function* () { + const items = [ + { type: "function_call", id: "__proto__", call_id: "toString", name: "first" }, + { type: "function_call", id: "constructor", call_id: "__proto__", name: "second" }, + { type: "function_call", id: "toString", call_id: "constructor", name: "third" }, + ] + const streamed: OpenResponses.Event[] = items.flatMap((item, output_index) => [ + { type: "response.output_item.added", output_index, item }, + { + type: "response.function_call_arguments.delta", + item_id: item.id, + delta: `{"value":"${item.name}"}`, + }, + ]) + const events = yield* collect(...streamed, { + type: "response.completed", + response: { + id: "resp_1", + output: items.toReversed().map((item) => ({ ...item, arguments: `{"value":"${item.name}"}` })), + }, + }) + expect(events.filter(LLMEvent.is.toolCall)).toEqual( + items.toReversed().map((item) => ({ + type: "tool-call", + id: item.call_id, + name: item.name, + input: { value: item.name }, + providerMetadata: { "openai-compatible": { itemId: item.id } }, + })), + ) + items.forEach((item) => { + expect(events.filter((event) => "id" in event && event.id === item.call_id).map((event) => event.type)).toEqual( + ["tool-input-start", "tool-input-delta", "tool-input-end", "tool-call"], + ) + }) + }), + ) + it.effect("recovers pending calls without reconciling terminal reasoning", () => Effect.gen(function* () { const events = yield* collect( From 9933c73a77f0570322dab98660a8176e26b019dd Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 28 Aug 2026 23:41:16 -0400 Subject: [PATCH 5/5] fix(ai): preserve response identity history --- packages/ai/src/protocols/open-responses.ts | 12 +- .../provider/open-responses-lifecycle.test.ts | 188 +++++++++++------- 2 files changed, 125 insertions(+), 75 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 0c9f4f8865b1..7e5c4d118ca1 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -392,6 +392,7 @@ export interface ParserState { readonly providerMetadataKey: string // Pending calls use generated keys. Wire call ids, item ids, and output // indexes are aliases only and can therefore collide without sharing state. + // Alias history is retained until this response's parser state is discarded. readonly tools: ToolStream.State readonly toolCalls: ReadonlyMap readonly toolItems: ReadonlyMap> @@ -876,9 +877,10 @@ const pendingToolID = (state: ParserState, event: Event) => { return id !== undefined && state.tools[id] !== undefined ? id : undefined } if (event.item_id === undefined) return undefined - const active = (state.toolItems.get(event.item_id) ?? []).filter((id) => state.tools[id] !== undefined) - if (active.length > 1) return null - return active[0] + const admitted = state.toolItems.get(event.item_id) ?? [] + if (admitted.length > 1) return null + const id = admitted[0] + return id !== undefined && state.tools[id] !== undefined ? id : undefined } const startReasoningSummaryPart = (state: ParserState, itemID: string, index: number): StepResult => { @@ -1037,7 +1039,9 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { toolCalls: new Map(state.toolCalls).set(item.call_id, id), toolItems: new Map(state.toolItems).set(itemID, [...(state.toolItems.get(itemID) ?? []), id]), toolOutputs: - event.output_index === undefined ? state.toolOutputs : new Map(state.toolOutputs).set(event.output_index, id), + event.output_index === undefined || state.toolOutputs.has(event.output_index) + ? state.toolOutputs + : new Map(state.toolOutputs).set(event.output_index, id), nextTool: state.nextTool + 1, }, [...events, LLMEvent.toolInputStart({ id: item.call_id, name: item.name ?? "", providerMetadata: metadata })], diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts index 356d451bd070..01ca58f0e100 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -394,78 +394,124 @@ describe("Open Responses basic-item lifecycles", () => { }), ) }) - - it.effect("tombstones a completed output index instead of resolving its late events through a shared item id", () => - Effect.gen(function* () { - const first = { type: "function_call", id: "shared", call_id: "call_1", name: "first" } - const second = { type: "function_call", id: "shared", call_id: "call_2", name: "second" } - const events = yield* collect( - { type: "response.output_item.added", output_index: 0, item: first }, - { type: "response.output_item.added", output_index: 1, item: second }, - { - type: "response.function_call_arguments.delta", - output_index: 0, - item_id: "shared", - delta: '{"first":"draft"}', - }, - { - type: "response.function_call_arguments.delta", - output_index: 1, - item_id: "shared", - delta: '{"second":"kept"}', - }, - { - type: "response.output_item.done", - output_index: 0, - item: { ...first, arguments: '{"first":"final"}' }, - }, - { type: "response.output_item.added", output_index: 0, item: first }, - { - type: "response.function_call_arguments.delta", - output_index: 0, - item_id: "shared", - delta: '{"wrong":"delta"}', - }, - { - type: "response.function_call_arguments.done", - output_index: 0, - item_id: "shared", - arguments: '{"wrong":"done"}', - }, - { - type: "response.output_item.done", - output_index: 0, - item: { ...first, arguments: '{"wrong":"item"}' }, - }, - { type: "response.output_item.done", output_index: 1, item: second }, - completed, - ) - expect(events.filter(LLMEvent.is.toolCall)).toEqual([ - { - type: "tool-call", - id: "call_1", - name: "first", - input: { first: "final" }, - providerMetadata: { "openai-compatible": { itemId: "shared" } }, - }, - { - type: "tool-call", - id: "call_2", - name: "second", - input: { second: "kept" }, - providerMetadata: { "openai-compatible": { itemId: "shared" } }, - }, - ]) - ;["call_1", "call_2"].forEach((id) => { - expect(events.filter((event) => "id" in event && event.id === id).map((event) => event.type)).toEqual([ - "tool-input-start", - "tool-input-delta", - "tool-input-end", - "tool-call", + ;["direct", "response"].forEach((completion) => { + it.effect(`keeps a completed output index tombstoned across ${completion} completion of its replacement`, () => + Effect.gen(function* () { + const first = { type: "function_call", id: "old", call_id: "call_1", name: "first" } + const second = { type: "function_call", id: "new", call_id: "call_2", name: "second" } + const terminal: OpenResponses.Event[] = + completion === "direct" + ? [{ type: "response.output_item.done", output_index: 0, item: second }, completed] + : [ + { + type: "response.completed", + response: { id: "resp_1", output: [second, { ...first, arguments: '{"wrong":"terminal"}' }] }, + }, + ] + const events = yield* collect( + { type: "response.output_item.added", output_index: 0, item: first }, + { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: "old", + delta: '{"first":"draft"}', + }, + { + type: "response.output_item.done", + output_index: 0, + item: { ...first, arguments: '{"first":"final"}' }, + }, + { type: "response.output_item.added", output_index: 0, item: second }, + { type: "response.output_item.added", output_index: 0, item: second }, + { + type: "response.function_call_arguments.delta", + item_id: "new", + delta: '{"second":"kept"}', + }, + { + type: "response.function_call_arguments.delta", + output_index: 0, + item_id: "new", + delta: '{"wrong":"delta"}', + }, + { + type: "response.function_call_arguments.done", + output_index: 0, + item_id: "new", + arguments: '{"wrong":"done"}', + }, + { + type: "response.output_item.done", + output_index: 0, + item: { ...first, arguments: '{"wrong":"item"}' }, + }, + ...terminal, + ) + expect(events.filter(LLMEvent.is.toolCall)).toEqual([ + { + type: "tool-call", + id: "call_1", + name: "first", + input: { first: "final" }, + providerMetadata: { "openai-compatible": { itemId: "old" } }, + }, + { + type: "tool-call", + id: "call_2", + name: "second", + input: { second: "kept" }, + providerMetadata: { "openai-compatible": { itemId: "new" } }, + }, ]) - }) - }), - ) + ;["call_1", "call_2"].forEach((id) => { + expect(events.filter((event) => "id" in event && event.id === id).map((event) => event.type)).toEqual([ + "tool-input-start", + "tool-input-delta", + "tool-input-end", + "tool-call", + ]) + }) + }), + ) + }) + ;[ + { type: "response.function_call_arguments.delta", item_id: "shared", delta: '{"wrong":"delta"}' }, + { type: "response.function_call_arguments.done", item_id: "shared", arguments: '{"wrong":"done"}' }, + ].forEach((late) => { + it.effect(`keeps a reused item id ambiguous for a late ${late.type}`, () => + Effect.gen(function* () { + const first = { type: "function_call", id: "shared", call_id: "call_1", name: "first" } + const second = { type: "function_call", id: "shared", call_id: "call_2", name: "second" } + const error = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "response.output_item.added", output_index: 0, item: first }, + { + type: "response.output_item.done", + output_index: 0, + item: { ...first, arguments: '{"first":"final"}' }, + }, + { type: "response.output_item.added", output_index: 1, item: second }, + { + type: "response.function_call_arguments.delta", + output_index: 1, + item_id: "shared", + delta: '{"second":"kept"}', + }, + late, + { type: "response.output_item.done", output_index: 1, item: second }, + completed, + ), + ), + ), + Effect.flip, + ) + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.message).toContain("ambiguous item_id without a matching output_index") + }), + ) + }) it.effect("treats prototype property names as ordinary call and item ids", () => Effect.gen(function* () {