fix(proxy): quarantine silent HTTP bridge sessions (takeover of #1405)

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

@@ -1303,6 +1303,9 @@ def _http_bridge_session_reusable_for_lookup(
     )
     return (
         live_or_retained
+        # A quarantined session has proven silent/wedged; a new request must
+        # take the fresh session path instead of re-attaching to it.
+        and not session.quarantined
         and _http_bridge_session_allows_api_key(session, api_key)
         and _http_bridge_session_reusable_for_request(
             session=session,

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

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass
+from typing import Any
+
+from app.modules.proxy._service.http_bridge.helpers import _log_http_bridge_event
+from app.modules.proxy._service.support import (
+    _REQUEST_TRANSPORT_HTTP,
+    _HTTPBridgeSession,
+    _HTTPBridgeSessionKey,
+    _WebSocketRequestState,
+)
+from app.modules.proxy.affinity import _extract_model_class
+
+logger = logging.getLogger("app.modules.proxy.service")
+
+# Quarantine is a bounded, in-memory, session-scoped (never account-scoped)
+# marker for HTTP bridge session keys that have proven silent/wedged: a later
+# request must not re-attach to them and must take the existing fresh
+# session/no-anchor path instead (#1534). It complements — and never replaces
+# — the in-flight recovery machinery: the eventless watchdog and bounded
+# replay (#1394) recover the request that is currently stuck, the fenced
+# durable-anchor clear (#1563) stops a *fully eventless* full-resend anchor
+# from being re-injected, and the durable retry circuit backs off in-place
+# retries. Quarantine covers what those leave open: the reattached stream
+# that delivers response events but never gets ``response.created`` (the
+# ``response_event_count == 0`` gates in the stale/eventless detection never
+# trip on it), and the repeated-wedge case where consecutive eventless
+# timeouts keep rebuilding the same reattach.
+_HTTP_BRIDGE_QUARANTINE_TTL_SECONDS = 600.0
+_HTTP_BRIDGE_QUARANTINE_EVENTLESS_TIMEOUT_THRESHOLD = 2
+_HTTP_BRIDGE_QUARANTINE_MAX_ENTRIES = 1024
+
+_HTTP_BRIDGE_QUARANTINE_WEDGED_REATTACH_REASON = "reattach_missing_response_created"
+_HTTP_BRIDGE_QUARANTINE_REPEATED_EVENTLESS_REASON = "repeated_eventless_timeout"
+
+
+@dataclass(slots=True)
+class _HTTPBridgeQuarantineEntry:
+    quarantined_until: float = 0.0
+    consecutive_eventless_timeouts: int = 0
+    last_touched_monotonic: float = 0.0
+    reason: str | None = None
+
+
+def _http_bridge_quarantine_registry(
+    service: Any,
+) -> dict[_HTTPBridgeSessionKey, _HTTPBridgeQuarantineEntry]:
+    registry = getattr(service, "_http_bridge_quarantined_keys", None)
+    if registry is None:
+        registry = {}
+        service._http_bridge_quarantined_keys = registry
+    return registry
+
+
+def _prune_http_bridge_quarantine_registry(
+    registry: dict[_HTTPBridgeSessionKey, _HTTPBridgeQuarantineEntry],
+    now: float,
+) -> None:
+    expiry = now - _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS
+    for key, entry in list(registry.items()):
+        if entry.last_touched_monotonic <= expiry and entry.quarantined_until <= now:
+            registry.pop(key, None)
+    overflow = len(registry) - _HTTP_BRIDGE_QUARANTINE_MAX_ENTRIES
+    if overflow > 0:
+        for stale_key in sorted(registry, key=lambda candidate: registry[candidate].last_touched_monotonic)[:overflow]:
+            registry.pop(stale_key, None)
+
+
+def _http_bridge_request_state_wedged_reattach(request_state: _WebSocketRequestState) -> bool:
+    """Identify the #1534 wedge shape on a request that is being failed/retired.
+
+    A reattached stream (proxy-injected ``previous_response_id``) whose
+    ``response.create`` was sent and that observed upstream response events,
+    but whose ``response.created`` was never assigned. This is only evaluated
+    when the request is already being failed or its session retired — never
+    against a live owned turn — so legitimate long event gaps (for example
+    deferred-reasoning streams) can never trip it, and any request whose
+    ``response.created`` was observed (``response_id`` or created latency set)
+    is excluded by construction.
+    """
+    return (
+        getattr(request_state, "transport", None) == _REQUEST_TRANSPORT_HTTP
+        and not getattr(request_state, "skip_request_log", False)
+        and getattr(request_state, "proxy_injected_previous_response_id", False)
+        and getattr(request_state, "response_create_sent_at", None) is not None
+        and getattr(request_state, "response_id", None) is None
+        and getattr(request_state, "latency_response_created_ms", None) is None
+        and getattr(request_state, "response_event_count", 0) > 0
+    )
+
+
+def _http_bridge_session_key_quarantined(service: Any, key: _HTTPBridgeSessionKey) -> bool:
+    registry = _http_bridge_quarantine_registry(service)
+    now = time.monotonic()
+    _prune_http_bridge_quarantine_registry(registry, now)
+    entry = registry.get(key)
+    return entry is not None and entry.quarantined_until > now
+
+
+def _quarantine_http_bridge_session(service: Any, session: _HTTPBridgeSession, *, reason: str) -> None:
+    """Quarantine a bridge session that has proven silent/wedged.
+
+    Session-scoped only: no account-health writes happen here, and the entry
+    is bounded by TTL, a registry size cap, and the healthy-completion clear.
+    """
+    now = time.monotonic()
+    registry = _http_bridge_quarantine_registry(service)
+    entry = registry.setdefault(session.key, _HTTPBridgeQuarantineEntry())
+    already_quarantined = entry.quarantined_until > now
+    entry.quarantined_until = max(entry.quarantined_until, now + _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS)
+    entry.last_touched_monotonic = now
+    entry.reason = reason
+    _prune_http_bridge_quarantine_registry(registry, now)
+    session.quarantined = True
+    if already_quarantined:
+        return
+    _log_http_bridge_event(
+        "session_quarantined",
+        session.key,
+        account_id=session.account.id,
+        model=session.request_model,
+        detail=f"reason={reason}, ttl_seconds={_HTTP_BRIDGE_QUARANTINE_TTL_SECONDS:.0f}",
+        cache_key_family=session.key.affinity_kind,
+        model_class=_extract_model_class(session.request_model) if session.request_model else None,
+    )
+
+
+def _record_http_bridge_quarantine_wedged_pending(
+    service: Any,
+    session: _HTTPBridgeSession,
+    request_states: Any,
+) -> bool:
+    """Quarantine the session when a failed/retired pending request proves the wedge shape."""
+    if not any(_http_bridge_request_state_wedged_reattach(request_state) for request_state in request_states):
+        return False
+    _quarantine_http_bridge_session(
+        service,
+        session,
+        reason=_HTTP_BRIDGE_QUARANTINE_WEDGED_REATTACH_REASON,
+    )
+    return True
+
+
+def _record_http_bridge_quarantine_eventless_timeout(service: Any, session: _HTTPBridgeSession) -> None:
+    """Count a ``missing_response_created_timeout`` retire; quarantine on repeats.
+
+    The first eventless timeout is left to the merged recovery machinery
+    (bounded pre-created retry, fenced durable-anchor clear). A second
+    consecutive one for the same session key proves that path is also
+    rebuilding a wedged attach, so later requests must stop re-attaching.
+    """
+    now = time.monotonic()
+    registry = _http_bridge_quarantine_registry(service)
+    # Prune before touching the entry: a strike whose TTL already lapsed must
+    # not be resurrected into a "consecutive" second strike hours later.
+    _prune_http_bridge_quarantine_registry(registry, now)
+    entry = registry.setdefault(session.key, _HTTPBridgeQuarantineEntry())
+    entry.consecutive_eventless_timeouts += 1
+    entry.last_touched_monotonic = now
+    if entry.consecutive_eventless_timeouts < _HTTP_BRIDGE_QUARANTINE_EVENTLESS_TIMEOUT_THRESHOLD:
+        return
+    _quarantine_http_bridge_session(
+        service,
+        session,
+        reason=_HTTP_BRIDGE_QUARANTINE_REPEATED_EVENTLESS_REASON,
+    )
+
+
+def _clear_http_bridge_quarantine(service: Any, session: _HTTPBridgeSession) -> None:
+    """A completed response on this key disproves the wedge; drop all state."""
+    registry = _http_bridge_quarantine_registry(service)
+    session.quarantined = False
+    entry = registry.pop(session.key, None)
+    if entry is None:
+        return
+    if entry.quarantined_until <= time.monotonic():
+        return
+    _log_http_bridge_event(
+        "session_quarantine_cleared",
+        session.key,
+        account_id=session.account.id,
+        model=session.request_model,
+        detail=f"reason={entry.reason}",
+        cache_key_family=session.key.affinity_kind,
+        model_class=_extract_model_class(session.request_model) if session.request_model else None,
+    )

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

@@ -84,6 +84,9 @@
     _register_http_bridge_turn_state_aliases_locked,
     _release_http_bridge_unanchored_handoff,
 )
+from app.modules.proxy._service.http_bridge.quarantine import (
+    _record_http_bridge_quarantine_wedged_pending,
+)
 from app.modules.proxy._service.http_bridge.service_stubs import (
     _call_with_supported_optional_kwargs,
     _classify_upstream_close,
@@ -1747,6 +1750,10 @@ async def _fail_stale_http_bridge_pending_requests(
                 stale_requests.append(request_state)
         if not stale_requests:
             return
+        # A stale gate holder that streamed response events without ever
+        # receiving ``response.created`` proves the reattach wedge (#1534)
+        # even when the session itself survives with other active requests.
+        _record_http_bridge_quarantine_wedged_pending(self, session, stale_requests)
         if response_events_seen == 0:
             await self._record_http_bridge_retry_circuit_failure(session, detail=detail)
         await self._fail_pending_websocket_requests(

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

@@ -101,6 +101,9 @@
     _reserve_http_bridge_unanchored_handoff,
     _trim_http_bridge_previous_response_input_items,
 )
+from app.modules.proxy._service.http_bridge.quarantine import (
+    _http_bridge_session_key_quarantined,
+)
 from app.modules.proxy._service.http_bridge.service_stubs import (
     _build_rewritten_stream_response_failed_event,
     _codex_keepalive_frame,
@@ -1411,6 +1414,27 @@ def classify_durable_full_resend(
                 and durable_lookup.latest_response_id is not None
                 and (not payload_looks_like_full_resend or durable_anchor_trimmable)
             )
+            if (
+                fresh_reattach_can_use_durable_anchor
+                and payload_looks_like_full_resend
+                and _http_bridge_session_key_quarantined(self, bridge_session_key)
+            ):
+                # The previous attach on this key proved silent/wedged
+                # (#1534). The client's own payload already carries the full
+                # conversation, so send it unanchored on the fresh path
+                # instead of rebuilding the same reattach. Delta-only
+                # payloads keep the anchor: it is their only way to convey
+                # prior context (same boundary as the fenced anchor clear).
+                fresh_reattach_can_use_durable_anchor = False
+                _log_http_bridge_event(
+                    "fresh_reattach_anchor_skipped_quarantined",
+                    bridge_session_key,
+                    account_id=durable_lookup.account_id,
+                    model=payload.model,
+                    detail=f"response_id={durable_lookup.latest_response_id}",
+                    cache_key_family=bridge_session_key.affinity_kind,
+                    model_class=_extract_model_class(payload.model) if payload.model else None,
+                )
             if (
                 fresh_reattach_can_use_durable_anchor
                 and payload_looks_like_full_resend

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

@@ -57,6 +57,11 @@
     _normalize_http_bridge_error_event,
     _record_http_bridge_stuck_retire,
 )
+from app.modules.proxy._service.http_bridge.quarantine import (
+    _clear_http_bridge_quarantine,
+    _record_http_bridge_quarantine_eventless_timeout,
+    _record_http_bridge_quarantine_wedged_pending,
+)
 from app.modules.proxy._service.http_bridge.service_stubs import (
     _assign_websocket_response_id,
     _await_cancelled_task,
@@ -677,6 +682,13 @@ async def _fail_http_bridge_reader_and_maybe_retire(
                 (getattr(request_state, "response_event_count", 0) for request_state in session.pending_requests),
                 default=0,
             )
+            pending_request_states = list(session.pending_requests)
+        # The #1534 wedge shape: a reattached stream that streamed response
+        # events whose ``response.created`` was never assigned. The eventless
+        # watchdog and the durable-anchor clear both key on
+        # ``response_event_count == 0`` and never trip on it, so quarantine
+        # the session here so later requests stop re-attaching to it.
+        _record_http_bridge_quarantine_wedged_pending(self, session, pending_request_states)
         observed_close_code = (
             upstream_close_code if upstream_close_code is not None else session.last_upstream_close_code
         )
@@ -920,6 +932,7 @@ async def _relay_http_bridge_upstream_messages(
                                 # generic reader-crash account penalty path.
                                 if expired_proxy_injected_anchor:
                                     await _clear_durable_http_bridge_response_anchor(self, session)
+                                _record_http_bridge_quarantine_eventless_timeout(self, session)
                                 session.closed = True
                                 await self._fail_http_bridge_reader_and_maybe_retire(
                                     session,
@@ -935,6 +948,10 @@ async def _relay_http_bridge_upstream_messages(
                             # to persist the durable-anchor invalidation.
                             if expired_proxy_injected_anchor:
                                 await _clear_durable_http_bridge_response_anchor(self, session)
+                            # Count the eventless retire toward the repeated-
+                            # wedge quarantine; the first strike still goes
+                            # through the bounded pre-created recovery below.
+                            _record_http_bridge_quarantine_eventless_timeout(self, session)
                             _record_http_bridge_stuck_retire(
                                 reason=_HTTP_BRIDGE_MISSING_RESPONSE_CREATED_TIMEOUT_DETAIL,
                                 session=session,
@@ -2035,6 +2052,7 @@ async def _process_parsed_http_bridge_upstream_event(
             and not terminal_request_state.skip_request_log
         ):
             await self._clear_http_bridge_retry_circuit(session)
+            _clear_http_bridge_quarantine(self, session)
 
         normalize_error_event = (
             terminal_request_state is None or terminal_request_state.enforce_openai_sdk_contract

app/modules/proxy/_service/support.py

@@ -1030,6 +1030,11 @@ class _HTTPBridgeSession:
     last_upstream_close_code: int | None = None
     last_upstream_close_generation: int = 0
     closed: bool = False
+    # Set when the session proved silent/wedged (reattached stream with
+    # response events but no ``response.created``, or repeated eventless
+    # timeouts). A quarantined session must never be selected for reuse or
+    # re-attach; later requests take the fresh session/no-anchor path.
+    quarantined: bool = False
     # Set while a reader handoff is replacing the socket. Idle pruning must
     # retain the registered session during this short transition even though
     # ``closed`` is fail-closed for normal request reuse.

openspec/changes/quarantine-silent-bridge-sessions/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-06