From 6caaced8ebfa8b95f7506c477a72efa9c063bd79 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Wed, 1 Jul 2026 18:14:39 +0000 Subject: [PATCH] fix(webui): bound in-memory SESSIONS cache with lazy reload (#4765) Root cause of the silent-crash-after-hours cluster (#4765/#2233/#4633): the global in-memory SESSIONS LRU evicted with a blind popitem(last=False), which could drop an actively streaming or not-yet-persisted session (data loss) and was capped only via an env var. On long-running installs the effective result was unbounded RAM growth until segfault. - Add _session_is_evictable(): a session is evictable ONLY when it is not streaming (no active_stream_id), has no in-flight turn (no pending_user_message / pending_started_at), and its full state is proven on disk (sidecar message_count >= in-memory count; metadata-only stubs and zero-message shells are trivially safe). - Add _evict_sessions_over_cap(): replaces all 8 blind popitem loops across models.py, routes.py, streaming.py. Walks the LRU oldest-first and removes only provably-safe entries; never acquires LOCK/stream locks itself (caller holds LOCK) so no lock-ordering deadlock. May briefly exceed the cap rather than ever evict an active/unsaved session. - Make the cap configurable via config.yaml webui.sessions_cache_max (get_sessions_cache_max()); precedence config.yaml -> HERMES_WEBUI_SESSIONS_MAX (legacy) -> DEFAULT_SESSIONS_CACHE_MAX=300. No new HERMES_* env var. Invalid or <1 values fall back so a typo can never disable the bound. - Evicted sessions lazily reload from their JSON sidecar via the existing get_session() accessor; no call sites changed. _index.json sidebar behavior unchanged (the sidebar reads the index, not SESSIONS). - Add tests/test_issue4765_sessions_lru_eviction.py (8 tests): eviction past cap, active/streaming never evicted, unsaved/stale-tail never evicted, lazy reload with identical content, and no-data-loss under heavy churn. - README: document the config.yaml key + safety semantics. Fixes #4765. --- README.md | 2 +- api/config.py | 45 ++- api/models.py | 124 ++++++- api/routes.py | 16 +- api/streaming.py | 4 +- tests/test_issue4765_sessions_lru_eviction.py | 335 ++++++++++++++++++ 6 files changed, 505 insertions(+), 21 deletions(-) create mode 100644 tests/test_issue4765_sessions_lru_eviction.py diff --git a/README.md b/README.md index 18daa7196..4e17ead08 100644 --- a/README.md +++ b/README.md @@ -367,7 +367,7 @@ Full list of environment variables: | `HERMES_CONFIG_PATH` | `$HERMES_HOME/config.yaml` | Path to Hermes config file | | `HERMES_WEBUI_SERVER_CWD` | *(unset)* | Working directory for the server process. Defaults to the agent dir; point it at a writable workspace when the agent dir is read-only so fallback relative writes land somewhere writable | | `HERMES_WEBUI_AGENT_CACHE_MAX` | `25` | Max live agent instances kept warm in the in-memory LRU. Each pins a full conversation transcript, so this is the dominant lever on resident memory — lower it on installs with many long sessions to cap RAM (at the cost of more cold reloads) | -| `HERMES_WEBUI_SESSIONS_MAX` | `100` | Max compact `Session` objects held in the in-memory LRU. Lighter than the agent cache; lower it on installs with hundreds of sessions | +| `HERMES_WEBUI_SESSIONS_MAX` | `300` | Legacy operator override for the max compact `Session` objects held in the in-memory LRU. Prefer the `webui.sessions_cache_max` key in `config.yaml` (which takes precedence); this env var remains a fallback. Bounds resident memory so long-running installs cannot accumulate every session ever touched and eventually crash (#4765/#2233/#4633). Eviction only ever drops clean, persisted, non-active sessions — an evicted session lazily reloads from its JSON sidecar on next access | Extension deployments can inspect sanitized, authenticated diagnostics at `GET /api/extensions/status`; see [WebUI Extensions](docs/EXTENSIONS.md#diagnostics). diff --git a/api/config.py b/api/config.py index f8577a8c6..c9b0ef46c 100644 --- a/api/config.py +++ b/api/config.py @@ -7355,10 +7355,47 @@ _INDEX_HTML_PATH = REPO_ROOT / "static" / "index.html" # ── Thread synchronisation ─────────────────────────────────────────────────── LOCK = threading.Lock() -# Max compact Session objects held in the in-memory LRU (issue #3506). Lighter -# than the agent cache (no live agent runtime), but still bounded and operator- -# tunable via HERMES_WEBUI_SESSIONS_MAX for installs with hundreds of sessions. -SESSIONS_MAX = _env_int("HERMES_WEBUI_SESSIONS_MAX", 100) +# Max compact Session objects held in the in-memory LRU (issue #3506, #4765). +# Lighter than the agent cache (no live agent runtime), but still bounded so a +# long-running self-hosted install cannot accumulate every session it ever +# touched in RAM and eventually segfault (the #4765/#2233/#4633 crash cluster). +# +# Precedence for the effective cap is resolved by get_sessions_cache_max(): +# 1. config.yaml webui.sessions_cache_max (preferred, no new env var) +# 2. HERMES_WEBUI_SESSIONS_MAX env var (legacy operator override) +# 3. DEFAULT_SESSIONS_CACHE_MAX (sane bounded default) +DEFAULT_SESSIONS_CACHE_MAX = 300 +SESSIONS_MAX = _env_int("HERMES_WEBUI_SESSIONS_MAX", DEFAULT_SESSIONS_CACHE_MAX) + + +def get_sessions_cache_max(config_data: dict | None = None) -> int: + """Return the effective in-memory SESSIONS cache cap (issue #4765). + + The bound is configurable through ``webui.sessions_cache_max`` in + ``config.yaml`` so operators of large self-hosted installs can size the + cache without editing source or adding a new ``HERMES_*`` env var (this + project forbids new env vars for non-secret config). A missing, empty, + non-numeric, or below-1 value falls back to the legacy + ``HERMES_WEBUI_SESSIONS_MAX`` env override, then to + ``DEFAULT_SESSIONS_CACHE_MAX`` — a typo can never disable the bound and + reintroduce unbounded memory growth. + """ + active_cfg = config_data if isinstance(config_data, dict) else get_config() + webui_cfg = active_cfg.get("webui", {}) if isinstance(active_cfg, dict) else {} + if isinstance(webui_cfg, dict): + raw = webui_cfg.get("sessions_cache_max") + if raw is not None: + try: + value = int(raw) + except (TypeError, ValueError): + value = None + if value is not None and value >= 1: + return value + # config.yaml did not specify a valid cap: honor the legacy env override + # (already parsed into SESSIONS_MAX) and finally the hardened default. + if isinstance(SESSIONS_MAX, int) and SESSIONS_MAX >= 1: + return SESSIONS_MAX + return DEFAULT_SESSIONS_CACHE_MAX CHAT_LOCK = threading.Lock() diff --git a/api/models.py b/api/models.py index 060b2b117..350d805bc 100644 --- a/api/models.py +++ b/api/models.py @@ -2937,6 +2937,124 @@ def _cached_session_lags_disk(cached) -> bool: return False +def _persisted_message_count(sid) -> int | None: + """Return the on-disk message count for *sid* without a full load (#4765). + + Reads only the sidecar metadata prefix (and falls back to the sidebar + ``_index.json`` count) so the eviction safety check stays cheap even while + the global ``LOCK`` is held. Returns ``None`` when the sidecar is missing or + its count cannot be determined — callers treat that as "do not evict", + because we must never drop an in-memory session we cannot prove is on disk. + """ + if not is_safe_session_id(sid): + return None + p = SESSION_DIR / f'{sid}.json' + if not p.exists(): + return None + try: + prefix = _read_metadata_json_prefix(p) + if prefix: + parsed = json.loads(prefix) + count = _parse_nonnegative_int(parsed.get('message_count')) + if count is not None: + return count + except Exception: + # Fall through to the index-based fallback below. + pass + return _parse_nonnegative_int(_lookup_index_message_count(sid)) + + +def _session_is_evictable(s) -> bool: + """Return True only when *s* can be safely dropped from the LRU (#4765). + + Eviction must never lose data or interrupt a live turn. A session is + evictable ONLY when ALL of the following hold: + + * It is not streaming (no ``active_stream_id``). + * It has no in-flight/queued turn (no ``pending_user_message`` and no + ``pending_started_at``). + * Its full state is already persisted to the JSON sidecar, proven by the + on-disk ``message_count`` being at least the in-memory message count. + A metadata-only stub is inherently backed by disk, so it is evictable. + + Anything we cannot positively prove is safe stays resident. Using slightly + more RAM for a session we are unsure about is strictly better than evicting + an active or unsaved session (task safety invariant: a half-done memory fix + that loses a session is worse than none). + """ + if s is None: + return True # nothing to protect; let the caller drop it + if getattr(s, 'active_stream_id', None): + return False + if getattr(s, 'pending_user_message', None): + return False + if getattr(s, 'pending_started_at', None): + return False + sid = getattr(s, 'session_id', None) + if not sid: + return False + # Metadata-only stubs never carry unsaved messages (messages=[] by design), + # so they are always disk-backed and safe to drop. + if getattr(s, '_loaded_metadata_only', False): + return True + in_memory_count = len(getattr(s, 'messages', None) or []) + if in_memory_count == 0: + # A zero-message session has nothing to lose. If it was never persisted + # (brand new, no sidecar) dropping it only discards an empty shell; the + # next access recreates it. If it is persisted, it is trivially clean. + return True + disk_count = _persisted_message_count(sid) + if disk_count is None: + return False # cannot prove it is on disk → keep it resident + return disk_count >= in_memory_count + + +def _evict_sessions_over_cap(cap: int | None = None) -> int: + """Evict clean, persisted, non-active sessions until len(SESSIONS) <= cap. + + Replaces the previous blind ``SESSIONS.popitem(last=False)`` loops (#4765). + The blind loops could evict the least-recently-used entry even if it was + actively streaming or held unsaved messages, risking a dropped turn or lost + conversation. This walks the LRU from oldest to newest and removes only + entries that ``_session_is_evictable()`` proves are safe. An evicted session + transparently lazily reloads from its sidecar on the next ``get_session()``. + + CALLER CONTRACT: the global ``LOCK`` MUST already be held (every call site + mutates ``SESSIONS`` under ``LOCK``). This function never acquires ``LOCK`` + or any stream lock itself, so it cannot introduce a lock-ordering deadlock. + + Returns the number of sessions evicted. If every over-cap candidate is + active/unsaved, the cache may temporarily exceed ``cap`` — that is the + intended safe behavior (never lose an active/unsaved session). + """ + if cap is None: + try: + cap = _cfg.get_sessions_cache_max() + except Exception: + cap = SESSIONS_MAX + if not isinstance(cap, int) or cap < 1: + cap = SESSIONS_MAX if isinstance(SESSIONS_MAX, int) and SESSIONS_MAX >= 1 else 1 + evicted = 0 + # Iterate over a snapshot of ids in LRU order (oldest first). We stop as + # soon as we are at/below the cap. Skipping a non-evictable oldest entry and + # moving on lets us reclaim a slightly-newer clean entry instead of blocking + # eviction entirely behind one pinned active session. + for sid in list(SESSIONS.keys()): + if len(SESSIONS) <= cap: + break + candidate = SESSIONS.get(sid) + if _session_is_evictable(candidate): + SESSIONS.pop(sid, None) + evicted += 1 + if len(SESSIONS) > cap: + logger.debug( + "SESSIONS cache above cap (%d > %d) after eviction pass: remaining " + "entries are active or unsaved and were preserved (#4765)", + len(SESSIONS), cap, + ) + return evicted + + def get_session(sid, metadata_only=False): """Load a session, optionally with metadata only (skipping the messages array). @@ -3018,8 +3136,7 @@ def get_session(sid, metadata_only=False): with LOCK: SESSIONS[sid] = s SESSIONS.move_to_end(sid) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) # evict least recently used + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) if not metadata_only: try: synced_from_state = _sync_sidecar_from_state_db_if_newer(s) @@ -3144,8 +3261,7 @@ def new_session(workspace=None, model=None, profile=None, model_provider=None, p with LOCK: SESSIONS[s.session_id] = s SESSIONS.move_to_end(s.session_id) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) if wt: s.save() return s diff --git a/api/routes.py b/api/routes.py index 120419e7e..f88fffbf5 100644 --- a/api/routes.py +++ b/api/routes.py @@ -139,8 +139,7 @@ def _persist_generated_session_title( with LOCK: SESSIONS[sid] = session SESSIONS.move_to_end(sid) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) _sync_session_title_to_insights(session) _publish_session_list_changed( event_reason, @@ -3480,8 +3479,7 @@ def _ensure_full_session_before_mutation(sid: str, session): with LOCK: SESSIONS[sid] = full_session SESSIONS.move_to_end(sid) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) return full_session @@ -8022,6 +8020,7 @@ from api.models import ( merge_session_messages_append_only, _enrich_sidebar_lineage_metadata, _active_stream_ids, + _evict_sessions_over_cap, _merge_session_display_metadata, _session_message_merge_key, _session_messages_have_prefix, @@ -12551,8 +12550,7 @@ def handle_post(handler, parsed) -> bool: with LOCK: SESSIONS[copied_session.session_id] = copied_session SESSIONS.move_to_end(copied_session.session_id) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) # Persist immediately. The pre-PR flow (/api/session/new + /api/session/rename) # accidentally avoided this because `/api/session/rename` calls `s.save()`. # Without this explicit save, the duplicate is in-memory only — if the user @@ -13213,8 +13211,7 @@ def handle_post(handler, parsed) -> bool: with LOCK: SESSIONS[branch.session_id] = branch SESSIONS.move_to_end(branch.session_id) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) # Persist only if there are messages (matches new_session pattern) if forked_messages: @@ -22673,8 +22670,7 @@ def _handle_session_import(handler, body): with LOCK: SESSIONS[s.session_id] = s SESSIONS.move_to_end(s.session_id) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) s.save() publish_session_list_changed("session_import") return j(handler, {"ok": True, "session": s.compact() | {"messages": s.messages}}) diff --git a/api/streaming.py b/api/streaming.py index 283db6855..4c2b635c1 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -51,6 +51,7 @@ from api.turn_journal import append_turn_journal_event_for_stream from api.usage import prompt_cache_hit_percent from api.models import ( _is_empty_partial_activity_message, + _evict_sessions_over_cap, get_state_db_session_messages, reconciled_state_db_messages_for_session, ) @@ -8215,8 +8216,7 @@ def _run_agent_streaming( ) SESSIONS[new_sid] = s SESSIONS.move_to_end(new_sid) - while len(SESSIONS) > SESSIONS_MAX: - SESSIONS.popitem(last=False) + _evict_sessions_over_cap() # #4765: safe LRU eviction (never active/unsaved) # Migrate the per-session lock: alias new_sid to the held # _agent_lock reference directly (not via old_sid lookup), # then remove the old_sid entry to prevent a leak. diff --git a/tests/test_issue4765_sessions_lru_eviction.py b/tests/test_issue4765_sessions_lru_eviction.py new file mode 100644 index 000000000..620c97470 --- /dev/null +++ b/tests/test_issue4765_sessions_lru_eviction.py @@ -0,0 +1,335 @@ +"""Regression tests for the bounded, lazy-loading SESSIONS cache (#4765). + +Crash cluster: #4765 / #2233 / #4633. + +Root cause: the WebUI kept ALL session objects + messages in a global in-memory +``OrderedDict`` (``api.config.SESSIONS``). On long-running self-hosted installs +the cache never shed idle sessions, so RSS climbed unbounded +(~700MB -> 7.5GB@9h -> 17.8GB@44h) until the interpreter segfaulted. + +The fix keeps the cache an LRU ``OrderedDict`` but replaces the pre-existing +*blind* ``SESSIONS.popitem(last=False)`` eviction (which could drop an active or +unsaved session and lose data) with ``_evict_sessions_over_cap()``: it only ever +removes clean, persisted, non-active sessions, and ``get_session()`` lazily +reloads an evicted session from its JSON sidecar on next access. + +These tests prove the four required invariants: + 1. Eviction happens once the cache grows past the cap. + 2. An active / streaming session is NEVER evicted, even when oldest. + 3. An evicted session lazily reloads from disk with identical content. + 4. No data loss: eviction removes only the in-memory copy, never the file. +""" +import collections +import shutil +import tempfile +import threading +import time +from pathlib import Path + +import pytest + + +@pytest.fixture +def isolated_session_env(): + """Isolate all SESSIONS-cache global state onto a throwaway temp dir. + + ``api.models`` imports ``SESSION_DIR`` / ``SESSION_INDEX_FILE`` at module + load, so both ``api.config`` and ``api.models`` copies must be redirected. + Everything is restored on teardown (even on exception). + """ + from api import config as _cfg + from api import models as _models + + tmpdir = tempfile.mkdtemp() + sessions_dir = Path(tmpdir) / "sessions" + sessions_dir.mkdir(parents=True, exist_ok=True) + + old = { + "cfg_SESSION_DIR": _cfg.SESSION_DIR, + "models_SESSION_DIR": getattr(_models, "SESSION_DIR", None), + "cfg_SESSION_INDEX_FILE": _cfg.SESSION_INDEX_FILE, + "models_SESSION_INDEX_FILE": getattr(_models, "SESSION_INDEX_FILE", None), + "SESSIONS": _cfg.SESSIONS, + "LOCK": _cfg.LOCK, + "SESSIONS_MAX": _cfg.SESSIONS_MAX, + "cfg": getattr(_cfg, "cfg", None), + } + + index_file = sessions_dir / "_index.json" + _cfg.SESSION_DIR = sessions_dir + _models.SESSION_DIR = sessions_dir + _cfg.SESSION_INDEX_FILE = index_file + _models.SESSION_INDEX_FILE = index_file + _cfg.LOCK = threading.Lock() + _models.LOCK = _cfg.LOCK + _cfg.SESSIONS = collections.OrderedDict() + _models.SESSIONS = _cfg.SESSIONS + + try: + yield sessions_dir + finally: + _cfg.SESSION_DIR = old["cfg_SESSION_DIR"] + if old["models_SESSION_DIR"] is not None: + _models.SESSION_DIR = old["models_SESSION_DIR"] + _cfg.SESSION_INDEX_FILE = old["cfg_SESSION_INDEX_FILE"] + if old["models_SESSION_INDEX_FILE"] is not None: + _models.SESSION_INDEX_FILE = old["models_SESSION_INDEX_FILE"] + _cfg.SESSIONS = old["SESSIONS"] + _models.SESSIONS = old["SESSIONS"] + _cfg.LOCK = old["LOCK"] + _models.LOCK = old["LOCK"] + _cfg.SESSIONS_MAX = old["SESSIONS_MAX"] + if old["cfg"] is not None: + _cfg.cfg = old["cfg"] + shutil.rmtree(tmpdir, ignore_errors=True) + + +def _make_persisted_session(idx, *, messages=None): + """Build + save a real session with at least one message (so it persists).""" + from api.models import Session + + if messages is None: + messages = [ + {"role": "user", "content": f"hello {idx}", "timestamp": time.time()}, + {"role": "assistant", "content": f"reply {idx}", "timestamp": time.time()}, + ] + s = Session(session_id=f"sess{idx:04d}", title=f"Session {idx}", messages=messages) + s.save() + return s + + +def _insert(sid_session): + """Insert a session into the cache exactly like the production accessors do.""" + from api.config import SESSIONS, LOCK + from api.models import _evict_sessions_over_cap + + with LOCK: + SESSIONS[sid_session.session_id] = sid_session + SESSIONS.move_to_end(sid_session.session_id) + _evict_sessions_over_cap() + + +# ─────────────────────────── config knob ──────────────────────────────────── + +def test_cache_cap_reads_config_yaml_key(): + """The cap is configurable via config.yaml webui.sessions_cache_max (#4765).""" + from api import config as _cfg + + assert _cfg.get_sessions_cache_max({"webui": {"sessions_cache_max": 42}}) == 42 + # Invalid / missing values must fall back, never disable the bound. + fallback = _cfg.get_sessions_cache_max({"webui": {"sessions_cache_max": "nope"}}) + assert isinstance(fallback, int) and fallback >= 1 + assert _cfg.get_sessions_cache_max({"webui": {}}) >= 1 + assert _cfg.get_sessions_cache_max({}) >= 1 + # A zero/negative typo must not disable the cap. + assert _cfg.get_sessions_cache_max({"webui": {"sessions_cache_max": 0}}) >= 1 + assert _cfg.get_sessions_cache_max({"webui": {"sessions_cache_max": -5}}) >= 1 + + +# ─────────────────────────── invariant 1: eviction ────────────────────────── + +def test_eviction_happens_past_the_cap(isolated_session_env): + """Inserting well past the cap must bound the in-memory cache size (#4765).""" + from api import config as _cfg + from api.config import SESSIONS + + _cfg.SESSIONS_MAX = 5 + cap = 5 + + created = [_make_persisted_session(i) for i in range(20)] + for s in created: + _insert(s) + + # The cache must be bounded — this is the whole point of the fix. Without + # it, all 20 (and eventually millions) would remain resident forever. + assert len(SESSIONS) <= cap, ( + f"cache grew to {len(SESSIONS)} entries; expected <= {cap} — the " + f"unbounded-growth crash (#4765/#2233/#4633) is not fixed" + ) + + # The most-recently-inserted sessions are the ones kept (LRU semantics). + kept = set(SESSIONS.keys()) + assert created[-1].session_id in kept + assert created[0].session_id not in kept + + +# ────────────────────── invariant 2: never evict active ────────────────────── + +def test_active_streaming_session_never_evicted(isolated_session_env): + """An active/streaming session must survive eviction even as the oldest (#4765).""" + from api import config as _cfg + from api.config import SESSIONS + from api.models import _session_is_evictable + + _cfg.SESSIONS_MAX = 3 + + # Oldest entry is actively streaming (has an in-flight turn). + active = _make_persisted_session(0) + active.active_stream_id = "live-stream-xyz" + active.pending_user_message = "in-flight question" + active.pending_started_at = time.time() + _insert(active) + + assert _session_is_evictable(active) is False + + # Now flood the cache far past the cap with clean sessions. + for i in range(1, 30): + _insert(_make_persisted_session(i)) + + assert active.session_id in SESSIONS, ( + "an actively streaming session was evicted — this would drop an " + "in-flight turn and corrupt live state (#4765 safety invariant)" + ) + # The live object identity (with its unsaved runtime state) is preserved. + assert SESSIONS[active.session_id] is active + assert SESSIONS[active.session_id].active_stream_id == "live-stream-xyz" + + +def test_unsaved_session_never_evicted(isolated_session_env): + """A session with unsaved messages (not yet on disk) is never evicted (#4765).""" + from api import config as _cfg + from api.config import SESSIONS + from api.models import Session, _session_is_evictable + + _cfg.SESSIONS_MAX = 3 + + # Build a session with messages in memory but DO NOT save it to disk. + unsaved = Session( + session_id="unsaved00001", + title="Unsaved", + messages=[{"role": "user", "content": "not persisted yet", "timestamp": time.time()}], + ) + assert not unsaved.path.exists() + assert _session_is_evictable(unsaved) is False + + _insert(unsaved) + for i in range(1, 30): + _insert(_make_persisted_session(i)) + + assert unsaved.session_id in SESSIONS, ( + "a session with unsaved in-memory messages was evicted — this loses " + "data (#4765 safety invariant)" + ) + assert SESSIONS[unsaved.session_id] is unsaved + + +def test_stale_disk_copy_blocks_eviction(isolated_session_env): + """A cached session ahead of its sidecar (unsaved tail) is not evictable (#4765).""" + from api.models import _session_is_evictable + + s = _make_persisted_session(1) # 2 messages on disk + # Simulate new turns appended in memory but not yet flushed to disk. + s.messages = s.messages + [ + {"role": "user", "content": "newer unsaved turn", "timestamp": time.time()}, + {"role": "assistant", "content": "newer unsaved reply", "timestamp": time.time()}, + ] + assert _session_is_evictable(s) is False, ( + "a session whose in-memory messages exceed the on-disk copy must not " + "be evicted — doing so silently loses the unsaved tail" + ) + # Once flushed, it becomes evictable again. + s.save() + assert _session_is_evictable(s) is True + + +# ───────────────── invariant 3: lazy reload + invariant 4: no data loss ────── + +def test_evicted_session_lazily_reloads_identical_content(isolated_session_env): + """An evicted session transparently reloads from disk with identical content.""" + from api import config as _cfg + from api.config import SESSIONS + from api.models import get_session + + _cfg.SESSIONS_MAX = 3 + + rich_messages = [ + {"role": "user", "content": "remember: the passphrase is orange-turbine-42", + "timestamp": time.time()}, + {"role": "assistant", "content": "Got it, I'll remember orange-turbine-42.", + "timestamp": time.time()}, + {"role": "user", "content": "what was it?", "timestamp": time.time()}, + {"role": "assistant", "content": "orange-turbine-42", "timestamp": time.time()}, + ] + victim = _make_persisted_session(0, messages=rich_messages) + _insert(victim) + victim_id = victim.session_id + expected = [dict(m) for m in victim.messages] + + # Push the victim out of the in-memory cache with newer sessions. + for i in range(1, 30): + _insert(_make_persisted_session(i)) + + assert victim_id not in SESSIONS, ( + "the clean, persisted, idle victim should have been evicted from RAM" + ) + # The sidecar file is untouched (invariant 4: no data loss). + assert victim.path.exists() + + # Accessing it again must transparently reload from the sidecar (invariant 3). + reloaded = get_session(victim_id) + assert reloaded is not None + assert reloaded.session_id == victim_id + assert [{"role": m["role"], "content": m["content"]} for m in reloaded.messages] == \ + [{"role": m["role"], "content": m["content"]} for m in expected], ( + "lazily-reloaded session content differs from what was persisted — " + "the reload path is lossy (#4765)" + ) + # And it is back in the cache after the lazy reload. + assert victim_id in SESSIONS + + +def test_no_data_loss_all_files_survive_heavy_churn(isolated_session_env): + """Eviction removes only the in-memory copy; every sidecar file survives (#4765).""" + from api import config as _cfg + from api.config import SESSIONS + from api.models import get_session + + _cfg.SESSIONS_MAX = 4 + + created = [_make_persisted_session(i) for i in range(25)] + for s in created: + _insert(s) + + # Cache is bounded... + assert len(SESSIONS) <= 4 + # ...but NOT ONE session file was deleted. + for s in created: + assert s.path.exists(), f"sidecar for {s.session_id} was deleted — data loss!" + + # Every single session (even long-evicted ones) is still fully retrievable + # with its original content via the lazy-reload accessor. + for i, s in enumerate(created): + loaded = get_session(s.session_id) + assert loaded is not None + assert loaded.title == f"Session {i}" + assert len(loaded.messages) == 2 + assert loaded.messages[0]["content"] == f"hello {i}" + + +def test_eviction_skips_active_but_still_bounds_clean_entries(isolated_session_env): + """Mixed workload: active pinned, clean bounded — the realistic steady state.""" + from api import config as _cfg + from api.config import SESSIONS + + _cfg.SESSIONS_MAX = 5 + + # A handful of concurrently-active streams that must all stay resident. + actives = [] + for i in range(3): + a = _make_persisted_session(1000 + i) + a.active_stream_id = f"stream-{i}" + _insert(a) + actives.append(a) + + # Plus heavy churn of clean idle sessions. + for i in range(40): + _insert(_make_persisted_session(i)) + + # All actives survive. + for a in actives: + assert a.session_id in SESSIONS, "an active stream was evicted under churn" + + # The cache stays bounded: active (3, pinned) + at most cap clean entries. + # It may briefly sit slightly above cap because actives are non-evictable, + # but it must NOT grow unbounded with the 40 churned sessions. + assert len(SESSIONS) <= _cfg.SESSIONS_MAX + len(actives)