fix(local-runtime): contain Windows runtimes and safely stop orphans

hermes_cli/local_runtime/endpoint.py

@@ -32,27 +32,23 @@ def _pid_alive(pid: int) -> bool:
 
 
 def _state_endpoint() -> dict | None:
-    from hermes_cli.local_runtime.supervisor import state_path
+    from hermes_cli.local_runtime.recovery import is_modern, read_state, recorded_process
 
-    path = state_path()
-    if not path.exists():
-        return None
-    try:
-        state = json.loads(path.read_text(encoding="utf-8"))
-    except (json.JSONDecodeError, OSError):
-        return None
+    state = read_state()
     base_url = state.get("base_url", "")
-    if not base_url:
-        return None
-    # Ownership proof: on the stable port a SECOND install (different HERMES_HOME) can own
-    # 127.0.0.1:18434 with a different api key while this install's state file still points
-    # there. /health is public and answers 200 for ANYONE's server — trusting it sent every
-    # request at a server that 401s our key, silently — so the recorded supervisor pid is the
-    # ONLY tiebreaker: a live pid is ours (healthy, or STARTING — state is written at spawn, and
-    # readiness probes racing the boot must see a configured provider, not missing credentials);
-    # a dead pid is a crashed-without-cleanup leftover, ignored so requests don't blackhole.
-    if not _pid_alive(int(state.get("pid") or 0)):
+    if not isinstance(base_url, str) or not base_url:
         return None
+    if is_modern(state):
+        if recorded_process(state) is None:
+            return None
+    else:
+        # Preserve the legacy endpoint shape, with malformed PID values rejected.
+        try:
+            pid = state.get("pid")
+            if isinstance(pid, bool) or not _pid_alive(int(pid or 0)):
+                return None
+        except (TypeError, ValueError, OverflowError):
+            return None
     return {"base_url": base_url, "api_key": state.get("api_key", "")}
 
 

hermes_cli/local_runtime/processes.py

@@ -0,0 +1,121 @@
+"""Contain each managed Windows router tree without adopting the owner process."""
+
+from __future__ import annotations
+
+import ctypes
+from ctypes import wintypes
+import subprocess
+import sys
+import threading
+
+import psutil
+
+
+class _BasicLimits(ctypes.Structure):
+    _fields_ = [
+        ("PerProcessUserTimeLimit", ctypes.c_longlong),
+        ("PerJobUserTimeLimit", ctypes.c_longlong),
+        ("LimitFlags", wintypes.DWORD),
+        ("MinimumWorkingSetSize", ctypes.c_size_t),
+        ("MaximumWorkingSetSize", ctypes.c_size_t),
+        ("ActiveProcessLimit", wintypes.DWORD),
+        ("Affinity", ctypes.c_size_t),
+        ("PriorityClass", wintypes.DWORD),
+        ("SchedulingClass", wintypes.DWORD),
+    ]
+
+
+class _IoCounters(ctypes.Structure):
+    _fields_ = [(name, ctypes.c_ulonglong) for name in (
+        "ReadOperationCount", "WriteOperationCount", "OtherOperationCount",
+        "ReadTransferCount", "WriteTransferCount", "OtherTransferCount",
+    )]
+
+
+class _ExtendedLimits(ctypes.Structure):
+    _fields_ = [
+        ("BasicLimitInformation", _BasicLimits),
+        ("IoInfo", _IoCounters),
+        ("ProcessMemoryLimit", ctypes.c_size_t),
+        ("JobMemoryLimit", ctypes.c_size_t),
+        ("PeakProcessMemoryUsed", ctypes.c_size_t),
+        ("PeakJobMemoryUsed", ctypes.c_size_t),
+    ]
+
+
+class _WindowsJob:
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._api = ctypes.WinDLL("kernel32", use_last_error=True)
+        for name, args, result in (
+            ("CreateJobObjectW", [ctypes.c_void_p, wintypes.LPCWSTR], wintypes.HANDLE),
+            ("SetInformationJobObject", [wintypes.HANDLE, ctypes.c_int,
+                                         ctypes.c_void_p, wintypes.DWORD], wintypes.BOOL),
+            ("AssignProcessToJobObject", [wintypes.HANDLE, wintypes.HANDLE], wintypes.BOOL),
+            ("CloseHandle", [wintypes.HANDLE], wintypes.BOOL),
+        ):
+            fn = getattr(self._api, name)
+            fn.argtypes = args
+            fn.restype = result
+        # NULL security attributes create a non-inheritable, unnamed owner handle.
+        self._handle = self._api.CreateJobObjectW(None, None)
+        if not self._handle:
+            raise ctypes.WinError(ctypes.get_last_error())
+        try:
+            limits = _ExtendedLimits()
+            # Neither BREAKAWAY_OK nor SILENT_BREAKAWAY_OK: descendants stay contained.
+            limits.BasicLimitInformation.LimitFlags = 0x2000  # KILL_ON_JOB_CLOSE
+            if not self._api.SetInformationJobObject(
+                    self._handle, 9, ctypes.byref(limits), ctypes.sizeof(limits)):
+                raise ctypes.WinError(ctypes.get_last_error())
+        except BaseException:
+            self.close()
+            raise
+
+    def assign(self, proc: subprocess.Popen) -> None:
+        # Popen retains the original process handle, avoiding a PID-reuse race.
+        if not self._api.AssignProcessToJobObject(self._handle, int(proc._handle)):
+            raise ctypes.WinError(ctypes.get_last_error())
+
+    def close(self) -> None:
+        """Terminate the contained tree; repeated closes are harmless."""
+        with self._lock:
+            if self._handle is not None:
+                if not self._api.CloseHandle(self._handle):
+                    raise ctypes.WinError(ctypes.get_last_error())
+                self._handle = None
+
+
+def spawn_server(cmd, **kwargs) -> tuple[subprocess.Popen, _WindowsJob | None]:
+    """Start a router, returning its process and an owner-held containment handle.
+
+    Keep the job until shutdown and call close() to terminate the entire tree.
+    Windows closes it automatically if the owner dies. Other hosts retain Popen's
+    ordinary behavior. Assignment happens before the child's first instruction.
+    """
+    if sys.platform != "win32":
+        return subprocess.Popen(cmd, **kwargs), None
+    job = _WindowsJob()
+    proc = None
+    try:
+        kwargs["creationflags"] = kwargs.get("creationflags", 0) | 0x00000004  # CREATE_SUSPENDED
+        proc = subprocess.Popen(cmd, **kwargs)
+        job.assign(proc)
+        psutil.Process(proc.pid).resume()
+        return proc, job
+    except BaseException:
+        try:
+            if proc is not None:
+                # Assignment may have failed: closing an empty job is not enough.
+                proc.kill()
+                proc.wait(timeout=10)
+        finally:
+            try:
+                job.close()
+            finally:
+                if proc is not None:
+                    for stream in (proc.stdin, proc.stdout, proc.stderr):
+                        if stream is not None:
+                            stream.close()
+                    proc._handle.Close()
+        raise

