"""Helpers for loading Hermes .env files consistently across entrypoints."""

from __future__ import annotations

import codecs
import io
import logging
import os
import sys
import threading
from pathlib import Path

from dotenv import load_dotenv
from utils import atomic_replace, fast_safe_load

logger = logging.getLogger(__name__)

# The ONLY env vars sanitized on load: we must not silently alter arbitrary user env vars, but
# credentials are known to require pure ASCII (they become HTTP header values).
_CREDENTIAL_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY")

# Once-per-process guards: load_hermes_dotenv() is called repeatedly (user env + project env,
# gateway hot-reload, lazy imports mid-turn, tests) so warnings/logs fire once per key/path/home.
_WARNED_KEYS: set[str] = set()          # credential names already given the non-ASCII warning
_WARNED_UTF32_PATHS: set[str] = set()   # .env paths already given the UTF-32 refuse-to-mangle warning
_SCOPED_SKIP_LOGGED: set[str] = set()   # routed profile homes whose multiplex dotenv skip was logged

# env-var name → source label ("bitwarden", …) for credentials injected by an external secret
# source. Setup / `hermes model` use it so users know WHERE a key came from when .env lacks it.
_SECRET_SOURCES: dict[str, str] = {}
# Applied values are immutable per-home snapshots: ``os.environ`` is shared across profiles and
# may be overwritten by a later home's source apply.
_SECRET_SOURCE_VALUES_BY_HOME: dict[str, dict[str, str]] = {}
# HERMES_HOME paths already pulled external secrets for. load_hermes_dotenv() runs at import time
# from several hot modules (cli.py, hermes_cli/main.py, run_agent.py, gateway/run.py, …); without
# this the Bitwarden status line prints 3-5x per startup and the config re-parse + ASCII sweep
# re-run each time (Bitwarden's own cache only saves the network call).
_APPLIED_HOMES: set[str] = set()
_SECRET_SOURCE_CACHE_LOCK = threading.RLock()

# Behavioral routing keys a parent Hermes process injects into child env that silently redirect a
# profile onto the wrong provider path. These — and ONLY these — are scrubbed from os.environ at
# startup when absent from the profile's .env. Credential keys are excluded: shell exports are a
# legitimate, documented way to supply them, and read-time secret-scope checks
# (agent/secret_scope.py) own cross-profile credential isolation.
_PROFILE_MANAGED_ENV_KEYS: frozenset[str] = frozenset({
    "HERMES_ACP_AUTH_METHOD", "HERMES_ACP_AUTO_APPROVE", "HERMES_COPILOT_ACP_COMMAND",
    "HERMES_COPILOT_ACP_ARGS", "COPILOT_CLI_PATH", "COPILOT_ACP_BASE_URL",
})


def _env_keys_defined_in_dotenv(path: Path) -> set[str]:
    """KEY names assigned in a dotenv file (including empty ``KEY=``). A fast line scanner, not full
    dotenv parsing, so it works during early bootstrap without importing python-dotenv; decode
    errors fall back to ``latin-1`` like ``_load_dotenv_with_fallback``."""
    keys: set[str] = set()
    try:
        text = path.read_text(encoding="utf-8", errors="replace")
    except Exception:
        try:
            text = path.read_text(encoding="latin-1", errors="replace")
        except Exception:
            return keys
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key = line.removeprefix("export ").split("=", 1)[0].strip()
        if key:
            keys.add(key)
    return keys


def _clear_known_keys_missing_from_dotenv(path: Path) -> None:
    """After ``.env`` loaded with ``override=True``, delete inherited ``_PROFILE_MANAGED_ENV_KEYS``
    it does not define. Deliberately NARROW: only keys that change *which provider path* is used."""
    if not path.exists():
        return
    defined = _env_keys_defined_in_dotenv(path)
    for key in _PROFILE_MANAGED_ENV_KEYS:
        if key not in defined and key in os.environ:
            del os.environ[key]


def get_secret_source(env_var: str) -> str | None:
    """Source label that supplied ``env_var`` (``"bitwarden"`` …), None for .env/shell keys. Metadata
    only -- credential-pool persistence may store it to explain a borrowed secret's origin but must
    never treat it as authorization to persist the raw value."""
    return _SECRET_SOURCES.get(env_var)


