docs: describe argument-aware command denylist entries

AGENTS.md

@@ -90,13 +90,13 @@ Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `
 
 Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 
-- **`workspace`**: path resolve under primary root **or** temporary allowlist roots **or** a directory-access grant (mode-aware: `read` < `write` < `exec`, default required `write`; a file grant matches only that file); path substring denylist (wins over grants); command basename denylist (CLI: timed human allow override, default deny on timeout; **not** skipped by YOLO); escape error hints `request_directory_access`.
+- **`workspace`**: path resolve under primary root **or** temporary allowlist roots **or** a directory-access grant (mode-aware: `read` < `write` < `exec`, default required `write`; a file grant matches only that file); path substring denylist (wins over grants); command denylist via `match_command_denylist` (checks every program of the wrapper/`run` chain, resolved by `tools/command_scan.py`; an entry is a program, plus optional argument words that make it dangerous — `systemctl reboot poweroff …` matches `systemctl reboot` and `systemctl start reboot.target`, not `systemctl status`; basenames are lower-cased, so `RM` / `rm.exe` count; defaults cover privilege `sudo`/`su`/`doas`/`pkexec`/`run0`/`systemd-run`, destructive `rm`/`mkfs`/`dd`/`chown`/`mount`/…, and power state `shutdown`/`reboot`/`poweroff`/`halt`/`init`/`telinit`/`kexec`/`pm-*`/`s2*`; CLI: timed human allow override, default deny on timeout; **not** skipped by YOLO); escape error hints `request_directory_access`.
 - **`access`** (`tools/access.py`): `request_directory_access` (LOCAL|INSTANCE|SESSION) asks the host hook for out-of-workspace access; grants are one `path → mode` pair in one of three stores — `WorkspacePolicy.config_allow` (static `[agent].allow_paths`), `SessionState.access_grants` (human-approved, persisted on the session row), `WorkspacePolicy.yolo_allow` (`--yes`/`/yolo`, process only). `AccessDecision(mode, persist)` from the hook; denylisted paths can never be granted; cap `MAX_ACCESS_GRANTS`. Not a sandbox: `exec` grants (and any command) reach the whole FS.
 - **`file`**: `read_file` (`with_lineno`, `max_chars` default 32k; results start with a 1-based `L{begin}-{end}` range), `write_file`, `listdir`, `tree` (model picks `format=markdown` default / `flat` paths / `decorated`; VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `regex_files` (Python `re` per line, not a glob; literal file/dir `path`, no glob expansion; skips VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range; only lines read via `with_lineno`; read state resets on any file write/edit/copy/move/delete), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind), `get_truncated` (resume any truncated result via its `truncate_token=...`).
 - **`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 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).
+- 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) — and detached runners (`nohup`), each resolved past env wrappers and `<launcher> run`; 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). 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`).
@@ -119,7 +119,7 @@ Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
 - 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 the in-flight turn (a session-scoped asyncio SIGINT handler routes it, `cli/interrupt.py` — a stray Ctrl+C while the loop waits never tears the REPL down); TTY confirms/ask prompts run off-loop and register their own SIGINT target, so Ctrl+C there cancels just the poll-based read (`wait` reports `cancelled by user`) or is ignored with a hint for readline prompts (`ask_user_*`, confirms) — never the turn, never the loop; auto-retry waits 5s/10s/15s/20s then +10s (10 attempts), its countdown is Ctrl+C-cancellable, and the budget **resets when an attempt completes a model round** (connection recovered); then `/retry`.
-- Pretty tool lines: every builtin tool prints its `* Verb 'target' ` prefix as the call starts and appends the outcome (`L1-4 (done)`, `(exit code 0)`, `(session 3)`) when the result lands, so a blocking tool stays visible while it runs; namespaced MCP tools share one generic line (`* MCP <server>:<tool> (done)`). Plugin tools and custom tool calls keep the `[tool] …` / `[tool ok] …` whole-line style. Interactive TTY only (`interactive_terminal()`): piped/plain output and `--verbose` keep whole lines. A second call before the first result (parallel batch) erases the open prefix — only that line, never the streamed line above it — and falls back to whole lines.
+- Pretty tool lines: every builtin tool prints its `* Verb 'target' ` prefix as the call starts and appends the outcome (`L1-4 (done)`, `(exit code 0)`, `(session 3)`) when the result lands, so a blocking tool stays visible while it runs; namespaced MCP tools share one generic line (`* MCP <server>:<tool> (done)`). Plugin tools and custom tool calls keep the `[tool] …` / `[tool ok] …` whole-line style. Interactive TTY only (`interactive_terminal()`): piped/plain output and `--verbose` keep whole lines. A second call before the first result (parallel batch), or a prompt box printed while the tool runs, *ends* the open prefix as its own line and falls back to whole lines for the rest of the batch — the cursor is never moved, so a prefix wider than the terminal cannot be half-erased.
 - **`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.
 

src/plyngent/cli/display.py

@@ -181,7 +181,10 @@ class PrettyLine:
 
     ``prefix`` is written as soon as the call starts — a blocking tool call is
     then visible while it runs — and ``detail`` is appended once its result
-    lands. Together they equal the single line the whole-line path prints.
+    lands, so the two land on one line. Together they equal the single line the
+    whole-line path prints. When something else is printed in between (a
+    parallel call, a confirm/ask prompt) the prefix stays as its own line and
+    the result is printed as a complete line instead.
     """
 
     prefix: Callable[[str], str]
@@ -867,14 +870,14 @@ def _echo_stream(text: str) -> None:
         _ = sys.stdout.flush()
 
 
-def _clear_streamed_lines(line_count: int, *, resume_above: bool = False) -> None:
+def _clear_streamed_lines(line_count: int) -> None:
     """Clear exactly *line_count* terminal lines ending at the cursor line.
 
     The cursor ends at column 0 of the topmost cleared line, so a re-render
-    overwrites the erased block in place instead of appending below it. With
-    ``resume_above`` the cursor steps one line further up: the next
-    ``"\\n"``-prefixed write then reuses the last cleared line — needed by the
-    parallel-batch fallback, which must not touch the line above the prefix.
+    overwrites the erased block in place instead of appending below it. Only the
+    assistant markdown re-render uses this: pretty tool lines never move the
+    cursor, because an interrupted prefix is *closed* as its own line instead
+    (see :class:`_PrettyToolStream`).
 
     ``line_count`` counts *physical* rows; callers fold wrapped lines via
     :func:`_line_count_for_clear`.
@@ -885,8 +888,6 @@ def _clear_streamed_lines(line_count: int, *, resume_above: bool = False) -> Non
     for _ in range(line_count - 1):
         _ = sys.stdout.write("\r\033[2K\033[1A")
     _ = sys.stdout.write("\r\033[2K")
-    if resume_above:
-        _ = sys.stdout.write("\033[1A")
     with contextlib.suppress(OSError):
         _ = sys.stdout.flush()
 
@@ -926,10 +927,12 @@ class _PrettyToolStream:
 
     Writing the prefix immediately keeps a blocking tool call visible; appending
     the detail later reproduces the single whole-line write byte for byte, so
