From 3b91f9c90a8541897913bd4a818b0db2faa11f00 Mon Sep 17 00:00:00 2001 From: Chewji9875 <126886556+Chewji9875@users.noreply.github.com> Date: Sat, 18 Jul 2026 08:24:47 -0300 Subject: [PATCH 1/3] fix(combo): failover when upstream SSE is truncated mid-lifecycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit User log 1784230812441-bf3789: a combo target returned an SSE stream that carried bytes but never sent a recognised terminator (`data: [DONE]`, `message_stop`, `message_delta` with `stop_reason`, or a `finish_reason`) and never produced a single parseable SSE frame. The streaming quality validator's generic done-branch gate only checked `!sawAnyBytes`, so any byte at all — even unparseable garbage — passed the stream through. The combo did not fail over to the next target and the downstream SSE client hung waiting for events that never arrived. Rebuilt against the current release/v3.8.49 tip instead of the original branch diff: the original diff predates and deletes two fixes already merged to release — issue #7285 (`OpenAiLifecycleFlags` / `applyOpenAiLifecycleEvent`, the OpenAI-shape "truncated without finish_reason" failover branch) and issue #1382 (`SseLifecycleFlags .hasRealContent`, the Claude real-content vs. empty-content_block nuance). Both are preserved untouched here. Two new flags are tracked in parallel to that existing machinery instead of replacing it: * sawStructuredSSE — any parseable `event:` or `data:` frame was seen, even one carrying no recognised content (ping/metadata) — keeps the #3399/#3685 pass-through contract for those streams. * sawTerminator — a recognised terminator arrived: `data: [DONE]`, an OpenAI `finish_reason` (mirrors `openAi.hasTerminalMarker`), a Claude `message_stop`/`message_delta` with `stop_reason` (mirrors `sse.hasLifecycleEnd`), or a terminal `usage`-only chunk (new). The generic done-branch gate now requires neither flag to be true before marking the stream invalid, replacing the old `!sawAnyBytes` check (now dead and removed). The #7285 and #1382 branches are untouched. Tests added in tests/unit/validate-response-quality.test.ts (adapted from the original branch, same scenarios): 1. incomplete lifecycle (the bug) -> invalid 2. `[DONE]` only -> valid (regression guard for #3685) 3. `event: ping` only -> valid (regression guard for #3399) 4. OpenAI `finish_reason`-only chunk (no `[DONE]`) -> valid, isolates the new finish_reason check Full touched-area regression set verified green (51/51): the new tests plus combo-streaming-openai-no-finish-reason-7285, streaming-empty- content-block-1382, combo-quality-validator-reasoning, masked-200- exhaustion-fallback-6427, combo-streaming-empty-content-failover, combo-empty-content-failover-5085, combo-response-validation-failover, and combo-response-validation. Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --- .../7545-combo-failover-truncated-sse.md | 1 + open-sse/services/combo/validateQuality.ts | 63 +++++++++++++++--- tests/unit/validate-response-quality.test.ts | 66 +++++++++++++++++++ 3 files changed, 121 insertions(+), 9 deletions(-) create mode 100644 changelog.d/fixes/7545-combo-failover-truncated-sse.md diff --git a/changelog.d/fixes/7545-combo-failover-truncated-sse.md b/changelog.d/fixes/7545-combo-failover-truncated-sse.md new file mode 100644 index 000000000000..1f0c9852c83f --- /dev/null +++ b/changelog.d/fixes/7545-combo-failover-truncated-sse.md @@ -0,0 +1 @@ +- **fix(combo):** streaming combo failover now fails over when an upstream SSE response is truncated mid-lifecycle with no recognised terminator (`data: [DONE]`, `finish_reason`, `message_stop`/`message_delta` with `stop_reason`, or a terminal `usage`-only chunk) and no structured SSE frame at all — previously any byte, even unparseable garbage, satisfied the generic done-branch gate and the truncated stream was passed through, leaving the client hung waiting for events that never arrived ([#7545](https://github.com/diegosouzapw/OmniRoute/pull/7545)) — thanks @Chewji9875 diff --git a/open-sse/services/combo/validateQuality.ts b/open-sse/services/combo/validateQuality.ts index 99805eb35ca4..970894c0b798 100644 --- a/open-sse/services/combo/validateQuality.ts +++ b/open-sse/services/combo/validateQuality.ts @@ -252,9 +252,23 @@ export async function validateResponseQuality( hasLifecycleEnd: false, }; let anyContentFound = false; - let sawAnyBytes = false; // #7285: OpenAI-shape lifecycle tracking, parallel to `sse` above. const openAi: OpenAiLifecycleFlags = { hasChoicePayload: false, hasTerminalMarker: false }; + // User log 1784230812441-bf3789: the previous `!sawAnyBytes` gate below let + // ANY byte — even unparseable garbage with no SSE framing at all — pass + // combo failover through. These two flags are tracked in parallel to + // `sse`/`openAi` above and only tighten the GENERIC done-branch gate + // further down; the #1382 (`sse.hasRealContent`) and #7285 + // (`openAi.hasTerminalMarker`) branches are untouched. + // - sawStructuredSSE — a parseable `event:` or `data:` frame was seen, + // even one that carries no recognised content (ping/metadata) — the + // #3399 pass-through contract for those streams is preserved. + // - sawTerminator — a recognised terminator arrived: `data: [DONE]`, + // an OpenAI `finish_reason` (mirrors `openAi.hasTerminalMarker`), a + // Claude `message_stop`/`message_delta` with `stop_reason` (mirrors + // `sse.hasLifecycleEnd`), or a terminal `usage`-only chunk (new). + let sawStructuredSSE = false; + let sawTerminator = false; const sseLineNormalizer = createSSEDataLineNormalizer(); let pendingEventType = ""; @@ -277,6 +291,9 @@ export async function validateResponseQuality( if (trimmed.startsWith("event:")) { pendingEventType = trimmed.slice(6).trim(); + // An `event:` line is structured SSE framing on its own, even + // before any `data:` payload arrives (e.g. a bare keepalive ping). + sawStructuredSSE = true; continue; } @@ -291,6 +308,7 @@ export async function validateResponseQuality( // #7285: `[DONE]` is itself a terminal marker for OpenAI-shape // streams, even when no earlier chunk carried `finish_reason`. openAi.hasTerminalMarker = true; + sawTerminator = true; continue; } @@ -301,12 +319,31 @@ export async function validateResponseQuality( continue; } + // A successfully parsed `data:` payload is structured SSE activity + // regardless of shape or content — tracked only for the generic + // done-branch gate below; the #1382/#7285 branches are unaffected. + sawStructuredSSE = true; + applyOpenAiLifecycleEvent(parsed, openAi); + if (openAi.hasTerminalMarker) sawTerminator = true; const eventType = (typeof parsed.type === "string" ? parsed.type : null) || pendingEventType || ""; pendingEventType = ""; + // Some providers send a terminal `usage`-only chunk (no `choices`) + // as the final SSE frame instead of a `[DONE]`/`finish_reason` + // marker. Excludes Responses API `response.*` events, which have + // their own dedicated handling via `isKnownNonClaudeStreamPayload`. + if ( + parsed.usage && + typeof parsed.usage === "object" && + !Array.isArray(parsed.choices) && + !eventType.startsWith("response.") + ) { + sawTerminator = true; + } + if (isKnownNonClaudeStreamPayload(parsed, eventType)) { return true; } @@ -314,6 +351,7 @@ export async function validateResponseQuality( if (applySseLifecycleEvent(eventType, parsed, sse)) { return true; } + if (sse.hasLifecycleEnd) sawTerminator = true; } return false; } @@ -384,15 +422,23 @@ export async function validateResponseQuality( } // Stream ended with a truly EMPTY body (e.g. Gemini returning HTTP - // 200 with zero bytes) — mark as invalid for combo failover so the - // sibling model gets tried. Streams that carried ANY SSE activity - // (an explicit `data: [DONE]`, ping/metadata events, an incomplete - // Claude lifecycle) keep the pass-through contract (#3399/#3685): - // those are handled by the stream-readiness timeout, not failover. - if (!anyContentFound && !sse.hasContentBlock && !sawAnyBytes) { + // 200 with zero bytes), or with bytes that never formed a single + // recognizable SSE frame and never signalled termination — mark as + // invalid for combo failover so the sibling model gets tried. + // Streams that carried ANY structured SSE activity (an explicit + // `data: [DONE]`, ping/metadata events, an incomplete Claude + // lifecycle) or a recognised terminator keep the pass-through + // contract (#3399/#3685): those are handled by the stream-readiness + // timeout, not failover. + // + // Tightened after user log 1784230812441-bf3789: the previous + // `!sawAnyBytes` check let ANY byte — even unparseable garbage that + // never produced a single structured SSE frame — pass through, + // leaving the downstream SSE parser hung on a half-finished stream. + if (!anyContentFound && !sse.hasContentBlock && !sawTerminator && !sawStructuredSSE) { log.warn?.( "COMBO", - "Streaming response ended with no recognized content — marking as invalid for combo failover" + "Streaming response ended with no recognized content or SSE terminator — marking as invalid for combo failover" ); return { valid: false, reason: "streaming no recognized content" }; } @@ -423,7 +469,6 @@ export async function validateResponseQuality( // Accumulate raw bytes for potential replay. bufferedChunks.push(value); - if (value && value.length > 0) sawAnyBytes = true; // Decode incrementally (stream:true keeps multi-byte char state). decodedSoFar += decoder.decode(value, { stream: true }); diff --git a/tests/unit/validate-response-quality.test.ts b/tests/unit/validate-response-quality.test.ts index c95df223f781..7ea04a15ba79 100644 --- a/tests/unit/validate-response-quality.test.ts +++ b/tests/unit/validate-response-quality.test.ts @@ -83,3 +83,69 @@ test("releaseQualityClone does not throw when there is no clonedResponse", () => const original = new Response("body"); assert.doesNotThrow(() => releaseQualityClone({} as Response, original, {})); }); + +// ── Combo fallback silent-stop regression (#3399/#3685 + user log 1784230812441) ── +// +// Bug: combo streamed an upstream SSE response that carried bytes but never sent +// `data: [DONE]`, `message_stop`, or any `content_block_*`. The validator saw +// `sawAnyBytes === true` and passed the response through; OpenCode then hung +// waiting for the next event. Reported via local dashboard log +// `1784230812441-bf3789` (no public GitHub issue). +// +// Fix: the streaming validator now passes through only when it actually saw a +// recognised SSE terminator ([DONE], `message_stop`/`message_delta` with +// `stop_reason`, OpenAI `finish_reason`, terminal `usage`) OR structured SSE +// activity (parsed `data:` / `event:` frames) — tracked alongside (not instead +// of) the existing #7285/#1382 lifecycle machinery. Raw bytes that never +// produced a parseable event now correctly mark invalid. + +function makeSseResponse(body: string): Response { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(body)); + controller.close(); + }, + }); + return new Response(stream, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); +} + +test("streaming incomplete lifecycle: bytes with no terminator and no structured SSE → invalid", async () => { + // Garbage bytes that look like SSE prefix but never produce a complete + // `data:` line, no `event:`, no [DONE], no message_stop. This is the + // exact failure mode from log 1784230812441-bf3789. + const res = makeSseResponse(": keepalive\n\npartial da"); + const verdict = await validateResponseQuality(res, true, {}); + assert.strictEqual(verdict.valid, false); + assert.match(verdict.reason ?? "", /streaming/); +}); + +test("streaming [DONE] only (no content) → still valid (regression guard for #3685)", async () => { + const res = makeSseResponse("data: [DONE]\n\n"); + const verdict = await validateResponseQuality(res, true, {}); + assert.strictEqual(verdict.valid, true); +}); + +test("streaming event: ping only (no content, no terminator) → still valid (regression guard for #3399)", async () => { + // Some upstream providers emit periodic SSE pings for keepalive. The + // validator must continue to pass them through so the downstream SSE + // parser receives them rather than dropping the connection mid-stream. + const res = makeSseResponse(": ping - 2026-07-17\n\nevent: ping\ndata: {}\n\n"); + const verdict = await validateResponseQuality(res, true, {}); + assert.strictEqual(verdict.valid, true); +}); + +test("streaming OpenAI finish_reason-only chunk (no content delta) → valid (recognised terminator)", async () => { + // Some reasoning models emit a final `finish_reason: "stop"` chunk with + // no content and no follow-up `data: [DONE]`. That's a legitimate empty + // completion, not a truncation. Sending the `finish_reason` chunk + // WITHOUT a trailing `[DONE]` isolates the new finish_reason check — + // removing it would flip this test to invalid. + const res = makeSseResponse( + 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n' + ); + const verdict = await validateResponseQuality(res, true, {}); + assert.strictEqual(verdict.valid, true); +}); From 0156e845be1537c559db77a68d0432e5dbbdcc83 Mon Sep 17 00:00:00 2001 From: Chewji9875 <126886556+Chewji9875@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:32:30 -0300 Subject: [PATCH 2/3] refactor(combo): extract consumeSseLine + isTerminalUsageOnlyChunk helpers (complexity gate on parseAccumulatedSse) Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --- open-sse/services/combo/validateQuality.ts | 95 ++++++++++++---------- 1 file changed, 52 insertions(+), 43 deletions(-) diff --git a/open-sse/services/combo/validateQuality.ts b/open-sse/services/combo/validateQuality.ts index 7694ce4fecad..1e8d038054a2 100644 --- a/open-sse/services/combo/validateQuality.ts +++ b/open-sse/services/combo/validateQuality.ts @@ -296,43 +296,63 @@ export async function validateResponseQuality( * is detected, or "error" when the upstream reports a failure before content. * Otherwise peeking continues. */ + // Some providers send a terminal `usage`-only chunk (no `choices`) as the + // final SSE frame instead of a `[DONE]`/`finish_reason` marker. Excludes + // Responses API `response.*` events, which have their own dedicated + // handling via `isKnownNonClaudeStreamPayload`. + function isTerminalUsageOnlyChunk(parsed: Record, eventType: string): boolean { + return Boolean( + parsed.usage && + typeof parsed.usage === "object" && + !Array.isArray(parsed.choices) && + !eventType.startsWith("response.") + ); + } + + // Consume one normalized SSE line: track `event:` framing / keepalives / + // `[DONE]` terminators in the enclosing state, and return the JSON-parsed + // `data:` payload when (and only when) the line carries one. + function consumeSseLine(line: string): Record | null { + const trimmed = line.trim(); + + if (trimmed.startsWith("event:")) { + pendingEventType = trimmed.slice(6).trim(); + // An `event:` line is structured SSE framing on its own, even + // before any `data:` payload arrives (e.g. a bare keepalive ping). + sawStructuredSSE = true; + return null; + } + + if (!trimmed.startsWith("data:")) { + if (!trimmed) pendingEventType = ""; + return null; + } + + const data = trimmed.slice(5).trim(); + if (!data) return null; + if (data === "[DONE]") { + // #7285: `[DONE]` is itself a terminal marker for OpenAI-shape + // streams, even when no earlier chunk carried `finish_reason`. + openAi.hasTerminalMarker = true; + sawTerminator = true; + return null; + } + + try { + return JSON.parse(data) as Record; + } catch { + return null; + } + } + function parseAccumulatedSse(): StreamingPeekOutcome { const lines = decodedSoFar.split(/\r?\n/); // Retain the potentially-incomplete trailing fragment. decodedSoFar = lines[lines.length - 1]; for (const line of sseLineNormalizer.normalize(lines.slice(0, -1))) { - const trimmed = line.trim(); - - if (trimmed.startsWith("event:")) { - pendingEventType = trimmed.slice(6).trim(); - // An `event:` line is structured SSE framing on its own, even - // before any `data:` payload arrives (e.g. a bare keepalive ping). - sawStructuredSSE = true; - continue; - } - - if (!trimmed.startsWith("data:")) { - if (!trimmed) pendingEventType = ""; - continue; - } - - const data = trimmed.slice(5).trim(); - if (!data) continue; - if (data === "[DONE]") { - // #7285: `[DONE]` is itself a terminal marker for OpenAI-shape - // streams, even when no earlier chunk carried `finish_reason`. - openAi.hasTerminalMarker = true; - sawTerminator = true; - continue; - } - - let parsed: Record; - try { - parsed = JSON.parse(data); - } catch { - continue; - } + const parsed = consumeSseLine(line); + if (!parsed) continue; // A successfully parsed `data:` payload is structured SSE activity // regardless of shape or content — tracked only for the generic @@ -350,18 +370,7 @@ export async function validateResponseQuality( return "error"; } - // Some providers send a terminal `usage`-only chunk (no `choices`) - // as the final SSE frame instead of a `[DONE]`/`finish_reason` - // marker. Excludes Responses API `response.*` events, which have - // their own dedicated handling via `isKnownNonClaudeStreamPayload`. - if ( - parsed.usage && - typeof parsed.usage === "object" && - !Array.isArray(parsed.choices) && - !eventType.startsWith("response.") - ) { - sawTerminator = true; - } + if (isTerminalUsageOnlyChunk(parsed, eventType)) sawTerminator = true; if (isKnownNonClaudeStreamPayload(parsed, eventType)) { return "content"; From 3a041fde1a990b1062e29ee2e747cee2d606f2e3 Mon Sep 17 00:00:00 2001 From: Chewji9875 <126886556+Chewji9875@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:34:31 -0300 Subject: [PATCH 3/3] refactor(combo): move parseJsonRecord to module scope (finish complexity-gate compensation) Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --- open-sse/services/combo/validateQuality.ts | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/open-sse/services/combo/validateQuality.ts b/open-sse/services/combo/validateQuality.ts index 1e8d038054a2..27f8e029f6bc 100644 --- a/open-sse/services/combo/validateQuality.ts +++ b/open-sse/services/combo/validateQuality.ts @@ -212,6 +212,14 @@ type StreamingPeekOutcome = "content" | "error" | null; * 1. Body is valid JSON * 2. Has at least one choice with non-empty content or tool_calls */ +function parseJsonRecord(data: string): Record | null { + try { + return JSON.parse(data) as Record; + } catch { + return null; + } +} + export async function validateResponseQuality( response: Response, isStreaming: boolean, @@ -338,11 +346,7 @@ export async function validateResponseQuality( return null; } - try { - return JSON.parse(data) as Record; - } catch { - return null; - } + return parseJsonRecord(data); } function parseAccumulatedSse(): StreamingPeekOutcome {