def get_secret_source_values(hermes_home: str | os.PathLike) -> dict[str, str]:
    """Return the external-secret value snapshot for ``hermes_home``."""
    return dict(_SECRET_SOURCE_VALUES_BY_HOME.get(str(Path(hermes_home).resolve()), {}))


def hydrate_profile_secret_sources(hermes_home: str | os.PathLike) -> dict[str, str]:
    """Resolve one profile's configured sources without mutating ``os.environ``.

    Multiplex gateways can route a first turn to a secondary profile that never ran the process-
    global dotenv startup path: resolve against a private mapping seeded from its own ``.env`` and
    record the per-home snapshot for ``build_profile_secret_scope()``. Fail-open and once-per-home
    semantics mirror ``_apply_external_secret_sources``; the result holds only values contributed
    by external sources, never plaintext ``.env`` entries.
    """
    with _SECRET_SOURCE_CACHE_LOCK:
        return _hydrate_profile_secret_sources(Path(hermes_home))


def _hydrate_profile_secret_sources(home: Path) -> dict[str, str]:
    """Locked implementation for :func:`hydrate_profile_secret_sources`."""
    home_key = str(home.resolve())
    if home_key in _APPLIED_HOMES:
        return get_secret_source_values(home)

    try:
        cfg = _load_secrets_config(home)
    except Exception:  # noqa: BLE001 — external sources must not block routing
        return {}
    if not cfg:
        return {}

    try:
        from agent.secret_scope import _is_global_env, load_env_file
        from agent.secret_sources.registry import apply_all

        local_env = {name: value for name, value in os.environ.items() if _is_global_env(name)}
        local_env.update(load_env_file(home / ".env"))
        # Mirror load_hermes_dotenv()'s .op.env bootstrap (1Password service-account token lives in
        # gitignored <home>/.op.env, not .env) or cold profiles fail 1Password hydration. .env wins.
        op_env = home / ".op.env"
        if op_env.exists():
            for _name, _value in load_env_file(op_env).items():
                local_env.setdefault(_name, _value)
        local_env["HERMES_HOME"] = str(home)
        report = apply_all(cfg, home, environ=local_env)
    except Exception:  # noqa: BLE001 — preserve fail-open startup behavior
        return {}

    if not report.sources:
        return {}

    _APPLIED_HOMES.add(home_key)
    values: dict[str, str] = {}
    for name, applied in report.provenance.items():
        value = local_env.get(name)
        if value is None:
            continue
        _SECRET_SOURCES[name] = applied.source
        values[name] = value
    if values:
        _SECRET_SOURCE_VALUES_BY_HOME[home_key] = values
    return dict(values)


def reset_secret_source_cache() -> None:
    """Forget applied homes so the next load re-pulls (tests, long-running processes after config edits)."""
    _APPLIED_HOMES.clear()
    _SECRET_SOURCES.clear()
    _SECRET_SOURCE_VALUES_BY_HOME.clear()


def format_secret_source_suffix(env_var: str) -> str:
    """``" (from Bitwarden)"``-style suffix; ``""`` for .env/shell credentials (the implicit default
    users already understand — only external sources get called out)."""
    source = get_secret_source(env_var)
    if not source:
        return ""
    if source == "bitwarden":
        return " (from Bitwarden)"
    # Registry label (e.g. "1Password"); raw name for unknown sources (uninstalled plugin, tests).
    try:
        from agent.secret_sources.registry import get_source

        registered = get_source(source)
        if registered is not None and registered.label:
            return f" (from {registered.label})"
    except Exception:  # noqa: BLE001 — label lookup must never raise
        pass
    return f" (from {source})"


def _format_offending_chars(value: str, limit: int = 3) -> str:
    """Compact ``U+XXXX ('c'), ...`` summary of non-ASCII codepoints."""
    seen: list[str] = []
    for ch in value:
        if ord(ch) > 127:
            label = f"U+{ord(ch):04X}"
            if ch.isprintable():
                label += f" ({ch!r})"
            if label not in seen:
                seen.append(label)
            if len(seen) >= limit:
                break
    return ", ".join(seen)


