diff --git a/hermes_cli/profiles.py b/hermes_cli/profiles.py index 0fc6151979..222ccd1c43 100644 --- a/hermes_cli/profiles.py +++ b/hermes_cli/profiles.py @@ -1184,6 +1184,11 @@ def delete_profile(name: str, yes: bool = False) -> Path: _released = _MemoryStore.release_all_under(profile_dir) if _released: print(f"✓ Released {_released} memory-store connection(s) held by this process") + with contextlib.suppress(Exception): + from hermes_state_registry import close_all_under as _close_session_dbs_under + _closed = _close_session_dbs_under(profile_dir) + if _closed: + print(f"✓ Released {_closed} session database connection(s) held by this process") # 3. Remove wrapper script if has_wrapper and remove_wrapper_script(canon): diff --git a/hermes_state_registry.py b/hermes_state_registry.py index f9415455c1..3d94c80a32 100644 --- a/hermes_state_registry.py +++ b/hermes_state_registry.py @@ -32,6 +32,7 @@ from __future__ import annotations import contextlib import logging +import os import threading from pathlib import Path from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Tuple @@ -308,6 +309,40 @@ def release(db: "SessionDB") -> bool: return True +def _path_is_under(path: Path, root: Path) -> bool: + """True when *path* is *root* or a file inside it (normcase, resolved).""" + try: + path_key = os.path.normcase(str(path.resolve())) + root_key = os.path.normcase(str(root.resolve())) + except OSError: + path_key = os.path.normcase(str(path)) + root_key = os.path.normcase(str(root)) + return path_key == root_key or path_key.startswith(root_key + os.sep) + + +def _teardown_swept_generations( + generations: List[_Generation], + teardown_barriers: Dict[Path, _TeardownBarrier], + active_teardowns: List[_TeardownBarrier], +) -> int: + """Close *generations* outside the registry lock; wait for already-admitted teardowns.""" + by_path: Dict[Path, List[_Generation]] = {} + for generation in generations: + by_path.setdefault(generation.path, []).append(generation) + for path, path_generations in by_path.items(): + with _lock: + lifecycle_lock = _path_lifecycle_lock_locked(path) + try: + with lifecycle_lock: + for generation in path_generations: + _teardown(generation.db) + finally: + _finish_teardown(path, teardown_barriers[path]) + for barrier in active_teardowns: + barrier.event.wait() + return len(generations) + + def close_all() -> int: """Close every shared SessionDB regardless of refcount; returns the count. For gateway shutdown, after all agents and cron jobs finished. Idempotent.""" @@ -321,26 +356,41 @@ def close_all() -> int: _retired.clear() for generation in generations: generation.retired = True - # Teardown outside the lock, one path at a time. Holding the lifecycle - # mutex across all generations for a path prevents an old retired handle - # and the current handle from checkpointing the same sidecars concurrently. - by_path: Dict[Path, List[_Generation]] = {} - for generation in generations: - by_path.setdefault(generation.path, []).append(generation) - for path, path_generations in by_path.items(): - with _lock: - lifecycle_lock = _path_lifecycle_lock_locked(path) - try: - with lifecycle_lock: - for generation in path_generations: - _teardown(generation.db) - finally: - _finish_teardown(path, teardown_barriers[path]) - # A final release that removed its generation before this sweep took _lock still owns - # its physical close; wait for it rather than return over a running teardown. - for barrier in active_teardowns: - barrier.event.wait() - return len(generations) + return _teardown_swept_generations(generations, teardown_barriers, active_teardowns) + + +def close_all_under(directory: str | Path) -> int: + """Force-close every shared SessionDB whose file lives under *directory*; returns the count. + + Profile delete rmtree (and a same-name recreate) fails while this process still holds + ``state.db``. Same contract as ``MemoryStore.release_all_under``: a live holder is + expected to fail afterward; a process that holds none is a no-op returning 0. + """ + try: + root = Path(directory).expanduser().resolve() + except OSError: + root = Path(directory).expanduser() + teardown_barriers: Dict[Path, _TeardownBarrier] = {} + with _lock: + generations = [ + generation + for generation in list(_generations.values()) + list(_retired.values()) + if _path_is_under(generation.path, root) + ] + if not generations: + return 0 + selected_paths = {generation.path for generation in generations} + active_teardowns = [ + barrier for path, barrier in _tearing_down.items() if path in selected_paths + ] + for path in selected_paths: + teardown_barriers[path] = _admit_teardown_locked(path) + for generation in generations: + generation.retired = True + if _generations.get(generation.path) is generation: + _generations.pop(generation.path, None) + _retired.pop(id(generation.db), None) + return _teardown_swept_generations(generations, teardown_barriers, active_teardowns) def live_shared_session_dbs() -> List["SessionDB"]: diff --git a/tests/hermes_cli/test_deleted_profile_tombstone.py b/tests/hermes_cli/test_deleted_profile_tombstone.py index fdabf90ae8..608e078ad4 100644 --- a/tests/hermes_cli/test_deleted_profile_tombstone.py +++ b/tests/hermes_cli/test_deleted_profile_tombstone.py @@ -106,6 +106,22 @@ class TestDeletedProfileTombstone: assert recreated.is_dir() assert "worker" in _named_homes(profile_env) + def test_delete_releases_this_process_session_db(self, profile_env): + import hermes_state_registry as registry + + profile_dir = create_profile("worker", no_alias=True, no_skills=True) + held = registry.acquire(profile_dir / "state.db") + assert held._conn is not None + try: + _delete("worker") + assert held._conn is None + recreated = create_profile("worker", no_alias=True, no_skills=True) + again = registry.acquire(recreated / "state.db") + assert again._conn is not None + registry.release(again) + finally: + registry.close_all() + def test_profile_exists_is_false_for_tombstoned_shell(self, profile_env): profile_dir = create_profile("worker", no_alias=True, no_skills=True) assert profile_exists("worker") is True diff --git a/tests/hermes_state/test_shared_session_db_registry.py b/tests/hermes_state/test_shared_session_db_registry.py index bc828b5f6a..8c421603a4 100644 --- a/tests/hermes_state/test_shared_session_db_registry.py +++ b/tests/hermes_state/test_shared_session_db_registry.py @@ -644,3 +644,30 @@ class TestMultiGenerationTeardownBarrier: fresh = registry.acquire(db_path) assert fresh is not db assert registry.release(fresh) is True + + +class TestCloseAllUnder: + def test_closes_connections_inside_directory_only(self, tmp_path): + profile_dir = tmp_path / "profiles" / "work" + profile_dir.mkdir(parents=True) + inside = registry.acquire(profile_dir / "state.db") + outside = registry.acquire(tmp_path / "other" / "state.db") + assert inside._conn is not None + assert outside._conn is not None + + closed = registry.close_all_under(profile_dir) + assert closed == 1 + assert inside._conn is None + assert outside._conn is not None + assert registry.close_all_under(profile_dir) == 0 + + again = registry.acquire(profile_dir / "state.db") + assert again._conn is not None + assert again is not inside + assert registry.release(again) is True + assert registry.release(outside) is True + + def test_noop_when_this_process_holds_nothing(self, tmp_path): + profile_dir = tmp_path / "profiles" / "empty" + profile_dir.mkdir(parents=True) + assert registry.close_all_under(profile_dir) == 0