hermes_cli/local_runtime/recovery.py

@@ -0,0 +1,159 @@
+"""Conservative recovery of the one router recorded by this managed runtime."""
+from __future__ import annotations
+
+import json
+import logging
+import math
+import os
+from pathlib import Path
+
+import psutil
+
+logger = logging.getLogger(__name__)
+
+
+def read_state() -> dict:
+    from hermes_cli.local_runtime.supervisor import state_path
+
+    try:
+        state = json.loads(state_path().read_text(encoding="utf-8"))
+        return state if isinstance(state, dict) else {}
+    except (OSError, ValueError):
+        return {}
+
+
+_MODERN_FIELDS = ("create_time", "executable", "owner_pid", "owner_create_time")
+
+
+def is_modern(state: dict) -> bool:
+    return any(key in state for key in _MODERN_FIELDS)
+
+
+def _valid_pid(value) -> bool:
+    return type(value) is int and value > 0
+
+
+def _valid_birth(value) -> bool:
+    return type(value) in (int, float) and math.isfinite(value) and value > 0
+
+
+def recorded_process(state: dict):
+    """Match a modern process incarnation, never just its reusable PID."""
+    try:
+        pid, created = state["pid"], state["create_time"]
+        owner_pid, owner_created = state["owner_pid"], state["owner_create_time"]
+        exe = state["executable"]
+        if (not _valid_pid(pid) or not _valid_birth(created)
+                or not _valid_pid(owner_pid) or not _valid_birth(owner_created)
+                or owner_created > created or not isinstance(exe, str) or not exe):
+            return None
+        proc = psutil.Process(pid)
+        if (not proc.is_running() or proc.create_time() != created
+                or Path(proc.exe()) != Path(exe)):
+            return None
+        # Windows retains the original parent PID; POSIX reparents orphans.
+        if os.name == "nt" and proc.ppid() != owner_pid:
+            return None
+        parent = proc.parent()
+        if parent is not None:
+            if parent.pid == owner_pid:
+                if parent.create_time() != owner_created:
+                    return None
+            elif os.name == "nt" or not _owner_is_dead(state):
+                return None  # POSIX may reparent a router whose recorded owner exited.
+        return proc
+    except (KeyError, TypeError, ValueError, OverflowError, OSError, psutil.Error):
+        return None
+
+
+def _owner_is_dead(state: dict) -> bool:
+    try:
+        pid, created = state["owner_pid"], state["owner_create_time"]
+        if not _valid_pid(pid) or not _valid_birth(created):
+            return False
+        try:
+            owner = psutil.Process(pid)
+            # A newer incarnation proves the recorded owner has exited.
+            return owner.create_time() > created or not owner.is_running()
+        except psutil.NoSuchProcess:
+            return True
+    except (KeyError, TypeError, ValueError, OverflowError, OSError, psutil.Error):
+        return False
+
+
+def _legacy_orphan_process(state: dict):
+    """Older state lacks birth times: require the exact installed binary and launch arguments."""
+    from urllib.parse import urlsplit
+    from hermes_cli.local_runtime.bootstrap import models_dir
+    from hermes_cli.local_runtime.supervisor import state_path
+
+    # A damaged new record must not fall back to weaker legacy evidence.
+    if os.name != "nt" or is_modern(state):
+        return None
+    try:
+        if not _valid_pid(state["pid"]):
+            return None
+        proc = psutil.Process(state["pid"])
+        root = state_path().parent
+        exe = Path(proc.exe())
+        relative = exe.relative_to(root)
+        if len(relative.parts) != 3 or exe.name.lower() != "llama-server.exe":
+            return None
+        if proc.parent() is not None or proc.ppid() <= 0:
+            return None
+        if proc.create_time() > state_path().stat().st_mtime:
+            return None  # the PID was reused after this record was written
+        url = urlsplit(state["base_url"])
+        if url.scheme != "http" or url.hostname != "127.0.0.1" or not url.port or not state["api_key"]:
+            return None
+        argv = proc.cmdline()
+
+        def value(flag):
+            if argv.count(flag) != 1:
+                return None
+            index = argv.index(flag) + 1
+            return argv[index] if index < len(argv) else None
+
+        if (value("--host") != "127.0.0.1" or value("--port") != str(url.port)
+                or value("--api-key") != state["api_key"]):
+            return None
+        preset, directory = value("--models-preset"), value("--models-dir")
+        if not ((preset and Path(preset) == root / "presets.ini" and directory is None)
+                or (directory and Path(directory) == models_dir() and preset is None)):
+            return None
+        return proc
+    except (KeyError, TypeError, ValueError, OSError, psutil.Error):
+        return None
+
+
+def stop_recorded_orphan() -> bool:
+    """Explicit user stop only. Refuse uncertain identity or a living recorded owner."""
+    from hermes_cli.local_runtime.supervisor import LlamaServerSupervisor, state_path
+
+    try:
+        state = json.loads(state_path().read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return True  # already stopped and no record
+    except (OSError, ValueError):
+        return False
+    if not isinstance(state, dict) or not _valid_pid(state.get("pid")):
+        return False
+    try:
+        if not psutil.pid_exists(state["pid"]):
+            return True  # keep the dead record; endpoint resolution rejects it
+        if is_modern(state):
+            proc = recorded_process(state)
+            if proc is None or not _owner_is_dead(state):
+                return False
+        else:
+            proc = _legacy_orphan_process(state)
+        if proc is None or read_state() != state or not proc.is_running():
+            return False
+        # Retain the psutil incarnation object: destructive methods guard PID reuse.
+        LlamaServerSupervisor._terminate_tree(proc, verified_root=True)
+        proc.wait(timeout=5)
+        logger.info("stopped orphaned managed llama-server pid=%s", proc.pid)
+        return True
+    except (OSError, psutil.Error):
+        logger.warning("could not verify or stop recorded managed llama-server")
+        return False

hermes_cli/local_runtime/supervisor.py

@@ -15,6 +15,7 @@ import logging
 import secrets
 import socket
 import subprocess
+import sys
 import threading
 import time
 import urllib.error
@@ -22,6 +23,7 @@ import urllib.request
 from pathlib import Path
 
 from hermes_cli.local_runtime.binaries import server_binary, runtimes_root
+from hermes_cli.local_runtime.processes import spawn_server
 
 logger = logging.getLogger(__name__)
 
@@ -113,9 +115,13 @@ class LlamaServerSupervisor:
         self.log_path = log_path or (self.models_dir.parent / "logs" / "llama-server.log")
         self.preset_path = preset_path
         self.proc: subprocess.Popen | None = None
+        self._job = None
         self.primary_model: str | None = None
         self._restarts = 0
         self._stopping = False
+        self._stop_event = threading.Event()
+        self._lifecycle_lock = threading.RLock()
+        self._state: dict | None = None
         self._watchdog: threading.Thread | None = None
         self._log_handle = None
         self._idle_since: dict[str, float] = {}
@@ -177,8 +183,8 @@ class LlamaServerSupervisor:
         self._log_handle.write(f"\n# spawn: {cmd}\n")
         self._log_handle.flush()
         # list-args, never a shell: spaced paths (user homes) must survive.
-        self.proc = subprocess.Popen(cmd, stdout=self._log_handle,
-                                     stderr=subprocess.STDOUT, cwd=str(exe.parent))
+        self.proc, self._job = spawn_server(cmd, stdout=self._log_handle,
+                                             stderr=subprocess.STDOUT, cwd=str(exe.parent))
         logger.info("llama-server router spawned pid=%s port=%s", self.proc.pid, self.port)
         # State goes down at SPAWN, not after health: endpoint resolution treats a
         # live-pid-but-not-yet-healthy server as "starting" rather than "unconfigured", so a
@@ -186,22 +192,33 @@ class LlamaServerSupervisor:
         self._write_state()
 
     def start(self, timeout_s: int = 120) -> None:
-        self._stopping = False
-        self._spawn()
+        with self._lifecycle_lock:
+            self._stopping = False
+            self._stop_event.clear()
+            self._spawn()
         self._wait_health(timeout_s)
-        self._write_state()
         self._watchdog = threading.Thread(target=self._watch, daemon=True, name="llamacpp-supervisor")
         self._watchdog.start()
 
     def _write_state(self) -> None:
+        import os
+        import psutil
+        from utils import atomic_json_write
+
+        proc = psutil.Process(self.proc.pid)
+        self._state = {"base_url": self.base_url, "api_key": self.api_key,
+                       "pid": proc.pid, "create_time": proc.create_time(),
+                       "executable": proc.exe(), "owner_pid": os.getpid(),
+                       "owner_create_time": psutil.Process().create_time()}
         path = state_path()
         path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text(json.dumps({"base_url": self.base_url, "api_key": self.api_key,
-                                    "pid": self.proc.pid if self.proc else None}), encoding="utf-8")
+        atomic_json_write(path, self._state, mode=0o600)
 
     def _wait_health(self, timeout_s: int) -> None:
         deadline = time.monotonic() + timeout_s
         while time.monotonic() < deadline:
+            if self._stop_event.is_set():
+                raise RuntimeError("llama-server startup cancelled")
             if self.proc and self.proc.poll() is not None:
                 raise RuntimeError(f"llama-server exited rc={self.proc.returncode} during startup "
                                    f"(log: {self.log_path})")
@@ -226,11 +243,15 @@ class LlamaServerSupervisor:
                 return
             backoff = _RESTART_BACKOFF_S[min(self._restarts, len(_RESTART_BACKOFF_S) - 1)]
             logger.warning("llama-server exited rc=%s; restart #%s in %ss", rc, self._restarts + 1, backoff)
-            time.sleep(backoff)
+            if self._stop_event.wait(backoff):
+                return
             self._restarts += 1
             try:
-                self._reap_orphaned_children()
-                self._spawn()
+                with self._lifecycle_lock:
+                    if self._stopping:
+                        return
+                    self._reap_orphaned_children()
+                    self._spawn()
                 self._wait_health(120)
                 if self.primary_model:
                     self.ensure_model_ready(self.primary_model)
@@ -238,16 +259,23 @@ class LlamaServerSupervisor:
                 logger.error("llama-server restart failed: %s", exc)
 
     def stop(self) -> None:
-        self._stopping = True
-        state_path().unlink(missing_ok=True)
-        if self.proc and self.proc.poll() is None:
-            self._terminate_tree(self.proc)
-        if self._log_handle:
-            self._log_handle.close()
-            self._log_handle = None
+        with self._lifecycle_lock:
+            self._stopping = True
+            self._stop_event.set()
+            try:
+                if self.proc and self.proc.poll() is None:
+                    self._terminate_tree(self.proc)
+            finally:
+                if self._job is not None:
+                    self._job.close()
+                    self._job = None
+            # Retain state: deleting it could race a replacement publication.
+            if self._log_handle:
+                self._log_handle.close()
+                self._log_handle = None
 
     @staticmethod
-    def _terminate_tree(proc: subprocess.Popen) -> None:
+    def _terminate_tree(proc: subprocess.Popen, *, verified_root: bool = False) -> None:
         """Terminate the router AND its model children.
 
         Each child holds gigabytes of VRAM; terminating only the router (TerminateProcess on
@@ -255,19 +283,30 @@ class LlamaServerSupervisor:
         FIRST (the parent must be alive to walk them), terminate all, escalate to kill.
         """
         children: list = []
-        with suppress(Exception):  # no psutil view; still stop the router
+        timeouts = (subprocess.TimeoutExpired,)
+        with suppress(ImportError):
             import psutil
 
-            children = psutil.Process(proc.pid).children(recursive=True)
-        proc.terminate()
-        for child in children:
-            _quiet(child.terminate)
+            timeouts += (psutil.TimeoutExpired,)
+            if verified_root:
+                # Recovery retains the birth identity; never rebuild it from a PID.
+                children = proc.children(recursive=True)
+                if not proc.is_running():
+                    raise psutil.NoSuchProcess(proc.pid)
+            else:
+                with suppress(psutil.Error):
+                    children = psutil.Process(proc.pid).children(recursive=True)
         try:
-            proc.wait(timeout=15)
-        except subprocess.TimeoutExpired:
-            proc.kill()
-        for child in children:
-            _quiet(lambda: child.is_running() and child.kill())
+            for child in children:
+                _quiet(child.terminate)
+            proc.terminate()
+            try:
+                proc.wait(timeout=15)
+            except timeouts:
+                proc.kill()
+        finally:
+            for child in children:
+                _quiet(lambda: child.is_running() and child.kill())
 
     def _reap_orphaned_children(self) -> None:
         """Kill model children orphaned by a router crash, before respawn.
@@ -276,6 +315,12 @@ class LlamaServerSupervisor:
         llama-server binary whose parent is gone is an orphan of a previous router. Its VRAM must
         come back before the new router loads models next to the ghosts.
         """
+        if self._job is not None:
+            self._job.close()
+            self._job = None
+            return
+        if sys.platform == "win32":
+            return  # Unrecorded processes are not ours merely because the binary matches.
         try:
             import psutil
 

hermes_cli/web_routers/local_models.py

@@ -773,21 +773,19 @@ async def local_models_quickstart(body: QuickstartBody):
 
 # ── server lifecycle: turn the engine on/off ─────────────────
 def _terminate_state_pid() -> None:
-    """Server owned by another process (or an orphan): terminate via the state file's pid, then clear the state."""
-    import psutil  # type: ignore
+    """Explicit recovery, never raw-PID termination of another live owner."""
+    from hermes_cli.local_runtime.recovery import stop_recorded_orphan
 
-    state = json.loads(supervisor.state_path().read_text(encoding="utf-8"))
-    pid = int(state.get("pid") or 0)
-    if pid > 0 and psutil.pid_exists(pid):
-        psutil.Process(pid).terminate()
-    supervisor.state_path().unlink(missing_ok=True)
+    if not stop_recorded_orphan():
+        raise HTTPException(status_code=409, detail=(
+            "Another Hermes process owns this server, or its ownership could not be verified"))
 
 
 def _stop_server() -> None:
     if bootstrap.get_supervisor() is not None:
         bootstrap.shutdown_local_runtime()
-    elif _state_endpoint() is not None:
-        _quiet(_terminate_state_pid, None)  # best-effort
+    else:
+        _terminate_state_pid()
     _set_runtime_enabled(False)
 
 
@@ -805,8 +803,12 @@ async def local_models_server(body: ServerActionBody):
     action = (body.action or "").strip().lower()
     if action not in _SERVER_ACTIONS:
         raise HTTPException(status_code=400, detail="action must be 'stop' or 'start'")
-    with _http_error(502):
+    try:
         await asyncio.to_thread(_SERVER_ACTIONS[action])
+    except HTTPException:
+        raise
+    except Exception as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
     return {"ok": True, "action": action}
 
 

tests/hermes_cli/test_local_abandoned_requests.py

@@ -122,7 +122,9 @@ def test_terminate_tree_terminates_children_too(monkeypatch):
             assert recursive is True
             return children
 
-    fake_psutil = types.SimpleNamespace(Process=_FakeParentProc)
+    import psutil
+    fake_psutil = types.SimpleNamespace(Process=_FakeParentProc,
+                                        TimeoutExpired=psutil.TimeoutExpired, Error=psutil.Error)
     monkeypatch.setitem(__import__("sys").modules, "psutil", fake_psutil)
 
     class _FakeRouter:
@@ -208,6 +210,8 @@ def test_reap_orphans_kills_only_our_parentless_binaries(tmp_path, monkeypatch):
     sup = LlamaServerSupervisor.__new__(LlamaServerSupervisor)
     sup.install_dir = tmp_path
     sup.proc = None
+    sup._job = None
     sup._reap_orphaned_children()
 
-    assert reaped == [300], f"reaped {reaped}; wanted only the orphan (300)"
+    expected = [] if __import__("sys").platform == "win32" else [300]
+    assert reaped == expected  # Windows uses retained job handles, never binary scans.

tests/hermes_cli/test_local_preset_admission.py

@@ -76,7 +76,8 @@ def test_supervisor_with_presets_does_not_scan_unadmitted_files(tmp_path, monkey
     ini.write_text("[allowed]\nmodel = allowed.gguf\nctx-size = 65536\n")
     calls = []
     monkeypatch.setattr(supervisor, "server_binary", lambda p: Path("llama-server"))
-    monkeypatch.setattr(supervisor.subprocess, "Popen", lambda cmd, **kw: calls.append(cmd) or SimpleNamespace(pid=123))
+    monkeypatch.setattr(supervisor, "spawn_server", lambda cmd, **kw: (calls.append(cmd) or SimpleNamespace(pid=123), None))
+    monkeypatch.setattr(supervisor.LlamaServerSupervisor, "_write_state", lambda self: None)
     sup = supervisor.LlamaServerSupervisor(tmp_path, tmp_path, port=1234, preset_path=ini)
     try:
         sup._spawn()

tests/hermes_cli/test_local_runtime_processes.py

@@ -0,0 +1,235 @@
+"""Native containment tests; all processes and homes are disposable."""
+
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+
+import psutil
+import pytest
+
+
+_SCRIPT = r'''
+import json, os, subprocess, sys, time
+from pathlib import Path
+import psutil
+root = Path(sys.argv[1])
+role = sys.argv[2]
+def record(name):
+    p = psutil.Process()
+    target = root / (name + '.json')
+    temp = target.with_suffix('.tmp')
+    temp.write_text(json.dumps({'pid': p.pid, 'created': p.create_time()}))
+    temp.replace(target)
+record(role)
+if role == 'owner':
+    from hermes_cli.local_runtime.processes import spawn_server
+    proc, job = spawn_server([sys.executable, __file__, str(root), 'router'],
+                             close_fds=False)
+    (root / 'ready').write_text('ready')
+    while not (root / 'stop').exists():
+        time.sleep(.02)
+    job.close()
+    job.close()
+    proc.wait(timeout=10)
+    (root / 'closed').write_text('closed')
+    while not (root / 'exit').exists():
+        time.sleep(.02)
+elif role == 'router':
+    subprocess.Popen([sys.executable, __file__, str(root), 'grandchild'])
+    time.sleep(90)
+else:
+    time.sleep(90)
+'''
+
+
+def _wait(predicate, timeout=10):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(.03)
+    return bool(predicate())
+
+
+def _read(path):
+    assert _wait(path.exists), f"missing child receipt: {path}"
+    return json.loads(path.read_text())
+
+
+def _alive(identity):
+    try:
+        proc = psutil.Process(identity['pid'])
+        if proc.create_time() != identity['created']:
+            return False
+        try:
+            proc.wait(timeout=0)
+            return False
+        except psutil.TimeoutExpired:
+            return True
+    except psutil.NoSuchProcess:
+        return False
+
+
+def _kill(identity):
+    if _alive(identity):
+        proc = psutil.Process(identity['pid'])
+        proc.kill()
+        proc.wait(timeout=10)
+
+
+@pytest.mark.windows_only
+@pytest.mark.parametrize('stop_mode', ['graceful', 'abrupt'])
+@pytest.mark.parametrize('nested', [False, True])
+def test_owner_exit_kills_router_tree_not_external(tmp_path, stop_mode, nested):
+    script = tmp_path / 'disposable server.py'
+    script.write_text(_SCRIPT)
+    env = dict(os.environ, HERMES_HOME=str(tmp_path / 'home'),
+               PYTHONPATH=str(Path(__file__).resolve().parents[2]))
+    launchers = []
+    outer_job = None
+    with (tmp_path / 'children.log').open('w') as log:
+        try:
+            for role in ('control', 'owner'):
+                cmd = [sys.executable, str(script), str(tmp_path), role]
+                if nested and role == 'owner':
+                    from hermes_cli.local_runtime.processes import spawn_server
+                    launcher, outer_job = spawn_server(cmd, env=env, stdout=log, stderr=log)
+                else:
+                    launcher = subprocess.Popen(cmd, env=env, stdout=log, stderr=log)
+                launchers.append(launcher)
+            owner = _read(tmp_path / 'owner.json')
+            router = _read(tmp_path / 'router.json')
+            grandchild = _read(tmp_path / 'grandchild.json')
+            control = _read(tmp_path / 'control.json')
+            assert _wait((tmp_path / 'ready').exists)
+            assert all(_alive(i) for i in (owner, router, grandchild, control))
+            print('disposable identities:', stop_mode, owner, router, grandchild, control)
+            if stop_mode == 'graceful':
+                (tmp_path / 'stop').write_text('stop')
+                assert _wait((tmp_path / 'closed').exists), 'close killed the owner itself'
+                assert _alive(owner)
+                assert _wait(lambda: not _alive(router) and not _alive(grandchild))
+                assert _alive(control)
+                (tmp_path / 'exit').write_text('exit')
+            else:
+                _kill(owner)  # The venv launcher is not necessarily the owner.
+            assert _wait(lambda: not _alive(owner))
+            assert _wait(lambda: not _alive(router) and not _alive(grandchild)), (
+                'uncontained router/grandchild survived owner exit', router, grandchild)
+            assert _alive(control), 'unrelated external control was terminated'
+        finally:
+            if outer_job is not None:
+                outer_job.close()
+            # Stop only identities/descendants created by this test, including red runs.
+            descendants = []
+            for launcher in launchers:
+                try:
+                    descendants.extend(psutil.Process(launcher.pid).children(recursive=True))
+                except psutil.NoSuchProcess:
+                    pass
+            for path in tmp_path.glob('*.json'):
+                _kill(json.loads(path.read_text()))
+            for proc in reversed(descendants):
+                _kill({'pid': proc.pid, 'created': proc.create_time()})
+            for launcher in launchers:
+                if launcher.poll() is None:
+                    launcher.kill()
+                launcher.wait(timeout=10)
+
+
+@pytest.mark.windows_only
+@pytest.mark.parametrize('failure', ['assign', 'resume', 'popen', 'configure'])
+def test_failed_setup_never_runs_child_and_releases_handles(tmp_path, monkeypatch, failure):
+    import ctypes
+    from ctypes import wintypes
+    from hermes_cli.local_runtime import processes
+
+    marker = tmp_path / 'child executed'
+    jobs, children, handles = [], [], []
+    real_init = processes._WindowsJob.__init__
+    real_assign = processes._WindowsJob.assign
+
+    def track_job(job):
+        jobs.append(job)
+        real_init(job)
+        handles.append(job._handle)
+
+    if failure == 'configure':
+        real_dll = ctypes.WinDLL
+
+        def failed_config_dll(*args, **kwargs):
+            api = real_dll(*args, **kwargs)
+
+            def fail_config(handle, *args):
+                handles.append(handle)
+                raise KeyboardInterrupt('injected cancellation configuring job')
+
+            api.SetInformationJobObject = fail_config
+            return api
+
+        monkeypatch.setattr(ctypes, 'WinDLL', failed_config_dll)
+
+    def assign(job, proc):
+        children.append(proc)
+        assert psutil.Process(proc.pid).status() == psutil.STATUS_STOPPED
+        assert not marker.exists()
+        # Query the actual kernel object, not implementation source/constants.
+        limits = processes._ExtendedLimits()
+        query = job._api.QueryInformationJobObject
+        query.argtypes = [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p,
+                          wintypes.DWORD, ctypes.c_void_p]
+        query.restype = wintypes.BOOL
+        assert query(job._handle, 9, ctypes.byref(limits), ctypes.sizeof(limits), None)
+        flags = limits.BasicLimitInformation.LimitFlags
+        assert flags & 0x2000  # KILL_ON_JOB_CLOSE
+        assert not flags & (0x800 | 0x1000)  # Neither breakaway limit.
+        info = job._api.GetHandleInformation
+        info.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)]
+        info.restype = wintypes.BOOL
+        inherited = wintypes.DWORD()
+        assert info(job._handle, ctypes.byref(inherited))
+        assert not inherited.value & 1  # HANDLE_FLAG_INHERIT
+        if failure == 'assign':
+            # Exercise a real WinAPI rejection while the child is suspended.
+            assert not job._api.AssignProcessToJobObject(job._handle, None)
+            raise ctypes.WinError(ctypes.get_last_error())
+        real_assign(job, proc)
+
+    def fail_resume(self):
+        raise KeyboardInterrupt('injected cancellation before resume')
+
+    monkeypatch.setattr(processes._WindowsJob, '__init__', track_job)
+    monkeypatch.setattr(processes._WindowsJob, 'assign', assign)
+    monkeypatch.setattr(psutil.Process, 'resume', fail_resume)
+    cmd = ([str(tmp_path / 'missing.exe')] if failure == 'popen' else
+           [sys.executable, '-c', 'from pathlib import Path; import sys; '
+            'Path(sys.argv[1]).write_text("ran")', str(marker)])
+    try:
+        with pytest.raises((OSError, KeyboardInterrupt)):
+            processes.spawn_server(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+        assert not marker.exists()
+        assert jobs and jobs[0]._handle is None
+        for proc in children:
+            assert proc.poll() is not None, 'failed setup left a live suspended child'
+            assert all(stream.closed for stream in (proc.stdin, proc.stdout, proc.stderr))
+            assert proc._handle.closed, 'failed setup leaked the Popen process handle'
+        api = ctypes.WinDLL('kernel32', use_last_error=True)
+        api.GetHandleInformation.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)]
+        api.GetHandleInformation.restype = wintypes.BOOL
+        for handle in handles:
+            flags = wintypes.DWORD()
+            assert not api.GetHandleInformation(handle, ctypes.byref(flags))
+    finally:
+        for proc in children:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait(timeout=10)
+            for stream in (proc.stdin, proc.stdout, proc.stderr):
+                stream.close()
+            proc._handle.Close()
+        for job in jobs:
+            job.close()

