core/agent: lead a confirmed tool result with how it was approved

A tool result said nothing about the answer that let the call run: the human
approved a delete, the model saw only "deleted x", and a resumed transcript
could not tell a YOLO auto-approval from a prompt.

The soft gate now returns an `Approval` (`user` / `grant` / `yolo` / `config`
with the scope that answer reaches) and `execute_output` leads the result with
`[approval] user (once)`. A handler that prompts on its own records its answer
through `tools/context.py` (`collect_approvals` around one call), and the one
line printed is the answer closest to a human. A call that needed no
confirmation, a denial and an `error:` result carry no line.
This commit is contained in:
2026-10-29 18:28:13 +08:00
parent 1c2587db82
commit f9ec812275
3 changed files with 117 additions and 17 deletions
+3
View File
@@ -28,6 +28,7 @@ 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 Approval as Approval
from .tools import BatchConfirmDecision as BatchConfirmDecision
from .tools import DangerClassifier as DangerClassifier
from .tools import DangerReason as DangerReason
@@ -38,6 +39,8 @@ from .tools import ToolDefinition as ToolDefinition
from .tools import ToolOutput as ToolOutput
from .tools import ToolRegistry as ToolRegistry
from .tools import ToolTag as ToolTag
from .tools import format_approval as format_approval
from .tools import most_specific_approval as most_specific_approval
from .tools import schema_from_callable as schema_from_callable
from .tools import tool as tool
from .usage import TokenUsage as TokenUsage
+85 -17
View File
@@ -102,6 +102,64 @@ class BatchConfirmDecision:
each: bool = False
@dataclass(frozen=True, slots=True)
class Approval:
"""How one call got past a confirmation, as the model reads it back.
``source`` names who answered — ``user`` (a human at a prompt), ``grant``
(trust recorded earlier, so nobody was asked *this* time), ``yolo`` (YOLO
mode dropped the soft confirm) or ``config`` (a configured allow, e.g.
``[agent].allow_paths`` or ``[skills].allow_write``) — and ``scope`` how far
that answer reaches: ``once``, ``round`` (a batch allow), ``session``,
``process`` or ``auto`` (mode-driven, not a grant).
"""
source: str
scope: str
@property
def label(self) -> str:
"""``user (once)`` — the words the result's first line carries."""
return f"{self.source} ({self.scope})"
# Who answered first when one call carries several records: a human prompt is the
# closest answer, an earlier grant sits second, a config allow third, and YOLO
# mode says the least about why the call was allowed.
_APPROVAL_PRIORITY = {"user": 0, "grant": 1, "config": 2, "yolo": 3}
def format_approval(approval: Approval) -> str:
"""The model-facing line naming how a call was approved."""
return f"[approval] {approval.label}"
def most_specific_approval(approvals: Sequence[Approval]) -> Approval | None:
"""The one record a result carries (ties keep the first recorded)."""
if not approvals:
return None
return min(approvals, key=lambda a: _APPROVAL_PRIORITY.get(a.source, len(_APPROVAL_PRIORITY)))
@dataclass(frozen=True, slots=True)
class _Gate:
"""What the soft gate decided for one call: a denial, else how it was allowed."""
denied: str | None = None
approval: Approval | None = None
def _with_approval(output: ToolOutput, approval: Approval | None) -> ToolOutput:
"""Lead *output* with its approval line (nothing to add: unchanged).
An ``error:`` result carries no line: the call did not finish, so the answer
that let it start is not what the model has to read first.
"""
if approval is None or output.text.startswith("error: "):
return output
return msgspec.structs.replace(output, text=f"{format_approval(approval)}\n{output.text}")
# One prompt for a round's requests; a single request keeps ``ToolConfirmHook``.
type ToolBatchConfirmHook = Callable[
[Sequence[ToolConfirmRequest]],
@@ -571,17 +629,17 @@ class ToolRegistry:
reason: str,
*,
tags: ToolTag,
) -> str | None:
) -> _Gate:
if self._on_confirm is None:
return f"error: tool {name!r} denied by policy ({reason}; no confirm hook)"
return _Gate(denied=f"error: tool {name!r} denied by policy ({reason}; no confirm hook)")
decision = self._on_confirm(name, args, reason)
if inspect.isawaitable(decision):
decision = await decision
if decision is True:
await self._record_grant(name, tags=tags)
return None
return _Gate(approval=Approval("user", "once"))
comment = decision if isinstance(decision, str) else None
return _denied_error(name, reason, comment)
return _Gate(denied=_denied_error(name, reason, comment))
async def _apply_batch_decision(
self,
@@ -590,11 +648,11 @@ class ToolRegistry:
decision: BatchConfirmDecision,
*,
tags: ToolTag,
) -> str | None:
) -> _Gate:
if decision.allowed:
await self._record_grant(name, tags=tags)
return None
return _denied_error(name, reason, decision.comment)
return _Gate(approval=Approval("user", "round"))
return _Gate(denied=_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.
@@ -678,7 +736,7 @@ class ToolRegistry:
*,
tags: ToolTag,
decision: BatchConfirmDecision | None = None,
) -> str | None:
) -> _Gate:
"""Soft-confirm gate driven by danger reason + tags + grants + YOLO.
Pipeline (soft gate only; hard denylists stay in tool handlers)::
@@ -690,20 +748,23 @@ class ToolRegistry:
else → on_confirm (a tool without the YOLO tag, or a reason YOLO
may not drop — a privilege escalation — still prompts
under YOLO mode)
Every allow names how it happened, so the result can say it; a call that
needed no confirmation carries no approval.
"""
if self._danger is None:
return None
return _Gate()
reason = _as_danger_reason(self._danger(name, args))
if reason is None:
return None
return _Gate()
if self._yolo and (tags & ToolTag.YOLO) and reason.yolo_skip:
return None
return _Gate(approval=Approval("yolo", "auto"))
async with self._confirm_lock:
if decision is not None:
return await self._apply_batch_decision(name, reason.text, decision, tags=tags)
if self._grant_allows(name, tags=tags):
return None
return _Gate(approval=Approval("grant", "session"))
return await self._prompt_soft_confirm(name, args, reason.text, tags=tags)
async def execute(
@@ -746,6 +807,8 @@ class ToolRegistry:
return ToolOutput(text=missing_error)
async def _run() -> ToolOutput:
from plyngent.tools.context import collect_approvals
missing = self._check_state_tags(definition)
if missing is not None:
return ToolOutput(text=missing)
@@ -754,11 +817,16 @@ 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, decision=decision)
if denied is not None:
return ToolOutput(text=denied)
result = await self._invoke(definition, args)
return _without_refused_images(result if isinstance(result, ToolOutput) else ToolOutput(text=result))
with collect_approvals() as recorded:
gate = await self._maybe_confirm(name, args, tags=definition.tags, decision=decision)
if gate.denied is not None:
return ToolOutput(text=gate.denied)
result = await self._invoke(definition, args)
output = _without_refused_images(result if isinstance(result, ToolOutput) else ToolOutput(text=result))
# The gate's answer and whatever the handler asked on its own;
# the result carries the one closest to a human (see Approval).
approvals = [*([gate.approval] if gate.approval is not None else []), *recorded]
return _with_approval(output, most_specific_approval(approvals))
# Bind host-provided state for tools that read contextvars (todo, workspace).
# Tag enforcement remains gated by ``auto_bind_state``; binding is always useful
+29
View File
@@ -14,6 +14,7 @@ if TYPE_CHECKING:
from pathlib import Path
from plyngent.agent.todo_stack import TodoStack
from plyngent.agent.tools import Approval
from plyngent.skills import SkillStore
from plyngent.tools.workspace import AccessMode, WorkspacePolicy
@@ -150,6 +151,34 @@ def bind_session(state: SessionState | None) -> Generator[SessionState | None]:
_read_only: ContextVar[bool] = ContextVar("plyngent_read_only_context", default=False)
# Approvals a handler records for the call it is answering. The registry binds a
# fresh list around one call and drains it into the result's first line.
_approvals: ContextVar[list[Approval] | None] = ContextVar("plyngent_tool_approvals", default=None)
@contextmanager
def collect_approvals() -> Generator[list[Approval]]:
"""Collect what the running tool call records about how it was approved."""
sink: list[Approval] = []
token = _approvals.set(sink)
try:
yield sink
finally:
_approvals.reset(token)
def record_approval(approval: Approval) -> None:
"""Record how the running call was approved (no-op outside a tool call).
A handler that asks the human for something policy-level — a denylisted
command, a skill write, a private host — says so here, so the model reads the
answer in the result it is about to see. The registry turns the records into
the single ``[approval]`` line that leads the tool result.
"""
sink = _approvals.get()
if sink is not None:
sink.append(approval)
@contextmanager
def read_only_context() -> Generator[None]: