hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(state): compact SessionMaintenanceMixin and state.db file helpers (-280 LOC, SQL-parity neutral)

Browse Source 
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.
This commit is contained in:
Teknium
2026-09-02 19:32:41 -07:00
parent 113f04616b
commit 1064a3a935
2 changed files with 296 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files

387
hermes_state_dbfile.py
View File

@@ -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.<name>`` 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.<name>`` 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/<pid>/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 <id>`.",
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 <id>`.",
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 ``<db>-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/<pid>/cmdline (world-readable even when the fd table is not)
as a space-joined string; None when unreadable (exited, hidepid mount)."""
"""Space-joined /proc/<pid>/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)

485
hermes_state_maintenance.py
View File

@@ -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)