hermes-agent/hermes_cli/config.py

"""Configuration management for Hermes Agent: config.yaml / .env loading, saving,
validation, migration, and the ``hermes config`` command."""

# Stale-module bridge — must run before ANY import below can bind a root-level symbol.
# A pre-handoff updater purges only package prefixes after the pull, so a root module
# (``utils``) stays cached from the OLD tree; the first fresh consumer of its new symbols
# dies with ImportError before any later heal point is reached. See hermes_cli.stale_modules.
from hermes_cli.stale_modules import drop_stale_root_modules

drop_stale_root_modules()

import copy
import difflib
import json
import logging
import os
import platform
import re
import shlex
import shutil
import stat
import subprocess
import sys
import tempfile
import threading
import time
import unicodedata
from contextlib import suppress
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Dict, Any, Optional, List, Tuple, Set

import hermes_yaml as yaml

from hermes_cli.cli_output import line_input
from hermes_cli.colors import Colors, color
from hermes_cli import managed_scope
from hermes_cli.default_soul import DEFAULT_SOUL_MD, is_legacy_template_soul
from hermes_cli.secret_prompt import masked_secret_prompt
# Managed-mode, container and HERMES_UID/GID policy live in hermes_constants (import-safe);
# re-exported here so existing callers/patch targets keep working.
from hermes_constants import (  # noqa: F401
    _IGNORED_MANAGED_VALUES, _LEGACY_MANAGED_SYSTEM, _MANAGED_FALSE_VALUES, _MANAGED_TRUE_VALUES,
    _chown_to_hermes_uid, _container_or_chmod_skipped, _resolve_hermes_uid_gid,
    apply_secure_dir_policy, get_managed_system)
# Re-export from hermes_constants — canonical definition lives there.
from hermes_constants import get_hermes_home, get_process_hermes_home  # noqa: F401
from utils import atomic_replace, fast_safe_load, file_signature
from hermes_cli.config_read_errors import (
    _CONFIG_PARSE_FAILURES, _FIX_PERMS, _FIX_YAML, FailedConfigRead, _backups_dir_display,
    _refuse_failed_read, _refuse_overwrite, _warn_config_parse_failure, _yaml_error_details,
    _yaml_error_location)

logger = logging.getLogger(__name__)


def is_uv_tool_install() -> bool:
    # Shim to stop the old updater doing work until relaunch, not select uv tool.
    return False


def is_unsupported_install_method(method: str) -> bool:
    # Shim to stop the old updater doing work until relaunch. no legacy detection.
    return False


def format_unsupported_install_warning(method: str) -> str:
    # Shim to stop the old updater doing work until relaunch. no obsolete advice.
    return ""


class InvalidUserConfigError(RuntimeError):
    """Raised when a run that cannot repair config finds invalid user YAML."""


_IS_WINDOWS = platform.system() == "Windows"
_ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")

# Env var names that influence how the next subprocess executes — never writable through
# ``save_env_value``: dynamic loader (LD_*/DYLD_*: attacker code loads before main()),
# interpreter init (PYTHON*, NODE_*: Hermes restarts through them), PATH (fix tool lookup
# with absolute paths instead), git rewrites (fire on every plugin install/update),
# implicitly-invoked commands (BROWSER/EDITOR/VISUAL/PAGER = RCE on next $EDITOR), SHELL,
# and Hermes runtime-location / security-policy flags (config.yaml is the supported surface).
#
# ``HERMES_*`` overall is NOT blocked — many integration credentials use that prefix
# (HERMES_LANGFUSE_PUBLIC_KEY, HERMES_SPOTIFY_CLIENT_ID, ...). The denylist is name-by-name so
# it cannot break provider setup wizards. Enforced on *write* only: pre-existing/out-of-band
# ``.env`` values keep working; the dashboard's writable surface just cannot escalate.

# Whole families whose every member steers execution or config injection, matched by prefix
# because enumeration cannot cover unbounded names (GIT_CONFIG_KEY_17 / GIT_CONFIG_VALUE_17).
_ENV_VAR_NAME_DENY_PREFIXES: tuple[str, ...] = (
    "LD_", "DYLD_",
    # PARAMETERS/COUNT/KEY_*/VALUE_* inject config pairs; GLOBAL/SYSTEM/NOSYSTEM redirect the
    # config sources _subprocess_compat already nulls for the same reason.
    "GIT_CONFIG_",
)

_ENV_VAR_NAME_DENYLIST: frozenset[str] = frozenset({
    # Loader / linker (the LD_/DYLD_ prefixes cover the family; kept name-by-name for clarity)
    "LD_PRELOAD", "LD_LIBRARY_PATH", "LD_AUDIT", "LD_DEBUG",
    "DYLD_INSERT_LIBRARIES", "DYLD_LIBRARY_PATH", "DYLD_FRAMEWORK_PATH",
    "DYLD_FALLBACK_LIBRARY_PATH", "DYLD_FALLBACK_FRAMEWORK_PATH",
    # Python / Node — init-time injection beyond the loader paths
    "PYTHONPATH", "PYTHONHOME", "PYTHONSTARTUP", "PYTHONUSERBASE",
    "PYTHONEXECUTABLE", "PYTHONNOUSERSITE", "PYTHONBREAKPOINT", "PYTHONCASEOK",
    "NODE_OPTIONS", "NODE_PATH",
    # Other interpreter / toolchain injection (same class as PYTHONPATH / NODE_OPTIONS)
    "PERL5OPT", "PERL5LIB", "PERLLIB", "RUBYOPT", "RUBYLIB", "CLASSPATH",
    "JAVA_TOOL_OPTIONS", "_JAVA_OPTIONS", "JDK_JAVA_OPTIONS",
    "GOFLAGS", "RUSTFLAGS",
    # General / git — executed helpers, repo/config redirection, and template hooks
    "PATH", "SHELL", "BROWSER", "EDITOR", "VISUAL", "PAGER", "MANPAGER",
    "GIT_SSH_COMMAND", "GIT_EXEC_PATH", "GIT_SHELL",
    "GIT_SSH", "GIT_ASKPASS", "SSH_ASKPASS", "SUDO_ASKPASS",
    "GIT_EDITOR", "GIT_SEQUENCE_EDITOR", "GIT_PAGER", "GIT_EXTERNAL_DIFF",
    "GIT_PROXY_COMMAND", "GIT_TEMPLATE_DIR", "GIT_DIR",
    # Shell init files / interactive hooks — sourced before or during execution
    "BASH_ENV", "ENV", "ZDOTDIR", "PROMPT_COMMAND", "VIMINIT", "EXINIT",
    # Hermes runtime location
    "HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV",
    "HERMES_CONFIG_PATH", "HERMES_ENV_PATH",
    # MCP catalog trust root; package-manager wrappers may still set it in the process env.
    "HERMES_OPTIONAL_MCPS",
    # Local ACP subprocess selection (executable/argv authority).
    "HERMES_COPILOT_ACP_COMMAND", "HERMES_COPILOT_ACP_ARGS",
    # Security policy / approval-routing context — set via their dedicated controls only.
    "HERMES_YOLO_MODE", "HERMES_ACCEPT_HOOKS", "HERMES_REDACT_SECRETS",
    "HERMES_INTERACTIVE", "HERMES_EXEC_ASK", "HERMES_GATEWAY_SESSION",
    "HERMES_CRON_SESSION", "HERMES_SINGLE_QUERY_SESSION",
    "HERMES_SESSION_KEY", "HERMES_SESSION_PLATFORM"})


def _env_var_policy_name(key: str, *, is_windows: Optional[bool] = None) -> str:
    """Name used for env policy comparisons: Windows env names are case-insensitive, POSIX not.
    The override keeps both semantics testable on any host."""
    windows = _IS_WINDOWS if is_windows is None else is_windows
    return key.upper() if windows else key


def validate_env_var_name_for_write(key: str) -> None:
    """Validate an env name before a generic persistence write (exposed for batch callers)."""
    if not _ENV_VAR_NAME_RE.match(key):
        raise ValueError(f"Invalid environment variable name: {key!r}")
    policy_name = _env_var_policy_name(key)
    if policy_name in _ENV_VAR_NAME_DENYLIST or policy_name.startswith(_ENV_VAR_NAME_DENY_PREFIXES):
        raise ValueError(
            f"Environment variable {key!r} is on the writer denylist. "
            "Names that influence subprocess execution (LD_PRELOAD, PYTHONPATH, PATH, EDITOR, ...) "
            "or Hermes runtime location and security policy (HERMES_HOME, HERMES_YOLO_MODE, ...) "
            "cannot be persisted via the env writer. If you really need this, edit ~/.hermes/.env "
            "directly.")


# Serializes all config read/write paths and guards the module-level caches below. libyaml's
# C extension is not thread-safe for concurrent safe_load() on one file, and tool threads
# (approval, browser, setup flows) load/save config concurrently during long agent runs.
# RLock because callers hold it across a read-modify-write and then call save_config(), which
# acquires it again (hermes_cli/plugins.py: `with ..., config_mod._CONFIG_LOCK:` then
# read_user_config_raw() + save_config()). save_config itself no longer re-enters via
# read_raw_config; it takes its raw mapping from require_readable_config_before_write.
_CONFIG_LOCK = threading.RLock()
# path -> last successfully loaded (expanded) config; served after a parse failure so a
# mid-edit broken YAML never silently drops user overrides (e.g. approvals.deny rules).
_LAST_EXPANDED_CONFIG_BY_PATH: Dict[str, Any] = {}
# path -> (user_mtime_ns, user_size, managed_mtime_ns, managed_size, merged, env_ref_snapshot).
# load_config() returns a deepcopy of the cached value while the signature matches (skips
# safe_load + merge + normalize + expand, ~13 ms). Writers use atomic_config_write (fresh inode
# -> new mtime_ns) so no explicit invalidation is needed. The managed-file signature is folded
# in so editing the managed-scope config.yaml invalidates, and the env snapshot invalidates
# when a referenced ${VAR} changes value (late .env load, in-process rotation).
# (path, mtime_ns, size) -> cached expanded config dict. load_config() returns a deepcopy of the cached
# value when the file hasn't changed since the last load, skipping yaml.safe_load + _deep_merge +
# _normalize_* + _expand_env_vars (~13 ms/call). save_config() + migrate_config() write via
# atomic_config_write which produces a fresh inode, so stat() sees a new signature and the next load
# repopulates automatically — no explicit invalidation hook. See #58514.
_LOAD_CONFIG_CACHE: Dict[str, Tuple[int, ...]] = {}
# path -> (mtime_ns, size, ino, ctime_ns, raw yaml dict) for read_raw_config() (no defaults merged in).
_RAW_CONFIG_CACHE: Dict[str, Tuple[int, ...]] = {}

# Env var names written to .env that aren't in OPTIONAL_ENV_VARS (managed by setup/provider
# flows directly). Also the set reload_env() may remove from os.environ.
_EXTRA_ENV_KEYS = frozenset({
    "OPENAI_API_KEY", "OPENAI_BASE_URL", "ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN",
    "DISCORD_HOME_CHANNEL", "DISCORD_HOME_CHANNEL_NAME",
    "TELEGRAM_HOME_CHANNEL", "TELEGRAM_HOME_CHANNEL_NAME",
    "SLACK_HOME_CHANNEL", "SLACK_HOME_CHANNEL_NAME",
    "SIGNAL_ACCOUNT", "SIGNAL_HTTP_URL", "SIGNAL_ALLOWED_USERS", "SIGNAL_GROUP_ALLOWED_USERS",
    "SIGNAL_HOME_CHANNEL", "SIGNAL_HOME_CHANNEL_NAME", "SMS_HOME_CHANNEL", "SMS_HOME_CHANNEL_NAME",
    "DINGTALK_CLIENT_ID", "DINGTALK_CLIENT_SECRET", "DINGTALK_HOME_CHANNEL", "DINGTALK_HOME_CHANNEL_NAME",
    "FEISHU_APP_ID", "FEISHU_APP_SECRET", "FEISHU_ENCRYPT_KEY", "FEISHU_VERIFICATION_TOKEN",
    "FEISHU_HOME_CHANNEL", "FEISHU_HOME_CHANNEL_NAME", "YUANBAO_HOME_CHANNEL", "YUANBAO_HOME_CHANNEL_NAME",
    "WECOM_BOT_ID", "WECOM_SECRET", "WECOM_CALLBACK_CORP_ID", "WECOM_CALLBACK_CORP_SECRET",
    "WECOM_CALLBACK_AGENT_ID", "WECOM_CALLBACK_TOKEN", "WECOM_CALLBACK_ENCODING_AES_KEY",
    "WECOM_CALLBACK_HOST", "WECOM_CALLBACK_PORT", "WECOM_HOME_CHANNEL", "WECOM_HOME_CHANNEL_NAME",
    "WEIXIN_ACCOUNT_ID", "WEIXIN_TOKEN", "WEIXIN_BASE_URL", "WEIXIN_CDN_BASE_URL",
    "WEIXIN_HOME_CHANNEL", "WEIXIN_HOME_CHANNEL_NAME", "WEIXIN_DM_POLICY", "WEIXIN_GROUP_POLICY",
    "WEIXIN_ALLOWED_USERS", "WEIXIN_GROUP_ALLOWED_USERS", "WEIXIN_ALLOW_ALL_USERS",
    "BLUEBUBBLES_SERVER_URL", "BLUEBUBBLES_PASSWORD", "BLUEBUBBLES_HOME_CHANNEL", "BLUEBUBBLES_HOME_CHANNEL_NAME",
    "QQ_APP_ID", "QQ_CLIENT_SECRET", "QQBOT_HOME_CHANNEL", "QQBOT_HOME_CHANNEL_NAME",
    "QQ_HOME_CHANNEL", "QQ_HOME_CHANNEL_NAME",  # legacy aliases (pre-rename, still read for back-compat)
    "QQ_ALLOWED_USERS", "QQ_GROUP_ALLOWED_USERS", "QQ_ALLOW_ALL_USERS", "QQ_MARKDOWN_SUPPORT",
    "QQ_STT_API_KEY", "QQ_STT_BASE_URL", "QQ_STT_MODEL",
    "IRC_SERVER", "IRC_PORT", "IRC_NICKNAME", "IRC_CHANNEL", "IRC_USE_TLS", "IRC_SERVER_PASSWORD",
    "IRC_NICKSERV_PASSWORD", "TERMINAL_ENV", "TERMINAL_SSH_KEY", "TERMINAL_SSH_PORT",
    # Deprecated (replaced by display.tool_progress) but STILL READ by the gateway as a
    # back-compat fallback. The boolean HERMES_TOOL_PROGRESS variant is unsupported (its only
    # consumer, the v3->4 migration, is below the v12 support floor); doctor flags it as ignored.
    "HERMES_TOOL_PROGRESS_MODE",
    "WHATSAPP_MODE", "WHATSAPP_ENABLED",
    "MATTERMOST_HOME_CHANNEL", "MATTERMOST_HOME_CHANNEL_NAME", "MATTERMOST_REPLY_MODE",
    "MATRIX_PASSWORD", "MATRIX_ENCRYPTION", "MATRIX_DEVICE_ID", "MATRIX_HOME_ROOM",
    "MATRIX_REQUIRE_MENTION", "MATRIX_FREE_RESPONSE_ROOMS", "MATRIX_AUTO_THREAD", "MATRIX_DM_AUTO_THREAD",
    "MATRIX_RECOVERY_KEY",
    # Langfuse observability plugin tuning keys + standard SDK vars (activation is via
    # plugins.enabled; credentials gate the plugin at runtime).
    "HERMES_LANGFUSE_ENV", "HERMES_LANGFUSE_RELEASE", "HERMES_LANGFUSE_SAMPLE_RATE",
    "HERMES_LANGFUSE_MAX_CHARS", "HERMES_LANGFUSE_CAPTURE", "HERMES_LANGFUSE_DEBUG",
    "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_BASE_URL",
    # ACP (Agent Client Protocol) keys — profile-isolable so profiles can use different backends.
    "HERMES_ACP_AUTH_METHOD", "HERMES_ACP_AUTO_APPROVE", "HERMES_COPILOT_ACP_COMMAND",
    "HERMES_COPILOT_ACP_ARGS", "COPILOT_CLI_PATH", "COPILOT_ACP_BASE_URL"})


# ---- Managed mode (NixOS declarative config) ----

_NIX_MANAGED_SYSTEMS = {"nixos", "home-manager"}
# Nix store root; identifies `nix run` / `nix profile install` installs (which don't set
# HERMES_MANAGED). Module-level so tests can patch it without touching /nix/store.
_NIX_STORE = Path("/nix/store")


def is_managed() -> bool:
    """Check if Hermes is running in package-manager-managed mode."""
    return get_managed_system() is not None


# Nix installs arrive by several routes (nix run, nix profile, system flake, home-manager) and
# the running process cannot tell which, so the text names the routes instead of one command.
_NIX_UPDATE_MSG = (
    "Update Hermes through the Nix source that installed it "
    "(e.g. nix profile upgrade, or update your flake input and rebuild with nixos-rebuild or home-manager switch)"
)


def get_managed_update_command() -> Optional[str]:
    """Return the preferred upgrade command for a managed install."""
    return _NIX_UPDATE_MSG if get_managed_system() in _NIX_MANAGED_SYSTEMS else None


# "apt" is the Termux APT distribution identifier, not a generic Debian/Ubuntu signal; another
# APT distribution needs its own method. "home-manager" is listed because the managed marker can
# return it and a stamp must name every method this function returns.
_SUPPORTED_INSTALL_METHODS = frozenset({"apt", "docker", "nix", "nixos", "home-manager", "git", "unknown"})


def _install_method_stamp(path: Path) -> Optional[str]:
    try:
        method = path.read_text(encoding="utf-8-sig").strip().lower()
    except OSError:
        return None
    return method if method in _SUPPORTED_INSTALL_METHODS else None


def detect_install_method(project_root: Optional[Path] = None) -> str:
    """Detect how Hermes was installed: apt/docker/nix/nixos/home-manager/git/unknown.
    Order: code-scoped ``<install tree>/.install_method`` stamp (authoritative) -> legacy
    ``$HERMES_HOME/.install_method`` -> managed marker -> /nix/store path -> .git dir -> unknown.
    The stamp lives next to the code because HERMES_HOME is shared data: a container and a host
    install can bind-mount the same home, so a home-scoped ``docker`` stamp would make the host
    ``hermes update`` refuse to run. A legacy ``docker`` value is therefore ignored unless we are
    really inside a container, and being in a container alone never implies 'docker'.

    Source installers clone a git checkout and publish ``install-stamp.json``;
    the ``.git`` fallback identifies it as a source install. Older installations
    may carry ``.install_method``, which remains authoritative for compatibility.
    The published image bakes a ``docker`` marker into ``/opt/hermes``. A manual
    clone in a container still resolves via ``.git``, not container presence alone.
    See issue #34397.
    """
    # The stamp is a property of the running code tree (parent of hermes_cli/), NOT of $HERMES_HOME,
    # so it survives two installs sharing a home.
    root = project_root if project_root is not None else get_project_root()
    method = _install_method_stamp(root / ".install_method")
    if method:
        return method

    method = _install_method_stamp(get_hermes_home() / ".install_method")
    if method and not (method == "docker" and not _running_in_container()):
        return method

    managed = get_managed_system()
    if managed:
        return managed.lower().replace(" ", "-")

    # Code under /nix/store/ is the hallmark of a nix-built install.
    try:
        resolved = root.resolve()
        if resolved != _NIX_STORE and _NIX_STORE in resolved.parents:
            return "nix"
    except OSError:
        pass

    # A .git directory, or a ``gitdir:`` pointer file for worktrees.
    git_path = root / ".git"
    try:
        if git_path.is_dir() or git_path.read_text(encoding="utf-8-sig").strip().startswith("gitdir:"):
            return "git"
    except OSError:
        pass
    return "unknown"


def _running_in_container() -> bool:
    """Import-safe wrapper around ``hermes_constants.is_container``."""
    try:
        from hermes_constants import is_container

        return is_container()
    except Exception:
        return False


def is_nix_install_method(method: str) -> bool:
    """True for every install method Nix owns ("nix", "nixos", "home-manager")."""
    return method == "nix" or method in _NIX_MANAGED_SYSTEMS


_UPDATE_COMMAND_BY_METHOD = {
    "docker": "docker pull nousresearch/hermes-agent:latest",
    "apt": "pkg upgrade hermes-agent",  # "apt" == Termux APT by contract; uses Termux's `pkg`.
}


def recommended_update_command_for_method(method: str) -> str:
    """Return the update command or guidance for a given install method."""
    if is_nix_install_method(method):
        return _NIX_UPDATE_MSG
    return _UPDATE_COMMAND_BY_METHOD.get(method, "hermes update")


def recommended_update_command() -> str:
    """Return the best update command for the current installation.
    Managed state wins over the code-scoped stamp: a managed install can carry a stale stamp
    naming an update path the managed guard refuses."""
    return get_managed_update_command() or recommended_update_command_for_method(
        detect_install_method(get_project_root()))


# Shared by ``cmd_update`` and ``_cmd_update_check`` (hermes_cli/main.py) so the wording never
# forks. The published image excludes ``.git``, so the git update path can never succeed there
# and the generic "reinstall via install.sh" fallback would install a NEW host-side Hermes.
_DOCKER_UPDATE_MESSAGE = """\
✗ ``hermes update`` doesn't apply inside the Docker container.

Hermes Agent runs as a published image (nousresearch/hermes-agent), not a
git checkout — the container has no working tree to pull into.  Update by
pulling a fresh image and restarting your container instead:

  docker pull nousresearch/hermes-agent:latest
  # then restart whatever started the container, e.g.:
  docker compose up -d --force-recreate hermes-agent
  # or, for ad-hoc runs, exit the current container and `docker run` again

Verify the new version after restart:
  docker run --rm nousresearch/hermes-agent:latest --version

Notes:
  • If you pinned a specific tag (e.g. ``:v0.14.0``) the ``:latest`` tag
    won't move your container — pull the newer tag you actually want, or
    switch to ``:latest`` / ``:main`` for rolling updates.  See available
    tags at https://hub.docker.com/r/nousresearch/hermes-agent/tags
  • On a ``-desktop`` tag (the one carrying Bot Screen)?  Keep the suffix:
    the unsuffixed image has no Xvnc/Xfce and no sudo to add them, so
    pulling it stops the bots' screens from starting.
  • Your config and session history live under ``$HERMES_HOME`` (``/opt/data``
    in the container, typically bind-mounted from the host) and persist
    across image upgrades — re-pulling doesn't lose any state.
  • Running a fork?  Build your own image with this repo's ``Dockerfile``
    and replace the ``docker pull`` step with your build/push pipeline."""


def format_docker_update_message() -> str:
    """Return the user-facing message for ``hermes update`` inside Docker."""
    return _DOCKER_UPDATE_MESSAGE


def format_managed_message(action: str = "modify this Hermes installation") -> str:
    """Build a user-facing error for managed installs."""
    managed_system = get_managed_system() or "a package manager"
    return (
        f"Cannot {action}: this Hermes installation is managed by {managed_system}.\n"
        "Use your package manager to upgrade or reinstall Hermes.")


def managed_error(action: str = "modify configuration"):
    """Print user-friendly error for managed mode."""
    print(format_managed_message(action), file=sys.stderr)


