aiturk-hermes-ide/tools/environments/local.py

"""Local execution environment — spawn-per-call with session snapshot."""

import logging
import ntpath
import os
import platform
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import threading
import time
from collections.abc import Mapping
from pathlib import Path

from hermes_constants import get_process_hermes_home
from tools.environments.base import BaseEnvironment, _pipe_stdin
from hermes_cli._subprocess_compat import windows_hide_flags

_IS_WINDOWS = platform.system() == "Windows"

logger = logging.getLogger(__name__)

# --- Terminal temp-cache pruning -------------------------------------------
#
# get_temp_dir() now defaults to HERMES_HOME/cache/terminal (real storage)
# instead of tmpfs /tmp, so stale session artifacts no longer disappear on
# reboot for free. Prune them ourselves: the gateway housekeeping loop calls
# cleanup_terminal_temp_cache() hourly (same contract as the other
# cleanup_*_cache helpers), and a once-per-process best-effort sweep covers
# CLI-only installs that never run the gateway.
#
# Background-process artifacts come in triplets (hermes_bg_<id>.log/.pid/
# .exit). A long-running server's .pid file never changes mtime while its
# .log keeps updating — so age is judged per GROUP (newest mtime among files
# sharing a stem) to avoid yanking the pid/exit files out from under a
# still-live background session.
TERMINAL_TEMP_MAX_AGE_HOURS = 72

_terminal_temp_prune_lock = threading.Lock()
_terminal_temp_pruned_once = False

_BG_GROUP_RE = re.compile(r"^(hermes_bg_[A-Za-z0-9_-]+)\.(log|pid|exit)$")


def _default_terminal_temp_dir() -> "Path | None":
    """Return HERMES_HOME/cache/terminal, or None if unresolvable."""
    try:
        from hermes_constants import get_hermes_home
        return get_hermes_home() / "cache" / "terminal"
    except Exception:
        return None


def cleanup_terminal_temp_cache(
    max_age_hours: int = TERMINAL_TEMP_MAX_AGE_HOURS,
) -> int:
    """Delete session temp artifacts older than *max_age_hours*.

    Same contract as the ``cleanup_*_cache`` helpers in
    ``gateway.platforms.base`` — returns the number of entries removed — so
    the gateway housekeeping loop can prune this dir on its hourly cadence.

    Only prunes the managed default dir (``HERMES_HOME/cache/terminal``).
    User-pointed ``terminal.temp_dir`` locations are the user's to manage —
    we never bulk-delete inside a directory we don't own.
    """
    root = _default_terminal_temp_dir()
    if root is None:
        return 0
    cutoff = time.time() - (max_age_hours * 3600)
    removed = 0
    try:
        entries = list(root.iterdir())
    except OSError:
        return 0

    # Newest mtime per hermes_bg_<id> group, so a live server's fresh .log
    # protects its stale-looking .pid/.exit siblings.
    group_newest: dict[str, float] = {}
    for f in entries:
        m = _BG_GROUP_RE.match(f.name)
        if m:
            try:
                mt = f.stat().st_mtime
            except OSError:
                continue
            key = m.group(1)
            group_newest[key] = max(group_newest.get(key, 0.0), mt)

    for f in entries:
        try:
            mt = f.stat().st_mtime
        except OSError:
            continue
        m = _BG_GROUP_RE.match(f.name)
        effective = group_newest.get(m.group(1), mt) if m else mt
        if effective >= cutoff:
            continue
        try:
            if f.is_dir():
                shutil.rmtree(f, ignore_errors=True)
            else:
                f.unlink()
            removed += 1
        except OSError:
            continue
    return removed


def _prune_terminal_temp_once() -> None:
    """Best-effort prune, at most once per process (CLI-only installs)."""
    global _terminal_temp_pruned_once
    with _terminal_temp_prune_lock:
        if _terminal_temp_pruned_once:
            return
        _terminal_temp_pruned_once = True
    try:
        cleanup_terminal_temp_cache()
    except Exception as exc:
        logger.debug("Terminal temp prune failed: %s", exc)


def _msys_to_windows_path(cwd: str) -> str:
    """Translate a Git Bash / MSYS-style POSIX path (``/c/Users/x``) to the
    native Windows form (``C:\\Users\\x``) so ``os.path.isdir`` and
    ``subprocess.Popen(..., cwd=...)`` can find it.

    Also accepts the Cygwin (``/cygdrive/c/...``) and WSL-mount
    (``/mnt/c/...``) spellings of a drive root. Multi-segment POSIX paths
    like ``/home/x`` or ``/tmp/foo`` are left untouched.

    No-ops on non-Windows hosts or for paths that aren't in MSYS form.
    Returns the input unchanged when no translation applies. This is
    idempotent — calling it on an already-Windows path returns it as-is.
    """
    if not _IS_WINDOWS or not cwd:
        return cwd
    # Match leading "/<single letter>/" or exactly "/<letter>" (bare drive root),
    # plus /cygdrive/<letter>/... and /mnt/<letter>/... variants.
    m = re.match(r'^/(?:(?:cygdrive|mnt)/)?([a-zA-Z])(/.*)?$', cwd)
    if not m:
        return cwd
    # Reject /cygdrive or /mnt with no drive letter — the optional group above
    # already requires the letter. Multi-char first segments (/home, /tmp)
    # fail the single-letter capture and fall through as no-ops.
    drive = m.group(1).upper()
    tail = (m.group(2) or "").replace('/', '\\')
    return f"{drive}:{tail or chr(92)}"  # chr(92) = backslash, avoid raw-string escape


def _resolve_local_initial_cwd(cwd: str) -> str:
    """Resolve the local backend's initial cwd to an absolute host path.

    ``TERMINAL_CWD`` can be populated from config.yaml before the terminal
    backend is created.  If that value is relative and happens to match the
    directory Hermes was already launched from (for example ``hermes-agent``
    while the process cwd is ``~/.hermes/hermes-agent``), passing it through
    unchanged makes the wrapper run ``cd hermes-agent`` *inside* the project
    and fail with a confusing nested-path error.  Anchor relative local cwd
    values once, up front, so both ``subprocess.Popen(cwd=...)`` and the
    in-shell ``cd`` use the same absolute directory.
    """
    expanded = os.path.expanduser(cwd) if cwd else os.getcwd()
    if _IS_WINDOWS:
        expanded = _msys_to_windows_path(expanded)
        # Use the Windows-aware check explicitly: when _IS_WINDOWS is
        # patched in tests on a POSIX host, os.path.isabs would reject
        # ``C:\Users\x`` and mangle it through the relative branch.
        import ntpath
        if ntpath.isabs(expanded):
            return expanded
    if os.path.isabs(expanded):
        return expanded

    candidate = os.path.abspath(expanded)
    current = os.getcwd()

    # Common recovery for config values like ``hermes-agent`` when Hermes was
    # launched from that directory already.  ``os.path.abspath`` would point at
    # a nonexistent nested ``./hermes-agent``; use the current directory instead.
    if not os.path.isdir(candidate):
        wanted_parts = Path(expanded).parts
        current_parts = Path(current).parts
        if wanted_parts and len(wanted_parts) <= len(current_parts):
            if current_parts[-len(wanted_parts):] == wanted_parts:
                return current

    return candidate


def _windows_to_msys_path(cwd: str) -> str:
    """Translate a native Windows path (``C:\\Users\\x``) to Git Bash /
    MSYS form (``/c/Users/x``) so ``builtin cd`` resolves it reliably.

    No-ops on non-Windows hosts or for paths that aren't drive-qualified
    native Windows paths. Returns the input unchanged when no translation
    applies.
    """
    if not _IS_WINDOWS or not cwd:
        return cwd
    m = re.match(r'^([a-zA-Z]):[\\/]*(.*)$', cwd)
    if not m:
        return cwd
    drive = m.group(1).lower()
    tail = (m.group(2) or "").replace('\\', '/').lstrip('/')
    return f"/{drive}/{tail}" if tail else f"/{drive}/"


def _bash_safe_path(path: str) -> str:
    """Return *path* in a form safe to embed in a Git Bash script.

    Native ``C:\\Users\\x`` / ``C:/Users/x`` → ``/c/Users/x`` via
    :func:`_windows_to_msys_path`. Mixed MSYS leftovers
    (``/c/Users\\Alexander\\Documents``) get backslashes normalized so
    bash does not eat ``\\U`` and trip the ``Directory \\drivers\\etc``
    failure class. No-op off Windows and for empty input.

    ``get_temp_dir`` already emits forward-slash ``C:/...`` forms for
    Python compatibility; those still need the ``/c/...`` rewrite —
    MSYS argument conversion treats ``C:/...`` as a Windows path and
    can corrupt the login-shell ``drivers\\etc`` lookup.
    """
    if not _IS_WINDOWS or not path:
        return path
    path = _windows_to_msys_path(path)
    if "\\" in path:
        path = path.replace("\\", "/")
    return path


def _quote_bash_path(path: str) -> str:
    """Quote *path* for safe interpolation into a Git Bash script on Windows."""
    import shlex

    return shlex.quote(_bash_safe_path(path))


def _cwd_usable(path: str) -> bool:
    """True when *path* is a directory this process can actually chdir into.

    ``os.path.isdir`` alone is not enough: stat() on ``/root`` succeeds for a
    non-root user (only ``/`` needs search permission), but
    ``subprocess.Popen(cwd='/root')`` then dies with ``PermissionError:
    [Errno 13] Permission denied: '/root'``. Seen in the wild when a
    root-launched CLI session leaks ``/root`` into shared state that a
    non-root gateway/cron process later reads (#65583) — every cron job's
    terminal/file tool then fails on every command, forever. Checking
    X_OK up front lets the caller fall back instead.
    """
    return os.path.isdir(path) and os.access(path, os.X_OK)


def _resolve_safe_cwd(cwd: str) -> str:
    """Return ``cwd`` if it exists as a directory this process can enter,
    else the nearest existing accessible ancestor.  Falls back to
    ``tempfile.gettempdir()`` only if walking up the path can't find any
    usable directory (effectively never on a healthy filesystem, but cheap
    belt-and-braces).

    On Windows, also normalizes Git Bash / MSYS-style POSIX paths
    (``/c/Users/x``) to native Windows form before the isdir check so a
    perfectly valid ``pwd -P`` result from bash doesn't get rejected as
    "missing" (see ``_msys_to_windows_path``).

    Used by ``_run_bash`` to recover when the configured cwd is gone — most
    commonly because a previous tool call deleted its own working directory
    (issue #17558) — or inaccessible to this user, e.g. ``/root`` leaking
    from a root-launched CLI session into a non-root gateway's cron jobs
    (issue #65583).  Without this guard, ``subprocess.Popen(..., cwd=...)``
    raises ``FileNotFoundError``/``PermissionError`` before bash starts,
    wedging every subsequent terminal call until the gateway restarts.
    """
    cwd = _msys_to_windows_path(cwd) if _IS_WINDOWS else cwd
    if cwd and _cwd_usable(cwd):
        return cwd
    if cwd and os.path.isdir(cwd):
        logger.warning(
            "Configured terminal cwd %r exists but is not accessible to "
            "this user (uid=%s) — falling back to the nearest usable "
            "directory. If this is a gateway/cron process, check for "
            "root-owned paths leaking into terminal.cwd / TERMINAL_CWD "
            "(#65583).",
            cwd, getattr(os, "getuid", lambda: "?")(),
        )
    parent = os.path.dirname(cwd) if cwd else ""
    while parent:
        if _cwd_usable(parent):
            return parent
        next_parent = os.path.dirname(parent)
        if next_parent == parent:
            # Reached the filesystem root and it doesn't exist either —
            # genuinely nothing to fall back to except the temp dir.
            break
        parent = next_parent
    return tempfile.gettempdir()


# Hermes-internal env vars that should NOT leak into terminal subprocesses.
_HERMES_PROVIDER_ENV_FORCE_PREFIX = "_HERMES_FORCE_"

