From 9469b9c79e167fc32b785c9c2a4a6664f724cb5e Mon Sep 17 00:00:00 2001 From: Harvey Doan Date: Sat, 22 Aug 2026 00:59:07 +0700 Subject: [PATCH] fix(sse): surface bare upstream close as response.failed for Responses clients (#10980) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ⭐5 — resolveSilentCloseOutcome tratava bare upstream close para clientes Claude (#7699) e OpenAI chat-completions (#10443), mas clientes Responses-format caíam por ambos os branches e recebiam um close silencioso no meio do stream. Estende o veredito para OPENAI_RESPONSES/OPENAI_RESPONSE: emite response.failed sintético preservando o conteúdo já entregue. 3 novos + 92 testes-irmãos (streamHandler) verdes. --- open-sse/utils/streamHandler.ts | 15 ++- ...nt-sse-close-responses-no-terminal.test.ts | 116 ++++++++++++++++++ 2 files changed, 127 insertions(+), 4 deletions(-) create mode 100644 tests/unit/silent-sse-close-responses-no-terminal.test.ts diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 0a7e41d5bb..4d8993c767 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -589,10 +589,7 @@ function resolveSilentCloseOutcome(input: { if (!input.bytesWereForwarded) return null; if (!input.clientTerminalSeen) { - if ( - input.clientResponseFormat === FORMATS.CLAUDE && - input.contentWatcher.sawContent() - ) { + if (input.clientResponseFormat === FORMATS.CLAUDE && input.contentWatcher.sawContent()) { // #7699 — upstream dropped after content reached the client on a Claude // stream. Keep the partial response: emit a clean max_tokens completion // instead of an error frame so Anthropic SDK / Claude Code don't report @@ -611,6 +608,16 @@ function resolveSilentCloseOutcome(input: { if (input.clientResponseFormat === FORMATS.OPENAI && input.contentWatcher.sawContent()) { return { kind: "error", reason: "Upstream stream ended without a terminal marker" }; } + // Responses-format clients (Codex CLI and other /v1/responses consumers): + // a healthy OpenAI Responses stream ALWAYS terminates with an explicit + // `response.completed` event — it is the format's only terminal marker and + // carries the final status/usage. Content forwarded without it is an + // upstream drop, the same class as #10443 for chat completions; surface a + // synthetic response.failed instead of a silent close so clients report + // the break instead of waiting on a completion event that never comes. + if (isResponsesClientFormat(input.clientResponseFormat) && input.contentWatcher.sawContent()) { + return { kind: "error", reason: "Upstream stream ended without a terminal marker" }; + } } const watcher = input.contentWatcher; diff --git a/tests/unit/silent-sse-close-responses-no-terminal.test.ts b/tests/unit/silent-sse-close-responses-no-terminal.test.ts new file mode 100644 index 0000000000..f3a23edeb9 --- /dev/null +++ b/tests/unit/silent-sse-close-responses-no-terminal.test.ts @@ -0,0 +1,116 @@ +/** + * Regression test — silent SSE close on OpenAI Responses-format clients. + * + * Companion to #10443 (OpenAI chat completions) and #7699 (Claude). A healthy + * OpenAI Responses stream ALWAYS terminates with an explicit + * `response.completed` event; it is the format's only terminal marker and + * carries the final status/usage. When the upstream forwards content deltas + * and then closes without that event, Responses-format clients (Codex CLI and + * other /v1/responses consumers) previously received a silent mid-stream + * close — indistinguishable from a healthy end, so the client waits on a + * completion event that never arrives. The close must surface a synthetic + * `response.failed` instead, keeping everything already forwarded. + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { createDisconnectAwareStream, createStreamController } = + await import("../../open-sse/utils/streamHandler.ts"); +const { FORMATS } = await import("../../open-sse/translator/formats.ts"); + +function createNoopAbortWritableStream(): { getWriter: () => { abort: () => Promise } } { + return { getWriter: () => ({ abort: () => Promise.resolve() }) }; +} + +async function drainStream(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const parts: Uint8Array[] = []; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + parts.push(value); + } + return new TextDecoder().decode( + parts.reduce((acc, c) => { + const merged = new Uint8Array(acc.length + c.length); + merged.set(acc, 0); + merged.set(c, acc.length); + return merged; + }, new Uint8Array(0)) + ); +} + +async function runClientStream( + upstreamChunks: string[], + clientResponseFormat: string | null +): Promise { + const upstream = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + for (const chunk of upstreamChunks) controller.enqueue(encoder.encode(chunk)); + controller.close(); + }, + }); + + const transform = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk); + }, + }); + const transformedBody = upstream.pipeThrough(transform); + + const sc = createStreamController({ + provider: "opencode-go", + model: "muse-spark-1.2-contributor", + clientResponseFormat, + }); + + const wrapped = createDisconnectAwareStream( + { readable: transformedBody, writable: createNoopAbortWritableStream() }, + sc + ); + + return drainStream(wrapped); +} + +test("Responses format: content then bare close emits response.failed, not a silent close", async () => { + const text = await runClientStream( + [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"partial"}\n\n', + ], + FORMATS.OPENAI_RESPONSES + ); + + // The forwarded content survives... + assert.match(text, /partial/); + // ...and the close must be flagged with the format's failure terminal. + assert.match(text, /response\.failed/); + assert.match(text, /Upstream stream ended without a terminal marker/); +}); + +test("Responses format: response.completed counts as terminal, no synthetic failure", async () => { + const text = await runClientStream( + [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"done"}\n\n', + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n', + ], + FORMATS.OPENAI_RESPONSES + ); + + assert.match(text, /done/); + assert.match(text, /response\.completed/); + assert.doesNotMatch(text, /response\.failed/); + assert.doesNotMatch(text, /Upstream stream ended without a terminal marker/); +}); + +test("Responses format: OPENAI_RESPONSE alias gets the same bare-close verdict", async () => { + const text = await runClientStream( + [ + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"partial"}\n\n', + ], + FORMATS.OPENAI_RESPONSE + ); + + assert.match(text, /partial/); + assert.match(text, /response\.failed/); +});