fix(proxy): integrate transport reliability fixes

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/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.

app/modules/proxy/_service/websocket/mixin.py

@@ -492,6 +492,9 @@ def _facade() -> Any:
 logger = logging.getLogger(__name__)
 
 _WEBSOCKET_PINNED_REFRESH_UNAVAILABLE_MESSAGE = "Account refresh is temporarily unavailable; retry later."
+# Scope teardown coordinates several request/lease finalizers; keep its normal
+# observation budget separate from the short generic child-task cancel bound.
+_WEBSOCKET_SCOPE_CLEANUP_TIMEOUT_SECONDS = 5.0
 _CAPABILITY_REQUIRED_NO_AUTHORIZED_ACCOUNTS_MESSAGE = (
     "This request requires Trusted Access for Cyber, but no eligible account is marked as "
     "security-work-authorized. codex-lb did not fall back to an ordinary account."
@@ -2510,11 +2513,19 @@ def take_reader_replay_request_state() -> _WebSocketRequestState | None:
             scope_cancelled = True
             raise
         finally:
-            cleanup_timeout = shutdown_state.remaining_drain_timeout_seconds()
-            if cleanup_timeout is None:
-                cleanup_timeout = _facade()._TASK_CANCEL_TIMEOUT_SECONDS
+            remaining_drain_timeout = shutdown_state.remaining_drain_timeout_seconds()
+            cleanup_timeout = (
+                _WEBSOCKET_SCOPE_CLEANUP_TIMEOUT_SECONDS
+                if remaining_drain_timeout is None
+                else max(float(remaining_drain_timeout), 0.0)
+            )
+            task_cleanup_timeout = (
+                _facade()._TASK_CANCEL_TIMEOUT_SECONDS if remaining_drain_timeout is None else cleanup_timeout
+            )
+            cleanup_phase = "not_started"
 
             async def finalize_websocket_scope() -> None:
+                nonlocal cleanup_phase
                 nonlocal replay_request_state
                 nonlocal request_state_failure_task
                 nonlocal request_state_to_fail
@@ -2527,13 +2538,15 @@ async def finalize_websocket_scope() -> None:
                     # release that wait.
                     reader_to_await.cancel()
                 if upstream is not None:
+                    cleanup_phase = "upstream_close"
                     await _close_websocket_upstream_for_cleanup(
                         proxy,
                         upstream,
                         timeout_seconds=cleanup_timeout,
                     )
                 if reader_to_await is not None:
                     try:
+                        cleanup_phase = "upstream_reader"
                         await _facade()._await_cancelled_task(
                             reader_to_await,
                             label="proxy websocket upstream reader",
@@ -2550,9 +2563,10 @@ async def finalize_websocket_scope() -> None:
                     upstream_reader = None
                 if retired_create_lease_release_task is not None:
                     try:
+                        cleanup_phase = "retired_create_lease"
                         await _facade()._await_cancelled_task(
                             retired_create_lease_release_task,
-                            timeout_seconds=cleanup_timeout,
+                            timeout_seconds=task_cleanup_timeout,
                             label="proxy websocket retired create lease release",
                             cancel=False,
                         )
@@ -2564,9 +2578,10 @@ async def finalize_websocket_scope() -> None:
                     retired_create_lease_release_task = None
                 if request_state_failure_task is not None:
                     try:
+                        cleanup_phase = "unsent_request"
                         await _facade()._await_cancelled_task(
                             request_state_failure_task,
-                            timeout_seconds=cleanup_timeout,
+                            timeout_seconds=task_cleanup_timeout,
                             label="proxy websocket unsent request finalization",
                             cancel=False,
                         )
@@ -2580,6 +2595,7 @@ async def finalize_websocket_scope() -> None:
                     replay_request_state = upstream_control.replay_request_state
                     upstream_control.replay_request_state = None
                 if request_state_to_fail is not None:
+                    cleanup_phase = "unsent_request"
                     await proxy._fail_pending_websocket_requests(
                         account=None,
                         account_id_value=account.id if account is not None else upstream_account_id,
@@ -2598,6 +2614,7 @@ async def finalize_websocket_scope() -> None:
                     )
                     request_state_to_fail = None
                 if replay_request_state is not None:
+                    cleanup_phase = "replay_request"
                     await proxy._fail_pending_websocket_requests(
                         account=None,
                         account_id_value=account.id if account is not None else upstream_account_id,
@@ -2615,6 +2632,7 @@ async def finalize_websocket_scope() -> None:
                         penalize_account=False,
                     )
                 client_disconnected = downstream_activity.disconnected
+                cleanup_phase = "pending_requests"
                 await proxy._fail_pending_websocket_requests(
                     account=None if client_disconnected or scope_cancelled else account,
                     account_id_value=account.id if account is not None else upstream_account_id,
@@ -2637,6 +2655,7 @@ async def finalize_websocket_scope() -> None:
                     penalize_account=not (client_disconnected or scope_cancelled),
                 )
                 try:
+                    cleanup_phase = "connection_lease"
                     await release_current_account_lease()
                 except Exception:
                     # Connection-lease cleanup must never replace cancellation
@@ -2645,6 +2664,7 @@ async def finalize_websocket_scope() -> None:
                         "Failed to release websocket connection lease during scope cleanup",
                         exc_info=True,
                     )
+                cleanup_phase = "complete"
 
             cleanup_task = asyncio.create_task(
                 finalize_websocket_scope(),
@@ -2670,8 +2690,9 @@ def log_scope_cleanup_failure(done_task: asyncio.Task[None]) -> None:
             if not done:
                 _facade().logger.warning(
                     "Websocket scope cleanup exceeded its remaining drain budget "
-                    "timeout_seconds=%.3f background_cleanup_tasks=%d",
+                    "timeout_seconds=%.3f cleanup_phase=%s background_cleanup_tasks=%d",
                     max(float(cleanup_timeout), 0.0),
+                    cleanup_phase,
                     sum(1 for task in proxy._background_cleanup_tasks if not task.done()),
                 )
 

openspec/changes/attribute-websocket-scope-cleanup-phase/proposal.md

@@ -0,0 +1,33 @@
+## Why
+
+When WebSocket scope cleanup exceeds its drain budget, the warning reports the
+timeout and total background cleanup task count but not the operation that is
+still blocked. Operators cannot distinguish an upstream-close stall from
+reader observation, request finalization, or lease release without reproducing
+the incident under instrumentation.
+
+## What Changes
+
+- Track the current WebSocket scope cleanup phase locally while the existing
+  finalization sequence runs.
+- Add that fixed, low-cardinality phase to the existing timeout warning.
+- Keep cleanup ordering, timeout budgets, retries, and ownership unchanged.
+- Do not log request ids, account ids, payloads, credentials, or exception
+  content in the phase field.
+
+## Capabilities
+
+### New Capabilities
+
+(none)
+
+### Modified Capabilities
+
+- `proxy-runtime-observability`: WebSocket scope cleanup timeout warnings MUST
+  identify the blocked cleanup phase with a fixed low-cardinality value.
+
+## Impact
+
+`app/modules/proxy/_service/websocket/mixin.py` and its route-level WebSocket
+cleanup regression coverage. No API, schema, setting, timeout, or dashboard
+change.