"""File passthrough registry for remote terminal backends (Docker, Modal, SSH).

Sandboxes start with no host files; this module tells them which credential files
(skill ``required_credential_files`` + ``terminal.credential_files`` config), skill
dirs, and host cache dirs to mount or sync in, at creation and before each command.
"""

from __future__ import annotations

import logging
import os
import posixpath
import stat
import time
from contextvars import ContextVar
from pathlib import Path
from typing import Callable, Dict, Iterator, List, Optional, Tuple

from hermes_cli.config import cfg_get
from hermes_constants import get_hermes_dir, get_hermes_home

from agent.provider_media import GENERATED_SUBDIR, MEDIA_CACHE_MAX_AGE_HOURS
from agent.skill_utils import EXCLUDED_SKILL_DIRS

try:  # pragma: no cover - exercised via the fail-closed test below
    from agent.file_safety import get_read_block_error
except ImportError:  # noqa: F401 - sentinel consumed in register_credential_file
    get_read_block_error = None  # type: ignore[assignment]

logger = logging.getLogger(__name__)

# Session-scoped registry; ContextVar prevents cross-session bleed in the gateway.
_registered_files_var: ContextVar[Dict[str, str]] = ContextVar("_registered_files")

# Cache for config-based file list, one entry per profile home (tests reset it).
_config_files: Dict[str, List[Dict[str, str]]] = {}
# Reused across calls so sanitized skill copies don't accumulate.
_safe_skills_tempdir: Path | None = None


def _get_registered() -> Dict[str, str]:
    val = _registered_files_var.get(None)
    if val is None:
        _registered_files_var.set(val := {})
    return val


def _mount(host_path: Path | str, container_path: str) -> Dict[str, str]:
    return {"host_path": str(host_path), "container_path": container_path}


def _contained_host_path(rel: str, hermes_home: Path, abs_msg: str, traversal_msg: str) -> Optional[Path]:
    """Resolve *rel* under HERMES_HOME, refusing absolute paths and escapes."""
    if os.path.isabs(rel):
        logger.warning(abs_msg, rel)
        return None
    host_path = hermes_home / rel
    from tools.path_security import validate_within_dir  # resolves symlinks and ``..`` before checking

    if containment_error := validate_within_dir(host_path, hermes_home):
        logger.warning(traversal_msg, rel, containment_error)
        return None
    return host_path.resolve()


def register_credential_file(relative_path: str, container_base: str = "/root/.hermes") -> bool:
    """Register a HERMES_HOME-relative credential file for mounting; True if it exists and was registered.

    Rejects absolute paths and traversal out of HERMES_HOME. Containment alone is not
    enough: HERMES_HOME holds the MASTER stores (``.env``, ``auth.json``, ``mcp-tokens/``),
    which are refused via the canonical read deny-list so the mount surface cannot hand a
    skill what the read surface denies. Fails CLOSED (logged) if the guard is unavailable or raises.
    """
    resolved = _contained_host_path(
        relative_path, get_hermes_home(),
        "credential_files: rejected absolute path %r (must be relative to HERMES_HOME)",
        "credential_files: rejected path traversal %r (%s)")
    if resolved is None:
        return False
    if not resolved.is_file():
        logger.debug("credential_files: skipping %s (not found)", resolved)
        return False
    # Master credential stores are never mountable, even though they sit inside HERMES_HOME and therefore
    # pass the containment check above. Fails CLOSED: if the canonical guard can't be consulted we refuse
    # the mount rather than risk bind-mounting auth.json into a sandbox. The import lives at module top (no
    # circular-import concern — file_safety is stdlib-only); the sentinel + logger.exception keep guard
    # failures debuggable instead of silently swallowed (#67665).
    if get_read_block_error is None:
        logger.error("credential_files: refusing %r — agent.file_safety could not be "
                     "imported, so the master-store deny-list cannot be consulted", relative_path)
        return False
    try:
        denied = get_read_block_error(str(resolved))
    except Exception:
        logger.exception("credential_files: refusing %r — read guard raised", relative_path)
        return False
    if denied:
        logger.warning("credential_files: refused %r — it is a credential store the agent "
                       "is denied from reading; a skill may mount its own service token, "
                       "not the master key files", relative_path)
        return False

    container_path = f"{container_base.rstrip('/')}/{relative_path}"
    _get_registered()[container_path] = str(resolved)
    logger.debug("credential_files: registered %s -> %s", resolved, container_path)
    return True


