diff --git a/hermes_cli/main_agent_cmds.py b/hermes_cli/main_agent_cmds.py index 5b8088f485..2e51c7f8c6 100644 --- a/hermes_cli/main_agent_cmds.py +++ b/hermes_cli/main_agent_cmds.py @@ -10,86 +10,80 @@ avoids an import cycle). import sys +def _cmd_memory_off(): + from hermes_cli.config import load_config, save_config + config = load_config() + if not isinstance(config.get("memory"), dict): + config["memory"] = {} + config["memory"]["provider"] = "" + save_config(config) + print("\n ✓ Memory provider: built-in only") + print(" Saved to config.yaml\n") + + +def _cmd_memory_reset(args): + from hermes_constants import get_hermes_home, display_hermes_home + mem_dir = get_hermes_home() / "memories" + target = getattr(args, "target", "all") + files_to_reset = [] + if target in {"all", "memory"}: + files_to_reset.append(("MEMORY.md", "agent notes")) + if target in {"all", "user"}: + files_to_reset.append(("USER.md", "user profile")) + + existing = [(f, desc) for f, desc in files_to_reset if (mem_dir / f).exists()] + if not existing: + print(f"\n Nothing to reset — no memory files found in {display_hermes_home()}/memories/\n") + return + + print("\n This will permanently erase the following memory files:") + for f, desc in existing: + size = (mem_dir / f).stat().st_size + print(f" ◆ {f} ({desc}) — {size:,} bytes") + + if not getattr(args, "yes", False): + try: + answer = input("\n Type 'yes' to confirm: ").strip().lower() + except (EOFError, KeyboardInterrupt): + print("\n Cancelled.\n") + return + if answer != "yes": + print(" Cancelled.\n") + return + + for f, desc in existing: + (mem_dir / f).unlink() + print(f" ✓ Deleted {f} ({desc})") + + print("\n Memory reset complete. New sessions will start with a blank slate.") + print(f" Files were in: {display_hermes_home()}/memories/\n") + + def cmd_memory(args): sub = getattr(args, "memory_command", None) if sub == "off": - from hermes_cli.config import load_config, save_config - - config = load_config() - if not isinstance(config.get("memory"), dict): - config["memory"] = {} - config["memory"]["provider"] = "" - save_config(config) - print("\n ✓ Memory provider: built-in only") - print(" Saved to config.yaml\n") + _cmd_memory_off() elif sub == "reset": - from hermes_constants import get_hermes_home, display_hermes_home - - mem_dir = get_hermes_home() / "memories" - target = getattr(args, "target", "all") - files_to_reset = [] - if target in {"all", "memory"}: - files_to_reset.append(("MEMORY.md", "agent notes")) - if target in {"all", "user"}: - files_to_reset.append(("USER.md", "user profile")) - - # Check what exists - existing = [ - (f, desc) for f, desc in files_to_reset if (mem_dir / f).exists() - ] - if not existing: - print( - f"\n Nothing to reset — no memory files found in {display_hermes_home()}/memories/\n" - ) - return - - print("\n This will permanently erase the following memory files:") - for f, desc in existing: - path = mem_dir / f - size = path.stat().st_size - print(f" ◆ {f} ({desc}) — {size:,} bytes") - - if not getattr(args, "yes", False): - try: - answer = input("\n Type 'yes' to confirm: ").strip().lower() - except (EOFError, KeyboardInterrupt): - print("\n Cancelled.\n") - return - if answer != "yes": - print(" Cancelled.\n") - return - - for f, desc in existing: - (mem_dir / f).unlink() - print(f" ✓ Deleted {f} ({desc})") - - print( - "\n Memory reset complete. New sessions will start with a blank slate." - ) - print(f" Files were in: {display_hermes_home()}/memories/\n") + _cmd_memory_reset(args) else: from hermes_cli.memory_setup import memory_command - memory_command(args) +# (args attribute, acp flag) — forwarded in this order. +_ACP_FLAGS = ( + ("acp_version", "--version"), + ("check", "--check"), + ("setup", "--setup"), + ("setup_browser", "--setup-browser"), + ("assume_yes", "--yes")) + + def cmd_acp(args): """Launch Hermes Agent as an ACP server.""" try: from acp_adapter.entry import main as acp_main - - acp_argv = [] - if getattr(args, "acp_version", False): - acp_argv.append("--version") - if getattr(args, "check", False): - acp_argv.append("--check") - if getattr(args, "setup", False): - acp_argv.append("--setup") - if getattr(args, "setup_browser", False): - acp_argv.append("--setup-browser") - if getattr(args, "assume_yes", False): - acp_argv.append("--yes") - acp_main(acp_argv) + acp_main([flag for attr, flag in _ACP_FLAGS if getattr(args, attr, False)]) except ImportError: print("ACP dependencies not installed.", file=sys.stderr) print("Install them with: pip install -e '.[acp]'", file=sys.stderr) @@ -101,16 +95,13 @@ def cmd_tools(args): action = getattr(args, "tools_action", None) if action in {"list", "disable", "enable"}: from hermes_cli.tools_config import tools_disable_enable_command - tools_disable_enable_command(args) elif action == "post-setup": from hermes_cli.tools_config import run_post_setup_command - sys.exit(run_post_setup_command(args)) else: _require_tty("tools") from hermes_cli.tools_config import tools_command - tools_command(args) @@ -119,7 +110,6 @@ def cmd_insights(args): try: from hermes_state import SessionDB from agent.insights import InsightsEngine - db = SessionDB() engine = InsightsEngine(db) report = engine.generate(days=args.days, source=args.source) @@ -134,24 +124,20 @@ def cmd_insights(args): pass +def _dict_or_empty(value) -> dict: + return value if isinstance(value, dict) else {} + + def cmd_monitoring(args): """Gateway monitoring status: health & diagnostics export posture.""" from hermes_cli.config import load_config - action = getattr(args, "monitoring_action", None) or "status" - config = load_config() - mon_raw = config.get("monitoring") - mon: dict = mon_raw if isinstance(mon_raw, dict) else {} + mon = _dict_or_empty(load_config().get("monitoring")) if action == "status": from agent.monitoring import otlp_exporter - - gh_raw = mon.get("gateway_health_export") - gh: dict = gh_raw if isinstance(gh_raw, dict) else {} - export_raw = mon.get("export") - export_cfg: dict = export_raw if isinstance(export_raw, dict) else {} - otlp_raw = export_cfg.get("otlp") - otlp: dict = otlp_raw if isinstance(otlp_raw, dict) else {} + gh = _dict_or_empty(mon.get("gateway_health_export")) + otlp = _dict_or_empty(_dict_or_empty(mon.get("export")).get("otlp")) print("Gateway monitoring") print(f" Health export: {'enabled' if gh.get('enabled') else 'disabled'} " @@ -180,38 +166,32 @@ def cmd_monitoring(args): def cmd_skills(args): - # Route 'config' action to skills_config module from hermes_cli.main import _require_tty - if getattr(args, "skills_action", None) == "config": + action = getattr(args, "skills_action", None) + if action == "config": _require_tty("skills config") from hermes_cli.skills_config import skills_command as skills_config_command - skills_config_command(args) - elif getattr(args, "skills_action", None) in ("trust", "untrust"): + elif action in ("trust", "untrust"): _cmd_skills_trust(args) else: from hermes_cli.skills_hub import skills_command - skills_command(args) def _cmd_skills_trust(args): - """``hermes skills trust [path]`` / ``hermes skills untrust [path]``. + """``hermes skills trust|untrust [path]`` — manage ``skills.trusted_project_dirs``. - Manages ``skills.trusted_project_dirs`` in config.yaml. With no path, - operates on the project root enclosing the current directory (nearest - ancestor with ``.git``). + With no path, operates on the project root enclosing the current directory + (nearest ancestor with ``.git``). """ from pathlib import Path - from agent.skill_utils import ( PROJECT_SKILLS_SUBDIRS, _candidate_project_skills_dirs, find_project_root, - iter_skill_index_files, - ) + iter_skill_index_files) from hermes_cli.config import load_config, save_config - action = args.skills_action raw_path = getattr(args, "path", None) if raw_path: @@ -224,8 +204,7 @@ def _cmd_skills_trust(args): if root is None: print( "Not inside a git checkout. Run from a project directory or " - "pass the project root path explicitly." - ) + "pass the project root path explicitly.") return config = load_config() @@ -236,8 +215,11 @@ def _cmd_skills_trust(args): trusted = [str(t) for t in trusted] root_str = str(root) + def _same(t: str) -> bool: + return str(Path(t).expanduser().resolve()) == root_str + if action == "untrust": - kept = [t for t in trusted if str(Path(t).expanduser().resolve()) != root_str] + kept = [t for t in trusted if not _same(t)] if len(kept) == len(trusted): print(f"{root} was not trusted.") return @@ -247,8 +229,7 @@ def _cmd_skills_trust(args): print("Project skills from this repo will no longer load.") return - # trust - if any(str(Path(t).expanduser().resolve()) == root_str for t in trusted): + if any(_same(t) for t in trusted): print(f"Already trusted: {root}") else: trusted.append(root_str) @@ -257,14 +238,13 @@ def _cmd_skills_trust(args): print(f"Trusted: {root}") # Show what this unlocks - count = 0 - for d in _candidate_project_skills_dirs(root): - count += sum(1 for _ in iter_skill_index_files(d, "SKILL.md")) + count = sum( + sum(1 for _ in iter_skill_index_files(d, "SKILL.md")) + for d in _candidate_project_skills_dirs(root)) if count: print( f"{count} project skill(s) will load in sessions started inside " - "this repo (they take precedence over same-named profile skills)." - ) + "this repo (they take precedence over same-named profile skills).") else: subdirs = " or ".join(PROJECT_SKILLS_SUBDIRS) print(f"No project skills found yet — add them under {subdirs}.") diff --git a/hermes_cli/main_install_repair.py b/hermes_cli/main_install_repair.py index a7fa5769d2..f8f3d45235 100644 --- a/hermes_cli/main_install_repair.py +++ b/hermes_cli/main_install_repair.py @@ -1,12 +1,11 @@ """Install/update recovery: interrupted-install markers, lazy-refresh repair, Windows shim quarantine, dependency verification. -Split out of ``hermes_cli/main.py``; every moved name is re-imported there, so -``hermes_cli.main.`` keeps resolving (and monkeypatching) as before. -Names that stay in main are imported lazily inside the functions that use them -(call-time resolution keeps ``hermes_cli.main.`` patches effective and -avoids an import cycle). +Split out of ``hermes_cli/main.py``; every moved name is re-imported there so +``hermes_cli.main.`` keeps resolving (and monkeypatching). Names that stay in main +are imported lazily inside the functions that use them (patch-effective, no import cycle). """ +import contextlib import logging import os import shlex @@ -23,46 +22,74 @@ from hermes_cli import _early_recovery as _early_recovery_mod logger = logging.getLogger("hermes_cli.main") -def _load_installable_optional_extras(group: str = "all") -> list[str]: - """Return optional extras referenced by a dependency group. - - ``group`` is usually ``all`` (desktop/server broad install) or - ``termux-all`` (Termux-compatible broad install). - """ +def _pyproject_project(debug_fmt: str | None = None) -> dict | None: + """``[project]`` table of pyproject.toml, or ``None`` when absent/unreadable.""" from hermes_cli.main import PROJECT_ROOT + pyproject = PROJECT_ROOT / "pyproject.toml" + if not pyproject.is_file(): + return None try: import tomllib - - with (PROJECT_ROOT / "pyproject.toml").open("rb") as handle: + with pyproject.open("rb") as handle: project = tomllib.load(handle).get("project", {}) - except Exception: - return [] + except Exception as exc: + if debug_fmt: + logger.debug(debug_fmt, exc) + return None + return project if isinstance(project, dict) else None - optional_deps = project.get("optional-dependencies", {}) + +def _naive_requirement(spec: str) -> tuple[str, str]: + """``(name, head)`` of a ``name OP version ; marker`` spec without ``packaging``.""" + head = spec.split(";", 1)[0].strip() + bare = head + for op in ("==", ">=", "<=", "~=", ">", "<", "!="): + if op in bare: + bare = bare.split(op, 1)[0] + break + return bare.strip().split("[", 1)[0].strip(), head + + +def _parse_requirements(raw_deps: list[str]) -> list[tuple[str, "object | None", str]]: + """``(name, marker, head)`` per dep spec — ``packaging`` when importable, else a naive split.""" + parsed: list[tuple[str, "object | None", str]] = [] + try: + from packaging.requirements import Requirement # type: ignore + for spec in raw_deps: + try: + req = Requirement(spec) + except Exception: + continue + parsed.append((req.name, req.marker, spec.split(";", 1)[0].strip())) + except Exception: + for spec in raw_deps: + name, head = _naive_requirement(spec) + if name: + parsed.append((name, None, head)) + return parsed + + +def _load_installable_optional_extras(group: str = "all") -> list[str]: + """Return optional extras referenced by a dependency group (``all`` or ``termux-all``).""" + optional_deps = (_pyproject_project() or {}).get("optional-dependencies", {}) if not isinstance(optional_deps, dict): return [] - - refs = optional_deps.get(group, []) referenced: list[str] = [] - for ref in refs: + for ref in optional_deps.get(group, []): if "[" in ref and "]" in ref: name = ref.split("[", 1)[1].split("]", 1)[0] if name in optional_deps: referenced.append(name) - return referenced # Install-scoped breadcrumbs live next to the venv (not under $HERMES_HOME) # because the venv is shared across profiles. -# -# ``.update-incomplete`` — generic core ``.[all]`` install was interrupted. -# Cleared only after a confirmed full dependency reinstall/recovery. -# -# ``.lazy-refresh-incomplete`` — lazy-backend refresh phase may have corrupted -# packages. Cleared only after import-probe repair confirms healthy (not when -# probes are unavailable/indeterminate). Narrow lazy probes must NEVER clear -# the generic core marker (#58004 review). +# ``.update-incomplete`` — generic core ``.[all]`` install was interrupted; +# cleared only after a confirmed full dependency reinstall/recovery. +# ``.lazy-refresh-incomplete`` — lazy-backend refresh may have corrupted packages; +# cleared only after import-probe repair confirms healthy (never on indeterminate). +# Narrow lazy probes must NEVER clear the generic core marker. def _update_marker_path() -> Path: from hermes_cli.main import PROJECT_ROOT return PROJECT_ROOT / ".update-incomplete" @@ -74,19 +101,10 @@ def _lazy_refresh_marker_path() -> Path: def _pytest_owns_live_checkout(root: Path) -> bool: - """True when running under pytest AND ``root`` is this checkout itself. - - Tests that drive update/recovery without sandboxing ``PROJECT_ROOT`` - must neither litter the live repo root with recovery breadcrumbs - (a leftover ``.lazy-refresh-incomplete`` / ``.update-incomplete`` - false-arms recovery on the developer's next real launch) nor run a real - reinstall against the executing venv. Sandboxed tests point at a - tmp_path and are unaffected (same posture as - ``managed_scope._under_pytest``).""" - return ( - "PYTEST_CURRENT_TEST" in os.environ - and root == Path(__file__).resolve().parent.parent - ) + """True under pytest when ``root`` is this checkout: unsandboxed update/recovery tests must + neither litter the live repo root with breadcrumbs (false-arming the developer's next launch) + nor run a real reinstall against the executing venv (cf. ``managed_scope._under_pytest``).""" + return "PYTEST_CURRENT_TEST" in os.environ and root == Path(__file__).resolve().parent.parent def _clear_marker_file(path: Path, *, label: str) -> None: @@ -110,54 +128,12 @@ def _clear_lazy_refresh_incomplete_marker() -> None: _clear_marker_file(_lazy_refresh_marker_path(), label="lazy-refresh-incomplete") -def _recover_from_interrupted_install() -> None: - """Finish update work left half-done by a prior ``hermes update``. +def _claim_recovery_lock(lock_path: Path) -> bool: + """Atomically claim the single-flight recovery lock; False when another process holds it. - Handles two independent breadcrumbs: - - - ``.update-incomplete`` — core ``.[all]`` install interrupted. Recovers - via full quarantined reinstall. Never cleared by the narrow lazy-refresh - import probes alone. - - ``.lazy-refresh-incomplete`` — lazy-backend refresh may have corrupted - packages. Recovers via package-only import probes; cleared only when - probes confirm healthy/repaired (indeterminate keeps the marker). - - Never raises: a recovery failure must not block launch. If it can't - self-heal it prints the manual command and leaves the relevant marker so - the next launch tries again. - - Concurrency: markers live next to the shared venv, so a gateway start - plus a CLI launch (or two profiles starting at once) can both see them. - An ``O_EXCL`` lockfile ensures only one process runs recovery; the - others skip and let the winner clear markers. - - Output: everything — our status lines AND the streamed pip/uv install - (which inherits fd 1) — is routed to stderr. Launches whose stdout is a - protocol stream (``hermes acp`` speaks JSON-RPC on stdout) must never get - install noise on stdout. - """ - from hermes_cli.main import PROJECT_ROOT, _clear_update_incomplete_marker, _pytest_owns_live_checkout, _recover_core_update_marker_locked, _update_marker_path - if _pytest_owns_live_checkout(PROJECT_ROOT): - return - core_marker = _update_marker_path().exists() - lazy_marker = _lazy_refresh_marker_path().exists() - if not core_marker and not lazy_marker: - return - - # Skip in managed/Docker installs and on PyPI installs with no git checkout: - # those don't run the source-tree update path, so a stray marker is not ours - # to act on. Just clear it. - if not (PROJECT_ROOT / "pyproject.toml").is_file(): - _clear_update_incomplete_marker() - _clear_lazy_refresh_incomplete_marker() - return - - # Single-flight guard: atomically claim the recovery lock. If another - # process holds it, skip — it is running the same reinstall into the same - # shared venv right now. A crashed holder leaves a stale lock; break it - # after an hour (well past any realistic install) so recovery can't be - # wedged forever. - lock_path = PROJECT_ROOT / ".update-incomplete.lock" + A crashed holder's stale lock is broken after an hour (well past any realistic install). + Failing to CREATE the lock (read-only fs, perms) proceeds unlocked — the install itself + will surface the real problem.""" try: fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) os.write(fd, f"{os.getpid()}\n".encode()) @@ -168,29 +144,26 @@ def _recover_from_interrupted_install() -> None: lock_path.unlink() except OSError: pass - return + return False except OSError as exc: - # Couldn't create the lock (read-only fs, perms). Proceed unlocked — - # the install itself will surface the real problem. logger.debug("Could not create install-recovery lock: %s", exc) + return True + +@contextlib.contextmanager +def _stdout_to_stderr(): + """Route Python prints AND the fd 1 that pip/uv inherit to stderr: launches whose stdout is + a protocol stream (``hermes acp`` speaks JSON-RPC on stdout) must never get install noise.""" saved_stdout_fd = None saved_sys_stdout = sys.stdout try: - # Route Python-level prints AND subprocess-inherited fd 1 to stderr - # for the duration of recovery (see docstring: ACP stdout safety). - try: - saved_stdout_fd = os.dup(1) - os.dup2(2, 1) - except OSError: - saved_stdout_fd = None - sys.stdout = sys.stderr - - if lazy_marker: - _recover_lazy_refresh_marker_locked() - - if _update_marker_path().exists(): - _recover_core_update_marker_locked() + saved_stdout_fd = os.dup(1) + os.dup2(2, 1) + except OSError: + saved_stdout_fd = None + sys.stdout = sys.stderr + try: + yield finally: sys.stdout = saved_sys_stdout if saved_stdout_fd is not None: @@ -199,6 +172,39 @@ def _recover_from_interrupted_install() -> None: os.close(saved_stdout_fd) except OSError: pass + + +def _recover_from_interrupted_install() -> None: + """Finish update work left half-done by a prior ``hermes update``. + + ``.update-incomplete`` recovers via full quarantined reinstall; ``.lazy-refresh-incomplete`` + via package-only import probes (cleared only when probes confirm healthy/repaired). Never + raises: on failure it prints the manual command and leaves the marker for the next launch. + Concurrent launches race on the shared venv, so an ``O_EXCL`` lockfile lets one process + recover while the others skip. + """ + from hermes_cli.main import PROJECT_ROOT, _clear_update_incomplete_marker, _pytest_owns_live_checkout, _recover_core_update_marker_locked, _update_marker_path + if _pytest_owns_live_checkout(PROJECT_ROOT): + return + lazy_marker = _lazy_refresh_marker_path().exists() + if not lazy_marker and not _update_marker_path().exists(): + return + # Managed/Docker installs and git-less PyPI installs never run the source-tree + # update path, so a stray marker is not ours to act on. Just clear it. + if not (PROJECT_ROOT / "pyproject.toml").is_file(): + _clear_update_incomplete_marker() + _clear_lazy_refresh_incomplete_marker() + return + lock_path = PROJECT_ROOT / ".update-incomplete.lock" + if not _claim_recovery_lock(lock_path): + return + try: + with _stdout_to_stderr(): + if lazy_marker: + _recover_lazy_refresh_marker_locked() + if _update_marker_path().exists(): + _recover_core_update_marker_locked() + finally: try: lock_path.unlink() except OSError: @@ -210,100 +216,79 @@ def _recover_lazy_refresh_marker_locked() -> None: from hermes_cli.main import _default_venv_install_target, _repair_venv_via_import_probes print( "⚠ A previous lazy-backend refresh may have left the venv unhealthy — " - "running import-based package repair..." - ) + "running import-based package repair...") install_prefix, install_env = _default_venv_install_target() status = _repair_venv_via_import_probes(install_prefix, env=install_env) if status in ("healthy", "repaired"): _clear_lazy_refresh_incomplete_marker() print("✓ Lazy-refresh venv recovery confirmed — install is healthy again.") return - if status == "indeterminate": - print( - " ⚠ Import probes unavailable — cannot confirm venv health. " - "Leaving `.lazy-refresh-incomplete` for the next launch." - ) - else: - print( - " ⚠ Lazy-refresh package repair incomplete. " - "Leaving `.lazy-refresh-incomplete` for the next launch." - ) - print(" Recover manually with:") - all_specs = _lazy_refresh_repair_specs( - sorted(set(_LAZY_REFRESH_REPAIR_PACKAGES.values())) - ) - print( - f" {' '.join(install_prefix)} install --force-reinstall " - + " ".join(shlex.quote(s) for s in all_specs) - ) + indeterminate = status == "indeterminate" + problem = ( + "Import probes unavailable — cannot confirm venv health." if indeterminate + else "Lazy-refresh package repair incomplete.") + print(f" ⚠ {problem} Leaving `.lazy-refresh-incomplete` for the next launch.") + if indeterminate: + return + print(" Recover manually with:") + all_specs = _lazy_refresh_repair_specs(sorted(set(_LAZY_REFRESH_REPAIR_PACKAGES.values()))) + print( + f" {' '.join(install_prefix)} install --force-reinstall " + + " ".join(shlex.quote(s) for s in all_specs)) def _recover_core_update_marker_locked() -> None: """Heal ``.update-incomplete`` via full ``.[all]`` reinstall only. - Narrow lazy-refresh import probes are not sufficient proof that a generic - interrupted core install finished — a missing dep outside that probe set - would otherwise look healthy and clear the breadcrumb too early. + Narrow lazy-refresh import probes are not proof that a generic interrupted core + install finished — a missing dep outside that probe set would look healthy and + clear the breadcrumb too early. """ from hermes_cli.main import PROJECT_ROOT, _clear_update_incomplete_marker, _default_venv_install_target, _repair_venv_via_import_probes print( "⚠ A previous `hermes update` was interrupted mid-install — " - "finishing dependency installation now..." - ) + "finishing dependency installation now...") - # Windows: a normal ``hermes.exe`` launch always has the launcher as an - # ancestor. Full editable reinstall uses quarantine so the live shim can - # still be replaced. Package-only import repair may help as first aid but - # must NEVER clear this core marker on its own (#58004 review). + # Windows: a ``hermes.exe`` launch has the launcher as an ancestor; the quarantined full + # reinstall can still replace it. Package-only repair is first aid and NEVER clears the marker. self_locked = _windows_running_hermes_launcher_locked() if self_locked: install_prefix, install_env = _default_venv_install_target() print( " → Running from hermes.exe; applying package-only first aid, " "then quarantined full reinstall (core marker stays until that " - "succeeds)..." - ) + "succeeds)...") _repair_venv_via_import_probes(install_prefix, env=install_env) - try: from hermes_cli import _install_repair as _ir - # ensure_uv bootstraps the installer itself when missing (the early - # pass's stdlib-only lookup cannot); keeping it here means the late - # path still self-heals a venv whose uv vanished mid-update. + # ensure_uv bootstraps uv itself when missing (the early pass's stdlib-only lookup + # cannot), so a venv whose uv vanished mid-update still heals. from hermes_cli.managed_uv import ensure_uv - ensure_uv() - - # Delegate the install itself to the shared stdlib executor so both - # this late path and the pre-import early pass run exactly the same - # reinstall. Called inside the same stdout→stderr redirect already - # established by _recover_from_interrupted_install, so - # run_core_install's own redirect nests harmlessly. + # Shared stdlib executor: late path and pre-import early pass run exactly the same + # reinstall. Its own stdout→stderr redirect nests harmlessly inside ours. _ir.run_core_install(PROJECT_ROOT) - _clear_update_incomplete_marker() print("✓ Dependency installation recovered — your install is healthy again.") except Exception as exc: - # Leave the marker in place so the next launch retries. Give the user - # the exact manual recovery command in the meantime. + # Leave the marker so the next launch retries; give the exact manual command. logger.debug("Interrupted-install recovery failed: %s", exc) print("✗ Could not auto-recover the interrupted install.") - if self_locked: - print( - " Hermes is still running from the launcher that needs " - "replacing. Close other Hermes windows, restart from a " - "different terminal, then run:" - ) - print(f' cd /d "{PROJECT_ROOT}"') - print( - f' "{sys.executable}" -m pip install -e ".[all]"' - ) - else: - print(" Recover manually with:") - print(f" cd {PROJECT_ROOT}") - print(f" {sys.executable} -m ensurepip --upgrade") - print(f" {sys.executable} -m pip install -e '.[all]'") + manual = ( + " Hermes is still running from the launcher that needs " + "replacing. Close other Hermes windows, restart from a " + "different terminal, then run:", + f' cd /d "{PROJECT_ROOT}"', + f' "{sys.executable}" -m pip install -e ".[all]"', + ) if self_locked else ( + " Recover manually with:", + f" cd {PROJECT_ROOT}", + f" {sys.executable} -m ensurepip --upgrade", + f" {sys.executable} -m pip install -e '.[all]'", + ) + for line in manual: + print(line) def _norm_exe_path(path) -> str: @@ -317,22 +302,11 @@ def _norm_exe_path(path) -> str: def _windows_shim_in_process_chain() -> Path | None: """The venv console shim this process runs from or under, if any. - ``venv\\Scripts\\hermes.exe`` is a launcher that runs the interpreter with - the shim itself as its script, and that keeps the shim open — without - ``FILE_SHARE_DELETE`` — for the whole process lifetime. So every - ``hermes ...`` command holds its own shim, and an editable install run - from one can never rewrite it (#88838, #89599). - - Two independent probes, because either can come up empty. Process - ancestry finds the launcher when it is a separate parent process, but - needs psutil. This process's own launch paths (``sys.argv[0]``, - ``__main__.__file__``, the module spec origin) cover the rest — the - runpy/zipapp launch puts ``\\__main__.py`` there, which a plain - argv[0] check misses. - - Candidates are intersected with the project venv's own shims, so a - ``hermes.exe`` belonging to some other install never matches. - """ + ``venv\\Scripts\\hermes.exe`` holds itself open (no ``FILE_SHARE_DELETE``) for the whole + process lifetime, so an editable install run from one can never rewrite it. Two probes, since + either can come up empty: own launch paths (argv[0], ``__main__`` file/spec origin — runpy/ + zipapp puts ``\\__main__.py`` there) and psutil ancestry. Candidates are intersected + with the project venv's own shims so a foreign ``hermes.exe`` never matches.""" from hermes_cli.main import _hermes_exe_shims, _is_windows, _venv_scripts_dir if not _is_windows(): return None @@ -349,20 +323,17 @@ def _windows_shim_in_process_chain() -> Path | None: path = path.parent return shims.get(_norm_exe_path(path)) - candidates: list[str] = list(sys.argv[:1]) main_mod = sys.modules.get("__main__") - for attr in (getattr(main_mod, "__file__", None), - getattr(getattr(main_mod, "__spec__", None), "origin", None)): - if attr: - candidates.append(attr) + candidates = [*sys.argv[:1], *filter(None, ( + getattr(main_mod, "__file__", None), + getattr(getattr(main_mod, "__spec__", None), "origin", None)))] for candidate in candidates: matched = _match(candidate) if matched is not None: return matched - try: + with contextlib.suppress(Exception): import psutil - me = psutil.Process() for proc in [me] + list(me.parents()): try: @@ -371,16 +342,11 @@ def _windows_shim_in_process_chain() -> Path | None: continue if matched is not None: return matched - except Exception: - return None return None def _windows_running_hermes_launcher_locked() -> bool: - """True when a venv ``hermes*.exe`` shim is this process or an ancestor. - - Best-effort: returns False when psutil is unavailable or inspection fails. - """ + """True when a venv ``hermes*.exe`` shim is this process or an ancestor (best-effort).""" from hermes_cli.main import _windows_shim_in_process_chain return _windows_shim_in_process_chain() is not None @@ -392,71 +358,38 @@ _UPDATE_REEXEC_ENV = "HERMES_UPDATE_REEXEC" def _reexec_dependency_sync_off_windows_shim() -> bool: """Hand the dependency sync to the venv interpreter, off the console shim. - Returns True when a child was spawned and the caller must exit at once, - releasing the shim before the child reaches ``pip install -e .``. Returns - False to continue the sync in-process. + Returns True when a child was spawned and the caller must exit at once (releasing the + shim before the child reaches ``pip install -e .``); False to continue in-process. - Called at the dependency-sync boundary, NOT at the top of the command — - the same placement rule as the native-module deferral beside it, and for - the same reason (#86735): a hand-off that fires before the fetch detaches - every run, including the ``Already up to date!`` no-op that never touches - the venv at all, and it takes the interactive prompts with it. By the time - we reach here the code swap is done and every question — stash, branch - switch, config migration — has already been asked and answered in the - user's own console. Only the venv rewrite is left, and that is the single - step that genuinely cannot run from inside the shim. - - ``venv\\Scripts\\hermes.exe`` is a launcher that runs the interpreter with - the shim as its script and holds it open without ``FILE_SHARE_DELETE`` for - the whole command, so the quarantine rename is refused and uv fails to - replace it with os error 32 (#88838, #89599). - - A child is required, and waiting on it cannot work: this process holds the - handle the child needs released, so a parent that waits deadlocks against - the work it is waiting for. Windows has no exec to escape with either. - The shell therefore returns while the install runs on; the child keeps the - console and prints its own result, and ``--gateway`` writes the true exit - code to ``.update_exit_code`` for the gateway watcher. - - The child re-runs ``hermes update``, so the whole remaining flow — the - dependency sync and the node/web/lazy-refresh tail behind it — still - happens exactly once. ``_UPDATE_REEXEC_ENV`` marks it so it cannot spawn - another child, and so the "already up to date" early return does not - swallow the sync it was spawned to perform (the checkout is current by - now; that is the point). - - The caller has already written ``.update-incomplete``, so a child that - dies mid-install is finished by the next launch's recovery instead of - leaving a half-synced venv. Anything that stops the hand-off (no venv - python, spawn refused) returns False and syncs in-process, where the - pre-existing os-error-32 path and its marker recovery still apply. - """ + Called at the dependency-sync boundary, NOT at the top of the command: by then the code swap + is done and every interactive question has been answered; only the venv rewrite — the one + step that cannot run inside the shim — remains. Earlier would detach every run (even the + ``Already up to date!`` no-op) and take the prompts along. Waiting on the child deadlocks + (we hold the handle it needs) and Windows has no exec, so the shell returns; the child keeps + the console, prints its own result, and ``--gateway`` writes the true exit code to + ``.update_exit_code``. The child re-runs ``hermes update`` so the sync and its tail happen + exactly once; ``_UPDATE_REEXEC_ENV`` stops it spawning again and stops the "already up to + date" early return from swallowing the sync. ``.update-incomplete`` is already written, so + a child that dies mid-install is finished by the next launch's recovery.""" from hermes_cli.main import _UPDATE_REEXEC_ENV, _windows_shim_in_process_chain if os.environ.get(_UPDATE_REEXEC_ENV) == "1": return False shim = _windows_shim_in_process_chain() if shim is None: return False - from hermes_constants import venv_python_path - python_exe = venv_python_path(shim.parent.parent, windows=True) cmd = [str(python_exe), "-m", "hermes_cli.main", *sys.argv[1:]] if python_exe.is_file(): try: subprocess.Popen( - cmd, - env={**os.environ, _UPDATE_REEXEC_ENV: "1"}, - stdin=subprocess.DEVNULL, - ) + cmd, env={**os.environ, _UPDATE_REEXEC_ENV: "1"}, stdin=subprocess.DEVNULL) print( f"→ Windows: {shim.name} cannot replace itself while it runs; " - "finishing the dependency install under the venv Python." - ) + "finishing the dependency install under the venv Python.") print( " The code update is already applied. The install continues " - "below and this shell returns right away." - ) + "below and this shell returns right away.") return True except OSError as exc: logger.debug("Dependency-sync hand-off via %s failed: %s", python_exe, exc) @@ -471,13 +404,11 @@ def _default_venv_install_target() -> tuple[list[str], dict[str, str] | None]: from hermes_cli.main import PROJECT_ROOT, _is_termux_env try: from hermes_cli.managed_uv import ensure_uv - uv_bin = ensure_uv() except Exception: uv_bin = None if uv_bin: from hermes_constants import project_venv_dir - venv_dir = project_venv_dir(PROJECT_ROOT) or PROJECT_ROOT / "venv" env = {**os.environ, "VIRTUAL_ENV": str(venv_dir)} if _is_termux_env(env): @@ -488,16 +419,11 @@ def _default_venv_install_target() -> tuple[list[str], dict[str, str] | None]: def _run_install_with_heartbeat( - cmd: list[str], - *, - env: dict[str, str] | None = None, - heartbeat_interval_seconds: int = 30, + cmd: list[str], *, env: dict[str, str] | None = None, heartbeat_interval_seconds: int = 30 ) -> None: - """Run dependency install command with periodic heartbeat output. + """Run a dependency install, printing an elapsed-time heartbeat while pip/uv is silent. - Some resolvers/build backends (especially when compiling Rust/C extensions) - can stay quiet for minutes. Emit a simple elapsed-time heartbeat so users - know ``hermes update`` is still progressing even if pip/uv itself is silent. + Resolvers/build backends compiling Rust/C extensions can stay quiet for minutes. """ from hermes_cli.main import PROJECT_ROOT done = threading.Event() @@ -510,23 +436,36 @@ def _run_install_with_heartbeat( print( f" … still installing dependencies ({elapsed}s elapsed)" " — compiling Rust/C extensions can take several minutes", - flush=True, - ) + flush=True) t = threading.Thread(target=_heartbeat, daemon=True) t.start() try: - subprocess.run( - cmd, - cwd=PROJECT_ROOT, - check=True, - env=env, - ) + subprocess.run(cmd, cwd=PROJECT_ROOT, check=True, env=env) finally: done.set() t.join(timeout=0.2) +def _run_repair_step(run, cmd: list[str], *, log_msg: str, fail_msg: str | None, **kwargs) -> bool: + """``run(cmd, **kwargs)``; on ``CalledProcessError`` log + print the failure and return False.""" + try: + run(cmd, **kwargs) + except subprocess.CalledProcessError as e: + logger.warning(log_msg, e) + if fail_msg is not None: + print(fail_msg) + return False + return True + + +def _report_still_missing(missing: list[str], hint: str, *, ok: str) -> None: + if missing: + print(f" ⚠ Still missing after repair: {', '.join(missing)}. {hint}") + else: + print(ok) + + def _is_windows() -> bool: return sys.platform == "win32" @@ -535,121 +474,80 @@ def _venv_scripts_dir() -> Path | None: """Return the venv Scripts directory if we're running inside the project venv.""" from hermes_cli.main import PROJECT_ROOT, _is_windows from hermes_constants import project_venv_dir, venv_bin_dir - venv_dir = project_venv_dir(PROJECT_ROOT) if venv_dir is None: return None - scripts = venv_bin_dir(venv_dir, windows=_is_windows()) return scripts if scripts.is_dir() else None def _hermes_exe_shims(scripts_dir: Path) -> list[Path]: - """Entry-point shims that uv may try to rewrite during ``pip install -e .``. - - On Windows these are .exe launchers generated by setuptools/uv. On POSIX - they're regular Python scripts which can be replaced atomically — no - self-replacement hazard exists outside Windows. - """ + """Entry-point shims uv may rewrite during ``pip install -e .`` — Windows .exe launchers + only; POSIX shims are plain scripts replaced atomically.""" from hermes_cli.main import _is_windows if not _is_windows(): return [] - names = set(_load_console_script_names()) or {"hermes", "hermes-agent", "hermes-acp"} - # The gateway shim is not a [project.scripts] entry point, but older - # update/install paths still rewrite and quarantine it. + # Not a [project.scripts] entry point, but older update/install paths still + # rewrite and quarantine it. names.add("hermes-gateway") return [scripts_dir / f"{name}.exe" for name in sorted(names)] +_QUARANTINE_BACKOFF_MS = (0, 100, 250, 500, 1000) + + +def _rename_with_backoff(source: Path, target: Path, attempts: int) -> OSError | None: + """Rename with the quarantine backoff ladder; returns the last ``OSError`` or ``None``.""" + for delay_ms in _QUARANTINE_BACKOFF_MS[:attempts]: + if delay_ms: + _time.sleep(delay_ms / 1000.0) + try: + source.rename(target) + return None + except OSError as e: + last_exc = e + return last_exc + + def _quarantine_running_hermes_exe( - scripts_dir: Path, *, max_attempts: int = 4, - failed_out: list[str] | None = None, + scripts_dir: Path, *, max_attempts: int = 4, failed_out: list[str] | None = None ) -> list[tuple[Path, Path]]: - """Pre-empt Windows file lock on the running ``hermes.exe``. + """Pre-empt the Windows file lock on the running ``hermes.exe``. - Windows allows RENAMING a mapped/running executable (the kernel tracks the - file by handle, not path), but blocks DELETE/REPLACE while it's loaded. uv - needs to overwrite the entry-point shims during ``pip install -e .``; - when ``hermes update`` runs, ``hermes.exe`` IS the live process, and uv - fails with ``Access is denied. (os error 5)``. - - We rename live shims to ``hermes.exe.old.`` first. uv then writes - fresh shims at the original paths. The ``.old`` files are cleaned up on - the next hermes invocation by ``_cleanup_quarantined_exes``. - - Rename can still fail when *another* process has opened the .exe without - ``FILE_SHARE_DELETE`` — typically AV real-time scanners with transient - handles (recovers in <1s), or the Hermes Desktop backend child process - (won't recover until the user closes it). We mitigate: - - 1. Retry up to ``max_attempts`` times with exponential backoff - (100/250/500/1000 ms). Handles the AV-scanner case. - 2. If all retries fail, print a clear warning naming the most likely - culprit (running Hermes Desktop / gateway / REPL). - - The updater's own launcher is no longer one of those culprits: an update - started from ``hermes.exe`` re-runs itself under the venv Python before - reaching here (``_reexec_dependency_sync_off_windows_shim``). - - Returns the list of (original, quarantined) pairs so the caller can roll - back if the install itself fails before uv writes a replacement. - - ``failed_out``: when provided, the names of shims whose rename failed on - every attempt are appended — callers that must not mutate a contended - venv (the update dependency sync, #87331) check it and refuse instead of - letting the install run into a half-broken state. + Windows allows RENAMING a running executable but blocks DELETE/REPLACE (uv fails with + ``Access is denied. (os error 5)``), so live shims are renamed to ``.old.`` + first; ``_cleanup_quarantined_exes`` sweeps the ``.old`` files next invocation. Rename can + still fail when another process holds the .exe without ``FILE_SHARE_DELETE`` (AV scanner: + transient; Hermes Desktop backend child: until closed) — retry with backoff, then warn + naming the likely culprit. Returns ``(original, quarantined)`` pairs for rollback; + ``failed_out`` collects shims whose rename failed every attempt so the update dependency + sync can refuse instead of stranding a half-broken venv. """ from hermes_cli.main import _hermes_exe_shims, _is_windows moved: list[tuple[Path, Path]] = [] if not _is_windows(): return moved - - import time - - stamp = int(time.time() * 1000) - # Backoff schedule: first attempt is immediate, subsequent ones sleep. - # 100ms / 250ms / 500ms covers the typical AV scanner re-scan window. - backoff_ms = [0, 100, 250, 500, 1000] - attempts = max(1, min(max_attempts, len(backoff_ms))) - + stamp = int(_time.time() * 1000) + # First attempt immediate; 100/250/500ms covers the typical AV re-scan window. + attempts = max(1, min(max_attempts, len(_QUARANTINE_BACKOFF_MS))) for shim in _hermes_exe_shims(scripts_dir): if not shim.exists(): continue target = shim.with_suffix(shim.suffix + f".old.{stamp}") - - last_exc: OSError | None = None - for attempt in range(attempts): - delay = backoff_ms[attempt] / 1000.0 - if delay: - time.sleep(delay) - try: - shim.rename(target) - moved.append((shim, target)) - last_exc = None - break - except OSError as e: - last_exc = e - continue - + last_exc = _rename_with_backoff(shim, target, attempts) if last_exc is None: + moved.append((shim, target)) continue - # Every rename failed. Deferring one to next boot via - # MOVEFILE_DELAY_UNTIL_REBOOT used to be the fallback here, but it - # cannot help: it needs elevation we don't have, and when it does - # land it frees nothing for the install running right now while - # queueing an operation that will move a later, freshly repaired shim - # aside at next boot. Report and let uv try its luck instead — - # sometimes its own retry handling pulls through. + # Every rename failed. MOVEFILE_DELAY_UNTIL_REBOOT is no fallback (needs elevation, frees + # nothing now, moves a later repaired shim aside at boot). Report; let uv try its luck. print( f" ⚠ Could not quarantine {shim.name} ({last_exc.__class__.__name__}: " - f"another process is holding it open)." - ) + f"another process is holding it open).") print( " Close Hermes Desktop, exit other `hermes` REPLs, stop the " - "gateway, or pause AV scanning, then re-run `hermes update`." - ) + "gateway, or pause AV scanning, then re-run `hermes update`.") if failed_out is not None: failed_out.append(shim.name) @@ -657,21 +555,13 @@ def _quarantine_running_hermes_exe( _PENDING_RENAME_KEY = r"SYSTEM\CurrentControlSet\Control\Session Manager" - - _PENDING_RENAME_VALUE = "PendingFileRenameOperations" -def _filter_pending_shim_renames( - entries: list[str], shims: list[Path] -) -> tuple[list[str], int]: - """Drop shim-quarantine pairs from a PendingFileRenameOperations value. - - The value is a flat REG_MULTI_SZ of (source, target) pairs, and other - installers share it, so only pairs matching our own - ```` -> ``.old.`` naming are removed. Returns the - entries to keep and how many pairs were dropped. - """ +def _filter_pending_shim_renames(entries: list[str], shims: list[Path]) -> tuple[list[str], int]: + """Drop our ```` -> ``.old.`` pairs from a PendingFileRenameOperations + value (a flat REG_MULTI_SZ of (source, target) pairs shared with other installers). + Returns the entries to keep and how many pairs were dropped.""" import ntpath def _norm(value: str) -> str: @@ -696,32 +586,22 @@ def _filter_pending_shim_renames( def _cleanup_pending_shim_renames(scripts_dir: Path) -> int: - """Drop reboot renames older Hermes versions queued for our shims. - - Hermes used to fall back to ``MoveFileExW(MOVEFILE_DELAY_UNTIL_REBOOT)`` - when the quarantine rename failed. Those entries outlive the update that - queued them, so at the next boot they move away whatever now sits at the - shim path — including a shim a later repair just wrote. Needs elevation - to remove (same as it needed to create); a no-op otherwise. - """ + """Drop reboot renames older Hermes versions queued for our shims: ``MOVEFILE_DELAY_UNTIL_REBOOT`` + fallbacks outlive the update that queued them and move away whatever sits at the shim path + at next boot — even a shim a later repair just wrote. Needs elevation; a no-op otherwise.""" from hermes_cli.main import _filter_pending_shim_renames, _hermes_exe_shims, _is_windows if not _is_windows(): return 0 try: import winreg - with winreg.OpenKey( - winreg.HKEY_LOCAL_MACHINE, - _PENDING_RENAME_KEY, - 0, + winreg.HKEY_LOCAL_MACHINE, _PENDING_RENAME_KEY, 0, winreg.KEY_QUERY_VALUE | winreg.KEY_SET_VALUE, ) as key: entries, value_type = winreg.QueryValueEx(key, _PENDING_RENAME_VALUE) if value_type != winreg.REG_MULTI_SZ or not isinstance(entries, list): return 0 - kept, removed = _filter_pending_shim_renames( - entries, _hermes_exe_shims(scripts_dir) - ) + kept, removed = _filter_pending_shim_renames(entries, _hermes_exe_shims(scripts_dir)) if not removed: return 0 if kept: @@ -734,67 +614,34 @@ def _cleanup_pending_shim_renames(scripts_dir: Path) -> int: def _restore_quarantined_exes(moved: list[tuple[Path, Path]]) -> None: - """Roll back ``_quarantine_running_hermes_exe`` if uv didn't write replacements. - - This is the safety-critical direction. A failed *quarantine* only aborts an - update; a failed *restore* leaves the install with no ``hermes`` on PATH, - and therefore no way to run the command that would repair it (#75584). The - outbound rename already retries a lock, so this one must too rather than - swallow the first ``OSError`` in silence. - - Delegates to the stdlib-only helper that the early-recovery copy in - ``_install_repair`` also uses, so the two cannot drift apart. - """ + """Roll back ``_quarantine_running_hermes_exe`` if uv didn't write replacements. Safety- + critical: a failed quarantine only aborts an update; a failed restore leaves no ``hermes`` + on PATH. Delegates to the stdlib-only retrying helper shared with ``_install_repair``.""" _early_recovery_mod.restore_quarantined_shims(moved) class ShimQuarantineError(RuntimeError): - """A live ``hermes*.exe`` shim could not be renamed aside (#87331). - - Raised by :func:`_run_quarantined_install` in ``strict_quarantine`` mode - BEFORE the install command runs. A shim that cannot even be renamed means - another process holds the venv hard enough that the dependency sync would - die partway and strand the install half-updated — the update must refuse, - not warn-and-continue. - """ + """A live ``hermes*.exe`` shim could not be renamed aside. Raised by + :func:`_run_quarantined_install` in ``strict_quarantine`` mode BEFORE the install runs: a + process holds the venv hard enough that the sync would die partway — refuse, don't warn.""" def __init__(self, failed_shims: list[str]): self.failed_shims = list(failed_shims) - super().__init__( - "could not quarantine live shim(s): " + ", ".join(self.failed_shims) - ) + super().__init__("could not quarantine live shim(s): " + ", ".join(self.failed_shims)) def _run_quarantined_install( - cmd: list[str], - *, - env: dict[str, str] | None = None, - scripts_dir: Path | None = None, + cmd: list[str], *, env: dict[str, str] | None = None, scripts_dir: Path | None = None, strict_quarantine: bool = False, ) -> None: """Run an editable install, quarantining the running ``hermes.exe`` first. - Any ``pip install -e .`` (or ``--reinstall``) rewrites the entry-point - shims, and on Windows the live ``hermes.exe`` is the running process — - pip can neither delete nor overwrite it, so without quarantine the shim - is left missing and ``hermes`` drops off PATH. This wraps - :func:`_run_install_with_heartbeat` with the same rename-out-of-the-way / - restore-on-failure dance that the primary install path uses, so EVERY - install that touches the shims is protected — including the - verification-repair reinstalls in - :func:`_verify_core_dependencies_installed`, which previously called - ``_run_install_with_heartbeat`` directly and bypassed quarantine. - - ``strict_quarantine=True`` (the update dependency sync, #87331): a shim - whose rename failed every retry means a process is holding the venv - without ``FILE_SHARE_DELETE`` — the install WILL hit the same lock on - .pyd files and strand the venv between versions. Roll the successful - renames back and raise :class:`ShimQuarantineError` WITHOUT running the - install. Non-strict callers (post-sync entry-point repair) keep the old - warn-and-try behavior: their venv is already mutated, so refusing buys - nothing. - - Off-Windows (``scripts_dir is None``) this is a thin pass-through. + Every editable install rewrites the entry-point shims; on Windows the live ``hermes.exe`` + can be neither deleted nor overwritten, so without quarantine ``hermes`` drops off PATH. + ``strict_quarantine=True`` (the update dependency sync): a shim whose rename failed every + retry proves a hard venv hold — the install WILL hit the same lock on .pyd files — so roll + back and raise :class:`ShimQuarantineError` without installing. Non-strict callers already + mutated the venv, so refusing buys nothing. ``scripts_dir is None`` is a pass-through. """ from hermes_cli.main import ShimQuarantineError, _quarantine_running_hermes_exe, _restore_quarantined_exes, _run_install_with_heartbeat moved: list[tuple[Path, Path]] = [] @@ -807,36 +654,21 @@ def _run_quarantined_install( try: _run_install_with_heartbeat(cmd, env=env) finally: - # Restore shims when the installer didn't write replacements — on - # FAILURE (install died before the entry-points step) and on SUCCESS - # too: uv audits an already-satisfied editable install as a no-op and - # rewrites no entry points, which would otherwise leave the shims - # quarantined aside and `hermes` missing from PATH after a green - # install (#75584). _restore_quarantined_exes skips any shim the - # installer actually replaced, so this never clobbers fresh output. - # Errors are not swallowed — the finally re-raises whatever escaped. + # Restore on FAILURE and SUCCESS: an already-satisfied editable install is a uv no-op + # that rewrites no entry points. Skips shims the installer replaced; finally re-raises. if scripts_dir is not None: _restore_quarantined_exes(moved) -# A quarantine file younger than this may belong to an update running RIGHT -# NOW in another process, whose restore step still needs it. Deleting one -# mid-flight destroys the only copy of that shim. +# A quarantine file younger than this may belong to an update running RIGHT NOW in +# another process, whose restore step still needs it — the only copy of that shim. _QUARANTINE_GRACE_SECONDS = 15 * 60 def _quarantine_stamp_ms(stale: Path) -> int | None: - """The ``.old.`` stamp in a quarantine filename, or ``None``. - - ``None`` means the name was not produced by - :func:`_quarantine_running_hermes_exe`. We neither rescue nor delete those: - the sweep should not destroy files whose provenance it cannot establish, and - they are not ours to put back. - - Parsed from the NAME rather than ``st_mtime`` because ``rename`` preserves - the original shim's mtime, which records when uv wrote the shim — days - earlier, in general — not when it was quarantined. - """ + """The ``.old.`` stamp in a quarantine filename; ``None`` (not ours — neither rescued + nor deleted) otherwise. Parsed from the NAME, not ``st_mtime``: ``rename`` preserves the + shim's mtime (when uv wrote it), not when it was quarantined.""" try: return int(stale.name.rsplit(".old.", 1)[1]) except (IndexError, ValueError): @@ -846,164 +678,94 @@ def _quarantine_stamp_ms(stale: Path) -> int | None: def _cleanup_quarantined_exes(scripts_dir: Path | None = None) -> None: """Sweep — and where necessary RESCUE — ``hermes.exe.old.*`` from updates. - Called early on every hermes invocation. Two cases the old unconditional - ``unlink()`` got wrong, both ending with ``hermes`` gone from PATH: - - 1. **Orphan rescue.** If ``hermes.exe`` is missing while - ``hermes.exe.old.*`` is present, that .old file is the ONLY surviving - copy of the shim — an update died, or its restore failed, between - the rename and uv writing a replacement (#75584). Deleting it converts a - one-rename recovery into a full reinstall. Put it back instead, through - the same retry-and-report helper the update-time restore uses. - 2. **Concurrency.** A fresh quarantine file may belong to an update in - flight in another process (the desktop update button racing a shell - ``hermes update`` does exactly this). Leave anything inside the grace - window alone; a later run sweeps it. - - Silent no-op on non-Windows, when there is nothing to do, or on - file-locked / permission errors. + Called early on every invocation. Two cases an unconditional ``unlink()`` gets wrong: + (1) orphan rescue — ``hermes.exe`` missing while ``hermes.exe.old.*`` exists means the .old + file is the ONLY surviving copy (update died between rename and uv's write); put it back via + the same retry-and-report helper the update-time restore uses. (2) concurrency — a fresh + quarantine file may belong to an update in flight in another process; leave anything inside + the grace window alone. Silent no-op on non-Windows, nothing to do, or locked/permission errors. """ from hermes_cli.main import _QUARANTINE_GRACE_SECONDS, _cleanup_pending_shim_renames, _is_windows, _quarantine_stamp_ms, _venv_scripts_dir if not _is_windows(): return - if scripts_dir is None: - scripts_dir = _venv_scripts_dir() + scripts_dir = scripts_dir if scripts_dir is not None else _venv_scripts_dir() if scripts_dir is None: return _cleanup_pending_shim_renames(scripts_dir) - now = _time.time() - try: candidates = [ - (stamp, stale) - for stale, stamp in ( - (p, _quarantine_stamp_ms(p)) for p in scripts_dir.glob("*.exe.old.*") - ) - if stamp is not None - ] + (stamp, stale) for stale in scripts_dir.glob("*.exe.old.*") + if (stamp := _quarantine_stamp_ms(stale)) is not None] except OSError: return - - # Newest first by PARSED stamp. Sorting the raw filenames lexicographically - # only tracks recency while every stamp shares a digit width: a stray - # ``.old.999`` sorts above a 13-digit epoch-ms stamp and would be the copy - # rescued onto the live shim name. + # Newest first by PARSED stamp: lexicographic order breaks when a stray ``.old.999`` exists. candidates.sort(key=lambda pair: pair[0], reverse=True) - for stamp, stale in candidates: try: original = stale.with_name(stale.name.rsplit(".old.", 1)[0]) - if not original.exists(): - # Orphan rescue: this is the last copy of the shim, so it gets - # the retry ladder and the recovery message, not a bare rename. + # Orphan rescue: last copy of the shim — retry ladder + recovery message. _early_recovery_mod.restore_quarantined_shims([(original, stale)]) continue - if now - stamp / 1000.0 < _QUARANTINE_GRACE_SECONDS: continue # may be a live quarantine from a concurrent update - stale.unlink() except OSError: pass # still locked or in use — try again next run -# Import probes for venv corruption after a failed lazy ``uv pip install``. -# Metadata can look fine while ``.py`` files were removed mid-install (#57828). -# Canonical tables live in the stdlib-only ``_early_recovery`` module (which -# also probes/repairs BEFORE this module's third-party imports can run) so the -# early and full recovery layers can never drift apart. +# Import probes for venv corruption after a failed lazy ``uv pip install`` (metadata can +# look fine while ``.py`` files were removed mid-install). Canonical tables live in the +# stdlib-only ``_early_recovery`` module so the early and full recovery layers never drift. _LAZY_REFRESH_IMPORT_PROBES: tuple[tuple[str, str], ...] = ( - _early_recovery_mod.LAZY_REFRESH_IMPORT_PROBES -) + _early_recovery_mod.LAZY_REFRESH_IMPORT_PROBES) +_LAZY_REFRESH_REPAIR_PACKAGES: dict[str, str] = _early_recovery_mod.LAZY_REFRESH_REPAIR_PACKAGES -_LAZY_REFRESH_REPAIR_PACKAGES: dict[str, str] = ( - _early_recovery_mod.LAZY_REFRESH_REPAIR_PACKAGES -) - - -def _run_package_only_install( - cmd: list[str], - *, - env: dict[str, str] | None = None, -) -> None: - """Run a package-only pip/uv install without quarantining entry-point shims. - - ``pip install --upgrade pip`` and ``--force-reinstall `` do not - rewrite ``hermes.exe``. The editable-install quarantine path would rename - shims without uv recreating them on Windows (#57828). - """ +def _run_package_only_install(cmd: list[str], *, env: dict[str, str] | None = None) -> None: + """Package-only pip/uv install — no shim quarantine: ``--force-reinstall `` never rewrites + ``hermes.exe``, and the quarantine path would rename shims uv then never recreates.""" from hermes_cli.main import _run_install_with_heartbeat _run_install_with_heartbeat(cmd, env=env) def _lazy_refresh_repair_specs(packages: list[str]) -> list[str]: """Map repair package names to their declared pin specs in pyproject.toml.""" - from hermes_cli.main import PROJECT_ROOT - try: - import tomllib # Python 3.11+ - except ImportError: # pragma: no cover + project = _pyproject_project("lazy refresh repair spec lookup failed: %s") + if project is None: return packages - - pyproject = PROJECT_ROOT / "pyproject.toml" - if not pyproject.is_file(): - return packages - - try: - with open(pyproject, "rb") as f: - raw_deps = tomllib.load(f).get("project", {}).get("dependencies", []) or [] - except Exception as exc: - logger.debug("lazy refresh repair spec lookup failed: %s", exc) - return packages - - name_to_spec: dict[str, str] = {} - try: - from packaging.requirements import Requirement # type: ignore - - for spec in raw_deps: - try: - req = Requirement(spec) - name_to_spec[req.name.lower()] = spec.split(";", 1)[0].strip() - except Exception: - continue - except Exception: - for spec in raw_deps: - head = spec.split(";", 1)[0].strip() - bare = head - for op in ("==", ">=", "<=", "~=", ">", "<", "!="): - if op in bare: - bare = bare.split(op, 1)[0] - break - key = bare.strip().split("[", 1)[0].strip().lower() - if key: - name_to_spec[key] = head - + name_to_spec = { + name.lower(): head + for name, _, head in _parse_requirements(project.get("dependencies", []) or [])} return [name_to_spec.get(pkg.lower(), pkg) for pkg in packages] -def _detect_broken_lazy_refresh_imports( - install_cmd_prefix: list[str], - *, - env: dict[str, str] | None = None, -) -> list[str] | None: - """Probe lazy-refresh packages via real imports. +def _venv_probe(venv_python: Path, script: str, *args: str, env: dict[str, str] | None): + """Run ``script`` in the target venv's interpreter, capturing UTF-8 stdout.""" + return subprocess.run( + [str(venv_python), "-c", script, *args], + capture_output=True, + text=True, encoding="utf-8", errors="replace", + check=False, + env=env) - Returns: - - ``[]`` when probes ran and every package imported cleanly - - ``[dist, ...]`` when probes ran and some packages failed - - ``None`` when the probe could not run (missing venv Python, subprocess - failure, non-zero probe exit) — this is *indeterminate*, not healthy - """ + +def _nonblank_lines(text: str) -> list[str]: + return [line.strip() for line in text.splitlines() if line.strip()] + + +def _detect_broken_lazy_refresh_imports( + install_cmd_prefix: list[str], *, env: dict[str, str] | None = None) -> list[str] | None: + """Probe lazy-refresh packages via real imports: ``[]`` all clean, ``[dist, ...]`` failures, + ``None`` when the probe could not run (no venv Python, subprocess failure, non-zero exit) + — *indeterminate*, not healthy.""" from hermes_cli.main import _resolve_install_target_python venv_python = _resolve_install_target_python(install_cmd_prefix, env) if venv_python is None: return None - probe_lines = "\n".join( - f" ({mod!r}, {attr!r})," for mod, attr in _LAZY_REFRESH_IMPORT_PROBES - ) + f" ({mod!r}, {attr!r})," for mod, attr in _LAZY_REFRESH_IMPORT_PROBES) check_script = ( "import os\n" "import sys\n" @@ -1025,161 +787,97 @@ def _detect_broken_lazy_refresh_imports( " broken.append(mod)\n" " except Exception:\n" " broken.append(mod)\n" - "print('\\n'.join(broken))\n" - ) + "print('\\n'.join(broken))\n") try: - result = subprocess.run( - [str(venv_python), "-c", check_script], - capture_output=True, - text=True, encoding="utf-8", errors="replace", - check=False, - env=env, - ) + result = _venv_probe(venv_python, check_script, env=env) except Exception as exc: logger.debug("lazy refresh import probe failed: %s", exc) return None - if result.returncode != 0: - logger.debug( - "lazy refresh import probe exited %s: %s", - result.returncode, - (result.stderr or "")[:200], - ) + logger.debug("lazy refresh import probe exited %s: %s", + result.returncode, (result.stderr or "")[:200]) return None - - broken_modules = [ - line.strip() for line in result.stdout.splitlines() if line.strip() - ] packages: list[str] = [] - seen: set[str] = set() - for mod in broken_modules: + for mod in _nonblank_lines(result.stdout): pkg = _LAZY_REFRESH_REPAIR_PACKAGES.get(mod) - if pkg and pkg not in seen: - seen.add(pkg) + if pkg and pkg not in packages: packages.append(pkg) return packages def _repair_broken_lazy_refresh_imports( - install_cmd_prefix: list[str], - packages: list[str], - *, - env: dict[str, str] | None = None, + install_cmd_prefix: list[str], packages: list[str], *, env: dict[str, str] | None = None ) -> bool: """Force-reinstall ``packages`` and re-probe imports. Never raises.""" from hermes_cli.main import _detect_broken_lazy_refresh_imports, _run_package_only_install if not packages: return True - specs = _lazy_refresh_repair_specs(packages) - try: - _run_package_only_install( - install_cmd_prefix + ["install", "--force-reinstall", *specs], - env=env, - ) - except subprocess.CalledProcessError as exc: - logger.warning("lazy refresh venv repair failed: %s", exc) + if not _run_repair_step( + _run_package_only_install, install_cmd_prefix + ["install", "--force-reinstall", *specs], + env=env, log_msg="lazy refresh venv repair failed: %s", fail_msg=None): return False - - after = _detect_broken_lazy_refresh_imports(install_cmd_prefix, env=env) # Indeterminate re-probe is not confirmed success. - return after == [] + return _detect_broken_lazy_refresh_imports(install_cmd_prefix, env=env) == [] def _repair_venv_via_import_probes( - install_cmd_prefix: list[str], - *, - env: dict[str, str] | None = None, -) -> str: + install_cmd_prefix: list[str], *, env: dict[str, str] | None = None) -> str: """Probe imports and force-reinstall any broken lazy-refresh packages. - Uses real ``import`` checks (not distribution metadata) so a venv where - METADATA remains but ``.py`` files were wiped mid-install is still - detected (#57828). Package-only reinstall — never rewrites ``hermes.exe``. - - Never raises. Returns one of: - - ``"healthy"`` — probes ran and found nothing broken - - ``"repaired"`` — probes found breakage and force-reinstall confirmed clean - - ``"failed"`` — probes found breakage and repair did not confirm clean - - ``"indeterminate"`` — probes could not run; do NOT treat as healthy + Real ``import`` checks (not distribution metadata) catch a venv where METADATA remains + but ``.py`` files were wiped mid-install. Package-only reinstall — never rewrites + ``hermes.exe``. Never raises. Returns ``"healthy"``, ``"repaired"``, ``"failed"`` + (repair did not confirm clean) or ``"indeterminate"`` (probes could not run; NOT healthy). """ from hermes_cli.main import _detect_broken_lazy_refresh_imports, _repair_broken_lazy_refresh_imports broken = _detect_broken_lazy_refresh_imports(install_cmd_prefix, env=env) if broken is None: - print( - " ⚠ Import probes unavailable — cannot confirm venv package health." - ) + print(" ⚠ Import probes unavailable — cannot confirm venv package health.") return "indeterminate" if not broken: return "healthy" print( " → Detected corrupted venv packages via import probes: " - f"{', '.join(broken)}; repairing..." - ) - if _repair_broken_lazy_refresh_imports( - install_cmd_prefix, broken, env=env - ): + f"{', '.join(broken)}; repairing...") + if _repair_broken_lazy_refresh_imports(install_cmd_prefix, broken, env=env): print(" ✓ Venv repair succeeded") return "repaired" - manual = " ".join( - shlex.quote(s) for s in _lazy_refresh_repair_specs(broken) - ) + manual = " ".join(shlex.quote(s) for s in _lazy_refresh_repair_specs(broken)) print(" ⚠ Venv repair incomplete. Run manually, then `hermes update`:") - print( - f" {' '.join(install_cmd_prefix)} install --force-reinstall {manual}" - ) + print(f" {' '.join(install_cmd_prefix)} install --force-reinstall {manual}") return "failed" def _is_uv_command(install_cmd_prefix: list[str]) -> bool: - """True when the install command is a uv/uvx invocation. - - Handles a bare uv binary (``uv`` / ``uvx``, any extension), a path to - one, and ``python -m uv`` / ``python -m uvx`` — the naive basename check - misses the module form and launcher wrappers whose name does not contain - "uv". - """ + """True for a uv/uvx binary (bare or path) or ``python -m uv`` / ``python -m uvx``.""" if not install_cmd_prefix: return False first = str(install_cmd_prefix[0]).lower() if "uv" in Path(first).name: return True - # python -m uv / python -m uvx - if len(install_cmd_prefix) >= 3 and first.endswith(("python", "python.exe")): - return install_cmd_prefix[1] == "-m" and install_cmd_prefix[2] in ( - "uv", - "uvx", - ) - return False + return ( + len(install_cmd_prefix) >= 3 + and first.endswith(("python", "python.exe")) + and install_cmd_prefix[1] == "-m" + and install_cmd_prefix[2] in ("uv", "uvx")) def _insert_python_pin(args: list[str]) -> list[str]: - """Insert ``--python `` into a uv command line. - - If the caller already passed ``--python``, its value wins (uv's last-wins - semantics are ambiguous; the explicit caller intent should not be - overridden by the fallback pin). - """ + """Insert ``--python `` into a uv command line; an explicit caller ``--python`` wins.""" if "--python" in args: return args return [args[0], "--python", str(sys.executable), *args[1:]] def _interpreter_scripts_dir() -> Path | None: - """Scripts/bin directory of the running interpreter (sys.executable). - - Used when pinning an install to ``sys.executable`` on a site-packages - install where ``PROJECT_ROOT / "venv"`` does not exist: the entry-point - shims uv rewrites live next to the interpreter, not under a project venv. - Layout comes from the canonical ``venv_bin_dir`` helper (#76105 — - hand-rolling Scripts/bin is lint-tested against). - """ + """Scripts/bin dir of ``sys.executable``: on a site-packages install ``PROJECT_ROOT/venv`` + does not exist and the shims uv rewrites live next to the interpreter. Layout via the + canonical ``venv_bin_dir`` (hand-rolling Scripts/bin is lint-tested against).""" from hermes_cli.main import _is_windows from hermes_constants import venv_bin_dir - exe = Path(sys.executable) - # sys.executable lives IN the bin/Scripts dir; its parent.parent is the - # env root venv_bin_dir derives from. + # sys.executable lives IN the bin/Scripts dir; parent.parent is the env root. cand = venv_bin_dir(exe.parent.parent, windows=_is_windows()) if cand.is_dir(): return cand @@ -1187,72 +885,38 @@ def _interpreter_scripts_dir() -> Path | None: def _install_python_dependencies_with_optional_fallback( - install_cmd_prefix: list[str], - *, - env: dict[str, str] | None = None, - group: str = "all", + install_cmd_prefix: list[str], *, env: dict[str, str] | None = None, group: str = "all" ) -> None: """Install base deps plus as many optional extras as the environment supports. - By default this targets ``.[all]``; Termux callers can pass - ``group='termux-all'`` to use the curated Android-compatible profile. - - On Windows, pre-renames live ``hermes.exe`` / ``hermes-gateway.exe`` shims - in the venv Scripts dir before each install attempt so uv can write fresh - copies (Windows blocks REPLACE on a running .exe but allows RENAME). See - ``_quarantine_running_hermes_exe`` for the rationale. - - When ``env`` carries a ``VIRTUAL_ENV`` that does not exist (a pip / - site-packages install whose ``PROJECT_ROOT`` is the interpreter's - ``site-packages`` directory, where ``PROJECT_ROOT / "venv"`` is never - created), ``uv pip`` fails with ``Failed to inspect Python interpreter from - active virtual environment`` before doing any work. Pin the install at the - running interpreter instead so the update/recovery path succeeds on those - installs (#71510 fixed the ZIP path, #83335 fixed lazy-deps; this closes the - shared helper for the remaining callers). + Targets ``.[all]`` by default; Termux callers pass ``group='termux-all'``. On Windows every + attempt quarantines the live ``hermes*.exe`` shims first. When ``env`` carries a + ``VIRTUAL_ENV`` that does not exist (pip / site-packages install), ``uv pip`` fails with + ``Failed to inspect Python interpreter from active virtual environment`` before doing any + work — pin the install at the running interpreter instead. """ from hermes_cli.main import _insert_python_pin, _interpreter_scripts_dir, _is_windows, _load_installable_optional_extras, _run_quarantined_install, _venv_scripts_dir, _verify_console_scripts_installed, _verify_core_dependencies_installed scripts_dir = _venv_scripts_dir() if _is_windows() else None - # A pip / site-packages install has no PROJECT_ROOT/venv; the caller still - # passes VIRTUAL_ENV=PROJECT_ROOT/venv, which does not exist. uv would fail - # before installing anything ("Failed to inspect Python interpreter from - # active virtual environment"). Detect the stale pointer and pin the target - # interpreter explicitly instead of trusting the nonexistent venv. - pin_python = False - if ( - env - and env.get("VIRTUAL_ENV") - and not Path(env["VIRTUAL_ENV"]).is_dir() - and install_cmd_prefix - and _is_uv_command(install_cmd_prefix) - ): - # Only uv needs the explicit pin; pip resolves the target from - # sys.executable itself and has no --python flag. - pin_python = True + # Only uv needs the explicit pin; pip resolves the target from sys.executable itself. + pin_python = bool( + env and env.get("VIRTUAL_ENV") and not Path(env["VIRTUAL_ENV"]).is_dir() + and install_cmd_prefix and _is_uv_command(install_cmd_prefix)) + if pin_python: env = {**env} env.pop("VIRTUAL_ENV", None) - # When we pin to sys.executable, the entry-point shims that uv will - # rewrite live in that interpreter's Scripts/bin directory, NOT in - # PROJECT_ROOT/venv (which does not exist on a site-packages install). - # Quarantining the wrong dir means the running hermes.exe stays locked - # on Windows and the install fails exactly like the original bug. Only - # override when the venv-derived dir is missing; otherwise keep it. + # Pinned to sys.executable, the shims uv rewrites live in THAT interpreter's Scripts + # dir, not PROJECT_ROOT/venv; quarantining the wrong dir leaves hermes.exe locked. if scripts_dir is None and _is_windows(): scripts_dir = _interpreter_scripts_dir() def _install(args: list[str]) -> None: if pin_python: args = _insert_python_pin(args) - # strict_quarantine: this is the UPDATE dependency sync. A shim that - # cannot be renamed aside proves a hard venv hold; running uv anyway - # is how installs strand half-updated (#87331). ShimQuarantineError - # propagates to the update's sync boundary, which defers via the - # update-incomplete marker instead of mutating a contended venv. + # strict_quarantine: this is the UPDATE dependency sync; ShimQuarantineError propagates + # to the sync boundary, which defers via the update-incomplete marker instead. _run_quarantined_install( - install_cmd_prefix + args, env=env, scripts_dir=scripts_dir, - strict_quarantine=True, - ) + install_cmd_prefix + args, env=env, scripts_dir=scripts_dir, strict_quarantine=True) try: _install(["install", "-e", f".[{group}]"]) @@ -1264,7 +928,6 @@ def _install_python_dependencies_with_optional_fallback( ) _install(["install", "-e", "."]) - failed_extras: list[str] = [] installed_extras: list[str] = [] for extra in _load_installable_optional_extras(group=group): @@ -1273,333 +936,168 @@ def _install_python_dependencies_with_optional_fallback( installed_extras.append(extra) except subprocess.CalledProcessError: failed_extras.append(extra) - if installed_extras: - print( - f" ✓ Reinstalled optional extras individually: {', '.join(installed_extras)}" - ) + print(f" ✓ Reinstalled optional extras individually: {', '.join(installed_extras)}") if failed_extras: - print( - f" ⚠ Skipped optional extras that still failed: {', '.join(failed_extras)}" - ) - - # Belt-and-suspenders: verify every declared core dependency from - # pyproject.toml's [project.dependencies] is actually importable in the - # target venv. uv's incremental resolver has — in the wild — produced - # partial installs where a newly added base dep (e.g. ``pathspec``) - # silently fails to land on top of a half-stale venv, and the only - # symptom is a downstream subprocess crashing with ModuleNotFoundError - # hours later inside ``hermes update``'s desktop-rebuild or skill-sync - # stage. Reinstall with --reinstall to force resolution if anything is - # missing, then re-verify so the failure surfaces here instead of - # downstream. + print(f" ⚠ Skipped optional extras that still failed: {', '.join(failed_extras)}") + # uv's incremental resolver has left newly added base deps silently missing on a half-stale + # venv, surfacing hours later as a downstream ModuleNotFoundError. Verify here instead. _verify_core_dependencies_installed(install_cmd_prefix, env=env, group=group) _verify_console_scripts_installed(install_cmd_prefix, env=env) def _load_console_script_names() -> list[str]: """Return ``[project.scripts]`` entry-point names from pyproject.toml.""" - from hermes_cli.main import PROJECT_ROOT - try: - import tomllib # Python 3.11+ - except ImportError: # pragma: no cover - return [] - - pyproject = PROJECT_ROOT / "pyproject.toml" - if not pyproject.is_file(): - return [] - - try: - with open(pyproject, "rb") as f: - data = tomllib.load(f) - scripts = data.get("project", {}).get("scripts", {}) or {} - return [str(name) for name in scripts if name] - except Exception as e: - logger.debug("console script verification: failed to read pyproject.toml: %s", e) - return [] + project = _pyproject_project("console script verification: failed to read pyproject.toml: %s") + scripts = (project or {}).get("scripts", {}) or {} + return [str(name) for name in scripts if name] def _verify_console_scripts_installed( - install_cmd_prefix: list[str], - *, - env: dict[str, str] | None = None, -) -> None: + install_cmd_prefix: list[str], *, env: dict[str, str] | None = None) -> None: """Ensure every declared console_script shim exists on disk after install. - On Windows, ``uv pip install -e .`` can register ``hermes.exe`` in the - wheel RECORD while the file never lands on disk — typically when the live - ``hermes.exe`` shim is locked during ``hermes update``, or when uv/distlib - skips a launcher write. The symptom is ``hermes-agent.exe`` and - ``hermes-acp.exe`` present but ``hermes.exe`` missing, so ``hermes`` drops - off PATH even though the install reported success (issue #52931). - - If any shim is missing we reinstall with ``--reinstall -e .`` under the - same quarantine dance as the primary install path, then re-check. - """ + On Windows ``uv pip install -e .`` can register ``hermes.exe`` in the wheel RECORD while the + file never lands (live shim locked, launcher write skipped), so ``hermes`` drops off PATH + after a "successful" install. Missing shims get ``--reinstall -e .`` under quarantine.""" from hermes_cli.main import _is_windows, _run_quarantined_install, _venv_scripts_dir if not _is_windows(): return - scripts_dir = _venv_scripts_dir() - if scripts_dir is None: - return - - names = _load_console_script_names() + names = _load_console_script_names() if scripts_dir is not None else [] if not names: return def _missing() -> list[str]: - return [ - name - for name in names - if not (scripts_dir / f"{name}.exe").is_file() - ] + return [name for name in names if not (scripts_dir / f"{name}.exe").is_file()] missing = _missing() if not missing: return - print( f" ⚠ Verification: {len(missing)} console script(s) missing on disk: " - f"{', '.join(missing)}" - ) + f"{', '.join(missing)}") print(" → Reinstalling entry points with --reinstall...") - - try: - _run_quarantined_install( - install_cmd_prefix + ["install", "--reinstall", "-e", "."], - env=env, - scripts_dir=scripts_dir, - ) - except subprocess.CalledProcessError as e: - logger.warning("console script verification: repair install failed: %s", e) - print( + if not _run_repair_step( + _run_quarantined_install, install_cmd_prefix + ["install", "--reinstall", "-e", "."], + env=env, scripts_dir=scripts_dir, + log_msg="console script verification: repair install failed: %s", + fail_msg=( " ⚠ Entry point repair failed; try `hermes update --force` after " - "closing other hermes processes." - ) + "closing other hermes processes.")): return - - still_missing = _missing() - if still_missing: - print( - f" ⚠ Still missing after repair: {', '.join(still_missing)}. " - "Workaround: python -m hermes_cli.main " - ) - else: - print(" ✓ All console entry points restored") + _report_still_missing( + _missing(), "Workaround: python -m hermes_cli.main ", + ok=" ✓ All console entry points restored") -def _verify_core_dependencies_installed( - install_cmd_prefix: list[str], - *, - env: dict[str, str] | None = None, - group: str = "all", -) -> None: - """Check that every base dep from pyproject.toml is importable; if not, retry. - - Reads ``pyproject.toml`` directly (so we don't trust the venv's stale - metadata), filters out deps gated by ``;`` environment markers that don't - apply to this platform, and runs ``importlib.metadata.version()`` in the - venv interpreter for each one. If anything is missing we reinstall the - base group with ``--reinstall`` to force uv to re-resolve, then check - again. We treat the final state as a warning rather than a hard failure - so a single broken-on-PyPI dep can't block an otherwise-successful - update — but the warning makes the partial install visible at the spot - that caused it, instead of hours later in a downstream subprocess. - """ - from hermes_cli.main import PROJECT_ROOT, _is_windows, _resolve_install_target_python, _run_install_with_heartbeat, _run_quarantined_install, _venv_scripts_dir - try: - import tomllib # Python 3.11+ - except ImportError: # pragma: no cover — Python < 3.11 unsupported but be safe - return - - pyproject = PROJECT_ROOT / "pyproject.toml" - if not pyproject.is_file(): - return - - try: - with open(pyproject, "rb") as f: - data = tomllib.load(f) - raw_deps = data.get("project", {}).get("dependencies", []) or [] - except Exception as e: - logger.debug("dep verification: failed to read pyproject.toml: %s", e) - return - - # Parse each "name OP version ; marker" string into (dist_name, marker_obj). - # We use packaging.requirements when available (it ships with pip/uv envs), - # falling back to a naive split that's good enough for the canonical - # ``name==version[; marker]`` style this repo uses. - deps: list[tuple[str, "object | None"]] = [] - try: - from packaging.requirements import Requirement # type: ignore - - for spec in raw_deps: - try: - req = Requirement(spec) - deps.append((req.name, req.marker)) - except Exception: - continue - except Exception: - for spec in raw_deps: - head = spec.split(";", 1)[0] - for op in ("==", ">=", "<=", "~=", ">", "<", "!="): - if op in head: - head = head.split(op, 1)[0] - break - name = head.strip().split("[", 1)[0].strip() - if name: - deps.append((name, None)) - - # Apply environment markers to drop deps that don't apply on this platform - # (e.g. ``ptyprocess ; sys_platform != 'win32'`` is correctly skipped on - # Windows). Without markers we'd false-positive every cross-platform exclusion. +def _applicable_dependency_names(raw_deps: list[str]) -> list[str]: + """Declared dep names whose ``;`` markers apply here (else ``ptyprocess ; sys_platform != + 'win32'`` would false-positive on Windows). An unevaluable marker counts as applicable.""" applicable: list[str] = [] - for name, marker in deps: - if marker is None: - applicable.append(name) - continue + for name, marker, _ in _parse_requirements(raw_deps): try: - if marker.evaluate(): # type: ignore[union-attr] + if marker is None or marker.evaluate(): # type: ignore[union-attr] applicable.append(name) except Exception: applicable.append(name) + return applicable + +_MISSING_DEPS_SCRIPT = ( + "import importlib.metadata as md, sys\n" + "missing=[]\n" + "for name in sys.argv[1:]:\n" + " try: md.version(name)\n" + " except md.PackageNotFoundError: missing.append(name)\n" + "print('\\n'.join(missing))\n") + + +def _verify_core_dependencies_installed( + install_cmd_prefix: list[str], *, env: dict[str, str] | None = None, group: str = "all" +) -> None: + """Check that every base dep from pyproject.toml is installed in the target venv; if not, retry. + + Reads ``pyproject.toml`` directly (not the venv's stale metadata), drops deps whose ``;`` + markers don't apply here, and probes ``importlib.metadata.version()`` in the venv + interpreter. Missing deps trigger a base-group ``--reinstall``, then a per-package force + install. The final state is a warning, not a hard failure, so one broken-on-PyPI dep can't + block an otherwise-successful update — but the partial install is visible where it happened. + """ + from hermes_cli.main import _is_windows, _resolve_install_target_python, _run_install_with_heartbeat, _run_quarantined_install, _venv_scripts_dir + project = _pyproject_project("dep verification: failed to read pyproject.toml: %s") + if project is None: + return + raw_deps = project.get("dependencies", []) or [] + applicable = _applicable_dependency_names(raw_deps) if not applicable: return - - # Run the check inside the venv Python — sys.executable here may be the - # outer Python that drove ``hermes update``, not the venv we just wrote - # to. The uv install_cmd_prefix encodes which environment we targeted - # (either ``[uv, pip]`` with VIRTUAL_ENV in env, or - # ``[sys.executable, -m, pip]`` for the in-process Python); resolve the - # right interpreter for the verification. + # Probe inside the venv Python — sys.executable may be the outer Python that drove + # ``hermes update``; the install prefix/env encode which environment we targeted. venv_python = _resolve_install_target_python(install_cmd_prefix, env) if venv_python is None: return def _missing_deps() -> list[str]: - check_script = ( - "import importlib.metadata as md, sys\n" - "missing=[]\n" - "for name in sys.argv[1:]:\n" - " try: md.version(name)\n" - " except md.PackageNotFoundError: missing.append(name)\n" - "print('\\n'.join(missing))\n" - ) try: - result = subprocess.run( - [str(venv_python), "-c", check_script, *applicable], - capture_output=True, - text=True, encoding="utf-8", errors="replace", - check=False, - env=env, - ) + result = _venv_probe(venv_python, _MISSING_DEPS_SCRIPT, *applicable, env=env) except Exception as e: logger.debug("dep verification: subprocess failed: %s", e) return [] - return [line.strip() for line in result.stdout.splitlines() if line.strip()] + return _nonblank_lines(result.stdout) missing = _missing_deps() if not missing: return - print( f" ⚠ Verification: {len(missing)} declared dep(s) missing after install: " - f"{', '.join(missing[:8])}{'...' if len(missing) > 8 else ''}" - ) + f"{', '.join(missing[:8])}{'...' if len(missing) > 8 else ''}") print(" → Reinstalling base group with --reinstall to repair...") - - # Reinstall base group with --reinstall so uv re-resolves from scratch - # against the current pyproject. We don't pass ``[{group}]`` here on - # purpose — the missing dep is in *base* deps; rerunning the full all- - # extras install can cost minutes and trips on whatever optional extra - # was already broken upstream. Base is fast and is what's actually wrong. - # - # Quarantine the running ``hermes.exe`` first: ``--reinstall -e .`` - # rewrites the entry-point shims, and on Windows pip can't overwrite the - # live launcher, which would leave ``hermes`` off PATH. + # Base group only, not ``[{group}]``: the missing dep is a *base* dep and the all-extras + # install costs minutes. Quarantine first: ``--reinstall -e .`` rewrites the shims. scripts_dir = _venv_scripts_dir() if _is_windows() else None - repair_args = ["install", "--reinstall", "-e", "."] - try: - _run_quarantined_install( - install_cmd_prefix + repair_args, env=env, scripts_dir=scripts_dir - ) - except subprocess.CalledProcessError as e: - logger.warning("dep verification: repair install failed: %s", e) - print(" ⚠ Repair install failed; check `hermes update` output above.") + if not _run_repair_step( + _run_quarantined_install, install_cmd_prefix + ["install", "--reinstall", "-e", "."], + env=env, scripts_dir=scripts_dir, + log_msg="dep verification: repair install failed: %s", + fail_msg=" ⚠ Repair install failed; check `hermes update` output above."): return - still_missing = _missing_deps() if not still_missing: print(" ✓ All declared core dependencies now installed") return - - # Last-ditch: install each remaining missing dep with its pin directly. - # Useful when uv's resolver thinks the env is satisfied but the on-disk - # package metadata says otherwise (rare but observed). - name_to_spec = {} - for spec in raw_deps: - head = spec.split(";", 1)[0].strip() - bare = head - for op in ("==", ">=", "<=", "~=", ">", "<", "!="): - if op in bare: - bare = bare.split(op, 1)[0] - break - name_to_spec[bare.strip().split("[", 1)[0].strip()] = head - + # Last-ditch: install each remaining missing dep with its pin directly — uv's + # resolver can think the env is satisfied while on-disk metadata disagrees. + name_to_spec = dict(_naive_requirement(spec) for spec in raw_deps) specs = [name_to_spec.get(n, n) for n in still_missing] - print( - f" → Force-installing remaining missing dep(s): {', '.join(specs)}" - ) - try: - _run_install_with_heartbeat( - install_cmd_prefix + ["install", "--reinstall", *specs], env=env - ) - except subprocess.CalledProcessError as e: - logger.warning("dep verification: per-package repair failed: %s", e) - print( + print(f" → Force-installing remaining missing dep(s): {', '.join(specs)}") + if not _run_repair_step( + _run_install_with_heartbeat, install_cmd_prefix + ["install", "--reinstall", *specs], + env=env, + log_msg="dep verification: per-package repair failed: %s", + fail_msg=( f" ⚠ Could not install: {', '.join(still_missing)}. " - "Run `hermes update --force` after closing other hermes processes." - ) + "Run `hermes update --force` after closing other hermes processes.")): return - - final_missing = _missing_deps() - if final_missing: - print( - f" ⚠ Still missing after repair: {', '.join(final_missing)}. " - "Run `hermes update --force` after closing other hermes processes." - ) - else: - print(" ✓ All declared core dependencies now installed") + _report_still_missing( + _missing_deps(), "Run `hermes update --force` after closing other hermes processes.", + ok=" ✓ All declared core dependencies now installed") def _resolve_install_target_python( - install_cmd_prefix: list[str], env: dict[str, str] | None -) -> Path | None: - """Figure out which Python interpreter the install just targeted. - - ``_install_python_dependencies_with_optional_fallback`` is called with - either ``[uv, pip]`` (and a ``VIRTUAL_ENV`` env var pointing at the - target venv) or ``[sys.executable, -m, pip]`` (the in-process Python). - The verification step needs the *resulting* environment's Python so - ``importlib.metadata`` queries the right site-packages. - """ + install_cmd_prefix: list[str], env: dict[str, str] | None) -> Path | None: + """Python interpreter the install targeted: ``VIRTUAL_ENV`` from ``env`` for the + ``[uv, pip]`` shape, else ``install_cmd_prefix[0]`` for ``[sys.executable, -m, pip]``.""" from hermes_cli.main import _is_windows if env and "VIRTUAL_ENV" in env: from hermes_constants import venv_python_path - - venv_root = Path(env["VIRTUAL_ENV"]) - candidate = venv_python_path(venv_root, windows=_is_windows()) + candidate = venv_python_path(Path(env["VIRTUAL_ENV"]), windows=_is_windows()) if candidate.exists(): return candidate - - # Fallback: assume install_cmd_prefix[0] is the python interpreter (the - # ``[sys.executable, -m, pip]`` shape). Skip if it looks like ``uv``. if install_cmd_prefix: first = Path(install_cmd_prefix[0]) if first.exists() and "uv" not in first.name.lower(): return first - return None @@ -1609,54 +1107,29 @@ def _is_termux_env(env: dict[str, str] | None = None) -> bool: def _is_windows_npm_path(npm_path: str) -> bool: - """Return True if ``npm_path`` points at a Windows npm shim. + """True if ``npm_path`` points at a Windows npm shim (WSL ``/mnt/c`` interop, ``.cmd``/``.exe``, UNC). - On WSL the Windows install dir is exposed through the ``/mnt/c`` drive - mount and PATH interop, so ``shutil.which("npm")`` can hand back - ``/mnt/c/Program Files/nodejs/npm`` (or the ``npm.cmd`` / ``npm.exe`` - shim). Those are detected here by their ``.exe``/``.cmd``/``.bat`` - suffix, a ``/mnt/`` drive-mount prefix, or an embedded backslash (a UNC - path). Callers use this only on a POSIX host — on native Windows an - ``npm.cmd`` shim is the correct executable. + Callers use this only on a POSIX host — on native Windows ``npm.cmd`` is correct. """ low = npm_path.lower() - return ( - low.endswith((".exe", ".cmd", ".bat")) - or low.startswith("/mnt/") - or "\\" in npm_path - ) + return low.endswith((".exe", ".cmd", ".bat")) or low.startswith("/mnt/") or "\\" in npm_path def _resolve_node_runtime_npm() -> str | None: """Resolve an npm executable that belongs to the host's Node runtime. - On WSL/Linux ``shutil.which("npm")`` may resolve a Windows npm exposed - through PATH interop. Running that Windows npm against the Linux checkout - operates over ``\\wsl.localhost\\...`` UNC paths and fails with EISDIR / - symlink errors in symlink-heavy trees like ``ui-tui`` (#30271). Refuse a - Windows npm on a POSIX host and re-scan PATH (skipping ``/mnt/*`` interop - entries) for a Linux-native npm. Returns the npm path, or ``None`` when - no suitable npm is reachable. - """ + On WSL, PATH interop can hand back a Windows npm that fails with EISDIR / symlink errors over + ``\\\\wsl.localhost\\...`` UNC paths. Refuse it on a POSIX host and re-scan PATH minus the + ``/mnt/*`` drive mounts. ``None`` when no suitable npm is reachable.""" from hermes_cli.main import _is_windows from hermes_constants import find_node_executable - npm = find_node_executable("npm") - - # On native Windows the platform npm (``npm.cmd``) is exactly what we - # want — only reject Windows shims when we're a POSIX/WSL process. if _is_windows(): return npm - if not npm: return None - if not _is_windows_npm_path(npm): return npm - - # The first resolution was a Windows npm. Re-scan PATH skipping the - # ``/mnt/*`` Windows drive mounts WSL injects, so a Linux-native npm that - # came later on PATH is still found. for directory in os.environ.get("PATH", "").split(os.pathsep): if not directory or directory.lower().startswith("/mnt/"): continue @@ -1667,11 +1140,5 @@ def _resolve_node_runtime_npm() -> str | None: def _resolve_update_branch(args) -> str: - """Normalize ``args.branch`` into a non-empty branch name. - - Centralizes the "default to main, accept --branch override, treat empty - or whitespace-only values as the default" parsing so every consumer of - ``--branch`` (check path, git-update path, ZIP-fallback path) agrees on - the same answer. - """ + """Normalize ``args.branch`` to a non-empty name (default ``main``; blank/whitespace = default).""" return (getattr(args, "branch", None) or "main").strip() or "main" diff --git a/hermes_cli/managed_scope.py b/hermes_cli/managed_scope.py index 89c89a276c..dc03b117c9 100644 --- a/hermes_cli/managed_scope.py +++ b/hermes_cli/managed_scope.py @@ -1,12 +1,10 @@ """Managed scope — IT-pushed, user-immutable config & env layer. -This is DISTINCT from ``hermes_cli.config.is_managed()`` / ``HERMES_MANAGED``, which is a coarse -package-manager write-lock (declarative-distro / formula installs). That lock blocks all mutation; -this layer injects specific immutable values. The two are independent and may coexist. - -v1 enforcement is filesystem permissions only — see ``docs/design/managed-scope.md`` §7. v1 is -Linux/POSIX-first; ``get_managed_dir()`` is the single seam for adding macOS / Windows native -locations later. +DISTINCT from ``hermes_cli.config.is_managed()`` / ``HERMES_MANAGED`` (a coarse package-manager +write-lock that blocks all mutation); this layer injects specific immutable values. The two are +independent and may coexist. v1 enforcement is filesystem permissions only (see +``docs/design/managed-scope.md`` §7); ``get_managed_dir()`` is the single seam for adding +macOS / Windows native locations later. """ from __future__ import annotations @@ -21,8 +19,7 @@ import yaml logger = logging.getLogger(__name__) -# POSIX default. Other-platform locations are a deliberate v2 item; when added, -# they belong ONLY inside get_managed_dir(). +# POSIX default. Other-platform locations belong ONLY inside get_managed_dir(). _DEFAULT_MANAGED_DIR = Path("/etc/hermes") _CACHE_LOCK = threading.Lock() @@ -32,24 +29,17 @@ _ENV_CACHE: Dict[str, tuple] = {} def _under_pytest() -> bool: - """True when running inside the test suite. - - Used to ignore the system default ``/etc/hermes`` under tests so a real managed scope on a - dev/CI box can't leak policy into the suite. An explicit ``HERMES_MANAGED_DIR`` is still - honored because that override runs before this guard. - """ + """True inside the test suite: ignore the system ``/etc/hermes`` so a real managed scope on a + dev/CI box can't leak policy into the suite. An explicit ``HERMES_MANAGED_DIR`` still wins.""" return "PYTEST_CURRENT_TEST" in os.environ def get_managed_dir() -> Optional[Path]: """Resolve the managed-scope directory, or None when no scope is present. - Resolution (highest priority first): 1. ``$HERMES_MANAGED_DIR`` — deployment/bootstrap path - override (IT-only; never persisted to any .env). Honored only when set to a non-empty value AND - the directory exists. 2. ``/etc/hermes`` — POSIX default, when it exists. - - A non-existent directory at either tier resolves to None (no managed scope), which is the common - case and must be cheap + side-effect-free. + Priority: ``$HERMES_MANAGED_DIR`` (IT-only bootstrap override; never persisted to any .env; + honored only when non-empty AND the directory exists), then ``/etc/hermes`` when it exists. + A missing directory resolves to None — the common case, so it must be cheap + side-effect-free. """ override = os.environ.get("HERMES_MANAGED_DIR", "").strip() if override: @@ -69,11 +59,11 @@ def invalidate_managed_cache() -> None: def _cached_read(path: Path, cache: Dict[str, tuple], parse): - """Shared (mtime_ns, size)-keyed read. Returns a deepcopy of the parsed value. + """Shared (mtime_ns, size)-keyed read; returns a deepcopy of the parsed value. - Returns ``None`` when the file is absent or fails to parse (fail-open). A parse failure is - logged LOUDLY — the admin needs to know their policy isn't being applied — but never raises, so - a malformed managed file can't brick startup. + ``None`` when the file is absent or fails to parse (fail-open). A parse failure is logged + LOUDLY — the admin needs to know their policy isn't applied — but never raises, so a malformed + managed file can't brick startup. """ try: st = path.stat() @@ -92,8 +82,7 @@ def _cached_read(path: Path, cache: Dict[str, tuple], parse): logger.warning( "managed scope: failed to parse %s: %s — IGNORING this managed file. " "Admin policy from this file is NOT being applied. Fix and restart.", - path, exc, - ) + path, exc) return None with _CACHE_LOCK: cache[path_key] = (*key, copy.deepcopy(parsed)) @@ -121,15 +110,11 @@ def load_managed_env() -> Dict[str, str]: def apply_managed_overlay(config: dict) -> dict: """Overlay administrator-pinned config values on top of an already-built dict. - * expand the managed config's ``${VAR}`` refs against the PROCESS env only (never user-config- - defined refs), so a user cannot shadow a managed literal via a ${VAR} they control; * normalize - the managed config's root ``model`` key (a bare ``model: x/y`` string is promoted to - ``model.default``) so it can't clobber the dict shape callers expect; * leaf-level deep-merge - managed ON TOP, so managed wins per-leaf while sibling keys stay user-controlled. - - Fail-open: returns ``config`` unchanged if no managed scope is present or on any error — managed - scope must never break a caller's startup. Mutates and returns ``config`` (callers pass a dict - they own). + ``${VAR}`` refs in the managed config expand against the PROCESS env only, so a user cannot + shadow a managed literal via a ref they control; a bare root ``model: x/y`` string is promoted + to ``model.default`` so it can't clobber the dict shape callers expect; managed values + deep-merge ON TOP per leaf while sibling keys stay user-controlled. Fail-open: returns + ``config`` unchanged when no scope is present or on any error. Mutates and returns ``config``. """ try: managed = load_managed_config() @@ -137,14 +122,10 @@ def apply_managed_overlay(config: dict) -> dict: return config # Imported lazily to avoid an import cycle (config imports managed_scope). from hermes_cli.config import _deep_merge, _expand_env_vars, _normalize_root_model_keys - managed_expanded = _normalize_root_model_keys(_expand_env_vars(managed)) - # A bare ``model: x/y`` string in the managed file must merge as - # ``model.default`` — otherwise _deep_merge would replace the caller's - # ``model`` dict with a string and break every ``cfg["model"]["..."]`` - # read. _normalize_root_model_keys only promotes the string when there - # are root provider/base_url keys to migrate, so handle the bare case - # here (matches cli.py's own string-model handling). + # _normalize_root_model_keys only promotes the string when root provider/base_url + # keys exist to migrate; handle the bare case here (matches cli.py) so _deep_merge + # never replaces the caller's ``model`` dict with a string. if isinstance(managed_expanded.get("model"), str): managed_expanded = dict(managed_expanded) managed_expanded["model"] = {"default": managed_expanded["model"]} diff --git a/hermes_cli/managed_uv.py b/hermes_cli/managed_uv.py index 61f243c220..2ae682ce6f 100644 --- a/hermes_cli/managed_uv.py +++ b/hermes_cli/managed_uv.py @@ -1,12 +1,13 @@ """Hermes-managed uv and Python runtime repair. -The Python backing the install is different: it is shared by every Hermes profile because the -checkout's ``venv`` is shared. Runtime repair therefore uses an install-scoped store under +The Python backing the install is shared by every Hermes profile because the checkout's ``venv`` +is shared. Runtime repair therefore uses an install-scoped store under ``/.hermes-runtime/python``. A vulnerable interpreter is never reinstalled in place. """ from __future__ import annotations +import contextlib import importlib import json import logging @@ -19,15 +20,13 @@ import tempfile import time import uuid from dataclasses import dataclass +from functools import partial from pathlib import Path from typing import Callable, Optional from hermes_constants import get_hermes_home from hermes_cli.sqlite_runtime import ( - SQLiteRuntimeInfo, - isolated_interpreter_env, - probe_sqlite_runtime, -) + SQLiteRuntimeInfo, isolated_interpreter_env, probe_sqlite_runtime) logger = logging.getLogger(__name__) @@ -38,25 +37,18 @@ _ALT_VENV_NAME = ".venv" _REPAIR_LOCK_NAME = "runtime-repair.lock" _MACOS_MANAGED_PYTHON_IDENTIFIER = "com.nousresearch.hermes.managed-python" -# --------------------------------------------------------------------------- -# Public helpers -# --------------------------------------------------------------------------- +_Provisioned = tuple[Path, Path, SQLiteRuntimeInfo] def managed_uv_path() -> Path: - """Return the path where Hermes keeps *its* uv binary (``$HERMES_HOME/bin/uv[.exe]``). - - The directory may not exist yet -- callers should use ``ensure_uv()`` to bootstrap it. - """ + """Path of Hermes' own uv binary (``$HERMES_HOME/bin/uv[.exe]``); may not exist yet.""" return get_hermes_home() / "bin" / ("uv.exe" if platform.system() == "Windows" else "uv") def resolve_uv() -> Optional[str]: """Return the managed uv path if it exists, else ``None``.""" p = managed_uv_path() - if p.is_file() and os.access(p, os.X_OK): - return str(p) - return None + return str(p) if p.is_file() and os.access(p, os.X_OK) else None def managed_python_install_dir(project_root: Path | None = None) -> Path: @@ -66,92 +58,54 @@ def managed_python_install_dir(project_root: Path | None = None) -> Path: def managed_python_env( - project_root: Path | None = None, - *, - install_dir: Path | None = None, - base_env: dict[str, str] | None = None, -) -> dict[str, str]: + project_root: Path | None = None, *, install_dir: Path | None = None, + base_env: dict[str, str] | None = None) -> dict[str, str]: """Return a sanitized environment for Hermes-private uv Python commands.""" target = ( - Path(install_dir) - if install_dir is not None - else managed_python_install_dir(project_root) - ) + Path(install_dir) if install_dir is not None else managed_python_install_dir(project_root)) env = dict(os.environ if base_env is None else base_env) for key in ( - "CONDA_DEFAULT_ENV", - "CONDA_PREFIX", - "UV_PROJECT_ENVIRONMENT", - "UV_NO_MANAGED_PYTHON", - "UV_PYTHON", - "UV_PYTHON_DOWNLOADS", - "UV_SYSTEM_PYTHON", - "VIRTUAL_ENV", - "PYTHONHOME", - "PYTHONPATH", - ): + "CONDA_DEFAULT_ENV", "CONDA_PREFIX", "UV_PROJECT_ENVIRONMENT", "UV_NO_MANAGED_PYTHON", + "UV_PYTHON", "UV_PYTHON_DOWNLOADS", "UV_SYSTEM_PYTHON", "VIRTUAL_ENV", "PYTHONHOME", + "PYTHONPATH"): env.pop(key, None) env.update({ - "UV_MANAGED_PYTHON": "1", - "UV_NO_CONFIG": "1", - "UV_PYTHON_INSTALL_BIN": "0", - "UV_PYTHON_INSTALL_DIR": str(target), - "UV_PYTHON_INSTALL_REGISTRY": "0", - }) + "UV_MANAGED_PYTHON": "1", "UV_NO_CONFIG": "1", "UV_PYTHON_INSTALL_BIN": "0", + "UV_PYTHON_INSTALL_DIR": str(target), "UV_PYTHON_INSTALL_REGISTRY": "0"}) return env def _macos_sign_managed_python(python: Path) -> bool: """Give a newly downloaded managed Python a stable macOS code identity. - python-build-standalone binaries are ad-hoc signed, which leaves macOS TCC with a cdhash-only - identity that changes whenever Hermes provisions a new runtime generation. An identifier-pinned - designated requirement gives those generations a stable identity even when no Developer ID - certificate is available locally. - - Signing is deliberately best effort. Runtime repair exists to remove a security vulnerability, - so an unavailable or incompatible ``codesign`` must not prevent the fixed interpreter from being - installed. + python-build-standalone binaries are ad-hoc signed, so TCC sees a cdhash-only identity that + changes every runtime generation; an identifier-pinned designated requirement keeps it stable + without a Developer ID. Best effort: a missing/incompatible ``codesign`` must not block repair. """ if platform.system() != "Darwin": return False - codesign = shutil.which("codesign") if not codesign: - logger.info( - "macOS codesign is unavailable; using the downloaded Python signature" - ) + logger.info("macOS codesign is unavailable; using the downloaded Python signature") return False - - requirement = ( - "=designated => identifier " - f'"{_MACOS_MANAGED_PYTHON_IDENTIFIER}"' - ) + requirement = f'=designated => identifier "{_MACOS_MANAGED_PYTHON_IDENTIFIER}"' try: + sign = [ + codesign, "--force", "--deep", "--sign", "-", "--timestamp=none", + "--identifier", _MACOS_MANAGED_PYTHON_IDENTIFIER, + "--requirements", requirement, str(python)] + verify = [codesign, "--verify", "--deep", "--strict", str(python)] steps = ( - ( - [ - codesign, "--force", "--deep", "--sign", "-", "--timestamp=none", - "--identifier", _MACOS_MANAGED_PYTHON_IDENTIFIER, - "--requirements", requirement, str(python), - ], - "could not stably sign managed Python %s: %s", - "codesign failed", - ), - ( - [codesign, "--verify", "--deep", "--strict", str(python)], - "macOS signature verification failed for managed Python %s: %s", - "verification failed", - ), - ) + (sign, "could not stably sign managed Python %s: %s", "codesign failed"), + (verify, "macOS signature verification failed for managed Python %s: %s", + "verification failed")) for cmd, warning, fallback in steps: result = subprocess.run( cmd, check=False, capture_output=True, text=True, encoding="utf-8", errors="replace" ) if result.returncode != 0: logger.warning( - warning, python, (result.stderr or result.stdout or fallback).strip() - ) + warning, python, (result.stderr or result.stdout or fallback).strip()) return False return True except Exception as exc: @@ -182,26 +136,19 @@ class _RepairLock: def _report_runtime_repair_failure(repair: RuntimeRepairResult) -> None: if repair.backup_venv is None: - print( - " ℹ Managed Python runtime was not replaced; " - f"the existing venv is unchanged ({repair.detail})." - ) - print( - " Sessions stay protected meanwhile: Hermes keeps databases " - "out of WAL mode on this SQLite build. The next `hermes update` " - "will retry." - ) + print(" ℹ Managed Python runtime was not replaced; " + f"the existing venv is unchanged ({repair.detail}).") + print(" Sessions stay protected meanwhile: Hermes keeps databases " + "out of WAL mode on this SQLite build. The next `hermes update` " + "will retry.") return print(f" ✗ Managed Python runtime cutover needs manual recovery: {repair.detail}") print(f" Previous venv: {repair.backup_venv}") class _UvResult(str): - """``ensure_uv()`` return value that survives an update boundary. - - POSIX only. This wrapper is **never** returned on Windows — see ``ensure_uv()`` for why the - ``__iter__`` override is unsafe there. - """ + """``ensure_uv()`` return value that survives an update boundary. POSIX only: a str subclass + with an overridden ``__iter__`` is unsafe as a Windows subprocess argument.""" fresh_bootstrap: bool @@ -211,41 +158,31 @@ class _UvResult(str): return self def __iter__(self): - # Tuple-unpacking hook for legacy ``uv_bin, fresh = ensure_uv()`` sites. - # First element mirrors the historical contract: the path string, or - # ``None`` when uv is unavailable. + # Tuple-unpacking hook for legacy ``uv_bin, fresh = ensure_uv()`` sites; the first + # element keeps the historical contract (path string, or None when unavailable). return iter(((str(self) or None), self.fresh_bootstrap)) def _ensure_uv_path( - *, - repair_observer: Callable[[RuntimeRepairResult], None] | None = None, -) -> Optional[str]: + *, repair_observer: Callable[[RuntimeRepairResult], None] | None = None) -> Optional[str]: """Resolve the managed uv path, installing it if necessary (plain ``str``/``None``).""" existing = resolve_uv() if existing: return existing - target = managed_uv_path() target.parent.mkdir(parents=True, exist_ok=True) - print(f" → Installing managed uv into {target.parent} ...") - try: _install_uv(target) except Exception as exc: logger.warning("Managed uv install failed: %s", exc) print(f" ✗ Failed to install managed uv: {exc}") return None - - # Verify result = resolve_uv() if result: print(f" ✓ Managed uv installed ({_uv_version(result)})") - # Compatibility boundary: an older, already-imported updater calls the - # freshly pulled ``ensure_uv()`` after bootstrapping uv. Repair here so - # that first update can migrate a vulnerable runtime without requiring - # a second ``hermes update``. + # Compatibility boundary: an older, already-imported updater calls the freshly pulled + # ``ensure_uv()``; repairing here lets that first update migrate a vulnerable runtime. _run_runtime_repair(result, repair_observer) else: print(" ✗ Managed uv install appeared to succeed but binary not found") @@ -260,11 +197,8 @@ def _uv_version(uv_bin: str) -> str: def _run_runtime_repair( - uv_bin: str, - repair_observer: Callable[[RuntimeRepairResult], None] | None, - *, - print_skip: bool = False, -) -> None: + uv_bin: str, repair_observer: Callable[[RuntimeRepairResult], None] | None, + *, print_skip: bool = False) -> None: """Run the vulnerable-runtime repair hook; never raises (repair is non-fatal).""" try: repair = repair_vulnerable_runtime(uv_bin) @@ -279,39 +213,29 @@ def _run_runtime_repair( def ensure_uv( - *, - repair_observer: Callable[[RuntimeRepairResult], None] | None = None, -): - """Return the managed uv path, installing it first if necessary. + *, repair_observer: Callable[[RuntimeRepairResult], None] | None = None): + """Return the managed uv path, installing it first if necessary; falsy on failure, never raises. - On **POSIX** the result is a :class:`_UvResult` (a ``str`` subclass) that is both usable - directly as the path *and* unpackable as ``(path, fresh_bootstrap)`` for older call sites parked - on a 2-tuple release — see :class:`_UvResult` for the update-boundary rationale. - - On failure the result is falsy — never raises — so callers can fall back to pip gracefully. - ``repair_observer``, when provided, receives the runtime repair result produced after a fresh uv - bootstrap. + On POSIX the result is a :class:`_UvResult` (``str`` subclass) usable as the path *and* + unpackable as ``(path, fresh_bootstrap)`` for older call sites. """ result = _ensure_uv_path(repair_observer=repair_observer) if platform.system() == "Windows": - # See docstring: a str subclass with an overridden __iter__ is unsafe as - # a Windows subprocess argument. Hand back the plain path (or None). + # See _UvResult: the __iter__ override is unsafe as a Windows subprocess argument. return result return _UvResult(result) def _uv_self_update_stamp() -> Path: from hermes_constants import get_hermes_home - return get_hermes_home() / "cache" / ".uv_self_update_stamp" def _uv_self_update_is_fresh(now: float | None = None) -> bool: - """Return True when ``uv self update`` ran recently enough to skip. + """True when ``uv self update`` ran recently enough to skip. uv releases roughly weekly while many users run ``hermes update`` daily; a blocking network - self-update on every run is waste and, offline, an unbounded hang risk. A stamp file under - HERMES_HOME caches the last successful self-update time. + self-update on every run is waste and, offline, an unbounded hang risk. """ try: age = (now if now is not None else time.time()) - _uv_self_update_stamp().stat().st_mtime @@ -321,51 +245,37 @@ def _uv_self_update_is_fresh(now: float | None = None) -> bool: def _touch_uv_self_update_stamp() -> None: - try: + with contextlib.suppress(OSError): stamp = _uv_self_update_stamp() stamp.parent.mkdir(parents=True, exist_ok=True) stamp.touch() - except OSError: - pass # uv ships releases ~weekly; refresh the managed binary at most this often. UV_SELF_UPDATE_INTERVAL_SECONDS = 7 * 24 * 3600 -# `uv self update` is a network call; unbounded it can hang forever on a -# blackholed connection (no default timeout in uv's downloader path). +# `uv self update` is a network call with no default timeout; unbounded it can hang forever. UV_SELF_UPDATE_TIMEOUT_SECONDS = 60 def update_managed_uv( - *, - repair_observer: Callable[[RuntimeRepairResult], None] | None = None, - force: bool = False, + *, repair_observer: Callable[[RuntimeRepairResult], None] | None = None, force: bool = False ) -> Optional[str]: - """Run ``uv self update`` on the managed uv binary. + """Run ``uv self update`` on the managed uv binary; returns its path, or ``None`` if absent. - Call this during ``hermes update`` so the managed copy stays current. Returns the managed path - when uv is available and ``None`` otherwise. ``repair_observer``, when provided, receives the - runtime repair result. - - The network self-update is skipped when it succeeded within the last - ``UV_SELF_UPDATE_INTERVAL_SECONDS`` (7 days) unless ``force=True``; the vulnerable-runtime - repair probe below ALWAYS runs — CVE-driven runtime repair must never be gated behind the - freshness stamp. + The network self-update is skipped when it succeeded within ``UV_SELF_UPDATE_INTERVAL_SECONDS`` + unless ``force=True``; the vulnerable-runtime repair probe ALWAYS runs — CVE-driven repair is + never gated behind the freshness stamp. """ existing = resolve_uv() if not existing: # Not installed yet — ensure_uv() will handle that elsewhere. return None - if force or not _uv_self_update_is_fresh(): try: result = subprocess.run( - [existing, "self", "update"], - capture_output=True, + [existing, "self", "update"], capture_output=True, text=True, encoding='utf-8', errors='replace', - check=False, - timeout=UV_SELF_UPDATE_TIMEOUT_SECONDS, - ) + check=False, timeout=UV_SELF_UPDATE_TIMEOUT_SECONDS) except subprocess.TimeoutExpired: logger.debug("uv self update timed out after %ss", UV_SELF_UPDATE_TIMEOUT_SECONDS) result = None @@ -374,33 +284,18 @@ def update_managed_uv( print(f" ✓ Managed uv updated ({_uv_version(existing)})") elif result is not None: # Non-fatal — old uv still works fine. - logger.debug( - "uv self update failed (rc=%d): %s", result.returncode, result.stderr - ) - - # Keep this hook inside the long-standing API. During an update, main.py is - # already imported from the old checkout, then ``git pull`` replaces this - # module on disk before the updater imports it. Calling the repair here is - # what makes the migration happen on that first update. Runtime refresh is - # deliberately non-fatal: the live venv was not touched unless a fully - # prepared candidate reached cutover. + logger.debug("uv self update failed (rc=%d): %s", result.returncode, result.stderr) + # Keep this hook inside the long-standing API: during an update main.py is already imported + # from the old checkout and ``git pull`` replaces this module before the updater imports it, + # so calling the repair here is what migrates the runtime on that first update. Non-fatal: + # the live venv is untouched unless a fully prepared candidate reached cutover. _run_runtime_repair(existing, repair_observer, print_skip=True) return existing -# --------------------------------------------------------------------------- -# Managed Python runtime repair -# --------------------------------------------------------------------------- - - def _reload_hermes_constants(): - """Re-execute ``hermes_constants`` from disk and return the fresh module. - - cannot import name 'venv_python_path' from 'hermes_constants' (~/.hermes/hermes- - agent/hermes_constants.py) - """ + """Re-execute ``hermes_constants`` from disk (the imported one may predate venv_python_path).""" import hermes_constants - return importlib.reload(hermes_constants) @@ -421,184 +316,143 @@ def _remove_tree(path: Path, *, boundary: Path) -> None: shutil.rmtree(path, ignore_errors=True) +def _reject(path: Path, boundary: Path, msg: str, *args) -> None: + """Log a rejected candidate and clean up its tree; always returns ``None``.""" + logger.warning(msg, *args) + _remove_tree(path, boundary=boundary) + return None + + +def _token() -> str: + return f"{int(time.time())}-{os.getpid()}-{uuid.uuid4().hex[:8]}" + + +def _dotted(parts) -> str: + return ".".join(str(p) for p in parts) + + def _make_world_traversable(path: Path) -> None: """Keep root/FHS-managed runtimes executable by non-root callers.""" - try: + with contextlib.suppress(OSError): path.chmod(path.stat().st_mode | 0o755) - except OSError: - pass def _runtime_request(info: SQLiteRuntimeInfo) -> str: - """Pin the candidate to the current CPython minor line (e.g. ``3.11``). - - Requesting the exact patch can never repair some installs: for a given patch, python-build- - standalone may have no artifact with fixed SQLite at all (e.g. every published 3.11.14 build - links SQLite 3.50.4; the fix only exists from 3.11.15). - """ - return ".".join(str(part) for part in info.python_version[:2]) + """Pin the candidate to the current CPython minor line (e.g. ``3.11``): requesting the exact + patch can never repair installs whose patch has no fixed-SQLite artifact at all.""" + return _dotted(info.python_version[:2]) -# Cap on how many newer patches we'll try, newest-first, before giving up. -# Bounded because each attempt is a real download+install+probe+delete cycle; -# in practice the fix is almost always in the very next patch or two. +# Cap on newer patches tried, newest-first, before giving up: each attempt is a real +# download+install+probe+delete cycle, and the fix is almost always in the next patch or two. _MAX_PATCH_RETRIES = 5 def _list_available_patches( - uv_bin: str, minor: str, *, cwd: Path, env: dict -) -> list[tuple[int, int, int]]: - """Return known patch versions for ``minor`` (e.g. "3.11"), newest first. - - Returns [] on any failure (network, parse) -- callers fall back to the original bare-minor - request in that case, preserving prior behavior. - """ + uv_bin: str, minor: str, *, cwd: Path, env: dict) -> list[tuple[int, int, int]]: + """Known patch versions for ``minor`` (e.g. "3.11"), newest first; [] on any failure + (network, parse), in which case callers fall back to the bare-minor request.""" try: result = subprocess.run( [ uv_bin, "python", "list", minor, "--all-versions", "--only-downloads", - "--output-format", "json", "--no-config", - ], - cwd=cwd, - env=env, - capture_output=True, - text=True, - check=False, - timeout=15, - ) + "--output-format", "json", "--no-config"], + cwd=cwd, env=env, capture_output=True, text=True, check=False, timeout=15) if result.returncode != 0 or not result.stdout.strip(): return [] - entries = json.loads(result.stdout) versions: list[tuple[int, int, int]] = [] - for entry in entries: + for entry in json.loads(result.stdout): if not isinstance(entry, dict): continue - # Only default/cpython builds -- skip pypy/graalpy/freethreaded - # variants, which aren't what this repair path wants. - if entry.get("implementation") not in (None, "cpython"): - continue - if entry.get("variant") not in (None, "default"): + # Only default/cpython builds -- skip pypy/graalpy/freethreaded variants. + if entry.get("implementation") not in (None, "cpython") or ( + entry.get("variant") not in (None, "default")): continue parts = entry.get("version_parts") or {} try: versions.append( - (int(parts["major"]), int(parts["minor"]), int(parts["patch"])) - ) + (int(parts["major"]), int(parts["minor"]), int(parts["patch"]))) except (KeyError, TypeError, ValueError): continue - # Deduplicate (list --all-versions can repeat a version across - # platforms/arches if filtering above didn't fully narrow it) and - # sort newest-first. + # Deduplicate (a version can repeat across platforms/arches) and sort newest-first. return sorted(set(versions), reverse=True) except Exception: return [] def _attempt_install_generation( - uv_bin: str, - request: str, - *, - project_root: Path, - python_root: Path, - current: SQLiteRuntimeInfo, - allow_minor_upgrade: bool = False, - tried_versions: set[tuple[int, int, int]] | None = None, -) -> tuple[Path, Path, SQLiteRuntimeInfo] | None: - """One install+probe attempt for a specific version request (bare minor like "3.11", or an explicit - patch like "3.11.15"). Each attempt gets its own generation directory so a rejected candidate's - files are fully cleaned up before the next attempt, matching --reinstall semantics. Returns None - (and cleans up) on any failure, including a vulnerable or off-line candidate. + uv_bin: str, request: str, *, project_root: Path, python_root: Path, + current: SQLiteRuntimeInfo, allow_minor_upgrade: bool = False, + tried_versions: set[tuple[int, int, int]] | None = None) -> _Provisioned | None: + """One install+probe attempt for ``request`` (bare minor "3.11" or explicit patch "3.11.15"). + + Each attempt gets its own generation directory so a rejected candidate is fully cleaned up + before the next attempt (--reinstall semantics). Returns None (and cleans up) on any failure. """ - token = f"{int(time.time())}-{os.getpid()}-{uuid.uuid4().hex[:8]}" - generation = python_root / f"generation-{token}" + generation = python_root / f"generation-{_token()}" generation.mkdir(parents=True, exist_ok=False) _make_world_traversable(generation) - def reject(msg: str, *args) -> None: - logger.warning(msg, *args) - _remove_tree(generation, boundary=python_root) - return None - + reject = partial(_reject, generation, python_root) env = managed_python_env(project_root, install_dir=generation) run = dict(cwd=project_root, env=env, capture_output=True, text=True, check=False) install = subprocess.run( - [uv_bin, "python", "install", request, "--reinstall", "--no-bin", "--no-registry", "--no-config"], - **run, - ) + [uv_bin, "python", "install", request, "--reinstall", "--no-bin", "--no-registry", + "--no-config"], + **run) if install.returncode != 0: return reject( "private Python install failed for %s (rc=%d): %s", - request, install.returncode, (install.stderr or install.stdout or "").strip(), - ) - + request, install.returncode, (install.stderr or install.stdout or "").strip()) found = subprocess.run( - [uv_bin, "python", "find", request, "--managed-python", "--no-config"], **run - ) + [uv_bin, "python", "find", request, "--managed-python", "--no-config"], **run) if found.returncode != 0 or not found.stdout.strip(): return reject( "private Python lookup failed for %s (rc=%d): %s", - request, found.returncode, (found.stderr or "").strip(), - ) - + request, found.returncode, (found.stderr or "").strip()) python = Path(found.stdout.strip().splitlines()[-1]) try: python.resolve().relative_to(generation.resolve()) except (OSError, ValueError): return reject("uv resolved Python outside the Hermes generation: %s", python) - - # Do this before the candidate is probed or promoted. On macOS, the - # stable identifier prevents each immutable generation from looking like - # a new TCC principal. Failure is non-fatal: the SQLite repair must still - # proceed when codesign is unavailable or rejects a particular artifact. + # Sign before the candidate is probed or promoted so each immutable generation does not look + # like a new TCC principal on macOS. Non-fatal: the SQLite repair proceeds regardless. _macos_sign_managed_python(python) - candidate = probe_sqlite_runtime(python) if candidate is None: return reject("could not probe candidate Python runtime: %s", python) if tried_versions is not None: tried_versions.add(candidate.python_version[:3]) if allow_minor_upgrade: - # When falling forward to a higher minor line (e.g. 3.11 → 3.12), - # only reject downgrades — allow the minor to differ. + # Falling forward to a higher minor line: only reject downgrades. if candidate.python_version < current.python_version: return reject( "candidate Python downgraded from %s: %s", - ".".join(str(p) for p in current.python_version), candidate.python_version, - ) + _dotted(current.python_version), candidate.python_version) elif candidate.python_version[:2] != current.python_version[:2] or ( - candidate.python_version < current.python_version - ): + candidate.python_version < current.python_version): return reject( "candidate Python drifted off the %s minor line or downgraded: %s", - ".".join(str(p) for p in current.python_version[:2]), candidate.python_version, - ) + _dotted(current.python_version[:2]), candidate.python_version) if candidate.wal_reset_vulnerable: return reject( "candidate Python still links vulnerable SQLite %s (%s)", - candidate.sqlite_version_string, candidate.sqlite_source_id, - ) + candidate.sqlite_version_string, candidate.sqlite_source_id) return generation, python, candidate def _retry_explicit_patches( - uv_bin: str, - request: str, - *, - project_root: Path, - python_root: Path, - current: SQLiteRuntimeInfo, - tried: set[tuple[int, int, int]], - allow_minor_upgrade: bool = False, - skip_at_or_below: tuple[int, int, int] | None = None, -) -> tuple[Path, Path, SQLiteRuntimeInfo] | None: - """Retry ``request``'s minor line with explicit patch versions, newest-first, at most - ``_MAX_PATCH_RETRIES`` attempts, skipping versions already in ``tried`` (retrying one explicitly - would spend a full download+install+probe+delete cycle to reach a certain rejection). + uv_bin: str, request: str, *, project_root: Path, python_root: Path, + current: SQLiteRuntimeInfo, tried: set[tuple[int, int, int]], + allow_minor_upgrade: bool = False, skip_at_or_below: tuple[int, int, int] | None = None, +) -> _Provisioned | None: + """Retry ``request``'s minor line with explicit patches, newest-first, at most + ``_MAX_PATCH_RETRIES`` attempts, skipping versions already in ``tried`` (a certain rejection + still costs a full download+install+probe+delete cycle). ``skip_at_or_below`` also skips patches at or below that version: only NEWER patches can carry - the SQLite fix, and the downgrade guard rejects the rest anyway. This matters on a uv whose - download catalog is stale: in #71250 the newest indexed 3.11 was 3.11.14, exactly the installed - version, so without this skip the loop burned all five retries walking backwards. + the fix and the downgrade guard rejects the rest; on a stale uv catalog the newest indexed + patch can be the installed one, and the loop would burn every retry walking backwards. """ env_for_list = managed_python_env(project_root, install_dir=python_root) patches = _list_available_patches(uv_bin, request, cwd=project_root, env=env_for_list) @@ -611,25 +465,34 @@ def _retry_explicit_patches( if skip_at_or_below is not None and version_tuple <= skip_at_or_below: continue tried.add(version_tuple) - explicit = ".".join(str(p) for p in version_tuple) + explicit = _dotted(version_tuple) print(f" → Retrying with explicit patch {explicit}...") attempts += 1 result = _attempt_install_generation( uv_bin, explicit, project_root=project_root, python_root=python_root, current=current, - allow_minor_upgrade=allow_minor_upgrade, - ) + allow_minor_upgrade=allow_minor_upgrade) if result is not None: return result return None +def _provision_line( + uv_bin: str, request: str, *, tried: set[tuple[int, int, int]], + allow_minor_upgrade: bool = False, skip_at_or_below: tuple[int, int, int] | None = None, + **common) -> _Provisioned | None: + """Try ``request`` once, then its explicit newer patches; None when the whole line fails.""" + result = _attempt_install_generation( + uv_bin, request, tried_versions=tried, allow_minor_upgrade=allow_minor_upgrade, **common) + if result is None: + result = _retry_explicit_patches( + uv_bin, request, tried=tried, allow_minor_upgrade=allow_minor_upgrade, + skip_at_or_below=skip_at_or_below, **common) + return result + + def _install_safe_python_generation( - uv_bin: str, - *, - project_root: Path, - current: SQLiteRuntimeInfo, -) -> tuple[Path, Path, SQLiteRuntimeInfo] | None: + uv_bin: str, *, project_root: Path, current: SQLiteRuntimeInfo) -> _Provisioned | None: runtime_root = project_root / _RUNTIME_DIR_NAME python_root = managed_python_install_dir(project_root) _make_world_traversable(runtime_root) @@ -639,46 +502,26 @@ def _install_safe_python_generation( request = _runtime_request(current) print(f" → Provisioning a private Python {request} runtime with fixed SQLite...") tried_versions = {current.python_version[:3]} - result = _attempt_install_generation(uv_bin, request, tried_versions=tried_versions, **common) - if result is not None: - return result - - # The bare minor-line request resolved to a still-vulnerable (or - # otherwise rejected) candidate. Rather than giving up immediately, - # query which patches on this minor line uv actually knows about and - # retry with explicit newer versions, newest-first -- this handles the - # case where the default resolution for a bare request picks an older - # cached/indexed patch even though a newer, non-vulnerable one is - # available (issue #71250). - result = _retry_explicit_patches( - uv_bin, request, tried=tried_versions, - skip_at_or_below=current.python_version[:3], **common, + # If the bare minor-line request resolves to a still-vulnerable (or otherwise rejected) + # candidate, the default resolution may have picked an older cached/indexed patch even though + # a newer, non-vulnerable one exists: retry with explicit newer patches, newest-first. + result = _provision_line( + uv_bin, request, tried=tried_versions, skip_at_or_below=current.python_version[:3], **common ) if result is not None: return result - - # All patches on the current minor line are vulnerable or rejected. - # Fall forward to the next supported minor (e.g. 3.11 → 3.12) so the - # user isn't stuck on every `hermes update` with no path to a fixed - # runtime (issue #76106). The requires-python constraint - # (>=3.11,<3.14) and the downstream import smoke-test gate - # compatibility; we only need to stay inside that window. + # All patches on the current minor line are vulnerable or rejected. Fall forward to the next + # supported minor (e.g. 3.11 → 3.12) so the user isn't stuck on every `hermes update`. The + # requires-python window (>=3.11,<3.14) and the import smoke-test gate compatibility. cur_major, cur_minor = current.python_version[:2] fb_tried: set[tuple[int, int, int]] = set(tried_versions) for next_minor in range(cur_minor + 1, 14): # up to 3.13 next_request = f"{cur_major}.{next_minor}" print( f" → No fixed {cur_major}.{cur_minor} build available; " - f"trying {next_request} as fallback..." - ) - result = _attempt_install_generation( - uv_bin, next_request, allow_minor_upgrade=True, tried_versions=fb_tried, **common - ) - if result is not None: - return result - result = _retry_explicit_patches( - uv_bin, next_request, tried=fb_tried, allow_minor_upgrade=True, **common - ) + f"trying {next_request} as fallback...") + result = _provision_line( + uv_bin, next_request, tried=fb_tried, allow_minor_upgrade=True, **common) if result is not None: return result return None @@ -691,91 +534,53 @@ def _smoke_candidate_venv(venv_dir: Path) -> tuple[bool, str, SQLiteRuntimeInfo if info is None: return False, f"could not execute {python}", None if info.wal_reset_vulnerable: - return ( - False, - f"candidate still links vulnerable SQLite {info.sqlite_version_string}", - info, - ) - + return False, f"candidate still links vulnerable SQLite {info.sqlite_version_string}", info check = ( "import dotenv, fastapi, openai, prompt_toolkit, pydantic, rich, uvicorn, yaml\n" - "import hermes_state\n" - ) + "import hermes_state\n") try: result = subprocess.run( - [str(python), "-I", "-c", check], - cwd=venv_dir.parent, - env=isolated_interpreter_env(), - capture_output=True, - text=True, - timeout=90, - check=False, - ) + [str(python), "-I", "-c", check], cwd=venv_dir.parent, env=isolated_interpreter_env(), + capture_output=True, text=True, timeout=90, check=False) except (OSError, subprocess.TimeoutExpired) as exc: return False, str(exc), info if result.returncode != 0: detail = (result.stderr or result.stdout or "core import smoke failed").strip() - last_line = detail.splitlines()[-1] if detail else "core import smoke failed" - return False, last_line, info + return False, detail.splitlines()[-1] if detail else "core import smoke failed", info return True, "", info def _stage_candidate_venv( - uv_bin: str, - *, - project_root: Path, - generation: Path, - python: Path, -) -> Path | None: + uv_bin: str, *, project_root: Path, generation: Path, python: Path) -> Path | None: runtime_root = project_root / _RUNTIME_DIR_NAME - token = f"{int(time.time())}-{os.getpid()}-{uuid.uuid4().hex[:8]}" - candidate = runtime_root / f"venv-candidate-{token}" + candidate = runtime_root / f"venv-candidate-{_token()}" env = managed_python_env(project_root, install_dir=generation) env.update({ - "UV_PROJECT_ENVIRONMENT": str(candidate), - "UV_PYTHON": str(python), - "UV_PYTHON_DOWNLOADS": "never", - "VIRTUAL_ENV": str(candidate), - }) - - def reject(msg: str, *args) -> None: - logger.warning(msg, *args) - _remove_tree(candidate, boundary=runtime_root) - return None + "UV_PROJECT_ENVIRONMENT": str(candidate), "UV_PYTHON": str(python), + "UV_PYTHON_DOWNLOADS": "never", "VIRTUAL_ENV": str(candidate)}) + reject = partial(_reject, candidate, runtime_root) print(" → Building a relocatable replacement environment...") created = subprocess.run( [ uv_bin, "venv", str(candidate), "--python", str(python), - "--managed-python", "--no-python-downloads", "--relocatable", "--no-config", - ], - cwd=project_root, - env=env, - capture_output=True, - text=True, - check=False, - ) + "--managed-python", "--no-python-downloads", "--relocatable", "--no-config"], + cwd=project_root, env=env, capture_output=True, text=True, check=False) if created.returncode != 0: return reject( "candidate venv creation failed (rc=%d): %s", - created.returncode, (created.stderr or created.stdout or "").strip(), - ) - + created.returncode, (created.stderr or created.stdout or "").strip()) if not (project_root / "uv.lock").is_file(): return reject("candidate dependency sync refused: uv.lock is missing") - # Locked sync must see project [tool.uv] exclude-newer; --no-config / - # UV_NO_CONFIG drops it and uv 0.12+ refuses --locked. + # Locked sync must see project [tool.uv] exclude-newer; --no-config / UV_NO_CONFIG drops it + # and uv 0.12+ refuses --locked. sync_env = dict(env) sync_env.pop("UV_NO_CONFIG", None) synced = subprocess.run( [uv_bin, "sync", "--extra", "all", "--locked", "--python", str(_venv_python(candidate))], - cwd=project_root, - env=sync_env, - check=False, - ) + cwd=project_root, env=sync_env, check=False) if synced.returncode != 0: return reject("candidate dependency sync failed (rc=%d)", synced.returncode) - healthy, detail, _ = _smoke_candidate_venv(candidate) if not healthy: return reject("candidate venv smoke failed: %s", detail) @@ -795,23 +600,18 @@ def _rename_with_retry(source: Path, destination: Path) -> None: def _cut_over_candidate( - candidate: Path, - *, - project_root: Path, - live: Path | None = None, + candidate: Path, *, project_root: Path, live: Path | None = None ) -> tuple[bool, Path | None, SQLiteRuntimeInfo | None, str]: live = live if live is not None else project_root / _VENV_NAME runtime_root = project_root / _RUNTIME_DIR_NAME - token = f"{int(time.time())}-{os.getpid()}-{uuid.uuid4().hex[:8]}" + token = _token() backup = live.with_name(f"{live.name}.stale.runtime-{token}") rejected = runtime_root / f"venv-rejected-{token}" - try: try: _rename_with_retry(live, backup) except OSError as exc: return False, None, None, f"could not park the existing venv: {exc}" - try: _rename_with_retry(candidate, live) except OSError as promote_error: @@ -820,25 +620,21 @@ def _cut_over_candidate( except OSError as rollback_error: return False, backup, None, ( "could not promote the replacement venv " - f"({promote_error}); rollback failed ({rollback_error})" - ) + f"({promote_error}); rollback failed ({rollback_error})") return False, None, None, f"could not promote the replacement venv: {promote_error}" - try: healthy, detail, info = _smoke_candidate_venv(live) except Exception as exc: healthy, detail, info = False, f"candidate smoke raised: {exc}", None if healthy: return True, backup, info, "" - try: _rename_with_retry(live, rejected) _rename_with_retry(backup, live) except OSError as exc: return False, backup, info, ( "post-cutover smoke failed " - f"({detail}); rollback failed ({exc}); rejected venv: {rejected}" - ) + f"({detail}); rollback failed ({exc}); rejected venv: {rejected}") _remove_tree(rejected, boundary=runtime_root) return False, None, info, f"post-cutover smoke failed: {detail}" except BaseException: @@ -848,10 +644,7 @@ def _cut_over_candidate( except OSError as exc: logger.error( "interrupted runtime cutover could not restore %s from %s: %s", - live, - backup, - exc, - ) + live, backup, exc) raise @@ -864,7 +657,6 @@ def _acquire_repair_lock(runtime_root: Path) -> _RepairLock | None: fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600) except OSError: return None - try: _flock(fd, acquire=True) except (ImportError, OSError): @@ -877,27 +669,22 @@ def _flock(fd: int, *, acquire: bool) -> None: """Non-blocking exclusive lock (or unlock) on *fd*, portable across msvcrt/fcntl.""" if os.name == "nt": import msvcrt - if acquire and os.fstat(fd).st_size == 0: os.write(fd, b"\0") os.lseek(fd, 0, os.SEEK_SET) msvcrt.locking(fd, msvcrt.LK_NBLCK if acquire else msvcrt.LK_UNLCK, 1) else: import fcntl - fcntl.flock(fd, (fcntl.LOCK_EX | fcntl.LOCK_NB) if acquire else fcntl.LOCK_UN) def _release_repair_lock(lock: _RepairLock) -> None: try: - _flock(lock.fd, acquire=False) - except (ImportError, OSError): - pass + with contextlib.suppress(ImportError, OSError): + _flock(lock.fd, acquire=False) finally: - try: + with contextlib.suppress(OSError): os.close(lock.fd) - except OSError: - pass def _windows_runtime_holders() -> tuple[bool, str]: @@ -918,12 +705,11 @@ def _windows_runtime_holders() -> tuple[bool, str]: def _windows_runtime_self_lock(live: Path) -> tuple[bool, str]: - """Detect the one holder the generic scan above is blind to: THIS process. + """Detect the one holder the generic scan is blind to: THIS process. - ``_detect_venv_python_processes`` excludes the calling process and its ancestors on purpose — a - CLI ``hermes update`` itself runs from the venv python — which is correct for the dependency- - sync path, where only a *loaded* ``.pyd`` image blocks the rewrite and a fresh child process - dodges it. + ``_detect_venv_python_processes`` excludes the calling process and its ancestors on purpose + (``hermes update`` itself runs from the venv python), which is correct for the dependency-sync + path where only a *loaded* ``.pyd`` image blocks the rewrite and a fresh child dodges it. """ if platform.system() != "Windows": return False, "" @@ -942,19 +728,14 @@ def _windows_runtime_self_lock(live: Path) -> tuple[bool, str]: resolved = str(path_value).lower() return resolved.startswith(live_res) + why = "Windows cannot rename a directory while a process executes from inside it" exe = sys.executable if _under_live(exe): - return True, ( - f"the updater itself runs from the live venv it must replace " - f"({exe}); Windows cannot rename a directory while a process " - "executes from inside it" - ) - # Belt-and-braces: the venv\Scripts\hermes.exe launcher stays mapped - # while it waits for this child, so an ancestor started from the venv - # blocks the rename too. - try: + return True, f"the updater itself runs from the live venv it must replace ({exe}); {why}" + # Belt-and-braces: the venv\Scripts\hermes.exe launcher stays mapped while it waits for this + # child, so an ancestor started from the venv blocks the rename too. + with contextlib.suppress(Exception): import psutil - for anc in psutil.Process().parents(): try: anc_exe = anc.exe() @@ -962,12 +743,7 @@ def _windows_runtime_self_lock(live: Path) -> tuple[bool, str]: continue if _under_live(anc_exe): return True, ( - f"ancestor process PID {anc.pid} runs from the live venv " - f"({anc_exe}); Windows cannot rename a directory while a " - "process executes from inside it" - ) - except Exception: - pass + f"ancestor process PID {anc.pid} runs from the live venv ({anc_exe}); {why}") return False, "" @@ -977,26 +753,20 @@ def _uv_version_string(uv_bin: str) -> str: result = subprocess.run( [uv_bin, "--version"], capture_output=True, text=True, encoding="utf-8", errors="replace", - check=False, timeout=15, - ) + check=False, timeout=15) except Exception: return "" return (result.stdout or "").strip() if result.returncode == 0 else "" def _refresh_managed_uv_catalog(uv_bin: str) -> bool: - """Re-bootstrap the managed uv binary to refresh its Python catalog. - - Re-running the official installer is the only supported refresh path for unmanaged installs. - Only the Hermes-managed binary is ever refreshed; a caller-supplied foreign uv path is left - alone. - """ + """Re-bootstrap the managed uv binary to refresh its Python catalog (the only supported + refresh path for unmanaged installs). A caller-supplied foreign uv path is left alone.""" managed = managed_uv_path() try: - is_managed = Path(uv_bin).resolve() == managed.resolve() + if Path(uv_bin).resolve() != managed.resolve(): + return False except OSError: - is_managed = False - if not is_managed: return False before = _uv_version_string(uv_bin) try: @@ -1009,33 +779,21 @@ def _refresh_managed_uv_catalog(uv_bin: str) -> bool: def _default_live_venv(root: Path) -> Path: - """Return the venv that runtime repair should target for *root*. - - ``venv`` wins when it holds an interpreter (managed layout takes precedence); otherwise fall - back to ``.venv`` when that one does. When neither has an interpreter, return the ``venv`` path - so the caller's existing ``not-applicable`` handling fires unchanged. + """Venv that runtime repair should target for *root*: ``venv`` when it holds an interpreter + (managed layout wins), else ``.venv`` when that does, else ``venv`` so ``not-applicable`` fires. """ primary, fallback = root / _VENV_NAME, root / _ALT_VENV_NAME - if not _venv_python(primary).is_file() and _venv_python(fallback).is_file(): - return fallback - return primary + use_fallback = not _venv_python(primary).is_file() and _venv_python(fallback).is_file() + return fallback if use_fallback else primary def _sweep_stale_runtime_backups( - live: Path, - *, - root: Path, - keep: Path | None = None, - min_age_seconds: float = 3600.0, -) -> None: - """Remove leftover ``venv.stale.runtime-*`` backups next to *live*. + live: Path, *, root: Path, keep: Path | None = None, min_age_seconds: float = 3600.0) -> None: + """Remove leftover ``venv.stale.runtime-*`` backups next to *live*. Best-effort: never raises. - On POSIX, deleting the tree is safe even while an older process still maps files from it — open - FDs and mmaps keep their inodes alive; the directory entry is what goes away. - - ``min_age_seconds`` guards against racing a concurrent repair in another process: a backup - parked seconds ago may still be that repair's rollback path, so only clearly-old markers are - swept. ``keep`` exempts the backup the current repair just created. Best-effort: never raises. + On POSIX this is safe while an older process still maps files from the tree (open FDs/mmaps + keep their inodes). ``min_age_seconds`` avoids racing a concurrent repair whose fresh backup + may still be its rollback path; ``keep`` exempts the backup this repair just created. """ try: candidates = list(live.parent.glob(f"{live.name}.stale.runtime-*")) @@ -1046,19 +804,100 @@ def _sweep_stale_runtime_backups( if keep is not None and candidate == keep: continue try: - age = now - candidate.stat().st_mtime + if now - candidate.stat().st_mtime < min_age_seconds: + continue except OSError: continue - if age < min_age_seconds: - continue _remove_tree(candidate, boundary=root) +def _result( + status: str, current: SQLiteRuntimeInfo, detail: str = "", **extra) -> RuntimeRepairResult: + return RuntimeRepairResult(status, detail, sqlite_before=current.sqlite_version_string, **extra) + + +def _repair_windows_preflight( + root: Path, live: Path, current: SQLiteRuntimeInfo) -> RuntimeRepairResult | None: + """Defer the repair when Windows holders make the venv rename impossible; else ``None``.""" + blocked, detail = _windows_runtime_holders() + if blocked: + print(f" ⚠ SQLite runtime repair deferred: {detail}") + return _result("skipped", current, detail) + self_locked, self_detail = _windows_runtime_self_lock(live) + if self_locked: + # Structural, not transient: this process maps the live venv's own executable, so the + # park rename fails identically on every run. Defer BEFORE provisioning — a candidate + # staged for a cutover that can never run only leaks an incomplete generation. + for line in ( + f" ⚠ SQLite runtime repair deferred: {self_detail}.", + " Retrying `hermes update` from inside this venv cannot help: " + "the mapped executable is released only when this process exits.", + " To complete the repair, run the updater from an interpreter " + "that lives outside this venv, e.g.:", + f" cd {root}", + " -m hermes_cli.main update", + " Sessions stay protected meanwhile: Hermes keeps databases " + "out of WAL mode on this SQLite build."): + print(line) + return _result("skipped", current, self_detail) + return None + + +def _repair_under_lock( + uv_bin: str, *, root: Path, live: Path, live_python: Path, runtime_root: Path +) -> RuntimeRepairResult: + """Provision, stage and cut over a fixed runtime; caller holds the repair lock.""" + # Re-probe under the install-scoped lock: another updater may have completed the repair + # while this process was entering the path. + current = probe_sqlite_runtime(live_python) + if current is None: + return RuntimeRepairResult("skipped", "live interpreter probe failed") + if not current.wal_reset_vulnerable: + return _result("safe", current, sqlite_after=current.sqlite_version_string) + print( + " ⚠ Hermes venv links SQLite " + f"{current.sqlite_version_string}, which has the WAL-reset bug.") + provisioned = _install_safe_python_generation(uv_bin, project_root=root, current=current) + # Likely a stale managed-uv catalog: python-build-standalone re-releases the same patch + # versions with fixed SQLite, but a frozen catalog keeps resolving the old vulnerable build + # and the patch-retry loop has no newer number to try. Refresh the binary and retry once. + if provisioned is None and _refresh_managed_uv_catalog(uv_bin): + print(" → Managed uv refreshed; retrying provisioning...") + provisioned = _install_safe_python_generation(uv_bin, project_root=root, current=current) + if provisioned is None: + return _result("failed", current, "could not provision a fixed private Python runtime") + generation, python, candidate_info = provisioned + + candidate = _stage_candidate_venv( + uv_bin, project_root=root, generation=generation, python=python) + if candidate is None: + _remove_tree(generation, boundary=managed_python_install_dir(root)) + return _result( + "failed", current, + "replacement environment did not pass dependency and import smoke tests", + sqlite_after=candidate_info.sqlite_version_string) + + cut_over, backup, final_info, cutover_detail = _cut_over_candidate( + candidate, project_root=root, live=live) + if not cut_over: + if backup is None: + _remove_tree(candidate, boundary=runtime_root) + _remove_tree(generation, boundary=managed_python_install_dir(root)) + return _result( + "failed", current, cutover_detail, + sqlite_after=final_info.sqlite_version_string if final_info is not None else "", + backup_venv=backup) + final_version = (final_info if final_info is not None else candidate_info).sqlite_version_string + print( + " ✓ Managed Python runtime repaired " + f"(SQLite {current.sqlite_version_string} → {final_version})") + if backup is not None and backup.exists(): + _remove_tree(backup, boundary=root) + return _result("repaired", current, sqlite_after=final_version, backup_venv=backup) + + def repair_vulnerable_runtime( - uv_bin: str, - *, - project_root: Path | None = None, - venv_dir: Path | None = None, + uv_bin: str, *, project_root: Path | None = None, venv_dir: Path | None = None ) -> RuntimeRepairResult: """Replace a vulnerable install venv without mutating it in place. @@ -1070,220 +909,62 @@ def repair_vulnerable_runtime( live_python = _venv_python(live) if not (root / "pyproject.toml").is_file() or not live_python.is_file(): return RuntimeRepairResult("not-applicable") - current = probe_sqlite_runtime(live_python) if current is None: - return RuntimeRepairResult( - "skipped", - f"could not probe live interpreter {live_python}", - ) + return RuntimeRepairResult("skipped", f"could not probe live interpreter {live_python}") if not current.wal_reset_vulnerable: - # The runtime is already fixed — any venv.stale.runtime-* markers - # next to the live venv are leftovers from a past repair (or from - # a build predating the post-repair cleanup) and will never be - # rolled back to. Sweep them so they don't leak ~1 GB each - # forever (issue #73109). Age-gated to avoid racing an in-flight - # repair in a sibling process. + # Already fixed: any venv.stale.runtime-* markers next to the live venv are leftovers + # from a past repair and will never be rolled back to. Sweep them so they don't leak + # ~1 GB each forever. Age-gated to avoid racing an in-flight repair in a sibling process. _sweep_stale_runtime_backups(live, root=root) - return RuntimeRepairResult( - "safe", - sqlite_before=current.sqlite_version_string, - sqlite_after=current.sqlite_version_string, - ) - - def deferred(detail: str) -> RuntimeRepairResult: - return RuntimeRepairResult( - "skipped", detail, sqlite_before=current.sqlite_version_string - ) - - blocked, detail = _windows_runtime_holders() - if blocked: - print(f" ⚠ SQLite runtime repair deferred: {detail}") - return deferred(detail) - - self_locked, self_detail = _windows_runtime_self_lock(live) - if self_locked: - # Structural, not transient: this process maps the live venv's own - # executable, so the park rename fails the same way on every run and - # no number of retries converges. Defer BEFORE provisioning — a - # candidate staged for a cutover that can never run only leaks an - # incomplete generation (#93032). - print(f" ⚠ SQLite runtime repair deferred: {self_detail}.") - print( - " Retrying `hermes update` from inside this venv cannot help: " - "the mapped executable is released only when this process exits." - ) - print( - " To complete the repair, run the updater from an interpreter " - "that lives outside this venv, e.g.:" - ) - print(f" cd {root}") - print(" -m hermes_cli.main update") - print( - " Sessions stay protected meanwhile: Hermes keeps databases " - "out of WAL mode on this SQLite build." - ) - return deferred(self_detail) - + return _result("safe", current, sqlite_after=current.sqlite_version_string) + deferred = _repair_windows_preflight(root, live, current) + if deferred is not None: + return deferred runtime_root = root / _RUNTIME_DIR_NAME lock = _acquire_repair_lock(runtime_root) if lock is None: detail = "another runtime repair is already in progress" print(f" ⚠ SQLite runtime repair deferred: {detail}") - return deferred(detail) - + return _result("skipped", current, detail) try: - # Re-probe under the install-scoped lock: another updater may have - # completed the repair while this process was entering the path. - current = probe_sqlite_runtime(live_python) - if current is None: - return RuntimeRepairResult("skipped", "live interpreter probe failed") - if not current.wal_reset_vulnerable: - return RuntimeRepairResult( - "safe", - sqlite_before=current.sqlite_version_string, - sqlite_after=current.sqlite_version_string, - ) - - print( - " ⚠ Hermes venv links SQLite " - f"{current.sqlite_version_string}, which has the WAL-reset bug." - ) - provisioned = _install_safe_python_generation( - uv_bin, - project_root=root, - current=current, - ) - # Likely a stale managed-uv catalog: python-build-standalone - # re-releases the same patch versions with fixed SQLite, but a - # frozen catalog keeps resolving the old vulnerable build and the - # patch-retry loop has no newer number to try (issue #72093). - # Refresh the managed binary and retry once. - if provisioned is None and _refresh_managed_uv_catalog(uv_bin): - print(" → Managed uv refreshed; retrying provisioning...") - provisioned = _install_safe_python_generation( - uv_bin, - project_root=root, - current=current, - ) - if provisioned is None: - return RuntimeRepairResult( - "failed", - "could not provision a fixed private Python runtime", - sqlite_before=current.sqlite_version_string, - ) - generation, python, candidate_info = provisioned - - candidate = _stage_candidate_venv( - uv_bin, - project_root=root, - generation=generation, - python=python, - ) - if candidate is None: - _remove_tree(generation, boundary=managed_python_install_dir(root)) - return RuntimeRepairResult( - "failed", - "replacement environment did not pass dependency and import smoke tests", - sqlite_before=current.sqlite_version_string, - sqlite_after=candidate_info.sqlite_version_string, - ) - - cut_over, backup, final_info, cutover_detail = _cut_over_candidate( - candidate, - project_root=root, - live=live, - ) - if not cut_over: - if backup is None: - _remove_tree(candidate, boundary=runtime_root) - _remove_tree(generation, boundary=managed_python_install_dir(root)) - return RuntimeRepairResult( - "failed", - cutover_detail, - sqlite_before=current.sqlite_version_string, - sqlite_after=final_info.sqlite_version_string if final_info is not None else "", - backup_venv=backup, - ) - - final_version = ( - final_info.sqlite_version_string - if final_info is not None - else candidate_info.sqlite_version_string - ) - print( - " ✓ Managed Python runtime repaired " - f"(SQLite {current.sqlite_version_string} → {final_version})" - ) - if backup is not None and backup.exists(): - _remove_tree(backup, boundary=root) - return RuntimeRepairResult( - "repaired", - sqlite_before=current.sqlite_version_string, - sqlite_after=final_version, - backup_venv=backup, - ) + return _repair_under_lock( + uv_bin, root=root, live=live, live_python=live_python, runtime_root=runtime_root) finally: _release_repair_lock(lock) -# --------------------------------------------------------------------------- -# Installer internals -# --------------------------------------------------------------------------- - - def _install_uv(target: Path) -> None: """Bootstrap uv into *target* using the official standalone installer. Sets ``UV_UNMANAGED_INSTALL`` (POSIX) / ``UV_INSTALL_DIR`` (Windows) so the installer writes into ``$HERMES_HOME/bin/`` instead of ``~/.local/bin/``. """ - system = platform.system() - env = { - **os.environ, - # Tell the astral installer to drop the binary in our dir, not - # ~/.local/bin. UV_UNMANAGED_INSTALL is the POSIX env var; Windows - # uses UV_INSTALL_DIR. - "UV_UNMANAGED_INSTALL": str(target.parent), - "UV_INSTALL_DIR": str(target.parent), - } - - (_install_uv_windows if system == "Windows" else _install_uv_posix)(env) + env = {**os.environ, "UV_UNMANAGED_INSTALL": str(target.parent), + "UV_INSTALL_DIR": str(target.parent)} + (_install_uv_windows if platform.system() == "Windows" else _install_uv_posix)(env) def _install_uv_posix(env: dict[str, str]) -> None: """Download + sh the POSIX installer (two-stage to avoid curl|sh pitfalls).""" with tempfile.NamedTemporaryFile(suffix=".sh", delete=False) as f: installer_path = f.name - try: subprocess.run( ["curl", "-LsSf", "https://astral.sh/uv/install.sh", "-o", installer_path], - check=True, - capture_output=True, - ) - subprocess.run( - ["sh", installer_path], - env=env, - check=True, - capture_output=True, - ) + check=True, capture_output=True) + subprocess.run(["sh", installer_path], env=env, check=True, capture_output=True) finally: - try: + with contextlib.suppress(OSError): os.unlink(installer_path) - except OSError: - pass def _install_uv_windows(env: dict[str, str]) -> None: """Invoke the PowerShell installer.""" cmd = "irm https://astral.sh/uv/install.ps1 | iex" subprocess.run( - ["powershell", "-ExecutionPolicy", "Bypass", "-c", cmd], - env=env, - check=True, - capture_output=True, - ) + ["powershell", "-ExecutionPolicy", "Bypass", "-c", cmd], env=env, check=True, + capture_output=True) def rebuild_venv(uv_bin: str, venv_dir: Path, python_version: str = "3.11") -> bool: diff --git a/hermes_cli/mem_trim.py b/hermes_cli/mem_trim.py index 761e5e9b97..9ed5856193 100644 --- a/hermes_cli/mem_trim.py +++ b/hermes_cli/mem_trim.py @@ -23,6 +23,11 @@ logger = logging.getLogger(__name__) _DEFAULT_COOLDOWN_SECONDS = 60.0 _DEFAULT_LOG_EVERY_N = 1 _DEFAULT_INFO_LOG_MIN_DELTA_MB = 0.0 +# Even forced trims honor a short floor: AIAgent.close() forces a trim, and delegate +# batches close N child subagents back-to-back in the SAME process — without a floor +# that stacks N+1 uncooled full gc.collect() passes (50-500ms each in a large gateway +# process). 5s coalesces the burst while keeping the parent's final close-trim effective. +_FORCE_FLOOR_SECONDS = 5.0 _trim_lock = threading.Lock() _last_trim_monotonic = 0.0 _probe_done = False @@ -31,17 +36,13 @@ _trim_call_count = 0 def _config_settings() -> tuple[bool, float, int, float]: - """Return fail-open settings from the normal Hermes config path.""" - enabled = True + """Return fail-open ``(enabled, cooldown, log_every_n, info_log_min_delta_mb)`` from config.""" settings: Any = None try: - # Read-only access: settings are only .get()ed and coerced, never - # mutated — use the no-deepcopy variant. This runs on EVERY trim - # attempt (before the cooldown check), and generating a full-config - # deepcopy per attempt is exactly the allocator garbage this module - # exists to release. + # Read-only, no-deepcopy variant: this runs on EVERY trim attempt (before the + # cooldown check), and a full-config deepcopy per attempt is exactly the + # allocator garbage this module exists to release. from hermes_cli.config import load_config_readonly - config = load_config_readonly() or {} context = config.get("context") if isinstance(config, dict) else None settings = context.get("memory_trim") if isinstance(context, dict) else None @@ -49,14 +50,12 @@ def _config_settings() -> tuple[bool, float, int, float]: pass if not isinstance(settings, dict): settings = {} - if isinstance(settings.get("enabled"), bool): - enabled = settings["enabled"] + enabled = settings["enabled"] if isinstance(settings.get("enabled"), bool) else True return ( enabled, _cooldown_seconds(settings.get("cooldown_seconds")), _coerce(settings.get("log_every_n"), _DEFAULT_LOG_EVERY_N, int, 1), - _coerce(settings.get("info_log_min_delta_mb"), _DEFAULT_INFO_LOG_MIN_DELTA_MB, float, 0.0), - ) + _coerce(settings.get("info_log_min_delta_mb"), _DEFAULT_INFO_LOG_MIN_DELTA_MB, float, 0.0)) def _coerce(value: Any, default, cast, floor): @@ -84,16 +83,12 @@ def _read_proc_status() -> str | None: def collect_memory_snapshot(history_bytes: int | None = None) -> dict[str, int | None]: - """Return lightweight process-memory telemetry for trim logs and canaries. + """Lightweight process-memory telemetry for trim logs and canaries. - ``VmRSS`` and ``RssAnon`` are Linux-only best effort fields. The helper is intentionally - dependency-free so allocation recovery never requires psutil. + ``VmRSS`` / ``RssAnon`` are Linux-only best effort; deliberately psutil-free. """ snapshot: dict[str, int | None] = { - "rss_kib": None, - "rss_anon_kib": None, - "thread_count": threading.active_count(), - } + "rss_kib": None, "rss_anon_kib": None, "thread_count": threading.active_count()} status = _read_proc_status() if status: for line in status.splitlines(): @@ -110,10 +105,9 @@ def collect_memory_snapshot(history_bytes: int | None = None) -> dict[str, int | def _should_log_trim( *, force: bool, log_every_n: int, call_count: int, before: dict[str, int | None], - after: dict[str, int | None], info_log_min_delta_mb: float, -) -> bool: - # trim_memory calls this only after malloc_trim reported success. A forced - # successful trim is an explicit observability event, regardless of RSS. + after: dict[str, int | None], info_log_min_delta_mb: float) -> bool: + # Called only after malloc_trim reported success; a forced successful trim is an + # explicit observability event regardless of RSS. if force: return True if call_count % log_every_n: @@ -136,8 +130,7 @@ def _probe_glibc_malloc_trim() -> Callable[[int], int] | None: try: if platform.libc_ver()[0].lower() != "glibc": return None - libc = ctypes.CDLL(None) - trim = libc.malloc_trim + trim = ctypes.CDLL(None).malloc_trim trim.argtypes = [ctypes.c_size_t] trim.restype = ctypes.c_int _malloc_trim = trim @@ -147,11 +140,7 @@ def _probe_glibc_malloc_trim() -> Callable[[int], int] | None: def trim_memory( - *, - force: bool = False, - reason: str = "", - cooldown_seconds: float | None = None, -) -> bool: + *, force: bool = False, reason: str = "", cooldown_seconds: float | None = None) -> bool: """Collect cycles and ask glibc to release free heap pages. Returns ``True`` only when ``malloc_trim(0)`` ran and reported success. Unsupported allocators, @@ -169,15 +158,8 @@ def trim_memory( return False now = time.monotonic() cooldown = configured_cooldown if cooldown_seconds is None else _cooldown_seconds(cooldown_seconds) - if not force and _last_trim_monotonic and now - _last_trim_monotonic < cooldown: - return False - # Even forced trims honor a short floor: AIAgent.close() forces a trim, - # and delegate batches close N child subagents back-to-back in the SAME - # process — without a floor that stacks N+1 uncooled full gc.collect() - # passes (50-500ms each in a large gateway process). 5s coalesces the - # burst while keeping the parent's final close-trim effective. - _FORCE_FLOOR_SECONDS = 5.0 - if force and _last_trim_monotonic and now - _last_trim_monotonic < _FORCE_FLOOR_SECONDS: + since_last = now - _last_trim_monotonic + if _last_trim_monotonic and since_last < (_FORCE_FLOOR_SECONDS if force else cooldown): return False # Record the attempt before calling into libc so repeated failures do not # turn every turn boundary into an expensive full collection. @@ -193,19 +175,16 @@ def trim_memory( _trim_call_count += 1 if released and _should_log_trim( force=force, log_every_n=log_every_n, call_count=_trim_call_count, - before=before, after=after, info_log_min_delta_mb=info_log_min_delta_mb, - ): + before=before, after=after, info_log_min_delta_mb=info_log_min_delta_mb): logger.info( "memory trim: reason=%s malloc_trim=%s rss_kib=%s->%s " "rss_anon_kib=%s->%s threads=%s duration_ms=%.1f", reason or "cleanup", trim_result, before.get("rss_kib"), after.get("rss_kib"), before.get("rss_anon_kib"), after.get("rss_anon_kib"), - after.get("thread_count"), duration_ms, - ) + after.get("thread_count"), duration_ms) return released except Exception as exc: logger.warning( - "memory trim failed after %s: %s: %s", reason or "cleanup", type(exc).__name__, exc, - ) + "memory trim failed after %s: %s: %s", reason or "cleanup", type(exc).__name__, exc) return False diff --git a/hermes_cli/migrate.py b/hermes_cli/migrate.py index 81afa057c5..11f96a7b8c 100644 --- a/hermes_cli/migrate.py +++ b/hermes_cli/migrate.py @@ -15,35 +15,29 @@ from hermes_cli.config import load_config def cmd_migrate(args: Any) -> int: """Dispatcher for ``hermes migrate ``.""" - sub = getattr(args, "migrate_type", None) - if sub == "xai": + if getattr(args, "migrate_type", None) == "xai": return cmd_migrate_xai(args) print("usage: hermes migrate xai [--apply] [--no-backup]", file=sys.stderr) return 2 +def _fail(message: str) -> int: + print(f" {color('✗', Colors.RED)} {message}", file=sys.stderr) + return 1 + + def cmd_migrate_xai(args: Any) -> int: """Run xAI May-15 model migration in dry-run or apply mode.""" from hermes_cli.xai_retirement import ( - MIGRATION_GUIDE_URL, - RETIREMENT_DATE, - apply_migration, - find_retired_xai_refs, - format_issue, - ) + MIGRATION_GUIDE_URL, RETIREMENT_DATE, apply_migration, find_retired_xai_refs, format_issue) apply = bool(getattr(args, "apply", False)) no_backup = bool(getattr(args, "no_backup", False)) - - config = load_config() - issues = find_retired_xai_refs(config) + issues = find_retired_xai_refs(load_config()) print() - print(color( - f"◆ xAI Model Retirement Migration ({RETIREMENT_DATE})", - Colors.CYAN, Colors.BOLD, - )) + print(color(f"◆ xAI Model Retirement Migration ({RETIREMENT_DATE})", Colors.CYAN, Colors.BOLD)) print() if not issues: @@ -65,30 +59,16 @@ def cmd_migrate_xai(args: Any) -> int: print(color( "Re-run with `hermes migrate xai --apply` to rewrite " f"{config_path} in-place (backup created automatically).", - Colors.DIM, - )) + Colors.DIM)) return 0 if not config_path or not config_path.exists(): - print( - f" {color('✗', Colors.RED)} Could not locate config.yaml " - f"(looked at: {config_path})", - file=sys.stderr, - ) - return 1 + return _fail(f"Could not locate config.yaml (looked at: {config_path})") try: - result = apply_migration( - config_path=config_path, - issues=issues, - backup=not no_backup, - ) + result = apply_migration(config_path=config_path, issues=issues, backup=not no_backup) except Exception as exc: - print( - f" {color('✗', Colors.RED)} Migration failed: {exc}", - file=sys.stderr, - ) - return 1 + return _fail(f"Migration failed: {exc}") if not result.config_changed: print(f" {color('⚠', Colors.YELLOW)} No changes written.") @@ -98,18 +78,13 @@ def cmd_migrate_xai(args: Any) -> int: print(f" {color('✓', Colors.GREEN)} Backup: {result.backup_path}") print( f" {color('✓', Colors.GREEN)} Updated {len(result.issues_resolved)} " - f"slot(s) in {result.file_path}" - ) + f"slot(s) in {result.file_path}") print() - print(color( - "Run `hermes doctor` to confirm no retired xAI models remain.", - Colors.DIM, - )) + print(color("Run `hermes doctor` to confirm no retired xAI models remain.", Colors.DIM)) return 0 def _resolve_config_path() -> Path: """Best-effort: locate the active config.yaml on disk.""" from hermes_cli.config import get_hermes_home - return get_hermes_home() / "config.yaml"