fix(proxy): detect stalled upstream websockets

app/core/clients/proxy_websocket.py

@@ -78,6 +78,9 @@
     rf"|{_LIVE_CALL_UUID_CORE})"
 )
 _LIVE_CALL_ID_PATTERN = re.compile(rf"{REALTIME_LIVE_CALL_ID_ROUTE_REGEX}\Z")
+UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE = "upstream_websocket_liveness_timeout"
+_WEBSOCKETS_KEEPALIVE_TIMEOUT_REASON = "keepalive ping timeout"
+_AIOHTTP_HEARTBEAT_TIMEOUT_PREFIX = "No PONG received after "
 
 
 class RealtimeWebSocketProtocol(StrEnum):
@@ -99,16 +102,19 @@ class _UpstreamWebSocketPolicy:
     preserve_close_semantics: bool
 
 
+# Responses turns may be silent at the application layer for minutes, but a
+# healthy transport still answers ping control frames. Keep both watchdogs on:
+# disabling them turns a black-holed VPN route into a multi-hour request stall.
 _RESPONSES_WEBSOCKET_POLICY = _UpstreamWebSocketPolicy(
     operation="responses websocket",
     include_responses_beta=True,
     archive_payloads=True,
-    enable_routed_heartbeat=False,
+    enable_routed_heartbeat=True,
     retry_handshake_status=True,
     preserve_handshake_status=False,
     credential_safe_connect_errors=False,
     retry_routed_network_errors=True,
-    enable_direct_ping_timeout=False,
+    enable_direct_ping_timeout=True,
     preserve_close_semantics=False,
 )
 _LIVE_SIDEBAND_WEBSOCKET_POLICY = _UpstreamWebSocketPolicy(
@@ -176,24 +182,73 @@ def __init__(self, message: str, *, error_code: str) -> None:
 
 
 def _websocket_transport_error_code(exc: BaseException, *, uses_proxy: bool) -> str:
+    if _is_websocket_liveness_timeout(exc):
+        return UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE
     return process_network_error_code(
         exc,
         fallback="upstream_unavailable",
         include_permanent_dns=not uses_proxy,
     )
 
 
+def is_account_neutral_websocket_error_code(error_code: str | None) -> bool:
+    """Return whether transport provenance rules out an account-health penalty."""
+
+    # These failures occur below the selected account's application protocol.
+    # They follow an ambiguous send, so relay owners must fail rather than
+    # replay while leaving the account eligible for unrelated requests. Keep
+    # the compatibility keepalive code here as long as adapters can emit it.
+    return error_code in {
+        PROCESS_NETWORK_UNAVAILABLE_CODE,
+        UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE,
+        "upstream_keepalive_timeout",
+    }
+
+
+def _is_websocket_liveness_timeout(exc: BaseException) -> bool:
+    if isinstance(exc, ConnectionClosedError):
+        # websockets emits this locally-sent 1011 when its own ping watchdog
+        # expires. A peer may acknowledge it, leaving both close frames on the
+        # exception; send-first ordering still proves the marker came from our
+        # watchdog without trusting a peer that sends the same code and reason.
+        return (
+            exc.sent is not None
+            and int(exc.sent.code) == 1011
+            and exc.sent.reason == _WEBSOCKETS_KEEPALIVE_TIMEOUT_REASON
+            and (exc.rcvd is None or exc.rcvd_then_sent is False)
+        )
+    # aiohttp surfaces its heartbeat watchdog through WSMsgType.ERROR with a
+    # ServerTimeoutError carrying this library-defined prefix.
+    return isinstance(exc, aiohttp.ServerTimeoutError) and str(exc).startswith(_AIOHTTP_HEARTBEAT_TIMEOUT_PREFIX)
+
+
+def _aiohttp_stored_liveness_exception(websocket: Any) -> Exception | None:
+    # When aiohttp's heartbeat expires between receive() calls, no waiter is
+    # available for WSMsgType.ERROR. aiohttp stores the timeout instead and the
+    # next receive returns CLOSED, so every post-connect path must consult it.
+    exception_getter = getattr(websocket, "exception", None)
+    if not callable(exception_getter):
+        return None
+    exception = exception_getter()
+    return exception if isinstance(exception, Exception) and _is_websocket_liveness_timeout(exception) else None
+
+
 def _relay_receive_error_code(error_code: str) -> str | None:
-    """Expose only account-neutral process failures across the adapter boundary."""
+    """Expose account-neutral transport failures across the adapter boundary."""
 
     # Relay owners map an absent code to their established stream_incomplete
     # contract. Leaking the adapter's generic fallback would bypass that path.
-    return error_code if error_code in {PROCESS_NETWORK_UNAVAILABLE_CODE, "upstream_keepalive_timeout"} else None
+    return error_code if is_account_neutral_websocket_error_code(error_code) else None
 
 
 def _is_keepalive_timeout_close(exc: ConnectionClosedError) -> bool:
     """Classify peer/proxy heartbeat failures without exposing socket details."""
 
+    # Treat the legacy text marker as trusted only when this endpoint initiated
+    # the close. A peer can send the same public code and reason, so peer-first
+    # ordering must retain ordinary close/error semantics.
+    if exc.sent is None or (exc.rcvd is not None and exc.rcvd_then_sent is not False):
+        return False
     reason = _close_reason_from_exception(exc)
     return "keepalive ping timeout" in f"{exc} {reason or ''}".lower()
 
@@ -282,11 +337,12 @@ async def receive(self) -> UpstreamWebSocketMessage:
                 )
             error_code = _websocket_transport_error_code(exc, uses_proxy=self._uses_proxy)
             await _rotate_after_websocket_network_failure(error_code)
