Merge branch 'simp/r3-16-F' into simp/integration3

This commit is contained in:
Teknium
2026-09-02 21:57:48 -07:00
7 changed files with 601 additions and 1442 deletions
+106 -241
View File
@@ -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.<unix-ms>``) 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.<unix-ms>``) 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.
+86 -208
View File
@@ -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 <name>`` that returns ``profiles\\<name>``, 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 <name>`` is
``profiles\\<name>`` 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
``<root>\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 ``<root>\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: ``<root>\\bin`` (in-checkout launcher dir
the update autostash could sweep) and ``<root>\\venv\\Scripts``
(shadowed the user's ``python``, #83797).
The legacy ``<root>\\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 ``<root>\\bin`` and
``<root>\\venv\\Scripts`` entries. Legacy ``<root>\\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):
+179 -495
View File
@@ -8,34 +8,19 @@ gateway, sessions, …) is built by ``hermes_cli/subcommands/<group>.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 <name>: 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 <name>: 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)
+56 -134
View File
@@ -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 "<redacted>"
# Conservative long-flag pass: preserve the flag name, replace the value
# and everything after it with ``<redacted>``. 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] + "<redacted>"
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()
main()
+22 -60
View File
@@ -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 ``<binary> --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)
+20 -43
View File
@@ -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 <sha>" 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 <sha>" 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
+132 -261
View File
@@ -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())