"""Terminal, search, and directory listing tool implementations."""

from __future__ import annotations

import json
import logging
import os
import subprocess
import threading
import time
from typing import IO, Any

from butler.execution_context import get_current_session_key
from butler.tools.interrupt import is_interrupted
from butler.tools.path_safety import (
    check_tool_path,
    default_tool_workdir,
    prepare_shell_command,
    safe_subprocess_env,
)
from butler.tools.terminal_approval import approval_allows_unsandboxed
from butler.tools.terminal_impl_ops import (
    close_pipe_safe,
    inc_terminal_metric_safe,
    is_selectable_pipe_safe,
    run_list_directory_loud,
    run_search_files_loud,
    run_subprocess_loud,
    run_terminal_with_gates_safe,
)
from butler.tools.terminal_sandbox import (
    classify_sandbox_failure,
    enrich_subprocess_env,
    format_sandbox_error_payload,
    load_terminal_sandbox_config,
    sandbox_runtime_available,
    scrub_credential_env,
    should_run_sandboxed,
    wrap_argv_with_bubblewrap,
)
from butler.defaults.env_defaults import ENABLE_TERMINAL_DEFAULT

logger = logging.getLogger(__name__)

MAX_TERMINAL_TIMEOUT_SECONDS = 120
MAX_TERMINAL_OUTPUT_CHARS = 50000


def _is_selectable_pipe(pipe: Any) -> bool:
    return bool(is_selectable_pipe_safe(pipe))


def _close_pipe(pipe: Any) -> None:
    close_pipe_safe(pipe)


def _communicate_limited(
    proc: subprocess.Popen[Any],
    *,
    timeout: int,
    max_output_chars: int,
) -> tuple[str, str, bool]:
    """Read process pipes incrementally so tool output cannot exhaust memory."""
    stdout_pipe = proc.stdout
    stderr_pipe = proc.stderr
    if not (
        stdout_pipe is not None
        and stderr_pipe is not None
        and _is_selectable_pipe(stdout_pipe)
        and _is_selectable_pipe(stderr_pipe)
    ):
        stdout, stderr = proc.communicate(timeout=timeout)
        return str(stdout or ""), str(stderr or ""), (
            len(str(stdout or "")) + len(str(stderr or "")) > max_output_chars
        )

    import selectors

    selector = selectors.DefaultSelector()
    stdout_io = stdout_pipe
    stderr_io = stderr_pipe
    selector.register(stdout_io, selectors.EVENT_READ, "stdout")
    selector.register(stderr_io, selectors.EVENT_READ, "stderr")
    stdout_parts: list[bytes] = []
    stderr_parts: list[bytes] = []
    total = 0
    truncated = False
    deadline = time.monotonic() + timeout

    try:
        while selector.get_map():
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise subprocess.TimeoutExpired(proc.args, timeout)
            events = selector.select(timeout=min(0.2, remaining))
            if not events:
                if proc.poll() is not None:
                    # Give pipes one final chance to report EOF.
                    continue
                continue
            for key, _mask in events:
                pipe = key.fileobj
                assert isinstance(pipe, IO)
                chunk = pipe.read1(8192) if hasattr(pipe, "read1") else pipe.read(8192)
                if not chunk:
                    selector.unregister(pipe)
                    continue
                if isinstance(chunk, str):
                    chunk = chunk.encode("utf-8", errors="replace")
                room = max(0, max_output_chars - total)
                if room:
                    kept = chunk[:room]
                    if key.data == "stdout":
                        stdout_parts.append(kept)
                    else:
                        stderr_parts.append(kept)
                    total += len(kept)
                if len(chunk) > room:
                    truncated = True
        proc.wait(timeout=max(0, deadline - time.monotonic()))
    finally:
        selector.close()

    stdout = b"".join(stdout_parts).decode("utf-8", errors="replace")
    stderr = b"".join(stderr_parts).decode("utf-8", errors="replace")
    return stdout, stderr, truncated


