release: plyngent 0.16.0

AGENTS.md

@@ -85,7 +85,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `
 - **`messages_bridge` / `messages_dispatch`**: chat ↔ Anthropic Messages; system fold; tool_use/tool_result; stream text + tool argument fragments; `thinking` from the request's `reasoning_effort` (level → token budget, floor 1024, `max_tokens` grown when the budget would leave no room) plus the client's configured `thinking_budget_tokens`; a thinking block the model refuses (from Claude 4.7 on, `thinking.type.enabled` is a 400 asking for adaptive thinking + `output_config.effort`; a model whose thinking is always on refuses `thinking.type.disabled`) gets one retry with the shape it takes — adaptive thinking with the level as the effort, or no block at all — and the refusal is remembered per model on the client, so only its first request pays the round trip (only a refusal that arrives before the stream yields anything is retried; a later failure has already been seen); a server tool (`server_tool_use` + its `web_search_tool_result`) is the surface's own business — it streams `input_json_delta` fragments like a call but nothing local asked for it, so only a `tool_use` block becomes a tool call and nothing is executed; `thinking` and the two search blocks are kept as `AssistantChatMessage.provider_blocks` (the thinking text also lands in `reasoning_content`, so it streams as reasoning) and go back with the round, which is also why a round that kept only those blocks is not an empty completion; a chat-completions request drops the field (`compat.coerce_chat_messages`).
 - **Images** (see `doc/images.md`): a user turn's `content` may be a part array (`ChatTextPart` + `ChatImagePart`), and a tool result carries its pictures in `ToolChatMessage.images`; each surface lifts them into what it can express — chat-completions content blocks (tool-result images move into the user turn after the tool run, since a `tool` message holds text only), Anthropic `image` blocks (inside `tool_result` for a tool result), Responses `input_image` parts (in a `function_call_output` item). Pictures belong to user turns: an image in a system or assistant message is a provider 400. `agent/images.py` sniffs the format from the bytes and caps an inline picture at 8 MiB, because every round re-sends it; `[providers.*.models.*].image_in` is the opt-in, and every image-carrying path checks it — an `@path` mention, `view_image`, `fetch_local`'s image body, and (as a safety net in the registry) an MCP server's or plugin's `ToolOutput`.
 - **Provider-side tools**: `provider_tools` on the OpenAI **and** DeepSeek presets (list of dicts; `[{type="web_search"}]` when omitted; `[]` disables) threaded via `ChatAgent`/`run_chat_loop` into Responses and Messages; never executed by `ToolRegistry`. The Messages surface declares its own server tool instead — a bare `web_search` becomes `{type="web_search_20250305", name="web_search"}` (the search then runs inside the response, and its `encrypted_content` results go back verbatim with the round — `AssistantChatMessage.provider_blocks`) — and a DeepSeek provider on a surface that ignores hosted tools (`config.routing.local_web_search`) gets the local `web_search` tool from `tools/net/search.py` instead.
-- **`@tool` / `ToolRegistry`**: decorator infers JSON Schema from type hints; execute tools by name. A handler returns its text, or a `ToolOutput(text, images)` when it also produced a picture: `execute` keeps the text-only contract and `execute_output` returns the pair, which `run_chat_loop` stores as the tool message's text plus its `images` (`view_image`, an image body from `fetch_local`, and an MCP `image` content block all take that path). Before running, `execute` answers a call that omits a required argument with the accepted names (`_missing_required_error`) — the handler's bare `TypeError` never reaches the model, and no confirm prompt fires for a call that cannot run. Shape mistakes inside a *present* argument are the tool's own job (see the `chat` bullet). `@tool(writes=("path",))` names a handler's file arguments: one batch applies calls that name the same file one at a time, in call order — the lock is taken before the confirm prompt, so a pending prompt cannot be overtaken — while different files keep running in parallel.
+- **`@tool` / `ToolRegistry`**: decorator infers JSON Schema from type hints; execute tools by name. A handler returns its text, or a `ToolOutput(text, images)` when it also produced a picture: `execute` keeps the text-only contract and `execute_output` returns the pair, which `run_chat_loop` stores as the tool message's text plus its `images` (`view_image`, an image body from `fetch_local`, and an MCP `image` content block all take that path). Before running, `execute` answers a call that omits a required argument with the accepted names (`_missing_required_error`) — the handler's bare `TypeError` never reaches the model, and no confirm prompt fires for a call that cannot run. Shape mistakes inside a *present* argument are the tool's own job (see the `chat` bullet), unless the tool declares its own pre-prompt check (`@tool(validate=…)`, answered in that same place). `run_argv` / `run_argv_batch` / `open_pty` / `write_pty_keys` / `edit_replace` / `edit_lineno` / `fetch_local` do: a `timeout_seconds` that is not a positive number (so a value that would only fail inside `wait_for` never spawns a process), an `argv` sent as a string, an unknown `<key>` marker, an empty `old_string` or `max_replaces < 1`, a backwards line range, a verb outside GET/POST/PUT/DELETE and a URL that cannot be parsed. `prepare_batch_confirm` leaves such a call out of the round's prompt, so nobody is asked about a call that cannot run. `@tool(writes=("path",))` names a handler's file arguments: one batch applies calls that name the same file one at a time, in call order — the lock is taken before the confirm prompt, so a pending prompt cannot be overtaken — while different files keep running in parallel.
 - **`run_chat_loop`**: multi-round tool loop; default **streaming** text deltas + stream tool-call merge; parallel tools; request-prep tool-result cap (each result stored whole, trimmed on the request copy with a `document_id=...` marker); soft context compact on request (**API-calibrated** after first usage when available); cooperative cancel points; optional `on_limit`; resolved `reasoning_effort` on every request (see Config). A round the provider cut off at the output limit is kept, not failed: the answer and its thinking blocks stay in `messages`, the round's tool calls are dropped (a half-written call must not run — the model re-issues it), and up to `max_truncation_continues` (2) further rounds carry on from a truncation notice; 0 spends the turn as unfinished instead, for `/retry`.
 - **`ChatAgent`**: optional `MemoryStore` (user message persisted immediately; **completed tool batches checkpointed** mid-turn; unfinished assistant suffix rolled back on failure); `stream`; system prompt; `provider_tools`; `retry()` continues incomplete turns (user-only, after committed tools — does not re-run those tools — or after a kept truncated answer) and `pending_retry_text` reads a truncation notice as unfinished work.
 - **`/compact`**: soft-compact tool dumps → model summary (no tools) via the same kind dispatch → **new** session seeded with summary message. Soft-compact keeps the last **12** tool results full-size by default (`DEFAULT_RECENT_TOOL_RESULTS`); older tool payloads shrink first. The default summarizer system prompt states that tool calls are unavailable (a `compact_system_prompt` override replaces that default wholesale).
@@ -100,7 +100,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 - **`workspace`**: path resolve under primary root **or** temporary allowlist roots **or** a directory-access grant (mode-aware: `read` < `write` < `exec`, default required `write`; a file grant matches only that file); path substring denylist (wins over grants); command denylist via `match_command_denylist` (checks every program of the wrapper/`run` chain, resolved by `tools/command_scan.py`; an entry is a program, plus optional argument words that make it dangerous — `systemctl reboot poweroff …` matches `systemctl reboot` and `systemctl start reboot.target`, not `systemctl status`; basenames are lower-cased, so `RM` / `rm.exe` count; defaults cover privilege `sudo`/`su`/`doas`/`pkexec`/`run0`/`systemd-run`, destructive `rm`/`mkfs`/`dd`/`chown`/`mount`/…, and power state `shutdown`/`reboot`/`poweroff`/`halt`/`init`/`telinit`/`kexec`/`pm-*`/`s2*`; CLI: timed human allow override, default deny on timeout; **not** skipped by YOLO); escape error names the `request_directory_access` mode the call needs (`call request_directory_access with mode='write'`).
 - **`access`** (`tools/access.py`): `request_directory_access` (LOCAL|INSTANCE|SESSION) asks the host hook for out-of-workspace access; grants are one `path → mode` pair in one of three stores — `WorkspacePolicy.config_allow` (static `[agent].allow_paths`), `SessionState.access_grants` (human-approved, persisted on the session row), `WorkspacePolicy.yolo_allow` (process only: an answer a host keeps without persisting — YOLO/`--yes` never fills it, an out-of-workspace grant is asked either way). `AccessDecision(mode, persist)` from the hook; denylisted paths can never be granted; cap `MAX_ACCESS_GRANTS`. Not a sandbox: `exec` grants (and any command) reach the whole FS.
 - **`file`**: `read_file` (`with_lineno`, `max_chars` default 32k; results start with a 1-based `L{begin}-{end}` range; an image file is refused with a pointer to `view_image`), `view_image` (a picture reaches the model as an image part; format sniffed from the bytes, `detail` = `auto` default / `low` / `high` / `original`), `write_file` (refuses to replace an existing file unless `overwrite=true`; that replacement is the one soft-confirmed), `listdir`, `tree` (model picks `format=markdown` default / `flat` paths / `decorated`; VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `regex_files` (Python `re` per line, not a glob; literal file/dir `path`, no glob expansion; skips VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range; only lines read via `with_lineno`; read state resets on any file write/edit/copy/move/delete), `copy_path` / `move_path` / `delete_path` (moves the path into the system trash — freedesktop trash / Finder trash / Recycle Bin — so a delete can be undone; `permanent=true`, `[trash] enabled = false`, or a path under the OS temp dir delete it for good, and a trashing that fails reports the reason and leaves the path alone; [doc/trash.md](doc/trash.md)), `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind), `get_truncated` (resume a truncated `read_file` slice or `@path` mention via its `truncate_token=...`), `read_document` / `search_document` (read a capped tool result by its `document_id`: a line window with optional line numbers, or a per-line regex), `document_info` / `list_documents` (describe or list the session's documents).
-- **`process`**: `run_argv` (argv, no shell, timeout, optional stdin/env); `run_argv_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (key markers) / `ask_into_pty` (human→PTY only; `secret` no-echo; answer never in tool result) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep).
+- **`process`**: `run_argv` (argv, no shell, timeout, optional stdin/env); `run_argv_batch` (a node is `argv` plus `then` / `else` / `pipe` continuations — one node or a list each — with `cwd` / `env` / `stdin` / `mix_stderr` / `stop_on_error` / `timeout_seconds`; a continuation inherits its parent's `cwd` / `env`, a `pipe` chain runs to its last command and only that exit code decides the branch, max 20 nodes and 4 levels, and a pipe step takes no `then` / `else`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (key markers) / `ask_into_pty` (human→PTY only; `secret` no-echo; answer never in tool result) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep).
 - PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`<c-c>`, `<esc>`, `<c-up>`, `<c-s-f5>`, `<up*3>`, `<m-x>`, `\xHH` — xterm modifier sequences where they exist; an unknown `<…>` is an error, never typed); secrets via `ask_into_pty` (not `write_pty` data); session limit/idle TTL/output budget; close terminate→kill.
 - CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget. The prompt accepts `y`/empty (once), `n`, or `yyy` = stop asking for the rest of the turn (`cli.limits` module flag, reset by `run_turn_with_retries`, `/btw`, session switches, and chat exit). `--auto-continue` / `[agent] auto_continue_limits` set the process-wide default (covers one-shot/PTY); the flag does not imply YOLO and vice versa.
 - Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. Soft confirm also for interpreters — bare shells/REPLs, scripts, and `-c` one-liners alike (`-c` is code only for an interpreter, so `grep -c` / `od -c` stay unflagged; argv + code preview) —, detached runners (`nohup`), and wrapper options that carry a command line (`env -S "…"`, `su -c "…"`, whose text is printed under its own `env -S:` label), each resolved past env wrappers and `<launcher> run` — every wrapper's options are spelled as a `getopt` spec (`command_scan._WRAPPERS`: flags as well as value options, so no option swallows the program), and an option the spec does not know stops the unwrap and is reviewed (`options not understood … the wrapped program is a guess`) while the wrapper itself stays the command the denylist judges; deny may include a user comment for the model. `ssh` / `mosh` are unwrapped like a wrapper (`command_scan`): the program reviewed is the one the client runs on the far end (`ssh prod rm -rf /` is refused as `rm`, `ssh prod bash -c …` prints the remote code, an escalation behind the hop keeps `yolo_skip=False`), and `-o BatchMode=yes` always asks because nothing on either end can prompt. A client with **no** command only opens a remote shell: nothing typed or piped into it (`write_pty`, `run_argv(stdin=…)`) is checked, so opening one is announced and confirmed instead — `remote shell on 'prod' (ssh with no command; its input is not checked)` — while the client itself stays outside the denylist (there is no program to name). The `delete_path` confirm names what will happen — `move path 'x' to the system trash`, or `delete path 'x' permanently` when the call asks for it, the host cleared `[trash]`, or the path is scratch — and the pretty line prints `(to trash)` for a delete that landed there. Session YOLO: `/yolo on|off|once` and `--yes` (skip the soft confirms of `YOLO`-tagged tools — interpreter runs included — while untagged ones still prompt and a command that escalates privileges keeps its confirm through `DangerReason(yolo_skip=False)`; hard denylists unchanged and a directory-access request is asked either way; `once` expires after the next user turn). A round of tool calls asks **once**: `ToolRegistry.prepare_batch_confirm` pre-scans it (unknown tools, undecodable arguments, calls missing a required argument, and calls YOLO already covers are left out — nobody is asked about a call that cannot run or will not ask), two or more soft reasons collapse into one `ToolBatchConfirmHook` prompt whose `BatchConfirmDecision` `execute(..., decision=…)` applies to each call (CLI box lists every call: `y` = allow these, `m` = review them one at a time through the per-call prompt, `N` = deny with one comment shared by all), the answer covers that round only (the next one asks again), and a lone request keeps the plain one-call prompt. Every allow says so in the result itself: the tool text leads with `[approval] user (once)`, `user (round)` (batch), `grant (session)` (a TRUSTABLE grant, so nobody was asked this time) or `yolo (auto)`, and a handler that prompts on its own records it with `tools.context.record_approval` — `workspace.check_command_allowed` (`user`/`grant (process)`), `tools/net/grants.py` (`user`/`grant (process)`) and the skill write path (`user (once|session)`, `grant (session)`, `config (policy)`). One line per call (`Approval` + `most_specific_approval`: the answer closest to a human wins), none for a call that needed no confirmation or ended in `error:`, and it leads so a wire-capped result and `read_document` keep it. A refusal names which no it was (`Denial` / `DenialKind`): `user declined` (with the comment), `no answer within 30s — denied by default` (the prompt's deadline), `prompt cancelled` (Ctrl+C / EOF), or `no confirm hook or non-interactive (denied)` — a timeout is never reported as a decision, and every timed prompt (`workspace.check_command_allowed`, `tools/net/grants.py`, `tools/skills.py`, `tools/access.py`) carries the reason it got into the tool message.
@@ -128,8 +128,8 @@ Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
 - `/btw` side questions: forked session; main transcript/DB untouched. `--tools=read` (default) exposes only `READ_ONLY` tools under `read_only_context()` (fetch=GET) with the main session view; `--tools=no` off; `--tools=full` whole-registry clone with fresh session. Tools off: default falls back to no tools; explicit `read`/`full` errors. The model gets an `aside` notice as its last message before the question (unsaved exchange + tool scope), shown to the user as `[notice] aside: …`.
 - Explicit `/resume` or `--session` from another workspace prompts: **keep** / **update** / **abort**. A session resumed by a **freshly started process** (`--session` / default latest) also gets a `resume` notice about what a restart drops (PTY sessions, running commands, temp workspaces) — resuming inside a live process pushes nothing.
 - Host notices (`agent/notices.py`): system-level messages appended as trailing `developer` rows, never system-prompt edits — `Notice.to_message()` for the model, `.summary` for the user via `cli.display.echo_notice` (`--quiet` hides the user line only). Durable hosts persist them (`ChatAgent.push_notice`); a side agent without memory keeps its notice local. `has_notice()` dedupes a session-scoped notice: the `skills` one is pushed once per session (feature enabled + tools on), lists nothing, and makes the model ask with `skill_list` / `skill_search`. Trailing developer messages never end a turn, so `/retry` still sees an interrupted turn behind a notice or checkpoint.
-- Failed/cancelled turns: user kept; **committed tool rounds kept**, and a batch cut short keeps the calls that already produced a result (`rollback_tail`; written to the store, so a resume continues from them instead of re-running them) — side effects are not re-run on `/retry`, and only unfinished assistant text is dropped; a round cut off at the output limit is not a failure at all: it is kept under a `truncated` notice (see the `run_chat_loop` bullet), which also keeps the turn unfinished for `/retry`; Ctrl+C cancels the in-flight turn (a session-scoped asyncio SIGINT handler routes it, `cli/interrupt.py` — a stray Ctrl+C while the loop waits never tears the REPL down); TTY confirms/ask prompts run off-loop and register their own SIGINT target, so Ctrl+C there cancels just the poll-based read (`wait` reports `cancelled by user`) or is ignored with a hint for readline prompts (`ask_user_*`, confirms) — never the turn, never the loop; auto-retry waits 5s/10s/15s/20s then +10s (10 attempts), its countdown is Ctrl+C-cancellable, and the budget **resets when an attempt completes a model round** (connection recovered); then `/retry`.
-- Pretty tool lines: every builtin tool prints its `* Verb 'target' ` prefix as the call starts and appends the outcome (`L1-4 (done)`, `(exit code 0)`, `(session 3)`) when the result lands, so a blocking tool stays visible while it runs; namespaced MCP tools share one generic line (`* MCP <server>:<tool> (done)`). Plugin tools and custom tool calls keep the `[tool] …` / `[tool ok] …` whole-line style. Interactive TTY only (`interactive_terminal()`): piped/plain output and `--verbose` keep whole lines. A second call of the same round holds its prefix back until the open row takes its own detail (results arrive in call order), so each call still prints one finished row; a prompt box (or other output) printed while the tool runs *ends* the open prefix as its own line and falls back to whole lines for the rest of the batch — the cursor is never moved, so a prefix wider than the terminal cannot be half-erased. A batch (one round's calls) that only touches todo tools prints one `* Todo x<N>: Push, Update` line plus its last result, whose text already carries the final stack, instead of one header and a full list render per call. `run_argv_batch` lists its commands above its row: one `$ command` line per step, a `pipe_out` chain joined into one line with ` | `, each line elided past 80 characters.
+- Failed/cancelled turns: user kept; **committed tool rounds kept**, and a batch cut short keeps the calls that already produced a result (`rollback_tail`; written to the store, so a resume continues from them instead of re-running them) — side effects are not re-run on `/retry`, and only unfinished assistant text is dropped; a round cut off at the output limit is not a failure at all: it is kept under a `truncated` notice (see the `run_chat_loop` bullet), which also keeps the turn unfinished for `/retry`; Ctrl+C cancels the in-flight turn (a session-scoped asyncio SIGINT handler routes it, `cli/interrupt.py` — a stray Ctrl+C while the loop waits never tears the REPL down; the handler is rebound to whichever loop owns the session, since asyncio drops it with its loop); TTY confirms/ask prompts run off-loop and register their own SIGINT target, so Ctrl+C there cancels just the poll-based read (`wait` reports `cancelled by user`) or is ignored with a hint for readline prompts (`ask_user_*`, confirms) — never the turn, never the loop (a slash command's own work is a task too — `/compact` cancels its summary request, `/btw` its side turn — and `handle_slash` answers that `CancelledError` with one `cancelled` line, so the REPL survives; before this, nothing was registered as a target during a slash command and the router dropped the signal); auto-retry waits 5s/10s/15s/20s then +10s (10 attempts), its countdown is Ctrl+C-cancellable, and the budget **resets when an attempt completes a model round** (connection recovered); then `/retry`.
+- Pretty tool lines: every builtin tool prints its `* Verb 'target' ` prefix as the call starts and appends the outcome (`L1-4 (done)`, `(exit code 0)`, `(session 3)`) when the result lands, so a blocking tool stays visible while it runs; namespaced MCP tools share one generic line (`* MCP <server>:<tool> (done)`). Plugin tools and custom tool calls keep the `[tool] …` / `[tool ok] …` whole-line style. Interactive TTY only (`interactive_terminal()`): piped/plain output and `--verbose` keep whole lines. A second call of the same round holds its prefix back until the open row takes its own detail (results arrive in call order), so each call still prints one finished row; a prompt box (or other output) printed while the tool runs *ends* the open prefix as its own line and falls back to whole lines for the rest of the batch — the cursor is never moved, so a prefix wider than the terminal cannot be half-erased. A batch (one round's calls) that only touches todo tools prints one `* Todo x<N>: Push, Update` line plus its last result, whose text already carries the final stack, instead of one header and a full list render per call. `run_argv_batch` lists its commands above its row: one `$ command` line per node, a `pipe` chain joined into one line with ` | `, a `then` / `else` body indented (two spaces per level) under the node it follows, each line elided past 80 characters.
 - **`plyngent providers`**, **`config path|edit`**. Config open: `$VISUAL`/`$EDITOR` (wait), else system default (`xdg-open` / `open` / `startfile`, non-blocking). `/edit` stays blocking-only. No providers → optional edit then reload when waited.
 - Tools default on; workspace defaults to cwd; `--max-rounds` default 32. Readline history under platformdirs (`repl_history`). PTY: `close_all` on chat exit.
 

README.md

@@ -315,7 +315,7 @@ Type `/help` in the REPL for the live list. Common ones:
 | `/sessions` | Sessions for this workspace |
 | `/new` `/resume` `/rename` `/delete` | Session lifecycle (`/delete` confirms) |
 | `/export [md\|json] [path]` | Transcript from DB (no secrets) |
-| `/compact` | Soft-compact + model summary into a **new** session |
+| `/compact` | Soft-compact + model summary into a **new** session (Ctrl+C drops the summary request, and no session is created) |
 | `/stream` `/verbose` `/markdown` `/tools` `/rounds` | Toggles and limits |
 | `/yolo [on\|off\|once]` | Soft destructive confirms: sticky skip, off, or next turn only |
 
@@ -362,7 +362,7 @@ Safety defaults:
   `write_pty` is literal text only; use `write_pty_keys` for keys and bytes: `<c-c>` `<esc>` `<enter>` `<s-tab>` `<up>` … `<f12>`, `<c-up>` `<s-home>` `<c-s-f5>` for modifier combinations, `<up*3>` to repeat, `<m-x>` for Alt, and `\xHH` / `\uHHHH` for raw bytes. An unknown `<…>` is an error, never text typed into the terminal.
   For passwords/sudo/ssh prompts use `ask_into_pty` (human types locally; answer never returns to the model).
 
-- Mis-typed tool arguments are answered, never guessed: a string where an array is expected (`run_argv` `argv`, `ask_user_choice` `options`, `ask_user_form` `fields`), a call that omits a required argument, or a non-numeric `wait` duration (numeric strings are accepted) returns an error showing the shape/names to send — the human never sees a garbage prompt, and no confirm prompt fires for a call that cannot run.
+- Mis-typed tool arguments are answered, never guessed: a string where an array is expected (`run_argv` `argv`, `ask_user_choice` `options`, `ask_user_form` `fields`), a call that omits a required argument, a non-numeric `wait` duration (numeric strings are accepted), or a `timeout_seconds` that is not a positive number returns an error showing the shape/names to send — the human never sees a garbage prompt, and no confirm prompt fires for a call that cannot run (a tool declares that pre-prompt check with `@tool(validate=…)`; `run_argv`, `run_argv_batch` (also for a tree it cannot build), `open_pty`, `write_pty_keys`, `edit_replace`, `edit_lineno` and `fetch_local` do).
 
 ## Usage / context (CLI)
 

pdm.lock

@@ -499,13 +499,13 @@ files = [
 
 [[package]]
 name = "platformdirs"
-version = "4.14.0"
+version = "4.14.3"
 requires_python = ">=3.11"
 summary = "A small Python package for determining appropriate platform-specific dirs, e.g. a `user data dir`."
 groups = ["default"]
 files = [
-    {file = "platformdirs-4.14.0-py3-none-any.whl", hash = "sha256:1dadfc59a8a5d7565b3d9270ae8fa71c065595352c4277b41c11ee6c222c4b91"},
-    {file = "platformdirs-4.14.0.tar.gz", hash = "sha256:83f009c1e8c5fc36869b5171755b291887bd2cade7965233cdfabd23d754720b"},
+    {file = "platformdirs-4.14.3-py3-none-any.whl", hash = "sha256:cba2d05fe5e18b26b1fa6f376b8e6234c012abe1f8696c486fface7d0054e931"},
+    {file = "platformdirs-4.14.3.tar.gz", hash = "sha256:021add1ab113483edbcd4694f5fbb2d688b893ea30d35e1155bb371cf4be0be8"},
 ]
 
 [[package]]
@@ -552,18 +552,18 @@ files = [
 
 [[package]]
 name = "pytest-asyncio"
-version = "1.4.1"
+version = "1.4.1.post1"
 requires_python = ">=3.10"
 summary = "Pytest support for asyncio"
 groups = ["dev"]
 dependencies = [
     "backports-asyncio-runner<2,>=1.1; python_version < \"3.11\"",
-    "pytest<10,>=8.4",
+    "pytest>=8.4",
     "typing-extensions>=4.12; python_version < \"3.13\"",
 ]
 files = [
-    {file = "pytest_asyncio-1.4.1-py3-none-any.whl", hash = "sha256:507afa78ec77c6661164ef1a2368bc7820383fd1a14b2a7f813de8b5a92949cc"},
-    {file = "pytest_asyncio-1.4.1.tar.gz", hash = "sha256:e8266eb094e5ee9ffdb2068c9836f33e209ea77a970eb846933d416b5602669b"},
+    {file = "pytest_asyncio-1.4.1.post1-py3-none-any.whl", hash = "sha256:0755e488cc0a2b9bd2eab7292cc6390d4cbb8390f1020745647bae5a708e0165"},
+    {file = "pytest_asyncio-1.4.1.post1.tar.gz", hash = "sha256:f582d9ea4a1c082195abe8bd29a1fe106aeb1cd435e3dad2d3bc2749935c36a2"},
 ]
 
 [[package]]
@@ -818,7 +818,7 @@ files = [
 
 [[package]]
 name = "winrt-runtime"
-version = "4.0.3"
+version = "4.1.0"
 requires_python = ">=3.11"
 summary = "The runtime that every PyWinRT projection package runs on"
 groups = ["notify"]
@@ -827,118 +827,118 @@ dependencies = [
     "typing-extensions>=4.12.2",
 ]
 files = [
-    {file = "winrt_runtime-4.0.3-cp314-cp314-win32.whl", hash = "sha256:fee35995e7dfb083b5024d7a11d63ad2d2c74e7a9fc6a58dcdede52c1f0a321d"},
-    {file = "winrt_runtime-4.0.3-cp314-cp314-win_amd64.whl", hash = "sha256:03d76d06172b6f5bb6312aa24f0d25761b1e45fe2b7c3c0b4ad0cd73a06bb293"},
-    {file = "winrt_runtime-4.0.3-cp314-cp314-win_arm64.whl", hash = "sha256:9731462174346a2ede57660b9d521e41a16cd7e59890dc292c4250c42d1642a9"},
-    {file = "winrt_runtime-4.0.3-cp314-cp314t-win32.whl", hash = "sha256:dc545f834a3dd2bcc69512193677f0918788ccce4871ca24937aa6f171ea7112"},
-    {file = "winrt_runtime-4.0.3-cp314-cp314t-win_amd64.whl", hash = "sha256:b0b06be21eb17248ce32d43a4a027ba22651065f9af5b42c3454575766879e1e"},
-    {file = "winrt_runtime-4.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:2b8d6a55d695e54a25e262a6d1134c510a1e291cf2c3d141990a8c209f87dc40"},
-    {file = "winrt_runtime-4.0.3-cp315-cp315-win32.whl", hash = "sha256:ff7920a2d42734c545984ef671376713ce09f8e59fc9a5b420395406e2c1eea7"},
-    {file = "winrt_runtime-4.0.3-cp315-cp315-win_amd64.whl", hash = "sha256:403a80372f33635851814760d1acd6d8bbdb0dedff7812203d0dd4cfba809e8b"},
-    {file = "winrt_runtime-4.0.3-cp315-cp315-win_arm64.whl", hash = "sha256:52eb937ab71e39ddf798f16dfe47d572a8c097ac6d1eb2ddb700fc7977591241"},
-    {file = "winrt_runtime-4.0.3-cp315-cp315t-win32.whl", hash = "sha256:325a701c70e32b31139bf650c31fa352f4eae108fb0488aad074fb7026b1e650"},
-    {file = "winrt_runtime-4.0.3-cp315-cp315t-win_amd64.whl", hash = "sha256:f0e5ce8fc5e6453fbad7ed3197010586337a9a4d42fce29847d8a5573f29d8e9"},
-    {file = "winrt_runtime-4.0.3-cp315-cp315t-win_arm64.whl", hash = "sha256:9580aaf167a9e1d9d790b57e9dfaa5499c8a625bd85f3872a5097915d25f1c24"},
-    {file = "winrt_runtime-4.0.3.tar.gz", hash = "sha256:b04877a5b1f2ca67084d62731546f7b512c8291e2b70b95def82a1c4d0bfe993"},
+    {file = "winrt_runtime-4.1.0-cp314-cp314-win32.whl", hash = "sha256:c51f2d9c0835e3628b3281eac1d5ef2f7cc772302d6a038f5d2bce9e5591d179"},
+    {file = "winrt_runtime-4.1.0-cp314-cp314-win_amd64.whl", hash = "sha256:af422bbcffa7249770ab118bc5b9aa1f8fcd401566abca478f91cf7bd9a8b7dd"},
+    {file = "winrt_runtime-4.1.0-cp314-cp314-win_arm64.whl", hash = "sha256:453b6d6f8568d639038fe313b37661acf372a8173040f41a36305be95dbf14f0"},
+    {file = "winrt_runtime-4.1.0-cp314-cp314t-win32.whl", hash = "sha256:d8c4654ec0f214a29354b22039d8bd9728968f3195c4cafcded8bfe57cb6fd36"},
+    {file = "winrt_runtime-4.1.0-cp314-cp314t-win_amd64.whl", hash = "sha256:77b79a83e8af56ea98ae8644d1e9e753376c8d97586986eb05068d49d653389f"},
+    {file = "winrt_runtime-4.1.0-cp314-cp314t-win_arm64.whl", hash = "sha256:4fd0f5779c793235aa10d09dfe851c2e7830739b4f41766cc865d4218d2a403f"},
+    {file = "winrt_runtime-4.1.0-cp315-cp315-win32.whl", hash = "sha256:a1418ed9c3592c48fa74f30004abf10d7e96df33f84e782aa61387af16013375"},
+    {file = "winrt_runtime-4.1.0-cp315-cp315-win_amd64.whl", hash = "sha256:281a6085d07dd941873ef16b7bc236f37ad45bd166435e8cabdeff00c67d2830"},
+    {file = "winrt_runtime-4.1.0-cp315-cp315-win_arm64.whl", hash = "sha256:a2bd0d3018f3c2add91f961305e000e56ef9ec222359d069f91cc2e0d4078ace"},
+    {file = "winrt_runtime-4.1.0-cp315-cp315t-win32.whl", hash = "sha256:db36eeae93d36d41e6e7e67b529de122b2798ad687d661e63b287169624f6d44"},
+    {file = "winrt_runtime-4.1.0-cp315-cp315t-win_amd64.whl", hash = "sha256:cf9f050cc3565163ec512eda2f834d7c201a2ece54232d7dda770bd9d232d352"},
+    {file = "winrt_runtime-4.1.0-cp315-cp315t-win_arm64.whl", hash = "sha256:c299e9b991b623413c5c8773dfaa3f72d039128bdce6ff0b9815c3cb07ba793e"},
+    {file = "winrt_runtime-4.1.0.tar.gz", hash = "sha256:d345452652ce7243ce3d97ebf63067f848717a8328cca8c4e851c2d0a53ef577"},
 ]
 
 [[package]]
 name = "winrt-windows-applicationmodel-core"
-version = "4!10.0.28000.2705"
+version = "4!10.0.28000.2957"
 requires_python = ">=3.11"
 summary = "PyWinRT projection of the Windows.ApplicationModel.Core namespace"
 groups = ["notify"]
 marker = "sys_platform == \"win32\""
 dependencies = [
     "winrt-runtime<5,>=4.0.0",
-    "winrt-windows-foundation-collections>=4!10.0.28000.2705",
-    "winrt-windows-foundation>=4!10.0.28000.2705",
+    "winrt-windows-foundation-collections>=4!10.0.28000.2957",
+    "winrt-windows-foundation>=4!10.0.28000.2957",
 ]
 files = [
-    {file = "winrt_windows_applicationmodel_core-4!10.0.28000.2705-py3-none-any.whl", hash = "sha256:c6c0a854e8ab3620899ae83671bb30853183cc43d7b7ab9d2cb11fe60f2e1138"},
-    {file = "winrt_windows_applicationmodel_core-4!10.0.28000.2705.tar.gz", hash = "sha256:f06920b88d030266cd2c0ef270dbcf3dec1f438e95e458933cba18103d35902b"},
+    {file = "winrt_windows_applicationmodel_core-4!10.0.28000.2957-py3-none-any.whl", hash = "sha256:55762c64133a38b3c1b50303030f440657bf37b3905f46929b5c5b445799fff1"},
+    {file = "winrt_windows_applicationmodel_core-4!10.0.28000.2957.tar.gz", hash = "sha256:eb1dd1d047a8aec26593956d52ed8204d9f3def71bf64a0fc6f676c0ba1b5663"},
 ]
 
 [[package]]
 name = "winrt-windows-data-xml-dom"
-version = "4!10.0.28000.2705"
+version = "4!10.0.28000.2957"
 requires_python = ">=3.11"
 summary = "PyWinRT projection of the Windows.Data.Xml.Dom namespace"
 groups = ["notify"]
 marker = "sys_platform == \"win32\""
 dependencies = [
     "winrt-runtime<5,>=4.0.0",
-    "winrt-windows-foundation-collections>=4!10.0.28000.2705",
-    "winrt-windows-foundation>=4!10.0.28000.2705",
-    "winrt-windows-storage-streams>=4!10.0.28000.2705",
+    "winrt-windows-foundation-collections>=4!10.0.28000.2957",
+    "winrt-windows-foundation>=4!10.0.28000.2957",
+    "winrt-windows-storage-streams>=4!10.0.28000.2957",
 ]
 files = [
-    {file = "winrt_windows_data_xml_dom-4!10.0.28000.2705-py3-none-any.whl", hash = "sha256:5784183aca5a7f0f297e21789de34312702207e767954dc96d4197588c714a5a"},
-    {file = "winrt_windows_data_xml_dom-4!10.0.28000.2705.tar.gz", hash = "sha256:9dade5d5e6768077009babc965c56e0d39be53c329e81ad06867bfb2b82516dd"},
+    {file = "winrt_windows_data_xml_dom-4!10.0.28000.2957-py3-none-any.whl", hash = "sha256:26a516c6484744260cfb15ec44510d5bf020ee29a946f87f38ef7eac7df3673e"},
+    {file = "winrt_windows_data_xml_dom-4!10.0.28000.2957.tar.gz", hash = "sha256:03643378e9123c85448c57965a056a7f76ebb63f67b00610bd4cf3095d2e9387"},
 ]
 
 [[package]]
 name = "winrt-windows-foundation"
-version = "4!10.0.28000.2705"
+version = "4!10.0.28000.2957"
 requires_python = ">=3.11"
 summary = "PyWinRT projection of the Windows.Foundation namespace"
 groups = ["notify"]
 marker = "sys_platform == \"win32\""
 dependencies = [
     "winrt-runtime<5,>=4.0.0",
-    "winrt-windows-foundation-collections>=4!10.0.28000.2705",
+    "winrt-windows-foundation-collections>=4!10.0.28000.2957",
 ]
 files = [
-    {file = "winrt_windows_foundation-4!10.0.28000.2705-py3-none-any.whl", hash = "sha256:b409530668b101cb17b3a6ce02061b9839665c92b64b0563c106186d87562571"},
-    {file = "winrt_windows_foundation-4!10.0.28000.2705.tar.gz", hash = "sha256:eb33cc2fdf0d8c5358cb7a731563118de0f16ddc75e256b8fd83c8117a6405a0"},
+    {file = "winrt_windows_foundation-4!10.0.28000.2957-py3-none-any.whl", hash = "sha256:0669de7ce480806a0f9a46d005837799e37f1738911b54c4bc04788b37974e69"},
+    {file = "winrt_windows_foundation-4!10.0.28000.2957.tar.gz", hash = "sha256:6f7d8e932e283991cd931741f84dc7c2e8bc595b4df09de73d16c1d580fc678a"},
 ]
 
 [[package]]
 name = "winrt-windows-foundation-collections"
-version = "4!10.0.28000.2705"
+version = "4!10.0.28000.2957"
 requires_python = ">=3.11"
 summary = "PyWinRT projection of the Windows.Foundation.Collections namespace"
 groups = ["notify"]
 marker = "sys_platform == \"win32\""
 dependencies = [
     "winrt-runtime<5,>=4.0.0",
-    "winrt-windows-foundation>=4!10.0.28000.2705",
+    "winrt-windows-foundation>=4!10.0.28000.2957",
 ]
 files = [
-    {file = "winrt_windows_foundation_collections-4!10.0.28000.2705-py3-none-any.whl", hash = "sha256:f2169228644b86d752dca6f4ba0bb55ee27c5cc1062f3a3f0fbb375f48894bd7"},
-    {file = "winrt_windows_foundation_collections-4!10.0.28000.2705.tar.gz", hash = "sha256:8d396ea352d3e57a534223314aa935c2fe3e3edaa6b38439b13d5ddfeb5a12ea"},
+    {file = "winrt_windows_foundation_collections-4!10.0.28000.2957-py3-none-any.whl", hash = "sha256:0401c1b6ed512d1fdad8ba1571698e9000ebee6a69781db9a01994ffcc54a503"},
+    {file = "winrt_windows_foundation_collections-4!10.0.28000.2957.tar.gz", hash = "sha256:38989873bbdace2346d24168ceb66359800b7429128ea2ef1a36da55b255af1e"},
 ]
 
 [[package]]
 name = "winrt-windows-storage-streams"
-version = "4!10.0.28000.2705"
+version = "4!10.0.28000.2957"
 requires_python = ">=3.11"
 summary = "PyWinRT projection of the Windows.Storage.Streams namespace"
 groups = ["notify"]
 marker = "sys_platform == \"win32\""
 dependencies = [
-    "winrt-runtime<5,>=4.0.0",
-    "winrt-windows-foundation-collections>=4!10.0.28000.2705",
-    "winrt-windows-foundation>=4!10.0.28000.2705",
+    "winrt-runtime<5,>=4.1.0",
+    "winrt-windows-foundation-collections>=4!10.0.28000.2957",
+    "winrt-windows-foundation>=4!10.0.28000.2957",
 ]
 files = [
-    {file = "winrt_windows_storage_streams-4!10.0.28000.2705-py3-none-any.whl", hash = "sha256:2824e516e328c9ba76ffc7926431255b124b4a9f52c656f0b1e0056a7f8ca7b8"},
-    {file = "winrt_windows_storage_streams-4!10.0.28000.2705.tar.gz", hash = "sha256:9511b19e9cc588dcc81969b2871e465385f2ab37a62a9019f90dc37e03a5dd16"},
+    {file = "winrt_windows_storage_streams-4!10.0.28000.2957-py3-none-any.whl", hash = "sha256:e113832320f2e7f8b09cedfb400cb7eaf0ac769ad1f2038ddbe88801497916c4"},
+    {file = "winrt_windows_storage_streams-4!10.0.28000.2957.tar.gz", hash = "sha256:6284341b4b01ab872007def0b8d29bf30d84d46f6922f6e42be1d738d6bf2ff7"},
 ]
 
 [[package]]
 name = "winrt-windows-ui-notifications"
-version = "4!10.0.28000.2705"
+version = "4!10.0.28000.2957"
 requires_python = ">=3.11"
 summary = "PyWinRT projection of the Windows.UI.Notifications namespace"
 groups = ["notify"]
 marker = "sys_platform == \"win32\""
 dependencies = [
     "winrt-runtime<5,>=4.0.0",
-    "winrt-windows-foundation-collections>=4!10.0.28000.2705",
-    "winrt-windows-foundation>=4!10.0.28000.2705",
+    "winrt-windows-foundation-collections>=4!10.0.28000.2957",
+    "winrt-windows-foundation>=4!10.0.28000.2957",
 ]
 files = [
-    {file = "winrt_windows_ui_notifications-4!10.0.28000.2705-py3-none-any.whl", hash = "sha256:909fdfe7dc7418adc59029fa723f24db5b28836bc017d791b03e4e7cda02117c"},
-    {file = "winrt_windows_ui_notifications-4!10.0.28000.2705.tar.gz", hash = "sha256:8fd5a28f81ef4b422b444b2889e2dec6a316692b7c2351a6daff39ccaf591bba"},
+    {file = "winrt_windows_ui_notifications-4!10.0.28000.2957-py3-none-any.whl", hash = "sha256:2d4e78018df29bb107d0490926ae0a7254be68b6b9c1bd81547d0df569682ea7"},
+    {file = "winrt_windows_ui_notifications-4!10.0.28000.2957.tar.gz", hash = "sha256:5ab89819245c7a90db2d5fe2a0d0cc88158d50bf6fbba12c96145b25bc04a9d3"},
 ]

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "plyngent"
-version = "0.15.0"
+version = "0.16.0"
 description = "LLM chat and agent toolkit"
 authors = [
   { name = "HivertMoZara", email = "worldmozara@163.com" },

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/cli/display.py

@@ -36,7 +36,8 @@ _TOOL_ARGS_PREVIEW = 80
 _ASK_SUBJECT_PREVIEW = 60
 _PTY_PAYLOAD_PREVIEW = 40
 # ``run_argv_batch`` lists the commands it runs above its ``* Run batch `` row — one
-# line per command, a ``pipe_out`` chain joined into one — elided past this width.
+# line per command, a pipe chain joined into one, a ``then`` / ``else`` body
+# indented under the node it follows — elided past this width.
 _RUN_BATCH_CMD_LIMIT = 80
 _MCP_TOOL_PREFIX = "mcp__"
 _SKILL_MAIN_FILE = "SKILL.md"
@@ -417,32 +418,54 @@ def _batch_steps(args_json: str) -> list[dict[str, object]]:
             value = json.loads(value)
         except ValueError:
             return []
+    return _batch_nodes(value)
+
+
+def _batch_nodes(value: object) -> list[dict[str, object]]:
+    """Nodes of one continuation: a single node, or a list of them."""
+    if isinstance(value, dict):
+        return [cast("dict[str, object]", value)]
     if not isinstance(value, list):
         return []
-    raw_steps = cast("list[object]", value)
-    return [cast("dict[str, object]", step) for step in raw_steps if isinstance(step, dict)]
+    return [cast("dict[str, object]", item) for item in cast("list[object]", value) if isinstance(item, dict)]
 
 
-def _batch_command_lines(args_json: str) -> list[str]:
-    """One command per line, a ``pipe_out`` chain joined into one line with ``|``."""
-    lines: list[str] = []
-    piped_from_previous = False
-    for step in _batch_steps(args_json):
-        argv = _str_list(step.get("argv"))
-        command = shlex.join(argv) if argv else "?"
-        if piped_from_previous and lines:
-            lines[-1] = f"{lines[-1]} | {command}"
-        else:
-            lines.append(command)
-        # Same flag spellings the tool itself accepts (bool, or "true"/"1"/…).
-        piped_from_previous = str(step.get("pipe_out") or "").strip().lower() in {"1", "true", "yes", "on"}
+def _batch_chain(node: dict[str, object]) -> str:
+    """``a | b | c`` for a node and the commands its pipe feeds."""
+    argv = _str_list(node.get("argv"))
+    parts = [shlex.join(argv) if argv else "?"]
+    parts.extend(_batch_chain(child) for child in _batch_nodes(node.get("pipe")))
+    return " | ".join(parts)
+
+
+def _batch_node_lines(
+    nodes: list[dict[str, object]],
+    depth: int,
+    prefix: str = "",
+) -> list[tuple[str, bool]]:
+    """One line per command, and whether it is a command (``$``) or a branch."""
+    lines: list[tuple[str, bool]] = []
+    for node in nodes:
+        lines.append((f"{'  ' * depth}{prefix}{_batch_chain(node)}", not prefix))
+        for branch in ("then", "else"):
+            children = _batch_nodes(node.get(branch))
+            if children:
+                lines.extend(_batch_node_lines(children, depth + 1, f"{branch} "))
     return lines
 
 
+def _batch_command_lines(args_json: str) -> list[tuple[str, bool]]:
+    """One line per command, a pipe chain joined into one line with ``|``."""
+    return _batch_node_lines(_batch_steps(args_json), 0)
+
+
 def _run_argv_batch_prefix(args_json: str) -> str:
     """``* Run batch `` with the commands it runs listed above it."""
-    commands = _batch_command_lines(args_json)
-    body = "\n".join(_pretty_prefix(f"$ {_preview(command, _RUN_BATCH_CMD_LIMIT)}") for command in commands)
+    lines = [
+        _pretty_prefix(f"$ {_preview(text, _RUN_BATCH_CMD_LIMIT)}" if is_command else f"{text}")
+        for text, is_command in _batch_command_lines(args_json)
+    ]
+    body = "\n".join(lines)
     header = _pretty_prefix("* Run batch ")
     return f"{body}\n{header}" if body else header
 

src/plyngent/cli/interrupt.py

@@ -67,6 +67,10 @@ class _RouterState:
     # Cancel targets, innermost last: a SIGINT cancels the innermost one.
     targets: list[InterruptTarget] = field(default_factory=list)
     installed: bool = False
+    # The loop that owns the handler. The flag alone is process-wide, so a later
+    # loop (a new session, another test) must rebind instead of believing the
+    # handler still follows it — asyncio detaches handlers with their loop.
+    loop: asyncio.AbstractEventLoop | None = None
     # Nesting depth of main-thread blocking reads. A plain counter (not a
     # ContextVar): asyncio.add_signal_handler freezes the ContextVar snapshot at
     # install time, so a ContextVar reset after reinstall would leave SIGINT
@@ -78,8 +82,14 @@ _router = _RouterState()
 
 
 def sigint_routed() -> bool:
-    """Whether the session SIGINT router currently owns SIGINT."""
-    return _router.installed
+    """Whether the session SIGINT router owns SIGINT on the running loop."""
+    if not _router.installed:
+        return False
+    try:
+        loop: asyncio.AbstractEventLoop | None = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = None
+    return loop is None or _router.loop is loop
 
 
 def install_sigint_router() -> bool:
@@ -95,13 +105,19 @@ def install_sigint_router() -> bool:
     Blocking main-thread reads suspend the router via
     :func:`pause_task_cancel_for_prompt`. Platforms without asyncio signal
     handlers (Windows proactor) return False and keep the plain
-    KeyboardInterrupt behaviour. Idempotent.
+    KeyboardInterrupt behaviour. Idempotent for the loop that owns it, and a
+    different running loop rebinds the handler to itself.
     """
-    if _router.installed:
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return False
+    if _router.installed and _router.loop is loop:
         return True
-    if not _add_router_handler():
+    if not _add_router_handler(loop):
         return False
     _router.installed = True
+    _router.loop = loop
     return True
 
 
@@ -111,21 +127,32 @@ def uninstall_sigint_router() -> None:
         return
     _router.installed = False
     _remove_router_handler()
+    _router.loop = None
     _router.targets.clear()
     install_keyboard_interrupt_sigint()
 
 
-def _add_router_handler() -> bool:
+def _add_router_handler(loop: asyncio.AbstractEventLoop | None = None) -> bool:
+    """Route SIGINT to the router on *loop* (the running one when omitted)."""
+    target = loop
+    if target is None:
+        try:
+            target = asyncio.get_running_loop()
+        except RuntimeError:
+            return False
     try:
-        asyncio.get_running_loop().add_signal_handler(signal.SIGINT, _route_sigint)
+        target.add_signal_handler(signal.SIGINT, _route_sigint)
     except RuntimeError, NotImplementedError, ValueError:
         return False
     return True
 
 
 def _remove_router_handler() -> None:
+    loop = _router.loop
+    if loop is None:
+        return
     with contextlib.suppress(NotImplementedError, RuntimeError, ValueError):
-        _ = asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)
+        _ = loop.remove_signal_handler(signal.SIGINT)
 
 
 def _route_sigint() -> None:

src/plyngent/cli/slash.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import os
 import shlex
 from typing import TYPE_CHECKING, Any, Literal, cast, override
@@ -1639,6 +1640,11 @@ async def handle_slash(state: ReplState, line: str) -> bool:
     args[0] = args[0].lower()
     try:
         await awaitlet.async_def(_run_slash_argv, args, state)
+    except asyncio.CancelledError:
+        # A command's own cancellable work was stopped (``/btw``'s side turn, a
+        # prompt): the interrupt belongs to that command. Letting it through
+        # would cancel this REPL task and end the session instead.
+        click.secho("cancelled", fg="yellow")
     except ReplExitError:
         return False
     except click.ClickException as exc:

src/plyngent/cli/state.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import contextlib
 import time
 from dataclasses import dataclass, field
@@ -1007,16 +1008,29 @@ class ReplState:
             hint = self.agent.last_request_usage.prompt_tokens
             # Approximate: calibrate against current full history char-est.
             sent_est = estimate_messages_tokens(messages)
-        summary = await summarize_messages(
-            self.client,
-            messages,
-            model=self.model,
-            max_context_tokens=self.agent.max_context_tokens,
-            prompt_tokens_hint=hint,
-            sent_estimate_tokens=sent_est,
-            system_prompt=self.config.agent_config.compact_system_prompt or None,
-            user_prefix=self.config.agent_config.compact_user_prefix or None,
-        )
+        # The summary is the only long wait in here, and nothing has been
+        # created yet, so it is the safe place to be interrupted: Ctrl+C drops
+        # the request and reports back (see the CancelledError below). Everything
+        # after it is a couple of DB writes that build the new session, and a
+        # cancel that landed there would leave it half-seeded.
+        from plyngent.cli.retry import run_cancellable
+
+        try:
+            summary = await run_cancellable(
+                summarize_messages(
+                    self.client,
+                    messages,
+                    model=self.model,
+                    max_context_tokens=self.agent.max_context_tokens,
+                    prompt_tokens_hint=hint,
+                    sent_estimate_tokens=sent_est,
+                    system_prompt=self.config.agent_config.compact_system_prompt or None,
+                    user_prefix=self.config.agent_config.compact_user_prefix or None,
+                )
+            )
+        except asyncio.CancelledError:
+            msg = "compaction cancelled (nothing changed)"
+            raise ValueError(msg) from None
         carried_todos = self.todo_stack.to_raw()
         session_name = name or f"compact-from-{old_id}"
         await self.new_session(name=session_name)

src/plyngent/config/models.py

@@ -76,7 +76,7 @@ confirm blocks the turn, so it costs time. Invoke the target command directly (a
 of wrapping it in `bash -c` / `python -c` / `node`, and reach for a dedicated tool \
 (`read_file` / `edit_*` / `vcs_*` / `fetch` / `fetch_local` / …) when one already does the job.
 - Several `run_argv` calls in one step may run in parallel; use `run_argv_batch` \
-for ordered pipelines (`pipe_out` / `mix_stderr` as needed).
+for ordered commands, with `then` / `else` / `pipe` continuations as needed.
 - Prefer `vcs_*` for status/diff/log/branch when enough.
 
 ### Network

src/plyngent/tools/chat/shape.py

@@ -14,20 +14,9 @@ tools themselves stay free of unnecessary-isinstance warnings.
 
 from __future__ import annotations
 
-_SHAPES: dict[type, str] = {
-    str: "a string",
-    list: "an array",
-    dict: "an object",
-    bool: "a boolean",
-    int: "a number",
-    float: "a number",
-    type(None): "null",
-}
+from plyngent.tools.shape import shape_of
 
-
-def shape_of(value: object) -> str:
-    """Model-facing name for a decoded JSON value (``a string``, ``an array``…)."""
-    return _SHAPES.get(type(value), type(value).__name__)
+__all__ = ["first_error", "list_error", "number_arg", "shape_of", "string_error"]
 
 
 def first_error(*errors: str | None) -> str | None:

src/plyngent/tools/danger.py

@@ -10,7 +10,7 @@ from plyngent.tools.trash import under_system_temp
 from plyngent.tools.workspace import AccessMode, WorkspaceError, get_trash_enabled, resolve_path
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping, Sequence
+    from collections.abc import Iterator, Mapping, Sequence
 
 
 def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | None:
@@ -188,6 +188,32 @@ def _batch_step_argv(item: object) -> list[str] | None:
     return parts or None
 
 
+def _batch_nodes(value: object) -> list[dict[str, object]]:
+    """Nodes of one continuation: a single node, or a list of them."""
+    if isinstance(value, dict):
+        return [cast("dict[str, object]", value)]
+    if not isinstance(value, list):
+        return []
+    return [cast("dict[str, object]", item) for item in cast("list[object]", value) if isinstance(item, dict)]
+
+
+def _walk_batch_argv(value: object, where: str) -> Iterator[tuple[str, list[str]]]:
+    """Every command of a batch tree, in the order it can run, with its path.
+
+    ``then`` / ``else`` / ``pipe`` hold the continuations, so a command hiding in
+    one is reviewed exactly like a top-level step.
+    """
+    for index, node in enumerate(_batch_nodes(value)):
+        label = f"{where}[{index}]"
+        argv = _batch_step_argv(node)
+        if argv is not None:
+            yield label, argv
+        for branch in ("then", "else", "pipe"):
+            child = node.get(branch)
+            if child is not None:
+                yield from _walk_batch_argv(child, f"{label}.{branch}")
+
+
 def _run_argv_batch_reason(args: Mapping[str, object]) -> DangerReason | None:
     """One confirm for the whole batch if any step runs code or detaches.
 
@@ -203,12 +229,11 @@ def _run_argv_batch_reason(args: Mapping[str, object]) -> DangerReason | None:
         raw = loaded
     if not isinstance(raw, list):
         return None
-    risky = [
-        reason
-        for index, item in enumerate(cast("list[object]", raw))
-        if (argv := _batch_step_argv(item)) is not None
-        and (reason := _command_reason(argv, via=f"run_argv_batch[{index}]")) is not None
-    ]
+    risky: list[DangerReason] = []
+    for label, argv in _walk_batch_argv(cast("list[object]", raw), "run_argv_batch"):
+        reason = _command_reason(argv, via=label)
+        if reason is not None:
+            risky.append(reason)
     if not risky:
         return None
     header = f"run_argv_batch: {len(risky)} risky step(s) (review before allow)"

src/plyngent/tools/file/edit_lineno.py

@@ -3,9 +3,11 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 from plyngent.agent import ToolTag, lineno_read_lines, lineno_read_mtime, tool
+from plyngent.tools.shape import shape_of
 from plyngent.tools.workspace import AccessMode, resolve_path
 
 if TYPE_CHECKING:
+    from collections.abc import Mapping
     from pathlib import Path
 
 # How many unread line numbers to preview in the error message.
@@ -15,6 +17,28 @@ _UNREAD_PREVIEW = 8
 _READ_HINT_MARGIN = 3
 
 
+def _range_shape_error(start_line: object, end_line: object) -> str | None:
+    """The two range checks that need nothing but the two numbers."""
+    if not isinstance(start_line, int) or not isinstance(end_line, int):
+        return None  # the schema declares integers; a bad pair stays the handler's to answer
+    if start_line < 1:
+        return "error: start_line must be >= 1"
+    if end_line < start_line:
+        return "error: end_line must be >= start_line"
+    return None
+
+
+def edit_lineno_shape(args: Mapping[str, object]) -> str | None:
+    """Answer a range that cannot be applied, before anyone is asked about it."""
+    shape_error = _range_shape_error(args.get("start_line"), args.get("end_line"))
+    if shape_error is not None:
+        return shape_error
+    content = args.get("new_content")
+    if content is not None and not isinstance(content, str):
+        return f"error: `edit_lineno` expects `new_content` as a string, got {shape_of(content)}"
+    return None
+
+
 def _detect_newline(lines: list[str]) -> str:
     for sample in lines:
         if sample.endswith("\r\n"):
@@ -43,10 +67,9 @@ def _to_keepends(new_content: str, newline: str, *, force_trailing: bool) -> lis
 
 
 def _validate_range(start_line: int, end_line: int, n: int) -> str | None:
-    if start_line < 1:
-        return "error: start_line must be >= 1"
-    if end_line < start_line:
-        return "error: end_line must be >= start_line"
+    shape_error = _range_shape_error(start_line, end_line)
+    if shape_error is not None:
+        return shape_error
     if start_line > n + 1:
         return f"error: start_line {start_line} past end of file ({n} lines)"
     if start_line == n + 1 and end_line != start_line:
@@ -124,7 +147,7 @@ def _replace_range(
     )
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, writes=("path",))
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, writes=("path",), validate=edit_lineno_shape)
 async def edit_lineno(path: str, start_line: int, end_line: int, new_content: str) -> str:
     """Replace lines ``start_line``..``end_line`` (1-based, inclusive) in a file.
 

src/plyngent/tools/file/edit_replace.py

@@ -1,8 +1,27 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from plyngent.agent import ToolTag, tool
+from plyngent.tools.shape import shape_of
 from plyngent.tools.workspace import AccessMode, resolve_path
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+
+def edit_replace_shape(args: Mapping[str, object]) -> str | None:
+    """Answer a replacement that cannot apply, before anyone is asked about it."""
+    old_string = args.get("old_string")
+    if isinstance(old_string, str) and not old_string:
+        return "error: old_string must not be empty"
+    if not isinstance(old_string, str) and old_string is not None:
+        return f"error: `edit_replace` expects `old_string` as a string, got {shape_of(old_string)}"
+    limit = args.get("max_replaces")
+    if isinstance(limit, int) and not isinstance(limit, bool) and limit < 1:
+        return "error: max_replaces must be >= 1"
+    return None
+
 
 def _count_non_overlapping(text: str, needle: str) -> int:
     """Count left-to-right non-overlapping matches (same as ``str.replace``)."""
@@ -48,7 +67,7 @@ def _success_message(
     )
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, writes=("path",))
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, writes=("path",), validate=edit_replace_shape)
 async def edit_replace(path: str, old_string: str, new_string: str, max_replaces: int = 1) -> str:
     """Replace occurrences of ``old_string`` with ``new_string`` in a file.
 

src/plyngent/tools/net/fetch_local.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from plyngent.agent import ToolOutput, ToolTag, tool
 from plyngent.agent.images import image_part_from_bytes
 from plyngent.tools.context import is_read_only_context
@@ -17,8 +19,36 @@ from plyngent.tools.net.policy import (
     parse_fetch_url,
     soft_confirm_reason,
 )
+from plyngent.tools.shape import shape_of, timeout_shape_error
 from plyngent.tools.workspace import image_input_refusal
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+
+def fetch_local_shape(args: Mapping[str, object]) -> str | None:
+    """Answer a request that cannot be built, before anyone is asked about it.
+
+    The verb and the URL are checked here (both are the request itself, and
+    ``parse_fetch_url`` touches nothing but the string); a host that needs a
+    human grant still asks in the handler, once the call can actually run.
+    """
+    url = args.get("url")
+    if not isinstance(url, str):
+        return f"error: `fetch_local` expects `url` as a string, got {shape_of(url)}"
+    method = args.get("method", "GET")
+    if not isinstance(method, str):
+        return f"error: `fetch_local` expects `method` as a string, got {shape_of(method)}"
+    timeout_error = timeout_shape_error(args.get("timeout_seconds"))
+    if timeout_error is not None:
+        return f"error: `fetch_local` {timeout_error}"
+    try:
+        _ = normalize_method(method)
+        _ = parse_fetch_url(url)
+    except FetchPolicyError as exc:
+        return f"error: {exc}"
+    return None
+
 
 def format_fetch_result(
     *,
@@ -59,7 +89,10 @@ def format_fetch_result(
     return "\n".join(parts)
 
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO | ToolTag.TRUSTABLE | ToolTag.READ_ONLY)
+@tool(
+    tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO | ToolTag.TRUSTABLE | ToolTag.READ_ONLY,
+    validate=fetch_local_shape,
+)
 async def fetch_local(
     url: str,
     *,

src/plyngent/tools/process/command_exec.py

@@ -19,6 +19,8 @@ from plyngent.tools.workspace import (
 
 DEFAULT_TIMEOUT_SECONDS = 30
 DEFAULT_MAX_BATCH_STEPS = 20
+# A batch node's continuations nest, so the tree needs its own ceiling.
+MAX_BATCH_DEPTH = 4
 
 
 def format_command_result(
@@ -65,7 +67,7 @@ class CommandStepResult:
     stdout: str
     stderr: str
     mix_stderr: bool
-    captured: str  # piped to the next step when provider sets pipe_out
+    captured: str  # stdout, fed to the next command of a pipe
 
 
 async def execute_argv(

src/plyngent/tools/process/open_pty.py

@@ -1,14 +1,26 @@
 from __future__ import annotations
 
 import shlex
+from typing import TYPE_CHECKING
 
 from plyngent.agent import ToolTag, tool
-from plyngent.tools.workspace import WorkspaceError
+from plyngent.tools.workspace import WorkspaceError, argv_shape_error
 
 from .pty_session import active_pty_manager
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
+
+def open_pty_shape(args: Mapping[str, object]) -> str | None:
+    """Answer a session that cannot start, before anyone is asked about it."""
+    shape_error = argv_shape_error(args.get("command"))
+    if shape_error is not None:
+        return f"error: `open_pty` {shape_error}"
+    return None
+
+
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, validate=open_pty_shape)
 async def open_pty(command: list[str], *, cwd: str = ".") -> str:
     """Open a PTY session running ``command`` (argv) under the workspace.
 

src/plyngent/tools/process/run_argv.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from plyngent.agent import ToolTag, tool
+from plyngent.tools.shape import timeout_shape_error
 from plyngent.tools.workspace import WorkspaceError, argv_shape_error
 
 from .command_exec import (
@@ -9,8 +12,22 @@ from .command_exec import (
     format_command_result,
 )
 
+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

@@ -2,14 +2,20 @@ from __future__ import annotations
 
 import json
 import shlex
-from typing import Any, cast
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, cast
 
 from plyngent.agent import ToolTag, tool
+from plyngent.tools.shape import timeout_shape_error
 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,
 )
@@ -19,6 +25,14 @@ from .command_exec import (
 BATCH_OUTPUT_BUDGET_CHARS = 32_000
 BATCH_OUTPUT_SOFT_CAP_FACTOR = 4
 
+# A node is one command plus up to three continuations: ``then`` runs when it
+# succeeds, ``else`` when it fails, ``pipe`` takes its stdout as stdin. Only a
+# pipe hands data to its child, so a pipe stage may not branch: the stage's
+# result belongs to the node that owns the pipe.
+_CONTINUATIONS = ("then", "else", "pipe")
+_COMMAND_KEYS = frozenset({"argv", "cwd", "env", "stdin", "mix_stderr", "stop_on_error", "timeout_seconds"})
+_ACCEPTED_KEYS = "argv, cwd, env, stdin, mix_stderr, stop_on_error, timeout_seconds, then, else, pipe"
+
 
 def _as_bool(value: object, *, default: bool = False) -> bool:
     if value is None:
@@ -49,36 +63,93 @@ def _as_str_dict(value: object) -> dict[str, str] | None:
     return {str(key): str(val) for key, val in cast("dict[object, object]", value).items()}
 
 
-def _parse_step(raw: object, index: int) -> dict[str, Any]:
+@dataclass
+class _Budget:
+    """Parsed node count: every node costs one, however deeply it nests."""
+
+    nodes: int = 0
+
+    def take(self, where: str) -> None:
+        self.nodes += 1
+        if self.nodes > DEFAULT_MAX_BATCH_STEPS:
+            msg = f"more than {DEFAULT_MAX_BATCH_STEPS} nodes in one batch (at {where})"
+            raise WorkspaceError(msg)
+
+
+def _parse_nodes(
+    value: object,
+    *,
+    where: str,
+    depth: int,
+    budget: _Budget,
+    fed: bool,
+) -> list[dict[str, Any]]:
+    """A continuation: one node or a list of them, in the order they run."""
+    if value is None:
+        return []
+    items = cast("list[object]", value) if isinstance(value, list) else [value]
+    parsed: list[dict[str, Any]] = []
+    for index, item in enumerate(items):
+        at = f"{where}[{index}]" if isinstance(value, list) else where
+        parsed.append(_parse_node(item, where=at, depth=depth, budget=budget, fed=fed))
+    return parsed
+
+
+def _parse_node(raw: object, *, where: str, depth: int, budget: _Budget, fed: bool) -> dict[str, Any]:
     if not isinstance(raw, dict):
-        msg = f"steps[{index}] must be an object"
+        msg = f"{where} must be an object"
+        raise WorkspaceError(msg)
+    if depth > MAX_BATCH_DEPTH:
+        msg = f"{where}: nodes nest at most {MAX_BATCH_DEPTH} levels deep"
         raise WorkspaceError(msg)
     data = cast("dict[str, object]", raw)
+    unknown = sorted(set(data) - _COMMAND_KEYS - set(_CONTINUATIONS))
+    if unknown:
+        hint = '; write a pipeline as {"pipe": [...]}, not "pipe_out"' if "pipe_out" in unknown else ""
+        msg = f"{where} has unknown key(s) {', '.join(unknown)}; accepted keys are {_ACCEPTED_KEYS}{hint}"
+        raise WorkspaceError(msg)
+    budget.take(where)
     argv = data.get("argv")
     if not isinstance(argv, list) or not argv:
-        msg = f"steps[{index}].argv must be a non-empty argv list"
+        msg = f"{where}.argv must be a non-empty argv list"
         raise WorkspaceError(msg)
     argv_parts = cast("list[object]", argv)
     argv_str = [part for part in argv_parts if isinstance(part, str)]
     if len(argv_str) != len(argv_parts):
-        msg = f"steps[{index}].argv must be a list of strings"
+        msg = f"{where}.argv must be a list of strings"
+        raise WorkspaceError(msg)
+    stdin = data.get("stdin")
+    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")
-    stdin = data.get("stdin")
     stop = data.get("stop_on_error", None)
-    return {
+    node: dict[str, Any] = {
         "argv": argv_str,
         "cwd": str(cwd) if isinstance(cwd, str) and cwd else None,
         "env": _as_str_dict(data.get("env")),
         "stdin": None if stdin is None else str(stdin),
-        "pipe_out": _as_bool(data.get("pipe_out"), default=False),
         "mix_stderr": _as_bool(data.get("mix_stderr"), default=False),
         "stop_on_error": None if stop is None else _as_bool(stop, default=True),
         "timeout_seconds": _as_float(data.get("timeout_seconds"), default=DEFAULT_TIMEOUT_SECONDS),
     }
+    node["then"] = _parse_nodes(data.get("then"), where=f"{where}.then", depth=depth + 1, budget=budget, fed=False)
+    node["else"] = _parse_nodes(data.get("else"), where=f"{where}.else", depth=depth + 1, budget=budget, fed=False)
+    node["pipe"] = _parse_nodes(data.get("pipe"), where=f"{where}.pipe", depth=depth + 1, budget=budget, fed=True)
+    if data.get("pipe") is not None and not node["pipe"]:
+        msg = f"{where}.pipe must not be empty"
+        raise WorkspaceError(msg)
+    if fed and (node["then"] or node["else"]):
+        msg = f"{where}: a pipe step takes no then/else; branch on the node that owns the pipe"
+        raise WorkspaceError(msg)
+    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)
@@ -89,18 +160,66 @@ 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 _format_step(index: int, result: CommandStepResult, *, pipe_out: bool) -> str:
-    return "\n".join(
+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."""
+
+    result: CommandStepResult
+    depth: int
+    via: tuple[str, int] | None
+    feeds: bool
+
+
+@dataclass(frozen=True)
+class _Stage:
+    """Exit code of a node's command stage — its pipe chain, for a piped node."""
+
+    code: int | None
+    failed: bool
+    captured: str
+
+
+def _format_step(index: int, step: _Step) -> str:
+    result = step.result
+    lines = [f"--- step {index} ---"]
+    if step.depth > 0:
+        lines.append(f"depth={step.depth}")
+    if step.via is not None:
+        lines.append(f"via={step.via[0]} of step {step.via[1]}")
+    lines.extend(
         [
-            f"--- step {index} ---",
             f"exit_code={result.exit_code if result.exit_code is not None else ''}",
             f"timed_out={'true' if result.timed_out else 'false'}",
             f"cwd={result.cwd_display}",
             f"cmd={shlex.join(result.command)}",
-            f"pipe_out={'true' if pipe_out else 'false'}",
+            f"pipe={'true' if step.feeds else 'false'}",
             f"mix_stderr={'true' if result.mix_stderr else 'false'}",
             "--- stdout ---",
             result.stdout,
@@ -108,6 +227,7 @@ def _format_step(index: int, result: CommandStepResult, *, pipe_out: bool) -> st
             result.stderr,
         ]
     )
+    return "\n".join(lines)
 
 
 def _merge_env(base_env: dict[str, str] | None, step_env: dict[str, str] | None) -> dict[str, str] | None:
@@ -116,46 +236,118 @@ def _merge_env(base_env: dict[str, str] | None, step_env: dict[str, str] | None)
     return {**(base_env or {}), **(step_env or {})}
 
 
-async def _run_batch_steps(
-    steps: list[dict[str, Any]],
-    *,
-    cwd: str,
-    env: dict[str, str] | None,
-    stop_on_error: bool,
-) -> tuple[list[tuple[dict[str, Any], CommandStepResult]], bool]:
-    results: list[tuple[dict[str, Any], CommandStepResult]] = []
-    prev_capture: str | None = None
-    prev_piped = False
-    stopped_early = False
-    total_chars = 0
+@dataclass
+class _Runner:
+    """Runs the node tree, remembering each command and when to stop."""
 
-    for index, step in enumerate(steps):
-        stdin_text = prev_capture if prev_piped else step["stdin"]
+    cwd: str
+    env: dict[str, str] | None
+    stop_on_error: bool
+    steps: list[_Step] = field(default_factory=list)
+    chars: int = 0
+    stopped: bool = False
+
+    async def run_list(
+        self,
+        nodes: list[dict[str, Any]],
+        *,
+        depth: int,
+        branch: str | None = None,
+        parent: int | None = None,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+    ) -> None:
+        via = (branch, parent) if branch is not None and parent is not None else None
+        for node in nodes:
+            if self.stopped:
+                return
+            _ = await self.run_node(node, depth=depth, via=via, cwd=cwd, env=env)
+
+    async def run_node(
+        self,
+        node: dict[str, Any],
+        *,
+        depth: int,
+        via: tuple[str, int] | None,
+        stdin: str | None = None,
+        also_feeds: bool = False,
+        stage_only: bool = False,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+    ) -> _Stage:
+        # A continuation runs where its parent ran, unless it names its own
+        # directory or environment; the batch's cwd/env are the outermost default.
+        node_cwd = node["cwd"] if node["cwd"] is not None else (self.cwd if cwd is None else cwd)
+        node_env = _merge_env(self.env if env is None else env, node["env"])
         result = await execute_argv(
-            step["argv"],
-            cwd=step["cwd"] if step["cwd"] is not None else cwd,
-            env=_merge_env(env, step["env"]),
-            stdin=stdin_text,
-            timeout_seconds=step["timeout_seconds"],
-            mix_stderr=step["mix_stderr"],
+            node["argv"],
+            cwd=node_cwd,
+            env=node_env,
+            stdin=stdin if stdin is not None else node["stdin"],
+            timeout_seconds=node["timeout_seconds"],
+            mix_stderr=node["mix_stderr"],
         )
-        results.append((step, result))
-        prev_capture = result.captured
-        prev_piped = bool(step["pipe_out"])
-        total_chars += len(result.stdout) + len(result.stderr)
+        index = len(self.steps)
+        self.steps.append(_Step(result, depth, via, feeds=bool(node["pipe"]) or also_feeds))
+        self.chars += len(result.stdout) + len(result.stderr)
+        stage = _Stage(result.exit_code, bool(result.timed_out) or result.exit_code != 0, result.captured)
 
-        failed = bool(result.timed_out) or result.exit_code != 0
-        effective_stop = stop_on_error if step["stop_on_error"] is None else bool(step["stop_on_error"])
-        if failed and effective_stop:
-            return results, True
-        if total_chars > BATCH_OUTPUT_BUDGET_CHARS * BATCH_OUTPUT_SOFT_CAP_FACTOR:
-            return results, True
-        del index  # used for clarity in loop only
+        chain: list[dict[str, Any]] = node["pipe"]
+        piped = result.captured
+        for position, child in enumerate(chain):
+            if self.stopped:
+                break
+            stage = await self.run_node(
+                child,
+                depth=depth + 1,
+                via=("pipe", index),
+                stdin=piped,
+                also_feeds=position + 1 < len(chain),
+                stage_only=True,
+                cwd=node_cwd,
+                env=node_env,
+            )
+            piped = stage.captured
 
-    return results, stopped_early
+        if self.chars > BATCH_OUTPUT_BUDGET_CHARS * BATCH_OUTPUT_SOFT_CAP_FACTOR:
+            self.stopped = True
+            return stage
+        # A pipe stage's failure is the pipe owner's to answer: the chain still
+        # ran to its last command, so only that result decides the branch.
+        if stage_only:
+            return stage
+        if stage.failed:
+            if node["else"]:
+                await self.run_list(
+                    node["else"],
+                    depth=depth + 1,
+                    branch="else",
+                    parent=index,
+                    cwd=node_cwd,
+                    env=node_env,
+                )
+            elif self._effective_stop(node):
+                self.stopped = True
+            return stage
+        if node["then"]:
+            await self.run_list(
+                node["then"],
+                depth=depth + 1,
+                branch="then",
+                parent=index,
+                cwd=node_cwd,
+                env=node_env,
+            )
+        return stage
+
+    def _effective_stop(self, node: dict[str, Any]) -> bool:
+        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,
     *,
@@ -163,53 +355,60 @@ async def run_argv_batch(
     env: dict[str, str] | None = None,
     stop_on_error: bool = True,
 ) -> str:
-    """Run a serial batch of argv commands (no shell).
+    """Run a batch of argv commands (no shell), each with optional continuations.
 
-    Each element of ``steps`` is an object::
+    A node is an object::
 
         {
-          "argv": ["git", "status"],     # required argv
-          "cwd": ".",                     # optional, else batch cwd
-          "env": {"FOO": "1"},            # optional overlay
-          "stdin": "...",                 # optional; unused if previous pipe_out
-          "pipe_out": false,              # if true, feed capture into *next* stdin
-          "mix_stderr": false,            # OS-level merge stderr into capture
-          "stop_on_error": null,          # override batch stop_on_error
+          "argv": ["pytest", "-q"],     # required argv
+          "then": {"argv": [...]},      # runs when argv succeeds (node or list)
+          "else": {"argv": [...]},      # runs when argv fails (node or list)
+          "pipe": {"argv": [...]},      # takes argv's stdout as its stdin
+          "cwd": ".",                   # optional, else batch cwd
+          "env": {"FOO": "1"},          # optional overlay
+          "stdin": "...",               # optional; a pipe step takes it from the pipe
+          "mix_stderr": false,          # OS-level merge stderr into capture
+          "stop_on_error": null,        # override batch stop_on_error
           "timeout_seconds": 30
         }
 
-    ``pipe_out`` is set on the **provider** step. Last-step ``pipe_out`` is an error.
-    Default ``stop_on_error`` is true. Max 20 steps; the batch stops early once
-    accumulated output passes the run budget (128k chars). Each step's output is
-    kept whole for the model; a long result carries a ``document_id`` to read on.
+    ``steps`` is a list of nodes, and ``then`` / ``else`` / ``pipe`` each take one
+    node or a list of them, so the tree nests. A continuation runs where its
+    parent ran: it inherits the parent's ``cwd`` and ``env`` (its own win, and
+    the batch's are the outermost default). A ``pipe`` chain runs to its last
+    command whatever the earlier ones did, and only that last exit code decides
+    the branch; a pipe step takes no ``then`` / ``else`` — branch on the node
+    that owns the pipe. A failure with no ``else`` stops the batch, unless that
+    node sets ``stop_on_error: false``: it then skips its own continuations and
+    the next sibling runs. A failure inside a continuation is nobody's to
+    answer, so it stops the batch the same way.
+
+    Max 20 nodes and 4 nesting levels; the batch stops early once accumulated
+    output passes the run budget (128k chars). Each command's output is kept
+    whole for the model (a long result carries a ``document_id`` to read on), in
+    a ``--- step N ---`` block that also says how it got there: ``depth=`` when
+    nested, ``via=then of step N`` (or ``else`` / ``pipe``) and ``pipe=true``
+    when its stdout fed another command.
     """
     try:
-        raw_steps = _normalize_steps(steps)
-        if not raw_steps:
-            return "error: steps must not be empty"
-        if len(raw_steps) > DEFAULT_MAX_BATCH_STEPS:
-            return f"error: at most {DEFAULT_MAX_BATCH_STEPS} steps per batch"
-
-        steps = [_parse_step(item, i) for i, item in enumerate(raw_steps)]
-        if steps[-1]["pipe_out"]:
-            return "error: last step has pipe_out=true but there is no next step"
-
-        results, stopped_early = await _run_batch_steps(steps, cwd=cwd, env=env, stop_on_error=stop_on_error)
-        body = "\n".join(
-            _format_step(i, result, pipe_out=bool(step["pipe_out"])) for i, (step, result) in enumerate(results)
-        )
-        last_exit = results[-1][1].exit_code if results else ""
-        return "\n".join(
-            [
-                f"steps={len(steps)} ran={len(results)} "
-                f"stop_on_error={'true' if stop_on_error else 'false'} "
-                f"stopped_early={'true' if stopped_early 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 ''}",
+        ]
+    )

src/plyngent/tools/process/write_pty_keys.py

@@ -1,13 +1,31 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from plyngent.agent import ToolTag, tool
+from plyngent.tools.shape import shape_of
 from plyngent.tools.workspace import WorkspaceError
 
 from .pty_terminal import KeySyntaxError, decode_write_data
 from .write_pty import write_pty_payload
 
+if TYPE_CHECKING:
+    from collections.abc import Mapping
 
-@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
+
+def write_pty_keys_shape(args: Mapping[str, object]) -> str | None:
+    """Answer a bad marker list, before anyone is asked about the call."""
+    data = args.get("data")
+    if not isinstance(data, str):
+        return f"error: `write_pty_keys` expects `data` as a string, got {shape_of(data)}"
+    try:
+        _ = decode_write_data(data)
+    except KeySyntaxError as exc:
+        return f"error: {exc}"
+    return None
+
+
+@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO, validate=write_pty_keys_shape)
 async def write_pty_keys(session_id: int, data: str) -> str:
     """Write to a PTY after expanding ``<key>`` markers (never for normal typing).
 

src/plyngent/tools/shape.py

@@ -0,0 +1,69 @@
+"""Pre-prompt argument checks shared by tools.
+
+Tool arguments arrive as decoded JSON, so a parameter declared ``float`` or
+``list[str]`` can still hold anything at runtime. A value the tool can judge
+without running a thing is answered *before* the call reaches a confirm prompt
+(a tool declares the check with ``@tool(validate=…)``, see
+:class:`plyngent.agent.tools.ToolDefinition`): the human is never asked about a
+call that could only fail, and a value that would only break halfway through
+(e.g. a deadline that fails inside ``wait_for``, after the process was spawned)
+never gets that far.
+
+Every check takes ``object``, so the isinstance calls stay meaningful and the
+tools themselves stay free of unnecessary-isinstance warnings.
+"""
+
+from __future__ import annotations
+
+_SHAPES: dict[type, str] = {
+    str: "a string",
+    list: "an array",
+    dict: "an object",
+    bool: "a boolean",
+    int: "a number",
+    float: "a number",
+    type(None): "null",
+}
+
+
+def shape_of(value: object) -> str:
+    """Model-facing name for a decoded JSON value (``a string``, ``an array``…)."""
+    return _SHAPES.get(type(value), type(value).__name__)
+
+
+def got(value: object) -> str:
+    """What a rejected value is, quoting a string (``'soon'``) instead of naming it."""
+    return repr(value) if isinstance(value, str) else shape_of(value)
+
+
+def number_value(value: object) -> float | None:
+    """The number *value* carries, or ``None`` when it is not one.
+
+    Numeric strings are accepted (models send them); a boolean is not a number
+    here, and neither is non-numeric text.
+    """
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int | float):
+        return float(value)
+    if isinstance(value, str):
+        try:
+            return float(value.strip())
+        except ValueError:
+            return None
+    return None
+
+
+def timeout_shape_error(value: object, *, name: str = "timeout_seconds") -> str | None:
+    """Error when *name* is not a positive number of seconds; ``None`` when it is.
+
+    ``None`` means the argument was omitted (the tool's default applies).
+    """
+    if value is None:
+        return None
+    seconds = number_value(value)
+    if seconds is None:
+        return f"{name} must be a number, not {got(value)}"
+    if seconds <= 0:
+        return f"{name} must be greater than 0 (got {value!r})"
+    return None

tests/test_agent/test_tool_shape.py

@@ -0,0 +1,82 @@
+"""A tool's own shape check answers a call before anyone is asked about it."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+
+from plyngent.agent import ToolRegistry
+from plyngent.tools.file.edit_lineno import edit_lineno
+from plyngent.tools.file.edit_replace import edit_replace
+from plyngent.tools.net.fetch_local import fetch_local
+from plyngent.tools.process.open_pty import open_pty
+from plyngent.tools.process.write_pty_keys import write_pty_keys
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+    from plyngent.agent import ToolDefinition
+
+
+def _everything_needs_a_confirm(name: str, _args: Mapping[str, object]) -> str:
+    """A danger classifier that prompts for every call, so a suppressed prompt is visible."""
+    return f"run {name}"
+
+
+def _registry(definition: ToolDefinition) -> tuple[ToolRegistry, list[str]]:
+    asked: list[str] = []
+
+    def on_confirm(name: str, _args: Mapping[str, object], _reason: str) -> bool:
+        asked.append(name)
+        return True
+
+    return ToolRegistry([definition], danger=_everything_needs_a_confirm, on_confirm=on_confirm), asked
+
+
+@pytest.mark.parametrize(
+    ("definition", "args", "expected"),
+    [
+        # A string where the tool needs an argv array would splat into characters.
+        (open_pty, {"command": "ls -la"}, "only accepts an array of args"),
+        # A typo in a <key> marker must not land in a shell prompt.
+        (write_pty_keys, {"session_id": 1, "data": "<nope>"}, "unknown key <nope>"),
+        (edit_replace, {"path": "a.py", "old_string": "", "new_string": "x"}, "old_string must not be empty"),
+        (
+            edit_replace,
+            {"path": "a.py", "old_string": "x", "new_string": "y", "max_replaces": 0},
+            "max_replaces must be >= 1",
+        ),
+        (
+            edit_lineno,
+            {"path": "a.py", "start_line": 0, "end_line": 1, "new_content": "x"},
+            "start_line must be >= 1",
+        ),
+        (
+            edit_lineno,
+            {"path": "a.py", "start_line": 3, "end_line": 2, "new_content": "x"},
+            "end_line must be >= start_line",
+        ),
+        (fetch_local, {"url": "https://example.com", "method": "PATCH"}, "method must be one of"),
+        (fetch_local, {"url": "ftp://example.com"}, "url scheme must be http or https"),
+        (fetch_local, {"url": "https://example.com", "timeout_seconds": "soon"}, "timeout_seconds must be a number"),
+    ],
+)
+async def test_a_bad_argument_is_answered_before_the_prompt(
+    definition: ToolDefinition,
+    args: dict[str, object],
+    expected: str,
+) -> None:
+    registry, asked = _registry(definition)
+    out = await registry.execute(definition.name, json.dumps(args))
+    assert asked == []
+    assert expected in out
+
+
+async def test_a_call_that_can_run_still_asks() -> None:
+    """The check answers only what it can judge; a runnable call keeps its prompt."""
+    registry, asked = _registry(write_pty_keys)
+    out = await registry.execute("write_pty_keys", json.dumps({"session_id": 1, "data": "<enter>"}))
+    assert asked == ["write_pty_keys"]
+    assert out.startswith(("[approval]", "error:"))

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_cli/test_compact_cmd.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from typing import TYPE_CHECKING, Literal, overload
 
 import pytest
@@ -57,6 +58,31 @@ class SummaryClient:
         )
 
 
+class HangingClient:
+    """A summarizer that never answers until the request is cancelled."""
+
+    def __init__(self) -> None:
+        self.entered = asyncio.Event()
+
+    @overload
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: Literal[False] = False
+    ) -> ChatCompletionResponse: ...
+
+    @overload
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: Literal[True]
+    ) -> AsyncIterator[ChatCompletionChunk]: ...
+
+    async def chat_completions(
+        self, param: ChatCompletionsParam, *, stream: bool = False
+    ) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]:
+        del param, stream
+        self.entered.set()
+        await asyncio.Event().wait()  # until the caller cancels
+        raise AssertionError("the summary request should have been cancelled")
+
+
 async def test_compact_to_new_session(tmp_path: Path) -> None:
     memory = await MemoryStore.open(DatabaseConfig())
     try:
@@ -128,6 +154,50 @@ async def test_compact_to_new_session(tmp_path: Path) -> None:
         await memory.close()
 
 
+async def test_compact_cancel_leaves_the_session_untouched(tmp_path: Path) -> None:
+    """Ctrl+C during the summary drops the request before anything is created."""
+    memory = await MemoryStore.open(DatabaseConfig())
+    try:
+        provider = OpenAIProvider(access_key_or_token="sk-test")
+        config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
+        config.providers = {"local": provider}
+        state = ReplState(
+            config=config,
+            memory=memory,
+            workspace=tmp_path,
+            provider_name="local",
+            provider=provider,
+            model="gpt-test",
+            tools_enabled=False,
+        )
+        await state.new_session("orig")
+        old_id = state.session_id
+        assert old_id is not None
+        client = HangingClient()
+        state.client = client
+        state.agent = ChatAgent(
+            state.client,
+            model=state.model,
+            memory=state.memory,
+            session_id=old_id,
+            system_prompt="Be brief.",
+        )
+        state.agent.messages = [UserChatMessage(content="do work")]
+        before = {row.sid for row in await memory.list_sessions(workspace=tmp_path)}
+
+        task = asyncio.ensure_future(state.compact_to_new_session())
+        await asyncio.wait_for(client.entered.wait(), timeout=5)
+        _ = task.cancel()
+        with pytest.raises(ValueError, match="compaction cancelled"):
+            await task
+
+        assert state.session_id == old_id
+        after = {row.sid for row in await memory.list_sessions(workspace=tmp_path)}
+        assert after == before
+    finally:
+        await memory.close()
+
+
 async def test_rebuild_client_preserves_persist_cursor(tmp_path: Path) -> None:
     memory = await MemoryStore.open(DatabaseConfig())
     try:

tests/test_cli/test_display.py

@@ -456,9 +456,9 @@ async def test_pretty_run_argv_batch_stopped(capsys: pytest.CaptureFixture[str])
     result = (
         "steps=3 ran=2 stop_on_error=true stopped_early=true\n"
         "--- step 1 ---\nexit_code=1\ntimed_out=false\ncwd=.\ncmd=git status\n"
-        "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
+        "pipe=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
         "--- step 2 ---\nexit_code=0\ntimed_out=false\ncwd=.\ncmd=git diff\n"
-        "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
+        "pipe=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
         "--- summary ---\nlast_exit=1\n"
     )
     await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)]))
@@ -495,16 +495,27 @@ async def test_pretty_run_argv_batch_lists_commands(capsys: pytest.CaptureFixtur
 
 
 async def test_pretty_run_argv_batch_joins_pipe_chain(capsys: pytest.CaptureFixture[str]) -> None:
-    """``pipe_out`` steps share one line; a step they do not feed keeps its own."""
+    """A pipe chain shares one line; a node it does not feed keeps its own."""
     args = (
-        '{"steps": [{"argv": ["grep", "-n", "foo", "a.py"], "pipe_out": true},'
-        ' {"argv": ["wc", "-l"]}, {"argv": ["git", "status"]}]}'
+        '{"steps": [{"argv": ["grep", "-n", "foo", "a.py"], "pipe": {"argv": ["wc", "-l"]}},'
+        ' {"argv": ["git", "status"]}]}'
     )
     await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
     out = capsys.readouterr().out
     assert "$ grep -n foo a.py | wc -l\n$ git status\n* Run batch (done)" in out
 
 
+async def test_pretty_run_argv_batch_indents_branches(capsys: pytest.CaptureFixture[str]) -> None:
+    """A ``then`` / ``else`` body sits below the node it follows."""
+    args = (
+        '{"steps": [{"argv": ["pytest", "-q"], "then": {"argv": ["ruff", "format", "."]},'
+        ' "else": [{"argv": ["git", "status"]}]}]}'
+    )
+    await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
+    out = capsys.readouterr().out
+    assert "$ pytest -q\n  then ruff format .\n  else git status\n* Run batch (done)" in out
+
+
 async def test_pretty_run_argv_batch_elides_a_long_command(capsys: pytest.CaptureFixture[str]) -> None:
     long_arg = "x" * 200
     args = f'{{"steps": [{{"argv": ["echo", "{long_arg}"]}}]}}'

tests/test_cli/test_interrupt.py

@@ -326,3 +326,33 @@ async def test_sleep_cancellable_cancelled_by_sigint(router: None) -> None:
     await asyncio.sleep(0)
     os.kill(os.getpid(), signal.SIGINT)
     assert await asyncio.wait_for(waiter, timeout=5) is False
+
+
+def test_a_second_loop_rebinds_the_router() -> None:
+    """The installed flag is process-wide, so a new loop needs its own handler.
+
+    asyncio drops a loop's signal handlers with the loop, so believing the flag
+    alone left the next loop unrouted: a stray Ctrl+C then became a
+    KeyboardInterrupt out of the loop's own frame instead of cancelling work.
+    """
+
+    class Counter:
+        hits = 0
+
+        def cancel(self) -> None:
+            self.hits += 1
+
+    async def routed_once() -> int:
+        assert install_sigint_router()
+        target = Counter()
+        with sigint_cancels(target):
+            os.kill(os.getpid(), signal.SIGINT)
+            await asyncio.sleep(0.05)
+        return target.hits
+
+    install_keyboard_interrupt_sigint()
+    try:
+        assert asyncio.run(routed_once()) == 1
+        assert asyncio.run(routed_once()) == 1
+    finally:
+        uninstall_sigint_router()

tests/test_cli/test_repl_commands.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
+import asyncio
 import base64
-from typing import TYPE_CHECKING, Literal, overload
+from typing import TYPE_CHECKING, Literal, cast, overload
 
 import pytest
 import tomlkit
@@ -28,7 +29,7 @@ from plyngent.tools import image_input_refusal
 from plyngent.tools.context import bind_instance
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator
+    from collections.abc import AsyncIterator, Sequence
     from pathlib import Path
 
 
@@ -110,6 +111,21 @@ async def test_help(state: ReplState) -> None:
     assert await handle_slash(state, "/help") is True
 
 
+async def test_a_cancelled_slash_command_keeps_the_repl(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A command's own work being cancelled must not cancel the REPL's task."""
+    from plyngent.cli import slash as slash_module
+
+    def cancelled(_args: Sequence[str], _state: object) -> None:
+        raise asyncio.CancelledError
+
+    monkeypatch.setattr(slash_module, "_run_slash_argv", cancelled)
+    assert await slash_module.handle_slash(cast("ReplState", cast("object", None)), "/btw hello") is True
+    assert "cancelled" in capsys.readouterr().out
+
+
 async def test_auto_continue_limits_installs_on_limit(tmp_path: Path) -> None:
     from plyngent.cli.limits import auto_continue_enabled, reset_auto_continue
 

tests/test_tools/test_danger.py

@@ -108,6 +108,34 @@ def test_classify_run_argv_batch_risky() -> None:
     )
 
 
+def test_classify_run_argv_batch_sees_nested_steps() -> None:
+    """A command hiding in a continuation is reviewed like a top-level step."""
+    reason = _danger(
+        "run_argv_batch",
+        {
+            "steps": [
+                {
+                    "argv": ["echo", "ok"],
+                    "then": {"argv": ["bash", "-c", "echo risky"]},
+                    "else": [{"argv": ["sh", "-c", "echo risky"]}],
+                }
+            ]
+        },
+    )
+    assert "run_argv_batch[0].then[0]" in reason.text
+    assert "run_argv_batch[0].else[0]" in reason.text
+    # A plain interpreter run is YOLO mode's to drop, wherever it sits.
+    assert reason.yolo_skip is True
+
+
+def test_a_nested_step_that_escalates_keeps_the_batch_prompt() -> None:
+    reason = _danger(
+        "run_argv_batch",
+        {"steps": [{"argv": ["echo", "ok"], "then": {"argv": ["sudo", "python", "-c", "print(1)"]}}]},
+    )
+    assert reason.yolo_skip is False
+
+
 def test_classify_shell_and_dash_c() -> None:
     r = _reason("run_argv", {"argv": ["bash", "-c", "rm -rf /"]})
     assert r is not None

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():
@@ -112,33 +119,31 @@ async def test_run_argv_batch_stop_on_error(workspace: object) -> None:
     assert "should-not-run" not in out
 
 
-async def test_run_argv_batch_pipe_out(workspace: object) -> None:
+async def test_run_argv_batch_pipe(workspace: object) -> None:
     del workspace
     out = await call_async(
         run_argv_batch,
         [
             {
                 "argv": _py("print('piped-payload', end='')"),
-                "pipe_out": True,
-            },
-            {
-                "argv": _py("import sys; print(sys.stdin.read())"),
+                "pipe": {"argv": _py("import sys; print(sys.stdin.read())")},
             },
         ],
     )
     assert "ran=2" in out
     assert "piped-payload" in out
-    assert "pipe_out=true" in out
+    assert "pipe=true" in out
 
 
-async def test_run_argv_batch_last_pipe_out_error(workspace: object) -> None:
+async def test_run_argv_batch_pipe_out_is_gone(workspace: object) -> None:
+    """The old provider-side flag is refused, pointing at the pipe node."""
     del workspace
     out = await call_async(
         run_argv_batch,
         [{"argv": _py("print(1)"), "pipe_out": True}],
     )
     assert "error:" in out
-    assert "pipe_out" in out
+    assert '{"pipe": [...]}' in out
 
 
 async def test_run_argv_batch_mix_stderr(workspace: object) -> None:
@@ -149,9 +154,8 @@ async def test_run_argv_batch_mix_stderr(workspace: object) -> None:
             {
                 "argv": _py("import sys; sys.stdout.write('out'); sys.stderr.write('err')"),
                 "mix_stderr": True,
-                "pipe_out": True,
+                "pipe": {"argv": _py("import sys; print(repr(sys.stdin.read()))")},
             },
-            {"argv": _py("import sys; print(repr(sys.stdin.read()))")},
         ],
     )
     assert "ran=2" in out
@@ -159,6 +163,238 @@ async def test_run_argv_batch_mix_stderr(workspace: object) -> None:
     assert "out" in out and "err" in out
 
 
+async def test_run_argv_batch_then_runs_on_success(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [{"argv": _py("print('first')"), "then": {"argv": _py("print('then-body')")}}],
+    )
+    assert "then-body" in out
+    assert "ran=2" in out
+    assert "depth=1" in out
+    assert "via=then of step 0" in out
+
+
+async def test_run_argv_batch_else_handles_the_failure(workspace: object) -> None:
+    """A failure with an ``else`` is answered, so the batch keeps going."""
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [
+            {
+                "argv": _py("import sys; sys.exit(2)"),
+                "else": {"argv": _py("print('else-body')")},
+            },
+            {"argv": _py("print('sibling-ran')")},
+        ],
+    )
+    assert "else-body" in out
+    assert "sibling-ran" in out
+    assert "via=else of step 0" in out
+    assert "stopped_early=false" in out
+
+
+async def test_run_argv_batch_runs_a_then_list_in_order(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [
+            {
+                "argv": _py("print('first')"),
+                "then": [{"argv": _py("print('second')")}, {"argv": _py("print('third')")}],
+            }
+        ],
+    )
+    assert "ran=3" in out
+    assert out.index("first") < out.index("second") < out.index("third")
+
+
+async def test_run_argv_batch_a_failing_branch_stops_the_batch(workspace: object) -> None:
+    """Nobody answered a failure inside a continuation, so the batch stops."""
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [
+            {"argv": _py("print('first')"), "then": {"argv": _py("import sys; sys.exit(4)")}},
+            {"argv": _py("print('sibling-ran')")},
+        ],
+    )
+    assert "stopped_early=true" in out
+    assert "sibling-ran" not in out
+
+
+async def test_run_argv_batch_stop_on_error_false_skips_only_its_continuations(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [
+            {
+                "argv": _py("import sys; sys.exit(2)"),
+                "stop_on_error": False,
+                "then": {"argv": _py("print('then-ran')")},
+            },
+            {"argv": _py("print('sibling-ran')")},
+        ],
+    )
+    assert "then-ran" not in out
+    assert "sibling-ran" in out
+    assert "stopped_early=false" in out
+
+
+async def test_run_argv_batch_branches_on_the_pipe_result(workspace: object) -> None:
+    """A piped node branches on the last command of its chain."""
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [
+            {
+                "argv": _py("print('payload', end='')"),
+                "pipe": {"argv": _py("import sys; sys.stdin.read(); sys.exit(3)")},
+                "then": {"argv": _py("print('pipe-ok')")},
+                "else": {"argv": _py("print('pipe-failed')")},
+            }
+        ],
+    )
+    assert "pipe-failed" in out
+    assert "pipe-ok" not in out
+    assert "ran=3" in out
+
+
+async def test_run_argv_batch_pipe_step_takes_no_branch(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [{"argv": _py("print(1)"), "pipe": {"argv": _py("print(2)"), "then": {"argv": _py("print(3)")}}}],
+    )
+    assert "error:" in out
+    assert "pipe step takes no then/else" in out
+
+
+async def test_run_argv_batch_pipe_step_takes_no_stdin(workspace: object) -> None:
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [{"argv": _py("print(1)"), "pipe": {"argv": _py("print(2)"), "stdin": "own"}}],
+    )
+    assert "error:" in out
+    assert "stdin comes from the pipe" in out
+
+
+async def test_run_argv_batch_rejects_a_deep_tree(workspace: object) -> None:
+    del workspace
+    node: dict[str, object] = {"argv": _py("print(1)")}
+    for _ in range(6):
+        node = {"argv": _py("print(1)"), "then": node}
+    out = await call_async(run_argv_batch, [node])
+    assert "error:" in out
+    assert "nest at most 4 levels deep" in out
+
+
+async def test_run_argv_batch_counts_nested_nodes(workspace: object) -> None:
+    """Every node of the tree costs one, so a nested tree hits the same cap."""
+    del workspace
+    out = await call_async(run_argv_batch, [{"argv": _py("print(1)")} for _ in range(21)])
+    assert "error:" in out
+    assert "more than 20 nodes" in out
+
+
+async def test_run_argv_batch_names_the_node_grammar(workspace: object) -> None:
+    del workspace
+    out = await call_async(run_argv_batch, [{"argv": _py("print(1)"), "when": "always"}])
+    assert "error:" in out
+    assert "unknown key(s) when" in out
+    assert "then, else, pipe" in out
+
+
+async def test_run_argv_batch_a_continuation_runs_where_its_parent_ran(workspace: Path) -> None:
+    """``then`` inherits the parent's ``cwd`` and ``env`` unless it names its own."""
+    (workspace / "sub").mkdir()
+    (workspace / "own").mkdir()
+    out = await call_async(
+        run_argv_batch,
+        [
+            {
+                "argv": _py("import os; print('parent', os.path.basename(os.getcwd()), os.environ['FOO'])"),
+                "cwd": "sub",
+                "env": {"FOO": "inherited"},
+                "then": [
+                    {"argv": _py("import os; print('then', os.path.basename(os.getcwd()), os.environ['FOO'])")},
+                    {
+                        "argv": _py("import os; print('own', os.path.basename(os.getcwd()), os.environ['FOO'])"),
+                        "cwd": "own",
+                        "env": {"FOO": "own"},
+                    },
+                ],
+            }
+        ],
+    )
+    assert "parent sub inherited" in out
+    assert "then sub inherited" in out
+    assert "own own own" in out
+
+
+async def test_run_argv_batch_a_pipe_runs_to_its_end(workspace: object) -> None:
+    """A failing pipe stage does not skip the rest; the last command decides."""
+    del workspace
+    out = await call_async(
+        run_argv_batch,
+        [
+            {
+                "argv": _py("import sys; sys.exit(3)"),
+                "pipe": {"argv": _py("import sys; print('second-stage-ran'); sys.exit(0)")},
+                "else": {"argv": _py("print('branch-on-pipe-end')")},
+            },
+            {"argv": _py("print('batch-continues')")},
+        ],
+    )
+    assert "second-stage-ran" in out
+    assert "branch-on-pipe-end" not in out  # the chain's last exit code was 0
+    assert "batch-continues" 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", "."])