From 1064a3a93515d33187769e686ba52a7b46095718 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:32:41 -0700 Subject: [PATCH] refactor(state): compact SessionMaintenanceMixin and state.db file helpers (-280 LOC, SQL-parity neutral) hermes_state_maintenance.py 557->418, hermes_state_dbfile.py 545->404. - _placeholders() replaces 4 inline ','.join('?'...) builders (identical output) - _write_guards_reject() unifies the lease/lock probe in sweep_orphaned_sessions and prune_sessions (same kwargs, same exception set; prune keeps set -= order) - _page_pragmas() absorbs the try/except-debug shape of logical_size_bytes and _freelist_ratio (log texts unchanged); _try_checkpoint() for the two WAL checkpoints in vacuum(); _seconds_since() for the two state_meta float parses - archived tri-state -> f'string' clause (byte-identical SQL) - dbfile: contextlib.suppress for pass-only excepts, lock closures collapsed, unreachable size<0 branch dropped, is_zeroed tail folded to one predicate - docstrings/comments compacted by hand; every WHY/lock-safety invariant kept SQL PARITY OK (1120 stmts), MSG PARITY OK, import smoke OK. --- hermes_state_dbfile.py | 387 +++++++++------------------- hermes_state_maintenance.py | 485 +++++++++++++----------------------- 2 files changed, 296 insertions(+), 576 deletions(-) diff --git a/hermes_state_dbfile.py b/hermes_state_dbfile.py index a6ee6c9c5a..d929e975b1 100644 --- a/hermes_state_dbfile.py +++ b/hermes_state_dbfile.py @@ -1,12 +1,11 @@ -"""state.db file-level health helpers. +"""state.db file-level health helpers, split out of ``hermes_state.py``. -Split out of ``hermes_state.py``: header probes (application_id / zeroed-file -detection), deleted-WAL-sidecar holder scans, quarantine of zeroed or -lock-poisoned databases, ``collect_state_db_stats`` and holder-process -classification. Every name is re-imported into ``hermes_state`` so -``hermes_state.`` keeps resolving — and tests that monkeypatch it keep -intercepting, because intra-module calls to patched helpers go through a -lazy ``from hermes_state import ...`` at call time. +Header probes (application_id / zeroed-file detection), deleted-WAL-sidecar +holder scans, quarantine of zeroed databases, ``collect_state_db_stats`` and +holder-process classification. Every name is re-imported into ``hermes_state`` +so ``hermes_state.`` keeps resolving — and tests that monkeypatch it keep +intercepting, because intra-module calls to patched helpers go through a lazy +``from hermes_state import ...`` at call time. """ from __future__ import annotations @@ -24,49 +23,35 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple from hermes_state_common import ( - FTS_REBUILD_DEFERRAL_KEY, - stat_db_file_identity as _stat_db_file_identity, + FTS_REBUILD_DEFERRAL_KEY, stat_db_file_identity as _stat_db_file_identity ) # Log-record parity with the origin module (caplog tests pin "hermes_state"). logger = logging.getLogger("hermes_state") -# _read_sqlite_application_id runs on EVERY write via _raise_if_db_replaced, -# against the LIVE state.db. A bare open()/read()/close() there is the -# howtocorrupt §2.2 bug: close() cancels every POSIX advisory lock this -# process holds on the file — one probe call drops the WAL-mode DMS shared -# lock the writer connection holds (see hermes_cli/sqlite_safe_read.py). With -# the DMS lock gone, a fresh opener in another process can treat this writer -# as dead and rerun WAL-index recovery underneath it. -# -# The probe therefore reads through a per-path fd cached for the life of the -# process: opening an fd never cancels locks (only close() does), and -# os.pread takes no shared file position. When the path is re-pointed at a -# new inode (the very replacement this probe exists to detect), the stale fd -# is RETIRED, never closed — closing it would cancel the live connection's -# locks on the old file. Replacement events are rare and halt writes anyway, -# so the leak is bounded. +# _read_sqlite_application_id runs on EVERY write (_raise_if_db_replaced) against the LIVE +# state.db. A bare open()/read()/close() there is the howtocorrupt §2.2 bug: close() cancels +# every POSIX advisory lock this process holds on the file, dropping the writer's WAL-mode DMS +# shared lock (see hermes_cli/sqlite_safe_read.py) so another process can treat this writer as +# dead and rerun WAL-index recovery underneath it. So the probe preads through a per-path fd +# cached for the life of the process (opening never cancels locks). When the path is re-pointed +# at a new inode (the very replacement this probe detects) the stale fd is RETIRED, never closed +# — closing it would cancel the live connection's locks. Replacements are rare and halt writes. _HEADER_PROBE_LOCK = threading.Lock() _HEADER_PROBE_FDS: "dict[str, tuple[int, int, int]]" = {} # key -> (fd, dev, ino) _RETIRED_HEADER_PROBE_FDS: "list[int]" = [] # intentionally never closed - _FTS_TABLE_NAMES = ("messages_fts", "messages_fts_trigram", "messages_fts_cjk") def _pread_db_header(db_path: Path, length: int) -> "Optional[bytes]": - """Lock-safe raw header read of a possibly-live SQLite database. - - POSIX: pread from a cached, never-closed fd (rebound when the path names - a new inode). Windows: plain read — advisory-lock cancellation is a - POSIX-only hazard and msvcrt locks do not share the failure mode. - """ + """Lock-safe raw header read of a possibly-live SQLite database: POSIX preads from a cached, + never-closed fd (rebound when the path names a new inode); Windows reads plainly, since + advisory-lock cancellation is a POSIX-only hazard.""" from hermes_state import _IS_WINDOWS if _IS_WINDOWS: - try: - with db_path.open("rb") as handle: - return handle.read(length) - except OSError: - return None + with contextlib.suppress(OSError), db_path.open("rb") as handle: + return handle.read(length) + return None key = str(db_path) try: st = os.stat(db_path) @@ -76,9 +61,8 @@ def _pread_db_header(db_path: Path, length: int) -> "Optional[bytes]": cached = _HEADER_PROBE_FDS.get(key) if cached is not None and (cached[1], cached[2]) != (st.st_dev, st.st_ino): # Path re-pointed at a new file. Retire (never close) the old fd. - _RETIRED_HEADER_PROBE_FDS.append(cached[0]) + _RETIRED_HEADER_PROBE_FDS.append(_HEADER_PROBE_FDS.pop(key)[0]) cached = None - del _HEADER_PROBE_FDS[key] if cached is None: try: fd = os.open(db_path, os.O_RDONLY) @@ -90,18 +74,13 @@ def _pread_db_header(db_path: Path, length: int) -> "Optional[bytes]": _RETIRED_HEADER_PROBE_FDS.append(fd) return None cached = _HEADER_PROBE_FDS[key] = (fd, fst.st_dev, fst.st_ino) - try: + with contextlib.suppress(OSError): return os.pread(cached[0], length, 0) - except OSError: - return None + return None def _read_sqlite_application_id(db_path: Path) -> "Optional[int]": - """Read application_id from the SQLite header without opening a connection. - - Routed through :func:`_pread_db_header`, which never issues a ``close()`` - that would cancel this process's POSIX locks on the file. - """ + """application_id from the SQLite header, via the lock-safe :func:`_pread_db_header`.""" from hermes_state import _STATE_DB_APPLICATION_ID_OFFSET end = _STATE_DB_APPLICATION_ID_OFFSET + 4 header = _pread_db_header(db_path, end) @@ -112,13 +91,9 @@ def _read_sqlite_application_id(db_path: Path) -> "Optional[int]": def _stat_sqlite_sidecar_identity(db_path: Path) -> Dict[str, tuple]: """Snapshot ``(st_dev, st_ino)`` for existing WAL/SHM sidecars.""" - identities: Dict[str, tuple] = {} base = os.fspath(db_path) - for suffix in ("-wal", "-shm"): - ident = _stat_db_file_identity(Path(base + suffix)) - if ident is not None: - identities[suffix] = ident - return identities + idents = {suffix: _stat_db_file_identity(Path(base + suffix)) for suffix in ("-wal", "-shm")} + return {suffix: ident for suffix, ident in idents.items() if ident is not None} def _canonical_sqlite_path(path: str) -> str: @@ -142,25 +117,15 @@ def _iter_proc_fd_targets(): except OSError: continue # process gone or not ours for fd in fds: - try: + with contextlib.suppress(OSError): yield int(pid_str), os.readlink(f"{fd_dir}/{fd}") - except OSError: - continue def iter_deleted_sqlite_sidecar_holders(db_path) -> List[Tuple[int, str]]: - """Return processes holding an unlinked ``state.db-wal`` / ``-shm``. - - Linux-only (``/proc//fd`` readlink). Windows and other hosts - return ``[]`` — Windows cannot unlink a sidecar another process still - holds, and macOS does not use the `` (deleted)`` suffix. - - The scan includes this process: on the SessionDB open/write refuse - path, the in-process writer that still holds the orphan inode is the - one that must not mint a replacement WAL (and must stop committing). - ``_foreign_state_db_holders`` keeps skipping this PID for FTS - maintenance so a process does not block its own optional repair. - """ + """Return processes holding an unlinked ``state.db-wal`` / ``-shm``. Linux-only; ``[]`` + elsewhere (Windows cannot unlink a held sidecar, macOS has no `` (deleted)`` suffix). + Includes this process: on the open/write refuse path the in-process writer holding the orphan + inode must not mint a replacement WAL (``_foreign_state_db_holders`` skips this PID).""" if not sys.platform.startswith("linux"): return [] holders: List[Tuple[int, str]] = [] @@ -175,11 +140,8 @@ def iter_deleted_sqlite_sidecar_holders(db_path) -> List[Tuple[int, str]]: def refuse_deleted_wal_generation(db_path) -> None: - """Raise if any process holds a deleted WAL/SHM generation for *db_path*. - - Called *before* ``sqlite3.connect`` so a second opener cannot mint a - replacement WAL inode while a live writer still holds the orphan. - """ + """Raise if any process holds a deleted WAL/SHM generation for *db_path*; called + *before* ``sqlite3.connect`` so a second opener cannot mint a replacement WAL inode.""" from hermes_state import DeletedWalGenerationError, _DELETED_WAL_GENERATION_MSG if not iter_deleted_sqlite_sidecar_holders(db_path): return @@ -188,83 +150,51 @@ def refuse_deleted_wal_generation(db_path) -> None: def _connect_tracked_db(path, tracking_path=None, **kwargs): - """``sqlite3.connect`` that registers the open fd for lock-safety. - - While a connection is live, byte-level probes of the same file are - refused: an ``open()``/``close()`` cancels every POSIX advisory lock this - process holds on it -- including a running VACUUM's EXCLUSIVE lock. - Released automatically on ``close()``. - - The ONLY tolerated fallback is the helper being absent entirely - (scaffold/embed installs that ship hermes_state without hermes_cli). A - real connection failure must propagate: silently retrying an *untracked* - connect would disable the guard for the lifetime of that connection. - """ + """``sqlite3.connect`` that registers the open fd so byte-level probes of a live file are + refused (an ``open()``/``close()`` would cancel every POSIX lock, even a running VACUUM's + EXCLUSIVE). The ONLY tolerated fallback is the helper being absent (scaffold/embed installs + without hermes_cli); a real connection failure must propagate — a silent untracked retry + would disable the guard for that connection.""" try: from hermes_cli.sqlite_safe_read import connect_tracked except ImportError: - logger.debug( - "hermes_cli.sqlite_safe_read unavailable; opening %s untracked " - "(byte-probe guard inactive in this install)", - path, - ) + logger.debug("hermes_cli.sqlite_safe_read unavailable; opening %s untracked " + "(byte-probe guard inactive in this install)", path) return sqlite3.connect(str(path), **kwargs) - # Open through THIS module's sqlite3.connect so callers (and tests) that - # patch hermes_state.sqlite3.connect keep control of connection creation; - # the helper still owns tracking. + # Open through THIS module's sqlite3.connect so tests patching hermes_state.sqlite3.connect keep control. return connect_tracked(path, tracking_path=tracking_path, connect_fn=sqlite3.connect, **kwargs) def is_zeroed_state_db(path: Path, *, probe_bytes: int = 100, force: bool = False) -> bool: - """Detect the zeroed state.db signature (0-byte or NUL header). - - Byte-level probe, so it is only safe BEFORE any connection to *path* - exists in this process: ``close()`` cancels every POSIX advisory lock the - process holds on the file, which can pull the EXCLUSIVE lock out from - under a running VACUUM and corrupt the database. The read is routed - through ``read_header_bytes_preopen``, which refuses (returning False - here) once a connection is live. Pass ``force=True`` only for offline - files -- quarantined copies, snapshots, archives. - - Prefer ``hermes_cli.backup.is_zeroed_sqlite_file`` when available; this - local copy keeps SessionDB openable without importing the CLI package - in constrained embed paths. - """ - try: + """Detect the zeroed state.db signature (0-byte or NUL header). Byte-level probe, so only + safe BEFORE any connection to *path* exists in this process (``close()`` cancels every POSIX + lock, even a running VACUUM's EXCLUSIVE); ``read_header_bytes_preopen`` refuses (-> False) + once a connection is live. Pass ``force=True`` only for offline files (quarantined copies, + snapshots). Prefers ``hermes_cli.backup.is_zeroed_sqlite_file``; this copy keeps SessionDB + openable without the CLI package in constrained embed paths.""" + with contextlib.suppress(Exception): from hermes_cli.backup import is_zeroed_sqlite_file - return is_zeroed_sqlite_file(path, probe_bytes=probe_bytes, force=force) - except Exception: - pass try: + # Special files (FIFO, device, socket) are never "zeroed", and probing + # a FIFO would block until a writer appears. if not path.is_file(): - # Special files (FIFO, device, socket) are never "zeroed", and - # probing a FIFO would block until a writer appears. return False - size = path.stat().st_size + path.stat() except OSError: return False - if size < 0: - return False from hermes_cli.sqlite_safe_read import has_live_connection, read_header_bytes_preopen - if not force and has_live_connection(path): return False head = read_header_bytes_preopen(path, length=max(16, probe_bytes), force=force) - if head is None: - return False - if len(head) == 0: - return True - if head.startswith(b"SQLite format 3"): - return False - return all(byte == 0 for byte in head) + # b"" (0-byte file) is zeroed; all() over an empty header is True. + return head is not None and not head.startswith(b"SQLite format 3") and all(b == 0 for b in head) @contextlib.contextmanager def quarantine_cross_process_lock(path: Path, timeout: float = 5.0): """Acquire the cross-process lock for path.quarantine.lock.""" import platform - lock_path = path.with_name(path.name + ".quarantine.lock") lock_path.parent.mkdir(parents=True, exist_ok=True) handle = lock_path.open("a+b") @@ -273,27 +203,21 @@ def quarantine_cross_process_lock(path: Path, timeout: float = 5.0): if platform.system() == "Windows": import msvcrt - def _try_lock(): + def _lock(mode): # msvcrt locks a byte range from the current position handle.seek(0) - msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + msvcrt.locking(handle.fileno(), mode, 1) - def _unlock(): - handle.seek(0) - msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + _try_lock = lambda: _lock(msvcrt.LK_NBLCK) # noqa: E731 + _unlock = lambda: _lock(msvcrt.LK_UNLCK) # noqa: E731 else: import fcntl - - def _try_lock(): - fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - - def _unlock(): - fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + _try_lock = lambda: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) # noqa: E731 + _unlock = lambda: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) # noqa: E731 deadline = time.monotonic() + timeout - while True: + while not acquired: try: _try_lock() acquired = True - break except OSError: if time.monotonic() >= deadline: break @@ -310,23 +234,16 @@ def quarantine_cross_process_lock(path: Path, timeout: float = 5.0): def quarantine_zeroed_state_db(path: Path, *, already_locked: bool = False) -> Optional[Path]: - """Move a zeroed state.db aside (preserve bytes) and return quarantine path. - - Uses a cross-process lock so two concurrent startups cannot race: the first - process moves the zeroed file and the second re-checks under the lock, - finding the file already gone (or a fresh DB in its place) instead of - clobbering the quarantine. - """ + """Move a zeroed state.db aside (preserve bytes) and return quarantine path. A cross-process + lock stops two concurrent startups racing: the second re-checks under the lock and finds the + file gone (or fresh) instead of clobbering the quarantine.""" def _do_quarantine(): if not path.exists(): logger.info("quarantine_zeroed_state_db: %s already moved by another process", path) return None if not is_zeroed_state_db(path): - logger.info( - "quarantine_zeroed_state_db: %s is no longer zeroed (another " - "process quarantined it and a fresh DB was created)", - path, - ) + logger.info("quarantine_zeroed_state_db: %s is no longer zeroed (another " + "process quarantined it and a fresh DB was created)", path) return None try: ts = time.strftime("%Y%m%d-%H%M%S") @@ -346,68 +263,44 @@ def quarantine_zeroed_state_db(path: Path, *, already_locked: bool = False) -> O for suffix in ("-wal", "-shm"): side = Path(str(path) + suffix) if side.exists(): - try: + with contextlib.suppress(OSError): side.rename(Path(str(dest) + suffix)) - except OSError: - pass return dest if already_locked: return _do_quarantine() with quarantine_cross_process_lock(path) as acquired: if not acquired: - logger.error( - "quarantine lock for %s not acquired within 5s — refusing to " - "quarantine without the cross-process lock. The zeroed file " - "is left in place. If sessions fail to load, restore from " - "state-snapshots via `hermes snapshot list` / " - "`hermes snapshot restore `.", - path, - ) + logger.error("quarantine lock for %s not acquired within 5s — refusing to " + "quarantine without the cross-process lock. The zeroed file " + "is left in place. If sessions fail to load, restore from " + "state-snapshots via `hermes snapshot list` / `hermes snapshot restore `.", + path) return None return _do_quarantine() def collect_state_db_stats(db_path: Path) -> Dict[str, Any]: - """Best-effort, strictly read-only stats snapshot of a state.db file. - - Opens the database with ``mode=ro`` (URI) and a short timeout so it can - run against a *live* database held by a gateway without ever taking a - write lock or mutating the file. Every field is collected independently: - a failed pragma/SELECT yields ``None`` for that field, and the helper - itself never raises. Deliberately does NOT instantiate :class:`SessionDB` - — its constructor runs schema DDL, which a diagnostics probe must never do. - - Returned keys (all present, any may be None on failure): ``page_count``, - ``page_size``, ``freelist_count``, ``logical_size_bytes`` (page_count * - page_size), ``wal_size_bytes`` (stat of ``-wal``, 0 when absent), - ``journal_mode``, ``messages`` / ``sessions`` row counts, ``fts_tables`` - ({name: present}), ``fts_storage_version`` (None = legacy inline layout), - ``fts_rebuild_pending`` (deferred backfill unfinished), - ``fts_rebuild_high_water`` / ``fts_rebuild_progress`` raw ints, and - ``fts_rebuild_deferral`` (durable blocked-repair diagnostic). - """ + """Best-effort, strictly read-only stats snapshot of a state.db file: ``mode=ro`` with a short + timeout so it can run against a *live* database without taking a write lock. Every field is + collected independently (a failed pragma/SELECT yields ``None`` for it); never raises. + Deliberately does NOT instantiate :class:`SessionDB` — its constructor runs DDL. + ``wal_size_bytes`` is 0 when the sidecar is absent; ``fts_storage_version`` None means the + legacy inline layout; ``fts_rebuild_deferral`` is the durable blocked-repair diagnostic.""" from hermes_state import _connect_tracked_db stats: Dict[str, Any] = dict.fromkeys(( - "page_count", "page_size", "freelist_count", "logical_size_bytes", "wal_size_bytes", - "journal_mode", "messages", "sessions", "fts_tables", "fts_storage_version", - "fts_rebuild_pending", "fts_rebuild_high_water", "fts_rebuild_progress", - "fts_rebuild_deferral", - )) + "page_count", "page_size", "freelist_count", "logical_size_bytes", "wal_size_bytes", "journal_mode", + "messages", "sessions", "fts_tables", "fts_storage_version", "fts_rebuild_pending", + "fts_rebuild_high_water", "fts_rebuild_progress", "fts_rebuild_deferral")) # WAL sidecar size needs no connection at all. - try: + with contextlib.suppress(OSError): wal_path = Path(str(db_path) + "-wal") stats["wal_size_bytes"] = wal_path.stat().st_size if wal_path.exists() else 0 - except OSError: - pass try: - # mode=ro refuses to create the file and refuses every write; a short - # timeout keeps doctor snappy when a writer holds the lock. The tracked - # connect lets byte-probe helpers see this connection and refuse raw - # opens that could cancel our POSIX locks mid-read. - conn = _connect_tracked_db( - f"file:{Path(db_path)}?mode=ro", tracking_path=Path(db_path), uri=True, timeout=2.0 - ) + # A short timeout keeps doctor snappy when a writer holds the lock. The tracked connect + # lets byte-probe helpers see this connection and refuse raw opens that would cancel locks. + conn = _connect_tracked_db(f"file:{Path(db_path)}?mode=ro", tracking_path=Path(db_path), + uri=True, timeout=2.0) except Exception as exc: logger.debug("collect_state_db_stats: cannot open %s read-only: %s", db_path, exc) return stats @@ -419,73 +312,54 @@ def collect_state_db_stats(db_path: Path) -> Dict[str, Any]: except Exception: return None - def _int(value) -> Optional[int]: + def _int(sql: str, params=()) -> Optional[int]: + value = _scalar(sql, params) return int(value) if value is not None else None def _meta_int(key: str) -> Optional[int]: - try: - return _int(_scalar("SELECT value FROM state_meta WHERE key = ?", (key,))) + try: # a non-numeric meta value must yield None, not fail the snapshot + return _int("SELECT value FROM state_meta WHERE key = ?", (key,)) except Exception: return None try: - stats["page_count"] = _int(_scalar("PRAGMA page_count")) - stats["page_size"] = _int(_scalar("PRAGMA page_size")) + stats["page_count"] = _int("PRAGMA page_count") + stats["page_size"] = _int("PRAGMA page_size") if stats["page_count"] is not None and stats["page_size"] is not None: stats["logical_size_bytes"] = stats["page_count"] * stats["page_size"] - stats["freelist_count"] = _int(_scalar("PRAGMA freelist_count")) + stats["freelist_count"] = _int("PRAGMA freelist_count") jm = _scalar("PRAGMA journal_mode") stats["journal_mode"] = str(jm) if jm is not None else None - stats["messages"] = _int(_scalar("SELECT COUNT(*) FROM messages")) - stats["sessions"] = _int(_scalar("SELECT COUNT(*) FROM sessions")) + stats["messages"] = _int("SELECT COUNT(*) FROM messages") + stats["sessions"] = _int("SELECT COUNT(*) FROM sessions") # FTS table presence via sqlite_master (never SELECTs from the # virtual tables themselves — a corrupt index must not fail stats). - try: - names = { - row[0] - for row in conn.execute( - "SELECT name FROM sqlite_master WHERE type = 'table' " - "AND name IN (?, ?, ?)", - _FTS_TABLE_NAMES, - ).fetchall() - } + with contextlib.suppress(Exception): + names = {row[0] for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name IN (?, ?, ?)", + _FTS_TABLE_NAMES).fetchall()} stats["fts_tables"] = {t: (t in names) for t in _FTS_TABLE_NAMES} - except Exception: - pass # Raw state_meta reads — cheap, and independent of SessionDB. stats["fts_storage_version"] = _meta_int("fts_storage_version") - high_water = _meta_int("fts_rebuild_high_water") - progress = _meta_int("fts_rebuild_progress") - stats["fts_rebuild_high_water"] = high_water - stats["fts_rebuild_progress"] = progress + stats["fts_rebuild_high_water"] = high_water = _meta_int("fts_rebuild_high_water") + stats["fts_rebuild_progress"] = progress = _meta_int("fts_rebuild_progress") stats["fts_rebuild_pending"] = False if high_water is None else (progress or 0) < high_water - try: - row = conn.execute( - "SELECT value FROM state_meta WHERE key = ? LIMIT 1", (FTS_REBUILD_DEFERRAL_KEY,) - ).fetchone() - if row: - parsed = json.loads(row[0]) - if isinstance(parsed, dict): - stats["fts_rebuild_deferral"] = parsed - except Exception: - pass + with contextlib.suppress(Exception): + row = conn.execute("SELECT value FROM state_meta WHERE key = ? LIMIT 1", + (FTS_REBUILD_DEFERRAL_KEY,)).fetchone() + parsed = json.loads(row[0]) if row else None + if isinstance(parsed, dict): + stats["fts_rebuild_deferral"] = parsed finally: - try: + with contextlib.suppress(Exception): conn.close() - except Exception: - pass return stats def count_db_holders(db_path: Path) -> Optional[int]: - """Best-effort count of processes holding ``db_path`` open (Linux only). - - Scans ``/proc/*/fd`` symlinks for the resolved database path. Returns - the number of distinct PIDs with the file open, or ``None`` on any - error or on non-Linux platforms. Never raises; no lsof dependency. - Unreadable per-process fd dirs (other users' processes without root) - are silently skipped, so the count is a lower bound. - """ + """Best-effort count of distinct PIDs holding ``db_path`` open (``/proc/*/fd`` scan); ``None`` + on any error or non-Linux host, never raises. Unreadable fd dirs (other users' processes + without root) are skipped, so this is a lower bound.""" try: if not sys.platform.startswith("linux"): return None @@ -495,40 +369,27 @@ def count_db_holders(db_path: Path) -> Optional[int]: return None -def _is_inactive_orphan_desktop_holder( - *, ppid: int, age_seconds: float, min_age_seconds: float, ephemeral_backend: bool, - connection_statuses: List[str], -) -> bool: +def _is_inactive_orphan_desktop_holder(*, ppid: int, age_seconds: float, min_age_seconds: float, + ephemeral_backend: bool, connection_statuses: List[str]) -> bool: """Pure safety predicate for the narrow Desktop holder reap.""" - return ( - ppid in (0, 1) - and age_seconds >= min_age_seconds - and ephemeral_backend - and "ESTABLISHED" not in connection_statuses - ) + return (ppid in (0, 1) and age_seconds >= min_age_seconds and ephemeral_backend + and "ESTABLISHED" not in connection_statuses) def _concrete_state_db_holder_pids(db_path: Path, holders: List[Tuple[int, str]]) -> List[int]: """Return unique PIDs proven to hold this DB or one of its sidecars.""" canonical_db = os.path.normcase(os.path.abspath(os.fspath(db_path))) watched = {canonical_db, canonical_db + "-wal", canonical_db + "-shm"} - pids: List[int] = [] - for pid, path in holders: - if pid <= 0 or pid in pids or _canonical_sqlite_path(path) not in watched: - continue - pids.append(pid) - return pids + return list(dict.fromkeys( + pid for pid, path in holders if pid > 0 and _canonical_sqlite_path(path) in watched)) def _read_proc_cmdline(pid: int) -> Optional[str]: - """Read /proc//cmdline (world-readable even when the fd table is not) - as a space-joined string; None when unreadable (exited, hidepid mount).""" + """Space-joined /proc//cmdline (readable even when the fd table is not); None if unreadable.""" try: with open(f"/proc/{pid}/cmdline", "rb") as f: raw = f.read() - if not raw: - return None - return raw.replace(b"\x00", b" ").decode("utf-8", "replace").strip() + return raw.replace(b"\x00", b" ").decode("utf-8", "replace").strip() if raw else None except OSError: return None @@ -538,8 +399,6 @@ _HERMES_CMDLINE_MARKERS = ("hermes_cli.main", "hermes_cli/main", "hermes serve", def _looks_like_hermes(cmdline: str) -> bool: - """Heuristic: does this cmdline look like a Hermes process? Decides whether - an uninspectable process (fd table unreadable, different user) is treated - as a potential state.db holder; system daemons are not flagged.""" - lower = cmdline.lower() - return any(marker in lower for marker in _HERMES_CMDLINE_MARKERS) + """Heuristic: is this a Hermes process? Decides whether an uninspectable process (fd + table unreadable, other user) counts as a potential state.db holder; daemons are not.""" + return any(marker in cmdline.lower() for marker in _HERMES_CMDLINE_MARKERS) diff --git a/hermes_state_maintenance.py b/hermes_state_maintenance.py index e6d1d38202..e53510f293 100644 --- a/hermes_state_maintenance.py +++ b/hermes_state_maintenance.py @@ -1,5 +1,4 @@ -"""Retention pruning, stale-session archiving and VACUUM policy mixin for -SessionDB.""" +"""Retention pruning, stale-session archiving and VACUUM policy mixin for SessionDB.""" from __future__ import annotations @@ -9,9 +8,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from hermes_state_common import ( - AUTO_VACUUM_MIN_FREELIST_RATIO, - _sql_session_last_active, - escape_like as _escape_like, + AUTO_VACUUM_MIN_FREELIST_RATIO, _sql_session_last_active, escape_like as _escape_like ) # caplog tests pin the "hermes_state" logger name. @@ -40,12 +37,22 @@ def _one(clause: str, conv=None): return lambda v: ([clause], [conv(v) if conv else v]) -# Prune/archive filters in evaluation order: (kwarg, applies-when, builder). -# ``applies-when`` is "notnone" (numeric/time bounds; 0 is a real bound) or -# "truthy" (strings; "" means unset). Builders return (clauses, params). +def _placeholders(n: int) -> str: + return ",".join("?" * n) + + +def _seconds_since(now: float, raw) -> Optional[float]: + """Age of a state_meta timestamp; None when unset or corrupt (= no prior run).""" + try: + return now - float(raw) if raw else None + except (TypeError, ValueError): + return None + + +# Prune/archive filters in evaluation order: (kwarg, applies-when, builder -> (clauses, params)). +# ``applies-when``: "notnone" (numeric/time bounds; 0 is a real bound) or "truthy" ("" = unset). _PRUNE_FILTERS = ( - # Orphan-swept rows age from the sweep, not their old activity, or the - # next prune pass deletes them before the user can recover. + # Orphan-swept rows age from the sweep, not old activity, or the next prune deletes them before recovery. ("last_active_before", "notnone", lambda v: ( [_LAST_ACTIVE_SQL + " < ?", "(COALESCE(s.end_reason, '') != 'startup_orphan_reap' OR s.ended_at < ?)"], @@ -81,9 +88,8 @@ class SessionMaintenanceMixin: def prune_empty_ghost_sessions(self, sessions_dir: "Optional[Path]" = None) -> int: """Remove empty TUI ghost sessions (no messages, no title, >24hr old).""" cutoff = time.time() - 86400 - def _do(conn): - rows = conn.execute(""" + ids = [r[0] for r in conn.execute(""" SELECT id FROM sessions WHERE source = 'tui' AND title IS NULL @@ -92,136 +98,95 @@ class SessionMaintenanceMixin: AND NOT EXISTS ( SELECT 1 FROM messages WHERE messages.session_id = sessions.id ) - """, (cutoff,)).fetchall() - ids = [r[0] for r in rows] + """, (cutoff,)).fetchall()] if ids: - placeholders = ",".join("?" * len(ids)) - conn.execute(f"DELETE FROM sessions WHERE id IN ({placeholders})", ids) + conn.execute(f"DELETE FROM sessions WHERE id IN ({_placeholders(len(ids))})", ids) self._delete_unreferenced_system_prompts(conn) return ids - removed_ids = self._execute_write(_do) or [] - if sessions_dir and removed_ids: - for sid in removed_ids: - self._remove_session_files(sessions_dir, sid) + for sid in removed_ids if sessions_dir else (): + self._remove_session_files(sessions_dir, sid) return len(removed_ids) + def _write_guards_reject(self, conn, sid: str, **kwargs) -> bool: + """True when a live turn lease / compression lock protects ``sid``; expired or + dead-holder guards are reclaimed and fenced as a side effect.""" + from hermes_state import SessionCompressionInProgressError, SessionTurnLeaseLostError + try: + self._check_transcript_write_guards( + conn, sid, compression_lock_holder=None, turn_lease_holder=None, + reject_active_turn_lease=True, reject_active_compression_lock=True, **kwargs) + except (SessionCompressionInProgressError, SessionTurnLeaseLostError): + return True + return False + def sweep_orphaned_sessions( - self, *, max_idle_seconds: float, - sources: Tuple[str, ...] = ("tui", "desktop", "subagent"), + self, *, max_idle_seconds: float, sources: Tuple[str, ...] = ("tui", "desktop", "subagent"), exclude_ids: Tuple[str, ...] = (), exclude_pinned: bool = False, heartbeat_staleness_seconds: Optional[float] = None, - heartbeat_ownership_grace_seconds: Optional[float] = None, - respect_gateway_heartbeats: bool = True, + heartbeat_ownership_grace_seconds: Optional[float] = None, respect_gateway_heartbeats: bool = True, ) -> List[str]: - """Close session rows orphaned by a dead gateway process. + """Close session rows orphaned by a dead gateway process (its in-process disconnect grace + timer died with it, leaving ``ended_at IS NULL`` forever). Rows of ``sources`` whose + ``started_at`` AND canonical last activity are both older than ``max_idle_seconds`` get + ``end_reason='startup_orphan_reap'`` (the ``started_at`` predicate protects fresh + compression/branch children whose copied activity is old). Only pass sources whose + lifecycle the caller owns — never messaging platforms like ``telegram`` (ending those + triggers a routing loop). ``exclude_ids`` spares rows this process still holds. + Non-destructive: messages kept, row resumable, first-reason-wins. - The TUI/desktop gateway reaps disconnected sessions with an in-process - grace timer; a restart destroys the timer and leaves ``ended_at IS - NULL`` forever. Closes rows for ``sources`` whose ``started_at`` AND - canonical last activity are both older than ``max_idle_seconds`` with - ``end_reason='startup_orphan_reap'`` (the separate ``started_at`` - predicate protects fresh compression/branch children whose copied - activity is old). Only pass sources whose lifecycle the caller owns — - never messaging platforms like ``telegram`` (ending those triggers a - routing loop). ``exclude_ids`` spares rows this process still holds in - memory. Non-destructive: messages are kept and the row stays - resumable; first-reason-wins via ``ended_at IS NULL``. - - Cross-backend liveness: with ``respect_gateway_heartbeats``, a row is - reaped only when stale AND no live backend (heartbeat within - ``heartbeat_staleness_seconds``, default ``2 * max_idle_seconds``) could - own it, where backend B owns session S if ``B.started_at <= S.started_at - + heartbeat_ownership_grace_seconds`` (default = staleness). The grace - covers a migrating backend whose sessions predate its first heartbeat - but is bounded so a PID-reuse respawn cannot protect rows forever. - Disable the gate only for sources owned by state.db itself. - - SELECT, live-lease validation and UPDATE run in one ``BEGIN IMMEDIATE`` - transaction; active turn leases / compression locks spare the row, and - expired guards are removed so their former owner is fenced. + With ``respect_gateway_heartbeats`` a row is reaped only when no live backend (heartbeat + within ``heartbeat_staleness_seconds``, default ``2 * max_idle_seconds``) could own it: B + owns S if ``B.started_at <= S.started_at + grace`` (default = staleness) — grace covers a + migrating backend whose sessions predate its first heartbeat, bounded so a PID-reuse + respawn cannot protect rows forever. Disable the gate only for state.db-owned sources. + SELECT, live-lease validation and UPDATE run in one ``BEGIN IMMEDIATE`` transaction; + active leases/locks spare the row, expired guards are removed so their owner is fenced. """ - from hermes_state import SessionCompressionInProgressError, SessionTurnLeaseLostError srcs = tuple(s for s in sources if s) if max_idle_seconds <= 0 or not srcs: return [] - hb_staleness = ( - heartbeat_staleness_seconds - if heartbeat_staleness_seconds and heartbeat_staleness_seconds > 0 - else max_idle_seconds * 2 - ) - hb_grace = ( - heartbeat_ownership_grace_seconds - if heartbeat_ownership_grace_seconds is not None and heartbeat_ownership_grace_seconds >= 0 - else hb_staleness - ) - now = time.time() - cutoff = now - max_idle_seconds - placeholders = ",".join("?" for _ in srcs) + hb_staleness, hb_grace = heartbeat_staleness_seconds, heartbeat_ownership_grace_seconds + if not (hb_staleness and hb_staleness > 0): + hb_staleness = max_idle_seconds * 2 + if not (hb_grace is not None and hb_grace >= 0): + hb_grace = hb_staleness + cutoff = (now := time.time()) - max_idle_seconds pin_scope = " AND COALESCE(pinned, 0) = 0" if exclude_pinned else "" orphan_predicate = f"started_at < ? AND {_sql_session_last_active('sessions')} < ?" heartbeat_params: Tuple[float, ...] = () if respect_gateway_heartbeats: - orphan_predicate += ( - " AND NOT EXISTS (" - "SELECT 1 FROM gateway_heartbeats h" - " WHERE h.last_heartbeat >= ?" - " AND h.started_at <= sessions.started_at + ?" - ")" - ) + orphan_predicate += (" AND NOT EXISTS (SELECT 1 FROM gateway_heartbeats h WHERE" + " h.last_heartbeat >= ? AND h.started_at <= sessions.started_at + ?)") heartbeat_params = (now - hb_staleness, hb_grace) - scope_sql = f" AND source IN ({placeholders}){pin_scope} AND {orphan_predicate}" + scope_sql = f" AND source IN ({_placeholders(len(srcs))}){pin_scope} AND {orphan_predicate}" scope_params = (*srcs, cutoff, cutoff, *heartbeat_params) - def _do(conn): - rows = conn.execute( - f"SELECT id FROM sessions WHERE ended_at IS NULL{scope_sql}", scope_params - ).fetchall() + rows = conn.execute(f"SELECT id FROM sessions WHERE ended_at IS NULL{scope_sql}", + scope_params).fetchall() excluded = {str(x) for x in exclude_ids if x} - victims = [] - for row in rows: - sid = str(row["id"]) - if sid in excluded: - continue - try: - self._check_transcript_write_guards( - conn, sid, compression_lock_holder=None, turn_lease_holder=None, - reject_active_turn_lease=True, reject_active_compression_lock=True, - ) - except (SessionCompressionInProgressError, SessionTurnLeaseLostError): - continue - victims.append(sid) + victims = [sid for sid in (str(row["id"]) for row in rows) + if sid not in excluded and not self._write_guards_reject(conn, sid)] if not victims: return [] - marks = ",".join("?" for _ in victims) # Re-apply every predicate under the write lock. conn.execute( f"UPDATE sessions SET ended_at = ?, end_reason = 'startup_orphan_reap'" - f" WHERE id IN ({marks}) AND ended_at IS NULL{scope_sql}", - (time.time(), *victims, *scope_params), - ) + f" WHERE id IN ({_placeholders(len(victims))}) AND ended_at IS NULL{scope_sql}", + (time.time(), *victims, *scope_params)) return victims - return self._execute_write(_do) or [] @staticmethod - def _prune_filter_where( - *, archived: Optional[bool] = None, include_pinned: bool = False, **filters - ) -> Tuple[str, list]: - """Shared WHERE clause for bulk prune/archive selection (alias ``s``). - - Filters (see ``_PRUNE_FILTERS``) AND together; only ended sessions are - ever candidates. ``archived`` is tri-state (None = both). ``*_like`` - filters are case-insensitive substrings; the rest are exact (provider - case-insensitive). Token bounds use input+output; cost bounds use - ``COALESCE(actual_cost_usd, estimated_cost_usd)``. - """ + def _prune_filter_where(*, archived: Optional[bool] = None, include_pinned: bool = False, + **filters) -> Tuple[str, list]: + """Shared WHERE clause for bulk prune/archive selection (alias ``s``): ``_PRUNE_FILTERS`` + AND together, only ended sessions are ever candidates, ``archived`` is tri-state + (None = both), ``*_like`` are case-insensitive substrings, the rest exact.""" unknown = set(filters) - _PRUNE_FILTER_NAMES if unknown: - raise TypeError( - "SessionMaintenanceMixin._prune_filter_where() got an unexpected " - f"keyword argument {sorted(unknown)[0]!r}" - ) + raise TypeError("SessionMaintenanceMixin._prune_filter_where() got an unexpected " + f"keyword argument {sorted(unknown)[0]!r}") clauses = ["s.ended_at IS NOT NULL"] params: list = [] for name, applies, build in _PRUNE_FILTERS: @@ -230,38 +195,26 @@ class SessionMaintenanceMixin: new_clauses, new_params = build(value) clauses.extend(new_clauses) params.extend(new_params) - if archived is True: - clauses.append("s.archived = 1") - elif archived is False: - clauses.append("s.archived = 0") - # Pinned is a durable "keep" flag: bulk prune/delete/archive exclude - # pinned rows unless the caller explicitly opts in. + if isinstance(archived, bool): + clauses.append(f"s.archived = {int(archived)}") + # Pinned is a durable "keep" flag: bulk prune/delete/archive exclude pinned rows unless opted in. if not include_pinned: clauses.append("COALESCE(s.pinned, 0) = 0") return " AND ".join(clauses), params - @staticmethod - def _apply_prune_age_filter(older_than_days: Optional[float], filters: Dict[str, Any]) -> None: - """Translate the legacy age window into the shared activity filter.""" - if ( - filters.get("last_active_before") is None - and filters.get("started_before") is None - and older_than_days is not None - ): - filters["last_active_before"] = time.time() - (older_than_days * 86400) - def _prune_where(self, older_than_days, source, filters) -> Tuple[str, list]: - self._apply_prune_age_filter(older_than_days, filters) + """Translate the legacy age window into the shared activity filter, then build WHERE.""" + if (older_than_days is not None and filters.get("last_active_before") is None + and filters.get("started_before") is None): + filters["last_active_before"] = time.time() - (older_than_days * 86400) return self._prune_filter_where(source=source, **filters) - def list_prune_candidates( - self, older_than_days: Optional[float] = None, source: str = None, **filters - ) -> List[Dict[str, Any]]: - """Sessions a matching prune/archive would touch (dry-run), oldest - first. Same filters as :meth:`_prune_filter_where`; ``older_than_days`` - is an inactivity threshold (latest message, else ``started_at``).""" + def list_prune_candidates(self, older_than_days: Optional[float] = None, source: str = None, + **filters) -> List[Dict[str, Any]]: + """Dry-run: sessions a matching prune/archive would touch, oldest first (``older_than_days`` + = inactivity threshold: latest message, else ``started_at``).""" where, params = self._prune_where(older_than_days, source, filters) - rows = self._read_all( + return [dict(row) for row in self._read_all( f"""SELECT s.id, s.source, s.title, s.model, s.started_at, COALESCE( (SELECT MAX(m.timestamp) FROM messages m @@ -270,25 +223,17 @@ class SessionMaintenanceMixin: ) AS last_active, s.ended_at, s.message_count, s.archived FROM sessions s WHERE {where} - ORDER BY last_active ASC, s.started_at ASC""", - params, - ) - return [dict(row) for row in rows] + ORDER BY last_active ASC, s.started_at ASC""", params)] - def count_prune_matches( - self, older_than_days: Optional[float] = None, source: str = None, **filters - ) -> int: - """Count-only variant of :meth:`list_prune_candidates` (the CLI uses it - to report how many pinned sessions are spared).""" + def count_prune_matches(self, older_than_days: Optional[float] = None, source: str = None, + **filters) -> int: + """Count-only :meth:`list_prune_candidates` (CLI reports spared pinned sessions).""" where, params = self._prune_where(older_than_days, source, filters) return int(self._read_one(f"SELECT COUNT(*) FROM sessions s WHERE {where}", params)[0]) - def count_open_prune_matches( - self, older_than_days: Optional[float] = None, source: str = None, **filters - ) -> int: - """Count open sessions a matching prune skips: every normal filter with - only the ``ended_at`` guard inverted. Visibility-only; live sessions - never become prune-eligible.""" + def count_open_prune_matches(self, older_than_days: Optional[float] = None, source: str = None, + **filters) -> int: + """Count open sessions a matching prune skips (``ended_at`` guard inverted); visibility-only.""" where, params = self._prune_where(older_than_days, source, filters) ended_guard = "s.ended_at IS NOT NULL" if not where.startswith(ended_guard): @@ -297,16 +242,11 @@ class SessionMaintenanceMixin: return int(self._read_one(f"SELECT COUNT(*) FROM sessions s WHERE {open_where}", params)[0]) def archive_stale_sessions(self, idle_days: float, *, exclude_pinned: bool = True) -> int: - """Archive every session untouched for ``idle_days`` (real recency: - freshest of ``last_activity_at`` / latest message / ``started_at``). - Unlike :meth:`archive_sessions`, this can archive unended sessions. - - Guards: ``pinned = 0`` when ``exclude_pinned``; ``archived = 0`` so - repeats are no-ops; only lineage tips (``end_reason <> 'compression'``) - are candidates — a stale tip archives its chain via - :meth:`set_session_archived`, so an old compressed-away root with a - recent continuation is never matched. Returns the count archived. - """ + """Archive every session untouched for ``idle_days`` (freshest of ``last_activity_at`` / + latest message / ``started_at``); may archive unended sessions. ``archived = 0`` makes + repeats no-ops; only lineage tips (``end_reason <> 'compression'``) are candidates — a + stale tip archives its chain via :meth:`set_session_archived`, so an old compressed-away + root with a recent continuation is never matched.""" if idle_days is None or idle_days < 0: return 0 cutoff = time.time() - float(idle_days) * 86400.0 @@ -319,143 +259,91 @@ class SessionMaintenanceMixin: {pin_clause} AND {_sql_session_last_active("s")} < ? ORDER BY s.started_at ASC - """, - (cutoff,), - ) - ids = [r[0] for r in rows] - for sid in ids: - self.set_session_archived(sid, True) - return len(ids) + """, (cutoff,)) + for row in rows: + self.set_session_archived(row[0], True) + return len(rows) - def prune_sessions( - self, older_than_days: Optional[float] = 90, source: str = None, - sessions_dir: Optional[Path] = None, exclude_active_write_guards: bool = False, - **filters, - ) -> int: - """Delete ended sessions matching the filters; returns the count. - - Default: inactive for ``older_than_days`` (latest message, else - ``started_at``), optionally by ``source``. Extra keyword filters are - those of :meth:`_prune_filter_where`; an explicit ``started_before`` / - ``last_active_before`` overrides the ``older_than_days`` cutoff - (pass ``older_than_days=None`` for no implicit age bound). - - Children outside the window are orphaned (parent NULLed), not cascade- - deleted. With *sessions_dir*, on-disk transcript files are removed - outside the DB transaction. ``exclude_active_write_guards`` (automatic - maintenance) skips rows under a live turn lease or compression lock, - while expired/dead holders are reclaimed and fenced in the same write. - """ - from hermes_state import SessionCompressionInProgressError, SessionTurnLeaseLostError + def prune_sessions(self, older_than_days: Optional[float] = 90, source: str = None, + sessions_dir: Optional[Path] = None, exclude_active_write_guards: bool = False, + **filters) -> int: + """Delete ended sessions inactive for ``older_than_days`` (an explicit ``started_before`` / + ``last_active_before`` overrides it; None = no implicit bound) matching the filters. + Children outside the window are orphaned (parent NULLed), not cascade-deleted. With + *sessions_dir*, transcript files are removed outside the DB transaction. + ``exclude_active_write_guards`` (automatic maintenance) skips rows under a live turn lease + or compression lock while expired/dead holders are reclaimed and fenced.""" where, where_params = self._prune_where(older_than_days, source, filters) removed_ids: list[str] = [] - def _do(conn): cursor = conn.execute(f"SELECT s.id FROM sessions s WHERE {where}", where_params) session_ids = {row["id"] for row in cursor.fetchall()} if exclude_active_write_guards: - protected = set() - for sid in session_ids: - try: - self._check_transcript_write_guards( - conn, sid, compression_lock_holder=None, turn_lease_holder=None, - reject_active_turn_lease=True, reject_active_compression_lock=True, - allow_closed_compression_parent=True, - ) - except (SessionCompressionInProgressError, SessionTurnLeaseLostError): - protected.add(sid) - session_ids.difference_update(protected) + session_ids -= {sid for sid in session_ids + if self._write_guards_reject(conn, sid, allow_closed_compression_parent=True)} if not session_ids: return 0 - placeholders = ",".join("?" * len(session_ids)) - conn.execute( - f"UPDATE sessions SET parent_session_id = NULL " - f"WHERE parent_session_id IN ({placeholders})", - list(session_ids), - ) + conn.execute(f"UPDATE sessions SET parent_session_id = NULL " + f"WHERE parent_session_id IN ({_placeholders(len(session_ids))})", list(session_ids)) for sid in session_ids: conn.execute("DELETE FROM messages WHERE session_id = ?", (sid,)) conn.execute("DELETE FROM sessions WHERE id = ?", (sid,)) removed_ids.append(sid) self._delete_unreferenced_system_prompts(conn) return len(session_ids) - count = self._execute_write(_do) for sid in removed_ids: self._remove_session_files(sessions_dir, sid) return count - def _page_pragmas(self, *names: str) -> Optional[list]: - """Read integer PRAGMAs over the existing connection (never a byte probe - of the live file); None if the connection is closed or a pragma fails.""" - with self._read_ctx() as conn: - if self._conn is None: - return None - return [conn.execute(f"PRAGMA {name}").fetchone()[0] for name in names] + def _page_pragmas(self, names: Tuple[str, ...], fail_msg: str) -> Optional[list]: + """Integer PRAGMAs over the existing connection (never a byte probe); None + debug log on failure.""" + try: + with self._read_ctx() as conn: + if self._conn is None: + return None + return [int(conn.execute(f"PRAGMA {name}").fetchone()[0]) for name in names] + except Exception as exc: + logger.debug(fail_msg, exc) + return None def logical_size_bytes(self) -> Optional[int]: - """``page_count * page_size``: the main-file size once the WAL is - checkpointed back in. Prefer over ``os.path.getsize`` when reporting a - VACUUM: in WAL mode the rewrite lands in ``-wal`` and the checkpoint is - refused while another connection holds a read-mark, so a stat() delta - understates the win and can go negative. None if pragmas fail. - """ - try: - values = self._page_pragmas("page_count", "page_size") - if values is None: - return None - page_count, page_size = values - return int(page_count) * int(page_size) - except Exception as exc: - logger.debug("Could not read logical DB size: %s", exc) - return None + """``page_count * page_size``: main-file size once the WAL is checkpointed in. Prefer + over ``os.path.getsize`` when reporting a VACUUM: in WAL mode the rewrite lands in + ``-wal`` and the checkpoint is refused while another connection holds a read-mark, so + a stat() delta understates the win and can go negative.""" + values = self._page_pragmas(("page_count", "page_size"), "Could not read logical DB size: %s") + return None if values is None else values[0] * values[1] def _freelist_ratio(self) -> Optional[float]: - """Reclaimable fraction (``freelist_count / page_count``); gates VACUUM - in :meth:`maybe_auto_prune_and_vacuum`. None if pragmas fail (callers - then fall back to the time throttle alone).""" + """Reclaimable fraction (``freelist_count / page_count``) gating VACUUM in + :meth:`maybe_auto_prune_and_vacuum`; None = fall back to the time throttle.""" + values = self._page_pragmas(("page_count", "freelist_count"), "Could not read freelist ratio: %s") + return None if values is None else (values[1] / values[0] if values[0] > 0 else 0.0) + + def _try_checkpoint(self, mode: str, fail_msg: str) -> None: try: - values = self._page_pragmas("page_count", "freelist_count") - if values is None: - return None - page_count, freelist = int(values[0]), int(values[1]) - if page_count <= 0: - return 0.0 - return freelist / page_count + self._conn.execute(f"PRAGMA wal_checkpoint({mode})") except Exception as exc: - logger.debug("Could not read freelist ratio: %s", exc) - return None + logger.debug(fail_msg, exc) def vacuum(self) -> int: - """VACUUM to reclaim space after large deletes (SQLite never shrinks - the file on its own). - - Rewrites the whole DB, cannot run inside a transaction, and takes an - exclusive lock — callers must ensure no other writers are active (safe - at startup before serving traffic). FTS5 segments are merged first via - :meth:`optimize_fts` so the VACUUM reclaims those pages too. Returns - the number of FTS indexes optimized (0 on merge failure / no FTS). - """ + """VACUUM to reclaim space after large deletes (SQLite never shrinks on its own). + Takes an exclusive lock — callers must ensure no other writers are active. FTS5 + segments are merged first (:meth:`optimize_fts`) so their pages are reclaimed too; + returns the number of FTS indexes optimized (0 on merge failure / no FTS).""" optimized = 0 try: optimized = self.optimize_fts() # manages its own lock except Exception as exc: logger.warning("FTS optimize before VACUUM failed: %s", exc) with self._lock: - # PASSIVE, not TRUNCATE: a manual `hermes sessions vacuum` runs in - # a transient CLI process, and a TRUNCATE reset here would race a - # live gateway writer and tear B-tree pages. - try: - self._conn.execute("PRAGMA wal_checkpoint(PASSIVE)") - except Exception as exc: - logger.debug("WAL checkpoint (PASSIVE) before VACUUM failed: %s", exc) + # PASSIVE, not TRUNCATE: a manual `hermes sessions vacuum` runs in a transient CLI + # process; a TRUNCATE reset here would race a live gateway writer. + self._try_checkpoint("PASSIVE", "WAL checkpoint (PASSIVE) before VACUUM failed: %s") self._conn.execute("VACUUM") - # VACUUM rewrites every page THROUGH the WAL; without this TRUNCATE - # a 3 GB database leaves a 3 GB -wal behind. - try: - self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") - except Exception as exc: - logger.debug("WAL checkpoint (TRUNCATE) after VACUUM failed: %s", exc) + # VACUUM rewrites every page THROUGH the WAL; without this TRUNCATE a 3 GB DB leaves a 3 GB -wal. + self._try_checkpoint("TRUNCATE", "WAL checkpoint (TRUNCATE) after VACUUM failed: %s") # TRUNCATE may replace the WAL inode; adopt the new sidecars so the # write-path generation guard does not halt this connection. self._record_db_file_identity() @@ -466,26 +354,17 @@ class SessionMaintenanceMixin: sessions_dir: Optional[Path] = None, min_vacuum_interval_days: int = 30, min_vacuum_freelist_ratio: float = AUTO_VACUUM_MIN_FREELIST_RATIO, ) -> Dict[str, Any]: - """Idempotent startup auto-maintenance: prune inactive sessions, - reap stale open state-owned rows, optional VACUUM. Never raises. - - Runs at most once per ``min_interval_hours`` (state_meta). VACUUM has - its own ``min_vacuum_interval_days`` throttle and additionally requires - ``freelist_count / page_count`` > ``min_vacuum_freelist_ratio`` so a - small prune on a dense multi-GB database never triggers a full rewrite. - With *sessions_dir*, pruned transcripts are removed from disk too. - - Stale-open reconciliation: cron/kanban/subagent/one-shot CLI rows never - set ``ended_at`` when their process dies, and prune only deletes ended - rows. After pruning, open rows from :attr:`_AUTO_PRUNE_STALE_OPEN_SOURCES` - older than ``retention_days`` are closed (``startup_orphan_reap``); they - stay resumable and age from their close, so they get one more full - retention window. Messaging and UI sources are never touched. - - Returns ``{"skipped", "pruned", "closed", "vacuumed"}`` plus - ``"freelist_ratio"`` when a VACUUM was considered and ``"error"`` on - failure. - """ + """Idempotent startup auto-maintenance (never raises): prune inactive sessions, reap stale + open state-owned rows, optional VACUUM. Runs at most once per ``min_interval_hours``; + VACUUM has its own ``min_vacuum_interval_days`` throttle and also requires + ``freelist_count / page_count`` > ``min_vacuum_freelist_ratio`` so a small prune on a + dense multi-GB database never triggers a full rewrite. Stale-open reconciliation: + cron/kanban/subagent/one-shot CLI rows never set ``ended_at`` when their process dies and + prune only deletes ended rows, so after pruning, open rows from + :attr:`_AUTO_PRUNE_STALE_OPEN_SOURCES` older than ``retention_days`` are closed + (``startup_orphan_reap``); they stay resumable and age from their close. Returns + ``{"skipped", "pruned", "closed", "vacuumed"}`` plus ``"freelist_ratio"`` when a VACUUM + was considered and ``"error"`` on failure.""" from hermes_state import _release_auto_maintenance_lock, _try_acquire_auto_maintenance_lock result: Dict[str, Any] = {"skipped": False, "pruned": 0, "closed": 0, "vacuumed": False} maintenance_lock = _try_acquire_auto_maintenance_lock(self.db_path) @@ -493,22 +372,14 @@ class SessionMaintenanceMixin: result["skipped"] = True return result try: - last_raw = self.get_meta("last_auto_prune") now = time.time() - if last_raw: - try: - if now - float(last_raw) < min_interval_hours * 3600: - result["skipped"] = True - return result - except (TypeError, ValueError): - pass # corrupt meta; treat as no prior run - + since_prune = _seconds_since(now, self.get_meta("last_auto_prune")) + if since_prune is not None and since_prune < min_interval_hours * 3600: + result["skipped"] = True + return result # Prune first: orphans closed below get a full retention window. - pruned = self.prune_sessions( - older_than_days=retention_days, sessions_dir=sessions_dir, - exclude_active_write_guards=True, - ) - result["pruned"] = pruned + result["pruned"] = pruned = self.prune_sessions( + older_than_days=retention_days, sessions_dir=sessions_dir, exclude_active_write_guards=True) closed = self.sweep_orphaned_sessions( max_idle_seconds=float(retention_days) * 86400.0, sources=self._AUTO_PRUNE_STALE_OPEN_SOURCES, exclude_pinned=True, @@ -517,16 +388,10 @@ class SessionMaintenanceMixin: result["closed"] = len(closed) # VACUUM only if rows were freed, the time throttle passed AND the # freelist ratio passed — it holds an exclusive lock for a full rewrite. - last_vacuum_raw = self.get_meta("last_vacuum") - vacuum_due = True - if last_vacuum_raw: - try: - vacuum_due = (now - float(last_vacuum_raw)) >= min_vacuum_interval_days * 86400 - except (TypeError, ValueError): - vacuum_due = True + since_vacuum = _seconds_since(now, self.get_meta("last_vacuum")) + vacuum_due = since_vacuum is None or since_vacuum >= min_vacuum_interval_days * 86400 if vacuum and pruned > 0 and vacuum_due: - ratio = self._freelist_ratio() - result["freelist_ratio"] = ratio + result["freelist_ratio"] = ratio = self._freelist_ratio() if ratio is None or ratio > min_vacuum_freelist_ratio: try: self.vacuum() @@ -535,19 +400,15 @@ class SessionMaintenanceMixin: except Exception as exc: logger.warning("state.db VACUUM failed: %s", exc) else: - logger.debug( - "state.db auto-maintenance: skipping VACUUM, only " - "%.1f%% of pages reclaimable (threshold %.0f%%)", - ratio * 100.0, min_vacuum_freelist_ratio * 100.0, - ) + logger.debug("state.db auto-maintenance: skipping VACUUM, only " + "%.1f%% of pages reclaimable (threshold %.0f%%)", + ratio * 100.0, min_vacuum_freelist_ratio * 100.0) # Record even when pruned == 0 so the throttle holds. self.set_meta("last_auto_prune", str(now)) if closed or pruned > 0: - logger.info( - "state.db auto-maintenance: closed %d stale open session(s), " - "pruned %d session(s) inactive for %d days%s", - len(closed), pruned, retention_days, " + VACUUM" if result["vacuumed"] else "", - ) + logger.info("state.db auto-maintenance: closed %d stale open session(s), " + "pruned %d session(s) inactive for %d days%s", + len(closed), pruned, retention_days, " + VACUUM" if result["vacuumed"] else "") except Exception as exc: # Maintenance must never block startup. logger.warning("state.db auto-maintenance failed: %s", exc)