# Hermes-managed AWS *inference* credentials for ``auth_type="aws_sdk"``
# providers (Bedrock).  Scoped DELIBERATELY NARROW: this lists only the
# Bedrock-specific bearer token, which is a Hermes inference secret exactly
# analogous to ``OPENAI_API_KEY`` — nobody drives the ``aws``/``terraform``/
# ``boto3`` toolchain off it, so stripping it from terminal/execute_code
# subprocesses costs no user capability.
#
# The GENERAL AWS credential chain (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY,
# AWS_SESSION_TOKEN, AWS_PROFILE, and the config/role pointers) is INTENTIONALLY
# left inheritable.  Per SECURITY.md §3.2 the local terminal is the user's
# trusted operator shell; the agent having the same general AWS access the
# user's own shell has is the intended posture, not a leak.  Hard-blocklisting
# those vars would (a) regress every user who runs aws/terraform/cdk/boto3 in
# the agent terminal — not just Bedrock users, since the registry is iterated
# unconditionally — and (b) be unrecoverable, because env_passthrough.py
# refuses to re-allow anything in this blocklist (GHSA-rhgp-j443-p4rf).  See
# issue #32314 discussion.
_AWS_SDK_CREDENTIAL_ENV_VARS = frozenset({
    "AWS_BEARER_TOKEN_BEDROCK",
})


def _build_provider_env_blocklist() -> frozenset:
    """Derive the blocklist from provider, tool, and gateway config."""
    blocked: set[str] = set()

    try:
        from hermes_cli.auth import PROVIDER_REGISTRY
        for pconfig in PROVIDER_REGISTRY.values():
            blocked.update(pconfig.api_key_env_vars)
            if pconfig.auth_type == "aws_sdk":
                blocked.update(_AWS_SDK_CREDENTIAL_ENV_VARS)
            if pconfig.base_url_env_var:
                blocked.add(pconfig.base_url_env_var)
    except ImportError:
        pass

    try:
        from hermes_cli.config import OPTIONAL_ENV_VARS
        for name, metadata in OPTIONAL_ENV_VARS.items():
            category = metadata.get("category")
            if category in {"tool", "messaging"}:
                blocked.add(name)
            elif category == "setting" and metadata.get("password"):
                blocked.add(name)
    except ImportError:
        pass

    blocked.update({
        "OPENAI_BASE_URL",
        "OPENAI_API_KEY",
        "OPENAI_API_BASE",
        "OPENAI_ORG_ID",
        "OPENAI_ORGANIZATION",
        "OPENROUTER_API_KEY",
        "ANTHROPIC_BASE_URL",
        "ANTHROPIC_API_KEY",
        "ANTHROPIC_TOKEN",
        "LLM_MODEL",
        "GOOGLE_API_KEY",
        # Path to a GCP service-account JSON, not a bare key, so
        # OPTIONAL_ENV_VARS marks it password=False and the loop above skips it.
        "VERTEX_CREDENTIALS_PATH",
        "GOOGLE_APPLICATION_CREDENTIALS",
        "DEEPSEEK_API_KEY",
        "MISTRAL_API_KEY",
        "GROQ_API_KEY",
        "TOGETHER_API_KEY",
        "PERPLEXITY_API_KEY",
        "COHERE_API_KEY",
        "FIREWORKS_API_KEY",
        "XAI_API_KEY",
        "HELICONE_API_KEY",
        "PARALLEL_API_KEY",
        "FIRECRAWL_API_KEY",
        "FIRECRAWL_API_URL",
        "TELEGRAM_HOME_CHANNEL",
        "TELEGRAM_HOME_CHANNEL_NAME",
        "DISCORD_HOME_CHANNEL",
        "DISCORD_HOME_CHANNEL_NAME",
        "DISCORD_REQUIRE_MENTION",
        "DISCORD_FREE_RESPONSE_CHANNELS",
        "DISCORD_AUTO_THREAD",
        "SLACK_HOME_CHANNEL",
        "SLACK_HOME_CHANNEL_NAME",
        "SLACK_ALLOWED_USERS",
        "WHATSAPP_ENABLED",
        "WHATSAPP_MODE",
        "WHATSAPP_ALLOWED_USERS",
        "SIGNAL_HTTP_URL",
        "SIGNAL_ACCOUNT",
        "SIGNAL_ALLOWED_USERS",
        "SIGNAL_GROUP_ALLOWED_USERS",
        "SIGNAL_HOME_CHANNEL",
        "SIGNAL_HOME_CHANNEL_NAME",
        "SIGNAL_IGNORE_STORIES",
        "HASS_TOKEN",
        "HASS_URL",
        "EMAIL_ADDRESS",
        "EMAIL_PASSWORD",
        "EMAIL_IMAP_HOST",
        "EMAIL_SMTP_HOST",
        "EMAIL_HOME_ADDRESS",
        "EMAIL_HOME_ADDRESS_NAME",
        "HERMES_DASHBOARD_SESSION_TOKEN",
        "GATEWAY_ALLOWED_USERS",
        "GH_TOKEN",
        "GITHUB_APP_ID",
        "GITHUB_APP_PRIVATE_KEY_PATH",
        "GITHUB_APP_INSTALLATION_ID",
        "MODAL_TOKEN_ID",
        "MODAL_TOKEN_SECRET",
        "DAYTONA_API_KEY",
        "GATEWAY_RELAY_ID",
        "GATEWAY_RELAY_SECRET",
        "GATEWAY_RELAY_DELIVERY_KEY",
        "VERCEL_OIDC_TOKEN",
        "VERCEL_TOKEN",
        "VERCEL_PROJECT_ID",
        "VERCEL_TEAM_ID",
    })
    # CLAUDE_CODE_OAUTH_TOKEN is deliberately NOT stripped.  It is set and
    # owned by the user's Claude Code install (subscription OAuth), not a
    # Hermes-managed inference credential — Claude subscription auth is not a
    # working Hermes provider path.  Stripping it broke agent-spawned
    # ``claude`` CLIs: the child fell through to the shared macOS Keychain /
    # ``~/.claude/.credentials.json`` store and, on auth failure, cleared it,
    # logging the user out of their interactive Claude sessions (#55878).
    # It arrives via the registry loop above (anthropic api_key_env_vars),
    # so remove it explicitly.
    blocked.discard("CLAUDE_CODE_OAUTH_TOKEN")
    # BUZZ_* is deliberately NOT discarded here, even for Buzz-managed agents
    # (BUZZ_MANAGED_AGENT set by the buzz-acp harness).  This blocklist is
    # shared by every scrub surface — the terminal paths, execute_code, and
    # the :func:`hermes_subprocess_env` Tier-2 strip (browser / TUI host /
    # copilot-executor spawns) — so an import-time discard would leak
    # BUZZ_PRIVATE_KEY into non-terminal children too.  The Buzz carve-out is
    # instead a TERMINAL-ONLY, context-gated scrub-path exemption: see
    # ``_TERMINAL_FIRST_PARTY_ENV_PREFIXES`` / ``_is_terminal_first_party_env``
    # below (issue #78026 / #76243, PRs #78065 + #78511).
    return frozenset(blocked)


_HERMES_PROVIDER_ENV_BLOCKLIST = _build_provider_env_blocklist()

# First-party platform credentials the agent's own platform adapters need in
# terminal children (e.g. the ``BUZZ_*`` vars for the Buzz messaging
# platform, which drive the platform-mandated ``buzz`` CLI: BUZZ_PRIVATE_KEY,
# BUZZ_AUTH_TAG, BUZZ_RELAY_URL, and the other BUZZ_* names). These are the
# agent's OWN credentials — a Buzz community agent is expected to operate the
# ``buzz`` CLI — so they are carved out of the terminal scrub.
#
# CONTEXT-GATED: the carve-out applies ONLY when this process/session is
# actually operating as a Buzz agent — either the process is a Buzz-ACP
# managed agent (``BUZZ_MANAGED_AGENT`` is set, only by Buzz Desktop's
# buzz-acp harness; see #76243 / #78511) or the current session's platform is
# ``buzz`` (the gateway's ``HERMES_SESSION_PLATFORM`` ContextVar; concurrency
# safe under a multi-session host). A Telegram/CLI/cron session on a host
# that also runs a Buzz gateway does NOT get BUZZ_PRIVATE_KEY in its terminal
# children — blanket passthrough of a signing key to every terminal child on
# the host would be wrong (maintainer triage note on #76243: don't expose the
# key to unrelated shell commands).
#
# Scope is TERMINAL ONLY: the foreground ``_make_run_env`` and background/PTY
# ``_sanitize_subprocess_env`` paths pass them through. ``_sanitize_subprocess_env``
# is also consumed by search workers (e.g. the ddgs web-search subprocess),
# the computer-use driver binary, and user-script runners (bang ``!``
# commands, quick commands, cron scripts, webhook-filter scripts), so those
# children receive the vars too — matching the approved background/PTY scope.
# Every other surface stays sealed — execute_code scrubbing,
# :func:`hermes_subprocess_env` (browser / TUI host / copilot-executor
# spawns), docker children, and ``env_passthrough`` registration (skills/config
# still cannot register these names). The GHSA-rhgp-j443-p4rf seal is
# preserved because no registration path is opened; this is a scrub-path
# exemption, not an allowlist addition.
#
# First-party matches use the merged env value directly — they are the
# process's own env values and are never scope-resolved (a profile secret
# scope under multiplex would otherwise raise UnscopedSecretError at
# passthrough-resolution call sites); only skill/config passthrough names
# resolve through the profile secret scope. The snapshot mechanism treats
# these names like profile-scoped passthrough names (see
# ``LocalEnvironment._additional_profile_scoped_passthrough_names``) so they
# never persist in the shared terminal snapshot across profiles.
#
# Prefix-based on purpose: future ``BUZZ_*`` names added by the platform's
# plugin.yaml (or a user's own credentials file) are covered without another
# code change. Contrast with CLAUDE_CODE_OAUTH_TOKEN above, which is discarded
# from the blocklist entirely because it is NOT a Hermes credential; these ARE
# Hermes-managed first-party platform credentials, so they stay IN the
# blocklist for every non-terminal surface.
#
# See issue #78026 (Buzz agents could not use ``buzz`` from the terminal tool)
# and #76243 (Buzz Desktop managed agent wakes but cannot reply).
_TERMINAL_FIRST_PARTY_ENV_PREFIXES = ("BUZZ_",)


def _matches_terminal_first_party_prefix(name: str) -> bool:
    """Pure name check: ``name`` is one of the first-party platform
    credential names (``BUZZ_*``), regardless of session context.  Used for
    the snapshot exclusion, which must stay conservative even when the
    carve-out itself is inactive."""
    return name.startswith(_TERMINAL_FIRST_PARTY_ENV_PREFIXES)


def _buzz_terminal_context_active() -> bool:
    """True when this process/session is operating as a Buzz agent.

    Two independent signals, either suffices:

    * ``BUZZ_MANAGED_AGENT`` in the process env — set exclusively by Buzz
      Desktop's buzz-acp harness when it spawns ``hermes acp`` (#76243).
      Gateway / CLI / cron / kanban processes never carry it.
    * The live session's platform is ``buzz`` — the gateway's
      ``HERMES_SESSION_PLATFORM`` ContextVar via
      :func:`gateway.session_context.get_session_env`, which is
      ContextVar-authoritative under a concurrent multi-session host, so a
      sibling Telegram/Discord session on the same gateway process resolves
      its OWN platform, not buzz.
    """
    if os.environ.get("BUZZ_MANAGED_AGENT"):
        return True
    try:
        from gateway.session_context import get_session_env

        return get_session_env("HERMES_SESSION_PLATFORM", "").strip().lower() == "buzz"
    except Exception:
        return False


def _is_terminal_first_party_env(name: str) -> bool:
    """Return True if ``name`` is a first-party platform credential that must
    reach terminal children (the ``BUZZ_*`` set) AND the current
    process/session context entitles it (Buzz-managed agent or a buzz-platform
    session — see :func:`_buzz_terminal_context_active`)."""
    if not _matches_terminal_first_party_prefix(name):
        return False
    return _buzz_terminal_context_active()


# Active-virtualenv markers that must NOT leak into terminal subprocesses.
# The gateway runs inside its own venv, so its process environment carries
# VIRTUAL_ENV (and possibly CONDA_PREFIX). If those leak into commands the
# agent runs against OTHER Python projects, tools like ``uv``/``poetry`` treat
# the inherited value as the active environment and build/sync that other
# project's dependencies into the Hermes venv path instead of the project's own
# ``.venv`` — silently clobbering the Hermes environment (e.g. a project pinned
# to a different Python version overwrites it and breaks the gateway). The
# Hermes venv stays reachable via PATH (its bin dir is first), so stripping
# these markers is safe and only prevents the cross-project clobber (#23473).
#
# PYTHONHOME is included because a gateway-inherited value redirects the
# standard-library search of ANY child interpreter — including unrelated
# system/venv Pythons — to the Hermes venv's stdlib, which crashes with
# version-mismatch errors before a child script even imports a package
# (#75018). Hermes itself treats PYTHONHOME as contamination in its own
# child processes (managed_uv.py, sqlite_runtime.py), so stripping it from
# subprocess envs is consistent. Users who need PYTHONHOME for a specific
# child can set it explicitly in the command.
#
# PYTHONPATH is NOT included here — it's handled by
# _strip_hermes_owned_pythonpath() which removes only Hermes-owned entries,
# preserving user-set paths.
_ACTIVE_VENV_MARKER_VARS = ("VIRTUAL_ENV", "CONDA_PREFIX", "PYTHONHOME")


def _is_hermes_internal_secret(key: str) -> bool:
    """Return True for Hermes-internal secrets injected under *dynamic* names.

    ``_HERMES_PROVIDER_ENV_BLOCKLIST`` is name-based and derived from the
    provider/tool registries, but the gateway and CLI also inject secrets into
    ``os.environ`` at runtime under names no static registry knows about:

    - ``AUXILIARY_<TASK>_API_KEY`` / ``AUXILIARY_<TASK>_BASE_URL`` — per-task
      side-LLM credentials bridged from ``config.yaml[auxiliary]`` by
      ``gateway/run.py`` and ``cli.py`` (vision, web_extract, approval,
      compression, and any plugin-registered auxiliary task). These are
      separate, often higher-spend API keys plus base URLs that may point at
      private endpoints; a model-authored shell command must never see them.
    - ``GATEWAY_RELAY_*_SECRET`` / ``GATEWAY_RELAY_*_KEY`` /
      ``GATEWAY_RELAY_*_TOKEN`` — relay-auth material provisioned by the
      gateway (``GATEWAY_RELAY_SECRET``, ``GATEWAY_RELAY_DELIVERY_KEY``).
      These are Tier-1 gateway secrets, like the messaging bot tokens in
      ``_ALWAYS_STRIP_KEYS``. Non-secret ``GATEWAY_RELAY_*`` routing hints
      (``GATEWAY_RELAY_URL``, ``GATEWAY_RELAY_PLATFORMS``, …) are NOT matched
      and remain visible.

    ``code_execution_tool.py`` already catches these via substring matching on
    ``KEY`` / ``SECRET`` / ``TOKEN``; the terminal backend's narrower name-based
    blocklist did not, which is the leak this predicate closes.

    This is the single source of truth for "Hermes-internal dynamic secret"
    across every spawn path — the terminal ``_make_run_env`` /
    ``_sanitize_subprocess_env`` filters, the Docker passthrough filter, and the
    non-terminal :func:`hermes_subprocess_env` helper all call it, so the
    dynamic patterns are stripped **unconditionally** regardless of
    ``env_passthrough`` skill registration or ``inherit_credentials``. Nothing
    a model-driving CLI legitimately needs matches these patterns.
    """
    upper = key.upper()
    if upper.startswith("AUXILIARY_") and (
        upper.endswith("_API_KEY") or upper.endswith("_BASE_URL")
    ):
        return True
    if upper.startswith("GATEWAY_RELAY_") and (
        upper.endswith("_SECRET") or upper.endswith("_KEY") or upper.endswith("_TOKEN")
    ):
        return True
    return False


def _plugin_terminal_env_strip_keys() -> frozenset:
    """Credential env keys owned by plugin-registered terminal backends.

    Computed at call time (not import time) because plugins register after
    this module is imported. Treated as Tier-1: stripped from every spawned
    subprocess unconditionally, exactly like MODAL_*/DAYTONA_API_KEY in
    ``_ALWAYS_STRIP_KEYS``. Fail-soft to an empty set.
    """
    try:
        from agent.terminal_env_registry import plugin_strip_env_keys

        return plugin_strip_env_keys()
    except Exception:
        return frozenset()


def _inject_context_hermes_home(env: dict) -> None:
    """Bridge the context-local Hermes home override into subprocess env."""
    try:
        from hermes_constants import get_hermes_home_override

        value = get_hermes_home_override()
        if value:
            env["HERMES_HOME"] = value
    except Exception:
        pass


def _inject_session_context_env(env: dict) -> None:
    """Bridge gateway session ContextVars into a subprocess environment dict.

    ContextVars don't propagate to child processes, so the live session vars
    (HERMES_SESSION_*) are bridged onto the child env here.

    🔴 Cross-session leak guard. The session vars also have a process-global
    os.environ mirror (written last-writer-wins as a CLI/cron fallback, never
    cleared). Under a concurrent multi-session host (the messaging gateway, ACP
    adapter, API server, TUI) that global belongs to *whichever turn wrote it
    last* — NOT necessarily this task. A subprocess spawned from a task whose
    ContextVar is _UNSET (e.g. a sibling message task that never bound, or one
    that inherited another session's context) would otherwise inherit the
    FOREIGN global and act on another session's identity.

    So once the session-context machinery is engaged in this process (any host
    has called set_session_vars), the session vars are ContextVar-authoritative:
    - ContextVar set (incl. explicitly-empty "") → that value wins, overriding
      any stale snapshot/global value.
    - ContextVar _UNSET → STRIP the var from the child env rather than inherit
      the possibly-foreign process-global.
    In a pure single-process CLI/one-shot that never engaged the session-context
    system there is no concurrency to leak across, so the inherited fallback is
    kept. See gateway/session_context.session_context_engaged and
    tests/tools/test_local_env_session_leak.py.
    """
    try:
        from gateway.session_context import (
            _UNSET,
            _VAR_MAP,
            session_context_engaged,
        )
    except Exception:
        return

    _engaged = session_context_engaged()
    for var_name, var in _VAR_MAP.items():
        value = var.get()
        if value is not _UNSET:
            # Explicitly bound (including "") — authoritative for this task.
            env[var_name] = "" if value is None else str(value)
        elif _engaged:
            # Unset for THIS task while a concurrent host is engaged: drop any
            # inherited global so a sibling session's value can't leak in.
            env.pop(var_name, None)


def _sanitize_subprocess_env(base_env: dict | None, extra_env: dict | None = None) -> dict:
    """Filter Hermes-managed secrets from a subprocess environment."""
    try:
        from tools.env_passthrough import (
            is_env_passthrough as _is_passthrough,
            resolve_passthrough_value as _resolve_passthrough_value,
        )
    except Exception:
        _is_passthrough = lambda _: False  # noqa: E731
        _resolve_passthrough_value = lambda _name, fallback: fallback  # noqa: E731

    sanitized: dict[str, str] = {}
    _plugin_strip = _plugin_terminal_env_strip_keys()

    for key, value in (base_env or {}).items():
        if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
            continue
        if _is_hermes_internal_secret(key):
            continue
        if key in _plugin_strip:
            continue
        first_party = _is_terminal_first_party_env(key)
        passthrough = _is_passthrough(key)
        if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not (passthrough or first_party):
            continue
        # First-party platform vars are the process's own env values: use them
        # directly, never scope-resolve (multiplex with no scope would raise
        # UnscopedSecretError — a regression where the script previously ran
        # without the var). Only skill/config passthrough names resolve.
        resolved = value
        if passthrough and not first_party:
            resolved = _resolve_passthrough_value(key, value)
        if resolved is not None:
            sanitized[key] = resolved

    for key, value in (extra_env or {}).items():
        if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
            real_key = key[len(_HERMES_PROVIDER_ENV_FORCE_PREFIX):]
            if _is_hermes_internal_secret(real_key):
                continue
            sanitized[real_key] = value
        elif _is_hermes_internal_secret(key):
            continue
        elif key in _plugin_strip:
            continue
        else:
            first_party = _is_terminal_first_party_env(key)
            passthrough = _is_passthrough(key)
            if key in _HERMES_PROVIDER_ENV_BLOCKLIST and not (passthrough or first_party):
                continue
            resolved = value
            if passthrough and not first_party:
                resolved = _resolve_passthrough_value(key, value)
            if resolved is not None:
                sanitized[key] = resolved

    _inject_context_hermes_home(sanitized)

    from hermes_constants import apply_subprocess_home_env
    apply_subprocess_home_env(sanitized)

    # Same cross-session leak guard as _make_run_env, for the background/PTY
    # spawn path (process_registry.spawn_local builds env via this function).
    _inject_session_context_env(sanitized)

    # Filter PYTHONPATH before removing VIRTUAL_ENV: legacy Windows launchers
    # can run the gateway under a base interpreter while VIRTUAL_ENV identifies
    # the separate Hermes runtime venv.  The filter validates that relationship
    # against the repo layout before trusting it.
    _strip_hermes_owned_pythonpath_and_runtime_markers(sanitized)

    # Keep bare ``hermes`` invocations available to child jobs even when the
    # gateway was launched by a service manager or cron without the console
    # script's directory on PATH.  The terminal environment already applies
    # this invariant; Cron scripts use this sanitizer directly (#92998).
    path_key = _path_env_key(sanitized)
    if path_key is not None:
        sanitized[path_key] = _prepend_hermes_bin_dir(sanitized.get(path_key, ""))

    _apply_windows_msys_bash_env_defaults(sanitized)

    sanitized = _scrub_delegated_child_kanban_env(sanitized)

    return sanitized


def _scrub_delegated_child_kanban_env(env: dict[str, str]) -> dict[str, str]:
    """Strip dispatcher-owned Kanban env from delegate_task child subprocesses."""
    try:
        from agent.delegation_context import (
            is_delegated_child_process_context,
            scrub_kanban_env,
        )

        if is_delegated_child_process_context():
            return scrub_kanban_env(env)
    except Exception:
        pass
    return env


# Tier-1 secrets: stripped from EVERY spawned subprocess unconditionally —
# even when the caller opts into credential inheritance for a model-driving
# CLI (claude / codex / gemini).  These are not LLM provider credentials; no
# legitimate child Hermes spawns needs them, and they are the highest-value
# secrets to keep out of a compromised dependency's reach (gateway bot tokens,
# GitHub auth, remote-compute tokens, dashboard session secret).  The set is a
# narrow subset of _HERMES_PROVIDER_ENV_BLOCKLIST; provider keys are handled by
# the conditional Tier-2 strip in hermes_subprocess_env().
_ALWAYS_STRIP_KEYS: frozenset[str] = frozenset({
    # GitHub auth
    "GH_TOKEN",
    "GITHUB_TOKEN",
    "GITHUB_APP_ID",
    "GITHUB_APP_PRIVATE_KEY_PATH",
    "GITHUB_APP_INSTALLATION_ID",
    # Gateway / messaging bot tokens and access control
    "TELEGRAM_BOT_TOKEN",
    "DISCORD_BOT_TOKEN",
    "SLACK_BOT_TOKEN",
    "SLACK_APP_TOKEN",
    "SLACK_SIGNING_SECRET",
    "GATEWAY_ALLOWED_USERS",
    "GATEWAY_ALLOW_ALL_USERS",
    # Gateway relay auth — the ID/secret/delivery-key triplet the gateway
    # provisions and persists to the 0600 .env. Stripped unconditionally on
    # EVERY spawn surface (terminal + model-driving CLIs) so it can't drift
    # between paths: _SECRET / _DELIVERY_KEY are also matched by
    # _is_hermes_internal_secret, but _ID has no secret suffix, so it must be
    # enumerated here to stay stripped on the inherit_credentials=True path
    # (codex / copilot), which skips the Tier-2 blocklist.
    "GATEWAY_RELAY_ID",
    "GATEWAY_RELAY_SECRET",
    "GATEWAY_RELAY_DELIVERY_KEY",
    "HASS_TOKEN",
    "EMAIL_PASSWORD",
    "HERMES_DASHBOARD_SESSION_TOKEN",
    # Remote-compute / infrastructure secrets
    "MODAL_TOKEN_ID",
    "MODAL_TOKEN_SECRET",
    "DAYTONA_API_KEY",
})


