"""MCP server config loading and stdio launch environment: ${VAR}/Cursor-style
interpolation, hidden-whitespace and suspicious-entry filtering, the filtered
subprocess env, command resolution, the cached-npx binary shortcut and the shared stderr log."""

import codecs
import hashlib
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import threading
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple
from hermes_cli.stderr_timestamp import stamp_line, timestamp
from tools.mcp_tool_common import _env_ref_name, _prepend_path

logger = logging.getLogger("tools.mcp_tool")

_mcp_stderr_log_fh: dict[str, Any] = {}  # profile home key -> handle
_mcp_stderr_log_lock = threading.Lock()


def _get_mcp_stderr_log() -> Any:
    """Shared append-mode handle for MCP subprocess stderr, cached until shutdown PER PROFILE HOME (a
    multiplexed gateway's secondary profile must log under ITS ``logs/``, not the launch profile's). Must
    expose a real fd (asyncio wires the child's stderr to it); falls back to ``/dev/null``, then real stderr."""
    from hermes_constants import get_hermes_home, hermes_home_key, mkdir_under_hermes_home
    home_key = hermes_home_key()
    with _mcp_stderr_log_lock:
        fh = _mcp_stderr_log_fh.get(home_key)
        if fh is None or fh.closed:
            try:
                log_dir = get_hermes_home() / "logs"
                mkdir_under_hermes_home(log_dir)
                # Line-buffered so output lands promptly; errors="replace" tolerates garbled binary.
                fh = open(log_dir / "mcp-stderr.log", "a", encoding="utf-8", errors="replace", buffering=1)
                fh.fileno()  # confirm a real fd before committing
            except Exception as exc:  # pragma: no cover — best-effort fallback
                logger.debug("Failed to open MCP stderr log, using devnull: %s", exc)
                try:
                    fh = open(os.devnull, "w", encoding="utf-8")
                except Exception:
                    fh = sys.stderr
            _mcp_stderr_log_fh[home_key] = fh
        return fh


def _close_mcp_stderr_logs(*, scope: Optional[str] = None) -> None:
    """Release cached parent handles after the selected MCP transports have stopped."""
    with _mcp_stderr_log_lock:
        keys = list(_mcp_stderr_log_fh) if scope is None else [scope]
        for key in keys:
            fh = _mcp_stderr_log_fh.pop(key, None)
            # The last-resort fallback is borrowed, not owned by MCP.
            if fh is None or fh is sys.stderr or fh is sys.__stderr__:
                continue
            try:
                fh.close()
            except OSError:
                logger.warning("Could not close MCP stderr log for %s", key, exc_info=True)


class _StderrTee:
    """A stdio child's stderr, copied into the shared log one stamped line at a time while the last few KB
    stay readable (raw), so a server that dies at startup can say why on the MCP status surfaces instead of
    only in the log (#124264). ``sink`` is handed to the child; ``close()`` returns the captured tail."""

    _TAIL_BYTES = 16384

    def __init__(self, log_fh: Any):
        read_fd, write_fd = os.pipe()
        self.sink = os.fdopen(write_fd, "wb", buffering=0)
        self._log, self._tail = log_fh, bytearray()
        self._reader = threading.Thread(target=self._pump, args=(read_fd,), name="mcp-stderr", daemon=True)
        self._reader.start()

    def _pump(self, read_fd: int) -> None:
        decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
        pending = ""
        with os.fdopen(read_fd, "rb", buffering=0) as source:
            while chunk := source.read(65536):
                self._tail = (self._tail + chunk)[-self._TAIL_BYTES:]
                # Stamp whole lines only; a partial line waits for its newline (or EOF).
                *lines, pending = (pending + decoder.decode(chunk)).split("\n")
                self._write_lines(lines)
        if rest := pending + decoder.decode(b"", final=True):
            self._write_lines([rest])

    def _write_lines(self, lines: list[str]) -> None:
        if not lines:
            return
        try:
            self._log.write("".join(stamp_line(line) for line in lines))
            self._log.flush()
        except (OSError, ValueError):  # log closed at shutdown: keep draining the child
            pass

    def close(self, timeout: float = 2.0) -> str:
        """Close our write end and give the reader *timeout* to drain (a surviving grandchild can keep
        the pipe open); the tail read so far."""
        self.sink.close()
        self._reader.join(timeout)
        return self._tail.decode("utf-8", errors="replace")


