"""Security layer — zeroclaw-style tiered autonomy.

Modes: supervised (risky ops ask y/n in terminal), auto (safe allowed, denied blocked),
yolo (everything allowed). Denylist always wins except in yolo mode.

Approvals are async: the interactive REPL installs a `confirmer` coroutine so approval
answers share the single stdin reader with new-task input; one-shot mode falls back to
a blocking input() on a thread.
"""
from __future__ import annotations

import asyncio
import re
from typing import Awaitable, Callable

from rich.console import Console

DENY_PATTERNS = [
    r"\brm\s+-rf\s+/", r"\bformat\b", r"\bmkfs", r"\bshutdown\b", r"\breboot\b",
    r"\bdel\s+/[sq]", r"Remove-Item.*-Recurse.*-Force.*(C:\\|/)\s*$",
    r"\breg\s+delete", r":\(\)\{.*\};:", r"\bdd\s+if=",
]

SAFE_SHELL_PREFIXES = (
    "dir", "ls", "cat", "type", "echo", "git status", "git log", "git diff",
    "python --version", "node --version", "pip show", "where", "whoami", "curl",
    "findstr", "grep", "head", "tail", "wc",
)

console = Console()


class Policy:
    def __init__(self, mode: str = "supervised"):
        self.mode = mode
        # Optional async confirm hook, installed by the REPL: () -> bool
        self.confirmer: Callable[[], Awaitable[bool]] | None = None

    async def _confirm(self) -> bool:
        if self.confirmer is not None:
            return await self.confirmer()
        try:
            return await asyncio.to_thread(
                lambda: input("   allow? [y/N] ").strip().lower() in ("y", "yes")
            )
        except (EOFError, KeyboardInterrupt):
            return False

    async def check_shell(self, command: str) -> tuple[bool, str]:
        """Returns (allowed, reason)."""
        if self.mode != "yolo":
            for pat in DENY_PATTERNS:
                if re.search(pat, command, re.IGNORECASE):
                    return False, f"blocked by denylist pattern: {pat}"
        if self.mode == "yolo":
            return True, "yolo"
        stripped = command.strip().lower()
        if any(stripped.startswith(p) for p in SAFE_SHELL_PREFIXES):
            return True, "safe prefix"
        if self.mode == "auto":
            return True, "auto mode"
        console.print(f"\n[bold yellow]⚡ agent wants to run:[/] [cyan]{command}[/]")
        if await self._confirm():
            return True, "user approved"
        return False, "user denied"

    async def check_write(self, path: str, workspace: str) -> tuple[bool, str]:
        from pathlib import Path
        p = Path(path).resolve()
        ws = Path(workspace).resolve()
        if ws in p.parents or p == ws:
            return True, "inside workspace"
        if self.mode in ("yolo", "auto"):
            return True, self.mode
        console.print(f"\n[bold yellow]⚡ agent wants to write outside its workspace:[/] [cyan]{path}[/]")
        return (True, "user approved") if await self._confirm() else (False, "user denied")