fix(http-bridge): stabilize silent and clean-close recovery

.all-contributorsrc

@@ -1082,6 +1082,25 @@
       "contributions": [
         "code"
       ]
+    },
+    {
+      "login": "shaqman",
+      "name": "Syakur Rahman",
+      "avatar_url": "https://avatars.githubusercontent.com/u/1113851?v=4",
+      "profile": "http://expressthisout.com/",
+      "contributions": [
+        "code",
+        "test"
+      ]
+    },
+    {
+      "login": "cigro-manager",
+      "name": "cigro-manager",
+      "avatar_url": "https://avatars.githubusercontent.com/u/219247995?v=4",
+      "profile": "https://github.com/cigro-manager",
+      "contributions": [
+        "code"
+      ]
     }
   ],
   "contributorsPerLine": 7,

README.md

@@ -265,6 +265,7 @@ Thanks goes to these wonderful people ([emoji key](https://allcontributors.org/e
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/alchemistkiv"><img src="https://avatars.githubusercontent.com/u/93935225?v=4?s=100" width="100px;" alt="alchemistkiv"/><br /><sub><b>alchemistkiv</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=alchemistkiv" title="Code">💻</a> <a href="https://github.com/Soju06/codex-lb/commits?author=alchemistkiv" title="Tests">⚠️</a></td>
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/mhooooo"><img src="https://avatars.githubusercontent.com/u/142386428?v=4?s=100" width="100px;" alt="mhooooo"/><br /><sub><b>mhooooo</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=mhooooo" title="Code">💻</a> <a href="https://github.com/Soju06/codex-lb/commits?author=mhooooo" title="Tests">⚠️</a></td>
       <td align="center" valign="top" width="14.28%"><a href="https://github.com/crowscc"><img src="https://avatars.githubusercontent.com/u/17800561?v=4?s=100" width="100px;" alt="crowscc"/><br /><sub><b>crowscc</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=crowscc" title="Code">💻</a></td>
+      <td align="center" valign="top" width="14.28%"><a href="http://expressthisout.com/"><img src="https://avatars.githubusercontent.com/u/1113851?v=4?s=100" width="100px;" alt="Syakur Rahman"/><br /><sub><b>Syakur Rahman</b></sub></a><br /><a href="https://github.com/Soju06/codex-lb/commits?author=shaqman" title="Code">💻</a> <a href="https://github.com/Soju06/codex-lb/commits?author=shaqman" title="Tests">⚠️</a></td>
     </tr>
   </tbody>
 </table>

app/core/clients/proxy.py

@@ -477,6 +477,7 @@ def __init__(
         upstream_status_code: int | None = None,
         upstream_error_code: str | None = None,
         failed_session: aiohttp.ClientSession | None = None,
+        retry_after_seconds: int | None = None,
     ) -> None:
         super().__init__(f"Proxy response error ({status_code})")
         self.status_code = status_code
@@ -488,6 +489,7 @@ def __init__(
         self.upstream_status_code = upstream_status_code
         self.upstream_error_code = upstream_error_code
         self.failed_session = failed_session
+        self.retry_after_seconds = retry_after_seconds
 
 
 def _process_network_failure_error(

app/core/clients/proxy_websocket.py

@@ -138,6 +138,24 @@ def normalize_realtime_call_id(value: str) -> str | None:
     return normalized.lower()
 
 
+def _consume_connection_lost_exception(done: asyncio.Future[Any]) -> None:
+    """Retrieve close exceptions before websockets shields the waiter.
+
+    websockets 16 waits on ``connection_lost_waiter`` through
+    ``asyncio.shield`` while completing ``ClientConnection.recv``.  A peer
+    keepalive/protocol close therefore leaves an exception on the waiter,
+    which asyncio reports as an ``exception in shielded future`` even though
+    ``recv`` translates it into an ``UpstreamWebSocketMessage``.  Consume it
+    at the adapter boundary; ``receive`` still classifies the close normally.
+    """
+    if done.cancelled():
+        return
+    try:
+        done.exception()
+    except asyncio.CancelledError:
+        return
+
+
 @dataclass(slots=True)
 class UpstreamWebSocketMessage:
     kind: str
@@ -170,7 +188,14 @@ def _relay_receive_error_code(error_code: str) -> str | None:
 
     # Relay owners map an absent code to their established stream_incomplete
     # contract. Leaking the adapter's generic fallback would bypass that path.
-    return error_code if error_code == PROCESS_NETWORK_UNAVAILABLE_CODE else None
+    return error_code if error_code in {PROCESS_NETWORK_UNAVAILABLE_CODE, "upstream_keepalive_timeout"} else None
+
+
+def _is_keepalive_timeout_close(exc: ConnectionClosedError) -> bool:
+    """Classify peer/proxy heartbeat failures without exposing socket details."""
+
+    reason = _close_reason_from_exception(exc)
+    return "keepalive ping timeout" in f"{exc} {reason or ''}".lower()
 
 
 async def _rotate_after_websocket_network_failure(error_code: str) -> None:
@@ -223,6 +248,9 @@ def __init__(
         self._connection = connection
         self._uses_proxy = uses_proxy
         self._preserve_close_semantics = preserve_close_semantics
+        connection_lost_waiter = getattr(connection, "connection_lost_waiter", None)
+        if isinstance(connection_lost_waiter, asyncio.Future):
+            connection_lost_waiter.add_done_callback(_consume_connection_lost_exception)
 
     async def send_text(self, text: str) -> None:
         try:
@@ -254,6 +282,11 @@ async def receive(self) -> UpstreamWebSocketMessage:
                 )
             error_code = _websocket_transport_error_code(exc, uses_proxy=self._uses_proxy)
             await _rotate_after_websocket_network_failure(error_code)
+            relay_error_code = (
+                "upstream_keepalive_timeout"
+                if _is_keepalive_timeout_close(exc)
+                else _relay_receive_error_code(error_code)
+            )
             # ConnectionClosedError describes an incomplete close handshake,
             # not generic transport provenance. Let Responses relay owners map
             # it to stream_incomplete while live relays preserve received closes.
@@ -265,7 +298,7 @@ async def receive(self) -> UpstreamWebSocketMessage:
                     if self._preserve_close_semantics
                     else str(exc)
                 ),
-                error_code=_relay_receive_error_code(error_code),
+                error_code=relay_error_code,
             )
         except Exception as exc:
             error_code = _websocket_transport_error_code(exc, uses_proxy=self._uses_proxy)

app/core/config/settings.py

@@ -299,6 +299,11 @@ class Settings(BaseSettings):
     http_responses_session_bridge_stuck_gate_retire_after_seconds: float = Field(default=300.0, gt=0)
     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(
+        default=2.0,
+        ge=0,
+        le=30.0,
+    )
     http_responses_session_bridge_gateway_safe_mode: bool = False
     http_responses_session_bridge_instance_id: str = Field(default_factory=_default_http_bridge_instance_id)
     http_responses_session_bridge_instance_ring: Annotated[list[str], NoDecode] = Field(default_factory=list)

app/core/metrics/prometheus.py

@@ -172,6 +172,18 @@ def labels(self, *args: str, **kwargs: str) -> "HistogramLike": ...
         ["strength"],
         registry=REGISTRY,
     )
+    bridge_handoff_compatibility_rejection_total = Counter(
+        "codex_lb_bridge_handoff_compatibility_rejection_total",
+        "Total closed HTTP bridge admission handoffs rejected for incompatible request context",
+        ["continuity_anchor", "preferred_account", "service_tier", "api_key_scope"],
+        registry=REGISTRY,
+    )
+    bridge_unanchored_handoff_recovery_total = Counter(
+        "codex_lb_bridge_unanchored_handoff_recovery_total",
+        "Total stale closed HTTP bridge admission handoffs recovered without a continuity anchor",
+        ["reason"],
+        registry=REGISTRY,
+    )
     bridge_local_rebind_total = Counter(
         "codex_lb_bridge_local_rebind_total",
         "Total bridge local rebinds by reason",
@@ -264,6 +276,24 @@ def labels(self, *args: str, **kwargs: str) -> "HistogramLike": ...
         ["reason", "affinity_kind", "model_class"],
         registry=REGISTRY,
     )
+    http_bridge_retry_circuit_total = Counter(
+        "codex_lb_http_bridge_retry_circuit_total",
+        "Total HTTP bridge automatic retry circuit outcomes",
+        ["outcome"],
+        registry=REGISTRY,
+    )
+    stream_keepalive_sent_total = Counter(
+        "codex_lb_stream_keepalive_sent_total",
+        "Total downstream SSE keepalive frames emitted by surface",
+        ["surface"],
+        registry=REGISTRY,
+    )
+    stream_idle_timeout_total = Counter(
+        "codex_lb_stream_idle_timeout_total",
+        "Total streams terminated after exceeding the configured idle window",
+        ["surface"],
+        registry=REGISTRY,
+    )
     cache_invalidation_bump_failures_total = Counter(
         "codex_lb_cache_invalidation_bump_failures_total",
         "Total cache invalidation version bumps that failed after retries",
@@ -315,6 +345,8 @@ def mark_process_dead() -> None:
     bridge_first_turn_timeout_total: CounterLike | None = None
     bridge_drain_recovery_allowed_total: CounterLike | None = None
     bridge_owner_mismatch_total: CounterLike | None = None
+    bridge_handoff_compatibility_rejection_total: CounterLike | None = None
+    bridge_unanchored_handoff_recovery_total: CounterLike | None = None
     bridge_local_rebind_total: CounterLike | None = None
     bridge_forward_latency_seconds: HistogramLike | None = None
     bridge_public_contract_error_total: CounterLike | None = None
@@ -329,6 +361,9 @@ def mark_process_dead() -> None:
     proxy_phase_latency_seconds: HistogramLike | None = None
     http_bridge_prewarm_total: CounterLike | None = None
     http_bridge_stuck_retire_total: CounterLike | None = None
+    http_bridge_retry_circuit_total: CounterLike | None = None
+    stream_keepalive_sent_total: CounterLike | None = None
+    stream_idle_timeout_total: CounterLike | None = None
     cache_invalidation_bump_failures_total: CounterLike | None = None
     cache_invalidation_poll_failures_total: CounterLike | None = None
 
@@ -355,6 +390,7 @@ def mark_process_dead() -> None:
     "bridge_durable_recover_total",
     "bridge_drain_recovery_allowed_total",
     "bridge_first_turn_timeout_total",
+    "bridge_handoff_compatibility_rejection_total",
     "bridge_local_rebind_total",
     "bridge_owner_forward_total",
     "bridge_owner_mismatch_total",
@@ -363,14 +399,18 @@ def mark_process_dead() -> None:
     "bridge_reattach_total",
     "bridge_same_account_takeover_total",
     "bridge_soft_local_rebind_total",
+    "bridge_unanchored_handoff_recovery_total",
     "cache_invalidation_bump_failures_total",
     "cache_invalidation_poll_failures_total",
     "cap_partition_replicas",
     "circuit_breaker_state",
     "continuity_fail_closed_total",
     "continuity_owner_resolution_total",
     "http_bridge_prewarm_total",
+    "http_bridge_retry_circuit_total",
     "http_bridge_stuck_retire_total",
+    "stream_keepalive_sent_total",
+    "stream_idle_timeout_total",
     "image_request_duration_seconds",
     "image_requests_total",
     "make_scrape_registry",

app/core/utils/sse.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import re
-from collections.abc import AsyncIterator, Mapping
+from collections.abc import AsyncIterator, Callable, Mapping
 
 from app.core.errors import ResponseFailedEvent
 from app.core.types import JsonValue
@@ -26,6 +26,7 @@ async def inject_sse_keepalives(
     interval_seconds: float,
     *,
     keepalive_frame: str = SSE_KEEPALIVE_FRAME,
+    on_keepalive: Callable[[], None] | None = None,
 ) -> AsyncIterator[str]:
     """Wrap an SSE event iterator and emit comment heartbeats on idle gaps.
 
@@ -56,6 +57,8 @@ async def _next_chunk(it: AsyncIterator[str]) -> str:
                     timeout=interval_seconds,
                 )
             except asyncio.TimeoutError:
+                if on_keepalive is not None:
+                    on_keepalive()
                 yield keepalive_frame
                 continue
             except StopAsyncIteration:

app/db/alembic/versions/20260717_000000_add_http_bridge_retry_circuits.py

@@ -0,0 +1,47 @@
+"""add durable HTTP bridge retry circuit state
+
+Revision ID: 20260717_000000_add_http_bridge_retry_circuits
+Revises: 20260717_000000_optimize_dashboard_hot_path_indexes
+Create Date: 2026-07-17
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "20260717_000000_add_http_bridge_retry_circuits"
+down_revision = "20260717_000000_optimize_dashboard_hot_path_indexes"
+branch_labels = None
+depends_on = None
+
+_TABLE_NAME = "http_bridge_retry_circuits"
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if inspector.has_table(_TABLE_NAME):
+        return
+    op.create_table(
+        _TABLE_NAME,
+        sa.Column("session_key_kind", sa.String(length=64), nullable=False),
+        sa.Column("session_key_hash", sa.String(length=64), nullable=False),
+        sa.Column("api_key_scope", sa.String(length=255), nullable=False),
+        sa.Column("consecutive_failures", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("cooldown_until_epoch", sa.Float(), nullable=False, server_default="0"),
+        sa.Column("last_detail", sa.String(length=255), nullable=True),
+        sa.Column("updated_at_epoch", sa.Float(), nullable=False),
+        sa.PrimaryKeyConstraint(
+            "session_key_kind",
+            "session_key_hash",
+            "api_key_scope",
+            name="pk_http_bridge_retry_circuits",
+        ),
+    )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if sa.inspect(bind).has_table(_TABLE_NAME):
+        op.drop_table(_TABLE_NAME)

app/db/alembic/versions/20260717_000001_merge_retry_circuits_and_dashboard_indexes.py

@@ -0,0 +1,22 @@
+"""merge retry circuits and dashboard index heads
+
+Revision ID: 20260717_000001_merge_retry_circuits_and_dashboard_indexes
+Revises:
+- 20260717_000000_add_http_bridge_retry_circuits
+Create Date: 2026-07-17 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+revision = "20260717_000001_merge_retry_circuits_and_dashboard_indexes"
+down_revision = ("20260717_000000_add_http_bridge_retry_circuits",)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

app/db/alembic/versions/20260724_000000_merge_request_log_schema_heads.py

@@ -0,0 +1,33 @@
+"""merge the deployed retry-circuit and request-log schema heads
+
+Revision ID: 20260724_000000_merge_request_log_schema_heads
+Revises:
+- 20260717_000001_merge_retry_circuits_and_dashboard_indexes
+- 20260722_000000_backfill_request_log_useragent_families
+Create Date: 2026-07-24 00:00:00.000000
+
+The deployed SQLite database was previously stamped at the retry-circuit
+merge revision while the request-log conversation-id branch was not applied.
+Keeping the retry-circuit merge and the request-usage rollup as parents lets
+Alembic apply the missing request-log revisions before converging on one head
+without leaving the rollup revision as a second head or rewriting migration
+history.
+"""
+
+from __future__ import annotations
+
+revision = "20260724_000000_merge_request_log_schema_heads"
+down_revision = (
+    "20260717_000001_merge_retry_circuits_and_dashboard_indexes",
+    "20260724_000000_add_request_usage_time_rollups",
+)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

app/db/alembic/versions/20260726_000000_repair_request_usage_rollups_after_merge.py

@@ -0,0 +1,37 @@
+"""repair request-usage rollups for databases stamped at the old merge head
+
+Revision ID: 20260726_000000_repair_request_usage_rollups_after_merge
+Revises: 20260724_000000_merge_request_log_schema_heads
+Create Date: 2026-07-26
+
+Some deployed databases were stamped at the request-log merge revision before
+the request-usage rollup child was connected to that merge. Changing the
+parent tuple cannot make Alembic replay an already-applied revision, so those
+databases need a forward-only repair step. The canonical rollup migration is
+idempotent and safely creates any missing tables or watermark column here.
+"""
+
+from __future__ import annotations
+
+import importlib
+from types import ModuleType
+
+revision = "20260726_000000_repair_request_usage_rollups_after_merge"
+down_revision = "20260724_000000_merge_request_log_schema_heads"
+branch_labels = None
+depends_on = None
+
+
+def _rollup_migration() -> ModuleType:
+    return importlib.import_module("app.db.alembic.versions.20260724_000000_add_request_usage_time_rollups")
+
+
+def upgrade() -> None:
+    _rollup_migration().upgrade()
+
+
+def downgrade() -> None:
+    # This revision repairs databases that were already stamped at the merge
+    # head. It must never remove objects owned by the canonical rollup
+    # revision, which remains an ancestor of that merge head on fresh installs.
+    pass