From 51e67babca3499c71e515b38971eb3e129b729f9 Mon Sep 17 00:00:00 2001
From: Brooklyn Nicholson <brooklyn.bb.nicholson@gmail.com>
Date: Thu, 27 Aug 2026 11:38:19 -0500
Subject: [PATCH] fix(cli): keep Desktop liveness leases when the session cap
 is off

Unlimited sessions used a no-op lease, so a sibling profile backend could
not see that the same durable session was still owned. Track liveness in
the profile registry without imposing a cap, and fail closed when the
registry cannot be inspected.

Co-authored-by: metamindedu <metamind@kakao.com>
---
 hermes_cli/active_sessions.py            | 414 +++++++++++++++++++----
 tests/hermes_cli/test_active_sessions.py | 320 ++++++++++++++++++
 2 files changed, 668 insertions(+), 66 deletions(-)

diff --git a/hermes_cli/active_sessions.py b/hermes_cli/active_sessions.py
index 13aa1e41b3..e1cd8b0307 100644
--- a/hermes_cli/active_sessions.py
+++ b/hermes_cli/active_sessions.py
@@ -9,18 +9,24 @@ from __future__ import annotations
 
 import json
 import logging
+import math
 import os
 import time
 import uuid
+from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any, Iterator, Optional
 
 from hermes_constants import get_hermes_home
 
 logger = logging.getLogger(__name__)
 
 
+class ActiveSessionRegistryError(RuntimeError):
+    """The liveness registry could not prove a safe ownership decision."""
+
+
 def coerce_max_concurrent_sessions(value: Any, key: str = "max_concurrent_sessions") -> Optional[int]:
     """Return a positive integer cap, or None when disabled/invalid."""
     if value is None:
@@ -114,16 +120,30 @@ def active_session_limit_message(
     )
 
 
-def _state_dir() -> Path:
-    return Path(get_hermes_home()) / "runtime"
+def _registry_home(registry_home: str | Path | None = None) -> Path:
+    return Path(registry_home) if registry_home is not None else Path(get_hermes_home())
 
 
-def _state_path() -> Path:
-    return _state_dir() / "active_sessions.json"
+def _state_dir(registry_home: str | Path | None = None) -> Path:
+    return _registry_home(registry_home) / "runtime"
 
 
-def _lock_path() -> Path:
-    return _state_dir() / "active_sessions.lock"
+def _state_path(registry_home: str | Path | None = None) -> Path:
+    return _state_dir(registry_home) / "active_sessions.json"
+
+
+def _lock_path(registry_home: str | Path | None = None) -> Path:
+    return _state_dir(registry_home) / "active_sessions.lock"
+
+
+def _lease_paths(
+    lease: Optional["ActiveSessionLease"] = None,
+    registry_home: str | Path | None = None,
+) -> tuple[Path, Path]:
+    if lease is not None and lease.state_path is not None and lease.lock_path is not None:
+        return lease.state_path, lease.lock_path
+    home = _registry_home(registry_home)
+    return _state_path(home), _lock_path(home)
 
 
 class _FileLock:
@@ -179,27 +199,100 @@ class _FileLock:
             self._fh = None
 
 
-def _read_entries(path: Path) -> list[dict[str, Any]]:
+def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]:
     try:
         with open(path, "r", encoding="utf-8") as fh:
             data = json.load(fh)
     except FileNotFoundError:
         return []
-    except Exception:
+    except Exception as exc:
+        if strict:
+            raise ActiveSessionRegistryError(
+                f"active session registry unreadable: {path}"
+            ) from exc
         logger.warning("Ignoring corrupt active session registry at %s", path)
         return []
     entries = data.get("entries") if isinstance(data, dict) else data
     if not isinstance(entries, list):
+        if strict:
+            raise ActiveSessionRegistryError(
+                f"active session registry has invalid shape: {path}"
+            )
         return []
