From 18e22cd82e61502afaef40820e0f192722f49b66 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Thu, 27 Aug 2026 23:56:42 -0500 Subject: [PATCH] refactor(ai): simplify response stream state (#45835) --- packages/ai/src/protocols/open-responses.ts | 81 ++- .../provider/open-responses-lifecycle.test.ts | 467 ++++++++++++++++++ 2 files changed, 498 insertions(+), 50 deletions(-) create mode 100644 packages/ai/test/provider/open-responses-lifecycle.test.ts diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 6e2fb755cdf..069e6e18f97 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -391,21 +391,19 @@ export interface ParserState { readonly name: string readonly providerMetadataKey: string readonly tools: ToolStream.State - // Call ids that already emitted their terminal tool-call event, so duplicate - // or late item events for the same call stay no-ops. Keyed by `call_id` - // because duplicate events may disagree on whether `item.id` is present. + // Call ids stay independent of item ids, which may be omitted or reused. readonly completedTools: ReadonlySet readonly hasFunctionCall: boolean readonly lifecycle: Lifecycle.State readonly outputItems: Readonly> - readonly messageItems: ReadonlySet - readonly messagePhases: Readonly> + readonly message: { readonly id: string; readonly phase: MessagePhase | null | undefined } | undefined readonly reasoningItems: Readonly> } type ReasoningSummaryStatus = "active" | "can-conclude" | "concluded" interface ReasoningStreamItem { + readonly open: boolean readonly encryptedContent: string | null | undefined // Keyed by the wire protocol's numeric `summary_index`. JS object keys coerce to // strings, but typing the map as `Record` documents intent @@ -830,16 +828,16 @@ const TERMINAL_TYPES = new Set(["error", "response.completed", "response.incompl export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type) const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => { - if (!event.delta || !state.messageItems.has(id)) return [state, NO_EVENTS] + if (!event.delta || state.message?.id !== id) return [state, NO_EVENTS] const events: LLMEvent[] = [] - const phase = state.messagePhases[id] + const phase = state.message.phase const metadata = providerMetadata(state, { itemId: id, ...(phase === undefined ? {} : { phase }) }) const lifecycle = Lifecycle.textStart(state.lifecycle, events, id, metadata) return [{ ...state, lifecycle: Lifecycle.textDelta(lifecycle, events, id, event.delta) }, events] } const onOutputTextDone = (state: ParserState, event: Event, id: string): StepResult => { - if (state.messageItems.has(id)) { + if (state.message?.id === id) { if (state.lifecycle.text.has(id) || event.text === undefined) return [state, NO_EVENTS] return onOutputTextDelta(state, { ...event, delta: event.text }, id) } @@ -852,8 +850,7 @@ export const outputItemID = (state: ParserState, event: Event) => const startReasoningSummaryPart = (state: ParserState, itemID: string, index: number): StepResult => { const item = state.reasoningItems[itemID] - if (!item || index === 0 || item.summaryParts[index] !== undefined) return [state, NO_EVENTS] - if (Object.values(item.summaryParts).every((status) => status === "concluded")) return [state, NO_EVENTS] + if (!item?.open || index === 0 || item.summaryParts[index] !== undefined) return [state, NO_EVENTS] const events: LLMEvent[] = [] const lifecycle = Object.entries(item.summaryParts) @@ -893,26 +890,19 @@ const startReasoningSummaryPart = (state: ParserState, itemID: string, index: nu export const onReasoningDelta = (state: ParserState, event: Event, itemID: string): StepResult => { const item = state.reasoningItems[itemID] - if (!event.delta || !item) return [state, NO_EVENTS] + if (!event.delta || !item?.open) return [state, NO_EVENTS] const index = event.summary_index ?? 0 if (item.summaryParts[index] === "concluded") return [state, NO_EVENTS] - // An unseen index cannot reopen an item whose parts have all concluded. - if ( - item.summaryParts[index] === undefined && - Object.values(item.summaryParts).every((status) => status === "concluded") - ) - return [state, NO_EVENTS] - const started: StepResult = - item.summaryParts[index] === undefined ? startReasoningSummaryPart(state, itemID, index) : [state, NO_EVENTS] - const current = started[0].reasoningItems[itemID] - if (!current) return started - const events: LLMEvent[] = [...started[1]] + const [started, emitted] = startReasoningSummaryPart(state, itemID, index) + const current = started.reasoningItems[itemID] + if (!current) return [started, emitted] + const events: LLMEvent[] = [...emitted] return [ { - ...started[0], - lifecycle: Lifecycle.reasoningDelta(started[0].lifecycle, events, `${itemID}:${index}`, event.delta), + ...started, + lifecycle: Lifecycle.reasoningDelta(started.lifecycle, events, `${itemID}:${index}`, event.delta), reasoningItems: { - ...started[0].reasoningItems, + ...started.reasoningItems, [itemID]: { ...current, deltaIndexes: new Set([...current.deltaIndexes, index]) }, }, }, @@ -925,7 +915,7 @@ export const onReasoningDelta = (state: ParserState, event: Event, itemID: strin // as a single delta unless that summary index already streamed one. export const onReasoningDone = (state: ParserState, event: Event, itemID: string): StepResult => { const item = state.reasoningItems[itemID] - if (!item || typeof event.text !== "string") return [state, NO_EVENTS] + if (!item?.open || typeof event.text !== "string") return [state, NO_EVENTS] const index = event.summary_index ?? 0 if (item.deltaIndexes.has(index)) return [state, NO_EVENTS] return onReasoningDelta(state, { ...event, delta: event.text }, itemID) @@ -949,14 +939,12 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { if (item?.type === "message" && item.id !== undefined) { const itemID = item.id const phase = messagePhase(item.phase) - // A new message item is an implicit boundary for every earlier message - // item: text still streaming is ended, and all older items leave the - // tracked set so their late deltas stay no-ops instead of overlapping. + // A new message closes earlier messages, including ones that never streamed. const events: LLMEvent[] = [] const lifecycle = [...state.lifecycle.text] .filter((id) => id !== itemID) .reduce((lifecycle, id) => { - const openPhase = state.messagePhases[id] + const openPhase = state.message?.id === id ? state.message.phase : undefined return Lifecycle.textEnd( lifecycle, events, @@ -964,13 +952,14 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { providerMetadata(state, { itemId: id, ...(openPhase === undefined ? {} : { phase: openPhase }) }), ) }, state.lifecycle) - const nextPhase = phase === undefined ? state.messagePhases[itemID] : phase return [ { ...state, lifecycle, - messageItems: new Set([itemID]), - messagePhases: nextPhase === undefined ? {} : { [itemID]: nextPhase }, + message: { + id: itemID, + phase: phase === undefined && state.message?.id === itemID ? state.message.phase : phase, + }, }, events, ] @@ -985,6 +974,7 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { reasoningItems: { ...state.reasoningItems, [item.id]: { + open: true, encryptedContent: item.encrypted_content, summaryParts: { 0: "active" }, deltaIndexes: new Set(), @@ -996,8 +986,6 @@ 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 - // Pending tools always store the call id, so this also catches duplicates - // that disagree on whether `item.id` is present. 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 @@ -1026,7 +1014,7 @@ const onReasoningSummaryPartAdded = (state: ParserState, event: Event): StepResu const onReasoningSummaryPartDone = (state: ParserState, event: Event): StepResult => { if (event.item_id === undefined || event.summary_index === undefined) return [state, NO_EVENTS] const item = state.reasoningItems[event.item_id] - if (!item) return [state, NO_EVENTS] + if (!item?.open) return [state, NO_EVENTS] if (item.summaryParts[event.summary_index] !== "active") return [state, NO_EVENTS] return [ { @@ -1083,11 +1071,8 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( if (item.type === "message" && item.id !== undefined) { const itemPhase = messagePhase(item.phase) - const phase = itemPhase === undefined ? state.messagePhases[item.id] : itemPhase + const phase = itemPhase === undefined && state.message?.id === item.id ? state.message.phase : itemPhase const events: LLMEvent[] = [] - const messageItems = new Set(state.messageItems) - messageItems.delete(item.id) - const { [item.id]: _phase, ...messagePhases } = state.messagePhases return [ { ...state, @@ -1097,8 +1082,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( item.id, providerMetadata(state, { itemId: item.id, ...(phase === undefined ? {} : { phase }) }), ), - messageItems, - messagePhases, + message: state.message?.id === item.id ? undefined : state.message, }, events, ] satisfies StepResult @@ -1157,14 +1141,13 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( const metadata = reasoningMetadata(state, item) const reasoningItem = state.reasoningItems[item.id] if (reasoningItem) { + if (!reasoningItem.open) return [state, NO_EVENTS] satisfies StepResult const lifecycle = Object.entries(reasoningItem.summaryParts) .filter((entry) => entry[1] === "active" || entry[1] === "can-conclude") .reduce( (lifecycle, entry) => Lifecycle.reasoningEnd(lifecycle, events, `${item.id}:${entry[0]}`, metadata), state.lifecycle, ) - // Keep the fully-concluded entry so duplicate or late events for this - // item remain no-ops instead of reopening lifecycle state. return [ { ...state, @@ -1173,10 +1156,8 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( ...state.reasoningItems, [item.id]: { ...reasoningItem, + open: false, encryptedContent: item.encrypted_content ?? reasoningItem.encryptedContent, - summaryParts: Object.fromEntries( - Object.keys(reasoningItem.summaryParts).map((index) => [index, "concluded" as const]), - ), }, }, }, @@ -1194,6 +1175,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( reasoningItems: { ...state.reasoningItems, [item.id]: { + open: false, encryptedContent: item.encrypted_content, summaryParts: { 0: "concluded" }, deltaIndexes: new Set(), @@ -1223,7 +1205,7 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( if ( id === undefined || ((item.type !== "function_call" || !current.tools[id]) && - (item.type !== "reasoning" || !current.reasoningItems[id])) + (item.type !== "reasoning" || !current.reasoningItems[id]?.open)) ) return Effect.succeed([current, events] satisfies StepResult) return onOutputItemDone(current, { type: "response.output_item.done", item }).pipe( @@ -1399,8 +1381,7 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse completedTools: new Set(), lifecycle: Lifecycle.initial(), outputItems: {}, - messageItems: new Set(), - messagePhases: {}, + message: undefined, reasoningItems: {}, }) diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts new file mode 100644 index 00000000000..065288c0bf3 --- /dev/null +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -0,0 +1,467 @@ +import { describe, expect } from "bun:test" +import { Effect, Stream } from "effect" +import { LLM, LLMEvent } from "../../src/index.js" +import { OpenResponses } from "../../src/protocols/open-responses.js" +import { configure } from "../../src/providers/openai-compatible-responses.js" +import { LLMClient } from "../../src/route.js" +import { it } from "../lib/effect.js" +import { fixedResponse } from "../lib/http.js" +import { sseEvents } from "../lib/sse.js" + +const request = LLM.request({ + model: configure({ apiKey: "test-key", baseURL: "https://responses.example.test/v1" }).model("example-model"), + prompt: "Respond.", +}) +const completed = { type: "response.completed", response: { id: "resp_1" } } + +const collect = (...input: OpenResponses.Event[]) => + Effect.gen(function* () { + const events = yield* LLMClient.stream(request).pipe( + Stream.runCollect, + Effect.provide(fixedResponse(sseEvents(...input))), + ) + expectLifecycle( + events, + input.some((event) => event.type === "response.completed"), + ) + return events + }) + +// Deliberately local to these basic-item fixtures, not a general stream validator. +function expectLifecycle(events: ReadonlyArray, completed: boolean) { + const active = { text: new Set(), reasoning: new Set() } + const tools = new Map() + events.forEach((event) => { + if (event.type === "text-start" || event.type === "reasoning-start") { + const blocks = event.type === "text-start" ? active.text : active.reasoning + expect(blocks.size).toBe(0) + blocks.add(event.id) + } + if (event.type === "text-delta" || event.type === "reasoning-delta") { + expect((event.type === "text-delta" ? active.text : active.reasoning).has(event.id)).toBe(true) + } + if (event.type === "text-end" || event.type === "reasoning-end") { + expect((event.type === "text-end" ? active.text : active.reasoning).delete(event.id)).toBe(true) + } + if (event.type === "tool-input-start") { + expect(tools.has(event.id)).toBe(false) + tools.set(event.id, "started") + } + if (event.type === "tool-input-delta") expect(tools.get(event.id)).toBe("started") + if (event.type === "tool-input-end") { + expect(tools.get(event.id)).toBe("started") + tools.set(event.id, "ended") + } + if (event.type === "tool-call") { + expect(tools.get(event.id)).toBe("ended") + tools.set(event.id, "called") + } + // Incomplete responses may leave pending tool inputs without a call. + if (event.type === "finish" && completed) { + expect(active.text.size).toBe(0) + expect(active.reasoning.size).toBe(0) + expect([...tools.values()].every((status) => status === "called")).toBe(true) + } + }) + expect(events.filter(LLMEvent.is.stepStart)).toHaveLength(1) + expect(events[0]?.type).toBe("step-start") + expect(events.filter(LLMEvent.is.stepFinish)).toHaveLength(1) + expect(events.filter(LLMEvent.is.finish)).toHaveLength(1) + expect(events.slice(-2).map((event) => event.type)).toEqual(["step-finish", "finish"]) +} + +describe("Open Responses basic-item lifecycles", () => { + it.effect("closes implicit summary boundaries and ignores late events for completed reasoning", () => + Effect.gen(function* () { + const item = { type: "reasoning", id: "rs_1", encrypted_content: "encrypted-state" } + const events = yield* collect( + { type: "response.output_item.added", output_index: 0, item: { ...item, encrypted_content: null } }, + { type: "response.output_item.added", item: { ...item, encrypted_content: null } }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", summary_index: 0, delta: "First" }, + { type: "response.reasoning_summary_part.added", item_id: "rs_1", summary_index: 1 }, + { type: "response.reasoning_summary_part.done", item_id: "rs_1", summary_index: 0 }, + { type: "response.reasoning_summary_text.done", item_id: "rs_1", summary_index: 1, text: "Second" }, + // The third part omits both explicit summary boundaries. + { + type: "response.reasoning_summary_text.delta", + output_index: 0, + item_id: "wrong", + summary_index: 2, + delta: "Third", + }, + { type: "response.output_item.done", item }, + { type: "response.output_item.done", item }, + { type: "response.output_item.added", item }, + { type: "response.reasoning_summary_part.added", item_id: "rs_1", summary_index: 3 }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", summary_index: 3, delta: "late" }, + { type: "response.reasoning_summary_text.done", item_id: "rs_1", summary_index: 2, text: "late final" }, + { type: "response.reasoning_summary_part.done", item_id: "rs_1", summary_index: 3 }, + completed, + ) + + expect(events.filter((event) => event.type.startsWith("reasoning-"))).toEqual([ + { + type: "reasoning-start", + id: "rs_1:0", + providerMetadata: { "openai-compatible": { itemId: "rs_1", reasoningEncryptedContent: null } }, + }, + { type: "reasoning-delta", id: "rs_1:0", text: "First" }, + { type: "reasoning-end", id: "rs_1:0", providerMetadata: { "openai-compatible": { itemId: "rs_1" } } }, + { + type: "reasoning-start", + id: "rs_1:1", + providerMetadata: { "openai-compatible": { itemId: "rs_1", reasoningEncryptedContent: null } }, + }, + { type: "reasoning-delta", id: "rs_1:1", text: "Second" }, + { type: "reasoning-end", id: "rs_1:1", providerMetadata: { "openai-compatible": { itemId: "rs_1" } } }, + { + type: "reasoning-start", + id: "rs_1:2", + providerMetadata: { "openai-compatible": { itemId: "rs_1", reasoningEncryptedContent: null } }, + }, + { type: "reasoning-delta", id: "rs_1:2", text: "Third" }, + { + type: "reasoning-end", + id: "rs_1:2", + providerMetadata: { "openai-compatible": { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, + }, + ]) + }), + ) + + it.effect("preserves done-only encrypted reasoning without replaying its summary or late events", () => + Effect.gen(function* () { + const item = { + type: "reasoning", + id: "rs_1", + encrypted_content: "encrypted-state", + summary: [{ type: "summary_text", text: "Not streamed" }], + } + const events = yield* collect( + { type: "response.output_item.done", item }, + { type: "response.output_item.done", item }, + { type: "response.output_item.added", item }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "late" }, + { type: "response.reasoning_summary_part.added", item_id: "rs_1", summary_index: 1 }, + { type: "response.reasoning_summary_text.done", item_id: "rs_1", summary_index: 1, text: "late final" }, + completed, + // Route termination must also prevent events after response completion. + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_after" } }, + ) + expect(events.filter((event) => event.type.startsWith("reasoning-"))).toEqual([ + { + type: "reasoning-start", + id: "rs_1", + providerMetadata: { "openai-compatible": { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, + }, + { + type: "reasoning-end", + id: "rs_1", + providerMetadata: { "openai-compatible": { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, + }, + ]) + }), + ) + + it.effect("forgets never-streamed messages at implicit boundaries and preserves refusal phases", () => + Effect.gen(function* () { + const events = yield* collect( + { type: "response.output_item.added", item: { type: "message", id: "msg_empty" } }, + { type: "response.output_item.added", item: { type: "message", id: "msg_1", phase: "commentary" } }, + { type: "response.output_text.done", item_id: "msg_1", text: "Checking" }, + { type: "response.output_text.done", item_id: "msg_1", text: "Duplicate" }, + { type: "response.output_item.added", item: { type: "message", id: "msg_2", phase: null } }, + { type: "response.output_text.delta", item_id: "msg_empty", delta: "stale" }, + { type: "response.output_text.done", item_id: "msg_empty", text: "stale final" }, + { type: "response.output_text.delta", item_id: "msg_1", delta: "late" }, + { type: "response.refusal.delta", item_id: "msg_2", delta: "Cannot help." }, + { type: "response.refusal.done", item_id: "msg_2", refusal: "Cannot help." }, + { type: "response.output_item.done", item: { type: "message", id: "msg_2", phase: "final_answer" } }, + { type: "response.output_item.added", item: { type: "message", id: "msg_3", phase: null } }, + { type: "response.refusal.done", item_id: "msg_3", refusal: "Done-only refusal." }, + { type: "response.output_item.done", item: { type: "message", id: "msg_3" } }, + completed, + ) + expect(events.filter((event) => event.type.startsWith("text-"))).toEqual([ + { + type: "text-start", + id: "msg_1", + providerMetadata: { "openai-compatible": { itemId: "msg_1", phase: "commentary" } }, + }, + { type: "text-delta", id: "msg_1", text: "Checking" }, + { + type: "text-end", + id: "msg_1", + providerMetadata: { "openai-compatible": { itemId: "msg_1", phase: "commentary" } }, + }, + { + type: "text-start", + id: "msg_2", + providerMetadata: { "openai-compatible": { itemId: "msg_2", phase: null } }, + }, + { type: "text-delta", id: "msg_2", text: "Cannot help." }, + { + type: "text-end", + id: "msg_2", + providerMetadata: { "openai-compatible": { itemId: "msg_2", phase: "final_answer" } }, + }, + { + type: "text-start", + id: "msg_3", + providerMetadata: { "openai-compatible": { itemId: "msg_3", phase: null } }, + }, + { type: "text-delta", id: "msg_3", text: "Done-only refusal." }, + { type: "text-end", id: "msg_3", providerMetadata: { "openai-compatible": { itemId: "msg_3", phase: null } } }, + ]) + }), + ) + it.effect("allows a message to be registered again without inheriting its previous phase", () => + Effect.gen(function* () { + const events = yield* collect( + { type: "response.output_item.added", item: { type: "message", id: "msg_1", phase: "commentary" } }, + { type: "response.output_text.delta", item_id: "msg_1", delta: "First" }, + { type: "response.output_item.done", item: { type: "message", id: "msg_1" } }, + { type: "response.output_item.added", item: { type: "message", id: "msg_1" } }, + { type: "response.output_text.delta", item_id: "msg_1", delta: "Second" }, + { type: "response.output_item.done", item: { type: "message", id: "msg_1" } }, + completed, + ) + expect(events.filter(LLMEvent.is.textEnd)).toEqual([ + { + type: "text-end", + id: "msg_1", + providerMetadata: { "openai-compatible": { itemId: "msg_1", phase: "commentary" } }, + }, + { type: "text-end", id: "msg_1", providerMetadata: { "openai-compatible": { itemId: "msg_1" } } }, + ]) + expect(events.filter(LLMEvent.is.textDelta).map((event) => event.text)).toEqual(["First", "Second"]) + }), + ) + ;[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 } }, + }, + ]) + }), + ) + + 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("recovers pending items in completed output order with terminal encrypted metadata", () => + 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" }, + }, + { type: "response.function_call_arguments.delta", item_id: "fc_1", delta: '{"query":"draft"}' }, + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_1", encrypted_content: null } }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "Thinking" }, + { type: "response.reasoning_summary_part.done", item_id: "rs_1", summary_index: 0 }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [ + { type: "reasoning", id: "rs_1", encrypted_content: "terminal-state" }, + { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: '{"query":"final"}' }, + { type: "function_call", id: "fc_unseen", call_id: "call_unseen", name: "lookup", arguments: "{}" }, + ], + }, + }, + ) + expect(events.slice(5, -2)).toEqual([ + { + type: "reasoning-end", + id: "rs_1:0", + providerMetadata: { "openai-compatible": { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }, + { + type: "tool-input-end", + id: "call_1", + name: "lookup", + providerMetadata: { "openai-compatible": { itemId: "fc_1" } }, + }, + { + type: "tool-call", + id: "call_1", + name: "lookup", + input: { query: "final" }, + providerMetadata: { "openai-compatible": { itemId: "fc_1" } }, + }, + ]) + }), + ) + + 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( + { type: "response.output_item.added", item: { type: "reasoning", id: "" } }, + { type: "response.output_item.added", item: { type: "message", id: "" } }, + { type: "response.output_item.added", item: { type: "reasoning", id: "" } }, + { type: "response.reasoning_summary_text.delta", item_id: "", delta: "Thinking" }, + { type: "response.output_text.delta", item_id: "", delta: "Answer" }, + { type: "response.output_item.done", item: { type: "reasoning", id: "", encrypted_content: "state" } }, + { type: "response.output_item.done", item: { type: "message", id: "" } }, + completed, + ) + expect(events.filter(LLMEvent.is.reasoningDelta).map((event) => event.text)).toEqual(["Thinking"]) + expect(events.filter(LLMEvent.is.textDelta).map((event) => event.text)).toEqual(["Answer"]) + expect(events.filter(LLMEvent.is.reasoningEnd)).toEqual([ + { + type: "reasoning-end", + id: ":0", + providerMetadata: { "openai-compatible": { itemId: "", reasoningEncryptedContent: "state" } }, + }, + ]) + }), + ) + + it.effect("does not recover a completed tool from a tracked message with the same id", () => + Effect.gen(function* () { + const events = yield* collect( + { type: "response.output_item.added", item: { type: "message", id: "item_1" } }, + { type: "response.output_text.delta", item_id: "item_1", delta: "Answer" }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "function_call", id: "item_1", call_id: "call_1", name: "lookup", arguments: "{}" }], + }, + }, + ) + expect(events.filter(LLMEvent.is.toolCall)).toEqual([]) + expect(events.filter(LLMEvent.is.finish).map((event) => event.reason.normalized)).toEqual(["stop"]) + }), + ) + + it.effect("flushes pending calls and open text when completed output is absent", () => + Effect.gen(function* () { + const events = yield* collect( + { type: "response.output_item.added", item: { type: "message", id: "msg_1", phase: "final_answer" } }, + { 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: "{}" }, + }, + completed, + ) + // Generic terminal closure does not repeat the message's phase metadata. + 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: "text-end", id: "msg_1" }, + ]) + }), + ) + + it.effect("does not reconcile pending calls or terminal reasoning metadata on incomplete responses", () => + Effect.gen(function* () { + const events = yield* collect( + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_1", encrypted_content: null } }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "Partial" }, + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup" }, + }, + { type: "response.function_call_arguments.delta", item_id: "fc_1", delta: '{"query":' }, + { + type: "response.incomplete", + response: { + id: "resp_1", + incomplete_details: { reason: "max_output_tokens" }, + output: [ + { type: "reasoning", id: "rs_1", encrypted_content: "not-reconciled" }, + { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"not-reconciled"}', + }, + ], + }, + }, + ) + expect(events.filter(LLMEvent.is.toolInputEnd)).toEqual([]) + expect(events.filter(LLMEvent.is.toolCall)).toEqual([]) + expect(events.filter(LLMEvent.is.reasoningEnd)).toEqual([{ type: "reasoning-end", id: "rs_1:0" }]) + expect(events.filter(LLMEvent.is.finish)).toEqual([ + { + type: "finish", + reason: { normalized: "length", raw: "max_output_tokens" }, + providerMetadata: { "openai-compatible": { responseId: "resp_1", serviceTier: undefined } }, + }, + ]) + }), + ) +})