def get_container_exec_info() -> Optional[dict]:
    """Read container mode metadata from HERMES_HOME/.container-mode.
    Written by the NixOS activation script when container.enable = true; tells the host CLI to
    exec into the container instead of running locally. None when container mode is off, when
    already inside the container, or when HERMES_DEV=1 is set. Only FileNotFoundError is
    swallowed; other errors (permissions, malformed data) propagate."""
    if os.environ.get("HERMES_DEV") == "1":
        return None

    from hermes_constants import is_container
    if is_container():
        return None

    try:
        info = {}
        with open(get_hermes_home() / ".container-mode", "r", encoding="utf-8-sig") as f:
            for line in f:
                line = line.strip()
                if "=" in line and not line.startswith("#"):
                    key, _, value = line.partition("=")
                    info[key.strip()] = value.strip()
    except FileNotFoundError:
        return None

    return {
        "backend": info.get("backend", "docker"),
        "container_name": info.get("container_name", "hermes-agent"),
        "exec_user": info.get("exec_user", "hermes"),
        "hermes_bin": info.get("hermes_bin", "/data/current-package/bin/hermes")}


# ---- Config paths / HERMES_HOME skeleton ----

def get_config_path() -> Path:
    """Get the main config file path."""
    return get_hermes_home() / "config.yaml"


def require_parseable_user_config(*, ignore_user_config: bool = False) -> None:
    """Reject an existing invalid config before a non-interactive agent run.
    Interactive surfaces keep ``load_config()``'s recovery behavior so the operator can repair
    the file; a one-shot run has no such chance, and defaults there could silently pick a hosted
    provider and spend against ``.env`` credentials. Missing/empty files stay valid first-run
    states; ``--ignore-user-config`` / HERMES_IGNORE_USER_CONFIG=1 remain authoritative."""
    if ignore_user_config or os.environ.get("HERMES_IGNORE_USER_CONFIG") == "1":
        return

    config_path = get_config_path()
    try:
        with open(config_path, encoding="utf-8-sig") as f:
            data = fast_safe_load(f)
    except FileNotFoundError:
        return
    except Exception as exc:
        parse_error = exc
    else:
        if data is None or isinstance(data, dict):
            return
        parse_error = TypeError(f"top-level YAML value must be a mapping, got {type(data).__name__}")

    from hermes_cli.config_backups import backup_config
    backup_path = backup_config(config_path, "corrupt")
    where = _yaml_error_location(parse_error)
    message = (
        f"Hermes stopped because your settings file ({config_path}) has a formatting error"
        f"{f' at {where}' if where else ''}. Fix it with `hermes config edit` and check with "
        "`hermes config check`, or add --ignore-user-config to run once with default settings.")
    if backup_path is not None:
        message += f" A copy of the broken file is at {backup_path}."
    message += f" Details: {_yaml_error_details(parse_error)}"
    logger.error(message)
    raise InvalidUserConfigError(message) from parse_error


def get_env_path() -> Path:
    """Get the .env file path (for API keys)."""
    return get_hermes_home() / ".env"


def get_project_root() -> Path:
    """Get the project installation directory."""
    return Path(__file__).parent.parent.resolve()


def _secure_dir(path):
    """chmod a directory owner-only (0700) and apply HERMES_UID/GID ownership. No-op when managed;
    in a container only an explicit HERMES_HOME_MODE is applied. HERMES_HOME_MODE (e.g. 0701)
    overrides the mode so a web server can traverse HERMES_HOME to a served subdirectory without
    directory listings.

    Also applies ``HERMES_UID``/``HERMES_GID``-based ownership when those env vars are set (#34107 — Docker
    deployments need this so profile subdirs created at runtime by kanban workers don't land as root:root
    and block subsequent uid-mapped workers).

    Delegates to the canonical import-safe primitive ``hermes_constants.apply_secure_dir_policy``
    so callers outside this package (``get_scratch_dir``) share one implementation (#117347).
    """
    return apply_secure_dir_policy(path)


def _secure_file(path):
    """chmod a file 0600. Skipped when managed (activation sets 0640 group-readable) or in a
    container (mounts often need broader permissions)."""
    if is_managed() or _container_or_chmod_skipped():
        return
    try:
        if os.path.exists(str(path)):
            os.chmod(path, 0o600)
    except (OSError, NotImplementedError):
        pass


def seed_config_file(config_path: Path, template: Optional[Path] = None) -> bool:
    """Create a missing config.yaml the way the installers do: copy cli-config.yaml.example (the display keys
    there are commented out), else write stripped DEFAULT_CONFIG. Never DEFAULT_CONFIG verbatim -- the gateway
    merges no defaults, so every written display key becomes a global that beats each platform's own default
    (#121230). Shared by ``hermes config edit`` and ``hermes doctor --fix`` so the seeders cannot drift.
    Returns True when the template was copied (the fallback, like save_config, writes get_config_path())."""
    template = template or get_project_root() / "cli-config.yaml.example"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    if template.exists():
        shutil.copy2(template, config_path)
        _secure_file(config_path)
        return True
    save_config(DEFAULT_CONFIG)
    return False


def _ensure_default_soul_md(home: Path) -> None:
    """Seed DEFAULT_SOUL_MD on first run; upgrade a legacy comment-only scaffold in place.
    A SOUL.md the user actually customized is never touched."""
    soul_path = home / "SOUL.md"
    if soul_path.exists():
        try:
            existing = soul_path.read_text(encoding="utf-8-sig")
        except (OSError, UnicodeDecodeError):
            return
        if not is_legacy_template_soul(existing):
            return
    try:
        soul_path.write_text(DEFAULT_SOUL_MD, encoding="utf-8")
    except OSError:
        if not soul_path.is_symlink():
            raise
        # A symlink the seed cannot write through — cyclic (``SOUL.md -> SOUL.md``, ELOOP) or
        # dangling into a missing directory (ENOENT) — can never hold an identity file, and the
        # OSError became HomeInitializationError on EVERY boot (launchd exit-75 relaunch storm,
        # #114592). Seed the default IN PLACE OF the link, never through it; mkstemp + replace
        # keeps concurrent gateway boots off one shared path. A working link is never reached
        # here: the write above succeeds through it.
        fd, tmp_name = tempfile.mkstemp(prefix=".SOUL.md.", suffix=".seed", dir=str(home))
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                handle.write(DEFAULT_SOUL_MD)
            os.replace(tmp_name, soul_path)
        except OSError:
            with suppress(OSError):
                os.unlink(tmp_name)
            raise
    _secure_file(soul_path)


# Home paths whose directory skeleton was created this process. Only successful passes are
# recorded, so a raised managed-mode/missing-profile error keeps re-checking on later loads.
_HERMES_HOME_ENSURED: set = set()
_HERMES_HOME_SUBDIRS = (
    "cron", "sessions", "logs", "logs/curator", "memories",
    "pairing", "hooks", "image_cache", "audio_cache", "skills")


def ensure_hermes_home():
    """Ensure the ~/.hermes directory skeleton exists with secure permissions.
    Memoized per home path: this runs on EVERY ``load_config()`` and the ~14 mkdir/chmod syscalls
    made repeated loads the dominant cost of hot read paths."""
    home = get_hermes_home()
    key = str(home)

    # Named profiles must be created explicitly. Check tombstones BEFORE the memo so a stale
    # empty shell cannot skip the deleted-profile guard.
    from hermes_constants import assert_named_profile_home_live
    assert_named_profile_home_live(home)
    if key in _HERMES_HOME_ENSURED and home.is_dir():
        return
    from hermes_cli.config_home import initialize_home
    initialize_home(home, _HERMES_HOME_SUBDIRS, _HERMES_HOME_ENSURED)


# ---- Config loading/saving ----

from hermes_cli.config_defaults import DEFAULT_CONFIG, OPTIONAL_ENV_VARS  # noqa: E402,F401
from hermes_cli.config_providers import (  # noqa: E402,F401  (re-exported; callers/tests use hermes_cli.config.<name>)
    _API_MODE_ALIASES, _CAMEL_ALIASES, _KNOWN_PROVIDER_KEYS, _PROVIDER_NORMALIZE_WARNED,
    _canonical_api_mode, _coerce_ssl_verify, _custom_provider_entry_to_provider_config,
    _entries_for_route, _normalize_custom_provider_entry, _normalize_provider_models,
    _pick_provider_base_url, _route_model_cfg, _warn_once_per_provider,
    apply_custom_provider_extra_headers_to_client_kwargs,
    apply_custom_provider_tls_to_client_kwargs, coerce_provider_id, find_provider_entry,
    get_compatible_custom_providers, get_custom_provider_api_mode, get_custom_provider_context_length,
    get_custom_provider_extra_headers, get_custom_provider_model_capability,
    get_custom_provider_session_affinity_header,
    get_custom_provider_tls_settings, is_provider_enabled, normalize_extra_headers,
    providers_dict_to_custom_providers, stringify_provider_map)
# Back-compat re-exports — :mod:`hermes_cli.personality` owns personality/overlay semantics.
from hermes_cli.personality import (  # noqa: E402,F401
    NEUTRAL_PERSONALITY_NAMES as _NEUTRAL_PERSONALITY_NAMES,
    prompt_text as _prompt_text,
    render_personality_prompt,
    resolve_ephemeral_system_prompt as resolve_ephemeral_system_prompt_from_config)

# ---- Config Migration System ----

# Env vars introduced per config version; migration only mentions vars new since the user's
# previous version.
ENV_VARS_BY_VERSION: Dict[int, List[str]] = {
    3: ["FIRECRAWL_API_KEY", "BROWSERBASE_API_KEY", "BROWSERBASE_PROJECT_ID", "FAL_KEY"],
    4: ["VOICE_TOOLS_OPENAI_KEY", "ELEVENLABS_API_KEY"],
    5: ["WHATSAPP_ENABLED", "WHATSAPP_MODE", "WHATSAPP_ALLOWED_USERS",
        "SLACK_BOT_TOKEN", "SLACK_APP_TOKEN", "SLACK_ALLOWED_USERS"],
    10: ["TAVILY_API_KEY"],
    11: ["TERMINAL_MODAL_MODE"]}

# Intentionally empty: the LLM provider is required but handled by the setup wizard's provider
# selection step, so no single env var is universally required.
REQUIRED_ENV_VARS = {}


def get_missing_env_vars(required_only: bool = False) -> List[Dict[str, Any]]:
    """Check which environment variables are missing."""
    groups = [(REQUIRED_ENV_VARS, True)]
    if not required_only:
        groups.append((OPTIONAL_ENV_VARS, False))
    return [
        {"name": var_name, **info, "is_required": is_required}
        for table, is_required in groups
        for var_name, info in table.items()
        if not get_env_value(var_name)]


def _split_key_path(key: str) -> list[str]:
    """Split a dotted config-key path, honoring backslash-escaped dots (``a\\.b`` -> ``a.b``).
    Backslashes before any other character are preserved verbatim.

    ``hermes config set`` uses ``.`` as the nesting separator, so a key that itself contains a literal dot
    (e.g. provider names like ``qwen3.5-397b-wafer``) was silently split into bogus nested segments
    (#84064).
    """
    parts: list[str] = []
    current: list[str] = []
    i = 0
    while i < len(key):
        ch = key[i]
        if ch == "\\" and key[i + 1:i + 2] == ".":
            current.append(".")
            i += 2
            continue
        if ch == ".":
            parts.append("".join(current))
            current = []
        else:
            current.append(ch)
        i += 1
    parts.append("".join(current))
    return parts


def _greedy_literal_match(container: dict, parts: list) -> Optional[Tuple[str, int]]:
    """Return ``(literal_key, n_consumed)`` for the longest dotted literal key present in
    *container*, or None. With no multi-segment literal this is the historic plain-split walk.

    Dots in config key names are the norm, not the exception — model IDs (``grok-4.6``, ``glm-5.3``), Matrix
    room IDs (``!room:chat.example.cc``), and versioned provider names all embed dots. Users typing
    ``providers.myprov.models.grok-4.6.context_length`` do not know the escape syntax exists, so when
    navigating an EXISTING mapping we prefer an existing literal key equal to the dot-join of the next N
    path segments (longest match wins) over blindly splitting. See #84064 / #80006 / 91095 / #91607 /
    #99124.
    """
    if not isinstance(container, dict) or not parts:
        return None
    return next(
        ((".".join(parts[:n]), n) for n in range(len(parts), 0, -1) if ".".join(parts[:n]) in container),
        None)


def _phantom_sibling(container: dict, part: str) -> Optional[str]:
    """Existing literal dotted key that creating an intermediate mapping ``part`` would shadow
    (``grok-4`` beside ``grok-4.5``) — the write would produce a phantom sibling the runtime never
    reads, so callers fail loudly instead.

    Called when a write is about to CREATE a new intermediate mapping named ``part``. See #84064.
    """
    if not isinstance(container, dict):
        return None
    prefix = part + "."
    return next((k for k in container if isinstance(k, str) and k.startswith(prefix)), None)


def _set_nested(config, dotted_key: str, value):
    """Set a value at a dotted key path, creating intermediate dicts on demand.
    Numeric segments index lists; the index must already exist (lists are never grown).

    Guards against #17876: before this fix the code unconditionally replaced any non-dict value (including
    lists) with ``{}``, silently destroying list-typed config like ``custom_providers`` whenever a caller
    used an indexed path.
    Dotted key names (#84064 family): when navigating an existing mapping, an existing literal key equal to
    the dot-join of the next N segments is preferred over blind splitting (see ``_greedy_literal_match``),
    so ``models.grok-4.6.supports_vision`` lands on the real ``grok-4.6`` entry. And when a write WOULD
    create a new intermediate mapping that shadows an existing dotted sibling (``grok-4`` beside
    ``grok-4.5``), it raises ``ValueError`` instead of silently writing a phantom the runtime never reads.
    """
    parts = _split_key_path(dotted_key)
    current = config
    i = 0
    while i < len(parts):
        remaining = parts[i:]
        at_leaf = len(remaining) == 1
        if isinstance(current, list):
            part = remaining[0]
            if at_leaf:
                current[int(part)] = value
                return
            try:
                current = current[int(part)]
            except (TypeError, ValueError):
                raise TypeError(
                    f"Cannot navigate into list at key {dotted_key!r}: "
                    f"segment {part!r} is not a numeric index")
            i += 1
        elif isinstance(current, dict):
            match = _greedy_literal_match(current, remaining)
            if match is not None:
                key, consumed = match
                if i + consumed == len(parts):
                    current[key] = value
                    return
                # Preserve dicts and lists; replace scalar with a fresh dict.
                if not isinstance(current.get(key), (dict, list)):
                    current[key] = {}
                current = current[key]
                i += consumed
                continue
            part = remaining[0]
            if at_leaf:
                current[part] = value
                return
            shadowed = _phantom_sibling(current, part)
            if shadowed is not None:
                escaped = shadowed.replace(".", "\\.")
                raise ValueError(
                    f"Refusing to create nested key {part!r} in {dotted_key!r}: the mapping "
                    f"already contains a literal key {shadowed!r} that contains a dot. If you "
                    f"meant that key, escape its dots with a backslash (e.g. {escaped}).")
            current = current.setdefault(part, {})
            i += 1
        else:
            raise TypeError(f"Cannot navigate into {type(current).__name__} at key {dotted_key!r}")


def clear_model_endpoint_credentials(
    model_cfg: Dict[str, Any], *, clear_api_key: bool = True, clear_api_mode: bool = True,
    clear_base_url: bool = False) -> Dict[str, Any]:
    """Remove stale inline endpoint credentials from a model config.
    ``model.api_key`` is valid only for explicit custom endpoints; built-in providers resolve
    credentials from env/auth.json/the pool. Leftovers keep secrets in config.yaml and can
    contaminate later custom resolution paths."""
    if not isinstance(model_cfg, dict):
        return model_cfg
    if clear_api_key:
        model_cfg.pop("api_key", None)
        model_cfg.pop("api", None)
        # key_env is a first-class credential POINTER (runtime_provider and
        # auxiliary_client resolve it), written by custom-endpoint activation.
        # Leaving it behind on a provider switch routes the NEW provider's
        # requests to the OLD endpoint's env var — same staleness class as an
        # inline api_key, so it clears under the same flag.
        model_cfg.pop("key_env", None)
        model_cfg.pop("api_key_env", None)
    if clear_api_mode:
        model_cfg.pop("api_mode", None)
    if clear_base_url:
        model_cfg.pop("base_url", None)
    return model_cfg


_MISSING = object()


def _locate_nested(config, parts: list):
    """Walk *parts* through nested dicts/lists (escape-aware, greedy-literal like ``_set_nested``).
    Returns ``(parents, container, key)`` where ``container[key]`` is the addressed leaf and
    ``parents`` lists the ``(container, key)`` hops above it, or ``None`` when any hop is missing,
    a list index is non-numeric/out of range, or a scalar is hit before the path is consumed."""
    parents = []
    current = config
    i = 0
    while True:
        remaining = parts[i:]
        if isinstance(current, list):
            try:
                key = int(remaining[0])
                current[key]
            except (TypeError, ValueError, IndexError):
                return None
            consumed = 1
        elif isinstance(current, dict):
            match = _greedy_literal_match(current, remaining)
            if match is None:
                return None
            key, consumed = match
        else:
            return None
        i += consumed
        if i == len(parts):
            return parents, current, key
        parents.append((current, key))
        current = current[key]


def _get_nested(config, dotted_key: str):
    """Return a dotted-path value (``_MISSING`` when absent); same navigation as ``_set_nested``
    so ``models.grok-4.6.context_length`` reads the real ``grok-4.6`` entry.

    Mirrors ``_set_nested``'s navigation: honors backslash-escaped dots and prefers an existing literal
    dotted key over blind splitting, so ``config get providers.p.models.grok-4.6.context_length`` reads the
    real ``grok-4.6`` entry instead of reporting the key unset (#84064).
    """
    loc = _locate_nested(config, _split_key_path(dotted_key))
    if loc is None:
        return _MISSING
    _, container, key = loc
    return container[key]


def _unset_nested(config, dotted_key: str) -> bool:
    """Remove a dotted-path value; True if it existed. Empty dict containers left behind are
    dropped, while user-authored empty lists and non-empty sibling branches are preserved.

    Same escape-aware, greedy-literal navigation as ``_set_nested`` / ``_get_nested`` (#84064): unsetting an
    unescaped dotted key removes the real literal entry rather than a phantom sibling.
    """
    loc = _locate_nested(config, _split_key_path(dotted_key))
    if loc is None:
        return False
    parents, current, key = loc
    del current[key]
    # ``parent[part] is current`` for every hop, so each now-empty dict container is dropped.
    for parent, part in reversed(parents):
        if current != {}:
            break
        del parent[part]
        current = parent
    return True


_ENV_CONFIG_KEYS = frozenset({
    'OPENROUTER_API_KEY', 'OPENAI_API_KEY', 'ANTHROPIC_API_KEY', 'VOICE_TOOLS_OPENAI_KEY',
    'EXA_API_KEY', 'PARALLEL_API_KEY', 'FIRECRAWL_API_KEY', 'FIRECRAWL_API_URL',
    'FIRECRAWL_GATEWAY_URL', 'TOOL_GATEWAY_URL', 'CONNECTOR_GATEWAY_URL',
    'TOOL_GATEWAY_DOMAIN', 'TOOL_GATEWAY_SCHEME',
    'TOOL_GATEWAY_USER_TOKEN', 'TAVILY_API_KEY', 'PERPLEXITY_API_KEY', 'API_SERVER_KEY',
    'BROWSERBASE_API_KEY', 'BROWSERBASE_PROJECT_ID', 'BROWSER_USE_API_KEY',
    'FAL_KEY', 'TELEGRAM_BOT_TOKEN', 'DISCORD_BOT_TOKEN',
    'TERMINAL_SSH_HOST', 'TERMINAL_SSH_USER', 'TERMINAL_SSH_KEY',
    'SUDO_PASSWORD', 'SLACK_BOT_TOKEN', 'SLACK_APP_TOKEN',
    'GITHUB_TOKEN', 'HONCHO_API_KEY'})


def _is_env_config_key(key: str) -> bool:
    """Return whether `hermes config set` routes this credential-shaped key to .env through the
    provider credential lifecycle. Non-secret env settings (``*_HOME_CHANNEL``, ``*_ALLOWED_USERS``)
    are ``config_env_routing.is_env_setting_key`` and take the plain ``.env`` path."""
    if "." in key:
        return False
    key_upper = key.upper()
    return (
        key_upper in _ENV_CONFIG_KEYS
        or key_upper.endswith(('_API_KEY', '_TOKEN', '_SECRET'))
        or key_upper.startswith('TERMINAL_SSH'))


def _format_config_get_value(value, *, as_json: bool) -> str:
    """Format a config value for command-line output."""
    if as_json:
        return json.dumps(value, ensure_ascii=False)
    if isinstance(value, bool):
        return "true" if value else "false"
    if value is None:
        return "null"
    if isinstance(value, (dict, list)):
        return yaml.safe_dump(value, sort_keys=False).rstrip()  # config-writer: ok — renders a value for display, never written to disk
    return str(value)


def get_missing_config_fields() -> List[Dict[str, Any]]:
    """Check which config fields are missing or outdated (recursive)."""
    missing = []

    def _check(defaults: dict, current: dict, prefix: str = ""):
        for key, default_value in defaults.items():
            if key.startswith('_'):
                continue
            full_key = key if not prefix else f"{prefix}.{key}"
            if key not in current:
                missing.append({"key": full_key, "default": default_value,
                                "description": f"New config option: {full_key}"})
            elif isinstance(default_value, dict) and isinstance(current.get(key), dict):
                _check(default_value, current[key], full_key)

    _check(DEFAULT_CONFIG, load_config())
    return missing


def get_missing_skill_config_vars() -> List[Dict[str, Any]]:
    """Return skill-declared config vars (``skills.config.<key>``) that are missing or empty."""
    try:
        from agent.skill_utils import discover_all_skill_config_vars, SKILL_CONFIG_PREFIX
    except Exception:
        return []

    try:
        all_vars = discover_all_skill_config_vars()
    except Exception as e:
        # A malformed SKILL.md must never break `hermes update`; this prompting is a nicety.
        logger.debug("discover_all_skill_config_vars failed: %s", e)
        return []
    if not all_vars:
        return []

    config = load_config()
    values = ((var, cfg_get(config, *f"{SKILL_CONFIG_PREFIX}.{var['key']}".split("."))) for var in all_vars)
    return [var for var, v in values if v is None or (isinstance(v, str) and not v.strip())]


def _coerce_config_version(value: Any) -> int:
    """Return a safe integer config version, treating invalid values as legacy."""
    if isinstance(value, bool):
        return 0
    try:
        version = int(value)
    except (TypeError, ValueError):
        return 0
    return max(version, 0)


