"""Hooks that gate the kernel's own tools: `run` and `files`.

The daemon fires hooks around the model's outer tool calls. A cell that calls
`run()` or `files.edit()` hides those calls inside one `python` call, so the
kernel asks the daemon's `hooks.fire` route itself, before and after, with
the Claude Code / Codex tool names scripts already match on: `Bash`, `Read`,
`Edit`, `Write`. Nothing happens unless the hooks extension installed a
blocking host call here.
"""

from __future__ import annotations

import asyncio
import inspect
import re
import threading
from collections.abc import Awaitable, Callable
from typing import cast

_sync: Callable[[str, dict[str, object]], object] | None = None

# Python callables the model registered through `hooks.register`: they run in
# the kernel, before the daemon's command hooks, and die with the kernel.
Callback = Callable[[dict[str, object]], object]
_callbacks: list[tuple[str, str | None, Callback]] = []

EVENTS = ("PreToolUse", "PostToolUse", "UserPromptSubmit", "Stop")


def register(
    event: str, callback: Callback, matcher: str | None = None
) -> Callable[[], None]:
    """Add a callback for `event` on tools matching the regex `matcher`; answers
    the function that removes it."""
    if event not in EVENTS:
        raise ValueError(f"event is one of {', '.join(EVENTS)}, not {event!r}")
    if not callable(callback):
        raise TypeError("callback must be callable")
    if matcher is not None:
        re.compile(matcher)
    entry = (event, matcher, callback)
    _callbacks.append(entry)

    def remove() -> None:
        if entry in _callbacks:
            _callbacks.remove(entry)

    return remove


def registered() -> list[tuple[str, str | None, Callback]]:
    return list(_callbacks)


def clear() -> None:
    _callbacks.clear()


def _matches(matcher: str | None, tool: str) -> bool:
    return matcher in (None, "", "*") or re.search(matcher, tool) is not None


def _settled(value: object) -> object:
    """A callback's answer, awaited off the (blocked) event loop if need be."""
    if not inspect.isawaitable(value):
        return value
    box: list[object] = []

    def work() -> None:
        try:
            box.append(asyncio.run(_await(cast(Awaitable[object], value))))
        except BaseException as error:
            box.append(error)

    thread = threading.Thread(target=work)
    thread.start()
    thread.join()
    if isinstance(box[0], BaseException):
        raise box[0]
    return box[0]


async def _await(value: Awaitable[object]) -> object:
    return await value


def _normal(output: object) -> dict[str, object]:
    """A callback's answer in the same shape the daemon's hooks answer: the
    Claude Code / Codex JSON output contract, read into block, halt,
    updated_input and context."""
    if not isinstance(output, dict):
        return {}
    answer: dict[str, object] = {"context": []}
    specific = output.get("hookSpecificOutput")
    specific = specific if isinstance(specific, dict) else {}
    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"] = [str(specific["additionalContext"])]
    return answer


def _callbacks_for(
    event: str, tool: str, tool_input: dict[str, object], response: object
) -> dict[str, object]:
    fields: dict[str, object] = {"tool_name": tool, "tool_input": tool_input}
    if response is not None:
        fields["tool_response"] = response
    return run_event(event, tool, fields)


def run_event(event: str, subject: str, fields: dict[str, object]) -> dict[str, object]:
    """Every callback registered for `event` whose matcher fits `subject`
    (the tool name; empty for events without one), merged into one answer:
    block, halt, updated_input, context. A tool event's callbacks see each
    other's input rewrites."""
    merged: dict[str, object] = {"context": []}
    context = cast(list[str], merged["context"])
    current = dict(fields)
    for kind, matcher, callback in list(_callbacks):
        if kind != event or not _matches(matcher, subject):
            continue
        payload: dict[str, object] = {**current, "hook_event_name": event}
        try:
            answer = _normal(_settled(callback(payload)))
        except Exception as error:  # a broken callback fails open, loudly
            print(
                f"[{event} hook] callback {getattr(callback, '__name__', callback)} failed: {error!r}"
            )
            continue
        context.extend(cast(list[str], answer.get("context") or []))
        for key in ("block", "halt"):
            if answer.get(key) and not merged.get(key):
                merged[key] = answer[key]
        if isinstance(answer.get("updated_input"), dict):
            current["tool_input"] = answer["updated_input"]
            merged["updated_input"] = answer["updated_input"]
    return merged


