core/agent: confirm a round's mutating calls in one prompt

ToolRegistry.prepare_batch_confirm pre-scans a round of tool calls, drops the ones that cannot run (unknown tool, undecodable arguments, missing required argument), and turns two or more soft reasons into a single ToolBatchConfirmHook prompt. The BatchConfirmDecision it returns is applied per call through execute(..., decision=...): allow, or deny with one comment shared by every call of that round.

_execute_tool_calls asks before it gathers, so a batch of moves or deletes no longer stacks one confirm box per call. A single request keeps the old one-call prompt, a host without a batch hook keeps the per-call behaviour, and the answer never outlives the round it was asked in.

src/plyngent/agent/__init__.py

@@ -28,8 +28,11 @@ from .notices import has_notice as has_notice
 from .notices import resume_notice as resume_notice
 from .notices import skill_tools_notice as skill_tools_notice
 from .notices import truncated_turn_notice as truncated_turn_notice
+from .tools import BatchConfirmDecision as BatchConfirmDecision
 from .tools import DangerClassifier as DangerClassifier
+from .tools import ToolBatchConfirmHook as ToolBatchConfirmHook
 from .tools import ToolConfirmHook as ToolConfirmHook
+from .tools import ToolConfirmRequest as ToolConfirmRequest
 from .tools import ToolDefinition as ToolDefinition
 from .tools import ToolRegistry as ToolRegistry
 from .tools import ToolTag as ToolTag

src/plyngent/agent/loop.py

@@ -64,7 +64,7 @@ if TYPE_CHECKING:
 
     from .todo_nag import TodoNagStrategy
     from .todo_stack import TodoStack
-    from .tools import ToolRegistry
+    from .tools import BatchConfirmDecision, ToolRegistry
     from .types import AnyLLMClient
 
     type LimitContinueHook = Callable[[str], bool | Awaitable[bool]]
@@ -89,10 +89,11 @@ async def _run_one_tool(
     call: AnyAssistantToolCall,
     *,
     max_result_chars: int,
+    decision: BatchConfirmDecision | None = None,
 ) -> tuple[ToolChatMessage, ErrorEvent | None]:
     if isinstance(call, AssistantFunctionToolCall):
         try:
-            result_text = await tools.execute(call.function.name, call.function.arguments)
+            result_text = await tools.execute(call.function.name, call.function.arguments, decision=decision)
         except asyncio.CancelledError:
             raise
         except Exception as exc:  # noqa: BLE001