-    non-interactive output is unchanged. The calls of one batch are yielded
-    before their results, so a second call while a prefix is open (parallel
-    tools) cannot append its detail in order: that prefix is erased and the
-    batch falls back to whole lines.
+    non-interactive output is unchanged. Nothing ever moves the cursor: when
+    something else is printed in between — a second call of the batch (parallel
+    tools) or a confirm/ask prompt — the open prefix is *closed* as its own line
+    and the rest of the batch prints whole lines (``prefix + detail``). A prefix
+    wider than the terminal therefore degrades cleanly instead of leaving a
+    half-erased row behind.
     """
 
     _interactive: bool
@@ -946,17 +949,15 @@ class _PrettyToolStream:
         if not (self._interactive and prefix) or self._whole_lines:
             return
         if self._open:
-            # Parallel batch: the open prefix could never receive its detail.
-            # Erase only the prefix line — the line above may be streamed
-            # reasoning — and let the fallback's leading newline reuse it.
-            _clear_streamed_lines(1, resume_above=True)
-            self._open = False
-            self._whole_lines = True
+            # Parallel batch: the open prefix can never receive its detail, so
+            # end it as its own line and print whole lines from here on.
+            self.close_line()
             return
         click.echo(f"\n{prefix}", nl=False)
         with contextlib.suppress(OSError):
             _ = sys.stdout.flush()
         self._open = True
+        _mark_open_prefix(self)
 
     def finish_call(self, detail: str) -> bool:
         """Append *detail* to the open prefix; False when a whole line is needed."""
@@ -964,6 +965,7 @@ class _PrettyToolStream:
             return False
         click.echo(f"{detail}\n", nl=False)
         self._open = False
+        _mark_open_prefix(None)
         return True
 
     def close_line(self) -> None:
@@ -973,6 +975,7 @@ class _PrettyToolStream:
         click.echo()
         self._open = False
         self._whole_lines = True
+        _mark_open_prefix(None)
 
     def end_batch(self) -> None:
         """Reset batch-scoped state once every call of the batch has a result."""
@@ -980,6 +983,29 @@ class _PrettyToolStream:
         self._whole_lines = False
 
 
+# The stream with a prefix line still open (cursor mid-line), or None. A confirm
+# box or ``ask_user_*`` prompt is printed from a worker thread while the tool
+# runs, so it needs a way to end that line before writing its own.
+_open_prefix_stream: _PrettyToolStream | None = None
+
+
+def _mark_open_prefix(stream: _PrettyToolStream | None) -> None:
+    global _open_prefix_stream  # noqa: PLW0603
+    _open_prefix_stream = stream
+
+
+def close_open_pretty_line() -> None:
+    """End an open pretty prefix line so other output starts on its own line.
+
+    Wired into the prompt backend (:func:`plyngent.prompting.set_prompt_output_hook`):
+    without it a confirm/ask prompt raised while a tool runs is written onto the
+    still-open prefix line, and the result would later be appended to that mess.
+    """
+    stream = _open_prefix_stream
+    if stream is not None:
+        stream.close_line()
+
+
 def print_markdown(text: str, *, label: str = "assistant:") -> None:
     """Render *text* as markdown via Rich; *label* on its own line when set."""
     from rich.console import Console

src/plyngent/cli/limits.py

@@ -7,6 +7,7 @@ import sys
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal
 
+from plyngent.cli.display import close_open_pretty_line
 from plyngent.cli.interrupt import off_loop_prompt, pause_task_cancel_for_prompt
 from plyngent.prompting import (
     ChoiceOption,
@@ -18,6 +19,7 @@ from plyngent.prompting import (
     configure_prompting,
     confirm,
     get_prompt_backend,
+    set_prompt_output_hook,
 )
 from plyngent.tools.access import AccessDecision
 from plyngent.tools.process.pty_session import PtyManager
@@ -580,4 +582,5 @@ def install_cli_limit_hooks() -> None:
     process-global hook).
     """
     configure_prompting(pause_factory=off_loop_prompt)
+    set_prompt_output_hook(close_open_pretty_line)
     PtyManager.set_limit_continue_hook(prompt_continue_limit)

src/plyngent/prompting.py

@@ -93,6 +93,30 @@ class PromptBackend(Protocol):
     def secho(self, message: str, *, fg: str | None = None, err: bool = False) -> None: ...
 
 
+# Hook run right before a backend writes prompt chrome (a confirm box, a question
+# label). Hosts install one so an open pretty tool line — see
+# ``plyngent.cli.display`` — is ended before the prompt takes that line over.
+_before_output: Callable[[], None] | None = None
+
+
+def set_prompt_output_hook(hook: Callable[[], None] | None) -> None:
+    """Install the hook run before prompt output (None clears it)."""
+    global _before_output  # noqa: PLW0603
+    _before_output = hook
+
+
+def get_prompt_output_hook() -> Callable[[], None] | None:
+    """The installed prompt-output hook, if any."""
+    return _before_output
+
+
+def _break_output() -> None:
+    """Run the prompt-output hook, if one is installed."""
+    hook = _before_output
+    if hook is not None:
+        hook()
+
+
 def _readline_input(prompt: str, *, completions: Sequence[str] | None = None) -> str:
     """``input()`` with optional Tab completion via readline when available."""
     try:
@@ -255,9 +279,11 @@ class ClickPromptBackend:
             raise NonInteractiveError(msg) from exc
 
     def echo(self, message: str = "", *, err: bool = False) -> None:
+        _break_output()
         click.echo(message, err=err)
 
     def secho(self, message: str, *, fg: str | None = None, err: bool = False) -> None:
+        _break_output()
         click.secho(message, fg=fg, err=err)
 
 
@@ -319,10 +345,11 @@ def get_prompt_backend() -> PromptBackend:
 
 
 def reset_prompting() -> None:
-    """Restore default Click backend, clear the pause hook and cancel flag (tests)."""
-    global _backend, _pause_factory  # noqa: PLW0603
+    """Restore the default backend; clear the pause/output hooks and cancel flag (tests)."""
+    global _backend, _pause_factory, _before_output  # noqa: PLW0603
     _backend = ClickPromptBackend()
     _pause_factory = None
+    _before_output = None
     _prompt_cancel.clear()
 
 

src/plyngent/tools/__init__.py

@@ -88,6 +88,7 @@ from .workspace import (
 )
 from .workspace import DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS as DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS
 from .workspace import AccessMode as AccessMode
+from .workspace import DenylistHit as DenylistHit
 from .workspace import WorkspaceError as WorkspaceError
 from .workspace import WorkspacePolicy as WorkspacePolicy
 from .workspace import active_workspace_policy as active_workspace_policy
@@ -104,6 +105,7 @@ 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 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
 from .workspace import parse_access_mode as parse_access_mode
 from .workspace import pop_owned_temporary_workspaces as pop_owned_temporary_workspaces

src/plyngent/tools/command_scan.py

