diff --git a/app/core/clients/proxy_websocket.py b/app/core/clients/proxy_websocket.py index 06eb2d5fbb..84c0bb39e8 100644 --- a/app/core/clients/proxy_websocket.py +++ b/app/core/clients/proxy_websocket.py @@ -21,6 +21,7 @@ InvalidProxy, InvalidStatus, ) +from websockets.protocol import State from websockets.typing import Origin, Subprotocol from app.core.clients.codex import ( @@ -157,6 +158,46 @@ def __init__(self, message: str, *, error_code: str) -> None: self.error_code = error_code +_SEND_NOT_DISPATCHED_CONSTRUCTOR_TOKEN = object() + + +class UpstreamWebSocketSendNotDispatchedError(UpstreamWebSocketTransportError): + """Proof that a websocket frame was rejected before dispatch began. + + The constructor token is module-private, subclassing is disabled, and the + adapters are the only production construction path. Python reflection can + still bypass these boundaries, so this is a runtime application contract, + not a security boundary against hostile in-process code. + """ + + def __init_subclass__(cls, **kwargs: object) -> None: + del cls, kwargs + raise TypeError("UpstreamWebSocketSendNotDispatchedError cannot be subclassed") + + def __init__( + self, + message: str, + *, + error_code: str, + _constructor_token: object, + ) -> None: + if _constructor_token is not _SEND_NOT_DISPATCHED_CONSTRUCTOR_TOKEN: + raise TypeError("Use the websocket adapter's closed-before-send check") + super().__init__(message, error_code=error_code) + + +def _websocket_send_not_dispatched_error( + *, + endpoint_id: str | None = None, +) -> UpstreamWebSocketSendNotDispatchedError: + endpoint_suffix = f" via endpoint {endpoint_id}" if endpoint_id else "" + return UpstreamWebSocketSendNotDispatchedError( + f"Upstream websocket was already closed before send{endpoint_suffix}", + error_code="upstream_websocket_closed_before_send", + _constructor_token=_SEND_NOT_DISPATCHED_CONSTRUCTOR_TOKEN, + ) + + def _websocket_transport_error_code(exc: BaseException, *, uses_proxy: bool) -> str: return process_network_error_code( exc, @@ -225,12 +266,16 @@ def __init__( self._preserve_close_semantics = preserve_close_semantics async def send_text(self, text: str) -> None: + if getattr(self._connection, "state", State.OPEN) is not State.OPEN: + raise _websocket_send_not_dispatched_error() try: await self._connection.send(text) except Exception as exc: await _raise_websocket_send_error(exc, uses_proxy=self._uses_proxy) async def send_bytes(self, data: bytes) -> None: + if getattr(self._connection, "state", State.OPEN) is not State.OPEN: + raise _websocket_send_not_dispatched_error() try: await self._connection.send(data) except Exception as exc: @@ -319,6 +364,8 @@ def __init__( self._response_headers = _normalize_response_headers(response_headers) async def send_text(self, text: str) -> None: + if bool(getattr(self._websocket, "closed", False)): + raise _websocket_send_not_dispatched_error(endpoint_id=self._endpoint_id) try: result = self._websocket.send_str(text) if asyncio.iscoroutine(result): @@ -327,6 +374,8 @@ async def send_text(self, text: str) -> None: await _raise_websocket_send_error(exc, endpoint_id=self._endpoint_id, uses_proxy=True) async def send_bytes(self, data: bytes) -> None: + if bool(getattr(self._websocket, "closed", False)): + raise _websocket_send_not_dispatched_error(endpoint_id=self._endpoint_id) try: result = self._websocket.send_bytes(data) if asyncio.iscoroutine(result): diff --git a/app/modules/proxy/_service/http_bridge/helpers.py b/app/modules/proxy/_service/http_bridge/helpers.py index ca98a7fa9f..de6df6855d 100644 --- a/app/modules/proxy/_service/http_bridge/helpers.py +++ b/app/modules/proxy/_service/http_bridge/helpers.py @@ -188,7 +188,7 @@ logger = logging.getLogger("app.modules.proxy.service") _HTTP_BRIDGE_BACKGROUND_CLOSE_TIMEOUT_SECONDS = 5.0 -_HTTP_BRIDGE_EVENTLESS_RESPONSE_CREATED_MAX_SECONDS = 240.0 +_HTTP_BRIDGE_EVENTLESS_RESPONSE_CREATED_MAX_SECONDS = 30.0 _HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL = "missing_response_created_timeout" T = TypeVar("T") diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index 0319e918ea..4036800e92 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -36,7 +36,10 @@ 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 ( + UpstreamWebSocketSendNotDispatchedError, + UpstreamWebSocketTransportError, +) from app.core.errors import ( openai_error, ) @@ -876,6 +879,25 @@ 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 UpstreamWebSocketSendNotDispatchedError: + # The adapter observed the warm socket closed before + # dispatch began, so the exact request remains safe to + # send once on a fresh socket with the same leased + # account. Mid-send failures retain the fail-closed path + # below because their dispatch state is ambiguous. + session.closed = True + session.upstream_control.reconnect_requested = True + session.upstream_control.retire_after_drain = True + recovered = await self._retry_http_bridge_request_on_fresh_upstream( + session, + request_state=request_state, + text_data=text_data, + require_same_account=True, + require_preferred_account=True, + dispatch_proven_absent=True, + ) + if not recovered: + raise except BaseException: # Publish retirement while lifecycle ownership is still # held; a gate waiter must never reuse an ambiguously sent @@ -1456,6 +1478,8 @@ async def _retry_http_bridge_request_on_fresh_upstream( text_data: str, send_request: bool = True, require_same_account: bool = False, + require_preferred_account: bool = False, + dispatch_proven_absent: bool = False, ) -> bool: require_same_account = require_same_account or is_http_bridge_account_neutral_replay( kind=session.key.affinity_kind, @@ -1463,7 +1487,7 @@ async def _retry_http_bridge_request_on_fresh_upstream( ) retry_text_data = text_data using_fresh_replay = False - if request_state.previous_response_id is not None and send_request: + if request_state.previous_response_id is not None and send_request and not dispatch_proven_absent: # After an ambiguous websocket send failure we cannot prove whether # upstream already accepted the continuation. Re-sending the same # previous_response_id request can fork continuity with duplicate @@ -1486,6 +1510,8 @@ async def _retry_http_bridge_request_on_fresh_upstream( if request_state.response_event_count > 0: return False request_state.replay_count += 1 + if require_preferred_account: + request_state.preferred_account_id = session.account.id _log_http_bridge_event( "retry_fresh_upstream", session.key, @@ -1501,6 +1527,7 @@ async def _retry_http_bridge_request_on_fresh_upstream( request_state=request_state, restart_reader=True, require_same_account=require_same_account, + require_preferred_account=require_preferred_account, ) if send_request: retry_text_data = self._http_bridge_text_with_account_installation_id( @@ -1530,12 +1557,13 @@ async def _retry_http_bridge_precreated_request( session: "_HTTPBridgeSession", *, request_state: _WebSocketRequestState | None = None, + require_current_account: bool = False, ) -> bool: account_neutral_recovery = is_http_bridge_account_neutral_replay( kind=session.key.affinity_kind, key=session.key.affinity_key, ) - hard_owner_bound = _http_bridge_key_strength(session.key) == "hard" + hard_owner_bound = _http_bridge_key_strength(session.key) == "hard" or require_current_account async with session.pending_lock: if request_state is not None: if ( @@ -1556,6 +1584,8 @@ async def _retry_http_bridge_precreated_request( if len(retryable_requests) != 1: return False request_state = retryable_requests[0] + if require_current_account: + request_state.preferred_account_id = session.account.id if request_state.previous_response_id is not None and not ( request_state.proxy_injected_previous_response_id and request_state.fresh_upstream_request_is_retry_safe @@ -1611,7 +1641,7 @@ async def _retry_http_bridge_precreated_request( request_text = _prepare_websocket_request_state_for_visible_output_replay(request_state) if request_text is None: return False - if account_neutral_recovery: + if account_neutral_recovery or require_current_account: request_state.preferred_account_id = session.account.id elif not request_state.file_required_preferred_account and not hard_owner_bound: request_state.preferred_account_id = None @@ -1637,6 +1667,7 @@ async def _retry_http_bridge_precreated_request( session, request_state=request_state, require_same_account=True, + require_preferred_account=require_current_account, ) elif require_preferred_reconnect: await self._reconnect_http_bridge_session( diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index a70367af9b..006bfef0e1 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -610,20 +610,63 @@ async def _relay_http_bridge_upstream_messages( async with session.pending_lock: if receive_task is not None and receive_task.done(): continue - expired_owner = any( - deadline is not None and deadline <= now - for request_state in session.pending_requests - if ( - deadline := _http_bridge_eventless_precreated_deadline( - request_state, - stuck_gate_retire_after_seconds=stuck_gate_retire_after_seconds, + eventless_owner = next( + ( + request_state + for request_state in session.pending_requests + if ( + deadline := _http_bridge_eventless_precreated_deadline( + request_state, + stuck_gate_retire_after_seconds=stuck_gate_retire_after_seconds, + ) ) - ) - is not None + is not None + and deadline <= now + ), + None, ) - if not expired_owner: + if eventless_owner is None: continue pending_count = len(session.pending_requests) + can_retry_eventless_owner = pending_count == 1 + receive_cancelled = True + if receive_task is not None: + cancel_requested = receive_task.cancel() + if not cancel_requested: + continue + try: + await receive_task + except asyncio.CancelledError: + relay_task = asyncio.current_task() + if relay_task is not None and relay_task.cancelling(): + raise + receive_task = None + except Exception: + # Preserve the completed task so the next + # loop iteration raises it outside the + # lifecycle lock and uses normal cleanup. + continue + else: + # A response may win the cancellation race. + # Leave the completed task in place so the + # next loop iteration processes its result. + continue + retried = False + if not session.closed and can_retry_eventless_owner and receive_cancelled: + try: + retried = await self._retry_http_bridge_precreated_request( + session, + request_state=eventless_owner, + require_current_account=True, + ) + except UpstreamWebSocketTransportError: + logger.warning( + "HTTP bridge missing response.created retry transport failed", + exc_info=True, + ) + if retried: + continue + async with session.pending_lock: for request_state in session.pending_requests: if request_state.failure_phase_override is None: request_state.failure_phase_override = "upstream" @@ -631,16 +674,9 @@ async def _relay_http_bridge_upstream_messages( request_state.failure_detail_override = ( _HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL ) - # Claim the session before cancelling receive so a + # Claim the session before terminal settlement so a # gate waiter cannot reopen this ambiguous socket. session.closed = True - if receive_task is not None: - receive_cancelled = await _cancel_http_bridge_reader_child( - receive_task, - label="HTTP bridge upstream receive after missing response.created", - ) - if receive_cancelled: - receive_task = None _record_http_bridge_stuck_retire( reason=_HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL, session=session, diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index 0d842e78f9..ecc09fd9d5 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -819,6 +819,11 @@ class _WebSocketRequestState: fresh_upstream_request_responses_lite_model: str | None = None request_stage: str = "first_turn" preferred_account_id: str | None = None + # A closed-before-send transport proof permits one exact replay, but the + # request's response-create lease belongs to the account selected for the + # undispatched attempt. Keep that account mandatory until the resend + # succeeds. + dispatch_absent_replay_account_id: str | None = None require_security_work_authorized: bool = False file_required_preferred_account: bool = False bridge_soft_capacity_reroute_allowed: bool = False diff --git a/app/modules/proxy/_service/websocket/mixin.py b/app/modules/proxy/_service/websocket/mixin.py index 4e292e0046..67f8d35786 100644 --- a/app/modules/proxy/_service/websocket/mixin.py +++ b/app/modules/proxy/_service/websocket/mixin.py @@ -57,6 +57,7 @@ from app.core.clients.proxy import transcribe_audio as core_transcribe_audio # noqa: F401 from app.core.clients.proxy_websocket import ( UpstreamWebSocket, + UpstreamWebSocketSendNotDispatchedError, UpstreamWebSocketTransportError, filter_inbound_websocket_headers, ) @@ -546,6 +547,36 @@ async def _close_downstream_after_sequenced_replay_refusal( ) +async def _claim_websocket_send_not_dispatched_replay( + request_state: _WebSocketRequestState, + pending_requests: deque[_WebSocketRequestState], + *, + pending_lock: anyio.Lock, + account_id: str, +) -> bool: + """Claim one exact resend after the adapter proves dispatch never began.""" + + async with pending_lock: + if ( + len(pending_requests) != 1 + or pending_requests[0] is not request_state + or not request_state.request_text + or request_state.replay_count >= 1 + or request_state.response_id is not None + or not request_state.awaiting_response_created + or request_state.response_event_count != 0 + or request_state.last_downstream_sequence_number is not None + or request_state.downstream_visible + or request_state.upstream_model_output_seen + ): + return False + pending_requests.popleft() + request_state.replay_count += 1 + request_state.preferred_account_id = account_id + request_state.dispatch_absent_replay_account_id = account_id + return True + + @contextmanager def _websocket_archive_request_context(request_id: str | None) -> Iterator[None]: token = set_request_id(request_id) @@ -1370,6 +1401,19 @@ async def retire_current_upstream() -> None: key: value for key, value in filtered_headers.items() if key.lower() != "x-codex-turn-state" } connect_headers = _facade()._headers_with_turn_state(filtered_headers, upstream_turn_state) + retained_connect_kwargs: dict[str, Any] = {} + if ( + account is not None + and account_lease is not None + and request_state.dispatch_absent_replay_account_id == account.id + ): + # This replay already owns the account's stream slot. + # Hand that exact lease to reconnect selection so a + # competing request cannot take it between sockets. + retained_connect_kwargs = { + "retained_account": account, + "retained_stream_lease": account_lease, + } account, upstream = await proxy._connect_proxy_websocket( connect_headers, sticky_key=request_affinity.selection_key, @@ -1384,7 +1428,10 @@ async def retire_current_upstream() -> None: api_key=api_key, client_send_lock=client_send_lock, websocket=websocket, + **retained_connect_kwargs, ) + if retained_connect_kwargs: + account_lease = None if upstream is None or account is None: proxy._cancel_request_state_api_key_reservation_heartbeat(request_state) if request_state_registered: @@ -1471,6 +1518,8 @@ async def retire_current_upstream() -> None: archive_request_id = None if request_state is None else request_state.archive_request_id with _websocket_archive_request_context(archive_request_id): await upstream.send_text(text_data) + if request_state is not None: + request_state.dispatch_absent_replay_account_id = None elif bytes_data is not None: archive_request_id = None if request_state is None else request_state.archive_request_id with _websocket_archive_request_context(archive_request_id): @@ -1499,6 +1548,74 @@ async def retire_current_upstream() -> None: ) continue except UpstreamWebSocketTransportError as exc: + if ( + isinstance(exc, UpstreamWebSocketSendNotDispatchedError) + and request_state is not None + and account is not None + and await _claim_websocket_send_not_dispatched_replay( + request_state, + pending_requests, + pending_lock=pending_lock, + account_id=account.id, + ) + ): + _facade().logger.info( + "Transparent websocket replay after closed-before-send proof request_id=%s", + request_state.request_log_id or request_state.request_id, + ) + replay_request_state = request_state + if upstream_control is not None: + upstream_control.reconnect_requested = True + if upstream_reader is not None: + reader_cancelled = await _facade()._await_cancelled_task( + upstream_reader, + label="proxy websocket upstream reader", + ) + if not reader_cancelled: + _facade().logger.warning( + "Refusing websocket replay because the retired upstream reader " + "did not stop request_id=%s", + request_state.request_log_id or request_state.request_id, + ) + replay_request_state = None + await proxy._fail_pending_websocket_requests( + account=account, + account_id_value=account.id, + pending_requests=deque([request_state]), + pending_lock=anyio.Lock(), + error_code="upstream_reader_cancellation_failed", + error_message="Upstream websocket reader did not stop before replay", + api_key=api_key, + websocket=websocket, + client_send_lock=client_send_lock, + response_create_gate=response_create_gate, + downstream_activity=downstream_activity, + penalize_account=False, + ) + downstream_activity.mark_disconnected() + try: + await websocket.close( + code=1011, + reason="upstream reader did not stop before replay", + ) + except Exception: + _facade().logger.debug( + "Failed to close downstream websocket after reader cancellation failure", + exc_info=True, + ) + break + upstream_reader = None + upstream_control = None + if upstream is not None: + try: + await upstream.close() + except Exception: + _facade().logger.debug( + "Failed to close closed-before-send upstream websocket", + exc_info=True, + ) + upstream = None + continue # 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. @@ -2012,6 +2129,8 @@ async def _connect_proxy_websocket( downstream_activity: _DownstreamWebSocketActivity | None = None, reallocate_sticky: bool = False, sticky_max_age_seconds: int | None = None, + retained_account: Account | None = None, + retained_stream_lease: AccountLease | None = None, ) -> tuple[Account | None, UpstreamWebSocket | None]: proxy = cast(_WebSocketServiceProtocol, self) _ = proxy @@ -2041,6 +2160,7 @@ async def _connect_proxy_websocket( is_retry = attempt > 0 forced_refresh_account_id = request_state.force_refresh_account_id preferred_account_id = forced_refresh_account_id or request_state.preferred_account_id + dispatch_absent_owner_required = request_state.dispatch_absent_replay_account_id is not None turn_state_owner_required = ( request_state.affinity_policy.codex_session_source == "turn_state" and request_state.preferred_account_id is not None @@ -2049,33 +2169,40 @@ async def _connect_proxy_websocket( (request_state.previous_response_id is not None and request_state.preferred_account_id is not None) or request_state.file_required_preferred_account or turn_state_owner_required + or dispatch_absent_owner_required ) - try: - account = await proxy._select_websocket_connect_account( - deadline, - sticky_key=sticky_key, - sticky_kind=sticky_kind, - prefer_earlier_reset=prefer_earlier_reset, - prefer_earlier_reset_window=prefer_earlier_reset_window, - routing_strategy=routing_strategy, - model=model, - request_state=request_state, - api_key=api_key, - client_send_lock=client_send_lock, - websocket=websocket, - downstream_activity=downstream_activity, - reallocate_sticky=True if is_retry else reallocate_sticky, - sticky_max_age_seconds=sticky_max_age_seconds, - exclude_account_ids=excluded_account_ids, - preferred_account_id=preferred_account_id, - require_security_work_authorized=request_state.require_security_work_authorized, - require_preferred_account=require_preferred_account, - defer_no_account_error=last_failover_exc is not None and not require_preferred_account, - ) - except _WebSocketConnectFailureEmitted: - return None, None - selected_stream_lease = request_state.websocket_stream_lease - request_state.websocket_stream_lease = None + if attempt == 0 and retained_account is not None and retained_stream_lease is not None: + account = retained_account + selected_stream_lease = retained_stream_lease + retained_account = None + retained_stream_lease = None + else: + try: + account = await proxy._select_websocket_connect_account( + deadline, + sticky_key=sticky_key, + sticky_kind=sticky_kind, + prefer_earlier_reset=prefer_earlier_reset, + prefer_earlier_reset_window=prefer_earlier_reset_window, + routing_strategy=routing_strategy, + model=model, + request_state=request_state, + api_key=api_key, + client_send_lock=client_send_lock, + websocket=websocket, + downstream_activity=downstream_activity, + reallocate_sticky=True if is_retry else reallocate_sticky, + sticky_max_age_seconds=sticky_max_age_seconds, + exclude_account_ids=excluded_account_ids, + preferred_account_id=preferred_account_id, + require_security_work_authorized=request_state.require_security_work_authorized, + require_preferred_account=require_preferred_account, + defer_no_account_error=last_failover_exc is not None and not require_preferred_account, + ) + except _WebSocketConnectFailureEmitted: + return None, None + selected_stream_lease = request_state.websocket_stream_lease + request_state.websocket_stream_lease = None if account is None: await proxy._load_balancer.release_account_lease(selected_stream_lease) if ( diff --git a/openspec/changes/retry-missing-response-created-once/.openspec.yaml b/openspec/changes/retry-missing-response-created-once/.openspec.yaml new file mode 100644 index 0000000000..3a038210f2 --- /dev/null +++ b/openspec/changes/retry-missing-response-created-once/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-25 diff --git a/openspec/changes/retry-missing-response-created-once/design.md b/openspec/changes/retry-missing-response-created-once/design.md new file mode 100644 index 0000000000..f812a78825 --- /dev/null +++ b/openspec/changes/retry-missing-response-created-once/design.md @@ -0,0 +1,117 @@ +## Context + +The HTTP Responses bridge records the monotonic time of each actual +`response.create` send. If the request still owns the response-create gate and +has no `response.created`, matched `response.*` event, response id, downstream +sequence, or visible output, the upstream reader currently fails and retires +the session after `min(stuck_gate_threshold, 240 seconds)`. + +The bridge already has a bounded `_retry_http_bridge_precreated_request` path. +It permits at most one replay, rejects ambiguous response progress, preserves +hard account ownership, protects account-scoped file references, and only +strips a continuation anchor when the proxy retained a fingerprint-safe full +resend body. + +## Goals / Non-Goals + +**Goals:** + +- Recover the production eventless failure before the client-safe timeout. +- Reuse the existing replay and ownership rules instead of creating another + retry policy. +- Settle and retire exactly once when recovery is not safe or does not work. +- Keep missing acknowledgement neutral to account health. + +**Non-Goals:** + +- Recover streams that have matched any `response.*` lifecycle event. +- Add durable cooldown or poison state across requests or replicas. +- Retry more than once, extend the original request budget, or change public + response framing. + +## Decisions + +### 1. Use a 30-second acknowledgement window + +The eventless watchdog uses +`min(http_responses_session_bridge_stuck_gate_retire_after_seconds, 30 +seconds)`, measured from the current send. Normal production TTFT is generally +sub-second to low-single-digit seconds; 30 seconds leaves margin for transient +startup delay while removing the four-minute dead period. Each real resend +replaces the timestamp, so a replay gets one fresh acknowledgement window +without extending the original request budget. + +### 2. Replay through the existing pre-created helper + +After eligibility is rechecked under lifecycle and pending-state locks, the +reader cancels the old socket receive task and invokes the existing pre-created +replay helper only when the eventless owner is the session's sole pending +request. A successful reconnect/resend returns control to the same reader loop, +which waits on the replacement socket while the downstream request stays open. + +Cancellation is conditional on the receive task still being pending. If +`response.created` or another upstream result wins that race, the reader leaves +the completed task intact and processes it through the normal event path +instead of replaying. + +The helper's existing `replay_count` bound makes this a single recovery +attempt. This timeout recovery reconnects on the current account because the +request still holds that account's response-create concurrency lease. +Continuations are replayed only from an explicitly retained retry-safe +full-resend body, and file ownership continues to require the preferred +account. + +### 3. Retire only after recovery is unavailable or exhausted + +If the helper declines replay, reconnect/resend fails, or the replacement send +also reaches the deadline, the reader applies the existing +`missing_response_created_timeout` overrides, records the stuck-retirement +metric and terminal log, settles pending requests, and retires the bridge. +Neither the retry nor terminal path marks the account unhealthy solely because +the acknowledgement was missing. + +## Failure Modes + +- **A warm socket is already closed before the next send begins.** The transport + adapter returns a sealed not-dispatched proof without calling its send + primitive. The HTTP bridge or direct WebSocket proxy reconnects once on the + same leased account and sends the exact request, including a continuation + anchor, because upstream could not have accepted the first attempt. +- **The socket closes during send.** Dispatch is ambiguous, so the existing + fail-closed 502 remains authoritative and no internal resend occurs. +- **The original send was accepted but its acknowledgement was lost.** Closing + the old socket discards any later output. Because no response lifecycle or + downstream-visible output was observed, client-side tools or other + downstream effects have not run; the bounded replay may spend extra upstream + compute but does not duplicate downstream effects. +- **The acknowledgement completes while cancellation begins.** The completed + receive remains authoritative and is processed normally; no replay occurs. +- **The relay is cancelled while awaiting child cancellation.** The relay's + own cancellation remains authoritative and propagates; a closed session + cannot reconnect or resend after ownership and leases are released. +- **The request is continuity- or file-bound without safe replay evidence.** + The existing helper declines replay and the request fails closed at 30 + seconds. +- **Another request is pending on the same socket.** Reconnecting could orphan + that sibling's response, so the proxy skips replay and retains the existing + whole-session terminal cleanup. +- **The reconnect or resend fails.** Existing typed retry errors are preserved + and the bridge is settled and retired exactly once. +- **The replacement socket also stays silent.** `replay_count` blocks a second + replay; terminal cleanup runs after the replacement's 30-second window. + +## Example + +A request sends at monotonic time 1,000 and receives no matched response event. +At 1,030 the reader cancels the old receive and safely resends once on a fresh +same-account socket. If the original receive completes during cancellation, +that result is processed and no replay occurs. If `response.created` arrives +from the replacement at 1,032, the original downstream stream continues +normally. If the fresh socket is still eventless at 1,060, the proxy returns +the existing explicit timeout and retires the bridge. + +A later compacted continuation finds its warm socket already closed. The +adapter rejects the frame before invoking the socket send operation, the bridge +reconnects once to the same account, and the original downstream stream +continues without a client-visible reconnect. If the socket instead fails while +the send operation is in progress, the bridge does not replay. diff --git a/openspec/changes/retry-missing-response-created-once/proposal.md b/openspec/changes/retry-missing-response-created-once/proposal.md new file mode 100644 index 0000000000..8f54c593bb --- /dev/null +++ b/openspec/changes/retry-missing-response-created-once/proposal.md @@ -0,0 +1,51 @@ +## Why + +Current `main` bounds an HTTP bridge request that receives no +`response.created` acknowledgement, but only after 240 seconds and by failing +the client request. Production evidence on issue #1393 shows that an immediate +fresh attempt commonly succeeds, so the proxy exposes a long avoidable failure +instead of using its existing pre-visible replay path. + +## What Changes + +- Reduce the eventless pre-`response.created` watchdog cap from 240 seconds to + 30 seconds. +- Detect a warm upstream socket that is already closed before a new HTTP-bridge + or direct-WebSocket send starts and transparently reconnect/resend once + without exposing a client reconnect. +- On the first eventless timeout, cancel the old receive wait and attempt one + replay through the existing pre-created replay guards and fresh-socket + reconnect path. +- Process an upstream event that wins the receive-cancellation race instead of + discarding it, and keep the replay on the account whose concurrency lease the + request already holds. +- Continue the original downstream stream when replay succeeds. +- Preserve the current account-neutral terminal settlement and whole-session + retirement when replay is unsafe, reconnect/resend fails, or the replay also + misses `response.created`. +- Keep hard-affinity and file-backed work on its required account and retain the + existing no-replay boundary after response lifecycle or downstream-visible + progress. +- Keep send errors after dispatch may have begun fail closed; only the + adapter's sealed pre-dispatch proof permits an immediate exact resend. + +## Capabilities + +### Modified Capabilities + +- `proxy-admission-control`: Recover one safely replayable eventless gate owner + before retiring the bridge. +- `responses-api-compat`: Keep the retry transparent and bounded before any + response lifecycle or downstream-visible output. + +## Impact + +- Affected code: websocket send classification, HTTP bridge and direct + WebSocket request submission, eventless deadline, and upstream-reader timeout + handling. +- Affected surface: streaming Responses requests served through the HTTP bridge + and direct WebSocket proxy. +- No new setting, dependency, endpoint, schema, migration, account-health + penalty, or durable coordinator. +- This partially addresses #1393. Cross-request cooldown and eventful + missing-created recovery remain separate work. diff --git a/openspec/changes/retry-missing-response-created-once/specs/proxy-admission-control/spec.md b/openspec/changes/retry-missing-response-created-once/specs/proxy-admission-control/spec.md new file mode 100644 index 0000000000..7d10e62dd3 --- /dev/null +++ b/openspec/changes/retry-missing-response-created-once/specs/proxy-admission-control/spec.md @@ -0,0 +1,208 @@ +## MODIFIED Requirements + +### Requirement: Stuck HTTP bridge response-create gate sessions are retired + +The proxy MUST retain the existing waiter-triggered retirement behavior for +stale HTTP bridge response-create gate owners and MUST additionally enforce an +owner-side deadline for a visible HTTP request whose current upstream +`response.create` send remains completely eventless before `response.created`. +The owner-side deadline MUST be measured from a monotonic timestamp recorded +immediately before the current upstream send, MUST use the smaller of the +configured stuck-gate retirement threshold and 30 seconds, MUST run without a +second gate waiter, and MUST remain active when periodic SSE keepalives are +disabled. + +The owner-side watchdog MUST apply only while the request owns the +response-create gate, awaits `response.created`, has neither a response id nor +recorded `response.created` latency, has received no matched `response.*` +lifecycle event, and has produced no downstream-visible output or sequence +evidence. Non-response telemetry such as `codex.rate_limits` MUST NOT suppress +this watchdog. Any matched `response.*` lifecycle event, response-created +milestone, or downstream-visible evidence MUST suppress the owner-side +watchdog and leave existing timeout behavior unchanged. + +Before invoking an upstream websocket send primitive, each supported websocket +adapter MUST check whether its connection is already closed. A closed adapter +MUST return an unforgeable not-dispatched result and MUST NOT invoke the send +primitive. When the HTTP bridge receives that result before any response +lifecycle or downstream-visible evidence, it MUST reconnect and send the exact +request at most once on the same leased account. This recovery MAY preserve a +client continuation anchor because the not-dispatched result proves that the +upstream did not receive the failed attempt. The replacement attempt MUST +remain within the original request budget and existing replay limit. + +The direct WebSocket proxy MUST apply the same closed-before-send recovery only +when the undispatched request is the socket's sole pending request. It MUST +remove that request from the retired reader before cancellation, retain its +response-create admission and account lease, require the same account for the +replacement connection, preserve the exact request including any continuation +anchor, and re-register it on the replacement reader. A sibling request, +response lifecycle evidence, downstream-visible output, or an exhausted replay +count MUST suppress this recovery and retain existing terminal settlement. +If cancellation of the retired reader cannot be confirmed, the proxy MUST fail +the claimed request, close the downstream socket, retain cleanup ownership of +the old reader, and MUST NOT open a replacement connection. + +Any failure raised after the adapter invokes its send primitive remains +dispatch-ambiguous and MUST retain the existing fail-closed behavior without an +internal resend. + +When the first owner-side deadline expires, the proxy MUST recheck eligibility, +cancel the stale receive wait, and attempt one transparent replay only through +the existing pre-created replay safety and ownership rules and only when the +eventless owner is the session's sole pending request. If the receive completes +while cancellation is attempted, the proxy MUST process that result and MUST +NOT replay the request. A replay MUST use the account whose response-create +concurrency lease the request holds. Hard-affinity work MUST remain on the +required account, account-scoped file ownership MUST be preserved, and a +continuation MUST be replayed only from an explicitly retained retry-safe +full-resend body. The retry MUST NOT extend the original request budget or mark +the selected account unhealthy solely because `response.created` was missing. + +If replay is unsafe, reconnect/resend fails, or the replacement send reaches +the deadline, the proxy MUST emit a structured low-cardinality timeout log and +the existing stuck-retirement Prometheus counter, terminally settle every +pending request exactly once, and retire the whole bridge session. It MUST NOT +attempt a second replay. Cancellation of the relay owner or closure of the +session MUST suppress reconnect and resend even when that cancellation is +observed while awaiting the stale receive task's cancellation. + +#### Scenario: Lone eventless gate owner recovers on a fresh socket + +- **GIVEN** a visible HTTP bridge request owns the response-create gate +- **AND** its current send has no matched `response.*` event, response id, or + downstream-visible output +- **WHEN** the smaller of the configured stuck threshold and 30 seconds elapses +- **THEN** the proxy cancels the stale receive and safely replays the request at + most once on a fresh upstream socket +- **AND** a successful replay continues the original downstream stream + +#### Scenario: Closed warm socket recovers before dispatch + +- **GIVEN** a warm bridge socket closed normally before a compacted + continuation starts sending +- **WHEN** the adapter checks the connection before invoking its send primitive +- **THEN** it returns the sealed not-dispatched result without sending +- **AND** the bridge reconnects once on the same leased account +- **AND** it sends the exact continuation through the original downstream + stream without requiring a client reconnect + +#### Scenario: Mid-send failure remains ambiguous + +- **GIVEN** an upstream socket appears open before send +- **WHEN** its send primitive raises after dispatch may have begun +- **THEN** the adapter returns the existing ambiguous transport failure +- **AND** the bridge does not reconnect and resend internally + +#### Scenario: Direct WebSocket continuation recovers before dispatch + +- **GIVEN** a direct WebSocket continuation is the sole pending request +- **AND** its warm upstream socket is already closed before send dispatch +- **WHEN** the adapter returns the sealed not-dispatched result +- **THEN** the proxy removes the request from the retired reader +- **AND** it reconnects once to the same leased account +- **AND** it resends the exact continuation through the existing downstream + WebSocket + +#### Scenario: Direct WebSocket closed-before-send recovery is bounded + +- **GIVEN** a direct WebSocket request already consumed its exact resend +- **WHEN** the replacement socket is also closed before dispatch +- **THEN** the proxy does not attempt another reconnect and resend +- **AND** it applies existing terminal settlement + +#### Scenario: Direct WebSocket reader cancellation failure suppresses replay + +- **GIVEN** a direct WebSocket request is claimed for closed-before-send + recovery +- **WHEN** cancellation of the retired upstream reader cannot be confirmed +- **THEN** the proxy terminally fails the claimed request and closes downstream +- **AND** it retains the old reader for cleanup +- **AND** it does not open a replacement connection + +#### Scenario: Completed receive wins cancellation + +- **GIVEN** an eventless gate owner reaches its deadline +- **WHEN** its pending upstream receive completes as cancellation is attempted +- **THEN** the proxy processes the completed receive through the normal path +- **AND** it does not replay the request + +#### Scenario: Relay shutdown wins receive cancellation + +- **GIVEN** an eventless gate owner reaches its deadline +- **AND** bridge shutdown cancels the relay while it awaits stale receive + cancellation +- **THEN** the relay propagates its own cancellation +- **AND** it does not reconnect or resend after session ownership is released + +#### Scenario: A pending sibling prevents socket replacement + +- **GIVEN** an eventless gate owner reaches its deadline +- **AND** another request is still pending on the same upstream socket +- **WHEN** recovery is evaluated +- **THEN** the proxy does not replace the socket for a transparent replay +- **AND** it retains the existing whole-session terminal settlement + +#### Scenario: Send time rather than request age anchors each deadline + +- **GIVEN** a request spends most of its budget waiting for admission +- **WHEN** the original request or its one replay sends `response.create` +- **THEN** the owner-side deadline begins from that current send +- **AND** prior admission time or the prior attempt does not make the send + immediately stale + +#### Scenario: Leading telemetry does not mask an eventless owner + +- **GIVEN** a pre-created gate owner receives `codex.rate_limits` but no matched + `response.*` lifecycle event +- **WHEN** the owner-side deadline elapses +- **THEN** the telemetry does not refresh or suppress the deadline +- **AND** the proxy applies the same one-replay policy + +#### Scenario: Eventless retry keeps its leased account + +- **GIVEN** an eventless gate owner holds an account-scoped response-create + lease +- **WHEN** the owner is safely replayed +- **THEN** the fresh socket uses the same account +- **AND** the resend does not bypass per-account concurrency admission + +#### Scenario: Response lifecycle evidence suppresses the narrow watchdog + +- **GIVEN** a pre-created request receives any matched `response.*` lifecycle + event, response id, recorded `response.created` latency, or + downstream-visible output +- **WHEN** the eventless owner-side deadline would otherwise elapse +- **THEN** this watchdog does not reconnect or retire the session +- **AND** existing stream, request-budget, and waiter-triggered behavior remains + authoritative + +#### Scenario: Unsafe or exhausted recovery fails closed + +- **GIVEN** an eventless pre-created owner reaches the owner-side deadline +- **AND** safe replay is unavailable, fails, or has already been attempted +- **WHEN** terminal cleanup runs +- **THEN** every pending request is settled exactly once and the whole session + is retired +- **AND** the selected account is not marked unhealthy solely because + `response.created` was missing +- **AND** no second replay is attempted + +#### Scenario: Old pending work blocks a visible gate waiter + +- **WHEN** a visible HTTP bridge request receives + `response_create_gate_timeout` +- **AND** at least one visible pending request on the same session is older than + the configured stuck-gate retirement threshold +- **THEN** the proxy retires the bridge session so later requests can create a + fresh session +- **AND** the waiter is rejected cleanly with `response_create_gate_timeout` + +#### Scenario: Healthy active stream is not retired during a normal wait + +- **WHEN** a visible HTTP bridge request times out waiting for the gate +- **AND** the session has no pending visible request older than the configured + stuck-gate retirement threshold +- **THEN** the proxy rejects only the waiter +- **AND** the bridge session remains available for the existing in-flight + request diff --git a/openspec/changes/retry-missing-response-created-once/specs/responses-api-compat/spec.md b/openspec/changes/retry-missing-response-created-once/specs/responses-api-compat/spec.md new file mode 100644 index 0000000000..1b111b2014 --- /dev/null +++ b/openspec/changes/retry-missing-response-created-once/specs/responses-api-compat/spec.md @@ -0,0 +1,147 @@ +## MODIFIED Requirements + +### Requirement: Failed precreated HTTP bridge replay retires stale sessions + +When an HTTP bridge request is still pending before upstream +`response.completed` and the upstream websocket closes or times out before the +pending request can be completed, the service MUST fail the pending request +terminally and retire the affected bridge session if precreated replay does not +reconnect and resend successfully. + +For an eventless `response.create` that reaches the owner-side +missing-`response.created` deadline, the service MUST cancel the old receive +wait and MAY transparently replay once only when the existing pre-created +replay predicate proves there is no matched response lifecycle, upstream model +output, downstream sequence, or downstream-visible output and the eventless +owner is the session's sole pending request. The service MUST process an +upstream receive result that completes while cancellation is attempted and +MUST NOT replay that accepted request. The replay MUST remain on the account +whose response-create concurrency lease the request holds and MUST preserve +hard-affinity and account-scoped file ownership. A request carrying +`previous_response_id` MUST fail closed unless the proxy retained an +explicitly retry-safe full-resend body that can be replayed without the anchor. + +If that replay succeeds, the original downstream stream MUST continue without +a terminal event. If replay is ineligible, reconnect/resend fails, or the +replacement send also times out, the service MUST terminally settle the request +and retire the session exactly once without an account-health penalty for the +missing acknowledgement. + +Separately, when a warm websocket is proven closed before the adapter invokes +its send primitive, both the HTTP bridge and direct WebSocket proxy MAY +reconnect and send the exact request once on the same leased account, including +a request carrying `previous_response_id`. A successful replacement send MUST +continue the original downstream stream without requiring a client retry. Send +failures without that pre-dispatch proof MUST retain the existing fail-closed +behavior. + +#### Scenario: Compacted continuation recovers from a closed warm socket + +- **GIVEN** a compacted continuation carries `previous_response_id` +- **AND** its warm upstream websocket is already closed before send dispatch +- **WHEN** the transport adapter returns the sealed not-dispatched proof +- **THEN** the service reconnects once on the same leased account +- **AND** it sends the exact continuation through the original downstream + stream +- **AND** the client does not need to reconnect + +#### Scenario: Ambiguous send failure does not replay + +- **GIVEN** an HTTP bridge request enters the upstream send primitive +- **WHEN** that operation fails without proof that dispatch was absent +- **THEN** the service returns the existing terminal transport failure +- **AND** it does not resend the request internally + +#### Scenario: First eventless timeout recovers transparently + +- **GIVEN** an HTTP bridge request has no response lifecycle or visible progress +- **AND** its first `response.create` reaches the missing-created deadline +- **WHEN** the existing replay guards accept the request and reconnect/resend + succeeds +- **THEN** the service continues reading the replacement upstream socket +- **AND** the downstream stream receives no terminal failure for the first + timeout + +#### Scenario: Response acknowledgement wins the cancellation race + +- **GIVEN** an eventless HTTP bridge request reaches its first deadline +- **WHEN** the pending upstream receive completes as cancellation is attempted +- **THEN** the service processes that completed result through the normal event + path +- **AND** it does not replay the request + +#### Scenario: Continuity without a safe full resend is not replayed + +- **GIVEN** an eventless HTTP bridge request carries `previous_response_id` +- **AND** the proxy has no explicitly retry-safe full-resend body +- **WHEN** the missing-created deadline elapses +- **THEN** the service does not replay the continuation +- **AND** it terminally settles the request and retires the bridge + +#### Scenario: Replay preserves hard account and file ownership + +- **GIVEN** an eventless request has hard affinity or an account-scoped file +- **WHEN** it is eligible for the one transparent replay +- **THEN** the replacement connection uses the required owner account +- **AND** the file reference is not moved to an account that does not own it + +#### Scenario: Replay preserves the account concurrency lease + +- **GIVEN** an eventless request holds an account-scoped response-create lease +- **WHEN** it is eligible for the one transparent replay +- **THEN** the replacement socket uses that same account +- **AND** the resend does not consume capacity on an unleased account + +#### Scenario: Pending sibling blocks transparent replay + +- **GIVEN** an eventless request shares its upstream socket with another pending + request +- **WHEN** the missing-created deadline elapses +- **THEN** the service does not replace the shared socket for replay +- **AND** it terminally settles the stale bridge through existing cleanup + +#### Scenario: Replacement timeout is terminal + +- **GIVEN** an eventless request was replayed once on a fresh upstream socket +- **WHEN** the replacement send also misses `response.created` +- **THEN** the service does not replay again +- **AND** it emits one terminal failure and retires the bridge session + +#### Scenario: Precreated replay fails after upstream disconnect + +- **WHEN** an HTTP bridge request is pending before `response.completed` +- **AND** the upstream websocket closes before the request completes +- **AND** precreated replay fails to reconnect and resend the request +- **THEN** the pending request is removed from the bridge queue +- **AND** the per-session response-create gate is released +- **AND** the bridge session is closed and removed from local reuse +- **AND** the terminal error preserves the original failure code such as + `stream_incomplete` or `upstream_request_timeout` + +#### Scenario: Terminal logging failure does not preserve stale bridge ownership + +- **WHEN** a failed pending HTTP bridge request is being logged as terminal +- **AND** request-log writing fails +- **THEN** the service still removes the stale bridge session from local reuse +- **AND** the service releases any durable bridge ownership for that stale + session + +#### Scenario: Concurrent waiter cannot submit on retired stale bridge + +- **WHEN** an HTTP bridge request is waiting on a session response-create gate +- **AND** the upstream reader retires that same bridge session after a failed + precreated replay +- **THEN** the waiting request or prewarm is rejected before it is appended to + pending requests or sent upstream +- **AND** the retired bridge session remains closed and removed from local reuse +- **AND** the post-admission ownership check, pending enqueue, and upstream send + are mutually exclusive with stale-session retirement + +#### Scenario: Unregistered stale bridge reference cannot submit after admission + +- **WHEN** an HTTP bridge request or prewarm holds a stale bridge session + reference +- **AND** that bridge session is no longer the registered local owner for its + session key +- **THEN** the request is rejected after response-create gate admission and + before it is appended or sent upstream diff --git a/openspec/changes/retry-missing-response-created-once/tasks.md b/openspec/changes/retry-missing-response-created-once/tasks.md new file mode 100644 index 0000000000..295423e4c7 --- /dev/null +++ b/openspec/changes/retry-missing-response-created-once/tasks.md @@ -0,0 +1,35 @@ +## 1. Specification + +- [x] 1.1 Define the bounded eventless retry and terminal fallback contracts. +- [x] 1.2 Strictly validate the OpenSpec change. + +## 2. Implementation + +- [x] 2.1 Reduce the eventless acknowledgement cap to 30 seconds. +- [x] 2.2 Cancel the stale receive and invoke one existing safe pre-created + replay before terminal settlement. +- [x] 2.3 Preserve the leased account, hard affinity, file ownership, request + budget, account neutrality, and whole-session retirement on exhaustion. +- [x] 2.4 Preserve and process a receive result that wins the cancellation race. +- [x] 2.5 Propagate relay-owner cancellation and suppress replay after session + closure. +- [x] 2.6 Seal the transport proof for a socket closed before dispatch and + transparently retry that exact request once on the leased account. +- [x] 2.7 Apply the same bounded exact resend to the direct WebSocket proxy + while preserving sole-owner, admission, and account-lease constraints. + +## 3. Verification + +- [x] 3.1 Add regressions for first-timeout recovery, telemetry-only silence, + cancellation-race acknowledgement, leased-account routing, unsafe replay, + second-timeout settlement, and relay shutdown during child cancellation. +- [x] 3.2 Run focused bridge tests, lint, format, type, architecture, and strict + OpenSpec validation. +- [x] 3.3 Review the final diff for replay widening, duplicate settlement, + affinity movement, account penalties, and unrelated edits. +- [x] 3.4 Add adapter construction/dispatch proofs and an externally visible + compacted-continuation regression for a closed warm socket. +- [x] 3.5 Add direct WebSocket regression coverage for transparent + closed-before-send continuation recovery. +- [x] 3.6 Prove a retired direct WebSocket reader that does not confirm + cancellation suppresses replay and is retained for cleanup. diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index c2139f2e64..deb682d8e8 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -21,6 +21,7 @@ import app.modules.proxy.load_balancer as load_balancer_module import app.modules.proxy.service as proxy_module +from app.core.clients import proxy_websocket as proxy_websocket_module from app.core.config.settings import Settings from app.core.openai.model_registry import ModelRegistry from app.core.utils.request_id import ( @@ -532,6 +533,34 @@ async def send_text(self, text: str) -> None: self.sent_text.append(text) +class _WarmThenClosedBeforeSendUpstreamWebSocket(_FakeBridgeUpstreamWebSocket): + def __init__(self) -> None: + super().__init__("resp_warm_socket") + self.closed_before_send = False + + async def send_text(self, text: str) -> None: + if self.closed_before_send: + raise proxy_websocket_module._websocket_send_not_dispatched_error() + await super().send_text(text) + + +class _CancellationBlockingUpstreamWebSocket(_SilentUpstreamWebSocket): + def __init__(self) -> None: + super().__init__() + self.receive_started = asyncio.Event() + self.receive_cancellation_started = asyncio.Event() + + async def receive(self) -> _FakeUpstreamMessage: + self.receive_started.set() + try: + await asyncio.Event().wait() + raise AssertionError("unreachable receive wait completed") + except asyncio.CancelledError: + self.receive_cancellation_started.set() + await asyncio.Event().wait() + raise + + class _RecordingUpstreamWebSocket(_FakeBridgeUpstreamWebSocket): pass @@ -8094,15 +8123,20 @@ async def fake_connect_responses_websocket( @pytest.mark.asyncio -async def test_v1_responses_http_bridge_retries_once_when_upstream_closes_before_response_created( +@pytest.mark.parametrize("first_failure", ["close", "silent"]) +async def test_v1_responses_http_bridge_retries_once_before_response_created( async_client, monkeypatch, + first_failure, ): _install_bridge_settings(monkeypatch, enabled=True) + proxy_module.get_settings().http_responses_session_bridge_stuck_gate_retire_after_seconds = 0.01 account_id = await _import_account(async_client, "acc_http_bridge_retry", "http-bridge-retry@example.com") account = await _get_account(account_id) - upstreams = [_PrecreatedCloseUpstreamWebSocket(), _FakeBridgeUpstreamWebSocket()] + first_upstream = _PrecreatedCloseUpstreamWebSocket() if first_failure == "close" else _SilentUpstreamWebSocket() + upstreams = [first_upstream, _FakeBridgeUpstreamWebSocket()] connect_count = 0 + preferred_account_ids: list[str | None] = [] async def fake_select_account_with_budget( self, @@ -8123,7 +8157,7 @@ async def fake_select_account_with_budget( api_key=None, preferred_account_id=None, ): - del preferred_account_id + preferred_account_ids.append(preferred_account_id) del ( self, deadline, @@ -8176,6 +8210,182 @@ async def fake_connect_responses_websocket( assert response.status_code == 200 assert connect_count == 2 + assert len(first_upstream.sent_text) == 1 + assert len(upstreams[1].sent_text) == 1 + if first_failure == "silent": + assert preferred_account_ids == [None, account.id] + + +@pytest.mark.asyncio +async def test_http_bridge_shutdown_during_eventless_receive_cancellation_does_not_replay( + app_instance, + monkeypatch, +): + _install_bridge_settings(monkeypatch, enabled=True) + proxy_module.get_settings().http_responses_session_bridge_stuck_gate_retire_after_seconds = 0.01 + service = get_proxy_service_for_app(app_instance) + upstream = _CancellationBlockingUpstreamWebSocket() + gate = asyncio.Semaphore(1) + await gate.acquire() + request_state = proxy_module._WebSocketRequestState( + request_id="req-shutdown-cancel-race", + model="gpt-5.6-sol", + service_tier=None, + reasoning_effort="high", + api_key_reservation=None, + started_at=time.monotonic() - 1.0, + transport="http", + response_create_gate=gate, + response_create_gate_acquired=True, + awaiting_response_created=True, + response_create_sent_at=time.monotonic() - 1.0, + event_queue=asyncio.Queue(), + request_text=json.dumps({"type": "response.create", "model": "gpt-5.6-sol"}), + ) + session = proxy_module._HTTPBridgeSession( + key=proxy_module._HTTPBridgeSessionKey("prompt_cache", "shutdown-cancel-race", None), + headers={}, + affinity=proxy_module._AffinityPolicy(key="shutdown-cancel-race"), + request_model="gpt-5.6-sol", + account=cast(Account, SimpleNamespace(id="acct-shutdown-cancel-race", status=AccountStatus.ACTIVE)), + upstream=cast(proxy_module.UpstreamWebSocket, upstream), + upstream_control=proxy_module._WebSocketUpstreamControl(), + pending_requests=deque([request_state]), + pending_lock=anyio.Lock(), + response_create_gate=gate, + queued_request_count=1, + last_used_at=time.monotonic(), + idle_ttl_seconds=120.0, + ) + retry = AsyncMock(return_value=True) + fail_pending = AsyncMock() + monkeypatch.setattr(service, "_retry_http_bridge_precreated_request", retry) + monkeypatch.setattr(service, "_fail_pending_websocket_requests", fail_pending) + + relay = asyncio.create_task(service._relay_http_bridge_upstream_messages(session)) + session.upstream_reader = relay + await _wait_for_event(upstream.receive_started) + await _wait_for_event(upstream.receive_cancellation_started) + + await asyncio.wait_for(service._close_http_bridge_session(session), timeout=1.0) + + retry.assert_not_awaited() + assert relay.cancelled() + assert session.closed is True + assert upstream.closed is True + + +@pytest.mark.asyncio +async def test_codex_http_bridge_retries_closed_warm_socket_before_send_without_client_reconnect( + async_client, + monkeypatch, +): + _install_bridge_settings(monkeypatch, enabled=True) + account_id = await _import_account( + async_client, + "acc_http_bridge_closed_warm_send", + "http-bridge-closed-warm-send@example.com", + ) + account = await _get_account(account_id) + warm_upstream = _WarmThenClosedBeforeSendUpstreamWebSocket() + replacement_upstream = _FakeBridgeUpstreamWebSocket("resp_replacement_socket") + upstreams = [warm_upstream, replacement_upstream] + connect_count = 0 + preferred_account_ids: list[str | None] = [] + + async def fake_select_account_with_budget( + self, + deadline, + *, + request_id, + kind, + request_stage="first_turn", + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids=None, + additional_limit_name=None, + api_key=None, + preferred_account_id=None, + ): + preferred_account_ids.append(preferred_account_id) + del ( + self, + deadline, + request_id, + kind, + request_stage, + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset_accounts, + routing_strategy, + model, + exclude_account_ids, + additional_limit_name, + api_key, + ) + return AccountSelection(account=account, error_message=None, error_code=None) + + async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds): + del self, force, timeout_seconds + return target + + async def fake_connect_responses_websocket( + headers, + access_token, + account_id_header, + *, + base_url=None, + session=None, + ): + del headers, access_token, account_id_header, base_url, session + nonlocal connect_count + upstream = upstreams[connect_count] + connect_count += 1 + return upstream + + monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget) + monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget) + monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket) + + headers = { + "session_id": "closed-warm-send-session", + "user-agent": "codex_cli_rs/0.145.0", + } + first = await async_client.post( + "/backend-api/codex/responses", + json={ + "model": "gpt-5.6-sol", + "instructions": "Return exactly OK.", + "input": "first turn", + }, + headers=headers, + ) + assert first.status_code == 200 + warm_upstream.closed_before_send = True + + second = await async_client.post( + "/backend-api/codex/responses", + json={ + "model": "gpt-5.6-sol", + "instructions": "Return exactly OK.", + "input": "compacted continuation", + "previous_response_id": "resp_warm_socket_1", + }, + headers=headers, + ) + + assert second.status_code == 200 + assert connect_count == 2 + assert len(warm_upstream.sent_text) == 1 + assert len(replacement_upstream.sent_text) == 1 + assert preferred_account_ids == [None, account.id] @pytest.mark.asyncio diff --git a/tests/integration/test_proxy_websocket_responses.py b/tests/integration/test_proxy_websocket_responses.py index 43c6b876a2..f9898d669c 100644 --- a/tests/integration/test_proxy_websocket_responses.py +++ b/tests/integration/test_proxy_websocket_responses.py @@ -15,9 +15,11 @@ 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 import proxy_websocket as proxy_websocket_module from app.core.utils.request_id import get_request_id from app.modules.proxy._service.websocket import mixin as websocket_mixin_module from app.modules.proxy.affinity import _codex_session_selection_key +from app.modules.proxy.load_balancer import AccountLease pytestmark = pytest.mark.integration @@ -136,6 +138,22 @@ async def send_text(self, text: str) -> None: raise RuntimeError("socket closed during send") +class _WarmClosedBeforeSendUpstreamWebSocket(_SequencedUpstreamWebSocket): + def __init__( + self, + messages: list[_FakeUpstreamMessage], + *, + deferred_message_batches: list[list[_FakeUpstreamMessage]], + ) -> None: + super().__init__(messages, deferred_message_batches=deferred_message_batches) + self.closed_before_send = False + + async def send_text(self, text: str) -> None: + if self.closed_before_send: + raise proxy_websocket_module._websocket_send_not_dispatched_error() + await super().send_text(text) + + class _DelayedUpstreamWebSocket(_FakeUpstreamWebSocket): def __init__(self, messages: list[_FakeUpstreamMessage], *, delays: list[float]) -> None: super().__init__(messages) @@ -7522,6 +7540,207 @@ async def fake_write_request_log(self, **kwargs): assert log_calls[0]["status"] == "error" +@pytest.mark.parametrize( + ("replacement_succeeds", "reader_cancellation_succeeds"), + [(True, True), (False, True), (True, False)], +) +def test_backend_responses_websocket_retries_closed_warm_socket_before_send( + app_instance, + monkeypatch, + replacement_succeeds, + reader_cancellation_succeeds, +): + first_response_id = "resp_ws_closed_warm_first" + second_response_id = "resp_ws_closed_warm_second" + first_upstream = _WarmClosedBeforeSendUpstreamWebSocket( + [], + deferred_message_batches=[ + [ + _FakeUpstreamMessage( + "text", + text=json.dumps( + { + "type": "response.created", + "response": {"id": first_response_id, "status": "in_progress"}, + }, + separators=(",", ":"), + ), + ), + _FakeUpstreamMessage( + "text", + text=json.dumps( + { + "type": "response.completed", + "response": {"id": first_response_id, "status": "completed"}, + }, + separators=(",", ":"), + ), + ), + ] + ], + ) + replacement_upstream = _WarmClosedBeforeSendUpstreamWebSocket( + [], + deferred_message_batches=( + [ + [ + _FakeUpstreamMessage( + "text", + text=json.dumps( + { + "type": "response.created", + "response": {"id": second_response_id, "status": "in_progress"}, + }, + separators=(",", ":"), + ), + ), + _FakeUpstreamMessage( + "text", + text=json.dumps( + { + "type": "response.completed", + "response": {"id": second_response_id, "status": "completed"}, + }, + separators=(",", ":"), + ), + ), + ] + ] + if replacement_succeeds + else [] + ), + ) + replacement_upstream.closed_before_send = not replacement_succeeds + upstreams = [first_upstream, replacement_upstream] + account = SimpleNamespace(id="acct_ws_closed_warm") + stream_lease = AccountLease( + lease_id="lease_ws_closed_warm", + account_id=account.id, + kind="stream", + acquired_at=0.0, + ) + connect_count = 0 + cancellation_failure_injected = False + + class _FakeSettingsCache: + async def get(self): + return _websocket_settings() + + async def allow_firewall(_websocket): + return None + + async def allow_proxy_api_key(_authorization: str | None, *, request: object | None = None): + return None + + async def fake_connect_proxy_websocket( + self, + headers, + *, + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset, + prefer_earlier_reset_window, + routing_strategy, + model, + request_state, + api_key, + client_send_lock, + websocket, + retained_account=None, + retained_stream_lease=None, + ): + del ( + self, + headers, + sticky_key, + sticky_kind, + reallocate_sticky, + sticky_max_age_seconds, + prefer_earlier_reset, + prefer_earlier_reset_window, + routing_strategy, + model, + api_key, + client_send_lock, + websocket, + ) + nonlocal connect_count + if connect_count == 1: + assert request_state.dispatch_absent_replay_account_id == account.id + assert request_state.preferred_account_id == account.id + assert retained_account is account + assert retained_stream_lease is stream_lease + else: + assert retained_account is None + assert retained_stream_lease is None + upstream = upstreams[connect_count] + connect_count += 1 + request_state.websocket_stream_lease = stream_lease + return account, upstream + + monkeypatch.setattr(proxy_api_module, "_websocket_firewall_denial_response", allow_firewall) + monkeypatch.setattr(proxy_api_module, "validate_proxy_api_key_authorization", allow_proxy_api_key) + monkeypatch.setattr(proxy_module, "get_settings_cache", lambda: _FakeSettingsCache()) + monkeypatch.setattr(proxy_module.ProxyService, "_connect_proxy_websocket", fake_connect_proxy_websocket) + original_await_cancelled_task = proxy_module._await_cancelled_task + + async def controlled_await_cancelled_task(task, **kwargs): + nonlocal cancellation_failure_injected + if not reader_cancellation_succeeds and not cancellation_failure_injected: + cancellation_failure_injected = True + task.cancel() + return False + return await original_await_cancelled_task(task, **kwargs) + + monkeypatch.setattr(proxy_module, "_await_cancelled_task", controlled_await_cancelled_task) + + first_request = { + "type": "response.create", + "model": "gpt-5.6-sol", + "instructions": "Return exactly OK.", + "input": [{"role": "user", "content": [{"type": "input_text", "text": "first"}]}], + "stream": True, + } + compacted_continuation = { + "type": "response.create", + "model": "gpt-5.6-sol", + "instructions": "Return exactly OK.", + "input": [{"role": "user", "content": [{"type": "input_text", "text": "compacted"}]}], + "previous_response_id": first_response_id, + "stream": True, + } + + with TestClient(app_instance) as client: + with client.websocket_connect("/backend-api/codex/responses") as websocket: + websocket.send_text(json.dumps(first_request)) + first_events = [json.loads(websocket.receive_text()) for _ in range(2)] + first_upstream.closed_before_send = True + + websocket.send_text(json.dumps(compacted_continuation)) + second_events = [ + json.loads(websocket.receive_text()) + for _ in range(2 if replacement_succeeds and reader_cancellation_succeeds else 1) + ] + + assert [event["type"] for event in first_events] == ["response.created", "response.completed"] + assert connect_count == (2 if reader_cancellation_succeeds else 1) + assert len(first_upstream.sent_text) == 1 + if not reader_cancellation_succeeds: + assert [event["type"] for event in second_events] == ["response.failed"] + assert second_events[0]["response"]["error"]["code"] == "upstream_reader_cancellation_failed" + assert replacement_upstream.sent_text == [] + elif replacement_succeeds: + assert [event["type"] for event in second_events] == ["response.created", "response.completed"] + assert len(replacement_upstream.sent_text) == 1 + assert json.loads(replacement_upstream.sent_text[0])["previous_response_id"] == first_response_id + else: + assert [event["type"] for event in second_events] == ["response.failed"] + assert second_events[0]["response"]["error"]["code"] == "upstream_websocket_closed_before_send" + assert replacement_upstream.sent_text == [] + + def test_backend_responses_websocket_rejects_oversized_response_create_before_upstream( app_instance, monkeypatch, diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 00dd232e8e..49a0f640ca 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -128,6 +128,80 @@ def _make_eventless_http_bridge_owner( ) +class _SilentTrackingUpstream: + def __init__(self, *, leading_telemetry: bool = False) -> None: + self.leading_telemetry = leading_telemetry + self.telemetry_emitted = False + self.blocking_receive_started = asyncio.Event() + self.receive_cancellations = 0 + self.closed = False + self.sent_texts: list[str] = [] + + async def receive(self) -> UpstreamWebSocketMessage: + if self.leading_telemetry and not self.telemetry_emitted: + self.telemetry_emitted = True + return UpstreamWebSocketMessage( + kind="text", + text=json.dumps( + { + "type": "codex.rate_limits", + "plan_type": "pro", + "rate_limits": {"allowed": True, "limit_reached": False}, + }, + separators=(",", ":"), + ), + ) + self.blocking_receive_started.set() + try: + await asyncio.Event().wait() + raise AssertionError("unreachable") + except asyncio.CancelledError: + self.receive_cancellations += 1 + raise + + async def send_text(self, text: str) -> None: + self.sent_texts.append(text) + + async def close(self) -> None: + self.closed = True + + +class _CreatedOnCancelUpstream: + def __init__(self) -> None: + self.receive_started = asyncio.Event() + self.cancel_races = 0 + self.receive_calls = 0 + self.closed = False + + async def receive(self) -> UpstreamWebSocketMessage: + self.receive_calls += 1 + self.receive_started.set() + try: + await asyncio.Event().wait() + raise AssertionError("unreachable") + except asyncio.CancelledError: + if self.receive_calls != 1: + raise + self.cancel_races += 1 + return UpstreamWebSocketMessage( + kind="text", + text=json.dumps( + { + "type": "response.created", + "response": { + "id": "resp-cancel-race", + "object": "response", + "status": "in_progress", + }, + }, + separators=(",", ":"), + ), + ) + + async def close(self) -> None: + self.closed = True + + def test_http_bridge_eventless_precreated_deadline_uses_current_send_and_client_safe_cap() -> None: request_state = _make_eventless_http_bridge_owner() @@ -136,7 +210,7 @@ def test_http_bridge_eventless_precreated_deadline_uses_current_send_and_client_ request_state, stuck_gate_retire_after_seconds=300.0, ) - == 340.0 + == 130.0 ) assert ( http_bridge_helpers_module._http_bridge_eventless_precreated_deadline( @@ -152,7 +226,7 @@ def test_http_bridge_eventless_precreated_deadline_uses_current_send_and_client_ request_state, stuck_gate_retire_after_seconds=300.0, ) - == 340.0 + == 130.0 ) @@ -16336,6 +16410,7 @@ async def test_retry_http_bridge_request_on_fresh_upstream_reconnects_without_re request_state=request_state, restart_reader=True, require_same_account=False, + require_preferred_account=False, ) send_text.assert_not_awaited() @@ -18457,6 +18532,231 @@ async def close(self) -> None: assert "http_bridge_event event=missing_response_created_timeout" in caplog.text +@pytest.mark.asyncio +@pytest.mark.parametrize("leading_telemetry", [False, True], ids=["silent", "leading-telemetry"]) +async def test_http_bridge_eventless_timeout_retries_once_on_fresh_same_account_socket( + monkeypatch: pytest.MonkeyPatch, + leading_telemetry: bool, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + old_upstream = _SilentTrackingUpstream(leading_telemetry=leading_telemetry) + replacement_upstream = _SilentTrackingUpstream() + session = _make_bridge_session(key_value=f"eventless-retry-{leading_telemetry}") + session.upstream = cast(UpstreamWebSocket, old_upstream) + service._http_bridge_sessions[session.key] = session + settings = _make_app_settings( + stream_idle_timeout_seconds=60.0, + http_responses_session_bridge_request_budget_seconds=60.0, + http_responses_session_bridge_stuck_gate_retire_after_seconds=0.1, + ) + monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + record_stuck_retire = Mock() + monkeypatch.setattr(proxy_service, "_record_http_bridge_stuck_retire", record_stuck_retire) + fail_reader = AsyncMock() + monkeypatch.setattr(service, "_fail_http_bridge_reader_and_maybe_retire", fail_reader) + + gate = session.response_create_gate + await gate.acquire() + owner = _make_eventless_http_bridge_owner( + request_id=f"req-eventless-retry-{leading_telemetry}", + sent_at=time.monotonic() if leading_telemetry else time.monotonic() - 1.0, + ) + owner.started_at = time.monotonic() + owner.response_create_gate = gate + owner.request_text = '{"type":"response.create","model":"gpt-5.6-sol","input":"hello"}' + owner.preferred_account_id = session.account.id + owner.account_response_create_lease = cast(Any, object()) + async with session.pending_lock: + session.pending_requests.append(owner) + session.queued_request_count = 1 + + async def reconnect( + target_session: proxy_service._HTTPBridgeSession, + **kwargs: object, + ) -> None: + assert target_session is session + assert kwargs["request_state"] is owner + assert kwargs["require_same_account"] is True + owner.response_create_sent_at = None + target_session.upstream = cast(UpstreamWebSocket, replacement_upstream) + + reconnect_mock = AsyncMock(side_effect=reconnect) + monkeypatch.setattr(service, "_reconnect_http_bridge_session", reconnect_mock) + + reader_task = asyncio.create_task(service._relay_http_bridge_upstream_messages(session)) + try: + await asyncio.wait_for(replacement_upstream.blocking_receive_started.wait(), timeout=1.0) + + reconnect_mock.assert_awaited_once_with( + session, + request_state=owner, + require_same_account=True, + require_preferred_account=True, + ) + assert replacement_upstream.sent_texts == [owner.request_text] + assert old_upstream.receive_cancellations == 1 + assert owner.replay_count == 1 + assert owner.response_create_sent_at is not None + assert owner.failure_detail_override is None + assert list(session.pending_requests) == [owner] + assert session.closed is False + assert owner.event_queue is not None + if leading_telemetry: + telemetry_block = owner.event_queue.get_nowait() + assert telemetry_block is not None + assert '"type":"codex.rate_limits"' in telemetry_block + assert owner.event_queue.empty() is True + fail_reader.assert_not_awaited() + record_stuck_retire.assert_not_called() + finally: + reader_task.cancel() + with pytest.raises(asyncio.CancelledError): + await reader_task + + assert replacement_upstream.receive_cancellations == 1 + + +@pytest.mark.asyncio +async def test_http_bridge_eventless_retry_second_timeout_settles_and_retires_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + old_upstream = _SilentTrackingUpstream() + replacement_upstream = _SilentTrackingUpstream() + session = _make_bridge_session(key_value="eventless-retry-exhausted") + session.upstream = cast(UpstreamWebSocket, old_upstream) + service._http_bridge_sessions[session.key] = session + settings = _make_app_settings( + stream_idle_timeout_seconds=60.0, + http_responses_session_bridge_request_budget_seconds=60.0, + http_responses_session_bridge_stuck_gate_retire_after_seconds=0.02, + ) + monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + monkeypatch.setattr(service, "_handle_stream_error", AsyncMock()) + write_request_log = AsyncMock() + monkeypatch.setattr(service, "_write_request_log", write_request_log) + record_stuck_retire = Mock() + monkeypatch.setattr(proxy_service, "_record_http_bridge_stuck_retire", record_stuck_retire) + original_fail_reader = service._fail_http_bridge_reader_and_maybe_retire + fail_reader = AsyncMock(wraps=original_fail_reader) + monkeypatch.setattr(service, "_fail_http_bridge_reader_and_maybe_retire", fail_reader) + + gate = session.response_create_gate + await gate.acquire() + owner = _make_eventless_http_bridge_owner( + request_id="req-eventless-retry-exhausted", + sent_at=time.monotonic() - 1.0, + ) + owner.started_at = time.monotonic() + owner.response_create_gate = gate + owner.request_text = '{"type":"response.create","model":"gpt-5.6-sol","input":"hello"}' + owner.preferred_account_id = session.account.id + owner.account_response_create_lease = cast(Any, object()) + event_queue = owner.event_queue + assert event_queue is not None + async with session.pending_lock: + session.pending_requests.append(owner) + session.queued_request_count = 1 + + async def reconnect( + target_session: proxy_service._HTTPBridgeSession, + **kwargs: object, + ) -> None: + assert target_session is session + assert kwargs["request_state"] is owner + assert kwargs["require_same_account"] is True + owner.response_create_sent_at = None + target_session.upstream = cast(UpstreamWebSocket, replacement_upstream) + + reconnect_mock = AsyncMock(side_effect=reconnect) + monkeypatch.setattr(service, "_reconnect_http_bridge_session", reconnect_mock) + + await asyncio.wait_for(service._relay_http_bridge_upstream_messages(session), timeout=1.0) + + event_blocks: list[str] = [] + while (event_block := await asyncio.wait_for(event_queue.get(), timeout=0.1)) is not None: + event_blocks.append(event_block) + assert len(event_blocks) == 1 + assert '"code":"upstream_request_timeout"' in event_blocks[0] + assert event_queue.empty() is True + reconnect_mock.assert_awaited_once_with( + session, + request_state=owner, + require_same_account=True, + require_preferred_account=True, + ) + assert replacement_upstream.sent_texts == [owner.request_text] + assert old_upstream.receive_cancellations == 1 + assert replacement_upstream.receive_cancellations == 1 + assert owner.replay_count == 1 + assert owner.failure_detail_override == "missing_response_created_timeout" + assert list(session.pending_requests) == [] + assert session.queued_request_count == 0 + assert session.closed is True + assert session.key not in service._http_bridge_sessions + assert gate.locked() is False + assert write_request_log.await_count == 1 + fail_reader.assert_awaited_once() + assert fail_reader.await_args.kwargs["penalize_account"] is False + record_stuck_retire.assert_called_once_with( + reason="missing_response_created_timeout", + session=session, + ) + + +@pytest.mark.asyncio +async def test_http_bridge_eventless_timeout_processes_response_created_that_wins_cancel_race( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = proxy_service.ProxyService(cast(Any, nullcontext())) + upstream = _CreatedOnCancelUpstream() + session = _make_bridge_session(key_value="eventless-created-cancel-race") + session.upstream = cast(UpstreamWebSocket, upstream) + service._http_bridge_sessions[session.key] = session + settings = _make_app_settings( + stream_idle_timeout_seconds=60.0, + http_responses_session_bridge_request_budget_seconds=60.0, + http_responses_session_bridge_stuck_gate_retire_after_seconds=0.01, + ) + monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + retry_precreated = AsyncMock(return_value=True) + monkeypatch.setattr(service, "_retry_http_bridge_precreated_request", retry_precreated) + + gate = session.response_create_gate + await gate.acquire() + owner = _make_eventless_http_bridge_owner( + request_id="req-eventless-created-cancel-race", + sent_at=time.monotonic() - 1.0, + ) + owner.started_at = time.monotonic() + owner.response_create_gate = gate + owner.request_text = '{"type":"response.create","model":"gpt-5.6-sol","input":"hello"}' + async with session.pending_lock: + session.pending_requests.append(owner) + session.queued_request_count = 1 + + reader_task = asyncio.create_task(service._relay_http_bridge_upstream_messages(session)) + try: + await asyncio.wait_for(upstream.receive_started.wait(), timeout=1.0) + for _ in range(100): + if owner.response_id == "resp-cancel-race": + break + await asyncio.sleep(0.01) + + assert owner.response_id == "resp-cancel-race" + assert owner.response_event_count == 1 + assert owner.awaiting_response_created is False + assert owner.replay_count == 0 + assert upstream.cancel_races == 1 + assert upstream.receive_calls >= 2 + assert gate.locked() is False + retry_precreated.assert_not_awaited() + finally: + reader_task.cancel() + with pytest.raises(asyncio.CancelledError): + await reader_task + + @pytest.mark.asyncio async def test_http_bridge_eventless_timeout_yields_to_locked_send_failure_cleanup( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py index 31910f68ca..61d3f3ded0 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -36569,6 +36569,7 @@ async def capture_send_text(_text: str) -> None: request_state=request_state, restart_reader=True, require_same_account=False, + require_preferred_account=False, ) send_text.assert_awaited_once_with('{"type":"response.create","model":"gpt-5.1","input":"retry"}') assert send_request_ids == ["archive_bridge_retry_fresh"] @@ -36767,6 +36768,7 @@ async def test_retry_http_bridge_precreated_request_keeps_hard_session_owner_bou session, request_state=request_state, require_same_account=True, + require_preferred_account=False, ) assert request_state.preferred_account_id is None assert request_state.excluded_account_ids == set() diff --git a/tests/unit/test_proxy_websocket_client.py b/tests/unit/test_proxy_websocket_client.py index 034d0ca6c2..463be9addb 100644 --- a/tests/unit/test_proxy_websocket_client.py +++ b/tests/unit/test_proxy_websocket_client.py @@ -14,6 +14,7 @@ from websockets.exceptions import ConnectionClosedError, InvalidHandshake, InvalidProxy, InvalidStatus from websockets.frames import Close from websockets.http11 import Response +from websockets.protocol import State import app.core.clients.proxy_websocket as proxy_websocket_module from app.core.clients.codex import CodexTransportError, CodexWebSocketResult @@ -21,6 +22,7 @@ from app.core.clients.proxy_websocket import ( CodexUpstreamWebSocket, RealtimeWebSocketProtocol, + UpstreamWebSocketSendNotDispatchedError, UpstreamWebSocketTransportError, WebsocketsUpstreamWebSocket, connect_live_websocket, @@ -55,6 +57,7 @@ class _FakeConnection: def __init__(self, *, subprotocol: str | None = None) -> None: self.sent: list[str | bytes] = [] self.closed = False + self.state = State.OPEN self.subprotocol = subprotocol async def send(self, data: str | bytes) -> None: @@ -67,6 +70,51 @@ async def close(self, code: int = 1000, reason: str = "") -> None: self.closed = True +@pytest.mark.asyncio +async def test_websockets_adapter_proves_closed_socket_was_not_dispatched() -> None: + connection = _FakeConnection() + connection.state = State.CLOSED + websocket = WebsocketsUpstreamWebSocket(cast(Any, connection)) + + with pytest.raises(UpstreamWebSocketSendNotDispatchedError) as exc_info: + await websocket.send_text('{"type":"response.create"}') + + assert exc_info.value.error_code == "upstream_websocket_closed_before_send" + assert connection.sent == [] + + +@pytest.mark.asyncio +async def test_codex_adapter_proves_closed_socket_was_not_dispatched() -> None: + def unexpected_send(_text: str) -> None: + raise AssertionError("closed websocket must not enter send_str") + + websocket = CodexUpstreamWebSocket( + SimpleNamespace( + closed=True, + send_str=unexpected_send, + ) + ) + + with pytest.raises(UpstreamWebSocketSendNotDispatchedError) as exc_info: + await websocket.send_text('{"type":"response.create"}') + + assert exc_info.value.error_code == "upstream_websocket_closed_before_send" + + +def test_send_not_dispatched_proof_cannot_be_constructed_or_subclassed() -> None: + with pytest.raises(TypeError, match="closed-before-send check"): + UpstreamWebSocketSendNotDispatchedError( + "forged", + error_code="upstream_websocket_closed_before_send", + _constructor_token=object(), + ) + + with pytest.raises(TypeError, match="cannot be subclassed"): + + class _ForgedNotDispatched(UpstreamWebSocketSendNotDispatchedError): + pass + + async def _local_proxy_tunnel_handler( reader: asyncio.StreamReader, writer: asyncio.StreamWriter,