def _write_stderr_log_header(server_name: str) -> None:
    """Session marker so operators can find each server's output in the shared log; it leads with the
    same stamp as every server line (``_StderrTee``) so ``hermes logs mcp --since`` can filter it."""
    fh = _get_mcp_stderr_log()
    try:
        fh.write(f"\n{timestamp()} ===== starting MCP server '{server_name}' =====\n")
        fh.flush()
    except Exception:
        pass


def _tail_server_stderr(server_name: str, *, max_bytes: int = 8192, max_lines: int = 8) -> str:
    """Trailing lines this server last wrote to the shared stderr log; ``''`` when none.

    A connect failure that only says ``Connection closed`` hides why the child died
    (#125300: a ``ModuleNotFoundError`` sat in the log while the agent log stayed mute).
    Best-effort: the segment header must still be inside the read window, and another
    server's output past that header is fair game — the header lines carry the name."""
    try:
        fh = _get_mcp_stderr_log()
        path = getattr(fh, "name", None)
        if not path or path == os.devnull:
            return ""
        with open(path, "rb") as raw:
            raw.seek(0, os.SEEK_END)
            size = raw.tell()
            raw.seek(max(0, size - max_bytes))
            chunk = raw.read().decode("utf-8", "replace")
        cut = chunk.rfind(f"starting MCP server '{server_name}'")
        if cut < 0:
            return ""
        lines = [line for line in chunk[cut:].splitlines() if line.strip()]
        return "\n".join(lines[-max_lines:]) if lines else ""
    except Exception:
        return ""


# Env vars safe to pass to stdio subprocesses (no secrets).
_SAFE_ENV_KEYS = frozenset({"PATH", "HOME", "USER", "LANG", "LC_ALL", "TERM", "SHELL", "TMPDIR"})

# Windows process/location vars needed by launcher-style tools (e.g. Docker Desktop's MCP plugin discovery).
_SAFE_ENV_KEYS_CASE_INSENSITIVE = frozenset({
    "ALLUSERSPROFILE", "APPDATA", "COMMONPROGRAMFILES", "COMMONPROGRAMFILES(X86)",
    "COMMONPROGRAMW6432", "COMPUTERNAME", "COMSPEC", "HOMEDRIVE", "HOMEPATH",
    "LOCALAPPDATA", "NUMBER_OF_PROCESSORS", "OS", "PATHEXT", "PROCESSOR_ARCHITECTURE",
    "PROGRAMDATA", "PROGRAMFILES", "PROGRAMFILES(X86)", "PROGRAMW6432", "PUBLIC",
    "SYSTEMDRIVE", "SYSTEMROOT", "TEMP", "TMP", "USERDOMAIN", "USERNAME",
    "USERPROFILE", "WINDIR"})

# ${VAR_NAME} interpolation; any non-} chars allowed so MY-VAR / my.var work.
_ENV_VAR_PATTERN = re.compile(r"\$\{([^}]+)\}")


def _workspace_folder() -> str:
    """Absolute workspace root for ``${workspaceFolder}``: the session's authoritative root
    (terminal cwd / task override / $TERMINAL_CWD), else cwd."""
    try:
        from tools.file_tools_paths import _authoritative_workspace_root
        root = _authoritative_workspace_root()
    except Exception:
        root = None
    return root or os.getcwd()


def _workspace_basename() -> str:
    root = _workspace_folder()
    return os.path.basename(root.rstrip("/\\")) or root


# Cursor's case-sensitive context vars -> resolver.
_CONTEXT_VAR_RESOLVERS = {
    "userHome": lambda: os.path.expanduser("~"), "workspaceFolder": lambda: _workspace_folder(),
    "workspaceFolderBasename": _workspace_basename, "pathSeparator": lambda: os.sep, "/": lambda: os.sep}


class _RenderedStr(str):
    """A config string after ``${...}`` rendering, carrying how it resolved, so later steps never
    guess from text that may itself look like ``${...}``. ``missed``: some env reference was unset
    (or empty) and kept its placeholder; a re-render (connect renders again under the owner's fresh
    scope) retries from ``template``, while a fully resolved value is final. ``unset_ref``: the
    template was one whole env reference that missed."""

    def __new__(cls, text: str = "", template: Optional[str] = None, missed: bool = False, unset_ref: bool = False):
        rendered = super().__new__(cls, text)
        rendered.template = text if template is None else template
        rendered.missed = missed
        rendered.unset_ref = unset_ref
        return rendered


