tools: answer a call's shape before the confirm prompt

src/plyngent/agent/__init__.py

@@ -41,6 +41,7 @@ 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 ToolValidator as ToolValidator
 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

src/plyngent/agent/tools.py

@@ -84,6 +84,8 @@ class Denial:
 
 
 type ToolHandler = Callable[..., Any | Awaitable[Any]]
+type ToolValidator = Callable[[Mapping[str, object]], str | None]
+"""A tool's pre-prompt shape check: the error to answer with, or ``None``."""
 type DangerClassifier = Callable[[str, Mapping[str, object]], str | DangerReason | None]
 # Returns True to allow, or a Denial (or a bare str comment) to deny.
 type ToolConfirmResult = bool | str | Denial
@@ -254,6 +256,11 @@ class ToolDefinition:
     ``writes`` names the arguments whose values name files the handler writes.
     Calls of one batch that name the same file apply one at a time, in call
     order, so the later one sees what the earlier one wrote.
+
+    ``validate`` answers an argument-shape problem the tool can see without
+    running anything (a value that is not the type it needs, a tree that cannot
+    be built). The registry asks it before any confirm prompt, so the human is
+    never shown a call that could only fail.
     """
 
     name: str
@@ -262,6 +269,7 @@ class ToolDefinition:
     handler: ToolHandler
     tags: ToolTag
     writes: tuple[str, ...]
+    validate: ToolValidator | None
 
     def __init__(
         self,
@@ -272,6 +280,7 @@ class ToolDefinition:
         *,
         tags: ToolTag = ToolTag.LOCAL,
         writes: tuple[str, ...] = (),
+        validate: ToolValidator | None = None,
     ) -> None:
         if not (tags & (ToolTag.LOCAL | ToolTag.PUBLIC)):
             msg = f"tool {name!r} tags must include LOCAL and/or PUBLIC"
@@ -282,6 +291,7 @@ class ToolDefinition:
         self.handler = handler
         self.tags = tags
         self.writes = writes
+        self.validate = validate
 
     def to_tool_item(self) -> ToolFunctionItem:
         return ToolFunctionItem(
@@ -342,6 +352,7 @@ def _build_definition(
     description: str | None,
     tags: ToolTag,
     writes: tuple[str, ...] = (),
+    validate: ToolValidator | None = None,
 ) -> ToolDefinition:
     tool_name = name or func.__name__
     tool_description = description if description is not None else (inspect.getdoc(func) or "")
@@ -352,6 +363,7 @@ def _build_definition(
         handler=func,
         tags=tags,
         writes=writes,
+        validate=validate,
     )
 
 
@@ -376,6 +388,7 @@ def tool[**PS, R](
     description: str | None = None,
     tags: ToolTag = ToolTag.LOCAL,
     writes: tuple[str, ...] = (),
+    validate: ToolValidator | None = None,
     register: bool = True,
 ) -> Callable[[Callable[PS, R]], ToolDefinition]: ...
 
@@ -388,6 +401,7 @@ def tool[**PS, R](
     description: str | None = None,
     tags: ToolTag = ToolTag.LOCAL,
     writes: tuple[str, ...] = (),
+    validate: ToolValidator | None = None,
     register: bool = True,
 ) -> ToolDefinition | Callable[[Callable[PS, R]], ToolDefinition]:
     """Define (and by default catalog-register) a function as an agent tool.