class Blocked(PermissionError):
    """A hook refused the call; the message is the hook's reason."""


def install(sync: Callable[[str, dict[str, object]], object] | None) -> None:
    global _sync
    _sync = sync


def active() -> bool:
    return _sync is not None or bool(_callbacks)


def _fire(
    event: str,
    tool: str,
    tool_input: dict[str, object],
    response: object = None,
    echo: bool = True,
) -> dict[str, object]:
    answer = _callbacks_for(event, tool, tool_input, response)
    if _sync is not None and not (answer.get("block") or answer.get("halt")):
        current = cast(dict[str, object], answer.get("updated_input") or tool_input)
        args: dict[str, object] = {
            "event": event,
            "tool_name": tool,
            "tool_input": current,
        }
        if response is not None:
            args["tool_response"] = response
        try:
            host = cast(dict[str, object], _sync("hooks.fire", args))
        except Exception as error:  # a broken hook channel never blocks the tool
            print(f"[{event} hook] unavailable: {error}")
            host = {}
        # A callback's rewrite is the host's input already; the host may rewrite again.
        answer["context"] = list(cast(list[str], answer["context"])) + list(
            cast(list[str], host.get("context") or [])
        )
        for key in ("block", "halt", "updated_input"):
            if host.get(key):
                answer[key] = host[key]
    if echo:
        for text in cast(list[str], answer.get("context") or []):
            print(f"[{event} hook] {text}")
    return answer


def pre(tool: str, tool_input: dict[str, object]) -> dict[str, object]:
    """Fire PreToolUse. Raises Blocked on a deny or halt; otherwise answers the
    input to use, which a hook may have rewritten."""
    if not active():
        return tool_input
    answer = _fire("PreToolUse", tool, tool_input)
    reason = answer.get("halt") or answer.get("block")
    if reason:
        raise Blocked(f"blocked by a hook: {reason}")
    updated = answer.get("updated_input")
    return cast(dict[str, object], updated) if isinstance(updated, dict) else tool_input


def post(tool: str, tool_input: dict[str, object], response: object) -> None:
    """Fire PostToolUse. A block cannot undo the call; its reason is printed."""
    if not active():
        return
    answer = _fire("PostToolUse", tool, tool_input, response)
    reason = answer.get("block")
    if reason:
        print(f"[PostToolUse hook] {reason}")


async def post_async(
    tool: str, tool_input: dict[str, object], response: object
) -> list[str]:
    """PostToolUse for work that finishes after its cell did (a `run()` job):
    fired off the event loop, answering the lines the hooks had to say."""
    if not active():
        return []

    def work() -> dict[str, object]:
        return _fire("PostToolUse", tool, tool_input, response, echo=False)

    answer = await asyncio.to_thread(work)
    lines = [
        f"[PostToolUse hook] {t}" for t in cast(list[str], answer.get("context") or [])
    ]
    if answer.get("block"):
        lines.append(f"[PostToolUse hook] {answer['block']}")
    return lines


def run_event_json(event: str, subject: str, fields_json: str) -> str:
    """`run_event` for the daemon, which fires the loop's own events (the
    outer tool call, the prompt, the end of a turn) through a hidden cell."""
    import json

    answer = run_event(event, subject, json.loads(fields_json))
    if "updated_input" in answer:
        answer["updated_input"] = json.dumps(answer["updated_input"])
    return json.dumps(answer)