docs: document SIGINT routing and MCP process-group isolation

Ctrl+C now cancels the turn (or the auto-retry countdown), cancels a poll-based
prompt read such as wait's, and is ignored with a hint for off-loop readline
prompts; it never exits the REPL and never reaches MCP servers, which spawn in
their own session. Also drop ask_into_pty's stale "Ctrl+C cancels" claim: that
prompt is readline-based and empty input is its cancel.

AGENTS.md

@@ -74,7 +74,7 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `
 - **`responses_bridge` / `responses_dispatch`**: chat ↔ OpenAI Responses; stream text/reasoning deltas; tool calls + usage on `response.completed`.
 - **`messages_bridge` / `messages_dispatch`**: chat ↔ Anthropic Messages; system fold; tool_use/tool_result; stream text + tool argument fragments.
 - **Provider-side tools**: `OpenAIProvider.provider_tools` (list of dicts; default `[{type="web_search"}]` when omitted; `[]` disables) threaded via `ChatAgent`/`run_chat_loop` into Responses only; never executed by `ToolRegistry`.
-- **`@tool` / `ToolRegistry`**: decorator infers JSON Schema from type hints; execute tools by name.
+- **`@tool` / `ToolRegistry`**: decorator infers JSON Schema from type hints; execute tools by name. 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).
 - **`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`.
 - **`ChatAgent`**: optional `MemoryStore` (user message persisted immediately; **completed tool batches checkpointed** mid-turn; unfinished assistant suffix rolled back on failure); `stream`; system prompt; `provider_tools`; `retry()` continues incomplete turns (user-only **or** after committed tools — does not re-run those tools).
 - **`/compact`**: soft-compact tool dumps → model summary (no tools) via the same kind dispatch → **new** session seeded with summary message. Soft-compact keeps the last **12** tool results full-size by default (`DEFAULT_RECENT_TOOL_RESULTS`); older tool payloads shrink first.
@@ -94,7 +94,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 - 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 shells/REPLs and `python|bash -c` one-liners (argv + code preview); deny may include a user comment for the model. Session YOLO: `/yolo on|off|once` and `--yes` (skip soft confirms; hard denylists unchanged; `once` expires after the next user turn).
 - **`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).
+- **`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 id; 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`).
 - **`net` / `fetch`**: HTTP GET/POST/PUT/DELETE via niquests (isolated session, not the LLM client). Manual redirects with per-hop SSRF checks (`asyncio` `getaddrinfo`). Private/loopback hosts need instance policy grant (CLI timed confirm; **not** YOLO). Public cleartext HTTP and mutating methods soft-confirm (`YOLO`/`TRUSTABLE`). HTTPS→HTTP redirects blocked unless `allow_http_downgrade`. `user_agent` tool arg (or headers) is never replaced by the default when provided. Caps: body bytes/chars, request body size.
 - **`mcp`** (`tools/mcp.py`): register one server's advertised tools into the catalog as namespaced `mcp__<server>__<tool>` definitions with `ToolSource(kind="mcp", plugin_id=server)`; tags LOCAL (+ READ_ONLY when the server config sets `read_only`) — never YOLO/TRUSTABLE. `register_mcp_tools` purges prior MCP entries then re-registers every connected server (hosts call it on tool-registry rebuild); connections must already be started by `McpManager`.
@@ -108,12 +108,12 @@ Shared interactive I/O: `ask` / `choose` / `form` / `confirm` with pluggable bac
 
 Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
 
-- **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--auto-continue` (auto-raise limits), `--stream/--no-stream`, `--quiet`. Root `--log-level`. Enabled `[mcp]` servers start before the REPL (`_start_mcp_manager`, per-server failures warn on stderr) and are closed on exit (`McpManager.aclose`).
+- **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--auto-continue` (auto-raise limits), `--stream/--no-stream`, `--quiet`. Root `--log-level`. Enabled `[mcp]` servers start before the REPL (`_start_mcp_manager`, per-server failures warn on stderr) and are closed on exit (`McpManager.aclose`); each server spawns in its own session/process group (`start_new_session` / `CREATE_NEW_PROCESS_GROUP`), so a terminal Ctrl+C does not kill it.
 - Slash: Click group in `cli/slash.py` + `awaitlet` for async work; Tab completer from registry + ParamType `shell_complete`. Multiline `"""` … `"""`; `/edit` via `$VISUAL`/`$EDITOR` (blocking only). `/yolo on|off|once` for soft destructive confirms. `/model --persist` / `/models --persist` write model catalog entries into TOML. `/todos` for human show/push/pop/clear of the todo stack. `/mcp` (`list` default) prints per-server status/tool counts + an `instructions` preview when the server sent usage guidance, + last stderr lines; `/mcp reconnect` re-reads config, restarts every enabled server, and rebuilds the tool registry. `/grants` lists directory-access grants (`revoke <index|all>`; config grants point at the TOML); `/status` includes the grant count.
 - `/btw` side questions: forked session; main transcript/DB untouched. `--tools=read` (default) exposes only `READ_ONLY` tools under `read_only_context()` (fetch=GET) with the main session view; `--tools=no` off; `--tools=full` whole-registry clone with fresh session. Tools off: default falls back to no tools; explicit `read`/`full` errors. The model gets an `aside` notice as its last message before the question (unsaved exchange + tool scope), shown to the user as `[notice] aside: …`.
 - Explicit `/resume` or `--session` from another workspace prompts: **keep** / **update** / **abort**. A session resumed by a **freshly started process** (`--session` / default latest) also gets a `resume` notice about what a restart drops (PTY sessions, running commands, temp workspaces) — resuming inside a live process pushes nothing.
 - Host notices (`agent/notices.py`): system-level messages appended as trailing `developer` rows, never system-prompt edits — `Notice.to_message()` for the model, `.summary` for the user via `cli.display.echo_notice` (`--quiet` hides the user line only). Durable hosts persist them (`ChatAgent.push_notice`); a side agent without memory keeps its notice local. Trailing developer messages never end a turn, so `/retry` still sees an interrupted turn behind a notice or checkpoint.
-- Failed/cancelled turns: user kept; **committed tool rounds kept** (side effects not re-run on `/retry`); only unfinished assistant rolled back; Ctrl+C cancels; TTY confirms off-loop; auto-retry waits 5s/10s/15s/20s then +10s (10 attempts) and the budget **resets when an attempt completes a model round** (connection recovered); then `/retry`.
+- Failed/cancelled turns: user kept; **committed tool rounds kept** (side effects not re-run on `/retry`); only unfinished assistant rolled back; Ctrl+C cancels the in-flight turn (a session-scoped asyncio SIGINT handler routes it, `cli/interrupt.py` — a stray Ctrl+C while the loop waits never tears the REPL down); TTY confirms/ask prompts run off-loop and register their own SIGINT target, so Ctrl+C there cancels just the poll-based read (`wait` reports `cancelled by user`) or is ignored with a hint for readline prompts (`ask_user_*`, confirms) — never the turn, never the loop; auto-retry waits 5s/10s/15s/20s then +10s (10 attempts), its countdown is Ctrl+C-cancellable, and the budget **resets when an attempt completes a model round** (connection recovered); then `/retry`.
 - Pretty tool lines: known tools (`read_file`, `run_argv`, `grep_files`, …) print their `* Verb 'target' ` prefix as the call starts and append the outcome (`L1-4 (done)`, `(exit code 0)`) when the result lands, so a blocking tool stays visible while it runs. Interactive TTY only (`interactive_terminal()`): piped/plain output and `--verbose` keep whole lines / `[tool ok]`. A second call before the first result (parallel batch) erases the open prefix and falls back to whole lines.
 - **`plyngent providers`**, **`config path|edit`**. Config open: `$VISUAL`/`$EDITOR` (wait), else system default (`xdg-open` / `open` / `startfile`, non-blocking). `/edit` stays blocking-only. No providers → optional edit then reload when waited.
 - Tools default on; workspace defaults to cwd; `--max-rounds` default 32. Readline history under platformdirs (`repl_history`). PTY: `close_all` on chat exit.

README.md

@@ -278,6 +278,8 @@ Type `/help` in the REPL for the live list. Common ones:
 
 User messages are saved immediately. On API error or Ctrl+C, partial assistant/tool output is discarded but the user message stays so `/retry` works after resume. Interactive auto-retry waits 5s, 10s, 15s, 20s, then +10s each step (10 attempts); an attempt that gets a full model round back **resets that budget**, so a connection that recovers and drops again starts over from 5s.
 
+Ctrl+C cancels the in-flight turn, and during the auto-retry countdown it cancels the retry. Model-initiated prompts run off the event loop: a `wait` prompt is cancelled by Ctrl+C (the tool reports `cancelled by user` and the turn continues), while `ask_user_*` and confirm prompts ignore it with a one-line hint — answer them to continue. Ctrl+C never exits the REPL (use Ctrl+D or `/quit`), and MCP servers run in their own process group, so it never kills them either.
+
 ## Workspace model
 
 - **Workspace** = root for file/process/VCS tools (default cwd).
@@ -300,6 +302,8 @@ Safety defaults:
   `write_pty` is literal text only; use `write_pty_keys` for `\xHH`, `ctrl+x`, `key=esc|enter|…`.
   For passwords/sudo/ssh prompts use `ask_into_pty` (human types locally; answer never returns to the model).
 
+- Mis-typed tool arguments are answered, never guessed: a string where an array is expected (`run_argv` `argv`, `ask_user_choice` `options`, `ask_user_form` `fields`), a call that omits a required argument, or a non-numeric `wait` duration (numeric strings are accepted) returns an error showing the shape/names to send — the human never sees a garbage prompt, and no confirm prompt fires for a call that cannot run.
+
 ## Usage / context (CLI)
 
 - **Context size** prefers API `prompt_tokens` from the last model call; otherwise a char-based estimate (~4 chars/token).

src/plyngent/agent/tools.py

