"""@-reference expansion (``@file:``, ``@folder:``, ``@diff``, ``@git:``, ``@url:`` + plugin prefixes)."""

from __future__ import annotations

import asyncio
import inspect
import json
import mimetypes
import os
import re
import subprocess
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Awaitable, Callable

from agent.model_metadata import CHARS_PER_TOKEN, estimate_tokens_rough
from hermes_cli._subprocess_compat import IS_WINDOWS, harden_git_argv, noninteractive_git_env, windows_hide_flags
from hermes_cli.sizefmt import format_bytes

# ── Plugin context-reference provider API ────────────────────────────────────

# --------------------------------------------------------------------------- Plugin context-reference
# provider API (Issue #26193) ---------------------------------------------------------------------------
BUILTIN_PREFIXES = frozenset({"diff", "staged", "file", "folder", "git", "url"})

_context_reference_providers: dict[str, "ContextReferenceProvider"] = {}


class ContextCompletionItem:
    """A single autocomplete result from a context reference provider."""

    __slots__ = ("text", "display", "meta")

    def __init__(self, text: str, display: str = "", meta: str = "") -> None:
        self.text = text
        self.display = display or text
        self.meta = meta


class ContextReferenceProvider(ABC):
    """Base class for plugin @-prefix providers, registered via ``PluginContext.register_context_reference()``."""

    prefix: str = ""  # e.g. "issue", "channel", "doc"
    description: str = ""  # shown in autocomplete meta column

    @abstractmethod
    async def autocomplete(self, query: str, *, limit: int = 10) -> list[ContextCompletionItem]:
        """Return autocomplete items for the given query string."""

    @abstractmethod
    async def expand(self, target: str) -> str | None:
        """Expand *target* to prompt content.  Return ``None`` to skip."""


def register_context_reference_provider(provider: ContextReferenceProvider) -> None:
    """Register a plugin context reference provider."""
    if not isinstance(provider, ContextReferenceProvider):
        raise TypeError("provider must be a ContextReferenceProvider instance")
    prefix = provider.prefix.lower().strip()
    if not prefix:
        raise ValueError("prefix must be a non-empty string")
    if prefix in BUILTIN_PREFIXES:
        raise ValueError(f"prefix '{prefix}' is reserved for built-in references")
    if prefix in _context_reference_providers:
        raise ValueError(f"prefix '{prefix}' is already registered")
    _context_reference_providers[prefix] = provider


def get_context_reference_providers() -> dict[str, ContextReferenceProvider]:
    """Return a snapshot of all registered plugin providers."""
    return dict(_context_reference_providers)


_QUOTED_REFERENCE_VALUE = r'(?:`[^`\n]+`|"[^"\n]+"|\'[^\'\n]+\')'
REFERENCE_PATTERN = re.compile(
    rf"(?<![\w/])@(?:(?P<simple>diff|staged)\b|(?P<kind>file|folder|git|url):(?P<value>{_QUOTED_REFERENCE_VALUE}(?::\d+(?:-\d+)?)?|\S+))"
)
# Plugin fallback: any @<word>:<value> the built-in regex did not claim.
_PLUGIN_REFERENCE_PATTERN = re.compile(
    rf"(?<![\w/])@(?P<kind>[a-zA-Z][a-zA-Z0-9_-]*):(?P<value>{_QUOTED_REFERENCE_VALUE}(?::\d+(?:-\d+)?)?|\S+)"
)
# ``@file:`` value: quoted path or bare path, each with an optional ``:start[-end]`` range.
_FILE_VALUE_PATTERN = re.compile(
    r'^(?:(?P<quote>`|"|\')(?P<qpath>.+?)(?P=quote)|(?P<path>.+?))(?::(?P<start>\d+)(?:-(?P<end>\d+))?)?$'
)