@@ -395,6 +409,8 @@ def tool[**PS, R](
     Schema is inferred from type hints; description defaults to the docstring.
     Default ``tags`` is :attr:`ToolTag.LOCAL`. ``writes`` names the arguments
     that carry the files the handler writes (:attr:`ToolDefinition.writes`).
+    ``validate`` answers an argument-shape problem **before** the call is put in
+    front of the human (:attr:`ToolDefinition.validate`).
     When ``register`` is true, the definition is added to the process
     :class:`~plyngent.tools.catalog.ToolCatalog` with the current registration
     source (builtin unless a plugin context is set). Catalog registration does
@@ -409,6 +425,7 @@ def tool[**PS, R](
             description=description,
             tags=tags,
             writes=writes,
+            validate=validate,
         )
         if register:
             _register_definition(definition)
@@ -443,6 +460,18 @@ def _missing_required_error(definition: ToolDefinition, args: Mapping[str, objec
     return f"error: `{definition.name}` is missing required argument(s): {', '.join(missing)}{hint}"
 
 
+def _shape_error(definition: ToolDefinition, args: Mapping[str, object]) -> str | None:
+    """Model-facing error when the tool's own pre-prompt check refuses the call.
+
+    Shape mistakes inside a *present* argument used to surface in the handler,
+    i.e. after the confirm prompt, so the human could be asked about a call that
+    could only fail. A tool that declares ``validate`` is answered first.
+    """
+    if definition.validate is None:
+        return None
+    return definition.validate(args)
+
+
 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:
@@ -730,6 +759,8 @@ class ToolRegistry:
         args = _decode_tool_args(arguments_json)
         if isinstance(args, str) or _missing_required_error(definition, args) is not None:
             return None
+        if _shape_error(definition, args) is not None:
+            return None
         reason = _as_danger_reason(self._danger(name, args))
         if reason is None or (self._yolo and (definition.tags & ToolTag.YOLO) and reason.yolo_skip):
             return None
@@ -866,6 +897,9 @@ class ToolRegistry:
         missing_error = _missing_required_error(definition, args)
         if missing_error is not None:
             return ToolOutput(text=missing_error)
+        check_error = _shape_error(definition, args)
+        if check_error is not None:
+            return ToolOutput(text=check_error)
 
         async def _run() -> ToolOutput:
             from plyngent.tools.context import collect_approvals

src/plyngent/tools/process/command_exec.py

@@ -45,6 +45,26 @@ def format_command_result(
     return "\n".join(parts)
 
 
+def timeout_shape_error(value: object) -> str | None:
+    """Model-facing error when ``timeout_seconds`` is not a positive number.
+
+    Numeric strings are accepted (models send them), anything else is answered
+    before the call can start: a value that only fails inside ``wait_for`` has
+    already spawned the process, so nothing would stop it.
+    """
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, int | float | str):
+        return f"timeout_seconds must be a number, not {type(value).__name__}"
+    try:
+        seconds = float(value)
+    except ValueError:
+        return f"timeout_seconds must be a number, not {value!r}"
+    if seconds <= 0:
+        return f"timeout_seconds must be greater than 0 (got {value!r})"
+    return None
+
+
 def resolve_workdir(cwd: str) -> tuple["Path", str]:
     """Return (absolute workdir, display path relative to workspace when possible)."""
     workdir = resolve_path(cwd, required=AccessMode.EXEC)

src/plyngent/tools/process/run_argv.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import WorkspaceError, argv_shape_error
 
@@ -7,10 +9,25 @@ from .command_exec import (
     DEFAULT_TIMEOUT_SECONDS,
     execute_argv,
     format_command_result,
+    timeout_shape_error,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
+
+def run_argv_shape(args: Mapping[str, object]) -> str | None:
+    """Answer a ``run_argv`` call whose arguments cannot run, before any prompt."""
+    shape_error = argv_shape_error(args.get("argv"))
+    if shape_error is not None:
+        return f"error: `run_argv` {shape_error}"
+    timeout_error = timeout_shape_error(args.get("timeout_seconds"))
+    if timeout_error is not None:
+        return f"error: `run_argv` {timeout_error}"
+    return None
+
+
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, validate=run_argv_shape)
 async def run_argv(
     argv: list[str],
     *,
@@ -20,9 +37,9 @@ async def run_argv(
     env: dict[str, str] | None = None,
 ) -> str:
     """Run ``argv`` (argv, no shell) under the workspace; capture stdout/stderr."""
-    shape_error = argv_shape_error(argv)
+    shape_error = run_argv_shape({"argv": argv, "timeout_seconds": timeout_seconds})
     if shape_error is not None:
-        return f"error: `run_argv` {shape_error}"
+        return shape_error
     try:
         result = await execute_argv(
             argv,

src/plyngent/tools/process/run_argv_batch.py

@@ -3,17 +3,21 @@ from __future__ import annotations
 import json
 import shlex
 from dataclasses import dataclass, field
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
 
 from plyngent.agent import ToolTag, tool
 from plyngent.tools.workspace import WorkspaceError
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
 from .command_exec import (
     DEFAULT_MAX_BATCH_STEPS,
     DEFAULT_TIMEOUT_SECONDS,
     MAX_BATCH_DEPTH,
     CommandStepResult,
     execute_argv,
+    timeout_shape_error,
 )
 
 # Execution budget: stop the batch once accumulated output crosses this factor
@@ -118,6 +122,10 @@ def _parse_node(raw: object, *, where: str, depth: int, budget: _Budget, fed: bo
     if fed and stdin is not None:
         msg = f"{where}: stdin comes from the pipe; drop stdin or split the command"
         raise WorkspaceError(msg)
+    timeout_error = timeout_shape_error(data.get("timeout_seconds"))
+    if timeout_error is not None:
+        msg = f"{where}.{timeout_error}"
+        raise WorkspaceError(msg)
     cwd = data.get("cwd")
     stop = data.get("stop_on_error", None)
     node: dict[str, Any] = {
@@ -141,7 +149,7 @@ def _parse_node(raw: object, *, where: str, depth: int, budget: _Budget, fed: bo
     return node
 
 
-def _normalize_steps(steps: list[dict[str, object]] | str) -> list[object]:
+def _normalize_steps(steps: object) -> list[object]:
     if isinstance(steps, str):
         try:
             loaded: object = json.loads(steps)
@@ -152,9 +160,33 @@ def _normalize_steps(steps: list[dict[str, object]] | str) -> list[object]:
             msg = "steps must be a JSON array of step objects"
             raise WorkspaceError(msg)
         return cast("list[object]", loaded)
+    if not isinstance(steps, list):
+        msg = "steps must be a JSON array of step objects"
+        raise WorkspaceError(msg)
     return cast("list[object]", steps)
 
 
+def _read_batch(steps: object) -> tuple[list[dict[str, Any]], int]:
+    """Parse a batch tree and count its nodes; raises when it cannot run."""
+    raw_steps = _normalize_steps(steps)
+    if not raw_steps:
+        msg = "steps must not be empty"
+        raise WorkspaceError(msg)
+    budget = _Budget()
+    return _parse_nodes(raw_steps, where="steps", depth=0, budget=budget, fed=False), budget.nodes
+
+
+def run_argv_batch_shape(args: Mapping[str, object]) -> str | None:
+    """Answer a batch that cannot run, before it reaches the human."""
+    try:
+        _ = _read_batch(args.get("steps"))
+    except WorkspaceError as exc:
+        return f"error: {exc}"
+    except (TypeError, ValueError) as exc:
+        return f"error: invalid batch arguments: {exc}"
+    return None
+
+
 @dataclass(frozen=True)
 class _Step:
     """One executed command, and whether its stdout fed another one."""
@@ -288,7 +320,10 @@ class _Runner:
         return self.stop_on_error if node["stop_on_error"] is None else bool(node["stop_on_error"])
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
+@tool(
+    tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO,
+    validate=run_argv_batch_shape,
+)
 async def run_argv_batch(
     steps: list[dict[str, object]] | str,
     *,
@@ -323,28 +358,24 @@ async def run_argv_batch(
     long result carries a ``document_id`` to read on.
     """
     try:
-        raw_steps = _normalize_steps(steps)
-        if not raw_steps:
-            return "error: steps must not be empty"
-
-        budget = _Budget()
-        nodes = _parse_nodes(raw_steps, where="steps", depth=0, budget=budget, fed=False)
-        runner = _Runner(cwd=cwd, env=env, stop_on_error=stop_on_error)
-        await runner.run_list(nodes, depth=0)
-
-        body = "\n".join(_format_step(i, step) for i, step in enumerate(runner.steps))
-        last_exit = runner.steps[-1].result.exit_code if runner.steps else ""
-        return "\n".join(
-            [
-                f"steps={budget.nodes} ran={len(runner.steps)} "
-                f"stop_on_error={'true' if stop_on_error else 'false'} "
-                f"stopped_early={'true' if runner.stopped else 'false'}",
-                body,
-                "--- summary ---",
-                f"last_exit={last_exit if last_exit is not None else ''}",
-            ]
-        )
+        nodes, count = _read_batch(steps)
     except WorkspaceError as exc:
         return f"error: {exc}"
     except (TypeError, ValueError) as exc:
         return f"error: invalid batch arguments: {exc}"
+
+    runner = _Runner(cwd=cwd, env=env, stop_on_error=stop_on_error)
+    await runner.run_list(nodes, depth=0)
+
+    body = "\n".join(_format_step(i, step) for i, step in enumerate(runner.steps))
+    last_exit = runner.steps[-1].result.exit_code if runner.steps else ""
+    return "\n".join(
+        [
+            f"steps={count} ran={len(runner.steps)} "
+            f"stop_on_error={'true' if stop_on_error else 'false'} "
+            f"stopped_early={'true' if runner.stopped else 'false'}",
+            body,
+            "--- summary ---",
+            f"last_exit={last_exit if last_exit is not None else ''}",
+        ]
+    )

tests/test_agent/test_tools_batch_confirm.py

@@ -105,6 +105,42 @@ async def test_round_asks_once_for_many_mutating_calls() -> None:
     assert single_prompts == []
 
 
+async def test_a_call_that_cannot_run_is_left_out_of_the_round_prompt() -> None:
+    """A tool's own shape check answers first, so the round prompt omits the call."""
+    batch_prompts: list[list[ToolConfirmRequest]] = []
+
+    def positive(args: Mapping[str, object]) -> str | None:
+        count = args.get("count")
+        if isinstance(count, int) and count <= 0:
+            return "error: count must be positive"
+        return None
+
+    @tool(tags=ToolTag.LOCAL, register=False, validate=positive)
+    async def scale(count: int = 1) -> str:
+        return f"scaled {count}"
+
+    def on_batch(requests: Sequence[ToolConfirmRequest]) -> BatchConfirmDecision:
+        batch_prompts.append(list(requests))
+        return BatchConfirmDecision(allowed=True)
+
+    reg = ToolRegistry([scale], danger=_danger, on_confirm_batch=on_batch)
+    calls = [("scale", '{"count": 2}'), ("scale", '{"count": 0}'), ("scale", '{"count": 3}')]
+    decisions = await reg.prepare_batch_confirm(calls)
+
+    assert len(batch_prompts) == 1
+    assert [request.args["count"] for request in batch_prompts[0]] == [2, 3]
+    assert decisions[1] is None
+
+    results = [
+        await reg.execute(name, arguments, decision=decisions[index]) for index, (name, arguments) in enumerate(calls)
+    ]
+    assert results == [
+        _approved("user (round)", "scaled 2"),
+        "error: count must be positive",
+        _approved("user (round)", "scaled 3"),
+    ]
+
+
 async def test_yolo_calls_are_left_out_of_the_prompt() -> None:
     """Under YOLO mode a YOLO-tagged call runs unprompted, batch prompt included."""
     batch_prompts: list[list[str]] = []

tests/test_tools/test_process.py

@@ -1,11 +1,15 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import sys
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 import pytest
 
+from plyngent.agent import ToolDefinition, ToolRegistry
+from plyngent.tools.danger import classify_danger
 from plyngent.tools.process import (
     ask_into_pty,
     close_pty,
@@ -20,6 +24,9 @@ from plyngent.tools.process.pty_session import PtyManager
 from plyngent.tools.workspace import set_command_denylist
 from tests.test_tools.helpers import call_async, call_sync
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
 
 def _session_id(opened: str) -> int:
     for line in opened.splitlines():
@@ -299,6 +306,49 @@ async def test_run_argv_batch_names_the_node_grammar(workspace: object) -> None:
     assert "then, else, pipe" in out
 
 
+def _asking_registry(*definitions: ToolDefinition) -> tuple[ToolRegistry, list[str]]:
+    """A registry whose confirm hook records the calls it was asked about."""
+    asked: list[str] = []
+
+    def on_confirm(name: str, args: Mapping[str, object], reason: str) -> bool:
+        del args, reason
+        asked.append(name)
+        return True
+
+    return ToolRegistry(list(definitions), danger=classify_danger, on_confirm=on_confirm), asked
+
+
+async def test_a_bad_timeout_is_answered_before_the_prompt(workspace: object) -> None:
+    """A call that cannot run is answered, never put in front of the human."""
+    del workspace
+    registry, asked = _asking_registry(run_argv)
+    out = await registry.execute(
+        "run_argv",
+        json.dumps({"argv": ["bash", "-c", "echo hi"], "timeout_seconds": "soon"}),
+    )
+    assert asked == []
+    assert "timeout_seconds" in out
+    assert "error:" in out
+
+
+async def test_a_broken_batch_is_answered_before_the_prompt(workspace: object) -> None:
+    del workspace
+    registry, asked = _asking_registry(run_argv_batch)
+    out = await registry.execute(
+        "run_argv_batch",
+        json.dumps(
+            {
+                "steps": [
+                    {"argv": ["bash", "-c", "echo hi"]},
+                    {"argv": ["echo", "x"], "pipe": {"argv": ["echo", "y"], "stdin": "own"}},
+                ]
+            }
+        ),
+    )
+    assert asked == []
+    assert "stdin comes from the pipe" in out
+
+
 async def test_run_argv_denied(workspace: object) -> None:
     del workspace
     out = await call_async(run_argv, ["rm", "-rf", "."])