release: plyngent 0.12.0

docs: say how wrapper options are read
The destructive-confirm bullet described the old hand tables. It now names the getopt spec per wrapper, and the rule that follows from it: an option the spec does not know stops the unwrap and is reviewed, while the wrapper itself stays the command the denylist judges.
2026-11-07 07:17:54 +08:00 · 2026-10-23 10:53:08 +08:00 · 2026-10-23 10:47:55 +08:00 · 2026-10-23 10:47:45 +08:00 · 2026-10-23 10:47:36 +08:00 · 2026-10-23 10:43:05 +08:00
26 changed files with 961 additions and 209 deletions
@@ -83,7 +83,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `
 - **Kind-based dispatch**: clients expose `kind` (`chat_completions` | `responses` | `messages`). Agent history stays chat-completions-shaped; transport adapters return synthetic `ChatCompletionResponse` / stream chunks.
 - **`responses_bridge` / `responses_dispatch`**: chat ↔ OpenAI Responses; stream text/reasoning deltas (`response.reasoning_text.done` repeats what its fragments spelled out, so the loop streams each part once and still keeps the whole text in the round's message); tool calls + usage on `response.completed`; a request carrying `tools` returns every round's reasoning as a `reasoning` item of `reasoning_text` parts — "必须完整回传 `reasoning_content` … 即使该轮模型未实际进行工具调用" (a missing item and an empty part are both a 400, so a round that kept none sends a blank line) — and a round's `function_call` items stay contiguous ahead of their outputs (`call, output, call, output` was a 400: the second call reads as a round of its own without reasoning); reasoning is read back from the item's `content` parts.
 - **`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.
