From 0b96fe9dc93e274c300789e79b417a34bc02286e Mon Sep 17 00:00:00 2001 From: anhtahaylove <37265396+anhtahaylove@users.noreply.github.com> Date: Thu, 17 Sep 2026 12:31:36 +0700 Subject: [PATCH] fix(combo): retry pre-content streaming failures (#13630) * fix(combo): retry pre-content streaming failures * fix(combo): allow same-target retry for native-pinned pre-content failures A native Codex turn pin forced maxRetries to 0 unconditionally, which also disabled same-target retries. A pre-content stream failure sends no bytes to the client, so retrying the same pinned target is safe and indistinguishable from a first attempt. Set retries stay disabled so a pinned turn can never fail over to a different target. Adds a regression test covering a pinned turn whose first attempt fails before any content is streamed. * docs(changelog): add fragment for pre-content streaming retry fix Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --------- Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> --- ...ombo-retry-precontent-streaming-failure.md | 1 + open-sse/services/combo.ts | 5 +- .../services/combo/executeTargetAttempt.ts | 10 +- .../services/combo/executeTargetClassify.ts | 36 ++++++ ...mbo-responses-sse-failure-fallback.test.ts | 113 ++++++++++++++++++ 5 files changed, 158 insertions(+), 7 deletions(-) create mode 100644 changelog.d/fixes/combo-retry-precontent-streaming-failure.md diff --git a/changelog.d/fixes/combo-retry-precontent-streaming-failure.md b/changelog.d/fixes/combo-retry-precontent-streaming-failure.md new file mode 100644 index 0000000000..5d3afbcc71 --- /dev/null +++ b/changelog.d/fixes/combo-retry-precontent-streaming-failure.md @@ -0,0 +1 @@ +- **fix(combo):** retry the same target once when a streaming response fails before any content reaches the client (`streaming upstream error`), including native-pinned Codex turns whose set retries stay disabled — previously the caller returned a 502 immediately instead of using the existing transient-retry loop ([#13630](https://github.com/diegosouzapw/OmniRoute/pull/13630)) diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 8b8da3a0e2..bd5720e94b 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -807,7 +807,10 @@ async function handleComboChatInner({ }); } - const maxRetries = activeNativeTurnPin ? 0 : (config.maxRetries ?? 1); + // Native Codex turns must stay on their pinned target, but a pre-content + // stream failure is safe to retry because no output reached the client. + // Keep set retries disabled while preserving same-target retries. + const maxRetries = config.maxRetries ?? 1; const maxSetRetries = activeNativeTurnPin ? 0 : (config.maxSetRetries ?? 0); const setRetryDelayMs = resolveDelayMs(config.setRetryDelayMs, 2000); diff --git a/open-sse/services/combo/executeTargetAttempt.ts b/open-sse/services/combo/executeTargetAttempt.ts index 2572dc2c20..1b2614f042 100644 --- a/open-sse/services/combo/executeTargetAttempt.ts +++ b/open-sse/services/combo/executeTargetAttempt.ts @@ -87,6 +87,8 @@ import { markAccountExhaustedFromCredits } from "../../../src/domain/quotaCache. import { classifyComboOutcome, redactConnectionLabel } from "./comboErrorAggregation.ts"; import { readConnectionForCooldownGate } from "./executeTargetGates.ts"; import { + handlePreContentStreamRetry, + qualityValidationFailure, remainderIsHomogeneous, shouldAbortOnInputBoundFailure, shouldSurfaceBodySpecific400, @@ -454,12 +456,8 @@ export async function executeTargetAttempt(opts: { latencyMs: Date.now() - deps.startTime, }); state.observeFailure(false, target.executionKey); - return protectedPriorityTarget - ? { - ok: false, - response: errorResponse(502, "Upstream response failed quality validation"), - } - : null; + if (handlePreContentStreamRetry(quality, retry, deps, modelStr)) continue; + return protectedPriorityTarget ? qualityValidationFailure() : null; } if (Boolean(deps.clientManagedResponsesContext) && effectiveConnectionId) { diff --git a/open-sse/services/combo/executeTargetClassify.ts b/open-sse/services/combo/executeTargetClassify.ts index ffb60db2cc..8f3640a2d5 100644 --- a/open-sse/services/combo/executeTargetClassify.ts +++ b/open-sse/services/combo/executeTargetClassify.ts @@ -10,6 +10,7 @@ import { isModelScoped400, isParamValidation400, } from "./comboPredicates.ts"; +import { errorResponse } from "../../utils/error.ts"; export function remainderIsHomogeneous( orderedTargets: { modelStr: string }[], @@ -19,6 +20,41 @@ export function remainderIsHomogeneous( return orderedTargets.slice(index + 1).every((nextInPool) => nextInPool.modelStr === modelStr); } +/** + * Handle a pre-content streaming upstream error: nothing reached the client + * yet, so re-dispatching the same target cannot duplicate output. Logs and + * returns true when the caller should retry, false to fall through. + */ +export function handlePreContentStreamRetry( + quality: { reason?: string | null }, + retry: number, + deps: { + maxRetries: number; + signal?: { aborted?: boolean } | null; + log: { info: (tag: string, msg: string) => void }; + }, + modelStr: string +): boolean { + if ( + quality.reason !== "streaming upstream error" || + retry >= deps.maxRetries || + deps.signal?.aborted + ) { + return false; + } + deps.log.info( + "COMBO", + `Retrying ${modelStr} after pre-content streaming upstream error ` + + `(attempt ${retry + 2}/${deps.maxRetries + 1})` + ); + return true; +} + +/** Protected-priority target whose upstream body failed quality validation. */ +export function qualityValidationFailure(): { ok: false; response: Response } { + return { ok: false, response: errorResponse(502, "Upstream response failed quality validation") }; +} + export function shouldAbortOnInputBoundFailure(opts: { structuredError: unknown; remainderIsHomogeneous: boolean; diff --git a/tests/unit/combo-responses-sse-failure-fallback.test.ts b/tests/unit/combo-responses-sse-failure-fallback.test.ts index 222a0b358d..ae327108f9 100644 --- a/tests/unit/combo-responses-sse-failure-fallback.test.ts +++ b/tests/unit/combo-responses-sse-failure-fallback.test.ts @@ -2,6 +2,10 @@ import test from "node:test"; import assert from "node:assert/strict"; import { handleComboChat, validateResponseQuality } from "../../open-sse/services/combo.ts"; +import { + clearNativeCodexTurnPinsForTests, + pinNativeCodexTurn, +} from "../../open-sse/services/combo/nativeCodexTurnPin.ts"; const encoder = new TextEncoder(); @@ -99,6 +103,115 @@ test("combo advances to the next target after a pre-content Responses SSE failur assert.match(await result.text(), /fallback ok/); }); +test("protected pre-content streaming quality rejection retries the same target once without advancing", async () => { + const calls: string[] = []; + const failed = failedResponsesSse(); + const healthy = [ + "event: response.output_text.delta", + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "retry ok" })}`, + "", + "", + ].join("\n"); + const combo = { + name: "protected-stream-quality-retry", + strategy: "priority", + models: [ + { + model: "openai/primary", + weight: 0, + fallbackOnlyOnQuotaExhaustion: true, + }, + { model: "anthropic/backup", weight: 0 }, + ], + config: { maxRetries: 1, retryDelayMs: 0 }, + }; + + const result = await handleComboChat({ + body: { stream: true, messages: [{ role: "user", content: "hello" }] }, + combo, + handleSingleModel: async (_body: unknown, model: string) => { + calls.push(model); + return calls.length === 1 ? sseResponse(failed) : sseResponse(healthy); + }, + isModelAvailable: async () => true, + log: silentLog(), + settings: null, + allCombos: [combo], + relayOptions: null as never, + }); + + assert.equal(result.ok, true); + assert.deepEqual(calls, ["openai/primary", "openai/primary"]); + assert.match(await result.text(), /retry ok/); +}); + +test("native pinned pre-content stream failure still retries the same target safely", async () => { + clearNativeCodexTurnPinsForTests(); + const body = { + stream: true, + messages: [{ role: "user", content: "hello" }], + client_metadata: { + "x-codex-turn-metadata": JSON.stringify({ thread_id: "t1", turn_id: "turn1" }), + }, + }; + const combo = { + name: "native-pinned-stream-quality-retry", + strategy: "priority", + models: [ + { + model: "codex/gpt-5.6-sol", + connectionId: "conn-1", + fallbackOnlyOnQuotaExhaustion: true, + weight: 0, + }, + ], + config: { maxRetries: 1, retryDelayMs: 0 }, + }; + pinNativeCodexTurn({ + body, + comboName: combo.name, + target: { + kind: "model", + stepId: "codex-step", + executionKey: "codex-step", + modelStr: "codex/gpt-5.6-sol", + provider: "codex", + providerId: null, + connectionId: "conn-1", + weight: 0, + label: null, + }, + connectionId: "conn-1", + }); + + const calls: string[] = []; + const healthy = [ + "event: response.output_text.delta", + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "retry ok" })}`, + "", + "", + ].join("\n"); + const result = await handleComboChat({ + body, + combo, + clientManagedResponsesContext: true, + handleSingleModel: async (_body: unknown, model: string) => { + calls.push(model); + return calls.length === 1 ? sseResponse(failedResponsesSse()) : sseResponse(healthy); + }, + isModelAvailable: async () => true, + log: silentLog(), + settings: null, + allCombos: [combo], + relayOptions: null as never, + }); + + assert.equal(result.ok, true); + assert.deepEqual(calls, ["codex/gpt-5.6-sol", "codex/gpt-5.6-sol"]); + assert.match(await result.text(), /retry ok/); + clearNativeCodexTurnPinsForTests(); +}); + test("combo cancels a discarded upstream stream after a pre-content Responses SSE failure", async () => { let resolvePrimaryCancelled: (() => void) | undefined; const primaryCancelled = new Promise((resolve) => {