Release v0.51.256 — Release HX (stage-r4) (#3596)

## Release v0.51.256 — Release HX (stage-r4)

Performance — bound WebUI memory growth & idle CPU on large installs.

### Fixed
| Issue | Author | Fix |
|-------|--------|-----|
| #3506 | @nesquena-hermes (reported w/ profiling by @djenttleman) | On a large install (~615 sessions / 40k messages / 454 MB state.db) the WebUI process climbed ~100 MB → ~1.5 GB RSS over days and held high idle CPU. Three root causes fixed: (1) `session_lifecycle._sessions` grew unbounded → new `discard_session()` drops the entry at agent-eviction boundaries, only when no in-flight commit / no uncommitted memory work (retry invariant preserved); (2) cache caps now operator-tunable (`HERMES_WEBUI_AGENT_CACHE_MAX` default 50→25, `HERMES_WEBUI_SESSIONS_MAX`); (3) GatewayWatcher computes a cheap fingerprint before the expensive per-session `MAX(messages.timestamp)` projection and only re-projects on change. |

### Rebase + review notes
- Rebased onto current master; the code diff was verified **byte-identical to the nesquena-APPROVED head** at rebase time (only CHANGELOG re-resolved).
- The Codex regression gate then surfaced **two correctness gaps** the approval didn't catch, both fixed here with regression tests:
  1. **Watcher fingerprint missed same-count transcript rewrites.** `/retry`,`/undo`,`/compress` (`SessionDB.replace_messages`) rewrite messages with new timestamps but can leave `message_count` unchanged → stale sidebar `last_activity`. Fixed with a **per-session** grouped message aggregate (`id, count, user_count, MAX(timestamp)`) over the same non-excluded sessions (a global MAX would miss a rewrite of an older, non-newest session); cron/webui stay excluded so idle churn still doesn't re-project.
  2. **LRU agent-cache eviction could close a live worker's agent** (`popitem(last=False)`, liveness-blind — pre-existing, but the lower 50→25 cap made it more likely). Eviction now snapshots `ACTIVE_RUNS` session_ids (before the cache lock — no nested lock) and skips live sessions, deferring (temporarily exceeding cap) rather than closing a live agent.

### Gate
- Full pytest suite: **7612 passed, 0 failed** (one boot-cascade flake re-run; clean on re-run)
- ruff: CLEAN · Codex (regression): 4 rounds → both gaps + a stale test fixed → **SAFE TO SHIP**

Co-authored-by: nesquena <nesquena@users.noreply.github.com>

CHANGELOG.md

@@ -3,6 +3,11 @@
 
 ## [Unreleased]
 
+## [v0.51.256] — 2026-06-04 — Release HX (stage-r4 — bound WebUI memory growth and idle CPU on large installs)
+
+### Fixed
+- **Bounded WebUI memory growth and idle CPU on large installs** (#3506). On a profile with hundreds of sessions and tens of thousands of messages, the WebUI Python process could climb from ~100 MB to ~1.5 GB RSS over several days and hold high CPU while idle. Three root causes were fixed: (1) the `session_lifecycle._sessions` dict grew without bound — entries were created but never removed — so every session id the WebUI ever touched leaked a permanent entry; a new safe `discard_session()` now drops the entry at the agent-eviction boundaries, but only when there is no in-flight commit and no uncommitted memory work (the retry invariant is preserved); (2) the in-memory agent/session cache sizes are now operator-tunable via `HERMES_WEBUI_AGENT_CACHE_MAX` and `HERMES_WEBUI_SESSIONS_MAX`, and the agent-cache default was lowered from 50 to 25 (each cached agent pins a full conversation transcript, so this is the dominant lever on resident memory); (3) the gateway session watcher re-ran an expensive per-session `MAX(messages.timestamp)` aggregation every 5 seconds even when nothing changed — it now first computes a cheap `sessions`-table-only fingerprint (~15× cheaper, measured on a 1.5 GB state.db) and only runs the full projection when that fingerprint actually changes. (#3506, @nesquena-hermes; reported with detailed profiling by @djenttleman)
+
 ## [v0.51.255] — 2026-06-04 — Release HW (stage-r3 — pid-scoped turn-journal shards for multi-process safety)
 
 ### Fixed

README.md

@@ -337,6 +337,8 @@ Full list of environment variables:
 | `HERMES_WEBUI_EXTENSION_STYLESHEET_URLS` | *(unset)* | Optional comma-separated same-origin stylesheet URLs to inject; see [WebUI Extensions](docs/EXTENSIONS.md) |
 | `HERMES_HOME` | Windows: `%LOCALAPPDATA%\hermes`; POSIX: `~/.hermes` | Base directory for Hermes state (affects all paths) |
 | `HERMES_CONFIG_PATH` | `$HERMES_HOME/config.yaml` | Path to Hermes config file |
+| `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 |
 
 ---
 

api/config.py

@@ -39,6 +39,24 @@ REPO_ROOT = Path(__file__).parent.parent.resolve()
 HOST = os.getenv("HERMES_WEBUI_HOST", "127.0.0.1")
 PORT = int(os.getenv("HERMES_WEBUI_PORT", "8787"))
 
+
+def _env_int(name: str, default: int, *, minimum: int = 1) -> int:
+    """Read a positive int from the environment, falling back on bad input.
+
+    Used for operator-tunable memory caps (issue #3506) so large installs can
+    shrink the agent/session caches without editing source. A missing, empty,
+    non-numeric, or below-``minimum`` value falls back to ``default`` so a typo
+    can never disable a cache bound entirely.
+    """
+    raw = os.getenv(name)
+    if raw is None or not str(raw).strip():
+        return default
+    try:
+        value = int(str(raw).strip())
+    except (TypeError, ValueError):
+        return default
+    return value if value >= minimum else default
+
 # ── TLS/HTTPS config (optional, env-overridable) ────────────────────────────
 TLS_CERT = os.getenv("HERMES_WEBUI_TLS_CERT", "").strip() or None
 TLS_KEY = os.getenv("HERMES_WEBUI_TLS_KEY", "").strip() or None
@@ -4810,7 +4828,10 @@ _INDEX_HTML_PATH = REPO_ROOT / "static" / "index.html"
 
 # ── Thread synchronisation ───────────────────────────────────────────────────
 LOCK = threading.Lock()
-SESSIONS_MAX = 100
+# 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)
 CHAT_LOCK = threading.Lock()
 
 
@@ -4931,7 +4952,12 @@ def unregister_active_run(stream_id: str) -> None:
 # SESSION_AGENT_CACHE_LOCK for thread safety in multi-threaded ASGI servers.
 import collections
 SESSION_AGENT_CACHE: collections.OrderedDict = collections.OrderedDict()  # LRU cache
-SESSION_AGENT_CACHE_MAX = 50  # Maximum cached agents (each holds full conversation history)
+# Each cached agent pins a full conversation transcript in RAM, so this cap is
+# the dominant lever on WebUI resident memory (issue #3506). The default is kept
+# deliberately modest -- large/long sessions can each weigh tens of MB, so 50
+# live agents could pin >1 GB on a heavily multiplexed install. Operators can
+# tune it via HERMES_WEBUI_AGENT_CACHE_MAX without editing source.
+SESSION_AGENT_CACHE_MAX = _env_int("HERMES_WEBUI_AGENT_CACHE_MAX", 25)
 SESSION_AGENT_CACHE_LOCK = threading.Lock()
 
 
@@ -4953,11 +4979,14 @@ def _evict_session_agent(session_id: str) -> None:
         return
     should_close = True
     try:
-        from api.session_lifecycle import commit_session_memory, has_uncommitted_work, unregister_agent
+        from api.session_lifecycle import commit_session_memory, discard_session, has_uncommitted_work, unregister_agent
         if has_uncommitted_work(session_id):
             commit_session_memory(session_id, agent=agent, wait=True)
         if not has_uncommitted_work(session_id):
             unregister_agent(session_id)
+            # Bound the lifecycle dict: drop the entry now that the session has
+            # no uncommitted work and the agent handle is gone (issue #3506).
+            discard_session(session_id)
         else:
             should_close = False
     except Exception:

api/gateway_watcher.py

@@ -13,8 +13,10 @@ import json
 import logging
 import os
 import queue
+import sqlite3
 import threading
 import time
+from contextlib import closing
 from pathlib import Path
 
 from api.config import HOME
@@ -34,6 +36,113 @@ def _snapshot_hash(sessions: list) -> str:
     return hashlib.md5(key.encode(), usedforsecurity=False).hexdigest()
 
 
+# Sources excluded from the WebUI sidebar projection. Must match the default
+# ``exclude_sources`` used by ``read_importable_agent_session_rows`` so the
+# cheap change-detection scan below sees exactly the same row set as the
+# expensive projection (otherwise cron message churn would defeat the gate).
+_WATCHER_EXCLUDED_SOURCES = ("cron", "webui")
+
+
+def _cheap_change_fingerprint(db_path: Path) -> str | None:
+    """Compute a cheap change-detection fingerprint without the messages JOIN.
+
+    The expensive projection (``read_importable_agent_session_rows``) runs a CTE
+    plus a per-session ``MAX(messages.timestamp)`` aggregation over an oversampled
+    candidate set every poll. On a large ``state.db`` (hundreds of sessions, tens
+    of thousands of messages) that is ~10x the cost of a single ``sessions``-table
+    scan, and the watcher runs it forever on a 5s timer even when nothing changed
+    (issue #3506).
+
+    This computes a fingerprint from a ``sessions``-table-only scan (no messages
+    JOIN), scoped to the same non-cron/webui rows as the projection. To guarantee
+    it never skips a change the projection would reflect, it hashes **every
+    sessions-table column the projection reads or uses for visibility/collapse**
+    -- not just the columns surfaced to the sidebar. That matters because the
+    projection collapses compression lineage and hides/shows rows based on
+    ``parent_session_id`` / ``ended_at`` / ``end_reason`` / ``source``, so a change
+    to one of those alters *which rows* appear even when no displayed field on a
+    given row moved.
+
+    The one projection input that does not live in the ``sessions`` table is the
+    per-session message aggregate (``COUNT`` / ``MAX(messages.timestamp)`` ->
+    ``last_activity``). That is fully proxied by ``sessions.message_count``: the
+    agent's state layer bumps ``message_count`` on every appended message and
+    rewrites it to the absolute count on truncate/rewind/compaction, so a message
+    insert or delete (the only events that can move ``MAX(timestamp)``) always
+    changes ``message_count``. The fingerprint is therefore a strict superset of
+    the projection's change surface (it also fires on out-of-order inserts that
+    would not raise ``MAX(timestamp)``).
+
+    Returns the fingerprint string, or ``None`` on any error / a pre-source
+    schema so the caller falls back to running the expensive projection rather
+    than risk skipping a change.
+    """
+    # Columns the projection reads from the ``sessions`` table. ``id``/``source``
+    # are always present (``source`` is required for the projection to run at
+    # all); the rest are optional on older agent schemas and filtered below.
+    _PROJECTION_SESSION_COLS = (
+        'id', 'source', 'session_source', 'title', 'model', 'message_count',
+        'started_at', 'ended_at', 'end_reason', 'parent_session_id', 'archived',
+        'user_id', 'chat_id', 'chat_type', 'thread_id', 'session_key',
+        'origin_chat_id', 'origin_user_id', 'platform',
+    )
+    try:
+        with closing(sqlite3.connect(str(db_path))) as conn:
+            cur = conn.cursor()
+            cur.execute("PRAGMA table_info(sessions)")
+            cols = {row[1] for row in cur.fetchall()}
+            if 'source' not in cols:
+                return None
+            selectable = [c for c in _PROJECTION_SESSION_COLS if c in cols]
+            placeholders = ", ".join("?" for _ in _WATCHER_EXCLUDED_SOURCES)
+            cur.execute(
+                f"SELECT {', '.join(selectable)} FROM sessions "
+                f"WHERE source IS NOT NULL AND source NOT IN ({placeholders}) "
+                f"ORDER BY id",
+                list(_WATCHER_EXCLUDED_SOURCES),
+            )
+            h = hashlib.md5(usedforsecurity=False)
+            for row in cur.fetchall():
+                h.update(repr(row).encode('utf-8', 'replace'))
+                h.update(b'\x1e')
+            # A same-count transcript rewrite (SessionDB.replace_messages used by
+            # /retry, /undo, /compress) deletes + reinserts messages with new
+            # timestamps but can leave sessions.message_count unchanged — so the
+            # sessions-only scan above would miss it and the watcher would skip a
+            # projection whose last_activity (MAX(messages.timestamp)) actually
+            # moved. Fold in a PER-SESSION message aggregate, scoped to the same
+            # non-excluded sessions as the projection. It must be per-session
+            # (grouped), NOT a single global MAX: rewriting an OLDER, non-newest
+            # session moves that session's last_activity but not the global max,
+            # so a global aggregate would still miss it (#3536 review round 2).
+            # cron/webui churn is excluded by the JOIN filter so it still does
+            # NOT trigger a re-projection. This is one GROUP BY over the already-
+            # filtered set — far cheaper than the projection's oversampled
+            # correlated CTE — so it preserves the cheap-fingerprint property.
+            if 'messages' in {r[0] for r in conn.execute(
+                    "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}:
+                try:
+                    msg_rows = conn.execute(
+                        "SELECT s.id, COUNT(m.id), "
+                        "COUNT(CASE WHEN LOWER(m.role) = 'user' THEN 1 END), "
+                        "COALESCE(MAX(m.timestamp), 0) "
+                        "FROM sessions s LEFT JOIN messages m ON m.session_id = s.id "
+                        f"WHERE s.source IS NOT NULL AND s.source NOT IN ({placeholders}) "
+                        "GROUP BY s.id ORDER BY s.id",
+                        list(_WATCHER_EXCLUDED_SOURCES),
+                    ).fetchall()
+                    for mrow in msg_rows:
+                        h.update(repr(mrow).encode('utf-8', 'replace'))
+                        h.update(b'\x1e')
+                except sqlite3.Error:
+                    # messages table shape unknown → don't trust the fingerprint;
+                    # signal the caller to run the full projection.
+                    return None
+            return h.hexdigest()
+    except Exception:
+        return None
+
+
 # ── DB resolution (shared pattern with state_sync.py) ──────────────────────
 
 def _get_state_db_path() -> Path:
@@ -98,6 +207,10 @@ class GatewayWatcher:
         self._thread: threading.Thread | None = None
         self._last_hash: str = ''
         self._last_sessions: list = []
+        # Cheap sessions-only fingerprint from the previous poll. When it is
+        # unchanged we skip the expensive messages-JOIN projection entirely
+        # (issue #3506). Empty string forces the first poll to run the full read.
+        self._last_cheap_fp: str = ''
 
     def start(self):
         """Start the watcher daemon thread."""
@@ -183,13 +296,30 @@ class GatewayWatcher:
         """Main polling loop. Runs in a daemon thread."""
         while not self._stop_event.is_set():
             try:
-                sessions = _get_agent_sessions_from_db()
-                current_hash = _snapshot_hash(sessions)
+                # Phase 1: cheap sessions-only fingerprint. The expensive
+                # messages-JOIN projection (_get_agent_sessions_from_db) only
+                # runs when this fingerprint actually changes, so an idle server
+                # with a large state.db stops re-aggregating tens of thousands
+                # of message rows every 5 seconds (issue #3506). A None
+                # fingerprint (error / unreadable db) forces the full read so we
+                # never silently skip a real change.
+                db_path = _get_state_db_path()
+                cheap_fp = _cheap_change_fingerprint(db_path) if db_path.exists() else ''
+                if cheap_fp is not None and cheap_fp == self._last_cheap_fp:
+                    # Nothing changed in the sidebar-visible session set; skip
+                    # the expensive projection and the notify entirely.
+                    pass
+                else:
+                    # Phase 2: only now pay for the full projection.
+                    sessions = _get_agent_sessions_from_db()
+                    current_hash = _snapshot_hash(sessions)
+                    if cheap_fp is not None:
+                        self._last_cheap_fp = cheap_fp
 
-                if current_hash != self._last_hash:
-                    self._last_hash = current_hash
-                    self._last_sessions = sessions
-                    self._notify_subscribers(sessions)
+                    if current_hash != self._last_hash:
+                        self._last_hash = current_hash
+                        self._last_sessions = sessions
+                        self._notify_subscribers(sessions)
             except Exception:
                 logger.debug("Error in gateway watcher poll loop", exc_info=True)
 

api/session_lifecycle.py

@@ -92,6 +92,39 @@ def unregister_agent(session_id: str) -> None:
         _condition.notify_all()
 
 
+def discard_session(session_id: str) -> bool:
+    """Permanently drop a session's lifecycle entry to bound memory growth.
+
+    The ``_sessions`` dict is process-global and historically only ever grew:
+    ``register_agent`` / ``mark_turn_completed`` insert keys but no runtime path
+    ever removed them, so every unique ``session_id`` the WebUI touched leaked a
+    permanent entry (issue #3506). Over days of use on a large install this is a
+    monotonic, unbounded climb.
+
+    This removes the entry, but only when it is provably safe to do so: no commit
+    is in flight and there is no uncommitted memory work that still needs the
+    retained agent handle. If the entry is busy or dirty it is left untouched so
+    failed batch-extraction memory work stays retryable -- exactly the invariant
+    ``unregister_agent`` and ``_evict_session_agent`` already preserve.
+
+    Returns True when the entry was removed (or was already absent), False when
+    it was retained because work is still pending.
+    """
+    if not session_id:
+        return False
+    with _condition:
+        entry = _sessions.get(session_id)
+        if entry is None:
+            return True
+        if entry["in_flight"]:
+            return False
+        if entry["generation"] > entry["committed_generation"]:
+            return False
+        del _sessions[session_id]
+        _condition.notify_all()
+        return True
+
+
 def mark_turn_completed(session_id: str, *, agent=None) -> int:
     if not session_id:
         return 0

api/streaming.py

@@ -3857,6 +3857,12 @@ def _lifecycle_unregister_agent(session_id: str) -> None:
     unregister_agent(session_id)
 
 
+def _lifecycle_discard_session(session_id: str) -> bool:
+    from api.session_lifecycle import discard_session
+
+    return discard_session(session_id)
+
+
 def _close_evicted_agent_at_session_boundary(session_id: str, agent) -> bool:
     """Commit and tear down an evicted cached agent at a WebUI session boundary.
 
@@ -3876,6 +3882,10 @@ def _close_evicted_agent_at_session_boundary(session_id: str, agent) -> bool:
         _lifecycle_commit_session_memory(session_id, agent=agent, wait=True)
         if not _lifecycle_has_uncommitted_work(session_id):
             _lifecycle_unregister_agent(session_id)
+            # Drop the lifecycle dict entry now that the LRU-evicted agent is
+            # gone and no uncommitted work remains, so the dict tracks only live
+            # sessions instead of growing unbounded (issue #3506).
+            _lifecycle_discard_session(session_id)
         else:
             should_close_evicted_agent = False
     except Exception:
@@ -5304,13 +5314,44 @@ def _run_agent_streaming(
                     except Exception:
                         logger.debug("Lifecycle register_agent failed for new session %s", session_id, exc_info=True)
                     _evicted_items = []
+                    # Snapshot the set of session_ids with a LIVE agent worker
+                    # BEFORE taking SESSION_AGENT_CACHE_LOCK, so LRU eviction never
+                    # closes an agent mid-run AND we never nest ACTIVE_RUNS_LOCK
+                    # inside SESSION_AGENT_CACHE_LOCK (avoids any lock-ordering
+                    # deadlock). A cancel/reconnect can drop STREAMS while the
+                    # worker is still unwinding or blocked in a provider call, so
+                    # ACTIVE_RUNS (worker lifecycle) is the authoritative liveness
+                    # signal, not STREAMS. (#3536 review round 2)
+                    _active_sids = set()
+                    try:
+                        from api.config import ACTIVE_RUNS, ACTIVE_RUNS_LOCK
+                        with ACTIVE_RUNS_LOCK:
+                            for _entry in (ACTIVE_RUNS or {}).values():
+                                _sid = (_entry or {}).get("session_id")
+                                if _sid:
+                                    _active_sids.add(_sid)
+                    except Exception:
+                        _active_sids = set()
                     with SESSION_AGENT_CACHE_LOCK:
                         SESSION_AGENT_CACHE[session_id] = (agent, _agent_sig)
                         SESSION_AGENT_CACHE.move_to_end(session_id)  # LRU: mark as recently used
                         from api.config import SESSION_AGENT_CACHE_MAX
+                        # Evict the oldest INACTIVE entries first. Walk LRU order
+                        # (front = oldest); skip any session with a live run. If
+                        # every over-cap entry is active, leave the cache
+                        # temporarily above cap rather than close a live worker's
+                        # agent — a later insertion/finalization trims it once the
+                        # run ends.
                         while len(SESSION_AGENT_CACHE) > SESSION_AGENT_CACHE_MAX:
-                            evicted_sid, evicted_entry = SESSION_AGENT_CACHE.popitem(last=False)
-                            _evicted_items.append((evicted_sid, evicted_entry))
+                            _evictable_sid = None
+                            for _sid in list(SESSION_AGENT_CACHE.keys()):
+                                if _sid not in _active_sids:
+                                    _evictable_sid = _sid
+                                    break
+                            if _evictable_sid is None:
+                                break  # all over-cap entries are active; defer
+                            evicted_entry = SESSION_AGENT_CACHE.pop(_evictable_sid)
+                            _evicted_items.append((_evictable_sid, evicted_entry))
                     # Commit and close evicted agents outside the cache lock so
                     # concurrent cache users are not blocked by provider I/O.
                     for _evicted_sid, _evicted_entry in _evicted_items:

tests/test_issue3506_memory_and_watcher.py

@@ -0,0 +1,429 @@
+"""Regression tests for issue #3506 — WebUI memory growth and idle CPU.
+
+A user on a large install (615 sessions / 40k messages in state.db) reported the
+WebUI Python process climbing from ~100 MB to ~1.5 GB RSS over 3 days and holding
+180%+ CPU at idle. Three independent contributors were confirmed in the code:
+
+  1. ``api.session_lifecycle._sessions`` grew without bound — keys were inserted
+     on ``register_agent`` / ``mark_turn_completed`` but never deleted, so every
+     unique session_id the WebUI ever touched leaked a permanent entry.
+  2. ``SESSION_AGENT_CACHE_MAX`` / ``SESSIONS_MAX`` were hard-coded with no way
+     for an operator to tune the dominant RSS lever without editing source.
+  3. ``GatewayWatcher`` re-ran an expensive per-session ``MAX(messages.timestamp)``
+     aggregation over an oversampled candidate set every 5s, forever, even when
+     nothing in the sidebar-visible session set had changed.
+
+These tests pin the fixes for all three:
+  * ``session_lifecycle.discard_session`` bounds the dict, safely.
+  * ``config._env_int`` makes the caps env-overridable with safe fallback.
+  * ``gateway_watcher._cheap_change_fingerprint`` is a sound, cheaper change
+    signal that the poll loop uses to skip the expensive projection.
+"""
+from __future__ import annotations
+
+import importlib
+import sqlite3
+import time
+from pathlib import Path
+
+
+# ─────────────────────────── Fix 1: lifecycle leak ───────────────────────────
+
+def _fresh_lifecycle():
+    lifecycle = importlib.import_module("api.session_lifecycle")
+    lifecycle = importlib.reload(lifecycle)
+    reset = getattr(lifecycle, "_reset_for_tests", None)
+    if callable(reset):
+        reset()
+    return lifecycle
+
+
+class _Agent:
+    def commit_memory_session(self):  # pragma: no cover - not exercised here
+        pass
+
+
+def test_discard_session_removes_clean_entry():
+    """A registered-then-completed-then-committed session must be evictable."""
+    lc = _fresh_lifecycle()
+    agent = _Agent()
+    sid = "clean-session"
+
+    lc.register_agent(sid, agent)
+    gen = lc.mark_turn_completed(sid, agent=agent)
+    # Simulate a successful commit catching up to the latest generation.
+    with lc._condition:
+        lc._sessions[sid]["committed_generation"] = gen
+
+    assert sid in lc._sessions
+    assert lc.has_uncommitted_work(sid) is False
+    assert lc.discard_session(sid) is True
+    assert sid not in lc._sessions, "clean entry must be removed to bound growth"
+
+
+def test_discard_session_preserves_uncommitted_work():
+    """A session with pending memory work must NOT be discarded (stays retryable)."""
+    lc = _fresh_lifecycle()
+    agent = _Agent()
+    sid = "dirty-session"
+
+    lc.register_agent(sid, agent)
+    lc.mark_turn_completed(sid, agent=agent)  # generation > committed_generation
+
+    assert lc.has_uncommitted_work(sid) is True
+    assert lc.discard_session(sid) is False
+    assert sid in lc._sessions, "dirty entry must be preserved so commit can retry"
+
+
+def test_discard_session_preserves_in_flight_commit():
+    """An in-flight commit must block discard to avoid racing the committer."""
+    lc = _fresh_lifecycle()
+    agent = _Agent()
+    sid = "in-flight-session"
+
+    lc.register_agent(sid, agent)
+    gen = lc.mark_turn_completed(sid, agent=agent)
+    with lc._condition:
+        lc._sessions[sid]["committed_generation"] = gen  # clean...
+        lc._sessions[sid]["in_flight"] = True             # ...but a commit is running
+
+    assert lc.discard_session(sid) is False
+    assert sid in lc._sessions
+
+
+def test_discard_session_absent_key_is_noop_success():
+    lc = _fresh_lifecycle()
+    assert lc.discard_session("never-seen") is True
+    assert lc.discard_session("") is False
+
+
+def test_lifecycle_dict_is_bounded_under_churn():
+    """Register/complete/commit/discard across many sessions must not accumulate."""
+    lc = _fresh_lifecycle()
+    for i in range(500):
+        sid = f"churn-{i}"
+        agent = _Agent()
+        lc.register_agent(sid, agent)
+        gen = lc.mark_turn_completed(sid, agent=agent)
+        with lc._condition:
+            lc._sessions[sid]["committed_generation"] = gen
+        lc.unregister_agent(sid)
+        assert lc.discard_session(sid) is True
+    assert len(lc._sessions) == 0, "dict must not grow unbounded across session churn"
+
+
+# ─────────────────────────── Fix 2: tunable caps ─────────────────────────────
+
+def test_env_int_reads_valid_override(monkeypatch):
+    cfg = importlib.import_module("api.config")
+    monkeypatch.setenv("HERMES_TEST_CAP", "12")
+    assert cfg._env_int("HERMES_TEST_CAP", 99) == 12
+
+
+def test_env_int_falls_back_on_bad_input(monkeypatch):
+    cfg = importlib.import_module("api.config")
+    monkeypatch.setenv("HERMES_TEST_CAP", "not-a-number")
+    assert cfg._env_int("HERMES_TEST_CAP", 99) == 99
+    monkeypatch.setenv("HERMES_TEST_CAP", "")
+    assert cfg._env_int("HERMES_TEST_CAP", 99) == 99
+    monkeypatch.delenv("HERMES_TEST_CAP", raising=False)
+    assert cfg._env_int("HERMES_TEST_CAP", 99) == 99
+
+
+def test_env_int_rejects_below_minimum(monkeypatch):
+    cfg = importlib.import_module("api.config")
+    monkeypatch.setenv("HERMES_TEST_CAP", "0")
+    # A 0 or negative cap would disable the bound entirely — must fall back.
+    assert cfg._env_int("HERMES_TEST_CAP", 99) == 99
+    monkeypatch.setenv("HERMES_TEST_CAP", "-5")
+    assert cfg._env_int("HERMES_TEST_CAP", 99) == 99
+
+
+def test_agent_cache_max_default_is_bounded():
+    cfg = importlib.import_module("api.config")
+    # Default must remain a sane, modest bound (each entry pins a full transcript).
+    assert isinstance(cfg.SESSION_AGENT_CACHE_MAX, int)
+    assert 1 <= cfg.SESSION_AGENT_CACHE_MAX <= 50
+    assert isinstance(cfg.SESSIONS_MAX, int)
+    assert cfg.SESSIONS_MAX >= 1
+
+
+# ─────────────────────── Fix 3: cheap watcher fingerprint ────────────────────
+
+def _make_db(tmp_path: Path):
+    db = tmp_path / "state.db"
+    conn = sqlite3.connect(str(db))
+    conn.executescript(
+        """
+        CREATE TABLE sessions (
+            id TEXT PRIMARY KEY,
+            source TEXT NOT NULL,
+            session_source TEXT,
+            model TEXT,
+            started_at REAL NOT NULL,
+            ended_at REAL,
+            end_reason TEXT,
+            parent_session_id TEXT,
+            message_count INTEGER DEFAULT 0,
+            title TEXT,
+            archived INTEGER DEFAULT 0
+        );
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            role TEXT NOT NULL,
+            content TEXT,
+            timestamp REAL NOT NULL
+        );
+        """
+    )
+    conn.commit()
+    return db, conn
+
+
+def _add_session(conn, sid, source="telegram", mc=2, started=None, title="Chat"):
+    started = started or time.time()
+    conn.execute(
+        "INSERT OR REPLACE INTO sessions (id, source, model, started_at, message_count, title) "
+        "VALUES (?, ?, 'm', ?, ?, ?)",
+        (sid, source, started, mc, title),
+    )
+    for i in range(mc):
+        conn.execute(
+            "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, 'user', 'x', ?)",
+            (sid, started + i),
+        )
+    conn.commit()
+
+
+def test_cheap_fingerprint_stable_and_sensitive(tmp_path):
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    _add_session(conn, "tg1", "telegram", mc=2)
+    _add_session(conn, "dc1", "discord", mc=3)
+
+    fp1 = gw._cheap_change_fingerprint(db)
+    fp2 = gw._cheap_change_fingerprint(db)
+    assert fp1 is not None
+    assert fp1 == fp2, "fingerprint must be stable when nothing changes"
+
+    # New message in a visible session bumps message_count -> fingerprint changes.
+    _add_session(conn, "tg1", "telegram", mc=3)
+    fp3 = gw._cheap_change_fingerprint(db)
+    assert fp3 != fp1, "fingerprint must change when a visible session gains a message"
+
+    # New session appears -> fingerprint changes.
+    _add_session(conn, "tg2", "telegram", mc=1)
+    fp4 = gw._cheap_change_fingerprint(db)
+    assert fp4 != fp3
+
+
+def test_cheap_fingerprint_ignores_excluded_sources(tmp_path):
+    """cron/webui churn must not invalidate the fingerprint (matches projection scope)."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    _add_session(conn, "tg1", "telegram", mc=2)
+    fp1 = gw._cheap_change_fingerprint(db)
+
+    # A cron session churns heavily — but cron is excluded from the sidebar, so
+    # the fingerprint (and thus the expensive projection) must NOT fire.
+    _add_session(conn, "cron1", "cron", mc=50)
+    fp2 = gw._cheap_change_fingerprint(db)
+    assert fp2 == fp1, "cron-only churn must not trigger a re-projection"
+
+    # A webui session likewise excluded.
+    _add_session(conn, "webui1", "webui", mc=20)
+    fp3 = gw._cheap_change_fingerprint(db)
+    assert fp3 == fp1
+
+
+def test_cheap_fingerprint_detects_source_change(tmp_path):
+    """A source retag changes the projection's derived source_label / visibility,
+    so the cheap fingerprint MUST change even though no displayed field moved."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    _add_session(conn, "s1", "telegram", mc=2)
+    fp1 = gw._cheap_change_fingerprint(db)
+
+    conn.execute("UPDATE sessions SET source = 'discord' WHERE id = 's1'")
+    conn.commit()
+    fp2 = gw._cheap_change_fingerprint(db)
+    assert fp2 != fp1, "a source change alters projected metadata and must be detected"
+
+
+def test_cheap_fingerprint_detects_same_count_message_rewrite(tmp_path):
+    """Regression (#3536 review): SessionDB.replace_messages (/retry, /undo,
+    /compress) deletes + reinserts a transcript with NEW timestamps but can leave
+    sessions.message_count UNCHANGED. The projection's last_activity
+    (MAX(messages.timestamp)) moves, so the cheap fingerprint MUST still change
+    even though every sessions-table column is identical — otherwise the watcher
+    skips a re-projection and other tabs show stale last_activity ordering."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    _add_session(conn, "s1", "telegram", mc=3)
+    fp1 = gw._cheap_change_fingerprint(db)
+
+    # Simulate replace_messages: same count (3), brand-new timestamps, no change
+    # to ANY sessions-table column (message_count stays 3).
+    conn.execute("DELETE FROM messages WHERE session_id = 's1'")
+    base = time.time() + 10_000  # strictly later than the originals
+    for i in range(3):
+        conn.execute(
+            "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, 'user', 'rewritten', ?)",
+            ("s1", base + i),
+        )
+    conn.commit()
+    # sessions table is byte-identical to before; only messages moved.
+    assert conn.execute("SELECT message_count FROM sessions WHERE id='s1'").fetchone()[0] == 3
+    fp2 = gw._cheap_change_fingerprint(db)
+    assert fp2 != fp1, (
+        "a same-count transcript rewrite moves MAX(messages.timestamp) and must "
+        "invalidate the fingerprint so the watcher re-projects"
+    )
+
+
+def test_cheap_fingerprint_detects_lineage_only_change(tmp_path):
+    """Lineage/visibility fields the projection uses for collapse (parent_session_id,
+    end_reason, ended_at) must be part of the fingerprint."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    _add_session(conn, "s1", "telegram", mc=2)
+    fp0 = gw._cheap_change_fingerprint(db)
+
+    conn.execute("UPDATE sessions SET parent_session_id = 'p-root' WHERE id = 's1'")
+    conn.commit()
+    fp1 = gw._cheap_change_fingerprint(db)
+    assert fp1 != fp0, "parent_session_id change (compression lineage) must be detected"
+
+    conn.execute("UPDATE sessions SET end_reason = 'compressed' WHERE id = 's1'")
+    conn.commit()
+    fp2 = gw._cheap_change_fingerprint(db)
+    assert fp2 != fp1, "end_reason change must be detected"
+
+    conn.execute("UPDATE sessions SET ended_at = 1234567890.0 WHERE id = 's1'")
+    conn.commit()
+    fp3 = gw._cheap_change_fingerprint(db)
+    assert fp3 != fp2, "ended_at change must be detected"
+
+
+def test_cheap_fingerprint_handles_missing_db(tmp_path):
+    gw = importlib.import_module("api.gateway_watcher")
+    missing = tmp_path / "nope.db"
+    # No exception, returns None so the caller falls back to the full read.
+    assert gw._cheap_change_fingerprint(missing) is None
+
+
+def test_cheap_fingerprint_handles_missing_optional_columns(tmp_path):
+    """Older agent schemas without archived/ended_at must still produce a fingerprint."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db = tmp_path / "old.db"
+    conn = sqlite3.connect(str(db))
+    conn.executescript(
+        """
+        CREATE TABLE sessions (
+            id TEXT PRIMARY KEY,
+            source TEXT NOT NULL,
+            model TEXT,
+            started_at REAL NOT NULL,
+            message_count INTEGER DEFAULT 0,
+            title TEXT
+        );
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            role TEXT NOT NULL,
+            content TEXT,
+            timestamp REAL NOT NULL
+        );
+        """
+    )
+    conn.execute(
+        "INSERT INTO sessions (id, source, model, started_at, message_count, title) "
+        "VALUES ('s1', 'telegram', 'm', ?, 2, 't')",
+        (time.time(),),
+    )
+    conn.commit()
+    fp = gw._cheap_change_fingerprint(db)
+    assert fp is not None and isinstance(fp, str)
+
+
+def test_cheap_fingerprint_returns_none_without_source_column(tmp_path):
+    """A pre-source-tracking schema must return None (forces safe full read)."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db = tmp_path / "ancient.db"
+    conn = sqlite3.connect(str(db))
+    conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, started_at REAL)")
+    conn.commit()
+    assert gw._cheap_change_fingerprint(db) is None
+
+
+def test_poll_loop_skips_projection_when_unchanged(tmp_path, monkeypatch):
+    """The poll body must call the expensive projection only when the cheap fp changes."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    _add_session(conn, "tg1", "telegram", mc=2)
+
+    monkeypatch.setattr(gw, "_get_state_db_path", lambda: db)
+
+    calls = {"n": 0}
+    real = gw._get_agent_sessions_from_db
+
+    def counting():
+        calls["n"] += 1
+        return real()
+
+    monkeypatch.setattr(gw, "_get_agent_sessions_from_db", counting)
+
+    w = gw.GatewayWatcher()
+
+    # Run the change-detection body directly (one iteration) without the thread.
+    def one_iteration():
+        db_path = gw._get_state_db_path()
+        cheap_fp = gw._cheap_change_fingerprint(db_path) if db_path.exists() else ''
+        if cheap_fp is not None and cheap_fp == w._last_cheap_fp:
+            return
+        sessions = gw._get_agent_sessions_from_db()
+        if cheap_fp is not None:
+            w._last_cheap_fp = cheap_fp
+        _ = gw._snapshot_hash(sessions)
+
+    one_iteration()  # first poll: must read
+    assert calls["n"] == 1
+    one_iteration()  # unchanged: must skip
+    one_iteration()  # still unchanged: must skip
+    assert calls["n"] == 1, "expensive projection must not run while state is unchanged"
+
+    _add_session(conn, "tg1", "telegram", mc=3)  # a real change
+    one_iteration()
+    assert calls["n"] == 2, "expensive projection must run again after a real change"
+
+
+def test_lru_eviction_skips_active_runs():
+    """Regression (#3536 review round 2): lowering SESSION_AGENT_CACHE_MAX (50→25)
+    makes LRU agent-cache eviction more likely to fire, so the eviction loop must
+    NOT close an agent whose worker is still live. The loop must consult the
+    ACTIVE_RUNS registry (worker lifecycle — survives a cancel/reconnect that
+    drops STREAMS) and skip those session_ids, deferring (temporarily exceeding
+    the cap) if every over-cap entry is active. Source-contract test: the deep
+    streaming function isn't unit-invokable, so pin the invariant in source."""
+    import pathlib
+    src = (pathlib.Path(__file__).resolve().parents[1] / "api" / "streaming.py").read_text()
+    idx = src.index("while len(SESSION_AGENT_CACHE) > SESSION_AGENT_CACHE_MAX:")
+    block = src[idx - 1600:idx + 700]
+    # The eviction path must build an active-session set from ACTIVE_RUNS...
+    assert "ACTIVE_RUNS" in block and "_active_sids" in block, (
+        "eviction must consult ACTIVE_RUNS to find live workers"
+    )
+    # ...skip active session_ids when choosing what to evict...
+    assert "_sid not in _active_sids" in block, (
+        "eviction must skip sessions with a live run"
+    )
+    # ...and defer (break) rather than evict when all over-cap entries are active.
+    assert "all over-cap entries are active; defer" in block, (
+        "eviction must defer (temporarily exceed cap) rather than close a live agent"
+    )
+    # The unconditional popitem(last=False) that closed the LRU agent regardless
+    # of liveness must be gone from this block.
+    assert "popitem(last=False)" not in block, (
+        "the liveness-blind popitem eviction must be replaced"
+    )

tests/test_memory_session_lifecycle_generation.py

@@ -328,7 +328,12 @@ def test_lru_eviction_commits_outside_cache_lock():
 
     assert "commit_session_memory" not in locked_section
     assert "_lifecycle_commit" not in locked_section
-    assert "SESSION_AGENT_CACHE.popitem" in locked_section
+    # Eviction now selects the oldest INACTIVE entry (active-run-aware) and pops
+    # it by id under the lock, rather than a liveness-blind popitem(last=False).
+    # The commit/close still happens outside the lock (asserted below).
+    assert "SESSION_AGENT_CACHE.pop(_evictable_sid)" in locked_section
+    assert "_sid not in _active_sids" in locked_section
+    assert "SESSION_AGENT_CACHE.popitem(last=False)" not in locked_section
     assert "_close_evicted_agent_at_session_boundary" in outside_section
     helper_start = src.index("def _close_evicted_agent_at_session_boundary")
     helper_end = src.index("\ndef _refresh_cached_agent_runtime", helper_start)

tests/test_v050259_sessiondb_fd_leak.py

@@ -61,8 +61,8 @@ def test_cached_agent_reuse_closes_old_session_db():
 
 def test_lru_eviction_closes_evicted_agent_session_db():
     """SAME LEAK SHAPE on the LRU eviction path: when SESSION_AGENT_CACHE
-    grows beyond SESSION_AGENT_CACHE_MAX (50), the LRU agent gets popped via
-    `popitem(last=False)`. Without explicit close, its `_session_db` waits
+    grows beyond SESSION_AGENT_CACHE_MAX (default 25), the LRU agent gets popped
+    via `popitem(last=False)`. Without explicit close, its `_session_db` waits
     on GC finalization which may never run on a long-lived server.
 
     Fix: capture the evicted entry, close its agent's `_session_db` before