def _sanitize_loaded_credentials() -> None:
    """Strip non-ASCII from credential env vars (``_CREDENTIAL_SUFFIXES``) after dotenv loads, so the
    rest of the codebase never sees non-ASCII API keys."""
    for key, value in list(os.environ.items()):
        if not any(key.endswith(suffix) for suffix in _CREDENTIAL_SUFFIXES):
            continue
        if value.isascii():
            continue
        cleaned = value.encode("ascii", errors="ignore").decode("ascii")
        os.environ[key] = cleaned
        if key in _WARNED_KEYS:
            continue
        _WARNED_KEYS.add(key)
        stripped = len(value) - len(cleaned)
        detail = _format_offending_chars(value) or "non-printable"
        print(
            f"  Warning: {key} contained {stripped} non-ASCII character"
            f"{'s' if stripped != 1 else ''} ({detail}) — stripped so the "
            f"key can be sent as an HTTP header.",
            file=sys.stderr,
        )
        print(
            "  This usually means the key was copy-pasted from a PDF, "
            "rich-text editor, or web page that substituted lookalike\n"
            "  Unicode glyphs for ASCII letters. If authentication fails "
            "(e.g. \"API key not valid\"), re-copy the key from the\n"
            "  provider's dashboard and run `hermes setup` (or edit the "
            ".env file in a plain-text editor).",
            file=sys.stderr,
        )


def _load_dotenv_with_fallback(path: Path, *, override: bool) -> None:
    try:
        # utf-8-sig strips a leading UTF-8 BOM (PowerShell 5.1 / Notepad); plain "utf-8" would keep
        # U+FEFF on the first key name and silently drop it from os.environ under its canonical name.
        load_dotenv(dotenv_path=path, override=override, encoding="utf-8-sig")
    except UnicodeDecodeError:
        # utf-8-sig can't strip a BOM once we fall back to latin-1 decode.
        raw = path.read_bytes()
        if raw.startswith(codecs.BOM_UTF8):
            raw = raw[len(codecs.BOM_UTF8) :]
        load_dotenv(stream=io.StringIO(raw.decode("latin-1")), override=override)
    # API keys must be pure ASCII (httpx encodes headers as ASCII); non-ASCII typically comes from
    # keys copy-pasted out of PDFs / rich-text editors that substitute lookalike glyphs.
    _sanitize_loaded_credentials()


def _sanitize_env_file_if_needed(path: Path) -> None:
    """Pre-sanitize a .env file before python-dotenv reads it.

    Sniffs a leading BOM *before* any text decode: UTF-16 (Notepad "Unicode") is decoded and
    rewritten as clean UTF-8; UTF-32 is refused (left untouched) so we never fall through to the
    errors=replace corruption path. ``hermes_cli.config._sanitize_env_lines`` normalizes line
    endings while treating content after the first ``=`` as opaque for boundary discovery.
    """
    if not path.exists():
        return
    try:
        from hermes_cli.config import _sanitize_env_lines
    except ImportError:
        return  # early bootstrap — config module not available yet

    try:
        raw = path.read_bytes()
    except Exception:
        return

    # ORDER MATTERS: BOM_UTF32_LE (FF FE 00 00) startswith BOM_UTF16_LE (FF FE); checking UTF-16
    # first would misdetect UTF-32-LE and mangle the file.
    force_utf8_rewrite = False
    if raw.startswith(codecs.BOM_UTF32_LE) or raw.startswith(codecs.BOM_UTF32_BE):
        path_key = str(path.resolve())
        if path_key not in _WARNED_UTF32_PATHS:
            _WARNED_UTF32_PATHS.add(path_key)
            logger.warning(
                "Skipping .env sanitize for %s: UTF-32 BOM detected; "
                "leaving file untouched to avoid corruption",
                path,
            )
        return
    if raw.startswith(codecs.BOM_UTF16_LE) or raw.startswith(codecs.BOM_UTF16_BE):
        # "utf-16" uses the BOM for endianness and strips it. newline=None matches open()'s
        # universal-newlines splitting (\n/\r\n/\r only — not splitlines()'s extra Unicode
        # boundaries like U+2028), so sanitize sees the same lines as the UTF-8 path.
        try:
            with io.TextIOWrapper(io.BytesIO(raw), encoding="utf-16", newline=None) as f:
                original = f.readlines()
        except UnicodeDecodeError:
            return
        force_utf8_rewrite = True  # always rewrite UTF-16 as UTF-8 so the dotenv load sees a canonical file
    else:
        # utf-8-sig strips a UTF-8 BOM; errors=replace so embedded NULs can be stripped below.
        try:
            with open(path, encoding="utf-8-sig", errors="replace") as f:
                original = f.readlines()
        except Exception:
            return
        # errors=replace turns undecodable leading bytes into U+FFFD; persisting that would glue
        # replacement chars onto the first key name permanently. Leave the file untouched instead.
        if original and original[0].startswith("\ufffd"):
            return

    try:
        # Strip NULs so they never reach python-dotenv (os.environ raises ValueError on them); this
        # also intentionally repairs BOM-less UTF-16 (NUL-padded ASCII) into clean UTF-8.
        stripped = [line.replace("\x00", "") for line in original]
        sanitized = _sanitize_env_lines(stripped)
        if sanitized != original or force_utf8_rewrite:
            import tempfile
            fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp", prefix=".env_")
            try:
                with os.fdopen(fd, "w", encoding="utf-8") as f:
                    f.writelines(sanitized)
                    f.flush()
                    os.fsync(f.fileno())
                atomic_replace(tmp, path)
            except BaseException:
                try:
                    os.unlink(tmp)
                except OSError:
                    pass
                raise
    except Exception:
        pass  # best-effort — don't block gateway startup