def hermes_subprocess_env(*, inherit_credentials: bool = False) -> dict[str, str]:
    """Build a sanitized environment dict for a spawned subprocess.

    Centralized helper for the **non-terminal** spawn surface (browser,
    ACP/CLI executors, computer-use driver, dep-ensure, TUI Node host,
    detached gateway).  Use this instead of copying ``os.environ`` directly
    so strip-by-default is the uniform policy across every spawn site, with a
    single source of truth (``_HERMES_PROVIDER_ENV_BLOCKLIST``).  The terminal
    / execute_code path keeps using :func:`_sanitize_subprocess_env`, which is
    skill-aware (``env_passthrough``); this helper is for spawns that have no
    skill-passthrough concept.

    Two-tier stripping:

    * **Tier 1 (always):** ``_ALWAYS_STRIP_KEYS`` — gateway bot tokens, GitHub
      auth, and remote-compute secrets are removed regardless of
      ``inherit_credentials``.  No child Hermes spawns legitimately needs them.
    * **Tier 2 (conditional):** the rest of ``_HERMES_PROVIDER_ENV_BLOCKLIST``
      (LLM provider API keys, tool secrets) is removed unless the caller passes
      ``inherit_credentials=True``.

    Pass ``inherit_credentials=True`` **only** when the child legitimately
    needs LLM provider credentials — a user-blessed ``claude`` / ``codex`` /
    ``gemini`` CLI executor, or the TUI Node host that makes model calls.  The
    flag is grep-able for audit: ``grep -rn 'inherit_credentials=True'`` lists
    every spawn site that still receives provider credentials.

    Callers that need a *specific* non-provider secret (e.g. the browser worker
    needs ``BROWSERBASE_API_KEY`` / ``FIRECRAWL_API_KEY``) should call with
    ``inherit_credentials=False`` and copy just those keys back from
    ``os.environ`` into the returned dict.
    """
    env = os.environ.copy()

    # Tier 1 — always strip.
    for key in _ALWAYS_STRIP_KEYS:
        env.pop(key, None)
    for key in _plugin_terminal_env_strip_keys():
        env.pop(key, None)
    # Internal routing hints and Hermes-internal dynamic secrets
    # (``AUXILIARY_<TASK>_API_KEY`` / ``_BASE_URL`` side-LLM credentials,
    # ``GATEWAY_RELAY_*`` relay-auth material) must never reach a child,
    # regardless of ``inherit_credentials`` — a model-driving CLI has no
    # legitimate use for them. See :func:`_is_hermes_internal_secret`.
    for key in list(env):
        if key.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
            env.pop(key, None)
        elif _is_hermes_internal_secret(key):
            env.pop(key, None)

    if not inherit_credentials:
        # Tier 2 — strip provider/tool credentials unless explicitly inherited.
        for key in _HERMES_PROVIDER_ENV_BLOCKLIST:
            env.pop(key, None)

    # Windows UTF-8 safety for spawned processes (#31420).
    env.setdefault("PYTHONUTF8", "1")

    _inject_context_hermes_home(env)
    from hermes_constants import apply_subprocess_home_env
    apply_subprocess_home_env(env)

    _strip_hermes_owned_pythonpath_and_runtime_markers(env)

    _apply_windows_msys_bash_env_defaults(env)

    # Cross-session leak guard, same as the terminal spawn paths: this helper
    # copies os.environ, whose HERMES_SESSION_* mirror is a last-writer-wins
    # global under a concurrent multi-session host. A caller that re-binds the
    # session identity explicitly (slash_worker/ACP via --session-key argv) is
    # unaffected — bound ContextVars win here — but a caller that spawns without
    # re-binding (e.g. tui_gateway cli.exec) would otherwise inherit a FOREIGN
    # session's identity. Strip _UNSET session vars when engaged so that can't
    # happen; single uniform policy across every spawn surface.
    _inject_session_context_env(env)

    # Non-terminal subprocess helpers (browser, lazy-deps, TUI/ACP hosts, etc.)
    # also need the delegate_task child lineage marker.  Otherwise a child
    # context that later imports Kanban DB code in the spawned process would
    # still see the parent's HERMES_HOME but lose the DB mutation guard.
    env = _scrub_delegated_child_kanban_env(env)

    return env


def build_subprocess_env(
    base: "Mapping[str, str] | None" = None,
    *,
    inherit_profile_home: bool = True,
    scrub_secrets: bool = True,
    extra: "Mapping[str, str] | None" = None,
) -> dict[str, str]:
    """Single factory for building a child-process environment.

    Every spawn site in the codebase should build its env through this
    function (or :func:`hermes_subprocess_env` for the model-driving-CLI
    surface) instead of copying ``os.environ`` directly, so profile-home
    propagation (``HERMES_HOME`` / subprocess ``HOME`` contract) and the
    Hermes secret-scrub policy have a single owner.  History: ~11 separate
    commits each fixed one more spawn site that missed profile-HOME or
    secret-scrub propagation; this factory is the fix for the class.

    Parameters:

    * ``base`` — starting environment.  ``None`` (default) snapshots
      ``os.environ``.  Pass an explicit mapping to build on a caller-prepared
      env instead.
    * ``scrub_secrets=True`` (default) — delegate to
      :func:`_sanitize_subprocess_env`, the long-standing owner of the scrub
      list (provider blocklist + ``_is_hermes_internal_secret`` dynamic
      patterns + kanban/venv-marker/session-context guards) **and** of
      ``HERMES_HOME`` / subprocess-HOME propagation.  On this path profile
      home propagation is inherent — ``inherit_profile_home`` is ignored
      (always applied), exactly matching today's sanitize semantics.
    * ``scrub_secrets=False`` — preserve the base env content byte-for-byte
      (no key is removed).  Use for children that intentionally receive
      secrets (git credential flows, ``bws``/``op`` secret CLIs) or where
      scrubbing could change behavior.  The site is still a win: it becomes
      grep-able and future-fixable.
    * ``inherit_profile_home`` — on the non-scrub path, when True, bridge the
      context-local Hermes home override into ``HERMES_HOME`` and apply the
      subprocess HOME contract (``hermes_constants.apply_subprocess_home_env``).
      Pass False to keep the inherited env untouched (exact legacy
      ``os.environ.copy()`` behavior).
    * ``extra`` — applied **last** on the non-scrub path so explicit caller
      overrides (e.g. a session-scoped ``HERMES_HOME``) always win.  On the
      scrub path it is forwarded as ``_sanitize_subprocess_env``'s
      ``extra_env`` (same force-prefix / blocklist handling as today).
    """
    if scrub_secrets:
        # _sanitize_subprocess_env already performs HERMES_HOME override
        # bridging + apply_subprocess_home_env unconditionally; delegating
        # wholesale keeps one owner and zero drift.
        return _sanitize_subprocess_env(
            dict(base) if base is not None else os.environ.copy(),
            dict(extra) if extra else None,
        )

    env: dict[str, str] = dict(base) if base is not None else os.environ.copy()
    if inherit_profile_home:
        _inject_context_hermes_home(env)
        from hermes_constants import apply_subprocess_home_env
        apply_subprocess_home_env(env)
    if extra:
        env.update(extra)
    return env


def _find_bash() -> str:
    """Find bash for command execution."""
    if not _IS_WINDOWS:
        return (
            shutil.which("bash")
            or ("/usr/bin/bash" if os.path.isfile("/usr/bin/bash") else None)
            or ("/bin/bash" if os.path.isfile("/bin/bash") else None)
            or os.environ.get("SHELL")
            or "/bin/sh"
        )

    candidates: list[str] = []

    custom = os.environ.get("HERMES_GIT_BASH_PATH")
    if custom and os.path.isfile(custom):
        candidates.append(custom)

    # Prefer our own portable Git install — a broken or partially-uninstalled
    # system Git (or a stale HERMES_GIT_BASH_PATH pointing at one) must not
    # brick the terminal.  install.ps1 drops PortableGit here when needed.
    #
    # Layouts (both checked so upgrades between MinGit and PortableGit
    # installs work transparently):
    #   PortableGit: %LOCALAPPDATA%\hermes\git\bin\bash.exe   (primary)
    #   MinGit:      %LOCALAPPDATA%\hermes\git\usr\bin\bash.exe (legacy/32-bit fallback)
    _local_appdata = os.environ.get("LOCALAPPDATA", "")
    _hermes_portable_git = os.path.join(_local_appdata, "hermes", "git") if _local_appdata else ""
    if _hermes_portable_git:
        for candidate in (
            os.path.join(_hermes_portable_git, "bin", "bash.exe"),        # PortableGit (primary)
            os.path.join(_hermes_portable_git, "usr", "bin", "bash.exe"), # MinGit fallback
        ):
            if os.path.isfile(candidate) and candidate not in candidates:
                candidates.append(candidate)

    # Check known Git for Windows install locations before PATH lookup.
    # On machines with both WSL and Git for Windows, shutil.which("bash")
    # may return WSL's bash (which doesn't understand Windows paths and
    # will fail silently).  Explicit Git-for-Windows paths avoid that.
    for candidate in (
        os.path.join(os.environ.get("ProgramFiles", r"C:\Program Files"), "Git", "bin", "bash.exe"),
        os.path.join(os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)"), "Git", "bin", "bash.exe"),
        os.path.join(_local_appdata, "Programs", "Git", "bin", "bash.exe") if _local_appdata else "",
    ):
        if candidate and os.path.isfile(candidate) and candidate not in candidates:
            candidates.append(candidate)

    found = shutil.which("bash")
    if found and found not in candidates:
        candidates.append(found)

    # Prefer the first candidate that can actually start.  A stale
    # HERMES_GIT_BASH_PATH pointing at a broken Git-for-Windows install
    # (``Directory \\drivers\\etc does not exist``) must not win over a
    # healthy portable Git under %LOCALAPPDATA%\\hermes\\git.
    for candidate in candidates:
        if _bash_starts(candidate):
            if candidate != custom and custom and os.path.isfile(custom):
                logger.warning(
                    "HERMES_GIT_BASH_PATH=%s fails to start; using %s instead",
                    custom,
                    candidate,
                )
            return candidate

    if candidates:
        probe_details = "\n".join(
            detail
            for candidate in candidates
            if (detail := _bash_probe_details_cache.get(candidate))
        )
        if _mandatory_aslr_enabled() is True or _looks_like_msys_spawn_failure(
            probe_details
        ):
            raise RuntimeError(_git_bash_aslr_help(candidates[0], probe_details))

        # Last resort for failures unrelated to the known MSYS/ASLR class:
        # return the first path so the caller still sees the real bash error
        # instead of the less useful "not found" message.
        return candidates[0]

    raise RuntimeError(
        "Git Bash not found. Hermes Agent requires Git for Windows on Windows.\n"
        "Install it from: https://git-scm.com/download/win\n"
        "Or set HERMES_GIT_BASH_PATH to your bash.exe location."
    )


_bash_starts_cache: dict[str, bool] = {}
_bash_probe_details_cache: dict[str, str] = {}
_mandatory_aslr_enabled_cache: "bool | None" = None

_BASH_EXTERNAL_PROGRAM_PROBE = "/usr/bin/true; /usr/bin/cat --version >/dev/null"


def _looks_like_msys_spawn_failure(details: str) -> bool:
    """Match Git-for-Windows child-launch failures associated with ASLR."""
    lowered = details.lower()
    return any(
        marker in lowered
        for marker in (
            "dofork:",
            "child_copy:",
            "0xc0000142",
            "0xc0000005",
        )
    )


def _mandatory_aslr_enabled() -> "bool | None":
    """Return Windows' system-wide ForceRelocateImages state when available."""
    global _mandatory_aslr_enabled_cache
    if _mandatory_aslr_enabled_cache is not None:
        return _mandatory_aslr_enabled_cache

    try:
        powershell = shutil.which("powershell.exe") or "powershell.exe"
        result = subprocess.run(
            [
                powershell,
                "-NoProfile",
                "-NonInteractive",
                "-Command",
                "(Get-ProcessMitigation -System).Aslr.ForceRelocateImages.ToString()",
            ],
            capture_output=True,
            text=True, encoding="utf-8", errors="replace",
            timeout=10,
            creationflags=windows_hide_flags(),
        )
        if result.returncode != 0:
            return None
        value = (result.stdout or "").strip().upper()
        if value == "ON":
            _mandatory_aslr_enabled_cache = True
            return True
        if value in {"OFF", "NOTSET"}:
            _mandatory_aslr_enabled_cache = False
            return False
    except Exception as exc:
        logger.debug("Could not query Windows Mandatory ASLR state: %s", exc)
    return None