-            relay_error_code = (
-                "upstream_keepalive_timeout"
-                if _is_keepalive_timeout_close(exc)
-                else _relay_receive_error_code(error_code)
-            )
+            relay_error_code = _relay_receive_error_code(error_code)
+            if relay_error_code is None and _is_keepalive_timeout_close(exc):
+                # Prefer the stable, provenance-checked watchdog code above.
+                # This text fallback preserves compatibility with keepalive
+                # failures whose exception shape lacks the local-send marker.
+                relay_error_code = "upstream_keepalive_timeout"
             # ConnectionClosedError describes an incomplete close handshake,
             # not generic transport provenance. Let Responses relay owners map
             # it to stream_incomplete while live relays preserve received closes.
@@ -357,35 +413,52 @@ async def send_text(self, text: str) -> None:
             if asyncio.iscoroutine(result):
                 await result
         except Exception as exc:
-            await _raise_websocket_send_error(exc, endpoint_id=self._endpoint_id, uses_proxy=True)
+            classification_exc = _aiohttp_stored_liveness_exception(self._websocket) or exc
+            await _raise_websocket_send_error(classification_exc, endpoint_id=self._endpoint_id, uses_proxy=True)
 
     async def send_bytes(self, data: bytes) -> None:
         try:
             result = self._websocket.send_bytes(data)
             if asyncio.iscoroutine(result):
                 await result
         except Exception as exc:
-            await _raise_websocket_send_error(exc, endpoint_id=self._endpoint_id, uses_proxy=True)
+            classification_exc = _aiohttp_stored_liveness_exception(self._websocket) or exc
+            await _raise_websocket_send_error(classification_exc, endpoint_id=self._endpoint_id, uses_proxy=True)
 
     async def receive(self) -> UpstreamWebSocketMessage:
         try:
             msg = await self._websocket.receive()
         except Exception as exc:
-            error_code = _websocket_transport_error_code(exc, uses_proxy=True)
+            classification_exc = _aiohttp_stored_liveness_exception(self._websocket) or exc
+            error_code = _websocket_transport_error_code(classification_exc, uses_proxy=True)
             await _rotate_after_websocket_network_failure(error_code)
             return UpstreamWebSocketMessage(
                 kind="error",
-                error=codex_transport_error_message("websocket receive", self._endpoint_id, exc),
+                error=codex_transport_error_message("websocket receive", self._endpoint_id, classification_exc),
                 error_code=_relay_receive_error_code(error_code),
             )
         if msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSING, aiohttp.WSMsgType.CLOSED):