def _build_safe_env(user_env: Optional[dict]) -> dict:
    """Filtered env for stdio subprocesses so API keys/tokens don't leak: the safe baseline
    keys, ``XDG_*``, vars injected by an external secret source (users configured that backend
    precisely so subprocesses can consume them), plus the server config's own ``env``."""
    from agent.secret_scope import get_secret
    from hermes_cli.env_loader import secret_source_names
    env = {
        key: value for key, value in os.environ.items()
        if key in _SAFE_ENV_KEYS or key.upper() in _SAFE_ENV_KEYS_CASE_INSENSITIVE or key.startswith("XDG_")}
    # Source-tagged names are process-wide (any profile's hydration tags them) while os.environ
    # holds only the LAUNCH profile's values, so the value must come from the active profile's
    # secret scope; a profile that lacks the name gets nothing, never another profile's token.
    for key in secret_source_names():
        value = get_secret(key)
        if value is not None:
            env[key] = value
    for key in ("HERMES_KANBAN_DB", "HERMES_KANBAN_BOARD"):
        if key in os.environ:
            env[key] = os.environ[key]
    if user_env:
        # An entry rendered from one whole ``${VAR}`` that is unset (or empty) counts as absent, so
        # a plugin can forward an optional var without the child seeing the literal placeholder.
        # The key keeps its baseline value, if any: the MCP SDK re-adds its own inherited set
        # (PATH, HOME, TERM, ...) at spawn anyway, so popping here would not unset it.
        for key, value in user_env.items():
            if not getattr(value, "unset_ref", False):
                env[key] = value
    from agent.delegation_context import delegated_child_subprocess_env
    return delegated_child_subprocess_env(env)


def _which_with_config_pathext(command: str, path_arg, env: dict):
    """Resolve *command* under the config env's PATHEXT (Windows only; ``shutil.which`` uses the PARENT's).

    The extension walk mirrors ``which`` itself (existing-suffix short-circuit, configured
    extensions in order) but reads nothing from and writes nothing to ``os.environ``: swapping
    the parent's PATHEXT around a ``which`` call would publish this server's per-profile value
    to every other thread for the duration, and a ``finally``-restore cannot undo that window."""
    cfg_pathext = next((v for k, v in env.items() if k.upper() == "PATHEXT" and isinstance(v, str) and v.strip()), None)
    if not cfg_pathext or cfg_pathext == os.environ.get("PATHEXT"):
        return None
    # PATHEXT is Windows-defined: ";"-separated even when resolved off-Windows
    exts = [ext for ext in cfg_pathext.split(";") if ext]
    candidates = [command + ext for ext in exts]
    if not candidates or any(command.lower().endswith(ext.lower()) for ext in exts):
        candidates = [command]
    directories = str(path_arg or "").split(os.pathsep)
    if sys.platform == "win32" and os.curdir not in directories:
        directories.insert(0, os.curdir)  # Windows resolves from the cwd first
    for raw in directories:
        directory = raw or os.curdir  # POSIX: an empty PATH component means the cwd
        if not os.path.isdir(directory):
            continue
        for candidate in candidates:
            resolved = os.path.join(directory, candidate)
            if os.path.isfile(resolved) and os.access(resolved, os.F_OK | os.X_OK):
                return resolved
    return None


# Bare MCP launchers Hermes ships through PM, keyed to the package that provides them.
_MANAGED_LAUNCHERS = {"npx": "npm", "npm": "npm", "node": "npm", "uv": "uv", "uvx": "uv"}


def _managed_launcher(command: str) -> Optional[tuple[str, list[str]]]:
    """PM's executable for a bare launcher name and the toolchain dirs its children need first on
    PATH (npx's ``env node``, uvx's sibling uv). Never the user's copy: a missing managed tool is
    provisioned through PM, which raises naming the remedy when it may not. None only when PM
    ships no build for this platform (Termux), where the platform's own copy IS the toolchain."""
    import pm

    package = _MANAGED_LAUNCHERS[command]
    if pm.get_package(package).missing_reason(pm.current_target()) is not None:
        return None
    pm.ensure(package)
    if package == "uv":
        launcher = pm.uv_launcher(command)
        dirs = [str(launcher.parent)] if launcher is not None else []
    else:
        from hermes_constants import with_hermes_node_path

        dirs = [d for d in with_hermes_node_path({"PATH": ""})["PATH"].split(os.pathsep) if d]
    executable = shutil.which(command, path=os.pathsep.join(dirs)) if dirs else None
    if executable is None:
        raise RuntimeError(f"Hermes-managed {command} is not installed; run `hermes pm install {package}`")
    return executable, dirs


