fix(proxy): unify eventless anchor poison settlement

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

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.modules.proxy._service.http_bridge.helpers import _log_http_bridge_event
+from app.modules.proxy._service.http_bridge.service_stubs import _service_get_settings
+from app.modules.proxy.affinity import _extract_model_class
+
+logger = logging.getLogger(__name__)
+
+
+def _log_durable_anchor_poison_clear_failed(session: Any, detail: str) -> None:
+    if session.durable_session_id is None:
+        return
+    _log_http_bridge_event(
+        "durable_anchor_poison_clear_failed",
+        session.key,
+        account_id=session.account.id,
+        model=session.request_model,
+        pending_count=len(session.pending_requests),
+        detail=detail,
+        cache_key_family=session.key.affinity_kind,
+        model_class=_extract_model_class(session.request_model) if session.request_model else None,
+    )
+
+
+async def _abandon_durable_http_bridge_continuity(
+    service: Any,
+    session: Any,
+    *,
+    detail: str = "repeated_zero_event_idle_timeout",
+    settle_circuit: bool = False,
+) -> bool:
+    """Clear durable continuity while the failed session still owns its row.
+
+    The write is fenced by the session owner epoch.  A failed or fenced clear
+    is deliberately reported as ``False`` so callers keep the retry circuit
+    open and do not treat cooldown expiry as proof that the poisoned anchor is
+    gone.
+    """
+    if session.durable_session_id is None or session.durable_owner_epoch is None:
+        session.anchor_poison_clear_failed = True
+        return False
+    try:
+        cleared = await service._durable_bridge.rebind_session_account(
+            session_id=session.durable_session_id,
+            api_key_id=session.key.api_key_id,
+            instance_id=_service_get_settings().http_responses_session_bridge_instance_id,
+            owner_epoch=session.durable_owner_epoch,
+            account_id=session.account.id,
+            clear_continuity=True,
+        )
+    except Exception:
+        session.anchor_poison_clear_failed = True
+        _log_durable_anchor_poison_clear_failed(session, detail)
+        logger.warning("Failed to abandon poisoned HTTP bridge continuity", exc_info=True)
+        return False
+    if not cleared:
+        session.anchor_poison_clear_failed = True
+        _log_durable_anchor_poison_clear_failed(session, detail)
+        logger.warning(
+            "Durable bridge continuity clear was fenced before poisoned anchor retirement",
+            extra={
+                "session_id": session.durable_session_id,
+                "account_id": session.account.id,
+            },
+        )
+        return False
+
+    session.anchor_poison_cleared = True
+    session.anchor_poison_clear_failed = False
+    _log_http_bridge_event(
+        "durable_anchor_poisoned",
+        session.key,
+        account_id=session.account.id,
+        model=session.request_model,
+        detail=detail,
+        cache_key_family=session.key.affinity_kind,
+        model_class=_extract_model_class(session.request_model) if session.request_model else None,
+    )
+    # The anchor is gone, so the circuit can be settled as the cause-removal
+    # step.  Keep this after the fenced continuity write: if circuit cleanup
+    # itself is unavailable, the durable poison row remains conservative and a
+    # later healthy response can still clear it.
+    if settle_circuit:
+        await service._clear_http_bridge_retry_circuit(session, settle_unfenced=True)
+    return True

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

@@ -206,6 +206,11 @@ def _clear_http_bridge_quarantine(
     """A completed response disproves the current and recovery-origin wedges."""
     registry = _http_bridge_quarantine_registry(service)
     session.quarantined = False
+    # A completed response establishes a new healthy continuity anchor. Any
+    # poison-settlement result from the previous lifecycle must not suppress
+    # future anchor handling on this session.
+    session.anchor_poison_cleared = False
+    session.anchor_poison_clear_failed = False
     keys = (session.key,) if additional_key is None or additional_key == session.key else (session.key, additional_key)
     for key in keys:
         entry = registry.pop(key, None)

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

@@ -78,7 +78,6 @@
 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,
@@ -96,9 +95,6 @@
 from app.modules.proxy._service.http_bridge.quarantine import (
     _record_http_bridge_quarantine_wedged_pending,
 )
-from app.modules.proxy._service.http_bridge.retry_circuit import (
-    _http_bridge_anchor_poison_detail,
-)
 from app.modules.proxy._service.http_bridge.service_stubs import (
     _call_with_supported_optional_kwargs,
     _classify_upstream_close,
@@ -128,9 +124,6 @@
     _websocket_auth_failure_requires_reauth,
     _websocket_request_text_is_account_neutral_fresh_replay,
 )
-from app.modules.proxy._service.http_bridge.upstream_events import (
-    _abandon_durable_http_bridge_continuity,
-)
 from app.modules.proxy._service.observability import (
     _hash_identifier as _hash_identifier,
 )
@@ -2927,46 +2920,11 @@ async def _retire_stale_pending_http_bridge_session(
         # that handoff, genuine pre-response failures disappear from circuit
         # accounting while idle closes and request failures look identical.
         if retired_request_count > 0 and response_events_seen == 0:
-            consecutive_failures = await self._record_http_bridge_retry_circuit_failure_for_attempt_selection(
+            await self._record_http_bridge_retry_circuit_failure_for_attempt_selection(
                 session,
                 detail=retry_circuit_detail or detail,
                 selection=retry_circuit_attempt_selection,
             )
-            poison_detail = _http_bridge_anchor_poison_detail(retry_circuit_detail or detail)
-            if (
-                poison_detail is not None
-                and consecutive_failures is not None
-                and consecutive_failures
-                >= _service_get_settings().http_responses_session_bridge_anchor_poison_failure_threshold
-            ):
-                # Consecutive eventless failures on one bridge key are
-                # same-anchor failures (the anchor only advances on a
-                # completed response, which resets the circuit). Clear the
-                # poisoned durable anchor while this session still owns the
-                # lease so the next attempt is not re-anchored into the same
-                # 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,
-                    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;
-                    # the next threshold failure re-attempts the clear.
-                    _log_http_bridge_event(
-                        "durable_anchor_poison_clear_failed",
-                        session.key,
-                        account_id=session.account.id,
-                        model=session.request_model,
-                        pending_count=retired_request_count,
-                        detail=poison_detail,
-                        cache_key_family=session.key.affinity_kind,
-                        model_class=_extract_model_class(session.request_model) if session.request_model else None,
-                    )
         session.closed = True
         async with self._http_bridge_lock:
             # Bounded close may return while resource finalization is still

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

@@ -9,6 +9,9 @@
 import anyio
 
 from app.core.metrics.prometheus import PROMETHEUS_AVAILABLE, http_bridge_retry_circuit_total
+from app.modules.proxy._service.http_bridge.anchor_poison import (
+    _abandon_durable_http_bridge_continuity,
+)
 from app.modules.proxy._service.http_bridge.quarantine import (
     _HTTP_BRIDGE_QUARANTINE_POISONED_ANCHOR_REASON,
     _quarantine_http_bridge_session,
@@ -80,6 +83,8 @@ class _HTTPBridgeRetryCircuitState:
     last_failure_monotonic: float = 0.0
     last_durable_load_monotonic: float = 0.0
     half_open_until: float = 0.0
+    anchor_poison_clear_in_flight: bool = False
+    anchor_poison_clear_retry_after: float = 0.0
 
 
 def _initialize_http_bridge_retry_circuit(service: Any, reset_transient_cache: Any = None) -> None:
@@ -159,6 +164,38 @@ async def _http_bridge_retry_circuit_generation_for_key(
                 local_cooldown_until,
             )
 
+    async def _http_bridge_retry_circuit_anchor_poisoned_for_key(
+        self: Any,
+        key: _HTTPBridgeSessionKey,
+    ) -> bool:
+        """Return whether a durable poison-class circuit still needs clearing.
+
+        ``last_detail`` plus the threshold is the durable marker. A successful
+        fenced anchor settlement clears the circuit row, while a failed clear
+        leaves this marker in place so a restarted replica cannot re-inject the
+        old anchor after the in-memory quarantine expires.
+        """
+        try:
+            persisted = await self._durable_bridge.lookup_retry_circuit(
+                session_key_kind=key.affinity_kind,
+                session_key_value=key.affinity_key,
+                api_key_id=key.api_key_id,
+            )
+        except Exception:
+            return False
+        if persisted is not None:
+            return bool(
+                persisted.consecutive_failures >= _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_THRESHOLD
+                and _http_bridge_anchor_poison_detail(persisted.last_detail) is not None
+            )
+        async with self._http_bridge_retry_circuit_lock:
+            state = self._http_bridge_retry_circuits.get(key)
+            return bool(
+                state is not None
+                and state.consecutive_failures >= _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_THRESHOLD
+                and _http_bridge_anchor_poison_detail(state.last_detail) is not None
+            )
+
     async def _http_bridge_retry_circuit_generation_is_not_newer(
         self: Any,
         *,
@@ -647,6 +684,7 @@ async def _record_http_bridge_retry_circuit_failure(
         poison_class_failure = _http_bridge_anchor_poison_detail(detail) is not None
         quarantine_poisoned_anchor = False
         quarantine_cooldown_remaining = 0.0
+        anchor_clear_required = False
         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
@@ -678,6 +716,13 @@ async def _record_http_bridge_retry_circuit_failure(
                     if poison_class_failure:
                         quarantine_poisoned_anchor = True
                         quarantine_cooldown_remaining = max(0.0, state.cooldown_until - now)
+                        if (
+                            not getattr(session, "anchor_poison_cleared", False)
+                            and not state.anchor_poison_clear_in_flight
+                            and now >= state.anchor_poison_clear_retry_after
+                        ):
+                            state.anchor_poison_clear_in_flight = True
+                            anchor_clear_required = True
                     if PROMETHEUS_AVAILABLE and http_bridge_retry_circuit_total is not None:
                         http_bridge_retry_circuit_total.labels(outcome="opened").inc()
                     logger.warning(
@@ -705,6 +750,38 @@ async def _record_http_bridge_retry_circuit_failure(
             )
         try:
             await self._persist_http_bridge_retry_circuit(session, state)
+            if poison_class_failure and not anchor_clear_required:
+                # A durable conflict can merge another replica's strike into
+                # this state while persistence is in flight. Re-evaluate the
+                # threshold after the merge so the clear cannot be skipped.
+                async with self._http_bridge_retry_circuit_lock:
+                    if (
+                        self._http_bridge_retry_circuits.get(session.key) is state
+                        and state.consecutive_failures >= threshold
+                        and not getattr(session, "anchor_poison_cleared", False)
+                        and not state.anchor_poison_clear_in_flight
+                        and time.monotonic() >= state.anchor_poison_clear_retry_after
+                    ):
+                        state.anchor_poison_clear_in_flight = True
+                        anchor_clear_required = True
+            if anchor_clear_required:
+                poison_detail = _http_bridge_anchor_poison_detail(detail)
+                durable_anchor_cleared = False
+                if poison_detail is not None:
+                    durable_anchor_cleared = await _abandon_durable_http_bridge_continuity(
+                        self,
+                        session,
+                        detail=poison_detail,
+                        settle_circuit=True,
+                    )
+                async with self._http_bridge_retry_circuit_lock:
+                    if self._http_bridge_retry_circuits.get(session.key) is state:
+                        state.anchor_poison_clear_in_flight = False
+                        if not durable_anchor_cleared:
+                            state.anchor_poison_clear_retry_after = max(
+                                state.anchor_poison_clear_retry_after,
+                                state.cooldown_until,
+                            )
             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:

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

@@ -1353,6 +1353,7 @@ 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)
+        anchor_poison_circuit_unresolved = False
         # 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
@@ -1514,6 +1515,27 @@ def classify_durable_full_resend(
                 durable_lookup.canonical_key,
                 bridge_session_key.api_key_id,
             )
+            anchor_poison_circuit_unresolved = await self._http_bridge_retry_circuit_anchor_poisoned_for_key(
+                bridge_session_key
+            )
+            if anchor_poison_circuit_unresolved and not payload_looks_like_full_resend:
+                _record_continuity_fail_closed(
+                    surface="http_bridge",
+                    reason="poisoned_anchor_unresolved",
+                    previous_response_id=payload.previous_response_id,
+                    session_id=bridge_session_key.affinity_key,
+                    upstream_error_code="retry_circuit_poisoned_anchor",
+                )
+                raise ProxyResponseError(
+                    503,
+                    openai_error(
+                        "upstream_request_timeout",
+                        "The previous response anchor is temporarily unavailable while bridge recovery completes.",
+                    ),
+                    retryable_same_contract=True,
+                    failure_phase="upstream",
+                    failure_detail="retry_circuit_poisoned_anchor",
+                )
             live_local_session_exists = await self._http_bridge_has_live_local_session(
                 key=bridge_session_key,
                 incoming_turn_state=incoming_turn_state_header,
@@ -1530,7 +1552,9 @@ 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):
+            if payload_looks_like_full_resend and (
+                _http_bridge_session_key_quarantined(self, bridge_session_key) or anchor_poison_circuit_unresolved
+            ):
                 # 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