"""Hooks a connected client answers (Claude Code's SDK hook callbacks) and the MCP tool that answers permission requests (``--permission-prompt-tool``).""" from __future__ import annotations import asyncio import json from pathlib import Path from typing import Any import pytest from noeta.core.agent import Agent from noeta.core.extension import Extension from noeta.plugins.hooks import HookCommand, Hooks from noeta.plugins.hooks.runners import run_callback from noeta.testing import reply from noeta.toolkit.hook_callbacks import HookCallbacks, parse_callbacks from tests.plugins.test_hooks_events import READ_A, AskAll, McpLike, Setup def with_hooks(s: Setup, hooks: Hooks, *extensions: Extension) -> Setup: s.hooks = hooks s.agent = Agent(name="a", model="m", extensions=[*extensions, hooks, *s.agent.extensions[-1:]]) return s def test_parse_callbacks() -> None: hooks = parse_callbacks({"PreToolUse": [{"matcher": "Read", "hookCallbackIds": ["h1", "h2"], "timeout": 5}], "Stop": [{"hookCallbackIds": []}]}) assert hooks == {"PreToolUse": [{"matcher": "Read", "hooks": [ {"type": "callback", "callback_id": "h1", "timeout": 5}, {"type": "callback", "callback_id": "h2", "timeout": 5}]}]} for bad in ([], {"Stop": {}}, {"Stop": [{"matcher": "x"}]}, {"Stop": [{"hookCallbackIds": [1]}]}): with pytest.raises(ValueError, match="hookCallbackIds"): parse_callbacks(bad) async def test_a_client_hook_answers_and_is_replaced(tmp_path: Path) -> None: asked: list[tuple[str, dict[str, Any], str | None]] = [] async def send(callback_id: str, payload: dict[str, Any], tool_use_id: str | None) -> dict[str, Any]: asked.append((callback_id, payload, tool_use_id)) return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": "the client says no"}} callbacks = HookCallbacks() callbacks.register(parse_callbacks({"PreToolUse": [{"matcher": "Read", "hookCallbackIds": ["h1"]}]}), send) hooks = Hooks(callbacks=callbacks) # made after the client registered: it still sees them s = with_hooks(Setup(tmp_path, {}, [reply("", READ_A), reply("ok")]), hooks) await s.start() [(callback_id, payload, tool_use_id)] = asked assert callback_id == "h1" and payload["tool_name"] == "Read" assert tool_use_id == payload["tool_use_id"] assert await s.results() == [("the client says no", True)] # A new registration replaces the old one; none leaves no hook. callbacks.register({}, send) assert "PreToolUse" not in hooks.matchers async def test_settings_files_cannot_declare_callback_hooks(tmp_path: Path) -> None: hooks = Hooks(hooks={"Stop": [{"hooks": [{"type": "callback", "callback_id": "x"}]}]}) assert hooks.matchers == {} async def test_a_callback_without_a_client_or_in_time_fails() -> None: hook = HookCommand(type="callback", callback_id="h", timeout=0.01) assert (await run_callback(hook, {}, None)).failed async def slow(*_a: Any) -> dict[str, Any]: await asyncio.sleep(1) return {} async def broken(*_a: Any) -> dict[str, Any]: raise RuntimeError("the client is gone") assert "timed out" in (await run_callback(hook, {}, slow)).stderr assert "the client is gone" in (await run_callback(hook, {}, broken)).stderr with pytest.raises(RuntimeError, match="no client"): await HookCallbacks().call("h", {}, None) async def test_the_permission_prompt_tool_answers_asks(tmp_path: Path) -> None: allow = McpLike(json.dumps({"behavior": "allow", "updatedInput": {"file_path": "b.txt"}})) s = with_hooks(Setup(tmp_path, {}, [reply("", READ_A), reply("ok")]), Hooks(permission_prompt_tool="mcp__chk__verify"), AskAll(), allow) await s.start() [call] = allow.calls assert call["tool_name"] == "Read" and call["input"] == {"file_path": "a.txt"} assert call["tool_use_id"] [(text, is_error)] = await s.results() assert not is_error and "other" in text deny = McpLike(json.dumps({"behavior": "deny", "message": "Not that."})) s = with_hooks(Setup(tmp_path / "2", {}, [reply("", READ_A), reply("ok")]), Hooks(permission_prompt_tool="mcp__chk__verify"), AskAll(), deny) await s.start() assert await s.results() == [("Not that.", True)] async def test_an_unusable_prompt_tool_answer_denies(tmp_path: Path) -> None: for answer in ("yes", json.dumps({"behavior": "deny"})): odd = McpLike(answer) s = with_hooks(Setup(tmp_path / str(len(answer)), {}, [reply("", READ_A), reply("ok")]), Hooks(permission_prompt_tool="mcp__chk__verify"), AskAll(), odd) await s.start() [(text, is_error)] = await s.results() assert is_error and "permission prompt tool chk/verify answered" in text with pytest.raises(ValueError, match="not an MCP tool name"): Hooks(permission_prompt_tool="Read") async def test_a_worker_that_cannot_ask_the_client_refuses_what_its_hooks_would_judge( tmp_path: Path) -> None: """Two workers on one store: the one holding the client asks it; the other refuses a call a recorded PreToolUse callback would judge, and an ask a recorded PermissionRequest callback would answer, instead of running them unjudged. Other calls run.""" asked: list[str] = [] async def send(callback_id: str, payload: dict[str, Any], tool_use_id: str | None) -> dict[str, Any]: asked.append(payload["tool_input"]["file_path"]) return {} holder = HookCallbacks() holder.register(parse_callbacks({ "PreToolUse": [{"matcher": "Read", "hookCallbackIds": ["h1"]}], "PermissionRequest": [{"matcher": "Glob", "hookCallbackIds": ["h2"]}]}), send) record = holder.summary() assert record == {"client": holder.client_id, "events": {"PreToolUse": ["Read"], "PermissionRequest": ["Glob"]}} s = Setup(tmp_path, {}, [reply("", READ_A), reply("refused"), reply("", ("Read", {"file_path": "b.txt"})), reply("read it")]) other = s.agent # the worker without the client: no callbacks with_hooks(s, Hooks(callbacks=holder)) holding = s.agent s.agent = other await s.start(client_hooks=record) [(refused, is_error)] = await s.results() assert is_error and "PreToolUse hook of the client" in refused and asked == [] s.agent = holding await s.send("try again") assert asked == ["b.txt"] and not (await s.results())[1][1] # A recorded PermissionRequest callback: the other worker denies the ask. s2 = Setup(tmp_path / "2", {}, [reply("", ("Glob", {"pattern": "*.txt"})), reply("", READ_A), reply("done")], extensions=(AskAll(),)) await s2.start(client_hooks=record) first, second = await s2.results() assert first[1] and "PermissionRequest hook of the client" in first[0] assert second[1] and "PreToolUse hook of the client" in second[0] async def test_a_child_task_carries_the_client_hooks_record() -> None: from noeta.core.loop.state import TaskState record = {"client": "c", "events": {"PreToolUse": [""]}} state = TaskState(task_id="t", agent="a", config={"client_hooks": record}) assert Hooks().inherit(state) == {"client_hooks": record} assert Hooks().inherit(TaskState(task_id="t", agent="a")) == {} def test_a_reconnected_client_with_the_same_callbacks_answers_an_older_record() -> None: """The record names events and matchers, not one client: after a reconnect (a new client id) the new client answers what the root and its children recorded, and a client lacking one of them does not.""" async def send(callback_id: str, payload: dict[str, Any], tool_use_id: str | None) -> dict[str, Any]: return {} hooks = {"PreToolUse": [{"matcher": "Read", "hookCallbackIds": ["h1"]}], "PermissionRequest": [{"matcher": "Glob", "hookCallbackIds": ["h2"]}]} old = HookCallbacks() old.register(parse_callbacks(hooks), send) record = old.summary() new = HookCallbacks() assert not new.answers(record) # no client registered here yet new.register(parse_callbacks(hooks), send) assert new.client_id != old.client_id and new.answers(record) fewer = HookCallbacks() fewer.register(parse_callbacks({"PreToolUse": hooks["PreToolUse"]}), send) assert not fewer.answers(record) async def test_a_reconnect_while_a_background_child_runs_lets_the_child_ask_the_new_client( tmp_path: Path) -> None: """A child task carries the copy written for the first client; once a new client with the same callbacks holds this process, the child's hooked call is asked of it, not refused.""" asked: list[str] = [] async def send(callback_id: str, payload: dict[str, Any], tool_use_id: str | None) -> dict[str, Any]: asked.append(payload["tool_input"]["file_path"]) return {} hooks = {"PreToolUse": [{"matcher": "Read", "hookCallbackIds": ["h1"]}]} first = HookCallbacks() first.register(parse_callbacks(hooks), send) copy = first.summary() # what the child inherited second = HookCallbacks() # the reconnected client in this process second.register(parse_callbacks(hooks), send) s = Setup(tmp_path, {}, [reply("", ("Read", {"file_path": "b.txt"})), reply("read it")]) with_hooks(s, Hooks(callbacks=second)) await s.start(client_hooks=copy) [(_text, is_error)] = await s.results() assert not is_error and asked == ["b.txt"] async def test_following_a_task_clears_its_open_children_too() -> None: from noeta.core.loop.state import CONFIG_SIGNAL from noeta.toolkit.interactive import FOLLOWED_CONFIG, follow class Row: def __init__(self, task_id: str) -> None: self.task_id = task_id tree = {("root", "waiting"): ["child"], ("child", "running"): ["grandchild"]} signalled: list[tuple[str, str, Any]] = [] class Fake: async def tasks(self, *, parent_id: str | None = None, status: Any = None, limit: int = 100) -> list[Any]: return [Row(t) for t in tree.get((parent_id or "", status), [])] async def signal(self, task_id: str, name: str, data: Any = None, *, dedupe_key: str | None = None) -> int: signalled.append((task_id, name, data)) return 0 await follow(Fake(), "root") # type: ignore[arg-type] assert sorted(t for t, _n, _d in signalled) == ["child", "grandchild", "root"] assert all(n == CONFIG_SIGNAL and d == dict(FOLLOWED_CONFIG) for _t, n, d in signalled)