def load_hermes_dotenv(
    *,
    hermes_home: str | os.PathLike | None = None,
    project_env: str | os.PathLike | None = None,
    load_external_secrets: bool = True,
) -> list[Path]:
    """Load Hermes env files, user config first: ``~/.hermes/.env`` overrides stale shell exports;
    project ``.env`` is a dev fallback that only fills gaps when the user env exists (and overrides
    stale shell vars when it does not)."""
    home_path = Path(hermes_home or os.getenv("HERMES_HOME", Path.home() / ".hermes"))

    # A multiplex gateway hosts every profile in one process: while a routed profile-home override
    # is active, copying that profile's .env into os.environ would expose its credentials to sibling
    # turns and every spawned child. An unscoped startup load is process configuration and keeps the
    # normal path. External sources still refresh, against the profile-local mapping.
    from agent.secret_scope import is_multiplex_active
    from hermes_constants import get_hermes_home_override

    if is_multiplex_active() and get_hermes_home_override() is not None:
        home_key = str(home_path.resolve())
        if home_key not in _SCOPED_SKIP_LOGGED:
            _SCOPED_SKIP_LOGGED.add(home_key)
            logger.debug(
                "multiplex: skipping process-global dotenv load for routed "
                "profile home %s (credentials resolve via the profile scope)",
                home_path,
            )
        if load_external_secrets:
            from hermes_cli import _early_recovery

            if not _early_recovery._should_skip_external_secret_sources():
                hydrate_profile_secret_sources(home_path)
        return []

    loaded: list[Path] = []
    user_env = home_path / ".env"
    project_env_path = Path(project_env) if project_env else None

    # Normalize safe formatting and remove invalid NUL bytes before parsing.
    if user_env.exists():
        _sanitize_env_file_if_needed(user_env)
    if project_env_path and project_env_path.exists():
        _sanitize_env_file_if_needed(project_env_path)

    if user_env.exists():
        _load_dotenv_with_fallback(user_env, override=True)
        loaded.append(user_env)
        # Mirror reload_env() known-key cleanup so inherited Hermes keys absent from this
        # profile's .env do not leak into the runtime.
        _clear_known_keys_missing_from_dotenv(user_env)

    # .op.env AFTER .env so .env wins, but the bootstrap OP_SERVICE_ACCOUNT_TOKEN is available to
    # apply_onepassword_secrets() even in cron / subprocesses with no shell state. It is gitignored
    # so the token never enters the committed .env; override=False lets a systemd
    # `EnvironmentFile=-…/.op.env` token take precedence.
    op_env = home_path / ".op.env"
    if op_env.exists() and not os.environ.get("OP_SERVICE_ACCOUNT_TOKEN"):
        _load_dotenv_with_fallback(op_env, override=False)

    if project_env_path and project_env_path.exists():
        _load_dotenv_with_fallback(project_env_path, override=not loaded)
        loaded.append(project_env_path)

    # External sources are skipped for the updater (dotenv + managed env still load):
    # 1. ``load_external_secrets=False`` — ``update`` must not import optional secret-manager libs
    #    (Bitwarden → cryptography → ``_rust.pyd``) into the process replacing that env on Windows.
    # 2. A fresh ``hermes update`` retry just completed a deferred dependency install; mapping native
    #    secret-source deps in that process would make the self-lock preflight exit 2 again.
    from hermes_cli import _early_recovery

    if load_external_secrets and not _early_recovery._should_skip_external_secret_sources():
        _apply_external_secret_sources(home_path)
    _apply_managed_env()

    # config.yaml owns terminal.* settings, but the override=True loads above let a stale
    # TERMINAL_ENV=docker in ~/.hermes/.env (older `hermes setup`) win on every reload; long-lived
    # processes (gateway per-turn reload, cron) call this repeatedly and used to flip the backend
    # back mid-session. Re-apply config.yaml's explicit terminal keys LAST, after
    # _apply_managed_env(), so the merged config (with the managed overlay) lands in the env.
    _reapply_terminal_config_bridge(home_path)

    return loaded


