diff --git a/tests/test_agent/test_confirm.py b/tests/test_agent/test_confirm.py index 858d34d..037ba5a 100644 --- a/tests/test_agent/test_confirm.py +++ b/tests/test_agent/test_confirm.py @@ -33,7 +33,7 @@ async def test_confirm_allow() -> None: danger=classify_danger, on_confirm=on_confirm, ) - assert await registry.execute("delete_path", '{"path": "a.txt"}') == "deleted a.txt" + assert await registry.execute("delete_path", '{"path": "a.txt"}') == "[approval] user (once)\ndeleted a.txt" assert len(decisions) == 1 assert await registry.execute("read_file", '{"path": "a.txt"}') == "read a.txt" assert len(decisions) == 1 diff --git a/tests/test_agent/test_parallel_tools.py b/tests/test_agent/test_parallel_tools.py index bc98ba0..ee732b8 100644 --- a/tests/test_agent/test_parallel_tools.py +++ b/tests/test_agent/test_parallel_tools.py @@ -253,5 +253,5 @@ async def test_a_write_holds_its_file_across_the_confirm() -> None: await asyncio.sleep(0) # the second write waits for the file, not for the human assert order == ["confirm write_a"] answered.set() - assert await asyncio.gather(first, second) == ["a", "b"] + assert await asyncio.gather(first, second) == ["[approval] user (once)\na", "b"] assert order == ["confirm write_a", "a", "b"] diff --git a/tests/test_agent/test_tool_approval.py b/tests/test_agent/test_tool_approval.py new file mode 100644 index 0000000..52ad848 --- /dev/null +++ b/tests/test_agent/test_tool_approval.py @@ -0,0 +1,154 @@ +"""The ``[approval]`` line: a confirmed call says how it got through.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from plyngent.agent import ( + Approval, + BatchConfirmDecision, + ToolRegistry, + ToolTag, + format_approval, + most_specific_approval, + tool, +) +from plyngent.agent.budget import cap_tool_message +from plyngent.lmproto.openai_compatible.model import ToolChatMessage +from plyngent.tools.context import SessionState, bind_tool_context, record_approval + +if TYPE_CHECKING: + from collections.abc import Mapping, Sequence + + from plyngent.agent import ToolConfirmRequest + + +def _danger(name: str, _args: Mapping[str, object]) -> str | None: + return f"mutate {name}" + + +def _allow(_name: str, _args: Mapping[str, object], _reason: str) -> bool: + return True + + +@tool(register=False) +async def quiet() -> str: + return "quiet" + + +@tool(register=False) +async def mutate() -> str: + return "ran" + + +async def test_a_call_that_needs_no_confirmation_carries_no_line() -> None: + """The common case stays byte-for-byte what the tool returned.""" + + def danger(_name: str, _args: Mapping[str, object]) -> str | None: + return None + + def on_confirm(_name: str, _args: Mapping[str, object], _reason: str) -> bool: + msg = "no prompt expected" + raise AssertionError(msg) + + reg = ToolRegistry([quiet], danger=danger, on_confirm=on_confirm) + assert await reg.execute("quiet", "{}") == "quiet" + + +async def test_a_single_prompt_leads_the_result() -> None: + reg = ToolRegistry([mutate], danger=_danger, on_confirm=_allow) + assert await reg.execute("mutate", "{}") == "[approval] user (once)\nran" + + +async def test_yolo_mode_says_it_dropped_the_prompt() -> None: + @tool(tags=ToolTag.LOCAL | ToolTag.YOLO, register=False) + async def yolo_ok() -> str: + return "y" + + reg = ToolRegistry([yolo_ok], danger=_danger, on_confirm=_allow, yolo=True) + assert await reg.execute("yolo_ok", "{}") == "[approval] yolo (auto)\ny" + + +async def test_a_batch_answer_covers_its_round() -> None: + def on_batch(_requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision: + return BatchConfirmDecision(allowed=True) + + reg = ToolRegistry([mutate], danger=_danger, on_confirm_batch=on_batch) + calls = [("mutate", "{}"), ("mutate", "{}")] + decisions = await reg.prepare_batch_confirm(calls) + results = [ + await reg.execute(name, arguments, decision=decisions[index]) for index, (name, arguments) in enumerate(calls) + ] + assert results == ["[approval] user (round)\nran", "[approval] user (round)\nran"] + + +async def test_a_session_grant_says_nobody_was_asked_this_time() -> None: + @tool(tags=ToolTag.LOCAL | ToolTag.TRUSTABLE, register=False) + async def trust_me() -> str: + return "ok" + + prompts: list[str] = [] + + def on_confirm(name: str, _args: Mapping[str, object], _reason: str) -> bool: + prompts.append(name) + return True + + session = SessionState(session_id=1) + reg = ToolRegistry([trust_me], danger=_danger, on_confirm=on_confirm, session_state=session) + with bind_tool_context(session=session): + assert await reg.execute("trust_me", "{}") == "[approval] user (once)\nok" + assert await reg.execute("trust_me", "{}") == "[approval] grant (session)\nok" + assert prompts == ["trust_me"] + + +async def test_a_denied_call_carries_no_line() -> None: + reg = ToolRegistry([mutate], danger=_danger, on_confirm=lambda *_args: False) + out = await reg.execute("mutate", "{}") + assert out.startswith("error: ") + assert "[approval]" not in out + + +async def test_a_failed_call_carries_no_line() -> None: + """An approved call that failed is an error first; the answer explains nothing.""" + + @tool(register=False) + async def boom() -> str: + return "error: boom" + + reg = ToolRegistry([boom], danger=_danger, on_confirm=_allow) + assert await reg.execute("boom", "{}") == "error: boom" + + +async def test_a_handler_record_leads_the_result() -> None: + @tool(register=False) + async def policy_tool() -> str: + record_approval(Approval("user", "process")) + return "ran" + + reg = ToolRegistry([policy_tool]) + assert await reg.execute("policy_tool", "{}") == "[approval] user (process)\nran" + + +async def test_the_closest_answer_wins() -> None: + """YOLO let the call run, but a human answered a policy prompt inside it.""" + + @tool(tags=ToolTag.LOCAL | ToolTag.YOLO, register=False) + async def yolo_policy() -> str: + record_approval(Approval("user", "process")) + return "ran" + + reg = ToolRegistry([yolo_policy], danger=_danger, on_confirm=_allow, yolo=True) + assert await reg.execute("yolo_policy", "{}") == "[approval] user (process)\nran" + + +async def test_the_line_survives_the_wire_cap() -> None: + approved = f"{format_approval(Approval('user', 'once'))}\n{'x' * 500}" + capped = cap_tool_message(ToolChatMessage(content=approved, tool_call_id="call-1"), 200) + assert capped.content.startswith("[approval] user (once)\n") + assert "document_id=call-1" in capped.content + + +def test_the_most_specific_record_wins() -> None: + records = [Approval("yolo", "auto"), Approval("grant", "session"), Approval("user", "once")] + assert most_specific_approval(records) == Approval("user", "once") + assert most_specific_approval([]) is None diff --git a/tests/test_agent/test_tools_batch_confirm.py b/tests/test_agent/test_tools_batch_confirm.py index f33ce21..05f5d84 100644 --- a/tests/test_agent/test_tools_batch_confirm.py +++ b/tests/test_agent/test_tools_batch_confirm.py @@ -26,6 +26,11 @@ def _danger(name: str, _args: Mapping[str, object]) -> str | None: return f"mutate {name}" +def _approved(scope: str, text: str) -> str: + """A result as the model reads it once a confirmation let the call run.""" + return f"[approval] {scope}\n{text}" + + async def test_loop_round_prompts_once_before_running_its_calls() -> None: """The loop aggregates a round's confirms and applies one answer to each call.""" batch_prompts: list[int] = [] @@ -54,7 +59,10 @@ async def test_loop_round_prompts_once_before_running_its_calls() -> None: assert batch_prompts == [2] assert single_prompts == [] - assert [getattr(message, "content", None) for message in messages] == ["stamped a", "stamped b"] + assert [getattr(message, "content", None) for message in messages] == [ + _approved("user (round)", "stamped a"), + _approved("user (round)", "stamped b"), + ] assert len(events) == len(tool_calls) * 2 # call event + result event per call @@ -88,7 +96,11 @@ async def test_round_asks_once_for_many_mutating_calls() -> None: results = [ await reg.execute(name, arguments, decision=decisions[index]) for index, (name, arguments) in enumerate(calls) ] - assert results == ["stamped a", "stamped b", "stamped c"] + assert results == [ + _approved("user (round)", "stamped a"), + _approved("user (round)", "stamped b"), + _approved("user (round)", "stamped c"), + ] assert ran == ["a", "b", "c"] assert single_prompts == [] @@ -129,7 +141,11 @@ async def test_yolo_calls_are_left_out_of_the_prompt() -> None: results = [ await reg.execute(name, arguments, decision=decisions[index]) for index, (name, arguments) in enumerate(calls) ] - assert results == ["stamped a", "stamped b", "stamped c"] + assert results == [ + _approved("yolo (auto)", "stamped a"), + _approved("user (round)", "stamped b"), + _approved("user (round)", "stamped c"), + ] assert single_prompts == [] @@ -153,7 +169,7 @@ async def test_single_request_keeps_the_one_call_prompt() -> None: reg = ToolRegistry([stamp], danger=_danger, on_confirm=on_single, on_confirm_batch=on_batch) assert await reg.prepare_batch_confirm([("stamp", '{"path": "a"}')]) == [None] assert batch_prompts == [] - assert await reg.execute("stamp", '{"path": "a"}') == "stamped a" + assert await reg.execute("stamp", '{"path": "a"}') == _approved("user (once)", "stamped a") assert single_prompts == ["stamp"] @@ -254,7 +270,10 @@ async def test_manual_hands_the_round_back_to_the_per_call_prompt() -> None: decisions = await reg.prepare_batch_confirm(calls) assert decisions == [None, None] - assert [await reg.execute(name, arguments) for name, arguments in calls] == ["stamped a", "stamped b"] + assert [await reg.execute(name, arguments) for name, arguments in calls] == [ + _approved("user (once)", "stamped a"), + _approved("user (once)", "stamped b"), + ] assert batch_prompts == [2] assert single_prompts == ["stamp", "stamp"] @@ -274,7 +293,10 @@ async def test_without_a_batch_hook_every_call_prompts_on_its_own() -> None: reg = ToolRegistry([stamp], danger=_danger, on_confirm=on_single) calls = [("stamp", '{"path": "a"}'), ("stamp", '{"path": "b"}')] assert await reg.prepare_batch_confirm(calls) == [None, None] - assert [await reg.execute(name, arguments) for name, arguments in calls] == ["stamped a", "stamped b"] + assert [await reg.execute(name, arguments) for name, arguments in calls] == [ + _approved("user (once)", "stamped a"), + _approved("user (once)", "stamped b"), + ] assert single_prompts == ["stamp", "stamp"] @@ -306,8 +328,9 @@ async def test_batch_allow_records_the_trustable_grant() -> None: calls = [("trust_me", '{"path": "a"}'), ("trust_me", '{"path": "b"}')] decisions = await reg.prepare_batch_confirm(calls) for path, decision in zip(("a", "b"), decisions, strict=True): - assert await reg.execute("trust_me", f'{{"path": "{path}"}}', decision=decision) == f"done {path}" + approve = _approved("user (round)", f"done {path}") + assert await reg.execute("trust_me", f'{{"path": "{path}"}}', decision=decision) == approve assert has_grant(session, "trust_me") # The TRUSTABLE grant, not the batch answer, keeps later calls unchecked. - assert await reg.execute("trust_me", '{"path": "c"}') == "done c" + assert await reg.execute("trust_me", '{"path": "c"}') == _approved("grant (session)", "done c") assert single_prompts == [] diff --git a/tests/test_cli/test_display.py b/tests/test_cli/test_display.py index b0c00e8..5be6685 100644 --- a/tests/test_cli/test_display.py +++ b/tests/test_cli/test_display.py @@ -1018,6 +1018,17 @@ def test_every_builtin_tool_has_a_pretty_line() -> None: assert missing == [] +def test_a_confirmed_result_keeps_its_pretty_detail() -> None: + """The approval line leads the model's text; a pretty detail reads past it.""" + from plyngent.cli.display import _pretty_line_for, _pretty_parts + + line = _pretty_line_for("write_file") + assert line is not None + approved = "[approval] user (once)\nwrote 120 characters to a.txt" + _prefix, detail = _pretty_parts(line, '{"path": "a.txt", "overwrite": true}', approved) + assert "(120 chars)" in detail + + async def test_pretty_skill_tools(capsys: pytest.CaptureFixture[str]) -> None: await render_events( _aiter( diff --git a/tests/test_tools/test_confirm_tags.py b/tests/test_tools/test_confirm_tags.py index c92e699..50f1e1c 100644 --- a/tests/test_tools/test_confirm_tags.py +++ b/tests/test_tools/test_confirm_tags.py @@ -33,9 +33,9 @@ async def test_yolo_only_skips_yolo_tagged_tools() -> None: on_confirm=confirm, yolo=True, ) - assert await reg.execute("yolo_ok", "{}") == "y" + assert await reg.execute("yolo_ok", "{}") == "[approval] yolo (auto)\ny" assert calls == [] - assert await reg.execute("no_yolo", "{}") == "n" + assert await reg.execute("no_yolo", "{}") == "[approval] user (once)\nn" assert calls == ["no_yolo"] @@ -55,7 +55,7 @@ async def test_yolo_off_keeps_the_yolo_tagged_prompt() -> None: return True reg = ToolRegistry([yolo_ok], danger=danger, on_confirm=confirm, yolo=False) - assert await reg.execute("yolo_ok", "{}") == "y" + assert await reg.execute("yolo_ok", "{}") == "[approval] user (once)\ny" assert calls == ["yolo_ok"] @@ -75,7 +75,7 @@ async def test_yolo_keeps_a_reason_that_refuses_the_skip() -> None: return True reg = ToolRegistry([escalated], danger=danger, on_confirm=confirm, yolo=True) - assert await reg.execute("escalated", "{}") == "x" + assert await reg.execute("escalated", "{}") == "[approval] user (once)\nx" assert seen == ["interpreter 'python' + privilege escalation sudo"] @@ -103,8 +103,10 @@ async def test_trustable_grant_once() -> None: session_state=session, ) with bind_tool_context(session=session): - assert await reg.execute("trust_me", "{}") == "ok" - assert await reg.execute("trust_me", "{}") == "ok" + assert await reg.execute("trust_me", "{}") == "[approval] user (once)\nok" + # The TRUSTABLE grant answers from here on: nobody is asked, and the + # result says so. + assert await reg.execute("trust_me", "{}") == "[approval] grant (session)\nok" assert calls == ["trust_me"] @@ -123,8 +125,8 @@ async def test_untagged_trustable_prompts_every_time() -> None: return True reg = ToolRegistry([always_ask], danger=danger, on_confirm=confirm, yolo=False) - assert await reg.execute("always_ask", "{}") == "ok" - assert await reg.execute("always_ask", "{}") == "ok" + assert await reg.execute("always_ask", "{}") == "[approval] user (once)\nok" + assert await reg.execute("always_ask", "{}") == "[approval] user (once)\nok" assert calls == ["always_ask", "always_ask"] @@ -151,7 +153,7 @@ async def test_trustable_grant_persists_to_session_data() -> None: yolo=False, session_state=session, ) - assert await reg.execute("trust_me", "{}") == "ok" + assert await reg.execute("trust_me", "{}") == "[approval] user (once)\nok" assert has_grant(session, "trust_me") loaded = await store.load() assert isinstance(loaded, dict) @@ -170,7 +172,7 @@ async def test_trustable_grant_persists_to_session_data() -> None: yolo=False, session_state=session2, ) - assert await reg2.execute("trust_me", "{}") == "ok" + assert await reg2.execute("trust_me", "{}") == "[approval] grant (session)\nok" assert calls == ["trust_me"] diff --git a/tests/test_tools/test_fetch_local.py b/tests/test_tools/test_fetch_local.py index f5057bc..bd07c5d 100644 --- a/tests/test_tools/test_fetch_local.py +++ b/tests/test_tools/test_fetch_local.py @@ -188,6 +188,35 @@ def test_classify_danger_fetch() -> None: assert isinstance(reason2, str) and "DELETE" in reason2 +async def test_a_private_host_answer_is_recorded_as_a_user_process_approval(workspace: Path) -> None: + """The prompt that opened loopback for this process is what the result says.""" + del workspace + from plyngent.agent import Approval + from plyngent.tools.context import collect_approvals + from plyngent.tools.net.grants import ensure_host_allowed, set_fetch_policy_confirm_hook + + set_fetch_policy_confirm_hook(lambda *_args: True) + try: + with collect_approvals() as recorded: + await ensure_host_allowed("http://127.0.0.1:9/") + finally: + set_fetch_policy_confirm_hook(None) + assert recorded == [Approval("user", "process")] + + +async def test_a_private_host_grant_is_recorded_as_a_process_grant(workspace: Path) -> None: + """A second fetch to the granted host asks nobody, and says so.""" + del workspace + from plyngent.agent import Approval + from plyngent.tools.context import collect_approvals + from plyngent.tools.net.grants import ensure_host_allowed + + grant_private_host("127.0.0.1", 9) + with collect_approvals() as recorded: + await ensure_host_allowed("http://127.0.0.1:9/") + assert recorded == [Approval("grant", "process")] + + async def test_fetch_get_post_put_delete(workspace: Path, http_server: str) -> None: del workspace base = http_server diff --git a/tests/test_tools/test_skills.py b/tests/test_tools/test_skills.py index c8e16b5..05b77a5 100644 --- a/tests/test_tools/test_skills.py +++ b/tests/test_tools/test_skills.py @@ -344,6 +344,30 @@ async def test_skill_write_session_grant_is_asked_once(tmp_path: Path) -> None: assert (root / "acme" / "notes.md").read_text(encoding="utf-8") == "note\n" +async def test_a_skill_write_says_which_answer_let_it_through(tmp_path: Path) -> None: + """Once, this chat, and the standing config each read differently.""" + from plyngent.agent import Approval + from plyngent.tools.context import collect_approvals + + root = _write_root(tmp_path) + with _bound(tmp_path, root, policy=SkillWritePolicy(confirm=lambda *_args: True)), collect_approvals() as once: + created = await skill_create.handler("once-skill", "d", "b") + + def for_the_chat(*_args: object) -> SkillWriteDecision: + return SkillWriteDecision(allow=True, session=True) + + with _bound(tmp_path, root, policy=SkillWritePolicy(confirm=for_the_chat)), collect_approvals() as chat: + _ = await skill_create.handler("chat-skill", "d", "b") + _ = await skill_edit.handler("chat-skill", append="\nMore.\n") + + with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)), collect_approvals() as config: + _ = await skill_create.handler("config-skill", "d", "b") + + assert once == [Approval("user", "once")], created + assert chat == [Approval("user", "session"), Approval("grant", "session")] + assert config == [Approval("config", "policy")] + + async def test_skill_write_denylist_wins_over_a_grant(tmp_path: Path) -> None: root = _write_root(tmp_path) with _bound(tmp_path, root, policy=SkillWritePolicy(allow_all=True)): diff --git a/tests/test_tools/test_workspace.py b/tests/test_tools/test_workspace.py index fbd8a3a..4690412 100644 --- a/tests/test_tools/test_workspace.py +++ b/tests/test_tools/test_workspace.py @@ -235,6 +235,41 @@ def test_command_denylist_policy_confirm_allow(workspace: object) -> None: clear_policy_allowed_commands() +def test_a_denylist_answer_is_recorded_as_a_user_process_approval(workspace: object) -> None: + """The answer that let the call through is what the result has to say.""" + del workspace + from plyngent.agent import Approval + from plyngent.tools.context import collect_approvals + from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook + + set_policy_confirm_hook(lambda *_args: True) + clear_policy_allowed_commands() + try: + with collect_approvals() as recorded: + check_command_allowed(["sudo", "id"]) + finally: + set_policy_confirm_hook(None) + clear_policy_allowed_commands() + assert recorded == [Approval("user", "process")] + + +def test_a_matching_denylist_grant_is_recorded_as_a_process_grant(workspace: object) -> None: + """Nobody is asked the second time, and the result says so.""" + del workspace + from plyngent.agent import Approval + from plyngent.tools.context import collect_approvals + from plyngent.tools.workspace import clear_policy_allowed_commands, grant_policy_command + + clear_policy_allowed_commands() + grant_policy_command("sudo") + try: + with collect_approvals() as recorded: + check_command_allowed(["sudo", "id"]) + finally: + clear_policy_allowed_commands() + assert recorded == [Approval("grant", "process")] + + def test_command_denylist_policy_confirm_deny(workspace: object) -> None: del workspace from plyngent.tools.workspace import (