fix(proxy): recover dead durable bridge anchors fast

app/core/config/settings.py

@@ -297,6 +297,7 @@ class Settings(BaseSettings):
     http_responses_session_bridge_codex_idle_ttl_seconds: float = Field(default=900.0, gt=0)
     http_responses_session_bridge_codex_prewarm_enabled: bool = False
     http_responses_session_bridge_stuck_gate_retire_after_seconds: float = Field(default=300.0, gt=0)
+    http_responses_session_bridge_anchor_poison_failure_threshold: int = Field(default=7, ge=1, le=100)
     http_responses_session_bridge_max_sessions: int = Field(default=256, gt=0)
     http_responses_session_bridge_queue_limit: int = Field(default=8, gt=0)
     http_responses_session_bridge_clean_close_retry_jitter_max_seconds: float = Field(

app/db/alembic/versions/20260806_120000_add_http_bridge_owner_process_epoch.py

@@ -0,0 +1,57 @@
+"""add owner process epoch to durable HTTP bridge sessions
+
+Revision ID: 20260806_120000_add_http_bridge_owner_process_epoch
+Revises: 20260808_000000_tune_usage_history_autovacuum
+Create Date: 2026-08-06 12:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.engine import Connection
+
+revision = "20260806_120000_add_http_bridge_owner_process_epoch"
+down_revision = "20260808_000000_tune_usage_history_autovacuum"
+branch_labels = None
+depends_on = None
+
+_TABLE = "http_bridge_sessions"
+_COLUMN = "owner_process_epoch"
+
+
+def _columns(connection: Connection) -> set[str]:
+    inspector = sa.inspect(connection)
+    if not inspector.has_table(_TABLE):
+        return set()
+    return {str(column["name"]) for column in inspector.get_columns(_TABLE) if column.get("name") is not None}
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if _COLUMN in _columns(bind):
+        return
+    with op.batch_alter_table(_TABLE) as batch_op:
+        batch_op.add_column(sa.Column(_COLUMN, sa.String(length=64), nullable=True))
+    op.drop_index("idx_http_bridge_sessions_owner_state", table_name=_TABLE, if_exists=True)
+    op.create_index(
+        "idx_http_bridge_sessions_owner_state",
+        _TABLE,
+        ["owner_instance_id", _COLUMN, "state"],
+        if_not_exists=True,
+    )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if _COLUMN not in _columns(bind):
+        return
+    op.drop_index("idx_http_bridge_sessions_owner_state", table_name=_TABLE, if_exists=True)
+    op.create_index(
+        "idx_http_bridge_sessions_owner_state",
+        _TABLE,
+        ["owner_instance_id", "state"],
+        if_not_exists=True,
+    )
+    with op.batch_alter_table(_TABLE) as batch_op:
+        batch_op.drop_column(_COLUMN)

app/db/models.py

@@ -1791,6 +1791,7 @@ class HttpBridgeSessionRecord(Base):
     session_key_hash: Mapped[str] = mapped_column(String(64), nullable=False)
     api_key_scope: Mapped[str] = mapped_column(String(255), nullable=False)
     owner_instance_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    owner_process_epoch: Mapped[str | None] = mapped_column(String(64), nullable=True)
     owner_epoch: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0"))
     lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     state: Mapped[HttpBridgeSessionState] = mapped_column(
@@ -2126,7 +2127,12 @@ class HttpBridgeRetryCircuit(Base):
 Index("idx_automation_runs_status_started_at", AutomationRun.status, AutomationRun.started_at)
 Index("idx_automation_runs_scheduled_for", AutomationRun.scheduled_for)
 Index("idx_automation_runs_cycle_key_started_at", AutomationRun.cycle_key, AutomationRun.started_at)
-Index("idx_http_bridge_sessions_owner_state", HttpBridgeSessionRecord.owner_instance_id, HttpBridgeSessionRecord.state)
+Index(
+    "idx_http_bridge_sessions_owner_state",
+    HttpBridgeSessionRecord.owner_instance_id,
+    HttpBridgeSessionRecord.owner_process_epoch,
+    HttpBridgeSessionRecord.state,
+)
 Index("idx_http_bridge_sessions_lease", HttpBridgeSessionRecord.lease_expires_at)
 Index("idx_http_bridge_sessions_last_seen", HttpBridgeSessionRecord.last_seen_at.desc())
 Index(

app/main.py

@@ -81,6 +81,7 @@
 from app.modules.proxy.cap_partitioning import refresh_cap_partition
 from app.modules.proxy.durable_bridge_coordinator import DurableBridgeSessionCoordinator
 from app.modules.proxy.durable_bridge_repository import missing_durable_bridge_tables
+from app.modules.proxy.durable_bridge_runtime import http_bridge_owner_process_epoch
 from app.modules.proxy.rate_limit_cache import get_rate_limit_headers_cache
 from app.modules.proxy.ring_membership import (
     RING_HEARTBEAT_INTERVAL_SECONDS,
@@ -276,6 +277,7 @@ async def lifespan(app: FastAPI):
         )
         deleted_bridge_rows = await DurableBridgeSessionCoordinator(SessionLocal).purge_owned_sessions_on_startup(
             instance_id=settings.http_responses_session_bridge_instance_id,
+            owner_process_epoch=http_bridge_owner_process_epoch(),
             ownerless_cutoff=ownerless_cutoff,
         )
         if deleted_bridge_rows > 0:

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

@@ -341,10 +341,10 @@ async def _record_http_bridge_retry_circuit_failure(
         session: _HTTPBridgeSession,
         *,
         detail: str,
-    ) -> None:
+    ) -> 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:
-            return
+            return None
 
         await self._load_http_bridge_retry_circuit(session)
         threshold = max(1, _HTTP_BRIDGE_RETRY_CIRCUIT_FAILURE_THRESHOLD)
@@ -385,6 +385,7 @@ async def _record_http_bridge_retry_circuit_failure(
         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)
+            return state.consecutive_failures
 
     async def _clear_http_bridge_retry_circuit(self: Any, session: _HTTPBridgeSession) -> None:
         if session.key.strength != "hard":

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

@@ -47,6 +47,7 @@
     DurableBridgeAliasRegistration,
     DurableBridgeAliasRegistrationReceipt,
 )
+from app.modules.proxy.durable_bridge_runtime import http_bridge_owner_process_epoch
 
 logger = logging.getLogger("app.modules.proxy.service")
 
@@ -433,6 +434,7 @@ async def _claim_durable_http_bridge_session(
         clear_latest_turn_state: bool = False,
     ) -> None:
         current_instance = _service_get_settings().http_responses_session_bridge_instance_id
+        current_process_epoch = http_bridge_owner_process_epoch()
         try:
             lookup: DurableBridgeLookup | None = None
             for claim_attempt in range(2):
@@ -441,6 +443,7 @@ async def _claim_durable_http_bridge_session(
                     session_key_value=session.key.affinity_key,
                     api_key_id=session.key.api_key_id,
                     instance_id=current_instance,
+                    owner_process_epoch=current_process_epoch,
                     lease_ttl_seconds=_http_bridge_durable_lease_ttl_seconds(),
                     account_id=claim_account_id or session.account.id,
                     model=session.request_model,