@@ -0,0 +1,290 @@
+"""Resolve the program a command really runs, past wrappers and launchers.
+
+Confirming a risky command (``danger``) and denying a policy-blocked one
+(``workspace.check_command_allowed``) must judge the program that will really
+run, not the outermost token: ``env FOO=1 python -c …``, ``nohup pdm run python
+…`` and ``timeout 5 bash -c …`` all reach the interpreter — or the denylisted
+program — further in.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+__all__ = ["UnwrappedCommand", "basename", "is_interpreter", "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.
+_RUN_LAUNCHERS: frozenset[str] = frozenset(
+    {"conda", "hatch", "mise", "npm", "pdm", "pipenv", "pnpm", "poetry", "uv", "yarn"}
+)
+
+# Programs that run code handed to them as text: for these, ``-c`` carries the
+# code (``grep -c`` / ``od -c`` is a plain flag, never a block of code).
+_INTERPRETERS: frozenset[str] = frozenset(
+    {
+        "bash",
+        "bun",
+        "cmd",
+        "csh",
+        "dash",
+        "deno",
+        "fish",
+        "ghci",
+        "ipython",
+        "ipython3",
+        "irb",
+        "jshell",
+        "julia",
+        "ksh",
+        "lua",
+        "mongo",
+        "mysql",
+        "node",
+        "nodejs",
+        "perl",
+        "php",
+        "powershell",
+        "pry",
+        "psql",
+        "pwsh",
+        "python",
+        "python2",
+        "python3",
+        "r",
+        "redis-cli",
+        "ruby",
+        "scala",
+        "sh",
+        "sqlite3",
+        "tcsh",
+        "zsh",
+    }
+)
+
+# Trailing version of an interpreter: ``python3.12`` / ``node20`` / ``php8.2``.
+_VERSION_SUFFIX = re.compile(r"[0-9][0-9.]*$")
+
+# ``FOO=bar`` in front of a program (env/sudo style).
+_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.
+_VALUE_OPTS: dict[str, frozenset[str]] = {
+    "command": frozenset({"-p", "-v", "-V"}),
+    "doas": frozenset({"-u", "--user"}),
+    "env": frozenset({"-C", "-S", "-u", "--chdir", "--split-string", "--unset"}),
+    "exec": frozenset({"-a", "-c", "-l"}),
+    "ionice": frozenset({"-c", "-n", "-p", "-P", "-u"}),
+    "nice": frozenset({"-n", "--adjustment"}),
+    "pkexec": frozenset({"--disable-internal-agent", "--user"}),
+    "stdbuf": frozenset({"-e", "-i", "-o"}),
+    "su": frozenset({"-c", "-g", "-G", "-p", "-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",
+        }
+    ),
+    "time": frozenset({"-f", "-o", "--format", "--output"}),
+    "timeout": frozenset({"-k", "-s", "--kill-after", "--signal"}),
+    "watch": frozenset({"-n"}),
+    "xargs": frozenset({"-a", "-d", "-E", "-e", "-I", "-L", "-n", "-P", "-s"}),
+}
+
+# Launcher options that consume a value (``uv run --with rich python``).
+_RUN_VALUE_OPTS: frozenset[str] = frozenset(
+    {
+        "-C",
+        "-e",
+        "-F",
+        "-G",
+        "-m",
+        "-n",
+        "-p",
+        "-s",
+        "-w",
+        "--config",
+        "--constraint",
+        "--default-index",
+        "--directory",
+        "--env",
+        "--env-file",
+        "--extra",
+        "--extras",
+        "--file",
+        "--group",
+        "--index",
+        "--module",
+        "--name",
+        "--overrides",
+        "--package",
+        "--prefix",
+        "--project",
+        "--python",
+        "--python-version",
+        "--script",
+        "--with",
+        "--without",
+    }
+)
+
+
+@dataclass(frozen=True)
+class UnwrappedCommand:
+    """A command plus the program it really runs, past wrappers and launchers."""
+
+    argv: tuple[str, ...]
+    offset: int
+    wrappers: tuple[str, ...]
+    base: str
+    self_review: tuple[str, ...]
+
+    @property
+    def command(self) -> tuple[str, ...]:
+        """The program and its arguments (the whole argv when nothing unwrapped)."""
+        return self.argv[self.offset :]
+
+
+def basename(argv0: str) -> str:
+    """Lower-cased basename of an executable token (``\\`` separators, ``.exe``)."""
+    name = argv0.replace("\\", "/").rsplit("/", 1)[-1].lower()
+    return name.removesuffix(".exe")
+
+
+def is_interpreter(base: str) -> bool:
+    """Whether *base* runs code from its own command line.
+
+    Version suffixes are ignored, so ``python3.12`` / ``node20`` / ``php8.2``
+    count as interpreters too.
+    """
+    return base in _INTERPRETERS or _VERSION_SUFFIX.sub("", base) in _INTERPRETERS
+
+
+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.
+    """
+    parts = list(argv)
+    wrappers: list[str] = []
+    index = 0
+    while index < len(parts):
+        name = basename(parts[index])
+        nxt = _skip_wrapper(name, parts, index + 1)
+        if nxt is None:
+            break
+        wrappers.append(name)
+        index = nxt
+    base = basename(parts[index]) if index < len(parts) else ""
+    chain = dict.fromkeys((*wrappers, base))
+    return UnwrappedCommand(
+        argv=tuple(parts),
+        offset=index,
+        wrappers=tuple(wrappers),
+        base=base,
+        self_review=tuple(name for name in chain if name in _SELF_REVIEW),
+    )
+
+
+def _skip_wrapper(name: str, parts: list[str], start: int) -> 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,
+            value_opts=_VALUE_OPTS.get(name, frozenset()),
+            leading_duration=name in _DURATION_WRAPPERS,
+        )
+        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, value_opts=_RUN_VALUE_OPTS, leading_duration=False)
+        return nxt if nxt < len(parts) else None
+    return None
+
+
+def _skip_options(
+    parts: list[str],
+    start: int,
+    *,
+    value_opts: frozenset[str],
+    leading_duration: bool,
+) -> int:
+    """Index of the first token after a wrapper's own options and assignments."""
+    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:
+                index += 1
+            continue
+        if leading_duration and not took_duration and _DURATION.fullmatch(token):
+            took_duration = True
+            index += 1
+            continue
+        return index
+    return index

src/plyngent/tools/danger.py

@@ -4,61 +4,13 @@ import json
 import shlex
 from typing import TYPE_CHECKING, cast
 
+from plyngent.tools.command_scan import is_interpreter, unwrap_command
 from plyngent.tools.workspace import AccessMode, WorkspaceError, resolve_path
 
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
 
 
-_SHELL_BASENAMES: frozenset[str] = frozenset(
-    {
-        "bash",
-        "sh",
-        "zsh",
-        "fish",
-        "dash",
-        "ksh",
-        "csh",
-        "tcsh",
-        "powershell",
-        "pwsh",
-        "cmd",
-        "cmd.exe",
-        "python",
-        "python3",
-        "python2",
-        "ipython",
-        "ipython3",
-        "node",
-        "nodejs",
-        "deno",
-        "bun",
-        "ruby",
-        "perl",
-        "php",
-        "lua",
-        "r",
-        "julia",
-        "irb",
-        "pry",
-        "ghci",
-        "scala",
-        "jshell",
-        "sqlite3",
-        "psql",
-        "mysql",
-        "mongo",
-        "redis-cli",
-    }
-)
-
-
-def _basename(argv0: str) -> str:
-    name = argv0.replace("\\", "/").rsplit("/", 1)[-1]
-    name = name.removesuffix(".exe")
-    return name.lower()
-
-
 def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | None:
     command = args.get(key)
     if not isinstance(command, list) or not command:
@@ -71,67 +23,60 @@ def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] |
     return out
 
 
-def _find_dash_c_code(argv: Sequence[str]) -> str | None:
-    """Return the argument after ``-c`` / ``-c…`` if present (python/bash/node style)."""
-    for index, part in enumerate(argv[1:], start=1):
+def _dash_c_code(command: Sequence[str]) -> tuple[int, str] | None:
+    """Index and text of the ``-c`` code in an interpreter argv, if present.
+
+    Only interpreters hand ``-c`` a block of code; for anything else it is a
+    plain flag (``grep -c``, ``od -c``) with nothing to review.
+    """
+    for index, part in enumerate(command[1:], start=1):
         if part == "-c":
-            return argv[index + 1] if index + 1 < len(argv) else ""
+            return index, command[index + 1] if index + 1 < len(command) else ""
         # Combined short forms are uncommon; only exact -c is supported.
     return None
 
 
 def _indent_block(text: str, *, prefix: str = "  ") -> str:
-    """Indent every line of *text* (for multi-line -c bodies in confirm prompts)."""
+    """Indent every line of *text* (for multi-line -c code in confirm prompts)."""
     if not text:
         return prefix
     return "\n".join(prefix + line if line else prefix.rstrip() for line in text.splitlines())
 
 
-def _format_argv_for_confirm(argv: Sequence[str], *, code: str | None) -> str:
-    """One-line argv summary; replace -c payload with ``$(command)`` when present."""
-    if code is None:
-        return shlex.join(list(argv))
-    parts: list[str] = []
-    skip_next = False
-    for part in argv:
-        if skip_next:
-            skip_next = False
-            continue
-        if part == "-c":
-            parts.append("-c")
-            parts.append("$(command)")
-            skip_next = True
-            continue
-        parts.append(part)
+def _format_argv_for_confirm(argv: Sequence[str], *, code_index: int | None) -> str:
+    """One-line argv summary; the ``-c`` code token becomes ``$(command)``."""
+    parts = list(argv)
+    if code_index is not None and code_index + 1 < len(parts):
+        parts[code_index + 1] = "$(command)"
     return shlex.join(parts)
 
 
-def _shell_or_dash_c_reason(argv: Sequence[str], *, via: str) -> str | None:
-    """Confirm interactive shells and ``*-c`` one-liners so the user can inspect argv.
+def _command_reason(argv: Sequence[str], *, via: str) -> str | None:
+    """Confirm a command that runs code or detaches, whatever wraps it.
 
     Multi-line reason (shown inside the CLI confirm box). ``via`` is a short
     label such as ``run_argv`` or ``open_pty``.
 
