fix(proxy): hold fenced hard turns through cooldown

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

@@ -218,6 +218,7 @@
 
 _REQUEST_TRANSPORT_HTTP = "http"
 _WEBSOCKET_AUTH_INVALIDATED_FAILURE_CODE = "account_auth_invalidated"
+_HTTP_BRIDGE_SAME_ANCHOR_PRECREATED_MAX_REPLAYS = 1
 _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 "
@@ -329,6 +330,23 @@ def _http_bridge_client_full_history_recovery_error() -> OpenAIErrorEnvelope:
     return payload
 
 
+def _http_bridge_can_replay_same_anchor_before_created(request_state: _WebSocketRequestState) -> bool:
+    if not request_state.request_text:
+        return False
+    if request_state.missing_response_created_retry_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
+        and not request_state.file_required_preferred_account
+    )
+
+
 async def _rollback_http_bridge_recovery_turn_state_registration(
     service: Any,
     receipt: DurableBridgeAliasRegistrationReceipt,
@@ -2856,16 +2874,25 @@ async def _retry_http_bridge_precreated_request(
         *,
         request_state: _WebSocketRequestState | None = None,
         restart_reader: bool = False,
+        allow_same_anchor_before_created: bool = False,
     ) -> bool:
         clean_close_retry_max_count = self._http_bridge_clean_close_retry_max_count()
         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"
+
         def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
             if _websocket_request_can_replay_before_visible_output(request_state):
                 return True
+            if (
+                allow_same_anchor_before_created
+                and hard_owner_bound
+                and _http_bridge_can_replay_same_anchor_before_created(request_state)
+            ):
+                return True
             if (
                 clean_close_retry_max_count <= 0
                 or request_state.replay_count != 1
@@ -2886,6 +2913,7 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
         fresh_hard_request_account_switch_candidate = False
         proof_gated_continuity_replay_candidate = False
         server_anchored_replay_candidate = False
+        eventless_same_anchor_replay_candidate = False
         if session.key.strength == "hard":
             async with session.pending_lock:
                 retryable_candidates = [
@@ -2911,20 +2939,28 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
                         and candidate.replay_count == 0
                     )
                     server_anchored_replay_candidate = _http_bridge_server_anchored_replay_enabled(candidate)
+                    eventless_same_anchor_replay_candidate = (
+                        allow_same_anchor_before_created
+                        and hard_owner_bound
+                        and _http_bridge_can_replay_same_anchor_before_created(candidate)
+                        and not (
+                            candidate.fresh_upstream_request_is_retry_safe and candidate.fresh_upstream_request_text
+                        )
+                    )
         if not await self._http_bridge_precreated_retry_allowed(
             session,
             allow_fresh_hard_account_switch=fresh_hard_request_account_switch_candidate,
             allow_proof_gated_continuity_replay=(
                 proof_gated_continuity_replay_candidate or server_anchored_replay_candidate
             ),
+            allow_eventless_same_anchor_replay=eventless_same_anchor_replay_candidate,
         ):
             return False
 
         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"
         async with session.pending_lock:
             if request_state is not None:
                 if (
@@ -2945,8 +2981,24 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
                     return False
                 request_state = retryable_requests[0]
             model_fallback_replay = request_state.precreated_replay_reason == _ACCOUNT_MODEL_UNSUPPORTED_ERROR_CODE
-            if request_state.previous_response_id is not None and not (
-                request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text
+            eventless_same_anchor_replay = (
+                allow_same_anchor_before_created
+                and hard_owner_bound
+                and _http_bridge_can_replay_same_anchor_before_created(request_state)
+                and not (
+                    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.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
@@ -3046,10 +3098,16 @@ def request_is_retryable(request_state: _WebSocketRequestState) -> bool:
                 request_state.clean_close_retry_close_generation = close_generation
             if additional_clean_close_retry:
                 request_state.clean_close_replay_count += 1
+            if eventless_same_anchor_replay:
+                request_state.missing_response_created_retry_count += 1
         retry_jitter_seconds = (
             self._http_bridge_clean_close_retry_jitter_seconds() if additional_clean_close_retry else 0.0
         )
-        retry_event = "retry_precreated_clean_close" if additional_clean_close_retry else "retry_precreated"
+        retry_event = (
+            "retry_precreated_same_anchor"
+            if eventless_same_anchor_replay
+            else ("retry_precreated_clean_close" if additional_clean_close_retry else "retry_precreated")
+        )
         _log_http_bridge_event(
             retry_event,
             session.key,

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

@@ -262,6 +262,7 @@ async def _http_bridge_precreated_retry_allowed(
         allow_fresh_hard_account_switch: bool = False,
         allow_proof_gated_continuity_replay: bool = False,
         allow_operation_fenced_continuity_replay: bool = False,
+        allow_eventless_same_anchor_replay: bool = False,
     ) -> bool:
         """Avoid replaying a repeatedly failing hard-affinity request in a tight loop."""
         if session.key.strength != "hard":
@@ -278,6 +279,7 @@ async def _http_bridge_precreated_retry_allowed(
                     and state.half_open_until > now
                     and not allow_fresh_hard_account_switch
                     and not allow_proof_gated_continuity_replay
+                    and not allow_eventless_same_anchor_replay
                 ):
                     if PROMETHEUS_AVAILABLE and http_bridge_retry_circuit_total is not None:
                         http_bridge_retry_circuit_total.labels(outcome="suppressed").inc()
@@ -324,6 +326,16 @@ async def _http_bridge_precreated_retry_allowed(
                     retry_after,
                 )
                 return True
+            if allow_eventless_same_anchor_replay:
+                logger.info(
+                    "http_bridge_retry_circuit event=bypass_eventless_same_anchor_replay bridge_kind=%s "
+                    "bridge_key=%s failures=%s retry_after_seconds=%.1f",
+                    session.key.affinity_kind,
+                    _hash_identifier(session.key.affinity_key),
+                    state.consecutive_failures,
+                    retry_after,
+                )
+                return True
             if PROMETHEUS_AVAILABLE and http_bridge_retry_circuit_total is not None:
                 http_bridge_retry_circuit_total.labels(outcome="suppressed").inc()
             logger.info(

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

@@ -3613,10 +3613,82 @@ async def retry_precreated_for_idle_recovery(
                     ),
                 )
 
+        def operation_fenced_cooldown_wait_enabled() -> bool:
+            """Allow a hard turn to wait until its durable fence can arbitrate recovery."""
+            return (
+                getattr(
+                    _service_get_settings(),
+                    "http_responses_session_bridge_ambiguous_continuation_recovery_mode",
+                    "fail_closed",
+                )
+                in {"server_anchored_replay_once", "server_indefinite_recovery"}
+                and request_state.hard_continuity_anchor
+                and session.durable_session_id is not None
+                and session.durable_owner_epoch is not None
+                and request_state.response_id is None
+                and request_state.response_event_count == 0
+            )
+
         def continuity_bound_without_safe_replay() -> bool:
             """Do not hold a client stream through a cooldown we cannot use."""
             return _http_bridge_continuity_bound_without_safe_replay(request_state) and not (
-                _http_bridge_server_anchored_replay_enabled(request_state)
+                _http_bridge_server_anchored_replay_enabled(request_state) or operation_fenced_cooldown_wait_enabled()
+            )
+
+        async def wait_through_operation_fenced_startup_cooldown() -> bool:
+            if session.key.strength != "hard" or not operation_fenced_cooldown_wait_enabled():
+                return False
+            retry_cooldown_seconds = await self._http_bridge_precreated_retry_cooldown_seconds(session)
+            if retry_cooldown_seconds <= 0:
+                return False
+            remaining_budget_seconds = request_deadline - _service_time().monotonic()
+            if remaining_budget_seconds <= 0:
+                return False
+            wait_seconds = min(retry_cooldown_seconds, remaining_budget_seconds)
+            _log_http_bridge_event(
+                "wait_operation_fenced_cooldown",
+                session.key,
+                account_id=session.account.id,
+                model=session.request_model,
+                detail="hard_turn_operation_fence",
+                cache_key_family=session.key.affinity_kind,
+            )
+            logger.info(
+                "HTTP bridge waiting through retry-circuit cooldown before durable hard-turn arbitration "
+                "request_id=%s wait_seconds=%.1f remaining_budget_seconds=%.1f",
+                request_state.request_id,
+                wait_seconds,
+                remaining_budget_seconds,
+            )
+            # No upstream request has been dispatched on this path.  After the
+            # cooldown, normal submission still has to create or claim the
+            # durable operation fence before response.create can be sent.
+            await asyncio.sleep(wait_seconds)
+            return True
+
+        async def operation_fenced_request_budget_terminal_event() -> str | None:
+            if not operation_fenced_cooldown_wait_enabled() or _service_time().monotonic() < request_deadline:
+                return None
+            await self._release_websocket_request_state_reservation(request_state)
+            request_state.api_key_reservation = None
+            if propagate_http_errors:
+                raise ProxyResponseError(
+                    503,
+                    openai_error(
+                        "upstream_request_timeout",
+                        "HTTP responses session bridge recovery exceeded the request budget.",
+                        error_type="server_error",
+                    ),
+                )
+            return format_sse_event(
+                cast(
+                    Mapping[str, JsonValue],
+                    response_failed_event(
+                        "stream_idle_timeout",
+                        "HTTP responses session bridge recovery exceeded the request budget",
+                        response_id=_websocket_downstream_response_id(request_state),
+                    ),
+                )
             )
 
         async def startup_continuity_cooldown_terminal_event() -> str | None:
@@ -3687,6 +3759,12 @@ async def startup_continuity_cooldown_terminal_event() -> str | None:
             )
 
         while True:
+            budget_terminal_event = await operation_fenced_request_budget_terminal_event()
+            if budget_terminal_event is not None:
+                yield budget_terminal_event
+                return
+            if await wait_through_operation_fenced_startup_cooldown():
+                continue
             startup_terminal_event = await startup_continuity_cooldown_terminal_event()
             if startup_terminal_event is not None:
                 yield startup_terminal_event

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

@@ -1288,14 +1288,19 @@ async def _relay_http_bridge_upstream_messages(
                                     _extract_model_class(session.request_model) if session.request_model else None
                                 ),
                             )
-                            # A fresh, self-contained hard request can use the
-                            # same bounded pre-created recovery as the idle
-                            # timeout path. Keep the session open until the
-                            # recovery routine claims the handoff; otherwise
-                            # its retry gate would reject the request as
-                            # already retired. Continuity-bound requests still
-                            # fail closed in _retry_http_bridge_precreated_request.
-                            retried = await self._retry_http_bridge_precreated_request(session)
+                            # A fresh, self-contained hard request, or the
+                            # narrower eventless same-anchor continuation
+                            # recovery, can use the same bounded pre-created
+                            # path. Keep the session open until the recovery
+                            # routine claims the handoff; otherwise its retry
+                            # gate would reject the request as already
+                            # retired. Continuity requests that do not satisfy
+                            # the narrow proof still fail closed in
+                            # _retry_http_bridge_precreated_request.
+                            retried = await self._retry_http_bridge_precreated_request(
+                                session,
+                                allow_same_anchor_before_created=True,
+                            )
                             if retried:
                                 continue
                             session.closed = True

app/modules/proxy/_service/support.py

@@ -817,6 +817,10 @@ class _WebSocketRequestState:
     request_usage_budget: ApiKeyRequestUsageBudget | None = None
     request_text: str | None = None
     replay_count: int = 0
+    # Counts the one watchdog-owned same-anchor recovery permitted after an
+    # eventless pre-response-created timeout. Keep this separate from
+    # ``replay_count``, which tracks client/security/fresh-replay attempts.
+    missing_response_created_retry_count: int = 0
     # Counts only the one extra replay permitted after the initial recovery
     # replay when the replacement upstream socket also closes cleanly before
     # producing any response event.