From 2f6a907a1ada2d6c39be4c0e189b1f5c095cdfa5 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Sat, 25 Jul 2026 14:45:15 -0500 Subject: [PATCH] fix(ai): reconcile terminal response items --- packages/ai/src/protocols/open-responses.ts | 188 +++++++++++++-- packages/ai/src/protocols/openai-responses.ts | 26 +- .../ai/test/provider/openai-responses.test.ts | 223 +++++++++++++++++- 3 files changed, 411 insertions(+), 26 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index 71f261f1a6..14f19716d2 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -280,6 +280,8 @@ export interface ParserState { readonly hasFunctionCall: boolean readonly completedItems: ReadonlySet readonly functionArguments: Readonly> + readonly outputIndexes: Readonly> + readonly outputSequence: ReadonlyArray readonly lifecycle: Lifecycle.State readonly messageItems: ReadonlySet readonly messagePhase: (value: unknown) => MessagePhase | null | undefined @@ -660,6 +662,12 @@ export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type) const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => { if (!event.delta) return [state, NO_EVENTS] + const key = + event.content_index === undefined + ? id + : event.content_index === 0 && state.outputText[`${id}:0`] === undefined && state.outputText[id] !== undefined + ? id + : `${id}:${event.content_index}` const events: LLMEvent[] = [] const phase = state.messagePhases[id] const metadata = phase === undefined ? undefined : providerMetadata(state, { phase }) @@ -668,7 +676,7 @@ const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepRe { ...state, lifecycle: Lifecycle.textDelta(lifecycle, events, id, event.delta), - outputText: { ...state.outputText, [id]: `${state.outputText[id] ?? ""}${event.delta}` }, + outputText: { ...state.outputText, [key]: `${state.outputText[key] ?? ""}${event.delta}` }, }, events, ] @@ -694,8 +702,14 @@ const onOutputTextDone = Effect.fn("OpenResponses.onOutputTextDone")(function* ( event: Event, id: string, ) { + const key = + event.content_index === undefined + ? id + : event.content_index === 0 && state.outputText[`${id}:0`] === undefined && state.outputText[id] !== undefined + ? id + : `${id}:${event.content_index}` const suffix = - event.text === undefined ? "" : yield* authoritativeSuffix(state, "output text", id, state.outputText[id] ?? "", event.text) + event.text === undefined ? "" : yield* authoritativeSuffix(state, "output text", key, state.outputText[key] ?? "", event.text) const [reconciled, deltaEvents] = onOutputTextDelta(state, { ...event, delta: suffix }, id) if (reconciled.messageItems.has(id)) return [reconciled, deltaEvents] satisfies StepResult const events: LLMEvent[] = [] @@ -927,29 +941,50 @@ const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgu return [{ ...state, lifecycle, tools: result.tools }, events] satisfies StepResult }) -const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (state: ParserState, event: Event) { +export const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( + state: ParserState, + event: Event, +) { const item = event.item if (!item) return [state, NO_EVENTS] satisfies StepResult if (item.type === "message" && item.id) { + const itemID = item.id const itemPhase = state.messagePhase(item.phase) const phase = itemPhase === undefined ? state.messagePhases[item.id] : itemPhase const messageItems = new Set([...state.messageItems, item.id]) const messagePhases = phase === undefined ? state.messagePhases : { ...state.messagePhases, [item.id]: phase } - const text = item.content - ?.filter((part) => part.type === "output_text" && part.text !== undefined) - .map((part) => part.text) - .join("") - const [reconciled, reconciledEvents] = - text === undefined - ? [{ ...state, messageItems, messagePhases }, NO_EVENTS] - : yield* onOutputTextDone({ ...state, messageItems, messagePhases }, { ...event, text }, item.id) + const reconciledEvents: LLMEvent[] = [] + const reconciled = yield* (item.content ?? []) + .map((part, contentIndex) => ({ part, contentIndex })) + .filter((entry) => entry.part.type === "output_text" && entry.part.text !== undefined) + .reduce>( + (effect, entry) => + effect.pipe( + Effect.flatMap( + Effect.fnUntraced(function* (current) { + const [next, nextEvents] = yield* onOutputTextDone( + current, + { ...event, content_index: entry.contentIndex, text: entry.part.text ?? "" }, + itemID, + ) + reconciledEvents.push(...nextEvents) + return next + }), + ), + ), + Effect.succeed({ ...state, messageItems, messagePhases }), + ) const events: LLMEvent[] = [] messageItems.delete(item.id) const { [item.id]: _phase, ...remainingPhases } = reconciled.messagePhases const metadata = phase === undefined ? undefined : providerMetadata(state, { phase }) const lifecycle = Lifecycle.textEnd(reconciled.lifecycle, events, item.id, metadata) - if (!events.length && state.outputText[item.id] !== undefined && metadata) + if ( + !events.length && + Object.keys(state.outputText).some((id) => id === item.id || id.startsWith(`${item.id}:`)) && + metadata + ) events.push(LLMEvent.textEnd({ id: item.id, providerMetadata: metadata })) return [ { @@ -957,6 +992,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( lifecycle, messageItems, messagePhases: remainingPhases, + completedItems: new Set([...reconciled.completedItems, item.id]), }, [...reconciledEvents, ...events], ] satisfies StepResult @@ -1007,6 +1043,47 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( const events: LLMEvent[] = [] const metadata = reasoningMetadata(state, item) const summaries = item.summary?.filter((part) => part.type === "summary_text" && part.text !== undefined) ?? [] + if (state.completedItems.has(item.id)) { + const reconciled = yield* summaries + .map((part, index) => ({ part, index })) + .reduce>( + (effect, entry) => + effect.pipe( + Effect.flatMap( + Effect.fnUntraced(function* (current) { + const [next, nextEvents] = yield* onReasoningDone(current, { + ...event, + item_id: item.id, + summary_index: entry.index, + text: entry.part.text, + }) + events.push(...nextEvents) + const endEvents: LLMEvent[] = [] + const lifecycle = Lifecycle.reasoningEnd( + next.lifecycle, + endEvents, + `${item.id}:${entry.index}`, + metadata, + ) + events.push( + ...(endEvents.length + ? endEvents + : [LLMEvent.reasoningEnd({ id: `${item.id}:${entry.index}`, providerMetadata: metadata })]), + ) + return { ...next, lifecycle } + }), + ), + ), + Effect.succeed(state), + ) + if (!summaries.length) { + const id = Object.keys(state.reasoningText) + .filter((id) => id === item.id || id.startsWith(`${item.id}:`)) + .at(-1) ?? `${item.id}:0` + events.push(LLMEvent.reasoningEnd({ id, providerMetadata: metadata })) + } + return [reconciled, events] satisfies StepResult + } if (summaries.length === 1 && !state.reasoningItems[item.id] && state.reasoningText[item.id] !== undefined) { const [reconciled, reconciledEvents] = yield* onReasoningDone(state, { ...event, @@ -1015,7 +1092,11 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( }) events.push(...reconciledEvents) return [ - { ...reconciled, lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata) }, + { + ...reconciled, + lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata), + completedItems: new Set([...reconciled.completedItems, item.id]), + }, events, ] satisfies StepResult } @@ -1049,7 +1130,13 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( text: entry.part.text, }) events.push(...nextEvents) - return next + const [done, doneEvents] = onReasoningSummaryPartDone(next, { + ...event, + item_id: item.id, + summary_index: entry.index, + }) + events.push(...doneEvents) + return done }), ), ), @@ -1064,20 +1151,38 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( reconciled.lifecycle, ) const { [item.id]: _removed, ...reasoningItems } = reconciled.reasoningItems - return [{ ...reconciled, lifecycle, reasoningItems }, events] satisfies StepResult + return [ + { + ...reconciled, + lifecycle, + reasoningItems, + completedItems: new Set([...reconciled.completedItems, item.id]), + }, + events, + ] satisfies StepResult } if (summaries.length) { events.push(LLMEvent.reasoningEnd({ id: `${item.id}:${summaries.length - 1}`, providerMetadata: metadata })) - return [reconciled, events] satisfies StepResult + return [ + { ...reconciled, completedItems: new Set([...reconciled.completedItems, item.id]) }, + events, + ] satisfies StepResult } if (!reconciled.lifecycle.reasoning.has(item.id)) { const lifecycle = Lifecycle.stepStart(reconciled.lifecycle, events) events.push(LLMEvent.reasoningStart({ id: item.id, providerMetadata: metadata })) events.push(LLMEvent.reasoningEnd({ id: item.id, providerMetadata: metadata })) - return [{ ...reconciled, lifecycle }, events] satisfies StepResult + return [ + { ...reconciled, lifecycle, completedItems: new Set([...reconciled.completedItems, item.id]) }, + events, + ] satisfies StepResult } return [ - { ...reconciled, lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata) }, + { + ...reconciled, + lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata), + completedItems: new Set([...reconciled.completedItems, item.id]), + }, events, ] satisfies StepResult } @@ -1085,14 +1190,37 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( return [state, NO_EVENTS] satisfies StepResult }) -const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (state: ParserState, event: Event) { +export const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( + state: ParserState, + event: Event, + onItemDone: (state: ParserState, event: Event) => Effect.Effect = onOutputItemDone, +) { const events: LLMEvent[] = [] - const reconciled = yield* (event.response?.output ?? []).reduce>( + const output = event.response?.output ?? [] + const terminalIndexes = Object.fromEntries( + output.flatMap((item, index) => (item.id === undefined ? [] : [[item.id, index] as const])), + ) + const emitted = state.outputSequence.filter((id) => terminalIndexes[id] !== undefined) + const expected = output.flatMap((item) => + item.id !== undefined && state.outputIndexes[item.id] !== undefined ? [item.id] : [], + ) + const missingBeforeEmitted = output.some( + (item, index) => + item.id !== undefined && + state.outputIndexes[item.id] === undefined && + Object.values(state.outputIndexes).some((seen) => seen > index), + ) + if (missingBeforeEmitted || emitted.some((id, index) => id !== expected[index])) + return yield* ProviderShared.eventError( + state.id, + `${state.name} terminal output conflicts with the streamed output order`, + ) + const reconciled = yield* output.reduce>( (effect, item) => effect.pipe( Effect.flatMap( Effect.fnUntraced(function* (current) { - const [next, nextEvents] = yield* onOutputItemDone(current, { ...event, item }) + const [next, nextEvents] = yield* onItemDone(current, { ...event, item }) events.push(...nextEvents) return next }), @@ -1140,7 +1268,21 @@ const providerError = (state: ParserState, event: Event, fallback: string) => { }) } -export const step = (state: ParserState, event: Event) => { +export const step = (state: ParserState, event: Event): Effect.Effect => { + const outputItemID = event.item_id ?? event.item?.id + if ( + outputItemID && + event.output_index !== undefined && + state.outputIndexes[outputItemID] === undefined + ) + return step( + { + ...state, + outputIndexes: { ...state.outputIndexes, [outputItemID]: event.output_index }, + outputSequence: [...state.outputSequence, outputItemID], + }, + { ...event, output_index: undefined }, + ) if (event.type === "response.output_text.delta" || event.type === "response.output_text.done") { if (!event.item_id) return ProviderShared.eventError(state.id, `${event.type} is missing item_id`) return event.type === "response.output_text.delta" @@ -1211,6 +1353,8 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse hasFunctionCall: false, completedItems: new Set(), functionArguments: {}, + outputIndexes: {}, + outputSequence: [], tools: ToolStream.empty(), lifecycle: Lifecycle.initial(), messageItems: new Set(), diff --git a/packages/ai/src/protocols/openai-responses.ts b/packages/ai/src/protocols/openai-responses.ts index 9c2a7db45e..510eedf531 100644 --- a/packages/ai/src/protocols/openai-responses.ts +++ b/packages/ai/src/protocols/openai-responses.ts @@ -194,6 +194,7 @@ const onHostedToolDone = Effect.fn("OpenAIResponses.onHostedToolDone")(function* state: OpenResponses.ParserState, item: HostedToolItem, ) { + if (state.completedItems.has(item.id)) return [state, []] satisfies OpenResponses.StepResult const tool = HOSTED_TOOLS[item.type] const providerMetadata = OpenResponses.providerMetadata(state, { itemId: item.id }) const events: LLMEvent[] = [] @@ -214,10 +215,28 @@ const onHostedToolDone = Effect.fn("OpenAIResponses.onHostedToolDone")(function* providerMetadata, }), ) - return [{ ...state, lifecycle }, events] satisfies OpenResponses.StepResult + return [ + { ...state, lifecycle, completedItems: new Set([...state.completedItems, item.id]) }, + events, + ] satisfies OpenResponses.StepResult }) +const onOutputItemDone = (state: OpenResponses.ParserState, event: OpenResponses.Event) => + event.item && isHostedToolItem(event.item) + ? onHostedToolDone(state, event.item) + : OpenResponses.onOutputItemDone(state, event) + const step = (state: OpenResponses.ParserState, event: OpenResponses.Event) => { + const outputItemID = event.item_id ?? event.item?.id + if (outputItemID && event.output_index !== undefined && state.outputIndexes[outputItemID] === undefined) + return step( + { + ...state, + outputIndexes: { ...state.outputIndexes, [outputItemID]: event.output_index }, + outputSequence: [...state.outputSequence, outputItemID], + }, + { ...event, output_index: undefined }, + ) if (event.type === "response.reasoning_text.delta" || event.type === "response.reasoning_summary.delta") return event.item_id ? Effect.succeed(OpenResponses.onReasoningDelta(state, event, event.item_id)) @@ -226,8 +245,9 @@ const step = (state: OpenResponses.ParserState, event: OpenResponses.Event) => { return event.item_id ? OpenResponses.onReasoningDone(state, event) : ProviderShared.eventError(ADAPTER, `${event.type} is missing item_id`) - if (event.type === "response.output_item.done" && event.item && isHostedToolItem(event.item)) - return onHostedToolDone(state, event.item) + if (event.type === "response.output_item.done") return onOutputItemDone(state, event) + if (event.type === "response.completed" || event.type === "response.incomplete") + return OpenResponses.onResponseFinish(state, event, onOutputItemDone) return OpenResponses.step(state, event) } diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index df0eac8ce6..2dccfe4ce7 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -913,6 +913,58 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("reconciles output text independently by content index", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "response.output_item.added", item: { type: "message", id: "msg_1" } }, + { + type: "response.output_text.delta", + item_id: "msg_1", + content_index: 0, + delta: "First", + }, + { + type: "response.output_text.done", + item_id: "msg_1", + content_index: 0, + text: "First.", + }, + { + type: "response.output_text.delta", + item_id: "msg_1", + content_index: 1, + delta: "Second", + }, + { + type: "response.output_text.done", + item_id: "msg_1", + content_index: 1, + text: "Second.", + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_1", + content: [ + { type: "output_text", text: "First." }, + { type: "output_text", text: "Second." }, + ], + }, + }, + { type: "response.completed", response: {} }, + ), + ), + ), + ) + + expect(response.text).toBe("First.Second.") + }), + ) + it.effect("recovers output from the authoritative terminal response", () => Effect.gen(function* () { const response = yield* LLMClient.generate(request).pipe( @@ -944,6 +996,52 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("rejects terminal output that cannot be restored to authoritative order", () => + Effect.gen(function* () { + const error = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { + type: "response.output_item.added", + output_index: 1, + item: { type: "message", id: "msg_later" }, + }, + { + type: "response.output_text.delta", + item_id: "msg_later", + output_index: 1, + content_index: 0, + delta: "Later", + }, + { + type: "response.completed", + response: { + output: [ + { + type: "message", + id: "msg_earlier", + content: [{ type: "output_text", text: "Earlier" }], + }, + { + type: "message", + id: "msg_later", + content: [{ type: "output_text", text: "Later" }], + }, + ], + }, + }, + ), + ), + ), + Effect.flip, + ) + + expect(error.reason._tag).toBe("InvalidProviderOutput") + expect(error.message).toContain("terminal output conflicts with the streamed output order") + }), + ) + it.effect("rejects authoritative text that conflicts with streamed deltas", () => Effect.gen(function* () { const error = yield* LLMClient.generate(request).pipe( @@ -1185,6 +1283,92 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("recovers terminal-only reasoning summaries sequentially", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents({ + type: "response.completed", + response: { + output: [ + { + type: "reasoning", + id: "rs_1", + encrypted_content: "encrypted-state", + summary: [ + { type: "summary_text", text: "First" }, + { type: "summary_text", text: "Second" }, + ], + }, + ], + }, + }), + ), + ), + ) + + expect(response.message.content.filter((part) => part.type === "reasoning")).toEqual([ + { + type: "reasoning", + text: "First", + providerMetadata: { openai: { itemId: "rs_1" } }, + }, + { + type: "reasoning", + text: "Second", + providerMetadata: { + openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" }, + }, + }, + ]) + expect( + response.events + .filter((event) => event.type.startsWith("reasoning-")) + .map((event) => `${event.type}:${event.id}`), + ).toEqual([ + "reasoning-start:rs_1:0", + "reasoning-delta:rs_1:0", + "reasoning-end:rs_1:0", + "reasoning-start:rs_1:1", + "reasoning-delta:rs_1:1", + "reasoning-end:rs_1:1", + ]) + }), + ) + + it.effect("updates completed reasoning metadata without adding a phantom block", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_1" } }, + { + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_1", encrypted_content: "stale", summary: [] }, + }, + { + type: "response.completed", + response: { + output: [{ type: "reasoning", id: "rs_1", encrypted_content: "final", summary: [] }], + }, + }, + ), + ), + ), + ) + + expect(response.message.content.filter((part) => part.type === "reasoning")).toEqual([ + { + type: "reasoning", + text: "", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "final" } }, + }, + ]) + }), + ) + it.effect("preserves encrypted reasoning metadata for continuation", () => Effect.gen(function* () { const response = yield* LLMClient.generate(request).pipe( @@ -1792,7 +1976,10 @@ describe("OpenAI Responses route", () => { const body = sseEvents( { type: "response.output_item.added", item }, { type: "response.output_item.done", item }, - { type: "response.completed", response: { usage: { input_tokens: 5, output_tokens: 1 } } }, + { + type: "response.completed", + response: { output: [item], usage: { input_tokens: 5, output_tokens: 1 } }, + }, ) const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body))) @@ -1851,6 +2038,40 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("recovers hosted tools from the terminal response", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents({ + type: "response.completed", + response: { + output: [ + { + type: "image_generation_call", + id: "ig_1", + status: "completed", + result: "AQID", + }, + ], + }, + }), + ), + ), + ) + + expect(response.events.find(LLMEvent.is.toolResult)).toMatchObject({ + id: "ig_1", + name: "image_generation", + providerExecuted: true, + result: { + type: "content", + value: [{ type: "file", uri: "data:image/png;base64,AQID", mime: "image/png" }], + }, + }) + }), + ) + it.effect("rejects malformed image generation base64", () => Effect.gen(function* () { const error = yield* LLMClient.generate(request).pipe(