def _read_config_version_stamp(*, raise_on_parse_error: bool = False) -> Tuple[Optional[int], int]:
    """Single raw read behind ``check_config_version()``: ``(stamp, latest_version)`` where
    *stamp* is ``None`` when config.yaml parsed but carries no ``_config_version`` key (a
    never-stamped current-schema file, not an ancient install — ``migrate_config()`` gives it only
    the legacy-key steps). A missing file, or malformed YAML under a tolerant caller, reads as
    ``latest`` exactly as ``check_config_version()`` always reported it."""
    latest = _coerce_config_version(DEFAULT_CONFIG.get("_config_version", 1)) or 1
    config_path = get_config_path()
    if not config_path.exists():
        return latest, latest

    try:
        with open(config_path, encoding="utf-8-sig") as f:
            config = fast_safe_load(f)
    except Exception as e:
        _warn_config_parse_failure(config_path, e)
        if raise_on_parse_error:
            raise InvalidUserConfigError(
                f"Cannot inspect {config_path}: config.yaml is not valid YAML ({e})"
            ) from e
        return latest, latest

    if config is None:
        config = {}  # empty file / bare document: valid first-run state
    if not isinstance(config, dict):
        # A list/scalar root parses fine but is just as unusable as broken YAML: save_config()
        # would refuse it later, after .env was already rewritten. Strict callers see it up front.
        if raise_on_parse_error:
            raise InvalidUserConfigError(
                f"Cannot inspect {config_path}: config.yaml top-level value must be "
                f"a mapping, got {type(config).__name__}"
            )
        config = {}
    if "_config_version" not in config:
        return None, latest
    return _coerce_config_version(config.get("_config_version")), latest


def check_config_version(*, raise_on_parse_error: bool = False) -> Tuple[int, int]:
    """Return ``(current_version, latest_version)`` from the raw on-disk config.
    Reads the raw file rather than ``load_config()``: the deep-merge would make a file lacking
    ``_config_version`` inherit the latest version, hiding that the schema was never migrated.
    Invalid YAML gets a parse warning, not an automatic schema rewrite. Tolerant runtime status
    callers keep the historical latest/latest fallback for malformed YAML; mutation and explicit
    validation paths set ``raise_on_parse_error`` so a parse failure or a non-mapping root cannot
    be mistaken for an up-to-date config. A file with no version key reads as 0."""
    stamp, latest = _read_config_version_stamp(raise_on_parse_error=raise_on_parse_error)
    return (0 if stamp is None else stamp), latest


# ---- Config structure validation ----

# DEFAULT_CONFIG is the single source of truth for documented roots; the set is derived so new
# defaults are accepted automatically. These optional/legacy roots are valid on disk but
# intentionally absent from DEFAULT_CONFIG (omitted when unused / alternate schema forms).
_EXTRA_KNOWN_ROOT_KEYS = {
    "custom_providers",  # legacy list form; modern equivalent is providers: {}
    "fallback_model",    # optional single dict or chain list; omitted when disabled
    "mcp_servers",       # MCP server definitions written by setup/tools flows
    "image_gen",         # agent/image_gen_registry.py
    "video_gen",         # agent/video_gen_registry.py
    "plugins",           # plugin enable/disable lists (hermes_cli/plugins_cmd.py)
    "smart_model_routing",   # written by the setup wizard
    "platform_toolsets",     # written by the setup wizard
    "known_plugin_toolsets", # hermes_cli/tools_config.py toolset-save flow
    "known_builtin_toolsets",  # ditto — builtin toolsets a platform's checklist has offered
    "tool_gateway_declined_tools",  # per-tool Tool Gateway offer declines
    # Top-level forms read/bridged by gateway/config.py:
    "group_sessions_per_user", "thread_sessions_per_user",
    "stt_echo_transcripts", "reset_triggers", "always_log_local", "filter_silence_narration",
    "multiplex_profiles", "profile_routes", "platforms", "require_mention",
    "unauthorized_dm_behavior", "signal", "allow_all_users",
    "timeouts",          # unified timeout resolution section (agent/deadline.py)
}
_KNOWN_ROOT_KEYS = frozenset(DEFAULT_CONFIG.keys()) | _EXTRA_KNOWN_ROOT_KEYS

# Valid fields inside a custom_providers list entry (key_env is read at runtime by
# runtime_provider.py and auxiliary_client.py).
_VALID_CUSTOM_PROVIDER_FIELDS = {
    "name", "base_url", "api_key", "api_mode", "model", "models",
    "context_length", "rate_limit_delay", "extra_body",
    "ssl_ca_cert", "ssl_verify", "key_env", "catalog_provider"}

# Fields that look like they should be inside custom_providers, not at root
_CUSTOM_PROVIDER_LIKE_FIELDS = {"base_url", "api_key", "rate_limit_delay", "api_mode"}


@dataclass
class ConfigIssue:
    """A detected config structure problem."""
    severity: str  # "error", "warning"
    message: str
    hint: str


def _issue(issues: List["ConfigIssue"], severity: str, message: str, hint: str) -> None:
    issues.append(ConfigIssue(severity, message, hint))


def _require_fields(
    issues: List["ConfigIssue"], entry: Dict[str, Any], label: str,
    fields: Tuple[Tuple[str, str], ...], suffix: str = "") -> None:
    """Append a warning for every falsy ``field`` of *entry* (message: ``<label> is missing '<f>' field``)."""
    for field, hint in fields:
        if not entry.get(field):
            _issue(issues, "warning", f"{label} is missing '{field}' field{suffix}", hint)


_CP_REQUIRED_FIELDS = (
    ("name", "Add a name, e.g.: name: my-provider"),
    ("base_url", "Add the API endpoint URL, e.g.: base_url: https://api.example.com/v1"))
_FB_REQUIRED_FIELDS = (
    ("provider", "Add: provider: openrouter (or another provider)"),
    ("model", "Add: model: <model-name>"))
_FB_SINGLE_REQUIRED_FIELDS = (
    ("provider", "Add: provider: openrouter (or another provider)"),
    ("model", "Add: model: anthropic/claude-sonnet-4 (or another model)"))


