"""Kernel-local Python callbacks and user-level Claude Code command hooks."""

from __future__ import annotations

import asyncio
import hashlib
import inspect
import json
import os
import re
import signal
import subprocess
import tempfile
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import albedo_proc

EVENTS = (
    "SessionStart",
    "SessionEnd",
    "UserPromptSubmit",
    "PreToolUse",
    "PostToolUse",
    "Stop",
)
Callback = Callable[[dict], object]
_callbacks: list[tuple[str, str | None, Callback]] = []
enabled: bool = False
started: bool = False  # a prompt has been seen: this kernel is a live session
_session, _cwd = "", None


def register(event: str, callback: Callback, matcher: str | None = None):
    if event not in EVENTS or not callable(callback):
        raise ValueError("expected a supported event and callable callback")
    if matcher not in (None, "", "*"):
        re.compile(matcher)
    entry = (event, matcher, callback)
    _callbacks.append(entry)

    def remove():
        if entry in _callbacks:
            _callbacks.remove(entry)

    return remove


registered, clear = _callbacks.copy, _callbacks.clear


def _matches(matcher, subject):
    return matcher in (None, "", "*") or re.search(matcher, subject) is not None


def _empty() -> dict:
    return dict(block=None, halt=None, updated_input=None, context=[], messages=[])


def _merge(answer, extra):
    for key in ("block", "halt", "updated_input"):
        if answer[key] is None:
            answer[key] = extra[key]
    for key in ("context", "messages"):
        answer[key].extend(extra[key])


def _clip(text):
    return text if len(text) <= 10000 else text[:9988] + "\n[truncated]"


def _normal(output):
    if not isinstance(output, dict):
        raise ValueError("hook output must be an object")
    answer = _empty()
    specific = output.get("hookSpecificOutput") or {}
    if output.get("continue") is False:
        answer["halt"] = str(output.get("stopReason") or "stopped by a hook")
    if output.get("decision") == "block":
        answer["block"] = str(output.get("reason") or "blocked by a hook")
    if specific.get("permissionDecision") == "deny":
        answer["block"] = str(specific.get("permissionDecisionReason") or "denied")
    if isinstance(specific.get("updatedInput"), dict):
        answer["updated_input"] = specific["updatedInput"]
    if specific.get("additionalContext"):
        answer["context"] = [_clip(str(specific["additionalContext"]))]
    if output.get("systemMessage"):
        answer["messages"] = [str(output["systemMessage"])]
    return answer


def _settled(value):
    if not inspect.isawaitable(value):
        return value

    async def wait():
        return await value

    with ThreadPoolExecutor(max_workers=1) as pool:
        return pool.submit(asyncio.run, wait()).result()


def _capture(command, payload, timeout):
    """Run one hook command: the payload on stdin, output in temp files so a
    backgrounded child holding a pipe cannot hang us, the whole process group
    killed afterwards."""
    with tempfile.TemporaryFile() as stdin, tempfile.TemporaryFile() as out:
        with tempfile.TemporaryFile() as err:
            stdin.write(json.dumps(payload).encode())
            stdin.seek(0)
            proc = subprocess.Popen(
                command,
                shell=True,
                cwd=payload["cwd"],
                stdin=stdin,
                stdout=out,
                stderr=err,
                start_new_session=True,
            )
            group = albedo_proc.Group(proc.pid, albedo_proc.leader_token(proc.pid))
            try:
                proc.wait(timeout)
            except subprocess.TimeoutExpired:
                raise TimeoutError("command hook timed out") from None
            finally:
                albedo_proc.send_signal(group, signal.SIGKILL)
                proc.wait()
            out.seek(0)
            err.seek(0)
            text = lambda f: f.read(1 << 20).decode("utf-8", errors="replace")  # noqa: E731
            return proc.returncode, text(out), text(err)


def _command(definition, payload):
    try:
        if definition.get("type") != "command" or not isinstance(
            definition.get("command"), str
        ):
            raise ValueError("expected a command hook")
        timeout = definition.get("timeout", 600)
        if payload["hook_event_name"] == "SessionEnd":
            timeout = min(timeout, 3)  # the kernel is shutting down
        if type(timeout) not in (int, float) or not 0 < timeout < float("inf"):
            raise ValueError("timeout must be a positive finite number")
        status, stdout, stderr = _capture(definition["command"], payload, timeout)
        answer = _empty()
        if status == 2:
            answer["block"] = stderr.strip() or "blocked by a hook"
        elif status != 0:
            raise ValueError(f"command hook exited {status}: {stderr.strip()}")
        elif stdout.strip():
            try:
                output = json.loads(stdout)
            except ValueError:
                if payload["hook_event_name"] not in (
                    "UserPromptSubmit",
                    "SessionStart",
                ):
                    raise ValueError("invalid hook JSON") from None
                answer["context"] = [_clip(stdout.strip())]
            else:
                answer = _normal(output)
        return answer
    except Exception as error:
        return {**_empty(), "messages": [f"Hook failed open: {error}"]}


