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 0000000000..1f0c9852c8 --- /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 5c751c2bcc..27f8e029f6 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, @@ -267,9 +275,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 = ""; @@ -282,41 +304,67 @@ 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; + } + + return parseJsonRecord(data); + } + 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(); + const parsed = consumeSseLine(line); + if (!parsed) continue; - if (trimmed.startsWith("event:")) { - pendingEventType = trimmed.slice(6).trim(); - 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; - continue; - } - - let parsed: Record; - try { - parsed = JSON.parse(data); - } catch { - 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 || ""; @@ -326,6 +374,8 @@ export async function validateResponseQuality( return "error"; } + if (isTerminalUsageOnlyChunk(parsed, eventType)) sawTerminator = true; + if (isKnownNonClaudeStreamPayload(parsed, eventType)) { return "content"; } @@ -333,6 +383,7 @@ export async function validateResponseQuality( if (applySseLifecycleEvent(eventType, parsed, sse)) { return "content"; } + if (sse.hasLifecycleEnd) sawTerminator = true; } return null; } @@ -411,15 +462,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" }; } @@ -450,7 +509,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 c95df223f7..7ea04a15ba 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); +});