fix(proxy): retry silent bridge response.create upstreams

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,23 +1561,45 @@ 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:
                 retryable_requests = [
                     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

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

@@ -624,27 +624,13 @@ 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,
                                     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,
-                            )
                             _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",

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,

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:

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)

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()

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()