def _git_root_from_bash(bash: str) -> str:
    """Resolve Git's root from either <root>/bin or <root>/usr/bin bash."""
    bin_dir = ntpath.dirname(ntpath.normpath(bash))
    if ntpath.basename(bin_dir).lower() != "bin":
        return ntpath.dirname(bin_dir)
    parent = ntpath.dirname(bin_dir)
    if ntpath.basename(parent).lower() == "usr":
        return ntpath.dirname(parent)
    return parent


def _git_bash_aslr_help(bash: str, details: str = "") -> str:
    """Build the targeted per-program Mandatory-ASLR remediation."""
    git_root = _git_root_from_bash(bash)
    escaped_root = git_root.replace("'", "''")
    detail_line = f"\nGit Bash probe output: {details[:500]}" if details else ""
    return (
        f"Git Bash at {bash} cannot launch required MSYS child processes while "
        "Windows Mandatory ASLR (ForceRelocateImages) is enabled, or its output "
        f"matches that Git-for-Windows failure class.{detail_line}\n"
        "Reinstalling Git will not change the Windows mitigation policy. Open "
        "PowerShell as Administrator and run:\n"
        f"$gitRoot = '{escaped_root}'\n"
        'Get-Item "$gitRoot\\bin\\bash.exe", "$gitRoot\\usr\\bin\\*.exe" '
        "-ErrorAction SilentlyContinue | ForEach-Object { "
        "Set-ProcessMitigation -Name $_.FullName -Disable ForceRelocateImages }\n"
        "Then restart Hermes. If the override is blocked or later re-applied, "
        "ask your Windows administrator to allow this per-program exception."
    )


def _bash_starts(bash: str) -> bool:
    """True if *bash* can launch external MSYS programs.

    Uses ``--noprofile --norc`` so a broken login post-install
    (``Directory \\drivers\\etc``) does not falsely condemn an otherwise
    usable bash. The external ``true`` and ``cat`` calls are intentional:
    a builtin-only ``exit 0`` probe misses Git-for-Windows fork/spawn failures
    under system-wide Mandatory ASLR. Cached per path for the process lifetime.
    """
    cached = _bash_starts_cache.get(bash)
    if cached is not None:
        return cached

    try:
        result = subprocess.run(
            [bash, "--noprofile", "--norc", "-c", _BASH_EXTERNAL_PROGRAM_PROBE],
            capture_output=True,
            text=True, encoding="utf-8", errors="replace",
            timeout=15,
            creationflags=windows_hide_flags() if _IS_WINDOWS else 0,
        )
        ok = result.returncode == 0
        if not ok:
            combined = f"{result.stdout or ''}{result.stderr or ''}"
            _bash_probe_details_cache[bash] = combined.strip()[:2000]
            logger.debug("bash probe failed for %s: %s", bash, combined.strip()[:200])
    except Exception as exc:
        _bash_probe_details_cache[bash] = str(exc)[:2000]
        logger.debug("bash probe error for %s: %s", bash, exc)
        ok = False

    _bash_starts_cache[bash] = ok
    return ok


_git_bash_bin_dirs_cache: "list[str] | None" = None


def _git_bash_bin_dirs() -> list[str]:
    """Git Bash's coreutils/binary dirs, in ``/etc/profile`` precedence order.

    A non-login ``bash -c`` (the fallback used when ``bash -l`` is broken —
    the classic Windows ``Directory \\drivers\\etc does not exist`` failure)
    never sources ``/etc/profile``, so it never gets ``…\\usr\\bin`` on PATH.
    That directory holds every coreutil the file/terminal tools shell out to
    (``cat``, ``mktemp``, ``mv``, ``wc``, ``head``, ``stat``, ``chmod``,
    ``mkdir``, ``find`` …).  Without it, ``write_file`` fails with an empty
    error (the failure text went to a missing binary's stderr) and terminal
    commands exit 127.  We derive these dirs from the resolved ``bash.exe`` so
    the fallback shell can find coreutils regardless of the login shell.

    Returns ``[]`` off Windows or when bash can't be located.  Dirs are
    returned in the order Git Bash's own ``/etc/profile`` prepends them
    (mingw first, then usr/bin, then bin) and only if they exist on disk.
    """
    global _git_bash_bin_dirs_cache
    if _git_bash_bin_dirs_cache is not None:
        return _git_bash_bin_dirs_cache

    if not _IS_WINDOWS:
        _git_bash_bin_dirs_cache = []
        return _git_bash_bin_dirs_cache

    dirs: list[str] = []
    try:
        bash = _find_bash()
    except Exception:
        _git_bash_bin_dirs_cache = []
        return _git_bash_bin_dirs_cache

    bin_dir = os.path.dirname(bash)          # <root>\bin  or  <root>\usr\bin
    parent = os.path.dirname(bin_dir)
    # MinGit ships bash under usr\bin; PortableGit/system Git under bin.
    root = os.path.dirname(parent) if os.path.basename(parent).lower() == "usr" else parent

    # Order mirrors Git-for-Windows /etc/profile so coreutils win over the
    # same-named Windows System32 tools (find.exe, sort.exe) inside the shell.
    for candidate in (
        os.path.join(root, "mingw64", "bin"),
        os.path.join(root, "mingw32", "bin"),
        os.path.join(root, "usr", "local", "bin"),
        os.path.join(root, "usr", "bin"),
        os.path.join(root, "bin"),
    ):
        if os.path.isdir(candidate) and candidate not in dirs:
            dirs.append(candidate)

    _git_bash_bin_dirs_cache = dirs
    return dirs


def _prepend_git_bash_dirs(existing_path: str) -> str:
    """Prepend Git Bash's binary dirs to ``existing_path`` if missing.

    No-op off Windows or when the dirs can't be resolved.  First-occurrence
    wins, so a PATH that already lists a dir keeps its position.  This is what
    lets the non-login ``bash -c`` fallback find coreutils; in the healthy
    case the session snapshot re-exports the full login PATH inside the shell,
    so this only matters when that snapshot is absent.
    """
    git_dirs = _git_bash_bin_dirs()
    if not git_dirs:
        return existing_path
    sep = os.pathsep
    entries = [e for e in existing_path.split(sep) if e] if existing_path else []
    missing = [d for d in git_dirs if d not in entries]
    if not missing:
        return existing_path
    return sep.join([*missing, *entries])


# POSIX-sh-family shells that understand the ``[shell, "-lic", "set +m; …"]``
# invocation spawn_local uses. $SHELL values outside this set (fish, csh/tcsh,
# nushell, elvish, xonsh, …) would error on that syntax, so _find_shell falls
# back to bash for them rather than honouring $SHELL. (#42203)
_SPAWN_COMPATIBLE_SHELLS = frozenset({"bash", "zsh", "sh", "dash", "ksh", "mksh"})


def _find_shell() -> str:
    """Find the user's login shell for background process spawning.

    Unlike ``_find_bash`` (which always returns a bash binary for callers
    that explicitly need bash), this function prefers the user's configured
    ``$SHELL`` on POSIX so that ``spawn_local`` uses the shell the user
    actually logs in with.

    On macOS Catalina+ the default login shell is zsh, but
    ``shutil.which("bash")`` still finds the system ``/bin/bash`` (GNU bash
    3.2).  When bash 3.2 is invoked with ``-l`` (login) and stdin is
    ``/dev/null``, it sources ``~/.bash_profile`` which on many macOS setups
    contains ``exec /bin/zsh -l``.  That ``exec`` replaces bash with zsh but
    drops the ``-c`` argument, so the background command never runs — the
    subprocess exits 0 with no output and no side effects.

    Preferring ``$SHELL`` (when it is a POSIX-``sh``-family shell) avoids this
    because zsh/bash/sh/dash/ksh handle ``-lic`` correctly even with
    redirected stdin.

    Only POSIX-sh-family shells are honoured: ``spawn_local`` invokes the
    shell as ``[shell, "-lic", "set +m; <cmd>"]``, and that ``-lic`` bundle +
    ``set +m`` job-control syntax is NOT understood by fish, csh/tcsh,
    nushell, elvish, xonsh, etc.  Returning such a ``$SHELL`` would trade the
    bash-3.2 swallow for a parse error on every background command, so for any
    non-allowlisted shell we fall back to ``_find_bash`` (the prior behaviour).

    On Windows, ``$SHELL`` is typically bash (Git Bash), so behaviour is
    unchanged — we fall through to ``_find_bash``.
    """
    if not _IS_WINDOWS:
        user_shell = os.environ.get("SHELL")
        if (
            user_shell
            and os.path.isfile(user_shell)
            and os.access(user_shell, os.X_OK)
            and Path(user_shell).name in _SPAWN_COMPATIBLE_SHELLS
        ):
            return user_shell
    return _find_bash()


# Standard PATH entries for environments with minimal PATH.
_SANE_PATH = (
    "/opt/homebrew/bin:/opt/homebrew/sbin:"
    "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
)

# Cached directory containing the ``hermes`` console-script.
# ``_SENTINEL`` distinguishes "not resolved yet" from a resolved ``None``.
_SENTINEL = object()
_HERMES_BIN_DIR: "str | None | object" = _SENTINEL


def _resolve_hermes_bin_dir() -> str | None:
    """Return the directory holding the ``hermes`` console-script, or None.

    The terminal tool runs in a freshly-spawned subshell whose PATH is the
    agent process's PATH plus a static set of system dirs (``_SANE_PATH``).
    When the gateway is launched by something that does NOT source the user's
    shell rc — systemd, a service manager, a desktop launcher, cron — the
    hermes install dir (``~/.local/bin``, the venv ``bin``/``Scripts``, pipx,
    nix) is absent from that PATH, so plugins shelling out to bare ``hermes``
    via the terminal tool hit ``command not found`` (exit 127) even though
    ``hermes`` works fine in the user's own interactive terminal.

    We resolve the install dir once (it never changes within a process) and
    prepend-if-missing it to the subshell PATH so bare ``hermes`` resolves
    regardless of how the gateway was started.

    Resolution order (cheap, no heavy imports):
      1. ``shutil.which("hermes")`` — normal PATH-installed shim.
      2. The directory of ``sys.argv[0]`` when it's an absolute path to a
         real ``hermes`` executable (covers nix-store / venv wrappers).
      3. The directory of ``sys.executable`` — the running interpreter's
         venv ``bin``/``Scripts`` is where its console-scripts live.
    """
    global _HERMES_BIN_DIR
    if _HERMES_BIN_DIR is not _SENTINEL:
        return _HERMES_BIN_DIR  # type: ignore[return-value]

    candidate: str | None = None

    which = shutil.which("hermes")
    if which:
        candidate = os.path.dirname(which)

    if candidate is None:
        argv0 = sys.argv[0] if sys.argv else ""
        base = os.path.basename(argv0).lower()
        if (
            os.path.isabs(argv0)
            and (base == "hermes" or base.startswith("hermes."))
            and os.path.isfile(argv0)
        ):
            candidate = os.path.dirname(argv0)

    if candidate is None:
        exe_dir = os.path.dirname(sys.executable) if sys.executable else ""
        if exe_dir:
            shim = "hermes.exe" if _IS_WINDOWS else "hermes"
            if os.path.isfile(os.path.join(exe_dir, shim)):
                candidate = exe_dir

    if candidate and not os.path.isdir(candidate):
        candidate = None

    _HERMES_BIN_DIR = candidate
    return candidate


def _prepend_hermes_bin_dir(existing_path: str) -> str:
    """Prepend the hermes install dir to ``existing_path`` if it's missing.

    Cross-platform (uses ``os.pathsep``). First-occurrence wins, so a PATH
    that already contains the dir is returned unchanged. Returns the input
    unchanged when the install dir can't be resolved.
    """
    bin_dir = _resolve_hermes_bin_dir()
    if not bin_dir:
        return existing_path
    sep = os.pathsep
    entries = [e for e in existing_path.split(sep) if e] if existing_path else []
    if bin_dir in entries:
        return existing_path
    return sep.join([bin_dir, *entries])


