51dd1970af
apply_scratch_tmp_env resolved the process home to derive the scratch dir; on Windows a child env built from nothing (no HERMES_HOME, no USERPROFILE) made Path.home() raise and took build_subprocess_env down with it. No home means no scratch dir to offer, so the vars are left alone.
1516 lines
63 KiB
Python
1516 lines
63 KiB
Python
"""Shared constants for Hermes Agent.
|
|
|
|
Import-safe, stdlib-only — importable from anywhere without circular-import risk.
|
|
"""
|
|
|
|
import contextlib
|
|
import os
|
|
import re
|
|
import shutil
|
|
import stat
|
|
import sys
|
|
from collections.abc import MutableMapping
|
|
from contextvars import ContextVar, Token
|
|
from pathlib import Path
|
|
|
|
_profile_fallback_warned: bool = False
|
|
_UNSET = object()
|
|
_HERMES_HOME_OVERRIDE: ContextVar[str | object] = ContextVar("_HERMES_HOME_OVERRIDE", default=_UNSET)
|
|
|
|
# TUI busy-indicator styles (CLI /indicator, TUI gateway config, /help registry).
|
|
# Keep in sync with INDICATOR_STYLES / DEFAULT_INDICATOR_STYLE in ui-tui/src/app/interfaces.ts.
|
|
INDICATOR_STYLES: tuple[str, ...] = ("ascii", "emoji", "kaomoji", "unicode")
|
|
DEFAULT_INDICATOR_STYLE: str = "kaomoji"
|
|
|
|
|
|
def set_hermes_home_override(path: str | Path | None) -> Token:
|
|
"""Set a context-local Hermes home override and return its reset token.
|
|
|
|
Deliberately does not mutate ``os.environ`` (shared by every thread in the process).
|
|
"""
|
|
value: str | object = _UNSET if path is None else str(path)
|
|
return _HERMES_HOME_OVERRIDE.set(value)
|
|
|
|
|
|
def reset_hermes_home_override(token: Token) -> None:
|
|
"""Restore the previous context-local Hermes home override."""
|
|
_HERMES_HOME_OVERRIDE.reset(token)
|
|
|
|
|
|
def get_hermes_home_override() -> str | None:
|
|
"""Return the active context-local Hermes home override, if any."""
|
|
override = _HERMES_HOME_OVERRIDE.get()
|
|
return str(override) if override is not _UNSET and override else None
|
|
|
|
|
|
def _expand_hermes_home(path: str) -> Path:
|
|
"""Expand environment and user-home syntax in a Hermes home path."""
|
|
return Path(os.path.expanduser(os.path.expandvars(path)))
|
|
|
|
|
|
def _get_platform_default_hermes_home() -> Path:
|
|
"""Return the platform-native default Hermes home path."""
|
|
if sys.platform == "win32":
|
|
local_appdata = os.environ.get("LOCALAPPDATA", "").strip()
|
|
base = Path(local_appdata) if local_appdata else Path.home() / "AppData" / "Local"
|
|
return base / "hermes"
|
|
return Path.home() / ".hermes"
|
|
|
|
|
|
def sudo_invoker_default_home() -> Path | None:
|
|
"""The invoking user's native ``~/.hermes`` when this process is root under ``sudo``, else None.
|
|
|
|
sudo strips HERMES_HOME and sets HOME=/root, so the process's own default is root's; the profile
|
|
store and the system service being operated on belong to SUDO_USER.
|
|
"""
|
|
if not hasattr(os, "geteuid") or os.geteuid() != 0:
|
|
return None
|
|
sudo_user = os.environ.get("SUDO_USER", "").strip()
|
|
if not sudo_user or sudo_user == "root":
|
|
return None
|
|
import pwd
|
|
|
|
try:
|
|
return Path(pwd.getpwnam(sudo_user).pw_dir) / ".hermes"
|
|
except KeyError: # SUDO_USER not in passwd (chroot/container)
|
|
return None
|
|
|
|
|
|
def _warn_profile_fallback_once() -> None:
|
|
"""Warn once when HERMES_HOME is unset but a non-default profile is sticky-active (wrong fallback)."""
|
|
global _profile_fallback_warned
|
|
if _profile_fallback_warned:
|
|
return
|
|
try:
|
|
fallback_home = _get_platform_default_hermes_home()
|
|
active_path = fallback_home / "active_profile"
|
|
active = active_path.read_text(encoding="utf-8").strip() if active_path.exists() else ""
|
|
except (UnicodeDecodeError, OSError):
|
|
active = ""
|
|
if active and active != "default":
|
|
_profile_fallback_warned = True
|
|
# Direct stderr, not logging: runs at import time (often before logging is
|
|
# configured) and root-logger propagation would double-emit.
|
|
msg = (
|
|
f"[HERMES_HOME fallback] HERMES_HOME is unset but active "
|
|
f"profile is {active!r}. Falling back to {fallback_home}, which "
|
|
f"is the DEFAULT profile — not {active!r}. Any data this "
|
|
f"process writes will land in the wrong profile. The "
|
|
f"subprocess spawner should pass HERMES_HOME explicitly "
|
|
f"(see issue #18594)."
|
|
)
|
|
with contextlib.suppress(Exception):
|
|
sys.stderr.write(msg + "\n")
|
|
sys.stderr.flush()
|
|
|
|
|
|
def get_hermes_home() -> Path:
|
|
"""Hermes home: context-local override → ``HERMES_HOME`` env var → platform default."""
|
|
override = get_hermes_home_override()
|
|
if override:
|
|
return _expand_hermes_home(override)
|
|
if not os.environ.get("HERMES_HOME", "").strip():
|
|
_warn_profile_fallback_once()
|
|
return get_process_hermes_home()
|
|
|
|
|
|
# Resolved keys, keyed by the path string that was handed in. Path.resolve()
|
|
# is a filesystem call, and this function sits under every ToolRegistry
|
|
# lookup through current_scope_key(), so without this the registry pays a
|
|
# syscall per lookup. A process only ever sees a handful of home paths, so
|
|
# the dict stays tiny. Only paths that really exist are stored, see below.
|
|
_HOME_KEY_CACHE: dict[str, str] = {}
|
|
|
|
|
|
def hermes_home_key(path: str | Path | None = None) -> str:
|
|
"""Stable registry key for a Hermes home/profile dir.
|
|
|
|
``strict=False`` so profiles whose directories don't exist yet still get a key.
|
|
|
|
The resolved value is remembered per input path. A directory that does
|
|
not exist yet is resolved without touching the cache, because the answer
|
|
can change once it is created (e.g. part of the path turns out to be a symlink).
|
|
"""
|
|
candidate = Path(path) if path is not None else get_hermes_home()
|
|
raw = str(candidate)
|
|
cached = _HOME_KEY_CACHE.get(raw)
|
|
if cached is not None:
|
|
return cached
|
|
expanded = candidate.expanduser()
|
|
try:
|
|
resolved = expanded.resolve(strict=True)
|
|
except OSError:
|
|
# Not on disk yet: lenient resolve, not stored, so the real answer is
|
|
# picked up once the directory appears.
|
|
return os.path.normcase(str(expanded.resolve(strict=False)))
|
|
key = os.path.normcase(str(resolved))
|
|
_HOME_KEY_CACHE[raw] = key
|
|
return key
|
|
|
|
|
|
def reset_hermes_home_key_cache() -> None:
|
|
"""Forget every remembered home key (for tests that move a home dir on disk)."""
|
|
_HOME_KEY_CACHE.clear()
|
|
|
|
|
|
def get_process_hermes_home() -> Path:
|
|
"""Hermes home of the running process, ignoring task overrides.
|
|
|
|
For process-level assets (theme YAML, dashboard plugin manifests) that must stay visible while a
|
|
request is scoped to another profile (e.g. embedded ``/chat`` under ``--open-profile``).
|
|
"""
|
|
val = os.environ.get("HERMES_HOME", "").strip()
|
|
return _expand_hermes_home(val) if val else _get_platform_default_hermes_home()
|
|
|
|
|
|
# Hermes-managed runtime downloads at the root of a home (GGUF models, llama.cpp runtimes,
|
|
# managed Node): re-downloadable on demand and routinely tens to hundreds of GB. Shared by
|
|
# ``hermes backup`` (excludes them) and ``profile create --clone-all`` (skips them from the
|
|
# default profile) so the two lists cannot drift apart.
|
|
LOCAL_RUNTIME_ROOT_DIRS: frozenset[str] = frozenset({"models", "runtimes", "node"})
|
|
|
|
# get_default_hermes_root() memo keyed on (native home, expanded HERMES_HOME) so it stays
|
|
# fresh when a test or plugin mutates either input; saves ~80us/call at 31+ sites.
|
|
_default_hermes_root_memo: "tuple[str, str, Path] | None" = None
|
|
|
|
|
|
def get_default_hermes_root() -> Path:
|
|
"""Root Hermes dir for profile-level ops: ``<root>`` when ``HERMES_HOME=<root>/profiles/<name>``."""
|
|
global _default_hermes_root_memo
|
|
native_home = _get_platform_default_hermes_home()
|
|
env_home = os.environ.get("HERMES_HOME", "").strip()
|
|
env_path = _expand_hermes_home(env_home) if env_home else None
|
|
memo_key = (str(native_home), str(env_path) if env_path is not None else "")
|
|
memo = _default_hermes_root_memo
|
|
if memo is not None and memo[:2] == memo_key:
|
|
return memo[2]
|
|
result = native_home
|
|
if env_path is not None:
|
|
try:
|
|
env_path.resolve().relative_to(native_home.resolve()) # under ~/.hermes (normal or profile mode)
|
|
except ValueError: # Docker/custom root: <root>/profiles/<name> -> <root>, else HERMES_HOME itself
|
|
result = env_path.parent.parent if env_path.parent.name == "profiles" else env_path
|
|
_default_hermes_root_memo = (*memo_key, result)
|
|
return result
|
|
|
|
|
|
# Tombstone lives beside the profile dir (not inside) so a stale mkdir or rmtree cannot erase it.
|
|
_DELETED_PROFILES_DIR = ".deleted"
|
|
# Files marking a real Hermes home; arbitrary dirs with a ``profiles`` segment lack them.
|
|
_HERMES_HOME_MARKERS = ("config.yaml", ".env", "state.db")
|
|
|
|
|
|
def _is_hermes_profiles_root(profiles_dir: Path) -> bool:
|
|
"""True when *profiles_dir* is provably ``<hermes-home>/profiles``.
|
|
|
|
Accepts the classic ``~/.hermes`` layout, a root carrying Hermes-home marker files, a
|
|
``profiles/.deleted`` tombstone dir (only ``profile delete`` creates it), or the default root.
|
|
"""
|
|
root = profiles_dir.parent
|
|
if root.name == ".hermes":
|
|
return True
|
|
try:
|
|
if (profiles_dir / _DELETED_PROFILES_DIR).is_dir() or any(
|
|
(root / marker).exists() for marker in _HERMES_HOME_MARKERS
|
|
):
|
|
return True
|
|
except OSError:
|
|
pass
|
|
try:
|
|
return root.resolve(strict=False) == get_default_hermes_root().resolve(strict=False)
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def named_profile_home(path: str | Path) -> Path | None:
|
|
"""Return ``<root>/profiles/<name>`` when *path* is that home or under it.
|
|
|
|
Requires ``<name>`` not to start with ``.`` and the ``profiles`` parent to be a real Hermes home;
|
|
a default home whose path merely contains a ``profiles`` segment is not a named profile.
|
|
"""
|
|
current = Path(path)
|
|
for candidate in (current, *current.parents):
|
|
if (candidate.parent.name == "profiles" and not candidate.name.startswith(".")
|
|
and _is_hermes_profiles_root(candidate.parent)):
|
|
return candidate
|
|
if candidate.name == ".hermes": # default home: a coincidental profiles/ ancestor is not a root
|
|
return None
|
|
return None
|
|
|
|
|
|
def profile_name_for_home(path: str | Path | None) -> str | None:
|
|
"""Return the canonical profile id owning *path*, or ``None`` when it is not a profile home.
|
|
|
|
The default home is the Hermes root itself, so its basename is an installation detail (``.hermes``
|
|
on POSIX and commonly ``hermes`` on Windows), not the profile id ``default``.
|
|
"""
|
|
if path is None or not str(path).strip():
|
|
return None
|
|
current = Path(path).expanduser()
|
|
try:
|
|
default_root = get_default_hermes_root()
|
|
for candidate in (current, current.resolve(strict=False)):
|
|
if candidate == default_root or candidate == default_root.resolve(strict=False):
|
|
return "default"
|
|
named = named_profile_home(candidate)
|
|
if named is not None:
|
|
return named.name
|
|
# A stored profile home is authoritative: its owner already resolved it, so the
|
|
# <root>/profiles/<name> shape names the profile even when <root> carries no markers.
|
|
if candidate.parent.name == "profiles" and not candidate.name.startswith("."):
|
|
return candidate.name
|
|
except (OSError, RuntimeError, ValueError):
|
|
return None
|
|
return None
|
|
|
|
|
|
def profile_tombstone_path(profile_home: Path) -> Path:
|
|
return profile_home.parent / _DELETED_PROFILES_DIR / profile_home.name
|
|
|
|
|
|
def named_profile_is_deleted(profile_home: str | Path) -> bool:
|
|
return profile_tombstone_path(Path(profile_home)).exists()
|
|
|
|
|
|
# A directory under profiles/ is a profile only when something identifies it as one.
|
|
# Runtime side-effects (cron heartbeats, log rotation, caches) create dirs that carry
|
|
# none of these; a pre-tombstone ghost shell or a stray infrastructure dir must never be
|
|
# listed, served, ticked, or seeded with the default install's credentials.
|
|
_PROFILE_IDENTITY_MARKERS = ("config.yaml", ".env", "SOUL.md", "profile.yaml", "auth.json", "state.db")
|
|
|
|
|
|
def named_profile_has_identity(profile_home: str | Path) -> bool:
|
|
# A dangling symlinked marker (clone/migration leftover) is still an identity claim:
|
|
# ``is_file()`` follows links, so it alone would make such a profile unlistable.
|
|
home = Path(profile_home)
|
|
return any((home / marker).is_file() or (home / marker).is_symlink() for marker in _PROFILE_IDENTITY_MARKERS)
|
|
|
|
|
|
def named_profile_is_live(profile_home: str | Path) -> bool:
|
|
"""A resolvable named profile: an existing dir with identity that has not been deleted.
|
|
``-p``/``--profile`` resolution and ``profile_exists`` share this so a stale ghost shell can
|
|
never be started as a backend (whose ``ensure_hermes_home`` would rebuild the full tree)."""
|
|
home = Path(profile_home)
|
|
return home.is_dir() and named_profile_has_identity(home) and not named_profile_is_deleted(home)
|
|
|
|
|
|
def mark_named_profile_deleted(profile_home: str | Path) -> None:
|
|
marker = profile_tombstone_path(Path(profile_home))
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
marker.write_text("deleted\n", encoding="utf-8")
|
|
|
|
|
|
def clear_named_profile_deleted(profile_home: str | Path) -> None:
|
|
profile_tombstone_path(Path(profile_home)).unlink(missing_ok=True)
|
|
|
|
|
|
def assert_named_profile_home_live(path: str | Path) -> None:
|
|
"""Refuse missing or tombstoned named profile homes."""
|
|
home = named_profile_home(path)
|
|
if home is not None and (named_profile_is_deleted(home) or not home.exists()):
|
|
raise FileNotFoundError(
|
|
f"Named profile home does not exist: {home}. "
|
|
"Create the profile explicitly before using it."
|
|
)
|
|
|
|
|
|
def mkdir_under_hermes_home(path: str | Path) -> Path:
|
|
"""Create *path*, but never materialize a deleted/missing named profile."""
|
|
target = Path(path)
|
|
assert_named_profile_home_live(target)
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
return target
|
|
|
|
|
|
def _packaged_dir(env_var: str, default: Path | None, subdir: str) -> Path:
|
|
"""Resolve a package-manager-relocatable directory.
|
|
|
|
Order: *env_var* (Nix wrapper / explicit override) → caller ``default`` (source checkout) →
|
|
``<HERMES_HOME>/<subdir>``.
|
|
"""
|
|
override = os.getenv(env_var, "").strip()
|
|
return Path(override) if override else default if default is not None else get_hermes_home() / subdir
|
|
|
|
|
|
def get_optional_skills_dir(default: Path | None = None) -> Path:
|
|
"""Return the optional-skills directory, honoring package-manager wrappers."""
|
|
return _packaged_dir("HERMES_OPTIONAL_SKILLS", default, "optional-skills")
|
|
|
|
|
|
def get_optional_mcps_dir(default: Path | None = None) -> Path:
|
|
"""Return the optional-mcps directory, honoring package-manager wrappers (``HERMES_OPTIONAL_MCPS``)."""
|
|
return _packaged_dir("HERMES_OPTIONAL_MCPS", default, "optional-mcps")
|
|
|
|
|
|
def get_bundled_skills_dir(default: Path | None = None) -> Path:
|
|
"""Return the bundled skills directory, honoring package-manager wrappers (``HERMES_BUNDLED_SKILLS``)."""
|
|
return _packaged_dir("HERMES_BUNDLED_SKILLS", default, "skills")
|
|
|
|
|
|
def get_hermes_dir(new_subpath: str, old_name: str, *, home: Path | None = None) -> Path:
|
|
"""Resolve a Hermes subdirectory, honouring a populated legacy ``<old_name>/`` (no migration).
|
|
|
|
An empty legacy dir does NOT count (install scaffolds, manual mkdir) so it cannot shadow the new path.
|
|
|
|
A bare empty ``<old_name>/`` directory does **not** count as "the legacy install is in use" — install
|
|
scaffolds, manual ``mkdir`` work, and cleared-then-abandoned locations all create empty stubs that would
|
|
otherwise silently shadow real data populated at ``<new_subpath>/``. See #27602 for the pairing-store
|
|
regression where a dormant empty ``pairing/`` orphaned approved-user data in ``platforms/pairing/``.
|
|
"""
|
|
home = home or get_hermes_home()
|
|
old_path = home / old_name
|
|
return old_path if _legacy_path_has_content(old_path) else home / new_subpath
|
|
|
|
|
|
def iter_hermes_node_dirs(home: Path | None = None) -> list[Path]:
|
|
"""Hermes-managed Node dirs in lookup order; both Windows and POSIX shapes so migrated installs work.
|
|
|
|
Keep in sync with hermesManagedNodePathEntries() in apps/desktop/electron/backend-env.ts.
|
|
"""
|
|
node_dir = (home or get_hermes_home()) / "node"
|
|
return [node_dir, node_dir / "bin"] if sys.platform == "win32" else [node_dir / "bin", node_dir]
|
|
|
|
|
|
_WINDOWS_NODE_SHIMS = {
|
|
"npm": ["npm.cmd", "npm.exe", "npm"], "npx": ["npx.cmd", "npx.exe", "npx"], "node": ["node.exe", "node"],
|
|
}
|
|
|
|
|
|
def _candidate_node_command_names(command: str) -> list[str]:
|
|
base = Path(command).name
|
|
if sys.platform != "win32" or "." in base:
|
|
return [base]
|
|
# Prefer npm.cmd: PowerShell may block npm.ps1 by policy; CreateProcess cannot launch a bare .ps1.
|
|
return _WINDOWS_NODE_SHIMS.get(base.lower(), [f"{base}.cmd", f"{base}.exe", base])
|
|
|
|
|
|
def _iter_managed_node_candidates(names: list[str], home: Path | None = None):
|
|
"""Yield existing (and on POSIX, executable) ``<node-dir>/<name>`` files."""
|
|
for directory in iter_hermes_node_dirs(home):
|
|
for name in names:
|
|
candidate = directory / name
|
|
if candidate.is_file() and (sys.platform == "win32" or os.access(candidate, os.X_OK)):
|
|
yield candidate
|
|
|
|
|
|
def _first_runnable_managed(names: list[str]) -> tuple[str | None, bool]:
|
|
"""Return ``(first runnable candidate, saw a broken one)``."""
|
|
broken = False
|
|
for candidate in map(str, _iter_managed_node_candidates(names)):
|
|
if node_tool_runnable(candidate):
|
|
return candidate, broken
|
|
broken = True
|
|
return None, broken
|
|
|
|
|
|
def _run_version_probe(argv: list[str], **kwargs):
|
|
"""Run a hidden ``--version`` probe; ``None`` when it cannot run."""
|
|
import subprocess
|
|
try:
|
|
from hermes_cli._subprocess_compat import windows_hide_flags
|
|
return subprocess.run(
|
|
argv, capture_output=True, timeout=10, creationflags=windows_hide_flags(), **kwargs
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired, ValueError):
|
|
return None
|
|
|
|
|
|
def _version_probe_ok(path: str) -> bool:
|
|
"""True when ``<path> --version`` exits 0 under the Hermes-managed Node PATH."""
|
|
result = _run_version_probe([path, "--version"], env=with_hermes_node_path())
|
|
return result is not None and result.returncode == 0
|
|
|
|
|
|
_HERMES_NODE_TARGET_MAJOR = int(os.environ.get("HERMES_NODE_TARGET_MAJOR", "22"))
|
|
_managed_node_heal_attempted = False
|
|
_NODE_BOOTSTRAP_SCRIPT = Path(__file__).resolve().parent / "scripts" / "lib" / "node-bootstrap.sh"
|
|
|
|
# Install tree root (this file lives at <install_root>/hermes_constants.py). Used by secure_parent_dir() to
|
|
# skip chmod on the install dir — chmodding it 0700 breaks hermes-user traversal in Docker (UID 10000). See
|
|
# #25821, #93050.
|
|
_INSTALL_ROOT = Path(__file__).resolve().parent
|
|
|
|
|
|
def _is_executable_file(path: str) -> bool:
|
|
"""``exists()`` follows symlinks, so a dangling link never spawns a probe."""
|
|
return os.path.exists(path) and os.access(path, os.X_OK)
|
|
|
|
|
|
def node_tool_runnable(path: str | None) -> bool:
|
|
"""True only when *path* is a Node/npm/npx binary that actually runs (``--version`` probe)."""
|
|
if not path:
|
|
return False
|
|
present = Path(path).is_file() if sys.platform == "win32" else _is_executable_file(path)
|
|
return present and _version_probe_ok(path)
|
|
|
|
|
|
def hermes_managed_node_tree_present(home: Path | None = None) -> bool:
|
|
"""Return True when any Hermes-managed node/npm/npx shim exists on disk."""
|
|
names = [n for c in ("node", "npm", "npx") for n in _candidate_node_command_names(c)]
|
|
return next(_iter_managed_node_candidates(names, home), None) is not None
|
|
|
|
|
|
def _path_under_any(path: str, roots: list[str]) -> bool:
|
|
"""True when *path* sits inside one of *roots* (same drive).
|
|
|
|
Compared via ``normcase``: psutil and env vars can disagree on Windows drive-letter casing.
|
|
"""
|
|
path_norm = os.path.normcase(os.path.normpath(path))
|
|
for root in roots:
|
|
root_norm = os.path.normcase(os.path.normpath(root))
|
|
try:
|
|
if os.path.commonpath([path_norm, root_norm]) == root_norm:
|
|
return True
|
|
except ValueError: # different drives on Windows
|
|
continue
|
|
return False
|
|
|
|
|
|
def managed_node_tree_in_use(home: Path | None = None) -> bool:
|
|
"""True when a running process executes from the managed Node tree.
|
|
|
|
Windows locks running executables against delete/overwrite, so the updater must not rewrite
|
|
``%HERMES_HOME%\\node`` while the desktop app holds it (``[WinError 5]`` on ``npm.cmd``).
|
|
|
|
Always ``False`` on POSIX, which has no equivalent lock semantics. See #80926.
|
|
"""
|
|
if sys.platform != "win32":
|
|
return False
|
|
try:
|
|
import psutil
|
|
except Exception:
|
|
return False
|
|
dirs: list[str] = []
|
|
for directory in iter_hermes_node_dirs(home):
|
|
try:
|
|
dirs.append(str(Path(directory).resolve()))
|
|
except OSError:
|
|
continue
|
|
if not dirs:
|
|
return False
|
|
try:
|
|
procs = psutil.process_iter(["exe", "cmdline"])
|
|
except Exception:
|
|
return False
|
|
for proc in procs:
|
|
try:
|
|
info = proc.info
|
|
except Exception:
|
|
continue
|
|
exe = info.get("exe")
|
|
if exe:
|
|
try:
|
|
exe = str(Path(exe).resolve())
|
|
except (OSError, ValueError):
|
|
exe = str(exe)
|
|
if any(_path_under_any(p, dirs) for p in ([exe] if exe else []) + list(info.get("cmdline") or [])):
|
|
return True
|
|
return False
|
|
|
|
|
|
_managed_node_in_use_notice_printed = False
|
|
|
|
|
|
def _print_managed_node_in_use_notice() -> None:
|
|
"""Print the managed-Node deferral notice once per process."""
|
|
global _managed_node_in_use_notice_printed
|
|
if _managed_node_in_use_notice_printed:
|
|
return
|
|
_managed_node_in_use_notice_printed = True
|
|
print(
|
|
"→ Hermes-managed Node.js is in use by a running app; deferring its "
|
|
"upgrade until the app is closed (re-run `hermes update` afterwards).", flush=True,
|
|
)
|
|
|
|
|
|
def _fetch_url(url: str, timeout: int) -> bytes | None:
|
|
import urllib.request
|
|
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as response:
|
|
return response.read()
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _stage_windows_node_zip(home: Path, node_arch: str) -> Path | None:
|
|
"""Download the target-major portable Node zip into a sibling ``node.new-*`` dir.
|
|
|
|
A sibling makes the later swap a same-volume rename. ``None`` on any failure.
|
|
"""
|
|
import tempfile
|
|
import uuid
|
|
import zipfile
|
|
|
|
index_url = f"https://nodejs.org/dist/latest-v{_HERMES_NODE_TARGET_MAJOR}.x/"
|
|
index_bytes = _fetch_url(index_url, 60)
|
|
if index_bytes is None:
|
|
return None
|
|
pattern = rf"node-v{_HERMES_NODE_TARGET_MAJOR}\.\d+\.\d+-win-{node_arch}\.zip"
|
|
match = re.search(pattern, index_bytes.decode("utf-8", errors="replace"))
|
|
if not match:
|
|
return None
|
|
zip_name = match.group(0)
|
|
zip_bytes = _fetch_url(f"{index_url}{zip_name}", 300)
|
|
if zip_bytes is None:
|
|
return None
|
|
staged = home / f"node.new-{uuid.uuid4().hex[:8]}"
|
|
try:
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp_path = Path(tmp_dir)
|
|
(tmp_path / zip_name).write_bytes(zip_bytes)
|
|
extract_dir = tmp_path / "extract"
|
|
extract_dir.mkdir()
|
|
with zipfile.ZipFile(tmp_path / zip_name) as archive:
|
|
archive.extractall(extract_dir)
|
|
extracted = next(extract_dir.glob("node-v*"), None)
|
|
if extracted is None or not extracted.is_dir():
|
|
return None
|
|
shutil.move(str(extracted), str(staged))
|
|
except OSError:
|
|
return None
|
|
return staged
|
|
|
|
|
|
def _swap_node_tree(target: Path, staged: Path) -> bool | None:
|
|
"""Rename the live tree aside (``node.old-*``) and *staged* into place.
|
|
|
|
``None`` when the OS refuses to move the live tree (in use; untouched, retried next time),
|
|
``False`` when the staged tree cannot be moved in (live tree rolled back).
|
|
"""
|
|
backup = target.parent / f"node.old-{staged.name.removeprefix('node.new-')}"
|
|
had_live = target.exists()
|
|
if had_live:
|
|
try:
|
|
os.replace(str(target), str(backup))
|
|
except OSError:
|
|
_print_managed_node_in_use_notice()
|
|
shutil.rmtree(staged, ignore_errors=True)
|
|
return None
|
|
# Rename preserves mtime: touch the backup (best-effort) so a concurrent heal's
|
|
# litter sweep never removes it mid-swap.
|
|
with contextlib.suppress(OSError):
|
|
os.utime(backup, None)
|
|
try:
|
|
os.replace(str(staged), str(target))
|
|
except OSError:
|
|
if had_live: # roll the live tree back
|
|
with contextlib.suppress(OSError):
|
|
os.replace(str(backup), str(target))
|
|
shutil.rmtree(staged, ignore_errors=True)
|
|
return False
|
|
if had_live:
|
|
# Locked files may keep the old tree on disk until the next heal; safe.
|
|
shutil.rmtree(backup, ignore_errors=True)
|
|
return True
|
|
|
|
|
|
def _heal_managed_node_windows(home: Path | None = None) -> bool | None:
|
|
"""Redownload the portable Node zip into ``%HERMES_HOME%\\node`` on Windows.
|
|
|
|
``True`` on success, ``False`` on genuine failure (offline, bad archive), ``None`` when the
|
|
tree is in use and the heal is deferred — callers must not record the once-per-process attempt
|
|
for ``None``. Staging-first (extract to ``node.new-*``, rename live aside, rename staged in) so
|
|
an interrupted heal cannot gut the install; a refused rename *is* the in-use signal.
|
|
|
|
The replacement is staging-first: the new tree is fully downloaded and extracted to a sibling
|
|
``node.new-*`` directory, then the live tree is renamed aside (``node.old-*``) and the staged tree
|
|
renamed into place. The live tree is never deleted before its replacement is ready, so an interrupted
|
|
heal cannot gut the running installation. Windows allows renaming a tree whose executables are running
|
|
(images are mapped with ``FILE_SHARE_DELETE`` — the same mechanism as the hermes.exe quarantine); when
|
|
the OS refuses the rename, that refusal *is* the in-use signal and the heal defers instead of forcing
|
|
the write and crashing with ``PermissionError: [WinError 5]`` on ``npm.cmd`` (#80926).
|
|
"""
|
|
import time
|
|
|
|
arch = (os.environ.get("PROCESSOR_ARCHITEW6432") or os.environ.get("PROCESSOR_ARCHITECTURE", "")).lower()
|
|
node_arch = {"amd64": "x64", "x86_64": "x64", "arm64": "arm64", "x86": "x86"}.get(arch)
|
|
if node_arch is None:
|
|
return False
|
|
home = home or get_hermes_home()
|
|
target = home / "node"
|
|
# Cheap pre-check; the rename-based swap is the authoritative guard.
|
|
if managed_node_tree_in_use(home):
|
|
_print_managed_node_in_use_notice()
|
|
return None
|
|
# Sweep litter from interrupted runs; only dirs >10 min old so a concurrent in-flight swap survives.
|
|
cutoff = time.time() - 600
|
|
for stale in (*home.glob("node.old-*"), *home.glob("node.new-*")):
|
|
try:
|
|
if stale.stat().st_mtime < cutoff:
|
|
shutil.rmtree(stale, ignore_errors=True)
|
|
except OSError:
|
|
continue
|
|
staged = _stage_windows_node_zip(home, node_arch)
|
|
if staged is None:
|
|
return False
|
|
return _swap_node_tree(target, staged) and node_tool_runnable(str(target / "node.exe"))
|
|
|
|
|
|
def _run_node_bootstrap(func: str, *, timeout: int, **extra_env: str) -> bool:
|
|
"""Source ``scripts/lib/node-bootstrap.sh`` and run shell function *func*."""
|
|
if not _NODE_BOOTSTRAP_SCRIPT.is_file():
|
|
return False
|
|
import subprocess
|
|
try:
|
|
result = subprocess.run(
|
|
["bash", "-c", f'source "{_NODE_BOOTSTRAP_SCRIPT}" && {func}'],
|
|
env={**os.environ, "HERMES_HOME": str(get_hermes_home()), **extra_env},
|
|
capture_output=True, timeout=timeout, check=False,
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return False
|
|
return result.returncode == 0
|
|
|
|
|
|
def bootstrap_hermes_managed_node() -> str | None:
|
|
"""Install a Hermes-managed Node tree under ``$HERMES_HOME/node`` and return its npm path.
|
|
|
|
Hermes never modifies a user-owned toolchain (system, nvm, brew, Nix) that fails ``engines``.
|
|
"""
|
|
existing = find_hermes_node_executable("npm")
|
|
if existing:
|
|
return existing
|
|
if sys.platform == "win32":
|
|
ok = _heal_managed_node_windows()
|
|
else:
|
|
# HERMES_NODE_SKIP_LINKS=1 keeps node/npm/npx out of ~/.local/bin: never shadow the user toolchain.
|
|
ok = _run_node_bootstrap("_nb_install_bundled_node", timeout=600, HERMES_NODE_SKIP_LINKS="1")
|
|
if not ok:
|
|
return None
|
|
return _first_runnable_managed(_candidate_node_command_names("npm"))[0]
|
|
|
|
|
|
def heal_hermes_managed_node() -> bool:
|
|
"""Redownload Hermes-managed Node when the tree exists but is broken; at most once per process.
|
|
|
|
A Windows in-use deferral does NOT record the attempt so a later call can heal once free.
|
|
|
|
POSIX installs shell out to ``heal_managed_node`` in ``scripts/lib/node-bootstrap.sh``; Windows
|
|
downloads the portable zip directly (same source as ``install.ps1``). See #80926.
|
|
"""
|
|
global _managed_node_heal_attempted
|
|
if _managed_node_heal_attempted or not hermes_managed_node_tree_present():
|
|
return False
|
|
if sys.platform == "win32":
|
|
result = _heal_managed_node_windows()
|
|
else:
|
|
result = _run_node_bootstrap("heal_managed_node", timeout=300)
|
|
if result is None: # in-use deferral: leave the attempt flag clear
|
|
return False
|
|
_managed_node_heal_attempted = True
|
|
return bool(result)
|
|
|
|
|
|
def _managed_node_tree_outdated(home: Path | None = None) -> bool:
|
|
"""True when the managed node runs but is below the target major (heals like a broken tree)."""
|
|
for candidate in _iter_managed_node_candidates(_candidate_node_command_names("node"), home):
|
|
result = _run_version_probe([str(candidate), "--version"])
|
|
if result is None:
|
|
return False # broken, not outdated — the runnable probe handles it
|
|
try:
|
|
version = result.stdout.decode().strip().lstrip("v")
|
|
major = int(version.split(".")[0])
|
|
except (ValueError, IndexError):
|
|
return False
|
|
# A pre-release is outdated whatever its major: nodejs.org publishes headers only for
|
|
# final releases, so node-gyp cannot build node-pty. Mirrors node_satisfies_build() in install.sh.
|
|
if "-" in version:
|
|
return True
|
|
return major < _HERMES_NODE_TARGET_MAJOR
|
|
return False
|
|
|
|
|
|
def find_hermes_node_executable(command: str) -> str | None:
|
|
"""Hermes-managed Node/npm path, healing broken/outdated trees; heal failure still returns old Node."""
|
|
names = _candidate_node_command_names(command)
|
|
resolved, broken_present = _first_runnable_managed(names)
|
|
needs_heal = broken_present or (resolved is not None and _managed_node_tree_outdated())
|
|
if needs_heal and heal_hermes_managed_node():
|
|
healed, _ = _first_runnable_managed(names)
|
|
if healed:
|
|
return healed
|
|
return resolved
|
|
|
|
|
|
def find_node_executable_on_path(command: str) -> str | None:
|
|
"""Node/npm from PATH; on Windows prefer ``.cmd``/``.exe`` (CreateProcess cannot run the bare shim)."""
|
|
if sys.platform != "win32":
|
|
return shutil.which(command)
|
|
command_str = str(command)
|
|
if any(sep and sep in command_str for sep in (os.sep, os.altsep, "/", "\\")):
|
|
return command_str if Path(command_str).is_file() else None
|
|
directories = [d for d in os.environ.get("PATH", "").split(os.pathsep) if d]
|
|
for name in _candidate_node_command_names(command_str):
|
|
for directory in directories:
|
|
if (Path(directory) / name).is_file():
|
|
return str(Path(directory) / name)
|
|
return None
|
|
|
|
|
|
def find_node_executable(command: str) -> str | None:
|
|
"""Resolve a Node command, preferring a healthy managed install.
|
|
|
|
A managed tree that exists but cannot be healed yields ``None`` rather than system Node.
|
|
"""
|
|
managed = find_hermes_node_executable(command)
|
|
if managed:
|
|
return managed
|
|
if hermes_managed_node_tree_present():
|
|
return None
|
|
return find_node_executable_on_path(command)
|
|
|
|
|
|
def with_hermes_node_path(env: dict[str, str] | None = None) -> dict[str, str]:
|
|
"""Return *env* with Hermes-managed Node directories prepended to PATH."""
|
|
merged = dict(os.environ if env is None else env)
|
|
parts = [p for p in merged.get("PATH", "").split(os.pathsep) if p]
|
|
for entry in reversed([str(path) for path in iter_hermes_node_dirs() if path.is_dir()]):
|
|
if entry not in parts:
|
|
parts.insert(0, entry)
|
|
merged["PATH"] = os.pathsep.join(parts)
|
|
return merged
|
|
|
|
|
|
def agent_browser_runnable(path: str | None) -> bool:
|
|
"""True when *path* is an agent-browser CLI that runs (``--version`` exits 0) or the npx fallback.
|
|
|
|
Dead/wrong-arch/hung binaries are rejected so callers try the next candidate.
|
|
|
|
A bare presence check (``shutil.which`` / ``Path.exists``) is not enough: agent-browser's npm
|
|
``postinstall`` re-points a *global* install symlink (e.g. ``/opt/homebrew/bin/agent-browser``) at our
|
|
local ``node_modules/agent-browser/bin/...`` binary, which then disappears on the next ``hermes update``
|
|
— leaving a **dangling symlink** that ``which`` still reports but exec fails on with exit 127 (issue
|
|
#48521). Callers that trust such a path silently break every browser tool.
|
|
"""
|
|
if not path:
|
|
return False
|
|
# The npx fallback is a two-token command string, not a path; npx validates at run time.
|
|
if " " in path and path.split()[0].endswith("npx"):
|
|
return True
|
|
return _is_executable_file(path) and _version_probe_ok(path)
|
|
|
|
|
|
def _legacy_path_has_content(path: Path) -> bool:
|
|
"""True iff *path* is a non-directory file or a populated directory.
|
|
|
|
Non-not-found ``OSError`` means "assume occupied" (never orphan legacy data). Symlinks are
|
|
judged on their target; a dangling one does NOT count.
|
|
"""
|
|
try:
|
|
st = path.lstat()
|
|
if stat.S_ISLNK(st.st_mode):
|
|
st = path.stat() # judge a symlink on its target; dangling → FileNotFoundError
|
|
except FileNotFoundError:
|
|
return False
|
|
except OSError: # e.g. PermissionError on a parent: assume occupied
|
|
return True
|
|
if not stat.S_ISDIR(st.st_mode):
|
|
return True
|
|
try:
|
|
next(path.iterdir())
|
|
except StopIteration:
|
|
return False
|
|
except OSError:
|
|
pass
|
|
return True
|
|
|
|
|
|
def display_hermes_home(home: Path | None = None) -> str:
|
|
"""User-facing ``~/`` display string for HERMES_HOME (``~/.hermes/profiles/coder``).
|
|
|
|
``home`` overrides the lookup for callers that run before the CLI has applied the sticky
|
|
``active_profile`` (``get_hermes_home()`` would emit the wrong-profile fallback warning there).
|
|
"""
|
|
if home is None:
|
|
home = get_hermes_home()
|
|
try: # as_posix(): str() on Windows yields chimeras like ~/AppData\Local\hermes/skills/
|
|
return "~/" + home.relative_to(Path.home()).as_posix()
|
|
except ValueError:
|
|
return str(home)
|
|
|
|
|
|
def profile_cli_selector() -> str:
|
|
"""``-p <name> `` (trailing space) pinning copy-pasteable ``hermes ...`` guidance to the
|
|
active NAMED profile, else ``""``: a bare ``hermes`` follows the sticky ``active_profile``
|
|
file, which can name a different database than the one that failed (#105887). A custom
|
|
home outside the profile tree has no selector (only HERMES_HOME names it)."""
|
|
name = profile_name_for_home(get_hermes_home())
|
|
return f"-p {name} " if name and name != "default" else ""
|
|
|
|
|
|
def secure_parent_dir(path: Path) -> None:
|
|
"""Chmod ``0o700`` on *path*'s parent, refusing ``/`` and top-level dirs (misresolved HERMES_HOME)."""
|
|
parent = path.parent.resolve()
|
|
if parent == Path("/") or len(parent.parts) < 3:
|
|
return
|
|
# Refuse the install tree: chmod 0700 breaks hermes-user traversal in Docker (UID 10000).
|
|
# A credential file here means HERMES_HOME misresolved; surface it (caused production lockouts).
|
|
# See #25821, #93050.
|
|
if parent == _INSTALL_ROOT or _INSTALL_ROOT in parent.parents:
|
|
import logging
|
|
|
|
logging.getLogger(__name__).warning(
|
|
"Not restricting permissions on %s: it is inside the "
|
|
"hermes-agent install directory (%s). Credential files are "
|
|
"normally stored under the hermes home directory instead.", parent, _INSTALL_ROOT,
|
|
)
|
|
return
|
|
with contextlib.suppress(OSError):
|
|
os.chmod(parent, 0o700)
|
|
|
|
|
|
def _norm_home_path(path: str | None) -> str:
|
|
"""Return a comparable absolute path string, or ``""`` for empty input."""
|
|
raw = (path or "").strip()
|
|
if not raw:
|
|
return ""
|
|
try:
|
|
return os.path.normcase(os.path.abspath(os.path.expanduser(raw)))
|
|
except Exception:
|
|
return os.path.normcase(raw)
|
|
|
|
|
|
def _profile_home_path(env: dict[str, str] | None = None) -> str | None:
|
|
"""Return ``{HERMES_HOME}/home`` when the profile-home directory exists."""
|
|
hermes_home = get_hermes_home_override() or (env or {}).get("HERMES_HOME") or os.getenv("HERMES_HOME")
|
|
if not hermes_home:
|
|
return None
|
|
profile_home = str(_expand_hermes_home(hermes_home) / "home")
|
|
return profile_home if os.path.isdir(profile_home) else None
|
|
|
|
|
|
def _is_profile_home(candidate: str | None, profile_home: str | None) -> bool:
|
|
return bool(candidate and profile_home and _norm_home_path(candidate) == _norm_home_path(profile_home))
|
|
|
|
|
|
def _env_get(env: dict[str, str], key: str, default: str = "") -> str:
|
|
"""Stripped *key* from *env*, falling back to the process environment."""
|
|
return str(env.get(key) or os.getenv(key, default)).strip()
|
|
|
|
|
|
def _iter_real_home_candidates(env: dict[str, str] | None = None) -> list[str]:
|
|
"""Return likely OS-user home candidates in trust order."""
|
|
env = env or {}
|
|
candidates = [_env_get(env, "HERMES_REAL_HOME"), _env_get(env, "HOME")]
|
|
with contextlib.suppress(Exception):
|
|
import pwd
|
|
candidates.append(pwd.getpwuid(os.getuid()).pw_dir.strip()) # windows-footgun: ok — POSIX-only module inside try/except
|
|
candidates.append(_env_get(env, "USERPROFILE"))
|
|
drive, path = _env_get(env, "HOMEDRIVE"), _env_get(env, "HOMEPATH")
|
|
if drive and path:
|
|
candidates.append(f"{drive}{path}" if path.startswith(("\\", "/")) else os.path.join(drive, path))
|
|
expanded = os.path.expanduser("~")
|
|
if expanded != "~":
|
|
candidates.append(expanded)
|
|
return [c for c in candidates if c]
|
|
|
|
|
|
def get_real_home(env: dict[str, str] | None = None) -> str:
|
|
"""The OS user's real home, avoiding the Hermes profile HOME.
|
|
|
|
``HOME`` belongs to the OS account and external CLIs keeping credentials under ``~``; a parent
|
|
already running with ``HOME={HERMES_HOME}/home`` is repaired back when possible.
|
|
"""
|
|
profile_home = _profile_home_path(env)
|
|
seen: set[str] = set()
|
|
for candidate in _iter_real_home_candidates(env):
|
|
key = _norm_home_path(candidate)
|
|
if not key or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
if not _is_profile_home(candidate, profile_home):
|
|
return candidate
|
|
import tempfile
|
|
try:
|
|
return tempfile.gettempdir()
|
|
except (RuntimeError, OSError):
|
|
# no HOME/USERPROFILE at all (env-less child on Windows): tempfile cannot expand ``~``
|
|
return "/tmp" # no-tmp: ok — last-resort fallback for an env with no home; not a write target we choose
|
|
|
|
|
|
_HOME_MODE_ALIASES = {"isolated": "profile", "profile_home": "profile", "profile-home": "profile",
|
|
"host": "real", "user": "real", "real_home": "real", "real-home": "real"}
|
|
|
|
|
|
def get_subprocess_home(env: dict[str, str] | None = None) -> str | None:
|
|
"""Subprocess ``HOME`` override, or ``None``.
|
|
|
|
``auto``: hosts keep real HOME (repairing a profile-home parent), containers use
|
|
``{HERMES_HOME}/home``; ``real``: always real HOME; ``profile``: always the profile home.
|
|
"""
|
|
env = env or {}
|
|
profile_home = _profile_home_path(env)
|
|
mode = _env_get(env, "TERMINAL_HOME_MODE", "auto").lower() or "auto"
|
|
mode = _HOME_MODE_ALIASES.get(mode, mode)
|
|
|
|
if mode == "profile":
|
|
return profile_home
|
|
real_home = get_real_home(env)
|
|
current_home = _env_get(env, "HOME")
|
|
repaired = real_home if _norm_home_path(real_home) != _norm_home_path(current_home) else None
|
|
if mode == "real":
|
|
return repaired
|
|
|
|
if profile_home and is_container():
|
|
return profile_home
|
|
if _is_profile_home(current_home, profile_home):
|
|
return repaired
|
|
return None
|
|
|
|
|
|
def apply_subprocess_home_env(env: MutableMapping[str, str]) -> None:
|
|
"""Apply Hermes' subprocess HOME contract to *env* in-place: ``HOME``/``HERMES_REAL_HOME``
|
|
per the home mode, and the temp vars re-pointed at ``env["HERMES_HOME"]``'s scratch dir."""
|
|
real_home = get_real_home(env)
|
|
if real_home:
|
|
env["HERMES_REAL_HOME"] = real_home
|
|
home = get_subprocess_home(env)
|
|
if home:
|
|
env["HOME"] = home
|
|
apply_scratch_tmp_env(env)
|
|
|
|
|
|
# --- Scratch dir: Hermes' own temp space, never the system /tmp ---
|
|
# System temp is tmpfs on most Linux distros and containers, so browser profiles, PTY probes,
|
|
# download spools and every ``tempfile.mkdtemp()`` a Hermes-launched script performs eat RAM
|
|
# and vanish on reboot. ``HERMES_HOME/cache/scratch`` is real storage with a fixed retention.
|
|
SCRATCH_TMP_ENV_VARS = ("TMPDIR", "TMP", "TEMP")
|
|
SCRATCH_DIR_MARKER_ENV = "HERMES_SCRATCH_DIR"
|
|
SCRATCH_MAX_AGE_HOURS = 72
|
|
_SCRATCH_PRUNE_STAMP = ".last_prune"
|
|
_SCRATCH_PRUNE_INTERVAL_SECONDS = 3600
|
|
_scratch_pruned_once = False
|
|
|
|
# AF_UNIX socket paths cap at 104 bytes (macOS) / 108 (Linux). Chrome appends
|
|
# ``com.google.Chrome.XXXXXX/SingletonSocket`` (~45) and the code kernel
|
|
# ``hermes_rpc_<32 hex>.sock`` (~49) to the temp root, so a root longer than this budget
|
|
# makes the bind fail (Chrome: "Socket path too long" at startup).
|
|
SOCKET_TMPDIR_MAX_LEN = 50
|
|
|
|
|
|
def socket_safe_tmpdir() -> str:
|
|
"""Temp root short enough for AF_UNIX sockets. The scratch dir usually fits; macOS
|
|
``TMPDIR`` never does and a deep profile home may not, so those fall back to the OS
|
|
default root for sockets only (everything else stays in the scratch dir)."""
|
|
import tempfile
|
|
if sys.platform == "darwin":
|
|
return "/tmp" # no-tmp: ok — AF_UNIX 104-byte socket path limit on darwin
|
|
candidate = tempfile.gettempdir()
|
|
if len(candidate) <= SOCKET_TMPDIR_MAX_LEN or not os.path.isdir("/tmp"): # no-tmp: ok — probe, not a write target
|
|
return candidate
|
|
return "/tmp" # no-tmp: ok — AF_UNIX 108-byte socket path limit on Linux
|
|
|
|
|
|
def get_scratch_dir(home: str | Path | None = None, *, prune: bool = True) -> Path:
|
|
"""``<home>/cache/scratch`` (created, owner-only); *home* defaults to the active Hermes home.
|
|
|
|
Every Hermes process and child gets ``TMPDIR``/``TMP``/``TEMP`` pointed here at boot (see
|
|
:func:`export_scratch_tmp_env`), so ``tempfile`` defaults land here without call sites
|
|
knowing. Entries older than ``SCRATCH_MAX_AGE_HOURS`` are pruned at most once per process
|
|
and once per hour across processes (stamp file), so a fan-out of children stays cheap.
|
|
"""
|
|
base = Path(home) if home is not None else get_hermes_home()
|
|
scratch = base / "cache" / "scratch"
|
|
try:
|
|
scratch.mkdir(parents=True, exist_ok=True)
|
|
if sys.platform != "win32":
|
|
os.chmod(scratch, 0o700)
|
|
except OSError:
|
|
pass
|
|
if prune:
|
|
_prune_scratch_dir_once(scratch)
|
|
return scratch
|
|
|
|
|
|
def prune_scratch_dir(scratch: Path | None = None, max_age_hours: float = SCRATCH_MAX_AGE_HOURS) -> int:
|
|
"""Delete top-level scratch entries untouched for *max_age_hours*; return the count removed."""
|
|
import time
|
|
root = scratch if scratch is not None else get_scratch_dir(prune=False)
|
|
cutoff = time.time() - max_age_hours * 3600
|
|
removed = 0
|
|
try:
|
|
entries = list(root.iterdir())
|
|
except OSError:
|
|
return 0
|
|
for entry in entries:
|
|
if entry.name == _SCRATCH_PRUNE_STAMP:
|
|
continue
|
|
try:
|
|
if entry.lstat().st_mtime >= cutoff:
|
|
continue
|
|
if entry.is_dir() and not entry.is_symlink():
|
|
shutil.rmtree(entry, ignore_errors=True)
|
|
else:
|
|
entry.unlink()
|
|
removed += 1
|
|
except OSError:
|
|
continue
|
|
return removed
|
|
|
|
|
|
def _prune_scratch_dir_once(scratch: Path) -> None:
|
|
global _scratch_pruned_once
|
|
if _scratch_pruned_once:
|
|
return
|
|
_scratch_pruned_once = True
|
|
import time
|
|
stamp = scratch / _SCRATCH_PRUNE_STAMP
|
|
try:
|
|
if time.time() - stamp.stat().st_mtime < _SCRATCH_PRUNE_INTERVAL_SECONDS:
|
|
return
|
|
except OSError:
|
|
pass
|
|
with contextlib.suppress(Exception):
|
|
stamp.touch()
|
|
prune_scratch_dir(scratch)
|
|
|
|
|
|
def scratch_dir_usage_bytes(scratch: Path | None = None) -> int:
|
|
"""Total bytes under the scratch dir (for ``hermes doctor``); 0 when unreadable."""
|
|
root = scratch if scratch is not None else get_scratch_dir(prune=False)
|
|
total = 0
|
|
for dirpath, _dirnames, filenames in os.walk(root, onerror=lambda _e: None):
|
|
for name in filenames:
|
|
with contextlib.suppress(OSError):
|
|
total += os.lstat(os.path.join(dirpath, name)).st_size
|
|
return total
|
|
|
|
|
|
def apply_scratch_tmp_env(env: MutableMapping[str, str]) -> bool:
|
|
"""Point ``TMPDIR``/``TMP``/``TEMP`` in *env* at the scratch dir of ``env["HERMES_HOME"]``.
|
|
|
|
A temp var the user (or the OS: macOS ``/var/folders``, Windows ``%TEMP%``) set is
|
|
respected and nothing changes. A value Hermes itself exported earlier — recognisable
|
|
because it equals ``HERMES_SCRATCH_DIR`` — is re-derived, so a child running under another
|
|
profile's home gets that home's scratch dir rather than its parent's. Returns True when
|
|
the vars were (re)written.
|
|
"""
|
|
ours = env.get(SCRATCH_DIR_MARKER_ENV, "")
|
|
for key in SCRATCH_TMP_ENV_VARS:
|
|
value = env.get(key, "").strip()
|
|
if value and value != ours:
|
|
return False
|
|
home = env.get("HERMES_HOME", "").strip()
|
|
try:
|
|
scratch = str(get_scratch_dir(_expand_hermes_home(home) if home else get_process_hermes_home()))
|
|
except (RuntimeError, OSError):
|
|
# No HERMES_HOME and no resolvable user home (a child env built from nothing on
|
|
# Windows): there is no scratch dir to point at; the child keeps the OS default.
|
|
return False
|
|
for key in SCRATCH_TMP_ENV_VARS:
|
|
env[key] = scratch
|
|
env[SCRATCH_DIR_MARKER_ENV] = scratch
|
|
return True
|
|
|
|
|
|
def export_scratch_tmp_env() -> bool:
|
|
"""Boot hook: apply :func:`apply_scratch_tmp_env` to this process and reset ``tempfile``'s
|
|
cached default so ``tempfile.gettempdir()`` follows. Call again after anything that
|
|
re-homes the process (``--profile`` resolution); a user-set temp var is never overridden."""
|
|
changed = apply_scratch_tmp_env(os.environ)
|
|
if changed:
|
|
import tempfile
|
|
tempfile.tempdir = None
|
|
return changed
|
|
|
|
|
|
VALID_REASONING_EFFORTS = ("minimal", "low", "medium", "high", "xhigh", "max", "ultra")
|
|
|
|
|
|
def parse_reasoning_effort(effort) -> dict | None:
|
|
"""Parse a reasoning effort level into a config dict.
|
|
|
|
``None`` for empty/unrecognized input (caller uses the default); ``{"enabled": False}`` for
|
|
"none"/"false"/"disabled"/YAML False — ``reasoning_effort: false`` must mean disabled.
|
|
|
|
The dict form ``{"enabled": true, "effort": "<level>"}`` passes ``effort`` through verbatim so
|
|
providers with bespoke thinking tiers (``fast``/``thinking`` relays) can be asked for their real
|
|
level; bare strings stay strict so a typo like ``hgih`` never reaches the wire. The wire layer
|
|
already tolerates unknown names (``agent.reasoning_effort.clamp_effort``).
|
|
"""
|
|
if effort is None or effort is True:
|
|
return None
|
|
if isinstance(effort, dict):
|
|
if effort.get("enabled", True) is False:
|
|
return {"enabled": False}
|
|
# ``or ""``: a falsy effort (0/False) is "no level", never the string "0" on the wire.
|
|
level = str(effort.get("effort") or "").strip()
|
|
return {"enabled": True, "effort": level} if level else None
|
|
effort = str(effort).strip().lower() # False -> "false" -> disabled; "" matches neither set
|
|
if effort in {"none", "false", "disabled"}:
|
|
return {"enabled": False}
|
|
if effort in VALID_REASONING_EFFORTS:
|
|
return {"enabled": True, "effort": effort}
|
|
return None
|
|
|
|
|
|
def _canonical_model_variants(model: str) -> list[str]:
|
|
"""Spelling variants for tolerant override matching, exact first, deduped in order.
|
|
|
|
Dot/dash recovery runs on EACH base form so ``x-4.5``, ``x-4-5`` and ``x.4.5`` share one variant set.
|
|
"""
|
|
_dash_to_dot = lambda s: re.sub(r'(\d)-(\d)', r'\1.\2', s)
|
|
_dot_to_dash = lambda s: re.sub(r'(\d)\.(\d)', r'\1-\2', s)
|
|
seen: set[str] = set()
|
|
variants: list[str] = []
|
|
|
|
def _add(*values):
|
|
for v in values:
|
|
if v and v not in seen:
|
|
seen.add(v)
|
|
variants.append(v)
|
|
|
|
def _add_with_derivatives(s):
|
|
dashed, dotted = s.replace('.', '-'), s.replace('-', '.')
|
|
_add(s, dashed, dotted, _dash_to_dot(s), _dot_to_dash(s), _dash_to_dot(dashed), _dot_to_dash(dotted))
|
|
_add_with_derivatives(model)
|
|
parts = model.split('/')
|
|
if len(parts) >= 2: # bare model (strip provider/aggregator prefix)
|
|
_add_with_derivatives(parts[-1])
|
|
if len(parts) >= 3: # strip aggregator only: "openrouter/anthropic/x" → "anthropic/x"
|
|
_add_with_derivatives('/'.join(parts[1:]))
|
|
known_providers = (
|
|
'anthropic', 'openai', 'google', 'openrouter', 'groq', 'mistral',
|
|
'xai', 'cohere', 'perplexity', 'together', 'fireworks', 'deepseek',
|
|
)
|
|
for v in [v for v in variants if '/' not in v]:
|
|
_add(*(f"{provider}/{v}" for provider in known_providers))
|
|
known_aggregators = ('openrouter', 'opencode', 'fireworks', 'groq', 'together')
|
|
for v in [v for v in variants if v.count('/') == 1]:
|
|
_add(*(f"{agg}/{v}" for agg in known_aggregators))
|
|
return variants
|
|
|
|
|
|
def resolve_per_model_reasoning_effort(model: str, overrides: dict | None) -> dict | None:
|
|
"""Per-model reasoning_effort override with spelling tolerance; first non-None parse wins.
|
|
|
|
Order: exact → dots↔dashes → provider stripped → aggregator stripped → known prefixes added →
|
|
reverse lookup of prefixed keys whose stripped forms match (custom provider slugs are not
|
|
enumerable, so a key like ``ollama-local/qwen3.6:27b`` must still match the bare
|
|
``qwen3.6:27b`` model string a fallback swap feeds after stripping the prefix).
|
|
"""
|
|
if not overrides or not isinstance(overrides, dict) or not model:
|
|
return None
|
|
variants = _canonical_model_variants(model)
|
|
for variant in variants:
|
|
if variant in overrides:
|
|
result = parse_reasoning_effort(overrides[variant])
|
|
if result is not None:
|
|
return result
|
|
# Reverse lookup: the key may carry a custom-provider prefix the model string lost
|
|
# (fallback entries and custom-provider resolution feed the bare slug, while the
|
|
# documented key spelling keeps the ``provider/model`` form). Direct and variant
|
|
# matches above still win, so provider-qualified keys stay most specific.
|
|
variant_set = set(variants)
|
|
for key, raw in overrides.items():
|
|
if not isinstance(key, str) or "/" not in key:
|
|
continue
|
|
parts = key.split("/")
|
|
key_forms = _canonical_model_variants(parts[-1])
|
|
if len(parts) >= 3:
|
|
key_forms += _canonical_model_variants("/".join(parts[1:]))
|
|
if any(form in variant_set for form in key_forms):
|
|
result = parse_reasoning_effort(raw)
|
|
if result is not None:
|
|
return result
|
|
return None
|
|
|
|
|
|
def resolve_per_model_provider_routing(model: str, models: dict | None) -> dict:
|
|
"""``provider_routing.models.<id>`` entry for *model*, spelling-tolerant like
|
|
``reasoning_overrides``; ``{}`` when none matches. Only the keys a user sets per model
|
|
are returned so unset ones fall through to the flat ``provider_routing`` values."""
|
|
if not model or not isinstance(models, dict):
|
|
return {}
|
|
for variant in _canonical_model_variants(model):
|
|
entry = models.get(variant)
|
|
if isinstance(entry, dict):
|
|
return entry
|
|
return {}
|
|
|
|
|
|
def resolve_reasoning_config(cfg: dict | None, model: str = "") -> dict | None:
|
|
"""Effective reasoning config for *model*: per-model override, then global ``agent.reasoning_effort``.
|
|
|
|
Single chokepoint for every surface (CLI, gateway, TUI, cron, ``/model``, fallback activation).
|
|
"""
|
|
cfg = cfg if isinstance(cfg, dict) else {}
|
|
agent_cfg = cfg.get("agent") if isinstance(cfg.get("agent"), dict) else {}
|
|
|
|
if not model:
|
|
model_cfg = cfg.get("model")
|
|
if isinstance(model_cfg, dict):
|
|
model_cfg = model_cfg.get("default") or model_cfg.get("model") or ""
|
|
model = model_cfg.strip() if isinstance(model_cfg, str) else ""
|
|
per_model = resolve_per_model_reasoning_effort(model, agent_cfg.get("reasoning_overrides") or {})
|
|
if per_model is not None:
|
|
return per_model
|
|
|
|
# Keep the raw value: ``or ""`` would turn a YAML False into "" and silently re-enable thinking.
|
|
effort = agent_cfg.get("reasoning_effort", "")
|
|
result = parse_reasoning_effort(effort)
|
|
if effort and str(effort).strip() and result is None:
|
|
import logging
|
|
logging.getLogger(__name__).warning("Unknown reasoning_effort '%s', using default (medium)", effort)
|
|
return result
|
|
|
|
|
|
def is_termux() -> bool:
|
|
"""True inside Termux (Android): ``TERMUX_VERSION`` or the Termux-specific ``PREFIX`` path."""
|
|
prefix = os.getenv("PREFIX", "")
|
|
return bool(os.getenv("TERMUX_VERSION") or "com.termux/files/usr" in prefix)
|
|
|
|
|
|
_wsl_detected: bool | None = None
|
|
|
|
|
|
def is_wsl() -> bool:
|
|
"""True inside WSL1/WSL2 (``microsoft`` marker in ``/proc/version``); cached per process."""
|
|
global _wsl_detected
|
|
if _wsl_detected is None:
|
|
try:
|
|
with open("/proc/version", "r", encoding="utf-8") as f:
|
|
_wsl_detected = "microsoft" in f.read().lower()
|
|
except Exception:
|
|
_wsl_detected = False
|
|
return _wsl_detected
|
|
|
|
|
|
def windows_path_to_wsl(path: str) -> str | None:
|
|
"""Convert a Windows drive path (``C:\\...``) to its ``/mnt/<drive>/...`` form."""
|
|
match = re.match(r"^([A-Za-z]):[\\/](.*)$", str(path or "").strip())
|
|
return f"/mnt/{match.group(1).lower()}/{match.group(2).replace(chr(92), '/')}" if match else None
|
|
|
|
|
|
def wsl_unc_path_to_posix(path: str) -> str | None:
|
|
"""Convert a ``\\\\wsl.localhost\\<distro>\\...`` (or legacy ``\\\\wsl$``) UNC path to POSIX."""
|
|
normalized = str(path or "").strip().replace("/", "\\")
|
|
match = re.match(r"^\\\\wsl(?:\.localhost|\$)\\[^\\]+\\(.*)$", normalized, re.IGNORECASE)
|
|
return "/" + match.group(1).replace("\\", "/") if match else None
|
|
|
|
|
|
def translate_cwd_for_wsl_backend(cwd: str) -> str:
|
|
"""Map a Windows-host cwd (drive path or ``\\\\wsl.localhost\\`` UNC) to POSIX when Hermes runs in WSL.
|
|
|
|
No-op off WSL and for paths already POSIX.
|
|
"""
|
|
if not is_wsl():
|
|
return cwd
|
|
for translator in (wsl_unc_path_to_posix, windows_path_to_wsl):
|
|
translated = translator(cwd)
|
|
if translated is not None:
|
|
return translated
|
|
return cwd
|
|
|
|
|
|
_container_detected: bool | None = None
|
|
|
|
|
|
def is_container() -> bool:
|
|
"""True inside a container (Docker/Podman/LXC/Kubernetes markers); cached per process.
|
|
|
|
See: NousResearch/hermes-agent#47111
|
|
"""
|
|
global _container_detected
|
|
if _container_detected is None:
|
|
_container_detected = _detect_container()
|
|
return _container_detected
|
|
|
|
|
|
def _read_proc(path: str) -> str:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return f.read()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def _proc_file_has_marker(path: str, markers: tuple[str, ...]) -> bool:
|
|
content = _read_proc(path)
|
|
return any(marker in content for marker in markers)
|
|
|
|
|
|
def _detect_container() -> bool:
|
|
if (
|
|
os.path.exists("/.dockerenv")
|
|
or os.path.exists("/run/.containerenv")
|
|
or os.environ.get("KUBERNETES_SERVICE_HOST")
|
|
or _proc_file_has_marker("/proc/1/cgroup", ("docker", "podman", "/lxc/", "kubepods", "containerd", "crio"))
|
|
):
|
|
return True
|
|
# cgroup v2: /proc/1/cgroup is just "0::/"; the runtime still shows in mountinfo — but ONLY on
|
|
# the root ("/") mount line. A host that merely *runs* containers exposes every container's
|
|
# overlay lowerdir (``lowerdir=/var/lib/containerd/...``) at non-root mount points, which a
|
|
# whole-file scan misread as "inside a container" and flipped subprocess HOME (#58135).
|
|
return _root_mount_has_marker("/proc/self/mountinfo", ("kubepods", "containerd", "crio"))
|
|
|
|
|
|
def _root_mount_has_marker(path: str, markers: tuple[str, ...]) -> bool:
|
|
"""mountinfo field 5 (index 4) is the mount point; only the root ("/") line is the process's own rootfs."""
|
|
root_lines = [line for line in _read_proc(path).splitlines() if len(f := line.split()) >= 5 and f[4] == "/"]
|
|
return any(marker in line for line in root_lines for marker in markers)
|
|
|
|
|
|
def get_config_path() -> Path:
|
|
"""Return the path to ``config.yaml`` under HERMES_HOME."""
|
|
return get_hermes_home() / "config.yaml"
|
|
|
|
|
|
def get_skills_dir() -> Path:
|
|
"""Return the path to the skills directory under HERMES_HOME."""
|
|
return get_hermes_home() / "skills"
|
|
|
|
|
|
def get_env_path() -> Path:
|
|
"""Return the path to the ``.env`` file under HERMES_HOME."""
|
|
return get_hermes_home() / ".env"
|
|
|
|
|
|
def apply_ipv4_preference(force: bool = False) -> None:
|
|
"""Monkey-patch ``socket.getaddrinfo`` to prefer IPv4 when *force* is True.
|
|
|
|
Broken-IPv6 hosts hang on AAAA for the full TCP timeout; ``AF_UNSPEC`` resolves as ``AF_INET``,
|
|
falling back to full resolution when no A record exists (pure-IPv6 hosts still work).
|
|
"""
|
|
if not force:
|
|
return
|
|
|
|
import socket
|
|
|
|
if getattr(socket.getaddrinfo, "_hermes_ipv4_patched", False):
|
|
return
|
|
_original_getaddrinfo = socket.getaddrinfo
|
|
|
|
def _ipv4_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
|
|
if family == 0: # AF_UNSPEC — caller didn't request a specific family
|
|
try:
|
|
return _original_getaddrinfo(host, port, socket.AF_INET, type, proto, flags)
|
|
except socket.gaierror: # no A record — pure-IPv6 host
|
|
return _original_getaddrinfo(host, port, family, type, proto, flags)
|
|
return _original_getaddrinfo(host, port, family, type, proto, flags)
|
|
_ipv4_getaddrinfo._hermes_ipv4_patched = True # type: ignore[attr-defined]
|
|
socket.getaddrinfo = _ipv4_getaddrinfo # type: ignore[assignment]
|
|
|
|
|
|
PARTIAL_STREAM_STUB_ID = "partial-stream-stub"
|
|
FINISH_REASON_LENGTH = "length"
|
|
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
|
|
OPENROUTER_MODELS_URL = f"{OPENROUTER_BASE_URL}/models"
|
|
|
|
# OpenRouter request-time routing variants (docs: guides/routing/model-variants).
|
|
# These suffixes are per-request routing modifiers valid on ANY model id —
|
|
# ":nitro" sorts the endpoint pool by throughput and admits priority-tier
|
|
# endpoints, ":floor" sorts by price and admits flex-tier endpoints, ":exacto"
|
|
# applies quality-first provider sorting, ":online" attaches the web plugin.
|
|
# They are never separate catalog entries: /models lists only the base id, so
|
|
# every catalog lookup must key on the BASE while the suffixed id stays on the
|
|
# wire.
|
|
# NOT in this set: ":free", ":batch", ":thinking", ":extended" — those ARE
|
|
# distinct catalog SKUs with their own /models entries (and their own context
|
|
# windows), so stripping them would resolve the wrong window.
|
|
OPENROUTER_VARIANT_SUFFIXES: frozenset[str] = frozenset(
|
|
{"nitro", "floor", "exacto", "online"}
|
|
)
|
|
|
|
|
|
def openrouter_variant_base(model_id: str) -> str | None:
|
|
"""Return the base model id when ``model_id`` carries a recognized
|
|
OpenRouter routing-variant suffix (e.g. ``x-ai/grok-4:nitro`` →
|
|
``x-ai/grok-4``), else ``None``.
|
|
|
|
Lives here rather than in ``hermes_cli.models`` so the metadata layer
|
|
(``agent.model_metadata``) can share one definition without importing the
|
|
CLI — this module is dependency-free by contract.
|
|
|
|
>>> openrouter_variant_base("x-ai/grok-4:nitro")
|
|
'x-ai/grok-4'
|
|
>>> openrouter_variant_base("x-ai/grok-4:free") is None
|
|
True
|
|
>>> openrouter_variant_base("x-ai/grok-4") is None
|
|
True
|
|
"""
|
|
base, sep, suffix = (model_id or "").rpartition(":")
|
|
if not sep or not base:
|
|
return None
|
|
if suffix.lower() in OPENROUTER_VARIANT_SUFFIXES:
|
|
return base
|
|
return None
|
|
|
|
|
|
AI_GATEWAY_BASE_URL = "https://ai-gateway.vercel.sh/v1"
|
|
|
|
|
|
def venv_bin_dir(venv_dir, *, windows: bool | None = None) -> Path:
|
|
"""Venv executable dir (``Scripts``/``bin``); *windows* lets tests exercise Windows paths on Linux.
|
|
|
|
Returned unconditionally — callers differ on whether a missing venv is an error.
|
|
|
|
Canonical helper for venv layout. This was open-coded in seven places across four ``hermes_cli`` modules
|
|
using three different Windows predicates (``platform.system()``, ``is_windows()``, ``_is_windows()``);
|
|
each new call site had to re-derive it, and #76091 shipped an eighth copy because the correct behaviour
|
|
lived 2400 lines away in another function. A few sites outside ``hermes_cli``
|
|
(``tools/code_execution_tool.py``, ``agent/lsp/install.py``, ``agent/lsp/servers.py``) still hand-roll
|
|
it — convert them as they are touched.
|
|
"""
|
|
windows = sys.platform == "win32" if windows is None else windows
|
|
return Path(venv_dir) / ("Scripts" if windows else "bin")
|
|
|
|
|
|
def project_venv_dir(project_root) -> Path | None:
|
|
"""The project's ``venv`` or ``.venv`` dir when one exists (``uv venv`` defaults to ``.venv``).
|
|
|
|
``uv venv`` defaults to ``.venv`` while our installers create ``venv``, so both layouts are in the wild.
|
|
Call sites that only knew about ``venv`` silently no-oped on a ``.venv`` install — that is how the
|
|
Windows shim-lock preflight skipped itself entirely (#79542). ``venv`` wins when both exist, matching
|
|
what the installers write.
|
|
"""
|
|
root = Path(project_root)
|
|
return next((root / n for n in ("venv", ".venv") if (root / n).is_dir()), None)
|
|
|
|
|
|
def venv_python_path(venv_dir, *, windows: bool | None = None) -> Path:
|
|
"""Path to the Python interpreter inside *venv_dir* (may not exist)."""
|
|
bin_dir = venv_bin_dir(venv_dir, windows=windows)
|
|
return bin_dir / ("python.exe" if bin_dir.name == "Scripts" else "python")
|
|
|
|
|
|
# First-party roots: an ImportError naming one means our own tree is inconsistent. The
|
|
# update post-probe shares this set so the guard that BLOCKS and the hint that EXPLAINS agree.
|
|
FIRST_PARTY_MODULE_ROOTS = frozenset({
|
|
"agent", "acp_adapter", "cli", "cron", "gateway", "model_tools", "plugins",
|
|
"providers", "tools", "toolsets", "run_agent", "tui_gateway", "utils",
|
|
})
|
|
|
|
|
|
def is_first_party_module(name: str | None) -> bool:
|
|
"""True when *name* ships with Hermes (exact first segment; ``startswith`` would claim ``agentops``)."""
|
|
root = str(name).split(".")[0] if name else ""
|
|
return bool(root) and (root in FIRST_PARTY_MODULE_ROOTS or root.startswith("hermes_"))
|
|
|
|
|
|
def partial_update_hint(exc: BaseException) -> list[str]:
|
|
"""Recovery guidance lines when *exc* looks like a half-updated tree, else ``[]``."""
|
|
# A missing third-party dep (bad venv, missing extra) is a different problem.
|
|
if (not isinstance(exc, ImportError) or isinstance(exc, ModuleNotFoundError)
|
|
or not is_first_party_module(getattr(exc, "name", None))):
|
|
return []
|
|
return [
|
|
"",
|
|
"This looks like a partially-updated install: one module was refreshed "
|
|
"and a related one was not.",
|
|
"Re-run the update to bring the whole tree to the same version:",
|
|
" hermes update",
|
|
"If that also fails, reinstall: https://hermes-agent.nousresearch.com",
|
|
]
|
|
|
|
|
|
def emit_partial_update_hint(exc: BaseException, *, file=None) -> bool:
|
|
"""Print recovery guidance for a half-updated tree."""
|
|
lines = partial_update_hint(exc)
|
|
if not lines:
|
|
return False
|
|
for line in (f"Error: {exc}", *lines):
|
|
print(line, file=sys.stderr if file is None else file)
|
|
return True
|