From 174ce8770d3ef1b7638d80789fbad08d42be98fd Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Sun, 9 Aug 2026 20:42:53 -0300 Subject: [PATCH] fix(state): arm the live-DB guard by process ancestry, not env alone MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production `state.db` files accumulate zero-message "open" gateway session rows carrying test-fixture identities (`chat-1` / `user-1` / `wx-chat`), with matching `gateway_routing` scopes pointing at `pytest-of-*` temp directories. The escape is structural. Hermetic isolation rides entirely on the process environment: `HERMES_HOME` says *where* to write, `PYTEST_CURRENT_TEST` / `PYTEST_VERSION` say *whether the guard is armed*. Both travel in the same carrier, so a child spawned with a rebuilt environment loses them together — it resolves the developer's real `state.db` *and* silences the only check that would have stopped it, in one step. The guard is a no-op in precisely the situation it was written for. Back the env probe with process ancestry, which survives an env rebuild: * `_process_looks_like_pytest()` matches a pytest launcher by argv token basename, so `/tmp/pytest-of-dev/...` paths in real argv cannot false-positive, and an unreadable process is never assumed to be a test. * `_has_pytest_ancestor()` walks parents via psutil, memoised, and fails open when psutil is unavailable — a real `hermes` run pays for at most one walk and keeps the previous behaviour if the walk errors. * `_in_test_context()` checks env first (two dict lookups, covers the in-process case) and only then ancestry. `_STATE_DB_GUARD_BYPASS` is a module global and cannot cross a process boundary, so ancestry-armed children would have had no way to opt out at all; `HERMES_STATE_DB_GUARD_BYPASS=1` is the env-carried twin. Also sweeps the rows already written. Bulk prune/archive cannot reach them: their shared selector is pinned to `ended_at IS NOT NULL` so a live session is never picked, which permanently excludes every never-closed row. Adds a narrower selector — keyed, still open, and with no messages, tokens, tool calls, API calls, activity or title — behind `hermes sessions prune --never-active` (default floor 30 days, honours --dry-run/--yes). Routing entries naming a deleted row go with it, so the gateway is never left resuming a session id that no longer exists; `pinned` and `archived` rows are excluded as explicit user intent. Closes #82770 --- hermes_cli/main.py | 10 + hermes_cli/sessions_cmd.py | 72 ++++++ hermes_state.py | 212 +++++++++++++++++- .../test_live_db_guard_ancestry.py | 154 +++++++++++++ .../test_never_active_keyed_prune.py | 149 ++++++++++++ 5 files changed, 595 insertions(+), 2 deletions(-) create mode 100644 tests/hermes_state/test_live_db_guard_ancestry.py create mode 100644 tests/hermes_state/test_never_active_keyed_prune.py diff --git a/hermes_cli/main.py b/hermes_cli/main.py index e13cedcef6..5356b8dce7 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -12683,6 +12683,16 @@ def main(): action="store_true", help="Also delete archived sessions (excluded by default)", ) + sessions_prune.add_argument( + "--never-active", + action="store_true", + help=( + "Instead of ended sessions, delete keyed gateway rows that were " + "opened and never used (no messages, tokens, tool calls or title) " + "and are older than AGE (default 30 days). Ordinary prune can " + "never reach these — it only ever selects ended sessions" + ), + ) sessions_archive = sessions_subparsers.add_parser( "archive", diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py index 2acfea5ca6..d5a33953dc 100644 --- a/hermes_cli/sessions_cmd.py +++ b/hermes_cli/sessions_cmd.py @@ -55,6 +55,72 @@ def _confirm_prompt(prompt: str) -> bool: return False +#: Default age floor for `hermes sessions prune --never-active`. Deliberately +#: generous: the rows are worthless but harmless, and a young never-active row +#: may simply be a chat that nobody has replied to yet. +_NEVER_ACTIVE_DEFAULT_DAYS = 30.0 + + +def _prune_never_active_keyed(db, args): + """`hermes sessions prune --never-active` — drop leaked/dead keyed rows. + + Targets keyed gateway rows that were opened and never used at all. The + population is dominated by escaped test fixtures (#82770), which the + hermetic-isolation guard can only stop from being *created* — rows already + written to a developer's state.db need a sweep to leave. + """ + from hermes_cli.session_filters import format_epoch, parse_duration_seconds + + older_than = getattr(args, "older_than", None) + if older_than is None: + days = _NEVER_ACTIVE_DEFAULT_DAYS + else: + seconds = parse_duration_seconds(str(older_than)) + if seconds is None: + print( + f"Error: --older-than '{older_than}' is not a duration. " + "Use a bare number of days or a form like '2d' / '1w'." + ) + return + days = seconds / 86400.0 + + candidates = db.list_never_active_keyed_sessions(older_than_days=days) + if not candidates: + print(f"No never-active keyed sessions older than {days:g} day(s).") + return + + shown = candidates if args.dry_run else candidates[:15] + print( + f"{len(candidates)} never-active keyed session(s) older than " + f"{days:g} day(s) — no messages, tokens, tool calls or title:" + ) + for s in shown: + print( + f" {s['id']} {format_epoch(s.get('started_at')):<17} " + f"{(s.get('source') or '-'):<10} {s.get('session_key') or '-'}" + ) + if len(candidates) > len(shown): + print(f" … {len(candidates) - len(shown)} more") + + if args.dry_run: + print("Dry run — nothing deleted.") + return + if not args.yes and not _confirm_prompt( + f"Delete {len(candidates)} session(s)? [y/N] " + ): + print("Aborted.") + return + + sessions_dir = get_hermes_home() / "sessions" + deleted, routing_deleted = db.prune_never_active_keyed_sessions( + older_than_days=days, sessions_dir=sessions_dir + ) + print( + f"Deleted {deleted} never-active session(s) and {routing_deleted} " + "stale routing entr(ies)." + ) + + def cmd_sessions(args, sessions_parser=None): import json as _json @@ -806,6 +872,12 @@ def cmd_sessions(args, sessions_parser=None): else: print(f"Session '{args.session_id}' not found.") + elif action == "prune" and getattr(args, "never_active", False): + # Separate branch on purpose: the shared prune/archive selector is + # pinned to `ended_at IS NOT NULL`, so never-closed rows sit outside + # it by construction and cannot be expressed as one more filter. + _prune_never_active_keyed(db, args) + elif action in ("prune", "archive"): from hermes_cli.session_filters import ( build_prune_filters, diff --git a/hermes_state.py b/hermes_state.py index d67dc076b6..6653b61956 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -423,6 +423,12 @@ def _default_db_path() -> Path: #: ``@pytest.mark.live_system_guard_bypass``; scripts may set it explicitly. _STATE_DB_GUARD_BYPASS = False +#: Env-carried twin of ``_STATE_DB_GUARD_BYPASS``. A module global cannot +#: cross a process boundary, so a test that deliberately points a *child* at +#: the live DB has no way to opt out once ancestry arms the guard there. +#: Export this in the child's env instead. +_STATE_DB_GUARD_BYPASS_ENV = "HERMES_STATE_DB_GUARD_BYPASS" + #: Additional production roots to refuse (beyond the platform default #: ``~/.hermes``). The test conftest injects the pre-sandbox production #: root here so custom-``HERMES_HOME`` deployments are covered too. @@ -463,6 +469,82 @@ def _running_under_pytest() -> bool: ) +#: Names that identify a pytest launcher in a process command line. Matched +#: against the *basename* of each argv token so ``/tmp/pytest-of-dev/...`` +#: paths — which do show up in real argv — cannot false-positive. +_PYTEST_LAUNCHER_NAMES = frozenset( + {"pytest", "py.test", "pytest.exe", "py.test.exe"} +) + +#: Memoised ancestry answer. The process tree above us does not change in a +#: way that matters here, and the walk must not cost anything on the hot path. +_PYTEST_ANCESTOR: Optional[bool] = None + + +def _process_looks_like_pytest(proc: Any) -> bool: + """True when *proc*'s command line is a pytest invocation. + + Covers both ``pytest ...`` (launcher on argv[0]) and ``python -m pytest`` + (launcher as a bare ``pytest`` token). A process whose command line we + cannot read is treated as "not pytest": guessing the other way would + refuse production opens for unrelated reasons. + """ + try: + cmdline = proc.cmdline() or [] + except Exception: + return False + for arg in cmdline: + try: + name = os.path.basename(str(arg).strip('"').strip("'")).lower() + except Exception: + continue + if name in _PYTEST_LAUNCHER_NAMES: + return True + return False + + +def _has_pytest_ancestor() -> bool: + """True when some ancestor process of this one is a pytest run. + + ``_running_under_pytest`` reads ``PYTEST_*`` env vars, which a child + spawned with a rebuilt environment loses at the same moment it loses the + ``HERMES_HOME`` redirect: that child aims at the production DB *and* + disarms the guard in one step (#82770). Ancestry is the one test-context + signal that survives an env rebuild, so it backs the env check up. + + Fails open (``False``) when ``psutil`` is unavailable or the walk errors — + that restores the previous env-only behaviour rather than blocking real + user runs on a psutil hiccup. + """ + global _PYTEST_ANCESTOR + if _PYTEST_ANCESTOR is not None: + return _PYTEST_ANCESTOR + found = False + if psutil is not None: + try: + for parent in psutil.Process().parents(): + if _process_looks_like_pytest(parent): + found = True + break + except Exception: + found = False + _PYTEST_ANCESTOR = found + return found + + +def _in_test_context() -> bool: + """True when this process is a test run, by environment or by ancestry. + + Order matters for cost: the env probe is two dict lookups and covers the + common in-process case, so the ancestry walk only runs for processes the + environment claims are ordinary user runs — and its answer is memoised, + so a real ``hermes`` invocation pays for at most one walk. + """ + if _running_under_pytest(): + return True + return _has_pytest_ancestor() + + def _production_state_roots() -> List[Path]: roots: List[Path] = [] real_root = _real_platform_state_root() @@ -501,8 +583,15 @@ def _ensure_test_isolation(db_path: Path) -> None: Raises ``RuntimeError`` before any connection, mkdir, journal-mode pragma, or byte probe can touch the live database. No-op outside pytest and for hermetic (tmp ``HERMES_HOME``) paths. + + "pytest context" means environment *or* process ancestry — see + :func:`_in_test_context`. Env alone is not enough: a child spawned with + a rebuilt environment loses ``PYTEST_*`` and ``HERMES_HOME`` together, + which is precisely the state in which it writes to production (#82770). """ - if _STATE_DB_GUARD_BYPASS or not _running_under_pytest(): + if _STATE_DB_GUARD_BYPASS or os.environ.get(_STATE_DB_GUARD_BYPASS_ENV): + return + if not _in_test_context(): return try: resolved = Path(db_path).expanduser().resolve() @@ -517,7 +606,9 @@ def _ensure_test_isolation(db_path: Path) -> None: "explicit tmp db_path or let the hermetic conftest redirect " "HERMES_HOME. If this test genuinely needs the live " "database, mark it with " - "@pytest.mark.live_system_guard_bypass." + "@pytest.mark.live_system_guard_bypass — or, for a spawned " + f"child process, export {_STATE_DB_GUARD_BYPASS_ENV}=1 in " + "its environment." ) # --------------------------------------------------------------------------- @@ -4539,6 +4630,123 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) self._execute_write(_do) + def list_never_active_keyed_sessions( + self, *, older_than_days: float + ) -> List[Dict[str, Any]]: + """Keyed gateway rows that were opened and then never used at all. + + Selects rows that are keyed (``session_key IS NOT NULL``), still open + (``ended_at IS NULL``) and carry no evidence of a single turn: no + messages, no tokens, no tool or API calls, no recorded activity, no + title. Such a row is indistinguishable from "never happened". + + That is exactly the shape of a leaked test fixture (#82770) — and + also of a chat that was routed but never answered. Both are safe to + drop: there is no transcript to lose, and the gateway mints a fresh + session on the next inbound message either way. + + ``bulk prune``/``archive`` cannot reach these rows: their shared + selector is pinned to ``ended_at IS NOT NULL`` so that a live session + is never picked, which permanently excludes every never-closed row. + Hence a separate, narrower selector rather than another filter flag. + + ``pinned`` and ``archived`` rows are excluded — both are explicit + user intent to keep the row around. + """ + cutoff = time.time() - (float(older_than_days) * 86400.0) + with self._lock: + rows = self._conn.execute( + """ + SELECT s.id, s.session_key, s.source, s.chat_id, + s.chat_type, s.user_id, s.started_at + FROM sessions s + WHERE s.session_key IS NOT NULL + AND s.ended_at IS NULL + AND s.title IS NULL + AND s.last_activity_at IS NULL + AND COALESCE(s.message_count, 0) = 0 + AND COALESCE(s.tool_call_count, 0) = 0 + AND COALESCE(s.api_call_count, 0) = 0 + AND COALESCE(s.input_tokens, 0) = 0 + AND COALESCE(s.output_tokens, 0) = 0 + AND COALESCE(s.pinned, 0) = 0 + AND COALESCE(s.archived, 0) = 0 + AND s.started_at IS NOT NULL + AND s.started_at < ? + AND NOT EXISTS ( + SELECT 1 FROM messages m WHERE m.session_id = s.id + ) + ORDER BY s.started_at + """, + (cutoff,), + ).fetchall() + return [dict(r) for r in rows] + + def _delete_routing_entries_for_sessions(self, session_ids: Set[str]) -> int: + """Drop ``gateway_routing`` rows pointing at any of *session_ids*. + + Routing entries are keyed by ``(scope, session_key)`` and record their + target session inside ``entry_json``, so there is no way to reach them + by session id in SQL — the match is done in Python over all scopes. + """ + if not session_ids: + return 0 + with self._lock: + rows = self._conn.execute( + "SELECT scope, session_key, entry_json FROM gateway_routing" + ).fetchall() + doomed: List[Tuple[str, str]] = [] + for row in rows: + try: + entry = json.loads(row["entry_json"] or "{}") + except Exception: + continue + if isinstance(entry, dict) and entry.get("session_id") in session_ids: + doomed.append((row["scope"], row["session_key"])) + if not doomed: + return 0 + + def _do(conn): + conn.executemany( + "DELETE FROM gateway_routing WHERE scope = ? AND session_key = ?", + doomed, + ) + + self._execute_write(_do) + return len(doomed) + + def prune_never_active_keyed_sessions( + self, + *, + older_than_days: float, + sessions_dir: Optional[Path] = None, + ) -> Tuple[int, int]: + """Delete never-active keyed rows and the routing entries naming them. + + Returns ``(sessions_deleted, routing_entries_deleted)``. + + The routing entries go first: a stale entry that outlived its target + would leave the gateway resuming a session id that no longer exists. + Deleting the pair is what leaving them both would have amounted to + anyway — the target had no transcript to resume. + + Deletion goes through :meth:`delete_session` rather than a bulk + ``DELETE`` so the delegate cascade, FTS bookkeeping and on-disk + transcript cleanup stay owned by one implementation. + """ + candidates = self.list_never_active_keyed_sessions( + older_than_days=older_than_days + ) + if not candidates: + return (0, 0) + ids = {str(row["id"]) for row in candidates} + routing_deleted = self._delete_routing_entries_for_sessions(ids) + deleted = 0 + for session_id in ids: + if self.delete_session(session_id, sessions_dir=sessions_dir): + deleted += 1 + return (deleted, routing_deleted) + def list_gateway_sessions( self, *, diff --git a/tests/hermes_state/test_live_db_guard_ancestry.py b/tests/hermes_state/test_live_db_guard_ancestry.py new file mode 100644 index 0000000000..713a7ca1fc --- /dev/null +++ b/tests/hermes_state/test_live_db_guard_ancestry.py @@ -0,0 +1,154 @@ +"""The live-DB guard must survive a scrubbed child environment (#82770). + +Forensic background: a read-only sweep of production ``state.db`` files found +hundreds of zero-message "open" gateway session rows carrying test-fixture +identities (``chat-1`` / ``user-1`` / ``wx-chat``), with matching +``gateway_routing`` scopes pointing at ``pytest-of-*`` temp directories. + +The escape is structural, not a one-off test bug. Hermetic isolation rides +entirely on the process environment: ``HERMES_HOME`` says *where* to write and +``PYTEST_CURRENT_TEST`` / ``PYTEST_VERSION`` say *whether the guard is armed*. +Both live in the same carrier, so a child spawned with a rebuilt environment +loses them together — it aims at the developer's real ``state.db`` and +silences the only check that would have stopped it, in one step. + +Process ancestry is the signal that survives an env rebuild, so these tests +pin that the guard is armed by ancestry when the environment no longer says +"pytest". + +These tests drive ``_ensure_test_isolation`` rather than constructing a real +``SessionDB``: if the guard regresses, the assertion must fail *without* the +test itself writing to the developer's live database. +""" + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +import hermes_state + +REPO_ROOT = Path(__file__).resolve().parents[2] + +# Probe run in the child: resolve the REAL platform state root (not a +# hardcoded ~/.hermes — that root is %LOCALAPPDATA%\hermes on Windows) and +# report whether the guard refuses it. +_CHILD_PROBE = """ +import sys +sys.path.insert(0, {repo!r}) +import hermes_state + +root = hermes_state._real_platform_state_root() +if root is None: + print("NO-ROOT") +else: + try: + hermes_state._ensure_test_isolation(root / "state.db") + except RuntimeError: + print("REFUSED") + else: + print("ALLOWED") +""" + + +def _scrubbed_env(**overrides): + """The environment a rebuilt-from-scratch child spawn ends up with.""" + env = { + k: v + for k, v in os.environ.items() + if not k.startswith("PYTEST_") and k != "HERMES_HOME" + } + env.update(overrides) + return env + + +def _run_probe(env): + result = subprocess.run( + [sys.executable, "-c", _CHILD_PROBE.format(repo=str(REPO_ROOT))], + env=env, + capture_output=True, + text=True, + cwd=str(REPO_ROOT), + timeout=120, + ) + verdict = result.stdout.strip().splitlines()[-1] if result.stdout.strip() else "" + if verdict == "NO-ROOT": + pytest.skip("no real platform state root resolvable on this machine") + assert verdict in ("REFUSED", "ALLOWED"), ( + f"probe produced no verdict.\nstdout={result.stdout!r}\n" + f"stderr={result.stderr!r}" + ) + return verdict + + +class TestScrubbedChildEnvironment: + def test_child_without_pytest_env_still_refuses_production_db(self): + """The #82770 escape: no PYTEST_* and no HERMES_HOME, yet still a test. + + This is the exact shape of the leak — the child resolves the real + ``state.db`` because ``HERMES_HOME`` is gone, and the env-only guard + sees a "normal user run" because ``PYTEST_*`` is gone with it. + """ + assert _run_probe(_scrubbed_env()) == "REFUSED" + + def test_child_inheriting_pytest_env_still_refuses_production_db(self): + """The pre-existing env path must keep working unchanged.""" + env = dict(os.environ) + env.pop("HERMES_HOME", None) + env.setdefault("PYTEST_CURRENT_TEST", "tests/x.py::test_x (call)") + assert _run_probe(env) == "REFUSED" + + def test_env_bypass_lets_a_deliberate_child_through(self): + """A test that genuinely needs the live DB in a child can opt out. + + ``_STATE_DB_GUARD_BYPASS`` is a module global and cannot cross a + process boundary, so ancestry-armed children need an env-carried + escape hatch or they would have no way to opt out at all. + """ + env = _scrubbed_env(**{hermes_state._STATE_DB_GUARD_BYPASS_ENV: "1"}) + assert _run_probe(env) == "ALLOWED" + + +class TestPytestProcessRecognition: + """Unit-level checks for the ancestry predicate's matching rules.""" + + class _FakeProc: + def __init__(self, cmdline): + self._cmdline = cmdline + + def cmdline(self): + return self._cmdline + + @pytest.mark.parametrize( + "cmdline", + [ + ["/usr/bin/python", "-m", "pytest", "tests/"], + ["/venv/bin/pytest", "-q"], + [r"C:\venv\Scripts\pytest.exe", "-q"], + ["/usr/bin/py.test", "tests/"], + ], + ) + def test_recognises_pytest_invocations(self, cmdline): + assert hermes_state._process_looks_like_pytest(self._FakeProc(cmdline)) + + @pytest.mark.parametrize( + "cmdline", + [ + ["hermes", "gateway", "start"], + ["/usr/bin/python", "-m", "hermes_cli.main", "sessions", "list"], + # A path that merely *contains* "pytest" is not a pytest process: + # tmp paths like /tmp/pytest-of-dev/... show up in real argv. + ["hermes", "run", "--file", "/tmp/pytest-of-dev/test0/input.txt"], + ], + ) + def test_ignores_non_pytest_invocations(self, cmdline): + assert not hermes_state._process_looks_like_pytest(self._FakeProc(cmdline)) + + def test_unreadable_process_is_not_pytest(self): + class _Denied: + def cmdline(self): + raise PermissionError("access denied") + + assert not hermes_state._process_looks_like_pytest(_Denied()) diff --git a/tests/hermes_state/test_never_active_keyed_prune.py b/tests/hermes_state/test_never_active_keyed_prune.py new file mode 100644 index 0000000000..b96129e1e1 --- /dev/null +++ b/tests/hermes_state/test_never_active_keyed_prune.py @@ -0,0 +1,149 @@ +"""Sweeping never-active keyed gateway rows (#82770). + +The live-DB guard stops *new* fixture escapes, but it cannot touch rows that +are already in a developer's ``state.db`` — and bulk prune/archive cannot +either: their shared selector is pinned to ``ended_at IS NOT NULL`` so a live +session is never picked, which permanently excludes every never-closed row. + +These tests pin the narrow selector that reaches them, and — more importantly +— pin the rows it must refuse to touch. +""" + +import json +import time + +import pytest + +from hermes_state import SessionDB + +DAY = 86400.0 + + +@pytest.fixture() +def db(tmp_path): + return SessionDB(db_path=tmp_path / "state.db") + + +def _insert(db, session_id, *, age_days, **overrides): + """Insert a keyed gateway row directly, defaulting to the junk shape.""" + row = { + "id": session_id, + "source": "telegram", + "user_id": "user-1", + "session_key": f"agent:main:telegram:dm:{session_id}", + "chat_id": "chat-1", + "chat_type": "dm", + "started_at": time.time() - age_days * DAY, + "message_count": 0, + "tool_call_count": 0, + "api_call_count": 0, + "input_tokens": 0, + "output_tokens": 0, + "archived": 0, + "pinned": 0, + } + row.update(overrides) + cols = ", ".join(row) + placeholders = ", ".join("?" for _ in row) + db._conn.execute( + f"INSERT INTO sessions ({cols}) VALUES ({placeholders})", list(row.values()) + ) + db._conn.commit() + return session_id + + +class TestSelector: + def test_selects_old_never_active_keyed_row(self, db): + _insert(db, "junk-old", age_days=45) + found = db.list_never_active_keyed_sessions(older_than_days=30) + assert [r["id"] for r in found] == ["junk-old"] + + def test_ignores_row_inside_the_age_floor(self, db): + _insert(db, "junk-young", age_days=3) + assert db.list_never_active_keyed_sessions(older_than_days=30) == [] + + def test_ignores_unkeyed_row(self, db): + _insert(db, "cli-row", age_days=45, session_key=None, source="cli") + assert db.list_never_active_keyed_sessions(older_than_days=30) == [] + + def test_ignores_ended_row(self, db): + """Ended rows belong to ordinary prune — this selector must not + double-claim them.""" + _insert(db, "ended", age_days=45, ended_at=time.time() - 40 * DAY) + assert db.list_never_active_keyed_sessions(older_than_days=30) == [] + + @pytest.mark.parametrize( + "field, value", + [ + ("message_count", 1), + ("tool_call_count", 1), + ("api_call_count", 1), + ("input_tokens", 12), + ("output_tokens", 12), + ("title", "kept by the user"), + ("last_activity_at", 1785354069.0), + ("pinned", 1), + ("archived", 1), + ], + ) + def test_any_sign_of_use_or_intent_protects_the_row(self, db, field, value): + _insert(db, "used", age_days=45, **{field: value}) + assert db.list_never_active_keyed_sessions(older_than_days=30) == [] + + def test_row_with_messages_is_protected_even_if_counter_says_zero(self, db): + """``message_count`` is a denormalised counter — trust the messages.""" + _insert(db, "stale-counter", age_days=45) + db._conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) " + "VALUES (?, ?, ?, ?)", + ("stale-counter", "user", "hello", time.time() - 44 * DAY), + ) + db._conn.commit() + assert db.list_never_active_keyed_sessions(older_than_days=30) == [] + + +class TestPrune: + def test_deletes_candidates_and_leaves_everything_else(self, db): + _insert(db, "junk-a", age_days=45) + _insert(db, "junk-b", age_days=60) + _insert(db, "keeper-young", age_days=1) + _insert(db, "keeper-used", age_days=45, message_count=3) + + deleted, _ = db.prune_never_active_keyed_sessions(older_than_days=30) + + assert deleted == 2 + surviving = { + r[0] for r in db._conn.execute("SELECT id FROM sessions").fetchall() + } + assert surviving == {"keeper-young", "keeper-used"} + + def test_drops_routing_entries_pointing_at_deleted_rows(self, db): + """A routing entry that outlived its target would leave the gateway + resuming a session id that no longer exists.""" + _insert(db, "junk", age_days=45) + _insert(db, "keeper", age_days=1) + db.save_gateway_routing_entry( + "agent:main:telegram:dm:junk", + json.dumps({"session_id": "junk"}), + scope="/tmp/pytest-of-dev/test0", + ) + db.save_gateway_routing_entry( + "agent:main:telegram:dm:keeper", + json.dumps({"session_id": "keeper"}), + scope="/home/dev/project", + ) + + deleted, routing_deleted = db.prune_never_active_keyed_sessions( + older_than_days=30 + ) + + assert (deleted, routing_deleted) == (1, 1) + remaining = db._conn.execute( + "SELECT session_key FROM gateway_routing" + ).fetchall() + assert [r[0] for r in remaining] == ["agent:main:telegram:dm:keeper"] + + def test_no_candidates_is_a_no_op(self, db): + _insert(db, "keeper", age_days=1) + assert db.prune_never_active_keyed_sessions(older_than_days=30) == (0, 0) + assert db._conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 1