diff --git a/hermes_cli/active_sessions.py b/hermes_cli/active_sessions.py index 4f21e6cd9a..4706909df4 100644 --- a/hermes_cli/active_sessions.py +++ b/hermes_cli/active_sessions.py @@ -18,7 +18,7 @@ from dataclasses import dataclass from pathlib import Path from typing import Any, Iterator, Optional -from hermes_constants import get_hermes_home +from hermes_constants import get_default_hermes_root, get_hermes_home logger = logging.getLogger(__name__) @@ -757,6 +757,42 @@ def transfer_active_session( return updated +def _drop_self_orphans( + entries: list[dict[str, Any]], own_live_lease_ids: set[str] | None +) -> list[dict[str, Any]]: + """Drop this process's leases only when its caller can vouch for owners.""" + if own_live_lease_ids is None: + return entries + pid = os.getpid() + return [ + entry + for entry in entries + if entry.get("pid") != pid + or str(entry.get("lease_id") or "") in own_live_lease_ids + ] + + +def _release_orphaned_leases_in_home( + registry_home: Path, live_lease_ids: set[str] +) -> int: + state_path = _state_path(registry_home) + if not state_path.exists(): + return 0 + with _FileLock(_lock_path(registry_home)): + try: + entries = _prune_dead(_read_entries(state_path, strict=True)) + except ActiveSessionRegistryError: + logger.warning( + "Active-session registry is unavailable; skipping orphaned-lease sweep" + ) + return 0 + kept = _drop_self_orphans(entries, live_lease_ids) + dropped = len(entries) - len(kept) + if dropped: + _write_entries(state_path, kept) + return dropped + + def release_orphaned_leases(live_lease_ids: set[str]) -> int: """Drop this process's registry entries that no live session owns. @@ -767,30 +803,26 @@ def release_orphaned_leases(live_lease_ids: set[str]) -> int: real, so it drops the rest itself — exact, with no heartbeat write on the turn path and no staleness threshold to tune. """ - pid = os.getpid() - state_path = _state_path() - # No registry file yet means no leases have ever been written under this - # home — don't take a lock (or create its file) on the idle-reaper tick. - if not state_path.exists(): - return 0 - with _FileLock(_lock_path()): + root = get_default_hermes_root() + homes = [root] + profiles_root = root / "profiles" + try: + homes.extend( + profile + for profile in profiles_root.iterdir() + if profile.is_dir() and not profile.name.startswith(".") + ) + except OSError: + pass + + dropped = 0 + for home in homes: try: - raw_entries = _read_entries(state_path, strict=True) - entries = _prune_dead(raw_entries) - except ActiveSessionRegistryError: - logger.warning( - "Active-session registry is unavailable; skipping orphaned-lease sweep" + dropped += _release_orphaned_leases_in_home(home, live_lease_ids) + except OSError as exc: + logger.debug( + "orphaned-lease sweep failed for %s: %s", home, exc ) - return 0 - kept = [ - entry - for entry in entries - if entry.get("pid") != pid - or str(entry.get("lease_id") or "") in live_lease_ids - ] - dropped = len(entries) - len(kept) - if dropped: - _write_entries(state_path, kept) return dropped @@ -812,6 +844,7 @@ def active_session_liveness_guard( session_id: str, *, registry_home: str | Path | None = None, + own_live_lease_ids: set[str] | None = None, ) -> Iterator[bool]: """Hold the registry lock while reporting whether ``session_id`` is leased. @@ -823,6 +856,7 @@ def active_session_liveness_guard( state_path, lock_path = _lease_paths(registry_home=registry_home) with _FileLock(lock_path): entries = _prune_dead(_read_entries(state_path, strict=True), strict=True) + entries = _drop_self_orphans(entries, own_live_lease_ids) _write_entries(state_path, entries) yield bool(target) and any( str(entry.get("session_id") or "") == target for entry in entries @@ -833,6 +867,8 @@ def active_session_liveness_guard( def release_active_session_liveness_guard( lease: ActiveSessionLease, session_id: str, + *, + own_live_lease_ids: set[str] | None = None, ) -> Iterator[bool]: """Remove ``lease`` and hold its registry lock through a lifecycle write. @@ -842,7 +878,9 @@ def release_active_session_liveness_guard( """ if not lease.enabled or lease.released: with active_session_liveness_guard( - session_id, registry_home=_registry_home_for_lease(lease) + session_id, + registry_home=_registry_home_for_lease(lease), + own_live_lease_ids=own_live_lease_ids, ) as active: yield active return @@ -857,6 +895,7 @@ def release_active_session_liveness_guard( for entry in entries if str(entry.get("lease_id") or "") != lease.lease_id ] + kept = _drop_self_orphans(kept, own_live_lease_ids) if len(kept) != len(entries): _write_entries(state_path, kept) lease.released = True diff --git a/tests/hermes_cli/test_active_sessions.py b/tests/hermes_cli/test_active_sessions.py index 5741aed178..e62ae64e78 100644 --- a/tests/hermes_cli/test_active_sessions.py +++ b/tests/hermes_cli/test_active_sessions.py @@ -172,6 +172,43 @@ def test_release_orphaned_leases_reclaims_only_unowned_own_pid_entries(tmp_path, assert orphan is not None +def test_release_orphaned_leases_sweeps_profile_runtime_registries( + tmp_path, monkeypatch +): + root = tmp_path / "hermes" + profile = root / "profiles" / "worker" + profile.mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(root)) + + root_lease, root_error = active_sessions.try_acquire_active_session( + session_id="root-orphan", surface="desktop", config={}, registry_home=root + ) + profile_lease, profile_error = active_sessions.try_acquire_active_session( + session_id="profile-orphan", + surface="desktop", + config={}, + registry_home=profile, + ) + assert root_lease is not None and root_error is None + assert profile_lease is not None and profile_error is None + + assert active_sessions.release_orphaned_leases(set()) == 2 + assert active_sessions.active_session_registry_snapshot(root) == [] + assert active_sessions.active_session_registry_snapshot(profile) == [] + + +def test_drop_self_orphans_spares_foreign_and_vouched_leases(): + own = os.getpid() + entries = [ + {"lease_id": "orphan", "pid": own}, + {"lease_id": "live", "pid": own}, + {"lease_id": "foreign", "pid": own + 1}, + ] + + assert active_sessions._drop_self_orphans(entries, None) == entries + assert active_sessions._drop_self_orphans(entries, {"live"}) == entries[1:] + + def test_release_under_profile_home_override_targets_acquisition_registry( tmp_path, monkeypatch ): diff --git a/tests/tui_gateway/test_cross_process_orphan_ownership.py b/tests/tui_gateway/test_cross_process_orphan_ownership.py index 481b837c07..c15d96512f 100644 --- a/tests/tui_gateway/test_cross_process_orphan_ownership.py +++ b/tests/tui_gateway/test_cross_process_orphan_ownership.py @@ -270,6 +270,79 @@ def test_automatic_cleanup_preserves_corrupt_registry_without_overwrite( assert state_path.read_text(encoding="utf-8") == corrupt +def test_own_live_lease_ids_reports_live_owners_and_skips_the_excluded( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _Lease: + def __init__(self, lease_id: str) -> None: + self.lease_id = lease_id + + first = _Lease("first") + second = _Lease("second") + monkeypatch.setattr( + server, + "_sessions", + { + "one": {"active_session_lease": first}, + "two": {"active_session_lease": second}, + "three": {"active_session_lease": None}, + }, + ) + + assert server._own_live_lease_ids() == {"first", "second"} + assert server._own_live_lease_ids(exclude=first) == {"second"} + + +def test_automatic_cleanup_reclaims_own_orphan_lease_not_treated_as_sibling( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile_home = tmp_path / "profile-home" + session_id = "own-orphan-session" + owner_lease, message = server._claim_active_session_slot( + session_id, + live_session_id="vanished-runtime", + surface="desktop", + profile_home=profile_home, + ) + assert owner_lease is not None and message is None + ended: list[tuple[str, str]] = [] + + class _FakeDB: + def get_session(self, target: str) -> dict[str, str]: + return {"id": target, "source": "desktop"} + + def end_session(self, target: str, reason: str) -> None: + ended.append((target, reason)) + + @contextlib.contextmanager + def _profile_db(_session: dict): + yield _FakeDB() + + monkeypatch.setattr(server, "_sessions", {}) + monkeypatch.setattr(server, "_session_db", _profile_db) + monkeypatch.setattr( + server, "_notify_session_boundary", lambda *args, **kwargs: None + ) + monkeypatch.setattr( + "tools.async_delegation.interrupt_for_session", lambda *args, **kwargs: None + ) + session = { + "active_session_lease": None, + "agent": None, + "history": [], + "history_lock": threading.Lock(), + "profile_home": str(profile_home), + "session_key": session_id, + "slash_worker": None, + "source": "desktop", + } + + server._finalize_session(session, end_reason="ws_orphan_reap") + + assert ended == [(session_id, "ws_orphan_reap")] + assert active_session_registry_snapshot(registry_home=profile_home) == [] + + def test_liveness_guard_serializes_cross_process_acquire(tmp_path: Path) -> None: home = tmp_path / "guard-home" waiting_file = tmp_path / "child-waiting" diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 52b0a72d38..4455b91e5f 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -764,6 +764,17 @@ def _release_active_session_slot(session: dict | None) -> bool: return False +def _own_live_lease_ids(*, exclude=None) -> set[str]: + """Snapshot leases still backed by this process's live session records.""" + with _sessions_lock: + return { + str(lease.lease_id) + for session in _sessions.values() + if (lease := session.get("active_session_lease")) is not None + and lease is not exclude + } + + @contextlib.contextmanager def _other_runtime_lease_guard(session_id: str, session: dict): """Release this runtime and lock sibling ownership through the DB write.""" @@ -784,13 +795,20 @@ def _other_runtime_lease_guard(session_id: str, session: dict): last_error: Exception | None = None stack = contextlib.ExitStack() + own_live_lease_ids = _own_live_lease_ids(exclude=lease) for attempt in range(3): try: if lease is not None and getattr(lease, "enabled", False): - guard = release_active_session_liveness_guard(lease, session_id) + guard = release_active_session_liveness_guard( + lease, + session_id, + own_live_lease_ids=own_live_lease_ids, + ) else: guard = active_session_liveness_guard( - session_id, registry_home=session.get("profile_home") + session_id, + registry_home=session.get("profile_home"), + own_live_lease_ids=own_live_lease_ids, ) active = stack.enter_context(guard) break @@ -1930,12 +1948,7 @@ def _reclaim_orphaned_leases() -> None: try: from hermes_cli.active_sessions import release_orphaned_leases - with _sessions_lock: - live = { - lease.lease_id - for session in _sessions.values() - if (lease := session.get("active_session_lease")) is not None - } + live = _own_live_lease_ids() if dropped := release_orphaned_leases(live): logger.info("Reclaimed %d orphaned active-session lease(s)", dropped) except Exception: