"""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_config 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_message_aggregate_does_not_read_role_payload( tmp_path, monkeypatch ): """The five-second fingerprint must stay on the covering session/timestamp index. Reading ``role`` forces a table lookup for every message row and made a 10 GB state.db fingerprint take tens of seconds even while WebUI was idle. Role-only changes are deliberately handled by the bounded periodic full projection, so this query can stay on the existing covering index. """ gw = importlib.import_module("api.gateway_watcher") db, conn = _make_db(tmp_path) conn.execute( "CREATE INDEX idx_messages_session ON messages(session_id, timestamp)" ) conn.commit() _add_session(conn, "s1", "telegram", mc=3) statements = [] real_open = gw.open_state_db_readonly def traced_open(path): traced = real_open(path) traced.set_trace_callback(statements.append) return traced monkeypatch.setattr(gw, "open_state_db_readonly", traced_open) assert gw._cheap_change_fingerprint(db) is not None aggregate = next( statement for statement in statements if "LEFT JOIN messages" in statement ) assert "m.role" not in aggregate.lower() assert "COUNT(m.id)" in aggregate assert "MAX(m.timestamp)" in aggregate plan = conn.execute("EXPLAIN QUERY PLAN " + aggregate).fetchall() assert any( "COVERING INDEX idx_messages_session" in str(row[3]) for row in plan ), plan def test_periodic_projection_recovers_role_only_sidebar_visibility_change(tmp_path): """Role-only mutations must not remain invisible forever.""" gw = importlib.import_module("api.gateway_watcher") db, conn = _make_db(tmp_path) _add_session(conn, "cli1", "cli", mc=1, title="Untitled") conn.execute( "CREATE INDEX idx_messages_session ON messages(session_id, timestamp)" ) conn.commit() watcher = gw.GatewayWatcher(state_db_path=db) subscriber = watcher.subscribe() assert watcher._poll_once(now=1.0) is True initial = subscriber.get_nowait() assert [row["session_id"] for row in initial["sessions"]] == ["cli1"] initial_fingerprint = watcher._last_cheap_fp conn.execute("UPDATE messages SET role = 'assistant' WHERE session_id = 'cli1'") conn.commit() assert gw._cheap_change_fingerprint(db) == initial_fingerprint before_deadline = 1.0 + watcher.PROJECTION_PARITY_INTERVAL - 1.0 assert watcher._poll_once(now=before_deadline) is False assert subscriber.empty() at_deadline = 1.0 + watcher.PROJECTION_PARITY_INTERVAL assert watcher._poll_once(now=at_deadline) is True event = subscriber.get_nowait() assert event["sessions"] == [] assert subscriber.empty() def test_marker_only_update_invalidates_projection_on_next_poll(tmp_path): """model_config is projection authority, so its mutation cannot wait for parity.""" gw = importlib.import_module("api.gateway_watcher") db, conn = _make_db(tmp_path) _add_session(conn, "compression-parent", "telegram", mc=1, started=100.0) _add_session(conn, "compression-child", "telegram", mc=1, started=199.0) conn.execute( "UPDATE sessions SET ended_at = 200.0, end_reason = 'compression' " "WHERE id = 'compression-parent'" ) conn.execute( "UPDATE sessions SET parent_session_id = 'compression-parent' " "WHERE id = 'compression-child'" ) conn.commit() watcher = gw.GatewayWatcher(state_db_path=db) subscriber = watcher.subscribe() assert watcher._poll_once(now=1.0) is True initial = subscriber.get_nowait() assert [row["session_id"] for row in initial["sessions"]] == [ "compression-child" ] # No timestamp, count, source, or message row changes. This one field turns # the previously-collapsed child into a direct branch and must be visible on # the ordinary five-second poll rather than the 60-second parity fallback. conn.execute( "UPDATE sessions SET model_config = ? WHERE id = 'compression-child'", ('{"_branched_from":"compression-parent"}',), ) conn.commit() assert watcher._poll_once(now=2.0) is True event = subscriber.get_nowait() assert {row["session_id"] for row in event["sessions"]} == { "compression-parent", "compression-child", } assert subscriber.empty() def test_snapshot_hash_covers_every_published_field(): """Any emitted field change must change the hash; ordering must not.""" gw = importlib.import_module("api.gateway_watcher") base = [ { "session_id": "b", "title": "Chat B", "model": "m", "message_count": 2, "created_at": 100.0, "updated_at": 150.0, "source": "telegram", "raw_source": "telegram", "session_source": "messaging", "source_label": "Telegram", }, { "session_id": "a", "title": "Chat A", "model": "m", "message_count": 1, "created_at": 50.0, "updated_at": 60.0, "source": "cli", "raw_source": "cli", "session_source": "cli", "source_label": "CLI", }, ] reference = gw._snapshot_hash(base) assert gw._snapshot_hash(list(reversed(base))) == reference, "order-independent" assert gw._snapshot_hash([dict(row) for row in base]) == reference, "deterministic" assert gw._snapshot_hash([]) != reference # Mutating any single published field, including the ones the old # id/updated_at/message_count triple ignored, changes the hash. for field in base[0]: mutated = [dict(row) for row in base] mutated[0][field] = f"changed-{field}" assert gw._snapshot_hash(mutated) != reference, field def test_marker_only_projection_change_emits_sse_event_without_row_churn(tmp_path): """ID / message count / activity stay fixed; title + segments change → event. Gate reproduction: a lineage-marker mutation reruns the projection (the cheap fingerprint covers ``model_config``) and changes the published title/segment projection of a row, but the session id set, every ``message_count`` and every ``updated_at`` in the payload are unchanged. The old snapshot hash saw no difference, so ``_last_sessions`` kept the stale projection and no subscriber event was queued. """ gw = importlib.import_module("api.gateway_watcher") db, conn = _make_db(tmp_path) # A compression chain whose root segment holds no messages (its # transcript moved to the continuation): the projection publishes ONE # row under the tip's id carrying the root's title and a segment count # of 2. A message-less root is hidden when it stands alone. _add_session(conn, "root-a", "telegram", mc=0, started=100.0, title="Alpha") _add_session(conn, "tip-b", "telegram", mc=1, started=150.0, title="Beta") conn.execute( "UPDATE sessions SET ended_at = 150.0, end_reason = 'compression' WHERE id = 'root-a'" ) conn.execute("UPDATE sessions SET parent_session_id = 'root-a' WHERE id = 'tip-b'") conn.commit() watcher = gw.GatewayWatcher(state_db_path=db) subscriber = watcher.subscribe() assert watcher._poll_once(now=1.0) is True initial = subscriber.get_nowait() initial_rows = {row["session_id"]: row for row in initial["sessions"]} assert set(initial_rows) == {"tip-b"} assert initial_rows["tip-b"]["title"] == "Alpha" assert initial_rows["tip-b"]["message_count"] == 1 assert initial_rows["tip-b"]["updated_at"] == 150.0 assert watcher._last_sessions == initial["sessions"] # Marker-only flip: stamp _reset_from on the tip. No timestamp, count, # source or message row changes anywhere. The tip is now a reset child # (its own conversation, own title, one segment) and the message-less # root drops out — the published id set, message_count and updated_at # are byte-identical to the previous snapshot. conn.execute( "UPDATE sessions SET model_config = ? WHERE id = 'tip-b'", ('{"_reset_from":"root-a"}',), ) conn.commit() assert gw._cheap_change_fingerprint(db) != watcher._last_cheap_fp assert watcher._poll_once(now=2.0) is True event = subscriber.get_nowait() rows = {row["session_id"]: row for row in event["sessions"]} assert set(rows) == {"tip-b"} assert rows["tip-b"]["title"] == "Beta" assert rows["tip-b"]["message_count"] == initial_rows["tip-b"]["message_count"] assert rows["tip-b"]["updated_at"] == initial_rows["tip-b"]["updated_at"] assert watcher._last_sessions == event["sessions"] assert subscriber.empty() # A title-only projection change on an otherwise identical row is also # delivered on the ordinary poll. conn.execute("UPDATE sessions SET title = 'Beta renamed' WHERE id = 'tip-b'") conn.commit() assert watcher._poll_once(now=3.0) is True event = subscriber.get_nowait() rows = {row["session_id"]: row for row in event["sessions"]} assert set(rows) == {"tip-b"} assert rows["tip-b"]["title"] == "Beta renamed" assert rows["tip-b"]["updated_at"] == initial_rows["tip-b"]["updated_at"] assert watcher._last_sessions == event["sessions"] assert subscriber.empty() # Unchanged payload: the parity projection reruns but emits nothing. assert watcher._poll_once(now=3.0 + watcher.PROJECTION_PARITY_INTERVAL) is True assert subscriber.empty() def test_initial_missing_db_does_not_publish_an_empty_snapshot(tmp_path): gw = importlib.import_module("api.gateway_watcher") watcher = gw.GatewayWatcher(state_db_path=tmp_path / "missing.db") subscriber = watcher.subscribe() assert watcher._poll_once(now=1.0) is False assert subscriber.empty() assert watcher._last_full_projection_at is None def test_initial_projection_failure_does_not_publish_empty_and_retries( tmp_path, monkeypatch ): gw = importlib.import_module("api.gateway_watcher") db, conn = _make_db(tmp_path) conn.close() attempts = [] def fail_once_then_project_empty(*args, **kwargs): attempts.append(True) if len(attempts) == 1: raise sqlite3.OperationalError("projection failed") return [] monkeypatch.setattr( gw, "read_importable_agent_session_rows", fail_once_then_project_empty ) watcher = gw.GatewayWatcher(state_db_path=db) subscriber = watcher.subscribe() assert watcher._poll_once(now=1.0) is False assert attempts == [True] assert subscriber.empty() assert watcher._last_sessions == [] assert watcher._last_hash == "" assert watcher._last_cheap_fp == "" assert watcher._last_full_projection_at is None assert watcher._poll_once(now=2.0) is True assert attempts == [True, True] assert subscriber.get_nowait()["sessions"] == [] assert watcher._last_full_projection_at == 2.0 def test_projection_failure_preserves_populated_state_and_parity_retry( tmp_path, monkeypatch ): gw = importlib.import_module("api.gateway_watcher") db, conn = _make_db(tmp_path) _add_session(conn, "tg1", "telegram", mc=2) watcher = gw.GatewayWatcher(state_db_path=db) subscriber = watcher.subscribe() assert watcher._poll_once(now=1.0) is True initial_event = subscriber.get_nowait() assert [row["session_id"] for row in initial_event["sessions"]] == ["tg1"] initial_sessions = watcher._last_sessions initial_hash = watcher._last_hash initial_fingerprint = watcher._last_cheap_fp real_projection = gw.read_importable_agent_session_rows attempts = [] def fail_once_then_project(*args, **kwargs): attempts.append(True) if len(attempts) == 1: raise sqlite3.OperationalError("projection failed") return real_projection(*args, **kwargs) monkeypatch.setattr( gw, "read_importable_agent_session_rows", fail_once_then_project ) at_deadline = 1.0 + watcher.PROJECTION_PARITY_INTERVAL assert watcher._poll_once(now=at_deadline) is False assert attempts == [True] assert subscriber.empty() assert watcher._last_sessions is initial_sessions assert watcher._last_hash == initial_hash assert watcher._last_cheap_fp == initial_fingerprint assert watcher._last_full_projection_at == 1.0 assert watcher._poll_once(now=at_deadline + 1.0) is True assert attempts == [True, True] assert subscriber.empty() assert watcher._last_sessions is initial_sessions assert watcher._last_hash == initial_hash assert watcher._last_cheap_fp == initial_fingerprint assert watcher._last_full_projection_at == at_deadline + 1.0 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) projected = [] def fake_projection(_path): projected.append(True) return [{"session_id": "tg1"}] monkeypatch.setattr(gw, "_get_agent_sessions_from_db", fake_projection) w = gw.GatewayWatcher(state_db_path=db) w.subscribe() # Idle watchers deliberately do not poll without observers. assert w._poll_once(now=1.0) is True assert projected == [True] # Mutate only an excluded WebUI row: irrelevant churn must not trigger another # expensive projection before the parity deadline. _add_session(conn, "webui2", "webui", mc=99, title="WebUI run 2") assert w._poll_once(now=2.0) is False assert projected == [True] _add_session(conn, "tg1", "telegram", mc=3) # a real change assert w._poll_once(now=3.0) is True assert projected == [True, True] 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" )