def register_credential_files(entries: list, container_base: str = "/root/.hermes") -> List[str]:
    """Register skill-frontmatter entries (str or dict with ``path``); return missing paths."""
    missing = []
    for entry in entries:
        if isinstance(entry, dict):
            entry = entry.get("path") or entry.get("name") or ""
        elif not isinstance(entry, str):
            continue
        rel_path = entry.strip()
        if rel_path and not register_credential_file(rel_path, container_base):
            missing.append(rel_path)
    return missing


def _load_config_files() -> List[Dict[str, str]]:
    """Load ``terminal.credential_files`` from config.yaml (cached per profile home: the
    multiplexed gateway must never mount the launch profile's credential files into a
    secondary profile's sandbox)."""
    from hermes_constants import hermes_home_key
    home_key = hermes_home_key()
    cached = _config_files.get(home_key)
    if cached is not None:
        return cached

    result: List[Dict[str, str]] = []
    try:
        from hermes_cli.config import read_raw_config
        hermes_home = get_hermes_home()
        cred_files = cfg_get(read_raw_config(), "terminal", "credential_files")
        for item in cred_files if isinstance(cred_files, list) else []:
            rel = item.strip() if isinstance(item, str) else ""
            if not rel:
                continue
            resolved_path = _contained_host_path(
                rel, hermes_home,
                "credential_files: rejected absolute config path %r",
                "credential_files: rejected config path traversal %r (%s)")
            if resolved_path is not None and resolved_path.is_file():
                result.append(_mount(resolved_path, f"/root/.hermes/{rel}"))
    except Exception as e:
        logger.warning("Could not read terminal.credential_files from config: %s", e)

    _config_files[home_key] = result
    return result


def get_credential_file_mounts() -> List[Dict[str, str]]:
    """Skill-registered + config credential files as ``host_path``/``container_path`` dicts (re-checked for existence)."""
    mounts = {cp: hp for cp, hp in _get_registered().items() if Path(hp).is_file()}
    for entry in _load_config_files():
        cp, hp = entry["container_path"], entry["host_path"]
        if cp not in mounts and Path(hp).is_file():
            mounts[cp] = hp
    return [_mount(hp, cp) for cp, hp in mounts.items()]


# --- Skills directory mounts ---

def _skill_dir_roots(container_base: str) -> Iterator[Tuple[Path, str]]:
    """Yield ``(host_dir, container_root)`` for every existing skills directory.

    Local skills mount at ``<base>/skills``, external at ``<base>/external_skills/<i>``, trusted
    project-local at ``<base>/project_skills/<i>`` (own namespace so paths stay stable if external_dirs change).
    """
    base = container_base.rstrip("/")
    skills_dir = get_hermes_home() / "skills"
    if skills_dir.is_dir():
        yield skills_dir, f"{base}/skills"
    try:
        from agent.skill_utils import get_external_skills_dirs, get_project_skills_dirs
    except ImportError:
        return
    for label, dirs in (("external_skills", get_external_skills_dirs()), ("project_skills", get_project_skills_dirs())):
        yield from ((d, f"{base}/{label}/{idx}") for idx, d in enumerate(dirs) if d.is_dir())