def _outside_package(launcher: str, managed: str, dirs: list[str]) -> tuple[str, list[str]]:
    """Hermes's npm/node *launcher* and toolchain *dirs*, swapped for PM's copies outside a
    sealed Windows payload (``pm.user_facing.copies``). npm starts each package bin through
    cmd.exe, and cmd.exe cannot start the packaged ``node.exe``."""
    from pm.user_facing import copies

    moved = copies(dirs)
    if moved == dirs:
        return managed, dirs
    # Plain which(): Hermes's own copy dirs hold one real node.exe/npx.cmd, so the parent's
    # PATHEXT is the right question (unlike a user server's config env, see _which_with_config_pathext).
    return shutil.which(launcher, path=os.pathsep.join(moved)) or managed, moved


def _is_hermes_managed_bin_dir(directory: str) -> bool:
    """True for the bin dirs Hermes' bootstrap prepends to this process's PATH: anything
    under the active hermes home (the sealed payload's venv, ``<home>/bin``, PM store
    runtimes) plus the running interpreter's own bin dir (a repo checkout's venv)."""
    try:
        resolved = Path(directory).resolve()
    except OSError:
        return False
    if resolved == Path(sys.executable).resolve().parent:
        return True
    from hermes_constants import get_hermes_home
    try:
        home = Path(get_hermes_home()).resolve()
    except Exception:
        return False
    return resolved == home / "bin" or home in resolved.parents


_WINDOWS_DEFAULT_PATHEXT = ".COM;.EXE;.BAT;.CMD;.VBS;.VBE;.JS;.JSE;.WSF;.WSH;.MSC"


def _pathext_suffixes(env: Optional[dict] = None, *, windows: Optional[bool] = None) -> list:
    """Executable suffixes a bare name resolves through, in order. The child env's PATHEXT
    comes first (``shutil.which`` reads the PARENT's, so a per-profile config value never
    reaches a plain ``which`` — same source as ``_which_with_config_pathext``), then the
    parent's, then the OS default. POSIX appends nothing. ``windows`` injectable for the
    same testability reason as ``_npx_bin_candidates``."""
    is_windows = os.name == "nt" if windows is None else windows
    if not is_windows:
        return [""]
    for source in (env or {}, os.environ):
        value = next((v for k, v in source.items()
                      if k.upper() == "PATHEXT" and isinstance(v, str) and v.strip()), None)
        if value:
            exts = [ext for ext in value.split(";") if ext]
            if exts:
                return exts
    return [ext for ext in _WINDOWS_DEFAULT_PATHEXT.split(";") if ext]


def _first_user_which_hit(command: str, path_arg: Optional[str],
                          env: Optional[dict] = None, *, windows: Optional[bool] = None) -> Optional[str]:
    """First PATH hit for *command* OUTSIDE Hermes-managed bin dirs, or ``None``.

    ``shutil.which`` stops at the first hit, and bootstrap prepends the managed runtime's
    bin dir, so a bare ``python3`` resolves to the bundled interpreter — which lacks the
    user's packages and kills the server on import (#125300). Candidates run through
    ``_pathext_suffixes`` rather than the npx cache layout's ``.cmd``/``.exe`` pair: a user
    install may only ship a ``.bat``/``.py`` wrapper, and missing it here would silently
    fall back to the managed hit this exists to step past."""
    exts = _pathext_suffixes(env, windows=windows)
    if any(ext and command.lower().endswith(ext.lower()) for ext in exts):
        names = [command]
    else:
        names = [command + ext for ext in exts]
    for directory in str(path_arg or "").split(os.pathsep):
        if not directory or _is_hermes_managed_bin_dir(directory):
            continue
        for name in names:
            candidate = os.path.join(directory, name)
            if os.path.isfile(candidate) and os.access(candidate, os.F_OK | os.X_OK):
                return candidate
    return None


