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

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();
-
-        if (trimmed.startsWith("event:")) {
-          pendingEventType = trimmed.slice(6).trim();
-          continue;
-        }
+        const parsed = consumeSseLine(line);
+        if (!parsed) 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,13 +374,16 @@ export async function validateResponseQuality(
           return "error";
         }
 
+        if (isTerminalUsageOnlyChunk(parsed, eventType)) sawTerminator = true;
+
         if (isKnownNonClaudeStreamPayload(parsed, eventType)) {
           return "content";
         }
 
         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);
+});