diff --git a/src/agents/openai-transport-stream.test.ts b/src/agents/openai-transport-stream.test.ts index 4f92e46ef56a..be02b34fa1b1 100644 --- a/src/agents/openai-transport-stream.test.ts +++ b/src/agents/openai-transport-stream.test.ts @@ -435,6 +435,216 @@ describe("openai transport stream", () => { ]); }); + it("collapses cumulative message snapshot items into one text block (#91959)", async () => { + const model = createAzureResponsesModel(); + const output = createResponsesAssistantOutput(model); + const pushSpy = vi.fn(); + const snapshot1 = "Scaled dot-product attention"; + const snapshot2 = "Scaled dot-product attention divides by sqrt(d_k)"; + const snapshot3 = "Scaled dot-product attention divides by sqrt(d_k) before softmax."; + const messageItem = (id: string, text: string) => ({ + type: "message", + id, + phase: "final_answer", + content: [{ type: "output_text", text }], + }); + + await testing.processResponsesStream( + streamChunks([ + { + type: "response.output_item.added", + item: { type: "message", id: "msg_1", phase: "final_answer" }, + }, + { type: "response.output_text.delta", delta: snapshot1 }, + { type: "response.output_item.done", item: messageItem("msg_1", snapshot1) }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_2", phase: "final_answer" }, + }, + { type: "response.output_item.done", item: messageItem("msg_2", snapshot2) }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_3", phase: "final_answer" }, + }, + { type: "response.output_item.done", item: messageItem("msg_3", snapshot3) }, + { + type: "response.completed", + response: { id: "resp-snapshots", status: "completed" }, + }, + ]), + output, + { push: pushSpy }, + model, + ); + + expect(output.content).toEqual([ + { + type: "text", + text: snapshot3, + textSignature: '{"v":1,"id":"msg_3","phase":"final_answer"}', + }, + ]); + // Balanced lifecycle: one text_start, all events on index 0, and each + // collapsed snapshot re-ends the same block. + const textEvents = pushSpy.mock.calls + .map(([event]) => event as { type: string; contentIndex?: number }) + .filter((event) => event.type.startsWith("text_")); + expect(textEvents.map((event) => [event.type, event.contentIndex])).toEqual([ + ["text_start", 0], + ["text_delta", 0], + ["text_end", 0], + ["text_end", 0], + ["text_end", 0], + ]); + }); + + it("keeps prefix-nested message items separated by a tool call as separate blocks", async () => { + const model = createAzureResponsesModel(); + const output = createResponsesAssistantOutput(model); + const messageEvents = (id: string, text: string) => [ + { type: "response.output_item.added", item: { type: "message", id } }, + { + type: "response.output_item.done", + item: { type: "message", id, content: [{ type: "output_text", text }] }, + }, + ]; + + await testing.processResponsesStream( + streamChunks([ + ...messageEvents("msg_1", "Done."), + { + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "write", + arguments: "{}", + }, + }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "write", + arguments: "{}", + }, + }, + ...messageEvents("msg_2", "Done."), + { + type: "response.completed", + response: { id: "resp-tool-boundary", status: "completed" }, + }, + ]), + output, + { push: vi.fn() }, + model, + ); + + // The post-tool message is a real reply, not a snapshot of the pre-tool one. + expect(output.content.map((block) => block.type)).toEqual(["text", "toolCall", "text"]); + expect(output.content[2]).toMatchObject({ type: "text", text: "Done." }); + }); + + it("collapses cumulative message snapshots in completed-response backfill (#91959)", async () => { + const model = createAzureResponsesModel(); + const output = createResponsesAssistantOutput(model); + + await testing.processResponsesStream( + streamChunks([ + { + type: "response.completed", + response: { + id: "resp-backfill-snapshots", + status: "completed", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "The answer" }], + }, + { + type: "message", + id: "msg_2", + role: "assistant", + content: [{ type: "output_text", text: "The answer is 42." }], + }, + { + type: "message", + id: "msg_3", + role: "assistant", + content: [{ type: "output_text", text: "The answer" }], + }, + ], + }, + }, + ]), + output, + { push: vi.fn() }, + model, + ); + + // msg_2 strictly extends msg_1 and collapses into it; msg_3 shrinks back + // and is an independently identified message, so it stays a real block. + expect(output.content).toEqual([ + { + type: "text", + text: "The answer is 42.", + textSignature: '{"v":1,"id":"msg_2"}', + }, + { + type: "text", + text: "The answer", + textSignature: '{"v":1,"id":"msg_3"}', + }, + ]); + }); + + it("keeps backfill message items separated by a reasoning item as distinct blocks", async () => { + const model = createAzureResponsesModel(); + const output = createResponsesAssistantOutput(model); + + await testing.processResponsesStream( + streamChunks([ + { + type: "response.completed", + response: { + id: "resp-backfill-reasoning-boundary", + status: "completed", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "Step one." }], + }, + { type: "reasoning", id: "rs_1", summary: [] }, + { + type: "message", + id: "msg_2", + role: "assistant", + content: [{ type: "output_text", text: "Step one. Step two." }], + }, + ], + }, + }, + ]), + output, + { push: vi.fn() }, + model, + ); + + // A reasoning item is a real boundary even in backfill: msg_2 must not + // collapse into msg_1 despite being a strict extension (mirrors streaming). + expect(output.content).toEqual([ + { type: "text", text: "Step one.", textSignature: '{"v":1,"id":"msg_1"}' }, + { type: "text", text: "Step one. Step two.", textSignature: '{"v":1,"id":"msg_2"}' }, + ]); + }); + it("backfills Azure Responses completed function calls when item events are absent", async () => { const model = createAzureResponsesModel(); const output = createResponsesAssistantOutput(model); diff --git a/src/agents/openai-transport-stream.ts b/src/agents/openai-transport-stream.ts index 3205189e362b..f53cad359101 100644 --- a/src/agents/openai-transport-stream.ts +++ b/src/agents/openai-transport-stream.ts @@ -38,6 +38,7 @@ import { isOpenAICompatibleAzureResponsesBaseUrl } from "../shared/azure-openai- import { isResponsesTextContentPartType, isResponsesTextDeltaEventType, + resolveResponsesMessageSnapshotCollapse, } from "../shared/openai-responses-stream-compat.js"; import { createReasoningTagTextPartitioner } from "../shared/text/reasoning-tag-text-partitioner.js"; import { CHARS_PER_TOKEN_ESTIMATE, estimateStringChars } from "../utils/cjk-chars.js"; @@ -1474,25 +1475,68 @@ async function processResponsesStream( ) { let currentItem: Record | null = null; let currentBlock: Record | null = null; + let lastTextBlock: { + block: Record; + index: number; + phase: "commentary" | "final_answer" | undefined; + } | null = null; + // While a message item may still be a cumulative snapshot of lastTextBlock, + // its public block is deferred so a collapsed item never leaves an + // unbalanced text_start behind (#91959). null = no deferral in progress. + let pendingMessageText: string | null = null; const streamStartedAt = Date.now(); let eventCount = 0; const eventTypes = new Map(); const sseDebugMode = resolveModelSseDebugMode(); const blockIndex = () => output.content.length - 1; + const appendPendingMessageDelta = (delta: string) => { + pendingMessageText = `${pendingMessageText ?? ""}${delta}`; + const priorText = stringifyUnknown(lastTextBlock?.block.text); + if (priorText.startsWith(pendingMessageText) || pendingMessageText.startsWith(priorText)) { + return; + } + // Diverged from the prior text: this is a distinct message, so open its + // block now and replay the withheld text as one delta. + currentBlock = { type: "text", text: pendingMessageText }; + output.content.push(currentBlock); + stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + stream.push({ type: "text_delta", contentIndex: blockIndex(), delta: pendingMessageText }); + pendingMessageText = null; + }; const appendCompletedResponseTextItem = (item: Record) => { const text = readResponsesOutputMessageText(item); if (!text) { return; } + const phase = (item.phase as "commentary" | "final_answer" | undefined) ?? undefined; + const collapse = resolveResponsesMessageSnapshotCollapse({ + prior: lastTextBlock && { + text: stringifyUnknown(lastTextBlock.block.text), + phase: lastTextBlock.phase, + }, + nextText: text, + nextPhase: phase, + }); + if (collapse.kind === "extend" && lastTextBlock) { + // Cumulative snapshot of the prior message item: replace, don't append; + // the newest item's signature carries the content for replay (#91959). + lastTextBlock.block.text = collapse.text; + lastTextBlock.block.textSignature = encodeTextSignatureV1(stringifyUnknown(item.id), phase); + stream.push({ + type: "text_end", + contentIndex: lastTextBlock.index, + content: collapse.text, + partial: output, + }); + return; + } const block: Record = { type: "text", text, - textSignature: encodeTextSignatureV1( - stringifyUnknown(item.id), - (item.phase as "commentary" | "final_answer" | undefined) ?? undefined, - ), + textSignature: encodeTextSignatureV1(stringifyUnknown(item.id), phase), }; output.content.push(block); + lastTextBlock = { block, index: blockIndex(), phase }; stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); stream.push({ type: "text_end", @@ -1534,7 +1578,12 @@ async function processResponsesStream( } if (rawItem.type === "message") { appendCompletedResponseTextItem(rawItem); - } else if (rawItem.type === "function_call") { + continue; + } + // Any non-message item (reasoning, tool call) is a real boundary; a later + // message must not collapse across it, mirroring the streaming path. + lastTextBlock = null; + if (rawItem.type === "function_call") { appendCompletedResponseToolCallItem(rawItem); } } @@ -1569,6 +1618,12 @@ async function processResponsesStream( output.responseId = stringifyUnknown((event.response as { id?: string } | undefined)?.id); } else if (type === "response.output_item.added") { const item = event.item as Record; + if (item.type !== "message") { + // Snapshot collapse only applies to back-to-back message items; any + // other item is a real boundary (see resolveResponsesMessageSnapshotCollapse). + lastTextBlock = null; + pendingMessageText = null; + } if (item.type === "reasoning") { currentItem = item; currentBlock = { type: "thinking", thinking: "" }; @@ -1576,9 +1631,14 @@ async function processResponsesStream( stream.push({ type: "thinking_start", contentIndex: blockIndex(), partial: output }); } else if (item.type === "message") { currentItem = item; - currentBlock = { type: "text", text: "" }; - output.content.push(currentBlock); - stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + if (lastTextBlock) { + currentBlock = null; + pendingMessageText = ""; + } else { + currentBlock = { type: "text", text: "" }; + output.content.push(currentBlock); + stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + } } else if (item.type === "function_call") { currentItem = item; currentBlock = { @@ -1602,13 +1662,17 @@ async function processResponsesStream( }); } } else if (isResponsesTextDeltaEventType(type) || type === "response.refusal.delta") { - if (currentItem?.type === "message" && currentBlock?.type === "text") { - currentBlock.text = `${stringifyUnknown(currentBlock.text)}${stringifyUnknown(event.delta)}`; - stream.push({ - type: "text_delta", - contentIndex: blockIndex(), - delta: stringifyUnknown(event.delta), - }); + if (currentItem?.type === "message") { + if (pendingMessageText !== null) { + appendPendingMessageDelta(stringifyUnknown(event.delta)); + } else if (currentBlock?.type === "text") { + currentBlock.text = `${stringifyUnknown(currentBlock.text)}${stringifyUnknown(event.delta)}`; + stream.push({ + type: "text_delta", + contentIndex: blockIndex(), + delta: stringifyUnknown(event.delta), + }); + } } } else if (type === "response.function_call_arguments.delta") { if (currentItem?.type === "function_call" && currentBlock?.type === "toolCall") { @@ -1623,6 +1687,10 @@ async function processResponsesStream( } } else if (type === "response.output_item.done") { const item = event.item as Record; + if (item.type !== "message") { + lastTextBlock = null; + pendingMessageText = null; + } if (item.type === "reasoning" && currentBlock?.type === "thinking") { const summary = Array.isArray(item.summary) ? item.summary @@ -1648,9 +1716,12 @@ async function processResponsesStream( partial: output, }); currentBlock = null; - } else if (item.type === "message" && currentBlock?.type === "text") { + } else if ( + item.type === "message" && + (currentBlock?.type === "text" || pendingMessageText !== null) + ) { const content = Array.isArray(item.content) ? item.content : []; - currentBlock.text = content + const finalText = content .map((part) => { const contentPart = part as { type?: string; text?: string; refusal?: string }; return isResponsesTextContentPartType(contentPart.type) @@ -1658,16 +1729,53 @@ async function processResponsesStream( : (contentPart.refusal ?? ""); }) .join(""); - currentBlock.textSignature = encodeTextSignatureV1( - stringifyUnknown(item.id), - (item.phase as "commentary" | "final_answer" | undefined) ?? undefined, - ); - stream.push({ - type: "text_end", - contentIndex: blockIndex(), - content: stringifyUnknown(currentBlock.text), - partial: output, - }); + const phase = (item.phase as "commentary" | "final_answer" | undefined) ?? undefined; + const collapse = + pendingMessageText !== null + ? resolveResponsesMessageSnapshotCollapse({ + prior: lastTextBlock && { + text: stringifyUnknown(lastTextBlock.block.text), + phase: lastTextBlock.phase, + }, + nextText: finalText, + nextPhase: phase, + }) + : ({ kind: "keep" } as const); + pendingMessageText = null; + if (collapse.kind === "extend" && lastTextBlock) { + // Cumulative snapshot of the prior message item: replace its text + // instead of appending another copy. The deferred block was never + // started publicly, and the newest item's signature is kept so + // replay carries the item that produced this content (#91959). + lastTextBlock.block.text = collapse.text; + lastTextBlock.block.textSignature = encodeTextSignatureV1( + stringifyUnknown(item.id), + phase, + ); + stream.push({ + type: "text_end", + contentIndex: lastTextBlock.index, + content: collapse.text, + partial: output, + }); + } else { + if (currentBlock?.type !== "text") { + // Deferred distinct message: open its block now, balanced with the + // text_end below. + currentBlock = { type: "text", text: "" }; + output.content.push(currentBlock); + stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + } + currentBlock.text = finalText; + currentBlock.textSignature = encodeTextSignatureV1(stringifyUnknown(item.id), phase); + lastTextBlock = { block: currentBlock, index: blockIndex(), phase }; + stream.push({ + type: "text_end", + contentIndex: blockIndex(), + content: stringifyUnknown(currentBlock.text), + partial: output, + }); + } currentBlock = null; } else if (item.type === "function_call") { const args = diff --git a/src/llm/providers/openai-responses-shared.test.ts b/src/llm/providers/openai-responses-shared.test.ts index 016bc387f7f1..94696883bb4d 100644 --- a/src/llm/providers/openai-responses-shared.test.ts +++ b/src/llm/providers/openai-responses-shared.test.ts @@ -587,6 +587,323 @@ describe("processResponsesStream", () => { "toolcall_end", ]); }); + + it("collapses cumulative message snapshot items into one text block (#91959)", async () => { + const output = createAssistantOutput(); + const stream = new AssistantMessageEventStream(); + const events: Array> = []; + const collect = (async () => { + for await (const event of stream) { + events.push(event as unknown as Record); + } + })(); + + const snapshot1 = "Self-attention computes"; + const snapshot2 = "Self-attention computes Q/K/V projections"; + const snapshot3 = "Self-attention computes Q/K/V projections for each token."; + const messageItem = (id: string, text: string) => ({ + type: "message", + id, + phase: "final_answer", + content: [{ type: "output_text", text }], + }); + + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + item: { type: "message", id: "msg_1", phase: "final_answer" }, + }, + { type: "response.content_part.added", part: { type: "output_text", text: "" } }, + { type: "response.output_text.delta", delta: snapshot1 }, + { type: "response.output_item.done", item: messageItem("msg_1", snapshot1) }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_2", phase: "final_answer" }, + }, + { type: "response.output_item.done", item: messageItem("msg_2", snapshot2) }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_3", phase: "final_answer" }, + }, + { type: "response.output_item.done", item: messageItem("msg_3", snapshot3) }, + { type: "response.completed", response: { id: "resp_1", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + stream.end(); + await collect; + + expect(output.content).toEqual([ + { + type: "text", + text: snapshot3, + textSignature: JSON.stringify({ v: 1, id: "msg_3", phase: "final_answer" }), + }, + ]); + // Balanced lifecycle: exactly one text_start, every event on index 0, and + // each collapsed snapshot re-ends the same block with its grown content. + expect(events.map((event) => [event.type, event.contentIndex])).toEqual([ + ["text_start", 0], + ["text_delta", 0], + ["text_end", 0], + ["text_end", 0], + ["text_end", 0], + ]); + expect( + events.filter((event) => event.type === "text_end").map((event) => event.content), + ).toEqual([snapshot1, snapshot2, snapshot3]); + }); + + it.each([ + ["identical", "Hello world.", "Hello world."], + ["shrinking", "Step one. Step two.", "Step one."], + ])("keeps %s adjacent same-phase message items as distinct blocks", async (_label, a, b) => { + const output = createAssistantOutput(); + const stream = new AssistantMessageEventStream(); + const events: Array> = []; + const collect = (async () => { + for await (const event of stream) { + events.push(event as unknown as Record); + } + })(); + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + item: { type: "message", id: "msg_1", phase: "final_answer" }, + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_1", + phase: "final_answer", + content: [{ type: "output_text", text: a }], + }, + }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_2", phase: "final_answer" }, + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_2", + phase: "final_answer", + content: [{ type: "output_text", text: b }], + }, + }, + { type: "response.completed", response: { id: "resp_1", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + stream.end(); + await collect; + + // Only strict extensions collapse; equal or shrinking items are real, + // independently identified messages and must never be removed. + expect(output.content).toEqual([ + { + type: "text", + text: a, + textSignature: JSON.stringify({ v: 1, id: "msg_1", phase: "final_answer" }), + }, + { + type: "text", + text: b, + textSignature: JSON.stringify({ v: 1, id: "msg_2", phase: "final_answer" }), + }, + ]); + // The deferred second item still opens and closes its own block. + expect(events.map((event) => [event.type, event.contentIndex])).toEqual([ + ["text_start", 0], + ["text_end", 0], + ["text_start", 1], + ["text_end", 1], + ]); + }); + + it("streams a deferred distinct message live once its text diverges from the prior block", async () => { + const output = createAssistantOutput(); + const stream = new AssistantMessageEventStream(); + const events: Array> = []; + const collect = (async () => { + for await (const event of stream) { + events.push(event as unknown as Record); + } + })(); + + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + item: { type: "message", id: "msg_1", phase: "final_answer" }, + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_1", + phase: "final_answer", + content: [{ type: "output_text", text: "Hello." }], + }, + }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_2", phase: "final_answer" }, + }, + { type: "response.content_part.added", part: { type: "output_text", text: "" } }, + { type: "response.output_text.delta", delta: "Good" }, + { type: "response.output_text.delta", delta: "bye" }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_2", + phase: "final_answer", + content: [{ type: "output_text", text: "Goodbye" }], + }, + }, + { type: "response.completed", response: { id: "resp_1", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + stream.end(); + await collect; + + expect(output.content).toEqual([ + { + type: "text", + text: "Hello.", + textSignature: JSON.stringify({ v: 1, id: "msg_1", phase: "final_answer" }), + }, + { + type: "text", + text: "Goodbye", + textSignature: JSON.stringify({ v: 1, id: "msg_2", phase: "final_answer" }), + }, + ]); + // The withheld prefix is replayed as one delta at divergence ("Good" + // diverges from "Hello."), then later deltas stream live. + expect(events.map((event) => [event.type, event.contentIndex, event.delta ?? null])).toEqual([ + ["text_start", 0, null], + ["text_end", 0, null], + ["text_start", 1, null], + ["text_delta", 1, "Good"], + ["text_delta", 1, "bye"], + ["text_end", 1, null], + ]); + }); + + it("keeps prefix-nested message items separated by a reasoning item as separate blocks", async () => { + const output = createAssistantOutput(); + const stream = new AssistantMessageEventStream(); + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + item: { type: "message", id: "msg_1", phase: "final_answer" }, + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_1", + phase: "final_answer", + content: [{ type: "output_text", text: "Step one." }], + }, + }, + { type: "response.output_item.added", item: { type: "reasoning" } }, + { + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_1", summary: [] }, + }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_2", phase: "final_answer" }, + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_2", + phase: "final_answer", + content: [{ type: "output_text", text: "Step one. Step two." }], + }, + }, + { type: "response.completed", response: { id: "resp_1", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + stream.end(); + + // Collapsing across the reasoning block would orphan it for replay. + expect(output.content.map((block) => block.type)).toEqual(["text", "thinking", "text"]); + expect(output.content[2]).toMatchObject({ type: "text", text: "Step one. Step two." }); + }); + + it("keeps prefix-nested message items with different phases as separate blocks", async () => { + const output = createAssistantOutput(); + const stream = new AssistantMessageEventStream(); + await processResponsesStream( + responseEvents([ + { + type: "response.output_item.added", + item: { type: "message", id: "msg_1", phase: "commentary" }, + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_1", + phase: "commentary", + content: [{ type: "output_text", text: "Done" }], + }, + }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_2", phase: "final_answer" }, + }, + { + type: "response.output_item.done", + item: { + type: "message", + id: "msg_2", + phase: "final_answer", + content: [{ type: "output_text", text: "Done." }], + }, + }, + { type: "response.completed", response: { id: "resp_1", status: "completed" } }, + ]), + output, + stream, + nativeOpenAIModel, + ); + stream.end(); + + expect(output.content).toEqual([ + { + type: "text", + text: "Done", + textSignature: JSON.stringify({ v: 1, id: "msg_1", phase: "commentary" }), + }, + { + type: "text", + text: "Done.", + textSignature: JSON.stringify({ v: 1, id: "msg_2", phase: "final_answer" }), + }, + ]); + }); }); describe("Azure OpenAI Responses content type support", () => { diff --git a/src/llm/providers/openai-responses-shared.ts b/src/llm/providers/openai-responses-shared.ts index 82dd8ccf34a5..799783afd3e6 100644 --- a/src/llm/providers/openai-responses-shared.ts +++ b/src/llm/providers/openai-responses-shared.ts @@ -20,6 +20,7 @@ import { type AzureResponsesTextDeltaEvent, isAzureResponsesTextDeltaEvent, isResponsesTextContentPartType, + resolveResponsesMessageSnapshotCollapse, } from "../../shared/openai-responses-stream-compat.js"; import { calculateCost, clampThinkingLevel } from "../model-utils.js"; import type { @@ -579,14 +580,48 @@ export async function processResponsesStream( | null = null; let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null = null; + let lastTextBlock: { + block: TextContent; + index: number; + phase: TextSignatureV1["phase"] | undefined; + } | null = null; + // While a message item may still be a cumulative snapshot of lastTextBlock, + // its public block is deferred so a collapsed item never leaves an + // unbalanced text_start behind (#91959). null = no deferral in progress. + let pendingMessageText: string | null = null; const blocks = output.content; const blockIndex = () => blocks.length - 1; + const appendPendingMessageDelta = (delta: string) => { + pendingMessageText = `${pendingMessageText ?? ""}${delta}`; + const priorText = lastTextBlock?.block.text ?? ""; + if (priorText.startsWith(pendingMessageText) || pendingMessageText.startsWith(priorText)) { + return; + } + // Diverged from the prior text: this is a distinct message, so open its + // block now and replay the withheld text as one delta. + currentBlock = { type: "text", text: pendingMessageText }; + blocks.push(currentBlock); + stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + stream.push({ + type: "text_delta", + contentIndex: blockIndex(), + delta: pendingMessageText, + partial: output, + }); + pendingMessageText = null; + }; for await (const event of openaiStream) { if (event.type === "response.created") { output.responseId = event.response.id; } else if (event.type === "response.output_item.added") { const item = event.item; + if (item.type !== "message") { + // Snapshot collapse only applies to back-to-back message items; any + // other item is a real boundary (see resolveResponsesMessageSnapshotCollapse). + lastTextBlock = null; + pendingMessageText = null; + } if (item.type === "reasoning") { currentItem = item; currentBlock = { type: "thinking", thinking: "" }; @@ -594,9 +629,14 @@ export async function processResponsesStream( stream.push({ type: "thinking_start", contentIndex: blockIndex(), partial: output }); } else if (item.type === "message") { currentItem = item; - currentBlock = { type: "text", text: "" }; - output.content.push(currentBlock); - stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + if (lastTextBlock) { + currentBlock = null; + pendingMessageText = ""; + } else { + currentBlock = { type: "text", text: "" }; + output.content.push(currentBlock); + stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + } } else if (item.type === "function_call") { currentItem = item; currentBlock = { @@ -666,48 +706,39 @@ export async function processResponsesStream( } } } else if (event.type === "response.output_text.delta") { - if (currentItem?.type === "message" && currentBlock?.type === "text") { + if (currentItem?.type === "message") { if (!currentItem.content || currentItem.content.length === 0) { continue; } const lastPart = currentItem.content[currentItem.content.length - 1]; if (isResponsesTextContentPartType(lastPart?.type)) { - currentBlock.text += event.delta; lastPart.text += event.delta; - stream.push({ - type: "text_delta", - contentIndex: blockIndex(), - delta: event.delta, - partial: output, - }); + if (pendingMessageText !== null) { + appendPendingMessageDelta(event.delta); + } else if (currentBlock?.type === "text") { + currentBlock.text += event.delta; + stream.push({ + type: "text_delta", + contentIndex: blockIndex(), + delta: event.delta, + partial: output, + }); + } } } } else if (isAzureResponsesTextDeltaEvent(event)) { - if (currentItem?.type === "message" && currentBlock?.type === "text") { + if (currentItem?.type === "message") { currentItem.content = currentItem.content || []; let lastPart = currentItem.content[currentItem.content.length - 1]; if (lastPart?.type !== "text") { lastPart = { type: "text", text: "" }; currentItem.content.push(lastPart); } - currentBlock.text += event.delta; lastPart.text += event.delta; - stream.push({ - type: "text_delta", - contentIndex: blockIndex(), - delta: event.delta, - partial: output, - }); - } - } else if (event.type === "response.refusal.delta") { - if (currentItem?.type === "message" && currentBlock?.type === "text") { - if (!currentItem.content || currentItem.content.length === 0) { - continue; - } - const lastPart = currentItem.content[currentItem.content.length - 1]; - if (lastPart?.type === "refusal") { + if (pendingMessageText !== null) { + appendPendingMessageDelta(event.delta); + } else if (currentBlock?.type === "text") { currentBlock.text += event.delta; - lastPart.refusal += event.delta; stream.push({ type: "text_delta", contentIndex: blockIndex(), @@ -716,6 +747,27 @@ export async function processResponsesStream( }); } } + } else if (event.type === "response.refusal.delta") { + if (currentItem?.type === "message") { + if (!currentItem.content || currentItem.content.length === 0) { + continue; + } + const lastPart = currentItem.content[currentItem.content.length - 1]; + if (lastPart?.type === "refusal") { + lastPart.refusal += event.delta; + if (pendingMessageText !== null) { + appendPendingMessageDelta(event.delta); + } else if (currentBlock?.type === "text") { + currentBlock.text += event.delta; + stream.push({ + type: "text_delta", + contentIndex: blockIndex(), + delta: event.delta, + partial: output, + }); + } + } + } } else if (event.type === "response.function_call_arguments.delta") { if (currentItem?.type === "function_call" && currentBlock?.type === "toolCall") { currentBlock.partialJson += event.delta; @@ -754,6 +806,10 @@ export async function processResponsesStream( } } else if (event.type === "response.output_item.done") { const item = event.item; + if (item.type !== "message") { + lastTextBlock = null; + pendingMessageText = null; + } if (item.type === "reasoning" && currentBlock?.type === "thinking") { const summaryText = item.summary?.map((s) => s.text).join("\n\n") || ""; @@ -767,18 +823,58 @@ export async function processResponsesStream( partial: output, }); currentBlock = null; - } else if (item.type === "message" && currentBlock?.type === "text") { + } else if ( + item.type === "message" && + (currentBlock?.type === "text" || pendingMessageText !== null) + ) { // Support both OpenAI "output_text" and Azure "text" content types - currentBlock.text = item.content + const finalText = item.content .map((c) => (c.type === "output_text" || c.type === "text" ? c.text : c.refusal)) .join(""); - currentBlock.textSignature = encodeTextSignatureV1(item.id, item.phase ?? undefined); - stream.push({ - type: "text_end", - contentIndex: blockIndex(), - content: currentBlock.text, - partial: output, - }); + const phase = item.phase ?? undefined; + const collapse = + pendingMessageText !== null + ? resolveResponsesMessageSnapshotCollapse({ + prior: lastTextBlock && { + text: lastTextBlock.block.text, + phase: lastTextBlock.phase, + }, + nextText: finalText, + nextPhase: phase, + }) + : ({ kind: "keep" } as const); + pendingMessageText = null; + if (collapse.kind === "extend" && lastTextBlock) { + // Cumulative snapshot of the prior message item: replace its text + // instead of appending another copy. The deferred block was never + // started publicly, and the newest item's signature is kept so + // replay carries the item that produced this content (#91959). + lastTextBlock.block.text = collapse.text; + lastTextBlock.block.textSignature = encodeTextSignatureV1(item.id, phase); + stream.push({ + type: "text_end", + contentIndex: lastTextBlock.index, + content: collapse.text, + partial: output, + }); + } else { + if (currentBlock?.type !== "text") { + // Deferred distinct message: open its block now, balanced with the + // text_end below. + currentBlock = { type: "text", text: "" }; + blocks.push(currentBlock); + stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output }); + } + currentBlock.text = finalText; + currentBlock.textSignature = encodeTextSignatureV1(item.id, phase); + lastTextBlock = { block: currentBlock, index: blockIndex(), phase }; + stream.push({ + type: "text_end", + contentIndex: blockIndex(), + content: currentBlock.text, + partial: output, + }); + } currentBlock = null; } else if (item.type === "function_call") { const args = diff --git a/src/shared/openai-responses-stream-compat.ts b/src/shared/openai-responses-stream-compat.ts index 749860c47264..e8ca1377d4b3 100644 --- a/src/shared/openai-responses-stream-compat.ts +++ b/src/shared/openai-responses-stream-compat.ts @@ -49,3 +49,30 @@ export function isAzureResponsesTextDeltaEvent(event: { }): event is AzureResponsesTextDeltaEvent { return isAzureResponsesTextDeltaEventType(event.type) && typeof event.delta === "string"; } + +export type ResponsesMessageSnapshotCollapse = { kind: "extend"; text: string } | { kind: "keep" }; + +// Some openai-responses providers re-emit the assistant message as cumulative +// snapshot items — each a strict prefix-superset of the previous one — instead +// of one final message item. A same-phase strict extension replaces the prior +// text block, or the visible reply repeats once per snapshot (#91959). +// Extension-only on purpose: equal or shrinking adjacent items stay distinct +// (the Responses protocol allows multiple message items per response), so a +// false positive can only merge rendering — it can never lose text. +// `prior` must be the immediately preceding output item: collapsing across +// reasoning/function_call boundaries would drop real post-tool messages and +// orphan reasoning items, which OpenAI replay rejects. +export function resolveResponsesMessageSnapshotCollapse(params: { + prior: { text: string; phase: string | undefined } | null; + nextText: string; + nextPhase: string | undefined; +}): ResponsesMessageSnapshotCollapse { + const { prior, nextText } = params; + if (!prior?.text || !nextText || prior.phase !== params.nextPhase) { + return { kind: "keep" }; + } + if (nextText.length > prior.text.length && nextText.startsWith(prior.text)) { + return { kind: "extend", text: nextText }; + } + return { kind: "keep" }; +}