fix(proxy): fail over dead account proxy routes before upstream dispatch

app/core/balancer/__init__.py

@@ -1,4 +1,5 @@
 from app.core.balancer.logic import (
+    ERROR_BACKOFF_THRESHOLD,
     HEALTH_TIER_DRAINING,
     HEALTH_TIER_HEALTHY,
     HEALTH_TIER_PROBING,
@@ -36,6 +37,7 @@
     "HEALTH_TIER_DRAINING",
     "HEALTH_TIER_HEALTHY",
     "HEALTH_TIER_PROBING",
+    "ERROR_BACKOFF_THRESHOLD",
     "REAUTH_REQUIRED_FAILURE_CODES",
     "AccountState",
     "RoutingCost",

app/core/balancer/logic.py

@@ -91,6 +91,7 @@
 DRAIN_SECONDARY_THRESHOLD_PCT = 90.0
 DRAIN_ERROR_WINDOW_SECONDS = 60.0
 DRAIN_ERROR_COUNT_THRESHOLD = 2
+ERROR_BACKOFF_THRESHOLD = 3
 PROBE_QUIET_SECONDS = 60.0
 PROBE_SUCCESS_STREAK_REQUIRED = 3
 ROUTING_POLICY_NORMAL = "normal"
@@ -481,8 +482,8 @@ def select_account(
             state.error_count = 0
         if state.cooldown_until and current < state.cooldown_until:
             continue
-        if state.error_count >= 3:
-            backoff = min(300, 30 * (2 ** (state.error_count - 3)))
+        if state.error_count >= ERROR_BACKOFF_THRESHOLD:
+            backoff = min(300, 30 * (2 ** (state.error_count - ERROR_BACKOFF_THRESHOLD)))
             if state.last_error_at and current - state.last_error_at < backoff:
                 in_error_backoff.append(state)
                 continue
@@ -519,7 +520,7 @@ def select_account(
         if allow_backoff_fallback and (len(in_error_backoff) > 1 or (in_error_backoff and hard_blocked_exists)):
 
             def _backoff_expires_at(s: AccountState) -> float:
-                backoff = min(300, 30 * (2 ** (s.error_count - 3)))
+                backoff = min(300, 30 * (2 ** (s.error_count - ERROR_BACKOFF_THRESHOLD)))
                 return (s.last_error_at or 0.0) + backoff
 
             available.append(min(in_error_backoff, key=_backoff_expires_at))

app/core/clients/codex.py

@@ -212,11 +212,20 @@ async def request_with_route_metadata(
                                 retryable_same_contract=False,
                             ) from None
                 return CodexRequestResult(response, candidate, index > 0)
-            except CodexTransportError:
-                if index == len(endpoints) - 1 or not allow_fallback:
+            except CodexTransportError as exc:
+                # A confirmed pre-dispatch connect failure proves the request
+                # never left for upstream, so trying the next endpoint in the
+                # same resolved pool is safe even for a non-idempotent POST.
+                # TLS verification failures are stable endpoint configuration
+                # errors rather than transient connect losses; they keep the
+                # idempotent-only rule.
+                if index == len(endpoints) - 1 or not (
+                    allow_fallback or (exc.retryable_same_contract and not exc.is_tls_verification_failure)
+                ):
                     raise
             except Exception as exc:
-                if index == len(endpoints) - 1 or not allow_fallback:
+                pre_dispatch = is_pre_dispatch_connection_failure(exc) and not isinstance(exc, aiohttp.ClientSSLError)
+                if index == len(endpoints) - 1 or not (allow_fallback or pre_dispatch):
                     raise _transport_error(
                         "request",
                         endpoint.id,

app/core/clients/proxy.py

@@ -492,6 +492,24 @@ def __init__(
         self.retry_after_seconds = retry_after_seconds
 
 
+def is_confirmed_pre_dispatch_transport_error(exc: ProxyResponseError) -> bool:
+    """Return whether the transport proved the upstream request never dispatched.
+
+    Only this provenance authorizes replaying a movable request on another
+    account: a typed connector failure while reaching the account's routed
+    proxy endpoint, before any request bytes could leave for upstream.
+    Host-wide network loss (``proxy_network_unavailable``) stays on its
+    account-neutral process recovery path instead of penalizing the selected
+    account, and ambiguous dispatch outcomes remain non-replayable.
+    """
+
+    if not (exc.retryable_same_contract and exc.failure_phase == "connect"):
+        return False
+    error = exc.payload.get("error")
+    error_code = error.get("code") if isinstance(error, dict) else None
+    return error_code != PROCESS_NETWORK_UNAVAILABLE_CODE
+
+
 def _process_network_failure_error(
     message: str,
     exc: Exception,

app/core/clients/proxy_websocket.py

@@ -802,9 +802,22 @@ async def _connect_upstream_websocket(
                 status_code if policy.preserve_handshake_status else 502,
                 openai_error(error_code, message, error_type="server_error"),
                 failure_phase="connect",
+                # Carry the client's dispatch provenance across the sanitizing
+                # boundary: a typed connector failure against the routed proxy
+                # proves no ``response.create`` frame could have reached
+                # upstream, so service-level failover may replay the request
+                # on another account. TLS verification failures are stable
+                # endpoint configuration errors and stay non-replayable.
                 retryable_same_contract=(
-                    policy.retry_routed_network_errors and error_code == PROCESS_NETWORK_UNAVAILABLE_CODE
+                    (policy.retry_routed_network_errors and error_code == PROCESS_NETWORK_UNAVAILABLE_CODE)
+                    or (exc.retryable_same_contract and not exc.is_tls_verification_failure)
                 ),
+                failure_detail=(
+                    "proxy_connect_pre_dispatch"
+                    if exc.retryable_same_contract and not exc.is_tls_verification_failure
+                    else "transport_error"
+                ),
+                failure_exception_type=type(exc).__name__,
             ) from exc
         except Exception:
             if owns_codex_client:

app/modules/proxy/_service/api_key_usage.py

@@ -14,6 +14,7 @@
 from app.core.exceptions import ProxyAuthError, ProxyRateLimitError
 from app.core.openai.models import CompactResponsePayload
 from app.core.utils.request_id import get_request_id
+from app.db.models import Account
 from app.modules.api_keys.service import (
     ApiKeyData,
     ApiKeyInvalidError,
@@ -64,6 +65,7 @@ class _ApiKeyUsageServiceProtocol(Protocol):
     _repo_factory: ProxyRepoFactory
     _background_cleanup_tasks: set[asyncio.Task[None]]
     _stream_api_key_release_retry_semaphore: asyncio.Semaphore
+    _load_balancer: Any
 
 
 def _normalize_service_tier_value(value: Any) -> str | None:
@@ -135,6 +137,30 @@ async def _release_websocket_request_state_reservation(
     ) -> None:
         self._cancel_request_state_api_key_reservation_heartbeat(request_state)
         await self._release_websocket_reservation(request_state.api_key_reservation)
+        request_state.api_key_reservation = None
+        lifecycle = request_state.deferred_account_backoff_lifecycle
+        if lifecycle is not None:
+            lifecycle.settlement_confirmed = True
+        pending_backoffs = (
+            lifecycle.pending_backoffs if lifecycle is not None else request_state.deferred_account_error_backoffs
+        )
+        if pending_backoffs:
+            await self._drain_deferred_account_error_backoffs(pending_backoffs)
+
+    async def _drain_deferred_account_error_backoffs(
+        self,
+        pending_backoffs: dict[str, Account],
+    ) -> None:
+        if not pending_backoffs:
+            return
+        proxy = cast(_ApiKeyUsageServiceProtocol, self)
+        while pending_backoffs:
+            account_id, account = pending_backoffs.popitem()
+            try:
+                await proxy._load_balancer.record_error_backoff(account)
+            except BaseException:
+                pending_backoffs.setdefault(account_id, account)
+                raise
 
     async def _maybe_touch_api_key_reservation(
         self,

app/modules/proxy/_service/http_bridge/helpers.py

@@ -757,6 +757,79 @@ def _http_bridge_session_has_visible_requests(session: "_HTTPBridgeSession") ->
     )
 
 
+async def _close_http_bridge_session(
+    service: Any,
+    session: "_HTTPBridgeSession",
+    *,
+    turn_state_lock_held: bool = False,
+    release_durable_session: bool = True,
+) -> None:
+    session.closed = True
+    if turn_state_lock_held:
+        service._unregister_http_bridge_turn_states_locked(session)
+        service._unregister_http_bridge_previous_response_ids_locked(session)
+    else:
+        await service._unregister_http_bridge_turn_states(session)
+        await service._unregister_http_bridge_previous_response_ids(session)
+    account_lease = getattr(session, "account_lease", None)
+    try:
+        await service._load_balancer.release_account_lease(account_lease)
+    except Exception:
+        logger.warning("Failed to release HTTP bridge account lease during close", exc_info=True)
+    finally:
+        session.account_lease = None
+    if release_durable_session and _http_bridge_durable_release_allowed(service, session):
+        try:
+            await service._durable_bridge.release_live_session(
+                session_id=session.durable_session_id,
+                instance_id=_service_get_settings().http_responses_session_bridge_instance_id,
+                owner_epoch=session.durable_owner_epoch,
+                draining=shutdown_state.is_bridge_drain_active(),
+            )
+        except Exception:
+            logger.warning("Failed to release durable HTTP bridge session", exc_info=True)
+    upstream_reader = session.upstream_reader
+    if upstream_reader is not None:
+        if upstream_reader is asyncio.current_task():
+            session.upstream_reader = None
+        else:
+            await _await_cancelled_task(
+                upstream_reader,
+                label="http bridge upstream reader",
+                cleanup_tasks=service._background_cleanup_tasks,
+            )
+            if session.upstream_reader is upstream_reader:
+                session.upstream_reader = None
+    try:
+        await session.upstream.close()
+    except Exception:
+        logger.debug("Failed to close HTTP bridge upstream websocket", exc_info=True)
+    pending_requests = getattr(session, "pending_requests", None)
+    pending_lock = getattr(session, "pending_lock", None)
+    response_create_gate = getattr(session, "response_create_gate", None)
+    if pending_requests is not None and pending_lock is not None:
+        async with pending_lock:
+            session.queued_request_count = 0
+        await service._fail_pending_websocket_requests(
+            account=session.account,
+            account_id_value=session.account.id,
+            pending_requests=pending_requests,
+            pending_lock=pending_lock,
+            error_code="stream_incomplete",
+            error_message="HTTP bridge session closed before response.completed",
+            api_key=None,
+            response_create_gate=response_create_gate,
+        )
+    _log_http_bridge_event(
+        "close",
+        session.key,
+        account_id=session.account.id,
+        model=session.request_model,
+        cache_key_family=session.key.affinity_kind,
+        model_class=_extract_model_class(session.request_model) if session.request_model else None,
+    )
+
+
 async def _close_http_bridge_session_bounded(
     service: Any,
     session: "_HTTPBridgeSession",