jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
hermes-webui/tests/test_issue3506_memory_and_watcher.py
T
Anthony Ruiz a41bfe8690 fix(lineage): honor _reset_from, fail closed on untrusted markers, hash full watcher payload 
Layer three additive pieces on top of the #7021 / #6931 continuation
predicate without changing its shape or CONTINUATION_STARTED_AT_TOLERANCE_SECONDS:

- _reset_from is a boundary marker alongside _branched_from/_delegate_from
  (Hermes Agent stamps it in gateway/session_recovery.py and filters it in
  _NON_CONTINUATION_CHILD_FILTER_SQL). The same predicate drives the sidebar
  listing, lineage metadata/report and the transcript stitcher.
- _branch_markers() returns ('none' | 'markers' | 'unknown', markers);
  unparsable, non-object, too-deep, or non-string marker payloads are
  'unknown' and fail closed as a lineage boundary instead of degrading to
  "no markers" (fail open).
- gateway_watcher._snapshot_hash() hashes the complete canonicalized published
  payload, and the cheap fingerprint covers model_config, so a projection change
  to title/created_at/metadata with fixed (id, updated_at, message_count) emits
  sessions_changed.
2026-09-24 11:53:47 +00:00

742 lines
29 KiB
Python
Raw Permalink Blame History

"""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"
)
Reference in new issue View Git Blame Copy Permalink