mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
NCBM
35595d8b04
Every place the CLI blocks on the human now announces itself: the single and batched tool confirms, the command-denylist and `fetch_local` policies, a skill write, a directory-access request, the `ask_user_*` questions and the limit-continue prompt. `_setup_hooks` installs the notifier from `[notify]`, and only for an interactive run — a one-shot run never waits on a human. An answer that came back through the notification is read as if it had been typed: `allow` / `deny` map onto the `y` / `n` the prompts already accept, and the confirmation prompts switch from click's readline to the poll-based read only when the notification actually offered answers — otherwise the terminal experience is untouched. "Go to terminal" never reaches the prompt at all. The stdin read behind the timed prompts now watches in 100ms slices instead of one full-length `select`, which is what lets a notification answer end it; an answer still loses to a human typing at the terminal. A `None` timeout waits for whichever arrives first, and a non-pollable stdin still gives up at once. `prompt_wait` announces only the outermost prompt of a nested read: a confirm that asks a follow-up question while it is already on screen would otherwise buzz the human a second time.
299 lines
11 KiB
Python
299 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
|
|
from plyngent.cli.display import close_open_pretty_line
|
|
from plyngent.cli.limits import (
|
|
auto_continue_enabled,
|
|
format_tool_confirm_box,
|
|
install_cli_limit_hooks,
|
|
prompt_confirm_tool,
|
|
prompt_continue_limit,
|
|
reset_auto_continue,
|
|
reset_auto_continue_turn,
|
|
set_auto_continue_default,
|
|
)
|
|
from plyngent.prompting import (
|
|
NonInteractiveBackend,
|
|
get_prompt_backend,
|
|
get_prompt_output_hook,
|
|
get_prompt_wait_hook,
|
|
temporary_backend,
|
|
)
|
|
from plyngent.tools.process.pty_session import PtyManager
|
|
from tests.test_prompting import ScriptedBackend
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Generator, Iterator
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_auto_continue() -> Iterator[None]:
|
|
reset_auto_continue()
|
|
yield
|
|
reset_auto_continue()
|
|
|
|
|
|
def test_prompt_continue_limit_yes() -> None:
|
|
backend = ScriptedBackend(["y"])
|
|
with temporary_backend(backend):
|
|
assert prompt_continue_limit("hit a wall") is True
|
|
|
|
|
|
def test_prompt_continue_limit_default_yes() -> None:
|
|
backend = ScriptedBackend([])
|
|
with temporary_backend(backend):
|
|
assert prompt_continue_limit("hit a wall") is True
|
|
|
|
|
|
def test_prompt_continue_limit_no() -> None:
|
|
backend = ScriptedBackend(["n"])
|
|
with temporary_backend(backend):
|
|
assert prompt_continue_limit("hit a wall") is False
|
|
|
|
|
|
def test_prompt_continue_limit_yyy_sets_turn_skip(clean_auto_continue: None) -> None:
|
|
del clean_auto_continue
|
|
backend = ScriptedBackend(["yyy"])
|
|
with temporary_backend(backend):
|
|
assert prompt_continue_limit("hit a wall") is True
|
|
assert auto_continue_enabled() is True
|
|
# No further prompt: the script is empty and must not be consumed.
|
|
assert prompt_continue_limit("hit a wall again") is True
|
|
|
|
|
|
def test_reset_auto_continue_turn_clears_yyy(clean_auto_continue: None) -> None:
|
|
del clean_auto_continue
|
|
with temporary_backend(ScriptedBackend(["yyy"])):
|
|
assert prompt_continue_limit("limit") is True
|
|
reset_auto_continue_turn()
|
|
assert auto_continue_enabled() is False
|
|
with temporary_backend(ScriptedBackend(["n"])):
|
|
assert prompt_continue_limit("limit") is False
|
|
|
|
|
|
def test_auto_continue_default_short_circuits_prompt(clean_auto_continue: None) -> None:
|
|
del clean_auto_continue
|
|
set_auto_continue_default(enabled=True)
|
|
assert auto_continue_enabled() is True
|
|
with temporary_backend(NonInteractiveBackend()):
|
|
assert prompt_continue_limit("limit") is True
|
|
|
|
|
|
def test_prompt_continue_limit_noninteractive_denies() -> None:
|
|
with temporary_backend(NonInteractiveBackend()):
|
|
assert prompt_continue_limit("limit") is False
|
|
|
|
|
|
def test_setup_hooks_installs_pty_hook_for_auto_continue(clean_auto_continue: None) -> None:
|
|
del clean_auto_continue
|
|
from plyngent.cli.app import _setup_hooks
|
|
from plyngent.config.models import NotifyConfig
|
|
|
|
_setup_hooks(interactive=False, auto_continue=True, notify=NotifyConfig())
|
|
assert PtyManager._limit_continue is not None
|
|
_setup_hooks(interactive=False, auto_continue=False, notify=NotifyConfig())
|
|
assert PtyManager._limit_continue is None
|
|
|
|
|
|
def test_pty_offer_raise_uses_auto_continue(clean_auto_continue: None) -> None:
|
|
del clean_auto_continue
|
|
PtyManager.set_limit_continue_hook(prompt_continue_limit)
|
|
set_auto_continue_default(enabled=True)
|
|
try:
|
|
with temporary_backend(NonInteractiveBackend()):
|
|
assert PtyManager._offer_raise("PTY session limit reached") is True
|
|
finally:
|
|
PtyManager.set_limit_continue_hook(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
|
|
|
|
|
|
def test_format_tool_confirm_box_multiline() -> None:
|
|
reason = "run_argv: python3 -c (review code before allow)\nargv:\n python3 -c print(1)\n-c code:\n print(1)"
|
|
box = format_tool_confirm_box("run_argv", reason)
|
|
assert "┌" in box and "└" in box
|
|
assert "confirm · tool 'run_argv'" in box
|
|
assert "python3 -c" in box
|
|
assert "print(1)" in box
|
|
# Distinct lines, not one jammed row
|
|
assert box.count("\n") >= 4
|
|
|
|
|
|
def test_install_cli_limit_hooks() -> None:
|
|
install_cli_limit_hooks()
|
|
assert callable(getattr(PtyManager, "_limit_continue", None))
|
|
# A prompt raised mid-call must end the open pretty tool line first.
|
|
assert get_prompt_output_hook() is close_open_pretty_line
|
|
PtyManager.set_limit_continue_hook(None)
|
|
# Backend remains usable after install.
|
|
assert get_prompt_backend().is_interactive() or True
|
|
|
|
|
|
def test_policy_confirm_noninteractive_denies() -> None:
|
|
from plyngent.cli.limits import prompt_policy_command_confirm
|
|
from plyngent.prompting import NonInteractiveBackend
|
|
|
|
with temporary_backend(NonInteractiveBackend()):
|
|
assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) is False
|
|
|
|
|
|
def test_directory_access_prompt_noninteractive_denies() -> None:
|
|
from pathlib import Path
|
|
|
|
from plyngent.cli.limits import prompt_directory_access_confirm
|
|
from plyngent.prompting import NonInteractiveBackend
|
|
from plyngent.tools.workspace import AccessMode
|
|
|
|
with temporary_backend(NonInteractiveBackend()):
|
|
assert prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 1.0) is None
|
|
|
|
|
|
def test_directory_access_prompt_yes_and_level(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from pathlib import Path
|
|
|
|
from plyngent.cli import limits
|
|
from plyngent.tools.access import AccessDecision
|
|
from plyngent.tools.workspace import AccessMode
|
|
|
|
target = Path("/tmp")
|
|
with temporary_backend(ScriptedBackend([])):
|
|
monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "y\n")
|
|
assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == AccessDecision(
|
|
AccessMode.READ
|
|
)
|
|
monkeypatch.setattr(limits, "_read_yes_no_line_with_timeout", lambda _timeout, **_kwargs: "exec\n")
|
|
assert limits.prompt_directory_access_confirm(target, AccessMode.READ, "why", 30.0) == AccessDecision(
|
|
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
|
|
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
|
|
|
|
|
|
class _AnswerSource:
|
|
"""A host surface offering answers to a waiting prompt."""
|
|
|
|
supports_actions = True
|
|
|
|
def __init__(self, *answers: str) -> None:
|
|
self._answers = list(answers)
|
|
|
|
def take(self) -> str | None:
|
|
return self._answers.pop(0) if self._answers else None
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _answered(*answers: str) -> Generator[_AnswerSource]:
|
|
"""Install a wait hook whose prompt can be answered by the notification."""
|
|
from plyngent.prompting import set_prompt_wait_hook
|
|
|
|
source = _AnswerSource(*answers)
|
|
previous = get_prompt_wait_hook()
|
|
set_prompt_wait_hook(lambda _wait: contextlib.nullcontext(source))
|
|
try:
|
|
yield source
|
|
finally:
|
|
set_prompt_wait_hook(previous)
|
|
|
|
|
|
def test_directory_access_takes_a_notification_answer() -> None:
|
|
"""``allow`` from the notification approves the requested level."""
|
|
from pathlib import Path
|
|
|
|
from plyngent.cli import limits
|
|
from plyngent.tools.access import AccessDecision
|
|
from plyngent.tools.workspace import AccessMode
|
|
|
|
with temporary_backend(ScriptedBackend([])), _answered("allow"):
|
|
assert limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0) == AccessDecision(
|
|
AccessMode.READ
|
|
)
|
|
|
|
|
|
def test_directory_access_denied_by_the_notification() -> None:
|
|
from pathlib import Path
|
|
|
|
from plyngent.cli import limits
|
|
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
|
|
|
|
|
|
def test_tool_confirm_takes_a_notification_answer() -> None:
|
|
with temporary_backend(ScriptedBackend([])), _answered("allow"):
|
|
assert prompt_confirm_tool("run_argv", {}, "runs a command") is True
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_batch_confirm_takes_a_notification_answer() -> None:
|
|
from plyngent.agent import ToolConfirmRequest
|
|
from plyngent.cli.limits import _prompt_confirm_batch_sync
|
|
|
|
requests = [
|
|
ToolConfirmRequest(name="run_argv", args={}, reason="one"),
|
|
ToolConfirmRequest(name="run_argv", args={}, reason="two"),
|
|
]
|
|
with temporary_backend(ScriptedBackend([])), _answered("allow"):
|
|
assert _prompt_confirm_batch_sync(requests).allowed is True
|
|
|
|
|
|
def test_batch_confirm_denied_by_the_notification() -> None:
|
|
from plyngent.agent import ToolConfirmRequest
|
|
from plyngent.cli.limits import _prompt_confirm_batch_sync
|
|
|
|
requests = [
|
|
ToolConfirmRequest(name="run_argv", args={}, reason="one"),
|
|
ToolConfirmRequest(name="run_argv", args={}, reason="two"),
|
|
]
|
|
with temporary_backend(ScriptedBackend([""])), _answered("deny"):
|
|
assert _prompt_confirm_batch_sync(requests).allowed is False
|
|
|
|
|
|
def test_every_confirm_announces_itself() -> None:
|
|
"""Each blocking confirm reports its kind, so a host can pick a surface."""
|
|
from pathlib import Path
|
|
|
|
from plyngent.agent import ToolConfirmRequest
|
|
from plyngent.cli import limits
|
|
from plyngent.tools.workspace import AccessMode
|
|
|
|
kinds: list[str] = []
|
|
|
|
@contextlib.contextmanager
|
|
def hook(wait: object) -> Generator[None]:
|
|
kinds.append(wait.kind) # type: ignore[attr-defined]
|
|
yield None
|
|
|
|
from plyngent.prompting import set_prompt_wait_hook
|
|
|
|
previous = get_prompt_wait_hook()
|
|
set_prompt_wait_hook(hook)
|
|
try:
|
|
with temporary_backend(ScriptedBackend(["y", "y", "y"], confirms=[True])):
|
|
_ = prompt_confirm_tool("run_argv", {}, "reason")
|
|
_ = limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0)
|
|
_ = limits._prompt_confirm_batch_sync(
|
|
[
|
|
ToolConfirmRequest(name="run_argv", args={}, reason="one"),
|
|
ToolConfirmRequest(name="run_argv", args={}, reason="two"),
|
|
]
|
|
)
|
|
_ = prompt_continue_limit("limit")
|
|
finally:
|
|
set_prompt_wait_hook(previous)
|
|
assert kinds == ["confirm", "access", "batch", "limit"]
|