fix(combo): honor recovery for pre-content transport failures

changelog.d/fixes/0-combo-fallback-wait-cancellation.md

@@ -0,0 +1 @@
+- **fix(combo):** Priority and round-robin fallback waits now stop immediately when the client has already disconnected and release abort listeners when the wait completes.

changelog.d/fixes/18258-combo-authorized-cooldown-retry.md

@@ -0,0 +1 @@
+- **fix(combo):** Preserve an authorized same-target retry across a short transient connection cooldown, within existing wait limits and cancellation deadlines (#18257).

changelog.d/fixes/18610-combo-precontent-transport-recovery.md

@@ -0,0 +1 @@
+- Honor enabled stream recovery for structured transport read failures before usable combo output, with bounded retries and client cancellation cleanup. First-output inspection limits conservatively prevent a new transport retry.

open-sse/services/combo.ts

@@ -63,6 +63,7 @@ import type {
   SingleModelTarget,
   ComboLogger,
   HandleComboChatOptions,
+  HandleSingleModel,
   ResolvedComboTarget,
   AutoProviderCandidate,
   HistoricalLatencyStatsEntry,
@@ -838,8 +839,16 @@ async function handleComboChatInner({
   const traceInvocationId = invocationId ?? createInvocationId();
   startComboTrace(traceInvocationId, { strategy, comboName: combo.name });
 
+  const preContentTransportRecovery =
+    clientRequestedStream && resilienceSettings.streamRecovery.enabled === true;
+  const guardedHandleSingleModel: HandleSingleModel = preContentTransportRecovery
+    ? async (...dispatchArgs) =>
+        signal?.aborted
+          ? errorResponse(499, "Client disconnected")
+          : handleSingleModel(...dispatchArgs)
+    : handleSingleModel;
   const handleSingleModelWithTimeout = buildTargetTimeoutRunner({
-    handleSingleModel,
+    handleSingleModel: guardedHandleSingleModel,
     comboTargetTimeoutMs,
     resolveTargetTimeoutMs: (target) =>
       resolveTargetTimeoutMsForTarget(
@@ -869,6 +878,8 @@ async function handleComboChatInner({
       handleSingleModelWithTimeout,
       log,
       hiddenModelsByProvider,
+      preContentTransportRecovery: resilienceSettings.streamRecovery.enabled === true,
+      signal,
     });
     if (pinnedDispatch) return pinnedDispatch;
   }

open-sse/services/combo/attemptLoopTypes.ts

@@ -102,6 +102,12 @@ export type AttemptLoopDeps = {
   signal?: AbortSignal | null;
   body: Record<string, unknown>;
   startTime: number;
+  /** Shares the existing request wait limits with an inline, already-authorized retry. */
+  retryCooldownWait?: {
+    attempt: { current: number };
+    budgetLeftMs: { current: number };
+    deadline: number;
+  };
   releaseStickyPinOnFailure: (
     messageHash: string | null | undefined,
     failedConnectionId: string | null | undefined

open-sse/services/combo/authorizedRetryWait.ts

@@ -0,0 +1,120 @@
+import { waitForCooldownAwareRetry } from "../../../src/sse/services/cooldownAwareRetry.ts";
+import type { AttemptLoopDeps, AttemptLoopState } from "./attemptLoopTypes.ts";
+import { COMBO_COOLDOWN_WAIT_MARGIN_MS, shouldWaitForComboCooldown } from "./comboCooldownRetry.ts";
+import { COMBO_LOOP_SAFETY_TIMEOUT_MS, resolveDelayMs } from "./comboPredicates.ts";
+import { readConnectionForCooldownGate } from "./executeTargetGates.ts";
+import {
+  resolvePersistedRetryCooldown,
+  type PersistedRetryCooldown,
+} from "./persistedConnectionCooldown.ts";
+
+type RetryCooldownWait = NonNullable<AttemptLoopDeps["retryCooldownWait"]>;
+
+/** dispatch = send the retry; client_disconnected = 499; stop = leave this target (null). */
+type AuthorizedRetryOutcome = "dispatch" | "client_disconnected" | "stop";
+
+type AuthorizedRetryOptions = {
+  state: AttemptLoopState;
+  deps: AttemptLoopDeps;
+  target: { modelStr: string; connectionId?: string | null };
+  targetSignal?: AbortSignal;
+  retryDelayMs: number;
+  cooldownWait: RetryCooldownWait;
+  allowRateLimitedConnection: boolean;
+};
+
+/** The request-wide wait limits, or a per-target fallback when the loop did not pass them. */
+export function resolveRetryCooldownWait(deps: AttemptLoopDeps): RetryCooldownWait {
+  if (deps.retryCooldownWait) return deps.retryCooldownWait;
+  const comboTimeoutMs = resolveDelayMs(deps.config.comboTimeoutMs, 0);
+  return {
+    attempt: { current: 0 },
+    budgetLeftMs: { current: deps.resilienceSettings.comboCooldownWait?.budgetMs ?? 0 },
+    deadline: deps.startTime + (comboTimeoutMs || COMBO_LOOP_SAFETY_TIMEOUT_MS),
+  };
+}
+
+function outcomeAfterAbort(deps: AttemptLoopDeps): AuthorizedRetryOutcome {
+  return deps.signal?.aborted ? "client_disconnected" : "stop";
+}
+
+function decidePersistedCooldownWait(
+  deps: AttemptLoopDeps,
+  persisted: PersistedRetryCooldown,
+  cooldownWait: RetryCooldownWait
+) {
+  const settings = deps.resilienceSettings.comboCooldownWait;
+  if (!settings) return null;
+  const { cooldown } = persisted;
+  return shouldWaitForComboCooldown({
+    reason: cooldown?.reason,
+    waitMs: cooldown ? cooldown.untilMs - Date.now() + COMBO_COOLDOWN_WAIT_MARGIN_MS : 0,
+    attempt: cooldownWait.attempt.current,
+    budgetLeftMs: cooldownWait.budgetLeftMs.current,
+    settings,
+  });
+}
+
+/** A concurrent request can extend or reclassify the row while this retry waits. */
+async function waitOutPersistedCooldown(
+  opts: AuthorizedRetryOptions,
+  retrySignal: AbortSignal,
+  expire: () => void
+): Promise<AuthorizedRetryOutcome> {
+  const { state, deps, target, cooldownWait } = opts;
+  let persistedCooldownWaited = false;
+  for (;;) {
+    const persisted = await resolvePersistedRetryCooldown(
+      target,
+      (id) => readConnectionForCooldownGate(id, true),
+      persistedCooldownWaited,
+      opts.allowRateLimitedConnection
+    );
+    if (Date.now() >= cooldownWait.deadline) expire();
+    if (retrySignal.aborted || state.comboExpired) return outcomeAfterAbort(deps);
+    if (!persisted.skipReason) return "dispatch";
+    const decision = decidePersistedCooldownWait(deps, persisted, cooldownWait);
+    if (!decision?.wait) {
+      deps.log.info("COMBO", persisted.skipReason);
+      state.fallbackCount++;
+      return "stop";
+    }
+    persistedCooldownWaited = true;
+    cooldownWait.attempt.current++;
+    cooldownWait.budgetLeftMs.current -= decision.waitMs;
+    if (!(await waitForCooldownAwareRetry(decision.waitMs, retrySignal))) {
+      return outcomeAfterAbort(deps);
+    }
+  }
+}
+
+/**
+ * Sleep out an already-authorized retry delay, then wait out (within the shared cooldown
+ * budget, combo deadline and abort signals) a transient cooldown persisted meanwhile.
+ */
+export async function waitForAuthorizedRetry(
+  opts: AuthorizedRetryOptions
+): Promise<AuthorizedRetryOutcome> {
+  const { state, deps, targetSignal, cooldownWait } = opts;
+  const deadlineController = new AbortController();
+  const retrySignal = AbortSignal.any(
+    [deps.signal, targetSignal, deadlineController.signal].filter((signal): signal is AbortSignal =>
+      Boolean(signal)
+    )
+  );
+  const expire = () => {
+    state.comboExpired = true;
+    deadlineController.abort();
+  };
+  const timeLeftMs = cooldownWait.deadline - Date.now();
+  const deadlineTimer = setTimeout(expire, Math.max(0, timeLeftMs));
+  if (timeLeftMs <= 0) expire();
+  try {
+    if (!(await waitForCooldownAwareRetry(opts.retryDelayMs, retrySignal))) {
+      return outcomeAfterAbort(deps);
+    }
+    return await waitOutPersistedCooldown(opts, retrySignal, expire);
+  } finally {
+    clearTimeout(deadlineTimer);
+  }
+}

open-sse/services/combo/comboAttemptLoop.ts

@@ -376,7 +376,16 @@ export async function dispatchWithCooldownRetry(opts: {
         return extra.executeAttempt({
           index: i,
           state,
-          deps,
+          deps: {
+            ...deps,
+            retryCooldownWait: {
+              attempt: extra.comboCooldownAttempt,
+              budgetLeftMs: extra.comboCooldownBudgetLeftMs,
+              deadline:
+                extra.comboStartTime +
+                (extra.comboTimeoutMs > 0 ? extra.comboTimeoutMs : COMBO_LOOP_SAFETY_TIMEOUT_MS),
+            },
+          },
           targetForAttempt: gate.targetForAttempt,
           profile: gate.profile,
           protectedPriorityTarget: gate.protectedPriorityTarget,