fix(profiles): never rmtree a profile home a live gateway runtime still owns

Mechanism: delete_profile tombstones the profile and asks the multiplexer to
rescan; the multiplexer answers {"pending": true} after its 5 s wait and keeps
tearing the profile down (adapters, authority retirement, then the gateway.lock
reservation it now holds per served profile via reserve_profile_home). The delete
went straight to rmtree, removing the held gateway.lock inode and the tree under
live writers that recreate it.

Fix: before rmtree, wait (up to 60 s) until exclusive_maintenance([profile_dir])
can take the profile's lock - the same reservation every gateway owner holds -
and abort with "a running gateway still owns this profile; retry" when it cannot.
The rmtree helpers move to hermes_cli/profiles_remove.py with the new wait so
profiles.py (over the file-line cap) shrinks. Class sweep (SWEEP_profile-home-owner.md):
rename_profile's multiplexed arm has the same shape (unroute, then move the dir) and
takes the same wait before the rename.

Test: tests/hermes_cli/test_profile_delete_held_lock.py holds the profile's lock
from a real separate process through process_ownership.reserve; red on r12-base
(the directory is removed while the owner holds it); green: delete completes
once the owner releases and refuses cleanly when it never does; the rename case
is red on base too (the home moves under the owner).

Review: dokterdok N60
(cherry picked from commit 2c4afb72c2c427a772cab73886cc6a85b224f66d)

hermes_cli/profiles.py

@@ -19,6 +19,7 @@ from typing import Dict, List, Optional, Tuple
 from agent.file_safety import HOME_CREDENTIAL_DIRS
 from hermes_cli.archive_safe import archive_root_dirs, make_targz, normalize_archive_parts, safe_extract_targz
 from hermes_cli.home_data_layout import PM_RUNTIME_ROOT_DIRS
