core/tools: timed human allow for denylisted commands

Denylist hits ask the user (default deny, timeout) instead of only hard
rejecting when a policy hook is installed. Independent of YOLO soft-confirm;
session grants skip re-prompt for the same basename. CLI installs the hook
via install_cli_limit_hooks.

CLAUDE.md

@@ -78,7 +78,7 @@ 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; path substring denylist; command basename denylist; clearer deny messages.
+- **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist (CLI: timed human allow override, default deny on timeout; **not** skipped by YOLO); clearer deny messages.
 - **`file`**: `read_file` (`with_lineno`), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind).
 - **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_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.

src/plyngent/cli/app.py

@@ -327,6 +327,10 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
         from plyngent.tools.temp_workspace import cleanup_temporary_workspaces
 
         PtyManager.close_all()
+        from plyngent.tools.workspace import clear_policy_allowed_commands, set_policy_confirm_hook
+
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
         _ = cleanup_temporary_workspaces()
 
 

src/plyngent/cli/limits.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import contextlib
+import selectors
 import shutil
+import sys
 from typing import TYPE_CHECKING, Literal
 
 from plyngent.cli.interrupt import pause_task_cancel_for_prompt
@@ -19,8 +22,7 @@ from plyngent.prompting import (
 from plyngent.tools.process.pty_session import PtyManager
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping
+    from collections.abc import Mapping, Sequence
-
 type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"]
 
 _BOX_MIN_WIDTH = 40
@@ -150,6 +152,86 @@ def prompt_confirm_tool(name: str, args: Mapping[str, object], reason: str) -> b
         return _prompt_confirm_tool_sync(name, args, reason)
 
 
+def _try_readline() -> str | None:
+    try:
+        return sys.stdin.readline()
+    except OSError, EOFError:
+        return None
+
+
+def _read_yes_no_line_with_timeout(timeout_seconds: float) -> str | None:
+    """Read one line from stdin with *timeout_seconds*; ``None`` on timeout/EOF.
+
+    Uses :mod:`selectors` on the process stdin FD. On Windows without selector
+    support, falls back to blocking readline (no true timeout).
+    """
+    try:
+        if not sys.stdin.isatty():
+            return None
+        fd = sys.stdin.fileno()
+    except OSError, ValueError, AttributeError:
+        return _try_readline()
+
+    try:
+        selector = selectors.DefaultSelector()
+        _ = selector.register(fd, selectors.EVENT_READ)
+    except OSError, ValueError, AttributeError:
+        return _try_readline()
+
+    try:
+        events = selector.select(timeout_seconds)
+        if not events:
+            return None
+        return sys.stdin.readline()
+    except OSError, ValueError, EOFError:
+        return None
+    finally:
+        selector.close()
+
+
+def prompt_policy_command_confirm(
+    basename: str,
+    argv: Sequence[str],
+    timeout_seconds: float,
+) -> bool:
+    """Ask whether to allow a denylisted command for this session (timed).
+
+    Independent of YOLO: always prompts when interactive. Default is **deny**.
+    Timeout, cancel, or non-interactive → False.
+    """
+    argv_preview = " ".join(str(part) for part in argv)
+    reason = (
+        f"policy denylist blocks basename {basename!r}\n"
+        f"argv: {argv_preview}\n"
+        f"Allow this basename for the rest of this process?\n"
+        f"(timeout {timeout_seconds:g}s defaults to DENY; not skipped by YOLO)"
+    )
+    try:
+        with pause_task_cancel_for_prompt():
+            backend = get_prompt_backend()
+            if not backend.is_interactive():
+                return False
+            backend.echo()
+            backend.secho(format_tool_confirm_box("policy", reason), fg="yellow")
+            backend.echo()
+            prompt = f"[policy] allow {basename!r}? [y/N] (timeout {timeout_seconds:g}s): "
+            with contextlib.suppress(OSError):
+                _ = sys.stderr.write(prompt)
+                _ = sys.stderr.flush()
+            raw = _read_yes_no_line_with_timeout(timeout_seconds)
+            if raw is None:
+                with contextlib.suppress(OSError, NonInteractiveError):
+                    backend.secho(
+                        f"[policy] timed out after {timeout_seconds:g}s — denied",
+                        fg="red",
+                        err=True,
+                    )
+                return False
+            return raw.strip().lower() in {"y", "yes"}
+    except NonInteractiveError, KeyboardInterrupt, EOFError:
+        return False
+
+
 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
@@ -268,6 +350,9 @@ async def prompt_workspace_mismatch_async(
 
 
 def install_cli_limit_hooks() -> None:
-    """Register interactive continue hooks and prompt cancel-pause for the CLI."""
+    """Register interactive continue hooks, policy confirm, and prompt cancel-pause."""
     configure_prompting(pause_factory=pause_task_cancel_for_prompt)
     PtyManager.set_limit_continue_hook(prompt_continue_limit)
+    from plyngent.tools.workspace import set_policy_confirm_hook
+
+    set_policy_confirm_hook(prompt_policy_command_confirm)

src/plyngent/tools/__init__.py

@@ -43,20 +43,26 @@ from .vcs import vcs_status as vcs_status
 from .workspace import (
     DEFAULT_COMMAND_DENYLIST as DEFAULT_COMMAND_DENYLIST,
 )
+from .workspace import DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS as DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS
 from .workspace import WorkspaceError as WorkspaceError
 from .workspace import add_workspace_allowlist as add_workspace_allowlist
 from .workspace import check_command_allowed as check_command_allowed
+from .workspace import clear_policy_allowed_commands as clear_policy_allowed_commands
 from .workspace import clear_workspace_allowlist as clear_workspace_allowlist
 from .workspace import clear_workspace_root as clear_workspace_root
 from .workspace import get_command_denylist as get_command_denylist
 from .workspace import get_path_denylist as get_path_denylist
+from .workspace import get_policy_confirm_timeout as get_policy_confirm_timeout
 from .workspace import get_workspace_root as get_workspace_root
+from .workspace import grant_policy_command as grant_policy_command
 from .workspace import list_workspace_allowlist as list_workspace_allowlist
 from .workspace import pop_owned_temporary_workspaces as pop_owned_temporary_workspaces
 from .workspace import remove_workspace_allowlist as remove_workspace_allowlist
 from .workspace import resolve_path as resolve_path
 from .workspace import set_command_denylist as set_command_denylist
 from .workspace import set_path_denylist as set_path_denylist
+from .workspace import set_policy_confirm_hook as set_policy_confirm_hook
+from .workspace import set_policy_confirm_timeout as set_policy_confirm_timeout
 from .workspace import set_workspace_root as set_workspace_root
 
 DEFAULT_TOOLS = [*FILE_TOOLS, *PROCESS_TOOLS, *VCS_TOOLS, *CHAT_TOOLS, *TODO_TOOLS]

src/plyngent/tools/workspace.py

@@ -1,6 +1,10 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Sequence
 
 DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset(
     {
@@ -29,6 +33,12 @@ DEFAULT_COMMAND_DENYLIST: frozenset[str] = frozenset(
 # Max concurrent temporary workspaces registered in one process.
 MAX_TEMPORARY_WORKSPACES = 16
 
+# Timed human override for command denylist (independent of YOLO soft-confirm).
+DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS = 30.0
+
+# Hook: (basename, argv, timeout_seconds) -> True allow for this session basename.
+type PolicyConfirmHook = Callable[[str, Sequence[str], float], bool]
+
 
 class WorkspaceError(ValueError):
     """Raised when a path or command violates workspace policy."""
@@ -42,10 +52,17 @@ class _WorkspaceState:
     allowlist: list[Path]
     # Paths created by new_temporary_workspace (subset of allowlist); cleaned on chat exit.
     temporary_owned: list[Path]
+    # Basenames the human allowed this process (denylist override, not permanent).
+    policy_allowed_commands: set[str]
+    policy_confirm_hook: PolicyConfirmHook | None
+    policy_confirm_timeout_seconds: float
 
     def __init__(self) -> None:
         self.allowlist = []
         self.temporary_owned = []
+        self.policy_allowed_commands = set()
+        self.policy_confirm_hook = None
+        self.policy_confirm_timeout_seconds = DEFAULT_POLICY_CONFIRM_TIMEOUT_SECONDS
 
 
 _state = _WorkspaceState()
@@ -87,12 +104,47 @@ def get_path_denylist() -> tuple[str, ...]:
 def set_command_denylist(names: list[str] | tuple[str, ...] | frozenset[str] | None) -> None:
     """Set denied command basenames (None restores defaults)."""
     _state.command_denylist = DEFAULT_COMMAND_DENYLIST if names is None else frozenset(names)
+    # Session overrides only make sense against the active denylist.
+    _state.policy_allowed_commands &= _state.command_denylist
 
 
 def get_command_denylist() -> frozenset[str]:
     return _state.command_denylist
 
 
+def set_policy_confirm_hook(hook: PolicyConfirmHook | None) -> None:
+    """Register a timed human confirm for denylisted commands (CLI installs this)."""
+    _state.policy_confirm_hook = hook
+
+
+def get_policy_confirm_hook() -> PolicyConfirmHook | None:
+    return _state.policy_confirm_hook
+
+
+def set_policy_confirm_timeout(seconds: float) -> None:
+    """Timeout for policy confirm prompts (must be > 0)."""
+    if seconds <= 0:
+        msg = "policy confirm timeout must be > 0"
+        raise WorkspaceError(msg)
+    _state.policy_confirm_timeout_seconds = float(seconds)
+
+
+def get_policy_confirm_timeout() -> float:
+    return _state.policy_confirm_timeout_seconds
+
+
+def clear_policy_allowed_commands() -> None:
+    """Drop session-scoped denylist overrides (tests / chat exit)."""
+    _state.policy_allowed_commands.clear()
+
+
+def grant_policy_command(basename: str) -> None:
+    """Allow *basename* for the rest of this process despite the denylist."""
+    name = basename.strip()
+    if name:
+        _state.policy_allowed_commands.add(name)
+
+
 def add_workspace_allowlist(root: Path | str, *, owned: bool = False) -> Path:
     """Allow tool paths under *root* in addition to the primary workspace.
 
@@ -180,11 +232,34 @@ def resolve_path(path: str | Path) -> Path:
 
 
 def check_command_allowed(argv: list[str]) -> None:
-    """Raise if argv is empty or the executable basename is denylisted."""
+    """Raise if argv is empty or the executable basename is denylisted.
+
+    Denylisted basenames 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.
+    """
     if not argv:
         msg = "command argv must not be empty"
         raise WorkspaceError(msg)
     binary = Path(argv[0]).name
-    if binary in _state.command_denylist:
+    if binary not in _state.command_denylist:
-        msg = f"command denied by policy (basename {binary!r} is blocked)"
+        return
+    if binary in _state.policy_allowed_commands:
+        return
+    hook = _state.policy_confirm_hook
+    if hook is not None:
+        timeout = _state.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:
+            _state.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)

tests/test_cli/test_limits.py

@@ -41,8 +41,20 @@ def test_format_tool_confirm_box_multiline() -> None:
 
 
 def test_install_cli_limit_hooks() -> None:
+    from plyngent.tools.workspace import get_policy_confirm_hook, set_policy_confirm_hook
+
     install_cli_limit_hooks()
     assert callable(getattr(PtyManager, "_limit_continue", None))
+    assert get_policy_confirm_hook() is not None
     PtyManager.set_limit_continue_hook(None)
+    set_policy_confirm_hook(None)
     # Backend remains usable after install.
     assert get_prompt_backend().is_interactive() or True
+
+
+def test_policy_confirm_noninteractive_denies() -> None:
+    from plyngent.cli.limits import prompt_policy_command_confirm
+    from plyngent.prompting import NonInteractiveBackend
+
+    with temporary_backend(NonInteractiveBackend()):
+        assert prompt_policy_command_confirm("sudo", ["sudo", "id"], 1.0) is False

tests/test_tools/test_workspace.py

@@ -43,12 +43,91 @@ def test_path_denylist(workspace: object) -> None:
 
 def test_command_denylist(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()
     with pytest.raises(WorkspaceError, match="basename 'rm' is blocked"):
         check_command_allowed(["rm", "-rf", "/"])
     check_command_allowed(["echo", "ok"])
     set_command_denylist(None)
 
 
+def test_command_denylist_policy_confirm_allow(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import (
+        clear_policy_allowed_commands,
+        set_policy_confirm_hook,
+    )
+
+    calls: list[tuple[str, float]] = []
+
+    def hook(basename: str, argv: object, timeout: float) -> bool:
+        del argv
+        calls.append((basename, timeout))
+        return basename == "sudo"
+
+    set_policy_confirm_hook(hook)
+    clear_policy_allowed_commands()
+    try:
+        check_command_allowed(["sudo", "echo", "test"])
+        assert calls == [("sudo", 30.0)]
+        # Session grant: second call does not re-prompt.
+        check_command_allowed(["sudo", "id"])
+        assert len(calls) == 1
+    finally:
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
+def test_command_denylist_policy_confirm_deny(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import (
+        clear_policy_allowed_commands,
+        set_policy_confirm_hook,
+    )
+
+    set_policy_confirm_hook(lambda *_a: False)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError, match="declined or timed out"):
+            check_command_allowed(["sudo", "echo", "x"])
+    finally:
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
+def test_command_denylist_policy_confirm_timeout_value(workspace: object) -> None:
+    del workspace
+    from plyngent.tools.workspace import (
+        clear_policy_allowed_commands,
+        set_policy_confirm_hook,
+        set_policy_confirm_timeout,
+    )
+
+    seen: list[float] = []
+
+    def hook(basename: str, argv: object, timeout: float) -> bool:
+        del basename, argv
+        seen.append(timeout)
+        return False
+
+    set_policy_confirm_timeout(5.0)
+    set_policy_confirm_hook(hook)
+    clear_policy_allowed_commands()
+    try:
+        with pytest.raises(WorkspaceError):
+            check_command_allowed(["rm", "x"])
+        assert seen == [5.0]
+    finally:
+        set_policy_confirm_timeout(30.0)
+        set_policy_confirm_hook(None)
+        clear_policy_allowed_commands()
+
+
 def test_root_required() -> None:
     clear_workspace_root()
     with pytest.raises(WorkspaceError, match="not set"):