diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 720f361708..eb7bd792ac 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -376,7 +376,7 @@ export function createStreamController({ return controller; } -function buildStreamErrorChunks( +export function buildStreamErrorChunks( errorMsg: string, statusCode: number, clientResponseFormat?: string | null @@ -409,7 +409,13 @@ function buildStreamErrorChunks( }, }; - return encodeSseEvent(errorEvent, { event: "error" }); + // #7699 — emit message_stop after event:error so Anthropic SDK / Claude Code + // see a proper terminal frame instead of a silent mid-response close. + // Without message_stop, clients report "Connection closed mid-response." + return [ + ...encodeSseEvent(errorEvent, { event: "error" }), + ...encodeSseEvent({ type: "message_stop" }, { event: "message_stop" }), + ]; } const errorEvent = { @@ -457,10 +463,12 @@ export function createDisconnectAwareStream(transformStream, streamController) { const terminalDecoder = new TextDecoder(); let terminalTail = ""; let clientTerminalSeen = false; + let bytesWereForwarded = false; const noteClientChunk = (chunk: unknown) => { - if (clientTerminalSeen) return; if (!(chunk instanceof Uint8Array)) return; + bytesWereForwarded = true; + if (clientTerminalSeen) return; terminalTail += terminalDecoder.decode(chunk, { stream: true }); if (terminalTail.length > 4096) { @@ -486,8 +494,43 @@ export function createDisconnectAwareStream(transformStream, streamController) { try { const { done, value } = await reader.read(); if (done) { - streamController.handleComplete(); - controller.close(); + // #7699 — upstream ended without a client-visible terminal marker. + // Scoped to Claude (/v1/messages) specifically, which is the + // issue's real scope: Anthropic's SSE spec permits a mid-stream + // event: error and Claude clients (Claude Code, Anthropic SDK) + // treat a stream that ends without message_stop as an error. For + // every other format (plain OpenAI chat completions included — + // see #7699's "Suggested Fix") a done-without-recognized-marker + // close is NOT necessarily a silent drop (many providers/formats + // legitimately have no [DONE]/response.completed equivalent), so + // injecting a synthetic error there would be a false positive. + if ( + bytesWereForwarded && + !clientTerminalSeen && + streamController.clientResponseFormat === FORMATS.CLAUDE + ) { + streamController.handleError( + Object.assign(new Error("Upstream stream ended without a terminal marker"), { + statusCode: 502, + }) + ); + try { + for (const chunk of buildStreamErrorChunks( + "Upstream stream ended without a terminal marker", + 502, + streamController.clientResponseFormat + )) { + controller.enqueue(chunk); + } + } catch { + // downstream may have closed; original error already recorded + } + } else { + streamController.handleComplete(); + } + try { + controller.close(); + } catch {} return; } controller.enqueue(value); diff --git a/tests/unit/silent-sse-close-7699.test.ts b/tests/unit/silent-sse-close-7699.test.ts new file mode 100644 index 0000000000..a24a6670d2 --- /dev/null +++ b/tests/unit/silent-sse-close-7699.test.ts @@ -0,0 +1,172 @@ +/** + * Regression test for #7699 — silent SSE close on mid-stream upstream failure (/v1/messages). + * + * When the upstream SSE stream fails mid-flight (after bytes have been forwarded + * to the client) and the upstream drops without emitting a terminal marker, + * OmniRoute used to silently close the connection with no terminal `event: error` + * or `message_stop` for Anthropic-format clients. Claude Code and the Anthropic SDK + * then report "Connection closed mid-response. The response above may be incomplete." + * + * Two code paths must emit a synthetic terminal frame: + * 1. `buildStreamErrorChunks` for the Claude format must follow `event: error` + * with `event: message_stop` (the Anthropic stream terminator). + * 2. `createDisconnectAwareStream`'s `if (done)` branch must emit a synthetic + * error + terminal frame when upstream ends without a client-visible + * terminal marker (silent mid-stream drop). + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { buildStreamErrorChunks, createDisconnectAwareStream, createStreamController } = + await import("../../open-sse/utils/streamHandler.ts"); +const { FORMATS } = await import("../../open-sse/translator/formats.ts"); + +function decodeChunks(chunks: Uint8Array[]): string { + return new TextDecoder().decode( + (chunks as Uint8Array[]).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)) + ); +} + +test("#7699 buildStreamErrorChunks (Claude) emits event:error AND event:message_stop", () => { + const chunks = buildStreamErrorChunks( + "Upstream stream error", + 502, + FORMATS.CLAUDE + ) as Uint8Array[]; + const text = decodeChunks(chunks); + + // Must include an error event... + assert.match(text, /event: error\r?\n/); + assert.match(text, /"type":\s*"error"/); + assert.match(text, /"message":\s*"Upstream stream error"/); + + // ...AND a message_stop terminator so Anthropic SDK / Claude Code + // don't see a silent mid-response close (#7699). + assert.match(text, /event: message_stop\r?\n/); + assert.match(text, /"type":\s*"message_stop"/); +}); + +test("#7699 buildStreamErrorChunks (Claude) emits error before message_stop", () => { + const chunks = buildStreamErrorChunks("rate limited", 429, FORMATS.CLAUDE) as Uint8Array[]; + const text = decodeChunks(chunks); + + const errorIdx = text.indexOf("event: error"); + const stopIdx = text.indexOf("event: message_stop"); + assert.notEqual(errorIdx, -1, "expected event: error in output"); + assert.notEqual(stopIdx, -1, "expected event: message_stop in output"); + assert.ok(errorIdx < stopIdx, "event: error must precede event: message_stop"); +}); + +test("#7699 buildStreamErrorChunks (OpenAI) still emits [DONE] terminator (unchanged)", () => { + const chunks = buildStreamErrorChunks("Upstream stream error", 502, null) as Uint8Array[]; + const text = decodeChunks(chunks); + + // OpenAI format: finish_reason error + [DONE] + assert.match(text, /"finish_reason":\s*"error"/); + assert.match(text, /data: \[DONE\]/); + // Must NOT include Claude-only markers + assert.doesNotMatch(text, /message_stop/); +}); + +test("#7699 buildStreamErrorChunks (Responses) emits response.failed (unchanged)", () => { + const chunks = buildStreamErrorChunks( + "Upstream stream error", + 502, + FORMATS.OPENAI_RESPONSES + ) as Uint8Array[]; + const text = decodeChunks(chunks); + + assert.match(text, /event: response\.failed\r?\n/); + // Must NOT include Claude-only markers + assert.doesNotMatch(text, /message_stop/); +}); + +/** + * Helper: build a minimal TransformStream that forwards bytes unchanged + * so createDisconnectAwareStream can wrap it. We then feed it a synthetic + * upstream that ends (`done`) without ever emitting a terminal SSE marker. + */ +function buildPassthroughTransform(): TransformStream { + return new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk); + }, + }); +} + +/** + * Collect all bytes from a ReadableStream into a string. + */ +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)) + ); +} + +test("#7699 createDisconnectAwareStream emits synthetic error when upstream ends without terminal marker (Claude)", async () => { + // Upstream sends some partial content then ends (done=true) without + // ever emitting message_stop — reproduces the silent mid-stream close. + const upstream = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + 'data: {"type":"content_block_delta","delta":{"text":"partial"}}\n\n' + ) + ); + controller.close(); + }, + }); + + const transform = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk); + }, + }); + + // Pipe the upstream through the transform; the result is a ReadableStream. + const transformedBody = upstream.pipeThrough(transform); + + const sc = createStreamController({ + provider: "test", + model: "test-model", + clientResponseFormat: FORMATS.CLAUDE, + }); + + // createDisconnectAwareStream expects { readable, writable } — mirrors + // the shape produced by pipeWithDisconnect. + const wrapped = createDisconnectAwareStream( + { readable: transformedBody, writable: createNoopAbortWritableStream() }, + sc + ); + + const text = await drainStream(wrapped); + + // Must contain the partial content that was forwarded... + assert.match(text, /content_block_delta/); + // ...AND the synthetic terminal frames (error + message_stop) — NOT a silent close. + assert.match(text, /event: error\r?\n/); + assert.match(text, /event: message_stop\r?\n/); + assert.match(text, /Upstream stream ended without a terminal marker/); +}); + +// Minimal noop writable for the test wiring (mirrors createNoopAbortWritable). +function createNoopAbortWritableStream(): { getWriter: () => { abort: () => Promise } } { + return { getWriter: () => ({ abort: () => Promise.resolve() }) }; +} diff --git a/tests/unit/stream-handler.test.ts b/tests/unit/stream-handler.test.ts index 6271cd5fa5..d19a13f351 100644 --- a/tests/unit/stream-handler.test.ts +++ b/tests/unit/stream-handler.test.ts @@ -355,6 +355,30 @@ test("createDisconnectAwareStream keeps newlines escaped for Claude SSE errors", assert.doesNotMatch(text, /^claude line two/m); }); +// #7699/#7816 — heuristic is scoped to FORMATS.CLAUDE (/v1/messages); a +// plain non-Claude completion with no [DONE]/message_stop must pass through. +test("createDisconnectAwareStream does not append a synthetic error to a plain non-SSE OpenAI completion", async () => { + const transformStream = { + readable: new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("plain forwarded bytes, no completion marker")); + controller.close(); + }, + }), + writable: { getWriter: () => ({ abort() {} }) }, + }; + + const stream = createDisconnectAwareStream( + transformStream, + createStreamController({ clientResponseFormat: FORMATS.OPENAI }) + ); + const text = await readStreamText(stream); + + assert.equal(text, "plain forwarded bytes, no completion marker"); + assert.doesNotMatch(text, /event: error/); + assert.doesNotMatch(text, /"finish_reason":"error"/); +}); + test("createDisconnectAwareStream cancel propagates disconnect reason and aborts the writer", async () => { let aborted = false; let disconnectEvent = null;