#!/usr/bin/env python3
"""Fail-closed Hermes tool surface for live no-post Leo OOS trials."""

from __future__ import annotations

import argparse
import json
import os
import shlex
import shutil
import subprocess
import uuid
from pathlib import Path
from typing import Any

READ_ONLY_BRIDGE_COMMANDS = frozenset(
    {
        "search",
        "context",
        "show",
        "evidence",
        "edges",
        "list-proposals",
        "search-proposals",
        "show-proposal",
        "status",
        "decision-matrix-status",
    }
)
HARMLESS_TRAILING_REDIRECT_TOKENS = frozenset({"2>/dev/null", "2> /dev/null"})


class ReadOnlyGuardError(RuntimeError):
    """The requested command cannot be proven read-only."""


class GuardArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        raise ReadOnlyGuardError(message)

    def exit(self, status: int = 0, message: str | None = None) -> None:
        raise ReadOnlyGuardError(message or f"argument parser exited with status {status}")


def _uuid_argument(value: str) -> str:
    try:
        return str(uuid.UUID(value))
    except ValueError as exc:
        raise argparse.ArgumentTypeError("expected a UUID") from exc


def validate_read_only_bridge_argv(argv: list[str]) -> None:
    """Accept only the bridge's bounded read commands and exact arguments."""

    parser = GuardArgumentParser(add_help=False)
    subparsers = parser.add_subparsers(dest="command", required=True)

    def command(name: str) -> argparse.ArgumentParser:
        item = subparsers.add_parser(name, add_help=False)
        item.add_argument("--help", action="store_true")
        item.add_argument("--format", choices=("markdown", "json"), default="markdown")
        return item

    for name in ("search", "context"):
        item = command(name)
        item.add_argument("query", nargs="+")
        item.add_argument("--limit", type=int, default=5)
        item.add_argument("--context-limit", type=int, default=5 if name == "search" else 6)
    show = command("show")
    show.add_argument("claim_id", type=_uuid_argument)
    for name in ("evidence", "edges"):
        item = command(name)
        item.add_argument("claim_id", type=_uuid_argument)
        item.add_argument("--limit", type=int, default=12)
    list_proposals = command("list-proposals")
    list_proposals.add_argument(
        "--status",
        choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"),
        default="pending_review",
    )
    list_proposals.add_argument("--limit", type=int, default=10)
    search_proposals = command("search-proposals")
    search_proposals.add_argument("query", nargs="+")
    search_proposals.add_argument(
        "--status",
        choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"),
        default="all",
    )
    search_proposals.add_argument("--limit", type=int, default=10)
    show_proposal = command("show-proposal")
    show_proposal.add_argument("proposal_id", type=_uuid_argument)
    command("status")
    command("decision-matrix-status")
    parsed = parser.parse_args(argv)
    for name, value in vars(parsed).items():
        if name.endswith("limit") and not 1 <= value <= 100:
            raise ReadOnlyGuardError(f"{name.replace('_', '-')} must be between 1 and 100")


def classify_terminal_command(
    wrapper: Path,
    temp_profile: Path,
    tool_args: dict[str, Any],
    *,
    allow_kb_reads: bool,
) -> tuple[list[str] | None, str | None]:
    """Return verified argv or a fail-closed reason without executing anything."""

    command = str(tool_args.get("command") or "")
    if len(command) > 8192:
        return None, "command is too long"
    if tool_args.get("background") or tool_args.get("pty") or tool_args.get("workdir"):
        return None, "background, PTY, and workdir overrides are disabled"
    if not allow_kb_reads:
        return None, "database tools are disabled for the no-DB ablation"
    try:
        argv = shlex.split(command)
    except ValueError as exc:
        return None, f"invalid command quoting: {exc}"
    while argv and argv[-1] in HARMLESS_TRAILING_REDIRECT_TOKENS:
        argv.pop()
    command_path = f"{temp_profile / 'bin'}:{os.environ.get('PATH', '')}"
    try:
        exact_wrapper = wrapper.resolve(strict=True)
        executable = shutil.which(argv[0], path=command_path) if argv else None
        invoked = Path(executable).resolve(strict=True) if executable else None
    except OSError:
        invoked = None
        exact_wrapper = wrapper.resolve(strict=False)
    if invoked != exact_wrapper:
        return None, "only the exact temporary-profile teleo-kb wrapper is available"
    if len(argv) < 2 or argv[1] not in READ_ONLY_BRIDGE_COMMANDS:
        return None, "only read-only Teleo KB bridge commands are available"
    try:
        validate_read_only_bridge_argv(argv[1:])
    except ReadOnlyGuardError as exc:
        return None, f"invalid read-only KB arguments: {exc}"
    return argv, None


def restricted_bridge_terminal_handler(
    wrapper: Path,
    temp_profile: Path,
    tool_args: dict[str, Any],
    *,
    allow_kb_reads: bool,
    **_kwargs: Any,
) -> str:
    argv, reason = classify_terminal_command(
        wrapper,
        temp_profile,
        tool_args,
        allow_kb_reads=allow_kb_reads,
    )
    if argv is None:
        return json.dumps({"output": "", "exit_code": 126, "error": reason})
    timeout = min(max(int(tool_args.get("timeout") or 60), 1), 120)
    child_environment = {
        "HOME": str(temp_profile),
        "HERMES_HOME": str(temp_profile),
        "PATH": f"{temp_profile / 'bin'}:{os.environ.get('PATH', '')}",
        "TELEO_KB_MODE": "local",
    }
    try:
        proc = subprocess.run(
            argv,
            cwd=temp_profile,
            env=child_environment,
            text=True,
            capture_output=True,
            check=False,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return json.dumps({"output": "", "exit_code": 124, "error": f"bridge command exceeded {timeout}s"})
    return json.dumps(
        {
            "output": proc.stdout,
            "exit_code": proc.returncode,
            "error": proc.stderr or None,
        }
    )


def install_read_only_tool_surface(temp_profile: Path, *, allow_kb_reads: bool) -> dict[str, Any]:
    """Replace Hermes' registry with skills plus one guarded terminal tool."""

    import tools.skills_tool
    import tools.terminal_tool  # noqa: F401
    import toolsets
    from hermes_cli import tools_config
    from tools.registry import registry

    toolset_name = "leo-oos-kb-readonly" if allow_kb_reads else "leo-oos-no-db-ablation"
    allowed_tools = ["skills_list", "skill_view", "terminal"]
    toolsets.TOOLSETS[toolset_name] = {
        "description": "Fail-closed Leo OOS tool surface",
        "tools": allowed_tools,
        "includes": [],
    }
    tools_config._get_platform_tools = lambda *_args, **_kwargs: {toolset_name}
    missing = sorted(name for name in allowed_tools if name not in registry._tools)
    if missing:
        raise ReadOnlyGuardError(f"required tools are not registered: {missing}")
    allowed_entries = {name: registry._tools[name] for name in allowed_tools}
    registry._tools.clear()
    registry._tools.update(allowed_entries)
    wrapper = temp_profile / "bin" / "teleo-kb"
    terminal_entry = registry._tools["terminal"]
    terminal_entry.handler = lambda tool_args, **kwargs: restricted_bridge_terminal_handler(
        wrapper,
        temp_profile,
        tool_args,
        allow_kb_reads=allow_kb_reads,
        **kwargs,
    )
    return {
        "mode": "read_only_kb" if allow_kb_reads else "no_db_ablation",
        "allowed_tools": allowed_tools,
        "actual_registry_tools": sorted(registry._tools),
        "read_only_bridge_commands": sorted(READ_ONLY_BRIDGE_COMMANDS) if allow_kb_reads else [],
        "send_message_tool_enabled": "send_message" in registry._tools,
        "terminal_restricted_to_exact_wrapper": True,
        "mutating_bridge_commands_exposed": False,
        "provider_credentials_forwarded_to_terminal": False,
    }