def _managed_runtime_path_entries() -> list[str]:
    """Return existing Hermes-managed runtime dirs for the terminal subshell PATH.

    The terminal tool spawns a subshell whose PATH is the agent process's PATH
    plus ``_SANE_PATH``. Neither carries the runtimes Hermes installs for
    itself, so on a machine where Hermes provisioned its own toolchain a
    command the agent runs resolves a system copy instead — or nothing at all:

    - ``$HERMES_HOME/node`` (+ ``/bin``) — installed to satisfy the desktop and
      browser toolchain. ``tools/browser_tool.py`` already does this for its own
      subprocesses; the agent's shell deserves the same.
    - ``$HERMES_HOME/bin`` — the managed ``uv``. ``install.sh`` writes it there
      and nothing has ever put that directory on PATH, so an install whose only
      uv is the managed one looks uv-less to both the agent and the model.

    Resolved per call rather than cached in a module constant because
    ``get_hermes_home()`` is profile-scoped and a managed tree can appear
    mid-process (``heal_hermes_managed_node``, a first browser install).
    """
    try:
        from hermes_constants import get_hermes_home, iter_hermes_node_dirs

        candidates = [*iter_hermes_node_dirs(), get_hermes_home() / "bin"]
        return [str(d) for d in candidates if d.is_dir()]
    except Exception:
        return []


def _append_missing_sane_path_entries(existing_path: str) -> str:
    """Return a normalised POSIX PATH with missing sane entries appended.

    On POSIX the caller-supplied PATH is rewritten (not merely appended to):
    empty entries and duplicate entries are dropped, preserving
    first-occurrence order, then each missing ``_SANE_PATH`` entry is appended
    once at the end so existing entries keep their precedence.

    Two intentional normalisations beyond the bare "add Homebrew dirs" fix:

    - **Empty entries are stripped.** A leading/trailing/double ``:`` encodes
      an empty PATH element, which POSIX shells interpret as the current
      working directory — a mild foot-gun in a default terminal environment.
      We drop these rather than carry them through.
    - **Duplicates are collapsed** (first occurrence wins), so a caller PATH
      that already contains repeats is not propagated verbatim.

    Hermes-managed runtime dirs are appended alongside the sane entries, not
    prepended: a tool the user deliberately put on their own PATH still wins,
    and the managed one only fills the gap where there would otherwise be
    nothing.

    For a well-formed PATH (no empties, no duplicates) the leading segment is
    byte-identical to the input and ordering is preserved; only the missing
    sane entries are appended. On Windows this is a no-op passthrough (the
    separator is ``;`` and the native PATH must not be touched).
    """
    if _IS_WINDOWS:
        return existing_path

    sane_entries = [entry for entry in _SANE_PATH.split(":") if entry]
    sane_entries.extend(
        entry for entry in _managed_runtime_path_entries() if entry not in sane_entries
    )
    if not existing_path:
        return ":".join(sane_entries)

    # De-duplicate the caller PATH (first occurrence wins) and drop empty
    # entries before merging in the sane fallbacks.
    seen: set[str] = set()
    ordered_entries: list[str] = []
    for entry in existing_path.split(":"):
        if not entry or entry in seen:
            continue
        seen.add(entry)
        ordered_entries.append(entry)

    # _SANE_PATH is a static, duplicate-free constant, so a membership check
    # against the caller entries is sufficient — no need to track `seen` here.
    for entry in sane_entries:
        if entry not in seen:
            ordered_entries.append(entry)

    return ":".join(ordered_entries)


def _apply_windows_msys_bash_env_defaults(env: dict) -> None:
    """Disable MSYS argument path conversion for Git Bash subprocesses.

    Git Bash rewrites arguments that look like Unix paths (``/FO``, ``/TN``,
    ``/Create``) into ``C:/.../git/FO``-style paths, which breaks native
    Windows commands such as ``tasklist``, ``schtasks``, and ``wmic``.  Hermes
    runs terminal commands through bash on Windows, so set the standard MSYS
    opt-out by default.  Users who need conversion can override in their env.
    Refs #56700.

    ``MSYS_NO_PATHCONV`` is honored by Git for Windows bash only.  MSYS2-proper
    and Cygwin bash (which ``_find_bash`` can still return via the final
    ``shutil.which`` fallback) ignore it and honor ``MSYS2_ARG_CONV_EXCL``
    instead, so set both.  ``*`` disables all argv conversion — the semantic
    equivalent of ``MSYS_NO_PATHCONV=1``.  Also fixes ``cmd /c`` mangling
    (#56147).
    """
    if not _IS_WINDOWS:
        return
    env.setdefault("MSYS_NO_PATHCONV", "1")
    env.setdefault("MSYS2_ARG_CONV_EXCL", "*")


def _path_env_key(run_env: dict) -> str | None:
    """Return the PATH env key to update without altering Windows casing.

    Note: this is deliberately a *second* Windows guard, distinct from the
    early-return in ``_append_missing_sane_path_entries``. Its job is to pick
    the correctly-cased key (``Path`` vs ``PATH``) so completion writes back to
    the key the caller already used; the helper's guard makes that helper safe
    to call standalone (it is, e.g. in the Windows unit tests). Both are
    intentional.
    """
    if not _IS_WINDOWS:
        return "PATH"
    for key in run_env:
        if key.upper() == "PATH":
            return key
    return None


def _make_run_env(env: dict) -> dict:
    """Build a run environment with a sane PATH and provider-var stripping."""
    try:
        from tools.env_passthrough import (
            is_env_passthrough as _is_passthrough,
            resolve_passthrough_value as _resolve_passthrough_value,
        )
    except Exception:
        _is_passthrough = lambda _: False  # noqa: E731
        _resolve_passthrough_value = lambda _name, fallback: fallback  # noqa: E731

    merged = dict(os.environ | env)
    run_env = {}
    for k, v in merged.items():
        if k.startswith(_HERMES_PROVIDER_ENV_FORCE_PREFIX):
            real_key = k[len(_HERMES_PROVIDER_ENV_FORCE_PREFIX):]
            if _is_hermes_internal_secret(real_key):
                continue
            run_env[real_key] = v
        elif _is_hermes_internal_secret(k):
            continue
        else:
            first_party = _is_terminal_first_party_env(k)
            passthrough = _is_passthrough(k)
            if k in _HERMES_PROVIDER_ENV_BLOCKLIST and not (passthrough or first_party):
                continue
            # First-party vars use the merged env value directly (see
            # _sanitize_subprocess_env); only passthrough names resolve.
            value = v
            if passthrough and not first_party:
                value = _resolve_passthrough_value(k, v)
            if value is not None:
                run_env[k] = value
    path_key = _path_env_key(run_env)
    if path_key is not None:
        new_path = _append_missing_sane_path_entries(run_env.get(path_key, ""))
        # On Windows, ensure Git Bash's coreutils dirs (…\usr\bin etc.) are on
        # PATH.  A non-login ``bash -c`` fallback (used when ``bash -l`` is
        # broken) never sources /etc/profile, so without this cat/mktemp/mv and
        # friends are missing and every write_file/terminal call fails (empty
        # error / exit 127).  No-op off Windows and when a login snapshot is
        # healthy (the snapshot re-exports the full PATH inside the shell).
        new_path = _prepend_git_bash_dirs(new_path)
        # Ensure the hermes install dir is reachable so plugins can shell out
        # to bare ``hermes`` via the terminal tool even when the gateway was
        # launched without it on PATH (systemd, service managers, cron, etc.).
        run_env[path_key] = _prepend_hermes_bin_dir(new_path)

    _inject_context_hermes_home(run_env)

    from hermes_constants import apply_subprocess_home_env
    apply_subprocess_home_env(run_env)

    # Bridge ContextVar-based session vars into the subprocess env (with the
    # cross-session leak guard — strips _UNSET vars when a concurrent host is
    # engaged so a sibling session's os.environ mirror can't leak in).
    _inject_session_context_env(run_env)

    _strip_hermes_owned_pythonpath_and_runtime_markers(run_env)

    _apply_windows_msys_bash_env_defaults(run_env)

    run_env = _scrub_delegated_child_kanban_env(run_env)

    return run_env


def _same_path(left: Path, right: Path) -> bool:
    """Compare path spellings with host filesystem case semantics."""
    left_parts = [os.path.normcase(part) for part in left.parts]
    right_parts = [os.path.normcase(part) for part in right.parts]
    return left_parts == right_parts


def _build_hermes_repo_root_aliases(
    resolved_root: Path,
    lexical_root: Path,
    configured_home: Path,
) -> tuple[Path, ...]:
    """Return exact repo-root spellings emitted by Hermes launchers.

    ``gateway_windows._preserve_hermes_home_path`` maps a physical path under
    the resolved HERMES_HOME back onto the configured HERMES_HOME spelling.
    Mirror that producer contract here so a junction-backed install is matched
    without treating arbitrary descendants of HERMES_HOME as Hermes-owned.
    Additionally, when the repo itself is a junction under the configured root
    (repo-level junction, possibly cross-drive), the single deterministic
    candidate <root>/<repo dirname> is accepted only when strict resolve
    proves it is the exact physical repo root.
    """
    aliases: list[Path] = []

    def add(candidate: Path) -> None:
        if not any(_same_path(candidate, existing) for existing in aliases):
            aliases.append(candidate)

    add(resolved_root)
    add(lexical_root)

    # Profile re-home: with --profile / sticky active_profile the configured
    # home becomes <root>/profiles/<name>.  The repo root then lives beside
    # the profiles directory (not under the profile home), so the home-
    # relative mapping below cannot reach it.  Derive the root spelling
    # lexically the same way get_default_hermes_root() does (parent of a
    # "profiles" component) and run the same exact-ownership mapping against
    # it -- this recovers the launcher's lexical root under profile re-home
    # while still never matching arbitrary descendants of HERMES_HOME.
    home_candidates = [configured_home]
    if configured_home.parent.name == "profiles":
        home_candidates.append(configured_home.parent.parent)

    for home in home_candidates:
        try:
            resolved_home = home.resolve()
            home_key = os.path.normcase(str(resolved_home))
            root_key = os.path.normcase(str(resolved_root))
            if os.path.commonpath([home_key, root_key]) == home_key:
                relative_root = os.path.relpath(str(resolved_root), str(resolved_home))
                add(home / relative_root)
        except (OSError, ValueError):
            pass

    # Repo-level junction recovery: the repository itself may be a
    # junction/symlink under the configured root (e.g. D:\hermes\hermes-agent
    # -> C:\...\hermes-agent) while the import spelling (editable install)
    # resolves to the physical location.  The home-relative mapping above
    # cannot express a cross-drive link (commonpath raises on different
    # drives), so prove the EXACT filesystem identity of the single
    # deterministic candidate -- <lexical root>/<repo dirname> -- with a
    # strict resolve before accepting it as Hermes-owned.  Fail-closed: a
    # missing path (strict resolve raises), a real directory that is not the
    # known physical root, or any unrelated spelling never becomes an alias.
    for home in home_candidates:
        repo_candidate = home / resolved_root.name
        try:
            if repo_candidate.resolve(strict=True) == resolved_root.resolve(strict=True):
                add(repo_candidate)
        except OSError:
            pass

    return tuple(aliases)


# --- Hermes venv / repo-root detection (module-level, computed once) ---

#: The Hermes repository root - three levels up from this file
#: (``tools/environments/local.py`` -> ``tools/environments`` -> ``tools``
#: -> repo root).  This is the directory the Electron app prepends to
#: PYTHONPATH so the backend can do ``import tools``, ``import hermes_cli``,
#: etc.  Subprocesses that are NOT the Hermes backend don't need it and it
#: can shadow local packages.
_hermes_repo_root: Path = Path(__file__).resolve().parents[2]

#: Alternate spellings of the repo root that Hermes launchers may emit.
#: ``Path(__file__).resolve()`` canonicalizes symlinks/junctions, but the
#: Windows gateway launcher deliberately renders Hermes-owned paths under
#: the configured HERMES_HOME spelling (which may be a junction to another
#: drive — see ``hermes_cli/gateway_windows.py::_preserve_hermes_home_path``).
#: ``Path(__file__)`` (unresolved) keeps that spelling, so a PYTHONPATH
#: entry written by the launcher still matches even though it differs
#: lexically from the resolved root.
_hermes_repo_root_aliases: tuple[Path, ...] = _build_hermes_repo_root_aliases(
    _hermes_repo_root,
    Path(__file__).absolute().parents[2],
    get_process_hermes_home(),
)

#: Whether the current interpreter is running inside a venv.  On Python 3.3+
#: ``sys.base_prefix != sys.prefix`` indicates a venv (or virtualenv).
#: ``sys.real_prefix`` is the old virtualenv (<20) marker.
_in_venv: bool = (
    getattr(sys, "base_prefix", sys.prefix) != sys.prefix
    or hasattr(sys, "real_prefix")
)

