From a8216c92feccfba2813acd6db3138e40c21f6609 Mon Sep 17 00:00:00 2001 From: Shixi Li <40780706+shixi-li@users.noreply.github.com> Date: Tue, 4 Aug 2026 21:06:25 +0800 Subject: [PATCH] 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 --- .../fixes/9022-stream-error-diagnostic.md | 1 + open-sse/handlers/chatCore.ts | 14 +-- open-sse/utils/streamReadiness.ts | 66 +++++++++++--- src/sse/handlers/chat.ts | 16 ++-- src/sse/handlers/chatPredicates.ts | 19 ++++ stryker.conf.json | 1 + tests/unit/stream-readiness.test.ts | 88 +++++++++++++++++++ 7 files changed, 178 insertions(+), 27 deletions(-) create mode 100644 changelog.d/fixes/9022-stream-error-diagnostic.md diff --git a/changelog.d/fixes/9022-stream-error-diagnostic.md b/changelog.d/fixes/9022-stream-error-diagnostic.md new file mode 100644 index 0000000000..041b27769a --- /dev/null +++ b/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 diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 941a5e9b45..7d2c31c111 100644 --- a/open-sse/handlers/chatCore.ts +++ b/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, diff --git a/open-sse/utils/streamReadiness.ts b/open-sse/utils/streamReadiness.ts index 4b76eafd65..23f57678e7 100644 --- a/open-sse/utils/streamReadiness.ts +++ b/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 { 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 ), }; } diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index 2d94cd141e..510314f455 100644 --- a/src/sse/handlers/chat.ts +++ b/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); } diff --git a/src/sse/handlers/chatPredicates.ts b/src/sse/handlers/chatPredicates.ts index fd14015c8d..6fee7bf078 100644 --- a/src/sse/handlers/chatPredicates.ts +++ b/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; +} diff --git a/stryker.conf.json b/stryker.conf.json index bf724daa64..831ee95a53 100644 --- a/stryker.conf.json +++ b/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", diff --git a/tests/unit/stream-readiness.test.ts b/tests/unit/stream-readiness.test.ts index 723aed3ee5..b2ea196818 100644 --- a/tests/unit/stream-readiness.test.ts +++ b/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; + 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] " + ); + + 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 () => {