From 3ca1188f64602d27baeb29ffca5ea090d0259897 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 4 Jun 2026 12:28:06 -0700 Subject: [PATCH] =?UTF-8?q?Release=20v0.51.256=20=E2=80=94=20Release=20HX?= =?UTF-8?q?=20(stage-r4)=20(#3596)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- CHANGELOG.md | 5 + README.md | 2 + api/config.py | 35 +- api/gateway_watcher.py | 142 +++++- api/session_lifecycle.py | 33 ++ api/streaming.py | 45 +- tests/test_issue3506_memory_and_watcher.py | 429 ++++++++++++++++++ ...est_memory_session_lifecycle_generation.py | 7 +- tests/test_v050259_sessiondb_fd_leak.py | 4 +- 9 files changed, 688 insertions(+), 14 deletions(-) create mode 100644 tests/test_issue3506_memory_and_watcher.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 455c8cd5..8c1710b8 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 797b90f5..0b8cac0d 100644 --- a/README.md +++ b/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 | --- diff --git a/api/config.py b/api/config.py index 35c16231..7e590430 100644 --- a/api/config.py +++ b/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: diff --git a/api/gateway_watcher.py b/api/gateway_watcher.py index c8bf8644..c4ff3e15 100644 --- a/api/gateway_watcher.py +++ b/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) diff --git a/api/session_lifecycle.py b/api/session_lifecycle.py index 7f73dff1..c33c32ee 100644 --- a/api/session_lifecycle.py +++ b/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 diff --git a/api/streaming.py b/api/streaming.py index 9b380f8c..ac6f4291 100644 --- a/api/streaming.py +++ b/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: diff --git a/tests/test_issue3506_memory_and_watcher.py b/tests/test_issue3506_memory_and_watcher.py new file mode 100644 index 00000000..08b33e21 --- /dev/null +++ b/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" + ) diff --git a/tests/test_memory_session_lifecycle_generation.py b/tests/test_memory_session_lifecycle_generation.py index 2132c105..8109ad1a 100644 --- a/tests/test_memory_session_lifecycle_generation.py +++ b/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) diff --git a/tests/test_v050259_sessiondb_fd_leak.py b/tests/test_v050259_sessiondb_fd_leak.py index 6d3b748e..3b80ac97 100644 --- a/tests/test_v050259_sessiondb_fd_leak.py +++ b/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