+- **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.
 - **`run_chat_loop`**: multi-round tool loop; default **streaming** text deltas + stream tool-call merge; parallel tools; tool-result char budget; 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`.
@@ -103,7 +103,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 - **`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` (escapes) / `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 (`\xHH`, `ctrl+x`, `key=esc`); 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`; deny may include a user comment for the model. 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; `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.
+- 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; `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.
 - **`vcs`**: read-only VCS tools (`vcs_kind` / `vcs_status` / `vcs_diff` / `vcs_log` / `vcs_branch`) via `VcsBackend` protocol; **git** implemented; detectors are pluggable for other systems.
 - **`chat`**: human prompts as tools — `ask_user_line` / `ask_user_choice` / `ask_user_form` (shared `prompting` core); `wait` (line prompt with timeout; Enter disturbs). Container args are shape-checked (`tools/chat/shape.py`): a mis-typed string `options`/`fields` (or a non-string `question`) returns an explained error — `options="a, b"` used to iterate characters into a per-character menu. `wait`'s `duration` accepts numeric strings (`"5"` = 5s) and rejects booleans/text/containers.
 - **`todo`**: LIFO stack of **task groups** (not a queue of tasks) — `todo_push` creates one group of siblings; `todo_pop` removes the whole top group; `todo_update` by the number a listing shows (render order, TOP group first; the stored `id` stays internal); DFS breakdown push[T1,T2]→push[T1.1…]→pop→push[T2.1…]; stored on session row; open items = unfinished work (`[TODO OPEN WORK]`); all-terminal non-empty = hygiene (`[TODO HYGIENE]`); nag channel via `[agent] todo_nag_strategy` (`developer`/`user` = prose nag; `synthetic_tool` = forged `todo_list` call + **real** `stack.render()` result + ToolCall/Result events; `none`).
@@ -117,7 +117,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.

 Shared interactive I/O: `ask` / `choose` / `form` / `confirm` with pluggable backend; non-TTY uses defaults or errors. CLI limit/confirm hooks and chat tools both use this. Async helpers serialize prompts (`run_prompt_async`).

-Three process-wide hooks let a host take over a prompt: `set_prompt_output_hook` (break an open pretty line before the prompt draws), `configure_prompting(pause_factory=…)` (the SIGINT guard around an off-loop read), and `set_prompt_wait_hook` — a prompt that blocks announces itself through `prompt_wait(kind, title, detail)`, which yields the `PromptAnswerSource` the read may consult (`take()` returns `allow` / `deny` / reply text without blocking) or `None` for notify-only. Only the outermost prompt of a nested read is announced; with no hook installed nothing changes. `read_line_with_timeout` accepts a `source` and a `None` deadline, and stdin always wins over a host answer. The CLI's implementation is `cli/notify.py` (`[notify]`, desktop via the optional `desktop-notifier` extra, terminal via OSC 777/9 to `/dev/tty`): a prompt is announced only after `grace_seconds`, the answer is treated like a typed one (never a session grant), and every failure falls through to the next channel instead of raising.
+Three process-wide hooks let a host take over a prompt: `set_prompt_output_hook` (break an open pretty line before the prompt draws), `configure_prompting(pause_factory=…)` (the SIGINT guard around an off-loop read), and `set_prompt_wait_hook` — a prompt that blocks announces itself through `prompt_wait(kind, title, detail)`, which yields the `PromptAnswerSource` the read may consult (`take()` returns `allow` / `deny` / reply text without blocking) or `None` for notify-only. Only the outermost prompt of a nested read is announced; with no hook installed nothing changes. A source that *will* offer an answer says so through `supports_actions` before it offers it, so the read polls from its first moment instead of deciding once. `read_line_with_timeout` accepts a `source` and a `None` deadline, and stdin always wins over a host answer. The CLI's implementation is `cli/notify.py` (`[notify]`, desktop via the optional `desktop-notifier` extra, terminal via OSC 777/9 to `/dev/tty`): a prompt is announced only after `grace_seconds`, the answer is treated like a typed one (never a session grant), and every failure falls through to the next channel instead of raising.

 ### CLI (`cli/`)

@@ -53,6 +53,40 @@ Images belong to **user** messages only: a picture in a system or assistant
 message is a 400 (`Image in assistant message is unsupported`). A tool result
 lives inside a user turn, which is why it can carry one.

+## Files API (upload once, reference by id)
+
+Every surface we speak to can take a picture as an uploaded file instead of
+inline bytes. The upload half of that is de-facto one shape; the object it
+returns and the way a message refers to it are not standardized at all. Field
+names below come from each vendor's own reference (DeepSeek `guides/files_api`,
+OpenAI `resources/files` + `guides/file-inputs`, Anthropic `api/beta/files`).
+
+| | OpenAI | Anthropic | DeepSeek |
+|---|---|---|---|
+| Upload | `POST /v1/files`, multipart: `file` + `purpose` (`user_data`, `vision`, …), optional `expires_after[anchor]` / `[seconds]` | `POST /v1/files`, multipart: `file` only; needs `anthropic-version` and `anthropic-beta: files-api-2025-04-14` | `POST /files`, OpenAI shape (`file` + `purpose="user_data"`, optional `expires_after[…]`); the same operations are also served under `/anthropic/v1/files` in the Anthropic shape |
+| Id | `file-abc123` | `file_01HX…` | `file-api-…` |
+| File object | `object:"file"`, `bytes`, `created_at` (unix seconds), `filename`, `purpose`, `expires_at` when it was asked for | `type:"file"`, `size_bytes`, `created_at` (RFC 3339), `filename`, `mime_type`, `downloadable` | both dialects: OpenAI shape on `/files`, Anthropic shape on `/anthropic/v1/files` |
+| Reference in a message | Responses `{"type":"input_image","file_id":…}` (mutually exclusive with `image_url`); chat completions cannot reference an image by id at all — its `image_url` part has no `file_id`, and its `file` part is documents only | `{"type":"image","source":{"type":"file","file_id":…}}`, plus the beta header | chat: `{"type":"file","file_id":…}`, or inline `file_data` + `filename`; Anthropic: `source.type="file"` plus beta; Responses: `input_image.file_id` |
+| Limits | 512 MB per file (what `purpose` allows) | not verified here (the API reference is unreachable from this host's region) | 64 MiB per file, 25 GiB and 10 000 files per account, expiry 1 h – 30 d or permanent |
+
+What is shared: `POST …/files` as `multipart/form-data` with a `file` part,
+and a string id back. What is not: the field names (`bytes` vs `size_bytes`),
+the type tag (`object` vs `type`), the timestamp format, the cursor and paging
+fields, the beta header, and every reference block above. A client therefore
+needs one uploader and three reference shapes — the split this module already
+has.
+
+Three rules if this is ever wired in:
+
+- a `file_id` is scoped to the provider (and account) that minted it, so a
+  stored id belongs to the surface that produced it, exactly like
+  `AssistantChatMessage.provider_blocks`; `/provider` cannot carry it across;
+- OpenAI's chat-completions surface cannot reference an image id, so on that
+  provider the feature implies the Responses convention;
+- the token bill does not change: the model still processes the picture in
+  every request that carries it. Uploading buys the 32 MiB inline cap, a
+  smaller request body, and not re-sending the same screenshot each round.
+
 ## Verified against the DeepSeek API

 The 2026-10-17 docs plus a live probe (`deepseek-flash`):
@@ -93,16 +127,32 @@ completed in its quoted form).

 ## Opting in

-`[providers.<name>.models."<model>"].image_in` is the gate — an `@image` mention
-is skipped, with a note naming the key, on a model that does not set it, and the
-default catalogs mark what takes pictures (`deepseek-flash` yes,
+`[providers.<name>.models."<model>"].image_in` is the gate, and it covers every
+way a picture can enter a turn — not only an `@image` mention:
+
+| Path | Where it is gated |
+|---|---|
+| an `@image` mention (`cli/mentions.py`) | skipped, with a note naming the key |
+| the `view_image` tool | `error: model '…' does not accept image input (set image_in = true under [providers.x.models."y"])` |
+| an image body from `fetch_local` | the metadata stays, the picture is left behind with the same reason |
+| an MCP server's image content, a plugin's `ToolOutput` | dropped by the registry's safety net (`agent/tools.py`), with the reason appended to the text |
+
+`ReplState.sync_image_capability` publishes the selected model's verdict on
+`InstanceState.workspace` and runs on every provider/model/client rebuild, so a
+`/model` switch flips the gate at once. A host that binds no instance declares
+nothing, and the picture is sent as usual.
+
+The point is to fail loudly: without the gate a text-only model answers
+`[Unsupported Image]` one round later, and a gateway that cannot read image
+blocks hands the model raw base64 — which reads like the picture arrived when it
+did not.
+
+The default catalogs mark what takes pictures (`deepseek-flash` yes,
 `deepseek-v4-pro` no, OpenAI's defaults yes). A model missing from the catalog
 has to be declared: the flag is an explicit opt-in.

 ## Not implemented

- **Files API** (`POST /files` → `file_id`): it lifts the 32 MiB per-image cap
-  and avoids re-uploading the same bytes, but it does not change the token bill
-  (the model still processes the picture in every request that carries it), so a
-  local 8 MiB cap plus resizing covers the common case.
+- **Files API**: see above — no image path uses it today; a local 8 MiB cap plus
+  resizing covers the common case, and the upload changes no token cost.
 - **Image output**: the API documents none.
@@ -45,6 +45,12 @@ session grant: the button means "this once". "Go to terminal" withdraws the
 notification and leaves the prompt waiting. A secret is never asked for in a
 notification, and never echoed into one.

+A prompt that can be answered from the notification reads both surfaces while it
+blocks, from the first moment — the notification itself only shows up after
+`grace_seconds`, so waiting for it to appear before listening would block on the
+terminal alone. Typing at the terminal always wins; a readline-only convenience
+(Tab completion on `ask_user_line`) is what a question gives up in exchange.
+
 ## Desktop channel

 The desktop channel needs the optional extra:
@@ -1,6 +1,6 @@
 [project]
 name = "plyngent"
-version = "0.11.0"
+version = "0.12.0"
 description = "LLM chat and agent toolkit"
 authors = [
  { name = "HivertMoZara", email = "worldmozara@163.com" },
@@ -58,6 +58,27 @@ class ToolOutput(Struct):
        return bool(self.images)


+def _without_refused_images(output: ToolOutput) -> ToolOutput:
+    """Drop a picture the selected model cannot take, and say why.
+
+    ``view_image`` and ``fetch_local`` refuse on their own, before any bytes are
+    read, so the model gets a clean error. This is the safety net for the rest —
+    an MCP server's image content, a plugin's ``ToolOutput`` — where the picture
+    would otherwise reach a model that answers ``[Unsupported Image]``, or a
+    gateway that hands it raw base64 (see ``doc/images.md``).
+    """
+    if not output.images:
+        return output
+    from plyngent.tools.workspace import image_input_refusal
+
+    refusal = image_input_refusal()
+    if refusal is None:
+        return output
+    note = refusal.removeprefix("error: ")
+    text = f"{output.text}\n(error: image dropped: {note})" if output.text else f"error: image dropped: {note}"
+    return ToolOutput(text=text)
+
+
@dataclass(frozen=True, slots=True)
 class ToolConfirmRequest:
    """One call of a batch that needs a soft confirm before it may run."""
@@ -737,7 +758,7 @@ class ToolRegistry:
                if denied is not None:
                    return ToolOutput(text=denied)
                result = await self._invoke(definition, args)
-                return result if isinstance(result, ToolOutput) else ToolOutput(text=result)
+                return _without_refused_images(result if isinstance(result, ToolOutput) else ToolOutput(text=result))

        # Bind host-provided state for tools that read contextvars (todo, workspace).
        # Tag enforcement remains gated by ``auto_bind_state``; binding is always useful
@@ -351,6 +351,7 @@ async def _run_chat(  # noqa: C901, PLR0915 — chat orchestration
        # instance.workspace (no process bag).
        state.instance_state.workspace.path_denylist = tuple(store.agent_config.path_denylist or ())
        state.instance_state.workspace.trash_enabled = store.trash_config.enabled
+        state.sync_image_capability()
        from plyngent.tools.net import set_ssrf_assume_public_cidrs

        # Clash Fake-IP etc.: treat configured CIDRs as public for fetch_local SSRF.
@@ -236,22 +236,7 @@ async def prompt_continue_limit_async(reason: str) -> bool:

 def _confirm_with(wait: PromptAnswerSource | None, prompt: str, *, default: bool) -> bool:
    """Yes/no, also taking an answer that arrived through a notification."""
-    source = _answer_source(wait)
-    if source is None:
-        return confirm(prompt, default=default)
-    suffix = "[Y/n]" if default else "[y/N]"
-    with contextlib.suppress(OSError):
-        _ = sys.stderr.write(f"{prompt} {suffix}: ")
-        _ = sys.stderr.flush()
-    try:
-        raw = _read_answer(None, wait)
-    except (KeyboardInterrupt, EOFError) as exc:
-        msg = "prompt cancelled"
-        raise NonInteractiveError(msg) from exc
-    token = (raw or "").strip().lower()
-    if not token:
-        return default
-    return token in {"y", "yes"}
+    return confirm(prompt, default=default, source=_answer_source(wait))


 def _prompt_confirm_tool_sync(name: str, args: Mapping[str, object], reason: str) -> bool | str:
@@ -4,7 +4,9 @@ A prompt blocks the turn, so someone who switched to another window learns about
 it only when they come back. The notifier announces a prompt once it has waited
 ``grace_seconds`` — answering at the terminal stays silent — and, where the
 platform allows it, offers the answer itself: allow / deny / review buttons for
-a confirmation, a reply field for a question.
+a confirmation, a reply field for a question. A prompt that can be answered from
+the notification reads from both surfaces from its first poll, since the
+announcement only appears once the grace period is over.

 Two channels. The desktop one is the optional ``desktop-notifier`` extra
 (``pip install plyngent[notify]``); the terminal one writes an OSC escape
@@ -38,8 +40,16 @@ ALLOW = "allow"
 DENY = "deny"
 REVIEW = "review"

-#: Kinds whose notification carries answer buttons (``ask`` gets a reply field).
-_ANSWER_KINDS = frozenset({"confirm", "batch", "policy", "access", "skills"})
+#: Kinds whose notification carries an answer, and the capability it needs: a
+#: confirmation gets buttons, a question gets a reply field.
+_ANSWER_CAPABILITY: dict[str, str] = {
+    "confirm": "buttons",
+    "batch": "buttons",
+    "policy": "buttons",
+    "access": "buttons",
+    "skills": "buttons",
+    "ask": "reply",
+}

 _BUTTONS: tuple[tuple[str, str], ...] = (
    (ALLOW, "Allow once"),
@@ -312,14 +322,16 @@ class WaitHandle:
    notifier: WaitNotifier
    wait: PromptWait
    identifier: str | None
-    #: Whether the notification offered answers, so a read may poll for them.
+    #: Whether an answer may arrive from the notification, so a read polls for
+    #: one from the first moment it blocks — the notification itself is only
+    #: shown after the grace period.
    supports_actions: bool

    def __init__(self, notifier: WaitNotifier, wait: PromptWait) -> None:
        self.notifier = notifier
        self.wait = wait
        self.identifier = None
-        self.supports_actions = False
+        self.supports_actions = notifier.answer_capability(wait.kind) is not None
        self._answers: SimpleQueue[str] = SimpleQueue()
        self._timer: threading.Timer | None = None
        self._closed: bool = False
@@ -377,6 +389,7 @@ class WaitNotifier:
    _config: NotifyConfig
    _resolve_desktop: bool
    _desktop: DesktopChannel | None
+    _resolve_lock: threading.Lock
    _write: Callable[[str], None]
    _env: Mapping[str, str]
    _clock: Callable[[], float]
@@ -398,6 +411,7 @@ class WaitNotifier:
        else:
            self._resolve_desktop = False
            self._desktop = desktop
+        self._resolve_lock = threading.Lock()
        self._write = write or write_terminal
        self._env = os.environ if env is None else env
        self._clock = clock
@@ -435,10 +449,39 @@ class WaitNotifier:
    def desktop_channel(self) -> DesktopChannel | None:
        """The desktop surface, resolved once (importing the extra on demand)."""
        if self._resolve_desktop:
-            self._desktop = load_desktop_channel()
-            self._resolve_desktop = False
+            with self._resolve_lock:
+                if self._resolve_desktop:
+                    self._desktop = load_desktop_channel()
+                    self._resolve_desktop = False
        return self._desktop

+    def answer_capability(self, kind: str) -> str | None:
+        """The capability *kind*'s answer needs (``buttons`` / ``reply``), or None.
+
+        Answered before the notification is shown, so a prompt knows from the
+        start of its wait whether an answer may arrive.
+        """
+        needed = _ANSWER_CAPABILITY.get(kind)
+        if needed is None or not self._config.actions or not self._wants("desktop"):
+            return None
+        channel = self.desktop_channel()
+        if channel is None or needed not in channel.capabilities():
+            return None
+        return needed
+
+    def warm(self) -> None:
+        """Resolve the desktop channel off the caller's thread.
+
+        Probing the platform's capabilities costs a round trip to the
+        notification server, so the first prompt of a run must not pay for it
+        while the human is already waiting.
+        """
+        if not self._resolve_desktop or not self._config.actions or not self._wants("desktop"):
+            return
+        if importlib.util.find_spec("desktop_notifier") is None:
+            return  # no extra: the terminal channel is all there is
+        threading.Thread(target=self.desktop_channel, name="plyngent-notify-warm", daemon=True).start()
+
    def notify_desktop(self, handle: WaitHandle) -> bool:
        """Show the desktop notification; False when it could not be shown."""
        if not self._wants("desktop"):
@@ -447,25 +490,20 @@ class WaitNotifier:
        if channel is None:
            return False
        wait = handle.wait
-        capabilities = channel.capabilities()
-        buttons: tuple[tuple[str, str], ...] = ()
-        if self._config.actions and wait.kind in _ANSWER_KINDS and "buttons" in capabilities:
-            buttons = _BUTTONS
-        reply = self._config.actions and wait.kind == "ask" and "reply" in capabilities
+        capability = self.answer_capability(wait.kind)
        try:
            handle.identifier = channel.send(
                DesktopRequest(
                    title=_clean(wait.title, _MAX_TITLE),
                    message=self._message(wait),
-                    buttons=buttons,
-                    reply_field=reply,
+                    buttons=_BUTTONS if capability == "buttons" else (),
+                    reply_field=capability == "reply",
                    on_answer=handle.answer,
                    on_reply=handle.answer,
                )
            )
        except Exception:  # noqa: BLE001 - a notification never breaks a prompt
            return False
-        handle.supports_actions = bool(buttons) or reply
        return True

    def notify_terminal(self, wait: PromptWait) -> None:
@@ -497,5 +535,6 @@ def install_notifier(config: NotifyConfig) -> WaitNotifier | None:
        set_prompt_wait_hook(None)
        return None
    notifier = WaitNotifier(config)
+    notifier.warm()
    set_prompt_wait_hook(notifier)
    return notifier
@@ -507,6 +507,7 @@ class ReplState:
        # Preserve live stream toggle if agent already exists.
        if hasattr(self, "agent"):
            self.stream_enabled = self.agent.stream
+        self.sync_image_capability()
        self.client = create_client(self.provider, model=self.model, reasoning=self.reasoning)
        self.agent = self._make_agent()
        # Restore history without re-marking already-stored messages as dirty.
@@ -517,6 +518,21 @@ class ReplState:
        if self._remote_models_key is not None and self._remote_models_key != self._models_cache_key():
            self.invalidate_remote_models()

+    def sync_image_capability(self) -> None:
+        """Publish the selected model's ``image_in`` to the image-carrying tools.
+
+        Same verdict the ``@path`` mention gate uses: only a model declared with
+        ``image_in = true`` takes pictures, so a text-only model (or one behind a
+        gateway that cannot read image blocks) is answered with an error instead
+        of a picture it could never describe.
+        """
+        from plyngent.config.routing import model_config_for
+
+        model_config = model_config_for(self.provider, self.model)
+        policy = self.instance_state.workspace
+        policy.image_in = model_config is not None and bool(model_config.image_in)
+        policy.image_model = (self.provider_name, self.model)
+
    def _models_cache_key(self) -> tuple[str, str]:
        # Remote model lists belong to the provider endpoint used for discovery,
        # not model-level overrides (the model is not selected yet there).
@@ -140,7 +140,12 @@ class PromptAnswerSource(Protocol):

    @property
    def supports_actions(self) -> bool:
-        """Whether :meth:`take` can answer this prompt at all."""
+        """Whether an answer may arrive while the prompt waits.
+
+        A surface that will offer the answer says so before it has offered it —
+        the notification is only shown after the host's grace period — so a read
+        polls :meth:`take` from its first moment instead of deciding once.
+        """
        ...

    def take(self) -> str | None:
@@ -619,8 +624,26 @@ def choose(
        backend.echo("Invalid choice; pick a listed option.")


-def confirm(prompt: str, *, default: bool = False) -> bool:
-    """Yes/no confirm via the active backend.
+#: Answers a surface spells out on top of the ``y`` a confirm already accepts.
+_YES = frozenset({"allow", "y", "yes"})
+
+
+def _confirm_from_source(prompt: str, source: PromptAnswerSource, *, default: bool) -> bool:
+    """Yes/no for a prompt a surface may answer, reading the terminal as well."""
+    hint = "[Y/n]" if default else "[y/N]"
+    try:
+        answer = _read_with_source(f"{prompt} {hint}", source, default="")
+    except (KeyboardInterrupt, EOFError) as exc:
+        msg = "prompt cancelled"
+        raise NonInteractiveError(msg) from exc
+    token = answer.strip().lower()
+    if not token:
+        return default
+    return token in _YES
+
+
+def confirm(prompt: str, *, default: bool = False, source: PromptAnswerSource | None = None) -> bool:
+    """Yes/no confirm via the active backend, or from *source* when given.

    Non-interactive backends return ``default`` without blocking.
    Cancel (Ctrl+C / EOF) raises :class:`NonInteractiveError`.
@@ -628,8 +651,13 @@ def confirm(prompt: str, *, default: bool = False) -> bool:
    backend = get_prompt_backend()
    if not backend.is_interactive():
        return default
+    if source is not None:
+        return _confirm_from_source(prompt, source, default=default)
    backend.echo()
-    with prompt_wait("confirm", prompt):
+    with prompt_wait("confirm", prompt) as wait:
+        offered = wait if wait is not None and wait.supports_actions else None
+        if offered is not None:
+            return _confirm_from_source(prompt, offered, default=default)
        return backend.confirm(prompt, default=default)


@@ -105,6 +105,7 @@ from .workspace import get_trash_enabled as get_trash_enabled
 from .workspace import get_workspace_root as get_workspace_root
 from .workspace import grant_policy_command as grant_policy_command
 from .workspace import granted_access_mode as granted_access_mode
+from .workspace import image_input_refusal as image_input_refusal
 from .workspace import list_workspace_allowlist as list_workspace_allowlist
 from .workspace import match_command_denylist as match_command_denylist
 from .workspace import mode_covers as mode_covers
@@ -113,6 +114,7 @@ from .workspace import pop_owned_temporary_workspaces as pop_owned_temporary_wor
 from .workspace import remove_workspace_allowlist as remove_workspace_allowlist
 from .workspace import resolve_path as resolve_path
 from .workspace import set_command_denylist as set_command_denylist
+from .workspace import set_image_in as set_image_in
 from .workspace import set_path_denylist as set_path_denylist
 from .workspace import set_policy_confirm_hook as set_policy_confirm_hook
 from .workspace import set_policy_confirm_timeout as set_policy_confirm_timeout
@@ -9,6 +9,7 @@ program — further in.

 from __future__ import annotations

+import getopt
 import re
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
@@ -19,6 +20,7 @@ if TYPE_CHECKING:
 __all__ = [
    "ESCALATION_PROGRAMS",
    "CommandString",
+    "RemoteTarget",
    "UnwrappedCommand",
    "basename",
    "dash_c_code",
@@ -27,34 +29,9 @@ __all__ = [
    "unwrap_command",
 ]

-# Wrappers that run the program named on their own command line.
-_ENV_WRAPPERS: frozenset[str] = frozenset(
-    {
-        "command",
-        "doas",
-        "env",
-        "exec",
-        "ionice",
-        "nice",
-        "nohup",
-        "pkexec",
-        "setsid",
-        "stdbuf",
-        "su",
-        "sudo",
-        "time",
-        "timeout",
-        "watch",
-        "xargs",
-    }
-)
-
 # Detached runners: review-worthy whatever program they wrap.
 _SELF_REVIEW: frozenset[str] = frozenset({"nohup"})

-# Wrappers that take a bare duration of their own (``timeout 5 cmd``).
-_DURATION_WRAPPERS: frozenset[str] = frozenset({"timeout"})
-
 # ``<launcher> run <program> …``: task runners and package managers that launch
 # a program in their environment. A package-manager script (``npm run build``)
 # names a script rather than a program; the launcher basename is still checked.
@@ -118,60 +95,186 @@ _ASSIGNMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=")
 # A bare duration, as ``timeout`` takes: ``5``, ``1.5``, ``30s``.
 _DURATION = re.compile(r"[0-9]+(?:\.[0-9]+)?[smhd]?")

-# Options that consume the next token, so ``sudo -u root python`` still finds
-# ``python``; jammed forms (``-n10``, ``--with=pkg``) need no entry, and unknown
-# options are skipped bare. List value-taking options only: an entry for a flag
-# eats the program behind it, which hid ``rm`` from ``exec -c rm -rf /`` and
-# judged ``exec -c git -c … init`` as ``init``. An option with an *optional*
-# value (``xargs -e``, ``watch -d``) needs no entry either — its bare form runs
-# the program, and the jammed form carries the value in the same token.
-_VALUE_OPTS: dict[str, frozenset[str]] = {
-    "doas": frozenset({"-u", "--user"}),
-    "env": frozenset({"-C", "-S", "-u", "--chdir", "--split-string", "--unset"}),
-    "exec": frozenset({"-a"}),
-    "ionice": frozenset({"-c", "-n", "-p", "-P", "-u"}),
-    "nice": frozenset({"-n", "--adjustment"}),
-    "pkexec": frozenset({"--user"}),
-    "stdbuf": frozenset({"-e", "-i", "-o"}),
-    "su": frozenset({"-c", "-g", "-G", "-s", "-u", "-w", "--command", "--group", "--session-command", "--shell"}),
-    "sudo": frozenset(
-        {
-            "-C",
-            "-D",
-            "-g",
-            "-h",
-            "-p",
-            "-r",
-            "-R",
-            "-t",
-            "-T",
-            "-u",
-            "-U",
-            "--close-from",
-            "--command-timeout",
-            "--group",
-            "--host",
-            "--other-user",
-            "--prompt",
-            "--role",
-            "--type",
-            "--user",
-        }
+# Every wrapper's own options, spelled the way ``getopt`` reads them: a short
+# option with ``:`` (``u:``) takes a value, a bare one is a flag, and a long
+# option ending in ``=`` takes one. The flags matter as much as the values — a
+# value-only list let a flag eat the program behind it (``exec -c rm -rf /``
+# resolved to the ``-c``, hiding ``rm``), and an unknown option is no longer
+# guessed at (see `_parse_options`).
+
+
+@dataclass(frozen=True)
+class _Spec:
+    """How one wrapper is invoked, so the program behind it can be found."""
+
+    short: str = ""
+    long: tuple[str, ...] = ()
+    #: Options whose value is a command line the wrapper runs (``env -S``).
+    command_opts: frozenset[str] = frozenset()
+    #: Takes a bare duration of its own before its command (``timeout 5 cmd``).
+    duration: bool = False
+    #: ``<launcher> run <program>``: only the ``run`` form wraps a program.
+    launcher: bool = False
+    #: Runs its command on another machine (``ssh`` / ``mosh``).
+    remote: bool = False
+
+
+_WRAPPERS: dict[str, _Spec] = {
+    "command": _Spec(short="pvV"),
+    "doas": _Spec(short="C:nsu:", long=("user=",)),
+    # ``env -S`` / ``su -c`` carry a *command line*: the value is code the review
+    # must surface (`_Spec.command_opts`), not an argument.
+    "env": _Spec(
+        short="0iC:u:S:v",
+        long=("chdir=", "debug", "ignore-environment", "null", "split-string=", "unset="),
+        command_opts=frozenset({"--split-string", "-S"}),
+    ),
+    "exec": _Spec(short="cla:"),
+    "ionice": _Spec(short="c:n:p:P:u:thV"),
+    "nice": _Spec(short="n:", long=("adjustment=",)),
+    "nohup": _Spec(),
+    "pkexec": _Spec(long=("disable-internal-agent", "help", "user=", "version")),
+    "setsid": _Spec(short="cfwV"),
+    "stdbuf": _Spec(short="e:i:o:", long=("error=", "input=", "output=")),
+    "su": _Spec(
+        short="c:g:G:lmps:u:w:",
+        long=(
+            "command=",
+            "group=",
+            "login",
+            "preserve-environment",
+            "session-command=",
+            "shell=",
+            "whitelist-environment=",
+        ),
+        command_opts=frozenset({"--command", "--session-command", "-c"}),
+    ),
+    "sudo": _Spec(
+        short="AbEHiKknPSsVvLlC:D:g:h:p:r:R:t:T:u:U:",
+        long=(
+            "askpass",
+            "background",
+            "chdir=",
+            "close-from=",
+            "command-timeout=",
+            "edit",
+            "group=",
+            "host=",
+            "list",
+            "login",
+            "non-interactive",
+            "other-user=",
+            "preserve-env",
+            "prompt=",
+            "remove-timestamp",
+            "reset-timestamp",
+            "role=",
+            "set-home",
+            "shell",
+            "stdin",
+            "type=",
+            "user=",
+            "validate",
+            "version",
+        ),
+    ),
+    "time": _Spec(short="af:o:pq", long=("append", "format=", "output=", "portability", "quiet", "verbose")),
+    "timeout": _Spec(
+        short="k:s:v",
+        long=("foreground", "kill-after=", "preserve-status", "signal="),
+        duration=True,
+    ),
+    "watch": _Spec(
+        short="bcdeghn:prtvx",
+        long=(
+            "beep",
+            "chgexit",
+            "color",
+            "differences",
+            "errexit",
+            "exec",
+            "interval=",
+            "no-rerun",
+            "no-title",
+            "precise",
+        ),
+    ),
+    "xargs": _Spec(
+        short="0a:d:eE:iI:L:ln:P:prs:txo",
+        long=(
+            "arg-file=",
+            "delimiter=",
+            "eof=",
+            "interactive",
+            "max-args=",
+            "max-chars=",
+            "max-lines=",
+            "max-procs=",
+            "no-run-if-empty",
+            "null",
+            "open-tty",
+            "replace=",
+            "show-limits",
+            "verbose",
+        ),
    ),
-    "time": frozenset({"-f", "-o", "--format", "--output"}),
-    "timeout": frozenset({"-k", "-s", "--kill-after", "--signal"}),
-    "watch": frozenset({"-n"}),
-    "xargs": frozenset({"-a", "-d", "-E", "-I", "-L", "-n", "-P", "-s"}),
 }

-# Options whose value is a *command line* the wrapper will run, so the value is
-# code rather than an argument: ``env -S "rm -rf /"`` and ``su -c "…"``. An
-# interpreter's own ``-c`` is the same idea, but there the interpreter is the
-# program under review; here a wrapper hides the command, so the review has to
-# surface it (:func:`plyngent.tools.danger.classify_danger`).
-_COMMAND_STRING_OPTS: dict[str, frozenset[str]] = {
-    "env": frozenset({"-S", "--split-string"}),
-    "su": frozenset({"-c", "--command", "--session-command"}),
+# ``<launcher> run <program> …``: the option sets of the launchers below differ,
+# so one union stands in for all of them — a value option is the only kind that
+# can swallow the program, and the common flags are listed to keep them quiet.
+_LAUNCHER_SPEC = _Spec(
+    short="C:e:F:G:m:n:p:s:w:",
+    long=(
+        "all-extras",
+        "all-groups",
+        "all-packages",
+        "config=",
+        "constraint=",
+        "default-index=",
+        "directory=",
+        "env=",
+        "env-file=",
+        "extra=",
+        "extras=",
+        "file=",
+        "frozen",
+        "group=",
+        "index=",
+        "locked",
+        "module=",
+        "name=",
+        "no-project",
+        "no-sync",
+        "offline",
+        "overrides=",
+        "package=",
+        "prefix=",
+        "project=",
+        "python=",
+        "python-version=",
+        "quiet",
+        "refresh",
+        "script=",
+        "verbose",
+        "with=",
+        "without=",
+    ),
+    launcher=True,
+)
+
+# ``ssh`` / ``mosh``: the program under review is the one after the destination,
+# which is the first token that is not one of their options.
+_REMOTE_WRAPPERS: dict[str, _Spec] = {
+    "mosh": _Spec(
+        short="46hP:p:v",
+        long=("client-port=", "family=", "help", "no-ssh-pty", "port=", "predict=", "server=", "ssh="),
+        remote=True,
+    ),
+    "ssh": _Spec(
+        short="46AaCfgGKkMNnqSTtVvXxYysB:b:c:D:E:e:F:I:i:J:L:l:m:O:o:P:p:Q:R:S:W:w:",
+        remote=True,
+    ),
 }

 # Launcher options that consume a value (``uv run --with rich python``).
@@ -226,6 +329,18 @@ class CommandString:
        return f"{self.wrapper} {self.option}"


+@dataclass(frozen=True)
+class RemoteTarget:
+    """The hop a remote wrapper (``ssh`` / ``mosh``) sends its command to."""
+
+    wrapper: str
+    host: str
+    #: ``-o BatchMode=yes``: the client never prompts, so the command runs.
+    batch: bool = False
+    #: No command followed: the client opens a shell whose input is not scanned.
+    shell: bool = False
+
+
@dataclass(frozen=True)
 class UnwrappedCommand:
    """A command plus the program it really runs, past wrappers and launchers."""
@@ -236,6 +351,10 @@ class UnwrappedCommand:
    base: str
    self_review: tuple[str, ...]
    command_strings: tuple[CommandString, ...] = ()
+    #: The remote hop, when the command runs on another machine.
+    remote: RemoteTarget | None = None
+    #: An option the wrapper's spec does not know: the program behind it is a guess.
+    unparsed: str = ""

    @property
    def command(self) -> tuple[str, ...]:
@@ -287,20 +406,39 @@ def unwrap_command(argv: Sequence[str]) -> UnwrappedCommand:
    """Resolve *argv* to the program behind any wrappers and ``run`` launchers.

    Wrappers are unwrapped outer → inner as long as a program still follows
-    them; their own options and ``FOO=bar`` assignments are skipped, and an
-    option that carries a command line is collected in ``command_strings``.
+    them; their own options and ``FOO=bar`` assignments are read by ``getopt``
+    (:func:`_parse_options`), and an option that carries a command line is
+    collected in ``command_strings``.
+
+    ``ssh`` / ``mosh`` are unwrapped the same way — the program under review is
+    the one after the destination (``ssh prod rm -rf /`` runs ``rm``), and the
+    hop is reported in :attr:`UnwrappedCommand.remote`. A client with no command
+    opens a remote shell (``remote.shell``); nothing behind it can be resolved
+    here.
+
+    An option the wrapper's spec does not know stops the unwrap at that wrapper
+    and is reported in :attr:`UnwrappedCommand.unparsed`: the program behind it
+    is unknown, so the review asks instead of guessing (the wrapper itself is
+    still judged, so ``sudo <new flag> rm -rf /`` keeps its ``sudo`` prompt).
    """
    parts = list(argv)
    wrappers: list[str] = []
    command_strings: list[CommandString] = []
+    remote: RemoteTarget | None = None
+    unparsed = ""
    index = 0
    while index < len(parts):
        name = basename(parts[index])
-        nxt = _skip_wrapper(name, parts, index + 1, command_strings=command_strings)
-        if nxt is None:
+        step = _step(name, parts, index + 1, command_strings=command_strings)
+        if step.unparsed:
+            unparsed = f"{name} {step.unparsed}"
+            break
+        if step.remote is not None:
+            remote = step.remote
+        if step.index is None:
            break
        wrappers.append(name)
-        index = nxt
+        index = step.index
    base = basename(parts[index]) if index < len(parts) else ""
    chain = dict.fromkeys((*wrappers, base))
    return UnwrappedCommand(
@@ -310,88 +448,108 @@ def unwrap_command(argv: Sequence[str]) -> UnwrappedCommand:
        base=base,
        self_review=tuple(name for name in chain if name in _SELF_REVIEW),
        command_strings=tuple(command_strings),
+        remote=remote,
+        unparsed=unparsed,
    )


-def _skip_wrapper(
-    name: str,
-    parts: list[str],
-    start: int,
-    *,
-    command_strings: list[CommandString],
-) -> int | None:
-    """Index of the program *name* wraps, or ``None`` when it is the whole command."""
-    if name in _ENV_WRAPPERS:
-        nxt = _skip_options(
-            parts,
-            start,
-            name=name,
-            value_opts=_VALUE_OPTS.get(name, frozenset()),
-            leading_duration=name in _DURATION_WRAPPERS,
-            command_strings=command_strings,
-        )
-        return nxt if nxt < len(parts) else None
-    if name in _RUN_LAUNCHERS and start < len(parts) and parts[start] == "run":
-        nxt = _skip_options(
-            parts,
-            start + 1,
-            name=name,
-            value_opts=_RUN_VALUE_OPTS,
-            leading_duration=False,
-            command_strings=command_strings,
-        )
-        return nxt if nxt < len(parts) else None
-    return None
+@dataclass(frozen=True)
+class _Parsed:
+    """A wrapper's own options, where its command starts, and what it did not know."""
+
+    options: tuple[tuple[str, str], ...] = ()
+    index: int = 0
+    unparsed: str = ""


-def _record_command_string(
-    name: str,
-    option: str,
-    text: str,
-    command_strings: list[CommandString],
-) -> None:
-    """Note *text* when *option* hands a command line to wrapper *name*."""
-    if option in _COMMAND_STRING_OPTS.get(name, frozenset()):
-        command_strings.append(CommandString(wrapper=name, option=option, text=text))
+@dataclass(frozen=True)
+class _Step:
+    """One wrapper resolved: its command, the hop it reaches, or its parse error."""
+
+    index: int | None = None
+    unparsed: str = ""
+    remote: RemoteTarget | None = None


-def _skip_options(
-    parts: list[str],
-    start: int,
-    *,
-    name: str,
-    value_opts: frozenset[str],
-    leading_duration: bool,
-    command_strings: list[CommandString],
-) -> int:
-    """Index of the first token after a wrapper's own options and assignments."""
+def _step(name: str, parts: list[str], start: int, *, command_strings: list[CommandString]) -> _Step:
+    """Resolve the wrapper *name* at ``parts[start - 1]`` past its own options."""
+    spec = _spec_for(name)
+    if spec is None:
+        return _Step()
    index = start
-    took_duration = False
-    while index < len(parts):
-        token = parts[index]
-        if token == "--":
-            return index + 1
-        if _ASSIGNMENT.match(token):
-            index += 1
-            continue
-        if token.startswith("-"):
-            index += 1
-            if token in value_opts:
-                value = parts[index] if index < len(parts) else ""
-                _record_command_string(name, token, value, command_strings)
-                index += 1
-                continue
-            # Jammed forms: ``--split-string=…`` and ``-S…`` carry the value in
-            # the token itself, so a command line may hide there too.
-            option, equal, attached = token.partition("=")
-            if equal:
-                _record_command_string(name, option, attached, command_strings)
-            elif token[:2] in _COMMAND_STRING_OPTS.get(name, frozenset()):
-                _record_command_string(name, token[:2], token[2:], command_strings)
-            continue
-        if leading_duration and not took_duration and _DURATION.fullmatch(token):
-            took_duration = True
-            index += 1
-            continue
-        return index
+    if spec.launcher:
+        if index >= len(parts) or parts[index] != "run":
+            return _Step()  # ``npm install``: the launcher is the whole command
+        index += 1
+    parsed = _parse_options(parts, index, spec)
+    if parsed.unparsed:
+        return _Step(unparsed=parsed.unparsed)
+    for option, value in parsed.options:
+        if option in spec.command_opts:
+            command_strings.append(CommandString(wrapper=name, option=option, text=value))
+    index = _skip_prefixes(parts, parsed.index, spec)
+    if spec.remote:
+        return _remote_step(name, parts, index, parsed.options)
+    return _Step(index=index) if index < len(parts) else _Step()
+
+
+def _spec_for(name: str) -> _Spec | None:
+    """The option spec of a wrapper, or ``None`` when *name* wraps nothing."""
+    spec = _WRAPPERS.get(name) or _REMOTE_WRAPPERS.get(name)
+    if spec is None and name in _RUN_LAUNCHERS:
+        return _LAUNCHER_SPEC
+    return spec
+
+
+def _skip_prefixes(parts: list[str], index: int, spec: _Spec) -> int:
+    """Skip the ``FOO=bar`` assignments and the bare duration a wrapper takes first."""
+    while index < len(parts) and _ASSIGNMENT.match(parts[index]):
+        index += 1
+    if spec.duration and index < len(parts) and _DURATION.fullmatch(parts[index]):
+        index += 1
    return index
+
+
+def _parse_options(parts: list[str], start: int, spec: _Spec) -> _Parsed:
+    """Read a wrapper's own options, stopping at the first token that is not one.
+
+    ``getopt`` (POSIX order — not ``gnu_getopt``, which permutes) consumes the
+    wrapper's options and leaves the rest, so the length of that remainder is
+    where the wrapped program starts. Jammed forms (``-p2222``,
+    ``-oBatchMode=yes``, ``--with=pkg``) are its business, not ours. An option
+    the spec does not know is *not* skipped as a flag — that is how a flag ended
+    up eating the program — but reported, so the review can ask.
+    """
+    try:
+        options, rest = getopt.getopt(list(parts[start:]), spec.short, list(spec.long))
+    except getopt.GetoptError as exc:
+        return _Parsed(index=start, unparsed=f"(cannot read its options: {exc})")
+    return _Parsed(options=tuple(options), index=len(parts) - len(rest))
+
+
+def _remote_step(name: str, parts: list[str], index: int, options: Sequence[tuple[str, str]]) -> _Step:
+    """The hop a remote client reaches, and where its command starts (if any).
+
+    The destination is the first token that is not one of the client's options;
+    the command follows it, or — for ``mosh`` — follows an explicit ``--``. No
+    token after the destination means the client just opens a remote shell.
+    """
+    if index >= len(parts):
+        return _Step()
+    host = parts[index]
+    index += 1
+    if name == "mosh" and index < len(parts) and parts[index] == "--":
+        index += 1
+    batch = _batch_mode(options)
+    if index >= len(parts):
+        return _Step(remote=RemoteTarget(wrapper=name, host=host, batch=batch, shell=True))
+    return _Step(index=index, remote=RemoteTarget(wrapper=name, host=host, batch=batch))
+
+
+def _batch_mode(options: Sequence[tuple[str, str]]) -> bool:
+    """Whether ``-o BatchMode=yes`` was given: the client never prompts."""
+    for option, value in options:
+        key, _, setting = value.partition("=")
+        if option == "-o" and key.strip().lower() == "batchmode":
+            return setting.strip().lower() in {"1", "on", "true", "yes"}
+    return False
@@ -59,12 +59,26 @@ def _command_reason(argv: Sequence[str], *, via: str) -> DangerReason | None:
    program or a wrapper) is confirmed with ``yolo_skip=False``. It never adds a
    reason of its own: the denylist gate in the handler always asks about the
    escalation, so a second prompt here would only repeat it.
+
+    A command on another machine (``ssh``/``mosh``) is reviewed by the program it
+    runs there — the client's destination and options are skipped. With
+    ``-o BatchMode=yes`` the client cannot prompt at either end, so the command
+    is always confirmed, whatever program it is. A client with no command opens a
+    remote shell, which is announced: nothing typed or piped into it can be
+    checked, so the human decides before it starts.
    """
    scan = unwrap_command(argv)
    interpreter = is_interpreter(scan.base)
    findings = [f"{name} (detached run)" for name in scan.self_review]
    if interpreter:
        findings.append(f"interpreter {scan.base!r}")
+    remote = scan.remote
+    if remote is not None and remote.shell:
+        findings.append(f"remote shell on {remote.host!r} ({remote.wrapper} with no command; its input is not checked)")
+    elif remote is not None and remote.batch:
+        findings.append(f"remote command on {remote.host!r} (BatchMode=yes, no prompt on either end)")
+    if scan.unparsed:
+        findings.append(f"options not understood {scan.unparsed} — the wrapped program is a guess")
    findings.extend(f"{span.label} runs a command line" for span in scan.command_strings)
    findings = list(dict.fromkeys(findings))
    if not findings:
@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, cast

 from plyngent.agent import ToolOutput, ToolTag, tool
 from plyngent.agent.images import image_part_from_bytes
-from plyngent.tools.workspace import AccessMode, resolve_path
+from plyngent.tools.workspace import AccessMode, image_input_refusal, resolve_path

 if TYPE_CHECKING:
    from plyngent.lmproto.openai_compatible.model import ImageDetail
@@ -27,7 +27,9 @@ async def view_image(path: str, detail: str = "auto") -> str | ToolOutput:
    picture as it is, and ``auto`` (the default) lets the provider decide.

    The format is read from the file's own bytes, so a misnamed file still
-    works; anything else comes back as an error.
+    works; anything else comes back as an error. A model whose
+    ``[providers.*.models.*].image_in`` is off gets an error naming that key
+    instead of an attachment it could never read.
    """
    if detail not in _DETAILS:
        return f"error: detail must be one of {', '.join(sorted(_DETAILS))}"
@@ -36,6 +38,9 @@ async def view_image(path: str, detail: str = "auto") -> str | ToolOutput:
        return f"error: file not found: {path}"
    if not target.is_file():
        return f"error: not a file: {path}"
+    refusal = image_input_refusal()
+    if refusal is not None:
+        return refusal
    part = image_part_from_bytes(target.read_bytes(), detail=cast("ImageDetail", detail))
    if isinstance(part, str):
        return f"error: {path}: {part.removeprefix('error: ')}"
@@ -19,6 +19,7 @@ from plyngent.tools.net.policy import (
    soft_confirm_reason,
 )
 from plyngent.tools.truncate_token import truncate_to_memory
+from plyngent.tools.workspace import image_input_refusal


 def format_fetch_result(
@@ -147,10 +148,14 @@ def _attach_image(text: str, result: FetchResult) -> str | ToolOutput:
    """An image response is handed over as a picture, beside the usual metadata.

    The preview line stays in the text so ``body_kind=image`` and the byte count
-    are still visible to the model.
+    are still visible to the model. A model without ``image_in`` keeps the
+    metadata and is told why the picture stayed behind.
    """
    if result.body_kind != "image" or not result.body_image:
        return text
+    refusal = image_input_refusal()
+    if refusal is not None:
+        return f"{text}\n{refusal}"
    part = image_part_from_bytes(result.body_image)
    if isinstance(part, str):
        return f"{text}\n{part}"
@@ -122,6 +122,14 @@ class WorkspacePolicy:
    static_read: list[Path] = field(default_factory=list)
    # ``[trash] enabled``: deletes go to the system trash unless cleared.
    trash_enabled: bool = True
+    # ``[providers.*.models.*].image_in`` of the model this host is talking to:
+    # ``None`` = the host did not say, so a picture is passed through unchanged.
+    # ``False`` makes every image-carrying tool refuse, instead of sending bytes
+    # the model cannot read (the provider answers ``[Unsupported Image]``, or a
+    # gateway that ignores the block hands the model raw base64).
+    image_in: bool | None = None
+    # ``(provider name, model id)`` that verdict came from, for the message.
+    image_model: tuple[str, str] | None = None


 def _bound_instance() -> InstanceState | None:
@@ -200,6 +208,40 @@ def get_trash_enabled() -> bool:
    return active_workspace_policy().trash_enabled


+def set_image_in(*, image_in: bool | None, provider: str | None = None, model: str | None = None) -> None:
+    """Record whether the current model takes pictures (``image_in``).
+
+    ``None`` means the host did not declare it, and an image is sent as usual;
+    ``False`` is the explicit opt-out the image-carrying tools and the agent's
+    tool registry check through :func:`image_input_refusal`.
+    """
+    policy = active_workspace_policy()
+    policy.image_in = image_in
+    policy.image_model = (provider, model) if provider and model else None
+
+
+def image_input_refusal() -> str | None:
+    """``error: …`` text for an image the current model cannot take, else ``None``.
+
+    Mirrors the ``@path`` mention gate (``cli/mentions.py``): a picture is not
+    sent to a model whose ``image_in`` is off. The message names the config key
+    that turns it on, so the fix is one line away. With no instance bound the
+    host has declared nothing, and the picture passes through.
+    """
+    instance = _bound_instance()
+    if instance is None:
+        return None
+    policy = instance.workspace
+    if policy.image_in is not False:
+        return None
+    named = policy.image_model
+    if named is None:
+        return "error: this model does not accept image input (set image_in = true in the provider config)"
+    provider, model = named
+    hint = f'set image_in = true under [providers.{provider}.models."{model}"]'
+    return f"error: model {model!r} does not accept image input ({hint})"
+
+
 def set_command_denylist(names: list[str] | tuple[str, ...] | frozenset[str] | None) -> None:
    """Set denied command basenames (None restores defaults)."""
    policy = active_workspace_policy()
@@ -16,6 +16,7 @@ from plyngent.lmproto.openai_compatible.model import (
    ToolChatMessage,
    UserChatMessage,
 )
+from plyngent.tools.context import InstanceState


 def _image() -> ChatImagePart:
@@ -100,3 +101,36 @@ def test_the_char_estimate_counts_images() -> None:
    assert estimate_message_chars(ToolChatMessage(content="shot", tool_call_id="call-1", images=[_image()])) == (
        len("shot") + len("call-1") + IMAGE_CHARS
    )
+
+
+def _registry_for(instance: InstanceState) -> ToolRegistry:
+    @tool(register=False)
+    def shot() -> ToolOutput:
+        """Pretend to take a screenshot."""
+        return ToolOutput(text="image: shot.png (2 KB)", images=[_image()])
+
+    return ToolRegistry([shot], instance_state=instance)
+
+
+async def test_a_text_only_model_never_receives_a_plugin_tool_image() -> None:
+    """The registry is the safety net for tools that do not check ``image_in``."""
+    instance = InstanceState()
+    instance.workspace.image_in = False
+    instance.workspace.image_model = ("ds", "text-only")
+
+    output = await _registry_for(instance).execute_output("shot", "{}")
+
+    assert output.text.startswith("image: shot.png (2 KB)")
+    assert "image dropped" in output.text
+    assert "image_in = true" in output.text
+    assert not output.has_images
+
+
+async def test_a_vision_model_keeps_a_plugin_tool_image() -> None:
+    instance = InstanceState()
+    instance.workspace.image_in = True
+
+    output = await _registry_for(instance).execute_output("shot", "{}")
+
+    assert output.text == "image: shot.png (2 KB)"
+    assert output.images == [_image()]
@@ -1,6 +1,7 @@
 from __future__ import annotations

 import contextlib
+import sys
 from typing import TYPE_CHECKING

 import pytest
@@ -253,6 +254,60 @@ def test_tool_confirm_denied_by_the_notification() -> None:
        assert prompt_confirm_tool("run_argv", {}, "runs a command") is False


+@pytest.mark.skipif(sys.platform == "win32", reason="a pty is POSIX-only")
+def test_a_button_replaces_the_terminal_read(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A button pressed in the notification answers the waiting prompt.
+
+    The read consults stdin *and* the notification while it blocks: the button
+    only exists once the notification is shown, which is after the read began.
+    """
+    import os
+    import pty
+    import threading
+
+    from plyngent.cli.notify import ALLOW, DesktopRequest, WaitNotifier
+    from plyngent.config.models import NotifyConfig
+    from plyngent.prompting import set_prompt_wait_hook
+
+    class PressesTheButton:
+        """A desktop surface that answers once the notification is up."""
+
+        def __init__(self) -> None:
+            self.requests: list[DesktopRequest] = []
+
+        def capabilities(self) -> frozenset[str]:
+            return frozenset({"buttons", "reply"})
+
+        def send(self, request: DesktopRequest) -> str | None:
+            self.requests.append(request)
+            answer = request.on_answer
+            if answer is not None:
+                threading.Timer(0.2, answer, args=(ALLOW,)).start()
+            return "id-1"
+
+        def clear(self, identifier: str | None) -> None:
+            del identifier
+
+    desktop = PressesTheButton()
+    notifier = WaitNotifier(
+        NotifyConfig(grace_seconds=0.0, min_interval_seconds=0.0),
+        desktop=desktop,
+        write=lambda _sequence: None,
+    )
+    master, slave = pty.openpty()
+    stream = os.fdopen(slave)
+    set_prompt_wait_hook(notifier)
+    monkeypatch.setattr(sys, "stdin", stream)
+    try:
+        with temporary_backend(ScriptedBackend([])):
+            assert prompt_confirm_tool("run_argv", {}, "runs a command") is True
+    finally:
+        set_prompt_wait_hook(None)
+        stream.close()
+        os.close(master)
+    assert desktop.requests
+
+
 def test_batch_confirm_takes_a_notification_answer() -> None:
    from plyngent.agent import ToolConfirmRequest
    from plyngent.cli.limits import _prompt_confirm_batch_sync
@@ -257,6 +257,34 @@ def test_a_confirmation_offers_three_buttons() -> None:
    assert request.reply_field is False


+def test_a_waiting_prompt_is_answerable_before_it_is_announced() -> None:
+    """The offer is decided when the wait starts, not when the notification shows.
+
+    A notification only appears after the grace period, so a prompt that learned
+    about its answers from ``notify_desktop`` would already be blocked on stdin.
+    """
+    desktop = _FakeDesktop()
+    notifier, _ = _notifier(desktop=desktop, grace_seconds=60.0)
+    with notifier(PromptWait("confirm", "Allow?")) as handle:
+        assert handle.supports_actions is True
+        assert desktop.requests == []  # not announced yet
+
+
+def test_a_question_is_answerable_before_it_is_announced() -> None:
+    desktop = _FakeDesktop()
+    notifier, _ = _notifier(desktop=desktop, grace_seconds=60.0)
+    with notifier(PromptWait("ask", "Which branch?")) as handle:
+        assert handle.supports_actions is True
+        assert desktop.requests == []
+
+
+def test_no_answer_is_expected_without_the_capability() -> None:
+    desktop = _FakeDesktop(capabilities=frozenset({"buttons"}))
+    notifier, _ = _notifier(desktop=desktop)
+    with notifier(PromptWait("ask", "Which branch?")) as handle:
+        assert handle.supports_actions is False
+
+
 def test_an_ask_offers_a_reply_field_instead_of_buttons() -> None:
    desktop = _FakeDesktop()
    notifier, _ = _notifier(desktop=desktop)
@@ -24,6 +24,8 @@ from plyngent.lmproto.openai_compatible.model import (
    UserChatMessage,
 )
 from plyngent.memory import MemoryStore
+from plyngent.tools import image_input_refusal
+from plyngent.tools.context import bind_instance

 if TYPE_CHECKING:
    from collections.abc import AsyncIterator
@@ -855,3 +857,23 @@ async def test_a_picture_mention_is_skipped_for_a_text_only_model(
    note = payload.notes[0]
    assert "does not accept image input" in note
    assert "image_in = true" in note
+
+
+async def test_the_tool_image_gate_follows_the_selected_model(state: ReplState) -> None:
+    """The image-carrying tools read the same verdict the mention gate uses."""
+    policy = state.instance_state.workspace
+
+    state.sync_image_capability()
+    assert policy.image_in is False
+    assert policy.image_model == (state.provider_name, state.model)
+
+    state.provider.models[state.model] = ModelConfig(image_in=True)
+    state.sync_image_capability()
+    assert policy.image_in is True
+
+    state.provider.models[state.model] = ModelConfig(image_in=False)
+    state.sync_image_capability()
+    with bind_instance(state.instance_state):
+        refusal = image_input_refusal()
+    assert refusal is not None
+    assert f'[providers.{state.provider_name}.models."{state.model}"]' in refusal
@@ -267,6 +267,25 @@ def test_non_interactive_confirm_uses_default() -> None:
        assert confirm("ok?", default=True) is True


+def test_confirm_takes_a_host_answer() -> None:
+    """``allow`` / ``deny`` from a notification answer a yes/no like ``y`` / ``n``."""
+    with temporary_backend(ScriptedBackend([])):
+        assert confirm("Go?", source=_AnswerSource("allow")) is True
+        assert confirm("Go?", source=_AnswerSource("deny")) is False
+        assert confirm("Go?", source=_AnswerSource("n")) is False
+        # Nothing arrived: the prompt falls back to its default.
+        assert confirm("Go?", default=True, source=_AnswerSource(None)) is True
+
+
+def test_confirm_takes_an_answer_the_hook_offers() -> None:
+    backend = ScriptedBackend([], confirms=[False])
+    set_prompt_wait_hook(lambda _wait: contextlib.nullcontext(_AnswerSource("allow")))
+    with temporary_backend(backend):
+        assert confirm("Go?") is True
+    # The answer came from the surface, so the backend was never asked.
+    assert backend.confirms == [False]
+
+
 def test_prompt_output_hook_runs_before_chrome(capsys: pytest.CaptureFixture[str]) -> None:
    """A registered hook runs before echo/secho write; clearing it removes it.

@@ -1,6 +1,6 @@
 from __future__ import annotations

-from plyngent.tools.command_scan import basename, escalators_in_chain, is_interpreter, unwrap_command
+from plyngent.tools.command_scan import basename, dash_c_code, escalators_in_chain, is_interpreter, unwrap_command


 def test_basename() -> None:
@@ -132,6 +132,92 @@ def test_unwrap_empty_argv() -> None:
    assert scan.command == ()


+def test_unwrap_remote_command() -> None:
+    """``ssh``/``mosh`` are unwrapped to the program they run on the far end."""
+    scan = unwrap_command(["ssh", "prod", "rm", "-rf", "/"])
+    assert scan.wrappers == ("ssh",)
+    assert scan.base == "rm"
+    assert scan.command == ("rm", "-rf", "/")
+    assert scan.remote is not None
+    assert (scan.remote.wrapper, scan.remote.host, scan.remote.batch) == ("ssh", "prod", False)
+
+    scan = unwrap_command(["ssh", "-p", "2222", "-i", "~/.ssh/id", "user@prod", "ls", "-la"])
+    assert scan.base == "ls"
+    assert scan.command == ("ls", "-la")
+    assert scan.remote is not None and scan.remote.host == "user@prod"
+
+    # Jammed values, and a hop behind a local wrapper.
+    scan = unwrap_command(["ssh", "-p2222", "prod", "uptime"])
+    assert scan.base == "uptime"
+    assert unwrap_command(["sudo", "ssh", "prod", "id"]).base == "id"
+    scan = unwrap_command(["mosh", "prod", "--", "htop"])
+    assert scan.base == "htop"
+    assert scan.remote is not None and scan.remote.wrapper == "mosh"
+
+
+def test_unwrap_remote_batch_mode() -> None:
+    """``-o BatchMode=yes`` is reported: the code runs with no prompt at either end."""
+    for argv in (
+        ["ssh", "-o", "BatchMode=yes", "prod", "deploy.sh"],
+        ["ssh", "-oBatchMode=yes", "prod", "deploy.sh"],
+        ["ssh", "-o", "batchmode=YES", "-p", "22", "prod", "deploy.sh"],
+    ):
+        scan = unwrap_command(argv)
+        assert scan.remote is not None, argv
+        assert scan.remote.batch is True, argv
+    scan = unwrap_command(["ssh", "-o", "BatchMode=no", "prod", "deploy.sh"])
+    assert scan.remote is not None and scan.remote.batch is False
+
+
+def test_unwrap_remote_client_alone() -> None:
+    """A client with no command opens a remote shell: nothing to unwrap."""
+    for argv in (["ssh", "prod"], ["ssh", "-N", "prod"], ["ssh", "-G", "prod"], ["mosh", "prod"]):
+        scan = unwrap_command(argv)
+        assert scan.wrappers == (), argv
+        assert scan.base == argv[0], argv
+        assert scan.remote is not None, argv
+        assert scan.remote.shell is True, argv
+        assert scan.remote.host == "prod", argv
+    # Without a destination there is no hop to announce.
+    for argv in (["ssh"], ["ssh", "-V"]):
+        assert unwrap_command(argv).remote is None, argv
+
+
+def test_unwrap_reads_options_with_getopt() -> None:
+    """Each wrapper's spec is a ``getopt`` pattern: flags and values are exact.
+
+    ``--`` ends the wrapper's options, jammed values work (``-p2222``), and a
+    launcher's common flags stay flags (``uv run --no-project pytest``).
+    """
+    assert unwrap_command(["sudo", "-u", "root", "--", "rm", "-rf", "/"]).base == "rm"
+    assert unwrap_command(["ssh", "-p2222", "prod", "uptime"]).base == "uptime"
+    assert unwrap_command(["uv", "run", "--no-project", "pytest", "-x"]).base == "pytest"
+    assert unwrap_command(["time", "-f", "%e", "python", "x.py"]).base == "python"
+    assert unwrap_command(["xargs", "-0", "--", "rm"]).base == "rm"
+
+
+def test_an_unknown_option_stops_the_unwrap_and_is_reported() -> None:
+    """The program behind an option we cannot read is a guess, so it is reported.
+
+    The wrapper itself stays the command: ``sudo <new flag> rm -rf /`` is still
+    judged as ``sudo`` by the denylist, so nothing hides behind the guess.
+    """
+    scan = unwrap_command(["sudo", "--new-sudo-flag", "rm", "-rf", "/"])
+    assert scan.base == "sudo", scan
+    assert scan.wrappers == ()
+    assert "new-sudo-flag" in scan.unparsed
+    # A known option is read, not reported.
+    assert unwrap_command(["sudo", "-E", "rm", "-rf", "/"]).unparsed == ""
+
+
+def test_unwrap_remote_command_still_reviews_what_runs() -> None:
+    """The remote slice is parsed like a local command (code, escalation)."""
+    scan = unwrap_command(["ssh", "prod", "bash", "-c", "rm -rf /"])
+    assert scan.base == "bash"
+    assert dash_c_code(scan.command) == (1, "rm -rf /")
+    assert escalators_in_chain(["ssh", "prod", "sudo", "rm", "-rf", "/"]) == ("sudo",)
+
+
 def test_is_interpreter_ignores_version_suffix() -> None:
    assert is_interpreter("python")
    assert is_interpreter("python3.12")
@@ -160,6 +160,58 @@ def test_versioned_interpreter_is_still_reviewed() -> None:
    assert "import os" in reason


+def test_a_remote_command_is_judged_by_what_runs_there() -> None:
+    """``ssh``/``mosh`` do not hide the program from the review."""
+    # The remote slice decides, exactly like a local command.
+    plain = _reason("run_argv", {"argv": ["ssh", "prod", "uptime"]})
+    assert plain is None
+    code = _reason("run_argv", {"argv": ["ssh", "prod", "bash", "-c", "rm -rf /"]})
+    assert code is not None
+    assert "bash -c" in code
+    assert "rm -rf" in code
+    # And it is the same confirm through a PTY session.
+    pty = _reason("open_pty", {"command": ["ssh", "prod", "python3"]})
+    assert pty is not None and "python3" in pty
+
+
+def test_an_unreadable_option_asks_instead_of_guessing() -> None:
+    """An option the spec does not know makes the wrapped program a guess."""
+    reason = _reason("run_argv", {"argv": ["sudo", "--new-sudo-flag", "uptime"]})
+    assert reason is not None
+    assert "options not understood" in reason
+    assert "new-sudo-flag" in reason
+    # A known option is read: no review for a harmless wrapped program.
+    assert _reason("run_argv", {"argv": ["sudo", "-E", "uptime"]}) is None
+
+
+def test_a_remote_shell_is_announced() -> None:
+    """``ssh host`` runs nothing this scan can see, so the human is told first."""
+    reason = _reason("open_pty", {"command": ["ssh", "prod"]})
+    assert reason is not None
+    assert "remote shell on 'prod'" in reason
+    assert "input is not checked" in reason
+    # The same for a one-shot client: stdin is another unchecked input channel.
+    assert "remote shell on 'prod'" in (_reason("run_argv", {"argv": ["ssh", "-p", "2222", "prod"]}) or "")
+    # A mosh shell counts too, and a hop with nothing after it is not silent.
+    assert "remote shell" in (_reason("open_pty", {"command": ["mosh", "prod"]}) or "")
+    # A remote *command* keeps its own review (no shell wording).
+    assert "remote shell" not in (_reason("run_argv", {"argv": ["ssh", "prod", "uptime"]}) or "")
+
+
+def test_batch_mode_is_always_reviewed() -> None:
+    """``BatchMode=yes`` cannot prompt at either end, so the command is shown."""
+    reason = _reason("run_argv", {"argv": ["ssh", "-o", "BatchMode=yes", "prod", "deploy.sh"]})
+    assert reason is not None
+    assert "BatchMode=yes" in reason
+    assert "deploy.sh" in reason
+    assert _danger("run_argv", {"argv": ["ssh", "-oBatchMode=yes", "prod", "deploy.sh"]}).yolo_skip is True
+    # An escalation behind the hop still never skips the confirm under YOLO.
+    escalated = _danger("run_argv", {"argv": ["ssh", "prod", "sudo", "python3", "-c", "x=1"]})
+    assert escalated.yolo_skip is False
+    assert "escalation sudo" in escalated.text
+    assert "interpreter 'python3'" in escalated.text
+
+
 def test_wrappers_reach_the_interpreter() -> None:
    for argv in (
        ["env", "FOO=1", "python", "-c", "print(1)"],
@@ -10,7 +10,7 @@ from typing import TYPE_CHECKING, override
 import pytest

 from plyngent.agent import ToolOutput
-from plyngent.tools.context import read_only_context
+from plyngent.tools.context import get_instance, read_only_context
 from plyngent.tools.danger import classify_danger
 from plyngent.tools.net import fetch_local, grant_private_host
 from plyngent.tools.net.policy import (
@@ -225,6 +225,24 @@ async def test_an_image_response_comes_back_as_the_picture(workspace: Path, http
    assert part.image_url.url.startswith("data:image/png;base64,")


+async def test_a_text_only_model_keeps_the_metadata_and_the_reason(
+    workspace: Path,
+    http_server: str,
+) -> None:
+    """The picture stays behind, and the model is told why (``image_in``)."""
+    grant_private_host("127.0.0.1", int(http_server.rsplit(":", 1)[1]))
+    instance = get_instance()
+    assert instance is not None
+    instance.workspace.image_in = False
+    instance.workspace.image_model = ("ds", "text-only")
+
+    output = await fetch_local.handler(f"{http_server}/png")
+
+    assert isinstance(output, str)
+    assert "body_kind=image" in output
+    assert "error: model 'text-only' does not accept image input" in output
+
+
 async def test_fetch_read_only_context_blocks_non_get(workspace: Path, http_server: str) -> None:
    """In a read-only context fetch_local only allows GET; outside it POST works."""
    del workspace
@@ -6,6 +6,7 @@ import base64
 from typing import TYPE_CHECKING

 from plyngent.agent import ToolOutput
+from plyngent.tools.context import get_instance
 from plyngent.tools.file.read import read_file
 from plyngent.tools.file.view_image import view_image

@@ -60,6 +61,21 @@ async def test_a_missing_file_is_an_error(workspace: Path) -> None:
    assert await view_image.handler("nope.png") == "error: file not found: nope.png"


+async def test_a_text_only_model_gets_the_reason_instead_of_the_picture(workspace: Path) -> None:
+    """``image_in`` gates the tool exactly like an ``@path`` mention."""
+    (workspace / "shot.png").write_bytes(PNG)
+    instance = get_instance()
+    assert instance is not None
+    instance.workspace.image_in = False
+    instance.workspace.image_model = ("deepseek", "deepseek-v4-pro")
+
+    result = await view_image.handler("shot.png")
+
+    assert isinstance(result, str)
+    assert result.startswith("error: model 'deepseek-v4-pro' does not accept image input")
+    assert 'set image_in = true under [providers.deepseek.models."deepseek-v4-pro"]' in result
+
+
 async def test_read_file_points_at_view_image_for_a_picture(workspace: Path) -> None:
    (workspace / "shot.png").write_bytes(PNG)

@@ -9,10 +9,12 @@ from plyngent.tools import (
    check_command_allowed,
    get_trash_enabled,
    get_workspace_root,
+    image_input_refusal,
    mode_covers,
    parse_access_mode,
    resolve_path,
    set_command_denylist,
+    set_image_in,
    set_path_denylist,
    set_trash_enabled,
 )
@@ -67,6 +69,34 @@ def test_trash_switch(workspace: object) -> None:
    assert get_trash_enabled() is True


+def test_image_switch(workspace: object) -> None:
+    """A picture is refused once the host says the model cannot take one."""
+    del workspace
+    assert image_input_refusal() is None
+
+    set_image_in(image_in=False)
+    unnamed = image_input_refusal()
+    assert unnamed is not None
+    assert "does not accept image input" in unnamed
+
+    set_image_in(image_in=False, provider="ds", model="text-only")
+    named = image_input_refusal()
+    assert named is not None
+    assert "model 'text-only' does not accept image input" in named
+    assert '[providers.ds.models."text-only"]' in named
+
+    set_image_in(image_in=True, provider="ds", model="vision")
+    assert image_input_refusal() is None
+
+
+def test_the_image_switch_needs_a_declaration(workspace: object) -> None:
+    """No host verdict (``None``) sends the picture: only ``False`` refuses."""
+    del workspace
+    policy = active_workspace_policy()
+    assert policy.image_in is None
+    assert image_input_refusal() is None
+
+
 def test_config_grant_modes(workspace: object, tmp_path_factory: pytest.TempPathFactory) -> None:
    del workspace
    outside = tmp_path_factory.mktemp("granted")
@@ -282,6 +312,26 @@ def test_command_denylist_reaches_through_wrappers(workspace: object) -> None:
        clear_policy_allowed_commands()


+def test_command_denylist_reaches_the_far_end_of_ssh(workspace: object) -> None:
+    """A remote command is judged by the program that runs there, not by ``ssh``."""
+    del workspace
+    from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
+
+    set_policy_confirm_hook(None)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["ssh", "prod", "rm", "-rf", "/"])
+        with pytest.raises(WorkspaceError, match="basename 'sudo' is blocked"):
+            check_command_allowed(["ssh", "-p", "2222", "prod", "sudo", "systemctl", "restart", "web"])
+        # A remote shell (no command) and harmless remote commands stay allowed.
+        check_command_allowed(["ssh", "prod"])
+        check_command_allowed(["ssh", "-o", "BatchMode=yes", "prod", "uptime"])
+        check_command_allowed(["mosh", "prod", "--", "htop"])
+    finally:
+        clear_policy_allowed_commands()
+
+
 def test_command_denylist_matches_lowercased_basename(workspace: object) -> None:
    del workspace
    from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
Author	SHA1	Message	Date
NCBM	d4485aff7a	release: plyngent 0.12.0	2026-10-23 10:53:08 +08:00
NCBM	0244a48998	docs: say how wrapper options are read The destructive-confirm bullet described the old hand tables. It now names the getopt spec per wrapper, and the rule that follows from it: an option the spec does not know stops the unwrap and is reviewed, while the wrapper itself stays the command the denylist judges.	2026-10-23 10:47:55 +08:00
NCBM	c1b40a7bf4	test: cover options read with getopt `--` ends a wrapper's options, jammed values (`-p2222`) and launcher flags (`uv run --no-project pytest`) resolve, and an unknown option stops the unwrap with the wrapper as the command — where the review asks and the denylist still sees `sudo`.	2026-10-23 10:47:45 +08:00
NCBM	12236cffb5	core/tools: read each wrapper's options with getopt Four hand-written tables said which wrapper options take a value, and the comment above them admitted the failure mode: an entry for a flag eats the program behind it, which is how `exec -c rm -rf /` hid `rm`. Guessing was the other half — an option nobody listed was skipped bare, so the token after it became the program. Each wrapper now carries a `getopt` spec instead (`command_scan._WRAPPERS`: short options with `:`, long options with `=`), which spells flags and values apart, understands jammed forms (`-p2222`, `--with=pkg`), `--`, and unique long prefixes, and stops at the first token that is not an option — exactly the wrapped program. The remote clients and the `run` launchers are the same thing with a flag on the spec. An option the spec does not know is no longer skipped as if it were a flag: the unwrap stops at that wrapper (`unparsed`) and the review asks — `options not understood sudo (cannot read its options: option --new-sudo-flag not recognized)`, with the whole argv. The wrapper itself stays the command the denylist judges, so a new `sudo` flag cannot hide `rm` from the hard gate.	2026-10-23 10:47:36 +08:00
NCBM	3cb4d490ed	core/tools: announce a remote shell instead of judging it `ssh prod` runs no command this scan can resolve, and what is typed or piped into the shell it opens never reaches any gate — `write_pty` and `run_argv(stdin=…)` are both unchecked input. The tool description said nothing about it, so the call looked like any other. The hop is now reported even when no command follows (`RemoteTarget.shell`), and the review announces it: `remote shell on 'prod' (ssh with no command; its input is not checked)`. It is a soft confirm like the other reviews (YOLO may skip it, an escalation behind a hop still may not) — the client itself stays outside the denylist, since there is no program to name.	2026-10-23 10:43:05 +08:00
NCBM	e1c857fd05	docs: record how far the command gate reaches The destructive-confirm bullet says what the review resolves past, so it now names remote clients: the far-end program is the one judged, `BatchMode=yes` always asks, and an interactive remote shell — `ssh host` with no command, or anything typed or piped into it — is outside the scan and stays a host decision.	2026-10-23 10:37:23 +08:00
NCBM	8c289db545	test: cover the far end of ssh What the scan resolves (destination, options, jammed `-p2222`, mosh's `--`, a hop behind `sudo`), what it leaves alone (a client with no command), and both gates: `check_command_allowed` now refuses `ssh prod rm -rf /` and prompts for the `sudo` behind the hop, while harmless remote commands stay allowed; the review prints a remote `bash -c` code block and always asks under `BatchMode=yes`.	2026-10-23 10:37:14 +08:00
NCBM	0cb740393e	core/tools: judge an ssh command by what runs on the far end The gate judged `ssh` itself: `unwrap_command` knew env wrappers and `<launcher> run`, not a remote client, so `ssh prod rm -rf /` reached `check_command_allowed` as "ssh" — not denylisted — and `classify_danger` found neither an interpreter nor a detached run. The command ran unannounced. `ssh` / `mosh` are wrappers now: their own options are skipped, the first token that is not an option is the destination, and the program under review is the one after it (after `--` for mosh). The hop is reported on the scan (`RemoteTarget`), so `ssh prod bash -c …` is reviewed with the code it carries, an escalation behind the hop keeps its confirm through YOLO, and the denylist matches the remote program (`rm`, `sudo`, `systemctl reboot`). `-o BatchMode=yes` cannot prompt at either end, so a command running with it is always confirmed, whatever program it is. A client with no command only opens a remote shell: nothing to judge at spawn time, and what is typed or piped into that shell never reaches the scan.	2026-10-23 10:37:04 +08:00
NCBM	d5c3f712ff	docs: describe when a prompt listens for the answer The page said a notification answer is treated like a typed one, without saying that the prompt has to be listening before the notification exists — the grace period is why. One paragraph says it, together with the price: a question that offers a reply field reads in polling mode, so no Tab completion for that prompt.	2026-10-23 10:25:40 +08:00
NCBM	16c5e01e44	test: cover answers from the notification Three seams: the offer is known when the wait starts (a confirm and a question, with and without the platform capability), a button pressed after the notification appears answers the tool confirm it belongs to — a pty stands in for the terminal so the read has to consult both surfaces — and `confirm()` takes an answer passed in and one offered by the wait hook, without asking the backend.	2026-10-23 10:24:56 +08:00
NCBM	80ad34f8fe	core/cli+prompting: hear the notification while a prompt waits Whether a prompt could be answered from its notification was decided by `notify_desktop`, which only runs once the grace period is over — by then the prompt had already committed to the plain terminal read. A pressed button or a typed reply landed on a queue nobody read, and the human had to come back to the terminal to answer the prompt the notification was showing. A wait now asks up front whether an answer may arrive (`answer_capability`: the kind, `actions`, the channel and the platform's capabilities) and listens on both surfaces from its first poll. The desktop channel is resolved once, in the background from `install_notifier`, so the capability probe never lands on a prompt that is already waiting. `confirm()` takes an optional `source` (the form's submit confirm and every limits.py confirmation go through it), which also leaves one place — not two — parsing `allow` / `deny` into a yes or no.	2026-10-23 10:24:46 +08:00
NCBM	b2d6acae26	docs: record the Files API shapes Uploading a picture and referencing it by id is the one image path the API offers that we do not use, and the vendor shapes do not line up: the upload is the same multipart `POST …/files` with a `file` part everywhere, but the object that comes back is `object`/`bytes`/unix time on OpenAI, `type`/`size_bytes`/ RFC 3339 on Anthropic, and DeepSeek serves both dialects under two base URLs; the reference block differs a third time per surface, and OpenAI's chat completions cannot name an image by id at all. Recorded next to the inline-path limits, with the three constraints that decide any future wiring: an id belongs to the provider that minted it (like `provider_blocks`), the OpenAI path implies the Responses convention, and the token bill is unchanged — uploading only buys the inline size cap, a smaller request body, and not re-sending the same bytes every round. The Anthropic per-file limit is marked unverified: its API reference is unreachable from this host's region.	2026-10-21 21:44:03 +08:00
NCBM	8926980a02	docs: describe the image gate `image_in` is not just the mention switch any more: the table names where each image-carrying path is gated, and says why the gate exists — a text-only model answers `[Unsupported Image]`, and a gateway that cannot read image blocks feeds the model raw base64, which is easy to mistake for a delivered picture.	2026-10-21 21:39:41 +08:00
NCBM	37e9395ac5	test: cover the image gate The gate has four entry points and each now has a test: `view_image` and `fetch_local` refuse with the message naming the config key, the registry drops a plugin tool's picture and says why, and the workspace helpers keep `None` (no host verdict) permissive. On the CLI side the verdict is asserted to follow the selected model, and to reach the tools once an instance is bound.	2026-10-21 21:39:33 +08:00
NCBM	da6f4b7dd7	core/tools+agent+cli: refuse a picture the model cannot take `[providers..models.].image_in` gated only the `@path` mention: `view_image` attached whatever it was asked for, and a tool result's picture (an MCP server's `image` content, a plugin's `ToolOutput`) went out the same way. On a model that does not read pictures the provider answers `[Unsupported Image]` one round later, and a gateway that cannot translate image blocks hands the model raw base64 — the model then reports "I can only see a base64 payload", which reads like the picture arrived when it did not. The verdict now lives on `InstanceState.workspace` (`image_in` + the `(provider, model)` it came from) and `tools.workspace.image_input_refusal` turns it into the one message every path shares: error: model 'deepseek-v4-pro' does not accept image input (set image_in = true under [providers.deepseek.models."deepseek-v4-pro"]) `view_image` and `fetch_local` refuse with it before reading any bytes (the fetch keeps its metadata line, so `body_kind=image` and the size stay visible); the registry drops an image from a `ToolOutput` it did not vouch for and appends the same reason to the text. That last one is the safety net for MCP and plugin tools, which know nothing about the model. `ReplState.sync_image_capability` publishes the verdict and runs from `rebuild_client`, so `/provider` and `/model` flip the gate immediately, and `_run_chat` seeds it before the first turn. A host that binds no instance declares nothing and keeps sending pictures, as before.	2026-10-21 21:39:24 +08:00