fix(proxy): sweep idle bridge sessions without request traffic

app/main.py

@@ -137,6 +137,31 @@ def _ensure_web_asset_mime_types() -> None:
 _ensure_web_asset_mime_types()
 
 
+async def run_http_bridge_heartbeat_maintenance(proxy_service: Any) -> None:
+    """Per-replica bridge upkeep driven by the ring heartbeat.
+
+    Both passes are request-independent by design: durable ownership must be
+    reconciled even on a replica nothing is routing to, and the idle sweep is
+    otherwise only reached from ``_get_or_create_http_bridge_session``, so a
+    replica that stops taking bridge requests would keep its idle sessions'
+    upstream WebSockets open until restart (issue #1354). Each pass is isolated
+    so one failing cannot skip the other or stop the heartbeat.
+    """
+    if proxy_service is None:
+        return
+    for attribute, failure_message in (
+        ("reconcile_durable_http_bridge_ownership", "HTTP bridge durable ownership reconciliation failed"),
+        ("prune_idle_http_bridge_sessions", "HTTP bridge idle sweep failed"),
+    ):
+        pass_callable = getattr(proxy_service, attribute, None)
+        if pass_callable is None:
+            continue
+        try:
+            await pass_callable()
+        except Exception:
+            logger.warning(failure_message, exc_info=True)
+
+
 def _log_abandoned_lease_release(task: asyncio.Task[None]) -> None:
     if task.cancelled():
         return
@@ -543,12 +568,7 @@ async def _heartbeat_only(svc: RingMembershipService, iid: str) -> None:
                 await svc.heartbeat(iid, endpoint_base_url=bridge_endpoint_base_url)
             except Exception:
                 logger.warning("Ring heartbeat failed", exc_info=True)
-            proxy_service = getattr(app.state, "proxy_service", None)
-            if proxy_service is not None and hasattr(proxy_service, "reconcile_durable_http_bridge_ownership"):
-                try:
-                    await proxy_service.reconcile_durable_http_bridge_ownership()
-                except Exception:
-                    logger.warning("HTTP bridge durable ownership reconciliation failed", exc_info=True)
+            await run_http_bridge_heartbeat_maintenance(getattr(app.state, "proxy_service", None))
             await refresh_cap_partition(svc.list_active, iid)
 
     async def _register_and_heartbeat(svc: RingMembershipService, iid: str) -> None:

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

@@ -3,6 +3,7 @@
 import asyncio
 import logging
 from collections.abc import Mapping
+from typing import Any
 
 from app.core.clients.proxy import ProxyResponseError
 from app.core.config.settings import Settings
@@ -73,6 +74,21 @@ def _requires_durable_recovery_alias_serialization(session: _HTTPBridgeSession)
 
 
 class _HTTPBridgeSessionRegistryMixin:
+    async def prune_idle_http_bridge_sessions(self: Any) -> int:
+        """Run the idle sweep off the request path (issue #1354).
+
+        The sweep is otherwise reached only from
+        ``_get_or_create_http_bridge_session``, so a replica that stops taking
+        bridge requests keeps idle sessions' upstream WebSockets open until
+        restart. Heartbeat-driven, so it runs without traffic or leadership.
+        """
+        async with self._http_bridge_lock:
+            pruned_sessions = self._prune_http_bridge_sessions_locked()
+        if not pruned_sessions:
+            return 0
+        self._schedule_http_bridge_session_closes(pruned_sessions, reason="idle_sweep")
+        return len(pruned_sessions)
+
     async def _register_http_bridge_turn_state(
         self: _HTTPBridgeServiceProtocol,
         session: _HTTPBridgeSession,

openspec/changes/sweep-idle-bridge-sessions-off-request-path/proposal.md

@@ -0,0 +1,21 @@
+## Why
+
+The HTTP-bridge idle sweep (`_prune_http_bridge_sessions_locked`) is reached from exactly one place: `_get_or_create_http_bridge_session`. It is therefore request-driven, so a replica that stops receiving bridge requests never evicts its idle sessions and holds their upstream WebSockets, registry entries, and durable claims until the process restarts.
+
+This is the residual leg of issue #1354. Fix (a) (#1476) already made account stream leases turn-scoped, so an idle session releases its cap slot as soon as its last turn detaches — the cap-exhaustion symptom is addressed there. What remains is resource hygiene on a quiet replica: in a multi-replica deployment, traffic moving off one replica leaves its warm sessions pinned indefinitely.
+
+## What Changes
+
+- Expose the existing sweep as `prune_idle_http_bridge_sessions()`: take the bridge lock, run the same `_prune_http_bridge_sessions_locked` selection the request path uses, and schedule the pruned sessions' closes through the existing bounded close scheduler.
+- Drive it from the per-replica ring heartbeat, alongside the durable-ownership reconcile that already runs there, so the sweep happens on every replica regardless of traffic, leadership, or durable-row cleanup.
+- No new selection logic and no new timing: eligibility, the idle TTL that protects a session freshly handed to a request, and the close path are unchanged.
+
+## Capabilities
+
+### New Capabilities
+
+None.
+
+### Modified Capabilities
+
+- `sticky-session-operations`: idle bridge-session eviction no longer depends on request traffic reaching the replica.

openspec/changes/sweep-idle-bridge-sessions-off-request-path/specs/sticky-session-operations/spec.md

@@ -0,0 +1,28 @@
+## ADDED Requirements
+
+### Requirement: Idle bridge sessions are swept without request traffic
+
+The system MUST evict idle HTTP-bridge sessions on every replica independently of whether that replica is receiving bridge requests. The sweep MUST reuse the same eligibility the request path applies — a session with pending or queued work, an admission waiter, a handoff in progress, or an unanchored reservation, and a session still inside its idle TTL, MUST NOT be evicted — and MUST close evicted sessions through the existing bounded close path so a slow upstream-reader cancellation cannot block the caller. A sweep failure MUST NOT interrupt the loop that drives it, and MUST NOT prevent the other per-replica bridge upkeep that shares that loop from running.
+
+#### Scenario: A replica with no bridge traffic still evicts idle sessions
+
+- **GIVEN** a replica holds an idle bridge session past its idle TTL and receives no further bridge requests
+- **WHEN** the sweep runs
+- **THEN** the session is detached from the registry and closed, releasing its upstream WebSocket
+
+#### Scenario: Sweep eligibility matches the request path
+
+- **GIVEN** a session with pending work whose idle TTL has elapsed, and a session used moments ago
+- **WHEN** the sweep runs
+- **THEN** neither session is evicted
+
+#### Scenario: One failing upkeep pass does not skip the other
+
+- **GIVEN** the durable-ownership reconcile raises on a heartbeat tick
+- **WHEN** that tick runs
+- **THEN** the idle sweep still runs and the heartbeat loop continues
+
+#### Scenario: Sweeping an empty registry does nothing
+
+- **WHEN** the sweep runs with no registered bridge sessions
+- **THEN** no session is closed and no cleanup work is scheduled

openspec/changes/sweep-idle-bridge-sessions-off-request-path/tasks.md

@@ -0,0 +1,18 @@
+## 1. Sweep entry point
+
+- [x] 1.1 Add `prune_idle_http_bridge_sessions()` to the bridge session-registry mixin (mixin.py is at its architecture line ratchet): take the bridge lock, reuse `_prune_http_bridge_sessions_locked`, and schedule closes via `_schedule_http_bridge_session_closes` with reason `idle_sweep`
+
+## 2. Heartbeat wiring
+
+- [x] 2.1 Extract the heartbeat's bridge upkeep into `run_http_bridge_heartbeat_maintenance()` in `app/main.py` and call the sweep there beside the durable-ownership reconcile, isolating each pass so one failing cannot skip the other or stop the heartbeat
+
+## 3. Tests
+
+- [x] 3.1 Idle session is evicted with no request traffic; a freshly-used session is spared
+- [x] 3.2 A session with pending work is spared even past its idle TTL
+- [x] 3.3 Empty registry is a no-op and schedules no cleanup task
+- [x] 3.4 Heartbeat maintenance runs both passes, isolates a failing one, and tolerates a missing service — so removing the wiring fails a test rather than silently restoring the leak
+
+## 4. Spec
+
+- [x] 4.1 Record that idle eviction does not depend on request traffic reaching the replica

tests/unit/test_proxy_http_bridge.py

@@ -26975,3 +26975,89 @@ async def test_http_bridge_eventless_timeout_signal_drains_after_repeated_sessio
     await http_bridge_upstream_events_module._record_http_bridge_account_timeout_signal(service, session)
     await http_bridge_upstream_events_module._record_http_bridge_account_timeout_signal(service, session)
     record_errors.assert_awaited_once_with(account, 2)
+
+
+@pytest.mark.asyncio
+async def test_prune_idle_http_bridge_sessions_evicts_without_request_traffic() -> None:
+    """The idle sweep is otherwise only reached from the request path, so a
+    replica that stops taking bridge requests would keep idle sessions'
+    upstream WebSockets open until restart (issue #1354)."""
+    service = proxy_service.ProxyService(cast(Any, nullcontext()))
+    idle = _make_bridge_session(key_value="idle-no-traffic")
+    idle.last_used_at = time.monotonic() - (idle.idle_ttl_seconds + 60.0)
+    fresh = _make_bridge_session(key_value="fresh-no-traffic")
+    fresh.last_used_at = time.monotonic()
+    service._http_bridge_sessions[idle.key] = idle
+    service._http_bridge_sessions[fresh.key] = fresh
+
+    pruned_count = await service.prune_idle_http_bridge_sessions()
+    await asyncio.gather(*service._background_cleanup_tasks)
+
+    assert pruned_count == 1
+    assert idle.key not in service._http_bridge_sessions
+    assert fresh.key in service._http_bridge_sessions
+    assert fresh.closed is False
+
+
+@pytest.mark.asyncio
+async def test_prune_idle_http_bridge_sessions_spares_sessions_with_pending_work() -> None:
+    """The sweep reuses _prune_http_bridge_sessions_locked, so a session with
+    in-flight work keeps its own lifecycle even past the idle TTL."""
+    service = proxy_service.ProxyService(cast(Any, nullcontext()))
+    busy = _make_bridge_session(key_value="busy-no-traffic", queued_request_count=1)
+    busy.last_used_at = time.monotonic() - (busy.idle_ttl_seconds + 60.0)
+    service._http_bridge_sessions[busy.key] = busy
+
+    assert await service.prune_idle_http_bridge_sessions() == 0
+    assert busy.key in service._http_bridge_sessions
+    assert busy.closed is False
+
+
+@pytest.mark.asyncio
+async def test_prune_idle_http_bridge_sessions_is_a_noop_on_an_empty_registry() -> None:
+    service = proxy_service.ProxyService(cast(Any, nullcontext()))
+
+    assert await service.prune_idle_http_bridge_sessions() == 0
+    assert not service._background_cleanup_tasks
+
+
+@pytest.mark.asyncio
+async def test_heartbeat_maintenance_runs_both_bridge_passes() -> None:
+    """The ring heartbeat is what makes these request-independent. Asserting the
+    sweep only through a direct call would still pass if the wiring were
+    removed, leaving the quiet-replica leak (issue #1354)."""
+    from app.main import run_http_bridge_heartbeat_maintenance
+
+    proxy_service_double = SimpleNamespace(
+        reconcile_durable_http_bridge_ownership=AsyncMock(return_value=0),
+        prune_idle_http_bridge_sessions=AsyncMock(return_value=0),
+    )
+
+    await run_http_bridge_heartbeat_maintenance(proxy_service_double)
+
+    proxy_service_double.reconcile_durable_http_bridge_ownership.assert_awaited_once()
+    proxy_service_double.prune_idle_http_bridge_sessions.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_heartbeat_maintenance_isolates_a_failing_pass() -> None:
+    """A failing reconcile must not skip the sweep, and neither may stop the
+    heartbeat loop."""
+    from app.main import run_http_bridge_heartbeat_maintenance
+
+    proxy_service_double = SimpleNamespace(
+        reconcile_durable_http_bridge_ownership=AsyncMock(side_effect=RuntimeError("durable read failed")),
+        prune_idle_http_bridge_sessions=AsyncMock(return_value=0),
+    )
+
+    await run_http_bridge_heartbeat_maintenance(proxy_service_double)
+
+    proxy_service_double.prune_idle_http_bridge_sessions.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_heartbeat_maintenance_tolerates_a_missing_service_or_pass() -> None:
+    from app.main import run_http_bridge_heartbeat_maintenance
+
+    await run_http_bridge_heartbeat_maintenance(None)
+    await run_http_bridge_heartbeat_maintenance(SimpleNamespace())