@@ -201,6 +201,30 @@ def tool[**PS, R](
     return decorator
 
 
+def _missing_required_error(definition: ToolDefinition, args: Mapping[str, object]) -> str | None:
+    """Model-facing error when a call omits a required argument.
+
+    The JSON schema already declares what is required, so the model gets the
+    exact argument names back instead of the handler's bare ``TypeError``
+    (``missing 1 required positional argument``). Unknown *extra* arguments are
+    left to the handler: a declared schema is not proof that no other key is
+    accepted (MCP servers may document only part of their input).
+    """
+    schema = definition.parameters
+    required = schema.get("required")
+    if not isinstance(required, list):
+        return None
+    missing = [name for name in cast("list[object]", required) if isinstance(name, str) and name not in args]
+    if not missing:
+        return None
+    properties = schema.get("properties")
+    accepted: list[str] = []
+    if isinstance(properties, dict):
+        accepted = sorted(str(key) for key in cast("dict[object, object]", properties))
+    hint = f"; it accepts: {', '.join(accepted)}" if accepted else ""
+    return f"error: `{definition.name}` is missing required argument(s): {', '.join(missing)}{hint}"
+
+
 class ToolRegistry:
     """Name → tool definition map with execution helpers."""
 
@@ -434,6 +458,11 @@ class ToolRegistry:
         if not isinstance(raw_args, dict):
             return "error: tool arguments must be a JSON object"
         args = {str(key): value for key, value in cast("dict[object, object]", raw_args).items()}
+        # Argument shape/coverage problems are answered before any confirm prompt:
+        # the human should never be asked about a call that cannot run.
+        missing_error = _missing_required_error(definition, args)
+        if missing_error is not None:
+            return missing_error
 
         async def _run() -> str:
             missing = self._check_state_tags(definition)

src/plyngent/cli/app.py

@@ -17,7 +17,11 @@ from plyngent.cli.editor import (
     resolve_config_path,
 )
 from plyngent.cli.exit_codes import EXIT_CANCELLED, EXIT_OK, EXIT_TURN_FAILED
-from plyngent.cli.interrupt import install_keyboard_interrupt_sigint
+from plyngent.cli.interrupt import (
+    install_keyboard_interrupt_sigint,
+    install_sigint_router,
+    uninstall_sigint_router,
+)
 from plyngent.cli.limits import install_cli_limit_hooks
 from plyngent.cli.repl import run_repl
 from plyngent.cli.retry import run_user_text_with_retries
@@ -251,6 +255,11 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
     yolo: YoloMode | None = "on" if yes else None
 
     memory = await MemoryStore.open(_database_config(store, quiet=quiet or oneshot))
+    # A SIGINT must never surface as a KeyboardInterrupt raised in the event
+    # loop's own frame (it would tear this run — and the shutdown below — down
+    # half-finished). The router cancels the innermost registered target instead
+    # and stays installed through cleanup.
+    _ = install_sigint_router()
     try:
         from plyngent.cli.provider_recovery import ensure_provider_ready
 
@@ -402,6 +411,7 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
             from plyngent.tools.process.pty_session import PtyManager
 
             PtyManager.close_all()
+        uninstall_sigint_router()
 
 
 def _configure_logging(level: str) -> None:

src/plyngent/cli/display.py

@@ -444,11 +444,13 @@ def _wait_prefix(_args_json: str) -> str:
 
 
 def _wait_detail(_args_json: str, result: str) -> str:
-    """``wait`` outcome: slept duration, disturbed, or error."""
+    """``wait`` outcome: slept duration, disturbed, cancelled, or error."""
     if result.startswith("error:"):
         return _pretty_segments((f"({result})", "red"))
     if result.startswith("waited "):
         return _pretty_segments((f"({result.removeprefix('waited ')})", "green"))
+    if result.startswith("cancelled by user"):
+        return _pretty_segments(("(cancelled)", "yellow"))
     return _pretty_segments(("(disturbed)", "yellow"))
 
 

src/plyngent/cli/interrupt.py

@@ -4,7 +4,10 @@ import asyncio
 import contextlib
 import signal
 from contextlib import contextmanager
-from typing import TYPE_CHECKING
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Protocol
+
+import click
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator
@@ -12,21 +15,17 @@ if TYPE_CHECKING:
 
     type SigHandler = Callable[[int, FrameType | None], None] | int | None
 
-# Depth of nested pause_task_cancel_for_prompt. Must be a plain int (not ContextVar):
-# asyncio.add_signal_handler freezes the ContextVar snapshot at install time, so a
-# ContextVar reset after reinstall would leave SIGINT permanently non-cancelling.
-_prompt_depth: int = 0
-_reinstall_holder: list[Callable[[], None] | None] = [None]
 
+class InterruptTarget(Protocol):
+    """Something a pending SIGINT can cancel (a task, a prompt read, a hint)."""
 
-def allow_task_cancel() -> bool:
-    """Whether the CLI SIGINT handler should cancel the in-flight turn task."""
-    return _prompt_depth == 0
+    def cancel(self) -> object:
+        """Cancel the pending work; the return value is ignored.
 
-
-def set_sigint_reinstall(callback: Callable[[], None] | None) -> None:
-    """Register how to re-bind SIGINT after a TTY prompt (set by run_cancellable)."""
-    _reinstall_holder[0] = callback
+        Declared loosely (``object``) so ``asyncio.Task.cancel`` — which returns
+        ``bool`` and takes an optional message — satisfies it.
+        """
+        ...
 
 
 def raise_keyboard_interrupt_sigint(signum: int, frame: FrameType | None) -> None:
@@ -56,57 +55,172 @@ def install_keyboard_interrupt_sigint() -> None:
         _ = signal.signal(signal.SIGINT, raise_keyboard_interrupt_sigint)
 
 
+# --------------------------------------------------------------------------- #
+# SIGINT routing (session scope)
+# --------------------------------------------------------------------------- #
+
+
+@dataclass(slots=True)
+class _RouterState:
+    """Session SIGINT routing state (one instance, kept in ``_router``)."""
+
+    # Cancel targets, innermost last: a SIGINT cancels the innermost one.
+    targets: list[InterruptTarget] = field(default_factory=list)
+    installed: bool = False
+    # Nesting depth of main-thread blocking reads. A plain counter (not a
+    # ContextVar): asyncio.add_signal_handler freezes the ContextVar snapshot at
+    # install time, so a ContextVar reset after reinstall would leave SIGINT
+    # permanently routed to the wrong place.
+    blocking_reads: int = 0
+
+
+_router = _RouterState()
+
+
+def sigint_routed() -> bool:
+    """Whether the session SIGINT router currently owns SIGINT."""
+    return _router.installed
+
+
+def install_sigint_router() -> bool:
+    """Install the session SIGINT handler; False when the loop cannot take it.
+
+    A SIGINT delivered while the event loop is parked in its selector is raised
+    in the *loop's* frame: it escapes ``run_until_complete``, the Runner cancels
+    the main task and the REPL (or a half-finished shutdown) is left broken —
+    the next Ctrl+C then kills the process. Keeps an asyncio-level SIGINT
+    handler installed for the whole session instead: it cancels the innermost
+    target registered with :func:`sigint_cancels` and never raises.
+
+    Blocking main-thread reads suspend the router via
+    :func:`pause_task_cancel_for_prompt`. Platforms without asyncio signal
+    handlers (Windows proactor) return False and keep the plain
+    KeyboardInterrupt behaviour. Idempotent.
+    """
+    if _router.installed:
+        return True
+    if not _add_router_handler():
+        return False
+    _router.installed = True
+    return True
+
+
+def uninstall_sigint_router() -> None:
+    """Drop the session SIGINT handler and restore the KeyboardInterrupt one."""
+    if not _router.installed:
+        return
+    _router.installed = False
+    _remove_router_handler()
+    _router.targets.clear()
+    install_keyboard_interrupt_sigint()
+
+
+def _add_router_handler() -> bool:
+    try:
+        asyncio.get_running_loop().add_signal_handler(signal.SIGINT, _route_sigint)
+    except RuntimeError, NotImplementedError, ValueError:
+        return False
+    return True
+
+
+def _remove_router_handler() -> None:
+    with contextlib.suppress(NotImplementedError, RuntimeError, ValueError):
+        _ = asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)
+
+
+def _route_sigint() -> None:
+    """Cancel the innermost target (none → ignore: never escape the loop)."""
+    if _router.targets:
+        _ = _router.targets[-1].cancel()
+
+
+@contextmanager
+def sigint_cancels(target: InterruptTarget) -> Generator[None]:
+    """Route SIGINT to ``target.cancel()`` while inside (innermost wins)."""
+    _router.targets.append(target)
+    try:
+        yield
+    finally:
+        for index in range(len(_router.targets) - 1, -1, -1):
+            if _router.targets[index] is target:
+                del _router.targets[index]
+                break
+
+
 @contextmanager
 def pause_task_cancel_for_prompt() -> Generator[None]:
-    """Disable turn-task cancel during blocking TTY prompts (confirm, etc.).
+    """Give SIGINT to a blocking *main-thread* read instead of routing it.
 
-    Must run on the main thread (signal handlers are main-thread only).
-    Restores the default SIGINT handler so ``click.confirm`` can receive
-    KeyboardInterrupt / Abort instead of the asyncio turn being cancelled.
+    The read (REPL entry prompt, synchronous click confirm/prompt) runs on the
+    main thread inside a coroutine, so the resulting KeyboardInterrupt is raised
+    in the reader's own frame: its handlers see it, and neither the in-flight
+    turn nor the event loop is cancelled. Use it only around blocking reads — an
+    ``await`` that parks the loop in its selector must keep the router installed
+    (:func:`install_sigint_router`).
 