#: Cached set of site-packages directories that belong to the running
#: interpreter's own venv.  Computed lazily (once) because ``site`` import
#: and path construction are not free and this function is called on every
#: subprocess spawn.
_hermes_site_packages: list[Path] | None = None


def _validated_runtime_venv(env: dict) -> Path | None:
    """Return a producer-owned runtime venv identified by VIRTUAL_ENV.

    A user may carry an unrelated VIRTUAL_ENV, so the variable alone is not
    provenance.  The legacy Windows base-Python gateway producer uses the exact
    ``<Hermes repo>/venv`` layout and a real venv marker; require both before
    accepting its separate runtime venv.
    """
    value = env.get("VIRTUAL_ENV")
    if not value:
        return None

    candidate = Path(value)
    if not any(_same_path(candidate, repo_root / "venv") for repo_root in _hermes_repo_root_aliases):
        return None

    try:
        if not (candidate / "pyvenv.cfg").is_file():
            return None
    except OSError:
        return None

    return candidate


def _get_hermes_site_packages(env: dict) -> list[Path]:
    """Return exact site-packages dirs owned by the Hermes runtime.

    Uses ``site.getsitepackages()`` when available for robustness (it respects
    ``.pth`` rewrites and platform conventions), with a manual fallback that
    constructs the canonical path from ``sys.prefix`` for POSIX and Windows.
    A validated Windows base-interpreter launch contributes its separate
    ``VIRTUAL_ENV/Lib/site-packages`` directory as an additional exact entry.
    """
    global _hermes_site_packages
    if _hermes_site_packages is not None:
        result = list(_hermes_site_packages)
    else:
        result = []
        if _in_venv:
            try:
                import site
                for sp in site.getsitepackages():
                    result.append(Path(sp))
            except Exception:
                pass

            # Fallback: construct manually.  On POSIX:
            #   sys.prefix / lib / python{X.Y} / site-packages
            # On Windows:
            #   sys.prefix / Lib / site-packages
            if not result:
                if _IS_WINDOWS:
                    result.append(Path(sys.prefix) / "Lib" / "site-packages")
                else:
                    pyver = f"python{sys.version_info[0]}.{sys.version_info[1]}"
                    result.append(Path(sys.prefix) / "lib" / pyver / "site-packages")

        _hermes_site_packages = list(result)

    runtime_venv = _validated_runtime_venv(env)
    if runtime_venv is not None:
        runtime_site_packages = runtime_venv / "Lib" / "site-packages"
        if not any(_same_path(runtime_site_packages, existing) for existing in result):
            result.append(runtime_site_packages)

    return result


def _strip_hermes_owned_pythonpath_and_runtime_markers(env: dict) -> None:
    """Strip Hermes-owned PYTHONPATH entries, then the runtime marker vars.

    Ordering is load-bearing: PYTHONPATH filtering must run BEFORE the
    markers are removed so a validated Windows base-interpreter launch
    (VIRTUAL_ENV -> <repo>/venv) can still prove ownership.
    """
    _strip_hermes_owned_pythonpath(env)
    for _marker in _ACTIVE_VENV_MARKER_VARS:
        env.pop(_marker, None)


def _strip_hermes_owned_pythonpath(env: dict) -> None:
    """Remove Hermes-owned PYTHONPATH entries from subprocess environments.

    Launchers prepend the Hermes repo root and the Hermes venv's
    site-packages so the backend can ``import tools``; leaking those into a
    child Python of a DIFFERENT version makes it load the backend's C
    extensions and crash (``numpy._core._multiarray_umath``, ``PIL._imaging``,
    ``cryptography``).  Blanket-removing PYTHONPATH would discard legitimate
    user entries, so only entries proven Hermes-owned are removed:

    1. The exact repo root (never direct children -- no launcher injects
       one, and user paths under the repo must survive).
    2. The exact runtime site-packages dirs (running interpreter's venv or
       a validated Windows base-Python runtime venv; descendants are user
       paths).

    Everything else -- user libs, Nix plugin paths, a pythonX.Y/site-packages
    entry meant for a DIFFERENT child version -- is preserved byte-for-byte:
    ownership is decided by path provenance, never by a cross-version
    heuristic (#74817 follow-up).
    """
    pp = env.get("PYTHONPATH")
    if not pp:
        return

    hermes_site_packages = _get_hermes_site_packages(env)

    kept: list[str] = []
    stripped: list[str] = []

    for entry in pp.split(os.pathsep):
        # Empty and non-normalized components are user-owned semantics.  In
        # particular, an empty component means the current working directory.
        # Preserve raw spelling unless the exact component is Hermes-owned.
        if entry == "":
            kept.append(entry)
            continue

        entry_path = Path(entry)
        should_strip = False

        # --- Check 1: Hermes venv site-packages ---
        # Producers inject the exact directory, never a descendant.  Exact
        # matching avoids deleting a user path nested below site-packages.
        for sp in hermes_site_packages:
            if _same_path(entry_path, sp):
                should_strip = True
                break
        if should_strip:
            stripped.append(entry)
            continue

        # --- Check 2: Hermes repo root ---
        # The Electron app prepends the repo root so ``import tools`` works
        # in the backend.  Subprocesses don't need it and it can shadow
        # local packages of the same name.  Only the EXACT root is stripped:
        # no launcher injects a direct child (``<repo>/tools`` etc.) as an
        # independent PYTHONPATH entry, and user paths that merely happen to
        # live under the repo directory must be preserved.  Both the
        # resolved and unresolved (HERMES_HOME/junction) spellings count as
        # Hermes-owned.
        if not should_strip:
            should_strip = any(
                _same_path(entry_path, repo_root)
                for repo_root in _hermes_repo_root_aliases
            )

        if should_strip:
            stripped.append(entry)
        else:
            kept.append(entry)

    if kept:
        env["PYTHONPATH"] = os.pathsep.join(kept)
    else:
        env.pop("PYTHONPATH", None)

    if stripped:
        logger.debug(
            "Stripped Hermes-owned entries from PYTHONPATH: %s",
            stripped,
        )


def _read_terminal_shell_init_config() -> tuple[list[str], bool]:
    """Return (shell_init_files, auto_source_bashrc) from config.yaml.

    Best-effort — returns sensible defaults on any failure so terminal
    execution never breaks because the config file is unreadable.
    """
    try:
        from hermes_cli.config import load_config

        cfg = load_config() or {}
        terminal_cfg = cfg.get("terminal") or {}
        files = terminal_cfg.get("shell_init_files") or []
        if not isinstance(files, list):
            files = []
        auto_bashrc = bool(terminal_cfg.get("auto_source_bashrc", True))
        return [str(f) for f in files if f], auto_bashrc
    except Exception:
        return [], True


def _resolve_shell_init_files() -> list[str]:
    """Resolve the list of files to source before the login-shell snapshot.

    Expands ``~`` and ``${VAR}`` references and drops anything that doesn't
    exist on disk, so a missing ``~/.bashrc`` never breaks the snapshot.
    The ``auto_source_bashrc`` path runs only when the user hasn't supplied
    an explicit list — once they have, Hermes trusts them.
    """
    explicit, auto_bashrc = _read_terminal_shell_init_config()

    candidates: list[str] = []
    if explicit:
        candidates.extend(explicit)
    elif auto_bashrc and not _IS_WINDOWS:
        # Build a login-shell-ish source list so tools like n / nvm / asdf /
        # pyenv that self-install into the user's shell rc land on PATH in
        # the captured snapshot.
        #
        # ~/.profile and ~/.bash_profile run first because they have no
        # interactivity guard — installers like ``n`` and ``nvm`` append
        # their PATH export there on most distros, and a non-interactive
        # ``. ~/.profile`` picks that up.
        #
        # ~/.bashrc runs last. On Debian/Ubuntu the default bashrc starts
        # with ``case $- in *i*) ;; *) return;; esac`` and exits early
        # when sourced non-interactively, which is why sourcing bashrc
        # alone misses nvm/n PATH additions placed below that guard. We
        # still include it so users who put PATH logic in bashrc (and
        # stripped the guard, or never had one) keep working.
        candidates.extend(["~/.profile", "~/.bash_profile", "~/.bashrc"])

    resolved: list[str] = []
    for raw in candidates:
        try:
            path = os.path.expandvars(os.path.expanduser(raw))
        except Exception:
            continue
        if path and os.path.isfile(path):
            resolved.append(path)
    return resolved


def _prepend_shell_init(cmd_string: str, files: list[str]) -> str:
    """Prepend ``source <file>`` lines (guarded + silent) to a bash script.

    Each file is wrapped so a failing rc file doesn't abort the whole
    bootstrap: ``set +e`` keeps going on errors, ``2>/dev/null`` hides
    noisy prompts, and ``|| true`` neutralises the exit status.
    """
    if not files:
        return cmd_string

    prelude_parts = ["set +e"]
    for path in files:
        # shlex.quote isn't available here without an import; the files list
        # comes from os.path.expanduser output so it's a concrete absolute
        # path.  Escape single quotes defensively anyway.
        safe = path.replace("'", "'\\''")
        prelude_parts.append(f"[ -r '{safe}' ] && . '{safe}' 2>/dev/null || true")
    prelude = "\n".join(prelude_parts) + "\n"
    return prelude + cmd_string


