diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 79fec2d2b8bd..7e5c4d118ca1 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -390,8 +390,14 @@ 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. + // Alias history is retained until this response's parser state is discarded. readonly tools: ToolStream.State - // Call ids stay independent of item ids, which may be omitted or reused. + readonly toolCalls: ReadonlyMap + readonly toolItems: ReadonlyMap> + readonly toolOutputs: ReadonlyMap + readonly nextTool: number readonly completedTools: ReadonlySet readonly hasFunctionCall: boolean readonly lifecycle: Lifecycle.State @@ -863,6 +869,20 @@ 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 && 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 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 => { const item = state.reasoningItems[itemID] if (!item?.open || index === 0 || item.summaryParts[index] !== undefined) return [state, NO_EVENTS] @@ -1000,9 +1020,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.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 const events: LLMEvent[] = [] const lifecycle = Lifecycle.stepStart(state.lifecycle, events) @@ -1016,6 +1036,13 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { input: item.arguments ?? "", providerMetadata: metadata, }), + 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.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 })], ] @@ -1053,15 +1080,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 +1103,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,18 +1151,13 @@ 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 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) - const id = registered ?? fallback + const admitted = state.toolCalls.get(callID) + const registered = admitted !== undefined && state.tools[admitted] !== undefined ? admitted : undefined + const id = registered ?? callID const tools = registered !== undefined ? state.tools - : ToolStream.start(state.tools, id, { + : ToolStream.start(ToolStream.empty(), id, { id: callID, name: item.name, providerMetadata: metadata, @@ -1153,7 +1182,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, @@ -1245,9 +1274,9 @@ 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) continue + 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 events.push(...emitted) @@ -1416,6 +1445,10 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse providerMetadataKey: request.model.route.providerMetadataKey ?? "openresponses", hasFunctionCall: false, tools: ToolStream.empty(), + toolCalls: new Map(), + toolItems: new Map(), + toolOutputs: new Map(), + 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 dfd3c8e4a4ab..01ca58f0e100 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -301,6 +301,256 @@ describe("Open Responses basic-item lifecycles", () => { }), ) }) + ;[ + { + 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", + } + 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 } }, + }) + }) + }), + ) + }) + ;["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* () { + 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* () { @@ -344,6 +594,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* () {