def _reapply_terminal_config_bridge(home_path: Path) -> None:
    """Re-assert config.yaml's explicit ``terminal.*`` keys over reloaded .env via the single shared
    bridge ``hermes_cli.config.apply_terminal_config_to_env`` (also used by terminal_tool's fallback
    and the TUI/dashboard launchers) so coverage, explicit-keys-only semantics, cwd placeholders and
    the managed-scope overlay can't drift between sites."""
    try:
        if Path(home_path).resolve() != _process_hermes_home().resolve():
            return
        from hermes_cli.config import apply_terminal_config_to_env

        apply_terminal_config_to_env(env=None)
    except Exception:  # noqa: BLE001 — early bootstrap / malformed config
        pass


def _apply_managed_env() -> None:
    """Apply the managed-scope .env last, with override, so it beats user/shell.

    Does NOT stop the agent from later mutating ``os.environ`` or ``export``-ing in a subprocess
    shell — that hard boundary is a documented v2 item (design §8.1); v1 relies on filesystem
    permissions. Fail-open: missing dir/.env is the common no-op; errors never block startup.
    """
    try:
        from hermes_cli import managed_scope

        managed_dir = managed_scope.get_managed_dir()
    except Exception:  # noqa: BLE001 — managed scope must never block startup
        return
    if managed_dir is None:
        return
    managed_env = managed_dir / ".env"
    if not managed_env.exists():
        return
    _sanitize_env_file_if_needed(managed_env)
    _load_dotenv_with_fallback(managed_env, override=True)