def _walk_skill_tree(root: Path) -> Iterator[Tuple[Path, List[Path]]]:
    """Yield ``(dir, regular_non_symlink_files)`` for every directory a sandbox should receive.

    Prunes ``EXCLUDED_SKILL_DIRS`` *before* descending so bookkeeping/dependency trees (``.hub``,
    ``.archive``, ``.curator_backups``, ``node_modules``, ``.git``, ...) the remote agent never reads
    are never even walked; sync thus agrees with discovery on what is skill content. Deliberately
    not ``is_excluded_skill_path()``: that also prunes ``references/``, ``templates/``, ``assets/``,
    ``scripts/`` — progressive-disclosure files and bundled scripts the sandbox does execute.
    """
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDED_SKILL_DIRS)
        base = Path(dirpath)
        yield base, [f for f in (base / n for n in filenames) if not f.is_symlink() and f.is_file()]


def get_skills_directory_mount(container_base: str = "/root/.hermes") -> list[Dict[str, str]]:
    """Directory mount entries for all skill dirs (local + external + project).

    Bind mounts follow symlinks, so a dir containing any symlink is replaced by a sanitized
    temp copy (regular files only); symlink-free dirs are returned directly, zero overhead.
    """
    return [_mount(_safe_skills_path(d), cp) for d, cp in _skill_dir_roots(container_base)]


def _safe_skills_path(skills_dir: Path) -> str:
    """Return *skills_dir* if symlink-free, else a sanitized temp copy (same exclusions as sync)."""
    global _safe_skills_tempdir

    symlinks = [p for p in skills_dir.rglob("*") if p.is_symlink()]
    if not symlinks:
        return str(skills_dir)
    for link in symlinks:
        logger.warning("credential_files: skipping symlink in skills dir: %s -> %s", link, os.readlink(link))

    import atexit
    import shutil
    import tempfile

    if _safe_skills_tempdir and _safe_skills_tempdir.is_dir():
        shutil.rmtree(_safe_skills_tempdir, ignore_errors=True)
    safe_dir = _safe_skills_tempdir = Path(tempfile.mkdtemp(prefix="hermes-skills-safe-"))

    for base, files in _walk_skill_tree(skills_dir):
        (safe_dir / base.relative_to(skills_dir)).mkdir(parents=True, exist_ok=True)
        for item in files:
            shutil.copy2(str(item), str(safe_dir / item.relative_to(skills_dir)))

    atexit.register(lambda: safe_dir.is_dir() and shutil.rmtree(safe_dir, ignore_errors=True))
    logger.info("credential_files: created symlink-safe skills copy at %s", safe_dir)
    return str(safe_dir)


def iter_skills_files(container_base: str = "/root/.hermes") -> List[Dict[str, str]]:
    """Per-file entries for all skills files (for backends that upload individually)."""
    return [_mount(item, f"{container_root}/{item.relative_to(host_dir).as_posix()}")
            for host_dir, container_root in _skill_dir_roots(container_base)
            for _base, files in _walk_skill_tree(host_dir) for item in files]


# --- Cache directory mounts (documents, images, audio, videos, screenshots) ---

