From 4d7f36015ab2fd1f1014e159723896d31713dc7c Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 1 Jul 2026 21:52:10 -0300 Subject: [PATCH] =?UTF-8?q?fix(sse):=20stop=20reasoning-summary=20drop=20+?= =?UTF-8?q?=20duplicated=20deltas=20on=20claude=E2=86=92codex=20streaming?= =?UTF-8?q?=20(#5786)=20(#5832)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.43 --- .../translator/response/openai-responses.ts | 89 ++++- open-sse/utils/stream.ts | 35 ++ .../streaming-reasoning-dedup-5786.test.ts | 340 ++++++++++++++++++ 3 files changed, 448 insertions(+), 16 deletions(-) create mode 100644 tests/unit/streaming-reasoning-dedup-5786.test.ts diff --git a/open-sse/translator/response/openai-responses.ts b/open-sse/translator/response/openai-responses.ts index d6c397fbad..3a9a1299da 100644 --- a/open-sse/translator/response/openai-responses.ts +++ b/open-sse/translator/response/openai-responses.ts @@ -644,6 +644,51 @@ function computeFinishReason(state): "tool_calls" | "stop" { return (state.toolCallIndex || 0) > 0 || state.currentToolCallId ? "tool_calls" : "stop"; } +// #5786 — remember that a reasoning delta was streamed for a given reasoning item, so +// the terminal `response.output_item.done` snapshot for that item is NOT re-emitted +// (which would duplicate the reasoning channel). Keyed by item_id when present, with a +// global fallback for streams whose deltas carry no item_id. +function markResponsesReasoningDeltaEmitted(state, itemId) { + state.reasoningDeltaEmitted = true; + const id = itemId != null ? String(itemId) : ""; + if (!id) return; + if (!(state.reasoningItemsWithDelta instanceof Set)) { + state.reasoningItemsWithDelta = new Set(); + } + state.reasoningItemsWithDelta.add(id); +} + +// #5786 — build a Chat-format reasoning delta chunk in the shape the client renders in +// its thinking panel (`reasoning_content`, or `reasoning_text` for Copilot-compatible +// clients). Mirrors the `response.reasoning_summary_text.delta` branch. +function buildResponsesReasoningDeltaChunk(state, text) { + const delta = state.copilotCompatibleReasoning + ? { reasoning_text: text } + : { reasoning_content: text }; + return { + id: state.chatId, + object: "chat.completion.chunk", + created: state.created, + model: state.model || "gpt-4", + choices: [ + { + index: 0, + delta, + finish_reason: null, + }, + ], + }; +} + +function extractResponsesReasoningSummaryText(item) { + if (!item || !Array.isArray(item.summary)) return ""; + return item.summary + .map((part) => + part && typeof part === "object" && typeof part.text === "string" ? part.text : "" + ) + .join(""); +} + /** * Translate OpenAI Responses API chunk to OpenAI Chat Completions format * This is for when Codex returns data and we need to send it to an OpenAI-compatible client @@ -983,6 +1028,7 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { ) { const reasoningDelta = data.delta || ""; if (!reasoningDelta) return null; + markResponsesReasoningDeltaEmitted(state, data.item_id); return { id: state.chatId, object: "chat.completion.chunk", @@ -1007,22 +1053,33 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { if (eventType === "response.reasoning_summary_text.delta") { const reasoningDelta = data.delta || ""; if (!reasoningDelta) return null; - const reasoningDeltaShape = state.copilotCompatibleReasoning - ? { reasoning_text: reasoningDelta } - : { reasoning_content: reasoningDelta }; - return { - id: state.chatId, - object: "chat.completion.chunk", - created: state.created, - model: state.model || "gpt-4", - choices: [ - { - index: 0, - delta: reasoningDeltaShape, - finish_reason: null, - }, - ], - }; + markResponsesReasoningDeltaEmitted(state, data.item_id); + return buildResponsesReasoningDeltaChunk(state, reasoningDelta); + } + + // #5786 — reasoning summary exposed ONLY as a terminal snapshot on + // `response.output_item.done` (no preceding reasoning_summary_text.delta events — e.g. + // Codex reasoning models that surface the summary once at item close). Without this the + // reasoning channel is silently dropped and never reaches the client's thinking panel. + // Only synthesize when NO reasoning delta was already streamed for this item, so normal + // delta streams are never duplicated. + if (eventType === "response.output_item.done" && data.item?.type === "reasoning") { + const item = data.item; + const itemId = item.id != null ? String(item.id) : ""; + const emittedForItem = + state.reasoningItemsWithDelta instanceof Set && + itemId && + state.reasoningItemsWithDelta.has(itemId); + // Deltas were streamed but carried no item_id: fall back to the global flag and + // suppress synthesis to avoid duplicating that same reasoning text. + const emittedWithoutItemId = + state.reasoningDeltaEmitted && + !(state.reasoningItemsWithDelta instanceof Set && state.reasoningItemsWithDelta.size > 0); + if (emittedForItem || emittedWithoutItemId) return null; + + const summaryText = extractResponsesReasoningSummaryText(item); + if (!summaryText) return null; + return buildResponsesReasoningDeltaChunk(state, summaryText); } // Ignore other events diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index b6feb86e0a..8357ce2506 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -684,6 +684,18 @@ export function createSSEStream(options: StreamOptions = {}) { let passthroughResponsesId: string | null = null; let passthroughResponsesCurrentFunctionCallKey: string | null = null; const passthroughResponsesReasoningSummarySeen = new Set(); + // #5786 — highest Responses-API `sequence_number` already forwarded on this stream. + // The Responses API guarantees a strictly increasing sequence_number, so any event at + // or below this watermark is an upstream reconnect/retry replay and must be dropped — + // otherwise the replayed deltas glue duplicated text into the client stream. Applies to + // both translate mode (openai-responses → claude/openai) and Responses passthrough. + let lastSeenResponsesSequenceNumber = -1; + const isDuplicateResponsesSequence = (value: unknown): boolean => { + if (typeof value !== "number" || !Number.isFinite(value)) return false; + if (value <= lastSeenResponsesSequenceNumber) return true; + lastSeenResponsesSequenceNumber = value; + return false; + }; const streamStartedAt = Date.now(); let lastToolCallChunkTime: number | null = null; @@ -1185,6 +1197,18 @@ export function createSSEStream(options: StreamOptions = {}) { }) : null; + // #5786 — drop replayed Responses-API events (a re-sent event carrying an + // already-seen sequence_number) so their deltas are not forwarded twice. + if ( + parsedPassthroughData && + typeof parsedPassthroughData.type === "string" && + parsedPassthroughData.type.startsWith("response.") && + isDuplicateResponsesSequence(parsedPassthroughData.sequence_number) + ) { + clearPendingPassthroughEvent(); + continue; + } + if (trimmed.startsWith("data:")) { const providerPayload = parsedPassthroughData ?? parseSSELine(trimmed); if (providerPayload) { @@ -1859,6 +1883,17 @@ export function createSSEStream(options: StreamOptions = {}) { const parsed = parseSSELine(trimmed); if (!parsed) continue; + + // #5786 — drop replayed Responses-API events (identical/lower sequence_number + // re-sent on an upstream reconnect) so their deltas are not glued twice into + // the translated client stream. + if ( + targetFormat === FORMATS.OPENAI_RESPONSES && + isDuplicateResponsesSequence((parsed as JsonRecord).sequence_number) + ) { + continue; + } + providerPayloadCollector.push(parsed); if (parsed && parsed.done) { diff --git a/tests/unit/streaming-reasoning-dedup-5786.test.ts b/tests/unit/streaming-reasoning-dedup-5786.test.ts new file mode 100644 index 0000000000..24d0f61125 --- /dev/null +++ b/tests/unit/streaming-reasoning-dedup-5786.test.ts @@ -0,0 +1,340 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { createSSETransformStreamWithLogger } from "@omniroute/open-sse/utils/stream.ts"; +import { FORMATS } from "@omniroute/open-sse/translator/formats.ts"; + +/** + * Regression guards for #5786 — streaming claude→codex. + * + * A Claude/Anthropic-format client (source_format=claude, e.g. Claude Code) routed + * to a Codex provider (ChatGPT OAuth, Responses-API reasoning model) hit two bugs on + * the STREAMING path: + * + * (A) Reassembled deltas were emitted TWICE, glued with no separator, whenever the + * upstream Responses-API stream replayed an already-seen event (identical + * `sequence_number`) — the gateway never de-duplicated on sequence_number. + * (B) A reasoning summary that arrived ONLY as a terminal snapshot on + * `response.output_item.done` (item.type === "reasoning", no preceding + * `reasoning_summary_text.delta` events) was silently DROPPED in translate mode, + * so the reasoning channel was never surfaced (`tokens_reasoning` null) — and in + * the reported production traces leaked into the visible assistant `content`. + * + * These tests drive the REAL streaming pipeline used for claude←codex: + * createSSETransformStreamWithLogger(OPENAI_RESPONSES → CLAUDE). + * The provider (Codex) speaks openai-responses; the client (Claude Code) speaks claude. + */ + +function sse(type: string, obj: Record): string { + return `event: ${type}\ndata: ${JSON.stringify({ type, ...obj })}\n\n`; +} + +type Reconstructed = { + raw: string; + content: string; + thinking: string; +}; + +/** + * Feed a raw Codex (Responses-API) SSE wire string through the real translate-mode + * transform stream (openai-responses → claude), optionally byte-chunked to stress the + * line-reassembly buffer, and reconstruct the client-visible Claude text/thinking. + */ +async function runClaudeFromCodex(rawSSE: string, chunkSize = 1_000_000): Promise { + const ts = createSSETransformStreamWithLogger( + FORMATS.OPENAI_RESPONSES, // targetFormat — provider (Codex) + FORMATS.CLAUDE, // sourceFormat — client (Claude Code) + "codex", + null, + null, + "gpt-5.5-high", + "conn-5786", + { model: "gpt-5.5-high" }, + null, + null, + null + ); + + const writer = ts.writable.getWriter(); + const reader = ts.readable.getReader(); + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + + const readAll = (async () => { + const out: string[] = []; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + out.push(decoder.decode(value)); + } + return out.join(""); + })(); + + for (let i = 0; i < rawSSE.length; i += chunkSize) { + await writer.write(encoder.encode(rawSSE.slice(i, i + chunkSize))); + } + await writer.close(); + + const raw = await readAll; + + let content = ""; + let thinking = ""; + for (const line of raw.split("\n")) { + if (!line.startsWith("data:")) continue; + const payload = line.slice(5).trim(); + if (!payload || payload === "[DONE]") continue; + try { + const evt = JSON.parse(payload); + if (evt.type === "content_block_delta") { + if (evt.delta?.type === "text_delta") content += evt.delta.text; + if (evt.delta?.type === "thinking_delta") thinking += evt.delta.thinking; + } + } catch { + // ignore metadata comment lines / non-JSON + } + } + return { raw, content, thinking }; +} + +test("#5786 (A) replayed Responses-API deltas (sequence_number <= last-seen) are dropped — no glued duplicate text", async () => { + // Upstream reconnect/retry replays the same output_text.delta event with the SAME + // sequence_number, producing the reported "...briefly.and I'll summarize briefly." glue. + const rawEvents = [ + sse("response.created", { sequence_number: 0, response: { id: "resp_1", status: "in_progress" } }), + sse("response.output_item.added", { + sequence_number: 1, + output_index: 0, + item: { id: "msg_1", type: "message", content: [], role: "assistant" }, + }), + sse("response.output_text.delta", { + sequence_number: 2, + item_id: "msg_1", + output_index: 0, + content_index: 0, + delta: "Sure, and I'll summarize briefly.", + }), + // Replay of the same event (identical sequence_number) — must be dropped. + sse("response.output_text.delta", { + sequence_number: 2, + item_id: "msg_1", + output_index: 0, + content_index: 0, + delta: "and I'll summarize briefly.", + }), + sse("response.output_item.done", { + sequence_number: 3, + output_index: 0, + item: { + id: "msg_1", + type: "message", + content: [{ type: "output_text", text: "Sure, and I'll summarize briefly." }], + role: "assistant", + }, + }), + sse("response.completed", { + sequence_number: 4, + response: { id: "resp_1", status: "completed", output: [], usage: { input_tokens: 10, output_tokens: 20 } }, + }), + ]; + + const { content } = await runClaudeFromCodex(rawEvents.join("")); + + assert.equal(content, "Sure, and I'll summarize briefly."); + assert.ok( + !content.includes("briefly.and I'll summarize briefly."), + `duplicated/glued text leaked into client content: ${JSON.stringify(content)}` + ); +}); + +test("#5786 (A-guard) a well-behaved stream with strictly increasing sequence_number is passed through UNCHANGED", async () => { + const rawEvents = [ + sse("response.created", { sequence_number: 0, response: { id: "resp_2", status: "in_progress" } }), + sse("response.output_item.added", { + sequence_number: 1, + output_index: 0, + item: { id: "msg_2", type: "message", content: [], role: "assistant" }, + }), + sse("response.output_text.delta", { + sequence_number: 2, + item_id: "msg_2", + output_index: 0, + content_index: 0, + delta: "Hello, ", + }), + sse("response.output_text.delta", { + sequence_number: 3, + item_id: "msg_2", + output_index: 0, + content_index: 0, + delta: "world.", + }), + sse("response.output_text.delta", { + sequence_number: 4, + item_id: "msg_2", + output_index: 0, + content_index: 0, + delta: " Bye.", + }), + sse("response.output_item.done", { + sequence_number: 5, + output_index: 0, + item: { + id: "msg_2", + type: "message", + content: [{ type: "output_text", text: "Hello, world. Bye." }], + role: "assistant", + }, + }), + sse("response.completed", { + sequence_number: 6, + response: { id: "resp_2", status: "completed", output: [], usage: { input_tokens: 5, output_tokens: 8 } }, + }), + ]; + + // Byte-chunked to also prove line reassembly does not corrupt the dedup gate. + const { content } = await runClaudeFromCodex(rawEvents.join(""), 1); + + assert.equal(content, "Hello, world. Bye."); +}); + +test("#5786 (B) reasoning summary that arrives ONLY on output_item.done is surfaced as a Claude thinking block (not dropped, not in content)", async () => { + const reasoningText = "I'm thinking I need to read the file again."; + const rawEvents = [ + sse("response.created", { sequence_number: 0, response: { id: "resp_3", status: "in_progress" } }), + // Reasoning item announced... + sse("response.output_item.added", { + sequence_number: 1, + output_index: 0, + item: { id: "rs_3", type: "reasoning", summary: [] }, + }), + // ...but the summary text is exposed ONLY on the terminal snapshot — no delta events. + sse("response.output_item.done", { + sequence_number: 2, + output_index: 0, + item: { + id: "rs_3", + type: "reasoning", + summary: [{ type: "summary_text", text: reasoningText }], + encrypted_content: "opaque-blob", + }, + }), + sse("response.output_item.added", { + sequence_number: 3, + output_index: 1, + item: { id: "msg_3", type: "message", content: [], role: "assistant" }, + }), + sse("response.output_text.delta", { + sequence_number: 4, + item_id: "msg_3", + output_index: 1, + content_index: 0, + delta: "Sure thing.", + }), + sse("response.output_item.done", { + sequence_number: 5, + output_index: 1, + item: { + id: "msg_3", + type: "message", + content: [{ type: "output_text", text: "Sure thing." }], + role: "assistant", + }, + }), + sse("response.completed", { + sequence_number: 6, + response: { id: "resp_3", status: "completed", output: [], usage: { input_tokens: 10, output_tokens: 20 } }, + }), + ]; + + const { content, thinking } = await runClaudeFromCodex(rawEvents.join("")); + + // Reasoning must surface in the thinking channel, exactly once... + assert.equal(thinking, reasoningText); + // ...and must NOT leak into the visible assistant content. + assert.equal(content, "Sure thing."); + assert.ok(!content.includes(reasoningText), `reasoning leaked into content: ${JSON.stringify(content)}`); +}); + +test("#5786 (B) a normal reasoning stream (with reasoning_summary_text.delta) still yields a thinking block and does NOT duplicate", async () => { + const rawEvents = [ + sse("response.created", { sequence_number: 0, response: { id: "resp_4", status: "in_progress" } }), + sse("response.output_item.added", { + sequence_number: 1, + output_index: 0, + item: { id: "rs_4", type: "reasoning", summary: [] }, + }), + sse("response.reasoning_summary_part.added", { + sequence_number: 2, + item_id: "rs_4", + output_index: 0, + summary_index: 0, + part: { type: "summary_text", text: "" }, + }), + sse("response.reasoning_summary_text.delta", { + sequence_number: 3, + item_id: "rs_4", + output_index: 0, + summary_index: 0, + delta: "Considering the request. ", + }), + sse("response.reasoning_summary_text.delta", { + sequence_number: 4, + item_id: "rs_4", + output_index: 0, + summary_index: 0, + delta: "Planning the answer.", + }), + sse("response.reasoning_summary_text.done", { + sequence_number: 5, + item_id: "rs_4", + output_index: 0, + summary_index: 0, + text: "Considering the request. Planning the answer.", + }), + // Terminal reasoning snapshot carries the FULL summary — must NOT re-emit it. + sse("response.output_item.done", { + sequence_number: 6, + output_index: 0, + item: { + id: "rs_4", + type: "reasoning", + summary: [{ type: "summary_text", text: "Considering the request. Planning the answer." }], + }, + }), + sse("response.output_item.added", { + sequence_number: 7, + output_index: 1, + item: { id: "msg_4", type: "message", content: [], role: "assistant" }, + }), + sse("response.output_text.delta", { + sequence_number: 8, + item_id: "msg_4", + output_index: 1, + content_index: 0, + delta: "Here is the answer.", + }), + sse("response.output_item.done", { + sequence_number: 9, + output_index: 1, + item: { + id: "msg_4", + type: "message", + content: [{ type: "output_text", text: "Here is the answer." }], + role: "assistant", + }, + }), + sse("response.completed", { + sequence_number: 10, + response: { id: "resp_4", status: "completed", output: [], usage: { input_tokens: 12, output_tokens: 15 } }, + }), + ]; + + const { content, thinking } = await runClaudeFromCodex(rawEvents.join(""), 7); + + assert.equal(thinking, "Considering the request. Planning the answer."); + assert.equal(content, "Here is the answer."); + // Duplication guard: the summary text must appear exactly once in the thinking channel. + const firstIdx = thinking.indexOf("Considering the request."); + const lastIdx = thinking.lastIndexOf("Considering the request."); + assert.equal(firstIdx, lastIdx, `reasoning summary duplicated in thinking: ${JSON.stringify(thinking)}`); +});