From 86596633e8d9f4bd3dd1e593099c8a0ab97255ea Mon Sep 17 00:00:00 2001 From: worldmozara Date: Sat, 17 Oct 2026 17:26:33 +0800 Subject: [PATCH] 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 | 3 + src/plyngent/agent/loop.py | 28 +- src/plyngent/agent/tools.py | 178 +++++++++++- tests/test_agent/test_tools_batch_confirm.py | 273 +++++++++++++++++++ 4 files changed, 464 insertions(+), 18 deletions(-) create mode 100644 tests/test_agent/test_tools_batch_confirm.py diff --git a/src/plyngent/agent/__init__.py b/src/plyngent/agent/__init__.py index 4edcd24..c251724 100644 --- a/src/plyngent/agent/__init__.py +++ b/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 diff --git a/src/plyngent/agent/loop.py b/src/plyngent/agent/loop.py index 256d70c..0147cfd 100644 --- a/src/plyngent/agent/loop.py +++ b/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 diff --git a/src/plyngent/agent/tools.py b/src/plyngent/agent/tools.py index 94b1002..4cca08f 100644 --- a/src/plyngent/agent/tools.py +++ b/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) diff --git a/tests/test_agent/test_tools_batch_confirm.py b/tests/test_agent/test_tools_batch_confirm.py new file mode 100644 index 0000000..fde30d3 --- /dev/null +++ b/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 == []