# (new_subpath, old_name) pairs matching hermes_constants.get_hermes_dir().
_GENERATED_CACHE = f"cache/{GENERATED_SUBDIR}"
_CACHE_DIRS: list[tuple[str, str]] = [
    ("cache/documents", "document_cache"),
    ("cache/images", "image_cache"),
    ("cache/audio", "audio_cache"),
    ("cache/videos", "video_cache"),
    ("cache/screenshots", "browser_screenshots"),
    ("cache/web", "web_cache"),
    ("cache/delegation", "delegation_cache"),
    ("cache/spillover", "cache/spillover"),  # oversized tool results; host side is canonical
    # Unswept generated image/video deliverables (#126445) need their own mount/sync
    # entry or remote backends never see them. No legacy alias exists.
    (_GENERATED_CACHE, _GENERATED_CACHE),
    # Flat top-level desktop staging dirs (tui_gateway attach RPCs; no legacy alias),
    # mounted so vision/file tools in sandboxes reach uploads and dropped files.
    # Mount it so vision can reach uploads inside sandbox containers (#69575). No legacy alias exists, so
    # both tuple slots are ``images``.
    ("images", "images"),
    # Mount it so the agent's file tools can read dropped binaries (zip/pdf/...) from inside sandbox
    # containers instead of dangling host paths (#76577).
    ("attachments", "attachments"),
    # Desktop stages a large plain-text paste as a `.txt` under this Hermes-managed dir
    # (apps/desktop/electron/composer-paste.ts; `COMPOSER_PASTES_DIRNAME` in
    # agent/context_references.py) and attaches it as `@file:`. Without a mount/sync
    # entry, remote execution backends (ssh/daytona/vercel_sandbox) never received the
    # bytes and `to_agent_visible_cache_path` left the gateway-host path dangling on
    # the remote host (#110174). No legacy alias, so both tuple slots match.
    ("composer-pastes", "composer-pastes"),
]


def _cache_dir_roots(container_base: str, *, create_missing: bool) -> Iterator[Tuple[Path, str]]:
    """Yield ``(host_dir, container_root)`` per cache dir; always maps to the *new* container layout."""
    base = container_base.rstrip("/")
    for new_subpath, old_name in _CACHE_DIRS:
        host_dir = get_hermes_dir(new_subpath, old_name)
        if not host_dir.is_dir():
            if not create_missing:
                continue
            # Docker snapshots this list at container CREATION, so a dir appearing later
            # would dangle for the container's life: create it now (empty bind mount is free).
            # get_hermes_dir already picked new-vs-legacy, so this can't shadow a legacy dir.
            try:
                # Create missing staging dirs instead of skipping them: Docker snapshots this mount list at
                # container CREATION, so a dir that appears later (first desktop attachment, first clipboard
                # image) would dangle for the whole life of a persistent container (#76577). An empty
                # bind-mounted dir costs nothing; a missing mount costs the feature. get_hermes_dir()
                # already resolved new-vs-legacy layout, so creating its answer cannot shadow a populated
                # legacy dir.
                host_dir.mkdir(parents=True, exist_ok=True)
            except OSError:
                continue  # unwritable home (tests, RO mounts) — skip as before
        yield host_dir, f"{base}/{new_subpath}"


def get_cache_directory_mounts(container_base: str = "/root/.hermes") -> List[Dict[str, str]]:
    """Bind-mount entries for each cache directory (host layout via ``get_hermes_dir``)."""
    return [_mount(h, c) for h, c in _cache_dir_roots(container_base, create_missing=True)]


def _remap_cache_path(path: str, container_base: str, src: str, dst: str, join: Callable[[str, Path], str]) -> Optional[str]:
    """Translate *path* from the *src* side of a cache mount to its *dst* side; None if unmounted."""
    for mount in get_cache_directory_mounts(container_base=container_base):
        if Path(path).is_relative_to(mount[src]):
            return join(mount[dst], Path(path).relative_to(mount[src]))
    return None


def map_cache_path_to_container(host_path: str, container_base: str = "/root/.hermes") -> Optional[str]:
    """POSIX container path for a host path under an auto-mounted cache dir, else None.

    Also matches through symlinks: ``@file:`` expansion hands over RESOLVED paths while the mount roots keep
    HERMES_HOME's configured spelling, so a symlinked home (``~/.hermes`` -> dotfiles, macOS ``/var`` ->
    ``/private/var``) left a staged attachment's host path in front of the sandboxed agent (#103147)."""
    def join(root: str, rel: Path) -> str:
        return posixpath.join(root, rel.as_posix())

    mapped = _remap_cache_path(host_path, container_base, "host_path", "container_path", join)
    if mapped is not None:
        return mapped
    try:
        real = Path(host_path).resolve()
        for mount in get_cache_directory_mounts(container_base=container_base):
            root = Path(mount["host_path"]).resolve()
            if real.is_relative_to(root):
                return join(mount["container_path"], real.relative_to(root))
    except (OSError, RuntimeError):
        pass
    return None


