Release — bound concurrent full-session resolution (#7127) (#7130)

* perf: bound concurrent full-session resolution

* Release — bound concurrent full-session resolution (#7127)

* ci: re-trigger fresh run (Playwright-install hang on prior run)

* ci: re-trigger (Playwright-install CDN hang on 3 shards, 12/15 passed)

* ci: re-trigger (sustained Playwright-CDN hang, run 4)

* ci: supersede 2h20m-zombie Playwright-hung run (run 6)

---------

Co-authored-by: Dr. Carlos Daniel Carvalho <292283903+Dandandad@users.noreply.github.com>
Co-authored-by: n <a@n>

CHANGELOG.md

@@ -5,6 +5,7 @@
 
 ### Fixed
 
+- **Loading several large sessions at once can no longer spike memory into an out-of-memory crash.** Full sidecar resolution (the expensive JSON parse for a large transcript) is now bounded by a small semaphore, and concurrent requests for the same session elect a single leader that parses once while the others wait for its cached result instead of each parsing their own copy — so a burst of tabs or reconnects against big sessions can't multiply transient memory. Metadata-only reads and cache hits stay on the lock-free fast path, and a failed parse always releases its slot (no capacity leak). Thanks @Dandandad. (#7127)
 - **Reconnecting to a live chat stream now resumes from the client's last received event instead of losing the tail or replaying the whole turn.** The `/api/chat/stream` endpoint honors the SSE `Last-Event-ID` header (and `after_event_id` / `after_seq` / `replay=1`) with a resume cursor that carries provenance, so a dropped connection re-subscribes at the right place: an in-range cursor replays only the gap, an unparseable/foreign/ahead cursor fails closed to replay-from-start (never silently skipping frames), and a genuinely truncated buffer still emits `recovery_control`. A prior edge where an ahead cursor with no known snapshot cutoff installed itself as the live dedup bound — discarding every queued frame including the terminal `stream_end` and stalling the reconnect on heartbeats — is fixed by treating an unknown cutoff as fence 0 before normalization. Thanks @Paladin173. (#6886)
 - **A parked no-run approval can no longer mask the real pending approval when a gateway producer is live.** No-run "legacy" approval mirrors are now stamped with a token derived from their own live producer, and whenever a session's gateway queue has live producers the reconciler tokenizes from the authoritative head and discards any unmatched or tokenless mirror instead of letting it linger — so a stale card can't sit in front of the genuine pending approval (previously gateway mode returned 409 and permanently retained the stale mirror, and local mode reported success while resolving nothing). Tokenless-orphan retention is preserved for the no-producer case, and responding to a stale id fails without waking a different producer. Thanks @VHSgunzo. (#7086)
 - **Toggling YOLO from an approval card no longer hangs the gateway run or leaks across a session switch.** The approval card's YOLO branch now takes the same single `gateway_yolo_handoff(sid)` as the Runs dispatcher (revalidating the exact mirror after acquisition and holding it through settlement, with no nested-lock path), and the client captures one immutable owner tuple (card-owner session, approval id, load generation) at click time — failing closed without a POST when the visible card is not the active session, and fencing every post-response mutation (card hide, pending clears, stale-clear requery, controls, toast, status) on that exact tuple so a stale card can't project its result into the successor session. Thanks @dss539. (#6945)

api/models.py

@@ -4684,7 +4684,46 @@ def get_session_for_scan(sid):
         return None
 
 
-def _resolve_session(sid, metadata_only=False, *, promote_cache=True, cache_on_miss=True):
+class _FullSessionResolveRequired(RuntimeError):
+    """Internal signal that a full sidecar load must enter the bounded path."""
+
+
+_FULL_SESSION_RESOLVE_MAX_CONCURRENT = 2
+_FULL_SESSION_RESOLVE_SLOTS = threading.BoundedSemaphore(
+    _FULL_SESSION_RESOLVE_MAX_CONCURRENT
+)
+_FULL_SESSION_RESOLVE_INFLIGHT: dict[str, threading.Event] = {}
+_FULL_SESSION_RESOLVE_INFLIGHT_LOCK = threading.Lock()
+_FULL_SESSION_RESOLVE_LOCAL = threading.local()
+
+
+def _claim_full_session_resolve(session_id: str) -> tuple[bool, threading.Event]:
+    """Return whether this caller owns the single-flight for ``session_id``."""
+    with _FULL_SESSION_RESOLVE_INFLIGHT_LOCK:
+        event = _FULL_SESSION_RESOLVE_INFLIGHT.get(session_id)
+        if event is not None:
+            return False, event
+        event = threading.Event()
+        _FULL_SESSION_RESOLVE_INFLIGHT[session_id] = event
+        return True, event
+
+
+def _finish_full_session_resolve(session_id: str, event: threading.Event) -> None:
+    """Release one single-flight owner and wake every waiter."""
+    with _FULL_SESSION_RESOLVE_INFLIGHT_LOCK:
+        if _FULL_SESSION_RESOLVE_INFLIGHT.get(session_id) is event:
+            _FULL_SESSION_RESOLVE_INFLIGHT.pop(session_id, None)
+    event.set()
+
+
+def _resolve_session_once(
+    sid,
+    metadata_only=False,
+    *,
+    promote_cache=True,
+    cache_on_miss=True,
+    allow_full_load=False,
+):
     """Resolve a session through the canonical freshness/recovery path.
 
     ``get_session_for_scan`` shares this resolver with normal reads so that
@@ -4715,6 +4754,8 @@ def _resolve_session(sid, metadata_only=False, *, promote_cache=True, cache_on_m
             cached = None
     if cached is not None:
         if not metadata_only and _cached_session_lags_disk(cached):
+            if not allow_full_load:
+                raise _FullSessionResolveRequired
             try:
                 disk_session = Session.load(sid)
                 with LOCK:
@@ -4727,6 +4768,8 @@ def _resolve_session(sid, metadata_only=False, *, promote_cache=True, cache_on_m
                     "cached session disk-freshness check failed for session %s", sid, exc_info=True,
                 )
         if not metadata_only and _inactive_cache_tail_needs_disk_check(cached):
+            if not allow_full_load:
+                raise _FullSessionResolveRequired
             try:
                 disk_session = Session.load(sid)
                 if _cache_has_stale_unsaved_user_tail(cached, disk_session):
@@ -4759,6 +4802,8 @@ def _resolve_session(sid, metadata_only=False, *, promote_cache=True, cache_on_m
         if s:
             return s
     else:
+        if not allow_full_load:
+            raise _FullSessionResolveRequired
         s = Session.load(sid)
     if s:
         if cache_on_miss:
@@ -4799,6 +4844,72 @@ def _resolve_session(sid, metadata_only=False, *, promote_cache=True, cache_on_m
     raise KeyError(sid)
 
 
+def _resolve_session(sid, metadata_only=False, *, promote_cache=True, cache_on_miss=True):
+    """Resolve a session while single-flighting heavyweight full sidecar loads.
+
+    Metadata-only reads and full sessions that remain current in the cache take
+    the lock-free path. A cache miss or stale cache entry elects one leader per
+    session; waiters retry the canonical resolver after that leader publishes its
+    result. Leaders for different sessions share a small global slot budget so
+    concurrent JSON parses cannot multiply a large sidecar's transient memory.
+    """
+    if metadata_only:
+        return _resolve_session_once(
+            sid,
+            metadata_only=True,
+            promote_cache=promote_cache,
+            cache_on_miss=cache_on_miss,
+        )
+
+    session_id = str(sid)
+    while True:
+        try:
+            return _resolve_session_once(
+                sid,
+                metadata_only=False,
+                promote_cache=promote_cache,
+                cache_on_miss=cache_on_miss,
+            )
+        except _FullSessionResolveRequired:
+            pass
+
+        active: set[str] = set(
+            getattr(_FULL_SESSION_RESOLVE_LOCAL, "active", ()) or ()
+        )
+        if active:
+            # Nested resolution already owns one global slot. Re-enter directly
+            # instead of waiting on this thread's own flight or consuming a
+            # second slot; the canonical resolver still owns all cache checks.
+            return _resolve_session_once(
+                sid,
+                metadata_only=False,
+                promote_cache=promote_cache,
+                cache_on_miss=cache_on_miss,
+                allow_full_load=True,
+            )
+
+        leader, event = _claim_full_session_resolve(session_id)
+        if not leader:
+            event.wait()
+            continue
+
+        active.add(session_id)
+        _FULL_SESSION_RESOLVE_LOCAL.active = active
+        try:
+            with _FULL_SESSION_RESOLVE_SLOTS:
+                return _resolve_session_once(
+                    sid,
+                    metadata_only=False,
+                    promote_cache=promote_cache,
+                    cache_on_miss=cache_on_miss,
+                    allow_full_load=True,
+                )
+        finally:
+            active.discard(session_id)
+            _FULL_SESSION_RESOLVE_LOCAL.active = active
+            _finish_full_session_resolve(session_id, event)
+
+
 def get_session(sid, metadata_only=False):
     """Load a session, optionally with metadata only (skipping messages)."""
     return _resolve_session(sid, metadata_only=metadata_only)

tests/test_full_session_resolve_gate.py

@@ -0,0 +1,238 @@
+"""Regression coverage for bounded full-session resolution."""
+
+from __future__ import annotations
+
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+
+def _install_lightweight_resolve_stubs(monkeypatch, models) -> None:
+    """Keep the tests focused on full sidecar materialization."""
+    monkeypatch.setattr(models, "_sync_sidecar_from_state_db_if_newer", lambda _session: False)
+    monkeypatch.setattr(models, "_repair_stale_pending", lambda _session: False)
+    monkeypatch.setattr(models, "_session_has_pending_journal_retry", lambda _session: False)
+
+
+def _empty_session(models, sid: str):
+    return models.Session(session_id=sid, messages=[])
+
+
+@pytest.fixture(autouse=True)
+def _clear_session_cache():
+    import api.models as models
+
+    with models.LOCK:
+        models.SESSIONS.clear()
+    yield
+    with models.LOCK:
+        models.SESSIONS.clear()
+
+
+def test_same_session_cold_load_is_single_flight(monkeypatch):
+    import api.models as models
+
+    _install_lightweight_resolve_stubs(monkeypatch, models)
+    workers = 4
+    start = threading.Barrier(workers)
+    state_lock = threading.Lock()
+    calls = 0
+    active = 0
+    peak = 0
+
+    def fake_load(cls, sid):
+        nonlocal calls, active, peak
+        with state_lock:
+            calls += 1
+            active += 1
+            peak = max(peak, active)
+        time.sleep(0.08)
+        with state_lock:
+            active -= 1
+        return _empty_session(models, sid)
+
+    monkeypatch.setattr(models.Session, "load", classmethod(fake_load))
+
+    def resolve():
+        start.wait(timeout=2)
+        return models.get_session("same-session")
+
+    with ThreadPoolExecutor(max_workers=workers) as executor:
+        results = [future.result(timeout=3) for future in [executor.submit(resolve) for _ in range(workers)]]
+
+    assert calls == 1
+    assert peak == 1
+    assert {result.session_id for result in results} == {"same-session"}
+
+
+def test_stale_cached_session_is_single_flight(monkeypatch):
+    import api.models as models
+
+    _install_lightweight_resolve_stubs(monkeypatch, models)
+    stale = _empty_session(models, "stale-session")
+    with models.LOCK:
+        models.SESSIONS[stale.session_id] = stale
+    monkeypatch.setattr(models, "_cached_session_lags_disk", lambda session: session is stale)
+    monkeypatch.setattr(models, "_inactive_cache_tail_needs_disk_check", lambda _session: False)
+
+    workers = 4
+    start = threading.Barrier(workers)
+    state_lock = threading.Lock()
+    calls = 0
+    active = 0
+    peak = 0
+
+    def fake_load(cls, sid):
+        nonlocal calls, active, peak
+        with state_lock:
+            calls += 1
+            active += 1
+            peak = max(peak, active)
+        time.sleep(0.08)
+        with state_lock:
+            active -= 1
+        return _empty_session(models, sid)
+
+    monkeypatch.setattr(models.Session, "load", classmethod(fake_load))
+
+    def resolve():
+        start.wait(timeout=2)
+        return models.get_session("stale-session")
+
+    with ThreadPoolExecutor(max_workers=workers) as executor:
+        results = [
+            future.result(timeout=3)
+            for future in [executor.submit(resolve) for _ in range(workers)]
+        ]
+
+    assert calls == 1
+    assert peak == 1
+    assert all(result is results[0] for result in results)
+    assert results[0] is not stale
+
+
+def test_distinct_cold_loads_have_a_global_concurrency_bound(monkeypatch):
+    import api.models as models
+
+    _install_lightweight_resolve_stubs(monkeypatch, models)
+    session_ids = ("session-a", "session-b", "session-c", "session-d")
+    start = threading.Barrier(len(session_ids))
+    state_lock = threading.Lock()
+    active = 0
+    peak = 0
+
+    def fake_load(cls, sid):
+        nonlocal active, peak
+        with state_lock:
+            active += 1
+            peak = max(peak, active)
+        time.sleep(0.08)
+        with state_lock:
+            active -= 1
+        return _empty_session(models, sid)
+
+    monkeypatch.setattr(models.Session, "load", classmethod(fake_load))
+
+    def resolve(sid):
+        start.wait(timeout=2)
+        return models.get_session(sid)
+
+    with ThreadPoolExecutor(max_workers=len(session_ids)) as executor:
+        results = [
+            future.result(timeout=3)
+            for future in [executor.submit(resolve, sid) for sid in session_ids]
+        ]
+
+    assert peak == 2
+    assert {result.session_id for result in results} == set(session_ids)
+
+
+def test_metadata_only_load_bypasses_full_resolve_slots(monkeypatch):
+    import api.models as models
+
+    _install_lightweight_resolve_stubs(monkeypatch, models)
+    heavy_entered = threading.Barrier(3)
+    release_heavy = threading.Event()
+
+    def fake_load(cls, sid):
+        heavy_entered.wait(timeout=2)
+        assert release_heavy.wait(timeout=2)
+        return _empty_session(models, sid)
+
+    def fake_load_metadata_only(cls, sid):
+        return _empty_session(models, sid)
+
+    monkeypatch.setattr(models.Session, "load", classmethod(fake_load))
+    monkeypatch.setattr(models.Session, "load_metadata_only", classmethod(fake_load_metadata_only))
+
+    with ThreadPoolExecutor(max_workers=3) as executor:
+        first = executor.submit(models.get_session, "heavy-a")
+        second = executor.submit(models.get_session, "heavy-b")
+        heavy_entered.wait(timeout=2)
+        metadata = executor.submit(models.get_session, "metadata", True)
+        try:
+            assert metadata.result(timeout=0.5).session_id == "metadata"
+        finally:
+            release_heavy.set()
+        assert first.result(timeout=2).session_id == "heavy-a"
+        assert second.result(timeout=2).session_id == "heavy-b"
+
+
+def test_fresh_cache_hit_bypasses_full_resolve_slots(monkeypatch):
+    import api.models as models
+
+    _install_lightweight_resolve_stubs(monkeypatch, models)
+    cached = _empty_session(models, "cached")
+    with models.LOCK:
+        models.SESSIONS[cached.session_id] = cached
+    monkeypatch.setattr(models, "_cached_session_lags_disk", lambda _session: False)
+    monkeypatch.setattr(models, "_inactive_cache_tail_needs_disk_check", lambda _session: False)
+
+    heavy_entered = threading.Barrier(3)
+    release_heavy = threading.Event()
+
+    def fake_load(cls, sid):
+        heavy_entered.wait(timeout=2)
+        assert release_heavy.wait(timeout=2)
+        return _empty_session(models, sid)
+
+    monkeypatch.setattr(models.Session, "load", classmethod(fake_load))
+
+    with ThreadPoolExecutor(max_workers=3) as executor:
+        first = executor.submit(models.get_session, "heavy-a")
+        second = executor.submit(models.get_session, "heavy-b")
+        heavy_entered.wait(timeout=2)
+        cache_hit = executor.submit(models.get_session, "cached")
+        try:
+            assert cache_hit.result(timeout=0.5) is cached
+        finally:
+            release_heavy.set()
+        first.result(timeout=2)
+        second.result(timeout=2)
+
+
+def test_failed_leader_releases_single_flight_state(monkeypatch):
+    import api.models as models
+
+    _install_lightweight_resolve_stubs(monkeypatch, models)
+    attempts = 0
+
+    def fake_load(cls, sid):
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise OSError("simulated sidecar failure")
+        return _empty_session(models, sid)
+
+    monkeypatch.setattr(models.Session, "load", classmethod(fake_load))
+
+    with pytest.raises(OSError, match="simulated sidecar failure"):
+        models.get_session("retryable")
+
+    recovered = models.get_session("retryable")
+
+    assert recovered.session_id == "retryable"
+    assert attempts == 2
+    assert models._FULL_SESSION_RESOLVE_INFLIGHT == {}