@@ -124,12 +125,31 @@ async def _execute_tool_calls(
     for call in tool_calls:
         yield ToolCallEvent(tool_call=call)
 
+    # One confirm prompt for the round's mutating calls, so a batch of moves or
+    # deletes asks once instead of once per call. Calls that need no confirm (or
+    # cannot run) keep ``None`` and go through their own path below.
+    requests: list[tuple[str, str]] = []
+    positions: list[int] = []
+    for index, call in enumerate(tool_calls):
+        if isinstance(call, AssistantFunctionToolCall):
+            requests.append((call.function.name, call.function.arguments))
+            positions.append(index)
+    decisions: list[BatchConfirmDecision | None] = [None] * len(tool_calls)
+    for index, decision in zip(positions, await tools.prepare_batch_confirm(requests), strict=True):
+        decisions[index] = decision
+
     if parallel and len(tool_calls) > 1:
         results = await asyncio.gather(
-            *[_run_one_tool(tools, call, max_result_chars=max_result_chars) for call in tool_calls]
+            *[
+                _run_one_tool(tools, call, max_result_chars=max_result_chars, decision=decision)
+                for call, decision in zip(tool_calls, decisions, strict=True)
+            ]
         )
     else:
-        results = [await _run_one_tool(tools, call, max_result_chars=max_result_chars) for call in tool_calls]
+        results = [
+            await _run_one_tool(tools, call, max_result_chars=max_result_chars, decision=decision)
+            for call, decision in zip(tool_calls, decisions, strict=True)
+        ]
 
     # Record every result before yielding any: a caller that goes away mid-batch
     # (cancel, display error) must not lose work that already ran, or the retry

src/plyngent/agent/tools.py

@@ -3,8 +3,9 @@ from __future__ import annotations
 import asyncio
 import inspect
 import types
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from contextlib import AsyncExitStack
+from dataclasses import dataclass
 from enum import Flag, auto
 from pathlib import Path
 from typing import Any, cast, get_args, get_origin, get_type_hints, overload
@@ -23,6 +24,39 @@ type ToolConfirmHook = Callable[
     ToolConfirmResult | Awaitable[ToolConfirmResult],
 ]
 
+
+@dataclass(frozen=True, slots=True)
+class ToolConfirmRequest:
+    """One call of a batch that needs a soft confirm before it may run."""
+
+    name: str
+    args: Mapping[str, object]
+    reason: str
+
+
+@dataclass(frozen=True, slots=True)
+class BatchConfirmDecision:
+    """Host answer for one round's soft-confirm requests.
+
+    It covers the calls of that round only; the next round asks again. With
+    ``each`` the host hands the calls back to the one-call prompt instead of
+    answering for the batch (``allowed``/``comment`` then carry no meaning).
+    """
+
+    allowed: bool
+    comment: str | None = None
+    each: bool = False
+
+
+# One prompt for a round's requests; a single request keeps ``ToolConfirmHook``.
+type ToolBatchConfirmHook = Callable[
+    [Sequence[ToolConfirmRequest]],
+    BatchConfirmDecision | Awaitable[BatchConfirmDecision],
+]
+
+MIN_BATCH_CONFIRM_REQUESTS = 2
+"""Below this many requests a round asks one call at a time (see ``prepare_batch_confirm``)."""
+
 _PRIMITIVE_SCHEMA: dict[type, JSONSchema] = {
     str: {"type": "string"},
     int: {"type": "integer"},
@@ -242,6 +276,24 @@ def _missing_required_error(definition: ToolDefinition, args: Mapping[str, objec
     return f"error: `{definition.name}` is missing required argument(s): {', '.join(missing)}{hint}"
 
 
+def _decode_tool_args(arguments_json: str) -> dict[str, object] | str:
+    """Decode one call's arguments; a returned string is the model-facing error."""
+    try:
+        raw_args: object = msgspec.json.decode(arguments_json.encode())
+    except (msgspec.DecodeError, UnicodeEncodeError) as exc:
+        return f"error: invalid tool arguments JSON: {exc}"
+    if not isinstance(raw_args, dict):
+        return "error: tool arguments must be a JSON object"
+    return {str(key): value for key, value in cast("dict[object, object]", raw_args).items()}
+
+
+def _denied_error(name: str, reason: str, comment: str | None) -> str:
+    """Model-facing text for a denied soft confirm (carries a user comment if one was given)."""
+    if comment is not None and comment.strip():
+        return f"error: tool {name!r} denied by user confirm ({reason}); user comment: {comment.strip()}"
+    return f"error: tool {name!r} denied by user confirm ({reason})"
+
+
 def _write_path_key(value: str) -> str:
     """Comparable key for one path argument of a writing tool.
 
@@ -257,6 +309,7 @@ class ToolRegistry:
     _tools: dict[str, ToolDefinition]
     _danger: DangerClassifier | None
     _on_confirm: ToolConfirmHook | None
+    _on_confirm_batch: ToolBatchConfirmHook | None
     _confirm_lock: asyncio.Lock
     # One lock per written file, kept for the registry's life (a session's edits
     # bound the map), so a batch's writes to one file apply one at a time.
@@ -272,6 +325,7 @@ class ToolRegistry:
         *,
         danger: DangerClassifier | None = None,
         on_confirm: ToolConfirmHook | None = None,
+        on_confirm_batch: ToolBatchConfirmHook | None = None,
         yolo: bool = False,
         auto_bind_state: bool = False,
         instance_state: object | None = None,
@@ -280,6 +334,7 @@ class ToolRegistry:
         self._tools = {}
         self._danger = danger
         self._on_confirm = on_confirm
+        self._on_confirm_batch = on_confirm_batch
         self._confirm_lock = asyncio.Lock()
         self._write_locks = {}
         self._yolo = yolo
@@ -339,6 +394,7 @@ class ToolRegistry:
             definitions,
             danger=self._danger,
             on_confirm=self._on_confirm,
+            on_confirm_batch=self._on_confirm_batch,
             yolo=self._yolo if yolo is None else yolo,
             auto_bind_state=self._auto_bind_state if auto_bind_state is None else auto_bind_state,
             instance_state=instance_state,
@@ -463,9 +519,92 @@ class ToolRegistry:
         if decision is True:
             await self._record_grant(name, tags=tags)
             return None
-        if isinstance(decision, str) and decision.strip():
-            return f"error: tool {name!r} denied by user confirm ({reason}); user comment: {decision.strip()}"
-        return f"error: tool {name!r} denied by user confirm ({reason})"
+        comment = decision if isinstance(decision, str) else None
+        return _denied_error(name, reason, comment)
+
+    async def _apply_batch_decision(
+        self,
+        name: str,
+        reason: str,
+        decision: BatchConfirmDecision,
+        *,
+        tags: ToolTag,
+    ) -> str | None:
+        if decision.allowed:
+            await self._record_grant(name, tags=tags)
+            return None
+        return _denied_error(name, reason, decision.comment)
+
+    def _prepare_confirm(self, name: str, arguments_json: str) -> ToolConfirmRequest | None:
+        """Soft-confirm request for one call, or ``None`` when it needs none.
+
+        Mirrors the front half of :meth:`execute`: a call whose arguments fail to
+        decode or miss a required argument is answered with an error there and
+        must not be put in front of the user now.
+        """
+        definition = self._tools.get(name)
+        if self._danger is None or definition is None:
+            return None
+        args = _decode_tool_args(arguments_json)
+        if isinstance(args, str) or _missing_required_error(definition, args) is not None:
+            return None
+        reason = self._danger(name, args)
+        return None if reason is None else ToolConfirmRequest(name=name, args=args, reason=reason)
+
+    def _batch_requests(
+        self,
+        calls: Sequence[tuple[str, str]],
+    ) -> tuple[list[ToolConfirmRequest], list[int]]:
+        """Requests for a round, plus the index each one came from."""
+        requests: list[ToolConfirmRequest] = []
+        positions: list[int] = []
+        for index, (name, arguments_json) in enumerate(calls):
+            request = self._prepare_confirm(name, arguments_json)
+            if request is not None:
+                requests.append(request)
+                positions.append(index)
+        return requests, positions
+
+    async def _record_batch_allow(
+        self,
+        requests: Sequence[ToolConfirmRequest],
+        decision: BatchConfirmDecision,
+    ) -> None:
+        """Record the trust a batch allow carries (the trustable grants)."""
+        if decision.allowed:
+            for request in requests:
+                definition = self._tools.get(request.name)
+                if definition is not None:
+                    await self._record_grant(request.name, tags=definition.tags)
+
+    async def prepare_batch_confirm(
+        self,
+        calls: Sequence[tuple[str, str]],
+    ) -> list[BatchConfirmDecision | None]:
+        """Ask once for every soft confirm a round's calls would raise.
+
+        Returns one entry per call: a decision to apply instead of prompting, or
+        ``None`` to fall through to the per-call path. Fewer than
+        :data:`MIN_BATCH_CONFIRM_REQUESTS` requests keep the plain one-call
+        prompt, a host without a batch hook keeps the per-call prompts, and a
+        :attr:`BatchConfirmDecision.each` answer hands this round back to them.
+        """
+        decisions: list[BatchConfirmDecision | None] = [None] * len(calls)
+        if self._danger is None or self._on_confirm_batch is None:
+            return decisions
+        requests, positions = self._batch_requests(calls)
+        if len(requests) < MIN_BATCH_CONFIRM_REQUESTS:
+            return decisions
+        async with self._confirm_lock:
+            decision = self._on_confirm_batch(requests)
+            if inspect.isawaitable(decision):
+                decision = await decision
+            if decision.each:
+                return decisions
+            await self._record_batch_allow(requests, decision)
+        for index in positions:
+            decisions[index] = decision
+        return decisions
 
     async def _maybe_confirm(
         self,
@@ -473,12 +612,14 @@ class ToolRegistry:
         args: dict[str, object],
         *,
         tags: ToolTag,
+        decision: BatchConfirmDecision | None = None,
     ) -> str | None:
         """Soft-confirm gate driven by danger reason + tags + grants.
 
         Pipeline (soft gate only; hard denylists stay in tool handlers)::
 
             no soft reason → run
+            batch decision for this call → apply it
             (tags & TRUSTABLE) and grant exists → allow
             else → on_confirm (decorator YOLO tag does not override
                    dangerous invocations — internal danger classification
@@ -491,22 +632,31 @@ class ToolRegistry:
             return None
 
         async with self._confirm_lock:
+            if decision is not None:
+                return await self._apply_batch_decision(name, reason, decision, tags=tags)
             if self._grant_allows(name, tags=tags):
                 return None
             return await self._prompt_soft_confirm(name, args, reason, tags=tags)
 
-    async def execute(self, name: str, arguments_json: str) -> str:
-        """Run a tool by name; returns a string result (errors become error text)."""
+    async def execute(
+        self,
+        name: str,
+        arguments_json: str,
+        *,
+        decision: BatchConfirmDecision | None = None,
+    ) -> str:
+        """Run a tool by name; returns a string result (errors become error text).
+
+        *decision* is the host's answer for this call from
+        :meth:`prepare_batch_confirm`; ``None`` means the call was not part of an
+        aggregated prompt and confirms on its own.
+        """
         definition = self._tools.get(name)
         if definition is None:
             return f"error: unknown tool {name!r}"
-        try:
-            raw_args: object = msgspec.json.decode(arguments_json.encode())
-        except (msgspec.DecodeError, UnicodeEncodeError) as exc:
-            return f"error: invalid tool arguments JSON: {exc}"
-        if not isinstance(raw_args, dict):
-            return "error: tool arguments must be a JSON object"
-        args = {str(key): value for key, value in cast("dict[object, object]", raw_args).items()}
+        args = _decode_tool_args(arguments_json)
+        if isinstance(args, str):
+            return args
         # Argument shape/coverage problems are answered before any confirm prompt:
         # the human should never be asked about a call that cannot run.
         missing_error = _missing_required_error(definition, args)
@@ -522,7 +672,7 @@ class ToolRegistry:
             async with AsyncExitStack() as held:
                 for lock in self._write_locks_for(definition, args):
                     await held.enter_async_context(lock)
-                denied = await self._maybe_confirm(name, args, tags=definition.tags)
+                denied = await self._maybe_confirm(name, args, tags=definition.tags, decision=decision)
                 if denied is not None:
                     return denied
                 return await self._invoke(definition, args)

tests/test_agent/test_tools_batch_confirm.py

@@ -0,0 +1,273 @@
+"""Aggregated soft confirm: one prompt for a round's mutating calls."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from plyngent.agent import (
+    BatchConfirmDecision,
+    ToolRegistry,
+    ToolTag,
+    tool,
+)
+from plyngent.agent.loop import _execute_tool_calls
+from plyngent.lmproto.openai_compatible.model import AssistantFunctionTool, AssistantFunctionToolCall
+from plyngent.tools.context import SessionState, bind_tool_context
+from plyngent.tools.grants import has_grant
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping, Sequence
+
+    from plyngent.agent import ToolConfirmRequest
+    from plyngent.lmproto.openai_compatible.model import AnyChatMessage
+
+
+def _danger(name: str, _args: Mapping[str, object]) -> str | None:
+    return f"mutate {name}"
+
+
+async def test_loop_round_prompts_once_before_running_its_calls() -> None:
+    """The loop aggregates a round's confirms and applies one answer to each call."""
+    batch_prompts: list[int] = []
+    single_prompts: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def stamp(path: str) -> str:
+        return f"stamped {path}"
+
+    def on_batch(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        batch_prompts.append(len(requests))
+        return BatchConfirmDecision(allowed=True)
+
+    async def on_single(name: str, _args: Mapping[str, object], _reason: str) -> bool:
+        single_prompts.append(name)
+        return True
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm=on_single, on_confirm_batch=on_batch)
+    tool_calls = [
+        AssistantFunctionToolCall(id=f"c{index}", function=AssistantFunctionTool(name="stamp", arguments=arguments))
+        for index, arguments in enumerate(['{"path": "a"}', '{"path": "b"}'])
+    ]
+    messages: list[AnyChatMessage] = []
+
+    events = [event async for event in _execute_tool_calls(reg, tool_calls, messages)]
+
+    assert batch_prompts == [2]
+    assert single_prompts == []
+    assert [getattr(message, "content", None) for message in messages] == ["stamped a", "stamped b"]
+    assert len(events) == len(tool_calls) * 2  # call event + result event per call
+
+
+async def test_round_asks_once_for_many_mutating_calls() -> None:
+    """Three calls in one round share one prompt instead of three."""
+    ran: list[str] = []
+    batch_prompts: list[list[ToolConfirmRequest]] = []
+    single_prompts: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL | ToolTag.YOLO, register=False)
+    async def stamp(path: str) -> str:
+        ran.append(path)
+        return f"stamped {path}"
+
+    def on_batch(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        batch_prompts.append(list(requests))
+        return BatchConfirmDecision(allowed=True)
+
+    async def on_single(name: str, _args: Mapping[str, object], _reason: str) -> bool:
+        single_prompts.append(name)
+        return True
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm=on_single, on_confirm_batch=on_batch)
+    calls = [("stamp", '{"path": "a"}'), ("stamp", '{"path": "b"}'), ("stamp", '{"path": "c"}')]
+    decisions = await reg.prepare_batch_confirm(calls)
+
+    assert all(decision is not None and decision.allowed for decision in decisions)
+    assert [request.args["path"] for request in batch_prompts[0]] == ["a", "b", "c"]
+    assert single_prompts == []
+
+    results = [
+        await reg.execute(name, arguments, decision=decisions[index]) for index, (name, arguments) in enumerate(calls)
+    ]
+    assert results == ["stamped a", "stamped b", "stamped c"]
+    assert ran == ["a", "b", "c"]
+    assert single_prompts == []
+
+
+async def test_single_request_keeps_the_one_call_prompt() -> None:
+    """One mutating call stays the plain per-call confirm."""
+    batch_prompts: list[int] = []
+    single_prompts: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def stamp(path: str) -> str:
+        return f"stamped {path}"
+
+    def on_batch(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        batch_prompts.append(len(requests))
+        return BatchConfirmDecision(allowed=True)
+
+    async def on_single(name: str, _args: Mapping[str, object], _reason: str) -> bool:
+        single_prompts.append(name)
+        return True
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm=on_single, on_confirm_batch=on_batch)
+    assert await reg.prepare_batch_confirm([("stamp", '{"path": "a"}')]) == [None]
+    assert batch_prompts == []
+    assert await reg.execute("stamp", '{"path": "a"}') == "stamped a"
+    assert single_prompts == ["stamp"]
+
+
+async def test_calls_that_cannot_run_are_left_out_of_the_prompt() -> None:
+    """Missing arguments, bad JSON and unknown tools never reach the user."""
+    batch_prompts: list[list[str]] = []
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def stamp(path: str, *, force: bool = False) -> str:
+        return f"stamped {path} ({force})"
+
+    def on_batch(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        batch_prompts.append([str(request.args.get("path")) for request in requests])
+        return BatchConfirmDecision(allowed=True)
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm_batch=on_batch)
+    decisions = await reg.prepare_batch_confirm(
+        [
+            ("stamp", "{}"),  # missing ``path``
+            ("stamp", "not json"),
+            ("nope", "{}"),
+            ("stamp", '{"path": "a"}'),
+            ("stamp", '{"path": "b", "force": true}'),
+        ]
+    )
+
+    assert decisions[:3] == [None, None, None]
+    assert all(decision is not None for decision in decisions[3:])
+    assert batch_prompts == [["a", "b"]]
+    missing = await reg.execute("stamp", "{}")
+    assert missing.startswith("error: `stamp` is missing required argument")
+
+
+async def test_batch_deny_reaches_every_call_with_one_comment() -> None:
+    """A deny stops the whole batch and hands the same comment to each call."""
+    ran: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def stamp(path: str) -> str:
+        ran.append(path)
+        return f"stamped {path}"
+
+    def on_batch(_requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        return BatchConfirmDecision(allowed=False, comment="not these two")
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm_batch=on_batch)
+    calls = [("stamp", '{"path": "a"}'), ("stamp", '{"path": "b"}')]
+    decisions = await reg.prepare_batch_confirm(calls)
+
+    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("user comment: not these two" in result for result in results)
+    assert ran == []
+
+
+async def test_each_round_asks_again() -> None:
+    """An answer covers one round: the next round's calls are asked again."""
+    batch_prompts: list[int] = []
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def stamp(path: str) -> str:
+        return f"stamped {path}"
+
+    def on_batch(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        batch_prompts.append(len(requests))
+        return BatchConfirmDecision(allowed=True)
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm_batch=on_batch)
+    for first, second in (("a", "b"), ("c", "d")):
+        calls = [("stamp", f'{{"path": "{first}"}}'), ("stamp", f'{{"path": "{second}"}}')]
+        decisions = await reg.prepare_batch_confirm(calls)
+        for index, (name, arguments) in enumerate(calls):
+            await reg.execute(name, arguments, decision=decisions[index])
+    assert batch_prompts == [2, 2]
+
+
+async def test_manual_hands_the_round_back_to_the_per_call_prompt() -> None:
+    """``each`` answers nothing for the batch: every call is asked on its own."""
+    batch_prompts: list[int] = []
+    single_prompts: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def stamp(path: str) -> str:
+        return f"stamped {path}"
+
+    def on_batch(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        batch_prompts.append(len(requests))
+        return BatchConfirmDecision(allowed=False, each=True)
+
+    async def on_single(name: str, _args: Mapping[str, object], _reason: str) -> bool:
+        single_prompts.append(name)
+        return True
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm=on_single, on_confirm_batch=on_batch)
+    calls = [("stamp", '{"path": "a"}'), ("stamp", '{"path": "b"}')]
+    decisions = await reg.prepare_batch_confirm(calls)
+
+    assert decisions == [None, None]
+    assert [await reg.execute(name, arguments) for name, arguments in calls] == ["stamped a", "stamped b"]
+    assert batch_prompts == [2]
+    assert single_prompts == ["stamp", "stamp"]
+
+
+async def test_without_a_batch_hook_every_call_prompts_on_its_own() -> None:
+    """A host that installs no batch hook keeps the per-call prompts."""
+    single_prompts: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL, register=False)
+    async def stamp(path: str) -> str:
+        return f"stamped {path}"
+
+    async def on_single(name: str, _args: Mapping[str, object], _reason: str) -> bool:
+        single_prompts.append(name)
+        return True
+
+    reg = ToolRegistry([stamp], danger=_danger, on_confirm=on_single)
+    calls = [("stamp", '{"path": "a"}'), ("stamp", '{"path": "b"}')]
+    assert await reg.prepare_batch_confirm(calls) == [None, None]
+    assert [await reg.execute(name, arguments) for name, arguments in calls] == ["stamped a", "stamped b"]
+    assert single_prompts == ["stamp", "stamp"]
+
+
+async def test_batch_allow_records_the_trustable_grant() -> None:
+    """A batch allow still carries the trust a single allow would record."""
+    single_prompts: list[str] = []
+
+    @tool(tags=ToolTag.LOCAL | ToolTag.TRUSTABLE, register=False)
+    async def trust_me(path: str) -> str:
+        return f"done {path}"
+
+    def on_batch(_requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        return BatchConfirmDecision(allowed=True)
+
+    async def on_single(name: str, _args: Mapping[str, object], _reason: str) -> bool:
+        single_prompts.append(name)
+        return True
+
+    session = SessionState(session_id=1)
+    reg = ToolRegistry(
+        [trust_me],
+        danger=_danger,
+        on_confirm=on_single,
+        on_confirm_batch=on_batch,
+        auto_bind_state=True,
+        session_state=session,
+    )
+    with bind_tool_context(session=session):
+        calls = [("trust_me", '{"path": "a"}'), ("trust_me", '{"path": "b"}')]
+        decisions = await reg.prepare_batch_confirm(calls)
+        for path, decision in zip(("a", "b"), decisions, strict=True):
+            assert await reg.execute("trust_me", f'{{"path": "{path}"}}', decision=decision) == f"done {path}"
+        assert has_grant(session, "trust_me")
+        # The TRUSTABLE grant, not the batch answer, keeps later calls unchecked.
+        assert await reg.execute("trust_me", '{"path": "c"}') == "done c"
+    assert single_prompts == []