From e9d677055ecc24b20a97a38fc0d737bbc7526333 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Fri, 10 Jul 2026 19:38:52 -0300 Subject: [PATCH] fix(api): Responses passthrough emits event-only SSE frames after filtering commentary output (#6561) (#6735) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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) --- .../6735-responses-commentary-event-frame.md | 1 + open-sse/utils/responsesCommentaryDrop.ts | 98 ++++++++++ open-sse/utils/stream.ts | 57 ++---- ...ponses-commentary-event-frame-6561.test.ts | 170 ++++++++++++++++++ 4 files changed, 283 insertions(+), 43 deletions(-) create mode 100644 changelog.d/fixes/6735-responses-commentary-event-frame.md create mode 100644 open-sse/utils/responsesCommentaryDrop.ts create mode 100644 tests/unit/responses-commentary-event-frame-6561.test.ts diff --git a/changelog.d/fixes/6735-responses-commentary-event-frame.md b/changelog.d/fixes/6735-responses-commentary-event-frame.md new file mode 100644 index 0000000000..6dce5931c9 --- /dev/null +++ b/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). diff --git a/open-sse/utils/responsesCommentaryDrop.ts b/open-sse/utils/responsesCommentaryDrop.ts new file mode 100644 index 0000000000..35ac68e8ec --- /dev/null +++ b/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; + +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, + commentaryIndexes: Set +): 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, + commentaryIndexes: Set +): 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, + commentaryIndexes: Set +): 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 + ) + ); +} diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 120644ac9d..0ed9f3aac6 100644 --- a/open-sse/utils/stream.ts +++ b/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); diff --git a/tests/unit/responses-commentary-event-frame-6561.test.ts b/tests/unit/responses-commentary-event-frame-6561.test.ts new file mode 100644 index 0000000000..99fcfeee7b --- /dev/null +++ b/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: \ndata: + * \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 { + 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"); +});