def _resolve_stdio_command(command: str, env: dict) -> tuple[str, dict]:
    """Resolve a stdio command against the exact subprocess env (bare launchers under a filtered PATH).

    Bare ``npx``/``npm``/``node``/``uv``/``uvx`` resolve to Hermes's PM-managed copies with their
    toolchain dirs first on the child PATH, never the user's (an absolute ``command:`` stays the
    user's choice). Anything else resolves on the child env's PATH only: ``shutil.which`` with
    ``path=None`` silently falls back to the PARENT's ``os.environ["PATH"]``, letting a command
    "resolve" against an env the child will never be spawned with. An absent child PATH is a
    miss; an explicitly empty one keeps its cwd-only meaning (same distinction the child's
    ``execvp`` will see); a miss stays as-written for an honest spawn failure."""
    resolved_command = os.path.expanduser(str(command).strip())
    resolved_env = dict(env or {})
    launcher = re.sub(r"\.(cmd|exe)$", "", resolved_command, flags=re.IGNORECASE)  # Windows spellings
    managed = _managed_launcher(launcher) if launcher in _MANAGED_LAUNCHERS else None
    if managed is not None:
        resolved_command, dirs = managed
        if _MANAGED_LAUNCHERS[launcher] == "npm":
            resolved_command, dirs = _outside_package(launcher, resolved_command, dirs)
        # Moved to the front even when already on PATH behind a user's copy.
        keys = {os.path.normcase(d) for d in dirs}
        rest = [p for p in resolved_env.get("PATH", "").split(os.pathsep) if p and os.path.normcase(p) not in keys]
        resolved_env["PATH"] = os.pathsep.join([*dirs, *rest])
        return resolved_command, resolved_env
    if os.sep not in resolved_command:
        path_arg = resolved_env.get("PATH")
        which_hit = shutil.which(resolved_command, path=path_arg) if path_arg is not None else None
        if which_hit is None and sys.platform == "win32" and resolved_env:
            which_hit = _which_with_config_pathext(resolved_command, path_arg, resolved_env)
        # A bare command keeps the USER's PATH semantics: bootstrap prepends the managed
        # runtime's bin dir to this process's PATH, so the first hit for a bare ``python3``
        # is Hermes' bundled interpreter, which lacks the user's packages and dies on import
        # (#125300). Step past managed dirs to the user's own hit; the explicit launcher
        # family keeps the managed-first resolution (that is the point of
        # ``_launcher_fallback``), and a managed-only PATH keeps the managed hit.
        if which_hit and resolved_command not in {"npx", "npm", "node", "uv", "uvx"}:
            which_hit = _first_user_which_hit(resolved_command, path_arg, resolved_env) or which_hit
        if which_hit:
            resolved_command = which_hit
    command_dir = os.path.dirname(resolved_command)
    if command_dir:
        resolved_env = _prepend_path(resolved_env, command_dir)
    return resolved_command, resolved_env


def _npx_bin_candidates(bin_dir: str, name: str, *, windows: Optional[bool] = None) -> list:
    """Launcher paths to try for *name* inside an npx cache's ``.bin``, in order. On Windows that
    directory holds the extensionless sh script plus ``<name>.cmd``/``<name>.ps1``; the sh one
    cannot be spawned there and ``os.access(X_OK)`` is only an existence check, so select by
    extension (same precedence as ``hermes_constants._candidate_node_command_names``). ``windows``
    is injectable so the branch is testable without patching ``os.name`` process-wide."""
    is_windows = os.name == "nt" if windows is None else windows
    if is_windows:
        return [os.path.join(bin_dir, name + ext) for ext in (".cmd", ".exe")]
    return [os.path.join(bin_dir, name)]


