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

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

@@ -1297,12 +1297,28 @@ def _http_bridge_session_reusable_for_lookup(
     require_preferred_account: bool,
     service_tier_supported: bool,
     allow_closed_admission_handoff: bool,
+    session_key_quarantined: bool,
 ) -> bool:
+    if session.quarantined and not session_key_quarantined:
+        # The quarantine registry TTL is the source of truth
+        # (``_http_bridge_session_key_quarantined``): pruning drops the
+        # registry entry without touching the per-session flag, so a live
+        # session that outlives its quarantine window must become reusable
+        # again instead of staying rejected forever. Reset the stale flag so
+        # session state agrees with the registry.
+        session.quarantined = False
     live_or_retained = _http_bridge_session_account_active(session) and (
         not session.closed or (allow_closed_admission_handoff and _http_bridge_session_has_admission_waiter(session))
     )
     return (
         live_or_retained
+        # A quarantined key has proven silent/wedged; a new request must take
+        # the fresh session path instead of re-attaching. The registry verdict
+        # is authoritative for the key in both directions: a freshly created
+        # replacement session (flag still False) under a still-quarantined key
+        # is not reusable either, so a concurrent full-resend cannot restore
+        # the suppressed durable anchor through session hydration.
+        and not session_key_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/mixin.py

@@ -125,6 +125,9 @@
 from app.modules.proxy._service.http_bridge.owner_forwarding import _HTTPBridgeOwnerForwardingMixin
 from app.modules.proxy._service.http_bridge.protocol import _HTTPBridgeServiceProtocol
 from app.modules.proxy._service.http_bridge.proxy_failover import _HTTPBridgePreDispatchFailover
+from app.modules.proxy._service.http_bridge.quarantine import (
+    _http_bridge_session_key_quarantined,
+)
 from app.modules.proxy._service.http_bridge.request_submit import _HTTPBridgeRequestSubmitMixin
 from app.modules.proxy._service.http_bridge.service_stubs import (
     _await_cancelled_task,
@@ -683,6 +686,7 @@ def bind_account_neutral_recovery_owner(session: _HTTPBridgeSession) -> None:
                     require_preferred_account=require_preferred_account,
                     service_tier_supported=_http_bridge_compatible(existing, request_model, request_service_tier),
                     allow_closed_admission_handoff=retained_handoff,
+                    session_key_quarantined=_http_bridge_session_key_quarantined(self, existing.key),
                 )
                 fork_key = _http_bridge_parallel_fork_key(
                     key=key,

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

@@ -68,6 +68,9 @@
     _normalized_http_bridge_instance_ring,
     _sticky_key_from_turn_state_header,
 )
+from app.modules.proxy._service.http_bridge.quarantine import (
+    _http_bridge_session_key_quarantined,
+)
 from app.modules.proxy._service.http_bridge.service_stubs import (
     _headers_with_authorization,
     _partial_output_proxy_error_event_block,
@@ -218,6 +221,16 @@ async def _http_bridge_has_live_local_session(
                 session = self._http_bridge_sessions.get(candidate_key)
                 if session is None or session.closed or not _http_bridge_session_account_active(session):
                     continue
+                if _http_bridge_session_key_quarantined(self, session.key):
+                    # A session under a quarantined key (#1534) is rejected and
+                    # detached at lookup time, so for durable-anchor selection
+                    # it must count as absent: a delta-only payload then keeps
+                    # the durable anchor on the fresh session instead of
+                    # silently losing its prior context. The registry verdict
+                    # is authoritative for the key — a freshly created
+                    # replacement session (flag still False) under a
+                    # still-quarantined key counts as absent too.
+                    continue
                 if _durable_recovery_supersedes_local_session(durable_lookup, session):
                     _drop_superseded_local_recovery_aliases_locked(
                         self,

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(
@@ -1818,6 +1825,14 @@ async def _retire_stale_pending_http_bridge_session(
         retry_circuit_detail: str | None = None,
         response_events_seen: int | None = None,
     ) -> None:
+        async with session.pending_lock:
+            retired_request_states = list(session.pending_requests)
+        # Direct retirement (for example the all-stale stuck-gate path, where
+        # the wedged reattach is the only pending request) cancels the reader
+        # and fails the pendings without passing the partial-cleanup hook or
+        # the reader-failure funnel, so evaluate the wedge shape (#1534) here
+        # too; recording is idempotent for callers that already quarantined.
+        _record_http_bridge_quarantine_wedged_pending(self, session, retired_request_states)
         if response_events_seen is None or response_events_seen == 0:
             await self._record_http_bridge_retry_circuit_failure(
                 session,

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,
@@ -1194,6 +1197,12 @@ async def release_unowned_bridge_lifecycle(
         durable_full_resend_fresh_bridge_proof: _VerifiedDurableFullResend | None = None
         force_local_recovery_creation = False
         payload_looks_like_full_resend = _http_bridge_payload_looks_like_full_resend(payload)
+        # Set when the quarantine check below suppresses the durable-anchor
+        # injection for a full-resend payload; the session hydration and the
+        # session-level anchor injection further down must honor it so the
+        # dispatch genuinely goes unanchored instead of rebuilding the same
+        # wedged reattach through the session-state side door.
+        fresh_reattach_anchor_suppressed_quarantined = False
 
         def classify_durable_full_resend(
             lookup: DurableBridgeLookup,
@@ -1411,6 +1420,31 @@ 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 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).
+                # Evaluated independently of the fresh-reattach eligibility
+                # above: even when that gate is already false (for example a
+                # conversation-scoped payload, a live alias session, or an
+                # active-owner forward that later falls back to a local
+                # rebind), the suppression must still reach the session
+                # hydration, session-level injection, and recovery injection
+                # paths below.
+                fresh_reattach_can_use_durable_anchor = False
+                fresh_reattach_anchor_suppressed_quarantined = True
+                _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
@@ -2164,6 +2198,10 @@ def switch_to_account_neutral_replay() -> None:
                 recovery_anchor_input_fingerprint: str | None = None
                 if (
                     not owner_forward_fresh_replay
+                    # The quarantine skip (#1534) applies to the local recovery
+                    # rebind as well: re-injecting the suppressed anchor here
+                    # would rebuild the same wedged reattach.
+                    and not fresh_reattach_anchor_suppressed_quarantined
                     and not durable_full_resend_has_safe_fresh_context
                     and recovery_payload.previous_response_id is None
                     and durable_lookup is not None
@@ -2293,7 +2331,12 @@ def switch_to_account_neutral_replay() -> None:
                 return
         session = session_or_forward
         if (
-            not durable_full_resend_has_safe_fresh_context
+            # A quarantine-suppressed anchor (#1534) must not be rehydrated
+            # into the session either: doing so would let the session-level
+            # injection below re-add the exact anchor the quarantine skipped
+            # and trim the prefix, rebuilding the wedged reattach.
+            not fresh_reattach_anchor_suppressed_quarantined
+            and not durable_full_resend_has_safe_fresh_context
             and durable_full_resend_anchor_count is not None
             and durable_full_resend_anchor_fingerprint is not None
             and durable_lookup is not None
@@ -2339,6 +2382,9 @@ def switch_to_account_neutral_replay() -> None:
         ) and (not _http_bridge_payload_looks_like_full_resend(effective_payload) or session_anchor_trimmable)
         if (
             session.codex_session
+            # Honor the quarantine decision end to end: the durable anchor
+            # skipped above must not come back as a session-level injection.
+            and not fresh_reattach_anchor_suppressed_quarantined
             and not proxy_injected_previous_response_id
             and effective_payload.previous_response_id is None
             and session.last_completed_response_id is not None