diff --git a/CLAUDE.md b/CLAUDE.md index 3129936..89bce3e 100644 --- a/CLAUDE.md +++ b/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. diff --git a/src/plyngent/agent/tools.py b/src/plyngent/agent/tools.py index bb53ad9..cb19c88 100644 --- a/src/plyngent/agent/tools.py +++ b/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).""" diff --git a/src/plyngent/cli/limits.py b/src/plyngent/cli/limits.py index 5321bc9..49e83a5 100644 --- a/src/plyngent/cli/limits.py +++ b/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( diff --git a/src/plyngent/tools/danger.py b/src/plyngent/tools/danger.py index cf976f0..862ad92 100644 --- a/src/plyngent/tools/danger.py +++ b/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 diff --git a/tests/test_tools/test_danger.py b/tests/test_tools/test_danger.py index 4ab69fc..43ce308 100644 --- a/tests/test_tools/test_danger.py +++ b/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