tests/hermes_cli/test_local_runtime_recovery.py

@@ -0,0 +1,426 @@
+"""Recovery only targets the recorded managed router, never a process-name sweep."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import time
+
+import psutil
+import pytest
+
+
+def _wait_for(predicate, timeout=10):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(0.05)
+    assert predicate(), "process transition did not complete"
+
+
+@pytest.mark.windows_only
+@pytest.mark.parametrize("case", [
+    "modern", "legacy", "live-owner", "pid-reused", "unknown-owner", "wrong-exe",
+    "legacy-key", "legacy-models", "busy", "state-replaced", "wrong-parent",
+])
+def test_startup_preserves_trees_and_explicit_stop_checks_owner(tmp_path, monkeypatch, case):
+    from hermes_cli.local_runtime import bootstrap, supervisor
+
+    root = tmp_path / "managed runtime"
+    root.mkdir()
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: root)
+    monkeypatch.setattr(bootstrap, "runtimes_root", lambda: root)
+    monkeypatch.setattr(bootstrap, "_SUPERVISOR", None)
+    monkeypatch.setattr(bootstrap, "_presets_stale", lambda: False)
+    monkeypatch.setattr(bootstrap, "_detect_gpu_vendor", lambda: None)
+    monkeypatch.setattr("hermes_cli.local_runtime.binaries.installed_tags", lambda: [])
+
+    # A copied native interpreter stands in for the installed server; no live model is touched.
+    exe = root / "test-build" / "cpu" / "llama-server.exe"
+    exe.parent.mkdir(parents=True)
+    shutil.copy2(sys._base_executable, exe)
+    env = os.environ.copy()
+    env["PYTHONHOME"] = str(Path(sys._base_executable).parent)
+    env["PATH"] = str(Path(sys._base_executable).parent) + os.pathsep + env.get("PATH", "")
+    env["PYTHONPATH"] = os.pathsep.join(sys.path)
+    mdir = tmp_path / "models"
+    mdir.mkdir()
+    monkeypatch.setattr(bootstrap, "models_dir", lambda: mdir)
+    # A real owner exits without cleanup, leaving a router and its model child.
+    ready = tmp_path / "router.json"
+    child_script = tmp_path / "router.py"
+    child_script.write_text(
+        "import json,os,sys,time,subprocess,psutil\n"
+        "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n"
+        "with open(sys.argv[1], 'w') as f:\n"
+        " json.dump({'router': os.getpid(), 'child': child.pid, "
+        "'create_time': psutil.Process().create_time()}, f)\n"
+        "time.sleep(60)\n", encoding="utf-8")
+    owner = subprocess.Popen([
+        sys.executable, "-c",
+        "import subprocess,sys,psutil,json,os,time; "
+        "p=subprocess.Popen(sys.argv[1:]); "
+        "print(json.dumps({'pid':os.getpid(),'create_time':psutil.Process().create_time()}),flush=True); "
+        f"time.sleep({60 if case == 'live-owner' else 1})",
+        str(exe), str(child_script), str(ready),
+        "--host", "127.0.0.1", "--port", "59999", "--api-key", "test-only",
+        "--models-dir", str(mdir),
+    ], stdout=subprocess.PIPE, text=True, env=env)
+    owner_identity = json.loads(owner.stdout.readline())
+    processes = []
+    try:
+        _wait_for(ready.exists)
+        record = json.loads(ready.read_text())
+        processes = [psutil.Process(record[k]) for k in ("router", "child")]
+        if case != "live-owner":
+            owner.wait(timeout=10)
+        # Retain the identity, not a still-open Popen handle keeping a dead PID visible.
+        monkeypatch.setattr(bootstrap, "staged_models", lambda: [tmp_path / "model.gguf"])
+        state = {
+            "pid": record["router"], "create_time": record["create_time"],
+            "owner_pid": owner_identity["pid"], "owner_create_time": owner_identity["create_time"],
+            "executable": processes[0].exe(),
+            "base_url": "http://127.0.0.1:59999/v1", "api_key": "test-only",
+        }
+        if case.startswith("legacy"):
+            state = {key: state[key] for key in ("pid", "base_url", "api_key")}
+        if case == "pid-reused":
+            state["create_time"] -= 10
+        if case == "unknown-owner":
+            state.pop("owner_create_time")
+        if case == "wrong-exe":
+            state["executable"] = str(tmp_path / "other.exe")
+        if case == "legacy-key":
+            state["api_key"] = "different-key"
+        if case == "legacy-models":
+            monkeypatch.setattr(bootstrap, "models_dir", lambda: tmp_path / "other models")
+        supervisor.state_path().write_text(json.dumps(state), encoding="utf-8")
+        from fastapi import HTTPException
+        from hermes_cli.local_runtime import endpoint
+        from hermes_cli.web_routers import local_models
+        if case == "wrong-parent":
+            state["owner_pid"] = os.getpid()
+            supervisor.state_path().write_text(json.dumps(state))
+        for _ in range(2):
+            bootstrap.ensure_local_runtime({"local_runtime": {"enabled": True}})
+            assert all(p.is_running() for p in processes), f"startup killed: {case}"
+            assert json.loads(supervisor.state_path().read_text()) == state
+        if case == "state-replaced":
+            state = {**state, "owner_pid": os.getpid(), "owner_create_time": psutil.Process().create_time()}
+            supervisor.state_path().write_text(json.dumps(state))
+        if case in ("modern", "legacy", "busy"):
+            if case == "modern":
+                from fastapi import FastAPI
+                from fastapi.testclient import TestClient
+                app = FastAPI()
+                app.include_router(local_models.router)
+                disabled = []
+                monkeypatch.setattr(local_models, "_set_runtime_enabled", lambda value: disabled.append(value))
+                response = TestClient(app).post("/api/local-models/server", json={"action": "stop"})
+                assert response.status_code == 200, response.text
+                assert disabled == [False]
+            else:
+                local_models._terminate_state_pid()
+            _wait_for(lambda: not any(p.is_running() for p in processes))
+            assert endpoint._state_endpoint() is None
+        else:
+            with pytest.raises(HTTPException) as exc:
+                local_models._terminate_state_pid()
+            assert exc.value.status_code == 409
+            assert all(p.is_running() for p in processes), f"unsafe recovery: {case}"
+        assert json.loads(supervisor.state_path().read_text()) == state
+    finally:
+        for proc in processes:
+            if proc.is_running():
+                proc.kill()
+            try:
+                proc.wait(timeout=5)
+            except psutil.NoSuchProcess:
+                pass
+        if owner.poll() is None:
+            owner.kill()
+            owner.wait(timeout=5)
+        owner.stdout.close()
+
+
+@pytest.mark.parametrize("failure", ["arrival", "truncated"])
+def test_startup_reuses_without_activity_probe(tmp_path, monkeypatch, failure):
+    import http.client
+    from hermes_cli.local_runtime import bootstrap, endpoint, recovery
+
+    monkeypatch.setattr(bootstrap, "_SUPERVISOR", None)
+    monkeypatch.setattr(bootstrap, "staged_models", lambda: [tmp_path / "model.gguf"])
+    monkeypatch.setattr(bootstrap, "_presets_stale", lambda: False)
+    state = {"base_url": "http://127.0.0.1:59999/v1", "api_key": "test-only"}
+    monkeypatch.setattr(endpoint, "_state_endpoint", lambda: state)
+    calls = []
+    def probe(*args, **kwargs):
+        calls.append(True)
+        if failure == "truncated":
+            raise http.client.IncompleteRead(b"partial", 100)
+        return {"data": []}  # work arrives immediately after an idle snapshot
+    monkeypatch.setattr(endpoint, "managed_get_json", probe)
+    monkeypatch.setattr(recovery, "read_state", lambda: state)
+    from types import SimpleNamespace
+    monkeypatch.setattr(recovery, "recorded_process",
+                        lambda state: SimpleNamespace(is_running=lambda: False))
+    monkeypatch.setattr(recovery, "_owner_is_dead", lambda state: True)
+    for _ in range(2):
+        assert bootstrap.ensure_local_runtime({"local_runtime": {"enabled": True}}) is None
+    assert not calls
+
+
+def test_shutdown_during_backoff_cannot_restart_or_remove_another_server(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+    from hermes_cli.local_runtime import supervisor
+
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path)
+    sup = supervisor.LlamaServerSupervisor(tmp_path, tmp_path, port=59998)
+    sup.proc = SimpleNamespace(pid=101, poll=lambda: 1)
+    other = {"pid": 202, "base_url": "http://127.0.0.1:59997/v1", "api_key": "test-only"}
+    supervisor.state_path().write_text(json.dumps(other))
+    spawned = []
+    monkeypatch.setattr(sup, "_spawn", lambda: spawned.append(True))
+    monkeypatch.setattr(sup, "_wait_health", lambda *a: None)
+    monkeypatch.setattr(sup, "_reap_orphaned_children", lambda: None)
+    monkeypatch.setattr(supervisor.time, "sleep", lambda *a: sup.stop())
+    # A real Event.wait is independently exercised by the native lifetime tests.
+    if hasattr(sup, "_stop_event"):
+        monkeypatch.setattr(sup._stop_event, "wait", lambda *a: (sup.stop() or True))
+    sup._watch()
+    assert not spawned, "stop during restart backoff resurrected the runtime"
+    assert json.loads(supervisor.state_path().read_text()) == other
+
+
+@pytest.mark.windows_only
+def test_supervisor_reaps_owned_job_even_after_router_exit(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+    from hermes_cli.local_runtime import supervisor
+
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path)
+    sup = supervisor.LlamaServerSupervisor(tmp_path, tmp_path, port=59998)
+    sup.proc = SimpleNamespace(pid=101, poll=lambda: 1)
+    closed = []
+    sup._job = SimpleNamespace(close=lambda: closed.append(True))
+    sup._reap_orphaned_children()
+    assert closed == [True], "crashed router's model children escaped cleanup"
+    assert sup._job is None
+    sup._job = SimpleNamespace(close=lambda: closed.append(True))
+    sup.stop()
+    assert closed == [True, True]
+    assert sup._job is None
+
+
+def test_spawn_state_records_process_incarnations(tmp_path, monkeypatch):
+    import os
+    from hermes_cli.local_runtime import supervisor
+
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path)
+    sup = supervisor.LlamaServerSupervisor(tmp_path, tmp_path, port=59998)
+    proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
+    try:
+        sup.proc = proc
+        sup._write_state()
+        state = json.loads(supervisor.state_path().read_text())
+        assert state.get("create_time") == psutil.Process(proc.pid).create_time()
+        assert state.get("owner_pid") == os.getpid()
+        assert state.get("owner_create_time") == psutil.Process().create_time()
+        assert state.get("executable") == psutil.Process(proc.pid).exe()
+    finally:
+        proc.kill()
+        proc.wait(timeout=5)
+
+
+def test_stopped_state_is_retained_without_unlink_race(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+    from hermes_cli.local_runtime import supervisor
+
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path)
+    sup = supervisor.LlamaServerSupervisor(tmp_path, tmp_path, port=59998)
+    sup.proc = SimpleNamespace(pid=101, poll=lambda: 1)
+    sup._state = {"pid": 101}
+    path = supervisor.state_path()
+    path.write_text(json.dumps(sup._state))
+    replacement = {"pid": os.getpid(), "base_url": "http://127.0.0.1:59997/v1", "api_key": "test-only"}
+    unlink = Path.unlink
+    def publish_before_unlink(self, *args, **kwargs):
+        if self == path:
+            self.write_text(json.dumps(replacement))
+        return unlink(self, *args, **kwargs)
+    with monkeypatch.context() as m:
+        m.setattr(Path, "unlink", publish_before_unlink)
+        sup.stop()
+    assert json.loads(path.read_text()) == sup._state
+    path.write_text(json.dumps(replacement))
+    sup.stop()
+    assert json.loads(path.read_text()) == replacement
+
+
+@pytest.mark.parametrize("kind", ["psutil", "subprocess", "missing-psutil", "wait-error"])
+def test_terminate_tree_escalates_and_always_cleans_children(monkeypatch, kind):
+    from types import SimpleNamespace
+    from unittest.mock import Mock
+    from hermes_cli.local_runtime.supervisor import LlamaServerSupervisor
+
+    child = Mock()
+    child.is_running.return_value = True
+    proc = Mock(pid=123)
+    error = (psutil.TimeoutExpired(15) if kind == "psutil" else
+             RuntimeError("wait failed") if kind == "wait-error" else
+             subprocess.TimeoutExpired("router", 15))
+    proc.wait.side_effect = error
+    if kind == "missing-psutil":
+        monkeypatch.setitem(sys.modules, "psutil", None)
+    else:
+        monkeypatch.setattr(psutil, "Process", lambda pid: SimpleNamespace(children=lambda **kw: [child]))
+    if kind == "wait-error":
+        with pytest.raises(RuntimeError, match="wait failed"):
+            LlamaServerSupervisor._terminate_tree(proc)
+    else:
+        LlamaServerSupervisor._terminate_tree(proc)
+        proc.kill.assert_called_once()
+    if kind != "missing-psutil":
+        child.terminate.assert_called_once()
+        child.kill.assert_called_once()
+
+
+@pytest.mark.parametrize("reuse_at", ["before-walk", "during-walk", "never"])
+def test_explicit_stop_preserves_verified_root_incarnation(tmp_path, monkeypatch, reuse_at):
+    from unittest.mock import Mock
+    from hermes_cli.local_runtime import recovery, supervisor
+
+    state = {"pid": 123, "create_time": 1.0}
+    path = tmp_path / "server.json"
+    path.write_text(json.dumps(state))
+    monkeypatch.setattr(supervisor, "state_path", lambda: path)
+    root = Mock(spec=psutil.Process, pid=123)
+    owned_child = Mock(spec=psutil.Process)
+    replacement_child = Mock(spec=psutil.Process)
+    replacement = Mock(spec=psutil.Process, pid=123)
+    replacement.children.return_value = [replacement_child]
+    replacement_child.is_running.return_value = True
+    owned_child.is_running.return_value = True
+    stale = False
+
+    def final_identity_check():
+        nonlocal stale
+        if stale:
+            return False
+        # The verified root exits and its PID is reused just after this check.
+        stale = reuse_at == "before-walk"
+        return True
+
+    def verified_children(*, recursive):
+        nonlocal stale
+        assert recursive is True
+        if stale:
+            raise psutil.NoSuchProcess(root.pid)
+        if reuse_at == "during-walk":
+            stale = True
+            return [replacement_child]
+        return [owned_child]
+
+    def verified_terminate():
+        if stale:
+            raise psutil.NoSuchProcess(root.pid)
+
+    root.is_running.side_effect = final_identity_check
+    root.children.side_effect = verified_children
+    root.terminate.side_effect = verified_terminate
+    factory = Mock(side_effect=lambda pid: replacement if stale else root)
+    monkeypatch.setattr(psutil, "Process", factory)
+    monkeypatch.setattr(psutil, "pid_exists", lambda pid: True)
+    monkeypatch.setattr(recovery, "recorded_process", lambda record: root)
+    monkeypatch.setattr(recovery, "_owner_is_dead", lambda record: True)
+
+    stopped = recovery.stop_recorded_orphan()
+
+    replacement_child.terminate.assert_not_called()
+    replacement_child.kill.assert_not_called()
+    replacement.terminate.assert_not_called()
+    replacement.kill.assert_not_called()
+    root.children.assert_called_once_with(recursive=True)
+    factory.assert_not_called()
+    assert stopped is (reuse_at == "never")
+    if reuse_at != "never":
+        root.terminate.assert_not_called()
+        root.kill.assert_not_called()
+        owned_child.terminate.assert_not_called()
+        owned_child.kill.assert_not_called()
+    else:
+        root.terminate.assert_called_once_with()
+        owned_child.terminate.assert_called_once_with()
+        owned_child.kill.assert_called_once_with()
+
+
+@pytest.mark.linux_only
+def test_reparented_router_keeps_its_endpoint(tmp_path, monkeypatch):
+    from hermes_cli.local_runtime import endpoint, supervisor
+
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path)
+    owner = subprocess.Popen([sys.executable, "-c", """
+import json, os, psutil, subprocess, sys
+p = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'], stdout=subprocess.DEVNULL)
+proc = psutil.Process(p.pid)
+print(json.dumps({'pid': proc.pid, 'create_time': proc.create_time(), 'executable': proc.exe(),
+                  'owner_pid': os.getpid(), 'owner_create_time': psutil.Process().create_time()}), flush=True)
+"""], stdout=subprocess.PIPE, text=True)
+    state = json.loads(owner.stdout.readline())
+    proc = psutil.Process(state["pid"])
+    try:
+        owner.wait(timeout=10)
+        route = {"base_url": "http://127.0.0.1:59999/v1", "api_key": "test-only"}
+        supervisor.state_path().write_text(json.dumps({**state, **route}))
+        assert proc.ppid() != state["owner_pid"]
+        assert endpoint._state_endpoint() == route
+    finally:
+        proc.kill()
+        proc.wait(timeout=10)
+        owner.stdout.close()
+
+
+@pytest.mark.windows_only
+@pytest.mark.parametrize("damage", ["valid", "birth", "exe", "bool-pid", "bool-birth", "nan", "inf", "owner-bool", "owner-nan", "parent", "partial", "list", "invalid", "unreadable"])
+def test_retained_endpoint_validates_identity(tmp_path, monkeypatch, damage):
+    from hermes_cli.local_runtime import endpoint, recovery, supervisor
+
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path)
+    proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
+    try:
+        real = psutil.Process(proc.pid)
+        state = {"pid": real.pid, "create_time": real.create_time(), "executable": real.exe(),
+                 "owner_pid": os.getpid(), "owner_create_time": psutil.Process().create_time(),
+                 "base_url": "http://127.0.0.1:59999/v1", "api_key": "test-only"}
+        changes = {"birth": {"create_time": real.create_time() - 10}, "exe": {"executable": str(tmp_path / "wrong.exe")},
+                   "bool-pid": {"pid": True}, "bool-birth": {"create_time": True},
+                   "nan": {"create_time": float("nan")}, "inf": {"create_time": float("inf")},
+                   "owner-bool": {"owner_create_time": True}, "owner-nan": {"owner_create_time": float("nan")},
+                   "parent": {"owner_pid": real.pid}}
+        state.update(changes.get(damage, {}))
+        if damage == "partial":
+            state.pop("create_time")
+        path = supervisor.state_path()
+        path.write_text("[]" if damage == "list" else "{" if damage == "invalid" else json.dumps(state))
+        if damage == "unreadable":
+            path.unlink()
+            path.mkdir()
+        got = endpoint._state_endpoint()
+        if damage == "valid":
+            assert got == {"base_url": state["base_url"], "api_key": state["api_key"]}
+            proc.terminate()
+            proc.wait(timeout=5)
+            assert endpoint._state_endpoint() is None
+            assert path.exists()
+        else:
+            assert got is None
+            if damage not in ("list", "invalid", "unreadable"):
+                assert recovery.recorded_process(state) is None
+            assert real.is_running()
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+        proc.wait(timeout=5)

tests/hermes_cli/test_local_server_lifecycle.py

@@ -112,3 +112,26 @@ def test_status_reports_loaded_models_from_live_router(client, monkeypatch):
         assert payload["loaded_models"] == {"m-loaded": "loaded", "m-loading": "loading"}
     finally:
         server.shutdown()
+
+
+@pytest.mark.parametrize("refuse", [False, True])
+def test_stop_forwards_recovery_and_preserves_conflict(client, tmp_path, monkeypatch, refuse):
+    from fastapi import HTTPException
+    from hermes_cli.web_routers import local_models
+    from hermes_cli.local_runtime import supervisor
+
+    monkeypatch.setattr(supervisor, "runtimes_root", lambda: tmp_path)
+    supervisor.state_path().write_text("{}")
+    monkeypatch.setattr(local_models.bootstrap, "get_supervisor", lambda: None)
+    monkeypatch.setattr(local_models, "_state_endpoint", lambda: None)
+    called, disabled = [], []
+    def recover():
+        called.append(True)
+        if refuse:
+            raise HTTPException(409, "Another Hermes process owns this server, or its ownership could not be verified")
+    monkeypatch.setattr(local_models, "_terminate_state_pid", recover)
+    monkeypatch.setattr(local_models, "_set_runtime_enabled", lambda value: disabled.append(value))
+    response = client.post("/api/local-models/server", json={"action": "stop"})
+    assert response.status_code == (409 if refuse else 200), response.text
+    assert called == [True]
+    assert disabled == ([] if refuse else [False])