def _apply_external_secret_sources(home_path: Path) -> None:
    """Pull secrets from every enabled external source into env.

    Runs AFTER dotenv loads (sources use .env values to locate bootstrap tokens) but BEFORE Hermes
    reads ``os.environ`` for credentials; failures are swallowed — sources must never block startup.
    Ordering, mapped-beats-bulk precedence, first-claim-wins conflicts and provenance live in
    ``agent.secret_sources.registry.apply_all``; this wrapper owns the once-per-HERMES_HOME guard,
    the post-apply ASCII sweep, the ``_SECRET_SOURCES`` map UI surfaces read, and status lines.
    """
    home_key = str(Path(home_path).resolve())
    if home_key in _APPLIED_HOMES:
        return

    try:
        cfg = _load_secrets_config(home_path)
    except Exception:  # noqa: BLE001 — config errors must not block startup
        # Deliberately NOT marked applied: a malformed config.yaml would otherwise permanently
        # disable secret loading for this process even after the user fixes the file.
        return
    if not cfg:
        # No secrets section. Not marked applied either — the re-parse is a cheap fast_safe_load and
        # an unmarked home picks up a config change on the next load_hermes_dotenv() instead of never.
        return

    # Defer the registry import until a source is enabled — agent.secret_sources.bitwarden eagerly
    # loads cryptography._rust.pyd, which makes the Windows updater self-lock before its preflight.
    # Detect by *shape* (dict with enabled flag), not hardcoded names, so plugin/test sources pass
    # and a generic dict entry never forces the crypto load on every launch.
    any_enabled = any(isinstance(v, dict) and v.get("enabled") is True for v in cfg.values())
    if not any_enabled:
        return

    try:
        from agent.secret_sources.registry import apply_all
    except ImportError:
        return

    try:
        report = apply_all(cfg, home_path)
    except Exception:  # noqa: BLE001 — belt-and-braces; apply_all shouldn't raise
        return

    if not report.sources:
        # No source enabled: keep retrying cheaply so flipping one on mid-process takes effect.
        return

    # A real fetch attempt happened (success OR error). Mark the home now so the 3-5 import-time
    # calls per startup don't re-fetch / re-print — error retries are opt-in via
    # reset_secret_source_cache(). Marking AFTER the attempt keeps the earlier failure paths retryable.
    _APPLIED_HOMES.add(home_key)

    if report.applied_any:
        # Vault values are user-supplied and may carry the same copy-paste corruption as .env.
        _sanitize_loaded_credentials()
        # Provenance lets setup / `hermes model` label credentials "(from Bitwarden)" etc.
        values: dict[str, str] = {}
        for name, applied in report.provenance.items():
            _SECRET_SOURCES[name] = applied.source
            if name in os.environ:
                values[name] = os.environ[name]
        _SECRET_SOURCE_VALUES_BY_HOME[home_key] = values

    for src in report.sources:
        if src.applied:
            print(
                f"  {src.label}: applied {len(src.applied)} "
                f"secret{'s' if len(src.applied) != 1 else ''}",
                file=sys.stderr,
            )
        if src.result.error:
            print(f"  {src.label}: {src.result.error}", file=sys.stderr)
            hint = _remediation_hint(src.name, src.result.error_kind, cfg, scope=home_key)
            if hint:
                print(f"  {src.label}: → {hint}", file=sys.stderr)
        for warn in src.result.warnings:
            print(f"  {src.label}: {warn}", file=sys.stderr)
    for conflict in report.conflicts:
        print(f"  Secret sources: {conflict}", file=sys.stderr)


def _remediation_hint(source_name: str, error_kind, secrets_cfg: dict, *, scope: str | None = None) -> str:
    """The failed source's one-line fix-it hint. remediation() is a pure mapping and shouldn't raise,
    but a plugin source could — and startup must never break on a status line."""
    try:
        from agent.secret_sources.registry import get_source

        source = get_source(source_name, scope=scope)
        if source is None:
            return ""
        src_cfg = secrets_cfg.get(source_name)
        src_cfg = src_cfg if isinstance(src_cfg, dict) else {}
        return str(source.remediation(error_kind, src_cfg) or "").strip()
    except Exception:  # noqa: BLE001 — hints must never block startup
        return ""


def _load_secrets_config(home_path: Path) -> dict:
    """Read just the ``secrets:`` section of config.yaml, isolated from the main config loader so a
    malformed config can't take down dotenv loading entirely."""
    config_path = home_path / "config.yaml"
    if not config_path.exists():
        return {}
    # Prefer the shared (mtime, size)-keyed raw-config cache: this is the first config.yaml read in
    # a normal startup, so populating it lets main.py's early bridge and hermes_logging reuse one
    # parse instead of 3-4. Falls back to a direct isolated parse (the shared reader also swallows
    # parse errors -> {}).
    if home_path == _process_hermes_home():
        try:
            from hermes_cli.config import read_raw_config

            data = read_raw_config() or {}
            return data.get("secrets") or {}
        except Exception:
            pass
    try:
        with open(config_path, "r", encoding="utf-8") as f:
            data = fast_safe_load(f) or {}
    except Exception:  # noqa: BLE001
        return {}
    return data.get("secrets") or {}


def _process_hermes_home() -> Path:
    """The HERMES_HOME the shared config cache is keyed to."""
    try:
        from hermes_constants import get_hermes_home

        return get_hermes_home()
    except Exception:
        return Path.home() / ".hermes"