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.
This commit is contained in:
2026-10-17 17:26:42 +08:00
parent 86596633e8
commit 0dd70ff31b
3 changed files with 164 additions and 10 deletions
+78 -9
View File
@@ -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,
+2 -1
View File
@@ -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,
@@ -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)