a2c0af430c
Three silent data-correctness bugs on a host where ONE process serves every profile: - The gateway constructor ran auto-archive and auto-prune/VACUUM once, on a handle pinned to the construction-time launch home, so a served secondary profile's state.db was never pruned or vacuumed by anybody. Both now run per SERVED profile, inside that profile's runtime scope, against its own SessionDB, its own `sessions:` config and its own transcript dir. - The dashboard/`hermes serve` auto-archive sweep archived an arbitrary profile's store but read the config through the PROCESS HERMES_HOME, so one profile's sessions.auto_archive_days governed every other profile's retention. It now loads the config of the home whose store it sweeps. - Auto-VACUUM admission read "no FOREIGN holder" as "the store is quiet", but the /proc scan skips our own pid by design. VACUUM plus its TRUNCATE checkpoint therefore retired a WAL generation another live SessionDB in THIS process still held. The path-keyed registry now answers the in-process half. Also: argv naming only the install root no longer dismisses a multiplexer as "another instance" when the store is a profile store beneath that root — under one-process-per-host that argv is exactly what the holder looks like.
450 lines
18 KiB
Python
450 lines
18 KiB
Python
"""Process and descriptor authority for state.db structural maintenance.
|
|
|
|
This module owns the proof that no foreign process still holds the active or
|
|
an unlinked SQLite DB/WAL/SHM generation. ``hermes_state`` supplies only the
|
|
SQLite connection factory needed by the final lock probe.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import errno
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Callable, List, Optional, Sequence, Set, Tuple
|
|
|
|
try: # Hard dependency, but tolerate scaffold-phase imports before pip install.
|
|
import psutil
|
|
except ImportError: # pragma: no cover - stripped/scaffold installs only
|
|
psutil = None # type: ignore[assignment]
|
|
|
|
|
|
def read_only_db_uri(db_path) -> str:
|
|
"""``file:`` URI for a ``mode=ro`` open. ``as_uri()`` percent-encodes ``?``/``#`` in the home
|
|
path; a raw ``f"file:{path}?mode=ro"`` truncates there and opens the wrong (empty) database."""
|
|
return Path(db_path).resolve().as_uri() + "?mode=ro"
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_IS_WINDOWS = sys.platform == "win32"
|
|
_HERMES_EXECUTABLES = frozenset({"hermes", "hermes-agent", "hermes-acp"})
|
|
_HERMES_PYTHON_MODULES = frozenset({"acp_adapter", "hermes_cli.main"})
|
|
_HERMES_PYTHON_SCRIPTS = frozenset({"hermes_cli/main.py", "run_agent.py"})
|
|
_PYTHON_SHORT_OPTIONS_WITH_OPERANDS = frozenset({"Q", "W", "X"})
|
|
_PYTHON_LONG_OPTIONS_WITH_OPERANDS = frozenset(
|
|
{"--check-hash-based-pycs", "--jit"}
|
|
)
|
|
|
|
|
|
def _read_proc_argv(pid: int) -> Optional[List[str]]:
|
|
"""Read /proc/<pid>/cmdline without losing argv boundaries."""
|
|
try:
|
|
with open(f"/proc/{pid}/cmdline", "rb") as handle:
|
|
raw = handle.read()
|
|
if not raw:
|
|
return None
|
|
argv = raw.decode("utf-8", "replace").split("\x00")
|
|
if argv[-1] == "":
|
|
argv.pop()
|
|
return argv or None
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def describe_holder_pid(pid: int) -> str:
|
|
"""``PID 123 (hermes gateway run)`` for operator-facing holder lists; /proc argv first, psutil elsewhere."""
|
|
argv = _read_proc_argv(pid)
|
|
if argv is None and psutil is not None:
|
|
try:
|
|
argv = psutil.Process(pid).cmdline() or None
|
|
except Exception:
|
|
argv = None
|
|
who = " ".join(" ".join([os.path.basename(argv[0]), *argv[1:]]).split())[:80] if argv else "command line unavailable"
|
|
return f"PID {pid} ({who})"
|
|
|
|
|
|
def _looks_like_python_executable(program: str) -> bool:
|
|
name = os.path.basename(program).lower().removesuffix(".exe")
|
|
for prefix in ("python", "pypy"):
|
|
if name.startswith(prefix):
|
|
suffix = name[len(prefix) :]
|
|
return not suffix or all(char.isdigit() or char == "." for char in suffix)
|
|
return False
|
|
|
|
|
|
def _python_execution_target(argv: Sequence[str]) -> Optional[Tuple[str, str]]:
|
|
"""Return the Python module or script selected by interpreter options."""
|
|
index = 1
|
|
while index < len(argv):
|
|
arg = argv[index]
|
|
if arg == "--":
|
|
index += 1
|
|
return ("script", argv[index]) if index < len(argv) else None
|
|
if arg in _PYTHON_LONG_OPTIONS_WITH_OPERANDS:
|
|
index += 2
|
|
continue
|
|
if arg.startswith("--check-hash-based-pycs=") or arg.startswith("--jit="):
|
|
index += 1
|
|
continue
|
|
if arg.startswith("--"):
|
|
index += 1
|
|
continue
|
|
if arg.startswith("-") and arg != "-":
|
|
options = arg[1:]
|
|
option_index = 0
|
|
consumed_next = False
|
|
while option_index < len(options):
|
|
option = options[option_index]
|
|
attached = options[option_index + 1 :]
|
|
if option == "c":
|
|
return None
|
|
if option == "m":
|
|
if attached:
|
|
return "module", attached
|
|
index += 1
|
|
return ("module", argv[index]) if index < len(argv) else None
|
|
if option in _PYTHON_SHORT_OPTIONS_WITH_OPERANDS:
|
|
consumed_next = not attached
|
|
break
|
|
option_index += 1
|
|
index += 2 if consumed_next else 1
|
|
continue
|
|
return "script", arg
|
|
return None
|
|
|
|
|
|
def _looks_like_hermes(argv: Sequence[str]) -> bool:
|
|
"""Return whether argv identifies a supported Hermes execution target."""
|
|
if not argv:
|
|
return False
|
|
program = os.path.basename(argv[0]).lower().removesuffix(".exe")
|
|
if program in _HERMES_EXECUTABLES:
|
|
return True
|
|
if not _looks_like_python_executable(program):
|
|
return False
|
|
target = _python_execution_target(argv)
|
|
if target is None:
|
|
return False
|
|
kind, value = target
|
|
normalized = value.lower().replace("\\", "/")
|
|
if kind == "module":
|
|
return normalized in _HERMES_PYTHON_MODULES
|
|
return any(
|
|
normalized == script or normalized.endswith(f"/{script}")
|
|
for script in _HERMES_PYTHON_SCRIPTS
|
|
)
|
|
|
|
|
|
def canonical_sqlite_path(path: str) -> str:
|
|
"""Normalize a /proc fd target, stripping the Linux `` (deleted)`` suffix."""
|
|
return os.path.normcase(os.path.abspath(path.removesuffix(" (deleted)")))
|
|
|
|
|
|
def _argv_scoped_to_other_home(argv: Sequence[str], db_path: Path) -> bool:
|
|
"""Return whether argv proves the process belongs to a DIFFERENT instance.
|
|
|
|
``state.db`` lives at the HERMES_HOME root, so an absolute-path token
|
|
containing a ``/.hermes`` segment (or naming a ``state.db``/WAL/SHM under
|
|
some other parent) identifies that token's own Hermes home. When at least
|
|
one such token exists AND no token references this instance's state.db,
|
|
its sidecars, or its home directory, the process provably works on a
|
|
different generation and must not be counted as an uninspectable holder
|
|
of ours (issue #92401: a second gateway under /home/demo/.hermes deferred
|
|
this instance's stale-FTS rebuild forever despite lsof proving zero open
|
|
handles). Ambiguous argv without absolute-path tokens returns False and
|
|
keeps the fail-closed suspicion.
|
|
|
|
One process per host serves EVERY profile, so a token naming the install
|
|
root is not evidence of another instance when ``db_path`` is a profile
|
|
store under it: a default-launched multiplexer holds
|
|
``<root>/profiles/<name>/state.db`` while its argv mentions only
|
|
``<root>``. Only a token under a DIFFERENT install root still counts.
|
|
"""
|
|
db_path_str = os.path.abspath(os.fspath(db_path))
|
|
this_home = os.path.dirname(db_path_str)
|
|
# ``<root>/profiles/<name>/state.db`` -> ``<root>``; the multiplexer scoped to the
|
|
# install root is a candidate holder of every profile store beneath it.
|
|
profiles_dir = os.path.dirname(this_home)
|
|
install_root = (
|
|
os.path.dirname(profiles_dir)
|
|
if os.path.basename(profiles_dir) == "profiles"
|
|
else None
|
|
)
|
|
ours = {
|
|
os.path.normcase(candidate)
|
|
for candidate in (
|
|
db_path_str,
|
|
db_path_str + "-wal",
|
|
db_path_str + "-shm",
|
|
this_home,
|
|
*((install_root,) if install_root else ()),
|
|
)
|
|
}
|
|
own_prefixes = tuple(
|
|
os.path.normcase(root) + os.sep
|
|
for root in (this_home, *((install_root,) if install_root else ()))
|
|
)
|
|
other_home_seen = False
|
|
for token in argv:
|
|
if not isinstance(token, str):
|
|
continue
|
|
if token.startswith("/"):
|
|
path_token = token
|
|
elif token.startswith("-") and "=" in token:
|
|
# ``--db=/abs/path``-style options carry a path value; anchor on
|
|
# the text after '=' so normpath does not prepend the option.
|
|
value = token.split("=", 1)[1]
|
|
path_token = value if value.startswith("/") else None
|
|
else:
|
|
path_token = None
|
|
if path_token is not None:
|
|
normalized = os.path.normcase(os.path.normpath(path_token))
|
|
if normalized in ours or normalized.startswith(own_prefixes):
|
|
return False
|
|
if "/.hermes" in normalized or normalized.endswith("/.hermes"):
|
|
other_home_seen = True
|
|
elif os.path.basename(normalized) in (
|
|
"state.db",
|
|
"state.db-wal",
|
|
"state.db-shm",
|
|
):
|
|
other_home_seen = True
|
|
return other_home_seen
|
|
|
|
|
|
def foreign_state_db_holders(db_path: Path) -> List[Tuple[int, str]]:
|
|
"""Return foreign holders of the DB or one of its WAL sidecars.
|
|
|
|
A scan failure is represented as an unknown holder. Structural maintenance
|
|
must not assume quiescence when an old, unlinked SQLite generation may
|
|
still be open by another process.
|
|
"""
|
|
if _IS_WINDOWS:
|
|
return []
|
|
|
|
# realpath, not abspath: psutil/libproc report the kernel-resolved pathname, so a symlinked
|
|
# HERMES_HOME would otherwise make every holder invisible and let maintenance proceed.
|
|
db_path_str = os.path.realpath(os.fspath(db_path))
|
|
watched = {
|
|
canonical_sqlite_path(db_path_str),
|
|
canonical_sqlite_path(db_path_str + "-wal"),
|
|
canonical_sqlite_path(db_path_str + "-shm"),
|
|
}
|
|
holders: List[Tuple[int, str]] = []
|
|
watched_ids: Set[Tuple[int, int]] = set()
|
|
db_dev: Optional[int] = None
|
|
for candidate in (db_path_str, db_path_str + "-wal", db_path_str + "-shm"):
|
|
try:
|
|
stat_result = os.stat(candidate)
|
|
except OSError as exc:
|
|
if exc.errno not in (errno.ENOENT, errno.ESRCH):
|
|
holders.append(
|
|
(-1, f"watched-file stat failed: {candidate}: {exc}")
|
|
)
|
|
continue
|
|
watched_ids.add((stat_result.st_dev, stat_result.st_ino))
|
|
if candidate == db_path_str:
|
|
db_dev = stat_result.st_dev
|
|
|
|
if sys.platform.startswith("linux"):
|
|
try:
|
|
own_pid = os.getpid()
|
|
for pid_str in os.listdir("/proc"):
|
|
if not pid_str.isdigit():
|
|
continue
|
|
pid = int(pid_str)
|
|
if pid == own_pid:
|
|
continue
|
|
fd_dir = f"/proc/{pid}/fd"
|
|
try:
|
|
fds = os.listdir(fd_dir)
|
|
except OSError:
|
|
argv = _read_proc_argv(pid)
|
|
if (
|
|
argv is not None
|
|
and _looks_like_hermes(argv)
|
|
and not _argv_scoped_to_other_home(argv, db_path)
|
|
):
|
|
cmdline = " ".join(argv)
|
|
holders.append((pid, f"uninspectable holder: {cmdline[:80]}"))
|
|
continue
|
|
for fd in fds:
|
|
fd_path = f"{fd_dir}/{fd}"
|
|
try:
|
|
target = os.readlink(fd_path)
|
|
except OSError as exc:
|
|
if exc.errno in (errno.ENOENT, errno.ESRCH):
|
|
continue
|
|
argv = _read_proc_argv(pid)
|
|
if (
|
|
argv is not None
|
|
and _looks_like_hermes(argv)
|
|
and not _argv_scoped_to_other_home(argv, db_path)
|
|
):
|
|
holders.append(
|
|
(
|
|
pid,
|
|
f"uninspectable descriptor: {fd_path}: {exc}",
|
|
)
|
|
)
|
|
continue
|
|
target_is_watched = canonical_sqlite_path(target) in watched
|
|
try:
|
|
fd_stat = os.stat(fd_path)
|
|
except OSError as exc:
|
|
if exc.errno in (errno.ENOENT, errno.ESRCH):
|
|
continue
|
|
if target_is_watched:
|
|
holders.append(
|
|
(pid, f"uninspectable descriptor: {target}: {exc}")
|
|
)
|
|
else:
|
|
argv = _read_proc_argv(pid)
|
|
if (
|
|
argv is not None
|
|
and _looks_like_hermes(argv)
|
|
and not _argv_scoped_to_other_home(argv, db_path)
|
|
):
|
|
holders.append(
|
|
(
|
|
pid,
|
|
"uninspectable descriptor: "
|
|
f"{target}: {exc}",
|
|
)
|
|
)
|
|
continue
|
|
if (fd_stat.st_dev, fd_stat.st_ino) in watched_ids or (
|
|
target_is_watched
|
|
and target.endswith(" (deleted)")
|
|
and db_dev is not None
|
|
and fd_stat.st_dev == db_dev
|
|
):
|
|
holders.append((pid, target))
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Could not prove state.db has no foreign holders; "
|
|
"deferring structural maintenance: %s",
|
|
exc,
|
|
)
|
|
holders.append((-1, f"open-file scan failed: {exc}"))
|
|
return holders
|
|
|
|
if psutil is None:
|
|
return [(-1, "open-file scan unavailable")]
|
|
try:
|
|
for process in psutil.process_iter(["pid", "open_files"]):
|
|
info = process.info
|
|
pid = int(info["pid"])
|
|
if pid == os.getpid():
|
|
continue
|
|
for opened in info.get("open_files") or ():
|
|
path = getattr(opened, "path", "")
|
|
if path and canonical_sqlite_path(os.path.realpath(path)) in watched:
|
|
holders.append((pid, path))
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Could not prove state.db has no foreign holders; "
|
|
"deferring structural maintenance: %s",
|
|
exc,
|
|
)
|
|
holders.append((-1, f"open-file scan failed: {exc}"))
|
|
return holders
|
|
|
|
|
|
def in_process_state_db_holders(
|
|
db_path: Path, *, exclude=None
|
|
) -> List[Tuple[int, str]]:
|
|
"""Return holders of ``db_path`` inside THIS process, other than *exclude*.
|
|
|
|
:func:`foreign_state_db_holders` skips ``os.getpid()`` by design, so it answers a
|
|
cross-PROCESS question only. Consumers that read "no holders" as "the store is quiet"
|
|
(auto-VACUUM admission) need this arm too: a VACUUM plus its TRUNCATE checkpoint retires
|
|
the generation a sibling SessionDB in this very process still holds.
|
|
"""
|
|
from hermes_state_registry import other_generations_for_path
|
|
|
|
return [
|
|
(os.getpid(), description)
|
|
for description in other_generations_for_path(db_path, exclude=exclude)
|
|
]
|
|
|
|
|
|
def held_store_refusal(db_path: Path, *, command: str, force_hint: Optional[str] = "--force") -> Optional[str]:
|
|
"""Operator-facing refusal for structural maintenance (VACUUM, index rebuild, bulk delete) while another
|
|
process holds ``db_path`` or a WAL sidecar; ``None`` when the store is provably quiet.
|
|
|
|
Running ``hermes sessions optimize-storage`` underneath a fleet of live gateways put every agent into
|
|
the retired-WAL refusal until all writers were stopped (#110054). Same fail-closed scan doctor and
|
|
repair use: an incomplete scan refuses too, it never reads as an all-clear.
|
|
"""
|
|
holders = foreign_state_db_holders(db_path)
|
|
if not holders:
|
|
return None
|
|
from hermes_constants import profile_cli_selector
|
|
from hermes_state_errors import STORAGE_RECOVERY_DOCS_URL
|
|
|
|
by_pid: dict[int, Set[str]] = {}
|
|
unknown: List[str] = []
|
|
for pid, target in holders:
|
|
if pid <= 0 or target.startswith("uninspectable"):
|
|
unknown.append(target)
|
|
else:
|
|
by_pid.setdefault(pid, set()).add(Path(target.removesuffix(" (deleted)")).name)
|
|
lines = [f"Refusing `hermes sessions {command}`: another process is using {db_path}."]
|
|
lines += [f" {describe_holder_pid(pid)}: {', '.join(sorted(by_pid[pid]))}" for pid in sorted(by_pid)]
|
|
if unknown:
|
|
lines.append(f" cannot prove the database is quiet (holder scan incomplete: {unknown[0][:120]})")
|
|
profile_arg = profile_cli_selector()
|
|
lines += [
|
|
"Rewriting the database under a live writer is how every agent ends up refusing turns with the "
|
|
"retired state.db-wal error. Nothing is lost.",
|
|
f"Stop them first (`hermes {profile_arg}gateway stop`, quit the Desktop app, pause cron), then re-run.",
|
|
]
|
|
if force_hint:
|
|
lines.append(f"Override with {force_hint} if you accept the risk.")
|
|
lines.append(f"Recovery guide: {STORAGE_RECOVERY_DOCS_URL}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def live_writer_holds_db(
|
|
db_path: Path,
|
|
*,
|
|
connect_repair_durable: Callable[..., sqlite3.Connection],
|
|
) -> bool:
|
|
"""Return whether repair lacks proven exclusive ownership of ``db_path``.
|
|
|
|
ANY foreign process holding the DB or a sidecar is a live holder (#103339): the lock probe below
|
|
cannot see a DELETE-mode reader (SHARED only) and cannot run at all on a malformed file, and those
|
|
are exactly the states repair/VACUUM/checkpoint get invoked in. The holder scan is the authority and
|
|
fails closed on its own failures (unknown/uninspectable sentinels); the probe only adds a positive
|
|
lock signal on top."""
|
|
if foreign_state_db_holders(db_path):
|
|
return True
|
|
|
|
probe = None
|
|
try:
|
|
probe = connect_repair_durable(db_path, timeout=0.0)
|
|
probe.execute("PRAGMA locking_mode=EXCLUSIVE")
|
|
probe.execute("BEGIN IMMEDIATE")
|
|
probe.execute("ROLLBACK")
|
|
return False
|
|
except sqlite3.OperationalError as exc:
|
|
lowered = str(exc).lower()
|
|
return "locked" in lowered or "busy" in lowered
|
|
except sqlite3.DatabaseError:
|
|
# Malformed/unreadable with no holder on the scan: nobody else has it open, so repair may run.
|
|
return False
|
|
finally:
|
|
if probe is not None:
|
|
try:
|
|
probe.execute("PRAGMA locking_mode=NORMAL")
|
|
except Exception:
|
|
pass
|
|
try:
|
|
probe.close()
|
|
except Exception:
|
|
pass
|