-    Nested prompts are supported via a depth counter. The asyncio SIGINT
-    handler is reinstalled only after depth returns to 0, and only after
-    cancel is re-enabled, so the handler never freezes ``allow=False``.
+    Suspends the router; nested calls restore it when the outermost exits.
     """
-
-    global _prompt_depth  # noqa: PLW0603
-    _prompt_depth += 1
-    loop_handler_removed = False
+    _router.blocking_reads += 1
+    resuming = _router.blocking_reads == 1 and _router.installed
+    if resuming:
+        _remove_router_handler()
     previous: SigHandler = signal.SIG_DFL
     try:
-        try:
-            loop = asyncio.get_running_loop()
-            _ = loop.remove_signal_handler(signal.SIGINT)
-            loop_handler_removed = True
-        except RuntimeError, NotImplementedError, ValueError:
-            loop_handler_removed = False
-
-        try:
-            previous = signal.getsignal(signal.SIGINT)
-            _ = signal.signal(signal.SIGINT, signal.default_int_handler)
-        except ValueError:
-            # Not on the main thread — skip OS signal rebinding.
-            previous = signal.SIG_DFL
+        previous = signal.getsignal(signal.SIGINT)
+        _ = signal.signal(signal.SIGINT, raise_keyboard_interrupt_sigint)
+    except ValueError:
+        # Not on the main thread — signal handlers cannot be rebound here.
+        previous = signal.SIG_DFL
+    try:
         yield
     finally:
         with contextlib.suppress(ValueError):
             _ = signal.signal(signal.SIGINT, previous)
-        _prompt_depth = max(0, _prompt_depth - 1)
-        # Reinstall only when fully out of prompts and cancel is allowed again.
-        # Order matters: decrement depth first so allow_task_cancel() is True
-        # before reinstall (and handlers never freeze allow=False).
-        if _prompt_depth == 0 and loop_handler_removed:
-            reinstall = _reinstall_holder[0]
-            if reinstall is not None:
-                with contextlib.suppress(RuntimeError, NotImplementedError, ValueError):
-                    reinstall()
+        _router.blocking_reads = max(0, _router.blocking_reads - 1)
+        if resuming:
+            _ = _add_router_handler()
 
 
-async def run_in_prompt_thread[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
-    """Run a blocking TTY prompt off the event loop with cancel paused.
+@dataclass(frozen=True, slots=True)
+class _CancelHook:
+    """SIGINT target for a poll-based prompt read (abort that read)."""
 
-    Pause/SIGINT rebinding happens on the main thread; only the prompt body
-    runs in a worker so the asyncio loop stays free and turn cancel is disabled.
+    cancel_read: Callable[[], None]
+
+    def cancel(self) -> None:
+        self.cancel_read()
+
+
+@dataclass(slots=True)
+class _WaitingPrompt:
+    """SIGINT target for a prompt read that cannot be interrupted.
+
+    A readline read blocks inside the C library on a worker thread; aborting it
+    would abandon that thread on stdin, where it would eat the next line the
+    human types. Explaining itself once is the best available answer.
     """
-    with pause_task_cancel_for_prompt():
-        return await asyncio.to_thread(func, *args, **kwargs)
+
+    _hinted: bool = False
+
+    def cancel(self) -> None:
+        if self._hinted:
+            return
+        self._hinted = True
+        click.secho(
+            "Ctrl+C ignored while a prompt is waiting — answer it to continue",
+            fg="yellow",
+            err=True,
+        )
+
+
+@contextmanager
+def off_loop_prompt(cancel: Callable[[], None] | None = None) -> Generator[None]:
+    """A human prompt is waiting in a worker thread.
+
+    While inside, SIGINT must not cancel the enclosing turn: poll-based reads
+    (``cancel`` hook) abort just that read, readline-based ones ignore it with a
+    one-line hint (see :class:`_WaitingPrompt`).
+    """
+    target: InterruptTarget = _WaitingPrompt() if cancel is None else _CancelHook(cancel)
+    with sigint_cancels(target):
+        yield

src/plyngent/cli/limits.py