def _validate_voice(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
    voice_cfg = config.get("voice")
    if not (isinstance(voice_cfg, dict) and "submit_mode" in voice_cfg):
        return
    submit_mode = voice_cfg.get("submit_mode")
    normalized = submit_mode.strip().lower() if isinstance(submit_mode, str) else None
    if normalized not in {"direct", "draft"}:
        _issue(issues, "error", f"voice.submit_mode must be 'direct' or 'draft', got {submit_mode!r}",
               "Set voice.submit_mode to direct (submit immediately) or draft (edit before sending)")


def _validate_timezone(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
    """``timezone`` must be an IANA name the runtime can load.

    ``hermes_time._get_zoneinfo()`` swallows an invalid name behind a single WARNING in the
    gateway log, then runs the agent clock AND every cron schedule on server-local time.
    Surface it here, where doctor and the startup check both look. Silent when the
    interpreter has no tz database at all (bare Windows without ``tzdata``) — nothing can be
    judged there.
    """
    if "timezone" not in config:
        return
    tz = config.get("timezone")
    hint = ("Use an IANA zone name such as America/New_York or Asia/Tokyo (see "
            "`timedatectl list-timezones`). With an invalid value the agent clock and cron "
            "schedules silently fall back to server-local time. HERMES_TIMEZONE overrides "
            "this key when set.")
    if tz is not None and not isinstance(tz, str):
        _issue(issues, "error", f"timezone must be an IANA zone name string, got {tz!r}", hint)
        return
    if not (isinstance(tz, str) and tz.strip()):
        return
    name = tz.strip()
    try:
        import zoneinfo
        zoneinfo.ZoneInfo("UTC")  # is a tz database available at all?
    except Exception:
        return
    try:
        zoneinfo.ZoneInfo(name)
    except Exception:
        _issue(issues, "error", f"timezone {name!r} is not a valid IANA zone name", hint)


def _validate_entry_list(
    entries: list, label: str, issues: List[ConfigIssue], fields, *, non_dict: Tuple[str, str, str],
) -> None:
    """Validate each list entry: ``non_dict`` = (severity, message-with-{i}-and-{type}, hint) for
    non-dict items; dict items get ``_require_fields`` with *fields*."""
    severity, message, hint = non_dict
    for i, entry in enumerate(entries):
        if not isinstance(entry, dict):
            _issue(issues, severity, message.format(i=i, type=type(entry).__name__), hint)
        else:
            _require_fields(issues, entry, f"{label}[{i}]", fields)


_CP_LIST_HINT = "Change to:\n  custom_providers:\n    - name: my-provider\n      base_url: https://...\n      api_key: ..."


def _validate_custom_providers(cp: Any, issues: List[ConfigIssue]) -> None:
    """custom_providers must be a list of dicts — a dict or a scalar is silently dropped by the runtime."""
    if isinstance(cp, dict):
        _issue(issues, "error",
               "custom_providers is a dict — it must be a YAML list (items prefixed with '-')", _CP_LIST_HINT)
        suspicious = set(cp.keys()) & _CUSTOM_PROVIDER_LIKE_FIELDS
        if suspicious:
            _issue(issues, "warning",
                   f"Root-level keys {sorted(suspicious)} look like custom_providers entry fields",
                   "These should be indented under a '- name: ...' list entry, not at root level")
    elif isinstance(cp, list):
        _validate_entry_list(cp, "custom_providers", issues, _CP_REQUIRED_FIELDS, non_dict=(
            "warning", "custom_providers[{i}] is not a dict (got {type})",
            "Each entry should have at minimum: name, base_url"))
    else:
        # get_compatible_custom_providers() returns [] for any non-list: the legacy entries vanish
        # ("0 endpoints") with nothing naming the cause.
        _issue(issues, "error",
               f"custom_providers is a {type(cp).__name__} — it must be a YAML list (items prefixed with '-'); "
               "legacy custom_providers entries are ignored until it is", _CP_LIST_HINT)


def _validate_fallback_model(fb: Any, issues: List[ConfigIssue]) -> None:
    """fallback_model: single dict OR list of dicts (chain)."""
    if isinstance(fb, list):
        _validate_entry_list(fb, "fallback_model", issues, _FB_REQUIRED_FIELDS, non_dict=(
            "error", "fallback_model[{i}] should be a dict, got {type}", "Each entry needs provider + model"))
    elif not isinstance(fb, dict):
        _issue(issues, "error",
               f"fallback_model should be a dict with 'provider' and 'model', got {type(fb).__name__}",
               "Change to:\n  fallback_model:\n    provider: openrouter\n    model: anthropic/claude-sonnet-4")
    elif fb:
        _require_fields(issues, fb, "fallback_model", _FB_SINGLE_REQUIRED_FIELDS,
                        suffix=" — fallback will be disabled")


def _validate_web_backends(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
    """A stale web backend selection otherwise fails only at the first web_search/web_extract
    call with a generic "no registered provider" error; warn at startup instead."""
    # See #99199.
    web_cfg = config.get("web")
    if not isinstance(web_cfg, dict):
        return
    try:
        from tools.tool_backend_helpers import removed_backend_note
    except Exception:
        return
    seen: set = set()
    for _key in ("backend", "search_backend", "extract_backend"):
        _val = str(web_cfg.get(_key) or "").strip().lower()
        if not _val or _val in seen:
            continue
        seen.add(_val)
        note = removed_backend_note("web", _val)
        if note:
            _issue(issues, "warning",
                   f"web.{_key} is set to '{_val}', but {note} — "
                   "web_search/web_extract will fail until it is changed",
                   "Run 'hermes tools' and pick a different Web Search & Extract provider")


def _container_slots() -> Dict[str, str]:
    """Dotted key -> ``"list"``/``"mapping"`` for every slot the schema fixes to a container:
    ``DEFAULT_CONFIG`` (sections included) plus the known-container table for roots it omits."""
    slots: Dict[str, str] = {}

    def walk(node: Dict[str, Any], prefix: str) -> None:
        for key, value in node.items():
            path = f"{prefix}.{key}" if prefix else key
            if isinstance(value, dict):
                slots[path] = "mapping"
                walk(value, path)
            elif isinstance(value, list):
                slots[path] = "list"

    walk(DEFAULT_CONFIG, "")
    slots.update(_KNOWN_CONTAINER_TYPES)
    return slots


def _validate_quoted_containers(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
    """A container slot holding ONE quoted string (``enabled: '["a","b"]'``) is skipped by every
    isinstance-gated reader while ``config get`` echoes it back, so plugins silently unmount and
    exclusions silently lapse (#83308, #105706). Finding only — the file is never rewritten."""
    for key, kind in _container_slots().items():
        # ``parse_config_string_list`` readers accept the quoted form; nothing is ignored there.
        if key in _SCALAR_AS_ONE_ITEM_LIST_KEYS:
            continue
        value = cfg_get(config, *key.split("."))
        if not isinstance(value, str) or not _looks_structured_value(value):
            continue
        try:
            parsed = yaml.safe_load(value)
        except yaml.YAMLError:
            continue
        if isinstance(parsed, (list, dict)):
            _issue(issues, "warning",
                   f"{key} is the quoted string {value!r} — Hermes expects a YAML {kind} here "
                   "and every reader ignores the string",
                   f"Run: hermes config set {key} {shlex.quote(value)}  (stores a real {kind}), "
                   "or remove the quotes in config.yaml")


def validate_config_structure(config: Optional[Dict[str, Any]] = None) -> List["ConfigIssue"]:
    """Validate config.yaml structure and return detected issues (accepts a pre-loaded dict).
    Catches common YAML mistakes that otherwise surface as confusing runtime errors."""
    if config is None:
        try:
            config = load_config()
        except Exception as exc:
            from hermes_cli.config_home import config_load_issue
            return [config_load_issue(exc)]

    issues: List[ConfigIssue] = []
    _validate_voice(config, issues)
    _validate_timezone(config, issues)
    cp = config.get("custom_providers")
    fb = config.get("fallback_model")
    for value, validator in ((cp, _validate_custom_providers), (fb, _validate_fallback_model)):
        if value is not None:
            validator(value, issues)

    if isinstance(cp, dict) and "fallback_model" not in config and "fallback_model" in (cp or {}):
        _issue(issues, "error", "fallback_model appears inside custom_providers instead of at root level",
               "Move fallback_model to the top level of config.yaml (no indentation)")

    if cp and not config.get("model"):
        _issue(issues, "warning",
               "custom_providers defined but no 'model' section — Hermes won't know which provider to use",
               "Add a model section:\n  model:\n    provider: custom\n    default: your-model-name\n"
               "    base_url: https://...")

    # Only provider-like fields are flagged as misplaced roots. Arbitrary unknown top-level keys
    # are deliberately NOT warned about: top-level scalars are bridged into os.environ so users
    # can feed skills/external apps env-style keys — a closed-world allowlist cannot enumerate those.
    for key in config:
        if not key.startswith("_") and key not in _KNOWN_ROOT_KEYS and key in _CUSTOM_PROVIDER_LIKE_FIELDS:
            _issue(issues, "warning",
                   f"Root-level key '{key}' looks misplaced — should it be under 'model:' or inside a 'custom_providers' entry?",
                   f"Move '{key}' under the appropriate section")

    _validate_web_backends(config, issues)
    _validate_quoted_containers(config, issues)
    return issues


def print_config_warnings(config: Optional[Dict[str, Any]] = None) -> None:
    """Print config structure warnings to stderr at startup; nothing if config is healthy."""
    try:
        issues = validate_config_structure(config)
    except Exception:
        issues = []
    if not issues:
        return

    lines = ["\033[33m⚠ Config issues detected in config.yaml:\033[0m"]
    for ci in issues:
        marker = "\033[31m✗\033[0m" if ci.severity == "error" else "\033[33m⚠\033[0m"
        lines.append(f"  {marker} {ci.message}")
    lines.append("  \033[2mRun 'hermes doctor' for fix suggestions.\033[0m")
    sys.stderr.write("\n".join(lines) + "\n\n")


def warn_deprecated_cwd_env_vars() -> None:
    """Warn if MESSAGING_CWD / TERMINAL_CWD is set in .env (canonical: terminal.cwd in config.yaml).
    Reads the file rather than ``os.environ`` because runtime bridges and session restoration
    legitimately set ``TERMINAL_CWD``."""
    try:
        env_map = load_env()
    except Exception:
        return

    lines: list[str] = []
    for name in ("MESSAGING_CWD", "TERMINAL_CWD"):
        val = str(env_map.get(name) or "").strip()
        if val:
            lines.append(f"  \033[33m⚠\033[0m {name}={val} found in .env — this is deprecated.")
    if lines:
        from hermes_constants import display_hermes_home

        hint_path = display_hermes_home()
        lines.insert(0, "\033[33m⚠ Deprecated .env settings detected:\033[0m")
        lines.append(
            "  \033[2mMove to config.yaml instead:  "
            "terminal:\\n    cwd: /your/project/path\033[0m")
        lines.append(f"  \033[2mThen remove the old entries from {hint_path}/.env\033[0m")
        sys.stderr.write("\n".join(lines) + "\n\n")


def _persist_migration(config: Dict[str, Any]) -> None:
    """Persist a migrated config under THE migration write invariant: a migration may only
    persist values that DIFFER from the schema default, plus explicit removals/renames of user
    data. Every migration step MUST write through here (``save_config`` with default-stripping
    ON, no ``merge_existing``) so the invariant cannot regress one migration at a time."""
    save_config(config)


def _prompt_and_save_env(name: str, info: Dict[str, Any], prompt: str, results: Dict[str, Any]) -> bool:
    """Prompt for one env var (masked when ``info['password']``), save it, record it; False if skipped."""
    value = masked_secret_prompt(prompt) if info.get("password") else line_input(prompt).strip()
    if not value:
        return False
    save_env_value(name, value)
    results["env_added"].append(name)
    print(f"  ✓ Saved {name}")
    return True


def _ask_yes_no(prompt: str) -> bool:
    try:
        answer = input(prompt).strip().lower()
    except (EOFError, KeyboardInterrupt):
        answer = "n"
    return answer in {"y", "yes"}


def migrate_config(interactive: bool = True, quiet: bool = False) -> Dict[str, Any]:
    """Migrate config to latest version, prompting for new required fields."""
    results = {"env_added": [], "config_added": [], "warnings": []}

    # Validate config.yaml before any migration side effect: sanitize_env_file() rewrites .env,
    # which must not happen when the migration will be refused for malformed YAML.
    stamp, latest_ver = _read_config_version_stamp(raise_on_parse_error=True)
    current_ver = 0 if stamp is None else stamp

    try:
        fixes = sanitize_env_file()
        if fixes and not quiet:
            print(f"  ✓ Normalized .env line formatting ({fixes} line(s) changed)")
    except Exception:
        pass  # best-effort; never block migration on sanitize failure

    # Auto-migration support floor (v12): an EXPLICIT on-disk ``_config_version`` below the
    # floor is NOT migrated and NOT rewritten — surface a message and leave the file untouched
    # (deep-merge supplies defaults at read time). A config with NO version key is not an
    # ancient install: it gets only the legacy-key steps and a version stamp.
    # Missing/unparseable files never trip the floor gate.
    # Imported lazily because the steps call back into this module.
    from hermes_cli.config_migrations import (
        SUPPORT_FLOOR_VERSION, run_migrations, support_floor_message)

    has_explicit_version = stamp is not None
    floor_refused = (
        has_explicit_version and current_ver < SUPPORT_FLOOR_VERSION and current_ver < latest_ver)
    if floor_refused:
        msg = support_floor_message()
        results["warnings"].append(msg)
        # stderr so it is visible even on quiet startup paths.
        sys.stderr.write(f"⚠ hermes config: {msg}\n")
        if not quiet:
            print(f"  ⚠ {msg}")
    else:
        run_migrations(current_ver, results, quiet, unversioned=not has_explicit_version)

    _disable_suspicious_mcp_servers(results, quiet)
    _warn_invalid_platform_toolsets(results, quiet)

    if current_ver < latest_ver and not quiet and not floor_refused:
        print(f"Config version: {current_ver} → {latest_ver}")

    missing_env = get_missing_env_vars(required_only=True)
    if missing_env and not quiet:
        print("\n⚠️  Missing required environment variables:")
        for var in missing_env:
            print(f"   • {var['name']}: {var['description']}")
    if interactive and missing_env:
        print("\nLet's configure them now:\n")
        for var in missing_env:
            if var.get("url"):
                print(f"  Get your key at: {var['url']}")
            if not _prompt_and_save_env(var["name"], var, f"  {var['prompt']}: ", results):
                results["warnings"].append(f"Skipped {var['name']} - some features may not work")
            print()

    if interactive and not quiet:
        _offer_new_optional_env_vars(current_ver, latest_ver, results)

    # New default keys are NOT materialised to disk (load_config() deep-merges DEFAULT_CONFIG at
    # read time); this list only feeds the "N new config option(s)" display.
    results["config_added"].extend(field["key"] for field in get_missing_config_fields())

    if current_ver < latest_ver and not floor_refused:
        config = read_raw_config()
        config["_config_version"] = latest_ver
        _persist_migration(config)

    missing_skill_config = get_missing_skill_config_vars()
    if missing_skill_config and interactive and not quiet:
        _offer_skill_config_vars(missing_skill_config, results)

    return results


def _disable_suspicious_mcp_servers(results: Dict[str, Any], quiet: bool) -> None:
    """Post-migration: disable exfiltration-shaped MCP stdio entries (hand-edited or from older
    installs). The stanza is preserved for auditability but marked disabled."""
    config = read_raw_config()
    # Preserve the stanza for auditability but mark it disabled so the next startup will not spawn it.
    # (#45620)
    raw_mcp_servers = config.get("mcp_servers")
    if not isinstance(raw_mcp_servers, dict):
        return
    try:
        from hermes_cli.mcp_security import validate_mcp_server_entry
    except Exception:
        return
    mcp_touched = False
    for server_name, entry in raw_mcp_servers.items():
        issues = validate_mcp_server_entry(server_name, entry) if isinstance(entry, dict) else None
        if not issues:
            continue
        entry["enabled"] = False
        mcp_touched = True
        results["warnings"].append(f"Disabled suspicious MCP server '{server_name}'")
        if not quiet:
            for issue in issues:
                print(f"  ⚠ {issue}")
            print(f"  ⚠ Disabled MCP server '{server_name}' pending review")
    if mcp_touched:
        config["mcp_servers"] = raw_mcp_servers
        _persist_migration(config)


def _warn_invalid_platform_toolsets(results: Dict[str, Any], quiet: bool) -> None:
    """Surface invalid toolset names in platform_toolsets: ``resolve_toolset()`` returns [] for an
    unknown name, silently disabling the affected tools. Best-effort; never blocks migration."""
    try:
        from toolsets import validate_toolset
        from hermes_cli.toolset_validation import validate_platform_toolsets
        from hermes_cli.toolset_scope import toolset_allowed_for_platform

        for w in validate_platform_toolsets(
                read_raw_config().get("platform_toolsets"), validate_toolset, toolset_allowed_for_platform):
            results["warnings"].append(w)
            if not quiet:
                print(f"  ⚠ {w}")
    except Exception as _ts_val_err:
        logger.debug("platform_toolsets validation skipped: %s", _ts_val_err)


def _offer_list(heading: str, items: List[str], question: str) -> bool:
    """Print a bulleted offer list and ask; False (with the "set later" hint) when declined."""
    print(heading)
    for item in items:
        print(f"    • {item}")
    print()
    if not _ask_yes_no(question):
        print("  Set later with: hermes config set <key> <value>")
        return False
    print()
    return True


def _offer_new_optional_env_vars(current_ver: int, latest_ver: int, results: Dict[str, Any]) -> None:
    """Interactively offer env vars that are NEW since the user's previous config version."""
    new_var_names: set = set()
    for ver in range(current_ver + 1, latest_ver + 1):
        new_var_names.update(ENV_VARS_BY_VERSION.get(ver, []))
    new_and_unset = [
        (name, OPTIONAL_ENV_VARS[name])
        for name in sorted(new_var_names)
        if not get_env_value(name) and name in OPTIONAL_ENV_VARS]
    if not new_and_unset or not _offer_list(
        f"\n  {len(new_and_unset)} new optional key(s) in this update:",
        [f"{name} — {info.get('description', '')}" for name, info in new_and_unset],
        "  Configure new keys? [y/N]: "):
        return
    for name, info in new_and_unset:
        print(f"  {info.get('description', name)}")
        if info.get("url"):
            print(f"  Get your key at: {info['url']}")
        _prompt_and_save_env(name, info, f"  {info.get('prompt', name)} (Enter to skip): ", results)
        print()


def _offer_skill_config_vars(missing_skill_config: List[Dict[str, Any]], results: Dict[str, Any]) -> None:
    """Prompt for skill-declared settings that are missing/empty and persist the answers."""
    if not _offer_list(
        f"\n  {len(missing_skill_config)} skill setting(s) not configured:",
        [f"{v['key']} — {v['description']} (from skill: {v.get('skill', 'unknown')})" for v in missing_skill_config],
        "  Configure skill settings? [y/N]: "):
        return
    config = read_raw_config()
    try:
        from agent.skill_utils import SKILL_CONFIG_PREFIX
    except Exception:
        SKILL_CONFIG_PREFIX = "skills.config"
    for var in missing_skill_config:
        default = var.get("default", "")
        default_hint = f" (default: {default})" if default else ""
        value = line_input(f"  {var['prompt']}{default_hint}: ").strip() or str(default or "")
        if value:
            _set_nested(config, f"{SKILL_CONFIG_PREFIX}.{var['key']}", value)
            results["config_added"].append(var["key"])
            print(f"  ✓ Saved {var['key']} = {value}")
        else:
            results["warnings"].append(
                f"Skipped {var['key']} — skill '{var.get('skill', '?')}' may ask for it later")
        print()
    _persist_migration(config)


def _merge_partial_save(raw: dict, override: dict) -> dict:
    """Merge *override* over *raw* for partial ``save_config`` writes.
    Omitted top-level sections are preserved; shared dict sections deep-merge so one nested key
    can change without dropping siblings on disk. Key REMOVALS are not supported here —
    migrations go through ``_persist_migration`` with a full ``read_raw_config()`` dict."""
    result = copy.deepcopy(override)
    for key, value in raw.items():
        if key not in result:
            result[key] = copy.deepcopy(value)
        elif isinstance(result.get(key), dict) and isinstance(value, dict):
            result[key] = _deep_merge(value, result[key])
    return result


def _deep_merge(base: dict, override: dict) -> dict:
    """Recursively merge *override* into *base*: dict-over-dict recurses (so overriding one leaf
    keeps sibling defaults), and ``None`` over a dict section is ignored.

    An empty section key in config.yaml (``terminal:`` with no value) parses as YAML ``None``; treating that
    as an override would replace the entire default dict with ``None`` and crash every downstream consumer
    that expects a mapping (#58277).
    """
    result = base.copy()
    for key, value in override.items():
        over_dict = isinstance(result.get(key), dict)
        if over_dict and isinstance(value, dict):
            result[key] = _deep_merge(result[key], value)
        elif not (over_dict and value is None):
            result[key] = value
    return result


def _strip_dotted_keys(cfg: dict, dotted_keys: set) -> Tuple[dict, set]:
    """Remove dotted leaf keys from *cfg* in place -> ``(cfg, keys_actually_present)``.
    ``save_config`` drops managed-scope leaves this way so a bulk write never persists a user
    value that would lose to the managed layer on the next load."""
    stripped: set = set()
    for dotted in dotted_keys:
        *parents, leaf = dotted.split(".")
        node = cfg_get(cfg, *parents)
        if isinstance(node, dict) and leaf in node:
            del node[leaf]
            stripped.add(dotted)
    return cfg, stripped


_ENV_REF_RE = re.compile(r"\${([^}]+)}")


def _env_ref_lookup(name: str) -> Optional[str]:
    """Resolve the env var behind a ``${VAR}`` / ``${env:VAR}`` ref — plain ``os.environ`` outside
    a profile secret scope (legacy behavior for the default profile).

    Inside a scope (a multiplexed gateway turn, a secondary profile's config load, a cron job) the read goes
    through ``agent.secret_scope.get_secret`` so the ref resolves against *that* profile's ``.env``: under
    multiplexing a miss is a miss, never another profile's ``os.environ`` value (#84079 — every profile
    "had" the default profile's ``${MATRIX_ACCESS_TOKEN}`` and fanned out). Same policy as
    ``gateway.config._getenv`` and ``get_env_value``.
    """
    try:
        from agent.secret_scope import current_secret_scope, get_secret as _get_secret
    except Exception:
        return os.environ.get(name)
    if current_secret_scope() is None:
        return os.environ.get(name)
    return _get_secret(name)


def _env_expand_match(m: re.Match) -> str:
    """Expand one ``${VAR}`` (legacy bare name) or ``${env:VAR}`` (Cursor-style SecretRef).
    Other SecretRef sources (``file:``, ``bitwarden:``, ``vault:``...) are NOT resolved here:
    external backends inject their values into the environment at startup (the ``secrets:``
    block), so a config ref only ever needs the env shape. Unresolved refs stay verbatim so
    callers can detect them."""
    raw = m.group(0)
    inner = m.group(1).strip()
    name = _env_ref_var_name(inner)
    if name is None:
        if not inner.startswith("env:") and _is_non_env_secret_ref(inner):
            logger.warning(
                "Config ref %r uses source %r which is not resolvable in "
                "config.yaml — external secret sources inject env vars at "
                "startup, so reference the variable as ${env:NAME} instead",
                raw, inner.split(":", 1)[0])
        return raw  # non-env source, or empty ``${env:}``
    val = _env_ref_lookup(name)
    if val is not None:
        return val
    if inner.startswith("env:"):
        logger.warning(
            "Config ref %r: %s is not set (check ~/.hermes/.env); "
            "keeping the literal placeholder", raw, name)
    return raw


def _is_non_env_secret_ref(ref: str) -> bool:
    """True for a SecretRef body with a non-``env`` source (``bitwarden:FOO``, ``vault:...``)."""
    return ":" in ref and re.match(r"^[a-z][a-z0-9_-]*:", ref) is not None


def _env_ref_var_name(ref: str) -> Optional[str]:
    """Env-var name a ``${...}`` body reads, or None for a non-env source / empty ``env:``."""
    ref = ref.strip()
    if ref.startswith("env:"):
        return ref[len("env:"):].strip() or None
    if _is_non_env_secret_ref(ref):
        return None
    return ref


def _expand_env_vars(obj):
    """Recursively expand ``${VAR}`` / ``${env:VAR}`` in string values (keys/non-strings untouched)."""
    if isinstance(obj, str):
        return _ENV_REF_RE.sub(_env_expand_match, obj)
    if isinstance(obj, dict):
        return {k: _expand_env_vars(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_expand_env_vars(item) for item in obj]
    return obj


def _env_ref_snapshot(obj, snapshot=None):
    """Map each env-sourced ``${...}`` ref in *obj* to its current value.
    Stored with cached ``load_config()`` results so a cache hit can detect that the expansion was
    made against a different environment (load before ``load_hermes_dotenv()``, in-process
    rotation) — file mtime/size alone cannot see either.

    See #58514.
    """
    if snapshot is None:
        snapshot = {}
    if isinstance(obj, str):
        for raw in _ENV_REF_RE.findall(obj):
            name = _env_ref_var_name(raw)
            if name is not None:
                snapshot[name] = _env_ref_lookup(name)
    elif isinstance(obj, dict):
        for value in obj.values():
            _env_ref_snapshot(value, snapshot)
    elif isinstance(obj, list):
        for item in obj:
            _env_ref_snapshot(item, snapshot)
    return snapshot


def _items_by_unique_name(items):
    """Return a name-indexed dict only when all items have unique string names."""
    if not isinstance(items, list):
        return None
    indexed = {}
    for item in items:
        if not isinstance(item, dict) or not isinstance(item.get("name"), str):
            return None
        name = item["name"]
        if name in indexed:
            return None
        indexed[name] = item
    return indexed


def _preserve_env_ref_templates(current, raw, loaded_expanded=None):
    """Restore raw ``${VAR}`` templates where the value is otherwise unchanged, so persisting a
    loaded (expanded) config never writes the plaintext secret back to ``config.yaml``."""
    if isinstance(current, str) and isinstance(raw, str) and _ENV_REF_RE.search(raw):
        if current in (raw, loaded_expanded) or _expand_env_vars(raw) == current:
            return raw
        return current

    if isinstance(current, dict) and isinstance(raw, dict):
        return {
            key: _preserve_env_ref_templates(
                value, raw.get(key),
                loaded_expanded.get(key) if isinstance(loaded_expanded, dict) else None)
            for key, value in current.items()}

    if isinstance(current, list) and isinstance(raw, list):
        # Match named objects (e.g. custom_providers) by name so reordering keeps templates;
        # with duplicate names fall back to positional matching rather than shadowing an entry.
        current_by_name = _items_by_unique_name(current)
        raw_by_name = _items_by_unique_name(raw)
        loaded_by_name = _items_by_unique_name(loaded_expanded)
        if current_by_name is not None and raw_by_name is not None:
            return [
                _preserve_env_ref_templates(
                    item, raw_by_name.get(item.get("name")),
                    loaded_by_name.get(item.get("name")) if loaded_by_name is not None else None)
                for item in current]
        return [
            _preserve_env_ref_templates(
                item,
                raw[index] if index < len(raw) else None,
                loaded_expanded[index]
                if isinstance(loaded_expanded, list) and index < len(loaded_expanded)
                else None)
            for index, item in enumerate(current)]

    return current


def _explicit_config_paths(config: Dict[str, Any]) -> Set[Tuple[str, ...]]:
    """Leaf paths explicitly present in a RAW (un-normalized) config, so values injected by
    normalisation are never mistaken for user-set ones. Feeds ``_strip_default_values``."""
    paths: Set[Tuple[str, ...]] = set()

    def _walk(value: Any, path: Tuple[str, ...]) -> None:
        if isinstance(value, dict):
            for key, child in value.items():
                _walk(child, path + (key,))
        elif path:
            paths.add(path)

    _walk(config, ())
    return paths


def _strip_default_values(
    config: Dict[str, Any], defaults: Dict[str, Any] = DEFAULT_CONFIG,
    preserve_keys: Optional[Set[Tuple[str, ...]]] = None) -> Dict[str, Any]:
    """Return *config* without keys whose values match *defaults*.
    Paths in *preserve_keys* (explicitly present in the user's raw config) are always kept even
    when equal to the default. Dicts whose every child is stripped are removed entirely so
    default-only subtrees never bloat ``config.yaml``."""
    preserve_keys = {("_config_version",)} | set(preserve_keys or ())
    # None is a valid authored value, not a signal to remove the node.
    dropped = object()

    def _strip(value: Any, default: Any, path: Tuple[str, ...]) -> Any:
        if path in preserve_keys:
            return copy.deepcopy(value)
        if isinstance(value, dict) and value:
            default_dict = default if isinstance(default, dict) else {}
            stripped = {k: _strip(v, default_dict.get(k), path + (k,)) for k, v in value.items()}
            return {k: v for k, v in stripped.items() if v is not dropped} or dropped
        return dropped if value == default else copy.deepcopy(value)

    stripped = _strip(config, defaults, ())
    return {} if stripped is dropped else stripped


def split_model_config_default(raw_default: Any) -> tuple[str, str]:
    """Canonicalize ``model.default``/``model.model`` -> ``(model, provider)``; a dict value pairs
    the model string with the provider it must be routed through."""
    if isinstance(raw_default, dict):
        provider = str(raw_default.get("provider") or "").strip()
        model = raw_default.get("model") or raw_default.get("default")
        return (str(model or "").strip(), provider)
    return (str(raw_default or "").strip(), "")


def _normalize_root_model_keys(config: Dict[str, Any]) -> Dict[str, Any]:
    """Canonicalize the ``model`` section at the single load/save chokepoint.
    Root-level ``provider``/``base_url``/``context_length`` (older layouts) are moved under
    ``model`` only when the corresponding ``model.*`` key is empty — never overriding. ``api_base``
    (the OpenAI-SDK/LiteLLM name users reach for) is an alias for ``base_url``; the runtime reads
    only ``model.base_url``. A dict-valued ``default``/``model``/``name`` is flattened so no reader
    sees a nested dict, and the id is canonicalized to ``default``.

    Also aliases ``api_base`` → ``base_url`` (issue #8919). ``api_base`` is the intuitive name OpenAI-SDK /
    LiteLLM users reach for, and ``hermes config set`` blindly accepts any dotted key — so
    ``model.api_base`` got written, confirmed, and then silently ignored by the runtime resolver (which
    reads only ``model.base_url``), causing requests to fall back to OpenRouter. We migrate the alias to the
    canonical key (fallback-only — never override an explicit ``base_url``) and drop the alias so it can't
    confuse later loads.
    Finally, canonicalizes the model-id key to ``model.default`` (issue #34500). The runtime resolver and
    ~14 other readers select the chat model via ``model.default``; ``model.model`` was already aliased
    inline at some sites but ``model.name`` was not, so a custom-provider config like ``model: {name: <id>,
    provider: <custom>}`` resolved to an empty model and the API request went out with ``model=`` (HTTP 400
    from OpenAI-compatible backends) — while display paths (``hermes status``/``dump``) read ``name`` and
    *showed* the model, making the failure silent. Normalizing here (the single load/save chokepoint) means
    every reader, present and future, sees a populated ``default`` and the stale alias is migrated out of
    config.yaml on the next save. Precedence: ``default`` > ``model`` > ``name`` (never overrides an
    explicit ``default``, so existing configs are unaffected).
    """
    model_in = config.get("model")
    model_provider = model_in.get("provider") if isinstance(model_in, dict) else None
    needs_model_work = (model_provider is not None and not isinstance(model_provider, str)) or (
        isinstance(model_in, dict) and (
            model_in.get("api_base")
            or model_in.get("model") or model_in.get("name")
            or any(isinstance(model_in.get(k), dict) for k in ("default", "model", "name"))))
    has_root = any(config.get(k) for k in ("provider", "base_url", "context_length", "api_base"))
    if not has_root and not needs_model_work:
        return config

    config = dict(config)
    model = config.get("model")
    model = dict(model) if isinstance(model, dict) else {"default": model} if model else {}
    config["model"] = model

    # Flatten ``{provider: <p>, model: <m>}``. The nested provider wins over the merged default
    # ``"auto"`` (which runtime resolution treats as authoritative) but never over a configured one.
    for _key in ("default", "model", "name"):
        _val = model.get(_key)
        if isinstance(_val, dict):
            _nested_model = _val.get("model") or _val.get("default")
            _nested_provider = str(_val.get("provider") or "").strip()
            model[_key] = str(_nested_model or "").strip()
            if _nested_provider:
                _outer_provider = str(model.get("provider") or "").strip()
                if not _outer_provider or _outer_provider == "auto":
                    model["provider"] = _nested_provider

    for key in ("provider", "base_url", "context_length"):
        root_val = config.get(key)
        if root_val and not model.get(key):
            model[key] = root_val
        config.pop(key, None)

    # Provider identity is a string (#117345): an unquoted YAML scalar (``provider: 2``)
    # loads as int, and downstream readers call ``(provider or "").strip()`` — a gateway
    # turn dies before the agent runs. Normalize at the load/save chokepoint so every
    # reader (and the next save, which rewrites config.yaml) heals the persisted value.
    # Guard on presence: coerce_provider_id(None) is "" — injecting an empty key into
    # provider-less configs would add churn to config.yaml on the next save.
    if model.get("provider") is not None:
        model["provider"] = coerce_provider_id(model.get("provider"))

    for alias_val in (config.get("api_base"), model.get("api_base")):
        if alias_val and not model.get("base_url"):
            model["base_url"] = alias_val
    config.pop("api_base", None)
    model.pop("api_base", None)

    # ``model``/``name`` are last-resort aliases (in that order), then dropped.
    alias = model.get("model") or model.get("name")
    if not model.get("default") and alias:
        model["default"] = alias
    if model.get("default"):
        model.pop("model", None)
        model.pop("name", None)

    return config


def _normalize_max_turns_config(config: Dict[str, Any]) -> Dict[str, Any]:
    """Move legacy root-level ``max_turns`` under ``agent``; the schema default is injected only
    when the user set max_turns somewhere (so save_config can otherwise omit it)."""
    config = dict(config)
    agent_config = dict(config.get("agent") or {})
    if "max_turns" in config and "max_turns" not in agent_config:
        agent_config["max_turns"] = config["max_turns"]
    if agent_config or "agent" in config:  # a sparse save must not grow an `agent: {}` section
        config["agent"] = agent_config
    config.pop("max_turns", None)
    return config


def _canonicalize_config(config: Dict[str, Any]) -> Dict[str, Any]:
    """The load/save normalization pipeline: max_turns relocation, then model-section canon."""
    return _normalize_root_model_keys(_normalize_max_turns_config(config))


# Sentinel for an unlimited turn budget. ``sys.maxsize`` survives the str->int round-trip through
# the HERMES_MAX_ITERATIONS env bridge, works in every ``<``/``>=``/``max - used`` comparison in
# the iteration budget without an "unlimited" special case, and is unreachable in practice.
TURN_LIMIT_UNLIMITED = sys.maxsize

# Spellings that mean "no limit" (compared lowercased, whitespace-stripped).
_UNLIMITED_SPELLINGS = frozenset({
    "none", "null", "unlimited", "infinite", "infinity", "inf", "∞", "-1", "0"})


def resolve_turn_limit(raw: Any, default: int = TURN_LIMIT_UNLIMITED) -> int:
    """Normalize a raw ``agent.max_turns`` value into an int iteration cap (always >= 1)."""
    # bool is a subclass of int; reject explicitly so True/False don't become 1/0.
    if raw is None or isinstance(raw, bool):
        return default
    if isinstance(raw, (int, float)):
        n = int(raw)
    elif isinstance(raw, str):
        s = raw.strip().lower()
        if not s:
            return default
        if s in _UNLIMITED_SPELLINGS:
            return TURN_LIMIT_UNLIMITED
        try:
            n = int(s)
        except ValueError:
            try:
                n = int(float(s))
            except ValueError:
                logger.debug("resolve_turn_limit: unparseable value %r → default %d", raw, default)
                return default
    else:
        # Unknown type (list, dict, …) — don't crash the agent over a bad config.
        logger.debug("resolve_turn_limit: unsupported type %s (%r) → default %d", type(raw).__name__, raw, default)
        return default
    return TURN_LIMIT_UNLIMITED if n <= 0 else n


def cfg_get(cfg: Optional[Dict[str, Any]], *keys: str, default: Any = None) -> Any:
    """Traverse nested dict keys safely, returning ``default`` on any miss.
    Explicit ``None`` values are returned as-is (``dict.get`` semantics: ``default`` only when the
    key is absent). Named ``cfg_get`` to avoid shadowing the ubiquitous ``cfg_path`` local."""
    if not isinstance(cfg, dict):
        return default
    node: Any = cfg
    for key in keys:
        if not isinstance(node, dict) or key not in node:
            return default
        node = node[key]
    return node


def _raw_config_cache_hit(path_key: str, cache_key: Tuple[Any, ...]) -> Optional[Dict[str, Any]]:
    """Pure lookup: the cached raw config for ``path_key`` if its signature equals ``cache_key``,
    else ``None``. Shared by the lock-free fast path and the locked re-check of
    ``_read_raw_config_impl`` so the predicate cannot drift between them."""
    cached = _RAW_CONFIG_CACHE.get(path_key)
    if cached is not None and cached[:len(cache_key)] == cache_key:
        return cached[len(cache_key)]
    return None


def _read_raw_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
    # Lock-free fast path for cache hits — same shape as `_load_config_impl`. `_RAW_CONFIG_CACHE`
    # publishes each entry as ONE `(*sig, data)` tuple replaced wholesale, so a reader sees either
    # the complete old entry or the complete new one; `_CONFIG_LOCK` only serializes the re-parse
    # and the writers (`save_config()` holds it across an atomic YAML write, which used to stall
    # every cached read for the duration). A lost race just falls through to the locked re-check.
    try:
        config_path = get_config_path()
        cache_key = file_signature(config_path.stat())
        hit = _raw_config_cache_hit(str(config_path), cache_key)
        if hit is not None:
            return copy.deepcopy(hit) if want_deepcopy else hit
    except Exception:
        pass

    with _CONFIG_LOCK:
        config_path = get_config_path()
        try:
            cache_key = file_signature(config_path.stat())
        except FileNotFoundError:
            return {}
        except OSError as e:
            return FailedConfigRead(error=e)

        path_key = str(config_path)
        hit = _raw_config_cache_hit(path_key, cache_key)
        if hit is not None:
            return copy.deepcopy(hit) if want_deepcopy else hit

        try:
            with open(config_path, encoding="utf-8-sig") as f:
                data = fast_safe_load(f) or {}
        except Exception as e:
            _warn_config_parse_failure(config_path, e)
            return FailedConfigRead(error=e)

        if not isinstance(data, dict):
            return FailedConfigRead(error=TypeError(f"top-level YAML must be a mapping, got {type(data).__name__}"))
        _CONFIG_PARSE_FAILURES.pop(path_key, None)  # the file reads now (a transient error left the record)
        # The cache stores its own deepcopy. The readonly path returns THAT object (identity
        # invariant: later cache hits return the same dict); the mutable path returns the parse.
        cached_copy = copy.deepcopy(data)
        _RAW_CONFIG_CACHE[path_key] = (*cache_key, cached_copy)
        return data if want_deepcopy else cached_copy


def read_raw_config() -> Dict[str, Any]:
    """Read config.yaml as-is (no defaults merged, no migration); ``{}`` if missing/unparseable.
    Cached on the file signature (mtime_ns, size, ino, ctime_ns); returns a deepcopy since callers mutate before ``save_config()``."""
    return _read_raw_config_impl(want_deepcopy=True)


def read_user_config_raw(config_path: Optional[Path] = None) -> Dict[str, Any]:
    """Read a user ``config.yaml`` EXACTLY as written (no defaults/overlay/expansion, no cache).
    ONLY legal for write-back round-trips and raw-file diagnostics — behavioral reads must use
    load_config()/load_config_readonly()."""
    if config_path is None:
        config_path = get_config_path()
    try:
        with open(config_path, encoding="utf-8-sig") as f:
            data = fast_safe_load(f) or {}
    except FileNotFoundError:
        return {}
    return data if isinstance(data, dict) else {}


def read_raw_config_readonly() -> Dict[str, Any]:
    """``read_raw_config()`` without the per-call deepcopy, for callers that ONLY READ.
    **Mutating the result corrupts the in-process cache for every subsequent caller.** Meant for
    per-turn policy checks that were paying a full config deepcopy 2-3x per agent turn."""
    return _read_raw_config_impl(want_deepcopy=False)


def require_readable_config_before_write(config_path: Optional[Path] = None) -> Dict[str, Any]:
    """Refuse to replace an existing config.yaml that cannot be read or parsed; return the mapping.
    Guards two collapse-to-empty failure modes that would let a read-then-write caller silently
    wipe user overrides: an unreadable file (permissions / broken mount) and an unparseable or
    non-mapping root — bare-``except`` loaders treat both as ``{}``, so a subsequent write would
    replace the recoverable file with only the caller's partial dict. Fails closed."""
    if config_path is None:
        config_path = get_config_path()
    try:
        config_path.stat()
    except FileNotFoundError:
        return {}
    except OSError as exc:
        raise _refuse_overwrite(config_path, "cannot be accessed", exc, _FIX_PERMS) from exc

    try:
        with open(config_path, encoding="utf-8-sig") as f:
            loaded = fast_safe_load(f)
    except OSError as exc:
        raise _refuse_overwrite(config_path, "cannot be read", exc, _FIX_PERMS) from exc
    except Exception as exc:
        _warn_config_parse_failure(config_path, exc, fallback="refuse-write")
        raise _refuse_overwrite(
            config_path, "has a formatting error", exc, _FIX_YAML.format(backups=_backups_dir_display())) from exc
    if loaded is None:
        return {}
    if not isinstance(loaded, dict):
        exc = TypeError(f"top-level YAML must be a mapping, got {type(loaded).__name__}")
        _warn_config_parse_failure(config_path, exc, fallback="refuse-write")
        raise _refuse_overwrite(
            config_path, f"must start with settings names, but its top level is a {type(loaded).__name__}",
            exc, _FIX_YAML.format(backups=_backups_dir_display())) from exc
    return loaded


def atomic_config_write(config_path: Path, data: Dict[str, Any], *, extra_content_on_create: Optional[str] = None) -> None:
    """THE ``config.yaml`` writer: fail-closed (``require_readable_config_before_write``) and
    comment-preserving (ruamel round-trip merge of *data* onto the on-disk document). Every code
    path that persists a config.yaml — ``save_config``, ``config set``, migrations, plugin
    bookkeeping, gateway/TUI RPCs, auth resets — goes through here; a PyYAML dump of a config
    path anywhere else is rejected by ``scripts/check_config_yaml_writers.py`` (#92554)."""
    from utils import atomic_roundtrip_yaml_save

    _refuse_failed_read(config_path, data)
    atomic_roundtrip_yaml_save(config_path, data, extra_content_on_create=extra_content_on_create)


def load_config() -> Dict[str, Any]:
    """Load the merged configuration (DEFAULT_CONFIG + config.yaml + managed scope, env-expanded).
    Cached on the file signature; returns a deepcopy since most call sites mutate the result.
    Read-only hot paths should use ``load_config_readonly()`` to skip the deepcopy."""
    return _load_config_impl(want_deepcopy=True)


def load_config_readonly() -> Dict[str, Any]:
    """``load_config()`` without the defensive deepcopy (~half of the 265us cache-hit cost).
    **Mutating the returned dict (or any nested structure) corrupts the in-process cache for
    every subsequent caller** — only for code paths that never write to the result."""
    return _load_config_impl(want_deepcopy=False)


def _ensure_dict(parent: Dict[str, Any], key: str) -> Dict[str, Any]:
    """Return ``parent[key]`` as a dict, replacing a missing or non-dict value with ``{}``."""
    child = parent.get(key)
    if not isinstance(child, dict):
        child = {}
        parent[key] = child
    return child


def write_platform_config_field(
    platform_key: str, field_key: str, value: Any, *, raw: bool = False) -> None:
    """Persist one scalar field under ``platforms.<platform_key>``.
    ``raw=True`` (CLI setup flows) edits only the user's raw file; dashboard routes use the
    default loaded-config path to keep their profile-scoped ``load_config`` behavior."""
    config = read_raw_config() if raw else load_config()
    platforms = _ensure_dict(config, "platforms")
    _ensure_dict(platforms, platform_key)[field_key] = value
    save_config(config)


# ``terminal.<key>`` -> env var read by tools.terminal_tool. Every key maps to ``TERMINAL_<KEY>``
# except ``backend`` (historically ``TERMINAL_ENV``).
TERMINAL_CONFIG_ENV_MAP = {
    "backend": "TERMINAL_ENV",
    **{
        key: f"TERMINAL_{key.upper()}"
        for key in (
            "modal_mode", "degraded_mode", "cwd", "temp_dir", "timeout", "lifetime_seconds",
            "docker_image", "docker_forward_env", "singularity_image", "modal_image",
            "daytona_image", "vercel_runtime", "ssh_host", "ssh_user", "ssh_port", "ssh_key",
            "container_cpu", "container_memory", "container_disk", "container_persistent",
            "docker_volumes", "docker_env", "docker_mount_cwd_to_workspace", "docker_network",
            "docker_extra_args", "docker_shm_size", "docker_run_as_host_user", "docker_snap_compat",
            "docker_persist_across_processes", "docker_shared_container_key",
            "docker_orphan_reaper", "sandbox_dir", "persistent_shell")}}


def _terminal_env_value(value: Any) -> str:
    return json.dumps(value) if isinstance(value, (list, dict)) else str(value)


def _terminal_config_value_is_bridgeable(key: str, value: Any) -> bool:
    """Return whether a terminal config value owns its mirrored env var."""
    return not (key == "cwd" and str(value or "").strip() in {".", "auto", "cwd"})


def terminal_config_owned_env_vars(terminal_config: Any) -> Set[str]:
    """Return env vars explicitly owned by a raw ``terminal`` config section."""
    if not isinstance(terminal_config, dict):
        return set()
    return {
        env_var
        for key, env_var in TERMINAL_CONFIG_ENV_MAP.items()
        if key in terminal_config
        and _terminal_config_value_is_bridgeable(key, terminal_config[key])}


def terminal_config_env_var_for_key(key: str) -> Optional[str]:
    """Return the env var mirrored by a ``terminal.*`` config key."""
    return TERMINAL_CONFIG_ENV_MAP.get(key[len("terminal."):]) if key.startswith("terminal.") else None


def _is_ssh_remote_tilde_cwd(backend: str, cwd: str) -> bool:
    """Whether the remote SSH shell must expand *cwd* itself: ``~`` expanded on the Hermes host
    would name the host/container home instead of the SSH user's."""
    return (backend or "").strip().lower() == "ssh" and (cwd == "~" or cwd.startswith("~/"))


def apply_terminal_config_to_env(
    *, env: Optional[Dict[str, str]] = None, config: Optional[Dict[str, Any]] = None,
    override: Optional[bool] = None) -> Dict[str, str]:
    """Bridge ``terminal.*`` config into the env vars terminal tools read.
    ``tools.terminal_tool`` is environment-driven because it also runs in child processes (TUI,
    dashboard PTY, gateway workers); this gives those launch paths the same bridge as the CLI
    without importing ``cli.py``. Explicit keys in the user's raw ``terminal`` section override
    matching env values; merged defaults only backfill missing env vars."""
    target = os.environ if env is None else env

    raw_terminal_cfg = read_raw_config().get("terminal")
    file_has_terminal_config = isinstance(raw_terminal_cfg, dict)
    raw_terminal_cfg = raw_terminal_cfg if file_has_terminal_config else {}
    should_override = file_has_terminal_config if override is None else override

    cfg = config if config is not None else load_config_readonly()
    terminal_cfg = cfg.get("terminal", {}) if isinstance(cfg, dict) else {}
    if not isinstance(terminal_cfg, dict):
        return target

    # A caller-supplied config is its own source of explicit keys; otherwise only keys present
    # in raw config.yaml may override existing env values (DEFAULT_CONFIG keys are backfill-only).
    explicit_keys = terminal_cfg.keys() if config is not None else raw_terminal_cfg.keys()
    backend_sources = (terminal_cfg.get("backend"), target.get("TERMINAL_ENV"))
    if not (config is not None or "backend" in raw_terminal_cfg):
        backend_sources = backend_sources[::-1]  # env wins when the file did not set backend
    terminal_backend = str(backend_sources[0] or backend_sources[1] or "")

    for cfg_key, env_var in TERMINAL_CONFIG_ENV_MAP.items():
        if cfg_key not in terminal_cfg:
            continue
        value = terminal_cfg[cfg_key]
        if not _terminal_config_value_is_bridgeable(cfg_key, value):
            continue
        if cfg_key == "cwd":
            raw_cwd = str(value or "").strip()
            if isinstance(value, str) and not _is_ssh_remote_tilde_cwd(terminal_backend, raw_cwd):
                value = os.path.expanduser(value)
        if (should_override and cfg_key in explicit_keys) or env_var not in target:
            target[env_var] = _terminal_env_value(value)
    return target


def _load_config_cache_sig(config_path: Path) -> Tuple[Optional[Tuple[int, int, int, int]], Optional[Tuple[int, ...]]]:
    """Return ``(user_sig, cache_sig)`` for ``_LOAD_CONFIG_CACHE``.
    The managed config file's signature is folded in ((0, 0, 0, 0) = none) so editing it invalidates
    the merged result. ``cache_sig`` is None only when neither file exists (nothing to cache on)."""
    try:
        st = config_path.stat()
        user_sig: Optional[Tuple[int, int, int, int]] = file_signature(st)
    except FileNotFoundError:
        user_sig = None
    managed_dir = managed_scope.get_managed_dir()
    try:
        mst = (managed_dir / "config.yaml").stat() if managed_dir else None
        managed_sig = file_signature(mst) if mst else (0, 0, 0, 0)
    except OSError:
        managed_sig = (0, 0, 0, 0)
    if user_sig is None and managed_sig == (0, 0, 0, 0):
        return None, None
    return user_sig, (*(user_sig or (0, 0, 0, 0)), *managed_sig)


def _last_known_good_fallback(config_path: Path, path_key: str, cache_sig, exc: Exception) -> Optional[Dict[str, Any]]:
    """Warn about a parse failure and return the last-known-good config, or None (-> defaults).
    A parse failure must not silently replace the effective config with defaults — that drops
    EVERY user override, including security-critical ``approvals.deny`` rules, when a gateway
    user mid-edits config.yaml into broken YAML. Keep serving the last good config until fixed."""
    # Falling through to DEFAULT_CONFIG here drops EVERY user override — including security-critical
    # ``approvals.deny`` rules, which are supposed to block commands even under yolo. Within a running
    # process we still have the last successfully loaded config — keep serving it until the file is fixed.
    # See #31188.
    lkg = _LAST_EXPANDED_CONFIG_BY_PATH.get(path_key)
    fallback = "last-known-good"
    if lkg is None:
        # Fresh process (CLI restart, `hermes config get`): nothing loaded yet in this process, so
        # fall back to the newest byte-exact copy the last successful parse left in backups/config/.
        # It holds the raw file (``${VAR}`` templates intact), so it goes through the same
        # canonicalize -> expand -> managed-overlay pipeline as a normal load.
        from hermes_cli.config_backups import load_newest_good_backup
        raw_good = load_newest_good_backup(config_path)
        if raw_good is not None:
            normalized = _canonicalize_config(_deep_merge(copy.deepcopy(DEFAULT_CONFIG), raw_good))
            expanded_good: Dict[str, Any] = _expand_env_vars(normalized)  # type: ignore[assignment]
            lkg, _ = _merge_managed_overlay(expanded_good)
            fallback = "last-known-good-backup"
    _warn_config_parse_failure(
        config_path, exc, fallback=fallback if lkg is not None else "defaults")
    if lkg is None:
        return None
    # save_config() stores the pre-expansion dict (templates preserved); the load path stores the
    # expanded one. Expand defensively — idempotent when already expanded.
    lkg_copy = FailedConfigRead(_expand_env_vars(copy.deepcopy(lkg)), error=exc)
    if cache_sig is not None:
        # Cache under the failed file's signature (empty env snapshot: always valid) so repeated
        # loads don't re-parse the fallback; fixing the file changes the signature and reloads
        # normally, and a read error is re-probed on every hit (_load_config_cache_hit).
        _LOAD_CONFIG_CACHE[path_key] = (*cache_sig, lkg_copy, {})
    return lkg_copy


def _merge_managed_overlay(expanded: Dict[str, Any]) -> Tuple[Dict[str, Any], Any]:
    """Apply the managed-scope overlay; returns ``(merged, managed_config_or_falsy)``.
    Managed wins at the leaf and is applied AFTER user expansion so a user ``${VAR}`` cannot shadow
    a managed literal: managed values expand only against the process environment. This
    deliberately inverts the usual env-over-config precedence for the keys the managed layer pins
    (docs/design/managed-scope.md §4.1)."""
    managed_config = managed_scope.load_managed_config()
    if not managed_config:
        return expanded, managed_config
    # Same canonicalization as the user config BEFORE merging (parity with
    # managed_scope.apply_managed_overlay) so the merged result never exposes a nested dict.
    managed_normalized = _normalize_root_model_keys(managed_config)
    if isinstance(managed_normalized.get("model"), str):
        managed_normalized = dict(managed_normalized)
        managed_normalized["model"] = {"default": managed_normalized["model"]}
    return _deep_merge(expanded, _expand_env_vars(managed_normalized)), managed_config


def _load_config_cache_hit(path_key: str, cache_sig: Any) -> Optional[Dict[str, Any]]:
    """Lookup: the cached expanded config for ``path_key`` if its signature equals
    ``cache_sig`` AND every ``${VAR}`` it was expanded against still has the same value, else
    ``None``. Signatures matching is not enough: a load before load_hermes_dotenv() would otherwise
    pin unexpanded literals (e.g. auxiliary.<task>.api_key) for the process lifetime (#58514).
    Shared by the lock-free fast path and the locked re-check of ``_load_config_impl``."""
    cached = _LOAD_CONFIG_CACHE.get(path_key)
    if cached is None or cache_sig is None or cached[:8] != cache_sig:
        return None
    hit = cached[8]
    if isinstance(hit, FailedConfigRead) and isinstance(hit.read_error, OSError):
        # A read error (EMFILE/EIO/sharing violation) can clear without touching the file's
        # signature: serve the fallback only while the file still cannot be read.
        try:
            with open(path_key, "rb") as f:
                f.read()
            return None
        except OSError:
            return hit
    env_snapshot = cached[9] if len(cached) > 9 else {}
    if all(_env_ref_lookup(k) == v for k, v in env_snapshot.items()):
        return hit
    return None


def _load_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
    # Lock-free fast path for cache hits — same publication contract as `_read_raw_config_impl`
    # above (whole-tuple replace, `_CONFIG_LOCK` only serializes rebuilds and writers). A hit costs
    # ~0.024ms; behind a lock held by `save_config()` the same read measured 10010ms, and on a
    # gateway that stalls every inbound message's hook path. A lost race falls through to the lock.
    try:
        config_path = get_config_path()
        path_key = str(config_path)
        if path_key in _LOAD_CONFIG_CACHE:
            _, fast_sig = _load_config_cache_sig(config_path)
            hit = _load_config_cache_hit(path_key, fast_sig)
            if hit is not None:
                return copy.deepcopy(hit) if want_deepcopy else hit
    except Exception:
        # Any surprise here falls through to the locked path, which is the
        # original fully-defensive implementation.
        pass

    with _CONFIG_LOCK:
        ensure_hermes_home()
        config_path = get_config_path()
        path_key = str(config_path)

        user_sig, cache_sig = _load_config_cache_sig(config_path)

        hit = _load_config_cache_hit(path_key, cache_sig)
        if hit is not None:
            return copy.deepcopy(hit) if want_deepcopy else hit

        config = copy.deepcopy(DEFAULT_CONFIG)

        if user_sig is not None:
            try:
                with open(config_path, encoding="utf-8-sig") as f:
                    user_config = fast_safe_load(f) or {}
                _CONFIG_PARSE_FAILURES.pop(path_key, None)  # the file reads now (a transient error left the record)

                if "max_turns" in user_config:
                    agent_user_config = dict(user_config.get("agent") or {})
                    if agent_user_config.get("max_turns") is None:
                        agent_user_config["max_turns"] = user_config["max_turns"]
                    user_config["agent"] = agent_user_config
                    user_config.pop("max_turns", None)

                config = _deep_merge(config, user_config)
                # A copy of the file that just parsed is what a FRESH process falls back to when the
                # next edit breaks the YAML (see _last_known_good_fallback). backup_config() skips
                # byte-identical repeats and keeps a bounded count, so steady-state loads cost one stat.
                from hermes_cli.config_backups import backup_config
                backup_config(config_path, "good")
            except Exception as e:
                lkg_copy = _last_known_good_fallback(config_path, path_key, cache_sig, e)
                if lkg_copy is not None:
                    return copy.deepcopy(lkg_copy) if want_deepcopy else lkg_copy
                # Defaults stand in for the unreadable file: never the next last-known-good,
                # never saveable, and cached like the LKG path.
                fallback = FailedConfigRead(
                    _merge_managed_overlay(_expand_env_vars(_canonicalize_config(config)))[0], error=e)
                if cache_sig is not None:
                    _LOAD_CONFIG_CACHE[path_key] = (*cache_sig, fallback, {})
                return copy.deepcopy(fallback) if want_deepcopy else fallback

        normalized = _canonicalize_config(config)
        expanded, managed_config = _merge_managed_overlay(_expand_env_vars(normalized))
        _LAST_EXPANDED_CONFIG_BY_PATH[path_key] = copy.deepcopy(expanded)
        if cache_sig is not None:
            # The cache stores its own deepcopy so load_config() callers can mutate freely while
            # load_config_readonly() callers all see the same stable object. The env snapshot
            # records the values this expansion was made against so later loads detect drift.
            cached_copy = copy.deepcopy(expanded)
            env_snapshot = _env_ref_snapshot(normalized)
            if managed_config:
                _env_ref_snapshot(managed_config, env_snapshot)
            _LOAD_CONFIG_CACHE[path_key] = (*cache_sig, cached_copy, env_snapshot)
            # Readonly path returns the same object later calls will see (identity invariant).
            if not want_deepcopy:
                return cached_copy
        else:
            _LOAD_CONFIG_CACHE.pop(path_key, None)
        # First-load result is a fresh dict (not aliased to the cache); safe to return directly.
        return expanded


_SECURITY_COMMENT = """
# ── Security ──────────────────────────────────────────────────────────
# Secret redaction is ON by default — strings that look like API keys,
# tokens, and passwords are masked in tool output, logs, and chat
# responses before the model or user ever sees them. Set redact_secrets
# to false to disable (e.g. when developing the redactor itself).
# tirith pre-exec scanning is enabled by default when the tirith binary
# is available. Configure via security.tirith_* keys or env vars
# (TIRITH_ENABLED, TIRITH_BIN, TIRITH_TIMEOUT, TIRITH_FAIL_OPEN).
#
# security:
#   redact_secrets: true
#   tirith_enabled: true
#   tirith_path: "tirith"
#   tirith_timeout: 5
#   tirith_fail_open: true
"""

_FALLBACK_COMMENT = """
# ── Fallback Model ────────────────────────────────────────────────────
# Automatic provider failover when primary is unavailable.
# Uncomment and configure to enable. Triggers on rate limits (429),
# overload (529), service errors (503), or connection failures.
#
# Supported providers:
#   openrouter   (OPENROUTER_API_KEY)  — routes to any model
#   openai-codex (OAuth — hermes auth) — OpenAI Codex
#   nous         (OAuth — hermes auth) — Nous Portal
#   zai          (ZAI_API_KEY)         — Z.AI / GLM
#   kimi-coding  (KIMI_API_KEY)        — Kimi / Moonshot
#   kimi-coding-cn (KIMI_CN_API_KEY)   — Kimi / Moonshot (China)
#   minimax      (MINIMAX_API_KEY)     — MiniMax
#   minimax-cn   (MINIMAX_CN_API_KEY)  — MiniMax (China)
#   bedrock      (AWS IAM / boto3)     — AWS Bedrock (Converse API)
#
# For custom OpenAI-compatible endpoints, add base_url and key_env.
#
# fallback_model:
#   provider: openrouter
#   model: anthropic/claude-sonnet-4
"""


def _strip_managed_keys_for_save(config: Dict[str, Any]) -> Dict[str, Any]:
    """Drop every leaf the managed layer pins (bulk safety net; single-key ``config set``
    hard-rejects) and tell the user what was not saved."""
    managed_keys = managed_scope.managed_config_keys()
    if not managed_keys:
        return config
    config, _stripped = _strip_dotted_keys(copy.deepcopy(config), managed_keys)
    if _stripped:
        print(
            f"Note: {len(_stripped)} managed setting(s) were not saved "
            f"(managed by your administrator): {', '.join(sorted(_stripped))}", file=sys.stderr)
    return config


def _commented_sections_for_save(normalized: Dict[str, Any]) -> Optional[str]:
    """Commented-out example blocks for features that are off/unconfigured."""
    parts = []
    if (normalized.get("security") or {}).get("redact_secrets") is None:
        parts.append(_SECURITY_COMMENT)
    fb = normalized.get("fallback_model", {})
    fb_entries = fb if isinstance(fb, list) else [fb]
    if not any(isinstance(e, dict) and e.get("provider") and e.get("model") for e in fb_entries):
        parts.append(_FALLBACK_COMMENT)
    return "".join(parts) or None


def save_config(
    config: Dict[str, Any], *, strip_defaults: bool = True,
    preserve_keys: Optional[Set[Tuple[str, ...]]] = None, merge_existing: bool = False):
    """Save configuration to ~/.hermes/config.yaml.
    Schema defaults are not written unless the user explicitly set them (the path exists in the
    raw config before normalisation), so config.yaml is never contaminated with defaults that
    would hide future default changes. ``merge_existing`` deep-merges the on-disk raw config
    under *config* so partial callers cannot drop sections they omitted."""
    with _CONFIG_LOCK:
        if is_managed():
            managed_error("save configuration")
            return

        config_path = get_config_path()
        _refuse_failed_read(config_path, config)
        config = _strip_managed_keys_for_save(config)

        ensure_hermes_home()
        # Explicit user paths come from the RAW dict BEFORE normalisation (which may inject
        # agent.max_turns) so _strip_default_values keeps exactly what the user set. The
        # fail-closed read is the single authority here: ``read_raw_config()`` is cached and
        # swallows transient stat/open errors into ``{}``, and a ``{}`` at this point makes the
        # strip pass drop every user section whose value matches a default (#113301).
        _raw_for_paths = require_readable_config_before_write(config_path)
        if merge_existing and _raw_for_paths:
            config = _merge_partial_save(_raw_for_paths, config)

        current_normalized = _canonicalize_config(config)
        normalized = current_normalized
        if _raw_for_paths:
            normalized = _preserve_env_ref_templates(
                normalized, _canonicalize_config(_raw_for_paths),
                _LAST_EXPANDED_CONFIG_BY_PATH.get(str(config_path)))

        if strip_defaults:
            # ``_strip_default_values`` always preserves ``_config_version`` itself.
            effective_preserve_keys = _explicit_config_paths(_raw_for_paths) | set(preserve_keys or ())
            normalized = _strip_default_values(normalized, DEFAULT_CONFIG, preserve_keys=effective_preserve_keys)

        atomic_config_write(config_path, normalized, extra_content_on_create=_commented_sections_for_save(normalized))
        _secure_file(config_path)
        _RAW_CONFIG_CACHE.pop(str(config_path), None)
        _LAST_EXPANDED_CONFIG_BY_PATH[str(config_path)] = copy.deepcopy(current_normalized)


def load_env() -> Dict[str, str]:
    """Load ~/.hermes/.env as a dict. Memoised inside ``load_env_file`` (``get_env_value()`` runs
    hundreds of times per interactive menu render). Each assignment's value is opaque data for
    boundary discovery."""
    from agent.secret_scope import load_env_file  # the one .env tokenizer; also installs profile scopes

    return load_env_file(get_env_path())


def _parse_env_value(raw_value: str) -> str:
    """Frozen compat surface name (tests/compat/old_updater_surface.json).

    Pre-PM updaters lazy-import ``hermes_cli.config._parse_env_value`` after the
    checkout swap. The tokenizer moved to ``agent.secret_scope._parse_env_value``
    (c849bc383a), so this forwards there — behavior-preserving by construction.
    """
    from agent.secret_scope import _parse_env_value as _parse

    return _parse(raw_value)


def invalidate_env_cache() -> None:
    """Drop the ``.env`` memo so the next ``load_env()`` sees a write even on coarse-mtime filesystems
    (save_env_value / remove_env_value / sanitize_env_file call this)."""
    from agent.secret_scope import invalidate_env_file_cache

    invalidate_env_file_cache()


def _sanitize_env_lines(lines: list) -> list:
    """Normalize .env line endings/whitespace without changing assignment semantics.
    Content after the first ``=`` is opaque value data: a known variable name embedded in a value
    must never be reinterpreted as another assignment, so concatenated lines stay on one line."""
    sanitized: list[str] = []
    for line in lines:
        raw = line.rstrip("\r\n")
        stripped = raw.strip()
        # Blank lines and comments are preserved verbatim.
        sanitized.append((raw if not stripped or stripped.startswith("#") else stripped) + "\n")
    return sanitized


def sanitize_env_file() -> int:
    """Rewrite ~/.hermes/.env with normalized line formatting; returns the number of changed lines."""
    env_path = get_env_path()
    if not env_path.exists():
        return 0
    with open(env_path, encoding="utf-8-sig", errors="replace") as f:
        original_lines = f.readlines()
    sanitized = _sanitize_env_lines(original_lines)
    if sanitized == original_lines:
        return 0
    fixes = abs(len(sanitized) - len(original_lines)) or sum(
        1 for a, b in zip(original_lines, sanitized) if a != b)
    _write_env_lines(env_path, sanitized, preserve_mode=False)
    invalidate_env_cache()
    return fixes


def _read_env_lines(env_path: Path) -> list:
    """Read ``.env`` lines, normalized. Explicit UTF-8 (Windows defaults to cp1252) with BOM
    tolerance (Notepad adds one)."""
    with open(env_path, encoding="utf-8-sig", errors="replace") as f:
        return _sanitize_env_lines(f.readlines())


def _write_env_lines(env_path: Path, lines: list, *, preserve_mode: bool) -> None:
    """Atomically replace ``.env`` (tmp file + fsync + rename).
    ``preserve_mode`` keeps the original file mode (e.g. 0640 for Docker volume mounts) instead of
    letting ``_secure_file`` tighten to 0600; a new file is always secured."""
    original_mode = None
    try:
        original_mode = stat.S_IMODE(env_path.stat().st_mode) if preserve_mode else None
    except OSError:
        pass
    fd, tmp_path = tempfile.mkstemp(dir=str(env_path.parent), suffix=".tmp", prefix=".env_")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.writelines(lines)
            f.flush()
            os.fsync(f.fileno())
        atomic_replace(tmp_path, env_path)
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise
    if original_mode is not None:
        try:
            os.chmod(env_path, original_mode)
        except OSError:
            pass
    else:
        _secure_file(env_path)


def _check_non_ascii_credential(key: str, value: str) -> str:
    """Strip non-ASCII characters from a credential (HTTP header values must be ASCII) and warn.
    Lookalike glyphs typically come from copy-pasting out of a PDF or rich-text editor."""
    if value.isascii():
        return value

    bad_chars = [f"  position {i}: {ch!r} (U+{ord(ch):04X})" for i, ch in enumerate(value) if ord(ch) > 127]
    sanitized = value.encode("ascii", errors="ignore").decode("ascii")

    print(
        f"\n  Warning: {key} contains non-ASCII characters that will break API requests.\n"
        f"  This usually happens when copy-pasting from a PDF, rich-text editor,\n"
        f"  or web page that substitutes lookalike Unicode glyphs for ASCII letters.\n\n"
        + "\n".join(f"  {line}" for line in bad_chars[:5])
        + ("\n  ... and more" if len(bad_chars) > 5 else "")
        + "\n\n  The non-ASCII characters have been stripped automatically.\n"
        "  If authentication fails, re-copy the key from the provider's dashboard.\n",
        file=sys.stderr)
    return sanitized


def _quote_env_value(value: str) -> str:
    """Quote .env values containing characters with special dotenv meaning. Any whitespace
    (including internal runs) is quoted so ``set -a; . file`` word-splitting keeps paths intact."""
    if value == "":
        return value
    if not ("#" in value or '"' in value or "'" in value or any(c.isspace() for c in value)):
        return value
    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def _env_line_defines_key(line: str, key: str, *, is_windows: Optional[bool] = None) -> bool:
    """True when a .env line assigns ``key`` — plain, ``export``-prefixed, or ``KEY = value``.
    Must match exactly the shapes ``load_env()`` parses; otherwise a hand-added line is invisible
    to save (duplicate appended) and remove (line survives -> the value resurrects on next load).

    ``load_env()`` accepts the bash-compatible ``export KEY=value`` form (#6659), so the writers must
    recognise the same shape.
    """
    stripped = line.strip()
    if stripped.startswith("export "):
        stripped = stripped[7:].lstrip()
    assigned_key, separator, _value = stripped.partition("=")
    if not separator:
        return False
    # load_env() strips whitespace around the parsed name, so `KEY = value` IS a live assignment. The
    # writers must match the same shape, or a hand-edited spaced line is invisible to save (duplicate
    # appended) and remove (line survives -> value resurrects on next load). #67488.
    return _env_var_policy_name(
        assigned_key.strip(), is_windows=is_windows
    ) == _env_var_policy_name(key, is_windows=is_windows)


def _publish_env_value(key: str, value: Optional[str]) -> None:
    """Publish a just-persisted ``.env`` change to the live process.
    Under a multiplexed gateway a routed profile's write must not land in the SHARED
    ``os.environ`` where every profile sees it; the installed scope mapping is updated instead so
    same-turn reads see the change. All other callers keep the legacy ``os.environ`` publish.

    ``save_env_value`` / ``remove_env_value`` already target the right file (``get_env_path()`` honors the
    profile-home override), but the in-process mirror historically went straight to ``os.environ``. See
    #77490, #88441.
    """
    try:
        from agent.secret_scope import current_secret_scope, serves_routed_profile

        scope, routed = current_secret_scope(), serves_routed_profile()
    except Exception:
        scope, routed = None, False
    # The launch profile's own body runs under a scope snapshot even single-profile (the TUI /
    # dashboard launch scope), so a same-request read after the write must see it there too; a
    # routed profile's value never reaches the shared process env.
    targets = [scope] if isinstance(scope, dict) else []
    if not routed and (scope is None or isinstance(scope, dict)):
        targets.append(os.environ)
    for target in targets:
        if value is None:
            target.pop(key, None)
        else:
            target[key] = value


def _env_write_blocked(key: str, action: str) -> bool:
    """Shared write-lock check for ``.env`` writers; prints the refusal and returns True when blocked.
    Two distinct locks: ``is_managed()`` (package-manager install) and the managed *scope*
    (administrator-pinned env key — the managed .env wins at load anyway)."""
    if is_managed():
        managed_error(f"{action} {key}")
        return True

    if managed_scope.is_env_managed(key):
        print(
            f"Cannot {action} {key}: it is managed by your administrator ({_managed_source('.env')}) "
            f"and cannot be changed.", file=sys.stderr)
        return True
    return False


def _managed_source(filename: str):
    """``<managed dir>/<filename>`` for refusal messages, or a generic label without a managed dir."""
    managed_dir = managed_scope.get_managed_dir()
    return (managed_dir / filename) if managed_dir else "the managed scope"


def save_env_value(key: str, value: str):
    """Save or update a value in ~/.hermes/.env (also matching ``export KEY=`` lines, so a save
    never appends a second line that a later delete would resurrect)."""
    if _env_write_blocked(key, "set"):
        return
    validate_env_var_name_for_write(key)
    value = value.replace("\n", "").replace("\r", "")
    value = _check_non_ascii_credential(key, value)
    ensure_hermes_home()
    env_path = get_env_path()

    lines = _read_env_lines(env_path) if env_path.exists() else []
    serialized_value = _quote_env_value(value)

    idx = next((i for i, line in enumerate(lines) if _env_line_defines_key(line, key)), None)
    if idx is not None:
        lines[idx] = f"{key}={serialized_value}\n"
    else:
        if lines and not lines[-1].endswith("\n"):
            lines[-1] += "\n"
        lines.append(f"{key}={serialized_value}\n")

    _write_env_lines(env_path, lines, preserve_mode=env_path.exists())
    _publish_env_value(key, value)
    invalidate_env_cache()


def custom_endpoint_key_env(identity: str) -> str:
    """Env var name holding a custom endpoint's API key.
    ``identity`` is the endpoint's own id (Desktop endpoint id, or ``host:port`` for CLI setup),
    so two endpoints on one host get separate slots. The fixed ``HERMES_CUSTOM_`` prefix keeps the
    name POSIX-valid when the slug starts with a digit (``save_env_value`` rejects those)."""
    slug = re.sub(r"[^A-Z0-9]+", "_", str(identity or "").upper()).strip("_")
    return f"HERMES_CUSTOM_{slug}_API_KEY" if slug else "HERMES_CUSTOM_API_KEY"


def remove_env_value(key: str) -> bool:
    """Remove a key from ~/.hermes/.env and os.environ; True if it was found and removed."""
    if _env_write_blocked(key, "remove"):
        return False
    if not _ENV_VAR_NAME_RE.match(key):
        raise ValueError(f"Invalid environment variable name: {key!r}")
    env_path = get_env_path()
    if not env_path.exists():
        _publish_env_value(key, None)
        return False

    lines = _read_env_lines(env_path)
    new_lines = [line for line in lines if not _env_line_defines_key(line, key)]
    found = len(new_lines) < len(lines)
    if found:
        _write_env_lines(env_path, new_lines, preserve_mode=True)
    _publish_env_value(key, None)
    invalidate_env_cache()
    return found


def _write_anthropic_slots(token: str, api_key: str, save_fn=None, *, token_first: bool = True):
    """Write both Anthropic credential slots (one holds the value, the other is cleared)."""
    writer = save_fn or save_env_value
    order = (("ANTHROPIC_TOKEN", token), ("ANTHROPIC_API_KEY", api_key))
    for name, value in order if token_first else reversed(order):
        writer(name, value)


def save_anthropic_oauth_token(value: str, save_fn=None):
    """Persist an Anthropic OAuth/setup token and clear the API-key slot."""
    _write_anthropic_slots(value, "", save_fn)


def use_anthropic_claude_code_credentials(save_fn=None):
    """Use Claude Code's own credential files instead of persisting env tokens."""
    _write_anthropic_slots("", "", save_fn)


def save_anthropic_api_key(value: str, save_fn=None):
    """Persist an Anthropic API key and clear the OAuth/setup-token slot."""
    _write_anthropic_slots("", value, save_fn, token_first=False)


def save_env_value_secure(key: str, value: str) -> Dict[str, Any]:
    """Save via the unified credential lifecycle (also refreshes any config.yaml mirror of the old
    value and lifts a prior env-source suppression)."""
    from hermes_cli.credential_lifecycle import save_provider_env_credential

    # Route through the unified credential lifecycle so a rotation via the secret-capture path also
    # refreshes any config.yaml mirror of the old value and lifts a prior env-source suppression (#62269 fix
    # family).
    save_provider_env_credential(key, value)
    return {"success": True, "stored_as": key, "validated": False}


def reload_env() -> int:
    """Re-read ~/.hermes/.env into os.environ; returns count of vars changed.
    Removes deleted vars only when known to Hermes (OPTIONAL_ENV_VARS and _EXTRA_ENV_KEYS) so
    unrelated environment is never clobbered."""
    env_vars = load_env()
    count = 0
    for key, value in env_vars.items():
        if os.environ.get(key) != value:
            os.environ[key] = value
            count += 1
    for key in (set(OPTIONAL_ENV_VARS) | _EXTRA_ENV_KEYS) - set(env_vars):
        if key in os.environ:
            del os.environ[key]
            count += 1
    return count


def _scoped_environ_get(key: str) -> Optional[str]:
    """Read ``key`` from ``os.environ`` through ``agent.secret_scope.get_secret`` so an active
    profile scope (multiplexed gateway turn) never leaks another profile's raw value. Falls back to
    a plain environ read when the scope module is unavailable; ``UnscopedSecretError`` and scope
    failures propagate -- a failed scoped read must never borrow the ambient env."""
    try:
        from agent.secret_scope import get_secret as _get_secret
    except Exception:
        return os.environ.get(key)
    return _get_secret(key)


def get_env_value(key: str) -> Optional[str]:
    """Get a value from ``os.environ`` (scope-aware) or ``~/.hermes/.env``.

    The ``os.environ`` read routes through ``agent.secret_scope.get_secret`` so that, under an active
    profile scope (multiplexed gateway turn), this is scope-checked rather than leaking another profile's
    raw ``os.environ`` value. ``get_secret`` encodes the whole policy: global vars pass through; scope is
    authoritative under multiplexing (miss -> None, no environ fallthrough); when multiplexing is off it
    behaves exactly like the legacy ``os.environ`` read. Its siblings ``get_env_value_prefer_dotenv`` and
    ``gateway.config._getenv`` already work this way — this was the last scope-blind reader of the trio
    (#67027).
    """
    val = _scoped_environ_get(key)
    return load_env().get(key) if val is None else val


def get_env_value_prefer_dotenv(key: str) -> Optional[str]:
    """Resolve a Hermes-managed credential preferring ``~/.hermes/.env`` over ``os.environ``, so a
    deliberate .env edit beats a stale value inherited from the parent shell."""
    return load_env().get(key) or _scoped_environ_get(key)


# ---- Config display ----

def redact_key(key: str) -> str:
    """Redact an API key for display."""
    from agent.redact import mask_secret
    return mask_secret(key, empty=color("(not set)", Colors.DIM))


# Key names (case-insensitive, exact match) whose VALUE is a credential and must be masked
# before printing any config dict. Exact-match so ``token_count`` / ``secret_santa`` stay visible.
# Bare ``auth`` is deliberately absent: ``mcp_servers.<s>.auth: oauth`` is a documented mode enum.
_SECRET_CONFIG_KEYS = frozenset({
    "api_key", "apikey", "key", "token", "access_token", "refresh_token", "id_token",
    "secret", "client_secret", "password", "passwd", "authorization",
    "private_key", "bearer", "jwt"})
# Env-map shapes (``mcp_servers.<s>.env.FOO_API_KEY``, ``FAL_KEY``, ``AWS_SECRET_ACCESS_KEY``) and
# the suffixes ``_is_env_config_key`` routes to .env. Suffix-only so ``token_count`` stays visible.
_SECRET_CONFIG_KEY_SUFFIXES = ("_api_key", "_token", "_secret", "_password", "_key", "_access_key")
# .env-routed keys are credentials by default; these suffixes name the non-secret exceptions
# (``TERMINAL_SSH_HOST``, ``TOOL_GATEWAY_URL``, ``BROWSERBASE_PROJECT_ID``).
_NON_SECRET_KEY_SUFFIXES = ("_url", "_host", "_user", "_id", "_domain", "_scheme")
_ENV_PLACEHOLDER_RE = re.compile(r"^\$\{[A-Za-z_][A-Za-z0-9_]*\}$")


def _is_secret_config_key(key: str) -> bool:
    """Whether the LAST segment of a config key names a credential value. Header names
    (``mcp_servers.<s>.headers.X-API-Key``) are folded to snake_case before matching."""
    leaf = key.rsplit(".", 1)[-1].lower().replace("-", "_")
    if _is_env_config_key(key):
        return not leaf.endswith(_NON_SECRET_KEY_SUFFIXES)
    return leaf in _SECRET_CONFIG_KEYS or leaf.endswith(_SECRET_CONFIG_KEY_SUFFIXES)


def redact_config_value(value: Any, _depth: int = 0) -> Any:
    """Copy of ``value`` with credential-shaped keys masked. ``print`` bypasses the logging
    redactor and opaque tokens miss the vendor-prefix regexes, so structural masking is required."""
    from agent.redact import mask_secret

    if _depth > 20:  # bound recursion for pathological/cyclic configs
        return value
    if isinstance(value, dict):
        return {
            k: mask_secret(v)
            if isinstance(k, str) and _is_secret_config_key(k) and isinstance(v, str) and v
            and not _ENV_PLACEHOLDER_RE.match(v)
            else redact_config_value(v, _depth + 1)
            for k, v in value.items()}
    if isinstance(value, list):
        return [redact_config_value(v, _depth + 1) for v in value]
    return value


def _section(title: str) -> None:
    print()
    print(color(f"◆ {title}", Colors.CYAN, Colors.BOLD))


def _show_managed_banner() -> None:
    """Surface administrator-pinned settings so the user knows why a config.yaml value may not
    be the effective one."""
    managed_keys = managed_scope.managed_config_keys()
    managed_env = managed_scope.load_managed_env()
    if not managed_keys and not managed_env:
        return
    print()
    print(color(
        f"  ⚷ Some settings are managed by your administrator ({managed_scope.get_managed_dir()}) "
        f"and cannot be changed", Colors.YELLOW, Colors.BOLD))
    for label, keys in (("config", managed_keys), ("env", managed_env)):
        if keys:
            print(color(f"    Managed {label} keys: {', '.join(sorted(keys))}", Colors.YELLOW))


_SHOW_CONFIG_API_KEYS = (
    ("OPENROUTER_API_KEY", "OpenRouter"),
    ("VOICE_TOOLS_OPENAI_KEY", "OpenAI (STT/TTS)"),
    ("EXA_API_KEY", "Exa"),
    ("PARALLEL_API_KEY", "Parallel"),
    ("FIRECRAWL_API_KEY", "Firecrawl"),
    ("TAVILY_API_KEY", "Tavily"),
    ("PERPLEXITY_API_KEY", "Perplexity"),
    ("BROWSERBASE_API_KEY", "Browserbase"),
    ("BROWSER_USE_API_KEY", "Browser Use"),
    ("FAL_KEY", "FAL"))


def _show_model_section(config: Dict[str, Any]) -> None:
    _section("Model")
    print(f"  Model:        {redact_config_value(config.get('model', 'not set'))}")
    cfg_max_turns = config.get('agent', {}).get('max_turns', DEFAULT_CONFIG['agent']['max_turns'])
    print(f"  Max turns:    {cfg_max_turns}")
    # Read the .env FILE directly so a stale HERMES_MAX_ITERATIONS ghost is caught even when the
    # gateway bridge already overrode os.environ.
    try:
        env_ghost = load_env().get("HERMES_MAX_ITERATIONS")
    except Exception:
        env_ghost = None
    if env_ghost is not None and str(env_ghost).strip() != str(cfg_max_turns).strip():
        print(color(f"                ⚠ .env has stale HERMES_MAX_ITERATIONS={env_ghost} "
                    f"(run 'hermes doctor --fix' to remove)", Colors.YELLOW))


def _show_display_section(config: Dict[str, Any]) -> None:
    _section("Display")
    display = config.get('display', {})
    try:
        from hermes_cli.personality import active_personality_name
        active_personality = active_personality_name(config) or 'none'
    except Exception:
        active_personality = display.get('personality') or 'none'
    on_off = lambda flag: 'on' if flag else 'off'  # noqa: E731
    print(f"  Personality:  {active_personality}")
    print(f"  Reasoning:    {on_off(display.get('show_reasoning', True))}")
    print(
        f"  Bell:         complete={on_off(display.get('bell_on_complete', False))}, "
        f"prompt={on_off(display.get('bell_on_prompt', False))}")
    ump = display.get('user_message_preview', {})
    ump = ump if isinstance(ump, dict) else {}
    print(f"  User preview: first {ump.get('first_lines', 2)} line(s), last {ump.get('last_lines', 2)} line(s)")


def _show_terminal_section(config: Dict[str, Any]) -> None:
    _section("Terminal")
    terminal = config.get('terminal', {})
    print(f"  Backend:      {terminal.get('backend', 'local')}")
    print(f"  Working dir:  {terminal.get('cwd', '.')}")
    print(f"  Timeout:      {terminal.get('timeout', 60)}s")

    configured = lambda *names: 'configured' if all(get_env_value(n) for n in names) else '(not set)'  # noqa: E731
    default_img = 'nikolaik/python-nodejs:python3.14-nodejs22'
    backend_lines = {
        'docker': lambda: [f"  Docker image: {terminal.get('docker_image', default_img)}"],
        'singularity': lambda: [f"  Image:        {terminal.get('singularity_image', 'docker://' + default_img)}"],
        'modal': lambda: [
            f"  Modal image:  {terminal.get('modal_image', default_img)}",
            f"  Modal token:  {configured('MODAL_TOKEN_ID')}"],
        'daytona': lambda: [
            f"  Daytona image: {terminal.get('daytona_image', default_img)}",
            f"  API key:      {configured('DAYTONA_API_KEY')}"],
        'vercel_sandbox': lambda: [
            f"  Vercel runtime: {terminal.get('vercel_runtime', 'node24')}",
            f"  Vercel auth:    {'configured' if get_env_value('VERCEL_OIDC_TOKEN') or (get_env_value('VERCEL_TOKEN') and get_env_value('VERCEL_PROJECT_ID') and get_env_value('VERCEL_TEAM_ID')) else '(not set)'}",
        ],
        'ssh': lambda: [
            f"  SSH host:     {get_env_value('TERMINAL_SSH_HOST') or '(not set)'}",
            f"  SSH user:     {get_env_value('TERMINAL_SSH_USER') or '(not set)'}"]}
    for line in backend_lines.get(terminal.get('backend'), list)():
        print(line)


def _show_compression_section(config: Dict[str, Any]) -> None:
    _section("Context Compression")
    compression = config.get('compression', {})
    enabled = compression.get('enabled', True)
    print(f"  Enabled:      {'yes' if enabled else 'no'}")
    if not enabled:
        return
    print(f"  Threshold:    {compression.get('threshold', 0.50) * 100:.0f}%")
    tt = compression.get('threshold_tokens')
    try:
        if tt is not None and int(tt) > 0:
            print(f"  Token cap:    {int(tt):,} tokens (takes lower of ratio vs absolute)")
    except (TypeError, ValueError):
        pass
    print(f"  Target ratio: {compression.get('target_ratio', 0.20) * 100:.0f}% of threshold preserved")
    print(f"  Protect last: {compression.get('protect_last_n', 20)} messages")
    print(f"  Protect first: {compression.get('protect_first_n', 3)} non-system head messages")
    aux_comp = config.get('auxiliary', {}).get('compression', {})
    print(f"  Model:        {aux_comp.get('model', '') or '(auto)'}")
    comp_provider = aux_comp.get('provider', 'auto')
    if comp_provider and comp_provider != 'auto':
        print(f"  Provider:     {comp_provider}")


def _show_aux_overrides(config: Dict[str, Any]) -> None:
    aux_tasks = {"Vision": config.get('auxiliary', {}).get('vision', {})}
    overrides = {
        label: (t.get('provider', 'auto'), t.get('model', ''))
        for label, t in aux_tasks.items()
        if t.get('provider', 'auto') != 'auto' or t.get('model', '')}
    if not overrides:
        return
    _section("Auxiliary Models (overrides)")
    for label, (prov, mdl) in overrides.items():
        parts = [f"provider={prov}"] + ([f"model={mdl}"] if mdl else [])
        print(f"  {label:12s}  {', '.join(parts)}")


def _show_skill_settings() -> None:
    try:
        from agent.skill_utils import discover_all_skill_config_vars, resolve_skill_config_values
        skill_vars = discover_all_skill_config_vars()
        if not skill_vars:
            return
        resolved = resolve_skill_config_values(skill_vars)
        _section("Skill Settings")
        for var in skill_vars:
            value = resolved.get(var["key"], "")
            display_val = str(value) if value else color("(not set)", Colors.DIM)
            skill_tag = color(f"[{var.get('skill', '')}]", Colors.DIM)
            print(f"  {var['key']:<20s} {display_val}  {skill_tag}")
    except Exception:
        pass


def show_config():
    """Display current configuration."""
    config = load_config()

    print()
    print(color("┌─────────────────────────────────────────────────────────┐", Colors.CYAN))
    print(color("│              ☤ Hermes Configuration                    │", Colors.CYAN))
    print(color("└─────────────────────────────────────────────────────────┘", Colors.CYAN))
    _show_managed_banner()

    _section("Paths")
    print(f"  Config:       {get_config_path()}")
    print(f"  Secrets:      {get_env_path()}")
    print(f"  Install:      {get_project_root()}")

    _section("API Keys")
    for env_key, name in _SHOW_CONFIG_API_KEYS:
        print(f"  {name:<14} {redact_key(get_env_value(env_key))}")
    from hermes_cli.auth import get_anthropic_key
    print(f"  {'Anthropic':<14} {redact_key(get_anthropic_key())}")

    _show_model_section(config)
    _show_display_section(config)
    _show_terminal_section(config)

    _section("Timezone")
    tz = config.get('timezone', '')
    print(f"  Timezone:     {tz or color('(server-local)', Colors.DIM)}")

    _show_compression_section(config)
    _show_aux_overrides(config)

    _section("Messaging Platforms")
    for label, env_key in (("Telegram", "TELEGRAM_BOT_TOKEN"), ("Discord", "DISCORD_BOT_TOKEN")):
        state = 'configured' if get_env_value(env_key) else color('not configured', Colors.DIM)
        print(f"  {label + ':':<13} {state}")

    _show_skill_settings()

    print()
    print(color("─" * 60, Colors.DIM))
    print(color("  hermes config edit     # Edit config file", Colors.DIM))
    print(color("  hermes config set <key> <value>", Colors.DIM))
    print(color("  hermes setup           # Run setup wizard", Colors.DIM))
    print()


def edit_config():
    """Open config file in user's editor."""
    if is_managed():
        managed_error("edit configuration")
        return
    config_path = get_config_path()
    if not config_path.exists():
        seed_config_file(config_path)
        print(f"Created {config_path}")

    # Windows lands on notepad even without Git Bash/nano; POSIX prefers nano/vim, which headless
    # servers are more likely to have.
    candidates = (['notepad', 'code', 'vim', 'vi', 'nano'] if sys.platform == "win32"
                  else ['nano', 'vim', 'vi', 'code', 'notepad'])
    editor = os.getenv('EDITOR') or os.getenv('VISUAL') or next(
        (cmd for cmd in candidates if shutil.which(cmd)), None)
    if not editor:
        print("No editor found. Config file is at:")
        print(f"  {config_path}")
        return

    print(f"Opening {config_path} in {editor}...")
    subprocess.run([editor, str(config_path)])


def _default_value_for_key(dotted_key: str):
    """Return the leaf value declared for *dotted_key* in ``DEFAULT_CONFIG`` (None for dicts/misses)."""
    node = cfg_get(DEFAULT_CONFIG, *_split_key_path(dotted_key))
    return None if isinstance(node, dict) else node


# Top-level keys that accept arbitrary user-supplied child keys (schema declares the dict, the
# user populates it): any path below is accepted without deep checking.
_OPEN_DICT_TOP_LEVEL_KEYS = frozenset({
    "providers", "credential_pool_strategies", "mcp_servers", "hooks", "quick_commands",
    "personalities", "command_allowlist", "model_catalog", "channel_prompts", "server_actions",
    "secrets", "goals", "loops"})

# Top-level keys whose sub-keys are partially schema-defined (e.g. a PlatformConfig dataclass) but
# where users may add fields DEFAULT_CONFIG doesn't enumerate: validate the FIRST segment only.
_SCHEMA_DEFINED_DICT_KEYS = frozenset({
    # Platform configs — PlatformConfig dataclass + dynamic extras
    "discord", "telegram", "slack", "whatsapp", "signal", "mattermost",
    "matrix", "feishu", "wecom", "weixin", "bluebubbles", "qqbot", "yuanbao",
    "email", "sms", "dingtalk",
    # MCP server template / dynamic auth dicts
    "sessions", "checkpoints",
    # Plugin enable/disable lists + per-plugin entries; absent from DEFAULT_CONFIG.
    "plugins"})

# Top-level keys that can be ANY user-supplied name.
_DYNAMIC_TOP_LEVEL_KEYS = frozenset({
    "custom_providers",  # list-shaped, but indexed by position
})

# Containers whose immediate child IS a user-supplied platform name (``platforms.<name>.<field>``),
# both top-level and under ``gateway``; anything below the name is accepted (open ``extra``).
_PLATFORM_CONTAINER_KEYS = frozenset({"platforms"})


# Top-level keys whose sub-keys are accepted without deep checking.
_OPEN_SUBKEY_TOP_LEVEL_KEYS = _OPEN_DICT_TOP_LEVEL_KEYS | _DYNAMIC_TOP_LEVEL_KEYS | _SCHEMA_DEFINED_DICT_KEYS


def _known_top_level_keys() -> set[str]:
    """Return the union of known top-level config keys for validation.

    ``_EXTRA_KNOWN_ROOT_KEYS`` are roots the runtime reads but DEFAULT_CONFIG deliberately
    omits (``platform_toolsets``, ``smart_model_routing``, ...); without them every path under
    such a root was flagged "not a recognized config key" with a difflib near-miss suggestion.
    """
    return set(DEFAULT_CONFIG) | _EXTRA_KNOWN_ROOT_KEYS | _OPEN_SUBKEY_TOP_LEVEL_KEYS


def _suggest_closest_key(key: str, candidates: set[str], cutoff: float = 0.6) -> Optional[str]:
    """Closest candidate key name for a typo'd ``key``, or None."""
    return next(iter(difflib.get_close_matches(key, sorted(candidates), n=1, cutoff=cutoff)), None)


def _validate_config_key(key: str) -> tuple[bool, Optional[str]]:
    """Validate a dotted config-key path against the known schema -> ``(is_known, suggestion)``.

    Headline case from #34067: ``gateway.discord.gateway_restart_notification`` was silently written, even
    though ``gateway`` only has 4 known sub-keys (``strict``, ``media_delivery_allow_dirs``,
    ``trust_recent_files``, ``trust_recent_files_seconds``). The correct path is
    ``discord.gateway_restart_notification`` (platform configs live at the top level, not under a
    ``platforms`` namespace).
    """
    if not key:
        return False, None

    segments = _split_key_path(key)
    top = segments[0]

    # A leading underscore on the FIRST segment marks an intentionally non-schema internal key
    # (test harnesses/tooling); only the first segment is exempt so ``agent._max_turns`` is caught.
    if top.startswith("_") or top in _PLATFORM_CONTAINER_KEYS:
        return True, None

    known = _known_top_level_keys()
    if top not in known:
        suggestion = _suggest_closest_key(top, known)
        if suggestion is None:
            return False, None
        rest = ".".join(segments[1:])
        return False, f"{suggestion}.{rest}" if rest else suggestion

    if top in _OPEN_SUBKEY_TOP_LEVEL_KEYS:
        return True, None

    # Walk DEFAULT_CONFIG: a nested ``platforms`` container, a scalar leaf, or an EMPTY dict hit
    # before the path is consumed all accept. An empty dict is a free-form mapping section
    # (``compression.model_thresholds.<model>``, ``terminal.docker_env.<VAR>``,
    # ``lsp.servers.<lang>``): its keys are user-chosen, so nothing under it can be a typo. An
    # unknown sub-key of a populated section fails with a same-level "did you mean" suggestion.
    node: Any = DEFAULT_CONFIG.get(top)
    consumed = [top]
    for seg in segments[1:]:
        if seg in _PLATFORM_CONTAINER_KEYS or not isinstance(node, dict) or not node:
            return True, None
        if seg not in node:
            # ``gateway.discord.<field>``: the path minus its wrong prefix is itself a known key.
            # Checked BEFORE the fuzzy sibling: a structural match is proof, a fuzzy match is a
            # guess, and ``agent.gateway.strict`` must be refused as ``gateway.strict`` rather
            # than written with a misleading ``agent.gateway_timeout`` did-you-mean.
            # Only DEFAULT_CONFIG / open-subkey roots qualify as the stripped prefix:
            # ``_EXTRA_KNOWN_ROOT_KEYS`` also holds the top-level FORMS of nested gateway
            # settings (``filter_silence_narration``, ``reset_triggers``, ...), and
            # ``gateway.filter_silence_narration`` is a runtime-read path, not a wrong prefix.
            rest = ".".join(segments[len(consumed):])
            if (
                _split_key_path(rest)[0] in set(DEFAULT_CONFIG) | _OPEN_SUBKEY_TOP_LEVEL_KEYS
                and _validate_config_key(rest)[0]
            ):
                return False, rest
            sibling = _suggest_closest_key(seg, set(node.keys()))
            if sibling is not None:
                return False, ".".join(consumed + [sibling])
            return False, None
        consumed.append(seg)
        node = node[seg]
    return True, None


def _is_wrong_prefix_suggestion(key: str, suggestion: Optional[str]) -> bool:
    """Whether *suggestion* proves that *key* has only an extra prefix.

    ``DEFAULT_CONFIG`` is not a complete registry of runtime-read settings, so a
    sibling spelling suggestion alone cannot prove an unseeded path is a typo.
    A known suffix, such as ``gateway.discord.gateway_restart_notification``
    -> ``discord.gateway_restart_notification``, is the narrow case where the
    pre-write refusal is safe.
    """
    if not suggestion:
        return False
    key_segments = _split_key_path(key)
    suggestion_segments = _split_key_path(suggestion)
    return (
        len(suggestion_segments) < len(key_segments)
        and key_segments[-len(suggestion_segments):] == suggestion_segments
        and _validate_config_key(suggestion)[0]
    )


def _looks_structured_value(value: str) -> bool:
    """True when *value* plausibly encodes a YAML/JSON list or mapping. Deliberately conservative:
    a bare leading ``-`` is not a trigger (``-5``, ``--flag`` must stay strings)."""
    stripped = value.lstrip()
    if stripped[:1] in ('[', '{'):
        return True
    if '\n' not in value:
        return False
    for line in value.splitlines():
        item = line.strip()
        if item == '-' or item.startswith('- '):
            return True
        # ``key: value`` / ``key:`` mapping-entry shape (no whitespace in the key).
        head, sep, _rest = item.partition(': ')
        if sep and head and ' ' not in head and not head.startswith('#'):
            return True
        if item.endswith(':') and ' ' not in item[:-1] and item[:-1]:
            return True
    return False


def _coerce_int(value: str):
    """int(value) for a clean integer literal (signs/whitespace/underscores OK), else None."""
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _coerce_float(value: str):
    """``float(value)`` only when the conversion preserves its decimal value; NaN/inf rejected.
    Decimal-looking identifiers more precise than a binary float must stay strings."""
    try:
        f = float(value)
        if f != f or f in (float("inf"), float("-inf")) or Decimal(value) != Decimal(str(f)):
            return None
    except (TypeError, ValueError, InvalidOperation):
        return None
    return f


_SCALAR_WORDS = {
    'true': True, 'yes': True, 'on': True,
    'false': False, 'no': False, 'off': False,
    # YAML null. Many DEFAULT_CONFIG leaves are "null/absent = off"; without this,
    # ``config set X null`` stored the truthy string "null" and the feature could never be cleared.
    'null': None, 'none': None, '~': None}


def _coerce_config_set_value(key: str, value: str) -> Any:
    """Auto-coerce a ``hermes config set`` string to bool/None/int/float/list/dict.
    String-typed settings (per ``DEFAULT_CONFIG``) are preserved verbatim so enum members such as
    ``approvals.mode="off"`` never become booleans. List/mapping literals are parsed so
    isinstance-gated readers see real structures; the trigger is conservative."""
    if isinstance(_default_value_for_key(key), str):
        return value
    stripped = value.strip()
    lower = stripped.lower()
    if lower in _SCALAR_WORDS:
        return _SCALAR_WORDS[lower]
    for coerce in (_coerce_int, _coerce_float):
        coerced = coerce(stripped)
        if coerced is not None:
            return coerced
    if not _looks_structured_value(value):
        return value
    try:
        parsed = yaml.safe_load(value)
    except yaml.YAMLError as exc:
        # Storing the text as a string here used to be a warning; every isinstance-gated reader
        # then ignored the value while `config get` echoed it back (#114471). Refuse instead.
        detail = str(getattr(exc, "problem", None) or exc).splitlines()[0]
        _exit_invalid(
            f"✗ Value for '{key}' looks like a list/mapping but is not valid YAML/JSON "
            f"({detail}) — nothing was written.\n"
            "  Fix the literal, or quote it (e.g. \"'[text'\") to store a plain string.")
    if isinstance(parsed, (list, dict)):
        return parsed
    # A quoted literal ("'[text'") parses to a scalar: that is the deliberate way to store one.
    return value


# Container roots absent from DEFAULT_CONFIG whose shape is nonetheless fixed by their readers,
# so the guardrail holds before anything is on disk (#114471: `model.aliases notamap`).
_KNOWN_CONTAINER_TYPES = {
    "custom_providers": "list",
    "providers": "mapping",
    "model.aliases": "mapping",
    "model_aliases": "mapping",
    # Omitted from DEFAULT_CONFIG on purpose (an empty default would clobber a user allow-list),
    # so without these rows `config set plugins.enabled foo` stored a string every reader ignored.
    "plugins.enabled": "list",
    "plugins.disabled": "list",
    "model_catalog.excluded_providers": "list",
}
# List slots whose readers go through ``parse_config_string_list``: a bare name is one entry.
_SCALAR_AS_ONE_ITEM_LIST_KEYS = frozenset({"agent.disabled_toolsets", "skills.disabled"})


def _expected_container_type(key: str, user_config: Dict[str, Any]) -> Optional[str]:
    """``"list"`` / ``"mapping"`` when the schema (``DEFAULT_CONFIG``, the known-container table,
    or the value already on disk) fixes *key* to a container; ``None`` for scalars and open paths.
    A single-segment key that is a mapping *section* in the schema skips the lookup: replacing a
    whole section is ``_guard_section_overwrite``'s call (``--force``, the bare ``model`` shorthand)."""
    parts = _split_key_path(key)
    schema_node = cfg_get(DEFAULT_CONFIG, *parts)
    if len(parts) == 1 and isinstance(schema_node, dict):
        schema_node = None
    existing = _get_nested(user_config, key)
    for node in (schema_node, _KNOWN_CONTAINER_TYPES.get(key), existing):
        if isinstance(node, dict) or node == "mapping":
            return "mapping"
        if isinstance(node, list) or node == "list":
            return "list"
    return None


def _refuse_container_type_mismatch(key: str, value: Any, user_config: Dict[str, Any], force: bool) -> Any:
    """Hard guardrail: never store a value of the wrong shape where the schema wants a list or a
    mapping — every reader would ignore it while ``config get`` echoed it back. ``--force`` keeps
    its documented meaning (replace a whole mapping section); a non-list in a list slot is never
    readable, so it has no override. Returns the value to store: a bare name for a
    ``parse_config_string_list``-read slot becomes a one-item list."""
    expected = _expected_container_type(key, user_config)
    if expected is None:
        return value
    if expected == "list" and isinstance(value, str) and key in _SCALAR_AS_ONE_ITEM_LIST_KEYS:
        return [value]
    ok = isinstance(value, list) if expected == "list" else isinstance(value, dict)
    if ok or (expected == "mapping" and force):
        return value
    got = type(value).__name__ if not isinstance(value, str) else "string"
    literal = "[item, ...]" if expected == "list" else "{key: value}"
    _exit_invalid(
        f"✗ Cannot set '{key}': it must be a {expected}, got a {got} — nothing was written.\n"
        f"  Pass a YAML/JSON literal, e.g.:\n    hermes config set {key} '{literal}'\n"
        "  or edit config.yaml directly.")


def _redirect_platform_display_key(key: str) -> tuple[str, Optional[str]]:
    """Canonicalize ``platforms.<name>.<display_setting>`` -> ``display.platforms.<name>.<setting>``.
    The gateway resolves per-platform display settings (streaming, show_reasoning, ...) from
    ``display.platforms``; the top-level ``platforms.<name>`` block holds only connection config.
    Only known display settings (``OVERRIDEABLE_KEYS``) are redirected. Returns ``(key, note)``;
    the gateway import is guarded so the CLI works where the gateway package is unavailable.

    Before #71047 a write such as ``hermes config set platforms.telegram.streaming false`` landed on a key
    the gateway never reads: ``config get`` echoed the new value back while the runtime kept the old
    ``display.platforms`` one — a silent no-op that looks like a duplicated key to the user.

    ``gateway.platforms.<name>.<field>`` is canonicalized to the top-level ``platforms.<name>.<field>``
    first (#115212): ``merge_platform_sections`` reads both blocks but the top-level one wins on
    shared keys, so a nested write beside an existing top-level value printed ``✓ Set`` while the
    gateway kept the old value.
    """
    segs = _split_key_path(key)
    note = None
    if len(segs) >= 3 and segs[0] == "gateway" and segs[1] == "platforms":
        segs = segs[1:]
        key = ".".join(segs)
        note = f"  (note: the top-level platforms.{segs[1]} block outranks gateway.platforms — saved as {key})"
    if len(segs) != 3 or segs[0] != "platforms":
        return key, note
    try:
        from gateway.display_config import OVERRIDEABLE_KEYS as _display_keys
    except Exception:
        return key, note
    if segs[2] not in _display_keys:
        return key, note
    canonical = f"display.platforms.{segs[1]}.{segs[2]}"
    return canonical, f"  (note: per-platform display setting — saved as {canonical})"


def _legacy_gateway_platforms_key(requested_key: str) -> Optional[str]:
    """The ``gateway.platforms.<name>.<field>`` spelling the user typed, when that is what they typed.
    ``merge_platform_sections`` still honours a value that lives only there, so ``get`` must fall
    back to it and ``unset``/``set`` must clear it, or the CLI reports "not set" / writes a value
    while the gateway keeps reading the nested one."""
    segs = _split_key_path(requested_key)
    if len(segs) >= 3 and segs[0] == "gateway" and segs[1] == "platforms":
        return ".".join(segs)
    return None


def _exit_if_key_managed(key: str, action: str) -> None:
    """A key pinned by the managed layer cannot be set/unset (the next load would reinstate it):
    hard-reject and name the source. Distinct from ``is_managed()``; env-shaped keys route to the
    .env writers, which carry their own guard."""
    if managed_scope.is_key_managed(key):
        print(
            f"Cannot {action} '{key}': it is managed by your administrator ({_managed_source('config.yaml')}) "
            f"and cannot be changed. Contact your administrator to modify it.", file=sys.stderr)
        sys.exit(1)


def _guard_section_overwrite(key: str, value: Any, user_config: Dict[str, Any], force: bool) -> str:
    """Refuse (or with ``force`` allow) a single-segment key overwriting a mapping with a scalar.
    Bare ``model`` is a documented shorthand — redirected to ``model.default`` so siblings survive.
    Returns the (possibly redirected) key."""
    existing = user_config.get(key)
    if "." in key or not isinstance(existing, dict):
        return key
    if key == "model":
        if force:
            print(
                f"⚠ Replacing entire 'model' section with a scalar "
                f"(discarding {len(existing)} existing sub-key(s))")
            return key
        print(
            f"✓ Redirecting bare 'model' to 'model.default' "
            f"(preserving {len(existing)} existing model sub-key(s))")
        return "model.default"
    if force:
        return key
    sub = [k for k in existing if isinstance(k, str)]
    err = [
        f"✗ Cannot set '{key}' to a scalar — '{key}' is a "
        f"configuration section with {len(sub)} sub-key(s)."]
    if sub:
        err.append(f"  Sub-keys: {', '.join(sub[:8])}")
        if len(sub) > 8:
            err.append(f"  ... and {len(sub) - 8} more")
    err += [
        "  Use a dotted path to set a specific leaf key:",
        f"    hermes config set {key}.<sub-key> <value>",
        "  Or use --force to replace the entire section:",
        f"    hermes config set --force {key} {value!r}"]
    print("\n".join(err), file=sys.stderr)
    sys.exit(1)


def _touch_skin_file(key: str, value: Any) -> None:
    """``display.skin`` set means "apply NOW": bump the skin file's mtime so the gateway watcher's
    (name, mtime) signature moves even when the name is unchanged. Best-effort."""
    if key == "display.skin" and isinstance(value, str) and value:
        try:
            skin_file = get_hermes_home() / "skins" / f"{value}.yaml"
            if skin_file.exists():
                skin_file.touch()
        except Exception:
            pass


def _exit_invalid(msg: str) -> None:
    print(msg, file=sys.stderr)
    sys.exit(1)


def _write_user_config(config_path: Path, user_config: Dict[str, Any]) -> None:
    """Write only the user's raw config back (never the merged defaults)."""
    ensure_hermes_home()
    atomic_config_write(config_path, user_config)


def _print_unknown_key_notice(key: str, suggestion: Optional[str]) -> None:
    print(color(
        f"⚠ '{key}' is not a recognized config key — it was saved anyway, "
        "but Hermes may not read it.", Colors.YELLOW))
    if suggestion:
        print(color(f"  Did you mean: {suggestion}", Colors.YELLOW))
    # The env bridge covers custom TOP-LEVEL keys only; an unseeded nested path (``stt.provider``)
    # is written but not bridged, so the footer would be a false promise there.
    if len(_split_key_path(key)) == 1:
        print(color(
            "  (Custom top-level keys are supported and bridged to the "
            "environment for skills/external tools. Use --force to skip "
            "this notice.)", Colors.DIM))
    else:
        print(color("  (Use --force to skip this notice.)", Colors.DIM))


def _unknown_subkey_refusal(key: str, suggestion: Optional[str]) -> str:
    lines = [color(f"✗ '{key}' is not a recognized config key — nothing was written.", Colors.RED)]
    if suggestion:
        lines.append(color(f"  Did you mean: {suggestion}", Colors.YELLOW))
    lines.append(color(
        "  (Custom top-level keys are supported; use --force to write this path anyway.)", Colors.DIM))
    return "\n".join(lines)


def set_config_value(key: str, value: str, force: bool = False):
    """Set a configuration value at a dotted ``key``; ``value`` is auto-coerced to bool/int/float.
    ``force`` writes a known key given under the wrong prefix (``gateway.discord.foo`` where
    ``discord.foo`` is known; otherwise refused — any other unknown path under a known section
    is written with a did-you-mean notice), skips the unknown-top-level-key notice AND
    authorizes replacing a mapping section with a scalar. Without it, scalar writes over mappings are refused and bare ``model`` is redirected
    to ``model.default``."""
    if is_managed():
        managed_error("set configuration values")
        return
    # Empty segments (``"agent."``) would write config["agent"][""] into a live schema section.
    if key != key.strip() or not key.strip():
        _exit_invalid(f"✗ Invalid config key: {key!r} (empty or surrounding whitespace).")
    if "" in _split_key_path(key):
        _exit_invalid(
            f"✗ Invalid config key: {key!r} — contains an empty path segment "
            "(leading, trailing, or doubled '.').")
    _exit_if_key_managed(key, "set")
    if _is_env_config_key(key):
        from hermes_cli.credential_lifecycle import save_provider_env_credential

        # Unified lifecycle: also rotates any config.yaml mirror of the old value so a stale
        # higher-precedence copy can't win (#62269).
        save_provider_env_credential(key.upper(), value)
        print(f"✓ Set {key} in {get_env_path()}")
        return
    from hermes_cli.config_env_routing import is_env_setting_key, save_env_setting

    if is_env_setting_key(key):
        # Every UPPER_SNAKE name is an environment setting: same file the platform setup flows and
        # /sethome write, and the only one os.getenv readers see. config.yaml never gets one from
        # here, --force included (#111848). The env writer's denylist (HERMES_YOLO_MODE, PATH, ...)
        # therefore also refuses the config.yaml detour that used to bridge those into os.environ.
        try:
            save_env_setting(key, value)
        except ValueError as exc:
            _exit_invalid(f"✗ {exc}")
        print(f"✓ Set {key.upper()} in {get_env_path()}")
        return

    # Canonicalize per-platform display keys BEFORE validation/coercion so both see the path the
    # runtime reads.
    legacy_key = _legacy_gateway_platforms_key(key)
    key, _redirect_note = _redirect_platform_display_key(key)
    if _redirect_note:
        print(_redirect_note)
    is_known, suggestion = _validate_config_key(key)
    # DEFAULT_CONFIG is an incomplete schema: runtime-read settings may deliberately have no
    # seeded default. Refuse only the positive wrong-prefix case from #112003; other unknown
    # paths keep the post-write warning so valid runtime settings remain configurable.
    if not is_known and not force and _is_wrong_prefix_suggestion(key, suggestion):
        _exit_invalid(_unknown_subkey_refusal(key, suggestion))

    # Read the RAW user config (not merged) so defaults are never dumped back; fail-closed.
    config_path = get_config_path()
    user_config = require_readable_config_before_write(config_path)
    value = _coerce_config_set_value(key, value)
    # A scalar ``model`` shorthand must become a dict before writing sub-keys, or _set_nested
    # replaces it with an empty dict and the model id is lost.
    _model_val = user_config.get("model")
    if key.strip().lower().startswith("model.") and isinstance(_model_val, str) and _model_val:
        user_config["model"] = {"default": _model_val}
    key = _guard_section_overwrite(key, value, user_config, force)
    value = _refuse_container_type_mismatch(key, value, user_config, force)
    _old_provider = _model_val.get("provider") if isinstance(_model_val, dict) else None
    try:
        _set_nested(user_config, key, value)
    except ValueError as e:
        _exit_invalid(f"✗ {e}")
    if legacy_key and _unset_nested(user_config, legacy_key):
        print(f"  (removed the shadowed {legacy_key} duplicate)")
    # A provider switch re-points ``model:`` at a new route; ``base_url``/``api_mode`` are route
    # state of the OLD provider, and the runtime honours them for whatever provider the block now
    # names — the new provider's key would be posted to the old endpoint (#113719, #40862). Sync
    # them the way a persisted ``/model`` switch does: the previous route goes unless it is the
    # new provider's own endpoint.
    _route_notice = ""
    _old_provider = str(_old_provider or "").strip() or "the previous provider"
    if key == "model.provider" and _old_provider.lower() != str(value).strip().lower():
        from hermes_cli.route_identity import drop_stale_model_route
        _popped, _unverified = drop_stale_model_route(user_config.get("model"), value, user_config)
        if _popped:
            _route_notice = (
                "  Cleared " + ", ".join(f"model.{k} ({v})" for k, v in _popped.items())
                + f" — that route belonged to {_old_provider}, not {value}. {value}'s endpoint resolves "
                "automatically; set model.base_url again if you meant a custom endpoint.")
        elif _unverified:
            _route_notice = color(
                f"⚠ model.base_url ({user_config['model'].get('base_url')}) was set under {_old_provider} and "
                f"still applies to {value} — requests go there. If it is not {value}'s endpoint: "
                "`hermes config unset model.base_url` (and model.api_mode).", Colors.YELLOW)
    # api_base -> base_url alias at set-time too (mirrors _normalize_root_model_keys).
    if key.strip().lower() in ("model.api_base", "api_base"):
        # Normalize the api_base → base_url alias at set-time too (issue #8919), so a fresh `hermes config
        # set model.api_base ...` lands on the canonical key the runtime resolver actually reads, instead of
        # being silently ignored.
        user_config = _normalize_root_model_keys(user_config)
        key = "model.base_url"
        print("  (note: 'api_base' is an alias — saved as model.base_url)")
    _write_user_config(config_path, user_config)

    # Keep .env in sync: terminal_tool reads TERMINAL_ENV etc. directly from env vars.
    env_var = terminal_config_env_var_for_key(key)
    if env_var and key != "terminal.cwd":
        save_env_value(env_var, _terminal_env_value(value))

    _touch_skin_file(key, value)

    # Mask the echoed value when the (possibly nested) key is credential-shaped, e.g.
    # ``model.api_key`` (lowercase, so it misses the .env routing above).
    _display_value = value
    if _is_secret_config_key(key) and isinstance(value, str) and value:
        from agent.redact import mask_secret
        _display_value = mask_secret(value)
    print(f"✓ Set {key} = {_display_value} in {config_path}")
    if _route_notice:
        print(_route_notice)

    # Post-write unknown-key notice (#34067): value IS saved, but tell the user the runtime may never read
    # it and suggest the likely-intended path.
    if not is_known and not force:
        _print_unknown_key_notice(key, suggestion)


def get_config_value(key: str, *, as_json: bool = False, raw: bool = False):
    """Print a resolved configuration value. Credentials are masked unless ``--raw`` or
    ``security.redact_secrets: false``: ``print`` bypasses the log redactor, and the agent runs
    this command from sessions whose transcripts persist (#84106, #110758)."""
    from hermes_cli.config_env_routing import is_env_setting_key, read_env_setting

    if _is_env_config_key(key):
        env_value = get_env_value(key.upper())
        value = _MISSING if env_value is None else env_value
    elif is_env_setting_key(key):
        env_value = read_env_setting(key)
        value = _MISSING if env_value is None else env_value
    else:
        # Mirror set_config_value: read the canonical display.platforms path.
        # See #71047.
        legacy_key = _legacy_gateway_platforms_key(key)
        key, _ = _redirect_platform_display_key(key)
        config = load_config()
        value = _get_nested(config, key)
        if value is _MISSING and legacy_key:
            value = _get_nested(config, legacy_key)

    if value is _MISSING:
        _exit_invalid(f"Config key not set: {key}")

    from agent.redact import _redact_enabled, mask_secret
    if not raw and _redact_enabled():
        if isinstance(value, str):
            if _is_secret_config_key(key) and not _ENV_PLACEHOLDER_RE.match(value):
                value = mask_secret(value)
        else:
            value = redact_config_value(value)

    print(_format_config_get_value(value, as_json=as_json), flush=True)

    # Phantom-key notice (#112348): a nested path under a KNOWN section that the schema does not
    # define (``compression.compressor.enabled``) is echoed straight from the user's file and is
    # usually read by nothing, so it must not look like a live setting. The check is a
    # DEFAULT_CONFIG walk and some live keys are deliberately unseeded (``browser.cloud_provider``,
    # ``stt.provider``, ``gateway.proxy_url``: a stored value counts as an explicit user pick), so
    # the wording hedges exactly like the set-path notice. Custom top-level keys stay exempt (they
    # are bridged into os.environ for skills) and ``_validate_config_key`` already accepts
    # open-subkey sections. stderr keeps stdout/--json parseable; the exit code stays 0.
    if _split_key_path(key)[0] in _known_top_level_keys():
        is_known, suggestion = _validate_config_key(key)
        if not is_known:
            print(color(
                f"⚠ '{key}' is not a recognized config key — Hermes may not read it; the value "
                "printed above comes from your config file.", Colors.YELLOW), file=sys.stderr)
            if suggestion:
                print(color(f"  Did you mean: {suggestion}", Colors.YELLOW), file=sys.stderr)


def unset_config_value(key: str):
    """Remove a user-set configuration or .env value."""
    if is_managed():
        managed_error("unset configuration values")
        return
    _exit_if_key_managed(key, "unset")

    if _is_env_config_key(key):
        # Unified lifecycle: also prunes env-seeded credential_pool entries and model-cache rows so
        # the provider is fully removed instead of left resurrectable.
        # See #51071.
        from hermes_cli.credential_lifecycle import remove_provider_env_credential

        if not remove_provider_env_credential(key.upper()).get("found"):
            _exit_invalid(f"Config key not set: {key}")
        print(f"✓ Unset {key} from {get_env_path()}")
        return
    from hermes_cli.config_env_routing import is_env_setting_key, remove_env_setting

    if is_env_setting_key(key):
        # Also drops a stale top-level config.yaml copy left by older `config set` runs (#111848).
        if not remove_env_setting(key):
            _exit_invalid(f"Config key not set: {key}")
        print(f"✓ Unset {key} from {get_env_path()}")
        return

    config_path = get_config_path()
    user_config = require_readable_config_before_write(config_path)

    legacy_key = _legacy_gateway_platforms_key(key)
    key, _redirect_note = _redirect_platform_display_key(key)
    if _redirect_note:
        # Mirror set_config_value's display.platforms canonicalization (#71047).
        print(_redirect_note.replace("saved as", "resolved as"))
    removed = _unset_nested(user_config, key)
    if legacy_key:
        removed = _unset_nested(user_config, legacy_key) or removed

    env_var = terminal_config_env_var_for_key(key)
    if env_var and key != "terminal.cwd":
        removed = remove_env_value(env_var) or removed

    if not removed:
        _exit_invalid(f"Config key not set: {key}")

    _write_user_config(config_path, user_config)
    print(f"✓ Unset {key} from {config_path}")


# ---- Command handler ----

def _usage_exit(usage: str, examples: List[str], extra: Optional[List[str]] = None) -> None:
    print(usage)
    print()
    print("Examples:")
    for line in examples:
        print(f"  {line}")
    for line in extra or ():
        print(line)
    sys.exit(1)


def _run_write_command(fn, *args) -> None:
    """Run a config writer, surfacing the fail-closed write guard's RuntimeError as a clean CLI
    error instead of a traceback."""
    try:
        fn(*args)
    except RuntimeError as exc:
        _exit_invalid(f"✗ {exc}")


_USAGE_GET = ("Usage: hermes config get <key> [--json] [--raw]", [
    "hermes config get model", "hermes config get terminal.backend",
    "hermes config get skills.config --json"], None)
_USAGE_SET = ("Usage: hermes config set [--force] <key> <value>", [
    "hermes config set model anthropic/claude-sonnet-4", "hermes config set terminal.backend docker",
    "hermes config set OPENROUTER_API_KEY sk-or-..."], [
    "", "  --force: skip the unknown-key notice for unrecognized keys,",
    "           and allow a scalar to replace a whole mapping section"])
_USAGE_UNSET = ("Usage: hermes config unset <key>", [
    "hermes config unset model", "hermes config unset terminal.backend",
    "hermes config unset OPENROUTER_API_KEY"], None)


def _cmd_config_get(args):
    key = getattr(args, 'key', None)
    if not key:
        _usage_exit(*_USAGE_GET)
    get_config_value(key, as_json=getattr(args, 'json', False), raw=bool(getattr(args, 'raw', False)))


def _cmd_config_set(args):
    key = getattr(args, 'key', None)
    value = getattr(args, 'value', None)
    if not key or value is None:
        _usage_exit(*_USAGE_SET)
    _run_write_command(set_config_value, key, value, bool(getattr(args, 'force', False)))


def _cmd_config_unset(args):
    key = getattr(args, 'key', None)
    if not key:
        _usage_exit(*_USAGE_UNSET)
    _run_write_command(unset_config_value, key)


def _tools_suffix(info: Dict[str, Any], fmt: str) -> str:
    tools = info.get("tools", [])
    return fmt.format(", ".join(tools[:2])) if tools else ""


def _print_banner(text: str) -> None:
    print()
    print(color(text, Colors.CYAN, Colors.BOLD))
    print()


def _cmd_config_migrate(args):
    _print_banner("🔄 Checking configuration for updates...")

    missing_env = get_missing_env_vars(required_only=False)
    missing_config = get_missing_config_fields()
    current_ver, latest_ver = check_config_version(raise_on_parse_error=True)

    if not missing_env and not missing_config and current_ver >= latest_ver:
        print(color("✓ Configuration is up to date!", Colors.GREEN))
        print()
        return

    if current_ver < latest_ver:
        print(f"  Config version: {current_ver} → {latest_ver}")

    if missing_config:
        print(f"\n  {len(missing_config)} new config option(s) will be added with defaults")

    required_missing = [v for v in missing_env if v.get("is_required")]
    optional_missing = [v for v in missing_env if not v.get("is_required") and not v.get("advanced")]
    for heading, group, suffix in (
        ("⚠️  {} required API key(s) missing:", required_missing, ""),
        ("ℹ️  {} optional API key(s) not configured:", optional_missing, " (enables: {})")):
        if group:
            print(f"\n  {heading.format(len(group))}")
            for var in group:
                print(f"     • {var['name']}{_tools_suffix(var, suffix) if suffix else ''}")

    print()
    results = migrate_config(interactive=True, quiet=False)
    print()
    if results["env_added"] or results["config_added"]:
        print(color("✓ Configuration updated!", Colors.GREEN))
    if results["warnings"]:
        print()
        for warning in results["warnings"]:
            print(color(f"  ⚠️  {warning}", Colors.YELLOW))
    print()


def _cmd_config_check(args):
    """Non-interactive report of what's missing."""
    _print_banner("📋 Configuration Status")

    current_ver, latest_ver = check_config_version(raise_on_parse_error=True)
    if current_ver >= latest_ver:
        print(f"  Config version: {current_ver} ✓")
    else:
        print(color(f"  Config version: {current_ver} → {latest_ver} (update available)", Colors.YELLOW))

    groups = (
        ("Required", REQUIRED_ENV_VARS, lambda n, i: color(f"    ✗ {n} (missing)", Colors.RED)),
        ("Optional", OPTIONAL_ENV_VARS,
         lambda n, i: color(f"    ○ {n}{_tools_suffix(i, ' → {}')}", Colors.DIM)))
    for title, table, missing_line in groups:
        print()
        print(color(f"  {title}:", Colors.BOLD))
        for var_name, info in table.items():
            print(f"    ✓ {var_name}" if get_env_value(var_name) else missing_line(var_name, info))

    missing_config = get_missing_config_fields()
    if missing_config:
        print()
        print(color(f"  {len(missing_config)} new config option(s) available", Colors.YELLOW))
        print("    Run 'hermes config migrate' to add them")

    print()


_CONFIG_SUBCOMMANDS = {
    None: lambda args: show_config(),
    "show": lambda args: show_config(),
    "edit": lambda args: edit_config(),
    "get": _cmd_config_get,
    "set": _cmd_config_set,
    "unset": _cmd_config_unset,
    "path": lambda args: print(get_config_path()),
    "env-path": lambda args: print(get_env_path()),
    "migrate": _cmd_config_migrate,
    "check": _cmd_config_check}

_CONFIG_USAGE = """Available commands:
  hermes config           Show current configuration
  hermes config edit      Open config in editor
  hermes config get <key>          Print a resolved config value
  hermes config set <key> <value>   Set a config value
  hermes config unset <key>        Remove a config value
  hermes config check     Check for missing/outdated config
  hermes config migrate   Update config with new options
  hermes config path      Show config file path
  hermes config env-path  Show .env file path"""


def config_command(args):
    """Handle config subcommands."""
    subcmd = getattr(args, 'config_command', None)
    handler = _CONFIG_SUBCOMMANDS.get(subcmd)
    if handler is not None:
        handler(args)
        return
    print(f"Unknown config command: {subcmd}")
    print()
    print(_CONFIG_USAGE)
    sys.exit(1)


# ---- OPTIONAL_ENV_VARS injection from provider profiles and platform plugins (once, at import) ----

def _inject_profile_env_vars() -> None:
    """Expose env_vars of every ``auth_type="api_key"`` provider in providers/ via OPTIONAL_ENV_VARS
    without editing this file."""
    try:
        from providers import list_providers
        for _pp in list_providers():
            if _pp.auth_type != "api_key":
                continue
            for _var in _pp.env_vars:
                if _var in OPTIONAL_ENV_VARS:
                    continue
                _is_key = not _var.endswith(("_BASE_URL", "_URL"))
                _label = _pp.display_name or _pp.name
                OPTIONAL_ENV_VARS[_var] = {
                    "description": f"{_label} {'API key' if _is_key else 'base URL override'}",
                    "prompt": f"{_label} {'API key' if _is_key else 'base URL (leave empty for default)'}",
                    "url": _pp.signup_url or None,
                    "password": _is_key,
                    "category": "provider",
                    "advanced": True}
    except Exception:
        pass


_inject_profile_env_vars()


def _platform_plugin_manifests():
    """Yield ``(dir_name, manifest_dict)`` for every platform plugin manifest: bundled
    ``plugins/platforms/*``, the user's ``<HERMES_HOME>/plugins/platforms/*`` category dir, and flat
    user installs ``<HERMES_HOME>/plugins/*`` that declare ``kind: platform`` (#46600)."""
    user_plugins = get_hermes_home() / "plugins"
    roots = (
        (get_project_root() / "plugins" / "platforms", False),
        (user_plugins / "platforms", False),
        (user_plugins, True),  # flat layout: only manifests that say they are platforms
    )
    for root, require_kind in roots:
        if not root.is_dir():
            continue
        for child in root.iterdir():
            manifest_path = next(
                (p for p in (child / "plugin.yaml", child / "plugin.yml") if child.is_dir() and p.exists()), None)
            if manifest_path is None:
                continue
            try:
                with open(manifest_path, "r", encoding="utf-8-sig") as f:
                    manifest = fast_safe_load(f) or {}
            except Exception:
                continue
            if not isinstance(manifest, dict) or (require_kind and manifest.get("kind") != "platform"):
                continue
            yield child.name, manifest


def _inject_platform_plugin_env_vars() -> None:
    """Populate OPTIONAL_ENV_VARS from platform plugin manifests (bundled AND user-installed) so
    Teams / IRC / Google Chat and third-party platforms are configurable in the ``hermes config`` /
    Desktop Gateway form without the core knowing they exist.

    ``requires_env`` / ``optional_env`` entries are a bare name or a dict with ``name`` plus
    optional ``description``/``url``/``password``/``prompt``/``category``. Failures are swallowed
    so a malformed plugin.yaml can't break CLI import.
    """
    try:
        for dir_name, manifest in _platform_plugin_manifests():
            label = manifest.get("label") or manifest.get("name") or dir_name
            for entry in [*(manifest.get("requires_env") or []), *(manifest.get("optional_env") or [])]:
                meta = {"name": entry} if isinstance(entry, str) else entry if isinstance(entry, dict) else {}
                name = meta.get("name")
                if not name or name in OPTIONAL_ENV_VARS:
                    continue  # hardcoded entry wins (back-compat)
                # *TOKEN / *SECRET / *KEY / *PASSWORD / *JSON are password fields unless overridden.
                is_secret = bool(meta.get("password") or meta.get("secret"))
                if not is_secret and not meta.get("password") is False:
                    is_secret = name.upper().endswith(("_TOKEN", "_SECRET", "_KEY", "_PASSWORD", "_JSON"))
                OPTIONAL_ENV_VARS[name] = {
                    "description": meta.get("description") or f"{label} configuration",
                    "prompt": meta.get("prompt") or name,
                    "url": meta.get("url") or None,
                    "password": is_secret,
                    "category": meta.get("category") or "messaging"}
    except Exception:
        pass


_inject_platform_plugin_env_vars()


# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
# Names external plugins imported from this module before the Sep 2026 decomposition.
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
# The whole block is removed by reverting the commit that added it.

def _install_method_project_root(project_root: Optional[Path] = None) -> Path:
    """Resolve the directory that holds the *running code* (the install tree).

    This is the parent of ``hermes_cli/`` — i.e. the git checkout for source
    installs, ``/opt/hermes`` inside the published image. It is a property of
    the running interpreter, NOT of ``$HERMES_HOME``, which is why a
    code-scoped stamp here is immune to two installs sharing one data
    directory.
    """
    if project_root is not None:
        return project_root
    return Path(__file__).parent.parent.resolve()

def stamp_install_method(method: str, project_root: Optional[Path] = None) -> None:
    """Write the install method next to the running code (code-scoped stamp).

    The stamp lives in the install tree (``<install tree>/.install_method``),
    not in ``$HERMES_HOME``, so that two installs sharing one data directory
    do not overwrite each other's marker. See ``detect_install_method`` for
    the full rationale.

    Best-effort: if the install tree is read-only (e.g. the immutable
    ``/opt/hermes`` in the published image, which instead bakes the stamp at
    build time) the write silently no-ops and detection falls back to its
    other signals.
    """
    root = _install_method_project_root(project_root)
    try:
        root.mkdir(parents=True, exist_ok=True)
        (root / ".install_method").write_text(method + "\n", encoding="utf-8")
    except OSError:
        pass


_PLUGIN_COMPAT_LAZY = {
    'normalize_route_base_url': ('hermes_cli.route_identity', 'normalize_route_base_url'),
}


def __getattr__(name):  # PEP 562 — lazy so no import cycles
    target = _PLUGIN_COMPAT_LAZY.get(name)
    if target is None:
        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
    import importlib
    from hermes_cli.plugin_compat import warn_once
    warn_once(__name__, name, *target)
    return getattr(importlib.import_module(target[0]), target[1])
# ---- END PLUGIN-COMPAT ----