def _tool_terminal(command: str, timeout: int = 30, workdir: str | None = None, **_: Any) -> str:
    if os.getenv("BUTLER_ENABLE_TERMINAL", ENABLE_TERMINAL_DEFAULT).strip() != "1":
        return json.dumps({
            "error": "Terminal tool is disabled by default. Set BUTLER_ENABLE_TERMINAL=1 to enable restricted commands."
        })

    thread_id = threading.get_ident()
    proc: subprocess.Popen[str] | None = None
    interrupted = False
    try:
        timeout = int(timeout)
    except (TypeError, ValueError):
        return json.dumps({"error": "Terminal timeout must be an integer number of seconds"})
    if timeout < 1 or timeout > MAX_TERMINAL_TIMEOUT_SECONDS:
        return json.dumps({
            "error": f"Terminal timeout must be between 1 and {MAX_TERMINAL_TIMEOUT_SECONDS} seconds"
        })
    command_safety = prepare_shell_command(command)
    if not command_safety.allowed:
        return json.dumps({"error": command_safety.error})

    cwd_hint = str(workdir or default_tool_workdir() or "")
    cmd_text = " ".join(command_safety.argv) if command_safety.argv else command
    session_key = str(get_current_session_key() or "")

    if workdir:
        safety = check_tool_path(workdir)
    else:
        safety = default_tool_workdir()
    if not safety.allowed:
        return json.dumps({"error": safety.error})
    if not safety.path.is_dir():
        return json.dumps({"error": f"Not a directory: {workdir or safety.path}"})
    resolved_workdir = str(safety.path)
    workspace_path = safety.path

    sandbox_config = load_terminal_sandbox_config(workspace_path)
    unsandboxed = approval_allows_unsandboxed(
        cmd_text,
        cwd=resolved_workdir,
        session_key=session_key,
    )
    run_sandboxed = should_run_sandboxed(
        sandbox_config,
        unsandboxed_approved=unsandboxed,
    )
    if run_sandboxed and not sandbox_runtime_available():
        if sandbox_config.fail_if_unavailable:
            return json.dumps({
                "error": "bubblewrap (bwrap) 未安装，且 BUTLER_TERMINAL_SANDBOX_FAIL_UNAVAILABLE=1",
                "code": "SANDBOX_UNAVAILABLE",
            })
        logger.warning("terminal sandbox enabled but bwrap missing; running unsandboxed")
        run_sandboxed = False
        inc_terminal_metric_safe("terminal_sandbox_unavailable_fallback")

    exec_argv = list(command_safety.argv)
    if run_sandboxed:
        try:
            exec_argv = wrap_argv_with_bubblewrap(
                exec_argv,
                workspace=workspace_path,
                config=sandbox_config,
            )
        except RuntimeError as exc:
            return json.dumps({"error": str(exc), "code": "SANDBOX_UNAVAILABLE"})

    def _watch() -> None:
        nonlocal interrupted
        while proc is not None and proc.poll() is None:
            if is_interrupted(thread_id):
                interrupted = True
                proc.kill()
                return
            time.sleep(0.2)

    def _run_subprocess() -> str:
        nonlocal proc, interrupted
        base_env = safe_subprocess_env()
        if run_sandboxed:
            base_env = scrub_credential_env(base_env, sandbox_config)
        proc_env = enrich_subprocess_env(base_env, sandboxed=run_sandboxed)
        if run_sandboxed:
            inc_terminal_metric_safe(
                "terminal_sandbox_run",
                labels={"sandboxed": "1"},
                session_key=session_key,
            )
        proc = subprocess.Popen(
            exec_argv,
            shell=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            cwd=resolved_workdir,
            env=proc_env,
        )
        assert proc is not None
        watcher = threading.Thread(target=_watch, daemon=True)
        watcher.start()
        try:
            try:
                stdout, stderr, truncated = _communicate_limited(
                    proc,
                    timeout=timeout,
                    max_output_chars=MAX_TERMINAL_OUTPUT_CHARS,
                )
            except subprocess.TimeoutExpired:
                proc.kill()
                try:
                    proc.wait(timeout=1)
                finally:
                    _close_pipe(proc.stdout)
                    _close_pipe(proc.stderr)
                return json.dumps({"error": f"Command timed out after {timeout}s"})
            if interrupted:
                return json.dumps({"error": "interrupted", "output": ""})
            output = stdout or ""
            if stderr:
                output += "\n[stderr]\n" + stderr
            if truncated or len(output) > MAX_TERMINAL_OUTPUT_CHARS:
                output = output[:MAX_TERMINAL_OUTPUT_CHARS] + "\n... (truncated)"
            failure = classify_sandbox_failure(
                exit_code=proc.returncode,
                stdout=stdout or "",
                stderr=stderr or "",
                sandboxed=run_sandboxed,
            )
            if failure is not None:
                inc_terminal_metric_safe(
                    "terminal_sandbox_failure",
                    labels={"constraint": failure.constraint},
                    session_key=session_key,
                )
                return json.dumps(
                    format_sandbox_error_payload(
                        failure,
                        command=cmd_text,
                        exit_code=proc.returncode,
                        output=output,
                    ),
                    ensure_ascii=False,
                )
            return json.dumps({
                "exit_code": proc.returncode,
                "output": output,
                "sandboxed": run_sandboxed,
            })
        finally:
            _close_pipe(proc.stdout)
            _close_pipe(proc.stderr)

    return str(
        run_terminal_with_gates_safe(
            cmd_text,
            cwd=cwd_hint,
            session_key=session_key,
            run_fn=lambda: run_subprocess_loud(_run_subprocess),
        )
    )


def _tool_search_files(pattern: str = "", path: str = ".", include: str | None = None, **_: Any) -> str:
    needle = str(pattern or "").strip() or "*"
    safety = check_tool_path(path)
    if not safety.allowed:
        return json.dumps({"error": safety.error})
    cmd = ["rg", "--no-config", "--json", "-m", "20"]
    if include:
        cmd.extend(["--glob", include])
    cmd.extend(["--", needle, str(safety.path)])

    def _run() -> str:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=15,
            env=safe_subprocess_env(),
        )
        matches = []
        for line in result.stdout.strip().split("\n"):
            if not line:
                continue
            try:
                obj = json.loads(line)
                if obj.get("type") == "match":
                    data = obj["data"]
                    path_text = data["path"]["text"]
                    match_safety = check_tool_path(path_text)
                    if not match_safety.allowed:
                        continue
                    matches.append({
                        "path": str(match_safety.path),
                        "line": data["line_number"],
                        "text": data["lines"]["text"].rstrip(),
                    })
            except (json.JSONDecodeError, KeyError):
                continue
        return json.dumps({"matches": matches[:50], "total": len(matches)})

    return str(run_search_files_loud(_run))


def _tool_list_directory(path: str = ".", **_: Any) -> str:
    safety = check_tool_path(path)
    if not safety.allowed:
        return json.dumps({"error": safety.error})
    p = safety.path
    if not p.is_dir():
        return json.dumps({"error": f"Not a directory: {path}"})

    def _run() -> str:
        entries = []
        for item in sorted(p.iterdir()):
            entries.append({
                "name": item.name,
                "type": "dir" if item.is_dir() else "file",
                "size": item.stat().st_size if item.is_file() else None,
            })
        return json.dumps({"path": str(p), "entries": entries[:200]})

    return str(run_list_directory_loud(_run))