-    For ``-c`` scripts, argv shows ``$(command)`` instead of inlining the body;
-    the full script is printed below untruncated, with every line indented.
+    The program is resolved past environment wrappers and ``<launcher> run``
+    forms (``env FOO=1 python …``, ``nohup pdm run python …``), and an
+    interpreter is confirmed the same way however it runs — a bare shell, a
+    script (``python x.py``), or a ``-c`` one-liner, whose code is printed below
+    ``command:`` instead of inline.
     """
-    if not argv:
+    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}")
+    if not findings:
         return None
-    base = _basename(argv[0])
-    code = _find_dash_c_code(argv)
-    display = _format_argv_for_confirm(argv, code=code)
-
+    code = _dash_c_code(scan.command) if interpreter else None
+    code_index = scan.offset + code[0] if code is not None else None
+    display = _format_argv_for_confirm(argv, code_index=code_index)
+    reason = f"{via}: {' + '.join(findings)} — review before allow\n  argv: {display}"
     if code is not None:
-        body = _indent_block(code, prefix="  ")
-        return f"{via}: {base} -c (review code before allow)\n  argv: {display}\n  command:\n{body}"
-
-    if base in _SHELL_BASENAMES and len(argv) == 1:
-        return f"{via}: interactive {base!r} (review before allow)\n  argv: {display}"
-
-    if base in _SHELL_BASENAMES and "-c" not in argv[1:]:
-        return f"{via}: shell/runtime {base!r} without -c (review before allow)\n  argv: {display}"
-
-    return None
+        reason += f"\n  command:\n{_indent_block(code[1])}"
+    return reason
 
 
 def _write_file_reason(args: Mapping[str, object]) -> str | None:
@@ -185,7 +130,7 @@ def _run_argv_reason(args: Mapping[str, object]) -> str | None:
     argv = _as_argv(args, key="argv")
     if argv is None:
         return None
-    return _shell_or_dash_c_reason(argv, via="run_argv")
+    return _command_reason(argv, via="run_argv")
 
 
 def _batch_step_argv(item: object) -> list[str] | None:
@@ -204,7 +149,7 @@ def _batch_step_argv(item: object) -> list[str] | None:
 
 
 def _run_argv_batch_reason(args: Mapping[str, object]) -> str | None:
-    """One confirm for the whole batch if any step is shell/REPL/-c."""
+    """One confirm for the whole batch if any step runs code or detaches."""
     raw = args.get("steps")
     if isinstance(raw, str):
         try:
@@ -218,7 +163,7 @@ def _run_argv_batch_reason(args: Mapping[str, object]) -> str | None:
         reason
         for index, item in enumerate(cast("list[object]", raw))
         if (argv := _batch_step_argv(item)) is not None
-        and (reason := _shell_or_dash_c_reason(argv, via=f"run_argv_batch[{index}]")) is not None
+        and (reason := _command_reason(argv, via=f"run_argv_batch[{index}]")) is not None
     ]
     if not risky:
         return None
@@ -230,7 +175,7 @@ def _open_pty_reason(args: Mapping[str, object]) -> str | None:
     argv = _as_argv(args)
     if argv is None:
         return None
-    return _shell_or_dash_c_reason(argv, via="open_pty")
+    return _command_reason(argv, via="open_pty")
 
 
 def _fetch_reason(args: Mapping[str, object]) -> str | None:
@@ -250,8 +195,9 @@ def classify_danger(name: str, args: Mapping[str, object]) -> str | None:  # noq
     """Return a short reason if ``name``/``args`` need user confirm, else ``None``.
 
     Hard denylists (paths/commands) still raise independently. This only covers
-    soft confirms for mutating tools and risky shell/REPL launches
-    (interactive shells and ``python -c`` / ``bash -c`` one-liners).
+    soft confirms for mutating tools and risky command launches — interpreters
+    (however invoked) and detached runners such as ``nohup``, resolved past
+    environment wrappers and ``<launcher> run``.
     Private/loopback fetch targets use a separate policy grant (not YOLO).
     """
     if name == "delete_path":

src/plyngent/tools/file/get_truncated.py

@@ -7,7 +7,6 @@ from typing import cast
 from plyngent.agent import ToolTag, tool
 from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS
 from plyngent.tools.file.read import line_range_label, read_file, read_raw_text
-from plyngent.tools.net.fetch import fetch
 from plyngent.tools.truncate_token import (
     TruncateToken,
     decode_truncate_token,
@@ -61,19 +60,18 @@ async def get_truncated(token: str, *, max_chars: int = DEFAULT_TOOL_RESULT_MAX_
     """Fetch the next chunk of a truncated result using its ``truncate_token``.
 
     Pass the token shown at the end of a truncated ``read_file``, ``fetch``,
-    ``run_command``, or earlier ``get_truncated`` result to keep reading without
+    ``run_argv``, or earlier ``get_truncated`` result to keep reading without
     re-requesting the whole source or raising limits. Truncated output carries a
     fresh token, so chunks chain indefinitely. File chunks resume in the same
     view as the original read: numbered reads continue as numbered lines
     (editable via ``edit_lineno``), plain reads as raw text with a 1-based
-    ``L{begin}-{end}`` range. Memory chunks (generic tool output) are served
-    from a short-lived in-memory store that is forgotten when the agent exits.
+    ``L{begin}-{end}`` range. Memory chunks (generic tool output, and ``fetch``
+    bodies) are served from a short-lived in-memory store that is forgotten when
+    the agent exits, so resuming a fetch never repeats the request.
     """
     parsed = decode_truncate_token(token)
     if parsed is None:
         return "error: invalid truncate token"
-    if parsed.kind == "http":
-        return await fetch.handler(parsed.location, offset=parsed.offset, max_chars=parsed.limit)
     if parsed.kind == "memory":
         return _memory_chunk(parsed, max_chars)
     text, err = read_raw_text(parsed.location)

src/plyngent/tools/net/fetch.py

@@ -17,7 +17,7 @@ from plyngent.tools.net.policy import (
     parse_fetch_url,
     soft_confirm_reason,
 )
-from plyngent.tools.truncate_token import truncate_with_token
+from plyngent.tools.truncate_token import truncate_to_memory
 
 
 def format_fetch_result(
@@ -37,16 +37,10 @@ def format_fetch_result(
     offset: int = 0,
 ) -> str:
     start = max(0, offset)
-    body = body_text[start:]
-    body, token = truncate_with_token(
-        body,
-        max_chars,
-        kind="http",
-        location=final_url,
-        offset=start,
-        limit=max_chars,
-        total_len=len(body_text),
-    )
+    # The body is already in hand, so the remainder is kept in memory rather than
+    # re-requested: resuming a fetch must not cost another round trip (latency)
+    # nor re-hit the remote (rate limits / risk control).
+    body, token = truncate_to_memory(body_text[start:], max_chars)
     char_truncated = token is not None
     warn_line = "; ".join(warnings) if warnings else ""
     parts = [

src/plyngent/tools/truncate_token.py

@@ -1,7 +1,7 @@
 """Truncate-token cursors for resuming truncated tool results.
 
 Every truncation site (the agent loop's generic tool-result cap,
-``run_command`` / ``run_command_batch`` per-stream caps, ``read_file``,
+``run_argv`` / ``run_argv_batch`` per-stream caps, ``read_file``,
 ``fetch``, and request-time compact shrinks) appends the same compact marker
 when output is cut short::
 
@@ -12,11 +12,13 @@ re-requesting the whole source or raising limits. Tokens chain: each truncated
 chunk carries a fresh token, so ``get_truncated`` keeps resuming until the
 source is exhausted (then no marker is emitted).
 
-Resumable sources (``file`` / ``http``) are re-read by ``location`` +
-``offset``. Arbitrary tool output (run_command stdout, todo renders, …) has no
-resumable source, so its remainder lives in a short-lived in-memory store
-(``kind="memory"``) that is forgotten when the agent process exits — the model
-must re-run the tool after that.
+A ``file`` remainder is re-read from disk by ``location`` + ``offset``, so it
+survives for as long as the file does. Everything else — arbitrary tool output
+(run_argv stdout, todo renders, …) and ``fetch`` bodies — has no source worth
+re-running (a re-fetch costs a round trip and re-hits the remote), so its
+remainder lives in a short-lived in-memory store (``kind="memory"``) that is
+forgotten when the agent process exits; the model must re-run the tool after
+that.
 """
 
 from __future__ import annotations
@@ -27,18 +29,18 @@ from typing import Literal
 
 import msgspec
 
-type TruncateKind = Literal["file", "http", "memory"]
+type TruncateKind = Literal["file", "memory"]
 
 
 class TruncateToken(msgspec.Struct, frozen=True):
     """Opaque cursor into a truncated source.
 
