fix(combo): detect empty content_block in streaming SSE peek (#7121)

* fix(combo): detect empty content_block in streaming SSE peek (port from 9router#1382)

The bounded SSE peek in validateResponseQuality() treated ANY content_block_start/delta/stop event as proof of real output and stopped buffering immediately, without checking whether the block actually carried text/tool_use content. Some upstreams (reported: DeepSeek, GLM via claude→openai translation) can open and close a text content_block with empty text and no tool_use on tool-heavy requests — the gateway logged success and forwarded a client-visible empty completion, and combo routing never failed over to the next model.

Track real content separately from 'a content_block_* event was seen': a tool_use/redacted_thinking block start is self-evidently real signal, a text/thinking block start is not (real content only confirmed via a subsequent delta carrying non-empty text/thinking, or an input_json_delta streaming tool arguments). A completed lifecycle (message_start + message_delta/stop) that never produced real content now fails validateResponseQuality(), matching the existing content_filter empty-stream detection path (#3685).

Reported-by: heishen6 (https://github.com/decolua/9router/issues/1382)

* refactor(combo): extract SSE lifecycle applier to keep the complexity ratchets at baseline

The #1382 empty-content_block peek added a branchy switch inline in
parseAccumulatedSse, pushing check:complexity to 2057 > baseline 2056.

Move the switch to a module-level applySseLifecycleEvent() and hold the
four lifecycle booleans in a single SseLifecycleFlags object threaded
through it, so the closure no longer copies flags in and out per event.
The per-event predicates (content_block_start / content_block_delta /
message_delta) are split into small guard helpers, which keeps the
applier flat — cognitive complexity punishes nesting, and an earlier
switch-only extraction traded the cyclomatic ratchet for a cognitive
regression at 891 > 890.

Logic is unchanged; both ratchets are now green (complexity 2055,
cognitive-complexity 890) and the #1382 regression tests still pass.

changelog.d/fixes/1382-streaming-empty-content-block.md

@@ -0,0 +1 @@
+- **fix(combo):** streaming Claude responses whose content block opens (`content_block_start`) and closes with no usable text/tool_use — a shape some upstreams return for tool-heavy requests on HTTP 200 — are now detected by `validateResponseQuality`'s SSE peek and trigger combo failover instead of being forwarded to the client as a silent empty completion (thanks @heishen6).

open-sse/services/combo/validateQuality.ts

@@ -54,6 +54,91 @@ function extractEnvelopeErrorText(json: Record<string, unknown>): string | null
   return parts.length > 0 ? parts.join(" ") : null;
 }
 
+/** Mutable lifecycle flags threaded through {@link applySseLifecycleEvent}. */
+interface SseLifecycleFlags {
+  hasMessageStart: boolean;
+  hasContentBlock: boolean;
+  hasRealContent: boolean;
+  hasLifecycleEnd: boolean;
+}
+
+/** Read `parsed.<key>` as a nested object bag, or null when absent/not an object. */
+function asObject(parsed: Record<string, unknown>, key: string): Record<string, unknown> | null {
+  const value = parsed[key];
+  return value && typeof value === "object" ? (value as Record<string, unknown>) : null;
+}
+
+/**
+ * A content_block_start is real signal only for tool_use / redacted_thinking —
+ * a tool call is meaningful even before its input_json_delta arrives. text and
+ * thinking blocks routinely open empty; keep peeking for a delta instead.
+ */
+function contentBlockStartIsRealSignal(parsed: Record<string, unknown>): boolean {
+  const blockType = asObject(parsed, "content_block")?.type;
+  return blockType === "tool_use" || blockType === "redacted_thinking";
+}
+
+/**
+ * A content_block_delta is real signal when it carries non-empty text/thinking,
+ * or any input_json_delta fragment — even an empty-string first chunk proves a
+ * tool_use block is actively streaming its arguments.
+ */
+function contentBlockDeltaIsRealSignal(parsed: Record<string, unknown>): boolean {
+  const delta = asObject(parsed, "delta");
+  if (!delta) return false;
+  const deltaType = typeof delta.type === "string" ? delta.type : "";
+  if (deltaType === "input_json_delta") return true;
+  if (deltaType !== "text_delta" && deltaType !== "thinking_delta") return false;
+  const text = delta.text ?? delta.thinking;
+  return typeof text === "string" && text.length > 0;
+}
+
+/** A message_delta closes the lifecycle once it carries a stop_reason. */
+function messageDeltaEndsLifecycle(parsed: Record<string, unknown>): boolean {
+  return asObject(parsed, "delta")?.stop_reason != null;
+}
+
+/**
+ * Apply a single parsed Claude SSE event to the peeked lifecycle `flags`
+ * (mutated in place). Extracted from `parseAccumulatedSse`'s inline switch to
+ * keep that function under the complexity/line ratchets — logic unchanged.
+ *
+ * Returns true once REAL content (not just an empty content_block_start) is
+ * detected — the caller should stop peeking and treat the stream as non-empty.
+ */
+function applySseLifecycleEvent(
+  eventType: string,
+  parsed: Record<string, unknown>,
+  flags: SseLifecycleFlags
+): boolean {
+  switch (eventType) {
+    case "message_start":
+      flags.hasMessageStart = true;
+      return false;
+    case "content_block_start":
+      flags.hasContentBlock = true;
+      if (!contentBlockStartIsRealSignal(parsed)) return false;
+      flags.hasRealContent = true;
+      return true;
+    case "content_block_delta":
+      flags.hasContentBlock = true;
+      if (!contentBlockDeltaIsRealSignal(parsed)) return false;
+      flags.hasRealContent = true;
+      return true;
+    case "content_block_stop":
+      flags.hasContentBlock = true;
+      return false;
+    case "message_stop":
+      flags.hasLifecycleEnd = true;
+      return false;
+    case "message_delta":
+      if (messageDeltaEndsLifecycle(parsed)) flags.hasLifecycleEnd = true;
+      return false;
+    default:
+      return false;
+  }
+}
+
 function responsesApiOutputHasContent(output: unknown): boolean {
   return (
     Array.isArray(output) &&
@@ -125,9 +210,22 @@ export async function validateResponseQuality(
     let decodedSoFar = "";
 
     // SSE lifecycle state.
-    let hasMessageStart = false;
-    let hasContentBlock = false;
-    let hasLifecycleEnd = false;
+    //
+    // #1382: hasContentBlock only means "a content_block_* event was observed"
+    // — it does NOT mean the block carried usable content. A content_block_start
+    // for a text/thinking block routinely opens with empty text (real content
+    // arrives via subsequent content_block_delta events); some upstreams
+    // (reported: DeepSeek/GLM via claude→openai translation on tool-heavy
+    // requests) open and close such a block without ever emitting a delta.
+    // hasRealContent tracks whether we've actually seen usable output: a
+    // tool_use/redacted_thinking block start (self-evidently real, even before
+    // any delta), or a delta carrying non-empty text/thinking/tool-input.
+    const sse: SseLifecycleFlags = {
+      hasMessageStart: false,
+      hasContentBlock: false,
+      hasRealContent: false,
+      hasLifecycleEnd: false,
+    };
     let anyContentFound = false;
     let sawAnyBytes = false;
     const sseLineNormalizer = createSSEDataLineNormalizer();
@@ -138,8 +236,9 @@ export async function validateResponseQuality(
      * flags in the closure. The last (potentially incomplete) line is kept in
      * `decodedSoFar` for the next iteration.
      *
-     * Returns true when a content_block_* event is detected — the caller
-     * should stop peeking and treat the stream as non-empty.
+     * Returns true once REAL content (not just an empty content_block_start)
+     * is detected — the caller should stop peeking and treat the stream as
+     * non-empty.
      */
     function parseAccumulatedSse(): boolean {
       const lines = decodedSoFar.split(/\r?\n/);
@@ -177,32 +276,8 @@ export async function validateResponseQuality(
           return true;
         }
 
-        switch (eventType) {
-          case "message_start":
-            hasMessageStart = true;
-            break;
-          case "content_block_start":
-          case "content_block_delta":
-          case "content_block_stop":
-            hasContentBlock = true;
-            // Signal caller to stop buffering immediately.
-            return true;
-          case "message_stop":
-            hasLifecycleEnd = true;
-            break;
-          case "message_delta": {
-            const delta = parsed.delta;
-            if (
-              delta &&
-              typeof delta === "object" &&
-              (delta as Record<string, unknown>).stop_reason != null
-            ) {
-              hasLifecycleEnd = true;
-            }
-            break;
-          }
-          default:
-            break;
+        if (applySseLifecycleEvent(eventType, parsed, sse)) {
+          return true;
         }
       }
       return false;
@@ -258,11 +333,17 @@ export async function validateResponseQuality(
           if (decodedSoFar.trim()) decodedSoFar += "\n\n";
           parseAccumulatedSse();
 
-          if (hasMessageStart && hasLifecycleEnd && !hasContentBlock) {
-            // Complete Claude lifecycle with zero content blocks → failover.
+          if (sse.hasMessageStart && sse.hasLifecycleEnd && !sse.hasRealContent) {
+            // Complete Claude lifecycle with zero content blocks, or with
+            // content_block_start/stop pairs that never carried real text/
+            // thinking/tool_use content (#1382 — tool-heavy claude→openai
+            // requests against upstreams like DeepSeek/GLM can "complete" a
+            // lifecycle around an empty block) → failover.
             log.warn?.(
               "COMBO",
-              "Streaming Claude response has complete lifecycle but zero content blocks (content_filter?) — marking as invalid for combo failover"
+              sse.hasContentBlock
+                ? "Streaming Claude response has complete lifecycle but its content block(s) carried no usable text/tool_use — marking as invalid for combo failover"
+                : "Streaming Claude response has complete lifecycle but zero content blocks (content_filter?) — marking as invalid for combo failover"
             );
             return { valid: false, reason: "streaming empty content block" };
           }
@@ -273,7 +354,7 @@ export async function validateResponseQuality(
           // (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 && !hasContentBlock && !sawAnyBytes) {
+          if (!anyContentFound && !sse.hasContentBlock && !sawAnyBytes) {
             log.warn?.(
               "COMBO",
               "Streaming response ended with no recognized content — marking as invalid for combo failover"

tests/unit/streaming-empty-content-block-1382.test.ts

@@ -0,0 +1,138 @@
+/**
+ * Issue #1382 (upstream decolua/9router) — a streaming Claude response that
+ * opens a `content_block_start` (type "text", initial text "") and then
+ * immediately `content_block_stop`s WITHOUT ever emitting a
+ * `content_block_delta` carrying real text/tool_use content must be treated
+ * as an empty/malformed response, not a valid completion.
+ *
+ * Before this fix, `validateResponseQuality`'s bounded SSE peek stopped
+ * buffering (and reported `valid: true`) as soon as ANY content_block_*
+ * event was observed — including a content_block_start whose block never
+ * carries usable text. Tool-heavy requests against backends that mishandle
+ * tool definitions (reported: DeepSeek, GLM via claude→openai translation)
+ * can emit exactly this shape: a lifecycle that "completes" successfully at
+ * the transport layer while the client receives no usable content. The
+ * combo loop never saw this as a failure, so no failover to the next model
+ * in the combo ever happened.
+ */
+import test from "node:test";
+import assert from "node:assert/strict";
+
+const { validateResponseQuality } = await import("../../open-sse/services/combo.ts");
+
+const encoder = new TextEncoder();
+const silentLog = { warn: () => {} };
+
+function claudeSseStream(events: string[]): ReadableStream<Uint8Array> {
+  const body = events.join("\n") + "\n";
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(encoder.encode(body));
+      controller.close();
+    },
+  });
+}
+
+/**
+ * Build a mock Claude 200 streaming response with a content_block_start/stop
+ * pair carrying EMPTY text and no tool_use block — the shape reported in
+ * #1382 for tool-heavy claude→openai requests against DeepSeek/GLM.
+ */
+function makeEmptyTextBlockStream(): Response {
+  const events = [
+    `event: message_start\ndata: ${JSON.stringify({
+      type: "message_start",
+      message: {
+        id: "msg_test_1382",
+        type: "message",
+        role: "assistant",
+        model: "deepseek-v4-pro-max",
+        content: [],
+        stop_reason: null,
+        stop_sequence: null,
+        usage: { input_tokens: 19882, output_tokens: 0 },
+      },
+    })}`,
+    "",
+    `event: content_block_start\ndata: ${JSON.stringify({
+      type: "content_block_start",
+      index: 0,
+      content_block: { type: "text", text: "" },
+    })}`,
+    "",
+    `event: content_block_stop\ndata: ${JSON.stringify({
+      type: "content_block_stop",
+      index: 0,
+    })}`,
+    "",
+    `event: message_delta\ndata: ${JSON.stringify({
+      type: "message_delta",
+      delta: { stop_reason: "end_turn", stop_sequence: null },
+      usage: { input_tokens: 0, output_tokens: 25 },
+    })}`,
+    "",
+    `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}`,
+    "",
+  ];
+
+  return new Response(claudeSseStream(events), {
+    status: 200,
+    headers: { "content-type": "text/event-stream" },
+  });
+}
+
+test("#1382 streaming Claude response with empty content_block (no text, no tool_use) is marked invalid", async () => {
+  const res = makeEmptyTextBlockStream();
+  const out = await validateResponseQuality(res, true, silentLog);
+  assert.equal(
+    out.valid,
+    false,
+    `expected invalid for empty content_block stream, got valid=true (reason: ${out.reason})`
+  );
+  assert.match(out.reason ?? "", /empty/i, `reason should mention 'empty', got: "${out.reason}"`);
+});
+
+test("#1382 streaming Claude response with a real tool_use content_block_start remains valid", async () => {
+  const events = [
+    `event: message_start\ndata: ${JSON.stringify({
+      type: "message_start",
+      message: {
+        id: "msg_test_1382_tool",
+        type: "message",
+        role: "assistant",
+        model: "deepseek-v4-pro-max",
+        content: [],
+        stop_reason: null,
+        stop_sequence: null,
+        usage: { input_tokens: 100, output_tokens: 0 },
+      },
+    })}`,
+    "",
+    `event: content_block_start\ndata: ${JSON.stringify({
+      type: "content_block_start",
+      index: 0,
+      content_block: { type: "tool_use", id: "toolu_1", name: "Bash", input: {} },
+    })}`,
+    "",
+    `event: content_block_stop\ndata: ${JSON.stringify({
+      type: "content_block_stop",
+      index: 0,
+    })}`,
+    "",
+    `event: message_delta\ndata: ${JSON.stringify({
+      type: "message_delta",
+      delta: { stop_reason: "tool_use", stop_sequence: null },
+      usage: { input_tokens: 0, output_tokens: 12 },
+    })}`,
+    "",
+    `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}`,
+    "",
+  ];
+  const res = new Response(claudeSseStream(events), {
+    status: 200,
+    headers: { "content-type": "text/event-stream" },
+  });
+  const out = await validateResponseQuality(res, true, silentLog);
+  assert.equal(out.valid, true, `expected valid for tool_use stream, got invalid: ${out.reason}`);
+  assert.ok(out.clonedResponse, "clonedResponse must be present for valid streaming response");
+});