mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-02 21:32:10 +03:00
* fix(stream): emit terminal SSE frames on mid-stream upstream failure (#7699) On /v1/messages (Anthropic format), when the upstream SSE stream fails mid-flight after bytes have been forwarded to the client, OmniRoute used to silently close the connection with no terminal event. Anthropic SDK and Claude Code report "Connection closed mid-response. The response above may be incomplete." Two fixes in open-sse/utils/streamHandler.ts: 1. buildStreamErrorChunks (Claude format) now emits event:message_stop after event:error — the Anthropic stream terminator that clients expect. Previously only event:error was sent, leaving the client hanging. 2. createDisconnectAwareStream pull() now detects upstream "done" without a client-visible terminal marker ([DONE] / response.completed / message_stop) and emits a synthetic terminal error frame instead of silently closing. This covers the case where the upstream drops the connection mid-stream without sending an error chunk. Adds tests/unit/silent-sse-close-7699.test.ts covering both code paths across Claude, OpenAI Chat, and OpenAI Responses formats. Refs: diegosouzapw/OmniRoute#7699 * fix(stream): scope terminal-marker missing detection to known formats with forwarded bytes Gate the done-path synthetic 502 error on bytesWereForwarded AND a known clientResponseFormat. Without this gate, any stream that closes cleanly without a terminal marker (including raw passthrough streams and non-API transforms) is incorrectly treated as a mid-stream drop. - Add bytesWereForwarded flag set on first Uint8Array chunk - Require clientResponseFormat to be set before injecting 502 - Fixes 3 broken stream-handler tests (pipes transformed bytes, slow upstream stall watchdog, normal completion watchdog) - Preserves #7699 fix: Claude-format streams that forwarded content but missed message_stop still get the synthetic terminal frame * fix(stream): scope terminal-marker heuristic to Claude only, add non-SSE regression #7699 is scoped to /v1/messages (Anthropic): Claude clients treat a stream that ends without message_stop as an error, and Anthropic's SSE spec explicitly permits a mid-stream event: error. The issue's own suggested fix says the current OpenAI silent-close "remains reasonable" — so the done-without-terminal-marker synthetic-502 heuristic must not fire for any other clientResponseFormat (gemini/codex/kiro/cursor/openai/openai-responses etc.), where a done stream with no [DONE]/response.completed/message_stop equivalent is not necessarily a silent drop. Narrows the gate from "any truthy clientResponseFormat" to "clientResponseFormat === FORMATS.CLAUDE" specifically. Adds a regression test asserting a plain non-SSE OpenAI-format completion (bytes forwarded, no terminal marker) is NOT mutated with a synthetic error frame. Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> * test(stream): trim new regression test to clear the 800-line test-file cap tests/unit/stream-handler.test.ts was 772 lines pre-#7816 (not in the frozen file-size baseline, so it's evaluated as new-file-cap 800). The added regression test pushed it to 807; trim boilerplate to land at 796. Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --------- Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
This commit is contained in:
Andrew B.
@@ -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);
|
||||
|
||||
172
tests/unit/silent-sse-close-7699.test.ts
Normal file
172
tests/unit/silent-sse-close-7699.test.ts
Normal file
@@ -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<Uint8Array, Uint8Array> {
|
||||
return new TransformStream({
|
||||
transform(chunk, controller) {
|
||||
controller.enqueue(chunk);
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Collect all bytes from a ReadableStream into a string.
|
||||
*/
|
||||
async function drainStream(stream: ReadableStream<Uint8Array>): Promise<string> {
|
||||
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<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(
|
||||
new TextEncoder().encode(
|
||||
'data: {"type":"content_block_delta","delta":{"text":"partial"}}\n\n'
|
||||
)
|
||||
);
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
const transform = new TransformStream<Uint8Array, Uint8Array>({
|
||||
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<void> } } {
|
||||
return { getWriter: () => ({ abort: () => Promise.resolve() }) };
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user