hermes_constants.py

"""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 Callable, Mapping, 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"

# Browser-hosted attachments are staged on the Hermes host and then read back
# as a base64 data URL for the normal remote attachment RPC. Keep the upload
# and read-back boundaries identical so a successful picker/drop can always be
# attached afterward.
WEBAPP_ATTACHMENT_MAX_BYTES: int = 16 * 1024 * 1024


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 _platform_default_hermes_home(environ: Mapping[str, str], user_home: Callable[[], Path]) -> Path:
    """The platform default that *environ* names, with the literal data-directory suffix."""
    suffix = environ.get("HERMES_DATA_DIR_SUFFIX", "")
    if sys.platform == "win32":
        local_appdata = environ.get("LOCALAPPDATA", "").strip()
        base = Path(local_appdata) if local_appdata else user_home() / "AppData" / "Local"
        return base / ("hermes" + suffix)
    return user_home() / (".hermes" + suffix)


def _get_platform_default_hermes_home() -> Path:
    """Return the platform default with the literal data-directory suffix."""
    return _platform_default_hermes_home(os.environ, Path.home)


def _account_home() -> Path:
    """The OS account's home from the password database, never ``$HOME``: harnesses and profile isolation
    repoint HOME (root under ``sudo`` has uid 0 too, which keeps the #108674 remap). Falls back to the
    shared real-home resolver where pwd cannot answer."""
    try:
        import pwd
        return Path(pwd.getpwuid(os.getuid()).pw_dir)  # windows-footgun: ok — POSIX-only module inside try/except
    except (ImportError, KeyError, OSError, AttributeError):
        return Path(get_real_home())


def account_default_hermes_home() -> Path:
    """The platform default of the OS ACCOUNT, never of a foreign process ``HOME``.

    Host service identities (``systemctl --user`` units, LaunchAgents) belong to the account, and their
    definitions are found under the account home. A process started with ``HOME=<scratch>`` used to call
    ``<scratch>/.hermes`` native, name the bare ``hermes-gateway`` unit, and stop the account's real
    gateway through it. When the process HOME is the account home the two answers are the same home.
    """
    account = _account_home()
    try:
        same = Path.home().resolve() == account.resolve()
    except (OSError, RuntimeError):
        same = False
    return _get_platform_default_hermes_home() if same else _platform_default_hermes_home(os.environ, lambda: account)


def _hermes_root_of(native_home: Path, env_path: Path | None) -> Path:
    """The root that holds *env_path*: the native home, a custom root, or *env_path* itself."""
    if env_path is None:
        return native_home
    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
        return env_path.parent.parent if env_path.parent.name == "profiles" else env_path
    return native_home


_ENV_REFERENCE = re.compile(r"%([^%]+)%|\$\{([^}]+)\}|\$([A-Za-z0-9_-]+)")


def hermes_root_for_environment(environ: Mapping[str, str]) -> Path | None:
    """The Hermes root that a process started with *environ* resolves, or None.

    Applies the root rules of ``get_default_hermes_root`` to another process's
    environment. ``%NAME%``, ``${NAME}`` and ``$NAME`` references and a leading
    ``~`` expand from *environ* alone, never from this process. Returns None
    when the answer depends on anything outside *environ*: a relative
    HERMES_HOME, an undefined variable, or a missing user profile folder.
    """
    names = {key.upper(): value for key, value in environ.items()}
    profile = names.get("USERPROFILE" if sys.platform == "win32" else "HOME", "").strip()

    def user_home() -> Path:
        if not profile:
            raise LookupError("no user profile folder")
        return Path(profile)

    raw = names.get("HERMES_HOME", "").strip()
    try:
        native_home = _platform_default_hermes_home(names, user_home)
        if not raw:
            return native_home
        expanded = _ENV_REFERENCE.sub(lambda match: names[match.group(match.lastindex).upper()], raw)
        if expanded == "~" or expanded.startswith(("~/", "~\\")):
            expanded = str(user_home()) + expanded[1:]
    except LookupError:
        return None
    env_path = Path(expanded)
    return _hermes_root_of(native_home, env_path) if env_path.is_absolute() else None


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
    # Latch on the FIRST check regardless of outcome (one-shot contract). Previously the latch
    # was only set on the warning branch, so with no active_profile (or "default") the stat +
    # read re-ran on every get_hermes_home() call (#90065).
    _profile_fallback_warned = True
    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":
        # 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``). Follows
    ``HERMES_HOME`` live on purpose: routed-profile DECISIONS compare against
    :func:`get_routing_process_hermes_home` instead (#119242).
    """
    val = os.environ.get("HERMES_HOME", "").strip()
    return _expand_hermes_home(val) if val else _get_platform_default_hermes_home()


# Host-pinned identity of the profile this process serves as its own (None: follow HERMES_HOME).
_PINNED_PROCESS_HERMES_HOME: str | None = None


def pin_process_hermes_home(path: str | Path | None) -> None:
    """Pin the home this process serves as its own profile, for "is this task routed?" decisions.

    An embedding host that serves several profiles and mirrors the active turn's profile into
    ``os.environ["HERMES_HOME"]`` for legacy readers (Hermes WebUI) otherwise makes every turn's own
    profile look like the launch profile: ``agent.secret_scope.serves_routed_profile()`` turns
    False and that turn's MCP connections fall back to bare, cross-profile names; the sibling
    launch-home checks (``secret_scope._is_process_home``, ``tools.environments.local._is_routed_home``,
    ``hermes_cli.env_loader._process_hermes_home``) misjudge the same way. ``None`` clears the pin.

    Process-global on purpose: it names the process's own identity, not a per-task value. It is NOT
    folded into :func:`get_process_hermes_home`: :func:`get_hermes_home` falls back to that for
    tasks carrying no override, and the host's mirror exists precisely so those readers see the
    served profile. Hosts that never mutate ``HERMES_HOME`` need not call this (no-op).
    """
    global _PINNED_PROCESS_HERMES_HOME
    _PINNED_PROCESS_HERMES_HOME = None if path is None else str(path)


def process_hermes_home_is_pinned() -> bool:
    return _PINNED_PROCESS_HERMES_HOME is not None


def get_routing_process_hermes_home() -> Path:
    """Launch home for routed-profile decisions: the pinned home, else :func:`get_process_hermes_home`."""
    pinned = _PINNED_PROCESS_HERMES_HOME
    return _expand_hermes_home(pinned) if pinned else get_process_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"})

# Where connector tool results save the files they return, under each profile's home.
CONNECTOR_DOWNLOADS_SUBDIR = Path("downloads", "connectors")

# 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(*, home: str | Path | None = None) -> Path:
    """Root of an explicit home, or the process home when none is supplied."""
    global _default_hermes_root_memo
    native_home = _get_platform_default_hermes_home()
    env_home = str(home).strip() if home is not None else 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 = _hermes_root_of(native_home, 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")
# Canonical named-profile id grammar; every profile-directory gate imports this one object.
PROFILE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")


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_has_servable_identity(profile_home: str | Path) -> bool:
    """Stricter than :func:`named_profile_has_identity`: is this dir a profile a host should change
    its own posture for?

    An EMPTY ``.env`` is all a crashed ``hermes profile create`` leaves behind, and it is enough for
    ``named_profile_has_identity``. Listing such a shell is harmless; counting it as a second tenant
    is not — it flips the whole host's credential reads fail-closed at the next boot. Every other
    marker, and a non-empty or symlinked ``.env``, still counts.
    """
    home = Path(profile_home)
    for marker in _PROFILE_IDENTITY_MARKERS:
        path = home / marker
        if path.is_symlink():
            return True
        try:
            if path.is_file() and (marker != ".env" or path.stat().st_size > 0):
                return True
        except OSError:
            continue
    return False


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)
    # Lazy so this module still imports alone; hermes_dir_acl (stdlib only) ships beside it in every
    # closure that creates home dirs, and a missing copy fails loudly instead of silently skipping the
    # elevated grant (tests/pm/test_bootstrap_import_closure.py).
    from hermes_dir_acl import make_user_dir
    make_user_dir(target, 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]:
    """Historical layout for old-updater diagnostics, never runtime selection."""
    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 _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(  # noqa: PLW1510 -- forwarding wrapper: callers pass subprocess kwargs via **kwargs (check may arrive through it); an explicit check=False would raise TypeError
            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


# 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:
    """Read-only legacy artifact detection for already-running old updaters."""
    names = [n for c in ("node", "npm", "npx") for n in _candidate_node_command_names(c)]
    return any((directory / name).is_file() for directory in iter_hermes_node_dirs(home) for name in names)


def find_node_executable(command: str) -> str | None:
    """Read PM's selected Node/npm/npx; ``None`` when PM has not installed it.

    Never falls back to the user's PATH copy (callers ``pm.ensure`` on ``None``):
    mixing toolchains breaks native-addon ABIs and npm caches. Explicit
    executable paths remain caller-owned. Discovery never installs, probes,
    repairs, or activates the retired ``HERMES_HOME/node`` layout.
    """
    command = str(command)
    if any(sep in command for sep in ("/", "\\")):
        if sys.platform == "win32":
            return command if Path(command).is_file() else None
        return shutil.which(command)
    base = command.lower()
    for suffix in (".cmd", ".exe", ".ps1"):
        base = base.removesuffix(suffix)
    package_name = {"node": "node", "npm": "npm", "npx": "npm"}.get(base)
    if package_name is not None:
        from pm import installed_package

        installed = installed_package(package_name)
        if installed is None or installed.binary is None:
            return None
        if base != "npx":
            return str(installed.binary)
        for name in _candidate_node_command_names("npx"):
            candidate = installed.binary.parent / name
            if candidate.is_file():
                return str(candidate)
        return None
    if sys.platform != "win32":
        return shutil.which(command)
    directories = [d for d in os.environ.get("PATH", "").split(os.pathsep) if d]
    for name in _candidate_node_command_names(command):
        for directory in directories:
            candidate = Path(directory) / name
            if candidate.is_file():
                return str(candidate)
    return None


def with_hermes_node_path(env: dict[str, str] | None = None) -> dict[str, str]:
    """Compose installed PM npm and its Node dependency without provisioning."""
    from pm import env_for

    return env_for("npm", base_env=env)


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 not current_home or _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 an IDLE retention:
# an entry lives while anything inside it is still being written and goes 24h after the last
# write anywhere in its subtree. A fixed age was wrong both ways — a directory's own mtime only
# moves when a direct child is added or removed, so a lane writing deep inside a tree looked
# untouched and lost its worktree at the deadline, while finished trees (a 7 GB clone with its
# own venv per campaign lane) sat for three days and filled the disk.
SCRATCH_TMP_ENV_VARS = ("TMPDIR", "TMP", "TEMP")
SCRATCH_DIR_MARKER_ENV = "HERMES_SCRATCH_DIR"
SCRATCH_MAX_IDLE_HOURS = 24
_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


# ---- Managed mode (NixOS declarative config) ----
# Canonical home of "is this install package-manager managed": ``hermes_cli.config`` re-exports
# these, and :func:`apply_secure_dir_policy` below reads them. Lives here because constants
# must stay import-safe from the CLI.
_MANAGED_TRUE_VALUES = ("true", "1", "yes")
# Only the NixOS module ever wrote a bare "true" or an empty marker.
_LEGACY_MANAGED_SYSTEM = "nixos"
# Homebrew is no longer a supported distribution: these markers fall through to git/unknown
# detection instead of blocking config writes.
_IGNORED_MANAGED_VALUES = frozenset({"brew", "homebrew"})
# Explicit opt-out (``HERMES_MANAGED=false``): without this a bool-shaped value became a package
# manager literally named "false" and is_managed() blocked `hermes update` (#12864).
_MANAGED_FALSE_VALUES = frozenset({"false", "0", "no", "off"})


def get_managed_system(home: str | Path | None = None) -> str | None:
    """Return the package manager owning this install, if any.
    Signals: HERMES_MANAGED env var (systemd service) or a ``.managed`` marker file in
    HERMES_HOME (NixOS activation script — interactive shells don't see the service env).
    An unreadable or empty marker still counts as managed (the legacy NixOS shape).

    ``home`` names the home whose marker file is read, for callers that already resolved it
    (:func:`get_scratch_dir` at boot, before ``--profile`` re-homes the process).
    Defaults to the effective home."""
    marker = os.getenv("HERMES_MANAGED", "").strip().lower() or None
    managed_marker = (Path(home) if home is not None else get_hermes_home()) / ".managed"
    if marker is None and managed_marker.exists():
        try:
            marker = managed_marker.read_text(encoding="utf-8", errors="replace").strip().lower()
        except OSError:
            marker = ""
    if marker is None or marker in _IGNORED_MANAGED_VALUES or marker in _MANAGED_FALSE_VALUES:
        return None
    if marker == "" or marker in _MANAGED_TRUE_VALUES:
        return _LEGACY_MANAGED_SYSTEM
    return marker


def _container_or_chmod_skipped() -> bool:
    """Container/chmod-skip detection: the ``HERMES_CONTAINER``/``HERMES_SKIP_CHMOD`` operator
    overrides on top of the canonical :func:`_detect_container` signals (same breadth as
    :func:`is_container` — Docker/Podman/LXC/Kubernetes, cgroup and root-mountinfo). The cached
    :func:`is_container` itself is deliberately avoided: it ignores these env overrides and is
    computed only once per process, so tests could not flip it. Volume-mounted config is not
    forced to owner-only in containers: gateway and dashboard may run as different UIDs, or
    the mount itself needs broader permissions."""
    if os.environ.get("HERMES_CONTAINER") or os.environ.get("HERMES_SKIP_CHMOD"):
        return True
    return _detect_container()


def _resolve_hermes_uid_gid() -> tuple[int | None, int | None]:
    """Read HERMES_UID / HERMES_GID (set by Docker deployments); (None, None) if unset/invalid/Windows.
    The entrypoint chowns HERMES_HOME once, but subdirs created at runtime (``profiles/<name>/``)
    need the same chown or they land root:root and block later uid-mapped workers.

    Docker containers running Hermes commonly set these to map the in-container user to a host user so
    volume-mounted state files end up with the right ownership. See #34107.
    """
    if sys.platform == "win32":
        return None, None

    def _env_int(name: str) -> int | None:
        try:
            return int(os.environ.get(name, "").strip() or None)
        except (TypeError, ValueError):
            return None

    return _env_int("HERMES_UID"), _env_int("HERMES_GID")


def _chown_to_hermes_uid(path) -> None:
    """Chown ``path`` to ``HERMES_UID:HERMES_GID`` when set; EPERM/ENOENT are non-fatal (the
    entrypoint's startup chown -R fixes ownership on the next restart).

    Used by :func:`apply_secure_dir_policy` to keep ownership consistent across all directories
    created by ``ensure_hermes_home`` on Docker deployments. See #34107.
    """
    uid, gid = _resolve_hermes_uid_gid()
    if uid is None and gid is None:
        return
    try:
        os.chown(path, uid if uid is not None else -1, gid if gid is not None else -1)
    except (OSError, AttributeError, NotImplementedError):
        pass


def apply_secure_dir_policy(path, *, home: str | Path | None = None) -> None:
    """Apply the canonical Hermes home-directory permission policy to *path*.

    Owner-only ``0700`` by default, but the operator's explicit and managed sharing choices
    win (#117347): managed installs are left exactly as the package manager / activation
    script set them (#77579); in a container only an explicit ``HERMES_HOME_MODE`` is applied
    (a bind-mounted data dir is often shared with sibling containers, #10757); elsewhere
    ``HERMES_HOME_MODE`` (e.g. ``0701``, ``2770``) is passed to the host's ``chmod``. Special
    bits may be cleared by the host filesystem (macOS commonly clears setgid on directories
    whose group the caller does not belong to). ``HERMES_UID`` / ``HERMES_GID`` ownership is
    applied when those env vars are set (#34107).

    ``home`` (keyword-only: both arguments are path-likes) names the home whose managed-mode
    marker is read, for callers that already resolved it; without it the effective home is
    consulted.

    Import-safe twin of ``hermes_cli.config._secure_dir`` (which delegates here), so callers
    outside the CLI package — like :func:`get_scratch_dir` — share one policy implementation.
    """
    if get_managed_system(home) is not None:
        return
    explicit_mode = os.environ.get("HERMES_HOME_MODE", "").strip()
    if _container_or_chmod_skipped() and not explicit_mode:
        _chown_to_hermes_uid(path)
        return
    try:
        mode = int(explicit_mode or "700", 8)
    except ValueError:
        mode = 0o700
    try:
        os.chmod(path, mode)
    except (OSError, NotImplementedError):
        pass
    _chown_to_hermes_uid(path)


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 idle for ``SCRATCH_MAX_IDLE_HOURS`` are pruned at most once per process
    and once per hour across processes (stamp file), so a fan-out of children stays cheap.
    The pass itself runs in a detached child process (``hermes_constants_scratch_detach``): the
    directory is created and permissioned before this returns, no caller (the boot hook
    included) waits on the idle scan or the ``rmtree`` of a multi-GB entry, and the pass still
    finishes when the caller is a ``hermes`` command that exits a moment later.

    Permissions follow :func:`apply_secure_dir_policy`, so an explicit ``HERMES_HOME_MODE`` or
    a managed/shared home is honored instead of a blanket ``0700`` (#117347).
    """
    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":
            # The caller's home decides the policy: re-reading the effective home here would
            # warn about a profile the CLI has not switched to yet (boot scratch setup).
            apply_secure_dir_policy(scratch, home=base)
    except OSError:
        pass
    if prune:
        _prune_scratch_dir_once(scratch)
    return scratch


def prune_scratch_dir(scratch: Path | None = None, max_idle_hours: float = SCRATCH_MAX_IDLE_HOURS) -> int:
    """Delete top-level scratch entries with no write anywhere in their subtree for
    *max_idle_hours*, reaping processes and git worktree registrations rooted in them
    first (``hermes_constants_scratch``); return the count removed. Each removal, each
    kill and each entry rescued mid-prune is recorded in ``<home>/logs/scratch-prune.log``
    (*scratch* is ``<home>/cache/scratch``)."""
    from hermes_constants_scratch import prune_idle_entries

    root = scratch if scratch is not None else get_scratch_dir(prune=False)
    log_file = root.parent.parent / "logs" / "scratch-prune.log"
    return prune_idle_entries(root, max_idle_hours, frozenset({_SCRATCH_PRUNE_STAMP}), log_file)


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()
        from hermes_constants_scratch_detach import spawn_prune

        spawn_prune(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:
    """Delegate Termux detection without making the bootstrap constants module depend on the full package."""
    from hermes_platform.host.runtime import is_termux as detect

    return detect()


def is_wsl() -> bool:
    """Delegate WSL detection without making the bootstrap constants module depend on the full package."""
    from hermes_platform.host.runtime import is_wsl as detect

    return detect()


def is_container() -> bool:
    """Delegate container detection without making the bootstrap constants module depend on the full package."""
    from hermes_platform.host.runtime import is_container as detect

    return detect()


def _detect_container() -> bool:
    """Keep the historical test seam while delegating canonical detection."""
    from hermes_platform.host.runtime import _detect_container as detect

    return detect()


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


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:
    """Frozen updater surface: pre-PM updaters import this name; pm.environments owns it."""
    from pm.environments import venv_bin_dir as resolve

    return resolve(venv_dir, windows=windows)


def project_venv_dir(project_root) -> Path | None:
    """The project's ``venv`` or ``.venv`` dir when one exists (``uv venv`` defaults to ``.venv``);
    for an install whose interpreter lives outside the checkout, the running interpreter's 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.

    Installers that keep the interpreter out of the checkout (``$HERMES_HOME/venvs/<name>``, the layout the
    shipped Windows launchers assume) have neither, and the ``project_venv_dir(root) or root / "venv"``
    idiom those call sites share then handed ``uv`` a ``VIRTUAL_ENV`` that does not exist: that one invented
    path skipped the import probe, reclassified every ``hermes tools`` dependency as missing and failed the
    reinstall with interpreter errors (#116148). The interpreter running this module is the only truthful
    answer to "which venv is live", so fall back to it — but only for the checkout it was loaded from. A
    foreign root (test temp dir, another clone) still resolves to ``None``: handing it someone else's venv
    would point the callers' writes at the wrong environment.
    """
    root = Path(project_root)
    in_tree = next((root / n for n in ("venv", ".venv") if (root / n).is_dir()), None)
    if in_tree is not None:
        return in_tree
    # Out-of-tree install: the path is real by construction (never invented), and non-venv installs
    # keep today's ``None`` so the ``or root / "venv"`` fallback cannot install into a base interpreter.
    running = Path(sys.prefix)
    if (Path(__file__).resolve().parent == root.resolve()
            and sys.prefix != sys.base_prefix
            and venv_python_path(running).is_file()
            and _venv_installs_checkout(running, root)):
        return running
    return None


def _venv_installs_checkout(venv: Path, root: Path) -> bool:
    """Is *venv*'s own ``hermes-agent`` installed from *root*?

    Where this module was loaded from does not answer that: ``PYTHONPATH=<checkout>
    <other install>/bin/python`` runs one checkout's code on another install's interpreter,
    and adopting that venv made a dev checkout's update rewrite the Desktop install's venv
    into an editable install of the dev tree. Every install of a checkout into a venv
    (installers, ``uv sync``) records the source tree in ``direct_url.json``.
    """
    import json
    from importlib.metadata import distributions
    from urllib.parse import urlparse
    from urllib.request import url2pathname

    from pm.environments import site_packages

    for dist in distributions(name="hermes-agent", path=[str(site_packages(venv))]):
        try:
            raw = dist.read_text("direct_url.json")  # windows-footgun: ok — importlib.metadata API, reads utf-8, no encoding=
            url = json.loads(raw or "{}").get("url", "")
        except ValueError:
            continue
        if url.startswith("file:") and Path(url2pathname(urlparse(url).path)).resolve() == root.resolve():
            return True
    return False


def venv_python_path(venv_dir, *, windows: bool | None = None) -> Path:
    """Frozen updater surface: pre-PM updaters import this name; pm.environments owns it."""
    from pm.environments import venv_python

    return venv_python(venv_dir, windows=windows)


# 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


def normalize_scope(scope: str | Path | None) -> str | None:
    """Normalize a WRITE-side registry scope key, preserving ``None``.

    Two different contracts live on the same registries — do not unify them:

    * **Write / slot paths** (``register_*``, ``snapshot_registration``,
      ``restore_registration``, tool-registry slot lookup): ``None`` means
      the process-global layer and must stay ``None``. Use this function.
    * **Read paths** (``list_providers``, ``get_provider``): ``None`` means
      "the active home's scope" and must go through :func:`hermes_home_key`
      (falsy input resolves to the active default home). Using this
      function there hides every scoped registration — the exact bug
      fixed after e66a627aa5.

    Both normalize non-None values identically (resolved absolute path,
    normcase on Windows) so writes and reads agree on the key.
    """
    return hermes_home_key(scope) if scope is not None else None