core/tools: soft-confirm shells, REPLs, and python/bash -c one-liners

classify_danger prompts on interactive shell/runtime launches and any -c
code (with argv + snippet preview). Confirm deny can attach a user comment
for the model. Restores ToolRegistry.execute after WIP breakage.

CLAUDE.md

@@ -83,7 +83,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 - **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `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`); 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.
-- Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. 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 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).
 - **`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; agent injects a developer review message if open todos and no todo tool use in the turn.

src/plyngent/agent/tools.py

@@ -13,9 +13,11 @@ from plyngent.typedef import JSONSchema  # noqa: TC001
 
 type ToolHandler = Callable[..., Any | Awaitable[Any]]
 type DangerClassifier = Callable[[str, Mapping[str, object]], str | None]
+# Returns True to allow, False to deny, or a non-empty str as denial reason for the model.
+type ToolConfirmResult = bool | str
 type ToolConfirmHook = Callable[
     [str, Mapping[str, object], str],
-    bool | Awaitable[bool],
+    ToolConfirmResult | Awaitable[ToolConfirmResult],
 ]
 
 _PRIMITIVE_SCHEMA: dict[type, JSONSchema] = {
@@ -212,7 +214,11 @@ class ToolRegistry:
         return msgspec.json.encode(result).decode()
 
     async def _maybe_confirm(self, name: str, args: dict[str, object]) -> str | None:
-        """Return an error string if the user denies a dangerous tool, else None."""
+        """If confirm is required and denied, return an error string for the model.
+
+        The confirm hook may return ``True`` (allow), ``False`` (deny), or a
+        non-empty string (deny with user comment for the model).
+        """
         if self._danger is None or self._on_confirm is None:
             return None
         reason = self._danger(name, args)
@@ -223,12 +229,14 @@ class ToolRegistry:
             reason = self._danger(name, args)
             if reason is None:
                 return None
-            allowed = self._on_confirm(name, args, reason)
-            if inspect.isawaitable(allowed):
-                allowed = await allowed
-            if allowed:
+            decision = self._on_confirm(name, args, reason)
+            if inspect.isawaitable(decision):
+                decision = await decision
+            if decision is True:
                 return None
-        return f"error: user denied tool {name!r} ({reason})"
+            if isinstance(decision, str) and decision.strip():
+                return f"error: tool {name!r} denied by user confirm ({reason}); user comment: {decision.strip()}"
+            return f"error: tool {name!r} denied by user confirm ({reason})"
 
     async def execute(self, name: str, arguments_json: str) -> str:
         """Run a tool by name; returns a string result (errors become error text)."""

src/plyngent/cli/limits.py

@@ -6,6 +6,8 @@ from plyngent.cli.interrupt import pause_task_cancel_for_prompt
 from plyngent.prompting import (
     ChoiceOption,
     NonInteractiveError,
+    ask,
+    ask_async,
     choose,
     choose_async,
     configure_prompting,
@@ -44,33 +46,55 @@ async def prompt_continue_limit_async(reason: str) -> bool:
         return False
 
 
-def _prompt_confirm_tool_sync(name: str, args: Mapping[str, object], reason: str) -> bool:
+def _prompt_confirm_tool_sync(name: str, args: Mapping[str, object], reason: str) -> bool | str:
+    """True allow; False deny; non-empty str = deny with comment for the model."""
     del args
     try:
-        return confirm(
-            f"[confirm] tool {name!r}: {reason}\nAllow this tool call?",
-            default=False,
-        )
+        prompt = f"[confirm] tool {name!r}:\n{reason}\nAllow this tool call?"
+        allowed = confirm(prompt, default=False)
     except NonInteractiveError:
         return False
+    if allowed:
+        return True
+    try:
+        comment = ask(
+            "Optional comment for the agent (why denied; empty to skip):",
+            default="",
+        ).strip()
+    except NonInteractiveError:
+        return False
+    return comment or False
 
 
-def prompt_confirm_tool(name: str, args: Mapping[str, object], reason: str) -> bool:
-    """Ask whether to allow a destructive tool call (TTY). Default is deny."""
+def prompt_confirm_tool(name: str, args: Mapping[str, object], reason: str) -> bool | str:
+    """Ask whether to allow a dangerous tool call (TTY). Default is deny.
+
+    On deny, optionally collect a free-text comment for the model.
+    """
     with pause_task_cancel_for_prompt():
         return _prompt_confirm_tool_sync(name, args, reason)
 
 
-async def prompt_confirm_tool_async(name: str, args: Mapping[str, object], reason: str) -> bool:
-    """Async variant: confirm off the event loop."""
+async def prompt_confirm_tool_async(name: str, args: Mapping[str, object], reason: str) -> bool | str:
+    """Async confirm: True allow, False deny, str = deny with user comment."""
     del args
     try:
-        return await confirm_async(
-            f"[confirm] tool {name!r}: {reason}\nAllow this tool call?",
-            default=False,
-        )
+        prompt = f"[confirm] tool {name!r}:\n{reason}\nAllow this tool call?"
+        allowed = await confirm_async(prompt, default=False)
     except NonInteractiveError:
         return False
+    if allowed:
+        return True
+    try:
+        comment = (
+            await ask_async(
+                "Optional comment for the agent (why denied; empty to skip):",
+                default="",
+            )
+        ).strip()
+    except NonInteractiveError:
+        return False
+    return comment or False
 
 
 def _prompt_workspace_mismatch_sync(

src/plyngent/tools/danger.py

@@ -1,11 +1,107 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from plyngent.tools.workspace import WorkspaceError, resolve_path
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping
+    from collections.abc import Mapping, Sequence
+
+_DISPLAY_ARGV_MAX = 200
+_CODE_PREVIEW_MAX = 120
+
+_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]) -> list[str] | None:
+    command = args.get("command")
+    if not isinstance(command, list) or not command:
+        return None
+    out: list[str] = []
+    for part_obj in cast("list[object]", command):
+        if not isinstance(part_obj, str):
+            return None
+        out.append(part_obj)
+    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):
+        if part == "-c":
+            return argv[index + 1] if index + 1 < len(argv) else ""
+        # Combined short forms are uncommon; only exact -c is supported.
+    return None
+
+
+def _shell_or_dash_c_reason(argv: Sequence[str], *, via: str) -> str | None:
+    """Confirm interactive shells/REPLs and ``-c`` one-liners so the user can inspect argv."""
+    if not argv:
+        return None
+    base = _basename(argv[0])
+    display = " ".join(argv)
+    if len(display) > _DISPLAY_ARGV_MAX:
+        display = display[:_DISPLAY_ARGV_MAX] + "…"
+
+    code = _find_dash_c_code(argv)
+    if code is not None:
+        preview = code if len(code) <= _CODE_PREVIEW_MAX else code[:_CODE_PREVIEW_MAX] + "…"
+        return f"{via}: {base} -c (review code before allow)\n  argv: {display}\n  -c: {preview!r}"
+
+    if base in _SHELL_BASENAMES and len(argv) == 1:
+        return f"{via}: interactive {base!r} (review before allow)\n  argv: {display}"
+
+    # e.g. python -i, bash --login without -c still needs a glance.
+    if base in _SHELL_BASENAMES:
+        return f"{via}: shell/runtime {base!r} (review before allow)\n  argv: {display}"
+
+    return None
 
 
 def _write_file_reason(args: Mapping[str, object]) -> str | None:
@@ -15,32 +111,80 @@ def _write_file_reason(args: Mapping[str, object]) -> str | None:
     try:
         target = resolve_path(path)
     except WorkspaceError:
+        return f"write file {path!r}"
+    return f"write file {path!r} ({target})"
+
+
+def _edit_replace_reason(args: Mapping[str, object]) -> str | None:
+    path = args.get("path")
+    if not isinstance(path, str) or not path:
         return None
-    if target.exists() or target.is_symlink():
-        return f"overwrite existing file {path!r}"
-    return None
+    return f"edit (replace) in {path!r}"
 
 
-def classify_danger(name: str, args: Mapping[str, object]) -> str | None:
+def _edit_lineno_reason(args: Mapping[str, object]) -> str | None:
+    path = args.get("path")
+    if not isinstance(path, str) or not path:
+        return None
+    start = args.get("start_line")
+    end = args.get("end_line")
+    return f"edit lines {start}-{end} in {path!r}"
+
+
+def _copy_path_reason(args: Mapping[str, object]) -> str | None:
+    src = args.get("src")
+    dst = args.get("dst")
+    return f"copy {src!r} → {dst!r}"
+
+
+def _move_path_reason(args: Mapping[str, object]) -> str | None:
+    src = args.get("src")
+    dst = args.get("dst")
+    return f"move {src!r} → {dst!r}"
+
+
+def _delete_path_reason(args: Mapping[str, object]) -> str | None:
+    path = args.get("path")
+    recursive = bool(args.get("recursive", False))
+    extra = " recursively" if recursive else ""
+    return f"delete path {path!r}{extra}"
+
+
+def _run_command_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="run_command")
+
+
+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")
+
+
+def classify_danger(name: str, args: Mapping[str, object]) -> str | None:  # noqa: PLR0911
     """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 that policy otherwise allows.
+    soft confirms for mutating tools and risky shell/REPL launches
+    (interactive shells and ``python -c`` / ``bash -c`` one-liners).
     """
-    reasons: dict[str, str | None] = {
-        "delete_path": (
-            f"delete path {args.get('path', '')!r} recursively"
-            if bool(args.get("recursive", False))
-            else f"delete path {args.get('path', '')!r}"
-        ),
-        "move_path": f"move {args.get('src', '')!r} -> {args.get('dst', '')!r}",
-        "copy_path": (
-            f"copy with overwrite {args.get('src', '')!r} -> {args.get('dst', '')!r}"
-            if bool(args.get("overwrite", False))
-            else None
-        ),
-        "write_file": _write_file_reason(args),
-    }
-    if name not in reasons:
-        return None
-    return reasons[name]
+    if name == "delete_path":
+        return _delete_path_reason(args)
+    if name == "move_path":
+        return _move_path_reason(args)
+    if name == "copy_path":
+        return _copy_path_reason(args)
+    if name == "write_file":
+        return _write_file_reason(args)
+    if name == "edit_replace":
+        return _edit_replace_reason(args)
+    if name == "edit_lineno":
+        return _edit_lineno_reason(args)
+    if name == "run_command":
+        return _run_command_reason(args)
+    if name == "open_pty":
+        return _open_pty_reason(args)
+    return None

tests/test_tools/test_danger.py

@@ -11,18 +11,71 @@ def test_classify_delete_and_move() -> None:
     assert "move" in (classify_danger("move_path", {"src": "a", "dst": "b"}) or "")
 
 
-def test_classify_copy_overwrite_only() -> None:
-    assert classify_danger("copy_path", {"src": "a", "dst": "b"}) is None
-    assert "overwrite" in (classify_danger("copy_path", {"src": "a", "dst": "b", "overwrite": True}) or "")
+def test_classify_copy() -> None:
+    assert "copy" in (classify_danger("copy_path", {"src": "a", "dst": "b"}) or "")
 
 
-def test_classify_write_file_overwrite(workspace: object) -> None:
-    assert isinstance(workspace, Path)
-    _ = (workspace / "x.txt").write_text("old", encoding="utf-8")
-    assert "overwrite" in (classify_danger("write_file", {"path": "x.txt", "content": "n"}) or "")
-    assert classify_danger("write_file", {"path": "new.txt", "content": "n"}) is None
+def test_classify_write_and_edits(tmp_path: Path) -> None:
+    from plyngent.tools.workspace import clear_workspace_root, set_workspace_root
+
+    set_workspace_root(tmp_path)
+    try:
+        reason = classify_danger("write_file", {"path": "x.txt"})
+        assert reason is not None and "write file" in reason
+        assert "edit" in (classify_danger("edit_replace", {"path": "x.txt"}) or "")
+        assert "lines" in (classify_danger("edit_lineno", {"path": "x.txt", "start_line": 1, "end_line": 2}) or "")
+    finally:
+        clear_workspace_root()
 
 
 def test_classify_safe_tools() -> None:
     assert classify_danger("read_file", {"path": "a"}) is None
+    assert classify_danger("listdir", {"path": "."}) is None
     assert classify_danger("run_command", {"command": ["echo", "hi"]}) is None
+    assert classify_danger("open_pty", {"command": ["true"]}) is None
+    assert classify_danger("run_command", {"command": ["ls", "-la"]}) is None
+
+
+def test_classify_shell_and_dash_c() -> None:
+    r = classify_danger("run_command", {"command": ["bash", "-c", "rm -rf /"]})
+    assert r is not None
+    assert "bash -c" in r
+    assert "rm -rf" in r
+
+    r2 = classify_danger("run_command", {"command": ["python3", "-c", "print(1)"]})
+    assert r2 is not None
+    assert "python" in r2
+    assert "-c" in r2
+    assert "print(1)" in r2
+
+    r_py = classify_danger("run_command", {"command": ["python", "-c", "import os"]})
+    assert r_py is not None and "python -c" in r_py
+
+    r3 = classify_danger("open_pty", {"command": ["bash"]})
+    assert r3 is not None and "bash" in r3
+    assert "interactive" in r3 or "review" in r3
+
+    r4 = classify_danger("open_pty", {"command": ["python3"]})
+    assert r4 is not None and "python" in r4
+
+    r5 = classify_danger("open_pty", {"command": ["python3", "-c", "x=1"]})
+    assert r5 is not None and "-c" in r5
+
+
+async def test_confirm_deny_with_comment() -> None:
+    from plyngent.agent.tools import ToolRegistry, tool
+    from plyngent.tools.danger import classify_danger as danger
+
+    @tool
+    def delete_path(path: str) -> str:
+        return f"deleted {path}"
+
+    async def deny_comment(name: str, args: object, reason: str) -> str:
+        del name, args, reason
+        return "too destructive for this session"
+
+    reg = ToolRegistry([delete_path], danger=danger, on_confirm=deny_comment)
+    out = await reg.execute("delete_path", '{"path": "x"}')
+    assert "denied" in out
+    assert "user comment:" in out
+    assert "too destructive" in out