From 94d03fb0773e8b1afd247858c9e558cdf37d9845 Mon Sep 17 00:00:00 2001 From: Roman Leventov Date: Tue, 4 Aug 2026 03:33:47 +0800 Subject: [PATCH 1/4] fix(proxy): detect stalled upstream websockets --- app/core/clients/proxy_websocket.py | 90 +++++- .../_service/http_bridge/request_submit.py | 73 +++-- .../_service/http_bridge/upstream_events.py | 56 +++- app/modules/proxy/_service/support.py | 11 + app/modules/proxy/_service/websocket/mixin.py | 115 ++++++-- .../.openspec.yaml | 2 + .../design.md | 56 ++++ .../proposal.md | 27 ++ .../specs/responses-api-compat/spec.md | 62 ++++ .../tasks.md | 18 ++ .../specs/responses-api-compat/context.md | 4 + openspec/specs/responses-api-compat/spec.md | 42 ++- tests/unit/test_proxy_http_bridge.py | 164 +++++++++++ tests/unit/test_proxy_utils.py | 268 +++++++++++++++++- tests/unit/test_proxy_websocket_client.py | 143 +++++++++- 15 files changed, 1035 insertions(+), 96 deletions(-) create mode 100644 openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/.openspec.yaml create mode 100644 openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md create mode 100644 openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/proposal.md create mode 100644 openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md create mode 100644 openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md diff --git a/app/core/clients/proxy_websocket.py b/app/core/clients/proxy_websocket.py index 06eb2d5fbb..2403eb01d6 100644 --- a/app/core/clients/proxy_websocket.py +++ b/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( @@ -158,6 +164,8 @@ 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", @@ -165,12 +173,52 @@ def _websocket_transport_error_code(exc: BaseException, *, uses_proxy: bool) -> ) +def is_account_neutral_websocket_error_code(error_code: str | None) -> bool: + """Return whether transport provenance rules out an account-health penalty.""" + + # Both 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. + return error_code in { + PROCESS_NETWORK_UNAVAILABLE_CODE, + UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + } + + +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 == PROCESS_NETWORK_UNAVAILABLE_CODE else None + return error_code if is_account_neutral_websocket_error_code(error_code) else None async def _rotate_after_websocket_network_failure(error_code: str) -> None: @@ -324,7 +372,8 @@ 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: @@ -332,27 +381,43 @@ async def send_bytes(self, data: bytes) -> 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 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 @@ -799,11 +864,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 ) diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index 0319e918ea..8f7242afa7 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -36,7 +36,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, ) @@ -962,31 +966,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 marked the session closed 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 diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index a70367af9b..99c4eaeb82 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -27,7 +27,12 @@ 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 UpstreamWebSocketMessage, UpstreamWebSocketTransportError +from app.core.clients.proxy_websocket import ( + UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + UpstreamWebSocketMessage, + UpstreamWebSocketTransportError, + is_account_neutral_websocket_error_code, +) from app.core.errors import response_failed_event from app.core.openai.parsing import parse_sse_event_payload from app.core.types import JsonValue @@ -705,18 +710,33 @@ async def _relay_http_bridge_upstream_messages( _archive_http_bridge_upstream_message(session, message, archive_request_state) session.last_upstream_close_code = message.close_code retried = False - # A process-network receive failure follows a successful send; + # Account-neutral transport failures follow a successful send; # replay is not safe merely because output is not visible. - if message.error_code != "proxy_network_unavailable": + account_neutral = is_account_neutral_websocket_error_code(message.error_code) + if not account_neutral: retried = await self._retry_http_bridge_precreated_request(session) if retried: continue async with session.lifecycle_lock: + if session.closed and message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE: + # A submitter holds this lock across send_text and marks + # the session closed before releasing it on ambiguous + # send failure. In that race the submitter owns terminal + # settlement; the reader must not settle the same state. + break await self._fail_http_bridge_reader_and_maybe_retire( session, error_code=message.error_code or "stream_incomplete", error_message=_upstream_websocket_disconnect_message(message), - penalize_account=message.error_code != "proxy_network_unavailable", + penalize_account=not account_neutral, + **( + # An admission waiter must not inherit a socket whose + # heartbeat already proved it dead. Other failures + # preserve the existing deferred-retirement handoff. + {"force_retire": True} + if message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + else {} + ), ) break except asyncio.CancelledError: @@ -729,18 +749,24 @@ async def _relay_http_bridge_upstream_messages( exc_info=True, ) error_code = exc.error_code if isinstance(exc, UpstreamWebSocketTransportError) else "stream_incomplete" - account_neutral = error_code == "proxy_network_unavailable" + account_neutral = is_account_neutral_websocket_error_code(error_code) async with session.lifecycle_lock: - await self._fail_http_bridge_reader_and_maybe_retire( - session, - error_code=error_code, - error_message=( - str(exc) - if isinstance(exc, UpstreamWebSocketTransportError) - else "HTTP bridge upstream reader crashed before response.completed" - ), - penalize_account=not account_neutral, - ) + if not (session.closed and error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE): + # Match the message path above when receive() raises while + # a concurrent send failure already owns settlement. + await self._fail_http_bridge_reader_and_maybe_retire( + session, + error_code=error_code, + error_message=( + str(exc) + if isinstance(exc, UpstreamWebSocketTransportError) + else "HTTP bridge upstream reader crashed before response.completed" + ), + penalize_account=not account_neutral, + # Preserve ordinary crash handoff behavior, but never hand + # a heartbeat-expired socket to an admission waiter. + **({"force_retire": True} if error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE else {}), + ) finally: await _cancel_http_bridge_reader_child( wakeup_task, diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index 141b2b5570..487f522ead 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -1060,6 +1060,17 @@ class _WebSocketUpstreamControl: downstream_sequence_request_state: _WebSocketRequestState | None = None downstream_sequence_number: int | None = None seen_tool_call_keys: dict[ToolCallDedupeKey, None] = field(default_factory=dict) + # One watchdog expiry can wake receive() while the concurrent send also + # fails. Claim synchronously, before either path awaits settlement: the + # winner owns the whole pending deque and the loser must not cancel it. + # A control object belongs to one upstream generation, so this never resets. + liveness_settlement_owner: Literal["send", "receive"] | None = None + liveness_settlement_done: asyncio.Event = field(default_factory=asyncio.Event) + + def claim_liveness_settlement(self, owner: Literal["send", "receive"]) -> bool: + if self.liveness_settlement_owner is None: + self.liveness_settlement_owner = owner + return self.liveness_settlement_owner == owner @dataclass(slots=True) diff --git a/app/modules/proxy/_service/websocket/mixin.py b/app/modules/proxy/_service/websocket/mixin.py index 63d5a5687d..66ebf5322f 100644 --- a/app/modules/proxy/_service/websocket/mixin.py +++ b/app/modules/proxy/_service/websocket/mixin.py @@ -56,9 +56,11 @@ 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 ( + UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, UpstreamWebSocket, UpstreamWebSocketTransportError, filter_inbound_websocket_headers, + is_account_neutral_websocket_error_code, ) from app.core.errors import ( OpenAIErrorEnvelope, @@ -1698,32 +1700,55 @@ async def retire_current_upstream() -> None: # send_str/send_bytes may fail after handing bytes to the # kernel. Delivery is uncertain, so replay could duplicate # a response.create even when no output is visible yet. - async with pending_lock: - sequenced_downstream_replay_refused = any( - state.last_downstream_sequence_number is not None for state in pending_requests - ) - await proxy._fail_pending_websocket_requests( - account=account, - account_id_value=account.id if account else None, - pending_requests=pending_requests, - pending_lock=pending_lock, - error_code=exc.error_code, - error_message=str(exc), - api_key=api_key, - websocket=websocket, - client_send_lock=client_send_lock, - response_create_gate=response_create_gate, - downstream_activity=downstream_activity, - penalize_account=exc.error_code != "proxy_network_unavailable", - suppress_sequenced_downstream_errors=sequenced_downstream_replay_refused, + liveness_timeout = exc.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + sender_owns_settlement = not liveness_timeout or ( + upstream_control is not None and upstream_control.claim_liveness_settlement("send") ) - if sequenced_downstream_replay_refused: - await _close_downstream_after_sequenced_replay_refusal( - websocket, - downstream_activity, - ) + if sender_owns_settlement: + try: + async with pending_lock: + sequenced_downstream_replay_refused = any( + state.last_downstream_sequence_number is not None for state in pending_requests + ) + await proxy._fail_pending_websocket_requests( + account=account, + account_id_value=account.id if account else None, + pending_requests=pending_requests, + pending_lock=pending_lock, + error_code=exc.error_code, + error_message=str(exc), + api_key=api_key, + websocket=websocket, + client_send_lock=client_send_lock, + response_create_gate=response_create_gate, + downstream_activity=downstream_activity, + penalize_account=not is_account_neutral_websocket_error_code(exc.error_code), + suppress_sequenced_downstream_errors=sequenced_downstream_replay_refused, + ) + if sequenced_downstream_replay_refused: + await _close_downstream_after_sequenced_replay_refusal( + websocket, + downstream_activity, + ) + finally: + if liveness_timeout and upstream_control is not None: + # Wake a receive path that observed the same + # watchdog failure only after terminal effects + # owned by this sender can no longer be cancelled. + upstream_control.liveness_settlement_done.set() + elif upstream_reader is not None: + # The reader already removed the states from the deque. + # Await its terminal writes, gate releases, and reservation + # settlement instead of cancelling it based on an empty deque. + await upstream_reader + upstream_reader = None + elif upstream_control is not None: + await upstream_control.liveness_settlement_done.wait() if upstream_reader is not None: - await _facade()._await_cancelled_task(upstream_reader, label="proxy websocket upstream reader") + await _facade()._await_cancelled_task( + upstream_reader, + label="proxy websocket upstream reader", + ) upstream_reader = None upstream_control = None if upstream is not None: @@ -3742,12 +3767,28 @@ async def _relay_upstream_websocket_messages( ) break continue + if ( + message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + and not upstream_control.claim_liveness_settlement("receive") + ): + # The sender owns every pending request. Wait until its + # terminal effects complete so this reader's finally block + # cannot close the downstream while those states remain. + await upstream_control.liveness_settlement_done.wait() + break replay_refusal_reasons: list[str] = [] replay_request_state = None - # A classified route/DNS receive failure happens after a - # completed send. Surface it account-neutrally rather than - # guessing that the upstream did not accept the request. - if message.error_code != "proxy_network_unavailable": + # Account-neutral transport failures happen after a completed + # send. Surface them rather than guessing that upstream did + # not accept the request. In particular, do not call the + # pre-created replay selector for a liveness timeout: the lost + # pong says nothing about whether response.create was accepted. + account_neutral = is_account_neutral_websocket_error_code(message.error_code) + if account_neutral: + async with pending_lock: + if any(state.last_downstream_sequence_number is not None for state in pending_requests): + replay_refusal_reasons.append("sequenced_downstream_frame") + else: replay_request_state = await _pop_replayable_precreated_websocket_request_state( pending_requests, pending_lock=pending_lock, @@ -3779,9 +3820,20 @@ async def _relay_upstream_websocket_messages( client_send_lock=client_send_lock, response_create_gate=response_create_gate, downstream_activity=downstream_activity, - penalize_account=message.error_code != "proxy_network_unavailable", + penalize_account=not account_neutral, suppress_sequenced_downstream_errors=sequenced_downstream_replay_refused, ) + if message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE: + # Release the routed-client/context owner immediately. The + # next downstream request must open a socket on the current + # host route instead of retaining this dead generation. + try: + await upstream.close() + except Exception: + _facade().logger.debug( + "Failed to retire upstream websocket after liveness timeout", + exc_info=True, + ) if sequenced_downstream_replay_refused: await _close_downstream_after_sequenced_replay_refusal( websocket, @@ -3867,6 +3919,11 @@ async def _relay_upstream_websocket_messages( downstream_activity=downstream_activity, ) finally: + if upstream_control.liveness_settlement_owner == "receive": + # A concurrent sender awaits the reader task itself, but the + # event also makes ownership completion explicit for callers + # that no longer retain that task handle. + upstream_control.liveness_settlement_done.set() async with pending_lock: has_pending_requests = bool(pending_requests) if not upstream_control.reconnect_requested and has_pending_requests: diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/.openspec.yaml b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/.openspec.yaml new file mode 100644 index 0000000000..e08b5f89a2 --- /dev/null +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-08-03 diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md new file mode 100644 index 0000000000..2568d454d4 --- /dev/null +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md @@ -0,0 +1,56 @@ +## Context + +Responses upstream WebSockets use two existing transports: `websockets` for direct egress and aiohttp for routed egress. Both libraries already support ping/pong liveness detection, and the same connection policy already enables it for Realtime live sideband traffic. The Responses policy disables both mechanisms because application-frame silence is valid during long turns. That distinction is unnecessary for ping/pong: control-frame replies prove transport liveness without requiring an application event. + +After a VPN disconnect, an established TCP connection can be black-holed without an immediate DNS, route, or socket exception. Downstream keepalives then keep the client attached while the upstream request remains pending. Once a request frame has been sent, however, the proxy cannot know whether upstream accepted it before connectivity was lost, so transparent replay could duplicate work or side effects. + +## Goals / Non-Goals + +**Goals:** + +- Bound silent upstream Responses WebSocket failure detection on both direct and routed egress. +- Preserve a stable classification from transport adapter through the direct WebSocket and HTTP bridge owners. +- Settle pending work once, without replaying an ambiguously delivered request or penalizing a healthy account. +- Ensure subsequent client retries open a fresh upstream connection and therefore use the current host route. + +**Non-Goals:** + +- Add a host route watcher, VPN-specific integration, background recovery coordinator, or proactive socket migration. +- Change the long Responses request budget or application-event idle timeout. +- Automatically resume a turn whose upstream acceptance is unknown. +- Add a configuration setting, dependency, persistence change, or operator-facing UI. + +## Decisions + +### Reuse transport ping/pong support and the existing timeout + +Enable `heartbeat` for routed aiohttp Responses sockets and `ping_timeout` for direct `websockets` Responses sockets. Both values come from `proxy_downstream_websocket_idle_timeout_seconds`, matching the existing live-sideband policy and keeping the fix zero-config. + +Application-level watchdogs were rejected because valid Responses turns may be silent for minutes and downstream synthetic keepalives are not evidence of upstream health. A host-network watcher was rejected because it is platform-specific, cannot reliably enumerate every network transition, and duplicates transport-layer failure detection. + +### Give liveness expiry a narrow stable classification + +Map only library-specific ping/pong timeout signals to `upstream_websocket_liveness_timeout`: the `websockets` locally sent close with reason `keepalive ping timeout`, and aiohttp's `ServerTimeoutError` produced by its heartbeat watchdog. Ordinary upstream closes and other receive exceptions keep their current behavior. + +A shared code predicate identifies account-neutral WebSocket failures so relay owners do not duplicate string comparisons as new neutral conditions are added. + +### Fail closed after ambiguous delivery + +Both relay owners treat the classified liveness timeout like a post-send network failure: no transparent replay, no account-health write, exact-once pending-request settlement, and retirement of the affected socket. The downstream error remains retryable at the client boundary, where a fresh client connection can safely establish a new upstream route under the client's existing retry semantics. + +For example, if `response.create` was written before a VPN route disappears and no pong returns, codex-lb emits a terminal `upstream_websocket_liveness_timeout` failure for that pending request and closes or retires the upstream session. It does not resend `response.create` on another account or socket. + +## Risks / Trade-offs + +- [Some intermediaries do not answer WebSocket pings correctly] → Use the established, configurable timeout already deployed for live sideband traffic; operators can adjust the existing value if needed. +- [Detection is not instantaneous] → A bounded delay is preferable to false positives and is far shorter than the multi-hour Responses request budget. +- [The client must retry the interrupted turn] → This avoids duplicate model work and tool side effects when delivery status is unknowable. +- [Library wording could change] → Pin direct detection to the concrete close code/reason emitted by the installed `websockets` API and cover both adapters with regression tests. + +## Migration Plan + +No data or configuration migration is required. Deploy normally; rollback restores the previous policy flags and classification behavior. + +## Open Questions + +None. diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/proposal.md b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/proposal.md new file mode 100644 index 0000000000..2d90716341 --- /dev/null +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/proposal.md @@ -0,0 +1,27 @@ +## Why + +An established upstream Responses WebSocket can remain silently black-holed after a host network transition such as VPN disconnection. Because Responses sockets currently disable the transports' existing ping/pong liveness checks, downstream keepalives can keep a conversation waiting until the much longer request timeout instead of terminating promptly so the client can reconnect. + +## What Changes + +- Enable the existing WebSocket transport liveness checks for direct and routed upstream Responses connections, using the current downstream WebSocket idle-timeout setting as the liveness budget. +- Classify transport-detected ping/pong timeouts with a stable internal error code. +- Treat a post-send liveness timeout as account-neutral and terminal for pending work, without transparent replay, because upstream request acceptance is ambiguous. +- Retire the affected upstream socket so a later client retry establishes a fresh network route. +- Add transport, direct WebSocket relay, and HTTP bridge regression coverage for the liveness and settlement invariants. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `responses-api-compat`: Require bounded upstream Responses WebSocket liveness detection and safe handling of liveness timeouts across direct WebSocket and HTTP bridge clients. + +## Impact + +- Affects the shared upstream WebSocket adapters and both Responses relay owners. +- Reuses existing aiohttp heartbeat, websockets ping timeout, and configuration; no new dependency, setting, API, schema, migration, or dashboard surface is introduced. +- A stalled conversation terminates after the configured liveness budget and relies on the downstream client to retry on a fresh connection. diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..a8eeda588d --- /dev/null +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md @@ -0,0 +1,62 @@ +## ADDED Requirements + +### Requirement: Responses upstream websocket liveness is bounded + +The proxy MUST configure direct and routed upstream Responses WebSocket transports with finite ping/pong liveness detection derived from `proxy_downstream_websocket_idle_timeout_seconds`. When an established Responses WebSocket is terminated because its transport did not receive the required pong, the adapter MUST classify the failure as `upstream_websocket_liveness_timeout`. Direct WebSocket and HTTP bridge relay owners MUST treat that failure as account neutral, MUST NOT transparently replay a pending request whose delivery is ambiguous, MUST finalize its pending request ownership exactly once, and MUST retire the affected upstream socket so a later client retry opens a fresh connection. + +#### Scenario: Direct Responses websocket loses pong liveness + +- **GIVEN** a direct upstream Responses WebSocket has been established +- **WHEN** the `websockets` keepalive watchdog terminates it after a pong timeout +- **THEN** the pending request fails with `upstream_websocket_liveness_timeout` +- **AND** the request is not transparently replayed +- **AND** the selected account receives no failure-health signal +- **AND** the affected upstream socket is retired + +#### Scenario: Routed Responses websocket loses pong liveness + +- **GIVEN** a routed upstream Responses WebSocket has been established for an HTTP bridge or direct WebSocket client +- **WHEN** the aiohttp heartbeat watchdog terminates it after a pong timeout +- **THEN** the pending request fails with `upstream_websocket_liveness_timeout` +- **AND** the request is not transparently replayed +- **AND** the selected account receives no failure-health signal +- **AND** the affected upstream socket is retired + +#### Scenario: Long turn remains healthy through control frames + +- **GIVEN** a Responses turn emits no application event within the liveness interval +- **WHEN** the upstream WebSocket continues replying to transport pings +- **THEN** the proxy keeps the upstream socket open +- **AND** the existing Responses request budget remains authoritative for the turn + +## MODIFIED Requirements + +### Requirement: Upstream websocket drops penalize affected accounts +When an upstream websocket closes while one or more streamed response requests are pending and have not reached a terminal event, the proxy MUST record a transient upstream error for the account before signaling failure for those pending requests, except when the close carries a classified process-wide network failure or upstream WebSocket liveness timeout. A classified process-wide network failure or upstream WebSocket liveness timeout MUST remain account neutral and use its classified error code. For other closes, the proxy MUST surface `stream_incomplete` to affected pending requests except when a direct Responses WebSocket request has already successfully emitted a finite integer `sequence_number`. For that sequenced direct-WebSocket case, the proxy MUST record the request outcome as `stream_incomplete` without emitting a synthetic terminal frame under the active response id, then MUST close the downstream WebSocket with code 1011. + +#### Scenario: websocket closes before pending responses complete + +- **GIVEN** a streamed response request is pending on an upstream websocket +- **AND** the direct downstream response has not emitted a numeric sequence, or the request uses another transport +- **WHEN** the websocket closes before a terminal response event is observed +- **AND** the close does not carry a classified process-wide network failure or upstream WebSocket liveness timeout +- **THEN** the pending request fails with `stream_incomplete` +- **AND** the account receives a transient upstream failure signal for routing + +#### Scenario: sequenced direct websocket closes before completion + +- **GIVEN** a direct Responses WebSocket request has successfully emitted a finite integer `sequence_number` +- **WHEN** the upstream websocket closes before a terminal response event is observed +- **AND** the close does not carry a classified process-wide network failure or upstream WebSocket liveness timeout +- **THEN** the request is recorded as failed with `stream_incomplete` +- **AND** no synthetic terminal frame is emitted under the active response id +- **AND** the downstream WebSocket closes with code 1011 +- **AND** the account receives a transient upstream failure signal for routing + +#### Scenario: websocket liveness timeout remains account neutral + +- **GIVEN** a streamed response request is pending on an upstream websocket +- **WHEN** its transport reports `upstream_websocket_liveness_timeout` +- **THEN** the pending request fails with that classified error code +- **AND** the account receives no failure-health signal +- **AND** the request is not transparently replayed diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md new file mode 100644 index 0000000000..ac3e5b368c --- /dev/null +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md @@ -0,0 +1,18 @@ +## 1. Transport liveness + +- [x] 1.1 Enable the existing finite heartbeat and ping timeout for routed and direct Responses WebSocket connections. +- [x] 1.2 Classify aiohttp heartbeat expiry and websockets keepalive expiry with the stable liveness-timeout code and a shared account-neutral predicate. + +## 2. Relay safety + +- [x] 2.1 Make direct Responses WebSocket liveness failures terminal, non-replayable, account-neutral, and fully settled. +- [x] 2.2 Apply the same no-replay, account-neutral, forced-retirement behavior to HTTP bridge upstream readers. + +## 3. Regression coverage + +- [x] 3.1 Cover direct and routed transport policy values and library-specific liveness classification. +- [x] 3.2 Cover direct WebSocket and HTTP bridge no-replay, account-health, settlement, and retirement invariants. + +## 4. Verification + +- [x] 4.1 Run focused tests, formatting, lint, type, architecture, and strict OpenSpec validation checks. diff --git a/openspec/specs/responses-api-compat/context.md b/openspec/specs/responses-api-compat/context.md index 3eb8c29881..a4e2d90f1c 100644 --- a/openspec/specs/responses-api-compat/context.md +++ b/openspec/specs/responses-api-compat/context.md @@ -32,6 +32,8 @@ See `openspec/specs/responses-api-compat/spec.md` for normative requirements. - `/v1/responses/compact` is supported only when the upstream implements it. - `prompt_cache_key` affinity on OpenAI-style routes is intentionally bounded by a dashboard-managed freshness window, unlike durable backend `session_id` or dashboard sticky-thread routing. - Codex-native direct websocket `/backend-api/codex/responses` treats upstream `previous_response_id` as an ephemeral anchor. If that anchor goes stale, the proxy must mask raw `previous_response_not_found` details and emit a sanitized `codex_previous_response_stale` classifier so compatible Codex clients can soft-reset and retry without `previous_response_id`. +- Upstream Responses WebSockets use transport ping/pong control frames to detect a black-holed connection without confusing valid application-event silence with an idle turn. Direct and routed connections reuse `proxy_downstream_websocket_idle_timeout_seconds` for this zero-config liveness budget. +- A post-send liveness timeout is delivery-ambiguous. It remains account-neutral, is never transparently replayed, and retires the affected upstream socket so a client retry opens a fresh route without risking duplicated model work or tool side effects. ## Fast Mode and Service Tiers @@ -114,6 +116,7 @@ when upstream reports a different actual tier. - **Codex websocket reconnects:** Reconnect continuity now depends on the client replaying the accepted `x-codex-turn-state`; generated turn-state is emitted on accept for backend Codex routes and echoed back when the client already supplies one. - **Codex websocket stale previous-response anchors:** Direct backend Codex websocket stale-anchor failures are surfaced as `response.failed` / `codex_previous_response_stale` without the raw upstream code or missing `resp_...` id; OpenAI-compatible `/v1/responses` websocket clients continue to receive generic `stream_incomplete` masking. - **Websocket handshake forbidden/not-found:** Auto transport now fails loud on `403` / `404` instead of silently hiding the websocket regression behind HTTP fallback. +- **Upstream websocket stops answering pings:** Pending direct-WebSocket and HTTP-bridge work fails with `upstream_websocket_liveness_timeout`; the account remains healthy and the request is not replayed because upstream acceptance is unknown. - **Invalid request payloads:** Return 4xx with `invalid_request_error`. ## Error Envelope Mapping (Reference) @@ -174,5 +177,6 @@ OpenSpec change first. - Post-deploy: monitor `capacity_exhausted_active_sessions`, Codex-session bridge reuse/evict counts, websocket handshake 403/404 rates after the narrower auto-fallback policy, and backend Codex HTTP vs websocket cache-ratio gaps. - When tracing compact incidents, confirm that request logs and upstream logs show direct `/codex/responses/compact` usage without surrogate `/codex/responses` fallback. - Post-deploy: monitor `no_accounts`, `stream_incomplete`, and `upstream_unavailable`. +- Post-deploy: monitor `upstream_websocket_liveness_timeout`; recurring failures indicate a host route, VPN, proxy, or intermediary that black-holes established WebSockets. - Post-deploy: monitor `codex_previous_response_stale` on `/backend-api/codex/responses`; recurring spikes mean clients are still relying on stale upstream anchors and should perform the documented full-context retry without `previous_response_id`. - Websocket/Codex CLI tier verification runbook: `openspec/specs/responses-api-compat/ops.md` diff --git a/openspec/specs/responses-api-compat/spec.md b/openspec/specs/responses-api-compat/spec.md index 3e871142ab..0911439bdf 100644 --- a/openspec/specs/responses-api-compat/spec.md +++ b/openspec/specs/responses-api-compat/spec.md @@ -83,15 +83,44 @@ The default compact request budget MUST be at least 180 seconds, and the default - **THEN** `compact_request_budget_seconds` is at least 180 seconds - **AND** `stream_idle_timeout_seconds` is at least 600 seconds +### Requirement: Responses upstream websocket liveness is bounded + +The proxy MUST configure direct and routed upstream Responses WebSocket transports with finite ping/pong liveness detection derived from `proxy_downstream_websocket_idle_timeout_seconds`. When an established Responses WebSocket is terminated because its transport did not receive the required pong, the adapter MUST classify the failure as `upstream_websocket_liveness_timeout`. Direct WebSocket and HTTP bridge relay owners MUST treat that failure as account neutral, MUST NOT transparently replay a pending request whose delivery is ambiguous, MUST finalize its pending request ownership exactly once, and MUST retire the affected upstream socket so a later client retry opens a fresh connection. + +#### Scenario: Direct Responses websocket loses pong liveness + +- **GIVEN** a direct upstream Responses WebSocket has been established +- **WHEN** the `websockets` keepalive watchdog terminates it after a pong timeout +- **THEN** the pending request fails with `upstream_websocket_liveness_timeout` +- **AND** the request is not transparently replayed +- **AND** the selected account receives no failure-health signal +- **AND** the affected upstream socket is retired + +#### Scenario: Routed Responses websocket loses pong liveness + +- **GIVEN** a routed upstream Responses WebSocket has been established for an HTTP bridge or direct WebSocket client +- **WHEN** the aiohttp heartbeat watchdog terminates it after a pong timeout +- **THEN** the pending request fails with `upstream_websocket_liveness_timeout` +- **AND** the request is not transparently replayed +- **AND** the selected account receives no failure-health signal +- **AND** the affected upstream socket is retired + +#### Scenario: Long turn remains healthy through control frames + +- **GIVEN** a Responses turn emits no application event within the liveness interval +- **WHEN** the upstream WebSocket continues replying to transport pings +- **THEN** the proxy keeps the upstream socket open +- **AND** the existing Responses request budget remains authoritative for the turn + ### Requirement: Upstream websocket drops penalize affected accounts -When an upstream websocket closes while one or more streamed response requests are pending and have not reached a terminal event, the proxy MUST record a transient upstream error for the account before signaling failure for those pending requests, except when the close carries a classified process-wide network failure. A classified process-wide network failure MUST remain account neutral and use its network error code. For other closes, the proxy MUST surface `stream_incomplete` to affected pending requests except when a direct Responses WebSocket request has already successfully emitted a finite integer `sequence_number`. For that sequenced direct-WebSocket case, the proxy MUST record the request outcome as `stream_incomplete` without emitting a synthetic terminal frame under the active response id, then MUST close the downstream WebSocket with code 1011. +When an upstream websocket closes while one or more streamed response requests are pending and have not reached a terminal event, the proxy MUST record a transient upstream error for the account before signaling failure for those pending requests, except when the close carries a classified process-wide network failure or upstream WebSocket liveness timeout. A classified process-wide network failure or upstream WebSocket liveness timeout MUST remain account neutral and use its classified error code. For other closes, the proxy MUST surface `stream_incomplete` to affected pending requests except when a direct Responses WebSocket request has already successfully emitted a finite integer `sequence_number`. For that sequenced direct-WebSocket case, the proxy MUST record the request outcome as `stream_incomplete` without emitting a synthetic terminal frame under the active response id, then MUST close the downstream WebSocket with code 1011. #### Scenario: websocket closes before pending responses complete - **GIVEN** a streamed response request is pending on an upstream websocket - **AND** the direct downstream response has not emitted a numeric sequence, or the request uses another transport - **WHEN** the websocket closes before a terminal response event is observed -- **AND** the close does not carry a classified process-wide network failure +- **AND** the close does not carry a classified process-wide network failure or upstream WebSocket liveness timeout - **THEN** the pending request fails with `stream_incomplete` - **AND** the account receives a transient upstream failure signal for routing @@ -99,11 +128,20 @@ When an upstream websocket closes while one or more streamed response requests a - **GIVEN** a direct Responses WebSocket request has successfully emitted a finite integer `sequence_number` - **WHEN** the upstream websocket closes before a terminal response event is observed +- **AND** the close does not carry a classified process-wide network failure or upstream WebSocket liveness timeout - **THEN** the request is recorded as failed with `stream_incomplete` - **AND** no synthetic terminal frame is emitted under the active response id - **AND** the downstream WebSocket closes with code 1011 - **AND** the account receives a transient upstream failure signal for routing +#### Scenario: websocket liveness timeout remains account neutral + +- **GIVEN** a streamed response request is pending on an upstream websocket +- **WHEN** its transport reports `upstream_websocket_liveness_timeout` +- **THEN** the pending request fails with that classified error code +- **AND** the account receives no failure-health signal +- **AND** the request is not transparently replayed + ### Requirement: Single HTTP bridge previous-response misses recover or fail closed When an HTTP bridge session receives an anonymous upstream `previous_response_not_found` error for a single pending follow-up request, the service MUST treat the error as an internal continuity-loss signal. It MUST either recover through the existing previous-response rebind path or rewrite the error to a retryable continuity failure instead of forwarding the raw upstream invalid-request error. diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 00dd232e8e..d974b10b4d 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -23,6 +23,7 @@ from app.core.auth.refresh import RefreshError from app.core.clients.proxy import CODEX_RESPONSES_LITE_WEBSOCKET_METADATA_KEY, ProxyResponseError from app.core.clients.proxy_websocket import ( + UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, CodexUpstreamWebSocket, UpstreamWebSocket, UpstreamWebSocketMessage, @@ -18633,6 +18634,169 @@ async def reconnect_during_failure(*_args: object, **_kwargs: object) -> bool: assert session.closed is False +@pytest.mark.asyncio +async def test_http_bridge_liveness_timeout_is_neutral_not_replayed_and_forces_retirement( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + request_state = proxy_service._WebSocketRequestState( + request_id="req-bridge-liveness-timeout", + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + awaiting_response_created=True, + request_text='{"type":"response.create","model":"gpt-5.4","input":"hello"}', + transport="http", + ) + session = _make_bridge_session( + key_value="bridge-liveness-timeout", + pending_requests=deque([request_state]), + queued_request_count=1, + ) + session.admission_waiter_count = 1 + session.upstream = cast( + UpstreamWebSocket, + SimpleNamespace( + receive=AsyncMock( + return_value=UpstreamWebSocketMessage( + kind="error", + error="Upstream websocket liveness failed", + error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + ) + ), + close=AsyncMock(), + ), + ) + monkeypatch.setattr(proxy_service, "get_settings", lambda: _make_app_settings()) + retry_precreated = AsyncMock(return_value=True) + fail_pending = AsyncMock() + retire = AsyncMock() + monkeypatch.setattr(service, "_retry_http_bridge_precreated_request", retry_precreated) + monkeypatch.setattr(service, "_fail_pending_websocket_requests", fail_pending) + monkeypatch.setattr(service, "_retire_stale_pending_http_bridge_session", retire) + + await service._relay_http_bridge_upstream_messages(session) + + retry_precreated.assert_not_awaited() + fail_pending.assert_awaited_once() + fail_pending_args = fail_pending.await_args + assert fail_pending_args is not None + assert fail_pending_args.kwargs["error_code"] == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + assert fail_pending_args.kwargs["penalize_account"] is False + retire.assert_awaited_once_with(session, detail=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE) + assert session.queued_request_count == 0 + assert session.closed is True + + +@pytest.mark.asyncio +async def test_http_bridge_liveness_send_receive_race_settles_request_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class RacingLivenessUpstream: + def __init__(self) -> None: + self.send_started = asyncio.Event() + self.receive_returned = asyncio.Event() + self.close = AsyncMock() + + async def send_text(self, _text: str) -> None: + self.send_started.set() + await self.receive_returned.wait() + # Let the reader queue on lifecycle_lock before the submitter + # publishes its send-side failure ownership and releases the lock. + await asyncio.sleep(0) + raise UpstreamWebSocketTransportError( + "Codex upstream websocket send failed: heartbeat expired", + error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + ) + + async def receive(self) -> UpstreamWebSocketMessage: + await self.send_started.wait() + self.receive_returned.set() + return UpstreamWebSocketMessage( + kind="error", + error="Codex upstream websocket receive failed: heartbeat expired", + error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + ) + + service = proxy_service.ProxyService(cast(Any, nullcontext())) + sibling_queue: asyncio.Queue[str | None] = asyncio.Queue() + sibling_state = proxy_service._WebSocketRequestState( + request_id="req-bridge-liveness-race-sibling", + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + response_id="resp-bridge-liveness-race-sibling", + event_queue=sibling_queue, + transport="http", + skip_request_log=True, + ) + request_queue: asyncio.Queue[str | None] = asyncio.Queue() + request_state = proxy_service._WebSocketRequestState( + request_id="req-bridge-liveness-race", + model="gpt-5.4", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + awaiting_response_created=True, + event_queue=request_queue, + request_text='{"type":"response.create","model":"gpt-5.4","input":"hello"}', + transport="http", + skip_request_log=True, + ) + upstream = RacingLivenessUpstream() + session = _make_bridge_session( + key_value="bridge-liveness-race", + pending_requests=deque([sibling_state]), + queued_request_count=1, + ) + session.upstream = cast(UpstreamWebSocket, upstream) + service._http_bridge_sessions[session.key] = session + fail_pending = AsyncMock(wraps=service._fail_pending_websocket_requests) + retire = AsyncMock() + retry_precreated = AsyncMock(return_value=True) + monkeypatch.setattr(proxy_service, "get_settings", lambda: _make_app_settings()) + monkeypatch.setattr(service, "_fail_pending_websocket_requests", fail_pending) + monkeypatch.setattr(service, "_retire_stale_pending_http_bridge_session", retire) + monkeypatch.setattr(service, "_retry_http_bridge_precreated_request", retry_precreated) + + reader_task = asyncio.create_task(service._relay_http_bridge_upstream_messages(session)) + try: + with pytest.raises(ProxyResponseError) as exc_info: + await service._submit_http_bridge_request( + session, + request_state=request_state, + text_data=request_state.request_text or "{}", + queue_limit=8, + ) + await asyncio.wait_for(reader_task, timeout=1.0) + finally: + if not reader_task.done(): + reader_task.cancel() + await asyncio.gather(reader_task, return_exceptions=True) + + assert exc_info.value.payload["error"]["code"] == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + fail_pending.assert_awaited_once() + failure_call = fail_pending.await_args + assert failure_call is not None + assert failure_call.kwargs["penalize_account"] is False + retry_precreated.assert_not_awaited() + for event_queue in (sibling_queue, request_queue): + terminal_event = await asyncio.wait_for(event_queue.get(), timeout=0.1) + assert terminal_event is not None + assert UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE in terminal_event + assert await asyncio.wait_for(event_queue.get(), timeout=0.1) is None + assert request_state.replay_count == 0 + assert session.pending_requests == deque() + assert session.queued_request_count == 0 + assert session.closed is True + retire.assert_awaited_once_with(session, detail=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE) + + @pytest.mark.asyncio async def test_http_bridge_retry_send_network_failure_is_neutral_and_not_replayed( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py index f5d77f1a88..1bd4215e48 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -36,6 +36,7 @@ from app.core.balancer.types import UpstreamError from app.core.clients.proxy import _build_upstream_headers, filter_inbound_headers from app.core.clients.proxy_websocket import ( + UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, CodexUpstreamWebSocket, UpstreamWebSocket, UpstreamWebSocketTransportError, @@ -24745,10 +24746,172 @@ async def connect(*_args: object, **_kwargs: object): "response.created", "response.failed", ] - assert downstream.close_calls == [(1011, "upstream replay requires a fresh request")] + assert downstream.close_calls[0] == (1011, "upstream replay requires a fresh request") handle_stream_error.assert_not_awaited() +@pytest.mark.asyncio +async def test_proxy_responses_websocket_liveness_race_awaits_reader_settlement(monkeypatch): + request_logs = _RequestLogsRecorder() + service = proxy_service.ProxyService(_repo_factory(request_logs)) + settings = _make_proxy_settings() + settings.stream_idle_timeout_seconds = 300.0 + settings.proxy_downstream_websocket_idle_timeout_seconds = 120.0 + monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache(settings)) + monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + handle_stream_error = AsyncMock() + monkeypatch.setattr(proxy_service.ProxyService, "_handle_stream_error", handle_stream_error) + + request_texts = [ + json.dumps( + { + "type": "response.create", + "model": "gpt-5.4", + "instructions": "", + "input": [{"role": "user", "content": label}], + "stream": True, + }, + separators=(",", ":"), + ) + for label in ("first", "second") + ] + + class RacingDownstreamWebSocket: + def __init__(self) -> None: + self.request_index = 0 + self.first_created = asyncio.Event() + self.done = asyncio.Event() + self.sent_text: list[str] = [] + + async def receive(self) -> dict[str, object]: + if self.request_index == 0: + self.request_index = 1 + return {"type": "websocket.receive", "text": request_texts[0]} + if self.request_index == 1: + await self.first_created.wait() + self.request_index = 2 + return {"type": "websocket.receive", "text": request_texts[1]} + await self.done.wait() + return {"type": "websocket.disconnect"} + + async def send_text(self, text: str) -> None: + self.sent_text.append(text) + payload = json.loads(text) + if payload.get("type") == "response.created": + self.first_created.set() + if sum(json.loads(item).get("type") == "response.failed" for item in self.sent_text) == 2: + self.done.set() + + async def send_bytes(self, _data: bytes) -> None: + return None + + async def close(self, code: int = 1000, reason: str | None = None) -> None: + del code, reason + self.done.set() + + settlement_started = asyncio.Event() + allow_settlement = asyncio.Event() + + class RacingUpstreamWebSocket: + def __init__(self) -> None: + self.send_count = 0 + self.receive_count = 0 + self.second_send_started = asyncio.Event() + self.closed = False + + async def send_text(self, _text: str) -> None: + self.send_count += 1 + if self.send_count == 2: + self.second_send_started.set() + await settlement_started.wait() + raise UpstreamWebSocketTransportError( + "Codex upstream websocket send failed: heartbeat expired", + error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + ) + + async def send_bytes(self, _data: bytes) -> None: + return None + + async def receive(self) -> SimpleNamespace: + self.receive_count += 1 + if self.receive_count == 1: + return SimpleNamespace( + kind="text", + text=json.dumps( + { + "type": "response.created", + "response": {"id": "resp_liveness_race", "status": "in_progress"}, + }, + separators=(",", ":"), + ), + data=None, + close_code=None, + error=None, + error_code=None, + ) + await self.second_send_started.wait() + return SimpleNamespace( + kind="error", + text=None, + data=None, + close_code=1011, + error="Codex upstream websocket receive failed: heartbeat expired", + error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + ) + + async def close(self) -> None: + self.closed = True + + released_request_ids: list[str] = [] + + async def controlled_release(request_state: proxy_service._WebSocketRequestState) -> None: + if not settlement_started.is_set(): + settlement_started.set() + await allow_settlement.wait() + released_request_ids.append(request_state.request_id) + + downstream = RacingDownstreamWebSocket() + upstream = RacingUpstreamWebSocket() + account = _make_account("acc_ws_liveness_race") + + async def connect(*_args: object, **_kwargs: object): + return account, upstream + + monkeypatch.setattr(proxy_service.ProxyService, "_connect_proxy_websocket", connect) + monkeypatch.setattr(service, "_release_websocket_request_state_reservation", controlled_release) + + proxy_task = asyncio.create_task( + service.proxy_responses_websocket( + cast(WebSocket, downstream), + {}, + codex_session_affinity=False, + openai_cache_affinity=False, + api_key=None, + ) + ) + try: + await asyncio.wait_for(settlement_started.wait(), timeout=1.0) + assert proxy_task.done() is False + allow_settlement.set() + await asyncio.wait_for(proxy_task, timeout=1.0) + finally: + allow_settlement.set() + if not proxy_task.done(): + proxy_task.cancel() + await asyncio.gather(proxy_task, return_exceptions=True) + + emitted = [json.loads(text) for text in downstream.sent_text] + failures = [payload for payload in emitted if payload.get("type") == "response.failed"] + assert len(failures) == 2 + assert {payload["response"]["error"]["code"] for payload in failures} == {UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE} + assert len(released_request_ids) == 2 + assert len(set(released_request_ids)) == 2 + assert upstream.send_count == 2 + assert upstream.closed is True + handle_stream_error.assert_not_awaited() + assert len(request_logs.calls) == 2 + + @pytest.mark.asyncio async def test_stream_api_key_settlement_detaches_and_closes_repo(monkeypatch): started = asyncio.Event() @@ -25393,12 +25556,21 @@ async def close(self) -> None: @pytest.mark.asyncio -async def test_relay_upstream_websocket_network_failure_is_neutral_and_not_replayed(monkeypatch): +@pytest.mark.parametrize( + "error_code", + ["proxy_network_unavailable", UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE], + ids=["process-network", "liveness-timeout"], +) +async def test_relay_upstream_websocket_account_neutral_failure_is_not_replayed( + monkeypatch: pytest.MonkeyPatch, + error_code: str, +) -> None: request_logs = _RequestLogsRecorder() service = proxy_service.ProxyService(_repo_factory(request_logs)) handle_stream_error = AsyncMock() + release_reservation = AsyncMock() monkeypatch.setattr(service, "_handle_stream_error", handle_stream_error) - monkeypatch.setattr(service, "_release_websocket_request_state_reservation", AsyncMock()) + monkeypatch.setattr(service, "_release_websocket_request_state_reservation", release_reservation) class _FakeDownstreamWebSocket: def __init__(self) -> None: @@ -25410,19 +25582,22 @@ async def send_text(self, text: str) -> None: async def close(self, code: int = 1000, reason: str | None = None) -> None: del code, reason - class _NetworkFailureUpstream: + class _AccountNeutralFailureUpstream: + def __init__(self) -> None: + self.closed = False + async def receive(self) -> SimpleNamespace: return SimpleNamespace( kind="error", text=None, data=None, close_code=None, - error="Codex upstream websocket receive failed via proxy endpoint ep_1: OSError", - error_code="proxy_network_unavailable", + error="Upstream websocket liveness failed", + error_code=error_code, ) async def close(self) -> None: - return None + self.closed = True request_state = proxy_service._WebSocketRequestState( request_id="ws_req_network_failure", @@ -25437,10 +25612,11 @@ async def close(self) -> None: pending_requests = deque([request_state]) upstream_control = proxy_service._WebSocketUpstreamControl() downstream = _FakeDownstreamWebSocket() + upstream = _AccountNeutralFailureUpstream() await service._relay_upstream_websocket_messages( cast(WebSocket, downstream), - cast(proxy_service.UpstreamWebSocket, _NetworkFailureUpstream()), + cast(proxy_service.UpstreamWebSocket, upstream), account=_make_account("acc_ws_network_failure"), account_id_value="acc_ws_network_failure", pending_requests=pending_requests, @@ -25458,8 +25634,82 @@ async def close(self) -> None: assert upstream_control.reconnect_requested is False assert list(pending_requests) == [] handle_stream_error.assert_not_awaited() + release_reservation.assert_awaited_once_with(request_state) + assert upstream.closed is (error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE) terminal = json.loads(downstream.sent_text[-1]) - assert terminal["response"]["error"]["code"] == "proxy_network_unavailable" + assert terminal["response"]["error"]["code"] == error_code + + +@pytest.mark.asyncio +async def test_relay_upstream_websocket_liveness_timeout_preserves_sequenced_failure_contract( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(_repo_factory(_RequestLogsRecorder())) + fail_pending = AsyncMock() + monkeypatch.setattr(service, "_fail_pending_websocket_requests", fail_pending) + + class _DownstreamWebSocket: + def __init__(self) -> None: + self.close_calls: list[tuple[int, str | None]] = [] + + async def close(self, code: int = 1000, reason: str | None = None) -> None: + self.close_calls.append((code, reason)) + + class _LivenessTimeoutUpstream: + def __init__(self) -> None: + self.closed = False + + async def receive(self) -> SimpleNamespace: + return SimpleNamespace( + kind="error", + text=None, + data=None, + close_code=1011, + error="Upstream websocket liveness failed", + error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + ) + + async def close(self) -> None: + self.closed = True + + request_state = proxy_service._WebSocketRequestState( + request_id="ws_req_sequenced_liveness_timeout", + model="gpt-5.1", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + request_text='{"type":"response.create","model":"gpt-5.1","input":"hi"}', + awaiting_response_created=False, + last_downstream_sequence_number=1, + ) + pending_requests = deque([request_state]) + downstream = _DownstreamWebSocket() + upstream = _LivenessTimeoutUpstream() + + await service._relay_upstream_websocket_messages( + cast(WebSocket, downstream), + cast(proxy_service.UpstreamWebSocket, upstream), + account=_make_account("acc_ws_sequenced_liveness_timeout"), + account_id_value="acc_ws_sequenced_liveness_timeout", + pending_requests=pending_requests, + pending_lock=anyio.Lock(), + client_send_lock=anyio.Lock(), + api_key=None, + upstream_control=proxy_service._WebSocketUpstreamControl(), + response_create_gate=asyncio.Semaphore(1), + proxy_request_budget_seconds=5.0, + stream_idle_timeout_seconds=5.0, + downstream_activity=proxy_service._DownstreamWebSocketActivity(), + ) + + fail_pending.assert_awaited_once() + fail_pending_args = fail_pending.await_args + assert fail_pending_args is not None + assert fail_pending_args.kwargs["penalize_account"] is False + assert fail_pending_args.kwargs["suppress_sequenced_downstream_errors"] is True + assert upstream.closed is True + assert downstream.close_calls[0] == (1011, "upstream replay requires a fresh request") @pytest.mark.asyncio diff --git a/tests/unit/test_proxy_websocket_client.py b/tests/unit/test_proxy_websocket_client.py index 034d0ca6c2..3e09e51a60 100644 --- a/tests/unit/test_proxy_websocket_client.py +++ b/tests/unit/test_proxy_websocket_client.py @@ -19,6 +19,7 @@ from app.core.clients.codex import CodexTransportError, CodexWebSocketResult from app.core.clients.proxy import ProxyResponseError from app.core.clients.proxy_websocket import ( + UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, CodexUpstreamWebSocket, RealtimeWebSocketProtocol, UpstreamWebSocketTransportError, @@ -134,6 +135,9 @@ async def recv(self) -> tuple[bytes, int]: async def receive(self) -> object: return b'{"type":"response.completed"}' + def exception(self) -> BaseException | None: + return None + async def close(self, *, code: int = 1000, message: bytes = b"") -> None: del code, message self.closed = True @@ -214,6 +218,121 @@ async def recv(self): assert message.error is None +@pytest.mark.asyncio +async def test_direct_adapter_classifies_keepalive_timeout() -> None: + class Connection: + async def recv(self): + raise ConnectionClosedError(None, Close(1011, "keepalive ping timeout")) + + websocket = WebsocketsUpstreamWebSocket(cast(Any, Connection())) + + message = await websocket.receive() + + assert message.kind == "error" + assert message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + + +@pytest.mark.asyncio +async def test_direct_adapter_classifies_keepalive_timeout_after_close_ack() -> None: + class Connection: + async def recv(self): + raise ConnectionClosedError( + Close(1000, "acknowledged"), + Close(1011, "keepalive ping timeout"), + False, + ) + + websocket = WebsocketsUpstreamWebSocket(cast(Any, Connection())) + + message = await websocket.receive() + + assert message.kind == "error" + assert message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + + +@pytest.mark.asyncio +async def test_direct_adapter_does_not_trust_peer_keepalive_timeout_marker() -> None: + class Connection: + async def recv(self): + raise ConnectionClosedError( + Close(1011, "keepalive ping timeout"), + Close(1011, "keepalive ping timeout"), + True, + ) + + websocket = WebsocketsUpstreamWebSocket(cast(Any, Connection())) + + message = await websocket.receive() + + assert message.kind == "error" + assert message.error_code is None + + +@pytest.mark.asyncio +async def test_routed_adapter_classifies_heartbeat_timeout() -> None: + websocket = CodexUpstreamWebSocket( + _FakeCodexErrorWebSocket(aiohttp.ServerTimeoutError("No PONG received after 60.0 seconds")) + ) + + message = await websocket.receive() + + assert message.kind == "error" + assert message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + + +@pytest.mark.asyncio +async def test_routed_adapter_classifies_heartbeat_timeout_stored_between_receive_calls() -> None: + heartbeat_timeout = aiohttp.ServerTimeoutError("No PONG received after 60.0 seconds") + + class ClosedWebSocket(_FakeCodexWebSocket): + async def receive(self) -> aiohttp.WSMessage: + return aiohttp.WSMessage(aiohttp.WSMsgType.CLOSED, None, None) + + def exception(self) -> BaseException | None: + return heartbeat_timeout + + websocket = CodexUpstreamWebSocket(ClosedWebSocket()) + + message = await websocket.receive() + + assert message.kind == "error" + assert message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "payload", + ["request", b"request"], + ids=["text", "bytes"], +) +async def test_routed_adapter_send_preserves_stored_heartbeat_timeout( + payload: str | bytes, +) -> None: + heartbeat_timeout = aiohttp.ServerTimeoutError("No PONG received after 60.0 seconds") + + class ClosedWebSocket(_FakeCodexWebSocket): + async def send_str(self, data: str) -> None: + del data + raise RuntimeError("Cannot write to closing transport") + + async def send_bytes(self, data: bytes) -> None: + del data + raise RuntimeError("Cannot write to closing transport") + + def exception(self) -> BaseException | None: + return heartbeat_timeout + + websocket = CodexUpstreamWebSocket(ClosedWebSocket()) + + with pytest.raises(UpstreamWebSocketTransportError) as exc_info: + if isinstance(payload, str): + await websocket.send_text(payload) + else: + await websocket.send_bytes(payload) + + assert exc_info.value.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + + @pytest.mark.asyncio async def test_codex_responses_websocket_closes_owned_client_when_context_exit_fails(): class _FailingContext: @@ -252,6 +371,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=False, ), @@ -280,7 +400,7 @@ async def fake_websocket_connect(url: str, **kwargs): assert kwargs["proxy"] is None assert kwargs["open_timeout"] == 7.0 assert "ping_interval" not in kwargs - assert kwargs["ping_timeout"] is None + assert kwargs["ping_timeout"] == 120.0 assert kwargs["max_size"] == 4321 assert "subprotocols" not in kwargs additional_headers = cast(dict[str, str], kwargs["additional_headers"]) @@ -314,6 +434,7 @@ async def recv(self) -> str: lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=False, ), @@ -352,6 +473,7 @@ async def test_connect_responses_websocket_routed_codex_call_preserves_size_limi lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=False, ), @@ -376,6 +498,7 @@ async def test_connect_responses_websocket_routed_codex_call_preserves_size_limi assert call["route"] is route assert call["timeout"] == 7.0 assert call["max_msg_size"] == 4321 + assert call["heartbeat"] == 120.0 assert "max_size" not in call assert "protocols" not in call assert websocket.response_header("x-codex-turn-state") == "turn-routed" @@ -652,6 +775,7 @@ async def test_connect_responses_websocket_routed_transport_error_maps_proxy_err lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=False, ), @@ -689,6 +813,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=False, ), @@ -724,6 +849,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=False, ), @@ -789,6 +915,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=False, ), @@ -825,6 +952,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -863,6 +991,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -910,6 +1039,7 @@ async def test_connect_responses_websocket_sanitizes_ws_error_payload(monkeypatc lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -965,6 +1095,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1007,6 +1138,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1051,6 +1183,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1095,6 +1228,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1137,6 +1271,7 @@ def upstream_websocket_proxy_env(self): lambda: _Settings( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1180,6 +1315,7 @@ def upstream_websocket_proxy_env(self): lambda: _Settings( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1216,6 +1352,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="http://chatgpt.local/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1274,6 +1411,7 @@ async def upstream_handler(connection): lambda: SimpleNamespace( upstream_base_url=f"http://127.0.0.1:{upstream_port}/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1312,6 +1450,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="http://chatgpt.local/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1359,6 +1498,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), @@ -1393,6 +1533,7 @@ async def fake_websocket_connect(url: str, **kwargs): lambda: SimpleNamespace( upstream_base_url="https://chatgpt.com/backend-api", upstream_connect_timeout_seconds=7.0, + proxy_downstream_websocket_idle_timeout_seconds=120.0, max_sse_event_bytes=4321, upstream_websocket_trust_env=True, ), From 424e133664c244d9a39f3b10f88026224452e72d Mon Sep 17 00:00:00 2001 From: Roman Leventov Date: Tue, 4 Aug 2026 12:43:07 +0800 Subject: [PATCH 2/4] fix(proxy): preserve websocket cleanup during cancellation --- .../proxy/_service/http_bridge/helpers.py | 7 +++++- tests/unit/test_proxy_http_bridge.py | 25 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/app/modules/proxy/_service/http_bridge/helpers.py b/app/modules/proxy/_service/http_bridge/helpers.py index 59eca389e0..76b17dbe6f 100644 --- a/app/modules/proxy/_service/http_bridge/helpers.py +++ b/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 diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index a0d4ec3575..0834b6b4af 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -8925,6 +8925,31 @@ async def stubborn_child() -> None: assert cleanup_tasks == set() +@pytest.mark.asyncio +async def test_await_cancelled_task_allows_outer_cancellation_cleanup_to_finish() -> None: + cleanup_finished = asyncio.Event() + + async def cancelled_owner() -> None: + child = asyncio.create_task(asyncio.Event().wait()) + try: + await asyncio.Event().wait() + finally: + await proxy_service._await_cancelled_task( + child, + timeout_seconds=1.0, + label="outer cancellation child", + ) + cleanup_finished.set() + + owner = asyncio.create_task(cancelled_owner()) + await asyncio.sleep(0) + owner.cancel() + + with pytest.raises(asyncio.CancelledError): + await owner + assert cleanup_finished.is_set() + + @pytest.mark.asyncio async def test_await_cancelled_task_defers_stubborn_child_cleanup() -> None: child_cancelled = asyncio.Event() From 463179f3602c44d918854ebfc6a9ad7359b1b450 Mon Sep 17 00:00:00 2001 From: Roman Leventov Date: Tue, 4 Aug 2026 13:51:23 +0800 Subject: [PATCH 3/4] test(proxy): synchronize websocket cleanup assertion --- tests/integration/test_proxy_websocket_responses.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/integration/test_proxy_websocket_responses.py b/tests/integration/test_proxy_websocket_responses.py index 1918dbaa19..b09920cac7 100644 --- a/tests/integration/test_proxy_websocket_responses.py +++ b/tests/integration/test_proxy_websocket_responses.py @@ -3,6 +3,7 @@ import asyncio import json import logging +import threading from collections import deque from datetime import datetime, timedelta, timezone from types import SimpleNamespace @@ -93,6 +94,7 @@ def __init__(self, messages: list[_FakeUpstreamMessage]) -> None: self.archived_receive_request_ids: list[str | None] = [] self.archived_receive_texts: list[str] = [] self.closed = False + self.closed_event = threading.Event() self._messages: asyncio.Queue[_FakeUpstreamMessage] = asyncio.Queue() for message in messages: self._messages.put_nowait(message) @@ -116,6 +118,7 @@ def archive_received(self, message: _FakeUpstreamMessage) -> None: async def close(self) -> None: self.closed = True + self.closed_event.set() class _SequencedUpstreamWebSocket(_FakeUpstreamWebSocket): @@ -5606,6 +5609,9 @@ async def fake_resolve_previous_response_owner( for call in log_calls ) assert any(call["status"] == "success" and call["request_id"] == "resp_ws_inflight" for call in log_calls) + # TestClient runs the ASGI task in a worker thread. Wait for its owned + # cancellation cleanup instead of racing that thread on the plain flag. + assert fake_upstream.closed_event.wait(timeout=1.0) assert fake_upstream.closed is True From 5036851b21db7eb5b58650b27c84a1940fabddd7 Mon Sep 17 00:00:00 2001 From: Roman Leventov Date: Tue, 4 Aug 2026 20:30:06 +0800 Subject: [PATCH 4/4] fix(proxy): make bridge liveness settlement explicit --- .../_service/http_bridge/request_submit.py | 17 ++- .../_service/http_bridge/upstream_events.py | 18 ++- app/modules/proxy/_service/support.py | 18 +++ .../design.md | 10 +- .../specs/responses-api-compat/spec.md | 10 +- .../tasks.md | 5 + .../specs/responses-api-compat/context.md | 1 + openspec/specs/responses-api-compat/spec.md | 10 +- .../test_proxy_websocket_responses.py | 59 +++++++++ tests/unit/test_proxy_http_bridge.py | 123 ++++++++++++++++++ 10 files changed, 258 insertions(+), 13 deletions(-) diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index b8e51df7fa..c97ffcfa81 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -1149,7 +1149,7 @@ 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 @@ -1157,6 +1157,15 @@ async def _submit_http_bridge_request_with_handoff( 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() @@ -1242,9 +1251,9 @@ async def _submit_http_bridge_request_with_handoff( # 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 marked the session closed while holding - # lifecycle_lock. It therefore owns the entire session deque, - # including older in-flight requests; settling only this + # 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( diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index cbc3659134..165857d733 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -965,11 +965,14 @@ async def _relay_http_bridge_upstream_messages( else None ) async with session.lifecycle_lock: - if session.closed and message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE: - # A submitter holds this lock across send_text and marks - # the session closed before releasing it on ambiguous - # send failure. In that race the submitter owns terminal - # settlement; the reader must not settle the same state. + if ( + session.liveness_settlement_owner == "send" + and message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + ): + # A submitter publishes this dedicated claim beside the + # failing send while holding lifecycle_lock. ``closed`` + # alone is only an admission/retirement state and must + # never suppress settlement of still-pending siblings. break await self._fail_http_bridge_reader_and_maybe_retire( session, @@ -1007,7 +1010,10 @@ async def _relay_http_bridge_upstream_messages( error_code = exc.error_code if isinstance(exc, UpstreamWebSocketTransportError) else "stream_incomplete" account_neutral = is_account_neutral_websocket_error_code(error_code) async with session.lifecycle_lock: - if not (session.closed and error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE): + if not ( + session.liveness_settlement_owner == "send" + and error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + ): # Match the message path above when receive() raises while # a concurrent send failure already owns settlement. await self._fail_http_bridge_reader_and_maybe_retire( diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index a04b1709d1..e4e21ec9b0 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -987,6 +987,13 @@ class _HTTPBridgeSession: last_upstream_close_code: int | None = None last_upstream_close_generation: int = 0 closed: bool = False + # ``closed`` rejects new admissions but is written by many unrelated + # retirement paths; it never proves that a sender owns pending settlement. + # Only the submitter may claim this, while holding ``lifecycle_lock``, when + # its own send reports a liveness timeout. The reader remains the default + # settlement owner for every other close, including an already-closed + # session whose still-running transport later loses heartbeat liveness. + liveness_settlement_owner: Literal["send"] | None = None # Set while a reader handoff is replacing the socket. Idle pruning must # retain the registered session during this short transition even though # ``closed`` is fail-closed for normal request reuse. @@ -1001,6 +1008,17 @@ class _HTTPBridgeSession: upstream_proxy_fallback_used: bool | None = None upstream_proxy_fail_closed_reason: str | None = None + def claim_liveness_settlement(self) -> bool: + """Claim whole-deque settlement for a liveness-failed submitter. + + The caller must hold ``lifecycle_lock`` across the failing send and + this synchronous claim so the reader cannot settle the same deque. + """ + + if self.liveness_settlement_owner is None: + self.liveness_settlement_owner = "send" + return self.liveness_settlement_owner == "send" + def _complete_http_bridge_handoff( session: _HTTPBridgeSession, diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md index 2568d454d4..134f628197 100644 --- a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/design.md @@ -38,6 +38,14 @@ A shared code predicate identifies account-neutral WebSocket failures so relay o Both relay owners treat the classified liveness timeout like a post-send network failure: no transparent replay, no account-health write, exact-once pending-request settlement, and retirement of the affected socket. The downstream error remains retryable at the client boundary, where a fresh client connection can safely establish a new upstream route under the client's existing retry semantics. +The HTTP bridge cannot infer settlement ownership from `session.closed`: that +flag also rejects admission after continuity-persistence and other failures +that settle only the submitting request. A submitter claims whole-deque +liveness settlement explicitly while holding the session lifecycle lock around +the failing send. The reader skips its normal settlement only when that claim +exists; a later liveness expiry on an otherwise closed session still settles +every pending sibling. + For example, if `response.create` was written before a VPN route disappears and no pong returns, codex-lb emits a terminal `upstream_websocket_liveness_timeout` failure for that pending request and closes or retires the upstream session. It does not resend `response.create` on another account or socket. ## Risks / Trade-offs @@ -45,7 +53,7 @@ For example, if `response.create` was written before a VPN route disappears and - [Some intermediaries do not answer WebSocket pings correctly] → Use the established, configurable timeout already deployed for live sideband traffic; operators can adjust the existing value if needed. - [Detection is not instantaneous] → A bounded delay is preferable to false positives and is far shorter than the multi-hour Responses request budget. - [The client must retry the interrupted turn] → This avoids duplicate model work and tool side effects when delivery status is unknowable. -- [Library wording could change] → Pin direct detection to the concrete close code/reason emitted by the installed `websockets` API and cover both adapters with regression tests. +- [Library wording could change] → Pin direct detection to the concrete close code/reason emitted by the installed `websockets` API and drive a real no-pong expiry in integration coverage, in addition to adapter unit tests. ## Migration Plan diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md index a8eeda588d..35ab978124 100644 --- a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/specs/responses-api-compat/spec.md @@ -2,7 +2,7 @@ ### Requirement: Responses upstream websocket liveness is bounded -The proxy MUST configure direct and routed upstream Responses WebSocket transports with finite ping/pong liveness detection derived from `proxy_downstream_websocket_idle_timeout_seconds`. When an established Responses WebSocket is terminated because its transport did not receive the required pong, the adapter MUST classify the failure as `upstream_websocket_liveness_timeout`. Direct WebSocket and HTTP bridge relay owners MUST treat that failure as account neutral, MUST NOT transparently replay a pending request whose delivery is ambiguous, MUST finalize its pending request ownership exactly once, and MUST retire the affected upstream socket so a later client retry opens a fresh connection. +The proxy MUST configure direct and routed upstream Responses WebSocket transports with finite ping/pong liveness detection derived from `proxy_downstream_websocket_idle_timeout_seconds`. When an established Responses WebSocket is terminated because its transport did not receive the required pong, the adapter MUST classify the failure as `upstream_websocket_liveness_timeout`. Direct WebSocket and HTTP bridge relay owners MUST treat that failure as account neutral, MUST NOT transparently replay a pending request whose delivery is ambiguous, MUST finalize its pending request ownership exactly once, and MUST retire the affected upstream socket so a later client retry opens a fresh connection. An HTTP bridge reader MUST suppress its own pending-deque settlement only when a concurrent submitter explicitly claimed liveness-settlement ownership under the session lifecycle lock; `session.closed` alone MUST NOT suppress settlement. #### Scenario: Direct Responses websocket loses pong liveness @@ -29,6 +29,14 @@ The proxy MUST configure direct and routed upstream Responses WebSocket transpor - **THEN** the proxy keeps the upstream socket open - **AND** the existing Responses request budget remains authoritative for the turn +#### Scenario: Closed bridge without a sender claim later loses pong liveness + +- **GIVEN** an HTTP bridge session has multiple pending requests +- **AND** a separate submit failure marks the session closed without claiming liveness-settlement ownership +- **WHEN** the still-running upstream transport later expires its heartbeat +- **THEN** the reader settles every pending request with `upstream_websocket_liveness_timeout` +- **AND** the selected account receives no failure-health signal + ## MODIFIED Requirements ### Requirement: Upstream websocket drops penalize affected accounts diff --git a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md index ac3e5b368c..ace907a846 100644 --- a/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md +++ b/openspec/changes/archive/2026-08-04-recover-responses-websocket-liveness/tasks.md @@ -16,3 +16,8 @@ ## 4. Verification - [x] 4.1 Run focused tests, formatting, lint, type, architecture, and strict OpenSpec validation checks. + +## 5. Maintainer review follow-up + +- [x] 5.1 Replace the HTTP bridge reader's overloaded `closed` ownership guard with an explicit submitter liveness-settlement claim and cover closed-session sibling settlement. +- [x] 5.2 Drive a real installed-library no-pong timeout so integration coverage pins the `websockets` watchdog shape used by classification. diff --git a/openspec/specs/responses-api-compat/context.md b/openspec/specs/responses-api-compat/context.md index a4e2d90f1c..9a62f360da 100644 --- a/openspec/specs/responses-api-compat/context.md +++ b/openspec/specs/responses-api-compat/context.md @@ -34,6 +34,7 @@ See `openspec/specs/responses-api-compat/spec.md` for normative requirements. - Codex-native direct websocket `/backend-api/codex/responses` treats upstream `previous_response_id` as an ephemeral anchor. If that anchor goes stale, the proxy must mask raw `previous_response_not_found` details and emit a sanitized `codex_previous_response_stale` classifier so compatible Codex clients can soft-reset and retry without `previous_response_id`. - Upstream Responses WebSockets use transport ping/pong control frames to detect a black-holed connection without confusing valid application-event silence with an idle turn. Direct and routed connections reuse `proxy_downstream_websocket_idle_timeout_seconds` for this zero-config liveness budget. - A post-send liveness timeout is delivery-ambiguous. It remains account-neutral, is never transparently replayed, and retires the affected upstream socket so a client retry opens a fresh route without risking duplicated model work or tool side effects. +- HTTP bridge settlement ownership is explicit: `closed` rejects new work but does not imply that a submitter owns existing siblings. Only a liveness-failed send claims whole-deque settlement under the lifecycle lock; otherwise the reader remains responsible for settling pending requests when the transport dies. ## Fast Mode and Service Tiers diff --git a/openspec/specs/responses-api-compat/spec.md b/openspec/specs/responses-api-compat/spec.md index 782660bfc7..7470a0574d 100644 --- a/openspec/specs/responses-api-compat/spec.md +++ b/openspec/specs/responses-api-compat/spec.md @@ -85,7 +85,7 @@ The default compact request budget MUST be at least 180 seconds, and the default ### Requirement: Responses upstream websocket liveness is bounded -The proxy MUST configure direct and routed upstream Responses WebSocket transports with finite ping/pong liveness detection derived from `proxy_downstream_websocket_idle_timeout_seconds`. When an established Responses WebSocket is terminated because its transport did not receive the required pong, the adapter MUST classify the failure as `upstream_websocket_liveness_timeout`. Direct WebSocket and HTTP bridge relay owners MUST treat that failure as account neutral, MUST NOT transparently replay a pending request whose delivery is ambiguous, MUST finalize its pending request ownership exactly once, and MUST retire the affected upstream socket so a later client retry opens a fresh connection. +The proxy MUST configure direct and routed upstream Responses WebSocket transports with finite ping/pong liveness detection derived from `proxy_downstream_websocket_idle_timeout_seconds`. When an established Responses WebSocket is terminated because its transport did not receive the required pong, the adapter MUST classify the failure as `upstream_websocket_liveness_timeout`. Direct WebSocket and HTTP bridge relay owners MUST treat that failure as account neutral, MUST NOT transparently replay a pending request whose delivery is ambiguous, MUST finalize its pending request ownership exactly once, and MUST retire the affected upstream socket so a later client retry opens a fresh connection. An HTTP bridge reader MUST suppress its own pending-deque settlement only when a concurrent submitter explicitly claimed liveness-settlement ownership under the session lifecycle lock; `session.closed` alone MUST NOT suppress settlement. #### Scenario: Direct Responses websocket loses pong liveness @@ -112,6 +112,14 @@ The proxy MUST configure direct and routed upstream Responses WebSocket transpor - **THEN** the proxy keeps the upstream socket open - **AND** the existing Responses request budget remains authoritative for the turn +#### Scenario: Closed bridge without a sender claim later loses pong liveness + +- **GIVEN** an HTTP bridge session has multiple pending requests +- **AND** a separate submit failure marks the session closed without claiming liveness-settlement ownership +- **WHEN** the still-running upstream transport later expires its heartbeat +- **THEN** the reader settles every pending request with `upstream_websocket_liveness_timeout` +- **AND** the selected account receives no failure-health signal + ### Requirement: Upstream websocket drops penalize affected accounts When an upstream websocket closes while one or more streamed response requests are pending and have not reached a terminal event, the proxy MUST record a transient upstream error for the account before signaling failure for those pending requests, except when the close carries a classified process-wide network failure or upstream WebSocket liveness timeout. A classified process-wide network failure or upstream WebSocket liveness timeout MUST remain account neutral and use its classified error code. For other closes, the proxy MUST surface `stream_incomplete` to affected pending requests except when a direct Responses WebSocket request has already successfully emitted a finite integer `sequence_number`. For that sequenced direct-WebSocket case, the proxy MUST record the request outcome as `stream_incomplete` without emitting a synthetic terminal frame under the active response id, then MUST close the downstream WebSocket with code 1011. diff --git a/tests/integration/test_proxy_websocket_responses.py b/tests/integration/test_proxy_websocket_responses.py index faa24dcc5e..f5b35840a6 100644 --- a/tests/integration/test_proxy_websocket_responses.py +++ b/tests/integration/test_proxy_websocket_responses.py @@ -1,6 +1,8 @@ from __future__ import annotations import asyncio +import base64 +import hashlib import json import logging import threading @@ -13,10 +15,15 @@ from fastapi.testclient import TestClient from httpx import Headers from starlette.websockets import WebSocketDisconnect +from websockets.asyncio.client import connect as websocket_connect import app.modules.proxy.api as proxy_api_module import app.modules.proxy.service as proxy_module from app.core.auth.refresh import RefreshError +from app.core.clients.proxy_websocket import ( + UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + WebsocketsUpstreamWebSocket, +) from app.core.utils.request_id import get_request_id from app.modules.api_keys.service import ApiKeyData from app.modules.proxy._service.websocket import mixin as websocket_mixin_module @@ -29,6 +36,58 @@ pytestmark = pytest.mark.integration +@pytest.mark.asyncio +async def test_real_websockets_keepalive_expiry_preserves_liveness_classification() -> None: + async def accept_without_answering_frames( + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + try: + request = await reader.readuntil(b"\r\n\r\n") + websocket_key = next( + line.split(b":", 1)[1].strip() + for line in request.split(b"\r\n") + if line.lower().startswith(b"sec-websocket-key:") + ) + accept = base64.b64encode(hashlib.sha1(websocket_key + b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11").digest()) + writer.write( + b"HTTP/1.1 101 Switching Protocols\r\n" + b"Upgrade: websocket\r\n" + b"Connection: Upgrade\r\n" + b"Sec-WebSocket-Accept: " + accept + b"\r\n\r\n" + ) + await writer.drain() + # Read and discard every frame. In particular, never answer pings, + # so the production client watchdog must terminate the connection. + while await reader.read(65536): + pass + finally: + writer.close() + try: + await writer.wait_closed() + except ConnectionError: + pass + + server = await asyncio.start_server(accept_without_answering_frames, "127.0.0.1", 0) + async with server: + assert server.sockets + port = server.sockets[0].getsockname()[1] + async with websocket_connect( + f"ws://127.0.0.1:{port}", + ping_interval=0.05, + ping_timeout=0.05, + close_timeout=0.05, + proxy=None, + ) as connection: + upstream = WebsocketsUpstreamWebSocket(connection) + message = await asyncio.wait_for(upstream.receive(), timeout=1.0) + + # This assertion pins the actual close code/reason emitted by the installed + # websockets watchdog to the adapter's stable, account-neutral classifier. + assert message.kind == "error" + assert message.error_code == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + + def _assert_previous_response_not_found_error(error: dict[str, object]) -> None: assert error["code"] == proxy_module.PREVIOUS_RESPONSE_NOT_FOUND_CODE assert error["message"] == proxy_module.PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index c703df89eb..2c27c57594 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -20056,6 +20056,129 @@ async def receive(self) -> UpstreamWebSocketMessage: assert session.pending_requests == deque() assert session.queued_request_count == 0 assert session.closed is True + assert session.liveness_settlement_owner == "send" + retire.assert_awaited_once_with( + session, + detail=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + response_events_seen=0, + ) + + +@pytest.mark.asyncio +async def test_http_bridge_closed_without_liveness_claim_still_settles_pending_siblings( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class DelayedLivenessUpstream: + def __init__(self) -> None: + self.receive_started = asyncio.Event() + self.release_liveness = asyncio.Event() + self.send_text = AsyncMock() + self.close = AsyncMock() + + async def receive(self) -> UpstreamWebSocketMessage: + self.receive_started.set() + await self.release_liveness.wait() + return UpstreamWebSocketMessage( + kind="error", + error="Codex upstream websocket receive failed: heartbeat expired", + error_code=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE, + ) + + def pending_sibling(request_id: str) -> proxy_service._WebSocketRequestState: + return proxy_service._WebSocketRequestState( + request_id=request_id, + model="gpt-5.6-sol", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + response_id=f"resp-{request_id}", + event_queue=asyncio.Queue(), + transport="http", + skip_request_log=True, + ) + + service = proxy_service.ProxyService(cast(Any, nullcontext())) + siblings = [pending_sibling("sibling-one"), pending_sibling("sibling-two")] + upstream = DelayedLivenessUpstream() + session = _make_bridge_session( + key_value="closed-without-liveness-claim", + pending_requests=deque(siblings), + queued_request_count=len(siblings), + ) + session.upstream = cast(UpstreamWebSocket, upstream) + session.durable_session_id = "durable-closed-without-liveness-claim" + session.durable_owner_epoch = 2 + service._http_bridge_sessions[session.key] = session + record_recovery_attempt = AsyncMock(return_value=None) + service._durable_bridge = cast( + Any, + SimpleNamespace( + lookup_retry_circuit=AsyncMock(return_value=None), + record_recovery_attempt=record_recovery_attempt, + release_live_session=AsyncMock(return_value=None), + ), + ) + third_request = proxy_service._WebSocketRequestState( + request_id="third-submit", + model="gpt-5.6-sol", + service_tier=None, + reasoning_effort=None, + api_key_reservation=None, + started_at=time.monotonic(), + awaiting_response_created=True, + request_text='{"type":"response.create","model":"gpt-5.6-sol","input":"hi"}', + fresh_upstream_request_text='{"type":"response.create","model":"gpt-5.6-sol","input":"hi"}', + fresh_upstream_request_is_retry_safe=True, + transport="http", + skip_request_log=True, + ) + fail_pending = AsyncMock(wraps=service._fail_pending_websocket_requests) + retire = AsyncMock() + retry_precreated = AsyncMock(return_value=True) + monkeypatch.setattr(proxy_service, "get_settings", lambda: _make_app_settings()) + monkeypatch.setattr(service, "_fail_pending_websocket_requests", fail_pending) + monkeypatch.setattr(service, "_retire_stale_pending_http_bridge_session", retire) + monkeypatch.setattr(service, "_retry_http_bridge_precreated_request", retry_precreated) + + reader_task = asyncio.create_task(service._relay_http_bridge_upstream_messages(session)) + try: + await asyncio.wait_for(upstream.receive_started.wait(), timeout=1.0) + with pytest.raises(ProxyResponseError) as exc_info: + await service._submit_http_bridge_request( + session, + request_state=third_request, + text_data=third_request.request_text or "{}", + queue_limit=8, + ) + + assert exc_info.value.payload["error"]["code"] == "bridge_continuity_persistence_failed" + record_recovery_attempt.assert_awaited_once() + assert session.closed is True + assert session.liveness_settlement_owner is None + assert session.pending_requests == deque(siblings) + + upstream.release_liveness.set() + await asyncio.wait_for(reader_task, timeout=1.0) + finally: + if not reader_task.done(): + reader_task.cancel() + await asyncio.gather(reader_task, return_exceptions=True) + + fail_pending.assert_awaited_once() + failure_call = fail_pending.await_args + assert failure_call is not None + assert failure_call.kwargs["error_code"] == UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE + assert failure_call.kwargs["penalize_account"] is False + retry_precreated.assert_not_awaited() + for sibling in siblings: + assert sibling.event_queue is not None + terminal_event = await asyncio.wait_for(sibling.event_queue.get(), timeout=0.1) + assert terminal_event is not None + assert UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE in terminal_event + assert await asyncio.wait_for(sibling.event_queue.get(), timeout=0.1) is None + assert session.pending_requests == deque() + assert session.queued_request_count == 0 retire.assert_awaited_once_with( session, detail=UPSTREAM_WEBSOCKET_LIVENESS_TIMEOUT_CODE,