def _npx_cached_bin(args: list) -> Optional[tuple]:
    """Resolve ``npx -y <pkg>`` to the already-installed binary, or None.

    ``npx`` resolves the package and then FORKS, staying resident as the real server's parent
    for nothing (~48 MB private memory per MCP server, measured); Hermes already supervises the
    child (shared death supervisor). When the package is in npx's cache we spawn its binary
    directly. Deliberately conservative — None (caller keeps plain ``npx``, so a cold machine
    still installs) for a cache miss, a version pin (``pkg@1.2.3``), extra npx flags, a manifest
    without one obvious bin, or any unreadable cache entry. Returns ``(binary_path, remaining_args)``."""
    if not isinstance(args, list) or not args:
        return None

    rest = list(args)
    while rest and rest[0] in ("-y", "--yes"):
        rest.pop(0)
    if not rest:
        return None
    # `npx pkg -y` (flag AFTER the spec) would hand the server a flag npx would have eaten.
    if any(str(a) in ("-y", "--yes") for a in rest[1:]):
        return None

    spec = str(rest[0])
    # Scoped names keep their leading '@', so only an '@' AFTER the scope is a version separator.
    if "@" in (spec.removeprefix("@")):
        return None
    if not spec or spec.startswith("-"):
        return None

    cache_root = os.environ.get("npm_config_cache") or os.path.join(os.path.expanduser("~"), ".npm")
    npx_root = os.path.join(cache_root, "_npx")
    if not os.path.isdir(npx_root):
        return None
    try:
        entries = os.listdir(npx_root)
    except OSError:
        return None

    for entry in entries:
        manifest = os.path.join(npx_root, entry, "package.json")
        try:
            with open(manifest, "r", encoding="utf-8-sig") as fh:
                deps = (json.load(fh) or {}).get("dependencies") or {}
        except (OSError, ValueError, TypeError):
            continue
        if spec not in deps:
            continue
        pkg_json = os.path.join(npx_root, entry, "node_modules", spec, "package.json")
        try:
            with open(pkg_json, "r", encoding="utf-8-sig") as fh:
                bin_field = (json.load(fh) or {}).get("bin")
        except (OSError, ValueError, TypeError):
            continue
        if isinstance(bin_field, str):
            names = [os.path.basename(spec)]
        elif isinstance(bin_field, dict) and len(bin_field) == 1:
            names = list(bin_field.keys())
        else:
            continue  # zero or several bins: which one npx would pick is not ours to guess
        bin_dir = os.path.join(npx_root, entry, "node_modules", ".bin")
        for candidate in _npx_bin_candidates(bin_dir, names[0]):
            if os.path.exists(candidate) and os.access(candidate, os.X_OK):
                return candidate, rest[1:]
    return None


def _packaged_npx_entry(command: str, args: list) -> Optional[tuple]:
    """``(node.exe, [bin script, *args])`` for ``npx [-y] <pkg> ...`` when *command* is a sealed
    Windows payload's own npx, else None.

    That npx starts the package bin through cmd.exe, and cmd.exe cannot start the packaged
    ``node.exe`` (``pm.user_facing``). Hermes gets here only without a host Node and without
    a copy that PM may make. The payload's Python CAN start the packaged ``node.exe``, so the
    package is installed with npm's own script and the bin's script runs on that node.
    Install scripts stay off, because npm runs them through cmd.exe too. None keeps plain
    npx (an honest failure) for any other argv shape, a non-registry spec (git, URL, path:
    npm installs those under a name the spec does not show), a failed install, or an
    unclear bin.

    Like npx, only an exact version is reused as installed. Any other spec re-resolves on
    each start and keeps the previous install when npm cannot run (offline).
    """
    from pm.user_facing import _sealed_dirs, applies

    if not applies() or not isinstance(args, list):
        return None
    npm_dirs = {os.path.normcase(d): d for d in _sealed_dirs("npm")}
    npm_dir = npm_dirs.get(os.path.normcase(os.path.dirname(str(command))))
    node = shutil.which("node", path=os.pathsep.join(_sealed_dirs("node")))
    npm_cli = os.path.join(npm_dir or "", "node_modules", "npm", "bin", "npm-cli.js")
    if npm_dir is None or node is None or not os.path.isfile(npm_cli):
        return None
    rest = list(args)
    while rest and rest[0] in ("-y", "--yes"):
        rest.pop(0)
    spec = str(rest[0]) if rest else ""
    if not spec or spec.startswith("-") or any(str(a) in ("-y", "--yes") for a in rest[1:]):
        return None
    if ":" in spec or spec.startswith((".", "/", "\\", "~")) or ("/" in spec and not spec.startswith("@")):
        return None
    scope, _, bare = spec.rpartition("/") if spec.startswith("@") else ("", "", spec)
    base, _, version = bare.partition("@")
    name = (f"{scope}/" if scope else "") + base
    from hermes_constants import get_hermes_home

    prefix = Path(get_hermes_home()) / "cache" / "mcp-npx" / hashlib.sha256(spec.encode()).hexdigest()[:16]
    manifest = prefix / "node_modules" / name / "package.json"
    exact = re.fullmatch(r"\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.+-]+)?", version) is not None
    if not (exact and manifest.is_file()):
        install = [node, npm_cli, "install", "--prefix", str(prefix), "--no-save", "--ignore-scripts",
                   "--no-audit", "--no-fund", "--loglevel=error", spec]
        try:
            result = subprocess.run(install, capture_output=True, text=True, encoding="utf-8", errors="replace",
                                    stdin=subprocess.DEVNULL, timeout=600, check=False)
        except (subprocess.SubprocessError, OSError) as exc:  # hang, unwritable prefix, AV lock, node not startable
            result = subprocess.CompletedProcess(install, 1, "", str(exc))
        if result.returncode != 0 or not manifest.is_file():
            logger.warning("npm install of %s with the packaged node failed: %s", spec,
                           (result.stderr or result.stdout).strip()[-500:])
            if not manifest.is_file():
                return None
    try:
        bin_field = (json.loads(manifest.read_text(encoding="utf-8-sig")) or {}).get("bin")
    except (OSError, ValueError):
        return None
    if isinstance(bin_field, dict):
        bin_field = bin_field.get(base) or (
            next(iter(bin_field.values())) if len(bin_field) == 1 else None)
    if not isinstance(bin_field, str):
        return None
    return node, [str(manifest.parent / bin_field), *rest[1:]]


