fix(proxy): quarantine poisoned bridge anchors at circuit open

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

@@ -5,7 +5,7 @@
 import logging
 import sys
 import time
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Iterable, Sequence
 from dataclasses import dataclass, replace
 from hashlib import sha256
 from ipaddress import ip_address
@@ -863,6 +863,23 @@ def _http_bridge_retry_circuit_attempt_selection_for_pending_requests(
     return _HTTPBridgeRetryCircuitAttemptSelection(kind="ineligible" if attempt_seen else "absent")
 
 
+def _http_bridge_continuity_bound_without_safe_replay(request_state: _WebSocketRequestState) -> bool:
+    """Return whether retrying would require replaying an unsafe continuation."""
+    if request_state.previous_response_id is not None:
+        return not (request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text)
+    return request_state.hard_continuity_anchor and not (
+        request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text
+    )
+
+
+def _http_bridge_abandonment_strands_requests(request_states: Iterable[Any]) -> bool:
+    """Return whether every request covered by an abandonment is stranded."""
+    states = [state for state in request_states if state is not None]
+    if not states:
+        return False
+    return all(_http_bridge_continuity_bound_without_safe_replay(state) for state in states)
+
+
 def _http_bridge_session_has_admission_waiter(session: object | None) -> bool:
     """Keep a closed bridge registered while an unsent request owns its handoff."""
     return session is not None and bool(getattr(session, "admission_waiter_count", 0))

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

@@ -35,6 +35,9 @@
 
 _HTTP_BRIDGE_QUARANTINE_WEDGED_REATTACH_REASON = "reattach_missing_response_created"
 _HTTP_BRIDGE_QUARANTINE_REPEATED_EVENTLESS_REASON = "repeated_eventless_timeout"
+# A retry circuit opened on an eventless poison-class failure. The anchor that
+# caused the opening must not be re-injected into the next half-open probe.
+_HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON = "retry_circuit_poisoned_anchor"
 
 
 @dataclass(slots=True)
@@ -112,18 +115,29 @@ def _http_bridge_quarantine_generation(service: Any, key: _HTTPBridgeSessionKey)
     return entry.generation
 
 
-def _quarantine_http_bridge_session(service: Any, session: _HTTPBridgeSession, *, reason: str) -> None:
+def _quarantine_http_bridge_session(
+    service: Any,
+    session: _HTTPBridgeSession,
+    *,
+    reason: str,
+    minimum_seconds: float | None = None,
+) -> 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.
+    ``minimum_seconds`` lets the retry circuit keep a poison quarantine alive
+    through its cooldown and half-open probe lease.
     """
     now = time.monotonic()
     registry = _http_bridge_quarantine_registry(service)
     entry = registry.setdefault(session.key, _HTTPBridgeQuarantineEntry())
     already_quarantined = entry.quarantined_until > now
     entry.generation += 1
-    entry.quarantined_until = max(entry.quarantined_until, now + _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS)
+    ttl_seconds = _HTTP_BRIDGE_QUARANTINE_TTL_SECONDS
+    if minimum_seconds is not None:
+        ttl_seconds = max(ttl_seconds, max(0.0, minimum_seconds))
+    entry.quarantined_until = max(entry.quarantined_until, now + ttl_seconds)
     entry.last_touched_monotonic = now
     entry.reason = reason
     _prune_http_bridge_quarantine_registry(registry, now)
@@ -135,7 +149,7 @@ def _quarantine_http_bridge_session(service: Any, session: _HTTPBridgeSession, *
         session.key,
         account_id=session.account.id,
         model=session.request_model,
-        detail=f"reason={reason}, ttl_seconds={_HTTP_BRIDGE_QUARANTINE_TTL_SECONDS:.0f}",
+        detail=f"reason={reason}, ttl_seconds={ttl_seconds:.0f}",
         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

@@ -78,6 +78,7 @@
 from app.modules.proxy._service.http_bridge.helpers import (
     _await_task_deferring_cancellation,
     _build_http_bridge_prewarm_text,
+    _http_bridge_abandonment_strands_requests,
     _http_bridge_durable_lease_ttl_seconds,
     _http_bridge_is_previous_response_owner_unavailable,
     _http_bridge_key_strength,
@@ -2946,7 +2947,12 @@ async def _retire_stale_pending_http_bridge_session(
                 # failure. Without this, only the admission-waiter reader
                 # path could ever poison an anchor, and an anchored session
                 # failing without waiters cooled down forever (issue #1830).
-                durable_cleared = await _abandon_durable_http_bridge_continuity(self, session, detail=poison_detail)
+                durable_cleared = await _abandon_durable_http_bridge_continuity(
+                    self,
+                    session,
+                    detail=poison_detail,
+                    settle_circuit=_http_bridge_abandonment_strands_requests(retired_request_states),
+                )
                 if not durable_cleared and session.durable_session_id is not None:
                     # Keep failed waiterless clears visible in the same
                     # poison-clear telemetry the admission-waiter path emits;

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

@@ -9,6 +9,10 @@
 import anyio
 
 from app.core.metrics.prometheus import PROMETHEUS_AVAILABLE, http_bridge_retry_circuit_total
+from app.modules.proxy._service.http_bridge.quarantine import (
+    _HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON,
+    _quarantine_http_bridge_session,
+)
 from app.modules.proxy._service.observability import _hash_identifier
 from app.modules.proxy._service.support import (
     _HTTPBridgeResponseCreateAttempt,
@@ -61,6 +65,11 @@ def _http_bridge_anchor_poison_detail(detail: str | None) -> str | None:
     return _HTTP_BRIDGE_ANCHOR_POISON_DETAILS.get(aliased)
 
 
+def _http_bridge_poison_quarantine_minimum_seconds(cooldown_remaining: float) -> float:
+    """Keep poison quarantine live through cooldown and the half-open probe."""
+    return max(0.0, cooldown_remaining) + _HTTP_BRIDGE_RETRY_CIRCUIT_HALF_OPEN_LEASE_SECONDS
+
+
 @dataclass(slots=True)
 class _HTTPBridgeRetryCircuitState:
     consecutive_failures: int = 0
@@ -610,6 +619,7 @@ async def _record_http_bridge_retry_circuit_failure(
         *,
         detail: str,
         attempt: _HTTPBridgeResponseCreateAttempt | None = None,
+        terminal_pre_response_frame: bool = False,
     ) -> int | None:
         detail = _HTTP_BRIDGE_RETRY_CIRCUIT_DETAIL_ALIASES.get(detail, detail)
         if session.key.strength != "hard" or detail not in _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_DETAILS:
@@ -623,7 +633,7 @@ async def _record_http_bridge_retry_circuit_failure(
                     attempt=scoped_attempt,
                     detail=detail,
                 )
-            if scoped_attempt.disarmed or scoped_attempt.response_observed:
+            if scoped_attempt.disarmed or (scoped_attempt.response_observed and not terminal_pre_response_frame):
                 return None
 
         await self._load_http_bridge_retry_circuit(session)
@@ -634,10 +644,15 @@ async def _record_http_bridge_retry_circuit_failure(
         now = time.monotonic()
         duplicate_attempt: _HTTPBridgeResponseCreateAttempt | None = None
         state: _HTTPBridgeRetryCircuitState | None = None
+        poison_class_failure = _http_bridge_anchor_poison_detail(detail) is not None
+        quarantine_poisoned_anchor = False
+        quarantine_cooldown_remaining = 0.0
         async with self._http_bridge_retry_circuit_lock:
             if scoped_attempt is not None and scoped_attempt.retry_circuit_failure_recorded:
                 duplicate_attempt = scoped_attempt
-            elif scoped_attempt is not None and (scoped_attempt.disarmed or scoped_attempt.response_observed):
+            elif scoped_attempt is not None and (
+                scoped_attempt.disarmed or (scoped_attempt.response_observed and not terminal_pre_response_frame)
+            ):
                 return None
             else:
                 state = self._http_bridge_retry_circuits.setdefault(
@@ -660,6 +675,9 @@ async def _record_http_bridge_retry_circuit_failure(
                     if detail == "clean_close":
                         backoff = min(backoff, clean_close_max_backoff)
                     state.cooldown_until = max(state.cooldown_until, now + backoff)
+                    if poison_class_failure:
+                        quarantine_poisoned_anchor = True
+                        quarantine_cooldown_remaining = max(0.0, state.cooldown_until - now)
                     if PROMETHEUS_AVAILABLE and http_bridge_retry_circuit_total is not None:
                         http_bridge_retry_circuit_total.labels(outcome="opened").inc()
                     logger.warning(
@@ -678,18 +696,54 @@ async def _record_http_bridge_retry_circuit_failure(
                 detail=detail,
             )
         assert state is not None
+        if quarantine_poisoned_anchor:
+            _quarantine_http_bridge_session(
+                self,
+                session,
+                reason=_HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON,
+                minimum_seconds=_http_bridge_poison_quarantine_minimum_seconds(quarantine_cooldown_remaining),
+            )
         try:
             await self._persist_http_bridge_retry_circuit(session, state)
+            merged_quarantine_cooldown_remaining = 0.0
             async with self._http_bridge_retry_circuit_lock:
                 if self._http_bridge_retry_circuits.get(session.key) is state:
                     self._http_bridge_retry_circuit_loaded_keys.add(session.key)
                 consecutive_failures = state.consecutive_failures
+                if poison_class_failure and consecutive_failures >= threshold and not quarantine_poisoned_anchor:
+                    # A durable conflict merge can open the circuit even when
+                    # this replica's local count was below the threshold.
+                    merged_quarantine_cooldown_remaining = max(0.0, state.cooldown_until - time.monotonic())
+            if merged_quarantine_cooldown_remaining or (
+                poison_class_failure and consecutive_failures >= threshold and not quarantine_poisoned_anchor
+            ):
+                _quarantine_http_bridge_session(
+                    self,
+                    session,
+                    reason=_HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON,
+                    minimum_seconds=_http_bridge_poison_quarantine_minimum_seconds(
+                        merged_quarantine_cooldown_remaining
+                    ),
+                )
             return consecutive_failures
         finally:
             if scoped_attempt is not None and scoped_attempt.retry_circuit_failure_settled is not None:
                 scoped_attempt.retry_circuit_failure_settled.set()
 
-    async def _clear_http_bridge_retry_circuit(self: Any, session: _HTTPBridgeSession) -> None:
+    async def _clear_http_bridge_retry_circuit(
+        self: Any,
+        session: _HTTPBridgeSession,
+        *,
+        settle_unfenced: bool = False,
+    ) -> None:
+        """Settle a hard key's retry circuit.
+
+        ``settle_unfenced`` is asserted by callers that have removed the cause
+        of the circuit rather than merely outlived it. The normal version
+        fence protects a durable row another replica may have created, while
+        cause-removal paths may also clear a circuit opened before its first
+        durable write completed.
+        """
         if session.key.strength != "hard":
             return
 
@@ -705,7 +759,7 @@ async def _clear_http_bridge_retry_circuit(self: Any, session: _HTTPBridgeSessio
         # concurrently, so leave the durable row untouched when no state was
         # observed. Preserve the existing best-effort clear on read failures,
         # which is still useful for settling a row after a transient outage.
-        if durable_load_succeeded and (state is None or expected_updated_at_epoch is None):
+        if durable_load_succeeded and (state is None or (expected_updated_at_epoch is None and not settle_unfenced)):
             return
         try:
             # Clearing is idempotent and must be attempted even when the

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

@@ -73,6 +73,7 @@
 )
 from app.modules.proxy._service.http_bridge.helpers import (
     _effective_http_bridge_idle_ttl_seconds,
+    _http_bridge_continuity_bound_without_safe_replay,
     _http_bridge_durable_lease_ttl_seconds,
     _http_bridge_durable_lookup_allows_turn_state_takeover,
     _http_bridge_is_context_overflow_error,
@@ -246,15 +247,6 @@
 _RESPONSE_CREATE_GATE_RETRY_SLEEP_SECONDS = 10.0
 
 
-def _http_bridge_continuity_bound_without_safe_replay(request_state: _WebSocketRequestState) -> bool:
-    """Return whether retrying would require replaying an unsafe continuation."""
-    if request_state.previous_response_id is not None:
-        return not (request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text)
-    return request_state.hard_continuity_anchor and not (
-        request_state.fresh_upstream_request_is_retry_safe and request_state.fresh_upstream_request_text
-    )
-
-
 def _http_bridge_durable_recovery_predecessor_proven(request_state: _WebSocketRequestState) -> bool:
     """Return whether the operation has a durable predecessor anchor."""
     return request_state.previous_response_id is not None or request_state.operation_parent_response_id is not None