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>

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))

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);
 

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) {

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;

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<void>((resolve) => {