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

refactor(hermes_state): one canonical_sqlite_path

Browse Source 
`hermes_state_dbfile._canonical_sqlite_path` was a byte-identical copy of
`hermes_state_holders.canonical_sqlite_path`; keep the public one and repoint
the two hermes_state call sites. No import cycle: holders is stdlib+psutil.
This commit is contained in:
kshitijk4poor
2026-09-14 20:59:06 +05:30
committed by kshitij
parent afeaabf629
commit efca6efdc8
2 changed files with 6 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
hermes_state.py
View File

@@ -47,7 +47,7 @@ from hermes_state_schema import SessionSchemaMixin
import hermes_state_holders as _state_holders
import hermes_state_lockguard as _lockguard
from hermes_state_dbfile import (
_canonical_sqlite_path, _connect_tracked_db, _fd_is_truly_unlinked, _prepare_connection_retirement,
_connect_tracked_db, _fd_is_truly_unlinked, _prepare_connection_retirement,
_read_sqlite_application_id, _stat_sqlite_sidecar_identity,
_watched_sqlite_sidecar_paths, has_invalid_sqlite_header_preopen, is_zeroed_state_db, quarantine_cross_process_lock,
quarantine_invalid_state_db,
@@ -1114,7 +1114,7 @@ class SessionDB(
watched = _watched_sqlite_sidecar_paths(self.db_path)
try:
for target, fd_path in _proc_fd_targets(os.getpid()):
canonical = _canonical_sqlite_path(target)
canonical = _state_holders.canonical_sqlite_path(target)
if (" (deleted)" in target and canonical in watched
and _fd_is_truly_unlinked(fd_path, watched[canonical])):
return True

12
hermes_state_dbfile.py
View File

@@ -23,6 +23,7 @@ import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple
from hermes_state_holders import canonical_sqlite_path
from hermes_state_common import (
FTS_REBUILD_DEFERRAL_KEY, stat_db_file_identity as _stat_db_file_identity
)
@@ -131,17 +132,12 @@ def _stat_sqlite_sidecar_identity(db_path: Path) -> Dict[str, tuple]:
return {suffix: ident for suffix, ident in idents.items() if ident is not None}
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 _watched_sqlite_sidecar_paths(db_path) -> Dict[str, str]:
"""Map each sidecar's canonical (/proc-comparable) form to its literal, still-named path,
so a canonical match can be re-``stat``'d for identity rather than trusted as text."""
base = os.path.abspath(os.fspath(db_path))
literal = (base + "-wal", base + "-shm")
return {_canonical_sqlite_path(path): path for path in literal}
return {canonical_sqlite_path(path): path for path in literal}
def _identity_is_truly_unlinked(identity: "Tuple[int, int]", watched_path: str) -> bool:
@@ -327,7 +323,7 @@ def iter_deleted_sqlite_sidecar_holders(db_path) -> List[Tuple[int, str]]:
elif sys.platform.startswith("linux"):
watched = _watched_sqlite_sidecar_paths(db_path)
for pid, target, fd_path in _iter_proc_fd_targets():
canonical = _canonical_sqlite_path(target)
canonical = canonical_sqlite_path(target)
if (" (deleted)" in target and canonical in watched
and _fd_is_truly_unlinked(fd_path, watched[canonical])):
holders.append((pid, target))
@@ -793,4 +789,4 @@ def _concrete_state_db_holder_pids(db_path: Path, holders: List[Tuple[int, str]]
canonical_db = os.path.normcase(os.path.abspath(os.fspath(db_path)))
watched = {canonical_db, canonical_db + "-wal", canonical_db + "-shm"}
return list(dict.fromkeys(
pid for pid, path in holders if pid > 0 and _canonical_sqlite_path(path) in watched))
pid for pid, path in holders if pid > 0 and canonical_sqlite_path(path) in watched))