TRAILING_PUNCTUATION = ",.;!?"
_OPENERS = {")": "(", "]": "[", "}": "{"}
_NEEDS_QUOTING = re.compile(r"""[\s()\[\]{}<>"'`]""")
_SENSITIVE_HOME_DIRS = (".ssh", ".aws", ".gnupg", ".kube", ".docker", ".azure", ".config/gh")
_SENSITIVE_HERMES_DIRS = (Path("skills") / ".hub",)
_SENSITIVE_HOME_FILES = tuple(Path(p) for p in (
    ".ssh/authorized_keys", ".ssh/id_rsa", ".ssh/id_ed25519", ".ssh/config", ".bashrc", ".zshrc",
    ".profile", ".bash_profile", ".zprofile", ".netrc", ".pgpass", ".npmrc", ".pypirc",
))
_TEXT_EXTENSIONS = (".py", ".md", ".txt", ".json", ".yaml", ".yml", ".toml", ".js", ".ts")
# Bound the work one message can force: each expanded ref reads at most a bounded prefix /
# window, and at most this many refs are expanded per message.
_MAX_EXPANDED_REFERENCES = 16
# Folder-listing line counts stop paying I/O past this size; bigger files report bytes only.
_LINE_COUNT_MAX_BYTES = 4 * 1024 * 1024
# Per-command stdout/stderr ceiling for the ``git``/``rg`` helpers; past it the child is
# killed and the nonzero returncode routes the caller to its fallback path.
_MAX_QUIET_OUTPUT_BYTES = 4 * 1024 * 1024
_OUTPUT_CAP_EXCEEDED_RETURNCODE = 137  # 128 + SIGKILL, the code a shell reports for a killed child
_FENCE_LANGUAGES = {
    ".py": "python", ".js": "javascript", ".ts": "typescript", ".tsx": "tsx", ".jsx": "jsx",
    ".json": "json", ".md": "markdown", ".sh": "bash", ".yml": "yaml", ".yaml": "yaml", ".toml": "toml",
}


@dataclass(frozen=True)
class ContextReference:
    raw: str
    kind: str
    target: str
    start: int
    end: int
    line_start: int | None = None
    line_end: int | None = None


@dataclass
class ContextReferenceResult:
    message: str
    original_message: str
    references: list[ContextReference] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    injected_tokens: int = 0
    expanded: bool = False
    blocked: bool = False


UrlFetcher = Callable[[str], str | Awaitable[str]] | None
Expansion = tuple[str | None, str | None]  # (warning, block) — exactly one side is set


def format_reference_value(value: str) -> str:
    """Quote a value so ``REFERENCE_PATTERN`` (bare alternative ``\\S+``) reads it back whole.
    Mirrors ``formatRefValue`` in the desktop's directive-text.tsx."""
    if not _NEEDS_QUOTING.search(value):
        return value
    for quote in ("`", '"', "'"):
        if quote not in value:
            return f"{quote}{value}{quote}"
    return value


def parse_context_references(message: str) -> list[ContextReference]:
    refs: list[ContextReference] = []
    if not message:
        return refs
    for match in REFERENCE_PATTERN.finditer(message):
        kind = match.group("simple") or match.group("kind")
        value = _strip_trailing_punctuation(match.group("value") or "")
        if match.group("simple"):
            target, line_start, line_end = "", None, None
        elif kind == "file":
            target, line_start, line_end = _parse_file_reference_value(value)
        else:
            target, line_start, line_end = _strip_reference_wrappers(value), None, None
        refs.append(ContextReference(match.group(0), kind, target, match.start(), match.end(), line_start, line_end))

    # Second pass: plugin-registered prefixes the built-in pattern missed.
    for match in _PLUGIN_REFERENCE_PATTERN.finditer(message) if _context_reference_providers else ():
        kind = match.group("kind")
        if kind in BUILTIN_PREFIXES or kind not in _context_reference_providers:
            continue
        if any(r.kind == kind and r.start == match.start() for r in refs):
            continue
        target = _strip_reference_wrappers(_strip_trailing_punctuation(match.group("value") or ""))
        refs.append(ContextReference(match.group(0), kind, target, match.start(), match.end()))
    return refs


def preprocess_context_references(
    message: str, *, cwd: str | Path, context_length: int, url_fetcher: UrlFetcher = None,
    allowed_root: str | Path | None = None,
) -> ContextReferenceResult:
    """Sync wrapper; safe both without a loop (CLI) and inside a running loop (gateway)."""
    coro = preprocess_context_references_async(
        message, cwd=cwd, context_length=context_length, url_fetcher=url_fetcher, allowed_root=allowed_root
    )
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    import concurrent.futures
    import contextvars
    # The side thread starts with an empty Context: without the caller's copy the served profile's
    # HERMES_HOME override is lost and the credential-path guard checks the launch profile's .env.
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
        return pool.submit(contextvars.copy_context().run, asyncio.run, coro).result()


