From 4bab7806031bf937152ab4d62b66ca34b0584365 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Thu, 29 Oct 2026 23:43:49 +0800 Subject: [PATCH] test: cover declined, timed out, cancelled and absent prompts The four kinds and the words each produces, at the registry level and through the four policy gates: a timeout names its deadline, a decline carries the comment, and neither is reported as the other. The existing denial assertions now expect the specific clause. --- tests/test_agent/test_tool_denial.py | 71 ++++++++++++++++++++ tests/test_agent/test_tools_batch_confirm.py | 3 +- tests/test_cli/test_grants_cmd.py | 4 +- tests/test_cli/test_limits.py | 25 +++++-- tests/test_cli/test_limits_batch_confirm.py | 6 +- tests/test_tools/test_access.py | 19 +++++- tests/test_tools/test_fetch_local.py | 15 +++++ tests/test_tools/test_skills.py | 18 ++++- tests/test_tools/test_workspace.py | 19 +++++- 9 files changed, 165 insertions(+), 15 deletions(-) create mode 100644 tests/test_agent/test_tool_denial.py diff --git a/tests/test_agent/test_tool_denial.py b/tests/test_agent/test_tool_denial.py new file mode 100644 index 0000000..862382b --- /dev/null +++ b/tests/test_agent/test_tool_denial.py @@ -0,0 +1,71 @@ +"""A denial says which no it was: the human's, the clock's, or nobody's.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from plyngent.agent import Denial, DenialKind, ToolRegistry, tool + +if TYPE_CHECKING: + from collections.abc import Mapping + + +def _danger(name: str, _args: Mapping[str, object]) -> str | None: + return f"mutate {name}" + + +@tool(register=False) +async def mutate() -> str: + return "ran" + + +def _registry(answer: object) -> ToolRegistry: + """A registry whose one soft confirm answers *answer*.""" + return ToolRegistry([mutate], danger=_danger, on_confirm=lambda *_args: answer) + + +async def test_a_declined_prompt_says_the_user_declined() -> None: + out = await _registry(Denial(DenialKind.DECLINED)).execute("mutate", "{}") + assert out.startswith("error: tool 'mutate' denied (") + assert "user declined" in out + assert "no answer" not in out + + +async def test_a_timeout_names_its_deadline_instead_of_the_user() -> None: + out = await _registry(Denial(DenialKind.TIMEOUT, seconds=30.0)).execute("mutate", "{}") + assert "no answer within 30s — denied by default" in out + assert "user declined" not in out + + +async def test_a_cancelled_prompt_says_so() -> None: + out = await _registry(Denial(DenialKind.CANCELLED)).execute("mutate", "{}") + assert "prompt cancelled" in out + + +async def test_no_prompt_says_nobody_was_asked() -> None: + out = await _registry(Denial(DenialKind.NO_PROMPT)).execute("mutate", "{}") + assert "no confirm hook or non-interactive" in out + + +async def test_a_bare_comment_still_reads_as_a_declined_human() -> None: + out = await _registry("my tax data").execute("mutate", "{}") + assert "user declined" in out + assert "user comment: my tax data" in out + + +def test_a_timeout_without_a_deadline_still_says_default_deny() -> None: + assert Denial(DenialKind.TIMEOUT).clause() == "no answer — denied by default" + + +def test_the_denial_clause_never_claims_a_human() -> None: + """A timeout must not be reported as a decision, nor an absent prompt as one.""" + assert "user" not in Denial(DenialKind.TIMEOUT, seconds=5.0).clause() + assert "user" not in Denial(DenialKind.NO_PROMPT).clause() + + +def test_from_answer_keeps_a_denial_and_reads_a_comment() -> None: + timeout = Denial(DenialKind.TIMEOUT, seconds=1.0) + assert Denial.from_answer(timeout) is timeout + assert Denial.from_answer("why not") == Denial(comment="why not") + assert Denial.from_answer(None) == Denial(DenialKind.DECLINED) + assert Denial.from_answer(False) == Denial(DenialKind.DECLINED) diff --git a/tests/test_agent/test_tools_batch_confirm.py b/tests/test_agent/test_tools_batch_confirm.py index 05f5d84..dc31703 100644 --- a/tests/test_agent/test_tools_batch_confirm.py +++ b/tests/test_agent/test_tools_batch_confirm.py @@ -222,7 +222,8 @@ async def test_batch_deny_reaches_every_call_with_one_comment() -> None: results = [ await reg.execute(name, arguments, decision=decisions[index]) for index, (name, arguments) in enumerate(calls) ] - assert all(result.startswith("error: tool 'stamp' denied by user confirm") for result in results) + assert all(result.startswith("error: tool 'stamp' denied (") for result in results) + assert all("user declined" in result for result in results) assert all("user comment: not these two" in result for result in results) assert ran == [] diff --git a/tests/test_cli/test_grants_cmd.py b/tests/test_cli/test_grants_cmd.py index d9221f5..3264e36 100644 --- a/tests/test_cli/test_grants_cmd.py +++ b/tests/test_cli/test_grants_cmd.py @@ -8,6 +8,7 @@ from typing import TYPE_CHECKING import pytest import tomlkit +from plyngent.agent import Denial, DenialKind from plyngent.cli.slash import handle_slash from plyngent.cli.state import ReplState from plyngent.config.models import DatabaseConfig, OpenAIProvider @@ -154,7 +155,8 @@ def test_yolo_denies_the_directory_access_prompt_when_nobody_answers( ) -> None: state.set_yolo("on") with temporary_backend(NonInteractiveBackend()): - assert state.directory_access_confirm_hook(tmp_path, AccessMode.READ, "", 1.0) is None + answer = state.directory_access_confirm_hook(tmp_path, AccessMode.READ, "", 1.0) + assert answer == Denial(DenialKind.NO_PROMPT) async def test_config_allow_paths_installed( diff --git a/tests/test_cli/test_limits.py b/tests/test_cli/test_limits.py index b10ab56..98692f0 100644 --- a/tests/test_cli/test_limits.py +++ b/tests/test_cli/test_limits.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING import pytest +from plyngent.agent import Denial, DenialKind from plyngent.cli.display import close_open_pretty_line from plyngent.cli.limits import ( auto_continue_enabled, @@ -130,7 +131,8 @@ def test_pty_offer_raise_uses_auto_continue(clean_auto_continue: None) -> None: def test_prompt_confirm_tool_default_deny() -> None: backend = ScriptedBackend([], confirms=[False]) with temporary_backend(backend): - assert prompt_confirm_tool("delete_path", {"path": "x"}, "delete path 'x'") is False + answer = prompt_confirm_tool("delete_path", {"path": "x"}, "delete path 'x'") + assert answer == Denial(DenialKind.DECLINED) def test_format_tool_confirm_box_multiline() -> None: @@ -159,7 +161,7 @@ def test_policy_confirm_noninteractive_denies() -> None: from plyngent.prompting import NonInteractiveBackend with temporary_backend(NonInteractiveBackend()): - assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) is False + assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) == Denial(DenialKind.NO_PROMPT) def test_directory_access_prompt_noninteractive_denies() -> None: @@ -170,7 +172,9 @@ def test_directory_access_prompt_noninteractive_denies() -> None: from plyngent.tools.workspace import AccessMode with temporary_backend(NonInteractiveBackend()): - assert prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 1.0) is None + assert prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 1.0) == Denial( + DenialKind.NO_PROMPT + ) def test_directory_access_prompt_yes_and_level(monkeypatch: pytest.MonkeyPatch) -> None: @@ -191,9 +195,14 @@ def test_directory_access_prompt_yes_and_level(monkeypatch: pytest.MonkeyPatch) AccessMode.EXEC ) monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "n\n") - assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) is None + assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == Denial( + DenialKind.DECLINED + ) + # Silence is its own answer: the deadline is named, not the human. monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: None) - assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) is None + assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == Denial( + DenialKind.TIMEOUT, seconds=30.0 + ) class _AnswerSource: @@ -243,7 +252,9 @@ def test_directory_access_denied_by_the_notification() -> None: from plyngent.tools.workspace import AccessMode with temporary_backend(ScriptedBackend([])), _answered("deny"): - assert limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0) is None + assert limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0) == Denial( + DenialKind.DECLINED + ) def test_tool_confirm_takes_a_notification_answer() -> None: @@ -253,7 +264,7 @@ def test_tool_confirm_takes_a_notification_answer() -> None: def test_tool_confirm_denied_by_the_notification() -> None: with temporary_backend(ScriptedBackend([""])), _answered("deny"): - assert prompt_confirm_tool("run_argv", {}, "runs a command") is False + assert prompt_confirm_tool("run_argv", {}, "runs a command") == Denial(DenialKind.DECLINED) @pytest.mark.skipif(sys.platform == "win32", reason="a pty is POSIX-only") diff --git a/tests/test_cli/test_limits_batch_confirm.py b/tests/test_cli/test_limits_batch_confirm.py index b00a055..2691a11 100644 --- a/tests/test_cli/test_limits_batch_confirm.py +++ b/tests/test_cli/test_limits_batch_confirm.py @@ -2,7 +2,7 @@ from __future__ import annotations -from plyngent.agent import BatchConfirmDecision, ToolConfirmRequest +from plyngent.agent import BatchConfirmDecision, Denial, DenialKind, ToolConfirmRequest from plyngent.cli.limits import ( _prompt_confirm_batch_sync, format_batch_confirm_box, @@ -69,13 +69,13 @@ def test_batch_prompt_denies_with_a_comment() -> None: def test_batch_prompt_defaults_to_deny() -> None: with temporary_backend(ScriptedBackend([])): decision = _prompt_confirm_batch_sync(_requests()) - assert decision == BatchConfirmDecision(allowed=False) + assert decision == BatchConfirmDecision(allowed=False, denial=Denial(DenialKind.DECLINED)) def test_batch_prompt_denies_when_not_interactive() -> None: with temporary_backend(NonInteractiveBackend()): decision = _prompt_confirm_batch_sync(_requests()) - assert decision == BatchConfirmDecision(allowed=False) + assert decision == BatchConfirmDecision(allowed=False, denial=Denial(DenialKind.NO_PROMPT)) async def test_batch_prompt_async_allows() -> None: diff --git a/tests/test_tools/test_access.py b/tests/test_tools/test_access.py index 53774e7..93dfa90 100644 --- a/tests/test_tools/test_access.py +++ b/tests/test_tools/test_access.py @@ -123,7 +123,24 @@ async def test_deny_none( outside = tmp_path_factory.mktemp("timeout").resolve() set_directory_access_confirm_hook(lambda *_a: None) out = await call_async(request_directory_access, str(outside)) - assert "denied (declined or timed out" in out + assert "denied (user declined)" in out + assert session.access_grants == {} + + +async def test_a_prompt_timeout_is_not_reported_as_a_decision( + workspace: object, + session: SessionState, + tmp_path_factory: pytest.TempPathFactory, +) -> None: + """The deadline the prompt waited out is what the model reads.""" + from plyngent.agent import Denial, DenialKind + + del workspace + outside = tmp_path_factory.mktemp("silence").resolve() + set_directory_access_confirm_hook(lambda *_a: Denial(DenialKind.TIMEOUT, seconds=12.0)) + out = await call_async(request_directory_access, str(outside)) + assert "denied (no answer within 12s — denied by default)" in out + assert "user" not in out assert session.access_grants == {} diff --git a/tests/test_tools/test_fetch_local.py b/tests/test_tools/test_fetch_local.py index bd07c5d..8eb28b0 100644 --- a/tests/test_tools/test_fetch_local.py +++ b/tests/test_tools/test_fetch_local.py @@ -204,6 +204,21 @@ async def test_a_private_host_answer_is_recorded_as_a_user_process_approval(work assert recorded == [Approval("user", "process")] +async def test_a_private_host_timeout_names_the_deadline(workspace: Path) -> None: + """A prompt that nobody answered says the clock, not a human's no.""" + del workspace + from plyngent.agent import Denial, DenialKind + from plyngent.tools.net.grants import ensure_host_allowed, set_fetch_policy_confirm_hook + from plyngent.tools.net.policy import FetchPolicyError + + set_fetch_policy_confirm_hook(lambda *_args: Denial(DenialKind.TIMEOUT, seconds=7.0)) + try: + with pytest.raises(FetchPolicyError, match="no answer within 7s"): + await ensure_host_allowed("http://127.0.0.1:9/") + finally: + set_fetch_policy_confirm_hook(None) + + 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 diff --git a/tests/test_tools/test_skills.py b/tests/test_tools/test_skills.py index 05b77a5..be099ca 100644 --- a/tests/test_tools/test_skills.py +++ b/tests/test_tools/test_skills.py @@ -309,10 +309,26 @@ async def test_skill_write_prompt_denial_is_reported(tmp_path: Path) -> None: with _bound(tmp_path, root, policy=SkillWritePolicy(confirm=_deny, timeout=3.0)): out = await skill_edit.handler("acme", old_string="Before.", new_string="After.") assert out.startswith("error: skill write to ") - assert out.endswith("denied (declined or timed out after 3s)") + assert out.endswith("denied (user declined)") assert "After." not in (root / "acme" / SKILL_FILE).read_text(encoding="utf-8") +async def test_a_skill_write_timeout_names_the_deadline(tmp_path: Path) -> None: + """Nobody answering is not the same message as a human saying no.""" + from plyngent.agent import Denial, DenialKind + + root = _write_root(tmp_path) + _ = _write_skill(root, "acme", body="Before.") + + def silent(*_args: object) -> SkillWriteDecision: + return SkillWriteDecision(allow=False, denial=Denial(DenialKind.TIMEOUT, seconds=3.0)) + + with _bound(tmp_path, root, policy=SkillWritePolicy(confirm=silent, timeout=3.0)): + out = await skill_edit.handler("acme", old_string="Before.", new_string="After.") + assert out.endswith("denied (no answer within 3s — denied by default)") + assert "user" not in out + + async def test_skill_write_confirm_sees_the_action_and_the_skill(tmp_path: Path) -> None: root = _write_root(tmp_path) seen: list[tuple[Path, str, str, float]] = [] diff --git a/tests/test_tools/test_workspace.py b/tests/test_tools/test_workspace.py index 4690412..7914f6b 100644 --- a/tests/test_tools/test_workspace.py +++ b/tests/test_tools/test_workspace.py @@ -280,13 +280,30 @@ def test_command_denylist_policy_confirm_deny(workspace: object) -> None: set_policy_confirm_hook(lambda *_a: False) clear_policy_allowed_commands() try: - with pytest.raises(WorkspaceError, match="declined or timed out"): + with pytest.raises(WorkspaceError, match="user declined"): check_command_allowed(["sudo", "echo", "x"]) finally: set_policy_confirm_hook(None) clear_policy_allowed_commands() +def test_command_denylist_policy_timeout_names_the_deadline(workspace: object) -> None: + """Silence is reported as the clock, never as the human's decision.""" + del workspace + from plyngent.agent import Denial, DenialKind + from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook + + set_policy_confirm_hook(lambda *_a: Denial(DenialKind.TIMEOUT, seconds=3.0)) + clear_policy_allowed_commands() + try: + with pytest.raises(WorkspaceError, match="no answer within 3s") as refusal: + check_command_allowed(["sudo", "echo", "x"]) + finally: + set_policy_confirm_hook(None) + clear_policy_allowed_commands() + assert "declined" not in str(refusal.value) + + def test_command_denylist_policy_confirm_timeout_value(workspace: object) -> None: del workspace from plyngent.tools.workspace import (