From 5acfc2dcb8a6e6ce82323475fa2e7282f3fd0d14 Mon Sep 17 00:00:00 2001 From: Mike <5853861+snotty@users.noreply.github.com> Date: Thu, 27 Aug 2026 05:08:42 +0800 Subject: [PATCH] fix(openai): reconcile rotated terminal message IDs (#122560) * fix(ai): reconcile rotated Responses output identities Retain output position, completion, and block ownership in one shared tracker for terminal text recovery and reasoning backfill. Preserve unindexed aliases and reject changed output types or stable tool-call IDs. Co-authored-by: snotty * docs: keep Copilot release notes in PR context --------- Co-authored-by: Peter Steinberger Co-authored-by: snotty --- .../openai-responses-stream-internal.ts | 62 +++++---- .../openai-responses-stream-slots-internal.ts | 66 ++++++++-- ...enai-responses-stream-terminal-internal.ts | 76 +++++------ ...responses-stream-terminal-recovery.test.ts | 118 +++++++++++++++++- 4 files changed, 241 insertions(+), 81 deletions(-) diff --git a/packages/ai/src/transports/openai-responses-stream-internal.ts b/packages/ai/src/transports/openai-responses-stream-internal.ts index 5349c43fffe8..a51b3b5e4b59 100644 --- a/packages/ai/src/transports/openai-responses-stream-internal.ts +++ b/packages/ai/src/transports/openai-responses-stream-internal.ts @@ -27,7 +27,7 @@ import { encodeTextSignatureV1 } from "./openai-responses-replay-internal.js"; import { adaptResponsesStream } from "./openai-responses-stream-observer-internal.js"; import { appendResponsesPendingTextDelta, - createResponsesOutputContentIndex, + createResponsesOutputTracker, createResponsesOutputSlotTracker, readResponsesOutputIndex, type ResponsesStreamOutputSlot, @@ -69,9 +69,7 @@ export async function processResponsesStream( >; const streamingToolCalls = createResponsesToolCallTracker(); const outputSlots = createResponsesOutputSlotTracker(); - const reasoningBlocksById = new Map(); - const outputItemContentIndexes = createResponsesOutputContentIndex(); - const startedTextBlocksByItemId = new Map(); + const outputs = createResponsesOutputTracker(); let terminalResponse: CompletedResponse | null | undefined; let lastTextBlock: TextBlockReference | null = null; const blocks = output.content; @@ -87,10 +85,10 @@ export async function processResponsesStream( item, block, contentIndex: blocks.length, + outputIndex: readResponsesOutputIndex(event), } satisfies ResponsesOutputSlot; blocks.push(block); - reasoningBlocksById.set(item.id, block); - outputItemContentIndexes.set(item, slot.contentIndex); + outputs.set(item, slot.contentIndex, slot.outputIndex); outputSlots.register(event, slot); stream.push({ type: "thinking_start", contentIndex: slot.contentIndex, partial: output }); return slot; @@ -112,17 +110,13 @@ export async function processResponsesStream( item: messageItem, block, contentIndex: block ? blocks.length : undefined, + outputIndex: readResponsesOutputIndex(event), pendingText: collapseCandidate ? "" : null, collapseCandidate, } satisfies ResponsesOutputSlot; if (block) { blocks.push(block); - outputItemContentIndexes.set(messageItem, slot.contentIndex ?? blocks.length - 1); - startedTextBlocksByItemId.set(messageItem.id, { - block, - index: slot.contentIndex ?? blocks.length - 1, - phase: messageItem.phase ?? undefined, - }); + outputs.set(messageItem, slot.contentIndex ?? blocks.length - 1, slot.outputIndex); } outputSlots.register(event, slot); if (slot.contentIndex !== undefined) { @@ -144,12 +138,6 @@ export async function processResponsesStream( } return readResponsesOutputIndex(event) === undefined ? undefined : outputSlots.get(event); }; - const getOrCreateOutputSlot = ( - event: object, - item: ResponseOutputItem | ResponsesStreamOutputMessage, - ): ResponsesOutputSlot | undefined => { - return resolveOutputItemSlot(event, item) ?? createOutputSlot(event, item); - }; const materializeDeferredTextSlot = ( slot: Extract, ): void => { @@ -166,12 +154,7 @@ export async function processResponsesStream( }; blocks.push(slot.block); slot.contentIndex = blocks.length - 1; - outputItemContentIndexes.set(slot.item, slot.contentIndex); - startedTextBlocksByItemId.set(slot.item.id, { - block: slot.block, - index: slot.contentIndex, - phase: slot.item.phase ?? undefined, - }); + outputs.set(slot.item, slot.contentIndex, slot.outputIndex); stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output }); if (text) { stream.push({ @@ -199,9 +182,7 @@ export async function processResponsesStream( stream, model, options, - reasoningBlocksById, - startedTextBlocksByItemId, - outputItemContentIndexes, + outputs, getLastTextBlock: () => lastTextBlock, setLastTextBlock: (block) => { lastTextBlock = block; @@ -262,7 +243,7 @@ export async function processResponsesStream( outputSlots.register(event, { type: "toolCall", toolCall: toolCallState }); } output.content.push(toolCallBlock); - outputItemContentIndexes.set(item, contentIndex); + outputs.set(item, contentIndex, readResponsesOutputIndex(event)); stream.push({ type: "toolcall_start", contentIndex, partial: output }); } } else if (event.type === "response.reasoning_summary_part.added") { @@ -455,13 +436,19 @@ export async function processResponsesStream( const existingOutputSlot = resolveOutputItemSlot(event, item); materializeDeferredTextSlots(existingOutputSlot); - const outputSlot = existingOutputSlot ?? getOrCreateOutputSlot(event, item); + const outputSlot = existingOutputSlot ?? createOutputSlot(event, item); compactionTracker.completed(item, blocks.length); if (item.type === "reasoning" && outputSlot?.type === "thinking") { const summaryText = item.summary?.map((s) => s.text).join("\n\n") || ""; const contentText = item.content?.map((c) => c.text).join("\n\n") || ""; outputSlot.block.thinking = summaryText || contentText || outputSlot.block.thinking; outputSlot.block.thinkingSignature = JSON.stringify(item); + outputs.set( + item, + outputSlot.contentIndex, + readResponsesOutputIndex(event) ?? outputSlot.outputIndex, + true, + ); if (item.encrypted_content && options?.reasoningReplayMetadata) { outputSlot.block[OPENAI_RESPONSES_REASONING_REPLAY_BLOCK_META_KEY] = options.reasoningReplayMetadata; @@ -516,7 +503,12 @@ export async function processResponsesStream( partial: output, }); lastTextBlock = outputSlot.collapseCandidate; - outputItemContentIndexes.set(item, outputSlot.collapseCandidate.index); + outputs.set( + item, + outputSlot.collapseCandidate.index, + readResponsesOutputIndex(event) ?? outputSlot.outputIndex, + true, + ); } else { if (!outputSlot.block) { // Deferred distinct message: open its block now, balanced with the @@ -541,7 +533,12 @@ export async function processResponsesStream( throw new Error("Responses stream finalized text without a content index"); } lastTextBlock = { block: outputSlot.block, index: contentIndex, phase }; - outputItemContentIndexes.set(item, contentIndex); + outputs.set( + item, + contentIndex, + readResponsesOutputIndex(event) ?? outputSlot.outputIndex, + true, + ); stream.push({ type: "text_end", contentIndex, @@ -550,7 +547,6 @@ export async function processResponsesStream( }); } outputSlots.forget(outputSlot); - startedTextBlocksByItemId.delete(item.id); } else if (item.type === "function_call") { const streamingToolCall = streamingToolCalls.resolve( event, @@ -624,7 +620,7 @@ export async function processResponsesStream( toolCall, partial: output, }); - outputItemContentIndexes.set(item, contentIndex); + outputs.set(item, contentIndex, readResponsesOutputIndex(event), true); } } else if (event.type === "response.completed" || event.type === "response.incomplete") { if (streamingToolCalls.hasActive()) { diff --git a/packages/ai/src/transports/openai-responses-stream-slots-internal.ts b/packages/ai/src/transports/openai-responses-stream-slots-internal.ts index 1de5eda7c481..fdb245ad7a4b 100644 --- a/packages/ai/src/transports/openai-responses-stream-slots-internal.ts +++ b/packages/ai/src/transports/openai-responses-stream-slots-internal.ts @@ -11,12 +11,14 @@ export type ResponsesStreamOutputSlot = item: ResponseReasoningItem; block: ResponsesThinkingBlock; contentIndex: number; + outputIndex: number | undefined; } | { type: "text"; item: TMessage; block: TextContent | null; contentIndex: number | undefined; + outputIndex: number | undefined; pendingText: string | null; collapseCandidate: TextBlockReference | null; } @@ -33,8 +35,18 @@ type ResponsesOutputIdentityItem = { call_id?: string | null; }; -export function createResponsesOutputContentIndex() { - const indexes = new Map(); +type ResponsesOutputState = { + type: string; + callId?: string | null; + outputIndex?: number; + contentIndex: number; + completed: boolean; +}; + +export type ResponsesOutputTracker = ReturnType; + +export function createResponsesOutputTracker() { + const outputs = new Map(); const identity = (item: ResponsesOutputIdentityItem): string | undefined => { if ((item.type === "reasoning" || item.type === "message") && item.id) { return `${item.type}:${item.id}`; @@ -43,15 +55,53 @@ export function createResponsesOutputContentIndex() { ? `function_call:${item.call_id ?? item.id ?? ""}` : undefined; }; + const get = (item: ResponsesOutputIdentityItem, outputIndex?: number) => { + const key = identity(item); + const output = + (outputIndex === undefined ? undefined : outputs.get(outputIndex)) ?? + (key === undefined ? undefined : outputs.get(key)); + if ( + !output || + (outputIndex !== undefined && + output.outputIndex !== undefined && + output.outputIndex !== outputIndex) + ) { + return undefined; + } + if ( + output.type !== item.type || + (output.callId && item.call_id && output.callId !== item.call_id) + ) { + throw new Error("Responses stream changed output item identity"); + } + return output; + }; return { - get(item: ResponsesOutputIdentityItem): number | undefined { - const key = identity(item); - return key === undefined ? undefined : indexes.get(key); - }, - set(item: ResponsesOutputIdentityItem, contentIndex: number): void { + get, + set( + item: ResponsesOutputIdentityItem, + contentIndex: number, + outputIndex?: number, + completed = false, + ): void { + const output: ResponsesOutputState = get(item, outputIndex) ?? { + type: item.type, + contentIndex, + completed, + }; + Object.assign(output, { contentIndex, completed }); + if (item.call_id) { + output.callId = item.call_id; + } + if (outputIndex !== undefined) { + output.outputIndex = outputIndex; + outputs.set(outputIndex, output); + } + // Output positions survive encrypted ID rotation. Aliases retain the + // supported unindexed stream contract without owning lifecycle state. const key = identity(item); if (key !== undefined) { - indexes.set(key, contentIndex); + outputs.set(key, output); } }, }; diff --git a/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts b/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts index 8302db68dec0..918228fa2f04 100644 --- a/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts +++ b/packages/ai/src/transports/openai-responses-stream-terminal-internal.ts @@ -30,6 +30,7 @@ import { type OpenAIResponsesReasoningReplayMetadata, } from "./openai-responses-contracts.js"; import { encodeTextSignatureV1 } from "./openai-responses-replay-internal.js"; +import type { ResponsesOutputTracker } from "./openai-responses-stream-slots-internal.js"; import { parseTerminalToolCallArguments } from "./transport-stream-shared.js"; export type ResponsesEventSink = { push(event: AssistantMessageEvent): void }; @@ -109,12 +110,7 @@ export function createResponsesTerminalController(params: { stream: ResponsesEventSink; model: Model; options?: TerminalOptions; - reasoningBlocksById: Map; - startedTextBlocksByItemId: Map; - outputItemContentIndexes: { - get: (item: ResponseOutputItem) => number | undefined; - set: (item: ResponseOutputItem, contentIndex: number) => void; - }; + outputs: ResponsesOutputTracker; getLastTextBlock: () => TextBlockReference | null; setLastTextBlock: (block: TextBlockReference | null) => void; markFinalized: () => void; @@ -122,12 +118,13 @@ export function createResponsesTerminalController(params: { const { output, stream, model, options } = params; const blocks = output.content; const backfillReasoning = (items: ResponseOutputItem[]) => { - for (const item of items) { + for (const [outputIndex, item] of items.entries()) { if (item.type !== "reasoning" || !item.encrypted_content) { continue; } - const block = params.reasoningBlocksById.get(item.id); - if (!block?.thinkingSignature) { + const tracked = params.outputs.get(item, outputIndex); + const block = tracked && blocks[tracked.contentIndex]; + if (block?.type !== "thinking" || !block.thinkingSignature) { continue; } const stored = JSON.parse(block.thinkingSignature) as ResponseReasoningItem; @@ -138,11 +135,13 @@ export function createResponsesTerminalController(params: { }); } if (options?.reasoningReplayMetadata) { - block[OPENAI_RESPONSES_REASONING_REPLAY_BLOCK_META_KEY] = options.reasoningReplayMetadata; + Object.assign(block, { + [OPENAI_RESPONSES_REASONING_REPLAY_BLOCK_META_KEY]: options.reasoningReplayMetadata, + }); } } }; - const appendText = (item: ResponseOutputMessage): number | undefined => { + const appendText = (item: ResponseOutputMessage, contentIndex?: number): number | undefined => { const text = (Array.isArray(item.content) ? item.content : []) .map((part) => { const content = part as { type: string; text?: string; refusal?: string }; @@ -151,30 +150,30 @@ export function createResponsesTerminalController(params: { : (content.refusal ?? ""); }) .join(""); - const started = params.startedTextBlocksByItemId.get(item.id); + const block = contentIndex === undefined ? undefined : blocks[contentIndex]; + const started = block?.type === "text" ? block : undefined; if (!text && !started) { return undefined; } const phase = item.phase ?? undefined; - if (started) { - const previousText = started.block.text; - started.block.text = text; - started.block.textSignature = encodeTextSignatureV1(item.id, phase); - params.setLastTextBlock({ block: started.block, index: started.index, phase }); - params.startedTextBlocksByItemId.delete(item.id); + if (started && contentIndex !== undefined) { + const previousText = started.text; + started.text = text; + started.textSignature = encodeTextSignatureV1(item.id, phase); + params.setLastTextBlock({ block: started, index: contentIndex, phase }); if (text.startsWith(previousText)) { const delta = text.slice(previousText.length); if (delta) { - stream.push({ type: "text_delta", contentIndex: started.index, delta }); + stream.push({ type: "text_delta", contentIndex, delta }); } } stream.push({ type: "text_end", - contentIndex: started.index, + contentIndex, content: text, partial: output, }); - return started.index; + return contentIndex; } const previous = params.getLastTextBlock(); const collapse = resolveResponsesMessageSnapshotCollapse({ @@ -193,14 +192,14 @@ export function createResponsesTerminalController(params: { }); return previous.index; } - const block: TextContent = { + const newBlock: TextContent = { type: "text", text, textSignature: encodeTextSignatureV1(item.id, phase), }; - blocks.push(block); + blocks.push(newBlock); const index = blocks.length - 1; - params.setLastTextBlock({ block, index, phase }); + params.setLastTextBlock({ block: newBlock, index, phase }); stream.push({ type: "text_start", contentIndex: index, partial: output }); stream.push({ type: "text_end", contentIndex: index, content: text, partial: output }); return index; @@ -221,19 +220,17 @@ export function createResponsesTerminalController(params: { }; const recoverTerminalOutput = (items: ResponseOutputItem[], includeToolCalls: boolean) => { let hasCompletedLaterOutput = false; - for (const item of items.toReversed()) { + for (const [outputIndex, item] of [...items.entries()].toReversed()) { + const tracked = params.outputs.get(item, outputIndex); if (item.type === "reasoning") { // Terminal snapshots only backfill streamed reasoning; missing reasoning is never emitted. - hasCompletedLaterOutput ||= params.outputItemContentIndexes.get(item) !== undefined; + hasCompletedLaterOutput ||= tracked !== undefined; continue; } if (item.type !== "message" && item.type !== "function_call") { continue; } - if ( - params.outputItemContentIndexes.get(item) !== undefined || - (item.type === "message" && params.startedTextBlocksByItemId.has(item.id)) - ) { + if (tracked) { hasCompletedLaterOutput = true; continue; } @@ -250,13 +247,13 @@ export function createResponsesTerminalController(params: { } for (const [terminalIndex, item] of items.entries()) { if (item.type === "message") { - const contentIndex = params.outputItemContentIndexes.get(item); - if (contentIndex !== undefined && !params.startedTextBlocksByItemId.has(item.id)) { + const tracked = params.outputs.get(item, terminalIndex); + if (tracked?.completed) { continue; } - const appendedIndex = appendText(item); + const appendedIndex = appendText(item, tracked?.contentIndex); if (appendedIndex !== undefined) { - params.outputItemContentIndexes.set(item, appendedIndex); + params.outputs.set(item, appendedIndex, terminalIndex, true); } } else { params.setLastTextBlock(null); @@ -267,8 +264,11 @@ export function createResponsesTerminalController(params: { output.providerReplay.data === item.encrypted_content; if (item.type === "compaction" && !alreadyCapturedCompaction) { let replayIndex = blocks.length; - for (const laterItem of items.slice(terminalIndex + 1)) { - const laterContentIndex = params.outputItemContentIndexes.get(laterItem); + for (const [laterIndex, laterItem] of items.entries()) { + if (laterIndex <= terminalIndex) { + continue; + } + const laterContentIndex = params.outputs.get(laterItem, laterIndex)?.contentIndex; if (laterContentIndex !== undefined) { replayIndex = laterContentIndex; break; @@ -282,10 +282,10 @@ export function createResponsesTerminalController(params: { options?.reasoningReplayMetadata, ); } else if (includeToolCalls && item.type === "function_call") { - if (params.outputItemContentIndexes.get(item) !== undefined) { + if (params.outputs.get(item, terminalIndex)) { continue; } - params.outputItemContentIndexes.set(item, appendToolCall(item)); + params.outputs.set(item, appendToolCall(item), terminalIndex, true); } } } diff --git a/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts b/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts index bf8fec77ab26..35e5e7888c4c 100644 --- a/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts +++ b/packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts @@ -692,10 +692,124 @@ const fixtures: ParityFixture[] = [ error: null, }, }, + { + name: "terminal reasoning backfill follows the completed output slot", + events: [ + { + type: "response.output_item.added", + output_index: 0, + item: { id: "rs_backfill", type: "reasoning", summary: [] }, + }, + { + type: "response.output_item.done", + output_index: 0, + item: { + id: "rs_backfill", + type: "reasoning", + summary: [{ type: "summary_text", text: "Checking the answer." }], + }, + }, + completed("resp_reasoning_backfill", [ + { + id: "rs_backfill", + type: "reasoning", + summary: [{ type: "summary_text", text: "Checking the answer." }], + encrypted_content: "fixture-reasoning-ciphertext", + }, + ]), + ], + canonical: { + events: [ + { type: "thinking_start", contentIndex: 0 }, + { type: "thinking_end", contentIndex: 0, content: "Checking the answer." }, + ], + content: [{ type: "thinking", thinking: "Checking the answer.", encrypted: true }], + responseId: "resp_reasoning_backfill", + stopReason: "stop", + error: null, + }, + }, + { + name: "equal text in distinct output slots remains distinct", + events: [ + ...[0, 1].flatMap((outputIndex) => [ + { + type: "response.output_item.added", + output_index: outputIndex, + item: { id: `msg_equal_${outputIndex}`, type: "message", content: [] }, + }, + { + type: "response.output_item.done", + output_index: outputIndex, + item: { + id: `msg_equal_${outputIndex}`, + type: "message", + content: [{ type: "output_text", text: "Again." }], + }, + }, + ]), + completed( + "resp_equal_messages", + [0, 1].map((outputIndex) => ({ + id: `msg_equal_${outputIndex}`, + type: "message", + content: [{ type: "output_text", text: "Again." }], + })), + ), + ], + canonical: { + events: [ + { type: "text_start", contentIndex: 0 }, + { type: "text_end", contentIndex: 0, content: "Again." }, + { type: "text_start", contentIndex: 1 }, + { type: "text_end", contentIndex: 1, content: "Again." }, + ], + content: [ + { type: "text", text: "Again." }, + { type: "text", text: "Again." }, + ], + responseId: "resp_equal_messages", + stopReason: "stop", + error: null, + }, + }, ]; -describe("Responses terminal recovery fixtures", () => { +describe.each(["stable", "rotated"])("Responses terminal recovery with %s item IDs", (ids) => { it.each(fixtures)("$name", async (fixture) => { - expect(await runFixture(fixture.events)).toEqual(fixture.canonical); + let sequence = 0; + const events: ParityFixture["events"] = JSON.parse( + JSON.stringify(fixture.events), + (key, value: unknown) => + ids === "rotated" && + (key === "id" || key === "item_id") && + typeof value === "string" && + /^(msg|rs)_/.test(value) + ? `${value}_event_${sequence++}` + : value, + ); + expect(await runFixture(events)).toEqual(fixture.canonical); }); }); + +it.each(["type", "call_id"])("rejects terminal output that changes its %s", async (field) => { + const streamed = { + type: "function_call", + id: "fc_original", + call_id: "call_original", + name: "lookup", + arguments: "{}", + status: "completed", + }; + const terminal = + field === "type" + ? { type: "message", id: "msg_changed", content: [{ type: "output_text", text: "Changed" }] } + : { ...streamed, call_id: "call_changed" }; + const result = await runFixture([ + { type: "response.output_item.done", output_index: 0, item: streamed }, + completed("resp_changed_identity", [terminal]), + ]); + expect(result.error).toBe("Responses stream changed output item identity"); + expect(result.content).toHaveLength(1); + expect(result.events.filter((event) => event.type === "toolcall_end")).toHaveLength(1); +});