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.
This commit is contained in:
2026-10-29 23:43:49 +08:00
parent 8cbb98602e
commit 4bab780603
9 changed files with 165 additions and 15 deletions
+71
View File
@@ -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)
+2 -1
View File
@@ -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 == []
+3 -1
View File
@@ -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(
+18 -7
View File
@@ -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")
+3 -3
View File
@@ -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:
+18 -1
View File
@@ -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 == {}
+15
View File
@@ -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
+17 -1
View File
@@ -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]] = []
+18 -1
View File
@@ -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 (