diff --git a/hermes_cli/local_runtime/__init__.py b/hermes_cli/local_runtime/__init__.py index 3186263840..94e612162d 100644 --- a/hermes_cli/local_runtime/__init__.py +++ b/hermes_cli/local_runtime/__init__.py @@ -1,42 +1,22 @@ """Managed llama.cpp runtime. -- ``binaries`` — resolve/download/verify official llama.cpp release zips into -``$HERMES_HOME/runtimes/llamacpp//``. - ``supervisor``— spawn and supervise one llama-server in -router mode; readiness is a touch generation, never health-200 alone. - ``detect`` — find an +``binaries`` resolves/downloads/verifies official llama.cpp release zips into +``$HERMES_HOME/runtimes/llamacpp//``; ``supervisor`` spawns and supervises one llama-server in +router mode (readiness is a touch generation, never health-200 alone); ``detect`` finds an already-running llama-server (external or ours). """ from hermes_cli.local_runtime.binaries import ( # noqa: F401 - BinaryResolutionError, - ensure_runtime_installed, - resolve_assets, - select_backend, -) -from hermes_cli.local_runtime.bootstrap import ( # noqa: F401 - ensure_local_runtime, - shutdown_local_runtime, -) + BinaryResolutionError, ensure_runtime_installed, resolve_assets, select_backend) +from hermes_cli.local_runtime.bootstrap import ensure_local_runtime, shutdown_local_runtime # noqa: F401 from hermes_cli.local_runtime.context_policy import ( # noqa: F401 - FLOOR, - growth_decision, - initial_window, - ladder, - launch_args, -) + FLOOR, growth_decision, initial_window, ladder, launch_args) from hermes_cli.local_runtime.growth import ( # noqa: F401 - clear_window_override, - load_window_overrides, - maybe_grow_window, - save_window_override, -) + clear_window_override, load_window_overrides, maybe_grow_window, save_window_override) from hermes_cli.local_runtime.detect import detect_server # noqa: F401 from hermes_cli.local_runtime.endpoint import resolve_llamacpp_endpoint # noqa: F401 from hermes_cli.local_runtime.estimator import ( # noqa: F401 - HardwareBudget, - ctx_bytes, - physics_check, - profile_from_gguf, -) + HardwareBudget, ctx_bytes, physics_check, profile_from_gguf) from hermes_cli.local_runtime.gguf import read_gguf_header # noqa: F401 from hermes_cli.local_runtime.hardware import probe_budget # noqa: F401 from hermes_cli.local_runtime.presets import generate_presets # noqa: F401 diff --git a/hermes_cli/local_runtime/binaries.py b/hermes_cli/local_runtime/binaries.py index ea67131e2e..8c9257bea2 100644 --- a/hermes_cli/local_runtime/binaries.py +++ b/hermes_cli/local_runtime/binaries.py @@ -5,6 +5,7 @@ from __future__ import annotations import hashlib import json import logging +import os import platform import shutil import subprocess @@ -19,20 +20,16 @@ logger = logging.getLogger(__name__) RELEASE_URL = "https://github.com/ggml-org/llama.cpp/releases/download/{tag}/{asset}" -# Windows CUDA zips ship per CUDA major; the runtime zip must be paired with -# its cudart zip so end users need no toolkit. 13.3 verified on 13.1 and -# 13.2 drivers. +# Windows CUDA zips ship per CUDA major; the runtime zip must be paired with its cudart zip so +# end users need no toolkit. 13.3 verified on 13.1 and 13.2 drivers. _WIN_CUDA_VERSION = "13.3" -# arm64 Windows CUDA prebuilts landed upstream (~b1036x) on CUDA 13.4 — -# verified against live asset lists (b10362, b10630, b10679). Tags at or before -# b10290 don't have them; resolution succeeds and the download 404s -# honestly on such tags, which only arises if a user pins backward. +# arm64 Windows CUDA prebuilts landed upstream (~b1036x) on CUDA 13.4. Tags at or before b10290 +# don't have them; resolution succeeds and the download 404s honestly if a user pins backward. _WIN_CUDA_VERSION_ARM64 = "13.4" -# Fallback when the config section is missing entirely (deep-merge normally -# guarantees the key). Single source: DEFAULT_CONFIG owns the shipped tag. def default_tag() -> str: + """Fallback when the config section is missing entirely; DEFAULT_CONFIG owns the shipped tag.""" from hermes_cli.config_defaults import DEFAULT_CONFIG return DEFAULT_CONFIG["local_runtime"]["tag"] @@ -56,65 +53,57 @@ class AssetPlan: def runtimes_root() -> Path: - """Machine-scoped, deliberately NOT profile-scoped. Engine binaries, presets, and server state + """Machine-scoped, deliberately NOT profile-scoped: engine binaries, presets and server state describe this machine's hardware and its one managed server (stable port) — a second profile re-downloading the engine or fighting over the port would be the bug. Profile-scoped things - (which model is the default, enabled) live in each profile's config.yaml as ever. - """ + (default model, enabled) live in each profile's config.yaml.""" from hermes_constants import get_default_hermes_root return get_default_hermes_root() / "runtimes" / "llamacpp" +def manifest_verified(manifest: Path) -> bool: + """True when an install manifest records a verified_version (missing/damaged -> False).""" + try: + return bool(json.loads(manifest.read_text(encoding="utf-8")).get("verified_version")) + except (json.JSONDecodeError, OSError): + return False + + +def _release_number(tag: str) -> int: + digits = "".join(ch for ch in tag if ch.isdigit()) + return int(digits) if digits else 0 + + def installed_tags() -> list[str]: - """Tags with a verified install (manifest carries verified_version), - newest first by release number. The boot ladder and the update check - both read installed-ness from here — one resolver, every caller.""" + """Tags with a verified install, newest first by release number. The boot ladder and the + update check both read installed-ness from here — one resolver, every caller.""" root = runtimes_root() if not root.exists(): return [] - found: list[str] = [] - for entry in root.iterdir(): - if not entry.is_dir() or entry.name == "downloads": - continue - for manifest in entry.glob("*/manifest.json"): - try: - if json.loads(manifest.read_text(encoding="utf-8")).get("verified_version"): - found.append(entry.name) - break - except (json.JSONDecodeError, OSError): - continue - - def _release_number(tag: str) -> int: - digits = "".join(ch for ch in tag if ch.isdigit()) - return int(digits) if digits else 0 - - return sorted(set(found), key=_release_number, reverse=True) + found = {entry.name for entry in root.iterdir() + if entry.is_dir() and entry.name != "downloads" + and any(manifest_verified(m) for m in entry.glob("*/manifest.json"))} + return sorted(found, key=_release_number, reverse=True) def _host_os_arch() -> tuple[str, str]: - """(os, arch) normalized to release-asset vocabulary. - - PITFALL: PROCESSOR_ARCHITECTURE lies under x64 emulation on ARM64 Windows. platform.machine() - reads the same env on some Pythons, so on Windows prefer PROCESSOR_IDENTIFIER's text when - present. - """ + """(os, arch) normalized to release-asset vocabulary. PITFALL: PROCESSOR_ARCHITECTURE lies + under x64 emulation on ARM64 Windows, and platform.machine() reads the same env on some + Pythons — so on Windows prefer PROCESSOR_IDENTIFIER's text when present.""" system = platform.system().lower() os_name = {"windows": "win", "darwin": "macos", "linux": "ubuntu"}.get(system, system) - machine = platform.machine().lower() - arch = "arm64" if machine in ("arm64", "aarch64") else "x64" + arch = "arm64" if platform.machine().lower() in ("arm64", "aarch64") else "x64" if os_name == "win": - import os as _os - ident = _os.environ.get("PROCESSOR_IDENTIFIER", "").lower() + ident = os.environ.get("PROCESSOR_IDENTIFIER", "").lower() if "armv8" in ident or "arm " in ident: arch = "arm64" return os_name, arch def select_backend(gpu_vendor: str | None, os_name: str | None = None) -> str: - """Backend choice per design: CUDA if NVIDIA, Metal on macOS, Vulkan if - a non-NVIDIA GPU is present, else CPU. ``--list-devices`` validates the - choice post-install; the supervisor's touch generation is ground truth.""" + """CUDA if NVIDIA, Metal on macOS, Vulkan if a non-NVIDIA GPU is present, else CPU. + ``--list-devices`` validates post-install; the supervisor's touch generation is ground truth.""" if os_name is None: os_name, _ = _host_os_arch() if os_name == "macos": @@ -127,8 +116,8 @@ def select_backend(gpu_vendor: str | None, os_name: str | None = None) -> str: return "cpu" -# Per-OS (human label, {backend: asset-name templates}). Windows CUDA pairs the -# runtime zip with its cudart zip; ubuntu ships tarballs, win ships zips. +# Per-OS (human label, {backend: asset-name templates}). Windows CUDA pairs the runtime zip with +# its cudart zip; ubuntu ships tarballs, win ships zips. _ASSET_TEMPLATES = { "ubuntu": ("linux", { "vulkan": ["llama-{tag}-bin-ubuntu-vulkan-{arch}.tar.gz"], @@ -147,11 +136,8 @@ _ASSET_TEMPLATES = { def resolve_assets(tag: str, backend: str, os_name: str | None = None, arch: str | None = None) -> AssetPlan: - """Compose the asset list for (tag, backend, platform). - - Raises BinaryResolutionError for platform/backend pairs the release ships no artifact for; - callers fall back down the backend ladder cuda -> vulkan -> cpu. - """ + """Compose the asset list for (tag, backend, platform). Raises BinaryResolutionError for pairs + the release ships no artifact for; callers fall back down the ladder cuda -> vulkan -> cpu.""" host_os, host_arch = _host_os_arch() os_name = os_name or host_os arch = arch or host_arch @@ -161,8 +147,8 @@ def resolve_assets(tag: str, backend: str, os_name: str | None = None, if os_name not in _ASSET_TEMPLATES: raise BinaryResolutionError(f"unsupported platform {os_name}-{arch}") if os_name == "ubuntu" and backend == "cuda": - # No prebuilt Linux CUDA zips at current tags — Linux CUDA users - # build from source or use vulkan; resolver is honest about it. + # No prebuilt Linux CUDA zips at current tags — Linux CUDA users build from source or + # use vulkan; the resolver is honest about it. raise BinaryResolutionError( f"no prebuilt linux CUDA asset at {tag}; use vulkan/cpu or a source build") if os_name == "win" and backend == "vulkan" and arch == "arm64": @@ -185,9 +171,8 @@ def _sha256(path: Path) -> str: def _download(url: str, dest: Path, progress: "Callable[[int, int], None] | None" = None) -> None: - """Stream url -> dest. ``progress(done_bytes, total_bytes)`` ticks per - chunk (total 0 when the server sends no Content-Length) — a several- - hundred-MB archive on a slow line must never look hung.""" + """Stream url -> dest. ``progress(done_bytes, total_bytes)`` ticks per chunk (total 0 when + the server sends no Content-Length) — a several-hundred-MB archive must never look hung.""" logger.info("downloading %s", url) tmp = dest.with_suffix(dest.suffix + ".part") with urllib.request.urlopen(url, timeout=120) as r, open(tmp, "wb") as f: @@ -206,8 +191,7 @@ def _download(url: str, dest: Path, def _extract(archive: Path, dest: Path, progress: "Callable[[int, int], None] | None" = None) -> None: - """Extract member by member so ``progress(done, total)`` can tick in - uncompressed bytes — big archives take real time on laptop disks.""" + """Extract member by member so ``progress(done, total)`` can tick in uncompressed bytes.""" if archive.name.endswith(".zip"): opener, list_members, size = zipfile.ZipFile, "infolist", "file_size" kwargs = {} @@ -227,8 +211,8 @@ def _extract(archive: Path, dest: Path, def server_binary(install_dir: Path) -> Path: - """Locate llama-server within an extracted runtime (zips differ in - whether they nest a build/bin directory).""" + """Locate llama-server within an extracted runtime (zips differ in whether they nest a + build/bin directory).""" names = ("llama-server.exe", "llama-server") for name in names: direct = install_dir / name @@ -242,8 +226,7 @@ def server_binary(install_dir: Path) -> Path: def verify_install(install_dir: Path, tag: str) -> str: - """Run --version; require the tag's build number in the output. - (The binary prints the tag WITHOUT the 'b' prefix.)""" + """Run --version; require the tag's build number in the output (printed WITHOUT the 'b').""" exe = server_binary(install_dir) out = subprocess.run([str(exe), "--version"], capture_output=True, text=True, encoding="utf-8", errors="replace", @@ -256,8 +239,8 @@ def verify_install(install_dir: Path, tag: str) -> str: def prune_old_tags(keep: list[str]) -> None: - """Retain only the tags in ``keep`` (current + previous — N-1 rollback). - The shared ``downloads/`` archive cache is not a tag and always survives.""" + """Retain only the tags in ``keep`` (current + previous — N-1 rollback). The shared + ``downloads/`` archive cache is not a tag and always survives.""" root = runtimes_root() if not root.exists(): return @@ -270,28 +253,26 @@ def prune_old_tags(keep: list[str]) -> None: def ensure_runtime_installed(tag: str, backend: str, expected_sha256: dict[str, str] | None = None, progress: "Callable[[str, int, int, str], None] | None" = None) -> Path: - """Idempotent: resolve, download, verify, extract, version-check. - - ``expected_sha256`` pins hashes per asset when the catalog has them; without pins the computed - hash is recorded in the manifest (trust on first download, verified on every reinstall). - ``progress(stage, done, total, label)`` ticks through download/extract/verify. Returns the - install dir containing llama-server. - """ + """Idempotent: resolve, download, verify, extract, version-check; returns the install dir. + ``expected_sha256`` pins hashes per asset; without pins the computed hash is recorded in the + manifest (trust on first download, verified on every reinstall). ``progress(stage, done, + total, label)`` ticks through download/extract/verify.""" plan = resolve_assets(tag, backend) install_dir = plan.install_dir manifest_path = install_dir / "manifest.json" - if manifest_path.exists(): - try: - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - if manifest.get("verified_version"): - return install_dir - except (json.JSONDecodeError, OSError): - pass # damaged manifest -> reinstall + if manifest_verified(manifest_path): + return install_dir install_dir.mkdir(parents=True, exist_ok=True) downloads = runtimes_root() / "downloads" downloads.mkdir(parents=True, exist_ok=True) + def stage_progress(stage: str, label: str): + if progress is None: + return None + tick = progress + return lambda d, t: tick(stage, d, t, label) + recorded: dict[str, str] = {} n_assets = len(plan.assets) for i, asset in enumerate(plan.assets, 1): @@ -299,8 +280,7 @@ def ensure_runtime_installed(tag: str, backend: str, archive = downloads / asset if not archive.exists(): _download(RELEASE_URL.format(tag=tag, asset=asset), archive, - progress=(lambda d, t, _l=label: progress("download", d, t, _l)) - if progress is not None else None) + progress=stage_progress("download", label)) if progress is not None: progress("verify", 0, 0, label) digest = _sha256(archive) @@ -310,16 +290,12 @@ def ensure_runtime_installed(tag: str, backend: str, raise BinaryResolutionError( f"sha256 mismatch for {asset}: expected {expected}, got {digest}") recorded[asset] = digest - _extract(archive, install_dir, - progress=(lambda d, t, _l=label: progress("extract", d, t, _l)) - if progress is not None else None) + _extract(archive, install_dir, progress=stage_progress("extract", label)) if progress is not None: progress("verify", 0, 0, "") version = verify_install(install_dir, tag) - manifest_path.write_text(json.dumps({ - "tag": tag, "backend": plan.backend, "assets": recorded, - "verified_version": version, - }, indent=2), encoding="utf-8") + manifest_path.write_text(json.dumps({"tag": tag, "backend": plan.backend, "assets": recorded, + "verified_version": version}, indent=2), encoding="utf-8") logger.info("installed llama.cpp %s (%s): %s", tag, backend, version) return install_dir diff --git a/hermes_cli/local_runtime/bootstrap.py b/hermes_cli/local_runtime/bootstrap.py index d204b48ee3..105dcd29b2 100644 --- a/hermes_cli/local_runtime/bootstrap.py +++ b/hermes_cli/local_runtime/bootstrap.py @@ -1,24 +1,22 @@ """Bootstrap for the managed runtime: config -> installed binaries -> running supervised server. -One public call, ``ensure_local_runtime(config)``, safe to call at any session start: - disabled or -already-running (state file answers /health) -> no-op - enabled -> install binaries if missing -(idempotent), spawn supervisor - -Kept import-light: callers gate on config before importing this module so sessions with -local_runtime disabled never pay the import. +One public call, ``ensure_local_runtime(config)``, safe at any session start: disabled or +already-running -> no-op; enabled -> serve the installed build under a supervisor. Kept +import-light: callers gate on config before importing so disabled sessions never pay the import. """ from __future__ import annotations +from contextlib import suppress import logging import os -import re import signal import subprocess import time from pathlib import Path from hermes_cli.local_runtime.binaries import runtimes_root +from hermes_cli.local_runtime.gguf import SPLIT_PART_RE, model_id_from_stem logger = logging.getLogger(__name__) @@ -26,101 +24,93 @@ _SUPERVISOR = None # process-wide singleton; one router per Hermes process def _detect_gpu_vendor() -> str | None: - """Best-effort GPU vendor for backend selection. NVIDIA via nvidia-smi (resolved by the hardware - probe's PATH-independent ladder — a stripped service PATH must not demote an NVIDIA box to - vulkan/cpu); anything else defers to select_backend's fallback ladder. - """ + """Best-effort GPU vendor for backend selection. NVIDIA via nvidia-smi resolved by the hardware + probe's PATH-independent ladder (a stripped service PATH must not demote an NVIDIA box to + vulkan/cpu); anything else defers to select_backend's fallback ladder.""" from hermes_cli.local_runtime.hardware import _nvidia_smi_path smi = _nvidia_smi_path() if smi is None: return None - try: + with suppress(OSError, subprocess.TimeoutExpired): out = subprocess.run( [smi, "--query-gpu=name", "--format=csv,noheader"], capture_output=True, text=True, timeout=10) if out.returncode == 0 and out.stdout.strip(): return "nvidia " + out.stdout.strip().splitlines()[0] - except (OSError, subprocess.TimeoutExpired): - pass return None def models_dir() -> Path: - """Machine-scoped, deliberately NOT profile-scoped: a 20 GB GGUF is a - machine asset, and every profile shares the one managed server that - serves it. See runtimes_root() for the same rule on the engine.""" + """Machine-scoped, deliberately NOT profile-scoped: a 20 GB GGUF is a machine asset, and every + profile shares the one managed server that serves it (same rule as runtimes_root()).""" from hermes_constants import get_default_hermes_root return get_default_hermes_root() / "models" def assets_dir() -> Path: - """Non-model companion files (mmproj vision projectors, spec-decode - draft models). A subdirectory so the router's model listing — and our - staged_models() — never mistakes an asset for a servable model.""" + """Non-model companion files (mmproj projectors, spec-decode drafts). A subdirectory so the + router's model listing — and staged_models() — never mistakes an asset for a servable model.""" return models_dir() / "assets" -def staged_models() -> "list[Path]": - """Servable staged models: single-file GGUFs count when present; a split GGUF counts once, by its - first part, and only when EVERY part is on disk — a mid-download split is not servable and must - not surface anywhere as a model. Continuation parts and assets/ never count. - """ - part = re.compile(r"-(\d{5})-of-(\d{5})\.gguf$") - files = sorted(models_dir().glob("*.gguf")) +def staged_in(models_dir: Path, *, require_complete: bool = True) -> "list[Path]": + """Servable GGUFs in a directory: single files, plus split GGUFs once by their first part. + With ``require_complete`` a split counts only when EVERY part is on disk — a mid-download split + is not servable and must not surface anywhere as a model.""" + files = sorted(models_dir.glob("*.gguf")) names = {p.name for p in files} out = [] for p in files: - m = part.search(p.name) + m = SPLIT_PART_RE.search(p.name) if m is None: out.append(p) continue if m.group(1) != "00001": continue - stem = p.name[: m.start()] - total = int(m.group(2)) - if all(f"{stem}-{i:05d}-of-{m.group(2)}.gguf" in names - for i in range(2, total + 1)): + stem, total = p.name[: m.start()], int(m.group(2)) + if not require_complete or all(f"{stem}-{i:05d}-of-{m.group(2)}.gguf" in names + for i in range(2, total + 1)): out.append(p) return out +def staged_models() -> "list[Path]": + """Servable staged models (continuation parts, incomplete splits and assets/ never count).""" + return staged_in(models_dir()) + + def staged_model_ids() -> "list[str]": - return [re.sub(r"-\d{5}-of-\d{5}$", "", p.stem) for p in staged_models()] + return [model_id_from_stem(p.stem) for p in staged_models()] def _presets_stale() -> bool: - """True when a staged model has no section in the preset INI — it - would autoload with stock fit instead of a policy decision.""" - try: + """True when a staged model has no section in the preset INI — it would autoload with stock + fit instead of a policy decision.""" + with suppress(Exception): from hermes_cli.local_runtime.presets import read_preset_decisions known = set(read_preset_decisions()) return any(mid not in known for mid in staged_model_ids()) - except Exception: # noqa: BLE001 - return False + return False def _stop_state_server(state: dict) -> None: """Best-effort stop of the server the state file points at (an incumbent this process doesn't - supervise). The state pid is ours by contract — the file only ever describes the managed server. - """ + supervise). The state pid is ours by contract — the file only ever describes the managed + server.""" from hermes_cli.local_runtime.endpoint import _pid_alive try: pid = int(state.get("pid")) - except (TypeError, ValueError): - return - if pid <= 0: - return - try: + if pid <= 0: + return os.kill(pid, signal.SIGTERM) - except (OSError, ValueError): + except (TypeError, ValueError, OSError): return - # Give it a moment to release the port and the GPU. Liveness via - # psutil — on Windows os.kill(pid, 0) TERMINATES the process, it is - # not a probe (the endpoint.py pitfall note; #local-models review). + # Give it a moment to release the port and the GPU. Liveness via psutil — on Windows + # os.kill(pid, 0) TERMINATES the process, it is not a probe. for _ in range(50): if not _pid_alive(pid): return @@ -128,12 +118,9 @@ def _stop_state_server(state: dict) -> None: def refresh_local_runtime() -> bool: - """Restart the managed server so it rescans the models directory. - - The router's model list is SPAWN-ONLY: a GGUF added after start is invisible to GET /models and - 400s on completion, so anything that changes the staged set while the server runs must bounce - it. - """ + """Restart the managed server so it rescans the models directory. The router's model list is + SPAWN-ONLY: a GGUF added after start is invisible to GET /models and 400s on completion, so + anything that changes the staged set while the server runs must bounce it.""" global _SUPERVISOR try: from hermes_cli.config import load_config @@ -144,8 +131,7 @@ def refresh_local_runtime() -> bool: state = _state_endpoint() if state is None: return False - logger.info("bouncing adopted llama-server (pid=%s) to rescan models", - state.get("pid")) + logger.info("bouncing adopted llama-server (pid=%s) to rescan models", state.get("pid")) _stop_state_server(state) else: shutdown_local_runtime() @@ -155,11 +141,41 @@ def refresh_local_runtime() -> bool: return False +def _generate_presets(mdir: Path, preset_path: Path) -> Path | None: + """Write the launch-policy INI for every staged model; returns the path to hand the router. + + Priced against CAPACITY, not live free VRAM: this runs while the outgoing server instance may + still hold the card (restart, refresh after a download), and its memory is freed before the new + instance loads anything. Pricing against live-free once pinned a fitting model's weights to CPU. + + Degradation ladder on failure: a STALE policy still beats no policy — stock fit (f16 KV at max + context, no placement) is the silent-busy-wait failure on Windows. Keep serving with the + previous INI when one exists; only a first boot with no INI at all falls to stock fit. + """ + from hermes_cli.local_runtime.hardware import probe_budget + from hermes_cli.local_runtime.presets import generate_presets + + try: + for entry in generate_presets(mdir, probe_budget(planning=True), preset_path): + if entry.refusal: + logger.warning("model refused by physics check: %s", entry.refusal) + return preset_path + except Exception as exc: # noqa: BLE001 — policy failure must not block serving + if preset_path.exists(): + logger.error("preset generation failed (%s); serving with the " + "PREVIOUS launch policies — models staged since " + "the last successful generation run unpoliced " + "until this is fixed", exc) + return preset_path + logger.error("preset generation failed (%s) and no previous " + "policy file exists; router runs stock fit", exc) + return None + + def ensure_local_runtime(config: dict, force: bool = False) -> "object | None": """Idempotent boot of the managed runtime. Returns the supervisor (or None when disabled/unavailable). Never raises into a session start — failures log and return None; chat - falls back to configured providers. - """ + falls back to configured providers.""" global _SUPERVISOR section = (config or {}).get("local_runtime") or {} if not force and not section.get("enabled"): @@ -167,21 +183,17 @@ def ensure_local_runtime(config: dict, force: bool = False) -> "object | None": if _SUPERVISOR is not None: return _SUPERVISOR - # Residency: no staged models means nothing to serve — don't boot an - # empty server. The walked-away story handled with zero configuration - # (delete your last model and boots stop); Use force-boots as ever. + # Residency: no staged models means nothing to serve — don't boot an empty server (delete + # your last model and boots stop). force boots as ever. if not force and not staged_models(): logger.info("local runtime enabled but no models staged; not booting") return None - # Another Hermes process may already be supervising — reuse via state, - # but ONLY while its launch policy still covers every staged model. A - # server whose preset file predates a download serves the new model - # with no policy at all (--models-autoload + stock fit: f16 KV at max - # context, no placement — the silent-demotion busy-wait on WDDM). A - # stale incumbent gets stopped and replaced by a fresh boot with - # regenerated presets; sessions ride through exactly like any other - # supervised restart (stable port + persisted key). + # Another Hermes process may already be supervising — reuse via state, but ONLY while its + # launch policy still covers every staged model. A server whose preset file predates a + # download serves the new model with no policy at all (--models-autoload + stock fit). A stale + # incumbent gets stopped and replaced by a fresh boot with regenerated presets; sessions ride + # through like any other supervised restart (stable port + persisted key). from hermes_cli.local_runtime.endpoint import _state_endpoint state = _state_endpoint() @@ -195,25 +207,17 @@ def ensure_local_runtime(config: dict, force: bool = False) -> "object | None": try: from hermes_cli.local_runtime.binaries import ( - default_tag, - ensure_runtime_installed, - installed_tags, - select_backend, - ) - from hermes_cli.local_runtime.hardware import probe_budget - from hermes_cli.local_runtime.presets import generate_presets + default_tag, ensure_runtime_installed, installed_tags, select_backend) from hermes_cli.local_runtime.supervisor import LlamaServerSupervisor backend = section.get("backend", "auto") if backend == "auto": backend = select_backend(_detect_gpu_vendor()) - # Boot ladder: serve what is INSTALLED, never download here. The - # configured tag (config root-of-trust; deep-merge supplies the - # Hermes-release default when unpinned) is preferred; when it isn't - # installed yet, the newest installed tag serves and the status - # endpoint reports the pending update — the download is a deliberate - # button click in the pane, not a boot-path surprise (a multi-minute - # inline download here is exactly how the onboarding bounce returns). + # Boot ladder: serve what is INSTALLED, never download here. The configured tag is + # preferred; when it isn't installed yet, the newest installed tag serves and the status + # endpoint reports the pending update — the download is a deliberate click in the pane, + # not a boot-path surprise (a multi-minute inline download here is exactly how the + # onboarding bounce returns). tag = section.get("tag") or default_tag() have = installed_tags() if tag not in have: @@ -228,56 +232,21 @@ def ensure_local_runtime(config: dict, force: bool = False) -> "object | None": mdir = models_dir() mdir.mkdir(parents=True, exist_ok=True) + preset_path = _generate_presets(mdir, runtimes_root() / "presets.ini") - # Context policy: one launch decision per staged model, carried to - # the router via the preset INI. Priced against CAPACITY, not live - # free VRAM: this runs while the outgoing server instance may still - # hold the card (restart, refresh after a download), and its memory - # is freed before the new instance loads anything. Pricing against - # live-free here once pinned a fitting model's weights to CPU - # because the probe saw the predecessor's VRAM as gone. - preset_path = runtimes_root() / "presets.ini" - try: - entries = generate_presets(mdir, probe_budget(planning=True), preset_path) - for entry in entries: - if entry.refusal: - logger.warning("model refused by physics check: %s", entry.refusal) - except Exception as exc: # noqa: BLE001 — policy failure must not block serving - # Degradation ladder: a STALE policy still beats no policy — - # stock fit (f16 KV at max context, no placement) is the - # silent-busy-wait failure on Windows. Keep serving with the - # previous INI when one exists; only a first boot with no INI - # at all falls to stock fit. - if preset_path.exists(): - logger.error("preset generation failed (%s); serving with the " - "PREVIOUS launch policies — models staged since " - "the last successful generation run unpoliced " - "until this is fixed", exc) - else: - logger.error("preset generation failed (%s) and no previous " - "policy file exists; router runs stock fit", exc) - preset_path = None - - sup = LlamaServerSupervisor( - install_dir, mdir, - models_max=int(section.get("models_max", 4)), - port=int(section.get("port", 0)) or None, - preset_path=preset_path, - ) + sup = LlamaServerSupervisor(install_dir, mdir, preset_path=preset_path, + models_max=int(section.get("models_max", 4)), + port=int(section.get("port", 0)) or None) try: sup.start() except Exception: - # start() can fail after the router process exists (health - # timeout, spawn error): leaving it running unsupervised - # strands its VRAM behind a port nothing will clean up. - try: + # start() can fail after the router process exists (health timeout): leaving it + # running unsupervised strands its VRAM behind a port nothing will clean up. + with suppress(Exception): sup.stop() - except Exception: # noqa: BLE001 — cleanup is best-effort - pass raise _SUPERVISOR = sup - logger.info("managed llama-server up at %s (backend=%s tag=%s)", - sup.base_url, backend, tag) + logger.info("managed llama-server up at %s (backend=%s tag=%s)", sup.base_url, backend, tag) _start_idle_sweeper(sup) return sup except Exception as exc: # noqa: BLE001 — never break session start @@ -293,15 +262,14 @@ def shutdown_local_runtime() -> None: def get_supervisor(): - """The process-local supervisor, or None (server may still be running - under another process — check the state file).""" + """The process-local supervisor, or None (a server may still run under another process — + check the state file).""" return _SUPERVISOR def _start_idle_sweeper(sup) -> None: - """Idle-residency loop: every couple of minutes, unload non-primary - models idle past the supervisor's threshold. Daemon thread tied to the - supervisor's lifetime — exits when the server stops.""" + """Idle-residency loop: every couple of minutes, unload models idle past the supervisor's + threshold. Daemon thread tied to the supervisor's lifetime — exits when the server stops.""" import threading def _loop(): @@ -312,5 +280,4 @@ def _start_idle_sweeper(sup) -> None: except Exception as exc: # noqa: BLE001 logger.debug("idle sweep skipped: %s", exc) - threading.Thread(target=_loop, daemon=True, - name="local-runtime-idle-sweep").start() + threading.Thread(target=_loop, daemon=True, name="local-runtime-idle-sweep").start() diff --git a/hermes_cli/local_runtime/capabilities.py b/hermes_cli/local_runtime/capabilities.py index 084d7a190a..f85a8b3253 100644 --- a/hermes_cli/local_runtime/capabilities.py +++ b/hermes_cli/local_runtime/capabilities.py @@ -8,98 +8,72 @@ screenshot silently leaving the machine). from __future__ import annotations -import json +from contextlib import suppress import logging -import urllib.request logger = logging.getLogger(__name__) from hermes_cli.local_runtime.endpoint import LLAMACPP_ALIASES as _LLAMACPP_ALIASES -# Image formats the managed server's decoder actually handles. llama.cpp -# decodes with stb_image: PNG/JPEG/GIF/BMP yes, WebP NO — and a WebP part -# fails SILENTLY (no HTTP error, no log line; the model just never sees an -# image and confabulates a description). Anything outside this set must be -# transcoded before the request. Measured against the live server: the -# same red square answered 'Red' as PNG and 'Unseen' as WebP. +# Image formats the managed server's decoder actually handles. llama.cpp decodes with stb_image: +# PNG/JPEG/GIF/BMP yes, WebP NO — and a WebP part fails SILENTLY (no HTTP error, no log line; the +# model just never sees an image and confabulates). Anything outside this set must be transcoded +# before the request. Measured live: the same red square answered 'Red' as PNG, 'Unseen' as WebP. ACCEPTED_IMAGE_MIMES = frozenset({"image/png", "image/jpeg"}) def is_managed_provider(provider: str, base_url: str = "") -> bool: - """True when this provider/base_url pair points at the managed server. - ``custom`` only counts when the base_url IS the managed endpoint — - background lookups must never claim someone else's custom server.""" + """True when this provider/base_url pair points at the managed server. ``custom`` only counts + when the base_url IS the managed endpoint — never claim someone else's custom server.""" p = (provider or "").strip().lower() if p in _LLAMACPP_ALIASES: return True if p != "custom" or not base_url: return False - try: + with suppress(Exception): from hermes_cli.local_runtime.growth import is_managed_endpoint return is_managed_endpoint(base_url) - except Exception: # noqa: BLE001 - return False + return False def _props_modalities(model_id: str) -> "bool | None": - """Ask the running server whether this loaded child sees images. None when the server is down, the - model isn't loaded, or the build doesn't report modalities. - """ - try: - from hermes_cli.local_runtime.endpoint import _state_endpoint + """Ask the running server whether this loaded child sees images. None when the server is down, + the model isn't loaded, or the build doesn't report modalities.""" + with suppress(Exception): + from hermes_cli.local_runtime.endpoint import managed_get_json, managed_root - state = _state_endpoint() - if state is None: + ep = managed_root() + if ep is None: return None - base = state["base_url"].rsplit("/v1", 1)[0] - req = urllib.request.Request( - f"{base}/props?model={model_id}", - headers={"Authorization": f"Bearer {state.get('api_key', '')}"}) - with urllib.request.urlopen(req, timeout=3) as r: - props = json.load(r) - modalities = props.get("modalities") + modalities = managed_get_json(*ep, f"/props?model={model_id}", timeout_s=3).get("modalities") if isinstance(modalities, dict) and "vision" in modalities: return bool(modalities["vision"]) - return None - except Exception: # noqa: BLE001 - return None + return None def managed_model_supports_vision(model_id: str) -> "bool | None": - """Ground-truth vision capability for a staged model, or None when the - model isn't ours / nothing is known (caller keeps falling through).""" + """Ground-truth vision capability for a staged model, or None when the model isn't ours / + nothing is known (caller keeps falling through).""" if not model_id: return None - # Only answer for models actually staged with us. - try: - from hermes_cli.local_runtime.bootstrap import staged_model_ids + with suppress(Exception): + from hermes_cli.local_runtime.bootstrap import assets_dir, staged_model_ids + from hermes_cli.local_runtime.catalog import entry_for_model + # Only answer for models actually staged with us. if model_id not in staged_model_ids(): return None - except Exception: # noqa: BLE001 - return None - - live = _props_modalities(model_id) - if live is not None: - return live - - # Staged but not loaded (or an older server build): the catalog knows - # whether this model ships a vision projector. - try: - from hermes_cli.local_runtime.bootstrap import assets_dir - from hermes_cli.local_runtime.catalog import find_entry_for_model - - hit = find_entry_for_model(model_id) - if hit is None: + live = _props_modalities(model_id) + if live is not None: + return live + # Staged but not loaded (or an older server build): the catalog knows whether this model + # ships a vision projector. Capability requires the projector to actually be on disk — a + # model downloaded before its mmproj (partial delete, old layout) genuinely cannot see. + entry = entry_for_model(model_id) + if entry is None: return None - entry = hit[0] - if entry.mmproj is None: - return False - # Capability requires the projector to actually be on disk — a - # model downloaded before its mmproj (partial delete, old layout) - # genuinely cannot see. - return (assets_dir() / entry.mmproj.local_name).exists() - except Exception: # noqa: BLE001 - return None + return (entry.mmproj is not None + and (assets_dir() / entry.mmproj.local_name).exists()) + return None diff --git a/hermes_cli/local_runtime/catalog.py b/hermes_cli/local_runtime/catalog.py index 09beef35f1..b13373fd6a 100644 --- a/hermes_cli/local_runtime/catalog.py +++ b/hermes_cli/local_runtime/catalog.py @@ -1,19 +1,15 @@ """Curated starter catalog for the managed local runtime. -Small and honest: every entry carries the estimator inputs (measured on real GGUFs) so the picker -can price a model BEFORE the user downloads gigabytes. Once a file is on disk, profile_from_gguf() -is the authority and the catalog numbers are only used for the download decision. - -Validation lifecycle: builds proven end-to-end on real hardware are marked validated. Day-0 entries -ship before that proof (they simply lack the validated flag) — ensure_model_ready's touch generation -still gates every first load at runtime. +Every entry carries the estimator inputs (measured on real GGUFs) so the picker can price a model +BEFORE the user downloads gigabytes; once a file is on disk, profile_from_gguf() is the authority. +Builds proven end-to-end on real hardware are marked validated; day-0 entries ship without the flag +— ensure_model_ready's touch generation still gates every first load at runtime. """ from __future__ import annotations import json import logging -import re import threading import time import urllib.request @@ -21,30 +17,20 @@ from dataclasses import dataclass, field from pathlib import PurePosixPath from hermes_cli.local_runtime.context_policy import ( - FLOOR, - RUNTIME_OVERHEAD_BYTES, - TARGET_WINDOW, - ub_logits_bytes, -) -from hermes_cli.local_runtime.estimator import ( - HardwareBudget, - LayerKind, - ModelProfile, - ctx_bytes, -) + FLOOR, RUNTIME_OVERHEAD_BYTES, TARGET_WINDOW, ub_logits_bytes) +from hermes_cli.local_runtime.estimator import HardwareBudget, LayerKind, ModelProfile, ctx_bytes +from hermes_cli.local_runtime.gguf import model_id_from_stem logger = logging.getLogger(__name__) -_PART_SUFFIX = re.compile(r"-\d{5}-of-\d{5}$") - @dataclass(frozen=True) class AssetFile: - """One downloadable file: repo-relative path and exact bytes (the size feeds the estimator and the - download progress bar; there is no download-time integrity check by design — a corrupt file - surfaces as a llama.cpp load error). ``local`` overrides the on-disk name (repos reuse generic - names like mmproj-BF16.gguf across models). Non-model extras live under the models dir's assets/ - subdirectory so the router never lists them. + """One downloadable file: repo-relative path and exact bytes (feeds the estimator and the + progress bar; no download-time integrity check by design — a corrupt file surfaces as a + llama.cpp load error). ``local`` overrides the on-disk name (repos reuse generic names like + mmproj-BF16.gguf). Non-model extras live under the models dir's assets/ subdirectory so the + router never lists them. """ path: str # repo-relative (may include a subdir) @@ -58,8 +44,8 @@ class AssetFile: @dataclass(frozen=True) class QuantVariant: - """One downloadable build of a model. Split GGUFs list every part in - files; the model loads from the first part.""" + """One downloadable build. Split GGUFs list every part in files; the model loads from the + first part.""" quant: str # e.g. "UD-Q4_K_M" files: tuple # AssetFile, first = the load target @@ -67,8 +53,7 @@ class QuantVariant: @property def model_id(self) -> str: - stem = PurePosixPath(self.files[0].path).name.removesuffix(".gguf") - return _PART_SUFFIX.sub("", stem) + return model_id_from_stem(PurePosixPath(self.files[0].path).name.removesuffix(".gguf")) @property def size_bytes(self) -> int: @@ -76,9 +61,8 @@ class QuantVariant: @property def weights_bytes(self) -> int: - """Pre-download weights estimate: GGUF bytes ≈ tensor bytes + a small header (<2%) — a safe, - slightly conservative stand-in until profile_from_gguf reads the real table. - """ + """Pre-download weights estimate: GGUF bytes ≈ tensor bytes + a <2% header — slightly + conservative until profile_from_gguf reads the real table.""" return self.size_bytes @@ -89,9 +73,9 @@ class CatalogEntry: description: str # one line, plain language repo: str # HF repo variants: tuple # QuantVariant (exactly one, Q4-class) - # Estimator inputs (measured or config-derived; quant changes weights, - # never KV). Entries with gated upstream configs carry a conservative - # same-family prior — the GGUF header is the authority after download. + # Estimator inputs (measured or config-derived; quant changes weights, never KV). Gated + # upstream configs carry a conservative same-family prior — the GGUF header is the authority + # after download. n_ctx_train: int full_layers: int recurrent_layers: int @@ -100,32 +84,25 @@ class CatalogEntry: swa_window: int = 0 moe: bool = False mtp: bool = False # ships MTP heads (spec decode when loaded) - # Speculative draft depth for MTP models. Per-model and measured: - # deeper drafting pays only while draft acceptance holds, and the - # break-even depth differs by model. + # Speculative draft depth for MTP models: per-model and measured — deeper drafting pays only + # while draft acceptance holds, and the break-even depth differs by model. mtp_draft_depth: int = 3 - # Vocab size prices the GPU logits buffers (ubatch x vocab x fp32, - # doubled under MTP backend sampling) — a multi-GiB term at large - # vocab sizes that a weights-only fit would miss. + # Vocab size prices the GPU logits buffers (ubatch x vocab x fp32, doubled under MTP backend + # sampling) — a multi-GiB term at large vocabs that a weights-only fit would miss. n_vocab: int = 0 mmproj: "AssetFile | None" = None # vision projector, downloads with model draft: "AssetFile | None" = None # spec-decode draft model (e.g. DSpark) sampling: dict = field(default_factory=dict) # INI long-form launch defaults - # Oldest llama.cpp release tag that can load this model (day-0 - # architectures need the release where their support landed). Empty - # means any installed engine. The pane gates download/activate on it. + # Oldest llama.cpp release tag that can load this model (day-0 architectures need the release + # where their support landed). Empty means any installed engine. min_engine: str = "" - # Editorial quality ordering (higher = smarter), authored once, - # globally, at catalog-authoring time — Artificial Analysis-informed - # where they cover the model (scripts/aa_quality_sync.py proposes, - # the commit decides), editorial elsewhere. Ranks entries for the - # per-machine recommendation; never displayed as a score (it grades - # the full-precision model, not our Q4 build). + # Editorial quality ordering (higher = smarter), authored once at catalog time — Artificial + # Analysis-informed where covered (scripts/aa_quality_sync.py proposes, the commit decides). + # Ranks entries for the per-machine recommendation; never displayed as a score (it grades the + # full-precision model, not our Q4 build). quality: int = 0 - # Fraction of the build's bytes read per decoded token: 1.0 for dense - # models (every weight streams every token), the active slice for MoE - # (attention + shared + routed experts over total). With memory - # bandwidth this predicts decode speed — the physics half of the + # Fraction of the build's bytes read per decoded token: 1.0 for dense, the active slice for + # MoE. With memory bandwidth this predicts decode speed — the physics half of the # recommendation. decode_fraction: float = 1.0 @@ -134,11 +111,9 @@ class CatalogEntry: + [(LayerKind.SWA, self.per_layer_f16)] * self.swa_layers + [(LayerKind.RECURRENT, 0)] * self.recurrent_layers) return ModelProfile( - name=variant.model_id, weights_bytes=variant.weights_bytes, - embd_table_bytes=0, n_ctx_train=self.n_ctx_train, - layers=layers, swa_window=self.swa_window, moe=self.moe, - n_vocab=self.n_vocab, - kv_scale=1.2 if self.mtp else 1.0) + name=variant.model_id, weights_bytes=variant.weights_bytes, embd_table_bytes=0, + n_ctx_train=self.n_ctx_train, layers=layers, swa_window=self.swa_window, moe=self.moe, + n_vocab=self.n_vocab, kv_scale=1.2 if self.mtp else 1.0) def download_files(self, variant: QuantVariant) -> tuple: """Everything a download job fetches for this variant, in order.""" @@ -151,8 +126,8 @@ class CatalogEntry: @dataclass(frozen=True) class VariantChoice: - """Selection result: which build this machine should download and why. - reason_key is a UI-copy discriminator, not display text.""" + """Which build this machine should download and why. reason_key is a UI-copy + discriminator, not display text.""" variant: QuantVariant zero_spill: bool @@ -160,14 +135,11 @@ class VariantChoice: def select_variant(entry: CatalogEntry, budget: HardwareBudget) -> VariantChoice | None: - """Fit the entry's one build (Q4-class) to this machine. + """Fit the entry's one Q4-class build to this machine; headroom buys a bigger window, never a + bigger quant. - Every entry ships exactly one variant (see the module docstring for why there is no quant - ladder); headroom buys a bigger window, never a bigger quant. The fit shapes: - - - "best-large-window": zero-spills at TARGET_WINDOW - "best-fits": zero-spills at the 64K floor - - "smallest-fits-spilled": weights spill to host RAM, priced honestly - None: even spilled, - physics refuses (the machine can't run it) + "best-large-window": zero-spill at TARGET_WINDOW; "best-fits": zero-spill at the 64K floor; + "smallest-fits-spilled": weights spill to host RAM, priced honestly; None: physics refuses. """ overhead = (RUNTIME_OVERHEAD_BYTES + (entry.mmproj.size_bytes if entry.mmproj else 0) @@ -189,35 +161,24 @@ def select_variant(entry: CatalogEntry, budget: HardwareBudget) -> VariantChoice # ── recommendation: best quality that fits and isn't miserably slow ── # -# Two axes, each living where it belongs. QUALITY is a judgment made once, -# globally, at authoring time (entry.quality — AA-informed, editorially -# owned). SPEED is physics computed per machine: decode is memory-bound, -# so predicted tok/s ≈ bandwidth / bytes-read-per-token, and the bytes per -# token are the build's size scaled by its decode fraction (dense reads -# everything; MoE reads the active slice). The pick: highest quality among -# entries that run resident and clear a pleasant speed floor; else the -# fastest resident entry; else the least-painful spilled one. -# -# The bandwidth axis is the `uma` flag for now: every discrete card that -# matters is 900+ GB/s GDDR while the unified-memory class measures ~1/5th -# of that, so the flag IS the high/low split. A measured per-machine -# bandwidth (one cached memcpy probe) can replace these class constants -# without touching the rule; predictions order candidates and gate the -# floor — they are not display values. +# QUALITY is a judgment made once at authoring time (entry.quality). SPEED is physics per machine: +# decode is memory-bound, so predicted tok/s ≈ bandwidth / bytes-read-per-token (build size scaled +# by decode fraction). The bandwidth axis is the `uma` flag: every discrete card that matters is +# 900+ GB/s GDDR while the unified-memory class measures ~1/5th of that. A measured per-machine +# bandwidth could replace these class constants without touching the rule; predictions order +# candidates and gate the floor — they are not display values. _DISCRETE_BANDWIDTH_GB_S = 1000.0 # representative GDDR6X/GDDR7 class _UMA_BANDWIDTH_GB_S = 210.0 # measured on unified-memory NVIDIA _HOST_BANDWIDTH_GB_S = 80.0 # spilled weights stream over host DRAM -# The one editorial constant in the tree: below this predicted decode -# speed a model stops feeling pleasant for agentic use (roughly reading -# speed with headroom for tool-call bursts). Distinct from the growth -# policy's 6 tok/s compress floor, which marks unusable, not unpleasant. +# Below this predicted decode speed a model stops feeling pleasant for agentic use (roughly +# reading speed with headroom for tool-call bursts). Distinct from the growth policy's 6 tok/s +# compress floor, which marks unusable, not unpleasant. PLEASANT_FLOOR_TOK_S = 20.0 -def predicted_decode_tok_s(entry: CatalogEntry, variant: QuantVariant, - budget: HardwareBudget, *, +def predicted_decode_tok_s(entry: CatalogEntry, variant: QuantVariant, budget: HardwareBudget, *, spilled: bool = False) -> float: """Memory-bound decode prediction for ordering and floor-gating.""" bandwidth = (_HOST_BANDWIDTH_GB_S if spilled @@ -230,16 +191,13 @@ def predicted_decode_tok_s(entry: CatalogEntry, variant: QuantVariant, def recommended_entry(budget: HardwareBudget, entries: "tuple[CatalogEntry, ...] | None" = None ) -> "tuple[CatalogEntry, str] | None": - """The catalog's default pick for THIS machine, with its reason. + """The catalog's default pick for THIS machine, with its reason key. Callers pass pre-filtered entries when some are ineligible for reasons the catalog can't know - (engine too old); default is the full catalog. - - best-quality-resident quality won among resident entries that clear the pleasant floor speed- - gated-quality same, but the floor eliminated a HIGHER quality candidate — the exact 'why not the - big model?' a unified-memory owner asks fastest-resident nothing resident clears the floor; the - quickest resident entry wins least-painful-spilled nothing runs resident; fastest from host - memory (MoE by construction) + (engine too old). Reasons: best-quality-resident (quality won among resident entries clearing + the pleasant floor); speed-gated-quality (same, but the floor eliminated a HIGHER quality + candidate); fastest-resident (nothing resident clears the floor); least-painful-spilled + (nothing runs resident; fastest from host memory — MoE by construction). """ pool = CATALOG if entries is None else entries fitting = [(e, c) for e in pool if (c := select_variant(e, budget)) is not None] @@ -257,22 +215,16 @@ def recommended_entry(budget: HardwareBudget, return (pick, "speed-gated-quality" if floor_gated else "best-quality-resident") if resident: return (max(resident, key=speed)[0], "fastest-resident") - # Everything spills: take the least painful — fastest predicted decode - # from host memory (MoE wins here by construction; a dense spill - # streams every weight over the host bus). return (max(fitting, key=lambda t: speed(t, spilled=True))[0], "least-painful-spilled") # ── catalog data: packaged JSON, refreshed from GitHub in memory ─ # -# The catalog DATA lives in catalog.json (checked in beside this module -# and shipped as package data); this module keeps all policy. At import -# we load the packaged copy — no network on the import path. A TTL-gated -# background refresh fetches the same file from the repo's main branch -# and swaps it in memory only: nothing on disk changes, so a git -# checkout never sees a dirty tracked file and the packaged copy remains -# the offline truth. A reverted commit on main heals every install on -# its next fetch, and day-0 entries reach users without an app release. +# catalog.json ships as package data and is loaded at import (no network on the import path). A +# TTL-gated background refresh fetches the same file from the repo's main branch and swaps it in +# MEMORY only: nothing on disk changes, so a git checkout never sees a dirty tracked file and the +# packaged copy remains the offline truth. A reverted commit on main heals every install on its +# next fetch, and day-0 entries reach users without an app release. _CATALOG_URL = ("https://raw.githubusercontent.com/NousResearch/hermes-agent" "/main/hermes_cli/local_runtime/catalog.json") @@ -285,12 +237,10 @@ _last_refresh_attempt = 0.0 def _asset_from(d: "dict | None") -> "AssetFile | None": if not d: return None - return AssetFile(path=d["path"], size_bytes=int(d["size_bytes"]), - local=d.get("local")) + return AssetFile(path=d["path"], size_bytes=int(d["size_bytes"]), local=d.get("local")) -# Scalar CatalogEntry fields parsed from JSON: key -> (coerce, default); -# a None default means the key is required. +# Scalar CatalogEntry fields parsed from JSON: key -> (coerce, default); None default = required. _SCALAR_FIELDS = { "n_ctx_train": (int, None), "full_layers": (int, None), "recurrent_layers": (int, None), "per_layer_f16": (int, None), @@ -302,19 +252,16 @@ _SCALAR_FIELDS = { def _load_catalog(doc: dict) -> "tuple[CatalogEntry, ...]": - """Parse a catalog document into entries. Unknown fields are ignored - (newer catalogs stay readable by older apps); a major schema bump is - the signal that they wouldn't be, and the caller skips the document.""" + """Parse a catalog document. Unknown fields are ignored (newer catalogs stay readable by older + apps); a major schema bump is the signal that they wouldn't be, and the caller skips it.""" if int(doc.get("schema_version", 0)) != _SCHEMA_VERSION: raise ValueError(f"catalog schema {doc.get('schema_version')!r} " f"(this build reads {_SCHEMA_VERSION})") entries = [] for m in doc["models"]: - variants = tuple( - QuantVariant(quant=v["quant"], - files=tuple(_asset_from(f) for f in v["files"]), - validated=bool(v.get("validated"))) - for v in m["variants"]) + variants = tuple(QuantVariant(quant=v["quant"], validated=bool(v.get("validated")), + files=tuple(_asset_from(f) for f in v["files"])) + for v in m["variants"]) scalars = {k: coerce(m[k] if default is None else m.get(k, default)) for k, (coerce, default) in _SCALAR_FIELDS.items()} entries.append(CatalogEntry( @@ -328,8 +275,7 @@ def _load_catalog(doc: dict) -> "tuple[CatalogEntry, ...]": def _packaged_catalog() -> "tuple[CatalogEntry, ...]": from importlib.resources import files - raw = files("hermes_cli.local_runtime").joinpath("catalog.json").read_text( - encoding="utf-8") + raw = files("hermes_cli.local_runtime").joinpath("catalog.json").read_text(encoding="utf-8") return _load_catalog(json.loads(raw)) @@ -337,12 +283,9 @@ CATALOG: "tuple[CatalogEntry, ...]" = _packaged_catalog() def refresh_catalog(force: bool = False) -> bool: - """Fetch the current catalog from the repo and swap it in memory. - - Best-effort by design: any failure (offline, GitHub down, unreadable schema) leaves the running - catalog untouched and retries after the TTL. Returns True when a fetched document replaced the - catalog. - """ + """Fetch the current catalog from the repo and swap it in memory. Best-effort: any failure + (offline, GitHub down, unreadable schema) leaves the running catalog untouched and retries + after the TTL. Returns True when a fetched document replaced the catalog.""" global CATALOG, _last_refresh_attempt now = time.monotonic() @@ -351,8 +294,7 @@ def refresh_catalog(force: bool = False) -> bool: return False _last_refresh_attempt = now try: - req = urllib.request.Request( - _CATALOG_URL, headers={"User-Agent": "hermes-local-runtime"}) + req = urllib.request.Request(_CATALOG_URL, headers={"User-Agent": "hermes-local-runtime"}) with urllib.request.urlopen(req, timeout=10) as r: fetched = _load_catalog(json.load(r)) except Exception as exc: # noqa: BLE001 @@ -365,13 +307,11 @@ def refresh_catalog(force: bool = False) -> bool: def refresh_catalog_soon() -> None: - """TTL-gated background refresh; returns immediately. The caller's current request serves the - catalog it already has — the refresh lands for the next one. - """ + """TTL-gated background refresh; returns immediately. The current request serves the catalog + it already has — the refresh lands for the next one.""" if time.monotonic() - _last_refresh_attempt < _REFRESH_TTL_S: return - threading.Thread(target=refresh_catalog, daemon=True, - name="catalog-refresh").start() + threading.Thread(target=refresh_catalog, daemon=True, name="catalog-refresh").start() def catalog_by_id() -> dict[str, CatalogEntry]: @@ -385,3 +325,8 @@ def find_entry_for_model(model_id: str) -> "tuple[CatalogEntry, QuantVariant] | if variant.model_id == model_id: return entry, variant return None + + +def entry_for_model(model_id: str) -> "CatalogEntry | None": + hit = find_entry_for_model(model_id) + return hit[0] if hit is not None else None diff --git a/hermes_cli/local_runtime/context_policy.py b/hermes_cli/local_runtime/context_policy.py index 682580f8f7..c004690285 100644 --- a/hermes_cli/local_runtime/context_policy.py +++ b/hermes_cli/local_runtime/context_policy.py @@ -1,10 +1,8 @@ """Context policy — the window ladder for managed local models. One contract: any model runs at any window up to its native max; hardware and session depth only -change tokens/s. Constants, not knobs — nothing in this module reads config. - -The policy encodes behavior measured on real hardware (llama.cpp, discrete NVIDIA GPUs on -Windows/WDDM, and unified-memory devices): +change tokens/s. Constants, not knobs — nothing in this module reads config. The policy encodes +behavior measured on real hardware (llama.cpp, discrete NVIDIA on Windows/WDDM, unified-memory). """ from __future__ import annotations @@ -12,12 +10,7 @@ from __future__ import annotations from dataclasses import dataclass, field from hermes_cli.local_runtime.estimator import ( - HardwareBudget, - ModelProfile, - PhysicsRefusal, - ctx_bytes, - physics_check, -) + HardwareBudget, ModelProfile, PhysicsRefusal, ctx_bytes, physics_check) FLOOR = 64 * 1024 # = target; one internal constant _LADDER_GROWTH = 1.5 @@ -25,22 +18,17 @@ _GROW_AT_OCCUPANCY = 0.85 # of the current window, at turn boundary SPEED_FLOOR_TOK_S = 6.0 # deepest measured spill bottomed near this _EARLY_COST_CTX_FRACTION = 0.15 # bounded early cost when weights spill -# TARGET_WINDOW: the smallest ladder rung at which compression becomes the -# exception rather than the routine. Measured over 161 real agentic -# sessions: 66% complete uncompressed in 64K, 82% in 96K, 91% in 144K — -# and the marginal gain past 144K (+6 points for 216K) falls below the -# quality cost of stepping down another quant. Quant selection prefers -# the best build that reaches this; the FLOOR remains the guarantee. +# The smallest ladder rung at which compression becomes the exception rather than the routine. +# Measured over 161 real agentic sessions: 66% complete uncompressed in 64K, 82% in 96K, 91% in +# 144K — and the marginal gain past 144K (+6 points for 216K) falls below the quality cost of +# stepping down another quant. The FLOOR remains the guarantee. TARGET_WINDOW = 144 * 1024 -# What a load really costs beyond weights + KV: CUDA contexts and compute -# buffers at the DEFAULT microbatch (-ub 512, no MTP). Measured on a -# 32 GiB card: a model estimated at 29.3 GiB (weights+KV) loaded at -# ~31.2 GiB resident and the server's own fit still shaved a layer to -# CPU. Microbatch/MTP logits buffers are priced separately per model -# (ub_logits_bytes — they scale with the model's vocab and doubled once -# packed a card 3.9 GiB past this constant). Callers add mmproj bytes on -# top. +# What a load really costs beyond weights + KV: CUDA contexts and compute buffers at the DEFAULT +# microbatch (-ub 512, no MTP). Measured on a 32 GiB card: a model estimated at 29.3 GiB loaded at +# ~31.2 GiB resident and fit still shaved a layer to CPU. Microbatch/MTP logits buffers are priced +# separately per model (ub_logits_bytes — they scale with vocab and once packed a card 3.9 GiB +# past this constant). Callers add mmproj bytes on top. RUNTIME_OVERHEAD_BYTES = int(1.5 * (1 << 30)) @@ -67,15 +55,14 @@ class WindowDecision: return self.spill_bytes > 0 -def initial_window(profile: ModelProfile, budget: HardwareBudget, - *, flash_attention: bool = True, +def initial_window(profile: ModelProfile, budget: HardwareBudget, *, flash_attention: bool = True, overhead_bytes: int = 0) -> WindowDecision | PhysicsRefusal: """The launch decision: largest cheap rung, never below the floor. Zero-spill rung: weights + ctx + overhead fit usable VRAM entirely. Bounded-early-cost rung (weights already exceed VRAM): largest rung whose ctx stays <= ~15% of usable VRAM. Floor - everywhere, capped at native. ``overhead_bytes`` is runtime cost beyond weights+KV; zero - keeps this pure physics for decision-table tests, production callers pass it. + everywhere, capped at native. ``overhead_bytes`` is runtime cost beyond weights+KV; zero keeps + this pure physics for decision-table tests, production callers pass it. """ refusal = physics_check(profile, budget, FLOOR, flash_attention=flash_attention) if refusal: @@ -84,38 +71,35 @@ def initial_window(profile: ModelProfile, budget: HardwareBudget, native = profile.n_ctx_train or FLOOR rungs = ladder(native) - reasons: list[str] = [] + def kv(rung: int) -> int: + return ctx_bytes(profile, rung, flash_attention=flash_attention) + best_zero_spill: int | None = None for rung in rungs: - need = (profile.weights_bytes + overhead_bytes - + ctx_bytes(profile, rung, flash_attention=flash_attention)) - if need <= budget.usable_vram_bytes: - best_zero_spill = rung - else: + if profile.weights_bytes + overhead_bytes + kv(rung) > budget.usable_vram_bytes: break + best_zero_spill = rung if best_zero_spill is not None and best_zero_spill >= min(FLOOR, native): window = best_zero_spill - reasons.append(f"largest zero-spill rung ({window // 1024}K)") + reason = f"largest zero-spill rung ({window // 1024}K)" else: - # Weights spill from turn one (steep-curve model on a small card) — - # hold the floor, bound the early ctx cost. + # Weights spill from turn one (steep-curve model on a small card) — hold the floor, bound + # the early ctx cost. cap = int(budget.usable_vram_bytes * _EARLY_COST_CTX_FRACTION) window = min(FLOOR, native) for rung in rungs: if rung < window: continue - if ctx_bytes(profile, rung, flash_attention=flash_attention) <= cap: - window = rung - else: + if kv(rung) > cap: break - reasons.append(f"floor held at {window // 1024}K; weights spill (deliberate price of the guarantee)") + window = rung + reason = f"floor held at {window // 1024}K; weights spill (deliberate price of the guarantee)" - kv = ctx_bytes(profile, window, flash_attention=flash_attention) - spill = max(0, profile.weights_bytes + kv - budget.usable_vram_bytes) - return WindowDecision(window=window, spill_bytes=spill, - kv_on_gpu=kv <= budget.usable_vram_bytes, - reasons=reasons) + kv_bytes = kv(window) + return WindowDecision(window=window, reasons=[reason], + spill_bytes=max(0, profile.weights_bytes + kv_bytes - budget.usable_vram_bytes), + kv_on_gpu=kv_bytes <= budget.usable_vram_bytes) @dataclass @@ -126,17 +110,15 @@ class GrowthDecision: def growth_decision(profile: ModelProfile, budget: HardwareBudget, *, - current_window: int, session_tokens: int, - measured_decode_tok_s: float | None, - server_idle: bool, - flash_attention: bool = True, + current_window: int, session_tokens: int, measured_decode_tok_s: float | None, + server_idle: bool, flash_attention: bool = True, occupancy_confirmed: bool = False) -> GrowthDecision: """One growth evaluation, END-OF-TURN ONLY (recurrent state cannot rewind mid-sequence). Gate order: occupancy (~85%) → native cap → idleness (growth only on an otherwise-idle - server) → speed floor (below it compression is the default) → re-fit against LIVE free - memory (the rung must fit NOW, not at launch). ``occupancy_confirmed`` skips gate 1 when the - caller's own compression gate already fired, so two edge definitions can't deadlock into + server) → speed floor (below it compression is the default) → re-fit against LIVE free memory + (the rung must fit NOW, not at launch). ``occupancy_confirmed`` skips gate 1 when the caller's + own compression gate already fired, so two edge definitions can't deadlock into compress-before-grow. """ if not occupancy_confirmed and session_tokens < current_window * _GROW_AT_OCCUPANCY: @@ -144,8 +126,7 @@ def growth_decision(profile: ModelProfile, budget: HardwareBudget, *, native = profile.n_ctx_train or current_window if current_window >= native: - return GrowthDecision("compress-default", - reason="at native window; compression is the only move") + return GrowthDecision("compress-default", reason="at native window; compression is the only move") if not server_idle: return GrowthDecision("hold", reason="server busy; re-grant deferred to idle") @@ -159,13 +140,11 @@ def growth_decision(profile: ModelProfile, budget: HardwareBudget, *, next_rung = next((r for r in ladder(native) if r > current_window), native) - # Re-fit against live free memory: allocation beyond residency is the - # slow path, so a rung that no longer fits doesn't get granted. + # Re-fit against live free memory: allocation beyond residency is the slow path, so a rung + # that no longer fits doesn't get granted. kv = ctx_bytes(profile, next_rung, flash_attention=flash_attention) - total_need = profile.weights_bytes + kv - if total_need > budget.usable_vram_bytes + budget.ram_available_bytes: - return GrowthDecision("compress-default", - reason="next rung exceeds physics; compression instead") + if profile.weights_bytes + kv > budget.usable_vram_bytes + budget.ram_available_bytes: + return GrowthDecision("compress-default", reason="next rung exceeds physics; compression instead") return GrowthDecision("grow", next_window=next_rung, reason=f"rung {current_window // 1024}K -> {next_rung // 1024}K") @@ -173,9 +152,8 @@ def growth_decision(profile: ModelProfile, budget: HardwareBudget, *, def spill_overrides(profile: ModelProfile) -> list[str]: """-ot placement for spilled configs: expert/FFN weights to host so attention + KV stay - GPU-resident. MoE gets the expert pattern; hybrids push recurrent-layer FFNs (their n_head_kv==0 - layers carry no KV worth protecting). - """ + GPU-resident. MoE gets the expert pattern; hybrids push recurrent-layer FFNs (their + n_head_kv==0 layers carry no KV worth protecting).""" if profile.moe: return ["-ot", r"blk\.\d+\.ffn_.*_exps\.weight=CPU"] if profile.recurrent_layer_count: @@ -183,20 +161,15 @@ def spill_overrides(profile: ModelProfile) -> list[str]: return [] # dense: fit's back-to-front layer cut is the only axis -def launch_args(profile: ModelProfile, decision: WindowDecision, *, - flash_attention: bool = True, - mtp_capable: bool = False, - mtp_draft_depth: int = 3, - uma: bool = False, +def launch_args(profile: ModelProfile, decision: WindowDecision, *, flash_attention: bool = True, + mtp_capable: bool = False, mtp_draft_depth: int = 3, uma: bool = False, mtp_prefill: bool = False) -> list[str]: """Per-model launch flags from a window decision. Explicit -c puts fit into spill-weights-and-hold-ctx; q8 KV cache wherever flash attention exists; -ot placement on - spilled configs — DISCRETE cards only. - """ + spilled configs — DISCRETE cards only.""" args = ["-c", str(decision.window)] if mtp_capable: - args += ["--spec-type", "draft-mtp", - "--spec-draft-n-max", str(mtp_draft_depth), + args += ["--spec-type", "draft-mtp", "--spec-draft-n-max", str(mtp_draft_depth), "--backend-sampling", "--spec-draft-backend-sampling"] if mtp_prefill: args += ["-b", "4096", "-ub", "2048"] @@ -209,12 +182,10 @@ def launch_args(profile: ModelProfile, decision: WindowDecision, *, return args -def ub_logits_bytes(n_vocab: int, *, mtp_capable: bool, - mtp_prefill: bool = False) -> int: - """GPU logits/compute-buffer cost of the microbatch posture chosen by launch_args, priced from the - model's own vocab and calibrated against measured server RSS (Qwen3.8 Q4, both postures, three - windows): - """ +def ub_logits_bytes(n_vocab: int, *, mtp_capable: bool, mtp_prefill: bool = False) -> int: + """GPU logits/compute-buffer cost of the microbatch posture chosen by launch_args, priced from + the model's own vocab and calibrated against measured server RSS (Qwen3.8 Q4, both postures, + three windows).""" v = max(0, int(n_vocab)) if mtp_capable and mtp_prefill: return int(2048 * v * 4 * 1.5) diff --git a/hermes_cli/local_runtime/detect.py b/hermes_cli/local_runtime/detect.py index 376054555a..120301f95c 100644 --- a/hermes_cli/local_runtime/detect.py +++ b/hermes_cli/local_runtime/detect.py @@ -1,9 +1,8 @@ """Detection of running llama-server instances. Probes well-known local roots and fingerprints genuine llama-server via /props (build_info + model -fields — Ollama and LM Studio answer /v1/models but not /props). The credential is reachability; -detection never needs a key, but honors one if the probed server requires it (401 -> detected, -auth_required=True). +fields — Ollama and LM Studio answer /v1/models but not /props). Detection never needs a key, but +honors one if the probed server requires it (401 -> detected, auth_required=True). """ from __future__ import annotations @@ -43,25 +42,19 @@ def probe_port(port: int) -> DetectedServer | None: root = f"http://127.0.0.1:{port}" status, props = _get(f"{root}/props") if status == 401: - return DetectedServer(base_url=f"{root}/v1", build_info="", model_path="", - n_ctx=None, router_mode=False, auth_required=True) + return DetectedServer(f"{root}/v1", "", "", None, router_mode=False, auth_required=True) if status != 200 or not isinstance(props, dict): return None build = str(props.get("build_info", "")) if not build: return None # answers /props but isn't llama-server dgs = props.get("default_generation_settings") - n_ctx = dgs.get("n_ctx") if isinstance(dgs, dict) else None models_status, models = _get(f"{root}/models") return DetectedServer( - base_url=f"{root}/v1", - build_info=build, - model_path=str(props.get("model_path", "")), - n_ctx=n_ctx, - router_mode=(models_status == 200 and isinstance(models, dict) - and "data" in models), - auth_required=False, - ) + base_url=f"{root}/v1", build_info=build, model_path=str(props.get("model_path", "")), + n_ctx=dgs.get("n_ctx") if isinstance(dgs, dict) else None, + router_mode=models_status == 200 and isinstance(models, dict) and "data" in models, + auth_required=False) def detect_server(extra_ports: tuple[int, ...] = ()) -> DetectedServer | None: diff --git a/hermes_cli/local_runtime/endpoint.py b/hermes_cli/local_runtime/endpoint.py index 01734465bf..ac1a2ffd81 100644 --- a/hermes_cli/local_runtime/endpoint.py +++ b/hermes_cli/local_runtime/endpoint.py @@ -1,16 +1,17 @@ """Endpoint resolution for llamacpp-alias requests (provider integration). -The seam between the existing provider mechanism and the managed runtime: ``provider: llamacpp`` -with no explicit base_url resolves, in order, to +``provider: llamacpp`` with no explicit base_url resolves, in order, to the managed server (state +file), a detected external llama-server, or — during a backend boot race — the managed server once +its state file appears. """ from __future__ import annotations +from contextlib import suppress import json import logging import threading import time -import urllib.error import urllib.request LLAMACPP_ALIASES = frozenset({"llamacpp", "llama.cpp", "llama-cpp"}) @@ -19,19 +20,15 @@ logger = logging.getLogger(__name__) def _pid_alive(pid: int) -> bool: - """Liveness for the state file's supervisor-child pid. - - psutil when available; otherwise fall back to True (optimistic) — on Windows ``os.kill(pid, 0)`` - TERMINATES the process, so it must never be used as a probe (windows-git-bash interop pitfall). - """ + """Liveness for the state file's supervisor-child pid: psutil when available, else True + (optimistic). On Windows ``os.kill(pid, 0)`` TERMINATES the process — never use it as a probe.""" if not pid or pid < 0: return False - try: + with suppress(Exception): import psutil # type: ignore return psutil.pid_exists(pid) - except Exception: # noqa: BLE001 - return True + return True def _state_endpoint() -> dict | None: @@ -47,30 +44,38 @@ def _state_endpoint() -> dict | None: base_url = state.get("base_url", "") if not base_url: return None - endpoint = {"base_url": base_url, "api_key": state.get("api_key", "")} - # Ownership proof: the stable port means a SECOND install (different - # HERMES_HOME — a scratch profile, say) can own 127.0.0.1:18434 with a - # different api key while this install's state file still points there. - # /health is a public route, so it answers 200 for ANYONE's server — - # trusting it alone sent every chat request and the load-progress - # watcher at a server that 401s our key, silently. The recorded - # supervisor pid is the tiebreaker: health-200 from a server whose - # recorded child is DEAD is someone else's server, never a starting one. - pid_ok = _pid_alive(int(state.get("pid") or 0)) - # Healthy server: done (when it's ours). - try: - health = base_url.rsplit("/v1", 1)[0] + "/health" - with urllib.request.urlopen(health, timeout=3) as r: - if r.status == 200: - return endpoint if pid_ok else None - except (urllib.error.URLError, OSError, TimeoutError): - pass - # Not healthy YET: a live supervisor child is a STARTING server (state - # is written at spawn; llama-server takes seconds to listen). Resolve - # optimistically so readiness probes racing the boot see a configured - # provider, not missing credentials. A dead pid is a crashed-without- - # cleanup leftover — ignore it so requests don't blackhole. - return endpoint if pid_ok else 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)): + return None + return {"base_url": base_url, "api_key": state.get("api_key", "")} + + +def managed_root() -> "tuple[str, str] | None": + """(base_root, api_key) of the managed router, or None. Resolved through the + ownership-guarded reader, not a raw state-file read: on the shared stable port a foreign + install's server answers /health for anyone, and a raw read would attach callers to someone + else's server.""" + with suppress(Exception): + state = _state_endpoint() + if state is None: + return None + base = str(state.get("base_url", "")).rsplit("/v1", 1)[0] + return (base, str(state.get("api_key", ""))) if base else None + return None + + +def managed_get_json(base: str, api_key: str, route: str, timeout_s: float) -> object: + """Authenticated GET against the managed router; raises on any transport/decode failure.""" + req = urllib.request.Request(f"{base}{route}", + headers={"Authorization": f"Bearer {api_key}"}) + with urllib.request.urlopen(req, timeout=timeout_s) as r: + return json.loads(r.read()) def resolve_llamacpp_endpoint(config: dict | None = None, @@ -78,8 +83,8 @@ def resolve_llamacpp_endpoint(config: dict | None = None, """Managed-first, detection-second endpoint for llamacpp aliases. Boot-race rung: on a fresh backend start there is NO state file yet — the lifespan boot thread - is still spawning the server (config load + preset generation + spawn ≈ 1-3 s) while the - desktop's readiness probe fires the moment the WebSocket connects. + is still spawning the server (≈1-3 s) while the desktop's readiness probe fires the moment the + WebSocket connects. """ managed = _state_endpoint() if managed: @@ -115,11 +120,8 @@ def _load_config_if_none(config: dict | None) -> dict | None: def _kick_managed_boot(config: dict | None) -> None: - """Actively start the managed server when resolution finds it missing. - - The wait loop above assumes some OTHER thread is bringing the server up — true only at backend - start (the lifespan boot thread). - """ + """Actively start the managed server when resolution finds it missing — the wait loop assumes + some OTHER thread is bringing it up, which is true only at backend start.""" if not _KICK_LOCK.acquire(blocking=False): return # a kick is already in flight @@ -138,24 +140,14 @@ def _kick_managed_boot(config: dict | None) -> None: def _boot_in_flight(config: dict | None) -> bool: - """True when the managed runtime is enabled and installed — the state - - Installed-ness is a verified-manifest scan under runtimes_root(), NOT a bare ``server_binary()`` - call: that helper needs an install_dir, and calling it bare made this gate throw-and-return - False forever, silently disabling the boot wait. - """ - try: + """True when the managed runtime is enabled and installed (a verified-manifest scan under + runtimes_root(), NOT a bare ``server_binary()`` call — that needs an install_dir, and calling + it bare once made this gate throw-and-return False forever, disabling the boot wait).""" + with suppress(Exception): config = _load_config_if_none(config) if not ((config or {}).get("local_runtime") or {}).get("enabled"): return False - from hermes_cli.local_runtime.binaries import runtimes_root + from hermes_cli.local_runtime.binaries import manifest_verified, runtimes_root - for manifest in runtimes_root().glob("*/*/manifest.json"): - try: - if json.loads(manifest.read_text(encoding="utf-8")).get("verified_version"): - return True - except (ValueError, OSError): - continue - return False - except Exception: # noqa: BLE001 - return False + return any(manifest_verified(m) for m in runtimes_root().glob("*/*/manifest.json")) + return False diff --git a/hermes_cli/local_runtime/estimator.py b/hermes_cli/local_runtime/estimator.py index 988a88a2a1..e549460aea 100644 --- a/hermes_cli/local_runtime/estimator.py +++ b/hermes_cli/local_runtime/estimator.py @@ -15,15 +15,13 @@ from hermes_cli.local_runtime.gguf import GGUFHeader _Q8_BYTES_PER_ELEM = 34 / 32 _F16_BYTES_PER_ELEM = 2.0 -# Architectures with a known SWA layer pattern: arch -> fraction of layers -# that are sliding-window. Unknown SWA archs conservatively treat every -# layer as full attention (overestimate; safe direction). +# Architectures with a known SWA layer pattern: arch -> fraction of layers that are +# sliding-window. Unknown SWA archs treat every layer as full attention (overestimate; safe). _SWA_LAYER_FRACTION = {"gemma3": 5 / 6, "gemma2": 1 / 2} -# Per-recurrent-layer state allowance (bytes/seq). Deliberately generous — -# Measured: an entire hybrid slot state is ~99 MB including 8K tokens of -# full-attn KV, so tens of MiB total is the right order; unknown SSM shapes -# must never underestimate. +# Per-recurrent-layer state allowance (bytes/seq). Deliberately generous: an entire measured +# hybrid slot state is ~99 MB including 8K tokens of full-attn KV, so tens of MiB total is the +# right order; unknown SSM shapes must never underestimate. _RECURRENT_STATE_PER_LAYER = 4 << 20 @@ -35,26 +33,22 @@ class LayerKind(Enum): @dataclass class ModelProfile: - """Everything the policy needs, decoupled from GGUF parsing so the decision-table tests can - construct profiles directly (design's verification plan). - """ + """Everything the policy needs, decoupled from GGUF parsing so decision-table tests can + construct profiles directly.""" name: str weights_bytes: int embd_table_bytes: int n_ctx_train: int - layers: list[tuple[LayerKind, int]] # (kind, kv_bytes_per_token_f16); - # SWA/recurrent reuse the same - # per-token figure, capped/ignored + layers: list[tuple[LayerKind, int]] # (kind, kv_bytes_per_token_f16); SWA capped, recurrent ignored swa_window: int = 0 moe: bool = False architecture: str = "" n_vocab: int = 0 # prices logits buffers (ubatch x vocab) - # Context-cost multiplier. MTP spec decode keeps a small draft - # context beside the main one. Calibrated against four measured - # server-RSS points on Qwen3.8 Q4 (128K/221K/256K, both postures): - # the draft adds ~17% to per-token KV; 1.2 rounds up so the error - # stays on the safe side (+250 MiB at 256K, never negative). + # Context-cost multiplier. MTP spec decode keeps a small draft context beside the main one; + # calibrated against four measured server-RSS points on Qwen3.8 Q4 (128K/221K/256K, both + # postures): the draft adds ~17% to per-token KV; 1.2 rounds up so the error stays on the safe + # side (+250 MiB at 256K, never negative). kv_scale: float = 1.0 @property @@ -69,12 +63,9 @@ class ModelProfile: @dataclass class HardwareBudget: - """Memory the physics check may budget against. - - Budget-source rule: discrete cards may trust the device query (measured honest); unified-memory - devices must budget from OS free physical memory minus headroom — their device queries have been - observed off by 3x. Callers construct this accordingly; the estimator just consumes it. - """ + """Memory the physics check may budget against. Discrete cards may trust the device query; + unified-memory devices must budget from OS free memory minus headroom (device queries observed + off by 3x). Callers construct this accordingly; the estimator just consumes it.""" usable_vram_bytes: int # live free (discrete) / derived (UMA) total_device_bytes: int @@ -97,37 +88,27 @@ def profile_from_gguf(header: GGUFHeader) -> ModelProfile: layers.append((LayerKind.RECURRENT, 0)) continue per_token = round(heads * (dk + dv) * _F16_BYTES_PER_ELEM) - # Distribute the SWA share across the first n_swa attention layers; - # only the full/SWA SPLIT matters to the totals, not which indexes. + # Distribute the SWA share across the first n_swa attention layers; only the full/SWA + # SPLIT matters to the totals, not which indexes. kind = LayerKind.SWA if n_attn_seen < n_swa else LayerKind.FULL layers.append((kind, per_token)) n_attn_seen += 1 return ModelProfile( - name=header.path, - weights_bytes=header.tensor_bytes, - embd_table_bytes=header.embd_table_bytes, - n_ctx_train=header.n_ctx_train, - layers=layers, - swa_window=header.sliding_window, - moe=header.expert_count > 0, - architecture=header.architecture, - n_vocab=header.n_vocab, - ) + name=header.path, weights_bytes=header.tensor_bytes, embd_table_bytes=header.embd_table_bytes, + n_ctx_train=header.n_ctx_train, layers=layers, swa_window=header.sliding_window, + moe=header.expert_count > 0, architecture=header.architecture, n_vocab=header.n_vocab) def kv_dtype_factor(flash_attention: bool) -> float: - """q8_0 with FA (every backend we ship); f16 on exotic non-FA fallbacks - — the 64K guarantee stands either way, the physics check just prices - the doubled KV (design: KV dtype is behavior, not config).""" + """q8_0 with FA (every backend we ship); f16 on exotic non-FA fallbacks — the 64K guarantee + stands either way, the physics check just prices the doubled KV.""" return (_Q8_BYTES_PER_ELEM / _F16_BYTES_PER_ELEM) if flash_attention else 1.0 -def ctx_bytes(profile: ModelProfile, window: int, *, - flash_attention: bool = True) -> int: - """Context memory for one window: full layers linear in T, SWA layers capped at the sliding window, - recurrent layers constant. Scaled by profile.kv_scale (MTP draft context). - """ +def ctx_bytes(profile: ModelProfile, window: int, *, flash_attention: bool = True) -> int: + """Context memory for one window: full layers linear in T, SWA layers capped at the sliding + window, recurrent layers constant. Scaled by profile.kv_scale (MTP draft context).""" factor = kv_dtype_factor(flash_attention) total = 0.0 for kind, per_token_f16 in profile.layers: @@ -142,8 +123,8 @@ def ctx_bytes(profile: ModelProfile, window: int, *, @dataclass class PhysicsRefusal: - """The only true refusal: weights + floor-KV + state exceed VRAM + RAM. - The remedy is a smaller quant, never a smaller window.""" + """The only true refusal: weights + floor-KV + state exceed VRAM + RAM. The remedy is a + smaller quant, never a smaller window.""" needed_bytes: int available_bytes: int @@ -156,11 +137,11 @@ def physics_check(profile: ModelProfile, budget: HardwareBudget, + ctx_bytes(profile, min(floor, profile.n_ctx_train or floor), flash_attention=flash_attention)) available = budget.usable_vram_bytes + budget.ram_available_bytes - if needed > available: - gib = 1 << 30 - return PhysicsRefusal( - needed_bytes=needed, available_bytes=available, - message=(f"{profile.name}: needs ~{needed / gib:.1f} GiB at the " - f"{floor // 1024}K floor but only ~{available / gib:.1f} GiB " - "of VRAM+RAM exist — try a smaller quant (UD-Q3/Q2)")) - return None + if needed <= available: + return None + gib = 1 << 30 + return PhysicsRefusal( + needed_bytes=needed, available_bytes=available, + message=(f"{profile.name}: needs ~{needed / gib:.1f} GiB at the " + f"{floor // 1024}K floor but only ~{available / gib:.1f} GiB " + "of VRAM+RAM exist — try a smaller quant (UD-Q3/Q2)")) diff --git a/hermes_cli/local_runtime/gguf.py b/hermes_cli/local_runtime/gguf.py index e73e67dd2d..4c56d44bb9 100644 --- a/hermes_cli/local_runtime/gguf.py +++ b/hermes_cli/local_runtime/gguf.py @@ -6,14 +6,25 @@ run at picker time on multi-GB files. from __future__ import annotations +import re import struct from dataclasses import dataclass, field from pathlib import Path _GGUF_MAGIC = b"GGUF" -# ggml tensor type sizes: type_id -> (block_bytes, block_elems). -# IQ-family sizes verified against ggml-common.h. +# Split GGUF naming: "-00001-of-00003.gguf"; the part suffix is not part of the model id. +SPLIT_PART_RE = re.compile(r"-(\d{5})-of-(\d{5})\.gguf$") +_PART_SUFFIX_RE = re.compile(r"-\d{5}-of-\d{5}$") + + +def model_id_from_stem(stem: str) -> str: + """Model id from a GGUF file stem (split-part suffix stripped).""" + return _PART_SUFFIX_RE.sub("", stem) + + +# ggml tensor type sizes: type_id -> (block_bytes, block_elems). IQ-family verified against +# ggml-common.h. _GGML_TYPE_SIZES = { 0: (4, 1), 1: (2, 1), 2: (18, 32), 3: (20, 32), 6: (22, 32), 7: (24, 32), 8: (34, 32), 9: (36, 32), 10: (84, 256), 11: (110, 256), 12: (144, 256), @@ -23,17 +34,20 @@ _GGML_TYPE_SIZES = { 26: (4, 1), 27: (8, 1), 28: (8, 1), 29: (56, 256), 30: (2, 1), } -# GGUF metadata value types. -_V_UINT8, _V_INT8, _V_UINT16, _V_INT16 = 0, 1, 2, 3 -_V_UINT32, _V_INT32, _V_FLOAT32, _V_BOOL = 4, 5, 6, 7 -_V_STRING, _V_ARRAY, _V_UINT64, _V_INT64, _V_FLOAT64 = 8, 9, 10, 11, 12 - +# GGUF metadata value types -> struct format; STRING (8) and ARRAY (9) are variable-length. +_V_STRING, _V_ARRAY = 8, 9 _SCALAR_FMT = { - _V_UINT8: " preset INI key. +_SAMPLING_INI_KEY = {"temp": "temp", "temperature": "temp", "top_p": "top-p", + "top_k": "top-k", "min_p": "min-p", + "repeat_penalty": "repeat-penalty", + "presence_penalty": "presence-penalty"} + @dataclass class GGUFHeader: @@ -42,8 +56,7 @@ class GGUFHeader: metadata: dict = field(default_factory=dict) n_tensors: int = 0 tensor_bytes: int = 0 # exact sum over the tensor table - embd_table_bytes: int = 0 # token_embd.weight (duplicated host-side - # when fully offloaded) + embd_table_bytes: int = 0 # token_embd.weight (duplicated host-side when fully offloaded) # ── typed accessors ────────────────────────────────────── @@ -54,52 +67,48 @@ class GGUFHeader: def _arch_key(self, suffix: str): return self.metadata.get(f"{self.architecture}.{suffix}") - @property - def n_layer(self) -> int: - return int(self._arch_key("block_count") or 0) + def _arch_int(suffix: str, doc: str = ""): # noqa: N805 — property factory, deleted below + return property(lambda self: int(self._arch_key(suffix) or 0), doc=doc) + + n_layer = _arch_int("block_count") + n_ctx_train = _arch_int("context_length") + n_embd = _arch_int("embedding_length") + sliding_window = _arch_int("attention.sliding_window") + expert_count = _arch_int("expert_count") + full_attention_interval = _arch_int( + "full_attention_interval", + "GDN-hybrid discriminator (qwen35 family): every Nth layer is full attention, the rest " + "are linear/recurrent. 0 = not present.") + del _arch_int @property def n_vocab(self) -> int: - """Vocabulary size: prices the GPU logits buffers (they scale ubatch x vocab). vocab_size - metadata when present, else the tokenizer list length. - """ + """Vocabulary size (prices the GPU logits buffers): vocab_size metadata when present, else + the tokenizer list length.""" v = self._arch_key("vocab_size") if v: return int(v) toks = self.metadata.get("tokenizer.ggml.tokens") return len(toks) if isinstance(toks, list) else 0 - @property - def n_ctx_train(self) -> int: - return int(self._arch_key("context_length") or 0) - @property def sampling_defaults(self) -> dict: - """Upstream's recommended sampling, when the file carries it. + """Upstream's recommended sampling as preset INI keys, when the file carries it. Publishers bake ``general.sampling.*`` keys into the GGUF (llama-server reads them as that - model's defaults), so the file is the source of truth -- it ships with the download and - updates with every re-upload, no catalog needed. Returned as preset INI keys; empty if - absent. + model's defaults), so the file is the source of truth — it ships with the download and + updates with every re-upload, no catalog needed. Empty if absent. """ - ini_key = {"temp": "temp", "temperature": "temp", "top_p": "top-p", - "top_k": "top-k", "min_p": "min-p", - "repeat_penalty": "repeat-penalty", - "presence_penalty": "presence-penalty"} out = {} for key, value in self.metadata.items(): if not key.startswith("general.sampling."): continue - name = ini_key.get(key.rsplit(".", 1)[-1]) + name = _SAMPLING_INI_KEY.get(key.rsplit(".", 1)[-1]) if name is not None and isinstance(value, (int, float)): num = round(float(value), 4) out[name] = str(int(num)) if num == int(num) else str(num) return out - @property - def n_embd(self) -> int: - return int(self._arch_key("embedding_length") or 0) - @property def n_head(self) -> int: v = self._arch_key("attention.head_count") @@ -107,18 +116,12 @@ class GGUFHeader: return int(max(v)) return int(v or 0) - @property - def full_attention_interval(self) -> int: - """GDN-hybrid discriminator (qwen35 family): every Nth layer is full - attention, the rest are linear/recurrent. 0 = not present.""" - return int(self._arch_key("full_attention_interval") or 0) - def head_counts_kv(self) -> list[int]: """Per-layer KV head counts; 0 marks a recurrent/linear layer (n_head_kv == 0). Three GGUF shapes: a per-layer array (nemotron_h_moe) is used as-is; a scalar plus ``full_attention_interval`` (qwen35) applies to every N-th layer (1-indexed) and is zero - elsewhere -- pricing all layers as attention was a 4x overestimate; a plain scalar (dense) + elsewhere — pricing all layers as attention was a 4x overestimate; a plain scalar (dense) broadcasts to every layer. """ v = self._arch_key("attention.head_count_kv") @@ -145,52 +148,43 @@ class GGUFHeader: return int(v) return self.head_dim_k - @property - def sliding_window(self) -> int: - return int(self._arch_key("attention.sliding_window") or 0) - - @property - def expert_count(self) -> int: - return int(self._arch_key("expert_count") or 0) - def read_gguf_header(path: str | Path) -> GGUFHeader: path = Path(path) + def read(f, fmt: str): + return struct.unpack(fmt, f.read(struct.calcsize(fmt))) + def read_str(f) -> str: - (n,) = struct.unpack(" dict: """model_id -> granted window (int). Empty on any read problem.""" - try: + with suppress(Exception): with open(window_overrides_path(), encoding="utf-8") as fh: data = json.load(fh) return {str(k): int(v) for k, v in data.items()} - except Exception: # noqa: BLE001 - return {} + return {} + + +def _write_overrides(overrides: dict) -> None: + path = window_overrides_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(overrides, indent=1), encoding="utf-8") def save_window_override(model_id: str, window: int) -> None: overrides = load_window_overrides() overrides[model_id] = int(window) - path = window_overrides_path() - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(overrides, indent=1), encoding="utf-8") + _write_overrides(overrides) def clear_window_override(model_id: str) -> None: @@ -41,22 +45,18 @@ def clear_window_override(model_id: str) -> None: overrides = load_window_overrides() if model_id in overrides: del overrides[model_id] - window_overrides_path().write_text( - json.dumps(overrides, indent=1), encoding="utf-8") + _write_overrides(overrides) def is_managed_endpoint(base_url: str) -> bool: """True when base_url is the server this process's state file points at.""" - try: + with suppress(Exception): from hermes_cli.local_runtime.endpoint import _state_endpoint state = _state_endpoint() - if state is None: - return False - return (base_url or "").rstrip("/") == str( - state.get("base_url", "")).rstrip("/") - except Exception: # noqa: BLE001 - return False + return state is not None and ( + (base_url or "").rstrip("/") == str(state.get("base_url", "")).rstrip("/")) + return False def maybe_grow_window(model_id: str, *, base_url: str, session_tokens: int, @@ -70,10 +70,7 @@ def maybe_grow_window(model_id: str, *, base_url: str, session_tokens: int, window — nothing rewinds. """ from hermes_cli.local_runtime.bootstrap import ( - get_supervisor, - refresh_local_runtime, - staged_models, - ) + get_supervisor, refresh_local_runtime, staged_models) from hermes_cli.local_runtime.context_policy import growth_decision from hermes_cli.local_runtime.estimator import profile_from_gguf from hermes_cli.local_runtime.gguf import read_gguf_header @@ -83,8 +80,7 @@ def maybe_grow_window(model_id: str, *, base_url: str, session_tokens: int, if sup is None or not is_managed_endpoint(base_url): return None - gguf = next((p for p in staged_models() - if p.stem.startswith(model_id) or model_id in p.stem), None) + gguf = next((p for p in staged_models() if p.stem.startswith(model_id) or model_id in p.stem), None) if gguf is None: return None @@ -100,18 +96,16 @@ def maybe_grow_window(model_id: str, *, base_url: str, session_tokens: int, server_idle = False decision = growth_decision( - # Capacity budget, not live-free: growth executes via a server - # bounce, so the grown instance loads onto a freed card. Live-free - # here is distorted by the very model being grown — it reads its - # own residency as unavailable and vetoes rungs that fit. + # Capacity budget, not live-free: growth executes via a server bounce, so the grown + # instance loads onto a freed card. Live-free is distorted by the very model being grown + # — it reads its own residency as unavailable and vetoes rungs that fit. profile, probe_budget(planning=True), current_window=current_window, session_tokens=session_tokens, measured_decode_tok_s=measured_decode_tok_s, server_idle=server_idle, - # The caller IS the occupancy signal: this runs from the agent's - # compression gate, which fired on its own threshold. Two - # separately-derived edges must not deadlock into + # The caller IS the occupancy signal: this runs from the agent's compression gate, which + # fired on its own threshold. Two separately-derived edges must not deadlock into # compress-before-grow. occupancy_confirmed=True, ) @@ -122,8 +116,8 @@ def maybe_grow_window(model_id: str, *, base_url: str, session_tokens: int, logger.info("context growth %s: %s", model_id, decision.reason) save_window_override(model_id, decision.next_window) if not refresh_local_runtime(): - # The override still lands at the next boot; report no growth NOW - # so the caller compresses instead of overflowing a stale window. + # The override still lands at the next boot; report no growth NOW so the caller + # compresses instead of overflowing a stale window. logger.warning("growth %s: server refresh failed; compression proceeds", model_id) return None return decision.next_window diff --git a/hermes_cli/local_runtime/hardware.py b/hermes_cli/local_runtime/hardware.py index e6bcd25e2e..9dc51bc9af 100644 --- a/hermes_cli/local_runtime/hardware.py +++ b/hermes_cli/local_runtime/hardware.py @@ -2,14 +2,13 @@ Budget-source rule: discrete cards may trust the device query (measured honest within rounding); unified-memory devices must budget from OS free physical memory minus headroom — their device -queries have been observed off by 3x in both directions. - -Every probe here must work under a stripped PATH — gateway and service sessions don't inherit the -interactive environment. +queries have been observed off by 3x in both directions. Every probe here must work under a +stripped PATH — gateway and service sessions don't inherit the interactive environment. """ from __future__ import annotations +from contextlib import suppress import logging import os import re @@ -24,44 +23,35 @@ from hermes_cli.local_runtime.estimator import HardwareBudget logger = logging.getLogger(__name__) _GIB = 1 << 30 -# Reserve carved off the card before any grant: the desktop's own -# co-residents (compositor, browser, Electron) measure ~2-2.5 GiB on a -# working machine, and a window granted into that space demotes silently -# under WDDM. 7% covers big cards; the 2 GiB floor is what the margin's -# old 512 MiB floor failed to cover in practice (a 221K grant measured -# 31.9/32.6 GiB with the desktop running — 'fits' by the math, demoted -# in reality). Small cards give up window to this; spill mode is their -# path to big models regardless. +# Reserve carved off the card before any grant: the desktop's own co-residents (compositor, +# browser, Electron) measure ~2-2.5 GiB, and a window granted into that space demotes silently +# under WDDM. 7% covers big cards; the 2 GiB floor is what a 512 MiB floor failed to cover (a +# 221K grant measured 31.9/32.6 GiB with the desktop running — 'fits' by the math, demoted in +# reality). Small cards give up window to this; spill mode is their path to big models. _MARGIN_FLOOR = 2 << 30 _MARGIN_FRACTION = 0.09 -# UMA headroom: on unified-memory machines (Apple Silicon, unified-memory -# NVIDIA) the model shares physical memory with the OS and every app, so -# budget from RAM minus this fraction. +# UMA headroom: on unified-memory machines the model shares physical memory with the OS and every +# app, so budget from RAM minus this fraction. _UMA_HEADROOM_FRACTION = 0.20 -# Engine-fallback gates for the unified-pool quirk — BOTH must hold, and -# no discrete card can meet either: (1) the allocator's pool exceeds the -# smi report by well past rounding/ECC slack (discrete cards agree within -# ~2%; carve-out disagreement runs to whole multiples), and (2) the pool is -# system-RAM-sized — a workstation card in a RAM-matched box fails (1) -# because its smi and allocator AGREE, and a big discrete card in a -# bigger box fails (2). The driver's INTEGRATED attribute, when -# readable, bypasses both gates in whichever direction it points. +# Engine-fallback gates for the unified-pool quirk — BOTH must hold, and no discrete card can +# meet either: (1) the allocator's pool exceeds the smi report by well past rounding/ECC slack +# (discrete cards agree within ~2%; carve-out disagreement runs to whole multiples), and (2) the +# pool is system-RAM-sized. The driver's INTEGRATED attribute, when readable, bypasses both gates +# in whichever direction it points. _POOL_DISAGREEMENT_FACTOR = 1.5 _POOL_RAM_FRACTION = 0.75 # cuDeviceGetAttribute enum: device is integrated with host memory. _CU_DEVICE_ATTRIBUTE_INTEGRATED = 18 -# One probe per process once a device answers (silicon doesn't change); -# a miss retries after this long so a runtime installed mid-session gets -# picked up by the engine fallback. +# One probe per process once a device answers (silicon doesn't change); a miss retries after this +# long so a runtime installed mid-session gets picked up by the engine fallback. _POOL_NEGATIVE_TTL_S = 60.0 _pool_probe_cache: tuple[float, "tuple[int, bool | None] | None"] | None = None # ' CUDA0: NVIDIA Example Device (1234-core Example GPU) (46464 MiB, 46284 MiB free)' -# — greedy .* pins the LAST parenthesized group, so device names carrying -# their own parentheses parse correctly. +# — greedy .* pins the LAST parenthesized group, so device names with parentheses parse. _DEVICE_LINE_RE = re.compile(r"CUDA\d+:.*\((\d+)\s*MiB,\s*\d+\s*MiB free\)\s*$") @@ -79,9 +69,8 @@ def _ram_bytes() -> tuple[int, int]: class MEMORYSTATUSEX(ctypes.Structure): _fields_ = ([("dwLength", ctypes.c_ulong), ("dwMemoryLoad", ctypes.c_ulong)] + [(name, ctypes.c_ulonglong) for name in ( - "ullTotalPhys", "ullAvailPhys", "ullTotalPageFile", - "ullAvailPageFile", "ullTotalVirtual", "ullAvailVirtual", - "ullAvailExtendedVirtual")]) + "ullTotalPhys", "ullAvailPhys", "ullTotalPageFile", "ullAvailPageFile", + "ullTotalVirtual", "ullAvailVirtual", "ullAvailExtendedVirtual")]) stat = MEMORYSTATUSEX() stat.dwLength = ctypes.sizeof(MEMORYSTATUSEX) @@ -91,51 +80,44 @@ def _ram_bytes() -> tuple[int, int]: pass try: if sys.platform == "darwin": - # macOS getconf has no _PHYS_PAGES/_AVPHYS_PAGES (exit 64, "no such - # configuration parameter") — the POSIX branch below returns (0, 0) - # and every model reads unavailable. sysctl is the platform truth. + # macOS getconf has no _PHYS_PAGES/_AVPHYS_PAGES (exit 64) — the POSIX branch would + # return (0, 0) and every model would read unavailable. sysctl is the platform truth. total = int(_stdout("/usr/sbin/sysctl", "-n", "hw.memsize").strip() or 0) if total <= 0: return 0, 0 avail = total // 2 # conservative fallback - try: + with suppress(OSError, ValueError): out = _stdout("/usr/bin/vm_stat") page_m = re.search(r"page size of (\d+)", out) page = int(page_m.group(1)) if page_m else 16384 - # free + inactive + purgeable ≈ reclaimable-on-demand; the - # speculative pool is dropped by the OS under pressure too. + # free + inactive + purgeable ≈ reclaimable-on-demand; the speculative pool is + # dropped by the OS under pressure too. pages = sum(int(m.group(1)) for key in ( "Pages free", "Pages inactive", "Pages purgeable", "Pages speculative") if (m := re.search(rf"{key}:\s+(\d+)\.", out))) if pages > 0: avail = pages * page - except (OSError, ValueError): - pass return total, avail # POSIX page = int(_stdout("getconf", "PAGE_SIZE") or 4096) total = int(_stdout("getconf", "_PHYS_PAGES") or 0) * page avail = total // 2 # conservative when _AVPHYS is unavailable - try: + with suppress(OSError, ValueError): avail = int(_stdout("getconf", "_AVPHYS_PAGES") or 0) * page or avail - except (OSError, ValueError): - pass return total, avail except (OSError, ValueError): return 0, 0 -# nvidia-smi lives at a fixed path under the driver install; PATH presence -# varies by session type (services and gateways often run with a minimal -# environment) and by driver generation (legacy NVSMI dir was never on -# PATH). Resolution result is cached: the driver doesn't move mid-process. +# nvidia-smi lives at a fixed path under the driver install; PATH presence varies by session type +# (services and gateways often run minimal environments) and by driver generation (the legacy +# NVSMI dir was never on PATH). Cached: the driver doesn't move mid-process. _smi_path_cache: "tuple[str | None] | None" = None def _nvidia_smi_path() -> str | None: - """Absolute path to nvidia-smi, or None. PATH first (respects user overrides), then the driver's - known install locations on Windows; on Linux/WSL the PATH lookup is the whole ladder. - """ + """Absolute path to nvidia-smi, or None. PATH first (respects user overrides), then the + driver's known Windows install locations; on Linux/WSL the PATH lookup is the whole ladder.""" global _smi_path_cache if _smi_path_cache is not None: return _smi_path_cache[0] @@ -159,7 +141,7 @@ def _nvidia_vram() -> tuple[int, int] | None: exe = _nvidia_smi_path() if exe is None: return None - try: + with suppress(OSError, ValueError, subprocess.TimeoutExpired): out = subprocess.run( [exe, "--query-gpu=memory.total,memory.free", "--format=csv,noheader,nounits"], @@ -168,16 +150,14 @@ def _nvidia_vram() -> tuple[int, int] | None: return None total_mib, free_mib = (int(x) for x in out.stdout.strip().splitlines()[0].split(",")) return total_mib << 20, free_mib << 20 - except (OSError, ValueError, subprocess.TimeoutExpired): - return None + return None def _cuda_driver_pool() -> "tuple[int, bool | None] | None": - """(allocator_total_bytes, integrated_or_None) from the CUDA driver API, or None when unreachable. - ctypes against the driver's own DLL/SO — no toolkit, no subprocess, ~ms. INTEGRATED is the - vendor's own unified-memory declaration; total is the pool the allocator will actually hand out - (on carve-out devices, several times what nvidia-smi reports). - """ + """(allocator_total_bytes, integrated_or_None) from the CUDA driver API via ctypes against the + driver's own DLL/SO — no toolkit, no subprocess, ~ms. INTEGRATED is the vendor's own + unified-memory declaration; total is the pool the allocator will actually hand out (on + carve-out devices, several times what nvidia-smi reports).""" import ctypes for name in ("nvcuda.dll", "libcuda.so.1", "libcuda.so"): @@ -188,7 +168,7 @@ def _cuda_driver_pool() -> "tuple[int, bool | None] | None": continue else: return None - try: + with suppress(OSError, AttributeError): if cuda.cuInit(0) != 0: return None dev = ctypes.c_int() @@ -204,27 +184,20 @@ def _cuda_driver_pool() -> "tuple[int, bool | None] | None": ctypes.byref(attr), _CU_DEVICE_ATTRIBUTE_INTEGRATED, dev) == 0: integrated = bool(attr.value) return total.value, integrated - except (OSError, AttributeError): - return None + return None def _engine_device_pool() -> "tuple[int, bool | None] | None": """(engine_total_bytes, None) from the installed runtime's own --list-devices, or None. The - fallback truth source when the driver API is unreachable: asks the exact binary that will do the - allocating. Carries no integrated verdict — callers must gate it. - """ - try: - from hermes_cli.local_runtime.binaries import ( - installed_tags, - runtimes_root, - server_binary, - ) + fallback when the driver API is unreachable: asks the exact binary that will do the + allocating. Carries no integrated verdict — callers must gate it.""" + with suppress(Exception): # a probe miss must never block budgeting + from hermes_cli.local_runtime.binaries import installed_tags, runtimes_root, server_binary tags = installed_tags() if not tags: return None - tag_dir = runtimes_root() / tags[0] - backend_dirs = [d for d in tag_dir.iterdir() if d.is_dir()] + backend_dirs = [d for d in (runtimes_root() / tags[0]).iterdir() if d.is_dir()] if not backend_dirs: return None exe = server_binary(backend_dirs[0]) @@ -236,15 +209,12 @@ def _engine_device_pool() -> "tuple[int, bool | None] | None": m = _DEVICE_LINE_RE.search(line) if m: return int(m.group(1)) << 20, None - return None - except Exception: # noqa: BLE001 — a probe miss must never block budgeting - return None + return None def _device_pool_view() -> "tuple[int, bool | None] | None": - """Best available allocator-side view, cached: a hit is permanent for the process, a miss retries - after a short TTL (the engine binary can appear mid-session via a pane install). - """ + """Best available allocator-side view, cached: a hit is permanent for the process, a miss + retries after a short TTL (the engine binary can appear mid-session via a pane install).""" global _pool_probe_cache now = time.monotonic() if _pool_probe_cache is not None: @@ -257,20 +227,17 @@ def _device_pool_view() -> "tuple[int, bool | None] | None": def _unified_pool_bytes(smi_total: int, ram_total: int) -> int | None: - """The real pool size when this NVIDIA device is unified memory behind + """The real pool size when this NVIDIA device is unified memory behind a carve-out, else None. - The driver's INTEGRATED attribute decides in BOTH directions when readable (0 pins discrete - even if numbers look odd; a declared-integrated driver is believed at modest pool sizes). Only - the attribute-less engine fallback needs the two numeric gates, both of which must hold. + The driver's INTEGRATED attribute decides in BOTH directions when readable; only the + attribute-less engine fallback needs the two numeric gates, both of which must hold. """ view = _device_pool_view() if view is None: return None pool, integrated = view - if integrated is False: - return None - if integrated is True: - return pool + if integrated is not None: + return pool if integrated else None if (smi_total > 0 and pool >= int(smi_total * _POOL_DISAGREEMENT_FACTOR) and ram_total > 0 and pool >= int(ram_total * _POOL_RAM_FRACTION)): return pool @@ -287,23 +254,18 @@ def probe_budget(*, planning: bool = False) -> HardwareBudget: """Construct the budget per the source rules above. ``planning=False``: LIVE budget (free VRAM now) for launch-time fit and growth re-grants. - ``planning=True``: CAPACITY budget (total minus margin) for catalog pricing and quant selection; - pricing against live-free while a model was loaded made every row read as too large and - degraded quant picks. The managed server unloads/relaunches itself, so capacity is real. + ``planning=True``: CAPACITY budget (total minus margin) for catalog pricing and quant + selection — pricing against live-free while a model was loaded made every row read as too + large. The managed server unloads/relaunches itself, so capacity is real. """ ram_total, ram_avail = _ram_bytes() vram = _nvidia_vram() - # Unified-memory NVIDIA: the CUDA allocator pool is the real - # capacity. Classification comes from the driver API/engine — it - # must not require nvidia-smi (stripped-PATH sessions lose smi but - # nvcuda loads via the system loader regardless). Crossing the - # carve-out costs nothing (effective bandwidth is flat through the - # boundary; smi's used/total merely saturate at it) — the carve-out - # is an OS accounting knob, not a GPU limit. Deliberately NOT - # clamped to OS RAM: carved-out memory is invisible to - # GlobalMemoryStatusEx (the OS reports correspondingly less total - # RAM), so a RAM clamp would throw away exactly the carved capacity. + # Unified-memory NVIDIA: the CUDA allocator pool is the real capacity. Classification comes + # from the driver API/engine and must not require nvidia-smi (stripped-PATH sessions lose smi + # but nvcuda loads via the system loader). Crossing the carve-out costs nothing — it is an OS + # accounting knob, not a GPU limit. Deliberately NOT clamped to OS RAM: carved-out memory is + # invisible to GlobalMemoryStatusEx, so a RAM clamp would throw away exactly that capacity. unified = _unified_pool_bytes(vram[0] if vram else 0, ram_total) if unified is not None: logger.info( @@ -314,19 +276,16 @@ def probe_budget(*, planning: bool = False) -> HardwareBudget: if planning: base = unified else: - # Live: dedicated-free plus what the OS can still give. smi's - # free saturates at the carve-out so this under-counts a bit — - # the safe direction (the pool edge is a measured soft cliff: - # decode collapses ~3.5x when concurrent demand hits it). - # Without smi, OS-available alone is the honest floor. + # Live: dedicated-free plus what the OS can still give. smi's free saturates at the + # carve-out so this under-counts a bit — the safe direction (the pool edge is a + # measured soft cliff: decode collapses ~3.5x when concurrent demand hits it). live = (vram[1] + ram_avail) if vram else ram_avail base = min(unified, live) return _uma_budget(base, unified) if vram is None: - # No NVIDIA device visible: Metal/Vulkan/CPU paths budget from RAM - # as UMA (Apple Silicon) — conservative for discrete AMD until a - # vendor probe lands (E3 hardware). + # No NVIDIA device visible: Metal/Vulkan/CPU paths budget from RAM as UMA (Apple + # Silicon) — conservative for discrete AMD until a vendor probe lands. return _uma_budget(ram_total if planning else ram_avail, ram_total) total, free = vram diff --git a/hermes_cli/local_runtime/hf_browse.py b/hermes_cli/local_runtime/hf_browse.py index d69e1fbc85..2289871e2e 100644 --- a/hermes_cli/local_runtime/hf_browse.py +++ b/hermes_cli/local_runtime/hf_browse.py @@ -1,11 +1,9 @@ """Browse Hugging Face for GGUF models the user can run. The curated catalog is the front page; this module is the firehose behind it — day-0 models not yet -in the catalog, community quants, anything. Three rules keep it safe and honest: - -1. Acquisition only. Nothing here serves a model: a browsed download lands in the machine-scoped -models dir and from that moment the normal machinery owns it — staleness bounce, preset generation -from the real GGUF header, fit policy, placement pills. 2. +in the catalog, community quants, anything. Acquisition only: nothing here serves a model. A browsed +download lands in the machine-scoped models dir and from that moment the normal machinery owns it — +staleness bounce, preset generation from the real GGUF header, fit policy, placement pills. """ from __future__ import annotations @@ -22,14 +20,12 @@ logger = logging.getLogger(__name__) _HF = "https://huggingface.co" _TIMEOUT_S = 15 -# Rough-fit fill-ins for pre-download pricing: a mid-size model's 64K-floor -# KV plus runtime overhead. Deliberately round numbers — the verdict bands -# are coarse (fits GPU / needs RAM / too big), not window grants. +# Rough-fit fill-in for pre-download pricing: a mid-size model's 64K-floor KV plus runtime +# overhead. Deliberately round — the verdict bands are coarse, not window grants. _ROUGH_KV_AND_OVERHEAD = 4 << 30 -# Tiny TTL cache: the pane fires a search per keystroke pause and re-opens -# repos the user flips between. Process-local, size-capped, no invalidation -# subtleties — upstream truth changes slowly at this granularity. +# Tiny TTL cache: the pane fires a search per keystroke pause and re-opens repos the user flips +# between. Process-local, size-capped; upstream truth changes slowly at this granularity. _CACHE: dict[str, tuple[float, object]] = {} _CACHE_TTL_S = 300 _CACHE_MAX = 128 @@ -68,14 +64,13 @@ class HFFileGroup: fit: str = "unknown" # fits-gpu | needs-ram | too-big | unknown -_QUANT_RE = re.compile( - r"(?:IQ|Q)\d[_A-Z0-9]*|F16|BF16|F32", re.IGNORECASE) +_QUANT_RE = re.compile(r"(?:IQ|Q)\d[_A-Z0-9]*|F16|BF16|F32", re.IGNORECASE) _SPLIT_RE = re.compile(r"-(\d{5})-of-(\d{5})\.gguf$", re.IGNORECASE) def search_models(query: str, limit: int = 20) -> list[HFModelHit]: - """Full-text search over HF models that ship GGUF files, most - downloaded first (the closest public signal to 'trending').""" + """Full-text search over HF models that ship GGUF files, most downloaded first (the closest + public signal to 'trending').""" q = urllib.parse.quote(query.strip()) url = (f"{_HF}/api/models?search={q}&filter=gguf&sort=downloads" f"&direction=-1&limit={max(1, min(int(limit), 50))}") @@ -93,9 +88,8 @@ def _quant_label(filename: str) -> str: def repo_files(repo: str) -> list[HFFileGroup]: - """The servable GGUFs in a repo, grouped: split parts collapse into one - entry (first part is what llama.cpp loads), mmproj/draft companions are - excluded (they aren't standalone models). Largest quant first.""" + """The servable GGUFs in a repo, grouped: split parts collapse into one entry (first part is + what llama.cpp loads); mmproj/draft companions are excluded. Largest quant first.""" url = f"{_HF}/api/models/{urllib.parse.quote(repo)}/tree/main?recursive=true" files = _get_json(url) @@ -106,13 +100,12 @@ def repo_files(repo: str) -> list[HFFileGroup]: if not path.lower().endswith(".gguf"): continue name = path.rsplit("/", 1)[-1].lower() - if name.startswith("mmproj") or name.startswith("dspark") or "draft" in name: + if name.startswith(("mmproj", "dspark")) or "draft" in name: continue size = int(f.get("size") or 0) m = _SPLIT_RE.search(path) if m: - stem = path[: m.start()] - splits.setdefault(stem, []).append((int(m.group(1)), path, size)) + splits.setdefault(path[: m.start()], []).append((int(m.group(1)), path, size)) else: singles.append((path, size)) @@ -120,18 +113,15 @@ def repo_files(repo: str) -> list[HFFileGroup]: for path, size in singles] for stem, parts in splits.items(): parts.sort() - groups.append(HFFileGroup( - label=_quant_label(stem), - paths=tuple(p for _, p, _ in parts), - total_bytes=sum(s for _, _, s in parts))) + groups.append(HFFileGroup(label=_quant_label(stem), paths=tuple(p for _, p, _ in parts), + total_bytes=sum(s for _, _, s in parts))) groups.sort(key=lambda g: g.total_bytes, reverse=True) return groups def rough_fit(total_bytes: int, budget) -> str: - """Coarse pre-download verdict from file size alone. The GGUF header refines this after download; - bands match the catalog pills' language. File size ≈ in-memory weights for GGUF (mmap'd as-is). - """ + """Coarse pre-download verdict from file size alone (GGUF file size ≈ in-memory weights); + the GGUF header refines this after download. Bands match the catalog pills' language.""" need = total_bytes + _ROUGH_KV_AND_OVERHEAD if need <= budget.usable_vram_bytes: return "fits-gpu" @@ -141,5 +131,4 @@ def rough_fit(total_bytes: int, budget) -> str: def priced_repo_files(repo: str, budget) -> list[HFFileGroup]: - return [replace(g, fit=rough_fit(g.total_bytes, budget)) - for g in repo_files(repo)] + return [replace(g, fit=rough_fit(g.total_bytes, budget)) for g in repo_files(repo)] diff --git a/hermes_cli/local_runtime/load_progress.py b/hermes_cli/local_runtime/load_progress.py index da86a6ba94..211215abbc 100644 --- a/hermes_cli/local_runtime/load_progress.py +++ b/hermes_cli/local_runtime/load_progress.py @@ -1,16 +1,14 @@ """Live model-load progress from the managed llama-server router. -llama-server's child processes emit per-tensor load progress ({stages, current, value}, throttled -upstream to ~200ms) which the router relays ONLY over its /models/sse stream — GET /models carries -just the coarse status string. - -The watcher starts on first call, reconnects with backoff (the router bounces on model -download/eject), and never raises into callers — no router, no state file, or no SSE support (older -engines) all read as "nothing loading". +Children emit per-tensor progress ({stages, current, value}) which the router relays ONLY over its +/models/sse stream — GET /models carries just the coarse status. The watcher starts on first call, +reconnects with backoff (the router bounces on download/eject), and never raises into callers: no +router, no state file, or no SSE support (older engines) all read as "nothing loading". """ from __future__ import annotations +from contextlib import suppress import json import logging import threading @@ -22,82 +20,74 @@ logger = logging.getLogger(__name__) _TEXT_STAGE_SHARE = 0.85 # composite range share for the text model _RECONNECT_DELAY_S = 3.0 _STALE_ENTRY_TTL_S = 120.0 # a loading entry with no events this long is dead +_LOAD_EVENTS = ("status_change", "model_status") _lock = threading.Lock() _watcher: threading.Thread | None = None _snapshot: dict[str, dict] = {} +def _pct(x: float) -> int: + return max(0, min(100, round(x * 100))) + + def _composite_percent(stages: list[str], current: str, value: float) -> int: """Map (stage, in-stage value) onto one 0-100 range, text-heavy.""" if not stages or current not in stages or len(stages) == 1: - return max(0, min(100, round(value * 100))) + return _pct(value) extras = [s for s in stages if s != "text_model"] extra_share = (1.0 - _TEXT_STAGE_SHARE) / len(extras) if extras else 0.0 offset = 0.0 for stage in stages: share = _TEXT_STAGE_SHARE if stage == "text_model" else extra_share if stage == current: - return max(0, min(100, round((offset + share * value) * 100))) + return _pct(offset + share * value) offset += share - return max(0, min(100, round(value * 100))) + return _pct(value) def _endpoint() -> "tuple[str, str] | None": - """(base_root, api_key) of the managed router, or None. + """(base_root, api_key) of the managed router via the ownership-guarded reader, or None.""" + from hermes_cli.local_runtime.endpoint import managed_root - Resolved through the endpoint module's ownership-guarded reader, not a raw state-file read: on - the shared stable port a foreign install's server answers /health for anyone, and a raw read - would attach this watcher to someone else's SSE stream. The dead-pid check is the ownership - proof. - """ - try: - from hermes_cli.local_runtime.endpoint import _state_endpoint - - state = _state_endpoint() - if state is None: - return None - base = str(state.get("base_url", "")).rsplit("/v1", 1)[0] - return (base, str(state.get("api_key", ""))) if base else None - except Exception: # noqa: BLE001 - return None + return managed_root() def _apply_event(model: str, event: str, data: dict) -> None: with _lock: status = str(data.get("status", "")) - if event in ("status_change", "model_status") and status == "loading": + if event in _LOAD_EVENTS and status == "loading": progress = data.get("progress") or {} stages = [str(s) for s in (progress.get("stages") or [])] current = str(progress.get("current", "")) value = progress.get("value") - entry = _snapshot.setdefault(model, {"stage": "", "value": 0.0, - "percent": 0, "ts": 0.0}) + entry = _snapshot.setdefault(model, {"stage": "", "value": 0.0, "percent": 0, "ts": 0.0}) entry["ts"] = time.monotonic() if current and isinstance(value, (int, float)): entry["stage"] = current entry["value"] = float(value) entry["percent"] = _composite_percent(stages, current, float(value)) - elif event in ("status_change", "model_status", "model_remove"): + elif event in (*_LOAD_EVENTS, "model_remove") and status != "loading": # Any terminal status (loaded/unloaded/failed) ends the load. - if status != "loading": - _snapshot.pop(model, None) + _snapshot.pop(model, None) + + +def _clear_snapshot() -> None: + with _lock: + _snapshot.clear() def _watch() -> None: while True: endpoint = _endpoint() if endpoint is None: - with _lock: - _snapshot.clear() + _clear_snapshot() time.sleep(_RECONNECT_DELAY_S) continue base, key = endpoint try: - req = urllib.request.Request( - f"{base}/models/sse", - headers={"Authorization": f"Bearer {key}", - "Accept": "text/event-stream"}) + req = urllib.request.Request(f"{base}/models/sse", headers={ + "Authorization": f"Bearer {key}", "Accept": "text/event-stream"}) with urllib.request.urlopen(req, timeout=60) as r: buf = b"" while True: @@ -110,19 +100,15 @@ def _watch() -> None: text = line.decode("utf-8", "replace").strip() if not text.startswith("data:"): continue - try: + with suppress(json.JSONDecodeError, TypeError): msg = json.loads(text[5:].strip()) - _apply_event(str(msg.get("model", "")), - str(msg.get("event", "")), + _apply_event(str(msg.get("model", "")), str(msg.get("event", "")), msg.get("data") or {}) - except (json.JSONDecodeError, TypeError): - continue except Exception as exc: # noqa: BLE001 — watcher must never die loud logger.debug("load-progress SSE reconnecting: %s", exc) - # Stream ended (router bounce, timeout, error): loading entries from - # the dead connection are unverifiable — drop rather than freeze. - with _lock: - _snapshot.clear() + # Stream ended (router bounce, timeout, error): loading entries from the dead connection + # are unverifiable — drop rather than freeze. + _clear_snapshot() time.sleep(_RECONNECT_DELAY_S) @@ -130,53 +116,43 @@ def _ensure_watcher() -> None: global _watcher with _lock: if _watcher is None or not _watcher.is_alive(): - _watcher = threading.Thread(target=_watch, daemon=True, - name="llamacpp-load-progress") + _watcher = threading.Thread(target=_watch, daemon=True, name="llamacpp-load-progress") _watcher.start() def get_loading_progress() -> dict[str, dict]: - """{model_id: {"stage", "value", "percent"}} for models loading right - now. Empty when nothing is loading (or nothing is knowable).""" + """{model_id: {"stage", "value", "percent"}} for models loading right now. Empty when + nothing is loading (or nothing is knowable).""" _ensure_watcher() now = time.monotonic() with _lock: - return {m: {"stage": e["stage"], "value": e["value"], - "percent": e["percent"]} - for m, e in _snapshot.items() - if now - e["ts"] < _STALE_ENTRY_TTL_S} + return {m: {"stage": e["stage"], "value": e["value"], "percent": e["percent"]} + for m, e in _snapshot.items() if now - e["ts"] < _STALE_ENTRY_TTL_S} def get_prefill_progress(model: str) -> "dict | None": """{"processed": tokens} while the managed server is prompt-processing for ``model``, or None (idle, decoding, unreachable, or foreign server). - llama-server's /slots exposes ``n_prompt_tokens_processed`` but no total, so callers supply the - denominator. The busiest processing slot wins: a parallel small request freezes its counter - during decode while a live prefill keeps climbing. One HTTP call per poll; every failure reads - as "no prefill" — garnish, never load-bearing. + /slots exposes ``n_prompt_tokens_processed`` but no total, so callers supply the denominator. + The busiest processing slot wins: a parallel small request freezes its counter during decode + while a live prefill keeps climbing. One HTTP call per poll; every failure reads as "no + prefill" — garnish, never load-bearing. """ ep = _endpoint() if ep is None: return None - base, key = ep try: from urllib.parse import quote - req = urllib.request.Request( - f"{base}/slots?model={quote(model)}", - headers={"Authorization": f"Bearer {key}"}) - with urllib.request.urlopen(req, timeout=2) as r: - slots = json.loads(r.read()) + from hermes_cli.local_runtime.endpoint import managed_get_json + + slots = managed_get_json(*ep, f"/slots?model={quote(model)}", timeout_s=2) except Exception: # noqa: BLE001 return None best = 0 for slot in slots if isinstance(slots, list) else []: - if not slot.get("is_processing"): - continue - try: - processed = int(slot.get("n_prompt_tokens_processed") or 0) - except (TypeError, ValueError): - continue - best = max(best, processed) + with suppress(TypeError, ValueError): + if slot.get("is_processing"): + best = max(best, int(slot.get("n_prompt_tokens_processed") or 0)) return {"processed": best} if best > 0 else None diff --git a/hermes_cli/local_runtime/presets.py b/hermes_cli/local_runtime/presets.py index a4008c22a9..f4b9b49a70 100644 --- a/hermes_cli/local_runtime/presets.py +++ b/hermes_cli/local_runtime/presets.py @@ -5,37 +5,22 @@ launch decisions. from __future__ import annotations import logging -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from hermes_cli.local_runtime.context_policy import ( - RUNTIME_OVERHEAD_BYTES, - WindowDecision, - initial_window, - launch_args, - ub_logits_bytes, -) + RUNTIME_OVERHEAD_BYTES, WindowDecision, initial_window, launch_args, ub_logits_bytes) from hermes_cli.local_runtime.estimator import ( - HardwareBudget, - PhysicsRefusal, - profile_from_gguf, -) -from hermes_cli.local_runtime.gguf import read_gguf_header + HardwareBudget, ModelProfile, PhysicsRefusal, ctx_bytes, profile_from_gguf) +from hermes_cli.local_runtime.gguf import model_id_from_stem, read_gguf_header logger = logging.getLogger(__name__) -# args list -> INI keys. Flags the policy owns; everything else stays out -# of the preset (recipe sampling defaults merge in a later pass). +# args list -> INI keys. Flags the policy owns; everything else stays out of the preset. _FLAG_TO_KEY = { - "-c": "ctx-size", - "-b": "batch-size", - "-ub": "ubatch-size", - "-ctk": "cache-type-k", - "-ctv": "cache-type-v", - "-fa": "flash-attn", - "-ot": "override-tensor", - "--spec-type": "spec-type", - "--spec-draft-n-max": "spec-draft-n-max", + "-c": "ctx-size", "-b": "batch-size", "-ub": "ubatch-size", + "-ctk": "cache-type-k", "-ctv": "cache-type-v", "-fa": "flash-attn", + "-ot": "override-tensor", "--spec-type": "spec-type", "--spec-draft-n-max": "spec-draft-n-max", } @@ -52,8 +37,7 @@ def _args_to_keys(args: list[str]) -> dict[str, str]: keys: dict[str, str] = {} i = 0 while i < len(args): - flag = args[i] - key = _FLAG_TO_KEY.get(flag) + key = _FLAG_TO_KEY.get(args[i]) if key is None: i += 1 continue @@ -62,147 +46,140 @@ def _args_to_keys(args: list[str]) -> dict[str, str]: return keys -def generate_presets(models_dir: Path, budget: HardwareBudget, - preset_path: Path, - mtp_capable: set[str] | None = None) -> list[PresetEntry]: - """Walk the staged models, run the launch decision per model, and write one INI. Refused models get - no section (the router simply won't have policy for them; the picker surfaces the refusal + - smaller-quant suggestion from the returned entries). - """ +def _asset_path(asset) -> "Path | None": + """On-disk path of a catalog companion asset, or None when it isn't downloaded.""" from hermes_cli.local_runtime.bootstrap import assets_dir - from hermes_cli.local_runtime.catalog import find_entry_for_model + + if asset is None: + return None + path = assets_dir() / asset.local_name + return path if path.exists() else None + + +def _choose_mtp_posture(profile: ModelProfile, budget: HardwareBudget, + fixed_overhead: int) -> tuple[bool, int]: + """(mtp_prefill, logits_bytes) for an MTP model — window first, prefill second. + + Price the launch under both postures and keep whichever grants the larger window: the stacked + posture's bigger compute buffer buys ~3x short-prompt prefill but costs ~2 GiB that would + otherwise be window (measured at 256K the ub512 posture still prefills at 2.7K tok/s), so + never trade context away for prefill. Same window -> stacked. + """ + plain_logits = ub_logits_bytes(profile.n_vocab, mtp_capable=True) + stacked_logits = ub_logits_bytes(profile.n_vocab, mtp_capable=True, mtp_prefill=True) + stacked = initial_window(profile, budget, overhead_bytes=fixed_overhead + stacked_logits) + plain = initial_window(profile, budget, overhead_bytes=fixed_overhead + plain_logits) + if (not isinstance(stacked, PhysicsRefusal) and not stacked.spilled + and (isinstance(plain, PhysicsRefusal) or stacked.window >= plain.window)): + return True, stacked_logits + return False, plain_logits + + +def _restore_grown_window(model_id: str, profile: ModelProfile, budget: HardwareBudget, + decision: WindowDecision, overhead: int) -> WindowDecision: + """Session growth (growth.py): a persisted override lifts the launch window to where the ladder + last grew it — capped at native, and only when physics still clears the bigger window on THIS + boot's budget (a smaller-VRAM day re-fits honestly back down).""" + try: + from hermes_cli.local_runtime.growth import load_window_overrides + + override = load_window_overrides().get(model_id) + native = profile.n_ctx_train or decision.window + if override and override > decision.window: + target = min(int(override), native) + kv = ctx_bytes(profile, target) + need = profile.weights_bytes + kv + overhead + if need <= budget.usable_vram_bytes + budget.ram_available_bytes: + return WindowDecision( + window=target, spill_bytes=max(0, need - budget.usable_vram_bytes), + kv_on_gpu=kv <= budget.usable_vram_bytes, + reasons=[f"grown window restored ({target // 1024}K)"]) + except Exception as exc: # noqa: BLE001 — overrides are advisory + logger.debug("window override skipped for %s: %s", model_id, exc) + return decision + + +def _preset_for(gguf: Path, budget: HardwareBudget, + mtp_capable: set[str]) -> PresetEntry | None: + """The launch decision for one staged model, or None when its header is unreadable.""" + from hermes_cli.local_runtime.catalog import entry_for_model + + model_id = model_id_from_stem(gguf.stem) + try: + header = read_gguf_header(gguf) + profile = profile_from_gguf(header) + except (ValueError, OSError) as exc: + logger.warning("preset skip %s: %s", gguf.name, exc) + return None + entry = entry_for_model(model_id) + is_mtp = entry.mtp if entry is not None else model_id in mtp_capable + if is_mtp and profile.kv_scale == 1.0: + # Header-derived profiles don't know about MTP's draft context; apply the calibrated KV + # multiplier so the launch fit prices what the server will actually allocate. + profile = replace(profile, kv_scale=1.2) + mmproj_path = _asset_path(entry.mmproj) if entry is not None else None + # Overhead beyond weights+KV: runtime buffers, the vision projector when present, and the + # logits buffers of whichever microbatch/MTP posture launch_args will choose — flag and price + # decided together, from the same facts. + fixed_overhead = RUNTIME_OVERHEAD_BYTES + ( + entry.mmproj.size_bytes if entry is not None and mmproj_path is not None else 0) + if is_mtp: + mtp_prefill, logits_bytes = _choose_mtp_posture(profile, budget, fixed_overhead) + else: + mtp_prefill, logits_bytes = False, ub_logits_bytes(profile.n_vocab, mtp_capable=False) + overhead = fixed_overhead + logits_bytes + decision = initial_window(profile, budget, overhead_bytes=overhead) + if isinstance(decision, PhysicsRefusal): + return PresetEntry(model_id=model_id, window=0, spilled=False, refusal=decision.message) + decision = _restore_grown_window(model_id, profile, budget, decision, overhead) + + # The launch flags MUST match the pricing above (same entry/is_mtp/posture). + keys = _args_to_keys(launch_args( + profile, decision, mtp_capable=is_mtp, uma=budget.uma, mtp_prefill=mtp_prefill, + mtp_draft_depth=entry.mtp_draft_depth if entry is not None else 3)) + if entry is not None and is_mtp: + # Integrated-MTP targets sample on the backend, and so does the draft (pairing validated + # against the vendor's published llama.cpp recipes). + keys["backend-sampling"] = "on" + keys["spec-draft-backend-sampling"] = "on" + + # Sampling deference ladder, under the policy keys (policy wins on clash): the GGUF's own + # general.sampling.* metadata is the publisher's recommendation and covers models the catalog + # has never heard of; catalog sampling applies only where the file is silent; a model + # carrying neither runs llama.cpp defaults. + for k, v in header.sampling_defaults.items(): + keys.setdefault(k, v) + if entry is not None: + for k, v in (entry.sampling or {}).items(): + keys.setdefault(k, v) + if mmproj_path is not None: + keys["mmproj"] = str(mmproj_path) + draft_path = _asset_path(entry.draft) if decision.spilled else None + if draft_path is not None: + keys["model-draft"] = str(draft_path) + keys["spec-type"] = "draft-dspark" + # Unsloth's measured cliff: acceptance 83% at 2-3 drafts, collapses at 4. + keys["spec-draft-n-max"] = "3" + return PresetEntry(model_id=model_id, window=decision.window, + spilled=decision.spilled, keys=keys) + + +def generate_presets(models_dir: Path, budget: HardwareBudget, preset_path: Path, + mtp_capable: set[str] | None = None) -> list[PresetEntry]: + """Walk the staged models, run the launch decision per model, and write one INI. Refused + models get no section (the picker surfaces the refusal from the returned entries).""" + from hermes_cli.local_runtime.bootstrap import staged_in entries: list[PresetEntry] = [] sections: list[str] = [] - for gguf in _staged_in(models_dir): - model_id = _strip_part(gguf.stem) - try: - header = read_gguf_header(gguf) - profile = profile_from_gguf(header) - except (ValueError, OSError) as exc: - logger.warning("preset skip %s: %s", gguf.name, exc) + for gguf in staged_in(models_dir, require_complete=False): + entry = _preset_for(gguf, budget, mtp_capable or set()) + if entry is None: continue - # Overhead beyond weights+KV: runtime buffers, the vision projector - # when this model ships one, and the logits buffers of whichever - # microbatch/MTP posture launch_args will choose — flag and price - # decided together, from the same facts. - hit = find_entry_for_model(model_id) - entry = hit[0] if hit is not None else None - is_mtp = (entry.mtp if entry is not None - else model_id in (mtp_capable or set())) - if is_mtp and profile.kv_scale == 1.0: - # Header-derived profiles don't know about MTP's draft - # context; apply the calibrated KV multiplier here so the - # launch fit prices what the server will actually allocate. - import dataclasses - - profile = dataclasses.replace(profile, kv_scale=1.2) - mmproj_bytes = 0 - if entry is not None and entry.mmproj is not None: - mmproj_path = assets_dir() / entry.mmproj.local_name - if mmproj_path.exists(): - mmproj_bytes = entry.mmproj.size_bytes - # MTP posture ladder — window first, prefill second: price the - # launch under both postures and keep whichever grants the larger - # window (the stacked posture's bigger compute buffer buys ~3x - # short-prompt prefill but costs ~2 GiB that would otherwise be - # window; measured at 256K the ub512 posture still prefills at - # 2.7K tok/s, so window wins ties only one way: never trade - # context away for prefill). Same window -> stacked. - mtp_prefill = False - logits_bytes = ub_logits_bytes(profile.n_vocab, mtp_capable=is_mtp) - if is_mtp: - stacked_logits = ub_logits_bytes(profile.n_vocab, mtp_capable=True, - mtp_prefill=True) - stacked_probe = initial_window( - profile, budget, - overhead_bytes=(RUNTIME_OVERHEAD_BYTES + mmproj_bytes - + stacked_logits)) - plain_probe = initial_window( - profile, budget, - overhead_bytes=(RUNTIME_OVERHEAD_BYTES + mmproj_bytes - + logits_bytes)) - if (not isinstance(stacked_probe, PhysicsRefusal) - and not stacked_probe.spilled - and (isinstance(plain_probe, PhysicsRefusal) - or stacked_probe.window >= plain_probe.window)): - mtp_prefill = True - logits_bytes = stacked_logits - decision = initial_window( - profile, budget, - overhead_bytes=RUNTIME_OVERHEAD_BYTES + mmproj_bytes + logits_bytes) - if isinstance(decision, PhysicsRefusal): - entries.append(PresetEntry(model_id=model_id, window=0, - spilled=False, refusal=decision.message)) - continue - - # Session growth (growth.py): a persisted override lifts the launch - # window to where the ladder last grew it — capped at native, and - # only when physics still clears the bigger window on THIS boot's - # budget (a smaller-VRAM day re-fits honestly back down). - try: - from hermes_cli.local_runtime.estimator import ctx_bytes - from hermes_cli.local_runtime.growth import load_window_overrides - - override = load_window_overrides().get(model_id) - native = profile.n_ctx_train or decision.window - if override and override > decision.window: - target = min(int(override), native) - kv = ctx_bytes(profile, target) - need = (profile.weights_bytes + kv - + RUNTIME_OVERHEAD_BYTES + mmproj_bytes + logits_bytes) - if need <= budget.usable_vram_bytes + budget.ram_available_bytes: - spill = max(0, need - budget.usable_vram_bytes) - decision = WindowDecision( - window=target, spill_bytes=spill, - kv_on_gpu=kv <= budget.usable_vram_bytes, - reasons=[f"grown window restored ({target // 1024}K)"]) - except Exception as exc: # noqa: BLE001 — overrides are advisory - logger.debug("window override skipped for %s: %s", model_id, exc) - - # (entry and is_mtp resolved above, where the overhead was priced — - # the launch flags below MUST match that pricing.) - args = launch_args(profile, decision, mtp_capable=is_mtp, - mtp_draft_depth=(entry.mtp_draft_depth - if entry is not None else 3), - uma=budget.uma, mtp_prefill=mtp_prefill) - keys = _args_to_keys(args) - - if entry is not None and is_mtp: - # Integrated-MTP targets sample on the backend, and so does - # the draft (pairing validated against the vendor's published - # llama.cpp recipes for these models). - keys["backend-sampling"] = "on" - keys["spec-draft-backend-sampling"] = "on" - - # Sampling deference ladder, under the policy keys (policy wins - # on clash). The GGUF's own general.sampling.* metadata is the - # publisher's recommendation — it arrives with the file, updates - # with every re-upload, and covers models the catalog has never - # heard of. Catalog sampling applies only where the file is - # silent; a model carrying neither runs llama.cpp defaults. - for k, v in header.sampling_defaults.items(): - keys.setdefault(k, v) - if entry is not None: - for k, v in (entry.sampling or {}).items(): - keys.setdefault(k, v) - if entry.mmproj is not None: - mmproj_path = assets_dir() / entry.mmproj.local_name - if mmproj_path.exists(): - keys["mmproj"] = str(mmproj_path) - if entry.draft is not None and decision.spilled: - draft_path = assets_dir() / entry.draft.local_name - if draft_path.exists(): - keys["model-draft"] = str(draft_path) - keys["spec-type"] = "draft-dspark" - # Unsloth's measured cliff: acceptance 83% at 2-3 - # drafts, collapses at 4. - keys["spec-draft-n-max"] = "3" - - entries.append(PresetEntry(model_id=model_id, window=decision.window, - spilled=decision.spilled, keys=keys)) - body = "\n".join(f"{k} = {v}" for k, v in keys.items()) - sections.append(f"[{model_id}]\n{body}\n") + entries.append(entry) + if entry.keys is not None: + body = "\n".join(f"{k} = {v}" for k, v in entry.keys.items()) + sections.append(f"[{entry.model_id}]\n{body}\n") preset_path.parent.mkdir(parents=True, exist_ok=True) preset_path.write_text("\n".join(sections), encoding="utf-8") @@ -211,9 +188,8 @@ def generate_presets(models_dir: Path, budget: HardwareBudget, def read_preset_decisions(preset_path: Path | None = None) -> dict[str, PresetEntry]: - """The launch decisions the running server was actually given, read back from the preset INI (the - INI is the record — it's what spawned the children). Missing/unparseable file returns {}. - """ + """The launch decisions the running server was actually given, read back from the preset INI + (the INI is the record — it's what spawned the children). Missing/unparseable -> {}.""" import configparser if preset_path is None: @@ -225,31 +201,9 @@ def read_preset_decisions(preset_path: Path | None = None) -> dict[str, PresetEn parser = configparser.ConfigParser() parser.read(preset_path, encoding="utf-8") for section in parser.sections(): - window = parser.getint(section, "ctx-size", fallback=0) - spilled = parser.has_option(section, "override-tensor") - out[section] = PresetEntry(model_id=section, window=window, - spilled=spilled) + out[section] = PresetEntry( + model_id=section, window=parser.getint(section, "ctx-size", fallback=0), + spilled=parser.has_option(section, "override-tensor")) except Exception as exc: # noqa: BLE001 logger.debug("preset read-back failed: %s", exc) return out - - -def _strip_part(stem: str) -> str: - import re - - return re.sub(r"-\d{5}-of-\d{5}$", "", stem) - - -def _staged_in(models_dir: Path) -> "list[Path]": - """Servable models in an arbitrary directory (split first-parts only) — - the validation harness points at non-default dirs.""" - import re - - part = re.compile(r"-(\d{5})-of-\d{5}\.gguf$") - out = [] - for p in sorted(models_dir.glob("*.gguf")): - m = part.search(p.name) - if m and m.group(1) != "00001": - continue - out.append(p) - return out diff --git a/hermes_cli/local_runtime/supervisor.py b/hermes_cli/local_runtime/supervisor.py index 23d2fb258b..e7fb788235 100644 --- a/hermes_cli/local_runtime/supervisor.py +++ b/hermes_cli/local_runtime/supervisor.py @@ -1,16 +1,15 @@ """Supervision of one llama-server in router mode. The router process is ours (restart with backoff on crash); router children are its problem — child -failures surface via GET /models exit_code, never auto-retried here. - -Readiness rules (each learned the hard way on real hardware): - health-200 is NOT readiness; every -readiness claim requires a touch generation (temp-0, expected token, generous budget, -reasoning_content scanned). - Always dial 127.0.0.1 — resolving localhost adds ~2s per request on -Windows via IPv6 fallback. +failures surface via GET /models exit_code, never auto-retried here. Learned on real hardware: +health-200 is NOT readiness — every readiness claim requires a touch generation (temp-0, expected +token, generous budget, reasoning_content scanned); always dial 127.0.0.1 — resolving localhost adds +~2s per request on Windows via IPv6 fallback. """ from __future__ import annotations +from contextlib import suppress import json import logging import secrets @@ -29,32 +28,35 @@ logger = logging.getLogger(__name__) TOUCH_PROMPT = "Reply with exactly one word: the capital of France." TOUCH_EXPECT = "paris" _RESTART_BACKOFF_S = (1, 5, 15, 60) +_RESIDENT = ("loaded", "ready") + +# Chosen once and reused across restarts: sessions persist the resolved base_url, so an ephemeral +# port would strand every resumed session after each restart. Deliberately NOT 8080 so we never +# collide with a user's own llama-server/Ollama-adjacent stack. +_DEFAULT_PORT = 18434 def state_path() -> Path: - """Endpoint state for other Hermes processes (provider resolution reads - this to route llamacpp-alias requests at the managed server).""" + """Endpoint state for other Hermes processes (provider resolution routes llamacpp-alias + requests at the managed server from this).""" return runtimes_root() / "server.json" +def _quiet(fn) -> None: + """Best-effort call; a child that vanished mid-walk is not an error.""" + with suppress(Exception): + fn() + + def _free_port() -> int: with socket.socket() as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] -# Default port for the managed server, chosen once and reused across -# restarts. Sessions persist the resolved base_url; an ephemeral port -# would strand every resumed session on a dead endpoint after each -# restart. Deliberately NOT 8080 so we never collide with a user's own -# llama-server/Ollama-adjacent stack. -_DEFAULT_PORT = 18434 - - def _stable_port() -> int: - """The stable default port, falling back to an ephemeral one only when something else already - listens there (and it isn't a leftover managed server, which stop() would have cleaned up). - """ + """The stable default port, or an ephemeral one only when something else already listens + there (a leftover managed server would have been cleaned up by stop()).""" try: with socket.socket() as s: s.bind(("127.0.0.1", _DEFAULT_PORT)) @@ -69,17 +71,15 @@ def _stable_port() -> int: def _stable_api_key() -> str: """One key for the life of the install, persisted beside the runtimes. - Endpoint identity must survive restarts as a UNIT — sessions persist the resolved base_url + - api_key, so a per-boot key strands every resumed session on HTTP 401 exactly the way a per-boot - port would strand them on connection errors. + Endpoint identity must survive restarts as a UNIT — sessions persist base_url + api_key, so a + per-boot key strands every resumed session on HTTP 401 exactly as a per-boot port would on + connection errors. """ key_path = runtimes_root() / ".api_key" - try: + with suppress(OSError): existing = key_path.read_text(encoding="utf-8").strip() if len(existing) >= 16: return existing - except OSError: - pass key = secrets.token_urlsafe(24) try: key_path.parent.mkdir(parents=True, exist_ok=True) @@ -93,6 +93,12 @@ def _stable_api_key() -> str: class LlamaServerSupervisor: """Own one llama-server router process for the life of a Hermes session.""" + # A model that has gone quiet gets its VRAM back after this long. A constant, not a knob: + # long enough that an active conversation never trips it, short enough that a wandered-off + # session frees ~20 GiB within the hour. No exemptions: demand reloads anything the user + # comes back to. + IDLE_UNLOAD_S = 15 * 60 + def __init__(self, install_dir: Path, models_dir: Path, *, models_max: int = 4, port: int | None = None, extra_args: list[str] | None = None, @@ -128,11 +134,8 @@ class LlamaServerSupervisor: headers = {"Authorization": f"Bearer {self.api_key}"} if json_type: headers["Content-Type"] = "application/json" - req = urllib.request.Request( - self._url(route), - data=json.dumps(body).encode() if body is not None else None, - headers=headers, - ) + req = urllib.request.Request(self._url(route), headers=headers, + data=json.dumps(body).encode() if body is not None else None) return urllib.request.urlopen(req, timeout=timeout_s) def _request(self, route: str, body: dict | None = None, timeout_s: int = 30) -> dict: @@ -151,19 +154,15 @@ class LlamaServerSupervisor: "--api-key", self.api_key, "--models-dir", str(self.models_dir), "--models-max", str(self.models_max), - # The residency contract at the layer that sees every message: - # a chat request to a staged-but-unloaded model loads it (slow - # first token) instead of failing with 'model not found' — - # without this flag, chat after an eject is a bare 400/404. + # Residency contract: a chat request to a staged-but-unloaded model loads it (slow + # first token) instead of a bare 400/404 after an eject. "--models-autoload", "--metrics", # opt-in flag; supervisor telemetry needs it "--slots", # /slots endpoint is also opt-in; is_idle reads it "--no-webui", "--jinja", - # Direct I/O on model load: bypasses the page cache, so a - # multi-GB load doesn't evict half the OS cache — measured - # faster loads on NVMe, and our router bounces (download/ - # delete/activate) reload models often enough to care. + # Direct I/O on model load bypasses the page cache so a multi-GB load doesn't evict + # half the OS cache — measured faster on NVMe, and our router bounces reload often. "-dio", ] if self.preset_path and self.preset_path.exists(): @@ -171,12 +170,8 @@ class LlamaServerSupervisor: cmd += self.extra_args self.log_path.parent.mkdir(parents=True, exist_ok=True) if self._log_handle is not None: - # The crash-restart loop calls _spawn repeatedly; without - # closing the prior handle each restart leaks one fd. - try: - self._log_handle.close() - except Exception: # noqa: BLE001 — best-effort - pass + # The crash-restart loop calls _spawn repeatedly; each restart would leak one fd. + _quiet(self._log_handle.close) self._log_handle = open(self.log_path, "a", encoding="utf-8", errors="replace") self._log_handle.write(f"\n# spawn: {cmd}\n") self._log_handle.flush() @@ -184,10 +179,9 @@ class LlamaServerSupervisor: self.proc = subprocess.Popen(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 readiness probe racing the boot doesn't - # throw the app back to onboarding (observed on first restart test). + # 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 + # readiness probe racing the boot doesn't throw the app back to onboarding. self._write_state() def start(self, timeout_s: int = 120) -> None: @@ -195,32 +189,25 @@ class LlamaServerSupervisor: self._spawn() self._wait_health(timeout_s) self._write_state() - self._watchdog = threading.Thread(target=self._watch, daemon=True, - name="llamacpp-supervisor") + self._watchdog = threading.Thread(target=self._watch, daemon=True, name="llamacpp-supervisor") self._watchdog.start() def _write_state(self) -> None: 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") + 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") def _wait_health(self, timeout_s: int) -> None: deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: 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})") - try: + raise RuntimeError(f"llama-server exited rc={self.proc.returncode} during startup " + f"(log: {self.log_path})") + with suppress(urllib.error.URLError, OSError, TimeoutError): with urllib.request.urlopen(self._url("/health"), timeout=3) as r: if r.status == 200: return - except (urllib.error.URLError, OSError, TimeoutError): - pass time.sleep(1) raise TimeoutError(f"llama-server not healthy after {timeout_s}s (log: {self.log_path})") @@ -237,8 +224,7 @@ class LlamaServerSupervisor: if self._stopping: 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) + logger.warning("llama-server exited rc=%s; restart #%s in %ss", rc, self._restarts + 1, backoff) time.sleep(backoff) self._restarts += 1 try: @@ -263,42 +249,31 @@ class LlamaServerSupervisor: def _terminate_tree(proc: subprocess.Popen) -> None: """Terminate the router AND its model children. - The router spawns one child llama-server per loaded model, each holding gigabytes of VRAM. - Terminating only the router (on Windows, TerminateProcess with no cleanup) orphans them: the - port goes quiet but the weights stay resident. Enumerate children FIRST (the parent must be - alive to walk them), then terminate all together, escalating to kill for stragglers. + Each child holds gigabytes of VRAM; terminating only the router (TerminateProcess on + Windows does no cleanup) orphans them with the weights still resident. Enumerate children + FIRST (the parent must be alive to walk them), terminate all, escalate to kill. """ children: list = [] - try: + with suppress(Exception): # no psutil view; still stop the router import psutil children = psutil.Process(proc.pid).children(recursive=True) - except Exception: # noqa: BLE001 — no psutil view; still stop the router - children = [] proc.terminate() for child in children: - try: - child.terminate() - except Exception: # noqa: BLE001 - pass + _quiet(child.terminate) try: proc.wait(timeout=15) except subprocess.TimeoutExpired: proc.kill() for child in children: - try: - if child.is_running(): - child.kill() - except Exception: # noqa: BLE001 - pass + _quiet(lambda: child.is_running() and child.kill()) def _reap_orphaned_children(self) -> None: """Kill model children orphaned by a router crash, before respawn. - A crashed router can't clean up its children, and a dead parent can't be walked — so match - by identity instead: any process running OUR llama-server binary whose parent is gone is an - orphan of a previous router. Their VRAM must come back before the new router loads models - next to the ghosts. + A dead parent can't be walked, so match by identity: any process running OUR + 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. """ try: import psutil @@ -306,59 +281,42 @@ class LlamaServerSupervisor: exe = str(server_binary(self.install_dir)) except Exception: # noqa: BLE001 return + own_pid = self.proc.pid if self.proc is not None else None for p in psutil.process_iter(["exe", "ppid"]): - try: - if p.info.get("exe") != exe: - continue - if self.proc is not None and p.pid == self.proc.pid: - continue + with suppress(psutil.NoSuchProcess, psutil.AccessDenied): ppid = p.info.get("ppid") or 0 - if ppid and psutil.pid_exists(ppid): + if (p.info.get("exe") != exe or p.pid == own_pid + or (ppid and psutil.pid_exists(ppid))): continue logger.warning("reaping orphaned llama-server child pid=%s", p.pid) p.kill() - except (psutil.NoSuchProcess, psutil.AccessDenied): - continue # ── model management (router endpoints) ────────────────── def models(self) -> dict: """{model_id: status_value} from GET /models.""" - data = self._request("/models") return {m["id"]: m.get("status", {}).get("value", "unknown") - for m in data.get("data", [])} + for m in self._request("/models").get("data", [])} def load_model(self, model_id: str, timeout_s: int = 600) -> None: self._request("/models/load", {"model": model_id}, timeout_s=timeout_s) def unload_model(self, model_id: str) -> None: - """Free the child's VRAM now. Route existence verified empirically on b10290 (POST - /models/unload; bogus name -> 400 'model is not found'). Momentary action: never touches - primary_model — the declaration is durable, an eject is not (residency design). - """ + """Free the child's VRAM now (POST /models/unload; bogus name -> 400). Momentary: never + touches primary_model — the declaration is durable, an eject is not.""" self._request("/models/unload", {"model": model_id}, timeout_s=120) deadline = time.monotonic() + 15 while time.monotonic() < deadline: try: - if self.models().get(model_id) not in ("loaded", "ready", "unloading"): + if self.models().get(model_id) not in (*_RESIDENT, "unloading"): return except Exception: # noqa: BLE001 return time.sleep(0.3) - # ── idle residency (non-primary models) ────────────────── - - # A model that has gone quiet gets its VRAM back after this long. A - # constant, not a knob: long enough that an active conversation never - # trips it, short enough that a wandered-off session frees ~20 GiB - # within the hour. No exemptions (residency v2): demand reloads - # anything the user comes back to. - IDLE_UNLOAD_S = 15 * 60 - def sweep_idle(self, now: float | None = None) -> list[str]: - """Unload models idle past IDLE_UNLOAD_S. Returns the model ids - unloaded. Idle means no busy slots and no queued work, tracked - per model across calls; a model seen busy resets its clock.""" + """Unload models idle past IDLE_UNLOAD_S; returns their ids. Idle = no busy slots and no + queued work, tracked per model across calls; a model seen busy resets its clock.""" now = time.monotonic() if now is None else now unloaded: list[str] = [] try: @@ -366,31 +324,28 @@ class LlamaServerSupervisor: except Exception: # noqa: BLE001 return unloaded for model_id, status in statuses.items(): - if status not in ("loaded", "ready") or not self.is_idle(model_id): + if status not in _RESIDENT or not self.is_idle(model_id): self._idle_since.pop(model_id, None) continue first_idle = self._idle_since.setdefault(model_id, now) - if now - first_idle >= self.IDLE_UNLOAD_S: - try: - self.unload_model(model_id) - self._idle_since.pop(model_id, None) - unloaded.append(model_id) - logger.info("idle-unloaded %s (idle %ds)", model_id, - int(now - first_idle)) - except Exception as exc: # noqa: BLE001 - logger.warning("idle unload of %s failed: %s", model_id, exc) + if now - first_idle < self.IDLE_UNLOAD_S: + continue + try: + self.unload_model(model_id) + self._idle_since.pop(model_id, None) + unloaded.append(model_id) + logger.info("idle-unloaded %s (idle %ds)", model_id, int(now - first_idle)) + except Exception as exc: # noqa: BLE001 + logger.warning("idle unload of %s failed: %s", model_id, exc) return unloaded def touch_generate(self, model_id: str, timeout_s: int = 300) -> bool: """The readiness proof. Generous budget + reasoning_content scan — small token budgets - false-fail reasoning models, which spend their first tokens thinking. - """ + false-fail reasoning models, which spend their first tokens thinking.""" try: resp = self._request("/v1/chat/completions", { - "model": model_id, - "messages": [{"role": "user", "content": TOUCH_PROMPT}], - "max_tokens": 512, "temperature": 0, - }, timeout_s=timeout_s) + "model": model_id, "messages": [{"role": "user", "content": TOUCH_PROMPT}], + "max_tokens": 512, "temperature": 0}, timeout_s=timeout_s) msg = resp["choices"][0]["message"] blob = (msg.get("content") or "") + " " + (msg.get("reasoning_content") or "") return TOUCH_EXPECT in blob.lower() @@ -403,23 +358,19 @@ class LlamaServerSupervisor: status = self.models().get(model_id) if status is None: raise KeyError(f"model {model_id} not present in models dir") - if status not in ("loaded", "ready"): + if status not in _RESIDENT: self.load_model(model_id, timeout_s=timeout_s) return self.touch_generate(model_id) # ── telemetry ──────────────────────────────────────────── def is_idle(self, model_id: str | None = None) -> bool: - """No processing requests and no busy slots. Router quirk: /slots and /metrics are per-child - and require ?model= (bare calls 400), and no KV-usage metric exists. With ``model_id`` - checks that one child; without, every loaded child. - """ + """No processing requests and no busy slots. Router quirk: /slots and /metrics are + per-child and require ?model= (bare calls 400). With ``model_id`` checks that one child; + without, every loaded child.""" try: - if model_id is not None: - loaded = [model_id] - else: - loaded = [m for m, status in self.models().items() - if status in ("loaded", "ready")] + loaded = ([model_id] if model_id is not None + else [m for m, status in self.models().items() if status in _RESIDENT]) for mid in loaded: slots = self._request(f"/slots?model={mid}") if any(s.get("is_processing") for s in slots): @@ -427,9 +378,9 @@ class LlamaServerSupervisor: with self._open(f"/metrics?model={mid}", timeout_s=10, json_type=False) as r: text = r.read().decode() for line in text.splitlines(): - if line.startswith("llamacpp:requests_processing"): - if float(line.split()[-1]) != 0.0: - return False + if (line.startswith("llamacpp:requests_processing") + and float(line.split()[-1]) != 0.0): + return False return True except Exception: # noqa: BLE001 return False