+            liveness_exception = _aiohttp_stored_liveness_exception(self._websocket)
+            if liveness_exception is not None:
+                return UpstreamWebSocketMessage(
+                    kind="error",
+                    close_code=_aiohttp_ws_close_code(self._websocket, msg),
+                    error=codex_transport_error_message(
+                        "websocket receive",
+                        self._endpoint_id,
+                        liveness_exception,
+                    ),
+                    error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE,
+                )
             return UpstreamWebSocketMessage(
                 kind="close",
                 close_code=_aiohttp_ws_close_code(self._websocket, msg),
                 close_reason=_aiohttp_ws_close_reason(msg),
             )
         if msg.type == aiohttp.WSMsgType.ERROR:
-            exception = msg.data if isinstance(msg.data, BaseException) else None
+            exception = (
+                msg.data if isinstance(msg.data, Exception) else _aiohttp_stored_liveness_exception(self._websocket)
+            )
             error_code = (
                 _websocket_transport_error_code(exception, uses_proxy=True)
                 if exception is not None
@@ -832,11 +905,8 @@ async def _connect_upstream_websocket(
         settings.upstream_websocket_proxy_env() if hasattr(settings, "upstream_websocket_proxy_env") else os.environ
     )
     proxy_url = resolve_websocket_proxy_from_env(url, proxy_env) if settings.upstream_websocket_trust_env else None
-    # Long Responses turns can spend minutes without application frames,
-    # so that existing transport keeps its own watchdog disabled. Live
-    # sideband traffic uses ping/pong liveness rather than an application-
-    # frame idle timeout because WebRTC media may remain healthy while the
-    # sideband itself is silent.
+    # Ping/pong control frames verify transport liveness without treating valid
+    # application-frame silence as an idle response.
     ping_timeout = (
         settings.proxy_downstream_websocket_idle_timeout_seconds if policy.enable_direct_ping_timeout else None
     )

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

@@ -1658,6 +1658,7 @@ async def _await_cancelled_task(
     cleanup_tasks: set[asyncio.Task[None]] | None = None,
 ) -> bool:
     caller_task = asyncio.current_task()
+    caller_cancelling_at_entry = caller_task.cancelling() if caller_task is not None else 0
     # Give a new child one scheduling turn before cancellation so
     # cancellation-resistant tasks enter the deferred-drain path.
     if not task.done():
@@ -1670,7 +1671,11 @@ async def _await_cancelled_task(
     try:
         await asyncio.wait_for(asyncio.shield(task), timeout=timeout_seconds)
     except asyncio.CancelledError:
-        if caller_task is not None and caller_task.cancelling():
+        # A caller may enter from its own finally block with cancellation
+        # already pending. The child's expected cancellation also surfaces as
+        # CancelledError; only a newly added caller cancellation should abort
+        # the rest of that finally block's transport and lease cleanup.
+        if caller_task is not None and caller_task.cancelling() > caller_cancelling_at_entry:
             _cancel_and_track_cancelled_task(task, label=label, cleanup_tasks=cleanup_tasks, cancel_task=False)
             raise
         return True

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

@@ -38,7 +38,11 @@
 from app.core.clients.proxy import codex_control_request as core_codex_control_request  # noqa: F401
 from app.core.clients.proxy import compact_responses as core_compact_responses  # noqa: F401
 from app.core.clients.proxy import transcribe_audio as core_transcribe_audio  # noqa: F401
-from app.core.clients.proxy_websocket import UpstreamWebSocketTransportError
+from app.core.clients.proxy_websocket import (
+    UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE,
+    UpstreamWebSocketTransportError,
+    is_account_neutral_websocket_error_code,
+)
 from app.core.errors import (
     openai_error,
 )
@@ -1145,14 +1149,23 @@ async def _submit_http_bridge_request_with_handoff(
                     upstream_send_started = True
                     try:
                         await _send_http_bridge_request_text_with_archive_id(session, request_state, text_data)
-                    except BaseException:
+                    except BaseException as exc:
                         request_state.recovery_attempt_dispatched = True
                         # Publish retirement while lifecycle ownership is still
                         # held; a gate waiter must never reuse an ambiguously sent
                         # response.create socket between unlock and cleanup.
                         session.closed = True
                         session.upstream_control.reconnect_requested = True
                         session.upstream_control.retire_after_drain = True
+                        if (
+                            isinstance(exc, UpstreamWebSocketTransportError)
+                            and exc.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE
+                        ):
+                            # Only this narrow claim, not ``closed``, tells the
+                            # reader that the submitter will settle siblings.
+                            # Keep it inside lifecycle_lock with the failing
+                            # send so the reader cannot observe an ownership gap.
+                            session.claim_liveness_settlement()
                         raise
                     request_state.recovery_attempt_dispatched = True
                     session.last_used_at = _service_time().monotonic()
@@ -1233,31 +1246,48 @@ async def _submit_http_bridge_request_with_handoff(
             # handed to the kernel. Never reconnect-and-resend from this path;
             # only failures proven to precede dispatch may be replayed.
             error_code = exc.error_code if isinstance(exc, UpstreamWebSocketTransportError) else "stream_incomplete"
-            account_neutral = error_code == "proxy_network_unavailable"
-            await self._cleanup_http_bridge_submit_interruption(
-                session,
-                request_state=request_state,
-                gate_acquired=gate_acquired,
-                request_enqueued=request_enqueued,
-                counted_in_queue=True,
-                admission_waiter_registered=admission_waiter_registered,
-            )
-            await self._fail_pending_websocket_requests(
-                account=session.account,
-                account_id_value=session.account.id,
-                pending_requests=deque([request_state]),
-                pending_lock=anyio.Lock(),
-                error_code=error_code,
-                error_message=str(exc) or "Upstream websocket closed before response.completed",
-                api_key=None,
-                response_create_gate=session.response_create_gate,
-                penalize_account=not account_neutral,
-            )
-            session.closed = True
-            try:
-                await session.upstream.close()
-            except Exception:
-                logger.debug("Failed to close HTTP bridge upstream websocket after send failure", exc_info=True)
+            # Liveness expiry and local network loss are transport failures,
+            # not evidence against the selected account. Keep this in sync
+            # with the reader path's shared provenance classification.
+            account_neutral = is_account_neutral_websocket_error_code(error_code)
+            if error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE:
+                # The sender claimed ownership beside the failing send while
+                # holding lifecycle_lock. It therefore owns the entire session
+                # deque, including older in-flight requests; settling only this
+                # request would strand its siblings after the reader yields.
+                async with session.lifecycle_lock:
+                    await self._fail_http_bridge_reader_and_maybe_retire(
+                        session,
+                        error_code=error_code,
+                        error_message=str(exc) or "Upstream websocket liveness failed",
+                        penalize_account=False,
+                        force_retire=True,
+                    )
+            else:
+                await self._cleanup_http_bridge_submit_interruption(
+                    session,
+                    request_state=request_state,
+                    gate_acquired=gate_acquired,
+                    request_enqueued=request_enqueued,
+                    counted_in_queue=True,
+                    admission_waiter_registered=admission_waiter_registered,
+                )
+                await self._fail_pending_websocket_requests(
+                    account=session.account,
+                    account_id_value=session.account.id,
+                    pending_requests=deque([request_state]),
+                    pending_lock=anyio.Lock(),
+                    error_code=error_code,
+                    error_message=str(exc) or "Upstream websocket closed before response.completed",
+                    api_key=None,
+                    response_create_gate=session.response_create_gate,
+                    penalize_account=not account_neutral,
+                )
+                session.closed = True
+                try:
+                    await session.upstream.close()
+                except Exception:
+                    logger.debug("Failed to close HTTP bridge upstream websocket after send failure", exc_info=True)
             # Always raise 502 so the client can retry with
             # previous_response_id intact.  Returning 400
             # previous_response_not_found causes the client to drop