fix(state): arm the live-DB guard by process ancestry, not env alone
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
This commit is contained in:
joaomarcos
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
+210
-2
@@ -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,
|
||||
*,
|
||||
|
||||
@@ -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())
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user