diff --git a/hermes_cli/local_runtime/endpoint.py b/hermes_cli/local_runtime/endpoint.py index ac1a2ffd81..e1649d5315 100644 --- a/hermes_cli/local_runtime/endpoint.py +++ b/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", "")} diff --git a/hermes_cli/local_runtime/processes.py b/hermes_cli/local_runtime/processes.py new file mode 100644 index 0000000000..d2a3168cd0 --- /dev/null +++ b/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 diff --git a/hermes_cli/local_runtime/recovery.py b/hermes_cli/local_runtime/recovery.py new file mode 100644 index 0000000000..8e5d2c18f4 --- /dev/null +++ b/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 diff --git a/hermes_cli/local_runtime/supervisor.py b/hermes_cli/local_runtime/supervisor.py index 67269f8058..1f20d1ac9b 100644 --- a/hermes_cli/local_runtime/supervisor.py +++ b/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 diff --git a/hermes_cli/web_routers/local_models.py b/hermes_cli/web_routers/local_models.py index b14c700e14..125ac96adc 100644 --- a/hermes_cli/web_routers/local_models.py +++ b/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} diff --git a/tests/hermes_cli/test_local_abandoned_requests.py b/tests/hermes_cli/test_local_abandoned_requests.py index 0f9d05260f..eea3460036 100644 --- a/tests/hermes_cli/test_local_abandoned_requests.py +++ b/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. diff --git a/tests/hermes_cli/test_local_preset_admission.py b/tests/hermes_cli/test_local_preset_admission.py index 31b05f5c42..ce4757d687 100644 --- a/tests/hermes_cli/test_local_preset_admission.py +++ b/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() diff --git a/tests/hermes_cli/test_local_runtime_processes.py b/tests/hermes_cli/test_local_runtime_processes.py new file mode 100644 index 0000000000..17227987f5 --- /dev/null +++ b/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() diff --git a/tests/hermes_cli/test_local_runtime_recovery.py b/tests/hermes_cli/test_local_runtime_recovery.py new file mode 100644 index 0000000000..a9ecc85e22 --- /dev/null +++ b/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) diff --git a/tests/hermes_cli/test_local_server_lifecycle.py b/tests/hermes_cli/test_local_server_lifecycle.py index 7a2c5300ef..3aa6cecce8 100644 --- a/tests/hermes_cli/test_local_server_lifecycle.py +++ b/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])