-    return [entry for entry in entries if isinstance(entry, dict)]
+    valid = [entry for entry in entries if isinstance(entry, dict)]
+    if not strict:
+        return valid
+    if len(valid) != len(entries):
+        raise ActiveSessionRegistryError(
+            f"active session registry contains invalid entries: {path}"
+        )
+    seen_leases: set[str] = set()
+    for entry in valid:
+        lease_id = entry.get("lease_id")
+        session_id = entry.get("session_id")
+        pid = entry.get("pid")
+        if not isinstance(lease_id, str) or not lease_id.strip():
+            raise ActiveSessionRegistryError(
+                f"active session registry contains an invalid lease id: {path}"
+            )
+        if lease_id in seen_leases:
+            raise ActiveSessionRegistryError(
+                f"active session registry contains a duplicate lease id: {path}"
+            )
+        seen_leases.add(lease_id)
+        if not isinstance(session_id, str) or not session_id.strip():
+            raise ActiveSessionRegistryError(
+                f"active session registry contains an invalid session id: {path}"
+            )
+        if isinstance(pid, bool) or not isinstance(pid, (int, str)):
+            pid_int = 0
+        else:
+            try:
+                pid_int = int(pid)
+            except (TypeError, ValueError):
+                pid_int = 0
+        if pid_int <= 0:
+            raise ActiveSessionRegistryError(
+                f"active session registry contains an invalid pid: {path}"
+            )
+        surface = entry.get("surface")
+        if surface is not None and not isinstance(surface, str):
+            raise ActiveSessionRegistryError(
+                f"active session registry contains an invalid surface: {path}"
+            )
+        tracked = entry.get("track_liveness")
+        if tracked is not None and not isinstance(tracked, bool):
+            raise ActiveSessionRegistryError(
+                f"active session registry contains an invalid liveness marker: {path}"
+            )
+        metadata = entry.get("metadata")
+        if metadata is not None and not isinstance(metadata, dict):
+            raise ActiveSessionRegistryError(
+                f"active session registry contains invalid metadata: {path}"
+            )
+        process_start = entry.get("process_start_time")
+        parsed_process_start = _optional_float(process_start)
+        if process_start not in (None, "") and (
+            parsed_process_start is None or not math.isfinite(parsed_process_start)
+        ):
+            raise ActiveSessionRegistryError(
+                f"active session registry contains an invalid process start time: {path}"
+            )
+    return valid
 
 
 def _write_entries(path: Path, entries: list[dict[str, Any]]) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     tmp = path.with_name(f"{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
-    with open(tmp, "w", encoding="utf-8") as fh:
-        json.dump({"entries": entries}, fh, sort_keys=True)
-    os.replace(tmp, path)
+    try:
+        with open(tmp, "w", encoding="utf-8") as fh:
+            json.dump({"entries": entries}, fh, sort_keys=True)
+        os.replace(tmp, path)
+    finally:
+        try:
+            tmp.unlink(missing_ok=True)
+        except OSError:
+            pass
 
 
 def _process_start_time(pid: int) -> Optional[float]:
@@ -222,6 +315,31 @@ def _optional_float(value: Any) -> Optional[float]:
         return None
 
 
+def _pid_liveness(pid: Any, process_start_time: Any = None) -> Optional[bool]:
+    """Return True/False for live/dead, or None when liveness is unknowable."""
+    try:
+        pid_int = int(pid)
+    except (TypeError, ValueError):
+        return None
+    if pid_int <= 0:
+        return None
+    try:
+        from gateway.status import _pid_exists
+
+        exists = bool(_pid_exists(pid_int))
+    except Exception:
+        return None
+    if not exists:
+        return False
+    expected_start = _optional_float(process_start_time)
+    if expected_start is None:
+        return True
+    current_start = _process_start_time(pid_int)
+    if current_start is None:
+        return None
+    return abs(current_start - expected_start) < 0.001
+
+
 def _pid_alive(pid: Any, process_start_time: Any = None) -> bool:
     try:
         pid_int = int(pid)
@@ -246,12 +364,24 @@ def _pid_alive(pid: Any, process_start_time: Any = None) -> bool:
     return abs(current_start - expected_start) < 0.001
 
 
-def _prune_dead(entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    return [
-        entry
-        for entry in entries
-        if _pid_alive(entry.get("pid"), entry.get("process_start_time"))
-    ]
+def _prune_dead(
+    entries: list[dict[str, Any]], *, strict: bool = False
+) -> list[dict[str, Any]]:
+    live: list[dict[str, Any]] = []
+    for entry in entries:
+        tracked = bool(entry.get("track_liveness"))
+        if strict or tracked:
+            state = _pid_liveness(entry.get("pid"), entry.get("process_start_time"))
+            if state is None:
+                raise ActiveSessionRegistryError(
+                    "active session owner liveness is unknown"
+                )
+            if state:
+                live.append(entry)
+            continue
+        if _pid_alive(entry.get("pid"), entry.get("process_start_time")):
+            live.append(entry)
+    return live
 
 
 @dataclass
@@ -269,6 +399,7 @@ class ActiveSessionLease:
     # phantom leases (#85431).
     state_path: Optional[Path] = None
     lock_path: Optional[Path] = None
+    track_liveness: bool = False
 
     def release(self) -> None:
         if self.released or not self.enabled:
@@ -276,30 +407,16 @@ class ActiveSessionLease:
         release_active_session(self)
 
 
-def try_acquire_active_session(
+def _lease_entry(
     *,
+    lease_id: str,
     session_id: str,
     surface: str,
-    config: Any,
     metadata: Optional[dict[str, Any]] = None,
-) -> tuple[Optional[ActiveSessionLease], Optional[str]]:
-    """Acquire an active-session slot.
-
-    Returns ``(lease, None)`` on success.  When the cap is disabled, the lease is
-    a no-op object so callers can unconditionally call ``release()``.
-    """
-    max_sessions = resolve_max_concurrent_sessions(config)
-    lease_id = uuid.uuid4().hex
-    if max_sessions is None:
-        return ActiveSessionLease(
-            lease_id=lease_id,
-            session_id=session_id,
-            surface=surface,
-            enabled=False,
-        ), None
-
+    track_liveness: bool = False,
+) -> dict[str, Any]:
     now = time.time()
-    entry = {
+    entry: dict[str, Any] = {
         "lease_id": lease_id,
         "session_id": str(session_id),
         "surface": str(surface),
@@ -308,20 +425,75 @@ def try_acquire_active_session(
         "started_at": now,
         "updated_at": now,
     }
+    if track_liveness:
+        entry["track_liveness"] = True
     if metadata:
         entry["metadata"] = {
             str(k): v for k, v in metadata.items() if isinstance(k, str)
         }
+    return entry
 
-    state_path = _state_path()
-    with _FileLock(_lock_path()):
-        raw_entries = _read_entries(state_path)
-        entries = _prune_dead(raw_entries)
+
+def try_acquire_active_session(
+    *,
+    session_id: str,
+    surface: str,
+    config: Any,
+    metadata: Optional[dict[str, Any]] = None,
+    registry_home: str | Path | None = None,
+    track_liveness: bool = False,
+) -> tuple[Optional[ActiveSessionLease], Optional[str]]:
+    """Acquire an active-session slot.
+
+    Returns ``(lease, None)`` on success.  When the cap is disabled, the lease is
+    a no-op object so callers can unconditionally call ``release()`` unless
+    ``track_liveness`` is true.  Liveness tracking keeps a real lease without
+    imposing a concurrency cap; ``registry_home`` lets profile-scoped backends
+    share the owning profile's registry even when launched from another home.
+    """
+    max_sessions = resolve_max_concurrent_sessions(config)
+    lease_id = uuid.uuid4().hex
+    if max_sessions is None and not track_liveness:
+        return ActiveSessionLease(
+            lease_id=lease_id,
+            session_id=session_id,
+            surface=surface,
+            enabled=False,
+        ), None
+
+    entry = _lease_entry(
+        lease_id=lease_id,
+        session_id=str(session_id),
+        surface=str(surface),
+        metadata=metadata,
+        track_liveness=track_liveness,
+    )
+
+    state_path, lock_path = _lease_paths(registry_home=registry_home)
+    with _FileLock(lock_path):
+        try:
+            raw_entries = _read_entries(state_path, strict=True)
+            entries = _prune_dead(raw_entries, strict=track_liveness)
+        except ActiveSessionRegistryError:
+            if track_liveness:
+                raise
+            logger.warning(
+                "Active-session registry is unavailable; allowing an "
+                "untracked session without overwriting it"
+            )
+            return ActiveSessionLease(
+                lease_id=lease_id,
+                session_id=session_id,
+                surface=surface,
+                enabled=False,
+                state_path=state_path,
+                lock_path=lock_path,
+            ), None
         pruned = len(raw_entries) - len(entries)
         if pruned:
             logger.info("Pruned %d stale active session lease(s)", pruned)
         active_count = len(entries)
-        if active_count >= max_sessions:
+        if max_sessions is not None and active_count >= max_sessions:
             _write_entries(state_path, entries)
             logger.info(
                 "Active session limit reached: active=%d max=%d surface=%s",
@@ -340,26 +512,37 @@ def try_acquire_active_session(
         session_id=str(session_id),
         surface=str(surface),
         state_path=state_path,
-        lock_path=_lock_path(),
+        lock_path=lock_path,
+        track_liveness=track_liveness,
     ), None
 
 
 def release_active_session(lease: ActiveSessionLease) -> None:
     # Prefer the registry the lease was acquired against: the caller may be
     # running under a profile HERMES_HOME override (#85431).
-    state_path = lease.state_path or _state_path()
-    lock_path = lease.lock_path or _lock_path()
-    try:
-        with _FileLock(lock_path):
-            entries = _prune_dead(_read_entries(state_path))
-            kept = [
-                entry
-                for entry in entries
-                if str(entry.get("lease_id") or "") != lease.lease_id
-            ]
-            if len(kept) != len(entries):
-                _write_entries(state_path, kept)
-    finally:
+    state_path, lock_path = _lease_paths(lease)
+    with _FileLock(lock_path):
+        if lease.released:
+            return
+        try:
+            raw_entries = _read_entries(state_path, strict=True)
+            entries = _prune_dead(raw_entries, strict=lease.track_liveness)
+        except ActiveSessionRegistryError:
+            if lease.track_liveness:
+                raise
+            logger.warning(
+                "Active-session registry is unavailable; preserving it while "
+                "releasing an untracked lease"
+            )
+            lease.released = True
+            return
+        kept = [
+            entry
+            for entry in entries
+            if str(entry.get("lease_id") or "") != lease.lease_id
+        ]
+        if len(kept) != len(entries):
+            _write_entries(state_path, kept)
         lease.released = True
 
 
@@ -379,10 +562,23 @@ def transfer_active_session(
         lease.session_id = new_session_id
         return True
 
-    state_path = lease.state_path or _state_path()
-    lock_path = lease.lock_path or _lock_path()
+    state_path, lock_path = _lease_paths(lease)
     with _FileLock(lock_path):
-        entries = _prune_dead(_read_entries(state_path))
+        # release() may have won after the optimistic precheck but before this
+        # thread acquired the file lock. Never resurrect a durably removed lease.
+        if lease.released:
+            return False
+        try:
+            raw_entries = _read_entries(state_path, strict=True)
+            entries = _prune_dead(raw_entries, strict=lease.track_liveness)
+        except ActiveSessionRegistryError:
+            if lease.track_liveness:
+                raise
+            logger.warning(
+                "Active-session registry is unavailable; refusing to overwrite "
+                "it during lease transfer"
+            )
+            return False
         updated = False
         for entry in entries:
             if str(entry.get("lease_id") or "") != lease.lease_id:
@@ -395,6 +591,17 @@ def transfer_active_session(
                 }
             updated = True
             break
+        if not updated and lease.track_liveness:
+            entries.append(
+                _lease_entry(
+                    lease_id=lease.lease_id,
+                    session_id=new_session_id,
+                    surface=lease.surface,
+                    metadata=metadata,
+                    track_liveness=True,
+                )
+            )
+            updated = True
         if updated:
             _write_entries(state_path, entries)
             lease.session_id = new_session_id
@@ -418,7 +625,14 @@ def release_orphaned_leases(live_lease_ids: set[str]) -> int:
     if not state_path.exists():
         return 0
     with _FileLock(_lock_path()):
-        entries = _prune_dead(_read_entries(state_path))
+        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"
+            )
+            return 0
         kept = [
             entry
             for entry in entries
@@ -431,10 +645,78 @@ def release_orphaned_leases(live_lease_ids: set[str]) -> int:
     return dropped
 
 
-def active_session_registry_snapshot() -> list[dict[str, Any]]:
+def active_session_registry_snapshot(
+    registry_home: str | Path | None = None,
+) -> list[dict[str, Any]]:
     """Return the pruned active-session registry for diagnostics/tests."""
-    state_path = _state_path()
-    with _FileLock(_lock_path()):
-        entries = _prune_dead(_read_entries(state_path))
-        _write_entries(state_path, entries)
+    state_path, lock_path = _lease_paths(registry_home=registry_home)
+    with _FileLock(lock_path):
+        raw_entries = _read_entries(state_path, strict=True)
+        entries = _prune_dead(raw_entries)
+        if entries != raw_entries:
+            _write_entries(state_path, entries)
         return entries
+
+
+@contextmanager
+def active_session_liveness_guard(
+    session_id: str,
+    *,
+    registry_home: str | Path | None = None,
+) -> Iterator[bool]:
+    """Hold the registry lock while reporting whether ``session_id`` is leased.
+
+    Keeping the lock across the caller's lifecycle mutation prevents a new
+    backend from acquiring a lease and reopening the row between the liveness
+    check and the corresponding ``end_session`` write.
+    """
+    target = str(session_id or "")
+    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)
+        _write_entries(state_path, entries)
+        yield bool(target) and any(
+            str(entry.get("session_id") or "") == target for entry in entries
+        )
+
+
+@contextmanager
+def release_active_session_liveness_guard(
+    lease: ActiveSessionLease,
+    session_id: str,
+) -> Iterator[bool]:
+    """Remove ``lease`` and hold its registry lock through a lifecycle write.
+
+    This makes automatic cleanup one atomic ownership decision: the local
+    runtime disappears, sibling liveness is checked, and the caller may end the
+    durable row before any new backend can acquire/reopen it.
+    """
+    if not lease.enabled or lease.released:
+        with active_session_liveness_guard(
+            session_id, registry_home=_registry_home_for_lease(lease)
+        ) as active:
+            yield active
+        return
+
+    target = str(session_id or "")
+    state_path, lock_path = _lease_paths(lease)
+    with _FileLock(lock_path):
+        raw_entries = _read_entries(state_path, strict=True)
+        entries = _prune_dead(raw_entries, strict=True)
+        kept = [
+            entry
+            for entry in entries
+            if str(entry.get("lease_id") or "") != lease.lease_id
+        ]
+        if len(kept) != len(entries):
+            _write_entries(state_path, kept)
+        lease.released = True
+        yield bool(target) and any(
+            str(entry.get("session_id") or "") == target for entry in kept
+        )
+
+
+def _registry_home_for_lease(lease: ActiveSessionLease) -> Path | None:
+    if lease.state_path is None:
+        return None
+    return lease.state_path.parent.parent
diff --git a/tests/hermes_cli/test_active_sessions.py b/tests/hermes_cli/test_active_sessions.py
index dcbc36af55..d2feff3ad9 100644
--- a/tests/hermes_cli/test_active_sessions.py
+++ b/tests/hermes_cli/test_active_sessions.py
@@ -2,10 +2,13 @@ import logging
 import os
 import subprocess
 import sys
+import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
+import pytest
+
 from hermes_cli import active_sessions
 
 
@@ -240,3 +243,320 @@ def test_transfer_under_profile_home_override_targets_acquisition_registry(
     root_registry = root / "runtime" / "active_sessions.json"
     entries = active_sessions._read_entries(root_registry)
     assert [entry["session_id"] for entry in entries] == ["after"]
+
+
+def test_liveness_registry_corruption_fails_closed_without_overwrite(
+    tmp_path, monkeypatch
+):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    state_path = home / "runtime" / "active_sessions.json"
+    state_path.parent.mkdir(parents=True)
+    corrupt = "{not-json"
+    state_path.write_text(corrupt, encoding="utf-8")
+
+    with pytest.raises(active_sessions.ActiveSessionRegistryError):
+        with active_sessions.active_session_liveness_guard("session-1"):
+            pass
+
+    with pytest.raises(active_sessions.ActiveSessionRegistryError):
+        active_sessions.active_session_registry_snapshot()
+
+    assert state_path.read_text(encoding="utf-8") == corrupt
+
+    with pytest.raises(active_sessions.ActiveSessionRegistryError):
+        active_sessions.try_acquire_active_session(
+            session_id="desktop-1",
+            surface="desktop",
+            config={},
+            track_liveness=True,
+        )
+    assert state_path.read_text(encoding="utf-8") == corrupt
+
+    # The pre-existing concurrency-cap path remains available/fail-open, but
+    # must not erase evidence that strict liveness ownership is unknown.
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="cli-1",
+        surface="cli",
+        config={"max_concurrent_sessions": 1},
+    )
+    assert lease is not None and message is None
+    assert lease.enabled is False
+    assert state_path.read_text(encoding="utf-8") == corrupt
+    lease.release()
+    assert state_path.read_text(encoding="utf-8") == corrupt
+
+
+def test_strict_registry_rejects_structurally_invalid_entries(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    state_path = home / "runtime" / "active_sessions.json"
+    base = {
+        "lease_id": "lease-1",
+        "session_id": "session-1",
+        "surface": "desktop",
+        "pid": os.getpid(),
+        "track_liveness": True,
+    }
+    invalid_entries = (
+        {key: value for key, value in base.items() if key != "lease_id"},
+        {**base, "lease_id": ""},
+        {key: value for key, value in base.items() if key != "session_id"},
+        {**base, "session_id": "  "},
+        {**base, "pid": 0},
+        {**base, "pid": 1.5},
+        {**base, "surface": 1},
+        {**base, "track_liveness": "yes"},
+        {**base, "metadata": []},
+        {**base, "process_start_time": "not-a-number"},
+        {**base, "process_start_time": "nan"},
+    )
+
+    for entry in invalid_entries:
+        active_sessions._write_entries(state_path, [entry])
+        original = state_path.read_text(encoding="utf-8")
+        with pytest.raises(active_sessions.ActiveSessionRegistryError):
+            with active_sessions.active_session_liveness_guard("session-1"):
+                pass
+        assert state_path.read_text(encoding="utf-8") == original
+
+
+@pytest.mark.parametrize(
+    "second_session_id",
+    ("session-a", "session-b"),
+    ids=("exact-duplicate", "conflicting-duplicate"),
+)
+def test_strict_registry_rejects_duplicate_lease_ids(
+    tmp_path, monkeypatch, second_session_id
+):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    state_path = home / "runtime" / "active_sessions.json"
+    active_sessions._write_entries(
+        state_path,
+        [
+            {
+                "lease_id": "duplicate-lease",
+                "session_id": "session-a",
+                "surface": "desktop",
+                "pid": os.getpid(),
+                "track_liveness": True,
+            },
+            {
+                "lease_id": "duplicate-lease",
+                "session_id": second_session_id,
+                "surface": "desktop",
+                "pid": os.getpid(),
+                "track_liveness": True,
+            },
+        ],
+    )
+    original = state_path.read_text(encoding="utf-8")
+
+    with pytest.raises(active_sessions.ActiveSessionRegistryError):
+        active_sessions.active_session_registry_snapshot()
+
+    assert state_path.read_text(encoding="utf-8") == original
+
+
+def test_cap_transfer_does_not_overwrite_registry_corruption(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    state_path = home / "runtime" / "active_sessions.json"
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="cli-old",
+        surface="cli",
+        config={"max_concurrent_sessions": 1},
+    )
+    assert lease is not None and message is None
+
+    corrupt = "{not-json"
+    state_path.write_text(corrupt, encoding="utf-8")
+    assert not active_sessions.transfer_active_session(
+        lease,
+        session_id="cli-new",
+    )
+    assert lease.session_id == "cli-old"
+    assert state_path.read_text(encoding="utf-8") == corrupt
+
+    lease.release()
+    assert lease.released is True
+    assert state_path.read_text(encoding="utf-8") == corrupt
+
+
+def test_liveness_guard_rejects_unknown_pid_state(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    state_path = home / "runtime" / "active_sessions.json"
+    active_sessions._write_entries(
+        state_path,
+        [
+            {
+                "lease_id": "unknown-owner",
+                "session_id": "session-1",
+                "surface": "desktop",
+                "pid": 12345,
+                "track_liveness": True,
+            }
+        ],
+    )
+    monkeypatch.setattr(
+        "gateway.status._pid_exists",
+        lambda _pid: (_ for _ in ()).throw(OSError("pid lookup unavailable")),
+    )
+
+    with pytest.raises(active_sessions.ActiveSessionRegistryError):
+        with active_sessions.active_session_liveness_guard("session-1"):
+            pass
+
+    original = state_path.read_text(encoding="utf-8")
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="cli-cap-session",
+        surface="cli",
+        config={"max_concurrent_sessions": 1},
+    )
+    assert lease is not None and message is None
+    assert lease.enabled is False
+    assert state_path.read_text(encoding="utf-8") == original
+
+
+def test_liveness_release_failure_is_retryable(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="session-1",
+        surface="desktop",
+        config={},
+        track_liveness=True,
+    )
+    assert lease is not None and message is None
+
+    original_write = active_sessions._write_entries
+    monkeypatch.setattr(
+        active_sessions,
+        "_write_entries",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("replace failed")),
+    )
+    with pytest.raises(OSError, match="replace failed"):
+        lease.release()
+    assert lease.released is False
+
+    monkeypatch.setattr(active_sessions, "_write_entries", original_write)
+    lease.release()
+    assert lease.released is True
+    assert active_sessions.active_session_registry_snapshot() == []
+
+
+def test_liveness_transfer_upserts_missing_entry_without_consuming_a_new_slot(
+    tmp_path, monkeypatch
+):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="session-old",
+        surface="desktop",
+        config={"max_concurrent_sessions": 1},
+        track_liveness=True,
+    )
+    assert lease is not None and message is None
+    (home / "runtime" / "active_sessions.json").unlink()
+
+    assert active_sessions.transfer_active_session(lease, session_id="session-new")
+    snapshot = active_sessions.active_session_registry_snapshot()
+    assert [(entry["lease_id"], entry["session_id"]) for entry in snapshot] == [
+        (lease.lease_id, "session-new")
+    ]
+
+    blocked, limit_message = active_sessions.try_acquire_active_session(
+        session_id="session-other",
+        surface="desktop",
+        config={"max_concurrent_sessions": 1},
+        track_liveness=True,
+    )
+    assert blocked is None
+    assert limit_message is not None
+    lease.release()
+
+
+def test_liveness_transfer_write_failure_keeps_old_id_for_retry(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="session-old",
+        surface="desktop",
+        config={},
+        track_liveness=True,
+    )
+    assert lease is not None and message is None
+
+    original_write = active_sessions._write_entries
+    monkeypatch.setattr(
+        active_sessions,
+        "_write_entries",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("replace failed")),
+    )
+    with pytest.raises(OSError, match="replace failed"):
+        active_sessions.transfer_active_session(lease, session_id="session-new")
+    assert lease.session_id == "session-old"
+
+    monkeypatch.setattr(active_sessions, "_write_entries", original_write)
+    assert active_sessions.transfer_active_session(lease, session_id="session-new")
+    assert lease.session_id == "session-new"
+    lease.release()
+
+
+def test_release_wins_against_transfer_waiting_on_same_lease_lock(
+    tmp_path, monkeypatch
+):
+    home = tmp_path / ".hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    lease, message = active_sessions.try_acquire_active_session(
+        session_id="session-old",
+        surface="desktop",
+        config={},
+        track_liveness=True,
+    )
+    assert lease is not None and message is None
+
+    release_wrote = threading.Event()
+    allow_release = threading.Event()
+    transfer_at_lock = threading.Event()
+    original_write = active_sessions._write_entries
+    original_enter = active_sessions._FileLock.__enter__
+
+    def _blocking_write(path, entries):
+        original_write(path, entries)
+        if threading.current_thread().name == "lease-release":
+            release_wrote.set()
+            assert allow_release.wait(timeout=5)
+
+    def _instrumented_enter(lock):
+        if threading.current_thread().name == "lease-transfer":
+            transfer_at_lock.set()
+        return original_enter(lock)
+
+    monkeypatch.setattr(active_sessions, "_write_entries", _blocking_write)
+    monkeypatch.setattr(active_sessions._FileLock, "__enter__", _instrumented_enter)
+    transfer_result: list[bool] = []
+    release_thread = threading.Thread(target=lease.release, name="lease-release")
+    transfer_thread = threading.Thread(
+        target=lambda: transfer_result.append(
+            active_sessions.transfer_active_session(lease, session_id="session-new")
+        ),
+        name="lease-transfer",
+    )
+
+    release_thread.start()
+    assert release_wrote.wait(timeout=5)
+    transfer_thread.start()
+    assert transfer_at_lock.wait(timeout=5)
+    allow_release.set()
+    release_thread.join(timeout=5)
+    transfer_thread.join(timeout=5)
+
+    assert not release_thread.is_alive()
+    assert not transfer_thread.is_alive()
+    assert transfer_result == [False]
+    assert lease.released is True
+    assert active_sessions.active_session_registry_snapshot() == []
+