fix(sse): stop reasoning-summary drop + duplicated deltas on claude→codex streaming (#5786) (#5832)

Integrated into release/v3.8.43

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

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<string>();
+  // #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) {

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, unknown>): 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<Reconstructed> {
+  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)}`);
+});