def from_agent_visible_cache_path(container_path: str, container_base: str = "/root/.hermes") -> str:
    """Inverse of :func:`to_agent_visible_cache_path`; unchanged unless Docker + cache dir."""
    if _terminal_backend() != "docker":
        return container_path
    mapped = _remap_cache_path(container_path, container_base, "container_path", "host_path", lambda root, rel: str(Path(root) / rel))
    return mapped if mapped is not None else container_path


# Backends whose file-sync lands under the remote home: ``~/.hermes`` is
# expanded by the remote shell, so it resolves regardless of the actual home.
_HOME_RELATIVE_BACKENDS = frozenset({"ssh", "daytona", "vercel_sandbox"})


def _terminal_backend() -> str:
    """Active ``TERMINAL_ENV`` through the per-turn terminal scope (a routed multiplex profile's
    backend, never the launch profile's process env)."""
    from tools.terminal_scope import terminal_env
    return (terminal_env("TERMINAL_ENV") or "local").strip().lower()


def to_agent_visible_cache_path(host_path: str, container_base: str = "/root/.hermes") -> str:
    """Translate a host cache path to where the active backend (TERMINAL_ENV) sees it.

    Mirrors ``_agent_cache_base_for_env`` in tools/image_generation_tool.py: docker/modal mount at
    ``/root/.hermes``; ssh/daytona/vercel_sandbox under ``~/.hermes``; plugin backends declare
    ``cache_path_base`` (None = host paths stay correct); local/singularity/unknown unchanged
    (Apptainer auto-binds the host home, so translation would dangle).

    * docker / modal — bind-mounted (docker) or per-file-synced (modal) at ``/root/.hermes`` (the
    *container_base* default). * ssh / daytona / vercel_sandbox — file-synced under the remote user's home;
    ``~/.hermes`` is shell-expanded by the remote shell, so tool commands resolve it regardless of the
    actual remote home. Previously these backends synced the bytes but still rendered the dangling host path
    (#76577 gap).
    """
    backend = _terminal_backend()
    if backend in _HOME_RELATIVE_BACKENDS:
        container_base = "~/.hermes"
    elif backend not in ("docker", "modal"):
        try:
            from agent.terminal_env_registry import provider_flag
            plugin_base = provider_flag(backend, "cache_path_base", None)
        except Exception:
            plugin_base = None
        if not plugin_base:
            return host_path
        container_base = str(plugin_base)

    mapped = map_cache_path_to_container(host_path, container_base=container_base)
    return mapped if mapped is not None else host_path


def iter_cache_files(container_base: str = "/root/.hermes") -> List[Dict[str, str]]:
    """Per-file cache entries (Modal upload/resync); skips symlinks. ``cache/generated`` is
    never swept, so only its files from the last ``MEDIA_CACHE_MAX_AGE_HOURS`` are synced —
    otherwise every remote sync would re-walk and upload the whole generation history."""
    generated_cutoff = time.time() - MEDIA_CACHE_MAX_AGE_HOURS * 3600
    gen_root = f"{container_base.rstrip('/')}/{_GENERATED_CACHE}"
    entries: List[Dict[str, str]] = []
    for host_dir, root in _cache_dir_roots(container_base, create_missing=False):
        for item in host_dir.rglob("*"):
            try:
                st = item.lstat()
            except OSError:
                continue
            if stat.S_ISREG(st.st_mode) and (root != gen_root or st.st_mtime >= generated_cutoff):
                entries.append(_mount(item, f"{root}/{item.relative_to(host_dir)}"))
    return entries


def clear_credential_files() -> None:
    """Reset the skill-scoped registry (e.g. on session reset)."""
    _get_registered().clear()