From 0dd70ff31b147f2d5266776f5d3d07f5500ef9ce Mon Sep 17 00:00:00 2001 From: worldmozara Date: Sat, 17 Oct 2026 17:26:42 +0800 Subject: [PATCH] core/cli: prompt for a round's tool calls once prompt_confirm_batch_async answers a ToolBatchConfirmHook with one box that numbers every call the round wants to make (continuation lines stay under their entry). The prompt is the plain ``Allow these N tool calls?`` confirm: a deny offers the same one comment for all of them, a non-interactive host denies, and the next round asks again. The ReplState registry installs the hook next to the per-call one, and format_tool_confirm_box now shares the box drawing with the batch header. --- src/plyngent/cli/limits.py | 87 ++++++++++++++++++--- src/plyngent/cli/state.py | 3 +- tests/test_cli/test_limits_batch_confirm.py | 84 ++++++++++++++++++++ 3 files changed, 164 insertions(+), 10 deletions(-) create mode 100644 tests/test_cli/test_limits_batch_confirm.py diff --git a/src/plyngent/cli/limits.py b/src/plyngent/cli/limits.py index c944394..b10bbcd 100644 --- a/src/plyngent/cli/limits.py +++ b/src/plyngent/cli/limits.py @@ -7,6 +7,7 @@ import sys from dataclasses import dataclass from typing import TYPE_CHECKING, Literal +from plyngent.agent import BatchConfirmDecision, ToolConfirmRequest from plyngent.cli.display import close_open_pretty_line from plyngent.cli.interrupt import off_loop_prompt, pause_task_cancel_for_prompt from plyngent.prompting import ( @@ -40,6 +41,11 @@ _CONTINUE_PROMPT = "Raise limit and continue? [Y/n/yyy] (yyy = stop asking this _CONTINUE_YES_TOKENS = frozenset({"", "y", "yes"}) _CONTINUE_AUTO_TOKEN = "yyy" +# Batch confirm answers: ``y`` allows the round's calls, ``m`` reviews them one +# at a time through the per-call prompt. +_BATCH_YES_TOKENS = frozenset({"y", "yes"}) +_BATCH_MANUAL_TOKENS = frozenset({"m", "manual", "each"}) + @dataclass class _AutoContinue: @@ -105,18 +111,13 @@ def _wrap_line(text: str, width: int) -> list[str]: return out -def format_tool_confirm_box(name: str, reason: str) -> str: - """Multi-line boxed confirm body (header + reason lines). - - Printed with :meth:`PromptBackend.echo` before a short ``confirm()`` prompt - so terminals keep newlines (unlike a single jammed readline line). - """ +def _box(header: str, lines: Sequence[str]) -> str: + """Rounded box: a header row, a rule, then the wrapped body lines.""" term = _terminal_width() inner = max(20, term - 4) # room for ``│ `` + `` │`` - header = f"confirm · tool {name!r}" body_lines: list[str] = [] - for raw in reason.replace("\r\n", "\n").replace("\r", "\n").split("\n"): - body_lines.extend(_wrap_line(raw, inner - _BOX_PAD)) + for raw in lines: + body_lines.extend(_wrap_line(raw.replace("\r\n", "\n").replace("\r", "\n"), inner - _BOX_PAD)) content = [header, "─" * min(inner, max(len(header), 12)), *body_lines] width = min(inner, max(len(line) for line in content) + _BOX_PAD) width = max(width, min(inner, len(header) + _BOX_PAD)) @@ -134,6 +135,15 @@ def format_tool_confirm_box(name: str, reason: str) -> str: return "\n".join(rows) +def format_tool_confirm_box(name: str, reason: str) -> str: + """Multi-line boxed confirm body for one tool call. + + Printed with :meth:`PromptBackend.echo` before a short ``confirm()`` prompt + so terminals keep newlines (unlike a single jammed readline line). + """ + return _box(f"confirm · tool {name!r}", reason.split("\n")) + + def _echo_tool_confirm(name: str, reason: str) -> None: backend = get_prompt_backend() backend.echo() @@ -482,6 +492,65 @@ async def prompt_confirm_tool_async(name: str, args: Mapping[str, object], reaso return comment or False +async def prompt_confirm_batch_async(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision: + """One confirm for one round's soft-confirm requests (``y`` / ``N`` / ``m``). + + The answer covers exactly those calls — the next round asks again; ``m`` + asks them one at a time instead, and a deny carries one comment shared by + every one of them. + """ + try: + from plyngent.prompting import run_prompt_async + + return await run_prompt_async(lambda: _prompt_confirm_batch_sync(requests)) + except NonInteractiveError: + return BatchConfirmDecision(allowed=False) + + +def _prompt_confirm_batch_sync(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision: + backend = get_prompt_backend() + if not backend.is_interactive(): + # Non-interactive: deny, like the default of the single prompt. + return BatchConfirmDecision(allowed=False) + backend.echo() + backend.secho(format_batch_confirm_box(requests), fg="yellow") + backend.secho("m = review them one call at a time", fg="bright_black") + backend.echo() + try: + token = ( + backend.read_line( + f"Allow these {len(requests)} tool calls? [y/N/m]:", + default="n", + ) + .strip() + .lower() + ) + except NonInteractiveError: + return BatchConfirmDecision(allowed=False) + if token in _BATCH_MANUAL_TOKENS: + return BatchConfirmDecision(allowed=False, each=True) + if token in _BATCH_YES_TOKENS: + return BatchConfirmDecision(allowed=True) + try: + comment = ask( + "Optional comment for the agent (why denied; empty to skip):", + default="", + ).strip() + except NonInteractiveError: + comment = "" + return BatchConfirmDecision(allowed=False, comment=comment or None) + + +def format_batch_confirm_box(requests: Sequence[ToolConfirmRequest]) -> str: + """Boxed body listing every call a round wants to make, one numbered block each.""" + lines: list[str] = [] + for index, request in enumerate(requests, start=1): + reason_lines = request.reason.split("\n") + lines.append(f"{index}. [{request.name}] {reason_lines[0]}") + lines.extend(f" {line}" for line in reason_lines[1:]) + return _box(f"confirm · {len(requests)} tool calls", lines) + + def _prompt_workspace_mismatch_sync( session_id: int, session_workspace: str, diff --git a/src/plyngent/cli/state.py b/src/plyngent/cli/state.py index a274bf4..ad6b19c 100644 --- a/src/plyngent/cli/state.py +++ b/src/plyngent/cli/state.py @@ -347,7 +347,7 @@ class ReplState: def _tool_registry(self) -> ToolRegistry | None: if not self.tools_enabled: return None - from plyngent.cli.limits import prompt_confirm_tool_async + from plyngent.cli.limits import prompt_confirm_batch_async, prompt_confirm_tool_async from plyngent.tools.catalog import register_builtin_tools from plyngent.tools.danger import classify_danger from plyngent.tools.plugins import load_plugin_tools @@ -375,6 +375,7 @@ class ReplState: tools, danger=classify_danger, on_confirm=prompt_confirm_tool_async, + on_confirm_batch=prompt_confirm_batch_async, yolo=yolo, auto_bind_state=True, instance_state=self.instance_state, diff --git a/tests/test_cli/test_limits_batch_confirm.py b/tests/test_cli/test_limits_batch_confirm.py new file mode 100644 index 0000000..b00a055 --- /dev/null +++ b/tests/test_cli/test_limits_batch_confirm.py @@ -0,0 +1,84 @@ +"""Aggregated soft-confirm prompt: one box for a round's tool calls.""" + +from __future__ import annotations + +from plyngent.agent import BatchConfirmDecision, ToolConfirmRequest +from plyngent.cli.limits import ( + _prompt_confirm_batch_sync, + format_batch_confirm_box, + format_tool_confirm_box, + prompt_confirm_batch_async, +) +from plyngent.prompting import NonInteractiveBackend, temporary_backend +from tests.test_prompting import ScriptedBackend + + +def _requests() -> list[ToolConfirmRequest]: + return [ + ToolConfirmRequest(name="delete_path", args={"path": "a.txt"}, reason="delete path 'a.txt'"), + ToolConfirmRequest(name="move_path", args={"src": "b", "dst": "c"}, reason="move 'b' → 'c'"), + ] + + +def test_batch_box_numbers_every_call() -> None: + box = format_batch_confirm_box(_requests()) + lines = box.splitlines() + assert "confirm · 2 tool calls" in lines[1] + assert any("[delete_path] delete path 'a.txt'" in line for line in lines) + assert any("[move_path] move 'b' → 'c'" in line for line in lines) + # Numbering keeps the listed order visible when the user answers. + assert any(line.startswith("│ 1. ") for line in lines) + assert any(line.startswith("│ 2. ") for line in lines) + + +def test_batch_box_indents_continuation_lines() -> None: + box = format_batch_confirm_box( + [ + ToolConfirmRequest(name="run_argv", args={"argv": ["python", "-c", "x"]}, reason="argv\n body"), + ToolConfirmRequest(name="delete_path", args={"path": "a"}, reason="delete path 'a'"), + ] + ) + assert any(line.startswith("│ ") and "body" in line for line in box.splitlines()) + + +def test_single_call_box_is_unchanged() -> None: + box = format_tool_confirm_box("delete_path", "delete path 'a.txt'") + assert "confirm · tool 'delete_path'" in box.splitlines()[1] + + +def test_batch_prompt_allows_every_call_on_yes() -> None: + with temporary_backend(ScriptedBackend(["y"])): + decision = _prompt_confirm_batch_sync(_requests()) + assert decision == BatchConfirmDecision(allowed=True) + + +def test_batch_prompt_hands_the_round_to_the_per_call_prompt_on_manual() -> None: + with temporary_backend(ScriptedBackend(["m"])): + decision = _prompt_confirm_batch_sync(_requests()) + assert decision.each is True + assert decision.allowed is False + + +def test_batch_prompt_denies_with_a_comment() -> None: + with temporary_backend(ScriptedBackend(["", "keep those files"])): + decision = _prompt_confirm_batch_sync(_requests()) + assert decision.allowed is False + assert decision.comment == "keep those files" + + +def test_batch_prompt_defaults_to_deny() -> None: + with temporary_backend(ScriptedBackend([])): + decision = _prompt_confirm_batch_sync(_requests()) + assert decision == BatchConfirmDecision(allowed=False) + + +def test_batch_prompt_denies_when_not_interactive() -> None: + with temporary_backend(NonInteractiveBackend()): + decision = _prompt_confirm_batch_sync(_requests()) + assert decision == BatchConfirmDecision(allowed=False) + + +async def test_batch_prompt_async_allows() -> None: + with temporary_backend(ScriptedBackend(["yes"])): + decision = await prompt_confirm_batch_async(_requests()) + assert decision == BatchConfirmDecision(allowed=True)