+from hermes_cli.profiles_remove import _rmtree_make_writable, _rmtree_with_retry, wait_profile_unowned
 from hermes_constants import (
     LOCAL_RUNTIME_ROOT_DIRS, PROFILE_ID_RE, clear_named_profile_deleted, mark_named_profile_deleted,
     named_profile_has_identity, named_profile_is_deleted, named_profile_is_live,
@@ -1764,44 +1765,6 @@ def _stop_bot_desktop(profile_dir: Path) -> None:
         reset_hermes_home_override(token)
 
 
-def _rmtree_make_writable(func, path, exc):
-    """onexc/onerror handler: add +w on PermissionError so rmtree can proceed. Covers NixOS-
-    style read-only copies where the path itself (0444) or its parent (0555) isn't writable."""
-    # onexc(func, path, exc_instance) on 3.12+; onerror(func, path, exc_info_tuple) on 3.11.
-    if isinstance(exc, tuple):
-        exc = exc[1]
-    if not isinstance(exc, PermissionError):
-        raise
-    for target in (path, os.path.dirname(path)):  # parent needed for unlink/rmdir
-        if target:
-            with contextlib.suppress(OSError):
-                os.chmod(target, os.stat(target).st_mode | stat.S_IWUSR)
-    func(path)
-
-
-def _rmtree_with_retry(profile_dir: Path, onexc_handler) -> None:
-    """``shutil.rmtree`` with a short retry loop: a just-terminated process can leave in-flight
-    writes (SQLite -wal/-shm checkpoints, sandbox temp files) landing after rmtree walked
-    past a directory — ENOTEMPTY on POSIX, transient PermissionError on Windows."""
-    attempts = 3
-    last_exc: OSError | None = None
-    for attempt in range(attempts):
-        try:
-            try:
-                shutil.rmtree(profile_dir, onexc=onexc_handler)
-            except TypeError:  # ``onexc`` is 3.12+; 3.11 has ``onerror``
-                shutil.rmtree(profile_dir, onerror=onexc_handler)
-            return
-        except OSError as e:
-            last_exc = e
-            if not profile_dir.exists():
-                return
-            if attempt < attempts - 1:
-                time.sleep(0.3 * (attempt + 1))
-    if last_exc is not None:
-        raise last_exc
-
-
 def _print_delete_summary(canon: str, profile_dir: Path, gw_running: bool, wrapper_path: Optional[Path]) -> None:
     """Show what ``delete_profile`` is about to remove."""
     model, provider = _read_config_model(profile_dir)
@@ -1925,6 +1888,10 @@ def delete_profile(name: str, yes: bool = False) -> Path:
     # 4. Remove profile directory
     remove_error: Exception | None = None
     try:
+        # A multiplexer still tearing the profile down holds its gateway.lock: never remove a home
+        # a live runtime owns (its writers would recreate the tree under the tombstone).
+        if not wait_profile_unowned(profile_dir):
+            raise RuntimeError("a running gateway still owns this profile; retry the delete once it settles")
         _rmtree_with_retry(profile_dir, _rmtree_make_writable)
         print(f"✓ Removed {profile_dir}")
     except Exception as e:
@@ -2460,6 +2427,8 @@ def rename_profile(old_name: str, new_name: str) -> Path:
     # 2. Rename directory. If the move fails (cross-device EXDEV, permissions, a racing writer),
     # undo the unroute so the profile is never stranded tombstoned-but-present.
     try:
+        if live_mux and not wait_profile_unowned(old_dir):  # same owner wait as delete_profile
+            raise RuntimeError("a running gateway still owns this profile; retry the rename once it settles")
         old_dir.rename(new_dir)
     except Exception:
         if live_mux:

hermes_cli/profiles_remove.py

@@ -0,0 +1,66 @@
+"""Profile directory removal: wait out the live owner, then rmtree with a short retry."""
+from __future__ import annotations
+
+import contextlib
+import os
+from pathlib import Path
+import shutil
+import stat
+import time
+
+# A multiplexer answers a delete's rescan ``pending`` after 5 s and keeps tearing the profile down;
+# its gateway.lock reservation is released last, once the profile's writers are gone.
+OWNER_RELEASE_WAIT_SECONDS = 60.0
+
+
+def wait_profile_unowned(profile_dir: Path, *, timeout: float | None = None) -> bool:
+    """True once no gateway runtime holds ``profile_dir``'s ``gateway.lock`` (taken and released
+    through ``exclusive_maintenance``); False when it is still held after ``timeout`` seconds."""
+    from gateway.runtime_ownership import OwnershipConflict, exclusive_maintenance
+    deadline = time.monotonic() + (OWNER_RELEASE_WAIT_SECONDS if timeout is None else timeout)
+    while True:
+        try:
+            with exclusive_maintenance([profile_dir]):
+                return True
+        except OwnershipConflict:
+            if time.monotonic() >= deadline:
+                return False
+            time.sleep(0.25)
+
+
+def _rmtree_make_writable(func, path, exc):
+    """onexc/onerror handler: add +w on PermissionError so rmtree can proceed. Covers NixOS-
+    style read-only copies where the path itself (0444) or its parent (0555) isn't writable."""
+    # onexc(func, path, exc_instance) on 3.12+; onerror(func, path, exc_info_tuple) on 3.11.
+    if isinstance(exc, tuple):
+        exc = exc[1]
+    if not isinstance(exc, PermissionError):
+        raise
+    for target in (path, os.path.dirname(path)):  # parent needed for unlink/rmdir
+        if target:
+            with contextlib.suppress(OSError):
+                os.chmod(target, os.stat(target).st_mode | stat.S_IWUSR)
+    func(path)
+
+
+def _rmtree_with_retry(profile_dir: Path, onexc_handler) -> None:
+    """``shutil.rmtree`` with a short retry loop: a just-terminated process can leave in-flight
+    writes (SQLite -wal/-shm checkpoints, sandbox temp files) landing after rmtree walked
+    past a directory — ENOTEMPTY on POSIX, transient PermissionError on Windows."""
+    attempts = 3
+    last_exc: OSError | None = None
+    for attempt in range(attempts):
+        try:
+            try:
+                shutil.rmtree(profile_dir, onexc=onexc_handler)
+            except TypeError:  # ``onexc`` is 3.12+; 3.11 has ``onerror``
+                shutil.rmtree(profile_dir, onerror=onexc_handler)
+            return
+        except OSError as e:
+            last_exc = e
+            if not profile_dir.exists():
+                return
+            if attempt < attempts - 1:
+                time.sleep(0.3 * (attempt + 1))
+    if last_exc is not None:
+        raise last_exc

tests/hermes_cli/test_profile_delete_held_lock.py

@@ -0,0 +1,85 @@
+"""A profile delete (or multiplexed rename) never removes/moves a home whose gateway.lock a live runtime still holds (N60).
+
+A multiplexer answers the delete's rescan ``pending`` after 5 s and keeps tearing the profile down;
+it releases the profile's ``gateway.lock`` reservation last. The holder here is a real separate
+process reserving through the same primitive the multiplexer uses (``process_ownership.reserve``).
+"""
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+from hermes_cli import profiles, profiles_remove
+
+_HOLDER = (
+    "import sys\nfrom pathlib import Path\nfrom gateway.runtime_ownership import process_ownership\n"
+    "home = Path(sys.argv[1])\nprocess_ownership.reserve([home])\nprint('held', flush=True)\n"
+    "sys.stdin.readline()\nprocess_ownership.release(home)\n")
+
+
+@pytest.fixture
+def profile(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    for name in ("_cleanup_gateway_service", "_maybe_unregister_gateway_service", "_stop_profile_backends",
+                 "_notify_multiplexer", "_stop_gateway_process", "_maybe_register_gateway_service"):
+        monkeypatch.setattr(profiles, name, lambda *_a, **_k: None)
+    return profiles.create_profile("owned-delete", no_alias=True)
+
+
+def _hold(profile_dir):
+    holder = subprocess.Popen([sys.executable, "-c", _HOLDER, str(profile_dir)], stdin=subprocess.PIPE,
+                              stdout=subprocess.PIPE, text=True, cwd=str(Path(profiles.__file__).parent.parent))
+    assert holder.stdout.readline().strip() == "held"
+    return holder
+
+
+def _release(holder):
+    holder.stdin.write("\n")
+    holder.stdin.flush()
+    assert holder.wait(timeout=10) == 0
+
+
+# Every site that removes or moves a profile home after unrouting it from a live multiplexer.
+CASES = [("delete", True), ("delete", False), ("rename", False)]
+
+
+@pytest.mark.parametrize(("op", "released"), CASES, ids=["delete-owner-releases", "delete-owner-keeps-it",
+                                                       "rename-owner-keeps-it"])
+def test_profile_home_waits_for_the_live_owner_to_release_it(profile, monkeypatch, op, released):
+    monkeypatch.setattr(profiles_remove, "OWNER_RELEASE_WAIT_SECONDS", 30.0 if released else 1.5)
+    monkeypatch.setattr(profiles, "_live_default_multiplexer", lambda: op == "rename")  # rename's unroute arm
+    holder = _hold(profile)
+    outcome = {}
+
+    def delete():
+        try:
+            if op == "rename":
+                profiles.rename_profile("owned-delete", "owned-renamed")
+            else:
+                profiles.delete_profile("owned-delete", yes=True)
+            outcome["ok"] = True
+        except RuntimeError as exc:
+            outcome["error"] = str(exc)
+
+    worker = threading.Thread(target=delete)
+    try:
+        worker.start()
+        worker.join(timeout=1.0)
+        assert profile.is_dir(), "the profile was removed while a live gateway still owned it"
+        if released:
+            _release(holder)
+        worker.join(timeout=30)
+        assert not worker.is_alive()
+    finally:
+        if holder.poll() is None:
+            _release(holder)
+    if released:
+        assert outcome == {"ok": True} and not profile.exists()
+    else:
+        assert "still owns this profile" in outcome["error"] and profile.is_dir()