diff --git a/app/modules/proxy/_service/http_bridge/request_submit.py b/app/modules/proxy/_service/http_bridge/request_submit.py index 0319e918ea..e84b7d46c8 100644 --- a/app/modules/proxy/_service/http_bridge/request_submit.py +++ b/app/modules/proxy/_service/http_bridge/request_submit.py @@ -191,6 +191,7 @@ logger = logging.getLogger("app.modules.proxy.service") _REQUEST_TRANSPORT_HTTP = "http" _WEBSOCKET_AUTH_INVALIDATED_FAILURE_CODE = "account_auth_invalidated" +_HTTP_BRIDGE_SAME_ANCHOR_PRECREATED_MAX_REPLAYS = 20 _NO_SECURITY_WORK_AUTHORIZED_ACCOUNTS_CODE = "no_security_work_authorized_accounts" _SECURITY_WORK_NO_AUTHORIZED_ACCOUNTS_MESSAGE = ( "Upstream flagged this request as possible cybersecurity work, but no account is marked as authorized for " @@ -199,6 +200,22 @@ ) +def _http_bridge_can_replay_same_anchor_before_created(request_state: _WebSocketRequestState) -> bool: + if not request_state.request_text: + return False + if request_state.replay_count >= _HTTP_BRIDGE_SAME_ANCHOR_PRECREATED_MAX_REPLAYS: + return False + return ( + request_state.previous_response_id is not None + and request_state.response_id is None + and request_state.awaiting_response_created + and request_state.response_event_count == 0 + and request_state.last_downstream_sequence_number is None + and not request_state.downstream_visible + and not request_state.upstream_model_output_seen + ) + + async def _rollback_http_bridge_recovery_turn_state_registration( service: Any, receipt: DurableBridgeAliasRegistrationReceipt, @@ -1530,6 +1547,7 @@ async def _retry_http_bridge_precreated_request( session: "_HTTPBridgeSession", *, request_state: _WebSocketRequestState | None = None, + allow_same_anchor_before_created: bool = False, ) -> bool: account_neutral_recovery = is_http_bridge_account_neutral_replay( kind=session.key.affinity_kind, @@ -1543,7 +1561,14 @@ async def _retry_http_bridge_precreated_request( or any(pending_request is not request_state for pending_request in session.pending_requests) or request_state.draining_until_terminal or not _http_bridge_request_counts_against_queue(request_state) - or not _websocket_request_can_replay_before_visible_output(request_state) + or not ( + _websocket_request_can_replay_before_visible_output(request_state) + or ( + allow_same_anchor_before_created + and hard_owner_bound + and _http_bridge_can_replay_same_anchor_before_created(request_state) + ) + ) ): return False else: @@ -1551,15 +1576,30 @@ async def _retry_http_bridge_precreated_request( request_state for request_state in session.pending_requests if not request_state.draining_until_terminal - and _websocket_request_can_replay_before_visible_output(request_state) + and ( + _websocket_request_can_replay_before_visible_output(request_state) + or ( + allow_same_anchor_before_created + and hard_owner_bound + and _http_bridge_can_replay_same_anchor_before_created(request_state) + ) + ) ] if len(retryable_requests) != 1: return False request_state = retryable_requests[0] - 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 - and request_state.fresh_upstream_request_text + 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 + and request_state.fresh_upstream_request_text + ) + and not ( + allow_same_anchor_before_created + and hard_owner_bound + and _http_bridge_can_replay_same_anchor_before_created(request_state) + ) ): # Once a continuation is pending upstream, reconnecting without # replay cannot complete the current request, while replaying it diff --git a/app/modules/proxy/_service/http_bridge/upstream_events.py b/app/modules/proxy/_service/http_bridge/upstream_events.py index a70367af9b..827a3073a2 100644 --- a/app/modules/proxy/_service/http_bridge/upstream_events.py +++ b/app/modules/proxy/_service/http_bridge/upstream_events.py @@ -624,16 +624,6 @@ async def _relay_http_bridge_upstream_messages( if not expired_owner: continue pending_count = len(session.pending_requests) - for request_state in session.pending_requests: - if request_state.failure_phase_override is None: - request_state.failure_phase_override = "upstream" - if request_state.failure_detail_override is None: - request_state.failure_detail_override = ( - _HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL - ) - # Claim the session before cancelling receive 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, @@ -641,10 +631,6 @@ async def _relay_http_bridge_upstream_messages( ) if receive_cancelled: receive_task = None - _record_http_bridge_stuck_retire( - reason=_HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL, - session=session, - ) _log_http_bridge_event( "missing_response_created_timeout", session.key, @@ -657,6 +643,29 @@ async def _relay_http_bridge_upstream_messages( _extract_model_class(session.request_model) if session.request_model else None ), ) + retried = await self._retry_http_bridge_precreated_request( + session, + allow_same_anchor_before_created=True, + ) + if retried: + continue + # Claim the session only after the safe pre-created + # retry path refuses it. Reconnecting first prevents + # a silent upstream websocket from stranding clients + # until the request budget expires. + session.closed = True + _record_http_bridge_stuck_retire( + reason=_HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL, + session=session, + ) + 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" + if request_state.failure_detail_override is None: + request_state.failure_detail_override = ( + _HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL + ) await self._fail_http_bridge_reader_and_maybe_retire( session, error_code="upstream_request_timeout", diff --git a/app/modules/proxy/_service/streaming/mixin.py b/app/modules/proxy/_service/streaming/mixin.py index 6189ae7cd5..a8d228cab8 100644 --- a/app/modules/proxy/_service/streaming/mixin.py +++ b/app/modules/proxy/_service/streaming/mixin.py @@ -298,6 +298,7 @@ _RetryableStreamError, _StreamSettlement, _TerminalStreamError, + _TransientStreamError, _ttft_event_latency_ms, _WebSocketUpstreamControl, ) @@ -491,6 +492,7 @@ async def _stream_once( enforce_openai_sdk_contract: bool = True, ) -> AsyncIterator[str]: proxy = cast(_StreamingServiceProtocol, self) + settlement.reset() account_id_value = account.id access_token = proxy._encryptor.decrypt(account.access_token_encrypted) account_id = _header_account_id(account.chatgpt_account_id) @@ -577,6 +579,8 @@ async def _stream_once( settlement.record_success = False settlement.account_health_error = True settlement.error = {"message": error_message} + if allow_transient_retry and payload.previous_response_id is not None: + raise _TransientStreamError(error_code, settlement.error) yield format_sse_event( response_failed_event( error_code, diff --git a/app/modules/proxy/_service/support.py b/app/modules/proxy/_service/support.py index 0d842e78f9..148c03b3f2 100644 --- a/app/modules/proxy/_service/support.py +++ b/app/modules/proxy/_service/support.py @@ -67,6 +67,8 @@ _TTFT_OUTPUT_ITEM_TYPES = _PENDING_TOOL_CALL_ITEM_TYPES - {"function_call"} _WEBSOCKET_FULL_REPLAY_WAIT_MIN_ITEMS = 20 _WEBSOCKET_FULL_REPLAY_WAIT_POLL_SECONDS = 0.05 +_WEBSOCKET_CREATED_ONLY_CLOSE_MAX_REPLAYS = 1 +_WEBSOCKET_TRANSPARENT_CLOSE_MAX_REPLAYS = 20 _HARD_HTTP_BRIDGE_AFFINITY_KINDS = frozenset( { "turn_state_header", @@ -1157,7 +1159,7 @@ def _record_response_event(request_state: _WebSocketRequestState | None, event_t def _websocket_request_can_replay_before_visible_output(request_state: _WebSocketRequestState) -> bool: if not request_state.request_text: return False - if request_state.replay_count >= 1: + if request_state.replay_count >= _WEBSOCKET_TRANSPARENT_CLOSE_MAX_REPLAYS: return False sequenced_created_only_prewarm = ( request_state.generate_false_prewarm @@ -1167,6 +1169,12 @@ def _websocket_request_can_replay_before_visible_output(request_state: _WebSocke and request_state.response_event_count == 1 and not request_state.downstream_visible ) + if ( + request_state.response_id is not None + and not sequenced_created_only_prewarm + and request_state.replay_count >= _WEBSOCKET_CREATED_ONLY_CLOSE_MAX_REPLAYS + ): + return False if request_state.last_downstream_sequence_number is not None and not sequenced_created_only_prewarm: return False if request_state.downstream_visible: diff --git a/tests/integration/test_http_responses_bridge.py b/tests/integration/test_http_responses_bridge.py index 989783c5af..b12770af52 100644 --- a/tests/integration/test_http_responses_bridge.py +++ b/tests/integration/test_http_responses_bridge.py @@ -8178,6 +8178,108 @@ async def fake_connect_responses_websocket( assert connect_count == 2 +@pytest.mark.asyncio +async def test_v1_responses_http_bridge_retries_when_upstream_never_acknowledges_response_create( + async_client, + monkeypatch, +): + _install_bridge_settings_with_limits( + 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_missing_created_retry", + "http-bridge-missing-created-retry@example.com", + ) + account = await _get_account(account_id) + silent_upstreams = [_SilentUpstreamWebSocket() for _ in range(5)] + recovered_upstream = _FakeBridgeUpstreamWebSocket() + upstreams = [*silent_upstreams, recovered_upstream] + connect_count = 0 + + 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, + ): + del 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) + + response = await asyncio.wait_for( + async_client.post( + "/v1/responses", + json={ + "model": "gpt-5.1", + "instructions": "Return exactly OK.", + "input": "retry missing response.created", + "prompt_cache_key": "missing-created-retry-key", + }, + ), + timeout=_TEST_SYNC_TIMEOUT_SECONDS, + ) + + assert response.status_code == 200 + assert connect_count == len(silent_upstreams) + 1 + assert all(upstream.closed for upstream in silent_upstreams) + assert [len(upstream.sent_text) for upstream in silent_upstreams] == [1] * len(silent_upstreams) + assert len(recovered_upstream.sent_text) == 1 + + @pytest.mark.asyncio async def test_backend_responses_http_bridge_retries_precreated_server_overload(async_client, monkeypatch): _install_bridge_settings(monkeypatch, enabled=True) diff --git a/tests/unit/test_proxy_http_bridge.py b/tests/unit/test_proxy_http_bridge.py index 00dd232e8e..499a42800d 100644 --- a/tests/unit/test_proxy_http_bridge.py +++ b/tests/unit/test_proxy_http_bridge.py @@ -18444,7 +18444,7 @@ async def close(self) -> None: assert owner.response_event_count == 0 if leading_telemetry: assert owner.latency_first_upstream_event_ms is not None - retry_precreated.assert_not_awaited() + retry_precreated.assert_awaited_once_with(session, allow_same_anchor_before_created=True) assert write_request_log.await_count == 2 assert {call.kwargs["error_code"] for call in write_request_log.await_args_list} == {"upstream_request_timeout"} fail_reader.assert_awaited_once() diff --git a/tests/unit/test_proxy_utils.py b/tests/unit/test_proxy_utils.py index 31910f68ca..69939bb96a 100644 --- a/tests/unit/test_proxy_utils.py +++ b/tests/unit/test_proxy_utils.py @@ -30073,6 +30073,67 @@ async def fake_stream(payload, headers, access_token, account_id, base_url=None, record_success.assert_not_awaited() +@pytest.mark.asyncio +async def test_stream_previsible_core_eof_with_previous_response_id_retries(monkeypatch): + settings = _make_proxy_settings() + request_logs = _RequestLogsRecorder() + service = proxy_service.ProxyService(_repo_factory(request_logs)) + account = _make_account("acc_previsible_core_eof") + request_logs.response_owner_by_id[("resp_parent", None, "sid-stream")] = account.id + handle_stream_error = AsyncMock() + record_success = AsyncMock() + stream_calls = 0 + + monkeypatch.setattr(proxy_service, "get_settings_cache", lambda: _SettingsCache(settings)) + monkeypatch.setattr(proxy_service, "get_settings", lambda: settings) + monkeypatch.setattr(proxy_service, "_MAX_TRANSIENT_SAME_ACCOUNT_RETRIES", 3) + monkeypatch.setattr(streaming_retry_module.ProcessNetworkRecovery, "wait", AsyncMock(return_value=None)) + monkeypatch.setattr(streaming_retry_module.asyncio, "sleep", AsyncMock()) + monkeypatch.setattr( + service._load_balancer, + "select_account", + AsyncMock(return_value=AccountSelection(account=account, error_message=None)), + ) + monkeypatch.setattr(service._load_balancer, "record_success", record_success) + monkeypatch.setattr(service, "_ensure_fresh", AsyncMock(return_value=account)) + monkeypatch.setattr(service, "_handle_stream_error", handle_stream_error) + monkeypatch.setattr(service, "_settle_stream_api_key_usage", AsyncMock(return_value=True)) + + async def fake_stream(payload, headers, access_token, account_id, base_url=None, raise_for_status=False, **kwargs): + nonlocal stream_calls + del payload, headers, access_token, account_id, base_url, raise_for_status, kwargs + stream_calls += 1 + if stream_calls == 1: + return + yield 'data: {"type":"response.completed","response":{"id":"resp_child_retry_ok"}}\n\n' + + monkeypatch.setattr(proxy_service, "core_stream_responses", fake_stream) + + payload = ResponsesRequest.model_validate( + { + "model": "gpt-5.1", + "instructions": "hi", + "input": [], + "stream": True, + "previous_response_id": "resp_parent", + } + ) + + chunks = [chunk async for chunk in service.stream_responses(payload, {"session_id": "sid-stream"})] + + completed = json.loads(chunks[-1].split("data: ", 1)[1]) + assert completed["type"] == "response.completed" + assert completed["response"]["id"] == "resp_child_retry_ok" + assert stream_calls == 2 + assert request_logs.lookup_calls == [("resp_parent", None, "sid-stream")] + assert await service.drain_persistence_tasks(timeout_seconds=1) + assert [call["status"] for call in request_logs.calls] == ["error", "success"] + assert request_logs.calls[0]["error_code"] == "stream_incomplete" + assert request_logs.calls[-1]["request_id"] == "resp_child_retry_ok" + handle_stream_error.assert_not_awaited() + record_success.assert_awaited_once_with(account) + + @pytest.mark.asyncio async def test_stream_missing_tool_output_proxy_error_is_masked_to_stream_incomplete(monkeypatch, caplog): settings = _make_proxy_settings()