fix(api): Responses passthrough emits event-only SSE frames after filtering commentary output (#6561) (#6735)

* fix(api): Responses passthrough emits event-only SSE frames after filtering commentary output (#6561)

The #6199 commentary-drop `continue;` branches in stream.ts skipped the
data: line for a dropped commentary event but never cleared the
already-buffered event: line for the same frame, so the next blank line
flushed the stale event: line alone -- an event-only SSE frame that
crashes the OpenAI Python SDK's json.loads(). Both drop sites now call
clearPendingPassthroughEvent() before continue. The commentary-drop
decision was extracted into a new responsesCommentaryDrop.ts module so
the fix does not grow the frozen stream.ts.

* chore(6735): re-sync onto release tip; CHANGELOG entry → changelog.d fragment (fragments-first)

changelog.d/fixes/6735-responses-commentary-event-frame.md

@@ -0,0 +1 @@
+- **fix(api):** Responses passthrough emitted event-only SSE frames (no `data:` line) for every dropped commentary event, breaking the OpenAI Python SDK's `sse.json()` parser ([#6561](https://github.com/diegosouzapw/OmniRoute/issues/6561)), follow-up to #6199/#6232 — the commentary-drop `continue;` branches in `open-sse/utils/stream.ts` skipped the `data:` line for a dropped commentary event but never cleared the already-buffered `event:` line for that same frame, so the next blank line flushed the stale `event:` line alone. Both drop sites now call `clearPendingPassthroughEvent()` before `continue`, discarding the buffered prefix along with the dropped payload; the commentary-drop decision itself was extracted into a new `open-sse/utils/responsesCommentaryDrop.ts` so the fix does not grow the frozen `stream.ts`. Regression guard: `tests/unit/responses-commentary-event-frame-6561.test.ts` (realistic `event:\ndata:\n\n` frames — the existing #6199 test only used bare `data:` lines and never exercised this path).

open-sse/utils/responsesCommentaryDrop.ts

@@ -0,0 +1,98 @@
+// open-sse/utils/responsesCommentaryDrop.ts
+//
+// #6199 / #6561 — statefully decide whether a Responses SSE event belongs to
+// an internal "commentary" phase item and must be dropped from the
+// passthrough stream. The `response.output_item.added` event announces the
+// phase; the follow-up delta/done events only carry `item_id`/`output_index`,
+// so we key off those (tracked across calls in the two Sets the caller owns).
+//
+// Extracted out of `stream.ts` (a frozen file — see
+// config/quality/file-size-baseline.json) so the #6561 fix (clearing the
+// buffered `event:` line alongside every drop) does not grow that file.
+import { isResponsesCommentaryMessageItem } from "../handlers/responseSanitizer.ts";
+
+type JsonRecord = Record<string, unknown>;
+
+function extractEventItem(parsed: JsonRecord): JsonRecord | null {
+  return parsed.item && typeof parsed.item === "object" && !Array.isArray(parsed.item)
+    ? (parsed.item as JsonRecord)
+    : null;
+}
+
+function extractEventItemId(parsed: JsonRecord, eventItem: JsonRecord | null): string | null {
+  if (typeof parsed.item_id === "string") return parsed.item_id;
+  if (eventItem && typeof eventItem.id === "string") return eventItem.id;
+  return null;
+}
+
+function extractEventOutputIndex(parsed: JsonRecord): number | null {
+  return typeof parsed.output_index === "number" ? parsed.output_index : null;
+}
+
+// The `response.output_item.added` event that announces a new commentary-phase
+// item. Records its identifiers so follow-up delta/done events are recognized.
+function isCommentaryStart(
+  eventType: string,
+  parsed: JsonRecord,
+  eventItemId: string | null,
+  eventOutputIndex: number | null,
+  commentaryItemIds: Set<string>,
+  commentaryIndexes: Set<number>
+): boolean {
+  const isAddedEvent = eventType === "response.output_item.added";
+  if (!isAddedEvent || !isResponsesCommentaryMessageItem(parsed.item)) return false;
+
+  if (eventItemId) commentaryItemIds.add(eventItemId);
+  if (eventOutputIndex !== null) commentaryIndexes.add(eventOutputIndex);
+  return true;
+}
+
+// A follow-up delta/done event for an item already tracked as commentary.
+// Untracks the item once its `output_item.done` event is seen.
+function isCommentaryContinuation(
+  eventType: string,
+  eventItemId: string | null,
+  eventOutputIndex: number | null,
+  commentaryItemIds: Set<string>,
+  commentaryIndexes: Set<number>
+): boolean {
+  const belongsToCommentary =
+    (eventItemId !== null && commentaryItemIds.has(eventItemId)) ||
+    (eventOutputIndex !== null && commentaryIndexes.has(eventOutputIndex));
+  if (!belongsToCommentary) return false;
+
+  if (eventType === "response.output_item.done") {
+    if (eventItemId) commentaryItemIds.delete(eventItemId);
+    if (eventOutputIndex !== null) commentaryIndexes.delete(eventOutputIndex);
+  }
+  return true;
+}
+
+export function shouldDropResponsesCommentaryEvent(
+  parsed: JsonRecord,
+  commentaryItemIds: Set<string>,
+  commentaryIndexes: Set<number>
+): boolean {
+  const eventType = parsed.type as string;
+  const eventItem = extractEventItem(parsed);
+  const eventItemId = extractEventItemId(parsed, eventItem);
+  const eventOutputIndex = extractEventOutputIndex(parsed);
+
+  return (
+    isCommentaryStart(
+      eventType,
+      parsed,
+      eventItemId,
+      eventOutputIndex,
+      commentaryItemIds,
+      commentaryIndexes
+    ) ||
+    isCommentaryContinuation(
+      eventType,
+      eventItemId,
+      eventOutputIndex,
+      commentaryItemIds,
+      commentaryIndexes
+    )
+  );
+}

open-sse/utils/stream.ts

@@ -30,9 +30,9 @@ import { STREAM_IDLE_TIMEOUT_MS, FETCH_BODY_TIMEOUT_MS, HTTP_STATUS } from "../c
 import {
   OMIT_STREAMING_CHUNK_MARKER,
   sanitizeStreamingChunk,
-  isResponsesCommentaryMessageItem,
 } from "../handlers/responseSanitizer.ts";
 import { isFeatureFlagEnabled } from "@/shared/utils/featureFlags";
+import { shouldDropResponsesCommentaryEvent } from "./responsesCommentaryDrop.ts";
 import { buildErrorBody } from "./error.ts";
 import { parseTextualToolCallCandidate, isValidToolCallHeaderPrefix } from "./textualToolCall.ts";
 import { recordToolLatency } from "../services/toolLatencyTracker.ts";
@@ -1308,48 +1308,19 @@ export function createSSEStream(options: StreamOptions = {}) {
                     parsed.type === "error");
 
                 if (isResponsesSSE) {
-                  // #6199 — statefully drop internal commentary-phase output. The
-                  // `response.output_item.added` announces the phase; the follow-up
-                  // delta/done events only carry `item_id`/`output_index`, so we key
-                  // off those. Happy-path (non-commentary) events are untouched.
-                  if (shouldDropResponsesCommentary) {
-                    const responsesEventType = parsed.type as string;
-                    const eventOutputIndex =
-                      typeof parsed.output_index === "number" ? parsed.output_index : null;
-                    const eventItem =
-                      parsed.item && typeof parsed.item === "object" && !Array.isArray(parsed.item)
-                        ? (parsed.item as JsonRecord)
-                        : null;
-                    const eventItemId =
-                      typeof parsed.item_id === "string"
-                        ? parsed.item_id
-                        : eventItem && typeof eventItem.id === "string"
-                          ? eventItem.id
-                          : null;
-
-                    if (
-                      responsesEventType === "response.output_item.added" &&
-                      isResponsesCommentaryMessageItem(parsed.item)
-                    ) {
-                      if (eventItemId) passthroughResponsesCommentaryItemIds.add(eventItemId);
-                      if (eventOutputIndex !== null)
-                        passthroughResponsesCommentaryIndexes.add(eventOutputIndex);
-                      continue;
-                    }
-
-                    const belongsToCommentary =
-                      (eventItemId !== null &&
-                        passthroughResponsesCommentaryItemIds.has(eventItemId)) ||
-                      (eventOutputIndex !== null &&
-                        passthroughResponsesCommentaryIndexes.has(eventOutputIndex));
-                    if (belongsToCommentary) {
-                      if (responsesEventType === "response.output_item.done") {
-                        if (eventItemId) passthroughResponsesCommentaryItemIds.delete(eventItemId);
-                        if (eventOutputIndex !== null)
-                          passthroughResponsesCommentaryIndexes.delete(eventOutputIndex);
-                      }
-                      continue;
-                    }
+                  // #6199/#6561 — statefully drop internal commentary-phase output (see
+                  // ./responsesCommentaryDrop.ts) and clear the buffered `event:` line
+                  // for the same frame, or it flushes alone as an event-only SSE frame.
+                  if (
+                    shouldDropResponsesCommentary &&
+                    shouldDropResponsesCommentaryEvent(
+                      parsed as JsonRecord,
+                      passthroughResponsesCommentaryItemIds,
+                      passthroughResponsesCommentaryIndexes
+                    )
+                  ) {
+                    clearPendingPassthroughEvent();
+                    continue;
                   }
 
                   const responsesIdsNormalized = normalizeResponsesSseIds(parsed as JsonRecord);

tests/unit/responses-commentary-event-frame-6561.test.ts

@@ -0,0 +1,170 @@
+/**
+ * Regression test for #6561 (follow-up to #6199 / #6232).
+ *
+ * The upstream Responses-compatible SSE stream is a real `event: <type>\ndata:
+ * <json>\n\n` frame, not a bare `data:` line. The passthrough loop in
+ * open-sse/utils/stream.ts buffers the `event:` line into
+ * `passthroughEventPrefix` (via `.remember()`) and only clears/prefixes it when
+ * a `data:` line is actually forwarded (`prefixData()`/`clearPendingPassthroughEvent()`).
+ *
+ * The #6199 commentary-drop `continue;` branches (stream.ts ~1337 / ~1351) never
+ * called `clearPendingPassthroughEvent()` before skipping the `data:` line. So the
+ * buffered `event:` line survived and got flushed alone on the next blank line
+ * (stream.ts ~1200-1207), producing an event-only SSE frame with NO `data:` line
+ * — which is exactly what the OpenAI Python SDK chokes on (`json.loads("")`).
+ */
+
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-commentary-6561-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+const core = await import("../../src/lib/db/core.ts");
+
+const { createSSEStream } = await import("../../open-sse/utils/stream.ts");
+
+const textEncoder = new TextEncoder();
+
+async function readTransformed(chunks: string[], options: object): Promise<string> {
+  const source = new ReadableStream({
+    start(controller) {
+      for (const chunk of chunks) {
+        controller.enqueue(textEncoder.encode(chunk));
+      }
+      controller.close();
+    },
+  });
+  return new Response(source.pipeThrough(createSSEStream(options))).text();
+}
+
+test.after(() => {
+  core.resetDbInstance();
+  if (fs.existsSync(TEST_DATA_DIR)) {
+    fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+  }
+});
+
+const COMMENTARY_TEXT = "internal chain-of-thought commentary that must stay hidden";
+const FINAL_TEXT = "The final answer visible to the user.";
+
+// Real upstream frames carry BOTH an `event:` line and a `data:` line, unlike
+// the #6199 regression test's bare `data:`-only helper.
+function sseFrame(eventType: string, payload: object): string {
+  return `event: ${eventType}\ndata: ${JSON.stringify(payload)}\n\n`;
+}
+
+function buildResponsesStreamWithEventLines(): string[] {
+  return [
+    sseFrame("response.created", {
+      type: "response.created",
+      response: { id: "resp_6561", output: [] },
+    }),
+    // --- commentary item (internal, must be dropped when filtering) ---
+    sseFrame("response.output_item.added", {
+      type: "response.output_item.added",
+      output_index: 0,
+      item: {
+        id: "msg_commentary",
+        type: "message",
+        role: "assistant",
+        phase: "commentary",
+        content: [],
+      },
+    }),
+    sseFrame("response.content_part.added", {
+      type: "response.content_part.added",
+      output_index: 0,
+      item_id: "msg_commentary",
+      content_index: 0,
+      part: { type: "output_text", text: "" },
+    }),
+    sseFrame("response.output_text.delta", {
+      type: "response.output_text.delta",
+      output_index: 0,
+      item_id: "msg_commentary",
+      content_index: 0,
+      delta: COMMENTARY_TEXT,
+    }),
+    sseFrame("response.output_text.done", {
+      type: "response.output_text.done",
+      output_index: 0,
+      item_id: "msg_commentary",
+      content_index: 0,
+      text: COMMENTARY_TEXT,
+    }),
+    sseFrame("response.output_item.done", {
+      type: "response.output_item.done",
+      output_index: 0,
+      item: {
+        id: "msg_commentary",
+        type: "message",
+        role: "assistant",
+        phase: "commentary",
+        content: [{ type: "output_text", text: COMMENTARY_TEXT }],
+      },
+    }),
+    // --- final answer item (must always be forwarded) ---
+    sseFrame("response.output_item.added", {
+      type: "response.output_item.added",
+      output_index: 1,
+      item: {
+        id: "msg_final",
+        type: "message",
+        role: "assistant",
+        phase: "final",
+        content: [],
+      },
+    }),
+    sseFrame("response.output_text.delta", {
+      type: "response.output_text.delta",
+      output_index: 1,
+      item_id: "msg_final",
+      content_index: 0,
+      delta: FINAL_TEXT,
+    }),
+    sseFrame("response.output_item.done", {
+      type: "response.output_item.done",
+      output_index: 1,
+      item: {
+        id: "msg_final",
+        type: "message",
+        role: "assistant",
+        phase: "final",
+        content: [{ type: "output_text", text: FINAL_TEXT }],
+      },
+    }),
+  ];
+}
+
+const PASSTHROUGH_RESPONSES_OPTIONS = {
+  mode: "passthrough",
+  provider: "openai",
+  clientResponseFormat: "openai-responses",
+};
+
+test("#6561: dropping commentary via event+data frames must not leave event-only frames", async () => {
+  const output = await readTransformed(buildResponsesStreamWithEventLines(), {
+    ...PASSTHROUGH_RESPONSES_OPTIONS,
+    dropResponsesCommentary: true,
+  });
+
+  // Sanity: the original #6199 leak is indeed fixed — no commentary text leaks.
+  assert.ok(!output.includes(COMMENTARY_TEXT), "commentary text must not leak");
+
+  // Split the raw SSE text into frames (separated by a blank line) and find any
+  // frame that has an `event:` line but NO `data:` line — that is the bug.
+  const frames = output.split("\n\n").filter((f) => f.trim().length > 0);
+  const eventOnlyFrames = frames.filter((f) => /^event:/m.test(f) && !/^data:/m.test(f));
+
+  assert.deepEqual(
+    eventOnlyFrames,
+    [],
+    `expected no event-only (data-less) SSE frames, got:\n${JSON.stringify(eventOnlyFrames, null, 2)}`
+  );
+
+  // The final answer must still be forwarded correctly with its data line intact.
+  assert.ok(output.includes(FINAL_TEXT), "final answer text must be forwarded");
+});