@@ -7,7 +7,7 @@ import sys
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
-from plyngent.cli.interrupt import pause_task_cancel_for_prompt
+from plyngent.cli.interrupt import off_loop_prompt, pause_task_cancel_for_prompt
 from plyngent.prompting import (
     ChoiceOption,
     NonInteractiveError,
@@ -521,10 +521,14 @@ async def prompt_workspace_mismatch_async(
 
 
 def install_cli_limit_hooks() -> None:
-    """Register interactive continue hooks and prompt cancel-pause.
+    """Register interactive continue hooks and prompt guards.
 
-    Command denylist policy confirm is set on :class:`~plyngent.tools.context.InstanceState`
-    when the REPL starts (no process-global hook).
+    Off-loop prompts (worker-thread reads) register their own SIGINT target so a
+    Ctrl+C there never cancels the enclosing turn; synchronous prompts read on
+    the main thread and take SIGINT as a KeyboardInterrupt instead. Command
+    denylist policy confirm is set on
+    :class:`~plyngent.tools.context.InstanceState` when the REPL starts (no
+    process-global hook).
     """
-    configure_prompting(pause_factory=pause_task_cancel_for_prompt)
+    configure_prompting(pause_factory=off_loop_prompt)
     PtyManager.set_limit_continue_hook(prompt_continue_limit)

src/plyngent/cli/repl.py

@@ -6,6 +6,7 @@ from typing import TYPE_CHECKING
 import click
 
 from plyngent.cli.input_text import read_repl_entry
+from plyngent.cli.interrupt import pause_task_cancel_for_prompt
 from plyngent.cli.readline_setup import setup_readline
 from plyngent.cli.retry import run_user_text_with_retries
 from plyngent.cli.slash import handle_slash
@@ -25,7 +26,7 @@ def _echo_user(text: str) -> None:
 
 def _echo_interrupted() -> None:
     # A further Ctrl+C during this output must stay benign (same rule as
-    # retry._echo_cancel_lines): after a turn the SIGINT handler is removed, so
+    # retry._echo_cancel_lines): on platforms without asyncio signal handlers
     # the next Ctrl+C raises KeyboardInterrupt at an arbitrary bytecode.
     with contextlib.suppress(KeyboardInterrupt):
         click.echo()
@@ -51,16 +52,22 @@ async def run_repl(state: ReplState) -> None:
     click.echo('Type /help for commands. Multiline: """ … """. Empty line is ignored.')
 
     while True:
+        # The prompt read blocks the main thread inside this coroutine, so hand
+        # SIGINT to it: Ctrl+C raises KeyboardInterrupt here (caught below and
+        # inside read_repl_entry) and re-prompts. Everywhere else the session
+        # SIGINT router owns the signal (turn cancel, prompt cancel).
         try:
-            entry = read_repl_entry()
+            with pause_task_cancel_for_prompt():
+                entry = read_repl_entry()
         except EOFError:
             click.echo()
             break
+        except KeyboardInterrupt:
+            # Stray interrupt outside the read (e.g. right after the router was
+            # suspended): re-prompt rather than leave the REPL.
+            _echo_interrupted()
+            continue
 
-        # Between turns the turn-task SIGINT handler is removed (run_cancellable),
-        # so a stray Ctrl+C raises KeyboardInterrupt at whatever bytecode is
-        # executing — echoing user text, expire_yolo_once, gaps between
-        # statements. Swallow it and re-prompt instead of exiting the REPL.
         try:
             if entry is None:
                 continue

src/plyngent/cli/retry.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
-import signal
 import time
 from typing import TYPE_CHECKING
 
@@ -10,7 +9,7 @@ import click
 
 from plyngent.agent import AssistantMessageEvent
 from plyngent.cli.display import render_events
-from plyngent.cli.interrupt import allow_task_cancel, set_sigint_reinstall
+from plyngent.cli.interrupt import install_sigint_router, sigint_cancels
 from plyngent.cli.limits import reset_auto_continue_turn
 
 if TYPE_CHECKING:
@@ -47,7 +46,9 @@ async def sleep_cancellable(seconds: float) -> bool:
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 return True
-            await asyncio.sleep(min(0.5, remaining))
+            sleeper = asyncio.ensure_future(asyncio.sleep(min(0.5, remaining)))
+            with sigint_cancels(sleeper):
+                await sleeper
     except asyncio.CancelledError:
         return False
     except KeyboardInterrupt:
@@ -57,56 +58,32 @@ async def sleep_cancellable(seconds: float) -> bool:
 async def run_cancellable[T](coro: Coroutine[object, object, T]) -> T:
     """Await ``coro`` as a task; Ctrl+C / SIGINT cancels the task.
 
-    Interactive prompts (max-rounds / destructive confirm) temporarily disable
-    task cancel so the user can answer instead of aborting the whole turn.
+    Interactive prompts (max-rounds / destructive confirm) register their own
+    SIGINT target for as long as they wait, so the turn is not cancelled while
+    the human is answering (see :func:`~plyngent.cli.interrupt.off_loop_prompt`).
 
     Raises:
         asyncio.CancelledError: If the task was cancelled (including via SIGINT).
     """
     task: asyncio.Task[T] = asyncio.create_task(coro)
-    loop = asyncio.get_running_loop()
-    installed = False
-    # What to restore once the turn-cancel handler is removed. remove_signal_handler
-    # leaves SIG_DFL, which would terminate the process on a stray Ctrl+C between
-    # turns; restore the handler from before (the CLI installs a
-    # KeyboardInterrupt-raising one so the REPL can catch it).
-    previous_sigint = signal.getsignal(signal.SIGINT)
-
-    def _on_sigint() -> None:
-        # allow_task_cancel() uses a process-level depth counter (not ContextVar)
-        # so this remains correct even if the handler was installed under a
-        # frozen context (asyncio signal handles capture contextvars).
-        if allow_task_cancel() and not task.done():
-            _ = task.cancel()
-
-    def _reinstall() -> None:
-        try:
-            loop.add_signal_handler(signal.SIGINT, _on_sigint)
-        except NotImplementedError, RuntimeError, ValueError:
-            return
-
+    # Keep a loop-level SIGINT handler in place for the rest of the session, so
+    # a Ctrl+C that lands while the loop is parked in its selector cancels this
+    # task instead of being raised in the loop's own frame.
+    _ = install_sigint_router()
     try:
-        loop.add_signal_handler(signal.SIGINT, _on_sigint)
-        installed = True
-        set_sigint_reinstall(_reinstall)
-    except NotImplementedError, RuntimeError, ValueError:
-        installed = False
-
-    try:
-        return await task
-    except KeyboardInterrupt:
-        if not task.done():
-            _ = task.cancel()
-            with contextlib.suppress(asyncio.CancelledError):
-                await task
-        raise asyncio.CancelledError from None
+        with sigint_cancels(task):
+            try:
+                return await task
+            except KeyboardInterrupt:
+                # Platforms without asyncio signal handlers (Windows proactor)
+                # keep the CLI's KeyboardInterrupt handler, so the interrupt
+                # surfaces here while the task is still awaiting.
+                if not task.done():
+                    _ = task.cancel()
+                    with contextlib.suppress(asyncio.CancelledError):
+                        await task
+                raise asyncio.CancelledError from None
     finally:
-        set_sigint_reinstall(None)
-        if installed:
-            with contextlib.suppress(NotImplementedError, RuntimeError, ValueError):
-                _ = loop.remove_signal_handler(signal.SIGINT)
-            with contextlib.suppress(ValueError):
-                _ = signal.signal(signal.SIGINT, previous_sigint)
         # Only cancel if still running (e.g. KeyboardInterrupt path above).
         # Do not cancel a finished task — that would mask success.
         if not task.done():
@@ -143,9 +120,10 @@ def _echo_turn_usage(agent: ChatAgent) -> None:
 def _echo_cancel_lines(*lines: str) -> None:
     """Print post-cancel lines, treating a further Ctrl+C as benign.
 
-    After a turn-cancel the asyncio SIGINT handler is removed, so a second
-    Ctrl+C during this output would otherwise raise KeyboardInterrupt and
-    exit the REPL instead of returning to the prompt.
+    The session SIGINT router keeps the loop safe between turns, but platforms
+    without asyncio signal handlers (Windows proactor) deliver a second Ctrl+C
+    here as a KeyboardInterrupt, which must not exit the REPL instead of
+    returning to the prompt.
     """
     try:
         for line in lines:

src/plyngent/prompting.py

@@ -4,6 +4,7 @@ import asyncio
 import contextlib
 import select
 import sys
+import threading
 import time
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
@@ -19,6 +20,30 @@ class NonInteractiveError(RuntimeError):
     """Raised when a prompt is required but no interactive backend is available."""
 
 
+class PromptCancelledError(RuntimeError):
+    """Raised by a poll-based prompt read when the human cancels it (Ctrl+C).
+
+    Only poll-based reads (``read_line_with_timeout``) can observe the cancel:
+    they check :func:`prompt_cancelled` between polls, so the reader thread
+    returns and nothing is left blocked on stdin.
+    """
+
+
+# Set by the SIGINT target registered for a poll-based prompt read (see
+# cli.interrupt.off_loop_prompt); cleared around each off-loop prompt.
+_prompt_cancel = threading.Event()
+
+
+def prompt_cancelled() -> bool:
+    """Whether the in-flight poll-based prompt read was cancelled by SIGINT."""
+    return _prompt_cancel.is_set()
+
+
+def cancel_prompt_read() -> None:
+    """Ask the in-flight poll-based prompt read to return (thread-safe)."""
+    _prompt_cancel.set()
+
+
 @dataclass(frozen=True, slots=True)
 class ChoiceOption:
     """One selectable option for :func:`choose`."""
@@ -117,6 +142,8 @@ def _read_line_with_timeout_posix(
 
     Returns the line (without trailing newline), ``""`` for an empty Enter, or
     ``None`` when the timeout elapses before any line is complete (EOF too).
+    Raises :class:`PromptCancelledError` when :func:`prompt_cancelled` turns true
+    while waiting (the poll loop notices within one interval).
     """
     _ = sys.stdout.write(prompt)
     _ = sys.stdout.flush()
@@ -125,6 +152,8 @@ def _read_line_with_timeout_posix(
         remaining = deadline - time.monotonic()
         if remaining <= 0:
             return None
+        if prompt_cancelled():
+            raise PromptCancelledError
         ready, _, _ = poll([stream], [], [], min(remaining, 0.1))
         if not ready:
             continue
@@ -143,6 +172,8 @@ def _read_line_with_timeout_windows(prompt: str, timeout: float) -> str | None:
     deadline = time.monotonic() + timeout
     buf: list[str] = []
     while time.monotonic() < deadline:
+        if prompt_cancelled():
+            raise PromptCancelledError
         if msvcrt.kbhit():
             ch = msvcrt.getwch()
             if ch in ("\r", "\n"):
@@ -171,7 +202,9 @@ def read_line_with_timeout(prompt: str, timeout: float) -> str | None:
 
     Returns the typed line (without trailing newline), ``""`` for an empty
     Enter, or ``None`` when the timeout elapses first (EOF too). Ctrl+C
-    propagates as :class:`KeyboardInterrupt`. POSIX polls stdin via ``select``;
+    propagates as :class:`KeyboardInterrupt`; :func:`cancel_prompt_read` (the
+    SIGINT target of an off-loop prompt read) raises
+    :class:`PromptCancelledError` instead. POSIX polls stdin via ``select``;
     Windows uses ``msvcrt`` char polling (best-effort).
     """
     if timeout < 0:
@@ -263,16 +296,18 @@ class NonInteractiveBackend:
 
 
 _backend: PromptBackend = ClickPromptBackend()
-_pause_factory: Callable[[], AbstractContextManager[None]] | None = None
+# CLI hook: context manager around an off-loop prompt. It receives the read's
+# cancel hook (poll-based reads) or None (reads that cannot be interrupted).
+_pause_factory: Callable[[Callable[[], None] | None], AbstractContextManager[None]] | None = None
 _prompt_lock = asyncio.Lock()
 
 
 def configure_prompting(
     *,
     backend: PromptBackend | None = None,
-    pause_factory: Callable[[], AbstractContextManager[None]] | None = None,
+    pause_factory: Callable[[Callable[[], None] | None], AbstractContextManager[None]] | None = None,
 ) -> None:
-    """Install process-wide prompt backend and optional cancel-pause context."""
+    """Install process-wide prompt backend and optional SIGINT guard for prompts."""
     global _backend, _pause_factory  # noqa: PLW0603
     if backend is not None:
         _backend = backend
@@ -284,10 +319,11 @@ def get_prompt_backend() -> PromptBackend:
 
 
 def reset_prompting() -> None:
-    """Restore default Click backend and clear pause hook (tests)."""
+    """Restore default Click backend, clear the pause hook and cancel flag (tests)."""
     global _backend, _pause_factory  # noqa: PLW0603
     _backend = ClickPromptBackend()
     _pause_factory = None
+    _prompt_cancel.clear()
 
 
 def _normalize_options(
@@ -464,13 +500,45 @@ def form(
         backend.echo("Starting over…")
 
 
-async def run_prompt_async[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
-    """Run a blocking prompt off the event loop, serialized, with optional SIGINT pause."""
+async def _prompt_off_loop[R](
+    func: Callable[..., R],
+    *args: object,
+    cancel_hook: Callable[[], None] | None,
+    **kwargs: object,
+) -> R:
+    """Shared body: run *func* off the event loop under the CLI's SIGINT guard."""
     async with _prompt_lock:
-        if _pause_factory is not None:
-            with _pause_factory():
-                return await asyncio.to_thread(func, *args, **kwargs)
-        return await asyncio.to_thread(func, *args, **kwargs)
+        _prompt_cancel.clear()
+        try:
+            if _pause_factory is not None:
+                with _pause_factory(cancel_hook):
+                    return await asyncio.to_thread(func, *args, **kwargs)
+            return await asyncio.to_thread(func, *args, **kwargs)
+        finally:
+            _prompt_cancel.clear()
+
+
+async def run_prompt_async[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
+    """Run a blocking prompt off the event loop, serialized, SIGINT-guarded.
+
+    While the prompt waits, a Ctrl+C must not cancel the enclosing turn: the CLI
+    guard registers the prompt as the SIGINT target. readline-based reads cannot
+    be interrupted without abandoning the reader thread on stdin, so they ignore
+    the interrupt with a one-line hint (see :func:`run_cancellable_prompt_async`
+    for poll-based reads).
+    """
+    return await _prompt_off_loop(func, *args, cancel_hook=None, **kwargs)
+
+
+async def run_cancellable_prompt_async[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
+    """Run a *poll-based* prompt read off the event loop; Ctrl+C cancels it.
+
+    Only for reads that observe :func:`prompt_cancelled`
+    (``read_line_with_timeout``): the SIGINT raises :class:`PromptCancelledError`
+    in the reader thread, so it returns to the caller instead of leaving a thread
+    blocked on stdin (and without cancelling the enclosing turn).
+    """
+    return await _prompt_off_loop(func, *args, cancel_hook=cancel_prompt_read, **kwargs)
 
 
 async def ask_async(prompt: str, *, default: str | None = None) -> str:

src/plyngent/runtime/mcp_client.py

@@ -11,11 +11,15 @@ import asyncio
 import importlib.metadata
 import json
 import os
+import subprocess
+import sys
 from collections import deque
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, cast
 
 if TYPE_CHECKING:
+    from typing import Any
+
     from plyngent.config.models import McpConfig, McpServerConfig
 
 MCP_PROTOCOL_VERSION = "2025-06-18"
@@ -45,6 +49,19 @@ def _client_version() -> str:
         return "0.0.0"
 
 
+def _isolated_spawn_kwargs() -> dict[str, Any]:
+    """Spawn kwargs that keep a server out of our process/console group.
+
+    Ctrl+C in the terminal is delivered to every process in the foreground
+    process group (POSIX) / console process group (Windows), so without this a
+    single Ctrl+C during a turn also killed every MCP server — killing in-flight
+    tool calls and leaving the connection dead until ``/mcp reconnect``.
+    """
+    if sys.platform == "win32":
+        return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP}
+    return {"start_new_session": True}
+
+
 class McpConnectionError(RuntimeError):
     """Server process failed to spawn, handshake, or answer in time."""
 
@@ -143,6 +160,7 @@ class McpServerConnection:
                 stderr=asyncio.subprocess.PIPE,
                 env=os.environ | self._config.env,
                 cwd=self._config.cwd or None,
+                **_isolated_spawn_kwargs(),
             )
         except (OSError, ValueError) as exc:
             self.error = f"spawn {self._config.command!r}: {exc}"

src/plyngent/tools/chat/ask.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from plyngent.agent import ToolTag, tool
 from plyngent.prompting import NonInteractiveError, ask_async
+from plyngent.tools.chat.shape import first_error, string_error
 
 
 @tool(name="ask_user_line", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
@@ -12,6 +13,12 @@ async def ask_user(question: str, default: str = "") -> str:
     or any input that is not a fixed menu. Optional ``default`` is used if the
     user submits empty input (and in non-interactive mode when provided).
     """
+    shape_error = first_error(
+        string_error("ask_user_line", "question", question),
+        string_error("ask_user_line", "default", default),
+    )
+    if shape_error is not None:
+        return f"error: {shape_error}"
     try:
         return await ask_async(question, default=default or None)
     except NonInteractiveError as exc:

src/plyngent/tools/chat/choose.py

@@ -4,6 +4,7 @@ from typing import cast
 
 from plyngent.agent import ToolTag, tool
 from plyngent.prompting import ChoiceOption, NonInteractiveError, choose_async
+from plyngent.tools.chat.shape import first_error, list_error, string_error
 
 
 def parse_options(options: list[object]) -> list[ChoiceOption]:
@@ -51,6 +52,19 @@ async def choose_user(
     When ``allow_custom`` is true (default), free-text answers are accepted.
     Returns the chosen option value (or custom text).
     """
+    shape_error = first_error(
+        string_error("ask_user_choice", "question", question),
+        list_error(
+            "ask_user_choice",
+            "options",
+            options,
+            items="strings or objects with a label",
+            example='["alpha", "beta"]',
+        ),
+        string_error("ask_user_choice", "default", default),
+    )
+    if shape_error is not None:
+        return f"error: {shape_error}"
     try:
         parsed = parse_options([cast("object", o) for o in options])
     except (TypeError, ValueError) as exc:

src/plyngent/tools/chat/form.py

@@ -1,21 +1,28 @@
 from __future__ import annotations
 
 import json
-from typing import cast
+from typing import TYPE_CHECKING, cast
 
 from plyngent.agent import ToolTag, tool
 from plyngent.prompting import FormField, NonInteractiveError, form_async
 from plyngent.tools.chat.choose import parse_options
+from plyngent.tools.chat.shape import first_error, list_error, string_error
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
 
 
-def parse_fields(fields: list[dict[str, object]]) -> list[FormField]:
+def parse_fields(fields: Sequence[object]) -> list[FormField]:
     """Normalize a list of field dicts into ``FormField`` list."""
     if not fields:
         msg = "fields must be a non-empty list"
         raise ValueError(msg)
     out: list[FormField] = []
     for item_obj in fields:
-        raw_map = {str(key): value for key, value in item_obj.items()}
+        if not isinstance(item_obj, dict):
+            msg = "each field must be an object with name/prompt"
+            raise TypeError(msg)
+        raw_map = {str(key): value for key, value in cast("dict[object, object]", item_obj).items()}
         name = raw_map.get("name")
         prompt = raw_map.get("prompt")
         if not isinstance(name, str) or not name:
@@ -61,6 +68,18 @@ async def form_user(
     optional ``allow_custom`` (default true).
     When ``confirm_submit`` is true, the human reviews a summary before submit.
     """
+    shape_error = first_error(
+        string_error("ask_user_form", "title", title),
+        list_error(
+            "ask_user_form",
+            "fields",
+            fields,
+            items="objects",
+            example='[{"name": "port", "prompt": "Which port?"}]',
+        ),
+    )
+    if shape_error is not None:
+        return f"error: {shape_error}"
     try:
         parsed = parse_fields(fields)
     except (TypeError, ValueError) as exc:

src/plyngent/tools/chat/shape.py

@@ -0,0 +1,74 @@
+"""Shape checks for ask-family tool arguments.
+
+Tool arguments arrive as decoded JSON, so a parameter declared ``list[str]`` can
+still be a plain string at runtime. Duck-typing then misleads instead of failing:
+``ask_user_choice(options="yes/no")`` iterates the characters and shows a menu of
+letters, and a string ``fields`` dies inside the form parser with an
+``AttributeError``. Each ask-family tool checks its container arguments up front
+and answers with a model-facing error that also shows the expected shape (same
+idea as :func:`plyngent.tools.workspace.argv_shape_error`).
+
+Every check takes ``object``, so the isinstance calls stay meaningful and the
+tools themselves stay free of unnecessary-isinstance warnings.
+"""
+
+from __future__ import annotations
+
+_SHAPES: dict[type, str] = {
+    str: "a string",
+    list: "an array",
+    dict: "an object",
+    bool: "a boolean",
+    int: "a number",
+    float: "a number",
+    type(None): "null",
+}
+
+
+def shape_of(value: object) -> str:
+    """Model-facing name for a decoded JSON value (``a string``, ``an array``…)."""
+    return _SHAPES.get(type(value), type(value).__name__)
+
+
+def first_error(*errors: str | None) -> str | None:
+    """First non-``None`` error (``None`` when every shape was acceptable)."""
+    return next((error for error in errors if error), None)
+
+
+def string_error(tool: str, field: str, value: object) -> str | None:
+    """Error when *value* must be a string but is not; ``None`` when it is."""
+    if isinstance(value, str):
+        return None
+    return f"`{tool}` expects `{field}` as a string, got {shape_of(value)}"
+
+
+def list_error(tool: str, field: str, value: object, *, items: str, example: str) -> str | None:
+    """Error when *value* must be a JSON array of *items* but is not; else ``None``.
+
+    Only the container is checked: item-level validation stays in the tool's own
+    parser (``parse_options`` / ``parse_fields``), which already reports what is
+    wrong with a bad entry.
+    """
+    if isinstance(value, str):
+        return f"`{tool}` expects `{field}` as a JSON array of {items}, not a string; pass e.g. {example}"
+    if not isinstance(value, list):
+        return f"`{tool}` expects `{field}` as a JSON array of {items}, got {shape_of(value)}; pass e.g. {example}"
+    return None
+
+
+def number_arg(tool: str, field: str, value: object, *, example: str) -> tuple[float | None, str | None]:
+    """Coerce *value* to a float, accepting numeric strings; else ``(None, error)``.
+
+    Models often send numbers as strings, which is accepted here. Booleans are
+    not numbers for this purpose, and neither is non-numeric text.
+    """
+    if isinstance(value, bool):
+        return None, f"`{tool}` expects `{field}` as a number, got a boolean; pass e.g. {example}"
+    if isinstance(value, (int, float)):
+        return float(value), None
+    if isinstance(value, str):
+        try:
+            return float(value.strip()), None
+        except ValueError:
+            return None, f"`{tool}` expects `{field}` as a number, got a non-numeric string; pass e.g. {example}"
+    return None, f"`{tool}` expects `{field}` as a number, got {shape_of(value)}; pass e.g. {example}"

src/plyngent/tools/chat/wait.py

@@ -7,12 +7,18 @@ import sys
 from click import style
 
 from plyngent.agent import ToolTag, tool
-from plyngent.prompting import get_prompt_backend, read_line_with_timeout, run_prompt_async
+from plyngent.prompting import (
+    PromptCancelledError,
+    get_prompt_backend,
+    read_line_with_timeout,
+    run_cancellable_prompt_async,
+)
+from plyngent.tools.chat.shape import first_error, number_arg, string_error
 
 _ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
 
 
-def _wait_prompt(duration: int, *, reason: str | None) -> str:
+def _wait_prompt(duration: float, *, reason: str | None) -> str:
     """Two-line wait prompt: status line, then the optional-reason input prompt.
 
     ``read_line_with_timeout`` writes the prompt verbatim to stdout (no click
@@ -20,7 +26,7 @@ def _wait_prompt(duration: int, *, reason: str | None) -> str:
     is not a TTY. On a terminal: cyan status, dimmed detail, yellow input
     prompt; plain text otherwise (tests, redirected output).
     """
-    status = style(f"Waiting for {duration}s", fg="cyan")
+    status = style(f"Waiting for {duration:g}s", fg="cyan")
     if reason:
         status += style(f" ({reason})", fg="bright_black")
     status += style(". Press Enter to disturb.", fg="bright_black")
@@ -31,28 +37,44 @@ def _wait_prompt(duration: int, *, reason: str | None) -> str:
     return rendered
 
 
+async def _interactive_wait(seconds: float, reason: str | None) -> str:
+    """Show the two-line wait prompt and report how the wait ended."""
+    label = f"{seconds:g}"
+    prompt = _wait_prompt(seconds, reason=reason)
+    try:
+        line = await run_cancellable_prompt_async(read_line_with_timeout, prompt, seconds)
+    except PromptCancelledError:
+        return "cancelled by user"
+    if line is None:
+        return f"waited {label}s"
+    text = line.strip()
+    if text:
+        return f"disturbed by user: {text}"
+    return "disturbed by user (no reason)"
+
+
 @tool(name="wait", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
 async def wait(duration: int, *, reason: str | None = None) -> str:
     """Wait ``duration`` seconds before continuing.
 
     Interactive sessions show a two-line prompt: a status line, then an
     optional-reason input — pressing Enter (optionally after typing a reason)
-    "disturbs" the wait so the turn continues immediately. Non-interactive runs
-    simply sleep the full duration.
+    "disturbs" the wait so the turn continues immediately, and Ctrl+C cancels
+    the wait (the turn continues). Non-interactive runs simply sleep the full
+    duration.
     """
-    if duration < 0:
+    seconds, duration_error = number_arg("wait", "duration", duration, example="5")
+    reason_error = None if reason is None else string_error("wait", "reason", reason)
+    shape_error = first_error(duration_error, reason_error)
+    if shape_error is not None:
+        return f"error: {shape_error}"
+    assert seconds is not None  # number_arg returns a value whenever it accepts
+    if seconds < 0:
         return "error: duration must not be negative"
-    if duration == 0:
+    if seconds == 0:
         return "waited 0s"
     backend = get_prompt_backend()
     if not backend.is_interactive():
-        await asyncio.sleep(duration)
-        return f"waited {duration}s"
-    prompt = _wait_prompt(duration, reason=reason)
-    line = await run_prompt_async(read_line_with_timeout, prompt, float(duration))
-    if line is None:
-        return f"waited {duration}s"
-    text = line.strip()
-    if text:
-        return f"disturbed by user: {text}"
-    return "disturbed by user (no reason)"
+        await asyncio.sleep(seconds)
+        return f"waited {seconds:g}s"
+    return await _interactive_wait(seconds, reason)

src/plyngent/tools/process/ask_into_pty.py

@@ -54,7 +54,7 @@ async def ask_into_pty(
 
     ``message`` is shown to the human (what to enter), not the secret itself.
     ``secret=true`` uses no-echo input. ``submit=true`` (default) appends a
-    newline after the answer. Empty input and Ctrl+C cancel without writing.
+    newline after the answer. Empty input cancels without writing.
     """
     label = message.strip() or ("Secret" if secret else "Input")
     try:

tests/test_agent/test_tools.py

@@ -70,6 +70,46 @@ async def test_registry_handler_error() -> None:
     assert "failed" in result
 
 
+async def test_registry_reports_missing_required_arguments() -> None:
+    """A missing required arg is explained with the accepted names, not a TypeError."""
+    called: list[str] = []
+
+    @tool(register=False)
+    def pick(question: str, options: list[str], *, allow_custom: bool = True) -> str:
+        del allow_custom
+        called.append(question)
+        return options[0]
+
+    registry = ToolRegistry([pick])
+    result = await registry.execute("pick", '{"question": "Which?"}')
+    assert result == (
+        "error: `pick` is missing required argument(s): options; it accepts: allow_custom, options, question"
+    )
+    assert called == []  # the handler never ran
+    assert await registry.execute("pick", '{"question": "Which?", "options": ["a"]}') == "a"
+    assert called == ["Which?"]
+
+
+async def test_registry_missing_arg_skips_confirm_hook() -> None:
+    """The human is not asked to confirm a call that can never run."""
+    confirms: list[str] = []
+
+    def confirm(name: str, args: object, reason: str) -> bool:
+        del args, reason
+        confirms.append(name)
+        return True
+
+    @tool(register=False)
+    def write(path: str, content: str) -> str:
+        del content
+        return path
+
+    registry = ToolRegistry([write], danger=lambda name, args: "writes a file", on_confirm=confirm)
+    result = await registry.execute("write", '{"path": "x"}')
+    assert "missing required argument(s): content" in result
+    assert confirms == []
+
+
 def test_tool_items() -> None:
     @tool(register=False)
     def f(x: int) -> int:

tests/test_cli/test_display.py

@@ -553,6 +553,13 @@ async def test_pretty_wait_disturbed(capsys: pytest.CaptureFixture[str]) -> None
     assert "* Wait (disturbed)" in out
 
 
+async def test_pretty_wait_cancelled(capsys: pytest.CaptureFixture[str]) -> None:
+    """Ctrl+C during the wait prompt reads as cancelled, not disturbed."""
+    await render_events(_aiter([_pretty_call("wait", '{"duration": 30}'), _result("cancelled by user")]))
+    out = capsys.readouterr().out
+    assert "* Wait (cancelled)" in out
+
+
 async def test_pretty_wait_error(capsys: pytest.CaptureFixture[str]) -> None:
     content = "error: duration must not be negative"
     await render_events(_aiter([_pretty_call("wait", '{"duration": -1}'), _result(content)]))

tests/test_cli/test_interrupt.py

@@ -1,124 +1,73 @@
 from __future__ import annotations
 
 import asyncio
+import os
 import signal
+import threading
+import time
 from typing import TYPE_CHECKING
 
 import pytest
 
 from plyngent.cli.interrupt import (
-    allow_task_cancel,
     install_keyboard_interrupt_sigint,
+    install_sigint_router,
+    off_loop_prompt,
     pause_task_cancel_for_prompt,
     raise_keyboard_interrupt_sigint,
-    run_in_prompt_thread,
-    set_sigint_reinstall,
+    sigint_cancels,
+    sigint_routed,
+    uninstall_sigint_router,
+)
+from plyngent.cli.limits import prompt_continue_limit
+from plyngent.cli.retry import run_cancellable, sleep_cancellable
+from plyngent.prompting import (
+    PromptCancelledError,
+    configure_prompting,
+    prompt_cancelled,
+    run_cancellable_prompt_async,
+    temporary_backend,
 )
-from plyngent.cli.limits import prompt_continue_limit, prompt_continue_limit_async
-from plyngent.cli.retry import run_cancellable
-from plyngent.prompting import temporary_backend
 from tests.test_prompting import ScriptedBackend
 
 if TYPE_CHECKING:
-    pass
+    from collections.abc import AsyncIterator
+
+    import pytest as pytest_module
 
 
-def test_pause_task_cancel_for_prompt() -> None:
-    assert allow_task_cancel() is True
-    with pause_task_cancel_for_prompt():
-        assert allow_task_cancel() is False
-    assert allow_task_cancel() is True
+class FakeTarget:
+    """Minimal ``InterruptTarget`` recording its cancels."""
+
+    def __init__(self, log: list[str], name: str = "cancel") -> None:
+        self.log = log
+        self.name = name
+
+    def cancel(self) -> None:
+        self.log.append(self.name)
 
 
-def test_nested_pause_depth() -> None:
-    assert allow_task_cancel() is True
-    with pause_task_cancel_for_prompt():
-        assert allow_task_cancel() is False
-        with pause_task_cancel_for_prompt():
-            assert allow_task_cancel() is False
-        assert allow_task_cancel() is False
-    assert allow_task_cancel() is True
-
-
-def test_prompt_continue_limit_under_pause(monkeypatch: pytest.MonkeyPatch) -> None:
-    backend = ScriptedBackend(["y"])
-
-    def _read_line(prompt: str, *, default: str | None = None, completions: object = None) -> str:
-        del prompt, default, completions
-        assert allow_task_cancel() is False
-        return "y"
-
-    monkeypatch.setattr(backend, "read_line", _read_line)
-    with temporary_backend(backend):
-        assert prompt_continue_limit("too many rounds") is True
-
-
-async def test_run_in_prompt_thread_pauses_cancel() -> None:
-    """Cancel is paused on the main thread for the whole to_thread call."""
-    assert allow_task_cancel() is True
-
-    def work() -> str:
-        return "ok"
-
-    result = await run_in_prompt_thread(work)
-    assert result == "ok"
-    assert allow_task_cancel() is True
-
-
-async def test_prompt_continue_limit_async(monkeypatch: pytest.MonkeyPatch) -> None:
-    del monkeypatch
-    with temporary_backend(ScriptedBackend(["y"])):
-        assert await prompt_continue_limit_async("too many rounds") is True
-
-
-async def test_sigint_cancels_after_prompt_pause() -> None:
-    """Regression: after a mid-turn prompt, SIGINT must still cancel the turn.
-
-    Previously, reinstalling the asyncio SIGINT handler while
-    allow_task_cancel was still False froze that value into the callback
-    forever (ContextVar snapshot at add_signal_handler).
-    """
-    loop = asyncio.get_running_loop()
+@pytest.fixture
+async def router() -> AsyncIterator[None]:
+    """Install the session SIGINT router; restore the CLI handler afterwards."""
+    install_keyboard_interrupt_sigint()
+    _ = install_sigint_router()
     try:
-        loop.add_signal_handler(signal.SIGINT, lambda: None)
-        loop.remove_signal_handler(signal.SIGINT)
-    except NotImplementedError, RuntimeError, ValueError:
-        pytest.skip("asyncio signal handlers not available on this platform")
+        yield
+    finally:
+        uninstall_sigint_router()
+        signal.signal(signal.SIGINT, signal.default_int_handler)
 
-    started = asyncio.Event()
-    cancelled = asyncio.Event()
 
-    async def hang() -> None:
-        started.set()
-        try:
-            await asyncio.sleep(3600)
-        except asyncio.CancelledError:
-            cancelled.set()
-            raise
-
-    async def turn() -> None:
-        # Simulate soft-confirm pause mid-turn, then keep streaming.
-        with pause_task_cancel_for_prompt():
-            assert allow_task_cancel() is False
-        assert allow_task_cancel() is True
-        await hang()
-
-    task = asyncio.create_task(run_cancellable(turn()))
-    await started.wait()
-    # Give run_cancellable a tick to install the handler after the pause reinstall.
-    await asyncio.sleep(0)
-    assert allow_task_cancel() is True
-    # Deliver SIGINT the same way Ctrl+C does under asyncio.
-    loop.call_soon(lambda: None)  # ensure loop is processing
-    # Invoke the process signal: raise SIGINT to this process.
-    import os
+async def _deliver_sigint() -> None:
+    """Send SIGINT to this process and let the loop run its signal callback.
 
+    asyncio's signal wakeup is only noticed while the loop parks in its
+    selector, so a real (short) wait is needed — ``sleep(0)`` never parks and
+    the callback would not run at all.
+    """
     os.kill(os.getpid(), signal.SIGINT)
-    with pytest.raises(asyncio.CancelledError):
-        await task
-    assert cancelled.is_set()
-    # Cleanup any leftover reinstall hook from run_cancellable
-    set_sigint_reinstall(None)
+    await asyncio.sleep(0.05)
 
 
 def test_install_keyboard_interrupt_sigint() -> None:
@@ -148,8 +97,6 @@ def test_asyncio_run_keeps_keyboard_interrupt_sigint() -> None:
     KeyboardInterrupt, which also cancelled ``memory.close()`` on a Ctrl+D exit.
     With a non-default handler installed first, Runner leaves SIGINT alone.
     """
-    import asyncio
-
     install_keyboard_interrupt_sigint()
     try:
 
@@ -162,28 +109,211 @@ def test_asyncio_run_keeps_keyboard_interrupt_sigint() -> None:
         signal.signal(signal.SIGINT, signal.default_int_handler)
 
 
-async def test_run_cancellable_restores_previous_sigint() -> None:
-    """Regression: after a turn SIGINT is restored, not left as SIG_DFL.
+async def test_sigint_cancels_registered_target(router: None) -> None:
+    """SIGINT cancels the target registered around the awaited turn task."""
+    del router
+    cancelled = asyncio.Event()
+
+    async def hang() -> None:
+        try:
+            await asyncio.sleep(3600)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    task = asyncio.create_task(hang())
+    await asyncio.sleep(0)
+    with sigint_cancels(task):
+        os.kill(os.getpid(), signal.SIGINT)
+        # Waiting on the effect (not on a fixed sleep) keeps this deterministic.
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=5)
+    assert cancelled.is_set()
+
+
+async def test_sigint_while_loop_is_idle_does_not_escape(router: None) -> None:
+    """Regression: a stray Ctrl+C while the loop is parked in its selector.
+
+    The interrupt used to be raised in the loop's own frame (not inside any
+    coroutine), escaping ``run_until_complete``: the Runner cancelled the main
+    task and the REPL was left half-shut-down, so the next Ctrl+C killed the
+    process instead of returning to the prompt.
+    """
+    del router
+
+    def _send() -> None:
+        time.sleep(0.05)
+        os.kill(os.getpid(), signal.SIGINT)
+
+    sender = threading.Thread(target=_send)
+    sender.start()
+    try:
+        await asyncio.sleep(0.2)  # parked in the selector when SIGINT lands
+    finally:
+        sender.join()
+    task = asyncio.current_task()
+    assert task is not None
+    assert task.cancelling() == 0
+    assert not task.cancelled()
+
+
+async def test_sigint_cancels_innermost_target_only(router: None) -> None:
+    """Nested targets: the innermost (the active turn or prompt) wins."""
+    del router
+    outer: list[str] = []
+    inner: list[str] = []
+    with sigint_cancels(FakeTarget(outer, "outer")):
+        await _deliver_sigint()
+        assert outer == ["outer"]
+        with sigint_cancels(FakeTarget(inner, "inner")):
+            await _deliver_sigint()
+            assert inner == ["inner"]
+        assert outer == ["outer"]  # the nested signal never reached it
+
+
+async def test_sigint_target_stops_after_context_exit(router: None) -> None:
+    """A popped target no longer sees SIGINT (and an empty stack ignores it)."""
+    del router
+    calls: list[str] = []
+    with sigint_cancels(FakeTarget(calls)):
+        await _deliver_sigint()
+    assert calls == ["cancel"]
+    await _deliver_sigint()
+    assert calls == ["cancel"]
+
+
+async def test_off_loop_prompt_keeps_turn_alive(
+    router: None,
+    capsys: pytest_module.CaptureFixture[str],
+) -> None:
+    """A prompt's SIGINT target shields the turn; afterwards the turn cancels.
+
+    Regression (badcb3c, revisited): the router resolves the innermost target
+    per signal instead of freezing cancel state into the handler, so a mid-turn
+    prompt neither aborts the turn nor freezes turn-cancel.
+    """
+    del router
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    async def hang() -> None:
+        started.set()
+        try:
+            await asyncio.sleep(3600)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    task = asyncio.create_task(hang())
+    await started.wait()
+    with sigint_cancels(task):
+        with off_loop_prompt():
+            await _deliver_sigint()
+            assert not task.cancelling()
+        err = capsys.readouterr().err
+        assert "Ctrl+C ignored while a prompt is waiting" in err
+        # The prompt returned (outer context still active): SIGINT cancels now.
+        os.kill(os.getpid(), signal.SIGINT)
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=5)
+    assert cancelled.is_set()
+
+
+async def test_off_loop_prompt_cancel_hook_is_preferred(router: None) -> None:
+    """A poll-based prompt read gets cancelled instead of the enclosing turn."""
+    del router
+    hooks: list[str] = []
+    turn_calls: list[str] = []
+    with sigint_cancels(FakeTarget(turn_calls, "turn")), off_loop_prompt(lambda: hooks.append("read")):
+        await _deliver_sigint()
+    assert hooks == ["read"]
+    assert turn_calls == []
+
+
+async def test_sigint_cancels_off_loop_prompt_read(router: None) -> None:
+    """Wiring: SIGINT aborts a poll-based read (not the turn) and clears the flag."""
+    del router
+    turn_calls: list[str] = []
+    seen: list[str] = []
+
+    def _poll_read(prompt: str, timeout: float) -> str:
+        """Stand-in for ``read_line_with_timeout``: polls the cancel flag."""
+        del prompt
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if prompt_cancelled():
+                seen.append("cancelled")
+                raise PromptCancelledError
+            time.sleep(0.01)
+        seen.append("timed out")
+        return ""
+
+    with temporary_backend(ScriptedBackend([])):
+        configure_prompting(pause_factory=off_loop_prompt)
+        with sigint_cancels(FakeTarget(turn_calls, "turn")):
+            reader = asyncio.create_task(run_cancellable_prompt_async(_poll_read, "p> ", 30.0))
+            await asyncio.sleep(0.05)
+            os.kill(os.getpid(), signal.SIGINT)
+            with pytest.raises(PromptCancelledError):
+                await asyncio.wait_for(reader, timeout=5)
+    assert seen == ["cancelled"]
+    assert turn_calls == []
+    assert not prompt_cancelled()  # cleared for the next prompt
+
+
+async def test_pause_task_cancel_for_prompt_swaps_sigint(router: None) -> None:
+    """A main-thread blocking read takes SIGINT as a KeyboardInterrupt."""
+    del router
+    routed = signal.getsignal(signal.SIGINT)
+    assert routed is not raise_keyboard_interrupt_sigint
+    with pause_task_cancel_for_prompt():
+        assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+        with pause_task_cancel_for_prompt():
+            assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+        # Inner exit must not resume the router while the outer read blocks.
+        assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+    assert signal.getsignal(signal.SIGINT) is routed
+
+
+async def test_prompt_continue_limit_under_pause(
+    router: None,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Synchronous prompts read on the main thread, where SIGINT is a KI."""
+    del router
+    backend = ScriptedBackend(["y"])
+
+    def _read_line(prompt: str, *, default: str | None = None, completions: object = None) -> str:
+        del prompt, default, completions
+        assert signal.getsignal(signal.SIGINT) is raise_keyboard_interrupt_sigint
+        return "y"
+
+    monkeypatch.setattr(backend, "read_line", _read_line)
+    with temporary_backend(backend):
+        assert prompt_continue_limit("too many rounds") is True
+
+
+async def test_run_cancellable_keeps_router_installed(router: None) -> None:
+    """Regression: after a turn SIGINT must not be left as SIG_DFL.
 
     ``loop.remove_signal_handler`` leaves SIG_DFL, which would terminate the
-    process on a stray Ctrl+C between turns instead of letting the REPL catch
-    a KeyboardInterrupt.
+    process on a stray Ctrl+C between turns; the session router stays installed.
     """
-    loop = asyncio.get_running_loop()
-    try:
-        loop.add_signal_handler(signal.SIGINT, lambda: None)
-        loop.remove_signal_handler(signal.SIGINT)
-    except NotImplementedError, RuntimeError, ValueError:
-        pytest.skip("asyncio signal handlers not available on this platform")
-
-    previous = signal.getsignal(signal.SIGINT)
+    del router
 
     async def quick() -> None:
         return None
 
-    try:
-        await run_cancellable(quick())
-        assert signal.getsignal(signal.SIGINT) is previous
-        assert signal.getsignal(signal.SIGINT) is not signal.SIG_DFL
-    finally:
-        signal.signal(signal.SIGINT, signal.default_int_handler)
+    await run_cancellable(quick())
+    assert sigint_routed()
+    assert signal.getsignal(signal.SIGINT) is not signal.SIG_DFL
+    assert signal.getsignal(signal.SIGINT) is not signal.default_int_handler
+
+
+async def test_sleep_cancellable_cancelled_by_sigint(router: None) -> None:
+    """The auto-retry countdown is cancelled by Ctrl+C (was a teardown window)."""
+    del router
+    waiter = asyncio.create_task(sleep_cancellable(30))
+    await asyncio.sleep(0)
+    os.kill(os.getpid(), signal.SIGINT)
+    assert await asyncio.wait_for(waiter, timeout=5) is False

tests/test_prompting.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import contextlib
 import io
+from typing import TYPE_CHECKING
 
 import pytest
 
@@ -9,15 +11,33 @@ from plyngent.prompting import (
     FormField,
     NonInteractiveBackend,
     NonInteractiveError,
+    PromptCancelledError,
     _read_line_with_timeout_posix,
     ask,
+    cancel_prompt_read,
     choose,
+    configure_prompting,
     confirm,
     form,
+    prompt_cancelled,
     read_line_with_timeout,
+    reset_prompting,
+    run_cancellable_prompt_async,
+    run_prompt_async,
     temporary_backend,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Generator, Iterator
+
+
+@pytest.fixture(autouse=True)
+def reset_prompting_state() -> Iterator[None]:
+    """Prompting state (backend, SIGINT guard, cancel flag) is process-global."""
+    reset_prompting()
+    yield
+    reset_prompting()
+
 
 def _ready_poll(r, w, x, t):
     return (r, w, x)
@@ -51,6 +71,43 @@ def test_read_line_with_timeout_negative_timeout() -> None:
     assert read_line_with_timeout("p> ", -1) is None
 
 
+def test_read_line_with_timeout_cancelled() -> None:
+    """A cancel request (SIGINT during an off-loop prompt) ends the read."""
+    stream = io.StringIO()
+    cancel_prompt_read()
+    try:
+        assert prompt_cancelled()
+        with pytest.raises(PromptCancelledError):
+            _read_line_with_timeout_posix("p> ", 5.0, stream=stream, poll=_never_poll)
+    finally:
+        stream.close()
+
+
+async def test_run_prompt_async_clears_cancel_flag() -> None:
+    """A stale cancel request must not leak into the next prompt read."""
+    cancel_prompt_read()
+    with temporary_backend(NonInteractiveBackend()):
+        assert await run_prompt_async(lambda: "ok") == "ok"
+    assert not prompt_cancelled()
+
+
+async def test_run_prompt_async_passes_cancel_hook_only_when_cancellable() -> None:
+    """The CLI guard gets the read's cancel hook for poll-based reads only."""
+    seen: list[object] = []
+
+    @contextlib.contextmanager
+    def _guard(cancel: object) -> Generator[None]:
+        seen.append(cancel)
+        yield
+
+    with temporary_backend(NonInteractiveBackend()):
+        # temporary_backend resets the guard, so install it inside.
+        configure_prompting(pause_factory=_guard)
+        assert await run_cancellable_prompt_async(lambda: "a") == "a"
+        assert await run_prompt_async(lambda: "b") == "b"
+    assert seen == [cancel_prompt_read, None]
+
+
 class ScriptedBackend:
     """Deterministic backend for unit tests."""
 

tests/test_runtime/test_mcp_client.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import sys
 from pathlib import Path
 
@@ -97,6 +98,26 @@ async def test_close_terminates_subprocess() -> None:
     assert proc.returncode is not None
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups only")
+async def test_server_spawned_outside_our_process_group() -> None:
+    """Regression: a terminal Ctrl+C must not reach the MCP server.
+
+    The tty delivers SIGINT to every process in the foreground process group, so
+    a server spawned in *our* group died on any Ctrl+C during a turn (killing
+    in-flight tool calls and leaving the connection dead). ``start_new_session``
+    moves the server into its own session/group.
+    """
+    connection = McpServerConnection("fake", _server_config())
+    await connection.start()
+    try:
+        proc = connection._proc
+        assert proc is not None
+        assert os.getpgid(proc.pid) != os.getpgid(0)
+        assert os.getsid(proc.pid) != os.getsid(0)
+    finally:
+        await connection.aclose()
+
+
 async def test_manager_start_status_and_call() -> None:
     manager = McpManager(McpConfig(servers={"fake": _server_config()}, disable=[]))
     await manager.ensure_started()

tests/test_tools/test_chat_tools.py

@@ -3,8 +3,10 @@ from __future__ import annotations
 import importlib
 import json
 
+import pytest
+
 from plyngent.agent import ToolRegistry
-from plyngent.prompting import NonInteractiveBackend, temporary_backend
+from plyngent.prompting import NonInteractiveBackend, PromptCancelledError, temporary_backend
 from plyngent.tools.chat import CHAT_TOOLS, ask_user, choose_user, form_user, wait
 from tests.test_prompting import ScriptedBackend
 
@@ -146,6 +148,20 @@ async def test_wait_tool_times_out(monkeypatch) -> None:
     assert out == "waited 3s"
 
 
+async def test_wait_tool_cancelled_by_user(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Ctrl+C during the wait prompt cancels the wait; the turn continues."""
+
+    def _cancelled(prompt: str, timeout: float) -> str:
+        del prompt, timeout
+        raise PromptCancelledError
+
+    monkeypatch.setattr(wait_module, "read_line_with_timeout", _cancelled)
+    with temporary_backend(ScriptedBackend([])):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", '{"duration": 30}')
+    assert out == "cancelled by user"
+
+
 async def test_wait_tool_prompt_two_lines_with_reason(monkeypatch) -> None:
     prompts: list[str] = []
     monkeypatch.setattr(
@@ -192,3 +208,170 @@ def test_wait_prompt_colored_on_tty(monkeypatch) -> None:
     prompt = wait_module._wait_prompt(5, reason="tests")
     assert "\x1b[36m" in prompt  # cyan status
     assert "\x1b[33m" in prompt  # yellow input prompt
+
+
+def test_shape_of_names_json_types() -> None:
+    from plyngent.tools.chat.shape import first_error, shape_of
+
+    assert shape_of("x") == "a string"
+    assert shape_of([]) == "an array"
+    assert shape_of({}) == "an object"
+    assert shape_of(True) == "a boolean"  # bool is not reported as a number
+    assert shape_of(3) == "a number"
+    assert shape_of(3.5) == "a number"
+    assert shape_of(None) == "null"
+    assert first_error(None, "boom", "later") == "boom"
+    assert first_error(None, None) is None
+
+
+async def test_ask_user_line_rejects_non_string_args() -> None:
+    """Mis-typed args get explained instead of reaching the human as garbage."""
+    backend = ScriptedBackend(["unused"])
+    with temporary_backend(backend):
+        registry = ToolRegistry([ask_user])
+        bad_question = await registry.execute("ask_user_line", json.dumps({"question": ["a", "b"]}))
+        bad_default = await registry.execute(
+            "ask_user_line",
+            json.dumps({"question": "Which?", "default": 5}),
+        )
+    assert bad_question.startswith("error:")
+    assert "`question`" in bad_question
+    assert "got an array" in bad_question
+    assert bad_default.startswith("error:")
+    assert "`default`" in bad_default
+    assert "got a number" in bad_default
+    assert backend.lines == ["unused"]  # the prompt never happened
+
+
+async def test_choose_user_rejects_string_options() -> None:
+    """``options="alpha, beta"`` used to build a menu of single characters."""
+    backend = ScriptedBackend(["1"])
+    with temporary_backend(backend):
+        registry = ToolRegistry([choose_user])
+        out = await registry.execute(
+            "ask_user_choice",
+            json.dumps({"question": "Pick", "options": "alpha, beta"}),
+        )
+    assert out.startswith("error:")
+    assert "`options`" in out
+    assert "not a string" in out
+    assert '["alpha", "beta"]' in out
+    assert backend.lines == ["1"]
+
+
+async def test_choose_user_rejects_object_options() -> None:
+    registry = ToolRegistry([choose_user])
+    out = await registry.execute(
+        "ask_user_choice",
+        json.dumps({"question": "Pick", "options": {"alpha": 1}}),
+    )
+    assert out.startswith("error:")
+    assert "got an object" in out
+
+
+async def test_form_user_rejects_string_fields() -> None:
+    backend = ScriptedBackend(["ncbm"], confirms=[True])
+    with temporary_backend(backend):
+        registry = ToolRegistry([form_user])
+        out = await registry.execute(
+            "ask_user_form",
+            json.dumps({"title": "Setup", "fields": '[{"name": "user"}]'}),
+        )
+    assert out.startswith("error:")
+    assert "`fields`" in out
+    assert "not a string" in out
+    assert backend.lines == ["ncbm"]
+
+
+async def test_form_user_rejects_non_object_field() -> None:
+    registry = ToolRegistry([form_user])
+    out = await registry.execute(
+        "ask_user_form",
+        json.dumps({"title": "Setup", "fields": ["user"]}),
+    )
+    assert out == "error: each field must be an object with name/prompt"
+
+
+async def test_wait_accepts_numeric_string_duration(monkeypatch) -> None:
+    """A model sending "5" instead of 5 still waits."""
+    slept: list[float] = []
+
+    async def fake_sleep(seconds: float) -> None:
+        slept.append(seconds)
+
+    monkeypatch.setattr(wait_module.asyncio, "sleep", fake_sleep)
+    with temporary_backend(NonInteractiveBackend()):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", json.dumps({"duration": "5"}))
+    assert out == "waited 5s"
+    assert slept == [5.0]
+
+
+async def test_wait_accepts_fractional_duration(monkeypatch) -> None:
+    slept: list[float] = []
+
+    async def fake_sleep(seconds: float) -> None:
+        slept.append(seconds)
+
+    monkeypatch.setattr(wait_module.asyncio, "sleep", fake_sleep)
+    with temporary_backend(NonInteractiveBackend()):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", json.dumps({"duration": 2.5}))
+    assert out == "waited 2.5s"
+    assert slept == [2.5]
+
+
+async def test_wait_rejects_bad_duration_shapes() -> None:
+    cases: list[tuple[object, str]] = [
+        ("soon", "non-numeric string"),
+        ([5], "an array"),
+        ({"seconds": 5}, "an object"),
+        (True, "a boolean"),
+    ]
+    with temporary_backend(NonInteractiveBackend()):
+        registry = ToolRegistry([wait])
+        for value, expected in cases:
+            out = await registry.execute("wait", json.dumps({"duration": value}))
+            assert out.startswith("error:"), f"{value!r} was accepted: {out!r}"
+            assert "`duration`" in out
+            assert expected in out, f"{value!r}: {out!r}"
+
+
+async def test_wait_rejects_non_string_reason() -> None:
+    with temporary_backend(NonInteractiveBackend()):
+        registry = ToolRegistry([wait])
+        out = await registry.execute("wait", json.dumps({"duration": 0, "reason": ["why"]}))
+    assert out.startswith("error:")
+    assert "`reason`" in out
+
+
+def test_wait_prompt_renders_fractional_seconds() -> None:
+    prompt = wait_module._wait_prompt(2.5, reason=None)
+    assert prompt.startswith("Waiting for 2.5s.")
+    assert wait_module._wait_prompt(5, reason=None).startswith("Waiting for 5s.")
+
+
+async def test_choose_user_reports_missing_options() -> None:
+    """A call without the required ``options`` lists what the tool accepts."""
+    backend = ScriptedBackend(["1"])
+    with temporary_backend(backend):
+        registry = ToolRegistry([choose_user])
+        out = await registry.execute("ask_user_choice", json.dumps({"question": "Pick"}))
+    assert out.startswith("error:")
+    assert "missing required argument(s): options" in out
+    assert "it accepts: allow_custom, default, options, question" in out
+    assert backend.lines == ["1"]  # the human was never asked
+
+
+async def test_form_user_reports_missing_fields() -> None:
+    registry = ToolRegistry([form_user])
+    out = await registry.execute("ask_user_form", json.dumps({"title": "Setup"}))
+    assert out.startswith("error:")
+    assert "missing required argument(s): fields" in out
+
+
+async def test_wait_reports_missing_duration() -> None:
+    registry = ToolRegistry([wait])
+    out = await registry.execute("wait", "{}")
+    assert out.startswith("error:")
+    assert "missing required argument(s): duration" in out