tests/plugins/test_hooks.py
Rama main · réplica de lectura del repositorio original.
"""Claude Code's command hooks, end to end with real ``sh`` commands: what
each event receives, adds for the model, and blocks."""
from __future__ import annotations
import asyncio
import json
import os
import signal
from pathlib import Path
from typing import Any
import pytest
from noeta.backends.execenvs.local import LocalExecEnv
from noeta.backends.stores.sqlite import SqliteStore
from noeta.core.agent import Agent
from noeta.core.client import Client
from noeta.core.errors import ContextOverflow
from noeta.core.events import Compacted, ExtensionRecorded, ToolFinished, Waiting
from noeta.core.extension import Extension, TaskView
from noeta.core.loop.state import fold
from noeta.core.runner import Runtime
from noeta.core.tool import Tool, ToolContext, ToolOutput, text_output
from noeta.plugins.compaction.transcript import full_conversation
from noeta.plugins.fs import FileTools
from noeta.plugins.hooks import Hooks, parse_hooks
from noeta.plugins.hooks.config import HookCommand
from noeta.plugins.hooks.runners import OUTPUT_CAP, run_command
from noeta.plugins.markdown_commands import MarkdownCommands
from noeta.plugins.skills import Skills
from noeta.testing import ScriptedProvider, reply
async def _no_sleep(_: float) -> None:
return None
def _hook(command: str, matcher: str = "") -> list[dict[str, Any]]:
return [{"matcher": matcher, "hooks": [{"type": "command", "command": command}]}]
async def _run(tmp_path: Path, hooks: dict[str, Any], steps: list[Any],
prompts: tuple[str, ...] = ("go",)) -> tuple[ScriptedProvider, Client, str]:
work = tmp_path / "w"
work.mkdir(exist_ok=True)
(work / "a.txt").write_text("hello\n")
store = SqliteStore(str(tmp_path / "t.db"))
provider = ScriptedProvider(steps)
agent = Agent(name="a", model="m", extensions=[Hooks(hooks=hooks), FileTools()])
rt = Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider,
env_for=lambda _s: LocalExecEnv(str(work)), sleep=_no_sleep)
client = Client(store)
tid = await client.create("a", prompts[0], config={"cwd": str(work)})
await rt.run_until_idle()
for p in prompts[1:]:
await client.send(tid, p)
await rt.run_until_idle()
return provider, client, tid
def _system(provider: ScriptedProvider, i: int) -> list[str]:
return [m.text for m in provider.requests[i].messages if m.origin == "system"]
async def test_user_prompt_submit_adds_context_and_sees_the_prompt(tmp_path: Path) -> None:
seen = tmp_path / "seen.json"
hooks = {"UserPromptSubmit": _hook(f"cat > {seen}; echo 'The build is green.'")}
provider, _client, _tid = await _run(tmp_path, hooks, [reply("a"), reply("b")],
prompts=("first", "second"))
green = "UserPromptSubmit hook success: The build is green."
assert _system(provider, 0) == [green]
assert _system(provider, 1) == [green, green]
payload = json.loads(seen.read_text())
assert payload["hook_event_name"] == "UserPromptSubmit"
assert payload["prompt"] == "second" # only the new message
assert payload["cwd"].endswith("/w")
async def test_user_prompt_submit_can_block_the_turn(tmp_path: Path) -> None:
hooks = {"UserPromptSubmit": _hook("echo 'no secrets in prompts' >&2; exit 2")}
provider, client, tid = await _run(tmp_path, hooks, [])
assert provider.requests == []
failed = [r.event for r in await client.events(tid)
if isinstance(r.event, ExtensionRecorded) and r.event.kind == "turn_failed"]
assert failed and failed[0].data["detail"] == "no secrets in prompts"
async def test_a_blocked_prompt_is_erased_from_later_requests(tmp_path: Path) -> None:
hooks = {"UserPromptSubmit": _hook("grep -q SECRET && { echo blocked >&2; exit 2; }; exit 0")}
provider, client, tid = await _run(tmp_path, hooks, [reply("ok")],
prompts=("my SECRET is 42", "hello"))
assert len(provider.requests) == 1
texts = [m.text for m in provider.requests[0].messages]
assert any("hello" in t for t in texts)
assert not any("SECRET" in t for t in texts)
# Nor is it in the whole conversation a compaction's transcript saves.
whole = await full_conversation(client.events, tid)
assert any("hello" in m.text for m in whole)
assert not any("SECRET" in m.text for m in whole)
async def _blocked_expansion(tmp_path: Path, extension: Any, prompt: str) -> list[str]:
hooks = {"UserPromptSubmit": _hook("grep -q SECRET && { echo blocked >&2; exit 2; }; "
"exit 0")}
work = tmp_path / "w"
store = SqliteStore(str(tmp_path / "t.db"))
provider = ScriptedProvider([reply("ok")])
agent = Agent(name="a", model="m", extensions=[extension, Hooks(hooks=hooks), FileTools()])
rt = Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider,
env_for=lambda _s: LocalExecEnv(str(work)), sleep=_no_sleep)
client = Client(store)
tid = await client.create("a", prompt, config={"cwd": str(work)})
await rt.run_until_idle()
await client.send(tid, "hello")
await rt.run_until_idle()
assert len(provider.requests) == 1
return [m.text for m in provider.requests[0].messages]
async def test_a_blocked_command_loses_its_expansion_too(tmp_path: Path) -> None:
(tmp_path / "w").mkdir()
commands = MarkdownCommands([{"name": "deploy", "body": "Deploy with key $ARGUMENTS now."}])
texts = await _blocked_expansion(tmp_path, commands, "/deploy SECRET42")
assert any("hello" in t for t in texts)
assert not any("SECRET" in t for t in texts)
async def test_a_blocked_skill_command_loses_its_expansion_too(tmp_path: Path) -> None:
skill = tmp_path / "w" / ".noeta" / "skills" / "deploy"
skill.mkdir(parents=True)
(skill / "SKILL.md").write_text("---\nname: deploy\ndescription: Ship it.\n---\n"
"Deploy with key $ARGUMENTS now.\n")
skills = Skills(global_dir=None, global_agents_dir=None)
texts = await _blocked_expansion(tmp_path, skills, "/deploy SECRET42")
assert any("hello" in t for t in texts)
assert not any("SECRET" in t for t in texts)
class _MakeRoom(Extension):
"""Compacts the whole conversation when the model reports an overflow."""
name = "room"
async def on_context_overflow(self, view: TaskView, request: Any) -> bool:
view.compacted(Compacted(summary="(summary)", through=view.state.last_seq,
lead=await view.residents(lead=True, keeps_tail=False),
after=await view.residents(keeps_tail=False)))
return True
async def test_a_blocked_skill_command_leaves_no_activation_behind(tmp_path: Path) -> None:
skill = tmp_path / "w" / ".noeta" / "skills" / "deploy"
skill.mkdir(parents=True)
(skill / "SKILL.md").write_text("---\nname: deploy\ndescription: Ship it.\n"
"allowed-tools: Bash(rm:*)\n---\n"
"Deploy with key $ARGUMENTS now.\n")
hooks = {"UserPromptSubmit": _hook("grep -q SECRET && { echo blocked >&2; exit 2; }; "
"exit 0")}
skills = Skills(global_dir=None, global_agents_dir=None)
store = SqliteStore(str(tmp_path / "t.db"))
provider = ScriptedProvider([
ContextOverflow("too long"), reply("fits"),
reply("", ("Skill", {"skill": "deploy", "args": "SECRET42"})), reply("done")])
agent = Agent(name="a", model="m",
extensions=[skills, Hooks(hooks=hooks), FileTools(), _MakeRoom()])
rt = Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider,
env_for=lambda _s: LocalExecEnv(str(tmp_path / "w")), sleep=_no_sleep)
client = Client(store)
tid = await client.create("a", "/deploy SECRET42", config={"cwd": str(tmp_path / "w")})
await rt.run_until_idle()
state = fold(await client.events(tid), {"skills": skills.reduce})
assert state.ext["skills"]["active"] == {}
await client.send(tid, "hello")
await rt.run_until_idle()
# The request rebuilt after the compaction re-attaches no blocked skill.
assert not any("SECRET" in m.text for m in provider.requests[1].messages)
await client.send(tid, "use it")
await rt.run_until_idle()
kinds = [r.event.kind for r in await client.events(tid)
if isinstance(r.event, ExtensionRecorded) and r.event.extension == "skills"]
# The model's own call with the same arguments loads the skill in full.
assert "reinvoked" not in kinds and kinds[-1] == "activated"
loaded = provider.requests[-1].messages[-1].text
assert "Deploy with key SECRET42 now." in loaded and "earlier" not in loaded
async def test_a_blocked_command_leaves_no_pre_approval_behind(tmp_path: Path) -> None:
(tmp_path / "w").mkdir()
commands = MarkdownCommands([{"name": "go", "body": "go $ARGUMENTS",
"allowed_tools": "Bash(rm:*)"}])
await _blocked_expansion(tmp_path, commands, "/go SECRET99")
store = SqliteStore(str(tmp_path / "t.db"))
client = Client(store)
[tid] = [t.task_id for t in await client.tasks()]
state = fold(await client.events(tid), {"markdown_commands": commands.reduce})
assert state.ext["markdown_commands"]["allow_rules"] == []
class _Interject(Extension):
"""A ``Say`` tool: the person's ``input["text"]`` arrives mid-turn."""
name = "interject"
def __init__(self, client: Client) -> None:
self.client = client
def tools(self, view: TaskView) -> list[Tool]:
return [_SayTool(self.client)]
class _SayTool:
name = "Say"
description = "Say."
input_schema: dict[str, Any] = {"type": "object"}
read_only = True
clearable = False
def __init__(self, client: Client) -> None:
self.client = client
async def run(self, input: dict[str, Any], ctx: ToolContext) -> ToolOutput:
await self.client.send(ctx.task_id, input["text"], when="now")
return text_output("said")
async def _mid_turn(tmp_path: Path, steps: list[Any],
prompts: tuple[str, ...]) -> ScriptedProvider:
hooks = {"UserPromptSubmit": _hook("grep -q SECRET && { echo blocked >&2; exit 2; }; "
"exit 0")}
work = tmp_path / "w"
work.mkdir()
store = SqliteStore(str(tmp_path / "t.db"))
provider = ScriptedProvider(steps)
client = Client(store)
agent = Agent(name="a", model="m", extensions=[_Interject(client), Hooks(hooks=hooks)])
rt = Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider,
env_for=lambda _s: LocalExecEnv(str(work)), sleep=_no_sleep)
tid = await client.create("a", prompts[0], config={"cwd": str(work)})
await rt.run_until_idle()
for p in prompts[1:]:
await client.send(tid, p)
await rt.run_until_idle()
return provider
async def test_a_message_sent_mid_turn_goes_through_the_hook(tmp_path: Path) -> None:
provider = await _mid_turn(tmp_path, [reply("", ("Say", {"text": "my SECRET is 42"})),
reply("ok")], ("go", "hello"))
assert all("SECRET" not in m.text for r in provider.requests for m in r.messages)
assert any(m.text == "hello" for m in provider.requests[-1].messages)
async def test_a_blocked_prompt_leaves_what_the_model_already_saw(tmp_path: Path) -> None:
provider = await _mid_turn(tmp_path, [reply("", ("Say", {"text": "also check docs"})),
reply("done"), reply("ok")],
("go", "my SECRET is 42", "hello"))
assert len(provider.requests) == 3
texts = [m.text for m in provider.requests[-1].messages]
assert "also check docs" in texts and "hello" in texts
assert not any("SECRET" in t for t in texts)
async def test_a_prompt_hook_added_later_checks_only_what_the_model_has_not_seen(
tmp_path: Path) -> None:
work = tmp_path / "w"
work.mkdir()
store = SqliteStore(str(tmp_path / "t.db"))
provider = ScriptedProvider([reply("noted"), reply("ok")])
client = Client(store)
def runtime(hooks: dict[str, Any]) -> Runtime:
agent = Agent(name="a", model="m", extensions=[Hooks(hooks=hooks)])
return Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider,
env_for=lambda _s: LocalExecEnv(str(work)), sleep=_no_sleep)
tid = await client.create("a", "my SECRET is 42", config={"cwd": str(work)})
await runtime({}).run_until_idle()
seen = tmp_path / "seen.json"
blocking = {"UserPromptSubmit": _hook(f"tee {seen} | grep -q SECRET && "
"{ echo blocked >&2; exit 2; }; exit 0")}
await client.send(tid, "hello")
await runtime(blocking).run_until_idle()
assert json.loads(seen.read_text())["prompt"] == "hello"
assert len(provider.requests) == 2
assert [m.text for m in provider.requests[1].messages
if m.role == "user"][-2:] == ["my SECRET is 42", "hello"]
async def test_pre_tool_use_blocks_matching_calls_only(tmp_path: Path) -> None:
hooks = {"PreToolUse": _hook("echo 'reads are off today' >&2; exit 2", matcher="Read|Glob")}
steps = [reply("", ("Read", {"file_path": "a.txt"}), ("Write", {"file_path": "b.txt", "content": "x"})),
reply("done")]
_provider, client, tid = await _run(tmp_path, hooks, steps)
results = {r.event.result.call_id: r.event.result for r in await client.events(tid)
if isinstance(r.event, ToolFinished)}
read, write = results.values()
assert read.is_error and "reads are off today" in read.content[0].text # type: ignore[union-attr]
assert not write.is_error
async def test_pre_tool_use_json_can_ask(tmp_path: Path) -> None:
out = json.dumps({"hookSpecificOutput": {"permissionDecision": "ask",
"permissionDecisionReason": "Read a.txt?"}})
told = tmp_path / "told.json"
hooks = {"PreToolUse": _hook(f"echo '{out}'"), "Notification": _hook(f"cat > {told}")}
_provider, client, tid = await _run(tmp_path, hooks,
[reply("", ("Read", {"file_path": "a.txt"}))])
waits = [r.event for r in await client.events(tid) if isinstance(r.event, Waiting)]
assert waits[-1].on == "approval"
for _ in range(100): # the Notification hook is not waited for
if told.exists() and told.read_text():
break
await asyncio.sleep(0.02)
told_json = json.loads(told.read_text())
assert told_json["message"] == "Noeta needs your permission to use Read"
assert told_json["notification_type"] == "permission_prompt"
async def test_post_tool_use_feedback_follows_the_result(tmp_path: Path) -> None:
hooks = {"PostToolUse": _hook("echo 'lint: 2 warnings' >&2; exit 2", matcher="Read")}
provider, _c, _t = await _run(tmp_path, hooks,
[reply("", ("Read", {"file_path": "a.txt"})), reply("ok")])
second = provider.requests[1].messages
assert second[-1].origin == "system"
assert second[-1].text == ("PostToolUse:Read hook blocking error from command: "
"\"echo 'lint: 2 warnings' >&2; exit 2\": lint: 2 warnings")
assert second[-2].role == "user" and second[-2].content[0].call_id # type: ignore[union-attr]
async def test_stop_hook_keeps_the_model_going_then_lets_it_stop(tmp_path: Path) -> None:
marker = tmp_path / "ran_once"
script = (f"if [ -f {marker} ]; then exit 0; fi; touch {marker}; "
"echo 'Run the tests before you stop.' >&2; exit 2")
hooks = {"Stop": _hook(script)}
provider, client, tid = await _run(tmp_path, hooks, [reply("done"), reply("tests pass")])
assert len(provider.requests) == 2
assert provider.requests[1].messages[-1].text == (
f'Stop hook blocking error from command: "{script}": Run the tests before you stop.')
assert (await client.get(tid)).status == "waiting"
async def test_stop_hook_that_always_blocks_is_capped(tmp_path: Path) -> None:
hooks = {"Stop": _hook("echo again >&2; exit 2")}
provider, _c, _t = await _run(tmp_path, hooks, [reply(str(i)) for i in range(20)])
assert len(provider.requests) == 9 # the first answer plus eight continuations
def test_parse_and_settings_files(tmp_path: Path) -> None:
settings = tmp_path / "settings.json"
settings.write_text(json.dumps({"hooks": {"Stop": _hook("true"), "Bogus": _hook("x")}}))
hooks = Hooks(settings_files=[settings, tmp_path / "missing.json"],
hooks={"Stop": _hook("false")})
assert [h.command for m in hooks.matchers["Stop"] for h in m.hooks] == ["true", "false"]
with pytest.raises(ValueError):
parse_hooks({"Stop": [{"hooks": [{"type": "command"}]}]})
[m] = parse_hooks({"PreToolUse": _hook("x", matcher="Edit|Write")})["PreToolUse"]
assert m.matches("Edit") and m.matches("Write") and not m.matches("EditX")
async def test_a_cancelled_command_hook_kills_its_process_group(tmp_path: Path) -> None:
pid_file = tmp_path / "pid"
hook = HookCommand(command=f"sleep 30 & echo $! > {pid_file}; wait", timeout=60)
run = asyncio.create_task(run_command(hook, {}, str(tmp_path)))
for _ in range(500):
if pid_file.exists() and pid_file.read_text().strip():
break
await asyncio.sleep(0.01)
pid = int(pid_file.read_text())
run.cancel()
with pytest.raises(asyncio.CancelledError):
await run
for _ in range(200):
try:
os.kill(pid, 0)
except ProcessLookupError:
break
await asyncio.sleep(0.01)
else:
os.kill(pid, signal.SIGKILL)
pytest.fail("the hook's child outlived the cancelled hook")
async def test_a_command_hook_keeps_only_the_head_of_a_flood(tmp_path: Path) -> None:
hook = HookCommand(command="head -c 3000000 /dev/zero | tr '\\0' x; echo oops >&2")
result = await run_command(hook, {"big": "x" * 200_000}, str(tmp_path))
assert result.exit_code == 0
assert result.stdout == "x" * OUTPUT_CAP and result.stderr == "oops\n"
async def test_a_command_hook_inherits_the_environment_but_noetas_credentials(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("JAVA_HOME", "/opt/jdk")
monkeypatch.setenv("NOETA_SERVER_TOKEN", "t")
hook = HookCommand(command='printf "%s|%s" "$JAVA_HOME" "${NOETA_SERVER_TOKEN:-unset}"')
result = await run_command(hook, {}, str(tmp_path))
assert result.stdout == "/opt/jdk|unset"
@pytest.mark.parametrize("turns", [0, 3])
async def test_a_prompt_hook_checks_a_continued_tasks_prompts(tmp_path: Path,
turns: int) -> None:
"""A task that continues a conversation (``history_from``) runs its
blocking prompt hooks on its own prompts: the source's checked position
is a seq of the source's log, past the new log's first prompts."""
work = tmp_path / "w"
work.mkdir()
hooks = {"UserPromptSubmit": _hook("grep -q SECRET && { echo blocked >&2; exit 2; }; exit 0")}
store = SqliteStore(str(tmp_path / "t.db"))
provider = ScriptedProvider([reply(f"r{i}") for i in range(10)])
agent = Agent(name="a", model="m", extensions=[Hooks(hooks=hooks), FileTools()])
rt = Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider,
env_for=lambda _s: LocalExecEnv(str(work)), sleep=_no_sleep)
client = Client(store)
tid = await client.create("a", "hello", config={"cwd": str(work)})
await rt.run_until_idle()
for i in range(turns):
await client.send(tid, f"more {i}")
await rt.run_until_idle()
n = len(provider.requests)
await client.create("a", "my SECRET is 42", config={"cwd": str(work)}, history_from=tid)
await rt.run_until_idle()
assert not [r for r in provider.requests[n:] if any("SECRET" in m.text for m in r.messages)]