fix(combo): failover when upstream SSE is truncated mid-lifecycle (#7545)

* fix(combo): failover when upstream SSE is truncated mid-lifecycle

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>

* refactor(combo): extract consumeSseLine + isTerminalUsageOnlyChunk helpers (complexity gate on parseAccumulatedSse)

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* refactor(combo): move parseJsonRecord to module scope (finish complexity-gate compensation)

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

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

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<string, unknown> | null {
+  try {
+    return JSON.parse(data) as Record<string, unknown>;
+  } 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<string, unknown>, 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<string, unknown> | 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<string, unknown>;
-        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 });

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<Uint8Array>({
+    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);
+});