async def preprocess_context_references_async(
    message: str, *, cwd: str | Path, context_length: int, url_fetcher: UrlFetcher = None,
    allowed_root: str | Path | None = None,
) -> ContextReferenceResult:
    refs = parse_context_references(message)
    if not refs:
        return ContextReferenceResult(message=message, original_message=message)
    cwd_path = Path(cwd).expanduser().resolve()
    # Default root = cwd so @ references cannot escape the workspace unless a caller widens it.
    allowed_root_path = Path(allowed_root).expanduser().resolve() if allowed_root is not None else cwd_path
    # Expand concurrently (each ref is independent; several @url: refs would otherwise
    # serialize web_extract round-trips). gather preserves order, so warnings/blocks
    # are assembled in ref order; the token-budget check runs once afterwards.
    hard_limit = max(1, int(context_length * 0.50))
    soft_limit = max(1, int(context_length * 0.25))
    tasks = (
        _expand_reference(ref, cwd_path, url_fetcher=url_fetcher, allowed_root=allowed_root_path,
                          max_inline_tokens=hard_limit)
        for ref in refs[:_MAX_EXPANDED_REFERENCES]
    )
    expanded = await asyncio.gather(*tasks)
    warnings = [warning for warning, _ in expanded if warning]
    warnings.extend(
        f"{ref.raw}: not expanded (maximum {_MAX_EXPANDED_REFERENCES} @-references per message)"
        for ref in refs[_MAX_EXPANDED_REFERENCES:]
    )
    blocks = [block for _, block in expanded if block]
    injected_tokens = sum(estimate_tokens_rough(block) for block in blocks)
    result = ContextReferenceResult(
        message=message, original_message=message, references=refs, warnings=warnings, injected_tokens=injected_tokens
    )

    if injected_tokens > hard_limit:
        warnings.append(f"@ context injection refused: {injected_tokens} tokens exceeds the 50% hard limit ({hard_limit}).")
        result.blocked = True
        return result
    if injected_tokens > soft_limit:
        warnings.append(f"@ context injection warning: {injected_tokens} tokens exceeds the 25% soft limit ({soft_limit}).")

    # The `@file:`/`@folder:` tokens stay where the user typed them: the token IS the
    # reference (clients render it as an inline chip); stripping it left a hole in the
    # sentence and forced the desktop to re-derive refs from the attached block.
    final = message
    if warnings:
        final = f"{final}\n\n--- Context Warnings ---\n" + "\n".join(f"- {warning}" for warning in warnings)
    if blocks:
        final = f"{final}\n\n--- Attached Context ---\n\n" + "\n\n".join(blocks)
    result.message = final.strip()
    result.expanded = bool(blocks or warnings)
    return result


# Git-backed reference kinds -> f(ref) -> git argv (the label is "git " + argv).
_GIT_REFERENCE_ARGS: dict[str, Callable[[ContextReference], list[str]]] = {
    "diff": lambda ref: ["diff"],
    "staged": lambda ref: ["diff", "--staged"],
    "git": lambda ref: ["log", f"-{max(1, min(int(ref.target or '1'), 10))}", "-p"],
}


async def _expand_reference(
    ref: ContextReference, cwd: Path, *, url_fetcher: UrlFetcher = None, allowed_root: Path | None = None,
    max_inline_tokens: int | None = None,
) -> Expansion:
    try:
        if ref.kind in ("file", "folder"):
            return _expand_path_reference(ref, cwd, allowed_root=allowed_root, max_inline_tokens=max_inline_tokens)
        if ref.kind in _GIT_REFERENCE_ARGS:
            git_args = _GIT_REFERENCE_ARGS[ref.kind](ref)
            return _expand_git_reference(ref, cwd, git_args, "git " + " ".join(git_args))
        if ref.kind == "url":
            content = await _fetch_url_content(ref.target, url_fetcher=url_fetcher)
            if not content:
                return f"{ref.raw}: no content extracted", None
            return None, f"🌐 {ref.raw} ({estimate_tokens_rough(content)} tokens)\n{content}"
    except Exception as exc:
        return f"{ref.raw}: {exc}", None
    provider = _context_reference_providers.get(ref.kind)
    if provider is not None:
        try:
            plugin_content = await provider.expand(ref.target)
            if plugin_content is not None:
                return None, f"📌 {ref.raw} ({estimate_tokens_rough(plugin_content)} tokens)\n{plugin_content}"
        except Exception as exc:
            return f"{ref.raw}: plugin expansion error: {exc}", None
    return f"{ref.raw}: unsupported reference type", None