class LocalEnvironment(BaseEnvironment):
    """Run commands directly on the host machine.

    Spawn-per-call: every execute() spawns a fresh bash process.
    Session snapshot preserves env vars across calls.
    CWD persists via file-based read after each command.
    """

    _profile_scoped_passthrough = True

    # Commands run on the Hermes host itself — controller-side platform
    # behavior (macOS TCC pruning, etc.) legitimately applies here.
    is_local = True

    def _additional_profile_scoped_passthrough_names(self) -> tuple[str, ...]:
        """Return first-party terminal env names (``BUZZ_*``) present in the
        current env, so they are excluded from the shared session snapshot.

        The login-shell snapshot (``init_session`` ``export -p`` dump and the
        per-command re-dump) captures the child env, which now includes the
        ``BUZZ_*`` vars the terminal carve-out passes through. The exclusion
        set is derived from ``get_all_passthrough()`` plus backend-specific
        additions — and ``BUZZ_*`` can NEVER be in it, because env_passthrough
        refuses blocklisted names (GHSA-rhgp-j443-p4rf). Under a multiplexed
        gateway, profile A's BUZZ_PRIVATE_KEY would land in
        ``hermes-snap-<id>.sh`` and a later command from profile B sharing
        this collapsed LocalEnvironment would ``source`` it: a cross-profile
        nsec leak that defeats profile isolation.

        Treating these names like profile-scoped passthrough names keeps them
        out of the dump and save/restores the current profile's value (or
        unsets the name) per command in ``_wrap_command``. The set is monotonic
        for the environment lifetime: once a name is seen it stays excluded,
        so a later profile that lacks the var still gets the unset-guard.
        """
        merged = dict(os.environ | self.env)
        return tuple(
            sorted(
                name
                for name in merged
                # Prefix-only on purpose: the snapshot exclusion stays
                # conservative even when the context-gated carve-out is
                # inactive (the var then never reaches the child env anyway,
                # but a monotonic exclusion is a cheap extra guard).
                if isinstance(name, str) and _matches_terminal_first_party_prefix(name)
            )
        )

    def __init__(self, cwd: str = "", timeout: int = 60, env: dict = None):
        cwd = _resolve_local_initial_cwd(cwd)
        super().__init__(cwd=cwd, timeout=timeout, env=env)
        self.init_session()

    def get_temp_dir(self) -> str:
        """Return a shell-safe writable temp dir for local execution.

        Termux does not provide /tmp by default, but exposes a POSIX TMPDIR.
        Prefer POSIX-style env vars when available, keep using /tmp on regular
        Unix systems, and only fall back to tempfile.gettempdir() when it also
        resolves to a POSIX path.

        Check the environment configured for this backend first so callers can
        override the temp root explicitly (for example via terminal.temp_dir,
        terminal.env, or a custom TMPDIR), then fall back to the host process
        environment.

        **Default (no override set):** a dedicated cache dir under
        ``HERMES_HOME`` (``~/.hermes/cache/terminal``) rather than ``/tmp``.
        On several distros (Arch and friends) ``/tmp`` is a small RAM-backed
        tmpfs, and Hermes session artifacts — background-process logs,
        code-execution sandboxes, spilled tool results — can fill it under
        load. Real storage is the safer default; stale artifacts are pruned
        by ``cleanup_terminal_temp_cache`` (gateway housekeeping + a
        once-per-process best-effort sweep) since we no longer get tmpfs
        reboot wipes for free.

        **Windows:** hardcoded ``/tmp`` is wrong in two ways — native Python
        can't open the path, and the Windows default temp (``%TEMP%``) often
        contains spaces (``C:\\Users\\Some Name\\AppData\\Local\\Temp``) that
        break unquoted bash interpolations.  Use a dedicated cache dir under
        ``HERMES_HOME`` instead — single-word path, guaranteed to exist, same
        string resolves in both Git Bash and native Python.
        """
        if _IS_WINDOWS:
            # Derive a Windows-safe temp dir under HERMES_HOME.  Using
            # forward slashes makes the same string work unchanged in bash
            # command interpolations AND in Python ``open()`` — Windows
            # accepts forward slashes in filesystem paths, and we control
            # the path so we can guarantee no spaces.
            try:
                from hermes_constants import get_hermes_home
                cache_dir = get_hermes_home() / "cache" / "terminal"
            except Exception:
                cache_dir = Path(tempfile.gettempdir()) / "hermes_terminal"
            cache_dir.mkdir(parents=True, exist_ok=True)
            _prune_terminal_temp_once()
            # Force forward slashes so the same string serves both contexts.
            return str(cache_dir).replace("\\", "/")

        # Explicit temp-dir override from terminal.temp_dir (TERMINAL_TEMP_DIR).
        # Honored ahead of the generic TMPDIR so users can redirect Hermes' temp
        # root to real storage when /tmp is a small tmpfs.
        configured = self.env.get("TERMINAL_TEMP_DIR") or os.environ.get("TERMINAL_TEMP_DIR")
        if configured and configured.startswith("/") and os.path.isdir(configured):
            return configured.rstrip("/") or "/"

        for env_var in ("TMPDIR", "TMP", "TEMP"):
            candidate = self.env.get(env_var) or os.environ.get(env_var)
            if candidate and candidate.startswith("/"):
                return candidate.rstrip("/") or "/"

        # Default: HERMES_HOME/cache/terminal — real storage, mirroring the
        # Windows branch above. /tmp is only a last-resort fallback now
        # because RAM-backed tmpfs /tmp fills up under Hermes load.
        try:
            from hermes_constants import get_hermes_home
            cache_dir = get_hermes_home() / "cache" / "terminal"
            cache_dir.mkdir(parents=True, exist_ok=True)
            resolved = str(cache_dir)
            if resolved.startswith("/") and os.access(resolved, os.W_OK | os.X_OK):
                _prune_terminal_temp_once()
                return resolved.rstrip("/") or "/"
        except Exception:
            pass

        if os.path.isdir("/tmp") and os.access("/tmp", os.W_OK | os.X_OK):
            return "/tmp"

        candidate = tempfile.gettempdir()
        if candidate.startswith("/"):
            return candidate.rstrip("/") or "/"

        return "/tmp"

    @staticmethod
    def _quote_cwd_for_cd(cwd: str) -> str:
        """Use native paths for Python, but Git Bash-friendly paths for cd."""
        return BaseEnvironment._quote_cwd_for_cd(_windows_to_msys_path(cwd))

    def _quote_shell_path(self, path: str) -> str:
        """Rewrite native/mixed Windows paths before quoting for Git Bash."""
        return _quote_bash_path(path)

    def _run_bash(self, cmd_string: str, *, login: bool = False,
                  timeout: int = 120,
                  stdin_data: str | None = None) -> subprocess.Popen:
        bash = _find_bash()
        # For login-shell invocations (used by init_session to build the
        # environment snapshot), prepend sources for the user's bashrc /
        # custom init files so tools registered outside bash_profile
        # (nvm, asdf, pyenv, …) end up on PATH in the captured snapshot.
        # Non-login invocations are already sourcing the snapshot and
        # don't need this.
        if login:
            init_files = _resolve_shell_init_files()
            if init_files:
                cmd_string = _prepend_shell_init(cmd_string, init_files)
        args = [bash, "-l", "-c", cmd_string] if login else [bash, "-c", cmd_string]
        run_env = _make_run_env(self.env)

        # Recover when the cwd has been deleted out from under us — usually by
        # a previous tool call that ran ``rm -rf`` on its own working dir
        # (issue #17558).  Popen would otherwise raise FileNotFoundError on
        # the cwd before bash starts, wedging every subsequent call until the
        # gateway restarts.
        #
        # On Windows, ``_resolve_safe_cwd`` also normalises Git Bash-style
        # POSIX paths (``/c/Users/...``) to native form so a perfectly valid
        # ``pwd -P`` result from bash isn't mistakenly treated as "missing"
        # and spammed as a warning on every command.
        safe_cwd = _resolve_safe_cwd(self.cwd)
        if safe_cwd != self.cwd:
            # MSYS → Windows translation alone shouldn't surface as a warning
            # (it's a benign normalization, not a recovery). Only warn when
            # the directory really doesn't exist on disk.
            normalized = _msys_to_windows_path(self.cwd) if _IS_WINDOWS else self.cwd
            if safe_cwd != normalized:
                logger.warning(
                    "LocalEnvironment cwd %r is missing on disk; "
                    "falling back to %r so terminal commands keep working.",
                    self.cwd,
                    safe_cwd,
                )
            self.cwd = safe_cwd

        _popen_cwd = self.cwd

        _popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {}

        proc = subprocess.Popen(
            args,
            text=True,
            env=run_env,
            encoding="utf-8",
            errors="replace",
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            stdin=subprocess.PIPE if stdin_data is not None else subprocess.DEVNULL,
            start_new_session=True,
            cwd=_popen_cwd,
            **_popen_kwargs,
        )
        if not _IS_WINDOWS:
            try:
                proc._hermes_pgid = os.getpgid(proc.pid)
            except ProcessLookupError:
                pass

        if stdin_data is not None:
            _pipe_stdin(proc, stdin_data)

        return proc

    def _kill_process(self, proc):
        """Kill the entire process group (all children)."""

        def _group_alive(pgid: int) -> bool:
            try:
                # POSIX-only: _IS_WINDOWS is handled before this helper is used.
                os.killpg(pgid, 0)  # windows-footgun: ok — POSIX process-group alive probe
                return True
            except ProcessLookupError:
                return False
            except PermissionError:
                # The group exists, even if this process cannot signal it.
                return True

        def _wait_for_group_exit(pgid: int, timeout: float) -> bool:
            deadline = time.monotonic() + timeout
            while time.monotonic() < deadline:
                # Reap the wrapper promptly. A dead but unreaped group leader
                # still makes killpg(pgid, 0) report the group as alive.
                try:
                    proc.poll()
                except Exception:
                    pass
                if not _group_alive(pgid):
                    return True
                time.sleep(0.05)
            try:
                proc.poll()
            except Exception:
                pass
            return not _group_alive(pgid)

        try:
            if _IS_WINDOWS:
                try:
                    from gateway.status import get_process_start_time, terminate_pid

                    terminate_pid(
                        proc.pid,
                        force=True,
                        expected_start_time=get_process_start_time(proc.pid),
                    )
                except Exception:
                    proc.kill()
                try:
                    proc.wait(timeout=2.0)
                except (subprocess.TimeoutExpired, OSError):
                    pass
            else:
                try:
                    pgid = os.getpgid(proc.pid)
                except ProcessLookupError:
                    pgid = getattr(proc, "_hermes_pgid", None)
                    if pgid is None:
                        raise

                # Snapshot the descendant set BEFORE the first signal: once
                # the wrapper dies its children reparent to init and a parent
                # walk finds nothing (same rationale as agent/deadline.py
                # kill_process_tree).  A descendant that called ``setsid``
                # escapes the process group entirely and would survive the
                # group-kill below — the #71148 class, terminal flavor
                # (issue #84967's local sibling).  The snapshot must never
                # break the kill path, so any failure just yields an empty
                # sweep set.
                descendants: list = []
                try:
                    import psutil

                    descendants = psutil.Process(proc.pid).children(recursive=True)
                except Exception:
                    descendants = []

                def _sweep_escaped_descendants() -> None:
                    """SIGKILL snapshotted survivors outside the (dead) group.

                    Runs after the TERM→KILL group escalation so in-group
                    members keep their SIGTERM grace window; only escapees
                    (own setsid sessions) are force-killed.  psutil's
                    identity-aware Process means recycled PIDs are skipped.

                    POSIX-only: reached solely from the non-_IS_WINDOWS
                    branch above (the win32 path returns earlier).
                    """
                    for child in descendants:
                        try:
                            if not child.is_running():
                                continue
                            try:
                                if os.getpgid(child.pid) == pgid:
                                    continue  # group-kill already covers it
                            except (ProcessLookupError, PermissionError, OSError):
                                pass
                            child.kill()
                        except Exception:
                            continue

                try:
                    os.killpg(pgid, signal.SIGTERM)  # windows-footgun: ok — POSIX process-group SIGTERM (guarded by _IS_WINDOWS above)
                except ProcessLookupError:
                    _sweep_escaped_descendants()
                    return

                # Wait on the process group, not just the shell wrapper. Under
                # load the wrapper can exit before grandchildren do; returning
                # at that point leaves orphaned process-group members behind.
                if _wait_for_group_exit(pgid, 1.0):
                    _sweep_escaped_descendants()
                    return

                try:
                    # POSIX-only: _IS_WINDOWS is handled by the outer branch.
                    os.killpg(pgid, signal.SIGKILL)  # windows-footgun: ok — POSIX process-group SIGKILL
                except ProcessLookupError:
                    _sweep_escaped_descendants()
                    return
                _wait_for_group_exit(pgid, 2.0)
                try:
                    proc.wait(timeout=0.2)
                except (subprocess.TimeoutExpired, OSError):
                    pass
                _sweep_escaped_descendants()
        except (ProcessLookupError, PermissionError, OSError):
            try:
                proc.kill()
            except Exception:
                pass

    def _update_cwd(self, result: dict):
        """Update cwd from the stdout marker emitted by the wrapped command.

        The base command wrapper already appends ``pwd -P`` to stdout inside a
        session-specific marker, so the local backend can share the same parser
        as remote backends instead of re-reading the temp file it just wrote.
        ``_extract_cwd_from_output`` keeps the local Windows normalization and
        stale-path rollback semantics intact.
        """
        self._extract_cwd_from_output(result)

    def _extract_cwd_from_output(self, result: dict):
        """Same semantics as the base class, but on Windows the value
        emitted by ``pwd -P`` inside Git Bash is in MSYS form
        (``/c/Users/x``). Normalize to native Windows form and validate
        the directory exists before assigning to ``self.cwd`` — otherwise
        ``_run_bash``'s safe-cwd recovery would warn on every subsequent
        command.

        Always defers to the base class for stripping the marker text from
        ``result["output"]`` so output formatting is identical.
        """
        # Snapshot pre-existing cwd, defer to base for parsing + marker
        # stripping, then validate / normalize whatever it assigned.
        prev_cwd = self.cwd
        super()._extract_cwd_from_output(result)
        if self.cwd != prev_cwd:
            normalized = _msys_to_windows_path(self.cwd) if _IS_WINDOWS else self.cwd
            if normalized and os.path.isdir(normalized):
                self.cwd = normalized
                result["cwd"] = normalized
            else:
                # Stale / non-existent path — keep previous cwd; _run_bash
                # will resolve a safe fallback on the next call if needed.
                # The rollback restores a value this command did not observe,
                # so it is not attributable to this command's session either.
                self.cwd = prev_cwd
                result.pop("cwd_observed", None)
                result.pop("cwd", None)

    def cleanup(self):
        """Clean up temp files."""
        for f in (self._snapshot_path, self._cwd_file):
            try:
                os.unlink(f)
            except OSError:
                pass
        # Remove any orphaned atomic-write temp snapshots (snap.tmp.<bashpid>)
        # a failed/interrupted mv could have left behind (#38249).
        try:
            import glob
            for tmp in glob.glob(f"{self._snapshot_path}.tmp.*"):
                try:
                    os.unlink(tmp)
                except OSError:
                    pass
        except Exception:
            pass