-    ``offset``/``limit`` are in characters (http bodies, memory remainders) or
-    lines (files is line-based via ``read_file``); ``get_truncated`` interprets
-    by ``kind``. For ``memory``, ``location`` is a key into the in-memory
-    remainder store and ``offset``/``limit`` index characters of that stored
-    remainder. ``numbered`` marks file tokens from ``read_file(with_lineno)``:
-    the continuation resumes the numbered view instead of raw text.
+    ``offset``/``limit`` are in characters (memory remainders) or lines (files
+    is line-based via ``read_file``); ``get_truncated`` interprets by ``kind``.
+    For ``memory``, ``location`` is a key into the in-memory remainder store and
+    ``offset``/``limit`` index characters of that stored remainder. ``numbered``
+    marks file tokens from ``read_file(with_lineno)``: the continuation resumes
+    the numbered view instead of raw text.
     """
 
     kind: TruncateKind
@@ -134,17 +136,17 @@ def truncate_with_token(
     return text + truncation_marker(max_chars, omitted, token), token
 
 
-def truncate_generic(text: str, max_chars: int) -> str:
-    """Cap arbitrary tool output; embed a memory truncate token (chainable).
+def truncate_to_memory(text: str, max_chars: int) -> tuple[str, TruncateToken | None]:
+    """Cap ``text`` (a whole source) with a memory token over a stored copy.
 
-    The full remainder is kept in the in-memory store so ``get_truncated`` can
-    resume it. Returns ``text`` unchanged (no marker) when it fits within
-    ``max_chars``.
+    Nothing is stored when the text fits, so a caller that only needs the
+    bounded text does not pay for a remainder. Returns ``(bounded_text, token)``
+    where ``token`` is ``None`` when the source is exhausted.
     """
     if max_chars < 1 or len(text) <= max_chars:
-        return text
+        return text, None
     key = store_remainder(text)
