diff --git a/hermes_cli/_early_recovery.py b/hermes_cli/_early_recovery.py index 50b30acd25..ad29b2ce14 100644 --- a/hermes_cli/_early_recovery.py +++ b/hermes_cli/_early_recovery.py @@ -1,11 +1,9 @@ """Dependency-light venv recovery that runs BEFORE hermes_cli.main's imports. -This module is deliberately **stdlib-only** so importing it can never fail on a corrupted venv. -``hermes_cli.main`` imports and calls :func:`recover_if_needed` at the very top of its module body, -before any third-party import. - -Scope: this early pass only repairs enough for ``hermes_cli.main`` to become importable again -(force-reinstall of the known-fragile core packages, using the pins from pyproject.toml). +Deliberately **stdlib-only** so importing it can never fail on a corrupted venv. ``hermes_cli.main`` +calls :func:`recover_if_needed` at the very top of its module body, before any third-party import. +Scope: repair only enough for ``hermes_cli.main`` to become importable again (force-reinstall of +the known-fragile core packages, using the pins from pyproject.toml). """ from __future__ import annotations @@ -18,54 +16,31 @@ import sys import time from pathlib import Path -# Core packages a failed lazy ``uv pip install`` is known to leave with intact -# distribution metadata but wiped import files (#57828). ``module`` is what we -# probe via a real import; ``attr`` guards against an empty/stub module. -# main.py's marker-recovery path reuses these tables — keep them here (the -# dependency-light module) so both layers probe and repair the same set. +# Core packages a failed lazy ``uv pip install`` is known to leave with intact distribution +# metadata but wiped import files. ``module`` is probed via a real import; ``attr`` guards against +# an empty/stub module. main.py's marker-recovery path reuses these tables — keep them here so +# both layers probe and repair the same set. LAZY_REFRESH_IMPORT_PROBES: tuple[tuple[str, str], ...] = ( - ("yaml", "SafeDumper"), - ("dotenv", "load_dotenv"), - ("click", "Command"), - ("certifi", "contents"), - ("rich", "print"), - ("cryptography", "__version__"), + ("yaml", "SafeDumper"), ("dotenv", "load_dotenv"), ("click", "Command"), + ("certifi", "contents"), ("rich", "print"), ("cryptography", "__version__"), ("jwt", "encode"), ) LAZY_REFRESH_REPAIR_PACKAGES: dict[str, str] = { - "yaml": "PyYAML", - "dotenv": "python-dotenv", - "click": "click", - "certifi": "certifi", - "rich": "rich", - "cryptography": "cryptography", - "jwt": "PyJWT", + "yaml": "PyYAML", "dotenv": "python-dotenv", "click": "click", "certifi": "certifi", + "rich": "rich", "cryptography": "cryptography", "jwt": "PyJWT", } - -# --- Windows entry-point shim quarantine ----------------------------------- -# -# ``hermes update`` renames the live ``hermes*.exe`` shims aside -# (``hermes.exe.old.``) so uv can write replacements. Putting them BACK -# is the safety-critical direction: losing that rename leaves the install with -# no ``hermes`` on PATH, and the command that would repair it IS ``hermes -# update`` (#75584). -# -# Three call sites restore a quarantined shim -- the updater, the -# early-recovery installer, and the startup sweep's orphan rescue. They used to -# be separate one-shot renames with swallowed errors; the two that had messages -# had already drifted apart. The logic lives here, in the one stdlib-only module -# all of them can import, so the ladder and the recovery wording stay in -# lockstep. - +# ``hermes update`` renames the live ``hermes*.exe`` shims aside (``hermes.exe.old.``) so +# uv can write replacements. Putting them BACK is the safety-critical direction: losing that rename +# leaves no ``hermes`` on PATH, and the command that would repair it IS ``hermes update``. The +# updater, the early-recovery installer and the startup orphan sweep all restore through this one +# stdlib-only helper so the retry ladder and the recovery wording cannot drift apart again. QUARANTINE_RESTORE_BACKOFF_MS: tuple[int, ...] = (0, 100, 250, 500, 1000) def restore_quarantined_shims( - moved: list[tuple[Path, Path]], - *, - stream=None, + moved: list[tuple[Path, Path]], *, stream=None, backoff_ms: tuple[int, ...] = QUARANTINE_RESTORE_BACKOFF_MS, ) -> list[tuple[Path, Path]]: """Rename quarantined shims back, retrying a lock instead of giving up. @@ -76,12 +51,9 @@ def restore_quarantined_shims( """ if stream is None: stream = sys.stderr - failed: list[tuple[Path, Path]] = [] - for original, quarantined in moved: last_exc: OSError | None = None - for delay_ms in backoff_ms: try: if os.path.exists(original) or not os.path.exists(quarantined): @@ -94,11 +66,8 @@ def restore_quarantined_shims( break except OSError as exc: last_exc = exc - continue - if last_exc is None: continue - failed.append((original, quarantined)) name = os.path.basename(str(original)) stem = name[:-4] if name.lower().endswith(".exe") else name @@ -111,16 +80,13 @@ def restore_quarantined_shims( f' move "{quarantined}" "{original}"', file=stream, ) - return failed -# Set only when this process successfully finishes a deferred core install for -# an ``update`` invocation. The normal CLI import that follows must not resolve -# external secret sources: a configured source can map cryptography._rust and -# immediately recreate the self-lock marker this fresh process just consumed. -# Process-local state is intentional so child processes do not inherit the -# bootstrap exception. +# Set only when this process successfully finishes a deferred core install for an ``update`` +# invocation. The CLI import that follows must not resolve external secret sources: a configured +# source can map cryptography._rust and immediately recreate the self-lock marker this fresh +# process just consumed. Process-local on purpose so children do not inherit the exception. _UPDATE_RETRY_RECOVERED = False @@ -134,11 +100,7 @@ def _project_root() -> Path: def _load_pyproject_project(root: Path) -> dict | None: - """``[project]`` table of ``root/pyproject.toml``; ``None`` when missing/unreadable. - - Stdlib-only (tomllib) — shared by every pyproject lookup in the repair path so ``packaging`` - is never needed in the broken-venv state. - """ + """``[project]`` table of ``root/pyproject.toml``; ``None`` when missing/unreadable.""" pyproject = root / "pyproject.toml" if not pyproject.is_file(): return None @@ -171,23 +133,12 @@ def _read_marker_attempts(marker_path: Path) -> int: def _run_ensurepip(root: Path) -> None: """Best-effort pip bootstrap — a killed install can leave the venv with no pip module at all.""" try: - subprocess.run( - [sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"], - cwd=root, - capture_output=True, - ) + subprocess.run([sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"], + cwd=root, capture_output=True) except Exception: pass -def _report_failed_install(result) -> bool: - """Replay the captured tail of a failed installer run to stderr; always ``False``.""" - tail = (result.stderr or result.stdout or "")[-2000:] - if tail: - print(tail, file=sys.stderr) - return False - - def _pid_is_running(pid: int) -> bool: """Best-effort stdlib-only process liveness probe. @@ -203,11 +154,7 @@ def _pid_is_running(pid: int) -> bool: synchronize = 0x00100000 kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) - kernel32.OpenProcess.argtypes = [ - ctypes.c_ulong, - ctypes.c_int, - ctypes.c_ulong, - ] + kernel32.OpenProcess.argtypes = [ctypes.c_ulong, ctypes.c_int, ctypes.c_ulong] kernel32.OpenProcess.restype = ctypes.c_void_p kernel32.WaitForSingleObject.argtypes = [ctypes.c_void_p, ctypes.c_ulong] kernel32.WaitForSingleObject.restype = ctypes.c_ulong @@ -252,16 +199,14 @@ def _marker_owner_is_live(marker: Path) -> bool: def _pinned_specs(packages: list[str], project_root: Path) -> list[str]: """Map bare package names to their pinned specs from pyproject.toml. - Stdlib-only (tomllib + naive requirement-head parsing — ``packaging`` may itself be broken in - the failure state this module exists for). Unknown packages fall back to their bare name. + Naive requirement-head parsing on purpose — ``packaging`` may itself be broken in the failure + state this module exists for. Unknown packages fall back to their bare name. """ project = _load_pyproject_project(project_root) if project is None: return packages - raw_deps = project.get("dependencies", []) or [] - name_to_spec: dict[str, str] = {} - for spec in raw_deps: + for spec in project.get("dependencies", []) or []: head = spec.split(";", 1)[0].strip() bare = head for op in ("==", ">=", "<=", "~=", ">", "<", "!="): @@ -278,8 +223,8 @@ def _certifi_bundle_broken() -> bool: """True when certifi imports but its ``cacert.pem`` is missing/corrupt. A brew Python upgrade or interrupted venv rebuild can leave certifi's metadata intact while - ``cacert.pem`` is gone or a dangling symlink; every TLS connection then fails opaquely deep in - httpx/requests. An attribute probe passes in that state, so validate the bundle path itself. + ``cacert.pem`` is gone or a dangling symlink; an attribute probe passes in that state and every + TLS connection then fails opaquely, so validate the bundle path itself. """ try: import certifi @@ -288,8 +233,7 @@ def _certifi_bundle_broken() -> bool: # <1 KiB cannot hold a single PEM certificate — treat as corrupt. return not bundle.is_file() or bundle.stat().st_size < 1024 except Exception: - # Import failure is caught by the regular probe table; a failure to - # even stat is treated as broken. + # Import failure is caught by the regular probe table; failing to even stat is broken. return True @@ -319,16 +263,12 @@ def _find_uv_binary() -> str | None: """Locate a ``uv`` binary without importing third-party modules. uv-managed base interpreters carry an ``EXTERNALLY-MANAGED`` marker, so the stdlib ``pip`` - fallback below refuses to touch them. In that state the only sanctioned installer is uv itself, - which Hermes already vendors (``~/.hermes/bin/uv.exe``) or the user has on PATH. Stdlib-only. + fallback refuses to touch them; the only sanctioned installer is then uv itself, which Hermes + vendors (``~/.hermes/bin/uv.exe``) or the user has on PATH. """ exe = "uv.exe" if sys.platform == "win32" else "uv" - candidates = [ - Path.home() / ".hermes" / "bin" / exe, - Path.home() / ".local" / "bin" / exe, - Path.home() / ".cargo" / "bin" / exe, - ] - for path in candidates: + for sub in ((".hermes", "bin"), (".local", "bin"), (".cargo", "bin")): + path = Path.home().joinpath(*sub, exe) if path.is_file(): return str(path) return shutil.which(exe) @@ -337,30 +277,42 @@ def _find_uv_binary() -> str | None: def _base_interpreter_is_externally_managed() -> bool: """True when ``sys.executable`` is a uv/standalone-builds managed install. - Those interpreters ship an ``EXTERNALLY-MANAGED`` marker next to their stdlib (PEP 668), so - ``python -m pip install`` aborts with ``externally-managed-environment``. The early repair must - then go through uv (or explicitly override pip) or the reinstall no-ops and the venv stays - broken (#83569). + Those ship an ``EXTERNALLY-MANAGED`` marker next to their stdlib (PEP 668), so + ``python -m pip install`` aborts with ``externally-managed-environment``; the early repair + must then go through uv (or explicitly override pip) or the venv stays broken. """ try: import sysconfig stdlib = Path(sysconfig.get_path("stdlib")) # uv 0.5+ moved the marker one level up next to a ``_uv_managed`` sentinel dir. - return (stdlib / "EXTERNALLY-MANAGED").exists() or ( - stdlib.parent / "EXTERNALLY-MANAGED" - ).exists() + return ((stdlib / "EXTERNALLY-MANAGED").exists() + or (stdlib.parent / "EXTERNALLY-MANAGED").exists()) except Exception: return False +def _run_installer(tool: str, cmd: list[str], root: Path, env: dict | None = None) -> bool: + """Run one installer command; captured output is replayed to stderr only on failure.""" + try: + result = subprocess.run(cmd, cwd=root, capture_output=True, text=True, encoding="utf-8", + errors="replace", env=env) + except Exception as exc: + print(f" ✗ Early venv repair could not run {tool}: {exc}", file=sys.stderr) + return False + if result.returncode == 0: + return True + tail = (result.stderr or result.stdout or "")[-2000:] + if tail: + print(tail, file=sys.stderr) + return False + + def _run_repair_install(specs: list[str], project_root: Path) -> bool: - """``uv pip`` (or stdlib ``pip``) force-reinstall of the given specs. + """``uv pip`` (or stdlib ``pip``) force-reinstall of the given specs. Never raises. - Streams nothing to stdout (``hermes acp`` speaks JSON-RPC on stdout); output is captured and - replayed to stderr only on failure. Never raises. - - Two installer paths, in priority order: + Streams nothing to stdout (``hermes acp`` speaks JSON-RPC on stdout). uv is preferred when the + base interpreter is externally managed; without uv, pip runs with the PEP 668 override. """ externally_managed = _base_interpreter_is_externally_managed() if externally_managed: @@ -369,68 +321,32 @@ def _run_repair_install(specs: list[str], project_root: Path) -> bool: env = {**os.environ, "VIRTUAL_ENV": str(project_root / "venv")} env.pop("PYTHONHOME", None) env.pop("PYTHONPATH", None) - try: - result = subprocess.run( - [uv, "pip", "install", "--force-reinstall", *specs], - cwd=project_root, - capture_output=True, - text=True, encoding="utf-8", errors="replace", - env=env, - ) - except Exception as exc: - print(f" ✗ Early venv repair could not run uv: {exc}", file=sys.stderr) - return False - return result.returncode == 0 or _report_failed_install(result) - # No uv available: fall through to pip with the PEP 668 override so - # the repair at least attempts to fix the venv instead of no-oping. - print( - " ⚠ Base interpreter is externally managed and no uv binary was " - "found; retrying repair via pip with PEP 668 override.", - file=sys.stderr, - ) - + return _run_installer("uv", [uv, "pip", "install", "--force-reinstall", *specs], + project_root, env) + print(" ⚠ Base interpreter is externally managed and no uv binary was " + "found; retrying repair via pip with PEP 668 override.", file=sys.stderr) _run_ensurepip(project_root) pip_cmd = [sys.executable, "-m", "pip", "install", "--force-reinstall"] if externally_managed: pip_cmd.append("--break-system-packages") - pip_cmd.extend(specs) - try: - result = subprocess.run( - pip_cmd, - cwd=project_root, - capture_output=True, - text=True, encoding="utf-8", errors="replace", - ) - except Exception as exc: - print(f" ✗ Early venv repair could not run pip: {exc}", file=sys.stderr) - return False - return result.returncode == 0 or _report_failed_install(result) + return _run_installer("pip", pip_cmd + specs, project_root) def _pytest_owns_live_checkout(root: Path) -> bool: - """True when running under pytest AND ``root`` is this module's own checkout — the one whose - venv is executing the suite right now. + """True under pytest when ``root`` is this module's own checkout — the venv running the suite. Lifecycle tests spawn real subprocesses that import ``hermes_cli.main`` with recovery armed and - inherit ``PYTEST_CURRENT_TEST``. Without this guard a broken dev venv would get a REAL ensurepip - + ``pip install --force-reinstall`` from inside a running suite. Tests sandboxing - ``project_root`` to a tmp_path are unaffected. + inherit ``PYTEST_CURRENT_TEST``; without this guard a broken dev venv would get a REAL + ensurepip + force-reinstall from inside a running suite. tmp_path roots are unaffected. """ - return ( - "PYTEST_CURRENT_TEST" in os.environ - and root == Path(__file__).resolve().parent.parent - ) + return "PYTEST_CURRENT_TEST" in os.environ and root == Path(__file__).resolve().parent.parent -def recover_if_needed( - project_root: Path | None = None, - argv: list[str] | None = None, -) -> None: +def recover_if_needed(project_root: Path | None = None, argv: list[str] | None = None) -> None: """Repair wiped core packages so ``hermes_cli.main`` can import at all. Fast path (no marker present) is two ``lstat`` calls. Only acts when a recovery marker from a prior ``hermes update`` exists AND an import probe confirms a core package is actually broken. - Never raises: on any failure the import of main.py proceeds and surfaces the real error. """ global _UPDATE_RETRY_RECOVERED @@ -444,82 +360,56 @@ def recover_if_needed( lazy_marker = root / ".lazy-refresh-incomplete" if not core_marker.exists() and not lazy_marker.exists(): return - # Managed/Docker/PyPI installs have no source tree here — the marker - # is not ours to act on; main.py's recovery clears it. + # Managed/Docker/PyPI installs have no source tree here — the marker is not ours to act + # on; main.py's recovery clears it. if not (root / "pyproject.toml").is_file(): return - # Pending core install (.update-incomplete) — complete it NOW, before - # any native extension can be imported by this process. The lazy - # import-probe path below only proves main.py is importable; the core - # install here guarantees the WHOLE dependency set is replaced while - # nothing pins venv .pyd files yet (#83569 self-lock: deferring the - # install to main()'s post-import recovery re-locks it on Windows). - # Bounded retries: a persistently failing install must not hammer - # every launch, so attempts past the ceiling are left for main.py's - # post-import recovery path (which can safely probe-import after this - # process already holds whatever extensions it needs). - # A live marker owner means another updater is currently inside the - # marker-to-install window. Never race it. A dead owner means this is - # a prior deferral/interruption and MUST be recovered even when this - # launch is itself `hermes update`: CLI and Desktop retries preserve + # Pending core install: complete it NOW, before any native extension is imported, so the + # WHOLE dependency set is replaced while nothing pins venv .pyd files yet (deferring to + # main()'s post-import recovery re-locks it on Windows). A live marker owner is another + # updater inside the marker-to-install window — never race it. A dead owner MUST be + # recovered even when this launch is itself `hermes update`: CLI and Desktop retries keep # that argv, and skipping solely on argv recreates the self-lock loop. if core_marker.exists(): if _marker_owner_is_live(core_marker): return - completed = _complete_pending_core_install(root, core_marker) - if completed and "update" in args: + if _complete_pending_core_install(root, core_marker) and "update" in args: _UPDATE_RETRY_RECOVERED = True return - # Keep the historical update-argv exclusion for the lazy-refresh - # marker. Unlike the core marker it is not a deferred native install, - # and the active update flow owns its probe/repair lifecycle. + # The lazy-refresh marker keeps the update-argv exclusion: it is not a deferred native + # install, and the active update flow owns its probe/repair lifecycle. if "update" in args: return broken = _probe_broken_packages() if not broken: - # Imports are fine — main.py will load and run full recovery. - return + return # main.py will load and run full recovery. - # Single-flight: share main.py's recovery lock so an early repair - # never races a concurrent full recovery into the same shared venv. + # Single-flight: share main.py's recovery lock so an early repair never races a + # concurrent full recovery into the same shared venv. if not _claim_recovery_lock(root): return - try: specs = _pinned_specs(broken, root) - print( - "⚠ Core package(s) broken by an interrupted update — " - f"repairing before launch: {', '.join(broken)}", - file=sys.stderr, - ) + print("⚠ Core package(s) broken by an interrupted update — " + f"repairing before launch: {', '.join(broken)}", file=sys.stderr) if _run_repair_install(specs, root) and not _probe_broken_packages(): print(" ✓ Core packages repaired.", file=sys.stderr) else: - print( - " ✗ Automatic repair incomplete. Recover manually with:", - file=sys.stderr, - ) - print( - f" {sys.executable} -m pip install --force-reinstall " - + " ".join(specs), - file=sys.stderr, - ) + print(" ✗ Automatic repair incomplete. Recover manually with:", file=sys.stderr) + print(f" {sys.executable} -m pip install --force-reinstall " + " ".join(specs), + file=sys.stderr) finally: _release_recovery_lock(root) except Exception: - # Never block launch — the import of main.py will surface the truth. - pass + pass # Never block launch — the import of main.py will surface the truth. -# Cap on automatic early-pass install retries. A persistently failing -# install (e.g. network down, index unreachable) must not reinstall-hammer -# every `hermes` launch: past this many attempts the early pass hands the -# marker to main.py's post-import recovery, which presents the manual -# recovery command. The counter lives inside the marker file itself (JSON -# body) and is bumped on each failed attempt. +# Cap on automatic early-pass install retries: a persistently failing install (network down) must +# not reinstall-hammer every launch. Past this the marker is left to main.py's post-import recovery, +# which presents the manual command. The counter lives in the marker's JSON body. _EARLY_CORE_INSTALL_MAX_ATTEMPTS = 3 @@ -539,8 +429,7 @@ def _claim_recovery_lock(root: Path) -> bool: pass return False except OSError: - # Read-only fs / perms — proceed unlocked; the install itself - # surfaces the real problem. Recoverable. + # Read-only fs / perms — proceed unlocked; the install itself surfaces the real problem. return True @@ -555,52 +444,32 @@ def _release_recovery_lock(root: Path) -> None: def _complete_pending_core_install(root: Path, core_marker: Path) -> bool: """Run the pending core install BEFORE main.py can import native modules. - ``recover_if_needed`` invokes this when ``.update-incomplete`` exists — a prior ``hermes - update`` (or the self-lock preflight, #83569) left the dependency sync deliberately unfinished. - Never raises: any failure leaves the marker for the post-import path and returns ``False``. Returns ``True`` only after the install succeeds. """ try: from hermes_cli import _install_repair as ir - # Retry backoff: read current attempts before claiming the lock so a - # persistently-failing install stops hammering early. After the - # increment the counter reflects THIS attempt. + # Read attempts before claiming the lock so a persistently-failing install stops early. attempts = _read_marker_attempts(core_marker) - if attempts >= _EARLY_CORE_INSTALL_MAX_ATTEMPTS: - print( - "⚠ Pending interrupted-update install has already failed " - f"{attempts} times in the early pass — leaving it for the " - "post-import recovery path.", - file=sys.stderr, - ) + print("⚠ Pending interrupted-update install has already failed " + f"{attempts} times in the early pass — leaving it for the " + "post-import recovery path.", file=sys.stderr) return False - if not _claim_recovery_lock(root): return False - try: - print( - "⚠ A previous `hermes update` was interrupted mid-install — " - "finishing dependency installation now (before any native " - "extensions load)...", - file=sys.stderr, - ) + print("⚠ A previous `hermes update` was interrupted mid-install — " + "finishing dependency installation now (before any native " + "extensions load)...", file=sys.stderr) ir.run_core_install(root) except Exception as exc: new_attempts = ir.bump_marker_attempts(core_marker) - print( - f" ✗ Early interrupted-install completion failed (attempt " - f"{new_attempts}/{_EARLY_CORE_INSTALL_MAX_ATTEMPTS}): {exc}", - file=sys.stderr, - ) - print( - " The next launch will retry; hermes will keep working from " - "the current venv in the meantime.", - file=sys.stderr, - ) + print(f" ✗ Early interrupted-install completion failed (attempt " + f"{new_attempts}/{_EARLY_CORE_INSTALL_MAX_ATTEMPTS}): {exc}", file=sys.stderr) + print(" The next launch will retry; hermes will keep working from " + "the current venv in the meantime.", file=sys.stderr) return False finally: _release_recovery_lock(root) @@ -609,11 +478,7 @@ def _complete_pending_core_install(root: Path, core_marker: Path) -> bool: core_marker.unlink() except OSError: pass - print( - " ✓ Dependency installation completed in the early pass.", - file=sys.stderr, - ) + print(" ✓ Dependency installation completed in the early pass.", file=sys.stderr) return True except Exception: - # Never block launch — the marker stays for the post-import path. - return False + return False # Never block launch — the marker stays for the post-import path. diff --git a/hermes_cli/_install_repair.py b/hermes_cli/_install_repair.py index 49aaa2600c..763e16c521 100644 --- a/hermes_cli/_install_repair.py +++ b/hermes_cli/_install_repair.py @@ -1,13 +1,10 @@ """Dependency install execution shared between early recovery and full recovery. -- ``hermes_cli._early_recovery.recover_if_needed`` — stdlib-only, runs BEFORE ``hermes_cli.main``'s -third-party imports, so it can complete a pending update while no native extension is mapped yet -(#83569). - ``hermes_cli.main._recover_core_update_marker_locked`` — the historical post-import -recovery path. - -This module is deliberately **stdlib-only** so importing it can never fail in the corrupted-venv -state it exists to repair. ``hermes_cli.main`` imports ``managed_uv``, ``hermes_constants``, and -friends only in its late path; the early path must not. +Callers: ``_early_recovery.recover_if_needed`` (stdlib-only, runs BEFORE ``hermes_cli.main``'s +third-party imports so a pending update completes while no native extension is mapped yet) and +``hermes_cli.main._recover_core_update_marker_locked`` (the post-import recovery path). +Deliberately **stdlib-only** so importing it can never fail in the corrupted-venv state it exists +to repair; ``managed_uv`` and friends belong to the late path only. """ from __future__ import annotations @@ -21,8 +18,7 @@ import sys import time from pathlib import Path -# Single source of truth for the recovery-lock lifecycle and uv lookup — -# _early_recovery already owns both, and importing it is free (stdlib-only). +# _early_recovery owns the recovery-lock lifecycle and uv lookup; importing it is free (stdlib). from hermes_cli import _early_recovery as _er @@ -44,7 +40,7 @@ def _stdout_to_stderr(): """Route fd 1 (and sys.stdout) to stderr for the duration of an install. ``hermes acp`` speaks JSON-RPC on stdout; an inherited-fd install child writing there would - corrupt the protocol. Mirrors ``main.py::_recover_from_interrupted_install``. + corrupt the protocol. """ saved_sys_stdout = sys.stdout try: @@ -67,9 +63,8 @@ def _stdout_to_stderr(): def _resolve_install_target(root: Path) -> tuple[list[str], dict | None]: """(install_cmd_prefix, env) for the project venv — stdlib uv lookup. - Mirrors ``main.py::_default_venv_install_target`` without ``managed_uv``. ``VIRTUAL_ENV`` - steers ``uv pip`` at the project venv even when invoked from the base interpreter (early - recovery). Termux strips leaked interpreter-path env vars so uv resolves the venv correctly. + ``VIRTUAL_ENV`` steers ``uv pip`` at the project venv even when invoked from the base + interpreter. Termux strips leaked interpreter-path env vars so uv resolves the venv correctly. """ uv_bin = _er._find_uv_binary() if uv_bin: @@ -84,24 +79,18 @@ def _resolve_install_target(root: Path) -> tuple[list[str], dict | None]: def _venv_scripts_dir(root: Path) -> Path | None: - """Project venv Scripts/bin dir, when present. stdlib-only.""" - # hermes_constants is stdlib-only, so the canonical layout helpers are safe - # to use from this corrupted-venv repair path (#76105: never open-code - # the Scripts/bin split). + """Project venv Scripts/bin dir, when present (hermes_constants is stdlib-only).""" from hermes_constants import project_venv_dir, venv_bin_dir venv_dir = project_venv_dir(root) if venv_dir is None: return None - scripts = venv_bin_dir(venv_dir, windows=_is_windows()) return scripts if scripts.is_dir() else None -#: Launcher command names install.ps1's Set-PathVariable exposes from the -#: managed binary dir (the default Hermes root's ``bin``, next to uv.exe) -#: on the user PATH. Keep in lockstep with the launcher list in -#: scripts/install.ps1. +#: Launcher names install.ps1's Set-PathVariable exposes from the managed binary dir (the default +#: Hermes root's ``bin``, next to uv.exe). Keep in lockstep with scripts/install.ps1. _WINDOWS_BIN_LAUNCHERS = ("hermes", "hermes-acp") @@ -114,9 +103,9 @@ def _launchers_missing(target: Path) -> bool: def _default_hermes_root() -> Path | None: - """Per-machine anchor for the managed-clone gate — the DEFAULT Hermes root, not - ``get_hermes_home()``: under ``hermes -p `` that returns ``profiles\\``, which - would fail the gate and silently skip the heal for profile users. ``None`` when unresolvable.""" + """The DEFAULT Hermes root (not ``get_hermes_home()``, which under ``hermes -p `` is + ``profiles\\`` and would fail the managed-clone gate for profile users); ``None`` when + unresolvable.""" from hermes_constants import get_default_hermes_root try: @@ -128,16 +117,12 @@ def _default_hermes_root() -> Path | None: def _venv_is_relocatable(venv_dir: Path) -> bool: r"""True when the venv's pyvenv.cfg declares ``relocatable = true``. - uv writes the flag; ``managed_uv`` builds replacement venvs ``--relocatable``. A relocatable - venv's console-script trampolines embed a RELATIVE interpreter reference, so a copy placed - outside ``venv\Scripts`` fails with ``uv trampoline failed to canonicalize script path``; - non-relocatable venvs embed the absolute path and survive copying. This decides which - launcher form a PATH dir gets. + A relocatable venv's console-script trampolines embed a RELATIVE interpreter reference, so a + copy placed outside ``venv\Scripts`` fails (``uv trampoline failed to canonicalize script + path``); non-relocatable venvs survive copying. Decides which launcher form a PATH dir gets. """ try: - cfg = (Path(venv_dir) / "pyvenv.cfg").read_text( - encoding="utf-8", errors="replace" - ) + cfg = (Path(venv_dir) / "pyvenv.cfg").read_text(encoding="utf-8", errors="replace") except OSError: return False return any( @@ -149,19 +134,14 @@ def _venv_is_relocatable(venv_dir: Path) -> bool: def _normalize_windows_path(value) -> str: """Windows path equality key: backslashes, no trailing separator, lowered. - Lowercase via ``.lower()`` (what ``ntpath.normcase`` does) rather than ``os.path.normcase`` — - that is an identity function on POSIX, and this comparison must behave Windows-correct even when - tests exercise the Windows branch from another host (same rationale as - ``venv_bin_dir(windows=...)``). + ``.lower()`` rather than ``os.path.normcase`` (identity on POSIX) so the comparison behaves + Windows-correct even when tests exercise the Windows branch from another host. """ return str(value).replace("/", "\\").rstrip("\\").lower() def _windows_user_path_entries() -> list[str]: - """User PATH entries from the registry — the value install.ps1 writes. - - Falls back to the process PATH when the registry is unreadable. Only called on Windows. - """ + """User PATH entries from the registry (what install.ps1 writes); process PATH fallback.""" try: import winreg @@ -174,48 +154,34 @@ def _windows_user_path_entries() -> list[str]: def ensure_windows_bin_launchers( - root, - *, - windows: bool | None = None, - user_path_entries: list[str] | None = None, + root, *, windows: bool | None = None, user_path_entries: list[str] | None = None, ) -> list[str]: r"""Re-stage the Windows ``hermes`` launchers when they vanish. On Windows, ``hermes`` resolves through launchers derived from the venv console scripts — never - ``venv\Scripts`` itself on PATH, which would shadow the user's ``python`` (#83797). - - - canonical managed binary dir: only when *root* is the managed clone (``root.parent == - get_default_hermes_root()``), so source checkouts elsewhere never gain launchers; - legacy - ``\bin``: only when that dir is on the user PATH (registry value, process PATH as - fallback), i.e. + ``venv\Scripts`` itself on PATH, which would shadow the user's ``python``. Targets: the + canonical managed binary dir (only when *root* is the managed clone, so source checkouts + elsewhere never gain launchers) and the legacy ``\bin`` (only while the user PATH still + points at it). Never raises. """ if windows is None: windows = _is_windows() if not windows: return [] - root = Path(root) - home = _default_hermes_root() if home is None: return [] targets: list[Path] = [] - - # Canonical target — gate on the managed-clone shape. This runs at - # every hermes_cli.main process start (right after the profile - # override), so the healthy path must stay at a couple of stat calls. + # Runs at every hermes_cli.main process start, so the healthy path must stay a few stat calls. if _normalize_windows_path(root.parent) == _normalize_windows_path(home): canonical = home / "bin" if _launchers_missing(canonical): targets.append(canonical) - - # Legacy transition target — the pre-migration in-checkout dir. Only - # re-staged while the user PATH still points at it (consent), compared - # as normalized literal strings: the installer wrote the long literal - # path, and realpath'ing arbitrary PATH entries could hang on dead - # network shares. An entry stored some other way (8.3 short path, - # subst drive) misses the re-stage, which fails safe: no-op. + # Legacy target: compared as normalized literal strings — the installer wrote the long literal + # path, and realpath'ing arbitrary PATH entries could hang on dead network shares. An entry + # stored another way (8.3 short path, subst drive) misses the re-stage, which fails safe. legacy = root / "bin" if _launchers_missing(legacy): if user_path_entries is None: @@ -223,7 +189,6 @@ def ensure_windows_bin_launchers( configured = {_normalize_windows_path(entry) for entry in user_path_entries} if _normalize_windows_path(legacy) in configured: targets.append(legacy) - if not targets: return [] @@ -233,11 +198,8 @@ def ensure_windows_bin_launchers( if venv_dir is None: return [] scripts_dir = venv_bin_dir(venv_dir, windows=windows) - sources = [ - (name, scripts_dir / f"{name}.exe") - for name in _WINDOWS_BIN_LAUNCHERS - if (scripts_dir / f"{name}.exe").is_file() - ] + sources = [(name, scripts_dir / f"{name}.exe") for name in _WINDOWS_BIN_LAUNCHERS + if (scripts_dir / f"{name}.exe").is_file()] if not sources: return [] relocatable = _venv_is_relocatable(venv_dir) @@ -255,9 +217,7 @@ def ensure_windows_bin_launchers( staging = target / f"{final.name}.heal.{os.getpid()}" try: if relocatable: - staging.write_text( - "@echo off\r\n" f'"{source}" %*\r\n', encoding="ascii" - ) + staging.write_text("@echo off\r\n" f'"{source}" %*\r\n', encoding="ascii") else: shutil.copy2(source, staging) os.replace(staging, final) @@ -266,22 +226,14 @@ def ensure_windows_bin_launchers( with contextlib.suppress(OSError): staging.unlink() if restored: - # Guarded like everything else in this never-raises helper: a - # closed/broken stderr must not turn a successful heal into a crash. + # A closed/broken stderr must not turn a successful heal into a crash. with contextlib.suppress(OSError, ValueError): - print( - " ✓ Restored hermes launcher(s): " + ", ".join(restored), - file=sys.stderr, - ) + print(" ✓ Restored hermes launcher(s): " + ", ".join(restored), file=sys.stderr) return restored def _read_user_path_raw() -> tuple[list[str], int]: - """Raw (unexpanded) user PATH entries + registry value type. - - Raw so a rewrite preserves ``%VARS%`` exactly as the user stored them (same discipline as - ``hermes_cli.uninstall``). Only called on Windows. - """ + """Raw (unexpanded) user PATH entries + registry value type (a rewrite preserves ``%VARS%``).""" import winreg with winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Environment") as key: @@ -296,51 +248,28 @@ def _write_user_path_raw(entries: list[str], kind: int) -> None: """Write the user PATH back, preserving the registry value type.""" import winreg - with winreg.OpenKey( - winreg.HKEY_CURRENT_USER, "Environment", 0, winreg.KEY_READ | winreg.KEY_WRITE - ) as key: + with winreg.OpenKey(winreg.HKEY_CURRENT_USER, "Environment", 0, + winreg.KEY_READ | winreg.KEY_WRITE) as key: winreg.SetValueEx(key, "Path", 0, kind, ";".join(entries)) def migrate_windows_bin_path( - root, - *, - windows: bool | None = None, - read_user_path=None, - write_user_path=None, + root, *, windows: bool | None = None, read_user_path=None, write_user_path=None, ) -> bool: - """One-time PATH migration to the ``HERMES_HOME\\bin`` launcher layout. - - Runs from the ``hermes update`` tail (and mirrors what install.ps1's - Set-PathVariable does on fresh installs/repairs, which never reach - existing installs — updates don't run install.ps1): - - 1. stage the launcher copies into the managed binary dir (via - :func:`ensure_windows_bin_launchers`); - 2. verify both launchers are present there — otherwise STOP, leaving - the user PATH untouched (never strip a working entry before its - replacement is proven); - 3. ensure the managed binary dir is on the user PATH (prepend); - 4. strip the legacy entries: ``\\bin`` (in-checkout launcher dir - the update autostash could sweep) and ``\\venv\\Scripts`` - (shadowed the user's ``python``, #83797). - - The legacy ``\\bin`` FILES are deliberately left in place: editor - and ACP configs that captured absolute launcher paths keep working - (the launchers run fine from there — only PATH resolution through a - dir git could sweep was the bug), and the dir is git-ignored so it - cannot dirty the tree. - - Registry writes preserve the stored value type and raw ``%VARS%``. - Never raises; returns True when the canonical layout is in place. + """One-time PATH migration to the ``HERMES_HOME\\bin`` launcher layout (``hermes update`` tail). + 1. stage launchers into the managed binary dir; 2. verify both are present — otherwise STOP, + leaving the user PATH untouched (never strip a working entry before its replacement is proven); + 3. prepend the managed binary dir to the user PATH; 4. strip the legacy ``\\bin`` and + ``\\venv\\Scripts`` entries. Legacy ``\\bin`` FILES stay: configs that captured + absolute launcher paths keep working and the dir is git-ignored. Registry writes preserve the + stored value type and raw ``%VARS%``. Never raises; True when the canonical layout is in place. *read_user_path*/*write_user_path* are injectable for tests. """ if windows is None: windows = _is_windows() if not windows: return False - root = Path(root) from hermes_constants import venv_bin_dir @@ -352,19 +281,15 @@ def migrate_windows_bin_path( return False # not the managed clone — nothing to migrate ensure_windows_bin_launchers(root, windows=windows, user_path_entries=[]) - home_bin = home / "bin" - if any( - not ((home_bin / f"{name}.exe").is_file() or (home_bin / f"{name}.cmd").is_file()) - for name in _WINDOWS_BIN_LAUNCHERS - ): + if any(not ((home_bin / f"{name}.exe").is_file() or (home_bin / f"{name}.cmd").is_file()) + for name in _WINDOWS_BIN_LAUNCHERS): return False # staging incomplete — leave the PATH alone if read_user_path is None: read_user_path = _read_user_path_raw if write_user_path is None: write_user_path = _write_user_path_raw - try: entries, kind = read_user_path() except (OSError, ImportError): @@ -372,10 +297,8 @@ def migrate_windows_bin_path( legacy_keys = { _normalize_windows_path(root / "bin"), - # The pre-#83797 installer put the venv's Scripts dir itself on PATH, - # always at the literal `venv` layout (never `.venv`) — this strips - # that stale entry, so it must match what the installer wrote then, - # not where the venv lives now. + # The old installer put the venv's Scripts dir itself on PATH, always at the literal + # `venv` layout (never `.venv`) — match what it wrote then, not where the venv lives now. _normalize_windows_path(venv_bin_dir(root / "venv", windows=True)), } home_bin_key = _normalize_windows_path(home_bin) @@ -384,21 +307,16 @@ def migrate_windows_bin_path( return _normalize_windows_path(os.path.expandvars(entry)) kept = [e for e in entries if _entry_key(e) not in legacy_keys] - have_home_bin = any(_entry_key(e) == home_bin_key for e in kept) - if not have_home_bin: + if not any(_entry_key(e) == home_bin_key for e in kept): kept = [str(home_bin)] + kept - if kept != entries: try: write_user_path(kept, kind) except (OSError, ImportError): return False with contextlib.suppress(OSError, ValueError): - print( - f" ✓ hermes launchers now resolve from {home_bin} " - "(legacy PATH entries removed)", - file=sys.stderr, - ) + print(f" ✓ hermes launchers now resolve from {home_bin} " + "(legacy PATH entries removed)", file=sys.stderr) return True @@ -413,18 +331,16 @@ def _load_console_script_names(root: Path) -> list[str]: class ShimQuarantineError(RuntimeError): - """A live shim could not be renamed aside — the venv is contended (#87331). + """A live shim could not be renamed aside — the venv is contended. - Raised BEFORE the install command runs. Callers (early-pass recovery, core-marker recovery) - catch it like any install failure: the update-incomplete marker survives and a later launch - retries once the holder exits — the contended venv is never mutated. + Raised BEFORE the install command runs. Callers catch it like any install failure: the + update-incomplete marker survives and a later launch retries once the holder exits — the + contended venv is never mutated. """ 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 _quarantine_running_hermes_exe( @@ -433,16 +349,13 @@ def _quarantine_running_hermes_exe( """Rename live hermes*.exe shims aside so the installer can rewrite them. Windows blocks REPLACE on a running .exe but allows RENAME. Best-effort: silently skips anything - that cannot be renamed. Returns (original, quarantined) pairs. stdlib-only — the console-script - set comes from pyproject ``[project.scripts]`` (fallback: the well-known trio). + that cannot be renamed (names appended to *failed_out*). Returns (original, quarantined) pairs. + The console-script set comes from pyproject ``[project.scripts]`` (fallback: well-known trio). """ if not _is_windows(): return [] names = set(_load_console_script_names(scripts_dir.parent.parent)) or { - "hermes", - "hermes-agent", - "hermes-acp", - } + "hermes", "hermes-agent", "hermes-acp"} names.add("hermes-gateway") moved: list[tuple[Path, Path]] = [] for name in sorted(names): @@ -460,42 +373,30 @@ def _quarantine_running_hermes_exe( def _restore_quarantined_exes(moved: list[tuple[Path, Path]]) -> None: - """Put quarantined shims back when the installer did not replace them. - - Delegates to the shared helper in the stdlib-only ``_early_recovery`` module: one retry ladder - and one recovery message for every restore site, instead of the near-identical copies that had - already drifted (#75584). - """ + """Put quarantined shims back when the installer did not replace them (shared retry ladder).""" _er.restore_quarantined_shims(moved) def _run_install_cmd(cmd: list[str], *, env: dict | None, root: Path) -> None: """Run an install command with quarantine protection for venv shims. - Fail-closed (#87331): when any live shim cannot be renamed aside, the venv is contended and the + Fail-closed: when any live shim cannot be renamed aside, the venv is contended and the installer would die partway on the same locks — raise :class:`ShimQuarantineError` WITHOUT - running it. The caller's marker-keeping failure handling turns that into "retry next launch". - - Raises CalledProcessError on install failure (callers implement the per-extra fallback ladder). + running it. Raises CalledProcessError on install failure (callers implement the per-extra + fallback ladder). """ scripts_dir = _venv_scripts_dir(root) if _is_windows() else None failed: list[str] = [] - moved = ( - _quarantine_running_hermes_exe(scripts_dir, failed_out=failed) - if scripts_dir - else [] - ) + moved = _quarantine_running_hermes_exe(scripts_dir, failed_out=failed) if scripts_dir else [] if failed: _restore_quarantined_exes(moved) raise ShimQuarantineError(failed) try: subprocess.run(cmd, cwd=root, check=True, env=env) finally: - # Restore runs on success AND failure: a SUCCESSFUL install can still - # skip the entry-points step entirely (uv audits an already-satisfied - # editable install as a no-op and rewrites nothing), which would leave - # the quarantined shims renamed aside and `hermes` gone from PATH - # (#75584). _restore_quarantined_exes only renames back when the + # Restore on success AND failure: a SUCCESSFUL install can skip the entry-points step + # entirely (uv audits an already-satisfied editable install as a no-op), which would leave + # the shims renamed aside and `hermes` gone from PATH. Restore only renames back when the # installer did NOT write a fresh shim, so this is safe in both cases. if scripts_dir is not None: _restore_quarantined_exes(moved) @@ -521,67 +422,44 @@ def _load_installable_optional_extras(root: Path, group: str) -> list[str]: def run_core_install(root: Path) -> None: """Full core ``.[all]`` editable reinstall — the recovery install. - Equal in behavior to the install half of ``main.py::_recover_core_update_marker_locked``: - - - bootstrap pip via ensurepip (a killed install can leave the venv with no pip module at all) - - prefer ``uv pip`` with VIRTUAL_ENV pointed at the project venv; fall back to ``python -m pip`` - when no uv binary is available - target ``.[all]`` (or ``.[termux-all]`` on Termux) with the - per-extra fallback ladder when the combined extras resolve fails - quarantine live - ``hermes*.exe`` shims on Windows so they can be replaced - route ALL install output to stderr - (acp/JSON-RPC safety) - Termux strips leaked PYTHONPATH/PYTHONHOME from the uv env + ensurepip bootstrap; ``uv pip`` with VIRTUAL_ENV at the project venv (``python -m pip`` + fallback); ``.[all]`` (``.[termux-all]`` on Termux) with the per-extra fallback ladder when the + combined resolve fails; live shims quarantined on Windows; ALL output routed to stderr. """ prefix, env = _resolve_install_target(root) group = "termux-all" if _is_termux_env(env) else "all" + def install(target: str) -> None: + _run_install_cmd(prefix + ["install", "-e", target], env=env, root=root) + with _stdout_to_stderr(): _er._run_ensurepip(root) - try: - _run_install_cmd( - prefix + ["install", "-e", f".[{group}]"], env=env, root=root - ) + install(f".[{group}]") return except subprocess.CalledProcessError: - print( - " ⚠ Optional extras failed, reinstalling base dependencies " - "and retrying extras individually..." - ) - - _run_install_cmd(prefix + ["install", "-e", "."], env=env, root=root) - + print(" ⚠ Optional extras failed, reinstalling base dependencies " + "and retrying extras individually...") + install(".") failed_extras: list[str] = [] installed_extras: list[str] = [] for extra in _load_installable_optional_extras(root, group): try: - _run_install_cmd( - prefix + ["install", "-e", f".[{extra}]"], env=env, root=root - ) + install(f".[{extra}]") installed_extras.append(extra) except subprocess.CalledProcessError: failed_extras.append(extra) if installed_extras: - print( - " ✓ Reinstalled optional extras individually: " - + ", ".join(installed_extras) - ) + print(" ✓ Reinstalled optional extras individually: " + ", ".join(installed_extras)) if failed_extras: - print( - " ⚠ Skipped optional extras that still failed: " - + ", ".join(failed_extras) - ) - - -# --------------------------------------------------------------------------- -# Marker metadata (attempt counter for early-pass retry backoff) -# --------------------------------------------------------------------------- + print(" ⚠ Skipped optional extras that still failed: " + ", ".join(failed_extras)) def bump_marker_attempts(marker_path: Path) -> int: - """Increment an attempts counter stored inside the marker file. + """Increment the attempts counter stored inside the marker file's JSON body. - The marker's existence is the signal; opportunistic JSON body carries the retry count so a - persistently failing install can back off instead of reinstall-hammering every launch. - Corrupt/missing bodies restart at 1. Returns the new attempt count. Never raises. + The marker's existence is the signal; the body carries the retry count so a persistently + failing install can back off. Corrupt/missing bodies restart at 1. Never raises. """ attempts = _er._read_marker_attempts(marker_path) + 1 with contextlib.suppress(OSError): diff --git a/hermes_cli/_parser.py b/hermes_cli/_parser.py index 577568ad85..3788a2418e 100644 --- a/hermes_cli/_parser.py +++ b/hermes_cli/_parser.py @@ -8,34 +8,19 @@ gateway, sessions, …) is built by ``hermes_cli/subcommands/.py`` and wi import argparse from functools import lru_cache +# `--profile` / `-p` is consumed by ``main._apply_profile_override`` before argparse runs +# (it sets ``HERMES_HOME`` and strips itself from ``sys.argv``), so it isn't on the parser. +# Listed here so all "carry over on relaunch" metadata lives in one file. +PRE_ARGPARSE_INHERITED_FLAGS: list[tuple[str, bool]] = [("--profile", True), ("-p", True)] -# `--profile` / `-p` is consumed by ``main._apply_profile_override`` before -# argparse runs (it sets ``HERMES_HOME`` and strips itself from ``sys.argv``), -# so it isn't on the parser. Listed here so all "carry over on relaunch" -# metadata lives in one file. -PRE_ARGPARSE_INHERITED_FLAGS: list[tuple[str, bool]] = [ - ("--profile", True), - ("-p", True), -] - - -# Static snapshot fallback for ``top_level_value_flag_sets`` — used only if -# introspecting the live parser fails (e.g. argparse surface broken mid-edit). -# The derived path is authoritative; a parity test in -# tests/hermes_cli/test_top_level_value_flags_parity.py fails CI if the parser -# grows a value-taking flag this snapshot lacks AND derivation regresses. -_VALUE_FLAGS_FALLBACK: frozenset[str] = frozenset( - { - "-z", "--oneshot", - "-m", "--model", - "--provider", "--reasoning", - "-t", "--toolsets", - "-r", "--resume", - "-s", "--skills", - "--usage-file", - "--in", - } -) +# Static snapshot fallback for ``top_level_value_flag_sets`` — used only if introspecting the +# live parser fails. The derived path is authoritative; tests/hermes_cli/ +# test_top_level_value_flags_parity.py fails CI if the parser grows a value-taking flag this +# snapshot lacks AND derivation regresses. +_VALUE_FLAGS_FALLBACK: frozenset[str] = frozenset({ + "-z", "--oneshot", "-m", "--model", "--provider", "--reasoning", "-t", "--toolsets", + "-r", "--resume", "-s", "--skills", "--usage-file", "--in", +}) _OPTIONAL_VALUE_FLAGS_FALLBACK: frozenset[str] = frozenset({"-c", "--continue"}) @@ -45,8 +30,8 @@ def top_level_value_flag_sets() -> tuple[frozenset[str], frozenset[str]]: Introspects ``build_top_level_parser()`` (every option with nargs != 0) so the argv scanners in ``main.py`` (``_first_positional_argv``, ``_apply_profile_override``) can never drift from the - argparse surface — the exact drift that made ``hermes --reasoning high chat …`` misread ``high`` - as the subcommand and forced eager plugin discovery (#93530). + argparse surface — the drift that made ``hermes --reasoning high chat …`` misread ``high`` as + the subcommand and forced eager plugin discovery. """ try: parser = build_top_level_parser()[0] @@ -63,11 +48,7 @@ def top_level_value_flag_sets() -> tuple[frozenset[str], frozenset[str]]: def _inherited_flag(parser, *args, **kwargs): - """Register a flag that ``hermes_cli.relaunch`` carries over when the CLI re-execs itself. - - ``parser.add_argument(...)`` plus tagging the Action with ``inherit_on_relaunch = True`` so - the relaunch table builder can find it via introspection. - """ + """``parser.add_argument`` + tag the Action ``inherit_on_relaunch`` for ``hermes_cli.relaunch``.""" action = parser.add_argument(*args, **kwargs) action.inherit_on_relaunch = True return action @@ -122,462 +103,172 @@ For more help on a command: def _add_top_level_flags(parser: argparse.ArgumentParser) -> None: """Top-level (pre-subcommand) flags; ``-m/--provider`` pair with ``-z`` without ``chat``.""" - parser.add_argument( - "--version", "-V", action="store_true", help="Show version and exit" - ) - parser.add_argument( - "-z", - "--oneshot", - metavar="PROMPT", - default=None, - help=( - "One-shot mode: send a single prompt and print ONLY the final " - "response text to stdout. No banner, no spinner, no tool " - "previews, no session_id line. Tools, memory, rules, and " - "AGENTS.md in the CWD are loaded as normal; approvals are " - "auto-bypassed. Intended for scripts / pipes." - ), - ) - parser.add_argument( - "--usage-file", - metavar="PATH", - default=None, - help=( - "One-shot mode only: after the run, write a JSON usage report " - "(estimated cost, token counts, model, api_calls) to PATH. " - "The report is written even when the run fails, so pipelines " - "can always account for spend. No effect outside -z/--oneshot." - ), - ) - # --model / --provider are accepted at the top level so they can pair - # with -z without needing the `chat` subcommand. If neither -z nor a - # subcommand consumes them, they fall through harmlessly as None. - # Mirrors `hermes chat --model ... --provider ...` semantics. - _inherited_flag( - parser, - "-m", - "--model", - default=None, - help=( - "Model override for this invocation (e.g. anthropic/claude-sonnet-4.6). " - "Applies to -z/--oneshot and --tui. Also settable via HERMES_INFERENCE_MODEL env var." - ), - ) - _inherited_flag( - parser, - "--provider", - default=None, - help=( - "Provider override for this invocation (e.g. openrouter, anthropic). " - "Applies to -z/--oneshot and --tui. The persistent provider lives in config.yaml " - "under model.provider — use `hermes setup` or edit the file to change it." - ), - ) - _inherited_flag( - parser, - "--reasoning", - default=None, - metavar="LEVEL", - help=( - "Reasoning effort for this invocation: none, minimal, low, medium, " - "high, xhigh, max, or ultra. Overrides agent.reasoning_effort in " - "config.yaml for this run only; the persistent level lives there " - "(or per-model under agent.reasoning_overrides)." - ), - ) - parser.add_argument( - "-t", - "--toolsets", - default=None, - help="Comma-separated toolsets to enable for this invocation. Applies to -z/--oneshot and --tui.", - ) - parser.add_argument( - "--resume", - "-r", - metavar="SESSION", - default=None, - help=( - "Resume a previous session by ID or title, or pass 'latest' for " - "the most recent session (workspace-scoped, like -c with no name)" - ), - ) - parser.add_argument( - "--no-restore-cwd", - action="store_true", - default=False, - help="Don't cd into a resumed session's recorded working directory.", - ) - parser.add_argument( - "--in", - dest="in_dir", - metavar="DIR", - default=None, - help=( - "Change into DIR before starting or resuming. Combined with " - "'--resume latest' or -c, the most recent session for DIR's " - "workspace is picked, and the session stays in DIR (skips the " - "recorded-cwd restore)." - ), - ) - parser.add_argument( - "--continue", - "-c", - dest="continue_last", - nargs="?", - const=True, - default=None, - metavar="SESSION_NAME", - help="Resume a session by name, or the most recent if no name given", - ) - parser.add_argument( - "--worktree", - "-w", - action="store_true", - default=False, - help="Run in an isolated git worktree (for parallel agents)", - ) - _inherited_flag( - parser, - "--accept-hooks", - action="store_true", - default=False, - help=( - "Auto-approve any unseen shell hooks declared in config.yaml " - "without a TTY prompt. Equivalent to HERMES_ACCEPT_HOOKS=1 or " - "hooks_auto_accept: true in config.yaml. Use on CI / headless " - "runs that can't prompt." - ), - ) - _inherited_flag( - parser, - "--skills", - "-s", - action="append", - default=None, - help="Preload one or more skills for the session (repeat flag or comma-separate)", - ) - _inherited_flag( - parser, - "--yolo", - action="store_true", - default=False, - help="Bypass all dangerous command approval prompts (use at your own risk)", - ) - _inherited_flag( - parser, - "--pass-session-id", - action="store_true", - default=False, - help="Include the session ID in the agent's system prompt", - ) - _inherited_flag( - parser, - "--ignore-user-config", - action="store_true", - default=False, - help="Ignore ~/.hermes/config.yaml and fall back to built-in defaults (credentials in .env are still loaded)", - ) - _inherited_flag( - parser, - "--ignore-rules", - action="store_true", - default=False, - help="Skip auto-injection of AGENTS.md, SOUL.md, .cursorrules, memory, and preloaded skills", - ) - _inherited_flag( - parser, - "--safe-mode", - action="store_true", - default=False, - help="Troubleshooting mode: disable ALL customizations — user config, AGENTS.md/memory injection, plugins, and MCP servers (implies --ignore-user-config and --ignore-rules)", - ) - _inherited_flag( - parser, - "--tui", - action="store_true", - default=False, - help="Launch the modern TUI instead of the classic REPL", - ) - _inherited_flag( - parser, - "--cli", - action="store_true", - default=False, - help="Force the classic prompt_toolkit REPL (overrides display.interface=tui)", - ) - _inherited_flag( - parser, - "--dev", - dest="tui_dev", - action="store_true", - default=False, - help="With --tui: run TypeScript sources via tsx (skip dist build)", - ) + add, inherited = parser.add_argument, _inherited_flag + add("--version", "-V", action="store_true", help="Show version and exit") + add("-z", "--oneshot", metavar="PROMPT", default=None, help=( + "One-shot mode: send a single prompt and print ONLY the final " + "response text to stdout. No banner, no spinner, no tool " + "previews, no session_id line. Tools, memory, rules, and " + "AGENTS.md in the CWD are loaded as normal; approvals are " + "auto-bypassed. Intended for scripts / pipes.")) + add("--usage-file", metavar="PATH", default=None, help=( + "One-shot mode only: after the run, write a JSON usage report " + "(estimated cost, token counts, model, api_calls) to PATH. " + "The report is written even when the run fails, so pipelines " + "can always account for spend. No effect outside -z/--oneshot.")) + # --model / --provider are accepted at the top level so they can pair with -z without the + # `chat` subcommand; if neither -z nor a subcommand consumes them, they fall through as None. + inherited(parser, "-m", "--model", default=None, help=( + "Model override for this invocation (e.g. anthropic/claude-sonnet-4.6). " + "Applies to -z/--oneshot and --tui. Also settable via HERMES_INFERENCE_MODEL env var.")) + inherited(parser, "--provider", default=None, help=( + "Provider override for this invocation (e.g. openrouter, anthropic). " + "Applies to -z/--oneshot and --tui. The persistent provider lives in config.yaml " + "under model.provider — use `hermes setup` or edit the file to change it.")) + inherited(parser, "--reasoning", default=None, metavar="LEVEL", help=( + "Reasoning effort for this invocation: none, minimal, low, medium, " + "high, xhigh, max, or ultra. Overrides agent.reasoning_effort in " + "config.yaml for this run only; the persistent level lives there " + "(or per-model under agent.reasoning_overrides).")) + add("-t", "--toolsets", default=None, + help="Comma-separated toolsets to enable for this invocation. Applies to -z/--oneshot and --tui.") + add("--resume", "-r", metavar="SESSION", default=None, help=( + "Resume a previous session by ID or title, or pass 'latest' for " + "the most recent session (workspace-scoped, like -c with no name)")) + add("--no-restore-cwd", action="store_true", default=False, + help="Don't cd into a resumed session's recorded working directory.") + add("--in", dest="in_dir", metavar="DIR", default=None, help=( + "Change into DIR before starting or resuming. Combined with " + "'--resume latest' or -c, the most recent session for DIR's " + "workspace is picked, and the session stays in DIR (skips the " + "recorded-cwd restore).")) + add("--continue", "-c", dest="continue_last", nargs="?", const=True, default=None, + metavar="SESSION_NAME", help="Resume a session by name, or the most recent if no name given") + add("--worktree", "-w", action="store_true", default=False, + help="Run in an isolated git worktree (for parallel agents)") + inherited(parser, "--accept-hooks", action="store_true", default=False, help=( + "Auto-approve any unseen shell hooks declared in config.yaml " + "without a TTY prompt. Equivalent to HERMES_ACCEPT_HOOKS=1 or " + "hooks_auto_accept: true in config.yaml. Use on CI / headless " + "runs that can't prompt.")) + inherited(parser, "--skills", "-s", action="append", default=None, + help="Preload one or more skills for the session (repeat flag or comma-separate)") + inherited(parser, "--yolo", action="store_true", default=False, + help="Bypass all dangerous command approval prompts (use at your own risk)") + inherited(parser, "--pass-session-id", action="store_true", default=False, + help="Include the session ID in the agent's system prompt") + inherited(parser, "--ignore-user-config", action="store_true", default=False, + help="Ignore ~/.hermes/config.yaml and fall back to built-in defaults (credentials in .env are still loaded)") + inherited(parser, "--ignore-rules", action="store_true", default=False, + help="Skip auto-injection of AGENTS.md, SOUL.md, .cursorrules, memory, and preloaded skills") + inherited(parser, "--safe-mode", action="store_true", default=False, + help="Troubleshooting mode: disable ALL customizations — user config, AGENTS.md/memory injection, plugins, and MCP servers (implies --ignore-user-config and --ignore-rules)") + inherited(parser, "--tui", action="store_true", default=False, + help="Launch the modern TUI instead of the classic REPL") + inherited(parser, "--cli", action="store_true", default=False, + help="Force the classic prompt_toolkit REPL (overrides display.interface=tui)") + inherited(parser, "--dev", dest="tui_dev", action="store_true", default=False, + help="With --tui: run TypeScript sources via tsx (skip dist build)") def _build_chat_parser(subparsers) -> argparse.ArgumentParser: - """The ``chat`` subparser (also the implicit default command).""" + """The ``chat`` subparser (also the implicit default command). + + Flags ALSO declared on the top-level parser use ``default=argparse.SUPPRESS``: for + ``hermes -m foo chat`` argparse first sets ``args.model`` from the top-level parser, then + dispatches to the chat subparser, which shares the namespace and ``dest`` — a plain ``None`` + default would silently clobber the top-level value. SUPPRESS keeps the subparser action a no-op + unless the flag is actually passed after the subcommand (tests/hermes_cli/ + test_argparse_flag_propagation.py). + """ chat_parser = subparsers.add_parser( - "chat", - help="Interactive chat with the agent", - description="Start an interactive chat session with Hermes Agent", - ) + "chat", help="Interactive chat with the agent", + description="Start an interactive chat session with Hermes Agent") + add, inherited, SUPPRESS = chat_parser.add_argument, _inherited_flag, argparse.SUPPRESS _query_group = chat_parser.add_mutually_exclusive_group() - _query_group.add_argument( - "-q", "--query", - help=( - "Query to run. On a real TTY the prompt seeds an interactive " - "session (submitted literally as the first turn); combined with " - "--oneshot or -Q, or on a non-TTY, it answers and exits." - ), - ) - _query_group.add_argument( - "--query-file", - metavar="PATH", - help=( - "Read the single query from a file instead of the command line " - "('-' reads stdin). Safe for arbitrary text: nothing is shell-" - "interpreted, so quotes, $(...), and backticks are preserved " - "verbatim. Mutually exclusive with -q." - ), - ) - chat_parser.add_argument( - "--oneshot", - dest="oneshot_exit", - action="store_true", - # Distinct dest: the top-level `-z/--oneshot PROMPT` is value-taking - # and its dispatch sites do `if args.oneshot: _run_and_exit_oneshot( - # args.oneshot)` — a shared boolean dest would be passed as the - # prompt. `oneshot_exit` keeps the surfaces independent. - default=False, - help=( - "With -q/--query-file: answer the query and exit (legacy " - "single-query behavior) instead of seeding an interactive " - "session. Implied on non-TTY stdio and by -Q/--quiet." - ), - ) - chat_parser.add_argument( - "--image", help="Optional local image path to attach to a single query" - ) - # `default=argparse.SUPPRESS` on flags that are ALSO declared on the - # top-level parser: when the user writes `hermes -m foo chat`, argparse - # first sets `args.model = "foo"` from the top-level parser, then - # dispatches to the chat subparser. Without SUPPRESS the chat subparser's - # own default (`None`) would silently clobber the top-level value because - # the subparser shares the same namespace and `dest`. SUPPRESS keeps the - # subparser action a no-op unless the user actually passes the flag after - # the subcommand. Matches the pattern already used for `-s/--skills` and - # the relaunch-inherited flags `-r/--resume`, `-c/--continue`, - # `-w/--worktree`, `--yolo`, etc. (see tests/hermes_cli/ - # test_argparse_flag_propagation.py). - _inherited_flag( - chat_parser, - "-m", "--model", - default=argparse.SUPPRESS, - help="Model to use (e.g., anthropic/claude-sonnet-4)", - ) - chat_parser.add_argument( - "-t", "--toolsets", - default=argparse.SUPPRESS, - help="Comma-separated toolsets to enable", - ) - _inherited_flag( - chat_parser, - "--reasoning", - default=argparse.SUPPRESS, - metavar="LEVEL", - help=( - "Reasoning effort for this session: none, minimal, low, medium, " - "high, xhigh, max, or ultra. Overrides agent.reasoning_effort for " - "this run only (same levels as the /reasoning slash command)." - ), - ) - _inherited_flag( - chat_parser, - "-s", - "--skills", - action="append", - default=argparse.SUPPRESS, - help="Preload one or more skills for the session (repeat flag or comma-separate)", - ) - _inherited_flag( - chat_parser, - "--provider", - # No `choices=` here: user-defined providers from config.yaml `providers:` - # are also valid values, and runtime resolution (resolve_runtime_provider) - # handles validation/error reporting consistently with the top-level - # `--provider` flag. - default=argparse.SUPPRESS, - help="Inference provider (default: auto). Built-in or a user-defined name from `providers:` in config.yaml.", - ) - chat_parser.add_argument( - "-v", - "--verbose", - action="store_true", - default=argparse.SUPPRESS, - help="Verbose output", - ) - chat_parser.add_argument( - "-Q", - "--quiet", - action="store_true", - help="Quiet mode for programmatic use: suppress banner, spinner, and tool previews. Only output the final response and session info.", - ) - chat_parser.add_argument( - "--resume", - "-r", - metavar="SESSION_ID", - default=argparse.SUPPRESS, - help=( - "Resume a previous session by ID (shown on exit), or 'latest' " - "for the most recent session" - ), - ) - chat_parser.add_argument( - "--no-restore-cwd", - action="store_true", - default=argparse.SUPPRESS, - help="Don't cd into a resumed session's recorded working directory.", - ) - chat_parser.add_argument( - "--in", - dest="in_dir", - metavar="DIR", - default=argparse.SUPPRESS, - help=( - "Change into DIR before starting or resuming (scopes " - "'--resume latest' / -c lookups to DIR's workspace)." - ), - ) - chat_parser.add_argument( - "--continue", - "-c", - dest="continue_last", - nargs="?", - const=True, - default=argparse.SUPPRESS, - metavar="SESSION_NAME", - help="Resume a session by name, or the most recent if no name given", - ) - chat_parser.add_argument( - "--create-if-missing", - action="store_true", - default=argparse.SUPPRESS, - help=( - "With -c/--continue : if no session matches the name, " - "create a new session with that title and proceed (instead of " - "failing with a not-found error). Programmatic callers that " - "want 'send to this named thread, making it if needed'." - ), - ) - chat_parser.add_argument( - "--worktree", - "-w", - action="store_true", - default=argparse.SUPPRESS, - help="Run in an isolated git worktree (for parallel agents on the same repo)", - ) - _inherited_flag( - chat_parser, - "--accept-hooks", - action="store_true", - default=argparse.SUPPRESS, - help=( - "Auto-approve any unseen shell hooks declared in config.yaml " - "without a TTY prompt (see also HERMES_ACCEPT_HOOKS env var and " - "hooks_auto_accept: in config.yaml)." - ), - ) - chat_parser.add_argument( - "--checkpoints", - action="store_true", - default=False, - help="Enable filesystem checkpoints before destructive file operations (use /rollback to restore)", - ) - chat_parser.add_argument( - "--max-turns", - type=int, - default=None, - metavar="N", - help="Maximum tool-calling iterations per conversation turn (default: 500, or agent.max_turns in config)", - ) - chat_parser.add_argument( - "--run-budget", - type=float, - default=None, - metavar="SECONDS", - dest="run_budget", - help=( - "Optional wall-clock budget in seconds for each conversation run. " - "At 80%% elapsed the agent gets a one-time wrap-up notice, and " - "implicit provider stale timeouts are capped to the remaining " - "budget so one hung call can't consume the run. Unset = off. " - "Also configurable as agent.run_budget_seconds in config.yaml. " - "Intended for one-shot/eval invocations with a hard ceiling." - ), - ) - _inherited_flag( - chat_parser, - "--yolo", - action="store_true", - default=argparse.SUPPRESS, - help="Bypass all dangerous command approval prompts (use at your own risk)", - ) - _inherited_flag( - chat_parser, - "--pass-session-id", - action="store_true", - default=argparse.SUPPRESS, - help="Include the session ID in the agent's system prompt", - ) - _inherited_flag( - chat_parser, - "--ignore-user-config", - action="store_true", - default=argparse.SUPPRESS, - help="Ignore ~/.hermes/config.yaml and fall back to built-in defaults (credentials in .env are still loaded). Useful for isolated CI runs, reproduction, and third-party integrations.", - ) - _inherited_flag( - chat_parser, - "--ignore-rules", - action="store_true", - default=argparse.SUPPRESS, - help="Skip auto-injection of AGENTS.md, SOUL.md, .cursorrules, memory, and preloaded skills. Combine with --ignore-user-config for a fully isolated run.", - ) - _inherited_flag( - chat_parser, - "--safe-mode", - action="store_true", - default=argparse.SUPPRESS, - help="Troubleshooting mode: disable ALL customizations — user config, AGENTS.md/memory injection, plugins, and MCP servers (implies --ignore-user-config and --ignore-rules). Use to isolate whether a problem comes from your setup or from Hermes itself.", - ) - chat_parser.add_argument( - "--source", - default=None, - help="Session source tag for filtering (default: cli). Use 'tool' for third-party integrations that should not appear in user session lists.", - ) - _inherited_flag( - chat_parser, - "--tui", - action="store_true", - default=argparse.SUPPRESS, - help="Launch the modern TUI instead of the classic REPL", - ) - _inherited_flag( - chat_parser, - "--cli", - action="store_true", - default=argparse.SUPPRESS, - help="Force the classic prompt_toolkit REPL (overrides display.interface=tui)", - ) - _inherited_flag( - chat_parser, - "--dev", - dest="tui_dev", - action="store_true", - default=argparse.SUPPRESS, - help="With --tui: run TypeScript sources via tsx (skip dist build)", - ) + _query_group.add_argument("-q", "--query", help=( + "Query to run. On a real TTY the prompt seeds an interactive " + "session (submitted literally as the first turn); combined with " + "--oneshot or -Q, or on a non-TTY, it answers and exits.")) + _query_group.add_argument("--query-file", metavar="PATH", help=( + "Read the single query from a file instead of the command line " + "('-' reads stdin). Safe for arbitrary text: nothing is shell-" + "interpreted, so quotes, $(...), and backticks are preserved " + "verbatim. Mutually exclusive with -q.")) + # Distinct dest: the top-level `-z/--oneshot PROMPT` is value-taking and its dispatch sites do + # `if args.oneshot: _run_and_exit_oneshot(args.oneshot)` — a shared boolean dest would be + # passed as the prompt. + add("--oneshot", dest="oneshot_exit", action="store_true", default=False, help=( + "With -q/--query-file: answer the query and exit (legacy " + "single-query behavior) instead of seeding an interactive " + "session. Implied on non-TTY stdio and by -Q/--quiet.")) + add("--image", help="Optional local image path to attach to a single query") + inherited(chat_parser, "-m", "--model", default=SUPPRESS, + help="Model to use (e.g., anthropic/claude-sonnet-4)") + add("-t", "--toolsets", default=SUPPRESS, help="Comma-separated toolsets to enable") + inherited(chat_parser, "--reasoning", default=SUPPRESS, metavar="LEVEL", help=( + "Reasoning effort for this session: none, minimal, low, medium, " + "high, xhigh, max, or ultra. Overrides agent.reasoning_effort for " + "this run only (same levels as the /reasoning slash command).")) + inherited(chat_parser, "-s", "--skills", action="append", default=SUPPRESS, + help="Preload one or more skills for the session (repeat flag or comma-separate)") + # No `choices=` on --provider: user-defined providers from config.yaml `providers:` are valid + # too; runtime resolution (resolve_runtime_provider) validates, same as the top-level flag. + inherited(chat_parser, "--provider", default=SUPPRESS, + help="Inference provider (default: auto). Built-in or a user-defined name from `providers:` in config.yaml.") + add("-v", "--verbose", action="store_true", default=SUPPRESS, help="Verbose output") + add("-Q", "--quiet", action="store_true", + help="Quiet mode for programmatic use: suppress banner, spinner, and tool previews. Only output the final response and session info.") + add("--resume", "-r", metavar="SESSION_ID", default=SUPPRESS, help=( + "Resume a previous session by ID (shown on exit), or 'latest' " + "for the most recent session")) + add("--no-restore-cwd", action="store_true", default=SUPPRESS, + help="Don't cd into a resumed session's recorded working directory.") + add("--in", dest="in_dir", metavar="DIR", default=SUPPRESS, help=( + "Change into DIR before starting or resuming (scopes " + "'--resume latest' / -c lookups to DIR's workspace).")) + add("--continue", "-c", dest="continue_last", nargs="?", const=True, default=SUPPRESS, + metavar="SESSION_NAME", help="Resume a session by name, or the most recent if no name given") + add("--create-if-missing", action="store_true", default=SUPPRESS, help=( + "With -c/--continue : if no session matches the name, " + "create a new session with that title and proceed (instead of " + "failing with a not-found error). Programmatic callers that " + "want 'send to this named thread, making it if needed'.")) + add("--worktree", "-w", action="store_true", default=SUPPRESS, + help="Run in an isolated git worktree (for parallel agents on the same repo)") + inherited(chat_parser, "--accept-hooks", action="store_true", default=SUPPRESS, help=( + "Auto-approve any unseen shell hooks declared in config.yaml " + "without a TTY prompt (see also HERMES_ACCEPT_HOOKS env var and " + "hooks_auto_accept: in config.yaml).")) + add("--checkpoints", action="store_true", default=False, + help="Enable filesystem checkpoints before destructive file operations (use /rollback to restore)") + add("--max-turns", type=int, default=None, metavar="N", + help="Maximum tool-calling iterations per conversation turn (default: 500, or agent.max_turns in config)") + add("--run-budget", type=float, default=None, metavar="SECONDS", dest="run_budget", help=( + "Optional wall-clock budget in seconds for each conversation run. " + "At 80%% elapsed the agent gets a one-time wrap-up notice, and " + "implicit provider stale timeouts are capped to the remaining " + "budget so one hung call can't consume the run. Unset = off. " + "Also configurable as agent.run_budget_seconds in config.yaml. " + "Intended for one-shot/eval invocations with a hard ceiling.")) + inherited(chat_parser, "--yolo", action="store_true", default=SUPPRESS, + help="Bypass all dangerous command approval prompts (use at your own risk)") + inherited(chat_parser, "--pass-session-id", action="store_true", default=SUPPRESS, + help="Include the session ID in the agent's system prompt") + inherited(chat_parser, "--ignore-user-config", action="store_true", default=SUPPRESS, + help="Ignore ~/.hermes/config.yaml and fall back to built-in defaults (credentials in .env are still loaded). Useful for isolated CI runs, reproduction, and third-party integrations.") + inherited(chat_parser, "--ignore-rules", action="store_true", default=SUPPRESS, + help="Skip auto-injection of AGENTS.md, SOUL.md, .cursorrules, memory, and preloaded skills. Combine with --ignore-user-config for a fully isolated run.") + inherited(chat_parser, "--safe-mode", action="store_true", default=SUPPRESS, + help="Troubleshooting mode: disable ALL customizations — user config, AGENTS.md/memory injection, plugins, and MCP servers (implies --ignore-user-config and --ignore-rules). Use to isolate whether a problem comes from your setup or from Hermes itself.") + add("--source", default=None, + help="Session source tag for filtering (default: cli). Use 'tool' for third-party integrations that should not appear in user session lists.") + inherited(chat_parser, "--tui", action="store_true", default=SUPPRESS, + help="Launch the modern TUI instead of the classic REPL") + inherited(chat_parser, "--cli", action="store_true", default=SUPPRESS, + help="Force the classic prompt_toolkit REPL (overrides display.interface=tui)") + inherited(chat_parser, "--dev", dest="tui_dev", action="store_true", default=SUPPRESS, + help="With --tui: run TypeScript sources via tsx (skip dist build)") return chat_parser @@ -589,15 +280,8 @@ def build_top_level_parser(): ``subparsers.add_parser(...)``. """ parser = argparse.ArgumentParser( - prog="hermes", - description="Hermes Agent - AI assistant with tool-calling capabilities", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=_EPILOGUE, - ) - + prog="hermes", description="Hermes Agent - AI assistant with tool-calling capabilities", + formatter_class=argparse.RawDescriptionHelpFormatter, epilog=_EPILOGUE) _add_top_level_flags(parser) - subparsers = parser.add_subparsers(dest="command", help="Command to run") - chat_parser = _build_chat_parser(subparsers) - - return parser, subparsers, chat_parser + return parser, subparsers, _build_chat_parser(subparsers) diff --git a/hermes_cli/_scan_venv_blockers.py b/hermes_cli/_scan_venv_blockers.py index 5f7ee24c5b..ab8768c2df 100644 --- a/hermes_cli/_scan_venv_blockers.py +++ b/hermes_cli/_scan_venv_blockers.py @@ -1,4 +1,4 @@ -"""``hermes_cli/_scan_venv_blockers.py`` — Standalone venv-process scan for JSON consumption. +"""Standalone venv-process scan for JSON consumption (``python -m hermes_cli._scan_venv_blockers``). Exits 0 for valid clear or blocked results. Non-zero exit signals probe failure (the detector itself crashed, psutil unavailable, etc.). Exactly one JSON document on stdout; diagnostics on stderr only. @@ -12,35 +12,26 @@ import sys from pathlib import PureWindowsPath from typing import NoReturn -# Long CLI flags whose argument value must be redacted from the cmdline. +# Long CLI flags whose argument value must be redacted from the cmdline. Short flags (-t, -k, -p) +# are intentionally not redacted — ambiguous and useful diagnostics (toolset, port, profile). _SENSITIVE_LONG_FLAGS: list[str] = [ - "--token", - "--api-key", - "--password", - "--secret", - "--authorization", - "--access-key", - "--private-key", - "--session-key", + "--token", "--api-key", "--password", "--secret", "--authorization", "--access-key", + "--private-key", "--session-key", ] +_PYTHON_PROCESS_NAMES = {"python.exe", "pythonw.exe", "python", "pythonw"} +_UPDATER_STOPPABLE_PURPOSES = ("serve", "dashboard") + def _probe_fail_json(diagnostic: str = "probe failed") -> str: - """Return the standard probe-failure JSON document. + """The standard probe-failure JSON document. ``ok: false`` plus ``probe_failed: true`` means the detector itself could not run — this is *not* a clear scan. Callers must treat ``ok is not True`` / non-zero exit as probe failure, - never as ``blocked: false`` "clear" (#83149). + never as ``blocked: false`` "clear". """ - return json.dumps( - { - "ok": False, - "probe_failed": True, - "blocked": False, - "processes": [], - "error": diagnostic, - } - ) + return json.dumps({"ok": False, "probe_failed": True, "blocked": False, "processes": [], + "error": diagnostic}) def _emit_probe_fail(diagnostic: str) -> NoReturn: @@ -51,67 +42,50 @@ def _emit_probe_fail(diagnostic: str) -> NoReturn: def _find_flag(text: str, flag: str) -> int: - """Return the index of *flag* when it starts the string or follows a space.""" - low = text.lower() - fl = flag.lower() - pos = 0 + """Index of *flag* (case-insensitive) at string start or after a space; -1 if absent.""" + low, fl, pos = text.lower(), flag.lower(), 0 while True: idx = low.find(fl, pos) - if idx == -1: - return -1 - if idx == 0 or text[idx - 1] == " ": + if idx == -1 or idx == 0 or text[idx - 1] == " ": return idx pos = idx + 1 def _redact_sensitive_cmdline(cmdline: str) -> str: - """Apply generic secret redaction then long-flag redaction.""" - # Generic pass: the project's shared secret redactor. + """Apply the shared secret redactor, then replace everything after a sensitive long flag.""" try: from agent.redact import redact_sensitive_text # noqa: PLC0415 cmdline = redact_sensitive_text(cmdline, force=True) except Exception: return "" - - # Conservative long-flag pass: preserve the flag name, replace the value - # and everything after it with ````. Short flags (-t, -k, -p) - # are intentionally not redacted — they are ambiguous and may be useful - # diagnostics (toolset, port, profile). + # --flag=value → preserve "--flag="; --flag value → preserve "--flag ". earliest = len(cmdline) for flag in _SENSITIVE_LONG_FLAGS: - # --flag=value → preserve "--flag="; --flag value → preserve "--flag " for suffix in ("=", " "): idx = _find_flag(cmdline, flag + suffix) if idx != -1 and idx + len(flag) + 1 < earliest: earliest = idx + len(flag) + 1 - if earliest < len(cmdline): return cmdline[:earliest] + "" return cmdline def _classify_local_preview_args(args: object) -> dict[str, object]: - """Return safe UI metadata for an exact ``python -m http.server`` argv. + """Safe UI metadata for an exact ``python -m http.server`` argv; ``{}`` otherwise. The general holder detector truncates its diagnostic command line; reading argv separately preserves a useful directory label without exposing an unbounded command line to the renderer. """ if not isinstance(args, (list, tuple)) or not all(isinstance(arg, str) for arg in args): return {} - - # For a real interpreter module invocation, ``-m`` is the first argument - # after the executable. A later ``-m http.server`` can merely be data passed - # to an unrelated script and must never authorize termination. + # ``-m`` must be the first argument after the executable: a later ``-m http.server`` can be + # data passed to an unrelated script and must never authorize termination. if len(args) < 3 or args[1] != "-m" or args[2].lower() != "http.server": return {} - port = 8000 - if len(args) > 3: - candidate = args[3] - if candidate.isdigit() and 0 < int(candidate) <= 65535: - port = int(candidate) - + if len(args) > 3 and args[3].isdigit() and 0 < int(args[3]) <= 65535: + port = int(args[3]) label = "" try: directory_index = args.index("--directory") @@ -119,19 +93,14 @@ def _classify_local_preview_args(args: object) -> dict[str, object]: label = PureWindowsPath(args[directory_index + 1]).name except ValueError: pass - - metadata: dict[str, object] = { - "kind": "local-preview", - "safeToStop": True, - "port": port, - } + metadata: dict[str, object] = {"kind": "local-preview", "safeToStop": True, "port": port} if label: metadata["label"] = label return metadata def _local_preview_metadata(pid: int, name: str) -> dict[str, object]: - if name.lower() not in {"python.exe", "pythonw.exe", "python", "pythonw"}: + if name.lower() not in _PYTHON_PROCESS_NAMES: return {} try: import psutil # noqa: PLC0415 @@ -146,16 +115,13 @@ def _local_preview_metadata(pid: int, name: str) -> dict[str, object]: def _terminate_safe_preview( - pid: int, - expected_create_time: float, - *, - psutil_module: object | None = None, + pid: int, expected_create_time: float, *, psutil_module: object | None = None, ) -> tuple[bool, str | None]: """Terminate one verified local preview process tree. A fresh ``psutil.Process`` identity check and exact argv classification occur immediately before - termination. psutil guards mutating Process methods against PID reuse, avoiding taskkill's - stale-PID race. + termination; psutil guards mutating Process methods against PID reuse (no taskkill stale-PID + race). """ try: if psutil_module is None: @@ -166,9 +132,7 @@ def _terminate_safe_preview( return False, "process identity changed" if not _classify_local_preview_args(process.cmdline()): return False, "process is no longer a local preview" - - children = process.children(recursive=True) - targets = [*reversed(children), process] + targets = [*reversed(process.children(recursive=True)), process] for target in targets: target.terminate() _gone, alive = psutil_module.wait_procs(targets, timeout=3) # type: ignore[attr-defined] @@ -182,15 +146,12 @@ def _terminate_safe_preview( def _is_pausable_gateway(cmdline: str) -> bool: - """Return True when *cmdline* is a gateway process the updater can pause. + """True when *cmdline* is a gateway process the updater can pause. - Only gateway invocations are exempted. Anything else running from the venv (an operator's REPL, - a stray script, a ``serve`` backend that survived the desktop's own teardown) has no pause - machinery downstream and must keep blocking the handoff. - - Delegates to ``gateway.status.looks_like_gateway_command_line`` — the canonical ``gateway run`` - matcher (profile-selector aware, shlex tokenization, ``run``-only) — so this exemption, the - pause discovery, and the updater's guard fallback all share one parser. + Only gateway invocations are exempted; anything else running from the venv has no pause + machinery downstream and must keep blocking the handoff. Delegates to the canonical + ``gateway run`` matcher so this exemption, pause discovery and the updater's guard share one + parser. """ try: from gateway.status import looks_like_gateway_command_line # noqa: PLC0415 @@ -200,23 +161,15 @@ def _is_pausable_gateway(cmdline: str) -> bool: def _is_updater_owned_backend(pid: int, cmdline: str) -> bool: - """Return True when *pid* is a Hermes backend the CLI updater can stop. - - The gateway exemption above keeps ``gateway run`` holders out of the blocker list because the - updater's own pause machinery stops and resumes them. - - Positive identity only — never name/substring matching (#90778, and the #99558 identity-guard - contract): - """ + """True when *pid* is a Hermes backend the CLI updater can stop (positive ledger identity).""" return _updater_owned_backend_entry(pid, cmdline) is not None def _updater_owned_backend_entry(pid: int, cmdline: str) -> dict | None: - """Ledger entry for a deferred backend, or ``None`` when it must block. + """Ledger entry for a deferred serve/dashboard backend, or ``None`` when it must block. - Same decision logic as ``_is_updater_owned_backend`` (which delegates here); returning the - matched ledger entry lets ``main()`` emit sanitized decision evidence — structured identity - fields only, never argv, which can carry tokens or private endpoints (#98350). + Returning the entry lets ``main()`` emit sanitized decision evidence — structured identity + fields only, never argv, which can carry tokens or private endpoints. """ try: from hermes_cli.update_cmd import _hermes_holder_subcommand # noqa: PLC0415 @@ -224,13 +177,10 @@ def _updater_owned_backend_entry(pid: int, cmdline: str) -> dict | None: purpose = _hermes_holder_subcommand(cmdline) except Exception: return None - if purpose not in ("serve", "dashboard"): + if purpose not in _UPDATER_STOPPABLE_PURPOSES: return None try: - from hermes_cli.process_identity import ( # noqa: PLC0415 - ledger_entries, - spawner_is_dead, - ) + from hermes_cli.process_identity import ledger_entries, spawner_is_dead # noqa: PLC0415 entries = ledger_entries() except Exception: @@ -238,43 +188,28 @@ def _updater_owned_backend_entry(pid: int, cmdline: str) -> dict | None: for entry in entries: if entry.get("pid") != pid: continue - if entry.get("purpose") not in ("serve", "dashboard"): + if entry.get("purpose") not in _UPDATER_STOPPABLE_PURPOSES: return None - dead = spawner_is_dead(entry) - if dead is not False: - # Spawner dead, unrecorded, or unprovable-but-registered: the - # updater's ledger rungs own this holder (reap or stop+relaunch). - return entry - if _spawner_is_this_handoff_desktop(entry): + # Spawner dead, unrecorded, or unprovable-but-registered: the updater's ledger rungs own + # this holder (reap or stop+relaunch). + if spawner_is_dead(entry) is not False or _spawner_is_this_handoff_desktop(entry): return entry return None return None def _deferred_backend_evidence(entries: list[dict]) -> list[dict]: - """Sanitized decision evidence for deferred serve/dashboard backends. - - Structured ledger fields only — pid, purpose, recorded port — never the command line, which can - carry tokens or private endpoints. Lets the scan result explain *why* a holder disappeared from - ``processes`` without echoing argv (#98350). - """ - evidence = [] - for entry in entries: - pid = entry.get("pid") - if not isinstance(pid, int): - continue - evidence.append( - {"pid": pid, "purpose": entry.get("purpose"), "port": entry.get("port")} - ) - return evidence + """Sanitized evidence (pid, purpose, recorded port — never argv) for deferred backends.""" + return [{"pid": entry.get("pid"), "purpose": entry.get("purpose"), "port": entry.get("port")} + for entry in entries if isinstance(entry.get("pid"), int)] def _spawner_is_this_handoff_desktop(entry: dict) -> bool: """True when the entry's live spawner is an ancestor of this scan. - The scan subprocess is spawned by the Desktop app's update preflight, so the Desktop performing - the hand-off is in our ancestor chain. Identity is verified by ``(pid, create_time)`` — a - recycled PID cannot forge the pair. + The scan is spawned by the Desktop app's update preflight, so the Desktop performing the + hand-off is in our ancestor chain. Identity is ``(pid, create_time)`` — a recycled PID cannot + forge the pair. """ spawner_pid = entry.get("spawner_pid") if not isinstance(spawner_pid, int) or spawner_pid <= 0: @@ -300,7 +235,6 @@ def main() -> None: import psutil # noqa: PLC0415, F401 except Exception as exc: _emit_probe_fail(f"psutil is not available: {exc}") - try: from hermes_cli.main import _detect_venv_python_processes # noqa: PLC0415 @@ -317,34 +251,23 @@ def main() -> None: continue deferred_entry = _updater_owned_backend_entry(pid, cmdline) if deferred_entry is not None: - # Ledger-verified serve/dashboard backend the CLI updater's own - # rungs stop (and relaunch) downstream — reporting it here would - # dead-end the hand-off before that machinery can run (#98336). + # Ledger-verified backend the updater's own rungs stop (and relaunch) downstream — + # reporting it here would dead-end the hand-off before that machinery can run. deferred_entries.append(deferred_entry) continue - process = { - "pid": pid, - "name": name, - # Truncate for display AFTER the gateway exemption has seen the - # full cmdline (long managed-runtime interpreter paths would - # otherwise swallow the `gateway run` argv). - "cmdline": _redact_sensitive_cmdline(cmdline)[:120], - } + # Truncate for display AFTER the gateway exemption has seen the full cmdline (long + # managed-runtime interpreter paths would otherwise swallow the `gateway run` argv). + process = {"pid": pid, "name": name, "cmdline": _redact_sensitive_cmdline(cmdline)[:120]} process.update(_local_preview_metadata(pid, name)) processes.append(process) + # pausable_gateways / deferred_backends / deferred_backend_evidence are diagnostic only. data = { "ok": True, "blocked": bool(processes), "processes": processes, - # Diagnostic only: gateway processes present but not counted as - # blockers because the downstream updater pauses them itself. "pausable_gateways": exempted_gateways, - # Diagnostic only: ledger-verified serve/dashboard backends deferred - # to the updater's stop/relaunch rungs (#98336). "deferred_backends": len(deferred_entries), - # Diagnostic only: sanitized evidence (structured ledger identity, - # never argv) explaining which holders the deferral consumed (#98350). "deferred_backend_evidence": _deferred_backend_evidence(deferred_entries), } print(json.dumps(data)) @@ -363,7 +286,6 @@ def _terminate_safe_main(argv: list[str]) -> NoReturn: except ValueError: print(json.dumps({"ok": False, "error": "invalid process identity"})) raise SystemExit(2) - stopped, error = _terminate_safe_preview(pid, create_time) print(json.dumps({"ok": stopped, "pid": pid, "error": error})) raise SystemExit(0 if stopped else 1) @@ -372,4 +294,4 @@ def _terminate_safe_main(argv: list[str]) -> NoReturn: if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "--terminate-safe": _terminate_safe_main(sys.argv[2:]) - main() \ No newline at end of file + main() diff --git a/hermes_cli/_secrets_common.py b/hermes_cli/_secrets_common.py index 3c38545b90..5e74a86a7a 100644 --- a/hermes_cli/_secrets_common.py +++ b/hermes_cli/_secrets_common.py @@ -59,10 +59,8 @@ def require_enabled(console: Console, cfg: dict, product: str, command: str) -> """Print the "integration is disabled" hint and return False unless ``cfg["enabled"]``.""" if cfg.get("enabled"): return True - console.print( - f"[yellow]{product} integration is disabled. Run " - f"`hermes secrets {command} setup` first.[/yellow]" - ) + console.print(f"[yellow]{product} integration is disabled. Run " + f"`hermes secrets {command} setup` first.[/yellow]") return False @@ -96,12 +94,8 @@ def print_table(console: Console, columns: Sequence, rows: Iterable, def cli_version(binary: Path) -> str: """Return the first line of `` --version`` or ``"version unknown"``.""" try: - res = subprocess.run( - [str(binary), "--version"], - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=5, - ) + res = subprocess.run([str(binary), "--version"], capture_output=True, text=True, + encoding='utf-8', errors='replace', timeout=5) if res.returncode == 0: return (res.stdout or res.stderr).strip().splitlines()[0] except (OSError, subprocess.TimeoutExpired): @@ -112,8 +106,8 @@ def cli_version(binary: Path) -> str: def secret_cli_env() -> dict: """Env for a secret-manager CLI child (``bws`` / ``op``). - Intentionally receives tokens — no scrub, no HOME rewrite (both CLIs store - state under the real user home). + Intentionally receives tokens — no scrub, no HOME rewrite (both CLIs store state under the + real user home). """ from tools.environments.local import build_subprocess_env @@ -122,41 +116,10 @@ def secret_cli_env() -> dict: return env -def prompt_new_token( - console: Console, - given: Optional[str], - token_env: str, - *, - flag: str, - intro: str, - prompt: str, -) -> Optional[str]: - """Return a non-empty token from ``given`` or a masked prompt; None (after printing) on failure.""" - token = (given or "").strip() - if not token: - if not sys.stdin.isatty(): - console.print(f"[red]No TTY — pass the token with {flag}.[/red]") - return None - console.print(intro) - token = masked_secret_prompt(prompt).strip() - if not token: - console.print("[red]Empty token, aborting.[/red]") - return None - return token - - def rotate_token( - console: Console, - given: Optional[str], - token_env: str, - *, - flag: str, - intro: str, - prompt: str, - verify: Optional[Callable[[str], bool]], - save: Callable[[str, str], object], - env_path: Callable[[], object], - clear_caches: Callable[[], object], + console: Console, given: Optional[str], token_env: str, *, flag: str, intro: str, prompt: str, + verify: Optional[Callable[[str], bool]], save: Callable[[str, str], object], + env_path: Callable[[], object], clear_caches: Callable[[], object], disabled_note: Optional[str], ) -> int: """Shared ``token`` subcommand: prompt, optionally verify, then persist. Returns the exit code. @@ -168,31 +131,30 @@ def rotate_token( in effect. Old cached pulls are keyed on the previous token's fingerprint; clearing them makes the next startup fetch fresh with the new credential. """ - token = prompt_new_token(console, given, token_env, flag=flag, intro=intro, prompt=prompt) - if token is None: + token = (given or "").strip() + if not token: + if not sys.stdin.isatty(): + console.print(f"[red]No TTY — pass the token with {flag}.[/red]") + return 1 + console.print(intro) + token = masked_secret_prompt(prompt).strip() + if not token: + console.print("[red]Empty token, aborting.[/red]") return 1 if verify is not None and not verify(token): return 1 save(token_env, token) os.environ[token_env] = token clear_caches() - console.print( - f"[green]✓[/green] stored in {env_path()} as {token_env}. " - "Takes effect on the next Hermes invocation." - ) + console.print(f"[green]✓[/green] stored in {env_path()} as {token_env}. " + "Takes effect on the next Hermes invocation.") if disabled_note: console.print(disabled_note) return 0 -def prompt_index( - console: Console, - prompt: str, - count: int, - *, - allow_empty: bool = False, - empty_message: Optional[str] = None, -) -> int: +def prompt_index(console: Console, prompt: str, count: int, *, + allow_empty: bool = False, empty_message: Optional[str] = None) -> int: """Loop until the user enters an integer in ``1..count``; return it. Blank input returns 0 when ``allow_empty``; otherwise ``empty_message`` (if any) diff --git a/hermes_cli/_startup_fast.py b/hermes_cli/_startup_fast.py index 33410bfca1..385639d7ce 100644 --- a/hermes_cli/_startup_fast.py +++ b/hermes_cli/_startup_fast.py @@ -1,12 +1,9 @@ """Pre-import startup fast paths — THE canonical lightweight helpers. -This module is imported by ``hermes_cli/main.py`` BEFORE its heavy import wall (config, argparse -tree, logging, providers). Everything here must stay **stdlib-only and cheap** (os/sys file probes; -no yaml, no hermes_cli.config, no argparse). - -Why this module exists (the bug class it kills): version-printing kept being reimplemented as -``*_fast()`` copies at the top of main.py (Termux first, then globally), each duplicating canonical -logic — project-root resolution, container detection, profile detection. +Imported by ``hermes_cli/main.py`` BEFORE its heavy import wall (config, argparse tree, logging, +providers). Everything here must stay **stdlib-only and cheap** (os/sys file probes; no yaml, no +hermes_cli.config, no argparse). Exists so version-printing stops being reimplemented as +``*_fast()`` copies in main.py that duplicate project-root / container / profile detection. """ from __future__ import annotations @@ -41,20 +38,16 @@ def ensure_project_root_on_path() -> None: """Put the project root at sys.path[0], deduping realpath-equivalents.""" project_root = project_root_str() normalized_root = os.path.normcase(os.path.realpath(project_root)) - sys.path[:] = [ - entry for entry in sys.path - if not entry or os.path.normcase(os.path.realpath(entry)) != normalized_root - ] + sys.path[:] = [entry for entry in sys.path + if not entry or os.path.normcase(os.path.realpath(entry)) != normalized_root] sys.path.insert(0, project_root) def is_termux_env() -> bool: """Tiny Termux check for pre-import startup shortcuts.""" prefix = os.environ.get("PREFIX", "") - return bool( - os.environ.get("TERMUX_VERSION") or "com.termux/files/usr" in prefix - or prefix.startswith("/data/data/com.termux/") - ) + return bool(os.environ.get("TERMUX_VERSION") or "com.termux/files/usr" in prefix + or prefix.startswith("/data/data/com.termux/")) def is_termux_fast_version_argv(argv: list[str]) -> bool: @@ -95,26 +88,21 @@ def container_mode_may_be_active() -> bool: """ if os.environ.get("HERMES_DEV") == "1" or is_container_startup_environment(): return False - hermes_home = os.environ.get("HERMES_HOME", "").strip() if hermes_home: if os.path.exists(os.path.join(hermes_home, ".container-mode")): return True parent_name = os.path.basename(os.path.dirname(os.path.normpath(hermes_home))) return parent_name != "profiles" and active_profile_may_override_home(hermes_home) - default_home = _default_home() return active_profile_may_override_home(default_home) or os.path.exists( - os.path.join(default_home, ".container-mode") - ) + os.path.join(default_home, ".container-mode")) def read_openai_version() -> str | None: """Read OpenAI SDK version without importing ``importlib.metadata``.""" for base in sys.path: - if not base: - base = os.getcwd() - version_file = os.path.join(base, "openai", "_version.py") + version_file = os.path.join(base or os.getcwd(), "openai", "_version.py") try: with open(version_file, encoding="utf-8") as handle: for line in handle: @@ -130,7 +118,7 @@ def read_openai_version() -> str | None: def read_install_method() -> str | None: - """Read the installer's ``.install_method`` stamp, if present. + """The installer's ``.install_method`` stamp, if present. Only the stamp (step 1 of ``config.detect_install_method``'s resolution order) — the managed/git/pip fallbacks need heavier imports and stay on the slow path. @@ -145,9 +133,8 @@ def print_fast_version_info(*, check_updates: bool = True) -> None: Every lazy block degrades gracefully — a broken/heavy import can never take the basic version output down. """ - # Line 1: registry-owned banner label (includes "· upstream " for - # git installs). banner.py keeps rich/prompt_toolkit lazy, so this - # import is light; fall back to the plain label if anything fails. + # Registry-owned banner label (includes "· upstream " for git installs); banner.py keeps + # rich/prompt_toolkit lazy, so this import is light. try: from hermes_cli.banner import format_banner_version_label @@ -156,13 +143,9 @@ def print_fast_version_info(*, check_updates: bool = True) -> None: from hermes_cli import __release_date__, __version__ print(f"Hermes Agent v{__version__} ({__release_date__})") - print(f"Install directory: {project_root_str()}") - - # Install method: authoritative resolver first (code-scoped stamp → - # managed → nix → git → pip; also self-heals poisoned shared-home - # 'docker' stamps). Fall back to the cheap stdlib stamp probe only if - # the resolver import/run fails. + # Authoritative resolver first (code-scoped stamp → managed → nix → git → pip; also self-heals + # poisoned shared-home 'docker' stamps); cheap stdlib stamp probe only if it fails. try: from pathlib import Path @@ -173,18 +156,13 @@ def print_fast_version_info(*, check_updates: bool = True) -> None: install_method = read_install_method() if install_method: print(f"Install method: {install_method}") - print(f"Python: {sys.version.split()[0]}") - openai_version = read_openai_version() print(f"OpenAI SDK: {openai_version}" if openai_version else "OpenAI SDK: Not installed") - if not check_updates: return - - # Update status (synchronous — acceptable since the user asked for - # version info). Bounded by check_for_updates' own subprocess/network - # timeouts and its 6-hour cache; any failure prints nothing. + # Synchronous update status — bounded by check_for_updates' own subprocess/network timeouts + # and its 6-hour cache; any failure prints nothing. try: from hermes_cli.banner import UPDATE_AVAILABLE_NO_COUNT, check_for_updates from hermes_cli.config import recommended_update_command @@ -204,9 +182,9 @@ def print_fast_version_info(*, check_updates: bool = True) -> None: def try_fast_version(argv: list[str] | None = None) -> bool: """Handle ``hermes --version`` before the heavy import wall. - Only ``--version``/``-V`` (the ``version`` subcommand was removed — ``--version`` now carries - the full output incl. update status), and never when container mode may need to route the - command into the container. Termux keeps the HERMES_TERMUX_DISABLE_FAST_CLI escape hatch. + Only ``--version``/``-V`` (``--version`` carries the full output incl. update status), and never + when container mode may need to route the command into the container. Termux keeps the + HERMES_TERMUX_DISABLE_FAST_CLI escape hatch. """ if argv is None: argv = sys.argv[1:] @@ -218,6 +196,5 @@ def try_fast_version(argv: list[str] | None = None) -> bool: return False elif not is_global_fast_version_argv(argv) or container_mode_may_be_active(): return False - print_fast_version_info() return True diff --git a/hermes_cli/_subprocess_compat.py b/hermes_cli/_subprocess_compat.py index af3ac81560..7f57ae2097 100644 --- a/hermes_cli/_subprocess_compat.py +++ b/hermes_cli/_subprocess_compat.py @@ -1,12 +1,9 @@ """Windows subprocess compatibility helpers. -* ``["npm", "install", ...]`` — on Windows ``npm`` is ``npm.cmd``, a batch shim. -``subprocess.Popen(["npm", ...])`` fails with WinError 193 ("not a valid Win32 application") because -CreateProcessW can't run a ``.cmd`` file without ``shell=True`` or PATHEXT resolution. - -* ``start_new_session=True`` — on POSIX, this maps to ``os.setsid()`` and actually detaches the -child. On Windows it's silently ignored; the Windows equivalent is the ``CREATE_NEW_PROCESS_GROUP | -CREATE_NO_WINDOW`` creationflags bundle, which Python only applies when you pass it explicitly. +* ``["npm", ...]`` — on Windows ``npm`` is ``npm.cmd``, a batch shim; ``Popen`` fails with + WinError 193 because CreateProcessW can't run a ``.cmd`` without ``shell=True``/PATHEXT. +* ``start_new_session=True`` — POSIX ``os.setsid()`` detach; silently ignored on Windows, whose + equivalent is the ``CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW`` creationflags bundle. """ from __future__ import annotations @@ -34,48 +31,36 @@ __all__ = [ "pid_is_hermes", ] -# Flags that neutralize *attribute-scoped* diff drivers on any diff-rendering -# git command (``diff``, ``log -p``, ``show``, ``blame``). A malicious repo can -# name a driver in ``.gitattributes`` (``* diff=evil``) and point it at an -# arbitrary program via ``[diff "evil"] command=/textconv=`` in ``.git/config``. -# Because the attacker chooses the driver name, ``GIT_CONFIG_KEY`` overrides in -# ``noninteractive_git_env`` cannot enumerate and disable it — only these -# command-line flags do. ``--no-ext-diff`` kills ``command=``; ``--no-textconv`` -# kills ``textconv=``. Both are required (verified empirically: each alone -# leaves the other live). Smudge/clean filters are neutralized by the env +# Flags that neutralize *attribute-scoped* diff drivers on any diff-rendering git command. A +# malicious repo can name a driver in ``.gitattributes`` (``* diff=evil``) and point it at an +# arbitrary program via ``[diff "evil"] command=/textconv=`` in ``.git/config``; because the +# attacker chooses the name, ``GIT_CONFIG_KEY`` overrides in ``noninteractive_git_env`` cannot +# enumerate it — only these flags do. ``--no-ext-diff`` kills ``command=``; ``--no-textconv`` kills +# ``textconv=``; each alone leaves the other live. Smudge/clean filters are neutralized by the env # layer's ``core.hooksPath`` + running against the index without checkout. NO_DRIVER_DIFF_FLAGS = ("--no-ext-diff", "--no-textconv") -# Subcommands that render diffs and therefore invoke ``.gitattributes``-scoped -# diff/textconv drivers. Only these accept ``NO_DRIVER_DIFF_FLAGS`` — ``status`` -# and friends reject the flags (``unknown option``), so the helper must gate on -# this set rather than blanket-prepending. +# Only these subcommands accept ``NO_DRIVER_DIFF_FLAGS`` — ``status`` and friends reject them +# (``unknown option``), so the helper gates on this set rather than blanket-prepending. _DIFF_RENDERING_SUBCOMMANDS = frozenset({"diff", "show", "log", "blame"}) +# Options that consume the FOLLOWING token, so that value is never mistaken for the subcommand +# (``-C diff`` is a path; ``-c diff=x`` is a config pair). +_GIT_VALUE_OPTS = {"-C", "-c", "--git-dir", "--work-tree", "--namespace", "--exec-path"} + def harden_git_argv(args: Sequence[str]) -> list[str]: - """Return a copy of subcommand-first git *args* with diff-driver flags - inserted for diff-rendering subcommands. + """Copy of subcommand-first git *args* (no leading ``"git"``) with :data:`NO_DRIVER_DIFF_FLAGS` + inserted right after a diff-rendering subcommand; other subcommands are returned unchanged. - *args* is the argument list WITHOUT the leading ``"git"`` (e.g. - ``["diff", "HEAD"]`` or ``["-c", "core.quotePath=false", "diff", ...]``). - The first non-option token is treated as the subcommand; if it is one of - :data:`_DIFF_RENDERING_SUBCOMMANDS`, :data:`NO_DRIVER_DIFF_FLAGS` is - inserted immediately after it. Non-diff subcommands are returned unchanged. - - Pair with :func:`noninteractive_git_env`: the env layer disables - fsmonitor/hooks/pager/editor/credential sinks, this closes the one class - (attacker-named attribute drivers) env overrides cannot reach. + Pair with :func:`noninteractive_git_env`: the env layer disables fsmonitor/hooks/pager/editor/ + credential sinks, this closes the one class (attacker-named attribute drivers) env cannot reach. """ out = list(args) - # Options that consume the FOLLOWING token as their value, so that value is - # never mistaken for the subcommand (``-C diff`` is a path; ``-c diff=x`` is - # a config pair — neither is the diff subcommand). - _value_opts = {"-C", "-c", "--git-dir", "--work-tree", "--namespace", "--exec-path"} i = 0 while i < len(out): tok = out[i] - if tok in _value_opts: + if tok in _GIT_VALUE_OPTS: i += 2 continue if tok.startswith("-"): @@ -83,9 +68,7 @@ def harden_git_argv(args: Sequence[str]) -> list[str]: continue if tok in _DIFF_RENDERING_SUBCOMMANDS: return out[: i + 1] + list(NO_DRIVER_DIFF_FLAGS) + out[i + 1 :] - # First non-option token is the subcommand; if it isn't a diff renderer - # there is nothing to harden. - return out + return out # first non-option token is a non-diff subcommand return out @@ -98,136 +81,89 @@ _WINDOWS_GATEWAY_BREAKAWAY_ENV = "_HERMES_GATEWAY_BREAKAWAY" def split_command_line(line: str) -> list[str]: """Split a user-supplied command line into tokens, Windows-safely. - ``shlex.split`` (posix=True) treats every backslash as an escape, silently mangling Windows - paths (the separators vanish, leaving a wrong relative name). On Windows use ``posix=False`` - and strip one layer of matching double quotes per token; on POSIX this is exactly - ``shlex.split``. Raises ValueError on unbalanced quotes. + ``shlex.split`` (posix=True) treats every backslash as an escape, mangling Windows paths. On + Windows use ``posix=False`` and strip one layer of matching quotes per token; on POSIX this is + exactly ``shlex.split``. Raises ValueError on unbalanced quotes. """ - if not IS_WINDOWS: - import shlex - - return shlex.split(line) import shlex - tokens = shlex.split(line, posix=False) + if not IS_WINDOWS: + return shlex.split(line) out: list[str] = [] - for tok in tokens: + for tok in shlex.split(line, posix=False): if len(tok) >= 2 and tok[0] == tok[-1] and tok[0] in ("'", '"'): tok = tok[1:-1] out.append(tok) return out -# ----------------------------------------------------------------------------- -# Node ecosystem launcher resolution -# ----------------------------------------------------------------------------- - - def resolve_node_command(name: str, argv: Sequence[str]) -> list[str]: - """Resolve a Node-ecosystem command name to an absolute-path argv. + """Resolve a Node-ecosystem command name (``npm``, ``npx``, ``yarn``…) to an absolute-path argv. - On Windows, commands like ``npm``, ``npx``, ``yarn``, ``pnpm``, ``playwright``, ``prettier`` - ship as ``.cmd`` files (batch shims). ``subprocess.Popen(["npm", "install"])`` fails with - WinError 193 because CreateProcessW doesn't execute batch files directly. - - ``shutil.which(name)`` *does* resolve ``.cmd`` via PATHEXT and returns the fully-qualified path - — which CreateProcessW accepts because the extension tells Windows to route through ``cmd.exe - /c``. + On Windows these ship as ``.cmd`` batch shims that CreateProcessW won't execute by bare name; + ``shutil.which`` resolves via PATHEXT to a fully-qualified path whose extension routes it + through ``cmd.exe /c``. """ resolved = shutil.which(name) - if resolved: - return [resolved, *argv] - return [name, *argv] + return [resolved or name, *argv] -# ----------------------------------------------------------------------------- -# Detached / hidden process creation -# ----------------------------------------------------------------------------- - - -# Win32 CreationFlags — defined here rather than imported from subprocess -# because CREATE_NO_WINDOW and DETACHED_PROCESS aren't guaranteed to be -# present on stdlib subprocess on older Pythons or non-Windows builds. +# Win32 CreationFlags — defined here because CREATE_NO_WINDOW / DETACHED_PROCESS aren't guaranteed +# to exist on stdlib subprocess for older Pythons or non-Windows builds. _CREATE_NEW_PROCESS_GROUP = 0x00000200 -# DETACHED_PROCESS is intentionally NOT part of any flag bundle here — do not -# re-add it. Two reasons (the recurring console-flash bug #54220 / #56747): -# -# 1. MSDN (Process Creation Flags): CREATE_NO_WINDOW "is ignored if used with -# either CREATE_NEW_CONSOLE or DETACHED_PROCESS". Combining them means -# DETACHED_PROCESS governs and the no-window bit is dead. -# 2. A DETACHED_PROCESS child has NO console at all, so every console-subsystem -# descendant it ever spawns (git, gh, cmd, node, wmic, powershell, …) must -# allocate its OWN console — a visible flash per spawn, including spawns -# inside third-party libraries that no per-call-site CREATE_NO_WINDOW sweep -# can reach. A CREATE_NO_WINDOW child instead OWNS a hidden console that -# all descendants inherit, making "no flashing windows" a property of the -# one daemon launch. Root cause isolated + A/B verified on Windows 11 by -# the desktop backend fix (commit aa2ae36c3f): with per-site hide flags -# neutered, naive git/gh/cmd spawns don't flash under a hidden-console -# parent and do flash under a console-less one. -_DETACHED_PROCESS = 0x00000008 # kept for reference; must stay out of bundles +# DETACHED_PROCESS (0x00000008) is intentionally NOT part of any flag bundle — do not re-add it +# (the recurring console-flash bug): (1) MSDN: CREATE_NO_WINDOW "is ignored if used with either +# CREATE_NEW_CONSOLE or DETACHED_PROCESS"; (2) a DETACHED_PROCESS child has NO console, so every +# console-subsystem descendant (git, gh, cmd, node, powershell, …) allocates its own — a visible +# flash per spawn, including inside third-party libraries no per-site sweep can reach. A +# CREATE_NO_WINDOW child instead OWNS a hidden console all descendants inherit (A/B verified on +# Windows 11 by the desktop backend fix). _CREATE_NO_WINDOW = 0x08000000 -# Escape any Win32 job object the parent process belongs to. Without this, -# a detached child still inherits its parent's job object membership, and -# when that parent (Electron, Tauri, Windows Terminal, the Desktop GUI's -# bootstrap-installer) dies, the OS tears down the whole job — taking the -# "detached" child with it. Critical for the post-update gateway watcher: -# Electron spawns the Tauri updater inside its own job, the updater spawns -# the watcher subprocess; without BREAKAWAY the watcher dies the instant -# Electron exits, so the gateway never gets respawned after a `hermes -# update` triggered from the GUI. See fix/windows-gateway-reliability. +# Escape any Win32 job object the parent belongs to. Without this a detached child inherits the +# parent's job, and when that parent (Electron, Tauri, Windows Terminal, the Desktop bootstrap +# installer) dies the OS tears down the whole job — taking the "detached" child with it. Critical +# for the post-update gateway watcher spawned from inside Electron's job. _CREATE_BREAKAWAY_FROM_JOB = 0x01000000 def windows_detach_flags() -> int: - """Return Win32 creationflags detaching a child from the parent console/group; 0 elsewhere. + """Win32 creationflags detaching a child from the parent console/group; 0 elsewhere. Pair with the default ``start_new_session=False`` (POSIX uses ``start_new_session=True``). CREATE_NEW_PROCESS_GROUP stops Ctrl+C propagating; CREATE_NO_WINDOW gives the child a hidden - console that descendants inherit, avoiding per-descendant console flashes (DETACHED_PROCESS - would make CREATE_NO_WINDOW ignored and re-create that bug); CREATE_BREAKAWAY_FROM_JOB escapes - Electron/Tauri job objects that would otherwise kill the child with the parent. A job that - forbids breakaway yields PermissionError from Popen -- callers catch OSError and fall back. + console descendants inherit; CREATE_BREAKAWAY_FROM_JOB escapes Electron/Tauri job objects. A + job that forbids breakaway yields PermissionError from Popen — callers catch OSError and fall + back to :func:`windows_detach_flags_without_breakaway`. """ if not IS_WINDOWS: return 0 - return ( - _CREATE_NEW_PROCESS_GROUP - | _CREATE_NO_WINDOW - | _CREATE_BREAKAWAY_FROM_JOB - ) + return _CREATE_NEW_PROCESS_GROUP | _CREATE_NO_WINDOW | _CREATE_BREAKAWAY_FROM_JOB def windows_detach_flags_without_breakaway() -> int: - """Same as :func:`windows_detach_flags` minus ``CREATE_BREAKAWAY_FROM_JOB``. - - Retry with this when the breakaway variant raises OSError (job disallows breakaway), instead - of hand-coding the bit mask at every site. Returns 0 on non-Windows. - """ + """:func:`windows_detach_flags` minus ``CREATE_BREAKAWAY_FROM_JOB``; 0 on non-Windows.""" if not IS_WINDOWS: return 0 return _CREATE_NEW_PROCESS_GROUP | _CREATE_NO_WINDOW def windows_hide_flags() -> int: - """Return Win32 creationflags hiding the child's console without detaching it; 0 elsewhere. + """Win32 creationflags hiding the child's console without detaching it; 0 elsewhere. - For short-lived helpers (``taskkill``, ``where``, version probes) run synchronously: no - flash, but the child stays in the parent's process group and job so Ctrl+C and job teardown - still propagate. Stdio is inherited, so ``capture_output=True`` works. + For short-lived synchronous helpers (``taskkill``, ``where``, version probes): no flash, but the + child stays in the parent's process group and job so Ctrl+C and job teardown still propagate. + Stdio is inherited, so ``capture_output=True`` works. """ - if not IS_WINDOWS: - return 0 - return _CREATE_NO_WINDOW + return _CREATE_NO_WINDOW if IS_WINDOWS else 0 def suppress_platform_ver_console() -> None: """Stub ``platform._syscmd_ver`` on Windows so it never flashes a console. No-op elsewhere. ``platform.win32_ver()`` shells out ``cmd /c ver`` without CREATE_NO_WINDOW, so a windowless - parent (pythonw gateway, kanban workers) flashes a visible cmd window whenever a dependency - touches ``platform.uname()`` at import. With the stub, ``win32_ver()`` takes its documented - fallback to ``sys.getwindowsversion()`` -- same data, in-process. Call before heavy imports. + parent (pythonw gateway, kanban workers) flashes a cmd window whenever a dependency touches + ``platform.uname()`` at import. With the stub, ``win32_ver()`` takes its documented fallback to + ``sys.getwindowsversion()`` — same data, in-process. Call before heavy imports. """ if not IS_WINDOWS: return @@ -241,96 +177,72 @@ def suppress_platform_ver_console() -> None: platform._syscmd_ver = _quiet_syscmd_ver except Exception: - # Purely cosmetic hardening — never let it break startup. - pass + pass # Purely cosmetic hardening — never let it break startup. def windows_detach_popen_kwargs() -> dict: - """Return Popen kwargs detaching a child on Windows, or ``start_new_session=True`` on POSIX. + """Popen kwargs detaching a child on Windows, or ``start_new_session=True`` on POSIX. - Replaces bare ``start_new_session=True``, which is accepted but has no effect on Windows: the - child stays attached to the parent console and dies when it closes. + Bare ``start_new_session=True`` is accepted but has no effect on Windows: the child stays + attached to the parent console and dies when it closes. """ if IS_WINDOWS: return {"creationflags": windows_detach_flags()} return {"start_new_session": True} -# ----------------------------------------------------------------------------- -# Non-interactive git environment (credential-prompt hang guard) -# ----------------------------------------------------------------------------- +# GIT_CONFIG_KEY_n/VALUE_n overrides for internal git children: no credential/askpass prompts, no +# repo-configured fsmonitor/hooks/pager/editor/external-diff programs. +_GIT_CONFIG_INJECT_PREFIXES = ("GIT_CONFIG_KEY_", "GIT_CONFIG_VALUE_") +_GIT_CONFIG_OVERRIDES = { + "credential.helper": "", + "core.askPass": "", + "core.fsmonitor": "false", + "core.untrackedCache": "false", + "core.hooksPath": os.devnull, + "core.pager": "cat", + "core.editor": "true", + "sequence.editor": "true", + "diff.external": "", +} -def noninteractive_git_env( - base: "Mapping[str, str] | None" = None, -) -> dict[str, str]: +def noninteractive_git_env(base: "Mapping[str, str] | None" = None) -> dict[str, str]: """Environment for *internal* git invocations that must never prompt. - * ``GIT_TERMINAL_PROMPT=0`` — git fails with "terminal prompts disabled" instead of prompting - for credentials. * ``GCM_INTERACTIVE=Never`` — Git Credential Manager (the default credential - helper on Windows installs) never pops its own dialog. - - Returns a copy of ``base`` (default ``os.environ``) with ``GIT_TERMINAL_PROMPT=0`` (fail instead - of prompting), ``GCM_INTERACTIVE=Never`` (no Git Credential Manager dialog), and isolated git - config: inherited ``GIT_CONFIG_*`` overrides, global/system config, pagers, editors, fsmonitor, - external diff, and hooks are all disabled for the child so a user's repo/global config cannot - hang or mutate Hermes's internal plumbing calls. - - ``GIT_ASKPASS`` / ``SSH_ASKPASS`` are deliberately left alone: when the user has a *working* - askpass helper or ssh-agent configured, auth should still succeed non-interactively. The env - only disables paths that block on a human. Pair with ``stdin=subprocess.DEVNULL``. Internal - plumbing only — the agent-facing terminal tool has its own policy layer and visible PTY. + Copy of ``base`` (default ``os.environ``) with ``GIT_TERMINAL_PROMPT=0`` (fail instead of + prompting), ``GCM_INTERACTIVE=Never`` (no Git Credential Manager dialog), and isolated git + config: inherited ``GIT_CONFIG_*`` injection, global/system config, pagers, editors, fsmonitor, + external diff and hooks are all disabled so a user's repo/global config cannot hang or mutate + Hermes's plumbing calls. ``GIT_ASKPASS``/``SSH_ASKPASS`` are deliberately left alone: a + *working* askpass helper or ssh-agent should still succeed non-interactively. Pair with + ``stdin=subprocess.DEVNULL``. Internal plumbing only — the agent-facing terminal tool has its + own policy layer and visible PTY. """ env = dict(base if base is not None else os.environ) env["GIT_TERMINAL_PROMPT"] = "0" env["GCM_INTERACTIVE"] = "Never" - - # Do not inherit caller-supplied config injection. We rebuild the - # GIT_CONFIG_COUNT block below so ambient -c values cannot re-enable - # pagers, hooks, fsmonitor, editors, or credential prompts. + # Drop caller-supplied config injection; the GIT_CONFIG_COUNT block is rebuilt below so + # ambient -c values cannot re-enable pagers, hooks, fsmonitor, editors or credential prompts. for key in list(env): - if ( - key == "GIT_CONFIG_PARAMETERS" - or key.startswith("GIT_CONFIG_KEY_") - or key.startswith("GIT_CONFIG_VALUE_") - ): + if key == "GIT_CONFIG_PARAMETERS" or key.startswith(_GIT_CONFIG_INJECT_PREFIXES): env.pop(key, None) env.pop("GIT_CONFIG_COUNT", None) - - devnull = os.devnull - env["GIT_CONFIG_GLOBAL"] = devnull - env["GIT_CONFIG_SYSTEM"] = devnull + env["GIT_CONFIG_GLOBAL"] = os.devnull + env["GIT_CONFIG_SYSTEM"] = os.devnull env["GIT_CONFIG_NOSYSTEM"] = "1" env["GIT_PAGER"] = "cat" env["PAGER"] = "cat" env["GIT_EDITOR"] = "true" - - config_overrides = { - "credential.helper": "", - "core.askPass": "", - "core.fsmonitor": "false", - "core.untrackedCache": "false", - "core.hooksPath": devnull, - "core.pager": "cat", - "core.editor": "true", - "sequence.editor": "true", - "diff.external": "", - } - env["GIT_CONFIG_COUNT"] = str(len(config_overrides)) - for idx, (key, value) in enumerate(config_overrides.items()): + env["GIT_CONFIG_COUNT"] = str(len(_GIT_CONFIG_OVERRIDES)) + for idx, (key, value) in enumerate(_GIT_CONFIG_OVERRIDES.items()): env[f"GIT_CONFIG_KEY_{idx}"] = key env[f"GIT_CONFIG_VALUE_{idx}"] = value - return env -# ----------------------------------------------------------------------------- -# Bounded, fail-open git probing (Windows post-kill deadlock guard) -# ----------------------------------------------------------------------------- - - def _process_start_time(pid: int) -> int | None: - """Return the repository's stable process-start fingerprint, if available.""" + """The repository's stable process-start fingerprint, if available.""" try: from gateway.status import get_process_start_time @@ -343,13 +255,10 @@ def _text_names_hermes(text: str) -> bool: r"""True when *text* names Hermes at a path-segment / token boundary. A bare ``"hermes" in text`` substring test would also match unrelated processes whose paths - merely contain the letters (``...\shermesa\...``), which is exactly the false-positive class - this guard exists to prevent. + merely contain the letters (``...\shermesa\...``) — the false-positive class this prevents. """ - for token in re.split(r"[\\/\s=,;\"']+", text.lower()): - if token.startswith("hermes") or token.startswith(".hermes"): - return True - return False + return any(token.startswith(("hermes", ".hermes")) + for token in re.split(r"[\\/\s=,;\"']+", text.lower())) def _process_command_is_hermes(pid: int) -> bool: @@ -365,12 +274,8 @@ def _process_command_is_hermes(pid: int) -> bool: return False -def pid_is_hermes( - pid: int, - *, - expected_start_time: int | None = None, -) -> bool: - """Return whether it is safe to use ``taskkill`` for *pid*. +def pid_is_hermes(pid: int, *, expected_start_time: int | None = None) -> bool: + """Whether it is safe to use ``taskkill`` for *pid*. The PID must be valid, currently exist, and identify a Hermes process. When the caller captured a start-time fingerprint before the destructive action, the live process must still have the @@ -385,17 +290,13 @@ def pid_is_hermes( return _process_start_time(pid) == expected_start_time except Exception: return False - try: current_start_time = _process_start_time(pid) except Exception: return False if current_start_time is None: return False - if ( - expected_start_time is not None - and current_start_time != expected_start_time - ): + if expected_start_time is not None and current_start_time != expected_start_time: return False try: return _process_command_is_hermes(pid) @@ -404,13 +305,11 @@ def pid_is_hermes( def kill_process_tree(proc: "subprocess.Popen") -> None: - """Best-effort terminate *proc* and its descendants on both platforms. + """Best-effort terminate *proc* and its descendants on both platforms; never raises. - ``proc.kill()`` alone only terminates the direct child. On Windows a suspended descendant (e.g. - - All failures are swallowed — this is cleanup on an already-failing path, and the caller's - contract is to fail open. ``kill()`` can raise (access denied, already reaped); an unhandled - raise here would escape the caller's ``except`` handler and break that contract. + ``proc.kill()`` alone only terminates the direct child. This is cleanup on an already-failing + path whose contract is to fail open, so every failure (access denied, already reaped) is + swallowed rather than escaping the caller's ``except``. """ try: from agent.deadline import kill_process_tree as _deadline_kill_tree @@ -419,8 +318,7 @@ def kill_process_tree(proc: "subprocess.Popen") -> None: except Exception: _legacy_kill_process_tree(proc) return - # Ensure Popen's own bookkeeping sees the exit (matches the legacy body: - # a direct kill() so communicate()/wait() cannot hang on a stale handle). + # Ensure Popen's own bookkeeping sees the exit so communicate()/wait() cannot hang. try: proc.kill() except OSError: @@ -428,14 +326,9 @@ def kill_process_tree(proc: "subprocess.Popen") -> None: def _legacy_kill_process_tree(proc: "subprocess.Popen") -> None: - """Pre-#85125 local tree-kill — fallback when agent.deadline is unavailable. - - Kept verbatim so ``kill_process_tree`` can honor its swallow-everything contract even when the - delegation path itself fails (partial install, import cycle during teardown). - """ + """Local tree-kill fallback when agent.deadline is unavailable (partial install, cycle).""" if not IS_WINDOWS: - # Group-kill first: verify the child actually leads its own process - # group before signalling it, so we never blast a shared group. + # Verify the child leads its own process group before signalling, never a shared group. try: import signal as _signal @@ -449,58 +342,39 @@ def _legacy_kill_process_tree(proc: "subprocess.Popen") -> None: except OSError: pass if IS_WINDOWS: - # No identity guard here on purpose: *proc* is our own retained - # ``Popen`` handle. The child cannot be reaped (and its PID cannot be - # recycled) while we still hold the handle, so an identity check could - # only ever false-refuse a legitimate cleanup. The fail-closed - # ``pid_is_hermes`` guard is for BARE pids from state files or process - # scans, where recycling is real. + # No identity guard on purpose: *proc* is our own retained Popen handle, so the PID cannot + # be recycled while we hold it. The fail-closed ``pid_is_hermes`` guard is for BARE pids. try: - subprocess.run( - ["taskkill", "/T", "/F", "/PID", str(proc.pid)], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - stdin=subprocess.DEVNULL, - timeout=2, - check=False, - creationflags=windows_hide_flags(), - ) + subprocess.run(["taskkill", "/T", "/F", "/PID", str(proc.pid)], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + stdin=subprocess.DEVNULL, timeout=2, check=False, + creationflags=windows_hide_flags()) except Exception: pass def bounded_probe_run( - argv: Sequence[str], - *, - timeout: float, - errors: str = "replace", + argv: Sequence[str], *, timeout: float, errors: str = "replace", env: "Mapping[str, str] | None" = None, ) -> "subprocess.CompletedProcess[str] | None": - """Deadlock-safe ``subprocess.run(argv, capture_output=True, timeout=...)`` for fail-open probe - call sites. Returns a ``CompletedProcess`` when the child finished within *timeout* (any exit - code), or ``None`` on spawn failure or timeout. + """Deadlock-safe ``subprocess.run(argv, capture_output=True, timeout=…)`` for fail-open probes. + + Returns a ``CompletedProcess`` when the child finished within *timeout* (any exit code), or + ``None`` on spawn failure or timeout. """ _popen_kwargs: dict = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {"process_group": 0} try: proc = subprocess.Popen( - list(argv), - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - stdin=subprocess.DEVNULL, - text=True, - encoding="utf-8", - errors=errors, - env=dict(env) if env is not None else None, - **_popen_kwargs, - ) + list(argv), stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, + text=True, encoding="utf-8", errors=errors, + env=dict(env) if env is not None else None, **_popen_kwargs) except Exception: return None try: stdout, stderr = proc.communicate(timeout=timeout) except Exception: - # Timeout OR any other communicate() failure (torn-down pipe, decode - # error): terminate the child + descendants and drain bounded. Leaving - # it running would leak the same suspended-descendant class this guards. + # Timeout OR any other communicate() failure (torn-down pipe, decode error): tree-kill and + # drain bounded — leaving it running would leak the suspended-descendant class this guards. kill_process_tree(proc) try: proc.communicate(timeout=1) @@ -513,18 +387,15 @@ def bounded_probe_run( def bounded_git_probe(argv: Sequence[str], *, timeout: float) -> str: """Run a short ``git`` probe and return stripped stdout, or ``""`` on ANY failure. - Deadlock-safe replacement for ``subprocess.run(["git", ...], timeout=...)`` at fail-open - probe sites. On Windows ``run()``'s post-timeout cleanup calls an unbounded ``communicate()``; - a suspended descendant git.exe holding the pipe handles then blocks forever. Here: bounded - ``communicate``, then tree-kill plus a 1s drain, then abandon the pipes. Spawn contract - matches ``run`` byte-for-byte; on POSIX the probe gets its own process group so cleanup - also takes down credential helpers and remote helpers. + On Windows ``run()``'s post-timeout cleanup calls an unbounded ``communicate()``; a suspended + descendant git.exe holding the pipe handles then blocks forever. Here: bounded ``communicate``, + tree-kill plus a 1s drain, then abandon the pipes; on POSIX the probe gets its own process + group so cleanup also takes down credential/remote helpers. - Security (GHSA-7x36-8jrh-v4pw): these probes run automatically against whatever directory - the session sits in, before any tool call or trust prompt, and an index refresh executes the - repo-configured ``core.fsmonitor`` program (hooks/pager/editor/credential helper are sinks - too). A repo delivered with its ``.git`` intact would get host code execution. Every probe - therefore runs under :func:`noninteractive_git_env`; diff-rendering callers additionally pass + Security (GHSA-7x36-8jrh-v4pw): these probes run automatically against whatever directory the + session sits in, before any tool call or trust prompt, and an index refresh executes the + repo-configured ``core.fsmonitor`` program. Every probe therefore runs under + :func:`noninteractive_git_env`; diff-rendering callers additionally pass :data:`NO_DRIVER_DIFF_FLAGS` (attribute-scoped drivers can't be disabled via env). """ result = bounded_probe_run(argv, timeout=timeout, env=noninteractive_git_env())