_warned: set[str] = set()


def _approved(path, answer) -> bool:
    """Whether the user approved this exact workspace hooks file: its sha256
    sits in `hooks.sha256` beside it, which the user's `/hooks` page writes."""
    digest = hashlib.sha256(path.read_bytes()).hexdigest()
    seal = path.with_name("hooks.sha256")
    if seal.exists() and seal.read_text().strip() == digest:
        return True
    if digest not in _warned:
        _warned.add(digest)
        answer["messages"].append(
            "This workspace's hooks are not approved, so they are skipped. "
            "Review and approve them with /hooks."
        )
    return False


def fire(event: str, subject: str, fields: dict, session="", cwd=None) -> dict:
    payload = dict(
        fields,
        session_id=session or _session,
        cwd=cwd or _cwd or os.getcwd(),
        hook_event_name=event,
    )
    payload.update(transcript_path="", permission_mode="default")
    answer = _empty()
    for kind, matcher, callback in list(_callbacks):
        if kind == event and _matches(matcher, subject):
            try:
                value = _settled(callback(dict(payload)))
                _merge(answer, _normal(value) if value is not None else _empty())
                if answer["updated_input"] is not None:
                    payload["tool_input"] = answer["updated_input"]
            except Exception as error:
                answer["messages"].append(f"Callback failed open: {error}")
    if answer["block"] is not None or answer["halt"] is not None:
        return answer
    home = Path(os.environ.get("ALBEDO_HOME") or Path.home() / ".albedo")
    workspace = home / "workspaces" / re.sub(r"[^A-Za-z0-9]", "-", payload["cwd"])
    definitions = []
    # The workspace's hooks come first, so they win conflicts. Its file is
    # model-writable, so (as in Codex) it runs only once its hash is approved;
    # the user's global hooks are their own and always run.
    for path, own in ((workspace / "hooks.json", False), (home / "hooks.json", True)):
        try:
            if not path.exists() or not (own or _approved(path, answer)):
                continue
            for group in json.loads(path.read_text()).get("hooks", {}).get(event, []):
                if _matches(group.get("matcher", ""), subject):
                    definitions.extend(group["hooks"])
        except Exception as error:
            answer["messages"].append(f"Hook config {path} failed open: {error}")
    with ThreadPoolExecutor(max_workers=8) as pool:
        for extra in pool.map(lambda d: _command(d, payload), definitions):
            _merge(answer, extra)
    return answer


def fire_json(event, subject, fields_json, session, cwd):
    global _session, _cwd, started
    _session, _cwd = session, cwd
    answer = fire(event, subject, json.loads(fields_json), session, cwd)
    if event == "UserPromptSubmit" and not started:
        # The first prompt of a kernel is when a session starts: SessionStart
        # context comes first, and it cannot block.
        started = True
        first = fire("SessionStart", "startup", {"source": "startup"}, session, cwd)
        first["block"] = None
        _merge(first, answer)
        answer = first
    if answer["updated_input"] is not None:
        answer["updated_input"] = json.dumps(answer["updated_input"])
    return json.dumps(answer)


def end():
    """SessionEnd, as the kernel shuts down."""
    if started:
        fire("SessionEnd", "other", {"reason": "other"})


def _lines(event, answer):
    texts = answer["context"] + answer["messages"]
    texts += [v for v in (answer["block"], answer["halt"]) if v]
    return [f"[{event} hook] {text}" for text in texts]


def pre(tool, tool_input):
    if not enabled:
        return tool_input
    answer = fire("PreToolUse", tool, dict(tool_name=tool, tool_input=tool_input))
    for line in _lines("PreToolUse", answer):
        print(line)
    if answer["halt"] or answer["block"]:
        raise PermissionError(f"blocked by a hook: {answer['halt'] or answer['block']}")
    updated = answer["updated_input"]
    return tool_input if updated is None else updated


def post_lines(tool, tool_input, response):
    fields = dict(tool_name=tool, tool_input=tool_input, tool_response=response)
    return _lines("PostToolUse", fire("PostToolUse", tool, fields))


def post(tool, tool_input, response):
    if enabled:
        for line in post_lines(tool, tool_input, response):
            print(line)