def _expand_path_reference(ref: ContextReference, cwd: Path, *, allowed_root: Path | None = None,
                           max_inline_tokens: int | None = None) -> Expansion:
    """``@file:`` / ``@folder:``: resolve, allow-check, then inline text / binary stub / listing."""
    is_folder = ref.kind == "folder"
    path = _resolve_path(cwd, ref.target, allowed_root=allowed_root)
    _ensure_reference_path_allowed(path)
    if not path.exists():
        return f"{ref.raw}: {ref.kind} not found", None
    if not (path.is_dir() if is_folder else path.is_file()):
        return f"{ref.raw}: path is not a {ref.kind}", None
    if is_folder:
        listing = _build_folder_listing(path, cwd, display_base=allowed_root)
        return None, f"📁 {ref.raw} ({estimate_tokens_rough(listing)} tokens)\n{listing}"
    if _is_binary_file(path):
        # A bare "not supported" warning was a dead end (the model gave up); the file IS
        # on disk where the agent's tools run, so hand it an actionable block instead.
        return None, _binary_reference_block(ref, path)
    if ref.line_start is not None:
        # A ranged ref wants a slice, not the file: stream to the window so a GB-scale
        # file serves :1-5 without being materialized. Lines are read in bounded pieces
        # (budget + 1) so even a single-line giant (minified JSON, one-line logs) cannot
        # force an unbounded string into memory; skipped lines are discarded, not kept.
        char_budget = None if max_inline_tokens is None else max_inline_tokens * CHARS_PER_TOKEN
        line_cap = None if char_budget is None else char_budget + 1

        def _next_line(fh, collect: bool, remaining: int | None = None) -> str | None:
            """One line ("" when skipping), None at EOF. While collecting, stop as soon as the
            pieces exceed ``remaining``: the caller returns the oversized block and never needs
            the rest of a giant line, so it is never materialized."""
            pieces, seen, collected = [], False, 0
            while True:
                piece = fh.readline() if line_cap is None else fh.readline(line_cap)
                if not piece:
                    break
                seen = True
                if collect:
                    pieces.append(piece)
                    collected += len(piece)
                    if remaining is not None and collected > remaining:
                        break
                if piece.endswith("\n"):
                    break
            return ("".join(pieces) if collect else "") if seen else None

        parts, total_chars = [], 0
        with path.open(encoding="utf-8") as fh:
            for _ in range(max(ref.line_start - 1, 0)):
                if _next_line(fh, collect=False) is None:
                    break
            for _ in range((ref.line_end or ref.line_start) - ref.line_start + 1):
                line = _next_line(fh, collect=True, remaining=None if char_budget is None else char_budget - total_chars)
                if line is None:
                    break
                total_chars += len(line)
                if char_budget is not None and total_chars > char_budget:
                    return None, _oversized_text_reference_block(ref, path, total_chars // CHARS_PER_TOKEN)
                parts.append(line)
        text = "".join(parts)
    else:
        # estimate_tokens_rough >= bytes/CHARS_PER_TOKEN for every encoding mix, so a
        # file past that byte ceiling is certainly oversized; refuse without reading it.
        size = path.stat().st_size
        if max_inline_tokens is not None and size > max_inline_tokens * CHARS_PER_TOKEN:
            return None, _oversized_text_reference_block(ref, path, size // CHARS_PER_TOKEN)
        text = path.read_text(encoding="utf-8")
    lang = _FENCE_LANGUAGES.get(path.suffix.lower(), "")
    text_tokens = estimate_tokens_rough(text)
    # Check BEFORE building the fenced block: an oversized file is not going to be
    # inlined, so don't build a second MB-scale string just to discard it.
    if max_inline_tokens is not None and text_tokens > max_inline_tokens:
        # One oversized file used to poison the aggregate check and refuse the whole
        # turn (#61987); the file stays readable via the agent's tools instead. The
        # block alone carries the message (same shape as the binary path) — a warning
        # would duplicate it in "--- Context Warnings ---".
        return None, _oversized_text_reference_block(ref, path, text_tokens)
    return None, f"📄 {ref.raw} ({text_tokens} tokens)\n```{lang}\n{text}\n```"


def _run_quiet(cmd: list[str], cwd: Path, timeout: int, env: dict | None = None) -> subprocess.CompletedProcess:
    """Captured text output with bounded pipes, no stdin, and no console flash on Windows.

    ``capture_output=True`` buffers the child's entire stdout — an unbounded ``git diff``
    or ``rg --files`` would materialize a hostile-size stream in memory. Each pipe is
    drained up to ``_MAX_QUIET_OUTPUT_BYTES``; past it the child is killed and the result
    carries a nonzero returncode that routes callers to their existing fallback paths. Set
    explicitly: a child that flushed everything and exited 0 before the drain thread crossed
    the cap is not killable and would otherwise report success with truncated output.
    """
    popen_kwargs: dict = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {}
    proc = subprocess.Popen(
        cmd, cwd=cwd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        **popen_kwargs, **({} if env is None else {"env": env}))

    truncated = threading.Event()

    def _drain(stream, sink: list[bytes]) -> None:
        total = 0
        while True:
            chunk = stream.read1(1 << 16)
            if not chunk:
                return
            total += len(chunk)
            if total <= _MAX_QUIET_OUTPUT_BYTES:
                sink.append(chunk)
            else:
                truncated.set()
                proc.kill()
                return

    sinks: list[list[bytes]] = [[], []]
    threads = [
        threading.Thread(target=_drain, args=(stream, sink), daemon=True)
        for stream, sink in zip((proc.stdout, proc.stderr), sinks)
    ]
    for thread in threads:
        thread.start()
    try:
        proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait()
        raise subprocess.TimeoutExpired(cmd, timeout)
    for thread in threads:
        thread.join()
    stdout, stderr = (b"".join(sink).decode("utf-8", "replace") for sink in sinks)
    returncode = proc.returncode
    if truncated.is_set() and returncode == 0:
        returncode = _OUTPUT_CAP_EXCEEDED_RETURNCODE
    return subprocess.CompletedProcess(cmd, returncode, stdout, stderr)


def _expand_git_reference(ref: ContextReference, cwd: Path, args: list[str], label: str) -> Expansion:
    try:
        # Repo-supplied config/attributes must never execute code (GHSA-7x36-8jrh-v4pw).
        result = _run_quiet(["git", *harden_git_argv(args)], cwd, 30, env=noninteractive_git_env())
    except subprocess.TimeoutExpired:
        return f"{ref.raw}: git command timed out (30s)", None
    if result.returncode != 0:
        return f"{ref.raw}: {(result.stderr or '').strip() or 'git command failed'}", None
    content = result.stdout.strip() or "(no output)"
    return None, f"🧾 {label} ({estimate_tokens_rough(content)} tokens)\n```diff\n{content}\n```"


async def _fetch_url_content(url: str, *, url_fetcher: UrlFetcher = None) -> str:
    content = (url_fetcher or _default_url_fetcher)(url)
    if inspect.isawaitable(content):
        content = await content
    return str(content or "").strip()


async def _default_url_fetcher(url: str) -> str:
    from tools.web_tools import web_extract_tool
    docs = json.loads(await web_extract_tool([url], format="markdown")).get("results", [])
    return str(docs[0].get("content") or docs[0].get("raw_content") or "").strip() if docs else ""


def _is_under(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
    except ValueError:
        return False
    return True


# Desktop persists a large plain-text paste as a `.txt` under this Hermes-managed
# directory (apps/desktop/electron/composer-paste.ts) and attaches it as `@file:`.
# The chat's cwd is rarely an ancestor of it, so it is the one anchored root the
# workspace guard admits besides `allowed_root` itself.
COMPOSER_PASTES_DIRNAME = "composer-pastes"


def _composer_paste_roots() -> list[Path]:
    from agent.file_safety import _hermes_dirs
    return [hermes_dir / COMPOSER_PASTES_DIRNAME for hermes_dir in _hermes_dirs()]


def _resolve_path(cwd: Path, target: str, *, allowed_root: Path | None = None) -> Path:
    from agent.file_safety import is_nt_namespace_path
    if is_nt_namespace_path(target):  # raw-string check: resolving such a path is the NTLM-leak trigger
        raise ValueError("path uses a Windows NT/device namespace prefix and cannot be attached")
    resolved = (cwd / Path(os.path.expanduser(target))).resolve()  # `/` keeps an absolute target as-is
    if (
        allowed_root is not None
        and not _is_under(resolved, allowed_root)
        and not any(_is_under(resolved, root) for root in _composer_paste_roots())
    ):
        raise ValueError("path is outside the allowed workspace")
    return resolved


def _ensure_reference_path_allowed(path: Path) -> None:
    """Refuse credential/internal paths. Fails CLOSED: the gateway feeds untrusted remote text here."""
    from hermes_constants import get_hermes_home
    home, hermes_home = Path(os.path.expanduser("~")).resolve(), get_hermes_home().resolve()
    blocked_exact = {home / rel for rel in _SENSITIVE_HOME_FILES} | {hermes_home / ".env"}
    blocked_dirs = [home / rel for rel in _SENSITIVE_HOME_DIRS] + [hermes_home / rel for rel in _SENSITIVE_HERMES_DIRS]
    if path in blocked_exact:
        raise ValueError("path is a sensitive credential file and cannot be attached")
    if any(_is_under(path, blocked_dir) for blocked_dir in blocked_dirs):
        raise ValueError("path is a sensitive credential or internal Hermes path and cannot be attached")
    # Anchor to the canonical read deny-list (agent/file_safety.get_read_block_error): the
    # narrow list above never caught auth.json, .anthropic_oauth.json, mcp-tokens/, webhook
    # secrets or project .env files, and it grows automatically with that deny-list.
    try:
        from agent.file_safety import get_read_block_error
        blocked = get_read_block_error(str(path)) is not None
    except ValueError:
        raise
    except Exception:
        # If the canonical lookup fails, falling through would re-open the exact hole this
        # guard closes; a spurious block is recoverable, a leaked credential is not.
        raise ValueError("path could not be verified against the credential deny-list and cannot be attached")
    if blocked:
        raise ValueError("path is a sensitive credential or internal Hermes path and cannot be attached")


def _strip_trailing_punctuation(value: str) -> str:
    stripped = value.rstrip(TRAILING_PUNCTUATION)
    # Drop unbalanced closers so "(see @file:x.py)" does not swallow the ")".
    while stripped.endswith((")", "]", "}")) and stripped.count(stripped[-1]) > stripped.count(_OPENERS[stripped[-1]]):
        stripped = stripped[:-1]
    return stripped


def _strip_reference_wrappers(value: str) -> str:
    return value[1:-1] if len(value) >= 2 and value[0] == value[-1] and value[0] in "`\"'" else value


def _parse_file_reference_value(value: str) -> tuple[str, int | None, int | None]:
    m = _FILE_VALUE_PATTERN.match(value)
    start = m and m.group("start")
    if not start:  # no line range: the whole value is the (possibly quoted) path
        return _strip_reference_wrappers(value), None, None
    return m.group("qpath") or m.group("path"), int(start), int(m.group("end") or start)


def _is_binary_file(path: Path) -> bool:
    mime = mimetypes.guess_type(path.name)[0]
    if mime and not mime.startswith("text/") and not path.name.endswith(_TEXT_EXTENSIONS):
        return True
    with path.open("rb") as fh:  # sniff only; read_bytes() materialized the whole file
        return b"\x00" in fh.read(4096)


def _build_folder_listing(path: Path, cwd: Path, limit: int = 200, display_base: Path | None = None) -> str:
    # The target may sit outside cwd when the caller widened allowed_root: show it relative to
    # cwd when possible, else relative to the allowed root, else the absolute path.
    shown: str | None = None
    for base in (cwd, display_base):
        if base is None:
            continue
        try:
            shown = f"{path.relative_to(base)}/"
            break
        except ValueError:
            continue
    if shown is None:
        shown = f"{path}/"
    lines = [shown]
    entries = _iter_visible_entries(path, cwd, limit=limit)
    for entry in entries:
        indent = "  " * max(len(entry.relative_to(path).parts) - 1, 0)
        lines.append(f"{indent}- {entry.name}/" if entry.is_dir() else f"{indent}- {entry.name} ({_file_metadata(entry)})")
    if len(entries) >= limit:
        lines.append("- ...")
    return "\n".join(lines)


def _iter_visible_entries(path: Path, cwd: Path, limit: int) -> list[Path]:
    """Files under ``path`` via ``rg --files`` (honours ignore files), else an os.walk fallback."""
    try:
        # Absolute path arg: rg echoes it as the output prefix, so results stay correct even
        # when the folder is outside cwd (a widened allowed_root target).
        rg = _run_quiet(["rg", "--files", str(path)], cwd, 10)
    except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
        rg = None
    if rg is not None and rg.returncode == 0:
        output: list[Path] = []
        seen_dirs: set[Path] = set()
        for line in [ln.strip() for ln in rg.stdout.splitlines() if ln.strip()][:limit]:
            full = cwd / Path(line)  # absolute lines pass through unchanged; defensive for relative
            for parent in full.parents:
                if parent in seen_dirs or path not in {parent, *parent.parents}:
                    continue
                seen_dirs.add(parent)
                output.append(parent)
            output.append(full)
        return sorted({p for p in output if p.exists()}, key=lambda p: (not p.is_dir(), str(p)))
    output = []
    for root, dirs, files in os.walk(path):
        dirs[:] = sorted(d for d in dirs if not d.startswith(".") and d != "__pycache__")
        files = sorted(f for f in files if not f.startswith("."))
        for name in dirs + files:
            output.append(Path(root) / name)
            if len(output) >= limit:
                return output
    return output


def _agent_visible_path(path: Path) -> str:
    # Under a container backend the host path dangles inside the sandbox: translate staged
    # files to their auto-mounted cache path; fall back to the host path (local backend /
    # translation failure). Run the idempotent TERMINAL_ENV bridge first so in-process
    # gateways that never bridged terminal.* config still see the active backend.
    try:
        from tools.terminal_tool import _ensure_terminal_env_bridged
        _ensure_terminal_env_bridged()
        from tools.credential_files import to_agent_visible_cache_path
        return to_agent_visible_cache_path(str(path))
    except Exception:
        return str(path)


def _on_disk_reference_block(ref: ContextReference, path: Path, descriptor: str, reason: str, guidance: str) -> str:
    """Shared 📎 shape: the file was not inlined, but it IS on disk where the agent's
    tools run — hand the model the path and a nudge instead of a dead-end warning."""
    try:
        size = format_bytes(path.stat().st_size)
    except OSError:
        size = "unknown size"
    return (
        f"📎 {ref.raw} ({descriptor}, {size}) — {reason} "
        f"It is available on disk at `{_agent_visible_path(path)}`. {guidance}"
    )


def _binary_reference_block(ref: ContextReference, path: Path) -> str:
    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    return _on_disk_reference_block(
        ref, path,
        descriptor=mime,
        reason="binary file, not inlined as text.",
        guidance="Use your tools to work with it (read or convert it, extract its text, "
                 "or view/render it as needed); do not tell the user the file type is unsupported.",
    )


def _oversized_text_reference_block(ref: ContextReference, path: Path, text_tokens: int) -> str:
    return _on_disk_reference_block(
        ref, path,
        descriptor=f"text file, approximately {text_tokens} tokens",
        reason="too large to inline safely.",
        guidance="Use read_file with a narrow line range, search_files, or terminal/code tools "
                 "to inspect only the relevant parts; do not load the entire file into context.",
    )


def _file_metadata(path: Path) -> str:
    try:
        size = path.stat().st_size
    except OSError:
        return "unknown size"
    # A listing line is a summary, not content: past the cap, byte size conveys the
    # same "how big is this" without a full scan per entry.
    if _is_binary_file(path) or size > _LINE_COUNT_MAX_BYTES:
        return f"{size} bytes"
    try:
        with path.open("rb") as fh:
            # UTF-8 never embeds 0x0A inside a multibyte sequence, so counting bytes
            # matches a decoded newline count while streaming instead of read_text.
            lines = sum(chunk.count(b"\n") for chunk in iter(lambda: fh.read(1 << 20), b""))
        return f"{lines + 1} lines"
    except Exception:
        return f"{size} bytes"