From d9c85d8d95bc7e514c49e688985461cfb5f3e042 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:31:00 -0500 Subject: [PATCH] refactor(ai): resolve responses item ids once at the stream boundary (#46885) --- packages/ai/src/protocols/open-responses.ts | 170 ++++++++------- packages/ai/src/protocols/openai-responses.ts | 7 +- .../protocols/utils/responses-hosted-tools.ts | 7 +- packages/ai/src/protocols/xai-responses.ts | 3 +- packages/ai/test/provider/compaction.test.ts | 13 +- .../provider/open-responses-lifecycle.test.ts | 201 +++++++++++------- .../openai-compatible-responses.test.ts | 65 ++---- .../ai/test/provider/openai-responses.test.ts | 125 +---------- 8 files changed, 242 insertions(+), 349 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 7043da8dbbe..8a17f703513 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -304,6 +304,11 @@ export const OpenResponsesUsage = Schema.Struct({ }) type OpenResponsesUsage = Schema.Schema.Type +// The spec requires `id` on every output item, but some gateways drop it from +// later item events (Bedrock Mantle renames it to `item_id` on +// `output_item.done` and `response.completed.output`). Decode it as optional +// and let `normalize` recover or mint it once before the parser runs. +// https://www.openresponses.org/specification#extending-items export const StreamItem = Schema.StructWithRest( Schema.Struct({ type: Schema.String, @@ -316,6 +321,7 @@ export const StreamItem = Schema.StructWithRest( [Schema.Record(Schema.String, Schema.Unknown)], ) export type StreamItem = Schema.Schema.Type +export type OutputItem = StreamItem & { readonly id: string } // The Responses schema puts streaming error details at the top level and // response failures under `response.error`. WebSocket failures use an @@ -395,6 +401,7 @@ export const Event = Schema.StructWithRest( [Schema.Record(Schema.String, Schema.Unknown)], ) export type Event = Schema.Schema.Type +export type NormalizedEvent = Event & { readonly item?: OutputItem | null } export interface ProviderAdapter { readonly id: string @@ -416,14 +423,13 @@ export interface ParserState { readonly name: string readonly providerMetadataKey: string readonly tools: ToolStream.State - // Call ids stay independent of item ids, which may be omitted or reused. + // Item ids are response-scoped identities. Keep completed ids tombstoned so + // reconnect replay cannot reopen fragments already emitted downstream. readonly completedTools: ReadonlySet readonly hasFunctionCall: boolean readonly lifecycle: Lifecycle.State readonly outputItems: Readonly> readonly message: { readonly id: string; readonly phase: MessagePhase | null | undefined } | undefined - // Item ids are response-scoped identities. Keep completed ids tombstoned so - // reconnect replay cannot reopen fragments already emitted downstream. readonly completedMessages: ReadonlySet readonly reasoningItems: Readonly> } @@ -875,9 +881,6 @@ export const providerMetadata = (state: ParserState, metadata: Record - item.type === "reasoning" && typeof item.id === "string" - export type StepResult = readonly [ParserState, ReadonlyArray] const NO_EVENTS: StepResult["1"] = [] @@ -921,9 +924,34 @@ const joinReasoningText = (parts: ReadonlyArray) => { return parts.filter((part) => part !== undefined).join("\n\n") } -export const outputItemID = (state: ParserState, event: Event) => +const outputItemID = (state: ParserState, event: Event) => event.output_index === undefined ? event.item_id : (state.outputItems[event.output_index] ?? event.item_id) +const ITEM_ID_PREFIX: Readonly> = { + message: "msg", + reasoning: "rs", + function_call: "fc", + compaction: "cmp", +} + +// Mirror Codex: an item that arrives without an id adopts the id of the item +// already open in its output slot, otherwise it gets a locally minted one. +const hasID = (item: StreamItem): item is OutputItem => item.id !== undefined + +const resolveItem = (state: ParserState, item: StreamItem, index: number | undefined): OutputItem => { + if (hasID(item)) return item + const slot = index === undefined ? undefined : state.outputItems[index] + return { ...item, id: slot ?? `${ITEM_ID_PREFIX[item.type] ?? "item"}_${crypto.randomUUID().replaceAll("-", "")}` } +} + +// Registered output slots are authoritative for `item_id` routing, and items +// are resolved here so everything downstream can rely on `item.id`. +export const normalize = (state: ParserState, input: Event): NormalizedEvent => ({ + ...input, + item_id: input.item_id === undefined ? undefined : outputItemID(state, input), + item: input.item ? resolveItem(state, input.item, input.output_index) : input.item, +}) + 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] @@ -997,7 +1025,7 @@ export const onReasoningDone = (state: ParserState, event: Event, itemID: string return onReasoningDelta(state, { ...event, delta: event.text }, itemID) } -const reasoningMetadata = (state: ParserState, item: StreamItem & { id: string }) => +const reasoningMetadata = (state: ParserState, item: OutputItem) => providerMetadata(state, { itemId: item.id, reasoningEncryptedContent: item.encrypted_content ?? null }) // Responses APIs normally stream reasoning items in this order: @@ -1010,18 +1038,18 @@ const reasoningMetadata = (state: ParserState, item: StreamItem & { id: string } // `onOutputItemAdded` seeds the per-item entry, while each later part start is // also an implicit boundary for the previous part. This keeps the common event // lifecycle ordered when a compatible provider omits or delays a part-done event. -const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { +const onOutputItemAdded = (state: ParserState, event: NormalizedEvent): StepResult => { const item = event.item - if (item?.type === "message" && item.id !== undefined) { - const itemID = item.id - if (state.completedMessages.has(itemID)) return [state, NO_EVENTS] + if (!item) return [state, NO_EVENTS] + if (item.type === "message") { + if (state.completedMessages.has(item.id)) return [state, NO_EVENTS] const phase = messagePhase(item.phase) const completedMessages = new Set(state.completedMessages) - if (state.message !== undefined && state.message.id !== itemID) completedMessages.add(state.message.id) + if (state.message !== undefined && state.message.id !== item.id) completedMessages.add(state.message.id) // A new message closes earlier messages, including ones that never streamed. const events: LLMEvent[] = [] const lifecycle = [...state.lifecycle.text] - .filter((id) => id !== itemID) + .filter((id) => id !== item.id) .reduce((lifecycle, id) => { completedMessages.add(id) const openPhase = state.message?.id === id ? state.message.phase : undefined @@ -1038,14 +1066,14 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { lifecycle, completedMessages, message: { - id: itemID, - phase: phase === undefined && state.message?.id === itemID ? state.message.phase : phase, + id: item.id, + phase: phase === undefined && state.message?.id === item.id ? state.message.phase : phase, }, }, events, ] } - if (item && isReasoningItem(item)) { + if (item.type === "reasoning") { if (state.reasoningItems[item.id] !== undefined) return [state, NO_EVENTS] const events: LLMEvent[] = [] return [ @@ -1065,18 +1093,16 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { events, ] } - 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] - const metadata = item.id !== undefined ? providerMetadata(state, { itemId: item.id }) : undefined + if (item.type !== "function_call" || !item.call_id) return [state, NO_EVENTS] + if (state.tools[item.id] !== undefined || state.completedTools.has(item.id)) return [state, NO_EVENTS] + const metadata = providerMetadata(state, { itemId: item.id }) const events: LLMEvent[] = [] const lifecycle = Lifecycle.stepStart(state.lifecycle, events) return [ { ...state, lifecycle, - tools: ToolStream.start(state.tools, id, { + tools: ToolStream.start(state.tools, item.id, { id: item.call_id, name: item.name ?? "", input: item.arguments ?? "", @@ -1148,13 +1174,13 @@ const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgu const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( state: ParserState, - item: Event["item"], + item: NormalizedEvent["item"], ) { if (!item) return [state, NO_EVENTS] satisfies StepResult if (item.type === "compaction") { - if (!item.id || typeof item.encrypted_content !== "string") - return yield* ProviderShared.eventError(state.id, "Compaction output is missing its id or encrypted content") + if (typeof item.encrypted_content !== "string") + return yield* ProviderShared.eventError(state.id, "Compaction output is missing its encrypted content") if (state.completedCompactions.has(item.id)) return [state, NO_EVENTS] satisfies StepResult const events: LLMEvent[] = [] const lifecycle = Lifecycle.stepStart(state.lifecycle, events) @@ -1171,7 +1197,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( ] satisfies StepResult } - if (item.type === "message" && item.id !== undefined) { + if (item.type === "message") { if (state.completedMessages.has(item.id)) return [state, NO_EVENTS] satisfies StepResult const completedMessages = new Set(state.completedMessages) completedMessages.add(item.id) @@ -1204,36 +1230,23 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( if (item.type === "function_call") { if (!item.call_id || !item.name) return [state, NO_EVENTS] satisfies StepResult - 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 tools = - registered !== undefined - ? state.tools - : ToolStream.start(state.tools, id, { - id: callID, - name: item.name, - providerMetadata: metadata, - }) + if (state.completedTools.has(item.id)) return [state, NO_EVENTS] satisfies StepResult + const metadata = providerMetadata(state, { itemId: item.id }) + const registered = state.tools[item.id] !== undefined + const tools = registered + ? state.tools + : ToolStream.start(state.tools, item.id, { id: item.call_id, name: item.name, providerMetadata: metadata }) const result = item.arguments === undefined - ? yield* ToolStream.finish(state.id, tools, id) - : yield* ToolStream.finishWithInput(state.id, tools, id, item.arguments) + ? yield* ToolStream.finish(state.id, tools, item.id) + : yield* ToolStream.finishWithInput(state.id, tools, item.id, item.arguments) const events: LLMEvent[] = [] const finished = result.events ?? [] // A done-only call never streamed a start event, so open its lifecycle here. const resultEvents = - registered !== undefined || finished.length === 0 + registered || finished.length === 0 ? finished - : [LLMEvent.toolInputStart({ id: callID, name: item.name, providerMetadata: metadata }), ...finished] + : [LLMEvent.toolInputStart({ id: item.call_id, name: item.name, providerMetadata: metadata }), ...finished] const lifecycle = resultEvents.length ? Lifecycle.stepStart(state.lifecycle, events) : state.lifecycle events.push(...resultEvents) return [ @@ -1244,13 +1257,13 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( resultEvents.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)) || state.hasFunctionCall, tools: result.tools, - completedTools: new Set([...state.completedTools, callID]), + completedTools: new Set([...state.completedTools, item.id]), }, events, ] satisfies StepResult } - if (isReasoningItem(item)) { + if (item.type === "reasoning") { if (state.reasoningItems[item.id]?.open === false) return [state, NO_EVENTS] satisfies StepResult const metadata = reasoningMetadata(state, item) const summaryParts: ReadonlyArray = Array.isArray(item.summary) ? item.summary : [] @@ -1334,21 +1347,17 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( let current = state const events: LLMEvent[] = [] if (event.type === "response.completed") { - for (const item of event.response?.output ?? []) { - if (item.type !== "compaction" && item.type !== "function_call") continue - if (item.type === "compaction") { - // Terminal recovery cannot insert a checkpoint before already-emitted content. - if (state.lifecycle.stepStarted && !state.completedCompactions.has(item.id ?? "")) - return yield* ProviderShared.eventError( - state.id, - "Cannot recover a compaction checkpoint after output has been emitted", - ) - } - if ( - item.type === "function_call" && - (!item.call_id || !Object.values(current.tools).some((tool) => tool?.id === item.call_id)) - ) - continue + // An output item's array position is its output index. + for (const item of (event.response?.output ?? []).map((item, index) => resolveItem(state, item, index))) { + // Terminal recovery cannot insert a checkpoint before already-emitted content. + if (item.type === "compaction" && state.lifecycle.stepStarted && !state.completedCompactions.has(item.id)) + return yield* ProviderShared.eventError( + state.id, + "Cannot recover a compaction checkpoint after output has been emitted", + ) + const recoverable = + item.type === "compaction" || (item.type === "function_call" && current.tools[item.id] !== undefined) + if (!recoverable) continue const [next, emitted] = yield* onOutputItemDone(current, item) current = next events.push(...emitted) @@ -1415,12 +1424,9 @@ export const providerFailure = (event: Event, fallback: string, body = ProviderS return new AIError({ reason }) } -export const step = (state: ParserState, input: Event) => { - // The OpenAPI requires string IDs but imposes no minLength; empty is not missing. - const event = - input.item_id !== undefined && outputItemID(state, input) !== input.item_id - ? { ...input, item_id: outputItemID(state, input) } - : input +// Callers must pass events through `normalize` first. The OpenAPI requires +// string IDs but imposes no minLength; empty is not missing. +export const step = (state: ParserState, event: NormalizedEvent) => { if (event.type === "response.output_text.delta" || event.type === "response.output_text.done") { if (event.item_id === undefined) return ProviderShared.eventError(state.id, `${event.type} is missing item_id`) return Effect.succeed( @@ -1460,20 +1466,16 @@ export const step = (state: ParserState, input: Event) => { ? Effect.succeed(onReasoningSummaryPartDone(state, event)) : ProviderShared.eventError(state.id, `${event.type} is missing item_id`) if (event.type === "response.output_item.added") { - if (event.item?.type === "message" && event.item.id === undefined) - return ProviderShared.eventError(state.id, `${event.type} message is missing id`) if ( - event.item && - isReasoningItem(event.item) && + event.item?.type === "reasoning" && state.reasoningItems[event.item.id] === undefined && state.lifecycle.reasoning.size > 0 ) return ProviderShared.eventError(state.id, `${event.type} started reasoning before the previous item ended`) - const id = event.item?.id ?? (event.item?.type === "function_call" ? event.item.call_id : undefined) return Effect.succeed( onOutputItemAdded( - event.output_index !== undefined && id !== undefined - ? { ...state, outputItems: { ...state.outputItems, [event.output_index]: id } } + event.output_index !== undefined && event.item + ? { ...state, outputItems: { ...state.outputItems, [event.output_index]: event.item.id } } : state, event, ), @@ -1483,11 +1485,7 @@ export const step = (state: ParserState, input: Event) => { return event.item_id !== undefined ? onFunctionCallArgumentsDelta(state, event) : ProviderShared.eventError(state.id, `${event.type} is missing item_id`) - if (event.type === "response.output_item.done") { - if (event.item?.type === "message" && event.item.id === undefined) - return ProviderShared.eventError(state.id, `${event.type} message is missing id`) - return onOutputItemDone(state, event.item) - } + if (event.type === "response.output_item.done") return onOutputItemDone(state, event.item) if (event.type === "response.completed" || event.type === "response.incomplete") return onResponseFinish(state, event) if (event.type === "response.failed") return providerFailure(event, `${state.name} response failed`) if (event.type === "error") @@ -1537,7 +1535,7 @@ export const protocol = Protocol.make({ stream: { event: Protocol.jsonEvent(Event), initial, - step, + step: (state: ParserState, event: Event) => step(state, normalize(state, event)), terminal, }, }) diff --git a/packages/ai/src/protocols/openai-responses.ts b/packages/ai/src/protocols/openai-responses.ts index 2a0b2bc892d..7ab98ec1611 100644 --- a/packages/ai/src/protocols/openai-responses.ts +++ b/packages/ai/src/protocols/openai-responses.ts @@ -201,12 +201,11 @@ const HOSTED_TOOLS = { }, } as const satisfies ResponsesHostedTools.Definitions -const step = (state: OpenResponses.ParserState, event: OpenResponses.Event) => { +const step = (state: OpenResponses.ParserState, input: OpenResponses.Event) => { + const event = OpenResponses.normalize(state, input) if (event.type === "response.reasoning_text.delta") return event.item_id !== undefined - ? Effect.succeed( - OpenResponses.onReasoningDelta(state, event, OpenResponses.outputItemID(state, event) ?? event.item_id), - ) + ? Effect.succeed(OpenResponses.onReasoningDelta(state, event, event.item_id)) : ProviderShared.eventError(ADAPTER, `${event.type} is missing item_id`) if (event.type === "response.output_item.done" && event.item && ResponsesHostedTools.isItem(event.item, HOSTED_TOOLS)) return ResponsesHostedTools.onDone(state, event.item, HOSTED_TOOLS) diff --git a/packages/ai/src/protocols/utils/responses-hosted-tools.ts b/packages/ai/src/protocols/utils/responses-hosted-tools.ts index 283b6e549fe..465ff23ed85 100644 --- a/packages/ai/src/protocols/utils/responses-hosted-tools.ts +++ b/packages/ai/src/protocols/utils/responses-hosted-tools.ts @@ -3,8 +3,7 @@ import { LLMEvent, type AIError, type ToolResultPart } from "../../schema/index. import { OpenResponses } from "../open-responses.js" import { Lifecycle } from "./lifecycle.js" -export type Item = OpenResponses.StreamItem & { - readonly id: string +export type Item = OpenResponses.OutputItem & { readonly status?: string readonly action?: unknown readonly queries?: unknown @@ -27,8 +26,8 @@ export interface Definition { export type Definitions = Readonly> -export const isItem = (item: OpenResponses.StreamItem, tools: Tools): item is Item => - item.type in tools && typeof item.id === "string" && item.id.length > 0 +export const isItem = (item: OpenResponses.OutputItem, tools: Tools): item is Item => + item.type in tools export const onDone: ( state: OpenResponses.ParserState, diff --git a/packages/ai/src/protocols/xai-responses.ts b/packages/ai/src/protocols/xai-responses.ts index c78df60bbc6..acd55f347dc 100644 --- a/packages/ai/src/protocols/xai-responses.ts +++ b/packages/ai/src/protocols/xai-responses.ts @@ -69,7 +69,8 @@ const HOSTED_TOOLS = { // Grok speaks the standard Responses reasoning dialect (`reasoning_summary_text.*`, // handled by the baseline); only its hosted tool vocabulary differs. -const step = (state: OpenResponses.ParserState, event: OpenResponses.Event) => { +const step = (state: OpenResponses.ParserState, input: OpenResponses.Event) => { + const event = OpenResponses.normalize(state, input) if (event.type === "response.output_item.done" && event.item && ResponsesHostedTools.isItem(event.item, HOSTED_TOOLS)) return ResponsesHostedTools.onDone(state, event.item, HOSTED_TOOLS) return OpenResponses.step(state, event) diff --git a/packages/ai/test/provider/compaction.test.ts b/packages/ai/test/provider/compaction.test.ts index 5f1a5411986..caccb5d6bf1 100644 --- a/packages/ai/test/provider/compaction.test.ts +++ b/packages/ai/test/provider/compaction.test.ts @@ -125,12 +125,15 @@ testEffect( response: { output: [{ type: "compaction", encrypted_content: "opaque" }] }, }), ), -).effect("rejects terminal checkpoints missing an id", () => +).effect("mints an id for terminal checkpoints that omit one", () => Effect.gen(function* () { - const error = yield* LLMClient.generate( + const response = yield* LLMClient.generate( LLM.request({ model: OpenAI.configure({ apiKey: "test" }).responses("fixture"), prompt: "hello" }), - ).pipe(Effect.flip) - expect(error.reason._tag).toBe("InvalidProviderOutput") - expect(error.message).toContain("missing its id") + ) + const part = response.message.content[0] + expect(part?.type).toBe("compaction") + if (part?.type !== "compaction") return + expect(part.id).toMatch(/^cmp_[0-9a-f]{32}$/) + expect(part.encrypted).toBe("opaque") }), ) diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts index 9a35b4c9a83..fd29b09593f 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -329,69 +329,126 @@ describe("Open Responses basic-item lifecycles", () => { ]) }), ) - ;[undefined, "fc_1"].forEach((id) => { - it.effect(`opens and closes a done-only tool ${id === undefined ? "without" : "with"} an item id`, () => - Effect.gen(function* () { - const item = { - type: "function_call", - ...(id === undefined ? {} : { id }), - call_id: "call_1", - name: "lookup", - arguments: '{"query":"weather"}', - } - const events = yield* collect( - { type: "response.output_item.done", item }, - { type: "response.output_item.done", item: { ...item, id: "fc_1" } }, - { type: "response.output_item.added", item }, - completed, - ) - const providerMetadata = id === undefined ? undefined : { "openai-compatible": { itemId: id } } - 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: "weather" }, providerMetadata }, - ]) - expect(events.filter(LLMEvent.is.finish)).toEqual([ - { - type: "finish", - reason: { normalized: "tool-calls", raw: undefined }, - providerMetadata: { "openai-compatible": { responseId: "resp_1", serviceTier: undefined } }, + // Captured from Bedrock Mantle (openai.gpt-oss-120b): the terminal function_call + // items rename `id` to `item_id` and carry a stray `output_index`. + it.effect("recovers a terminal function_call id from its output slot", () => + Effect.gen(function* () { + const terminal = { + type: "function_call", + item_id: "fc_828bee50dee1d029", + call_id: "call_bc1eb4b42e70ee53", + name: "get_weather", + arguments: '{\n "city": "Paris"\n}', + output_index: 1, + status: "completed", + } + const events = yield* collect( + { + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "msg_879a68b589198b4c" }, + }, + { type: "response.output_item.done", output_index: 0, item: { type: "reasoning", id: "msg_879a68b589198b4c" } }, + { + type: "response.output_item.added", + output_index: 1, + item: { + type: "function_call", + id: "fc_828bee50dee1d029", + call_id: "call_bc1eb4b42e70ee53", + name: "get_weather", + arguments: "", + status: "in_progress", }, - ]) - }), - ) + }, + { + type: "response.function_call_arguments.delta", + output_index: 1, + item_id: "fc_828bee50dee1d029", + delta: '{\n "city": "Paris"\n}', + }, + { + type: "response.function_call_arguments.done", + output_index: 1, + item_id: "fc_828bee50dee1d029", + arguments: '{\n "city": "Paris"\n}', + }, + { type: "response.output_item.done", output_index: 1, item: terminal }, + { + type: "response.completed", + response: { id: "resp_1", output: [{ type: "reasoning", id: "msg_879a68b589198b4c" }, terminal] }, + }, + ) + const providerMetadata = { "openai-compatible": { itemId: "fc_828bee50dee1d029" } } + expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([ + { type: "tool-input-start", id: "call_bc1eb4b42e70ee53", name: "get_weather", providerMetadata }, + { + type: "tool-input-delta", + id: "call_bc1eb4b42e70ee53", + name: "get_weather", + text: '{\n "city": "Paris"\n}', + input: { city: "Paris" }, + }, + { type: "tool-input-end", id: "call_bc1eb4b42e70ee53", name: "get_weather", providerMetadata }, + { + type: "tool-call", + id: "call_bc1eb4b42e70ee53", + name: "get_weather", + input: { city: "Paris" }, + providerMetadata, + }, + ]) + }), + ) - it.effect(`deduplicates a pending call whose item id is ${id === undefined ? "introduced" : "omitted"} later`, () => - Effect.gen(function* () { - const item = { type: "function_call", call_id: "call_1", name: "lookup" } - const first = { ...item, ...(id === undefined ? {} : { id }) } - const duplicate = { ...item, ...(id === undefined ? { id: "fc_1" } : {}) } - const events = yield* collect( - { type: "response.output_item.added", item: first }, - { type: "response.function_call_arguments.delta", item_id: id ?? "call_1", delta: '{"query":"weather"}' }, - { type: "response.output_item.added", item: duplicate }, - { type: "response.output_item.done", item: duplicate }, - { type: "response.output_item.done", item: first }, - { type: "response.output_item.added", item: duplicate }, - completed, - ) - // Identity metadata comes from the first admission, not the duplicate. - const providerMetadata = id === undefined ? undefined : { "openai-compatible": { itemId: id } } - expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([ - { type: "tool-input-start", id: "call_1", name: "lookup", providerMetadata }, - { - type: "tool-input-delta", - id: "call_1", - name: "lookup", - text: '{"query":"weather"}', - input: { query: "weather" }, - }, - { type: "tool-input-end", id: "call_1", name: "lookup", providerMetadata }, - { type: "tool-call", id: "call_1", name: "lookup", input: { query: "weather" }, providerMetadata }, - ]) - }), - ) - }) + it.effect("mints an id for a done-only tool that never had one", () => + Effect.gen(function* () { + const events = yield* collect( + { + type: "response.output_item.done", + output_index: 0, + item: { type: "function_call", call_id: "call_1", name: "lookup", arguments: '{"query":"weather"}' }, + }, + completed, + ) + const call = events.find(LLMEvent.is.toolCall) + expect(call).toMatchObject({ id: "call_1", name: "lookup", input: { query: "weather" } }) + expect(call?.providerMetadata?.["openai-compatible"]).toMatchObject({ + itemId: expect.stringMatching(/^fc_[0-9a-f]{32}$/), + }) + }), + ) + + it.effect("opens and closes a done-only tool once", () => + Effect.gen(function* () { + const item = { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"weather"}', + } + const events = yield* collect( + { type: "response.output_item.done", item }, + { type: "response.output_item.done", item }, + { type: "response.output_item.added", item }, + 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: "lookup", providerMetadata }, + { type: "tool-input-end", id: "call_1", name: "lookup", providerMetadata }, + { type: "tool-call", id: "call_1", name: "lookup", input: { query: "weather" }, providerMetadata }, + ]) + expect(events.filter(LLMEvent.is.finish)).toEqual([ + { + type: "finish", + reason: { normalized: "tool-calls", raw: undefined }, + providerMetadata: { "openai-compatible": { responseId: "resp_1", serviceTier: undefined } }, + }, + ]) + }), + ) it.effect("recovers pending calls without reconciling terminal reasoning", () => Effect.gen(function* () { @@ -436,21 +493,6 @@ describe("Open Responses basic-item lifecycles", () => { }), ) - it.effect("preserves call identity and pending order when an item id is reused", () => - Effect.gen(function* () { - const first = { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: "{}" } - const events = yield* collect( - { type: "response.output_item.added", item: first }, - { type: "response.output_item.added", item: { ...first, id: "fc_2", call_id: "call_2" } }, - { type: "response.output_item.done", item: first }, - { type: "response.output_item.added", item: { ...first, call_id: "call_3" } }, - { type: "response.output_item.done", item: first }, - completed, - ) - expect(events.filter(LLMEvent.is.toolCall).map((event) => event.id)).toEqual(["call_1", "call_2", "call_3"]) - }), - ) - it.effect("keeps text and reasoning identities separate even with empty item ids", () => Effect.gen(function* () { const events = yield* collect( @@ -500,14 +542,15 @@ describe("Open Responses basic-item lifecycles", () => { { type: "response.output_text.delta", item_id: "msg_1", delta: "Answer" }, { type: "response.output_item.added", - item: { type: "function_call", call_id: "call_1", name: "lookup", arguments: "{}" }, + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: "{}" }, }, completed, ) // Generic terminal closure does not repeat the message's phase metadata. + const providerMetadata = { "openai-compatible": { itemId: "fc_1" } } expect(events.slice(4, -2)).toEqual([ - { type: "tool-input-end", id: "call_1", name: "lookup" }, - { type: "tool-call", id: "call_1", name: "lookup", input: {} }, + { type: "tool-input-end", id: "call_1", name: "lookup", providerMetadata }, + { type: "tool-call", id: "call_1", name: "lookup", input: {}, providerMetadata }, { type: "text-end", id: "msg_1" }, ]) }), diff --git a/packages/ai/test/provider/openai-compatible-responses.test.ts b/packages/ai/test/provider/openai-compatible-responses.test.ts index a24b0c7c5a9..dec79a23abc 100644 --- a/packages/ai/test/provider/openai-compatible-responses.test.ts +++ b/packages/ai/test/provider/openai-compatible-responses.test.ts @@ -586,23 +586,21 @@ describe("Open Responses-compatible route", () => { Effect.gen(function* () { yield* Effect.forEach(["response.output_item.added", "response.output_item.done"], (type) => Effect.forEach(fixtures, (fixture) => - Effect.forEach( - fixture.item.type === "message" ? [undefined, null, 0, false, {}, []] : [null, 0, false, {}, []], - (id) => - Effect.gen(function* () { - const error = yield* LLMClient.generate(request).pipe( - Effect.provide( - fixedResponse( - sseEvents( - { type, item: { ...fixture.item, id } }, - { type: "response.completed", response: { id: "resp_1" } }, - ), + Effect.forEach([null, 0, false, {}, []], (id) => + Effect.gen(function* () { + const error = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type, item: { ...fixture.item, id } }, + { type: "response.completed", response: { id: "resp_1" } }, ), ), - Effect.flip, - ) - expect(error.reason._tag).toBe("InvalidProviderOutput") - }), + ), + Effect.flip, + ) + expect(error.reason._tag).toBe("InvalidProviderOutput") + }), ), ), ) @@ -610,43 +608,6 @@ describe("Open Responses-compatible route", () => { ) }) - it.effect("streams function calls without optional item ids through the shared baseline", () => - Effect.gen(function* () { - const model = configure({ - apiKey: "test-key", - baseURL: "https://responses.example.test/v1", - provider: "example", - }).model("example-model") - const item = { type: "function_call", call_id: "call_1", name: "lookup", arguments: "" } - const response = yield* LLMClient.generate(LLM.request({ model, prompt: "Look it up." })).pipe( - Effect.provide( - fixedResponse( - sseEvents( - { type: "response.output_item.added", output_index: 1, item }, - { - type: "response.function_call_arguments.delta", - output_index: 1, - item_id: "opaque_item", - delta: '{"query":"shared"}', - }, - { - type: "response.output_item.done", - output_index: 1, - item: { ...item, arguments: '{"query":"complete"}' }, - }, - { type: "response.completed", response: { id: "resp_1" } }, - ), - ), - ), - ) - - expect(response.events.filter(LLMEvent.is.toolCall)).toEqual([ - expect.objectContaining({ id: "call_1", name: "lookup", input: { query: "complete" } }), - ]) - expect(response.events.find(LLMEvent.is.toolCall)?.providerMetadata).toBeUndefined() - }), - ) - it.effect("finalizes pending function calls from completed response output", () => Effect.gen(function* () { const model = configure({ diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index 4f88aa774de..0dff1271a1c 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -469,7 +469,7 @@ describe("OpenAI Responses route", () => { }), ) - it.effect("continues an item-id-less tool call with only the new tool output", () => + it.effect("continues a streamed tool call with only the new tool output", () => Effect.gen(function* () { const firstRequest = { type: "response.create", @@ -485,6 +485,7 @@ describe("OpenAI Responses route", () => { type: "response.output_item.done", item: { type: "function_call", + id: "fc_1", status: "completed", call_id: "call_1", name: "weather", @@ -2129,47 +2130,6 @@ describe("OpenAI Responses route", () => { }), ) - it.effect("routes item-id-less function arguments by output index and prefers item completion", () => - Effect.gen(function* () { - const item = { type: "function_call", call_id: "call_1", name: "lookup", arguments: "" } - const response = yield* LLMClient.generate(request).pipe( - Effect.provide( - fixedResponse( - sseEvents( - { type: "response.output_item.added", output_index: 2, item }, - { - type: "response.function_call_arguments.delta", - output_index: 2, - item_id: "opaque_delta", - delta: '{"query":"streamed"}', - }, - { - type: "response.function_call_arguments.done", - output_index: 2, - item_id: "opaque_done", - arguments: '{"query":"arguments-done"}', - }, - { - type: "response.output_item.done", - output_index: 2, - item: { ...item, arguments: '{"query":"output-item-done"}' }, - }, - { type: "response.completed", response: { id: "resp_1" } }, - ), - ), - ), - ) - - expect(response.events.filter((event) => event.type === "tool-input-delta")).toMatchObject([ - { id: "call_1", text: '{"query":"streamed"}' }, - ]) - expect(response.events.filter(LLMEvent.is.toolCall)).toEqual([ - expect.objectContaining({ id: "call_1", name: "lookup", input: { query: "output-item-done" } }), - ]) - expect(response.events.find(LLMEvent.is.toolCall)?.providerMetadata).toBeUndefined() - }), - ) - it.effect("routes reasoning summary events by output index", () => Effect.gen(function* () { const response = yield* LLMClient.generate(request).pipe( @@ -2389,7 +2349,7 @@ describe("OpenAI Responses route", () => { { type: "response.output_item.added", output_index: 0, - item: { type: "function_call", call_id: "call_1", name: "lookup", arguments: "" }, + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: "" }, }, event, { type: "response.completed", response: { id: "resp_1" } }, @@ -2931,14 +2891,10 @@ describe("OpenAI Responses route", () => { arguments: '{"query":"weather"}', }, }, - // Duplicates that drop the item id still resolve the same call. - { - type: "response.output_item.done", - item: { type: "function_call", call_id: "call_1", name: "lookup", arguments: '{"query":"weather"}' }, - }, + // A completed item that is re-added stays closed. { type: "response.output_item.added", - item: { type: "function_call", call_id: "call_1", name: "lookup", arguments: "" }, + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: "" }, }, { type: "response.completed", response: { id: "resp_1" } }, ), @@ -3793,43 +3749,6 @@ describe("OpenAI Responses route", () => { }), ) - it.effect("finalizes and replays a completed function call without an optional item id", () => - Effect.gen(function* () { - const response = yield* LLMClient.generate(request).pipe( - Effect.provide( - fixedResponse( - sseEvents( - { - type: "response.output_item.done", - item: { type: "function_call", call_id: "call_1", name: "lookup", arguments: '{"query":"weather"}' }, - }, - { type: "response.completed", response: { id: "resp_1" } }, - ), - ), - ), - ) - - expect(response.events.filter(LLMEvent.is.toolCall)).toEqual([ - expect.objectContaining({ id: "call_1", name: "lookup", input: { query: "weather" } }), - ]) - expect(response.events.find(LLMEvent.is.toolCall)?.providerMetadata).toBeUndefined() - - const prepared = yield* compileRequest( - LLM.request({ - model, - messages: [ - response.message, - Message.tool({ id: "call_1", name: "lookup", resultType: "json", result: { forecast: "sunny" } }), - ], - }), - ) - expect(prepared.body.input).toEqual([ - { type: "function_call", call_id: "call_1", name: "lookup", arguments: '{"query":"weather"}' }, - { type: "function_call_output", call_id: "call_1", output: '{"forecast":"sunny"}' }, - ]) - }), - ) - it.effect("emits only missing function arguments from the arguments done event", () => Effect.gen(function* () { const body = sseEvents( @@ -4017,7 +3936,7 @@ describe("OpenAI Responses route", () => { }), ) - it.effect("uses completed response output when item completion and its terminal item id are missing", () => + it.effect("uses completed response output when output item completion is missing", () => Effect.gen(function* () { const body = sseEvents( { @@ -4032,6 +3951,7 @@ describe("OpenAI Responses route", () => { output: [ { type: "function_call", + id: "fc_item_1", call_id: "call_1", name: "lookup", arguments: '{"query":"weather"}', @@ -4053,37 +3973,6 @@ describe("OpenAI Responses route", () => { }), ) - it.effect("reconciles an item-id-less pending function call from completed response output", () => - Effect.gen(function* () { - const item = { type: "function_call", call_id: "call_1", name: "lookup", arguments: "" } - const response = yield* LLMClient.generate(request).pipe( - Effect.provide( - fixedResponse( - sseEvents( - { type: "response.output_item.added", output_index: 0, item }, - { - type: "response.function_call_arguments.delta", - output_index: 0, - item_id: "opaque_delta", - delta: '{"query":"partial', - }, - { - type: "response.completed", - response: { id: "resp_1", output: [{ ...item, arguments: '{"query":"complete"}' }] }, - }, - ), - ), - ), - ) - - expect(response.events.filter(LLMEvent.is.toolCall)).toEqual([ - expect.objectContaining({ id: "call_1", name: "lookup", input: { query: "complete" } }), - ]) - expect(response.events.find(LLMEvent.is.toolCall)?.providerMetadata).toBeUndefined() - expect(response.events.filter(LLMEvent.is.toolInputEnd)).toHaveLength(1) - }), - ) - it.effect("lets completed response output override arguments done", () => Effect.gen(function* () { const body = sseEvents(