diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 375e47fb9b..d52dac30db 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -121,6 +121,7 @@ import { import { validateResponseQuality, releaseQualityClone, + releaseRejectedQualityResponse, toRetryAfterDisplayValue, } from "./combo/validateQuality.ts"; import { resolveComboCooldownWaitDecision } from "./combo/comboCooldownRetry.ts"; @@ -799,6 +800,7 @@ export async function handleComboChat({ ); releaseQualityClone(pinnedClone, pinnedResult, pinnedQuality); if (pinnedQuality.valid) return pinnedResult; + releaseRejectedQualityResponse(pinnedClone, pinnedResult); log.warn( "COMBO", `Pinned model ${pinnedModel} returned 200 but failed quality check: ${pinnedQuality.reason}, falling through to combo retry/fallback` @@ -1774,6 +1776,7 @@ export async function handleComboChat({ ); releaseQualityClone(qualityClone, result, quality); if (!quality.valid) { + releaseRejectedQualityResponse(qualityClone, result); log.warn( "COMBO", `Model ${modelStr} returned 200 but failed quality check: ${quality.reason}` @@ -2906,6 +2909,7 @@ async function handleRoundRobinCombo({ ); releaseQualityClone(rrClone, result, quality); if (!quality.valid) { + releaseRejectedQualityResponse(rrClone, result); log.warn( "COMBO-RR", `${modelStr} returned 200 but failed quality check: ${quality.reason}` diff --git a/open-sse/services/combo/validateQuality.ts b/open-sse/services/combo/validateQuality.ts index f5330ae773..5c751c2bcc 100644 --- a/open-sse/services/combo/validateQuality.ts +++ b/open-sse/services/combo/validateQuality.ts @@ -186,6 +186,21 @@ function responsesApiOutputHasContent(output: unknown): boolean { ); } +function isRecord(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +function isStreamingUpstreamError(parsed: unknown, eventType: string): boolean { + if (eventType === "response.failed" || eventType === "error") return true; + if (!isRecord(parsed)) return false; + if (parsed.error != null) return true; + + const nestedResponse = isRecord(parsed.response) ? parsed.response : null; + return nestedResponse?.status === "failed" && nestedResponse.error != null; +} + +type StreamingPeekOutcome = "content" | "error" | null; + /** * Validate that a successful (HTTP 200) non-streaming response actually contains * meaningful content. Returns { valid: true } or { valid: false, reason }. @@ -263,11 +278,11 @@ export async function validateResponseQuality( * flags in the closure. The last (potentially incomplete) line is kept in * `decodedSoFar` for the next iteration. * - * 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. + * Returns "content" once REAL content (not just an empty content_block_start) + * is detected, or "error" when the upstream reports a failure before content. + * Otherwise peeking continues. */ - function parseAccumulatedSse(): boolean { + function parseAccumulatedSse(): StreamingPeekOutcome { const lines = decodedSoFar.split(/\r?\n/); // Retain the potentially-incomplete trailing fragment. decodedSoFar = lines[lines.length - 1]; @@ -307,15 +322,19 @@ export async function validateResponseQuality( (typeof parsed.type === "string" ? parsed.type : null) || pendingEventType || ""; pendingEventType = ""; + if (isStreamingUpstreamError(parsed, eventType)) { + return "error"; + } + if (isKnownNonClaudeStreamPayload(parsed, eventType)) { - return true; + return "content"; } if (applySseLifecycleEvent(eventType, parsed, sse)) { - return true; + return "content"; } } - return false; + return null; } /** @@ -366,7 +385,15 @@ export async function validateResponseQuality( const tail = decoder.decode(undefined, { stream: false }); if (tail) decodedSoFar += tail; if (decodedSoFar.trim()) decodedSoFar += "\n\n"; - parseAccumulatedSse(); + const terminalOutcome = parseAccumulatedSse(); + + if (terminalOutcome === "error") { + log.warn?.( + "COMBO", + "Streaming response reported an upstream error before content — marking as invalid for combo failover" + ); + return { valid: false, reason: "streaming upstream error" }; + } if (sse.hasMessageStart && sse.hasLifecycleEnd && !sse.hasRealContent) { // Complete Claude lifecycle with zero content blocks, or with @@ -427,9 +454,20 @@ export async function validateResponseQuality( // Decode incrementally (stream:true keeps multi-byte char state). decodedSoFar += decoder.decode(value, { stream: true }); - const foundContent = parseAccumulatedSse(); + const outcome = parseAccumulatedSse(); - if (foundContent) { + if (outcome === "error") { + // Do not await cancellation of a Response.clone() tee branch: the + // promise may remain pending until the client-facing branch drains. + reader.cancel().catch(() => {}); + log.warn?.( + "COMBO", + "Streaming response reported an upstream error before content — marking as invalid for combo failover" + ); + return { valid: false, reason: "streaming upstream error" }; + } + + if (outcome === "content") { anyContentFound = true; // A content_block_* event was found — stop peeking. Return a // clonedResponse that replays all buffered bytes (the current chunk @@ -512,7 +550,9 @@ export async function validateResponseQuality( if (errorIsMeaningful) { const envelopeText = extractEnvelopeErrorText(json); const errMsg = - rawError && typeof rawError === "object" && typeof (rawError as Record).message === "string" + rawError && + typeof rawError === "object" && + typeof (rawError as Record).message === "string" ? ((rawError as Record).message as string) : envelopeText || JSON.stringify(rawError).substring(0, 200); return { valid: false, reason: `upstream error in 200 body: ${errMsg}` }; @@ -520,8 +560,7 @@ export async function validateResponseQuality( { const envelopeText = extractEnvelopeErrorText(json); if (envelopeText && EXHAUSTION_MARKER_PATTERN.test(envelopeText)) { - const snippet = - envelopeText.length > 80 ? `${envelopeText.slice(0, 80)}…` : envelopeText; + const snippet = envelopeText.length > 80 ? `${envelopeText.slice(0, 80)}…` : envelopeText; return { valid: false, reason: `upstream exhaustion marker in 200 body: ${snippet}` }; } } @@ -651,3 +690,19 @@ export function releaseQualityClone( if (clone === original) return; void quality.clonedResponse?.body?.cancel().catch(() => {}); } + +/** + * Cancel every response branch after a failed quality check when the caller is + * discarding the upstream response and falling back to another target. + * + * Streaming validation cancels its reader, but a reader on a `Response.clone()` + * tee cannot cancel the shared source until the untouched original branch is + * cancelled too. Best-effort cancellation of both branches also releases an + * unread quality clone for non-streaming failures. + */ +export function releaseRejectedQualityResponse(clone: Response, original: Response): void { + if (clone !== original) { + void clone.body?.cancel().catch(() => {}); + } + void original.body?.cancel().catch(() => {}); +} diff --git a/tests/unit/combo-responses-sse-failure-fallback.test.ts b/tests/unit/combo-responses-sse-failure-fallback.test.ts new file mode 100644 index 0000000000..222a0b358d --- /dev/null +++ b/tests/unit/combo-responses-sse-failure-fallback.test.ts @@ -0,0 +1,180 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { handleComboChat, validateResponseQuality } from "../../open-sse/services/combo.ts"; + +const encoder = new TextEncoder(); + +function sseResponse(body: string): Response { + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(body)); + controller.close(); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ); +} + +function silentLog() { + return { info() {}, warn() {}, error() {}, debug() {} }; +} + +function failedResponsesSse(): string { + return [ + "event: response.failed", + `data: ${JSON.stringify({ + type: "response.failed", + response: { + status: "failed", + error: { code: "no_capacity", message: "peak capacity" }, + }, + })}`, + "", + "", + ].join("\n"); +} + +test("streaming quality rejects a pre-content response.failed event", async () => { + const result = await validateResponseQuality( + sseResponse(failedResponsesSse()), + true, + silentLog() + ); + + assert.equal(result.valid, false); + assert.equal(result.reason, "streaming upstream error"); +}); + +test("streaming quality rejects a pre-content top-level error envelope", async () => { + const body = [ + "event: error", + `data: ${JSON.stringify({ + error: { type: "server_error", message: "temporarily unavailable" }, + })}`, + "", + "", + ].join("\n"); + + const result = await validateResponseQuality(sseResponse(body), true, silentLog()); + + assert.equal(result.valid, false); + assert.equal(result.reason, "streaming upstream error"); +}); + +test("combo advances to the next target after a pre-content Responses SSE failure", async () => { + const calls: string[] = []; + const healthy = [ + "event: response.output_text.delta", + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "fallback ok" })}`, + "", + "", + ].join("\n"); + + const result = await handleComboChat({ + body: { stream: true, messages: [{ role: "user", content: "hello" }] }, + combo: { + name: "responses-sse-failure-fallback", + strategy: "priority", + models: [ + { model: "openai/primary", weight: 0 }, + { model: "openai/secondary", weight: 0 }, + ], + config: { maxRetries: 0, retryDelayMs: 0 }, + }, + handleSingleModel: async (_body: unknown, model: string) => { + calls.push(model); + return model.endsWith("/primary") ? sseResponse(failedResponsesSse()) : sseResponse(healthy); + }, + isModelAvailable: async () => true, + log: silentLog(), + settings: null, + allCombos: null, + relayOptions: null as never, + }); + + assert.equal(result.ok, true); + assert.deepEqual(calls, ["openai/primary", "openai/secondary"]); + assert.match(await result.text(), /fallback ok/); +}); + +test("combo cancels a discarded upstream stream after a pre-content Responses SSE failure", async () => { + let resolvePrimaryCancelled: (() => void) | undefined; + const primaryCancelled = new Promise((resolve) => { + resolvePrimaryCancelled = resolve; + }); + const primary = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(failedResponsesSse())); + }, + cancel() { + resolvePrimaryCancelled?.(); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ); + const healthy = [ + "event: response.output_text.delta", + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "fallback ok" })}`, + "", + "", + ].join("\n"); + + const result = await handleComboChat({ + body: { stream: true, messages: [{ role: "user", content: "hello" }] }, + combo: { + name: "responses-sse-failure-cancellation", + strategy: "priority", + models: [ + { model: "openai/primary", weight: 0 }, + { model: "openai/secondary", weight: 0 }, + ], + config: { maxRetries: 0, retryDelayMs: 0 }, + }, + handleSingleModel: async (_body: unknown, model: string) => + model.endsWith("/primary") ? primary : sseResponse(healthy), + isModelAvailable: async () => true, + log: silentLog(), + settings: null, + allCombos: null, + relayOptions: null as never, + }); + + assert.equal(result.ok, true); + assert.match(await result.text(), /fallback ok/); + + let timeout: ReturnType | undefined; + try { + await Promise.race([ + primaryCancelled, + new Promise((_, reject) => { + timeout = setTimeout( + () => reject(new Error("discarded primary stream was not cancelled")), + 250 + ); + }), + ]); + } finally { + if (timeout) clearTimeout(timeout); + } +}); + +test("streaming quality still replays normal Responses lifecycle and content", async () => { + const body = [ + "event: response.created", + `data: ${JSON.stringify({ type: "response.created", response: { id: "resp_1" } })}`, + "", + "event: response.output_text.delta", + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "hello" })}`, + "", + "", + ].join("\n"); + + const result = await validateResponseQuality(sseResponse(body), true, silentLog()); + + assert.equal(result.valid, true); + assert.ok(result.clonedResponse); + assert.equal(await result.clonedResponse.text(), body); +});