From d0ff56613cf36c1ba9f925e22ca1ca6a5999a53c Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Sun, 5 Jul 2026 18:20:59 -0300 Subject: [PATCH] =?UTF-8?q?wip(sse):=20extract=20responsesCommentaryFilter?= =?UTF-8?q?=20from=20stream.ts=20(rescue;=20behavior=20j=C3=A1=20shipou=20?= =?UTF-8?q?via=20#6232)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- open-sse/utils/responsesCommentaryFilter.ts | 59 +++++++++++++++++++++ open-sse/utils/stream.ts | 56 +++++-------------- 2 files changed, 72 insertions(+), 43 deletions(-) create mode 100644 open-sse/utils/responsesCommentaryFilter.ts diff --git a/open-sse/utils/responsesCommentaryFilter.ts b/open-sse/utils/responsesCommentaryFilter.ts new file mode 100644 index 0000000000..2a8a5410d1 --- /dev/null +++ b/open-sse/utils/responsesCommentaryFilter.ts @@ -0,0 +1,59 @@ +/** + * Responses API passthrough — internal commentary-phase suppression (#6199). + * + * The Responses streaming passthrough must not forward internal commentary-phase + * output to clients. A commentary item is announced by `response.output_item.added` + * (which carries the `phase`), but its follow-up `response.output_text.delta` / + * `response.output_text.done` / `response.output_item.done` events only carry + * `item_id` / `output_index`. This helper tracks the announced commentary item id + + * output index in two caller-owned Sets and reports whether the current event should + * be dropped, so the hot per-chunk loop in stream.ts stays a single call. + */ + +import { isResponsesCommentaryMessageItem } from "../handlers/responseSanitizer.ts"; + +type JsonRecord = Record; + +/** + * Decide whether a single Responses SSE event is internal commentary that must be + * dropped before forwarding. Mutates the two tracking Sets: an `output_item.added` + * for a commentary item is recorded (and dropped); its trailing events are dropped + * while tracked; the item's `output_item.done` clears the tracking. + * + * @returns true when the event should be dropped (the caller should `continue`). + */ +export function shouldDropResponsesCommentaryEvent( + parsed: JsonRecord, + commentaryItemIds: Set, + commentaryIndexes: Set +): boolean { + const eventType = 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 (eventType === "response.output_item.added" && isResponsesCommentaryMessageItem(parsed.item)) { + if (eventItemId) commentaryItemIds.add(eventItemId); + if (eventOutputIndex !== null) commentaryIndexes.add(eventOutputIndex); + return true; + } + + 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; +} diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 120644ac9d..6fa9b29b86 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -30,8 +30,8 @@ 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 { shouldDropResponsesCommentaryEvent } from "./responsesCommentaryFilter.ts"; import { isFeatureFlagEnabled } from "@/shared/utils/featureFlags"; import { buildErrorBody } from "./error.ts"; import { parseTextualToolCallCandidate, isValidToolCallHeaderPrefix } from "./textualToolCall.ts"; @@ -1308,48 +1308,18 @@ 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 — statefully drop internal commentary-phase output + // (logic extracted to responsesCommentaryFilter.ts). Happy-path + // (non-commentary) events are untouched. + if ( + shouldDropResponsesCommentary && + shouldDropResponsesCommentaryEvent( + parsed as JsonRecord, + passthroughResponsesCommentaryItemIds, + passthroughResponsesCommentaryIndexes + ) + ) { + continue; } const responsesIdsNormalized = normalizeResponsesSseIds(parsed as JsonRecord);