-    bounded, _ = truncate_with_token(
+    return truncate_with_token(
         text,
         max_chars,
         kind="memory",
@@ -153,4 +155,14 @@ def truncate_generic(text: str, max_chars: int) -> str:
         limit=max_chars,
         total_len=len(text),
     )
+
+
+def truncate_generic(text: str, max_chars: int) -> str:
+    """Cap arbitrary tool output; embed a memory truncate token (chainable).
+
+    The full remainder is kept in the in-memory store so ``get_truncated`` can
+    resume it. Returns ``text`` unchanged (no marker) when it fits within
+    ``max_chars``.
+    """
+    bounded, _ = truncate_to_memory(text, max_chars)
     return bounded

src/plyngent/tools/workspace.py

@@ -5,6 +5,8 @@ from enum import IntEnum
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
+from plyngent.tools.command_scan import basename, unwrap_command
+
 if TYPE_CHECKING:
     from collections.abc import Callable, Mapping, Sequence
 
@@ -16,6 +18,10 @@ DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset(
         "su",
         "doas",
         "pkexec",
+        # Privileged systemd entry points: run0 is sudo's replacement,
+        # systemd-run executes a command as a transient system unit.
+        "run0",
+        "systemd-run",
         "rm",
         "rmdir",
         "mkfs",
@@ -31,6 +37,18 @@ DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset(
         "chown",
         "mount",
         "umount",
+        "init",
+        "telinit",
+        "kexec",
+        "pm-suspend",
+        "pm-hibernate",
+        "s2disk",
+        "s2ram",
+        # Machine power state: only these systemctl/loginctl words are blocked,
+        # so read-only calls (``systemctl status``) stay allowed.
+        "systemctl reboot poweroff halt suspend hibernate hybrid-sleep "
+        "suspend-then-hibernate soft-reboot kexec isolate rescue emergency default",
+        "loginctl terminate-session terminate-user kill-session kill-user reboot poweroff soft-reboot",
     }
 )
 
@@ -403,12 +421,88 @@ def argv_shape_error(argv: object) -> str | None:
     return None
 
 
-def check_command_allowed(argv: list[str]) -> None:
-    """Raise if argv is empty or the executable basename is denylisted.
+@dataclass(frozen=True)
+class DenylistHit:
+    """A command denylist entry that a command matches.
 
-    Denylisted basenames are not hard-rejected when a policy confirm hook is
+    ``program`` names the program to prompt for (and grant by); ``keyword`` is
+    the argument word that made a subcommand entry dangerous, or ``None`` for an
+    entry that blocks the whole program.
+    """
+
+    entry: str
+    program: str
+    keyword: str | None = None
+
+
+def match_command_denylist(argv: Sequence[str], denylist: frozenset[str]) -> list[DenylistHit]:
+    """Return the denylist entries *argv* matches, outermost program first.
+
+    An entry is a program basename, optionally followed by the argument words
+    that make it dangerous (``systemctl reboot poweroff …``). A whole-program
+    entry matches any program of the wrapper chain; a subcommand entry matches
+    only the program that runs, whose arguments are searched for one of its
+    words — so ``systemctl reboot`` and ``systemctl start reboot.target`` are
+    blocked while ``systemctl status`` is not.
+    """
+    scan = unwrap_command(argv)
+    words = _argument_words(scan.command)
+    hits: list[DenylistHit] = []
+    seen: set[str] = set()
+    for name in (*scan.wrappers, scan.base):
+        hit = _denylist_hit(name, base=scan.base, words=words, denylist=denylist)
+        if hit is not None and hit.program not in seen:
+            seen.add(hit.program)
+            hits.append(hit)
+    return hits
+
+
+def _denylist_hit(
+    name: str,
+    *,
+    base: str,
+    words: frozenset[str],
+    denylist: frozenset[str],
+) -> DenylistHit | None:
+    """First denylist entry matching program *name*, if any."""
+    for entry in sorted(denylist):
+        parts = entry.split()
+        if not parts or basename(parts[0]) != name:
+            continue
+        if len(parts) == 1:
+            return DenylistHit(entry=entry, program=name)
+        if name != base:
+            # Argument words describe the program that runs, not a wrapper.
+            continue
+        keyword = next((word for word in parts[1:] if word.lower() in words), None)
+        if keyword is not None:
+            return DenylistHit(entry=entry, program=name, keyword=keyword.lower())
+    return None
+
+
+def _argument_words(command: Sequence[str]) -> frozenset[str]:
+    """Lower-cased argument words of *command*, plus ``*.target`` stems."""
+    words: set[str] = set()
+    for part in command[1:]:
+        token = part.lower()
+        words.add(token)
+        if token.endswith(".target"):
+            words.add(token.removesuffix(".target"))
+    return frozenset(words)
+
+
+def check_command_allowed(argv: list[str]) -> None:
+    """Raise if argv is empty or any program in the command chain is denylisted.
+
+    The chain is the program that really runs plus every wrapper it is reached
+    through (``env FOO=1 rm``, ``pdm run rm``, ``sudo -u x rm``), so a denylisted
+    program cannot hide behind one. Basenames are compared lower-cased, so
+    ``RM`` and ``rm.exe`` count as ``rm``; entries that list dangerous arguments
+    (``systemctl reboot …``) match those words in the program's own arguments.
+
+    Denylisted programs are not hard-rejected when a policy confirm hook is
     installed: the human is asked (with a timeout; default deny). Session grants
-    skip re-prompting for the same basename. Independent of YOLO soft-confirm.
+    skip re-prompting for the same program. Independent of YOLO soft-confirm.
     """
     shape_error = argv_shape_error(argv)
     if shape_error is not None:
@@ -418,25 +512,25 @@ def check_command_allowed(argv: list[str]) -> None:
         msg = "command argv must not be empty"
         raise WorkspaceError(msg)
     policy = active_workspace_policy()
-    binary = Path(argv[0]).name
-    if binary not in policy.command_denylist:
-        return
-    if binary in policy.policy_allowed_commands:
-        return
-    hook = policy.policy_confirm_hook
-    if hook is not None:
+    for hit in match_command_denylist(argv, policy.command_denylist):
+        binary = hit.program
+        if binary in policy.policy_allowed_commands:
+            continue
+        why = f": {hit.keyword!r}" if hit.keyword else ""
+        hook = policy.policy_confirm_hook
+        if hook is None:
+            msg = f"command denied by policy (basename {binary!r} is blocked{why})"
+            raise WorkspaceError(msg)
         timeout = policy.policy_confirm_timeout_seconds
         try:
             allowed = bool(hook(binary, list(argv), timeout))
         except Exception as exc:
             msg = f"command denied by policy (basename {binary!r}; confirm failed: {exc})"
             raise WorkspaceError(msg) from exc
-        if allowed:
-            policy.policy_allowed_commands.add(binary)
-            return
-        msg = (
-            f"command denied by policy (basename {binary!r} is blocked; user declined or timed out after {timeout:g}s)"
-        )
-        raise WorkspaceError(msg)
-    msg = f"command denied by policy (basename {binary!r} is blocked)"
-    raise WorkspaceError(msg)
+        if not allowed:
+            msg = (
+                f"command denied by policy (basename {binary!r} is blocked{why}; "
+                f"user declined or timed out after {timeout:g}s)"
+            )
+            raise WorkspaceError(msg)
+        policy.policy_allowed_commands.add(binary)

tests/test_cli/test_display.py

@@ -7,6 +7,7 @@ from plyngent.cli.display import (
     _clear_streamed_lines,
     _line_count_for_clear,
     _pretty_line_for,
+    close_open_pretty_line,
     get_markdown_enabled,
     markdown_render_available,
     print_markdown,
@@ -951,11 +952,11 @@ async def test_pretty_verbose_keeps_full_result_output(
     assert "* Read 'a.txt' " not in out
 
 
-async def test_pretty_parallel_batch_erases_open_prefix(
+async def test_pretty_parallel_batch_closes_the_open_prefix(
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],
 ) -> None:
-    """A second call in the batch erases the open prefix; results print whole."""
+    """A second call in the batch ends the open prefix; results print whole lines."""
     monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
     await render_events(
         _aiter(
@@ -968,20 +969,20 @@ async def test_pretty_parallel_batch_erases_open_prefix(
         )
     )
     out = capsys.readouterr().out
-    assert "\x1b[2K\x1b[1A" in out  # the in-flight prefix was erased
-    assert "* Read 'a.txt' L1-4 (done)\n" in out
-    assert "* Read 'b.txt' L1-9 (done)\n" in out
+    # The prefix survives as its own line; whole lines take over from there.
+    assert out == ("\n* Read 'a.txt' \n\n* Read 'a.txt' L1-4 (done)\n\n* Read 'b.txt' L1-9 (done)\n\n")
+    assert "\x1b[" not in out  # the cursor is never moved
 
 
 async def test_pretty_parallel_batch_keeps_the_line_above_the_prefix(
     monkeypatch: pytest.MonkeyPatch,
     capsys: pytest.CaptureFixture[str],
 ) -> None:
-    """Regression: erasing the open prefix must not eat the line above it.
+    """Ending the open prefix must not touch the line above it.
 
     Reasoning streams without a trailing newline, so the prefix line sits right
-    below the reasoning text; the erase used to clear one row too many and wiped
-    that reasoning line out of the screen.
+    below the reasoning text; moving the cursor up used to wipe that reasoning
+    line out of the screen.
     """
     monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
     await render_events(
@@ -997,13 +998,31 @@ async def test_pretty_parallel_batch_keeps_the_line_above_the_prefix(
     )
     out = capsys.readouterr().out
     assert out == (
-        "\nreasoning:\nthink"
-        "\n* Read 'a.txt' "
-        "\r\x1b[2K\x1b[1A"
-        "\n* Read 'a.txt' L1-4 (done)\n"
-        "\n* Read 'b.txt' L1-9 (done)\n"
-        "\n\n"
+        "\nreasoning:\nthink\n* Read 'a.txt' \n\n* Read 'a.txt' L1-4 (done)\n\n* Read 'b.txt' L1-9 (done)\n\n\n"
     )
+    assert "\x1b[" not in out
+
+
+async def test_pretty_line_closed_by_out_of_band_output(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A prompt/confirm printed while the call runs ends the open prefix line.
+
+    The prompt backend runs :func:`close_open_pretty_line` before writing, so its
+    box starts on a fresh line and the result then prints whole.
+    """
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+
+    async def _events():
+        yield _pretty_call("read_file", '{"path": "a.txt"}')
+        close_open_pretty_line()  # what a confirm box does mid-call
+        yield _result("L1-4\none\n")
+
+    await render_events(_events())
+    out = capsys.readouterr().out
+    assert out == ("\n* Read 'a.txt' \n\n* Read 'a.txt' L1-4 (done)\n\n")
+    assert "\x1b[" not in out
 
 
 def test_clear_streamed_lines_clears_exactly_the_given_rows(capsys: pytest.CaptureFixture[str]) -> None:
@@ -1012,12 +1031,6 @@ def test_clear_streamed_lines_clears_exactly_the_given_rows(capsys: pytest.Captu
     assert capsys.readouterr().out == "\r\x1b[2K\x1b[1A\r\x1b[2K"
 
 
-def test_clear_streamed_lines_resume_above_steps_back_one_row(capsys: pytest.CaptureFixture[str]) -> None:
-    """``resume_above`` clears the cursor row only, then steps above it."""
-    _clear_streamed_lines(1, resume_above=True)
-    assert capsys.readouterr().out == "\r\x1b[2K\x1b[1A"
-
-
 def test_clear_streamed_lines_ignores_non_positive_counts(capsys: pytest.CaptureFixture[str]) -> None:
     _clear_streamed_lines(0)
     _clear_streamed_lines(-1)

tests/test_cli/test_limits.py

@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
 
 import pytest
 
+from plyngent.cli.display import close_open_pretty_line
 from plyngent.cli.limits import (
     auto_continue_enabled,
     format_tool_confirm_box,
@@ -14,7 +15,7 @@ from plyngent.cli.limits import (
     reset_auto_continue_turn,
     set_auto_continue_default,
 )
-from plyngent.prompting import NonInteractiveBackend, get_prompt_backend, temporary_backend
+from plyngent.prompting import NonInteractiveBackend, get_prompt_backend, get_prompt_output_hook, temporary_backend
 from plyngent.tools.process.pty_session import PtyManager
 from tests.test_prompting import ScriptedBackend
 
@@ -121,6 +122,8 @@ def test_format_tool_confirm_box_multiline() -> None:
 def test_install_cli_limit_hooks() -> None:
     install_cli_limit_hooks()
     assert callable(getattr(PtyManager, "_limit_continue", None))
+    # A prompt raised mid-call must end the open pretty tool line first.
+    assert get_prompt_output_hook() is close_open_pretty_line
     PtyManager.set_limit_continue_hook(None)
     # Backend remains usable after install.
     assert get_prompt_backend().is_interactive() or True

tests/test_prompting.py

@@ -19,11 +19,14 @@ from plyngent.prompting import (
     configure_prompting,
     confirm,
     form,
+    get_prompt_backend,
+    get_prompt_output_hook,
     prompt_cancelled,
     read_line_with_timeout,
     reset_prompting,
     run_cancellable_prompt_async,
     run_prompt_async,
+    set_prompt_output_hook,
     temporary_backend,
 )
 
@@ -257,3 +260,22 @@ def test_non_interactive_confirm_uses_default() -> None:
     with temporary_backend(NonInteractiveBackend()):
         assert confirm("ok?", default=False) is False
         assert confirm("ok?", default=True) is True
+
+
+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.
+
+    The CLI registers one so an open pretty tool line ends before a prompt box.
+    """
+    calls: list[str] = []
+    assert get_prompt_output_hook() is None
+    set_prompt_output_hook(lambda: calls.append("break"))
+    try:
+        backend = get_prompt_backend()
+        backend.echo("box")
+        backend.secho("question", fg="yellow")
+    finally:
+        set_prompt_output_hook(None)
+    assert calls == ["break", "break"]
+    assert get_prompt_output_hook() is None
+    assert "box" in capsys.readouterr().out

tests/test_tools/test_command_scan.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from plyngent.tools.command_scan import basename, is_interpreter, unwrap_command
+
+
+def test_basename() -> None:
+    assert basename("/usr/bin/python3.12") == "python3.12"
+    assert basename("C:\\Tools\\RM.EXE") == "rm"
+    assert basename("pdm") == "pdm"
+
+
+def test_unwrap_plain_command() -> None:
+    scan = unwrap_command(["grep", "-c", "needle", "f.txt"])
+    assert scan.wrappers == ()
+    assert scan.offset == 0
+    assert scan.base == "grep"
+    assert scan.command == ("grep", "-c", "needle", "f.txt")
+    assert scan.self_review == ()
+
+
+def test_unwrap_env_assignment_and_options() -> None:
+    scan = unwrap_command(["env", "FOO=1", "nice", "-n", "10", "python", "-c", "print(1)"])
+    assert scan.wrappers == ("env", "nice")
+    assert scan.offset == 5
+    assert scan.base == "python"
+    assert scan.command == ("python", "-c", "print(1)")
+
+
+def test_unwrap_privileged_wrapper() -> None:
+    scan = unwrap_command(["sudo", "-u", "root", "--", "python", "script.py"])
+    assert scan.wrappers == ("sudo",)
+    assert scan.base == "python"
+    assert scan.command == ("python", "script.py")
+
+
+def test_unwrap_wrapper_duration_of_its_own() -> None:
+    scan = unwrap_command(["timeout", "30s", "bash", "-c", "ls"])
+    assert scan.wrappers == ("timeout",)
+    assert scan.base == "bash"
+    scan = unwrap_command(["timeout", "5"])
+    assert scan.wrappers == ()
+    assert scan.base == "timeout"
+
+
+def test_unwrap_run_launcher_skips_its_options() -> None:
+    scan = unwrap_command(["uv", "run", "--with", "rich", "python", "-c", "print(1)"])
+    assert scan.wrappers == ("uv",)
+    assert scan.base == "python"
+    assert scan.command == ("python", "-c", "print(1)")
+
+
+def test_unwrap_nested_wrapper_and_launcher() -> None:
+    scan = unwrap_command(["nohup", "pdm", "run", "python", "script.py"])
+    assert scan.wrappers == ("nohup", "pdm")
+    assert scan.offset == 3
+    assert scan.base == "python"
+    assert scan.self_review == ("nohup",)
+
+
+def test_unwrap_stops_when_no_program_follows() -> None:
+    for argv in (["env"], ["timeout", "5"], ["pdm", "run"], ["nohup"]):
+        scan = unwrap_command(argv)
+        assert scan.wrappers == (), argv
+        assert scan.offset == 0, argv
+        assert scan.base == argv[0], argv
+
+
+def test_unwrap_empty_argv() -> None:
+    scan = unwrap_command([])
+    assert scan.base == ""
+    assert scan.command == ()
+
+
+def test_is_interpreter_ignores_version_suffix() -> None:
+    assert is_interpreter("python")
+    assert is_interpreter("python3.12")
+    assert is_interpreter("node20")
+    assert is_interpreter("php8.2")
+    assert is_interpreter("sqlite3")
+    assert is_interpreter("bash")
+    assert not is_interpreter("grep")
+    assert not is_interpreter("od")
+    assert not is_interpreter("phpmyadmin")
+    assert not is_interpreter("")

tests/test_tools/test_danger.py

@@ -96,6 +96,65 @@ def test_classify_shell_and_dash_c() -> None:
     assert r5 is not None and "-c" in r5
 
 
+def test_dash_c_means_code_only_for_interpreters() -> None:
+    # ``-c`` is a plain flag elsewhere: no code to review, no confirm.
+    assert classify_danger("run_argv", {"argv": ["grep", "-c", "needle", "f.txt"]}) is None
+    assert classify_danger("run_argv", {"argv": ["od", "-c", "f.bin"]}) is None
+    assert classify_danger("run_argv", {"argv": ["sort", "-c", "f.txt"]}) is None
+
+
+def test_interpreter_review_does_not_split_on_dash_c() -> None:
+    script = classify_danger("run_argv", {"argv": ["python", "script.py"]})
+    one_liner = classify_danger("run_argv", {"argv": ["python", "-c", "print(1)"]})
+    bare_shell = classify_danger("open_pty", {"command": ["bash"]})
+    assert script is not None and "interpreter 'python'" in script
+    assert one_liner is not None and "interpreter 'python'" in one_liner
+    assert bare_shell is not None and "interpreter 'bash'" in bare_shell
+    # Only the ``-c`` form has code to print below the argv line.
+    assert "command:" not in script
+    assert "command:" in one_liner
+    assert "print(1)" in one_liner
+
+
+def test_versioned_interpreter_is_still_reviewed() -> None:
+    reason = classify_danger("run_argv", {"argv": ["python3.12", "-c", "import os"]})
+    assert reason is not None and "interpreter 'python3.12'" in reason
+    assert "import os" in reason
+
+
+def test_wrappers_reach_the_interpreter() -> None:
+    for argv in (
+        ["env", "FOO=1", "python", "-c", "print(1)"],
+        ["sudo", "-u", "root", "python", "-c", "print(1)"],
+        ["timeout", "5", "bash", "-c", "rm -rf /"],
+        ["pdm", "run", "python", "-c", "print(1)"],
+        ["uv", "run", "--with", "rich", "python", "-c", "print(1)"],
+    ):
+        reason = classify_danger("run_argv", {"argv": argv})
+        assert reason is not None, argv
+        assert "interpreter 'python'" in reason or "interpreter 'bash'" in reason
+        assert "$(command)" in reason
+
+
+def test_only_the_interpreter_dash_c_becomes_a_placeholder() -> None:
+    reason = classify_danger("run_argv", {"argv": ["ionice", "-c", "2", "python", "-c", "print(1)"]})
+    assert reason is not None
+    argv_line = next(line for line in reason.splitlines() if line.startswith("  argv:"))
+    assert "ionice -c 2" in argv_line
+    assert argv_line.count("$(command)") == 1
+    assert "print(1)" in reason
+
+
+def test_detached_runner_is_always_reviewed() -> None:
+    assert classify_danger("run_argv", {"argv": ["sleep", "5"]}) is None
+    reason = classify_danger("run_argv", {"argv": ["nohup", "sleep", "5"]})
+    assert reason is not None and "nohup (detached run)" in reason
+    nested = classify_danger("run_argv", {"argv": ["nohup", "python", "-c", "print(1)"]})
+    assert nested is not None
+    assert "nohup (detached run)" in nested and "interpreter 'python'" in nested
+    assert "print(1)" in nested
+
+
 async def test_confirm_deny_with_comment() -> None:
     from plyngent.agent.tools import ToolRegistry, tool
     from plyngent.tools.danger import classify_danger as danger
@@ -124,7 +183,7 @@ def test_shell_confirm_formats_command_placeholder() -> None:
     assert reason is not None
     assert "$(command)" in reason
     assert "line1" in reason and "line2" in reason and "line3" in reason
-    argv_line = next(ln for ln in reason.splitlines() if "argv:" in ln)
+    argv_line = next(ln for ln in reason.splitlines() if ln.startswith("  argv:"))
     assert "line1" not in argv_line
     lines = reason.splitlines()
     idx = lines.index("  command:")

tests/test_tools/test_fetch.py

@@ -269,13 +269,10 @@ async def test_fetch_char_truncation_embeds_token(workspace: Path, http_server:
     token_str = token_line.split("truncate_token=", 1)[1].rstrip("]")
     token = decode_truncate_token(token_str)
     assert token is not None
-    assert token.kind == "http"
+    # The body is kept in memory, so resuming never repeats the request.
+    assert token.kind == "memory"
     assert token.limit == 1000
     assert 0 < token.offset < 1000
-    # Continuation resumes exactly where the chunk stopped (no gap).
-    out2 = await call_async(fetch, f"{http_server}/big", max_chars=1000, offset=token.offset)
-    body2 = out2.split("--- body ---", 1)[-1].strip()
-    assert body2.startswith("x")
 
 
 async def test_fetch_offset_skips_body(workspace: Path, http_server: str) -> None:

tests/test_tools/test_get_truncated.py

@@ -84,18 +84,40 @@ async def test_get_truncated_file_missing(workspace: Path) -> None:
     assert out.startswith("error:")
 
 
-async def test_get_truncated_http_passthrough(monkeypatch) -> None:
-    calls: list[tuple[str, int, int]] = []
+async def test_get_truncated_serves_a_fetch_body_without_refetching(monkeypatch) -> None:
+    """A fetch body's token reads the stored remainder; the request is not repeated."""
+    from plyngent.tools.net.fetch import format_fetch_result
 
-    async def fake_fetch(url: str, *, offset: int, max_chars: int) -> str:
-        calls.append((url, offset, max_chars))
-        return f"body-from-{offset}"
+    fetch_module = importlib.import_module("plyngent.tools.net.fetch")
 
-    monkeypatch.setattr(get_truncated_module.fetch, "handler", fake_fetch)
-    token = encode_truncate_token(TruncateToken(kind="http", location="https://example.com/x", offset=500, limit=400))
+    async def _must_not_refetch(*_args: object, **_kwargs: object) -> str:
+        msg = "resuming a fetch body must not re-run the request"
+        raise AssertionError(msg)
+
+    monkeypatch.setattr(fetch_module, "http_fetch", _must_not_refetch)
+    first = format_fetch_result(
+        status=200,
+        final_url="https://example.com/x",
+        content_type="text/plain",
+        body_text="x" * 600,
+        body_bytes=600,
+        truncated=False,
+        redirects=0,
+        method="GET",
+        security="public",
+        warnings=[],
+        body_kind="text",
+        max_chars=200,
+    )
+    token = _extract_token(first)
+    assert token is not None
+    parsed = decode_truncate_token(token)
+    assert parsed is not None
+    assert parsed.kind == "memory"
     out = await call_async(get_truncated, token, max_chars=200)
-    assert out == "body-from-500"
-    assert calls == [("https://example.com/x", 500, 400)]
+    content, _, _ = out.partition("\n[Truncated")
+    assert content == "x" * len(content)
+    assert 0 < len(content) <= 200
 
 
 async def test_get_truncated_memory_chains() -> None:

tests/test_tools/test_truncate_token.py

@@ -26,7 +26,7 @@ def test_decode_garbage_returns_none() -> None:
 def test_decode_rejects_invalid_fields() -> None:
     bad = TruncateToken(kind="file", location="x", offset=-1, limit=10)
     assert decode_truncate_token(encode_truncate_token(bad)) is None
-    bad_limit = TruncateToken(kind="http", location="u", offset=0, limit=0)
+    bad_limit = TruncateToken(kind="file", location="u", offset=0, limit=0)
     assert decode_truncate_token(encode_truncate_token(bad_limit)) is None
 
 
@@ -95,7 +95,7 @@ def test_truncate_with_token_end_reached_no_token() -> None:
     out, token = truncate_with_token(
         text,
         40,
-        kind="http",
+        kind="file",
         location="https://example.com/x",
         offset=0,
         limit=40,
@@ -111,14 +111,14 @@ def test_truncate_with_token_nonzero_start() -> None:
     out, token = truncate_with_token(
         text,
         200,
-        kind="http",
+        kind="file",
         location="https://example.com/x",
         offset=150,
         limit=200,
         total_len=700,
     )
     assert token is not None
-    assert token.kind == "http"
+    assert token.kind == "file"
     content, _, _ = out.partition("\n[Truncated")
     assert token.offset == 150 + len(content)
     assert "[Truncated (200 chars max.; 350 omitted). truncate_token=" in out

tests/test_tools/test_workspace.py

@@ -215,6 +215,124 @@ def test_command_denylist_policy_confirm_timeout_value(workspace: object) -> Non
         clear_policy_allowed_commands()
 
 
+def test_command_denylist_reaches_through_wrappers(workspace: object) -> None:
+    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(["env", "FOO=1", "rm", "-rf", "/"])
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["pdm", "run", "rm", "-rf", "/"])
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["timeout", "5", "rm", "-rf", "/"])
+        # The outermost denylisted program is the one named (``sudo`` here).
+        with pytest.raises(WorkspaceError, match="basename 'sudo' is blocked"):
+            check_command_allowed(["sudo", "-u", "root", "rm", "-rf", "/"])
+        # Wrapped programs that are not denylisted stay allowed.
+        check_command_allowed(["pdm", "run", "pytest", "-x"])
+        check_command_allowed(["env", "FOO=1", "echo", "ok"])
+    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
+
+    set_policy_confirm_hook(None)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["RM", "-rf", "/"])
+        with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
+            check_command_allowed(["C:\\Tools\\rm.exe", "-rf", "/"])
+    finally:
+        clear_policy_allowed_commands()
+
+
+def test_command_denylist_prompts_for_the_wrapped_program(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
+
+    asked: list[str] = []
+
+    def hook(basename: str, argv: object, timeout: float) -> bool:
+        del argv, timeout
+        asked.append(basename)
+        return True
+
+    set_policy_confirm_hook(hook)
+    clear_policy_allowed_commands()
+    try:
+        check_command_allowed(["pdm", "run", "rm", "-rf", "build"])
+        assert asked == ["rm"]
+        # Session grant: the same wrapped program does not re-prompt.
+        check_command_allowed(["pdm", "run", "rm", "-rf", "dist"])
+        assert asked == ["rm"]
+    finally:
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
+def test_match_command_denylist_entries() -> None:
+    from plyngent.tools import match_command_denylist
+
+    denylist = frozenset({"rm", "systemctl reboot poweroff", "loginctl kill-session"})
+    hits = match_command_denylist(["sudo", "rm", "-rf", "/"], denylist)
+    assert [(hit.program, hit.keyword) for hit in hits] == [("rm", None)]
+    assert [hit.program for hit in match_command_denylist(["env", "FOO=1", "rm", "x"], denylist)] == ["rm"]
+    hits = match_command_denylist(["sudo", "sudo", "rm", "x"], frozenset({"sudo", "rm"}))
+    assert [hit.program for hit in hits] == ["sudo", "rm"]
+    # Subcommand entries match the program's own arguments, targets included.
+    hits = match_command_denylist(["systemctl", "reboot"], denylist)
+    assert [(hit.program, hit.keyword, hit.entry) for hit in hits] == [
+        ("systemctl", "reboot", "systemctl reboot poweroff")
+    ]
+    assert match_command_denylist(["systemctl", "start", "poweroff.target"], denylist)[0].keyword == "poweroff"
+    assert match_command_denylist(["pdm", "run", "loginctl", "kill-session"], denylist)[0].program == "loginctl"
+    # Read-only and unrelated calls stay clean.
+    assert match_command_denylist(["systemctl", "status", "sshd"], denylist) == []
+    assert match_command_denylist(["loginctl", "list-sessions"], denylist) == []
+
+
+def test_command_denylist_power_subcommands(workspace: object) -> None:
+    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 'systemctl' is blocked: 'reboot'"):
+            check_command_allowed(["systemctl", "reboot"])
+        with pytest.raises(WorkspaceError, match="basename 'systemctl' is blocked: 'reboot'"):
+            check_command_allowed(["systemctl", "start", "reboot.target"])
+        with pytest.raises(WorkspaceError, match="basename 'systemctl' is blocked: 'poweroff'"):
+            check_command_allowed(["systemctl", "--user", "poweroff"])
+        with pytest.raises(WorkspaceError, match="basename 'systemctl' is blocked: 'emergency'"):
+            check_command_allowed(["timeout", "3", "systemctl", "emergency"])
+        with pytest.raises(WorkspaceError, match="basename 'loginctl' is blocked: 'kill-session'"):
+            check_command_allowed(["loginctl", "kill-session", "3"])
+        with pytest.raises(WorkspaceError, match="basename 'pm-suspend' is blocked"):
+            check_command_allowed(["pm-suspend"])
+        with pytest.raises(WorkspaceError, match="basename 'run0' is blocked"):
+            check_command_allowed(["run0", "id"])
+        with pytest.raises(WorkspaceError, match="basename 'systemd-run' is blocked"):
+            check_command_allowed(["systemd-run", "--user", "echo", "hi"])
+        # The outermost denylisted program is reported first.
+        with pytest.raises(WorkspaceError, match="basename 'sudo' is blocked"):
+            check_command_allowed(["sudo", "systemctl", "reboot"])
+        # Read-only systemd calls stay allowed.
+        check_command_allowed(["systemctl", "status", "sshd"])
+        check_command_allowed(["systemctl", "list-units", "--type=service"])
+        check_command_allowed(["systemctl", "show", "-p", "CanReboot"])
+        check_command_allowed(["loginctl", "list-sessions"])
+    finally:
+        clear_policy_allowed_commands()
+
+
 def test_root_required() -> None:
     from plyngent.tools.context import InstanceState, bind_instance