From 3b67eee227047fededcdd147a906924863796af6 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:47:34 -0700 Subject: [PATCH] refactor(hermes_cli): install_repair/managed_uv pass 1 --- hermes_cli/main_agent_cmds.py | 180 +++-- hermes_cli/main_install_repair.py | 1195 +++++++++++------------------ hermes_cli/managed_uv.py | 877 ++++++++------------- 3 files changed, 825 insertions(+), 1427 deletions(-) diff --git a/hermes_cli/main_agent_cmds.py b/hermes_cli/main_agent_cmds.py index 5b8088f485..415c4a9844 100644 --- a/hermes_cli/main_agent_cmds.py +++ b/hermes_cli/main_agent_cmds.py @@ -10,86 +10,84 @@ 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) @@ -134,24 +132,22 @@ 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,14 +176,14 @@ 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 @@ -196,11 +192,10 @@ def cmd_skills(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 @@ -208,8 +203,7 @@ def _cmd_skills_trust(args): 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 @@ -224,8 +218,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 +229,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 +243,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 +252,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..7f8caab98e 100644 --- a/hermes_cli/main_install_repair.py +++ b/hermes_cli/main_install_repair.py @@ -7,6 +7,7 @@ Names that stay in main are imported lazily inside the functions that use them avoids an import cycle). """ +import contextlib import logging import os import shlex @@ -23,46 +24,76 @@ 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" @@ -76,17 +107,11 @@ 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 - ) + Unsandboxed update/recovery tests must neither litter the live repo root with + breadcrumbs (they false-arm recovery on the developer's next launch) nor run a + real reinstall against the executing venv (same posture as ``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 +135,13 @@ 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. + A crashed holder leaves a stale lock; break it after an hour (well past any realistic + install) so recovery can't be wedged forever. Failing to CREATE the lock (read-only fs, + perms) proceeds unlocked — the install itself will surface the real problem. """ - 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" 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 +152,29 @@ 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 on stdout. + """ 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 +183,43 @@ 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: if it can't self-heal it prints the manual command and leaves the marker + so the next launch retries. Markers sit next to the shared venv, so concurrent launches + (gateway + CLI, two profiles) race — an ``O_EXCL`` lockfile lets one process recover + while the others skip and let the winner clear markers. + """ + 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 + + # 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: @@ -223,27 +244,25 @@ def _recover_lazy_refresh_marker_locked() -> None: " ⚠ 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) - ) + return + 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) + ) 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( @@ -251,10 +270,9 @@ def _recover_core_update_marker_locked() -> None: "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 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 is first aid only and must NEVER clear this core marker. self_locked = _windows_running_hermes_launcher_locked() if self_locked: install_prefix, install_env = _default_venv_install_target() @@ -268,25 +286,20 @@ def _recover_core_update_marker_locked() -> None: 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 the installer 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: this late path and the 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: @@ -296,9 +309,7 @@ def _recover_core_update_marker_locked() -> None: "different terminal, then run:" ) print(f' cd /d "{PROJECT_ROOT}"') - print( - f' "{sys.executable}" -m pip install -e ".[all]"' - ) + print(f' "{sys.executable}" -m pip install -e ".[all]"') else: print(" Recover manually with:") print(f" cd {PROJECT_ROOT}") @@ -317,21 +328,13 @@ 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`` runs the interpreter with the shim itself as its script + and holds it open — without ``FILE_SHARE_DELETE`` — for the whole process lifetime, so + an editable install run from one can never rewrite it. Two probes, because either can + come up empty: this process's own launch paths (``sys.argv[0]``, ``__main__.__file__``, + the spec origin — the runpy/zipapp launch puts ``\\__main__.py`` there) and the + psutil process ancestry. Candidates are intersected with the project venv's own shims + so a ``hermes.exe`` of some other install never matches. """ from hermes_cli.main import _hermes_exe_shims, _is_windows, _venv_scripts_dir if not _is_windows(): @@ -377,10 +380,7 @@ def _windows_shim_in_process_chain() -> Path | 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,44 +392,24 @@ _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. + Called at the dependency-sync boundary, NOT at the top of the command: by then the code + swap is done and every interactive question (stash, branch switch, config migration) has + been answered in the user's console; only the venv rewrite — the one step that cannot run + from inside the shim — remains. A hand-off before the fetch would detach every run, + including the ``Already up to date!`` no-op, and take the prompts with it. - ``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. + ``venv\\Scripts\\hermes.exe`` holds itself open without ``FILE_SHARE_DELETE``, so the + quarantine rename is refused and uv fails with os error 32. Waiting on the child + deadlocks (this process holds the handle the child needs) and Windows has no exec, so + the shell returns while 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 node/web/lazy-refresh 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": @@ -493,11 +473,9 @@ def _run_install_with_heartbeat( 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() @@ -516,17 +494,31 @@ def _run_install_with_heartbeat( 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" @@ -547,17 +539,15 @@ def _venv_scripts_dir() -> Path | 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. + Only Windows .exe launchers matter: 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)] @@ -566,50 +556,30 @@ def _quarantine_running_hermes_exe( 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)``. + Windows allows RENAMING a running executable but blocks DELETE/REPLACE, so uv fails + with ``Access is denied. (os error 5)`` when rewriting the live shim. Rename live shims + to ``.old.`` first; uv writes fresh shims and ``_cleanup_quarantined_exes`` + sweeps the ``.old`` files on the next invocation. - 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 opened the .exe without ``FILE_SHARE_DELETE`` + (AV scanners: transient, recovers in <1s; a Hermes Desktop backend child: not until + closed). Retry with backoff, then warn naming the likely culprit. The updater's own + launcher is not a culprit: ``_reexec_dependency_sync_off_windows_shim`` moves the update + under the venv Python before reaching here. - 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. + Returns ``(original, quarantined)`` pairs for rollback. ``failed_out`` collects shims + whose rename failed every attempt, so callers that must not mutate a contended venv + (the update dependency sync) can refuse instead of stranding a half-broken install. """ 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. + stamp = int(_time.time() * 1000) + # First attempt immediate; 100/250/500ms covers the typical AV re-scan window. backoff_ms = [0, 100, 250, 500, 1000] attempts = max(1, min(max_attempts, len(backoff_ms))) @@ -622,7 +592,7 @@ def _quarantine_running_hermes_exe( for attempt in range(attempts): delay = backoff_ms[attempt] / 1000.0 if delay: - time.sleep(delay) + _time.sleep(delay) try: shim.rename(target) moved.append((shim, target)) @@ -635,13 +605,9 @@ def _quarantine_running_hermes_exe( if last_exc is None: 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: it needs + # elevation, frees nothing for the install running now, and would move a later, + # freshly repaired shim aside at next boot. Report and let uv try its luck. print( f" ⚠ Could not quarantine {shim.name} ({last_exc.__class__.__name__}: " f"another process is holding it open)." @@ -657,20 +623,15 @@ 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]: +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. + The value is a flat REG_MULTI_SZ of (source, target) pairs shared with other + installers, so only our own ```` -> ``.old.`` pairs are removed. + Returns the entries to keep and how many pairs were dropped. """ import ntpath @@ -698,11 +659,9 @@ 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. + Old ``MoveFileExW(MOVEFILE_DELAY_UNTIL_REBOOT)`` fallbacks outlive the update that queued + them and move away whatever sits at the shim path at next boot — including a shim a later + repair just wrote. Needs elevation to remove (as it did to create); a no-op otherwise. """ from hermes_cli.main import _filter_pending_shim_renames, _hermes_exe_shims, _is_windows if not _is_windows(): @@ -719,9 +678,7 @@ def _cleanup_pending_shim_renames(scripts_dir: Path) -> int: 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: @@ -736,33 +693,24 @@ 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. + Safety-critical direction: a failed quarantine only aborts an update; a failed restore + leaves no ``hermes`` on PATH and no way to run the repair. Delegates to the stdlib-only + retrying helper shared with the early-recovery copy in ``_install_repair``. """ _early_recovery_mod.restore_quarantined_shims(moved) class ShimQuarantineError(RuntimeError): - """A live ``hermes*.exe`` shim could not be renamed aside (#87331). + """A live ``hermes*.exe`` shim could not be renamed aside. - 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. + Raised by :func:`_run_quarantined_install` in ``strict_quarantine`` mode BEFORE the + install runs: a shim that cannot even be renamed means another process holds the venv + hard enough that the sync would die partway — the update must refuse, not 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( @@ -774,27 +722,13 @@ def _run_quarantined_install( ) -> 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 ``pip install -e .`` / ``--reinstall`` 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 (post-sync entry-point repair) already mutated the venv, + so refusing buys nothing. Off-Windows (``scripts_dir is None``) is a thin 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,35 +741,25 @@ 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 on SUCCESS: uv audits an already-satisfied editable + # install as a no-op and rewrites no entry points, which would leave the shims + # quarantined aside. _restore_quarantined_exes skips any shim the installer + # actually replaced. Errors are not swallowed — the 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. + ``None`` means the name was not produced by :func:`_quarantine_running_hermes_exe`; those + are neither rescued nor deleted. Parsed from the NAME rather than ``st_mtime`` because + ``rename`` preserves the shim's mtime (when uv wrote it), not when it was quarantined. """ try: return int(stale.name.rsplit(".old.", 1)[1]) @@ -846,22 +770,13 @@ 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 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. Leave anything inside the grace window alone. + Silent no-op on non-Windows, when nothing to do, or on 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(): @@ -877,18 +792,14 @@ def _cleanup_quarantined_exes(scripts_dir: Path | None = None) -> None: try: candidates = [ (stamp, stale) - for stale, stamp in ( - (p, _quarantine_stamp_ms(p)) for p in scripts_dir.glob("*.exe.old.*") - ) + for stale, stamp in ((p, _quarantine_stamp_ms(p)) for p in scripts_dir.glob("*.exe.old.*")) if stamp 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 filename order only tracks recency while + # every stamp shares a digit width (a stray ``.old.999`` would sort above epoch-ms). candidates.sort(key=lambda pair: pair[0], reverse=True) for stamp, stale in candidates: @@ -896,8 +807,7 @@ def _cleanup_quarantined_exes(scripts_dir: Path | None = None) -> None: 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 @@ -909,31 +819,18 @@ def _cleanup_quarantined_exes(scripts_dir: Path | None = None) -> None: 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. -_LAZY_REFRESH_IMPORT_PROBES: tuple[tuple[str, str], ...] = ( - _early_recovery_mod.LAZY_REFRESH_IMPORT_PROBES -) +# 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 +_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: + """Package-only pip/uv install — no shim quarantine. - -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). + ``pip install --upgrade pip`` / ``--force-reinstall `` do not rewrite ``hermes.exe``; + the editable-install 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) @@ -941,48 +838,31 @@ def _run_package_only_install( 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 _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, + ) + + +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], *, @@ -990,20 +870,16 @@ def _detect_broken_lazy_refresh_imports( ) -> list[str] | None: """Probe lazy-refresh packages via real imports. - 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 + Returns ``[]`` when every package imported cleanly, ``[dist, ...]`` for failures, and + ``None`` when the probe could not run (missing venv Python, subprocess failure, + non-zero probe 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 - ) + probe_lines = "\n".join(f" ({mod!r}, {attr!r})," for mod, attr in _LAZY_REFRESH_IMPORT_PROBES) check_script = ( "import os\n" "import sys\n" @@ -1028,13 +904,7 @@ def _detect_broken_lazy_refresh_imports( "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 @@ -1047,15 +917,10 @@ def _detect_broken_lazy_refresh_imports( ) 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 @@ -1072,18 +937,13 @@ def _repair_broken_lazy_refresh_imports( 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( @@ -1093,22 +953,15 @@ def _repair_venv_via_import_probes( ) -> 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" @@ -1116,50 +969,32 @@ def _repair_venv_via_import_probes( " → Detected corrupted venv packages via import probes: " f"{', '.join(broken)}; repairing..." ) - if _repair_broken_lazy_refresh_imports( - install_cmd_prefix, broken, env=env - ): + 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:]] @@ -1168,18 +1003,15 @@ def _insert_python_pin(args: list[str]) -> list[str]: 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). + On a site-packages install ``PROJECT_ROOT / "venv"`` does not exist; the entry-point + shims uv rewrites live next to the interpreter. Layout comes from the canonical + ``venv_bin_dir`` helper (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 @@ -1194,31 +1026,19 @@ def _install_python_dependencies_with_optional_fallback( ) -> 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 (see + ``_quarantine_running_hermes_exe``). When ``env`` carries a ``VIRTUAL_ENV`` that does + not exist (a pip / site-packages install, 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. """ 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. + # Only uv needs the explicit pin; pip resolves the target from sys.executable + # itself and has no --python flag. pin_python = False if ( env @@ -1227,28 +1047,21 @@ def _install_python_dependencies_with_optional_fallback( 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 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 the + # running hermes.exe locked on Windows. Only override when the venv dir is missing. 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. A shim that cannot be + # renamed aside proves a hard venv hold; 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, @@ -1275,48 +1088,22 @@ def _install_python_dependencies_with_optional_fallback( 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)}" - ) + 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. + # uv's incremental resolver has produced partial installs where a newly added base + # dep silently fails to land on a half-stale venv, surfacing hours later as a + # ModuleNotFoundError in a downstream subprocess. Verify here so it surfaces now. _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( @@ -1326,15 +1113,10 @@ def _verify_console_scripts_installed( ) -> 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, or uv/distlib skipping a launcher write), so + ``hermes`` drops off PATH after a "successful" install. Missing shims are reinstalled + with ``--reinstall -e .`` under the same quarantine dance, then re-checked. """ from hermes_cli.main import _is_windows, _run_quarantined_install, _venv_scripts_dir if not _is_windows(): @@ -1349,11 +1131,7 @@ def _verify_console_scripts_installed( 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: @@ -1364,29 +1142,46 @@ def _verify_console_scripts_installed( 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." - ) + ), + ): return + _report_still_missing( + _missing(), "Workaround: python -m hermes_cli.main ", + ok=" ✓ All console entry points restored", + ) - 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") + +def _applicable_dependency_names(raw_deps: list[str]) -> list[str]: + """Declared dep names whose ``;`` environment markers apply on this platform. + + Without markers every cross-platform exclusion (``ptyprocess ; sys_platform != 'win32'``) + would false-positive on Windows. An unevaluable marker counts as applicable. + """ + applicable: list[str] = [] + for name, marker, _ in _parse_requirements(raw_deps): + try: + 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( @@ -1395,109 +1190,37 @@ def _verify_core_dependencies_installed( env: dict[str, str] | None = None, group: str = "all", ) -> None: - """Check that every base dep from pyproject.toml is importable; if not, retry. + """Check that every base dep from pyproject.toml is installed in the target venv; 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. + Reads ``pyproject.toml`` directly (not the venv's stale metadata), drops deps whose + ``;`` markers don't apply here, and runs ``importlib.metadata.version()`` in the venv + interpreter. Anything missing triggers a ``--reinstall`` of the base group, 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 at the spot that caused it. """ - 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 + 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 - - 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. - applicable: list[str] = [] - for name, marker in deps: - if marker is None: - applicable.append(name) - continue - try: - if marker.evaluate(): # type: ignore[union-attr] - applicable.append(name) - except Exception: - applicable.append(name) - + 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: @@ -1509,24 +1232,16 @@ def _verify_core_dependencies_installed( ) 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; the full + # all-extras install costs minutes and trips on whatever extra was already broken + # upstream. Quarantine first: ``--reinstall -e .`` rewrites the entry-point 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() @@ -1534,67 +1249,40 @@ def _verify_core_dependencies_installed( 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." - ) + ), + ): 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. - """ + """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(): @@ -1609,54 +1297,33 @@ 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; running it against the Linux checkout + goes through ``\\\\wsl.localhost\\...`` UNC paths and fails with EISDIR / symlink errors + in symlink-heavy trees. Refuse it on a POSIX host and re-scan PATH minus the ``/mnt/*`` + drive mounts. Returns ``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 +1334,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_uv.py b/hermes_cli/managed_uv.py index 61f243c220..07efcbb53f 100644 --- a/hermes_cli/managed_uv.py +++ b/hermes_cli/managed_uv.py @@ -1,7 +1,7 @@ """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. """ @@ -24,10 +24,7 @@ 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 +35,20 @@ _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] +# Public helpers + 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,37 +58,22 @@ 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, + 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", + "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 @@ -104,45 +81,28 @@ def managed_python_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( @@ -150,8 +110,7 @@ def _macos_sign_managed_python(python: Path) -> bool: ) 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,15 +141,11 @@ 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}") @@ -199,8 +154,8 @@ def _report_runtime_repair_failure(repair: RuntimeRepairResult) -> None: 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. + POSIX only: never returned on Windows, where a str subclass with an overridden ``__iter__`` + is unsafe as a subprocess argument. """ fresh_bootstrap: bool @@ -211,41 +166,32 @@ 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()`` after bootstrapping uv. Repair here so that first update can migrate a + # vulnerable runtime without requiring a second ``hermes update``. _run_runtime_repair(result, repair_observer) else: print(" ✗ Managed uv install appeared to succeed but binary not found") @@ -260,10 +206,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, + 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: @@ -279,23 +223,16 @@ def _run_runtime_repair( def ensure_uv( - *, - repair_observer: Callable[[RuntimeRepairResult], None] | None = None, -): + *, repair_observer: Callable[[RuntimeRepairResult], None] | None = None): """Return the managed uv path, installing it first if necessary. - 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. Falsy on failure — never + raises. ``repair_observer`` receives the repair result produced after a fresh bootstrap. """ 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) @@ -307,11 +244,10 @@ def _uv_self_update_stamp() -> Path: 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 @@ -331,41 +267,30 @@ def _touch_uv_self_update_stamp() -> None: # 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 during ``hermes update``. - 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. + Returns the managed path when uv is available, else ``None``. 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,30 +299,21 @@ 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) + Needed when the already-imported module predates ``venv_python_path``. """ import hermes_constants @@ -421,6 +337,21 @@ 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: @@ -432,26 +363,22 @@ def _make_world_traversable(path: Path) -> None: 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). + Requesting the exact patch can never repair some installs: python-build-standalone may have + no artifact with fixed SQLite for that patch at all (the fix may only exist from the next). """ - return ".".join(str(part) for part in info.python_version[:2]) + 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. + 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. - Returns [] on any failure (network, parse) -- callers fall back to the original bare-minor - request in that case, preserving prior behavior. + Returns [] on any failure (network, parse); callers then fall back to the bare-minor request. """ try: result = subprocess.run( @@ -459,146 +386,106 @@ def _list_available_patches( 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, - ) + 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, + 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. +) -> _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 + return _reject(generation, python_root, msg, *args) 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 +498,35 @@ 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 +536,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,26 +568,14 @@ 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: @@ -721,27 +586,17 @@ def _smoke_candidate_venv(venv_dir: Path) -> tuple[bool, str, SQLiteRuntimeInfo 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), + "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 + return _reject(candidate, runtime_root, msg, *args) print(" → Building a relocatable replacement environment...") created = subprocess.run( @@ -749,33 +604,22 @@ def _stage_candidate_venv( 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, - ) + 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 +639,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 +659,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 +683,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 +696,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,14 +708,12 @@ 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) @@ -918,12 +747,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, "" @@ -944,14 +772,10 @@ def _windows_runtime_self_lock(live: Path) -> tuple[bool, str]: 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. + return True, (f"the updater itself runs from the live venv it must replace ({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: import psutil @@ -962,10 +786,8 @@ 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" - ) + f"ancestor process PID {anc.pid} runs from the live venv ({anc_exe}); " + "Windows cannot rename a directory while a process executes from inside it") except Exception: pass return False, "" @@ -977,8 +799,7 @@ 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 "" @@ -988,15 +809,13 @@ 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. + 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 +828,23 @@ 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 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`` wins when it holds an interpreter (managed layout takes precedence), else ``.venv`` + when that one does. When neither has one, return ``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-*")) @@ -1054,11 +863,99 @@ def _sweep_stale_runtime_backups( _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 _safe_result(current: SQLiteRuntimeInfo) -> RuntimeRepairResult: + return _result("safe", current, sqlite_after=current.sqlite_version_string) + + +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 _safe_result(current) + 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,167 +967,32 @@ 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 _safe_result(current) + 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. @@ -1238,36 +1000,20 @@ def _install_uv(target: Path) -> None: 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: os.unlink(installer_path) @@ -1279,11 +1025,8 @@ 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: