diff --git a/changelog.d/fixes/10156-responses-commentary-completed-snapshot.md b/changelog.d/fixes/10156-responses-commentary-completed-snapshot.md new file mode 100644 index 0000000000..7a976ab85d --- /dev/null +++ b/changelog.d/fixes/10156-responses-commentary-completed-snapshot.md @@ -0,0 +1 @@ +- **fix(sse):** Responses-passthrough `response.completed` snapshots now drop `phase:"commentary"` items the same way live SSE frames already do, so the terminal `response.output` array no longer echoes internal commentary text that was already suppressed from the stream (#10156). diff --git a/open-sse/utils/responsesStreamHelpers.ts b/open-sse/utils/responsesStreamHelpers.ts index ce40999eb8..a2cba80fc1 100644 --- a/open-sse/utils/responsesStreamHelpers.ts +++ b/open-sse/utils/responsesStreamHelpers.ts @@ -121,6 +121,29 @@ export function pushUniqueResponsesOutputItems(target: unknown[], items: readonl } } +/** + * #10156 — strip items matched by `isCommentaryItem` (the same predicate used + * to drop live commentary-phase SSE frames, #6199) from a `response.completed` + * output array before it is forwarded or buffered for backfill. Upstreams may + * echo an already-dropped commentary item back inside a non-empty terminal + * `output` array; without this, the live stream and the terminal snapshot + * silently disagree about what the client actually saw. + */ +export function filterResponsesCommentaryFromItems( + items: readonly unknown[], + isCommentaryItem: (item: unknown) => boolean +): { items: unknown[]; changed: boolean } { + let changed = false; + const filtered = items.filter((item) => { + if (isCommentaryItem(item)) { + changed = true; + return false; + } + return true; + }); + return { items: filtered, changed }; +} + export function backfillResponsesCompletedOutput( parsed: unknown, collectedItems: readonly unknown[] diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 275669e983..4eab8ea7fc 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -36,6 +36,7 @@ import { import { STREAM_IDLE_TIMEOUT_MS, FETCH_BODY_TIMEOUT_MS, HTTP_STATUS } from "../config/constants.ts"; import { OMIT_STREAMING_CHUNK_MARKER, + isResponsesCommentaryMessageItem, sanitizeStreamingChunk, } from "../handlers/responseSanitizer.ts"; import { isFeatureFlagEnabled } from "@/shared/utils/featureFlags"; @@ -59,6 +60,7 @@ import { } from "../services/sessionManager.ts"; import { backfillResponsesCompletedOutput, + filterResponsesCommentaryFromItems, normalizeResponsesCompletedUsage as normalizeUsage, normalizeResponsesSseIds, pushUniqueResponsesOutputItems, @@ -1565,11 +1567,26 @@ export function createSSEStream(options: StreamOptions = {}) { } } } + let responsesCommentaryStrippedFromCompleted = false; if ( parsed.type === "response.completed" && Array.isArray(parsed.response?.output) && parsed.response.output.length > 0 ) { + // #10156 — an upstream may echo a `phase:"commentary"` item back + // inside a non-empty terminal `output` array even though its live + // SSE frames were already dropped above. Keep both representations + // consistent by applying the same drop here. + if (shouldDropResponsesCommentary) { + const { items, changed } = filterResponsesCommentaryFromItems( + parsed.response.output, + isResponsesCommentaryMessageItem + ); + if (changed) { + parsed.response.output = items; + responsesCommentaryStrippedFromCompleted = true; + } + } pushUniqueResponsesOutputItems( passthroughResponsesOutputItems, parsed.response.output @@ -1613,9 +1630,19 @@ export function createSSEStream(options: StreamOptions = {}) { ]) as typeof parsed; } const stripped = stripResponsesLifecycleEcho(parsed); + // Belt-and-suspenders for #10156: filter the backfill buffer itself + // before it can seed an empty `response.completed.response.output`, + // in case a future code path pushes a commentary item into it + // without going through the response.completed branch above. + const backfillCandidates = shouldDropResponsesCommentary + ? filterResponsesCommentaryFromItems( + passthroughResponsesOutputItems, + isResponsesCommentaryMessageItem + ).items + : passthroughResponsesOutputItems; const backfilled = backfillResponsesCompletedOutput( parsed, - passthroughResponsesOutputItems + backfillCandidates ); const usageNormalized = normalizeUsage(parsed); if ( @@ -1623,7 +1650,8 @@ export function createSSEStream(options: StreamOptions = {}) { backfilled || textualToolCallBackfilled || responsesIdsNormalized || - usageNormalized + usageNormalized || + responsesCommentaryStrippedFromCompleted ) { output = `data: ${JSON.stringify(parsed)}\n\n`; injectedUsage = true; diff --git a/tests/unit/responses-commentary-passthrough-6199.test.ts b/tests/unit/responses-commentary-passthrough-6199.test.ts index 2b041c48d8..c1fce80786 100644 --- a/tests/unit/responses-commentary-passthrough-6199.test.ts +++ b/tests/unit/responses-commentary-passthrough-6199.test.ts @@ -364,3 +364,63 @@ test("Claude to Responses translation includes canonical Codex usage", async () assert.equal(completed.response.usage.output_tokens, 6); assert.equal(completed.response.usage.total_tokens, 94); }); + +// #10156 — the live-frame drop above works correctly, but real upstreams (as in +// the issue's repro) echo the ALREADY-DROPPED commentary item back inside the +// terminal `response.completed.response.output` array. Because that array is +// non-empty, `backfillResponsesCompletedOutput` never touches it, so the +// terminal snapshot silently disagreed with the events already delivered to +// the client. This must stay filtered too. +test("response.completed strips a commentary item the upstream echoes back non-empty (#10156)", async () => { + const output = await readTransformed( + [ + ...buildResponsesStream().slice(0, -1), + sse({ + type: "response.completed", + response: { + id: "resp_10156", + output: [ + { + id: "msg_commentary", + type: "message", + role: "assistant", + phase: "commentary", + content: [{ type: "output_text", text: COMMENTARY_TEXT }], + }, + { + id: "msg_final", + type: "message", + role: "assistant", + phase: "final", + content: [{ type: "output_text", text: FINAL_TEXT }], + }, + ], + }, + }), + ], + { ...PASSTHROUGH_RESPONSES_OPTIONS, dropResponsesCommentary: true } + ); + + assert.ok( + !output.includes(COMMENTARY_TEXT), + "commentary text must never reach the client, live or in the terminal snapshot" + ); + assert.ok( + !output.includes("msg_commentary"), + "the commentary item id must not appear anywhere in the forwarded stream" + ); + + const completedLine = output + .split(/\r?\n/) + .find((line) => line.startsWith("data:") && line.includes('"response.completed"')); + assert.ok(completedLine, "the terminal Responses event must be forwarded"); + const completed = JSON.parse(completedLine.slice(5).trim()); + assert.ok( + !completed.response.output.some((item: { phase?: string }) => item.phase === "commentary"), + "BUG #10156: response.completed.response.output must not retain the commentary item once its live SSE frames were suppressed — live stream and terminal snapshot must stay consistent" + ); + assert.ok( + completed.response.output.some((item: { id?: string }) => item.id === "msg_final"), + "the final answer item must still be present in the terminal snapshot" + ); +});