def _interpolate_env_vars(value):
    """Recursively resolve ``${VAR}`` / Cursor ``${env:VAR}`` placeholders and context vars. Env
    refs resolve from the active profile's secret scope when multiplexing (the routed profile's
    value, not another profile's in ``os.environ``). Unset (or empty) vars keep the literal placeholder;
    a templated string comes back as a ``_RenderedStr`` recording how it resolved."""
    from agent.secret_scope import get_secret as _get_secret
    if isinstance(value, str):
        if isinstance(value, _RenderedStr) and not value.missed:
            return value
        template = getattr(value, "template", value)
        missed = []

        def _replace(m):
            resolver = _CONTEXT_VAR_RESOLVERS.get(m.group(1).strip())
            if resolver is not None:
                return resolver()
            resolved = _get_secret(_env_ref_name(m.group(1)), None)
            if not resolved:
                missed.append(m.group(0))
            return resolved or m.group(0)
        if not _ENV_VAR_PATTERN.search(template):
            return value
        text = _ENV_VAR_PATTERN.sub(_replace, template)
        return _RenderedStr(text, template, missed=bool(missed), unset_ref=missed == [template])
    if isinstance(value, dict):
        return {k: _interpolate_env_vars(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_interpolate_env_vars(v) for v in value]
    return value


def _require_rendered_remote(server_name: str, config: dict) -> dict:
    """*config* back, unless it is a remote server whose ``url`` / ``headers`` still carry a literal
    ``${VAR}`` after rendering: sending that is a guaranteed 401 that reads as a bad credential
    (#119092), so fail closed naming the variable instead."""
    if "url" not in config:
        return config
    values = [config.get("url") or "", *(config.get("headers") or {}).values()]
    unresolved = sorted({m.group(1) for value in values for m in _ENV_VAR_PATTERN.finditer(str(value))})
    if unresolved:
        refs = ", ".join(f"${{{ref}}}" for ref in unresolved)
        unset = ValueError(f"MCP server '{server_name}': {refs} in url/headers is not set in this profile's "
                           ".env or secret source")
        unset.failure_class = "missing_credentials"  # type: ignore[attr-defined]
        raise unset
    return config


# (server_name, dotted key path) pairs already warned about: config loads repeat per discovery pass.
_whitespace_warned: set[tuple[str, str]] = set()


def _warn_hidden_whitespace(server_name: str, config: dict) -> list[str]:
    """Warn once per (server, key path) about string values with leading/trailing whitespace (a
    pasted newline causes opaque auth failures). Advisory only: values are never mutated (could be
    intentional) nor logged (often secrets). Returns flagged paths."""
    flagged: list[str] = []

    def _walk(value: Any, path: str) -> None:
        if isinstance(value, str) and value != value.strip():
            flagged.append(path)
        elif isinstance(value, dict):
            for k, v in value.items():
                _walk(v, f"{path}.{k}" if path else str(k))
        elif isinstance(value, list):
            for i, v in enumerate(value):
                _walk(v, f"{path}[{i}]")
    _walk(config, "")
    for key_path in flagged:
        if (server_name, key_path) not in _whitespace_warned:
            _whitespace_warned.add((server_name, key_path))
            logger.warning(
                "MCP server '%s': config value '%s' has hidden leading or trailing whitespace — this often "
                "causes authentication or connection failures. Check for stray spaces/newlines in config.yaml "
                "(or the referenced env var).", server_name, key_path)
    return flagged


def _malformed_cost_table(name: str, cfg: Any) -> Optional[str]:
    """Why a server's ``tool_costs``/``spend_limit``/``cost_unit`` is unusable, else None. A budget the
    user declared must never be silently dropped while its server keeps spending, so the server is skipped."""
    from tools.mcp_tool_costs import config_cost_policy
    try:
        config_cost_policy(name, cfg if isinstance(cfg, dict) else None)
    except ValueError as exc:
        return str(exc)
    return None


def _filter_suspicious_mcp_servers(servers: dict[str, dict]) -> dict[str, dict]:
    """Drop exfiltration-shaped MCP configs, and servers with a malformed cost table, before any stdio spawn path."""
    try:
        from hermes_cli.mcp_security import validate_mcp_server_entry
    except Exception:
        validate_mcp_server_entry = None
    safe_servers = {}
    for name, cfg in servers.items():
        issues = validate_mcp_server_entry(name, cfg) if validate_mcp_server_entry and isinstance(cfg, dict) else None
        if issues:
            logger.warning("Skipping suspicious MCP server '%s': %s", name, "; ".join(issues))
        elif (bad_costs := _malformed_cost_table(name, cfg)) is not None:
            logger.error("Skipping MCP server '%s': %s (fix tool_costs/spend_limit/cost_unit in config.yaml)",
                         name, bad_costs)
        else:
            safe_servers[name] = cfg
    return safe_servers


def _portable_mcp_servers(safe_servers: dict[str, dict], rejected: frozenset[str] = frozenset()) -> None:
    """Merge plugin-provided (portable) MCP servers into *safe_servers*; native config wins on a clash. Never raises.

    The package's declared costs are facts about its server, so a same-name native entry (the way a user widens
    a package server's trust) keeps them unless it sets the cost key itself; ``spend_limit: 0`` turns the limit off."""
    from tools.mcp_tool_costs import CONFIG_KEYS
    try:
        from hermes_cli.plugins import discover_plugins, get_plugin_manager
        discover_plugins()
        portable = get_plugin_manager().get_portable_mcp_servers()
        for name, cfg in _filter_suspicious_mcp_servers(portable).items():
            if name in rejected:  # the user's own entry for this name has a malformed budget: stay off
                logger.warning("Portable MCP server '%s' skipped: the native entry of that name was rejected", name)
            elif name in safe_servers:
                native = safe_servers[name]
                carried = {k: cfg[k] for k in CONFIG_KEYS if k in cfg and k not in native}
                if carried:
                    safe_servers[name] = {**native, **carried}
                    logger.warning("Portable MCP server '%s' conflicts with native config; using the native entry "
                                   "with the package's declared %s", name, ", ".join(carried))
                else:
                    logger.warning("Portable MCP server '%s' conflicts with native config; skipping", name)
            else:
                safe_servers[name] = dict(cfg)
    except Exception:
        logger.debug("Failed to load portable MCP servers", exc_info=True)


def _load_mcp_config() -> dict[str, dict]:
    """``mcp_servers`` from config.yaml as ``{name: config}`` (empty on error / safe mode), ``${VAR}`` interpolated."""
    from agent.safe_worker_policy import safe_worker_enabled
    if safe_worker_enabled():
        return {}
    try:
        from hermes_cli.config import load_config
        from utils import env_var_enabled as _env_enabled
        if _env_enabled("HERMES_SAFE_MODE"):
            return {}
        servers = load_config().get("mcp_servers")
        try:  # ensure .env vars are available for interpolation
            from hermes_cli.env_loader import load_hermes_dotenv
            load_hermes_dotenv()
        except Exception:
            pass
        safe_servers: dict[str, dict] = {}
        servers = servers if isinstance(servers, dict) else {}
        kept = _filter_suspicious_mcp_servers(servers)
        for name, cfg in kept.items():
            interpolated = _interpolate_env_vars(cfg)
            if isinstance(interpolated, dict):
                _warn_hidden_whitespace(name, interpolated)
                safe_servers[name] = interpolated
        # A same-name package must not revive a server whose native budget was rejected (a mistyped
        # tighter limit would fall back to the package's looser one while the server keeps spending).
        rejected = frozenset(n for n in servers if n not in kept and _malformed_cost_table(n, servers[n]) is not None)
        _portable_mcp_servers(safe_servers, rejected)
        return safe_servers
    except Exception as exc:
        logger.debug("Failed to load MCP config: %s", exc)
        return {}