fix(sse): preserve error-only stream diagnostics (#9022)

* fix(sse): preserve error-only stream diagnostics

* test(ci): register stream readiness mutation coverage

* chore(changelog): finalize PR 9022 fragment

changelog.d/fixes/9022-stream-error-diagnostic.md

@@ -0,0 +1 @@
+- **fix(sse):** error-only streams now preserve sanitized executor diagnostics for operators without changing stream-readiness fallback classification ([#9022](https://github.com/diegosouzapw/OmniRoute/pull/9022)) — thanks @shixi-li

open-sse/handlers/chatCore.ts

@@ -4651,12 +4651,7 @@ export async function handleChatCore({
   });
   if (streamReadiness.ok === false) {
     const { response: failureResponse, reason } = streamReadiness;
-    const failure = {
-      status: failureResponse.status,
-      message: reason,
-      code: streamReadiness.code,
-      type: streamReadiness.type,
-    };
+    const { classificationReason, upstreamDiagnostic } = streamReadiness;
     trackPendingRequest(model, provider, connectionId, false);
     appendRequestLog({
       model,
@@ -4668,7 +4663,11 @@ export async function handleChatCore({
       status: failureResponse.status,
       error: reason,
       providerRequest: finalBody || translatedBody,
-      clientResponse: buildErrorBody(failureResponse.status, reason),
+      clientResponse: buildErrorBody(
+        failureResponse.status,
+        classificationReason,
+        upstreamDiagnostic ? { error: { message: upstreamDiagnostic } } : undefined
+      ),
       claudeCacheMeta: claudePromptCacheLogMeta,
       cacheSource: "upstream",
     });
@@ -4680,6 +4679,7 @@ export async function handleChatCore({
       success: false,
       status: failureResponse.status,
       error: reason,
+      classificationError: classificationReason,
       errorType: streamReadiness.type,
       errorCode: streamReadiness.code,
       response: failureResponse,

open-sse/utils/streamReadiness.ts

@@ -1,4 +1,5 @@
 import { HTTP_STATUS } from "../config/constants.ts";
+import { buildErrorBody, sanitizeErrorMessage } from "./error.ts";
 
 type StreamReadinessLogger = {
   debug?: (tag: string, message: string) => void;
@@ -7,7 +8,18 @@ type StreamReadinessLogger = {
 
 export type StreamReadinessResult =
   | { ok: true; response: Response }
-  | { ok: false; response: Response; reason: string; code: string; type: string };
+  | {
+      ok: false;
+      response: Response;
+      /** Sanitized operator-facing context for logs and persisted diagnostics. */
+      reason: string;
+      /** Stable internal text for retry, quota, and account-health classification. */
+      classificationReason: string;
+      /** First non-empty sanitized message from an error-only SSE payload. */
+      upstreamDiagnostic?: string;
+      code: string;
+      type: string;
+    };
 
 function isRecord(value: unknown): value is Record<string, unknown> {
   return !!value && typeof value === "object" && !Array.isArray(value);
@@ -233,6 +245,7 @@ type StreamReadinessSignalState = {
   currentEvent: string;
   dataLines: string[];
   pendingLine: string;
+  upstreamDiagnostic: string | null;
 };
 
 function resetCurrentEvent(state: StreamReadinessSignalState): void {
@@ -248,7 +261,23 @@ function processStreamReadinessEvent(state: StreamReadinessSignalState): boolean
   if (isPingEventType(eventType) || !data || data === "[DONE]") return false;
 
   try {
-    return hasNonPingStructuredPayload(JSON.parse(data), eventType);
+    const payload: unknown = JSON.parse(data);
+    if (
+      !state.upstreamDiagnostic &&
+      isRecord(payload) &&
+      isErrorOnlyStructuredPayload(payload)
+    ) {
+      const error = payload.error;
+      const rawMessage =
+        typeof error === "string"
+          ? error
+          : isRecord(error) && typeof error.message === "string"
+            ? error.message
+            : "";
+      const diagnostic = sanitizeErrorMessage(rawMessage).trim();
+      if (diagnostic) state.upstreamDiagnostic = diagnostic;
+    }
+    return hasNonPingStructuredPayload(payload, eventType);
   } catch {
     return data.length > 0;
   }
@@ -294,6 +323,7 @@ export function hasStreamReadinessSignal(text: string): boolean {
     currentEvent: "",
     dataLines: [],
     pendingLine: "",
+    upstreamDiagnostic: null,
   };
   if (appendStreamReadinessSignal(state, text)) return true;
   return finishStreamReadinessSignal(state);
@@ -303,16 +333,18 @@ function createErrorResponse(
   status: number,
   message: string,
   code: string,
-  type: string
+  type: string,
+  upstreamDiagnostic?: string
 ): Response {
   return new Response(
-    JSON.stringify({
-      error: {
+    JSON.stringify(
+      buildErrorBody(
+        status,
         message,
-        type,
-        code,
-      },
-    }),
+        upstreamDiagnostic ? { error: { message: upstreamDiagnostic } } : undefined,
+        { code, type }
+      )
+    ),
     { status, headers: { "Content-Type": "application/json" } }
   );
 }
@@ -385,6 +417,7 @@ export async function ensureStreamReadiness(
     currentEvent: "",
     dataLines: [],
     pendingLine: "",
+    upstreamDiagnostic: null,
   };
   const startedAt = Date.now();
   const effectiveTimeoutMs = Math.max(0, Math.floor(options.timeoutMs));
@@ -414,6 +447,7 @@ export async function ensureStreamReadiness(
         return {
           ok: false,
           reason,
+          classificationReason: reason,
           code: "STREAM_READINESS_TIMEOUT",
           type: "stream_timeout",
           response: createErrorResponse(
@@ -438,6 +472,7 @@ export async function ensureStreamReadiness(
         return {
           ok: false,
           reason,
+          classificationReason: reason,
           code: "STREAM_READINESS_TIMEOUT",
           type: "stream_timeout",
           response: createErrorResponse(
@@ -460,7 +495,11 @@ export async function ensureStreamReadiness(
           return { ok: true, response: buildReadyResponse() };
         }
 
-        const reason = "Stream ended before producing a non-ping SSE event";
+        const classificationReason = "Stream ended before producing a non-ping SSE event";
+        const upstreamDiagnostic = readinessState.upstreamDiagnostic || undefined;
+        const reason = upstreamDiagnostic
+          ? `${classificationReason}: ${upstreamDiagnostic}`
+          : classificationReason;
         options.log?.warn?.(
           "STREAM",
           `${reason} (${options.provider || "provider"}/${options.model || "unknown"})`
@@ -468,13 +507,16 @@ export async function ensureStreamReadiness(
         return {
           ok: false,
           reason,
+          classificationReason,
+          ...(upstreamDiagnostic ? { upstreamDiagnostic } : {}),
           code: "STREAM_EARLY_EOF",
           type: "stream_early_eof",
           response: createErrorResponse(
             HTTP_STATUS.BAD_GATEWAY,
-            reason,
+            classificationReason,
             "STREAM_EARLY_EOF",
-            "stream_early_eof"
+            "stream_early_eof",
+            upstreamDiagnostic
           ),
         };
       }

src/sse/handlers/chat.ts

@@ -77,6 +77,7 @@ import {
 import {
   isAntigravityMissingProjectError,
   PROVIDER_BREAKER_FAILURE_STATUSES,
+  resolveStreamReadinessClassificationError,
   shouldTripProviderBreakerForResult,
 } from "./chatPredicates";
 import { connectionHasExtraKeys } from "@omniroute/open-sse/services/apiKeyRotator.ts";
@@ -1491,10 +1492,9 @@ async function handleSingleModelChat(
         return result.response;
       }
 
-      // Missing Cloud Code project assignment is an account configuration error, not a
-      // transient upstream/account failure. Preserve the executor's typed fail-closed 422;
-      // marking the connection unavailable here would trigger cooldown redispatch and repeat
-      // bootstrap within the same logical request.
+      // Missing Cloud Code project assignment is configuration, not a transient failure.
+      // Preserve the typed fail-closed 422; marking it unavailable would trigger cooldown
+      // redispatch and repeat bootstrap within the same logical request.
       if (isAntigravityMissingProjectError(provider, result)) {
         return withSelectedConnectionHeader(result.response, credentials.connectionId);
       }
@@ -1537,10 +1537,11 @@ async function handleSingleModelChat(
       }
 
       if (isAntigravityStreamReadinessFailure) {
+        const classificationError = resolveStreamReadinessClassificationError(result);
         const { shouldFallback, cooldownMs } = await markAccountUnavailable(
           credentials.connectionId,
           result.status || HTTP_STATUS.BAD_GATEWAY,
-          result.error || result.errorCode || "Antigravity stream ended before useful content",
+          classificationError,
           provider,
           model,
           providerProfile,
@@ -1570,13 +1571,12 @@ async function handleSingleModelChat(
             }
           }
           excludedConnectionIds.add(credentials.connectionId);
-          lastError = result.error;
+          lastError = classificationError;
           lastStatus = result.status;
-          requestRetryLastError = result.error;
+          requestRetryLastError = classificationError;
           requestRetryLastStatus = result.status;
           continue;
         }
-
         return withSelectedConnectionHeader(result.response, credentials?.connectionId);
       }
 

src/sse/handlers/chatPredicates.ts

@@ -32,3 +32,22 @@ export function isAntigravityMissingProjectError(
     result.errorType === "oauth_missing_project_id"
   );
 }
+
+/**
+ * Keep stream-readiness routing decisions on the stable gate diagnostic.
+ * The operator-facing error can contain arbitrary upstream words such as
+ * "quota" or "retry after", which must not change account/combo classification.
+ */
+export function resolveStreamReadinessClassificationError(
+  result: {
+    classificationError?: unknown;
+    error?: unknown;
+    errorCode?: unknown;
+  },
+  fallback = "Antigravity stream ended before useful content"
+): string {
+  for (const value of [result.classificationError, result.error, result.errorCode]) {
+    if (typeof value === "string" && value.trim()) return value;
+  }
+  return fallback;
+}

stryker.conf.json

@@ -291,6 +291,7 @@
       "tests/unit/sse-auth-antigravity-credits.test.ts",
       "tests/unit/sse-auth-resource-404.test.ts",
       "tests/unit/sse-auth.test.ts",
+      "tests/unit/stream-readiness.test.ts",
       "tests/unit/strict-random-deck.test.ts",
       "tests/unit/strip-reasoning-header.test.ts",
       "tests/unit/system-role-extraction.test.ts",

tests/unit/stream-readiness.test.ts

@@ -6,6 +6,8 @@ import {
   hasStreamReadinessSignal,
   hasUsefulStreamContent,
 } from "../../open-sse/utils/streamReadiness.ts";
+import { checkFallbackError } from "../../open-sse/services/accountFallback.ts";
+import { resolveStreamReadinessClassificationError } from "../../src/sse/handlers/chatPredicates.ts";
 
 const encoder = new TextEncoder();
 
@@ -576,7 +578,93 @@ test("ensureStreamReadiness returns 502 when stream ends without a non-ping SSE
 
   const result = await ensureStreamReadiness(response, { timeoutMs: 100 });
   assert.equal(result.ok, false);
+  if (result.ok) assert.fail("keepalive-only SSE payload must remain a readiness failure");
   assert.equal(result.response.status, 502);
+  assert.equal(result.reason, "Stream ended before producing a non-ping SSE event");
+  assert.equal(result.classificationReason, result.reason);
+  const body = (await result.response.json()) as Record<string, unknown>;
+  assert.equal("upstream_details" in body, false);
+});
+
+test("ensureStreamReadiness preserves sanitized error-only diagnostics on early EOF (#8972)", async () => {
+  const warnings: string[] = [];
+  const response = new Response(
+    streamFromChunks([
+      `data: ${JSON.stringify({
+        error: {
+          message:
+            "UPSTREAM_DETAIL quota exhausted; retry after 2s; empty content " +
+            "Bearer TOP_SECRET /srv/omniroute/handler.ts:42",
+        },
+      })}\n\n`,
+      `data: ${JSON.stringify({ error: { message: "SECOND_DETAIL" } })}\n\n`,
+    ]),
+    { status: 200, headers: { "Content-Type": "text/event-stream" } }
+  );
+
+  const result = await ensureStreamReadiness(response, {
+    timeoutMs: 100,
+    provider: "test-provider",
+    model: "test-model",
+    log: {
+      warn: (_tag, message) => warnings.push(message),
+    },
+  });
+
+  assert.equal(result.ok, false);
+  if (result.ok) assert.fail("error-only SSE payload must remain a readiness failure");
+  assert.equal(result.response.status, 502);
+  assert.equal(result.code, "STREAM_EARLY_EOF");
+  assert.equal(result.type, "stream_early_eof");
+  assert.equal(
+    result.classificationReason,
+    "Stream ended before producing a non-ping SSE event"
+  );
+  assert.equal(
+    result.upstreamDiagnostic,
+    "UPSTREAM_DETAIL quota exhausted; retry after 2s; empty content Bearer [REDACTED] <path>"
+  );
+
+  const body = (await result.response.json()) as {
+    error: { message: string; code: string; type: string };
+    upstream_details: { error: { message: string } };
+  };
+  assert.equal(body.error.message, result.classificationReason);
+  assert.doesNotMatch(body.error.message, /quota|retry after|empty content/i);
+  assert.equal(body.error.code, "STREAM_EARLY_EOF");
+  assert.equal(body.error.type, "stream_early_eof");
+  assert.equal(body.upstream_details.error.message, result.upstreamDiagnostic);
+  assert.equal(warnings.length, 1);
+
+  for (const surfaced of [
+    result.reason,
+    body.upstream_details.error.message,
+    warnings[0],
+  ]) {
+    assert.match(surfaced, /UPSTREAM_DETAIL/);
+    assert.doesNotMatch(
+      surfaced,
+      /SECOND_DETAIL|TOP_SECRET|\/srv\/omniroute\/handler\.ts/
+    );
+  }
+});
+
+test("stream-readiness diagnostics cannot reclassify Antigravity account exhaustion (#8972)", () => {
+  const classificationError = "Stream ended before producing a non-ping SSE event";
+  const diagnostic = "UPSTREAM_DETAIL quota exhausted; retry after 2s; empty content";
+  const routedError = resolveStreamReadinessClassificationError({
+    classificationError,
+    error: `${classificationError}: ${diagnostic}`,
+    errorCode: "STREAM_EARLY_EOF",
+  });
+
+  assert.equal(routedError, classificationError);
+  assert.equal(checkFallbackError(502, routedError, 0, null, "antigravity").reason, "server_error");
+  assert.equal(
+    checkFallbackError(502, diagnostic, 0, null, "antigravity").reason,
+    "quota_exhausted",
+    "the regression fixture must prove that leaking the operator diagnostic changes routing"
+  );
 });
 
 test("ensureStreamReadiness accepts a final event without a trailing blank line", async () => {