From 27618774606dfc1c5afadfaeddbc67e916c3e12a Mon Sep 17 00:00:00 2001 From: worldmozara Date: Fri, 17 Jul 2026 14:55:34 +0800 Subject: [PATCH] core/cli: YOLO mode on|off|once for soft destructive confirms Session /yolo and --yes skip classify_danger prompts; once expires after the next user turn. Hard path/command denylists unchanged. --- src/plyngent/agent/tools.py | 5 +++ src/plyngent/cli/app.py | 12 ++++-- src/plyngent/cli/repl.py | 14 +++++-- src/plyngent/cli/slash.py | 50 +++++++++++++++++++++-- src/plyngent/cli/state.py | 41 +++++++++++++++---- tests/test_agent/test_confirm.py | 10 +++++ tests/test_cli/test_readline_setup.py | 2 + tests/test_cli/test_repl_commands.py | 58 +++++++++++++++++++++++++++ 8 files changed, 175 insertions(+), 17 deletions(-) diff --git a/src/plyngent/agent/tools.py b/src/plyngent/agent/tools.py index 428f3ad..bb53ad9 100644 --- a/src/plyngent/agent/tools.py +++ b/src/plyngent/agent/tools.py @@ -191,6 +191,11 @@ class ToolRegistry: def __len__(self) -> int: return len(self._tools) + @property + def soft_confirm(self) -> bool: + """True when dangerous tools are gated by ``on_confirm``.""" + return self._danger is not None and self._on_confirm is not None + async def _invoke(self, definition: ToolDefinition, args: dict[str, object]) -> str: try: result = definition.handler(**args) diff --git a/src/plyngent/cli/app.py b/src/plyngent/cli/app.py index fe9831c..67518ce 100644 --- a/src/plyngent/cli/app.py +++ b/src/plyngent/cli/app.py @@ -171,9 +171,12 @@ async def _run_oneshot(state: ReplState, prompt_text: str) -> int: try: ok = await run_user_text_with_retries(state.agent, prompt_text, delays=()) except asyncio.CancelledError: + state.expire_yolo_once(quiet=True) return EXIT_CANCELLED except KeyboardInterrupt: + state.expire_yolo_once(quiet=True) return EXIT_CANCELLED + state.expire_yolo_once(quiet=True) return EXIT_OK if ok else EXIT_TURN_FAILED @@ -206,7 +209,10 @@ async def _run_chat( # noqa: C901, PLR0912 — chat orchestration _warn_recoverable_providers(store.recoverable_providers) _setup_workspace_and_hooks(store, workspace, interactive=interactive) - confirm_destructive: bool | None = False if yes else None + # --yes forces sticky YOLO; else derive from config.confirm_destructive. + from plyngent.cli.state import YoloMode + + yolo: YoloMode | None = "on" if yes else None memory = await MemoryStore.open(_database_config(store, quiet=quiet or oneshot)) try: @@ -255,7 +261,7 @@ async def _run_chat( # noqa: C901, PLR0912 — chat orchestration max_rounds=max_rounds, stream_enabled=stream, interactive_limits=interactive, - confirm_destructive=confirm_destructive, + yolo=yolo, ) if not quiet and not oneshot: click.secho(f"workspace: {state.workspace}", fg="bright_black", err=True) @@ -367,7 +373,7 @@ def main(ctx: click.Context, log_level: str) -> None: "--yes", is_flag=True, default=False, - help="Allow destructive tools without confirm (one-shot / non-interactive).", + help="Enable YOLO: skip destructive-tool confirms (sticky for this process).", ) @click.option( "--quiet", diff --git a/src/plyngent/cli/repl.py b/src/plyngent/cli/repl.py index 4ff609f..76b7dd2 100644 --- a/src/plyngent/cli/repl.py +++ b/src/plyngent/cli/repl.py @@ -25,12 +25,14 @@ def _echo_user(text: str) -> None: async def run_repl(state: ReplState) -> None: """Interactive chat loop with readline editing, history, and Tab completion.""" setup_readline(state) + yolo = state.effective_yolo() + yolo_part = f" yolo={yolo}" if yolo != "off" else "" click.echo( f"plyngent chat provider={state.provider_name} model={state.model} " f"session={state.session_id} tools={'on' if state.tools_enabled else 'off'} " f"rounds={state.max_rounds} messages={len(state.agent.messages)} " f"stream={'on' if state.agent.stream else 'off'} " - f"verbose={'on' if state.verbose else 'off'}" + f"verbose={'on' if state.verbose else 'off'}{yolo_part}" ) click.echo('Type /help for commands. Multiline: """ … """. Empty line is ignored.') @@ -52,8 +54,14 @@ async def run_repl(state: ReplState) -> None: text = state.pending_user_text state.pending_user_text = None _echo_user(text) - _ = await run_user_text_with_retries(state.agent, text) + try: + _ = await run_user_text_with_retries(state.agent, text) + finally: + state.expire_yolo_once() continue _echo_user(entry) - _ = await run_user_text_with_retries(state.agent, entry) + try: + _ = await run_user_text_with_retries(state.agent, entry) + finally: + state.expire_yolo_once() diff --git a/src/plyngent/cli/slash.py b/src/plyngent/cli/slash.py index 37b2f83..f46dedb 100644 --- a/src/plyngent/cli/slash.py +++ b/src/plyngent/cli/slash.py @@ -23,13 +23,14 @@ from plyngent.runtime import ProviderNotSupportedError if TYPE_CHECKING: from collections.abc import Awaitable, Sequence - from plyngent.cli.state import ReplState + from plyngent.cli.state import ReplState, YoloMode from plyngent.lmproto.openai_compatible.model import AnyChatMessage _DEFAULT_HISTORY_LINES = 20 _CONTENT_PREVIEW = 200 _COMPACT_PREVIEW = 400 _ON_OFF_CHOICES = ("on", "off") +_YOLO_MODE_CHOICES = ("on", "off", "once") _EXPORT_FORMAT_CHOICES = ("md", "json") HELP_FOOTER = ( @@ -38,7 +39,7 @@ HELP_FOOTER = ( "works after resume, not only via readline history). Auto-retry: 10s/20s/30s.\n" "\n" "Tab completes slash commands and some arguments (provider, model, tools,\n" - "stream, verbose, export). Use --session ID or /resume to continue a prior\n" + "stream, verbose, yolo, export). Use --session ID or /resume to continue a prior\n" "chat after restart.\n" "\n" 'Multiline: start a message with """ then end a later line with """.\n' @@ -86,6 +87,30 @@ class OnOffParam(click.ParamType[bool]): ON_OFF = OnOffParam() +class YoloModeParam(click.ParamType[str]): + """Accept on|off|once for soft destructive-tool confirms.""" + + name: str = "yolo_mode" + + @override + def convert(self, value: Any, param: click.Parameter | None, ctx: click.Context | None) -> str: + if isinstance(value, str) and value in _YOLO_MODE_CHOICES: + return value + token = str(value).strip().lower() + if token in _YOLO_MODE_CHOICES: + return token + msg = "expected on, off, or once" + raise click.BadParameter(msg, ctx=ctx, param=param) + + @override + def shell_complete(self, ctx: click.Context, param: click.Parameter, incomplete: str) -> list[CompletionItem]: + del ctx, param + return _filter_choices(incomplete, _YOLO_MODE_CHOICES) + + +YOLO_MODE = YoloModeParam() + + class ExportFormatParam(click.ParamType[str]): """First token of /export: md|json (or a path if not a format).""" @@ -355,7 +380,8 @@ def status_cmd(state: ReplState) -> None: f"tools={'on' if state.tools_enabled else 'off'} " f"rounds={state.max_rounds} " f"stream={'on' if state.agent.stream else 'off'} " - f"verbose={'on' if state.verbose else 'off'}\n" + f"verbose={'on' if state.verbose else 'off'} " + f"yolo={state.effective_yolo()}\n" f"context_tokens={ctx_tilde}{ctx_tokens}/{ctx_budget} ({ctx_tag}) " f"context_chars={ctx_chars} " f"tool_result_max={state.agent.max_tool_result_chars}\n" @@ -698,6 +724,24 @@ def tools_cmd(state: ReplState, enabled: bool | None) -> None: # noqa: FBT001 click.echo(f"tools={'on' if enabled else 'off'}") +@slash.command("yolo") +@click.argument("mode", required=False, type=YOLO_MODE, metavar="[on|off|once]") +@click.pass_obj +def yolo_cmd(state: ReplState, mode: str | None) -> None: + """Show or set YOLO mode for soft destructive-tool confirms. + + ``off`` (default when config ``confirm_destructive`` is true): prompt on + delete/move/overwrite (deny in non-TTY). ``on``: skip confirms for the + process. ``once``: skip for the next user turn only, then return to ``off``. + Path/command denylists still apply. Omit the argument to print the value. + """ + if mode is None: + click.echo(f"yolo={state.effective_yolo()}") + return + state.set_yolo(cast("YoloMode", mode)) + click.echo(f"yolo={state.effective_yolo()}") + + @slash.command("stream") @click.argument("enabled", required=False, type=ON_OFF, metavar="[on|off]") @click.pass_obj diff --git a/src/plyngent/cli/state.py b/src/plyngent/cli/state.py index 19569e5..7fda333 100644 --- a/src/plyngent/cli/state.py +++ b/src/plyngent/cli/state.py @@ -4,7 +4,7 @@ import contextlib import time from dataclasses import dataclass, field from pathlib import Path -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Literal, cast from plyngent.agent import ChatAgent, ChatClient, ToolRegistry from plyngent.agent.loop import DEFAULT_MAX_ROUNDS @@ -25,6 +25,8 @@ if TYPE_CHECKING: from plyngent.memory import MemoryStore from plyngent.memory.database.schema import Session as SessionRow +type YoloMode = Literal["off", "on", "once"] + @dataclass class ReplState: @@ -44,8 +46,9 @@ class ReplState: markdown_enabled: bool = True # One-shot / scripts: never prompt to raise tool-loop limits. interactive_limits: bool = True - # When False, skip destructive-tool confirms (e.g. --yes). - confirm_destructive: bool | None = None + # Soft destructive-tool confirms: None → derive from config.confirm_destructive. + # off = confirm; on = skip (sticky); once = skip next user turn then off. + yolo: YoloMode | None = None # Set by /edit; REPL sends as the next user turn then clears. pending_user_text: str | None = None client: ChatClient = field(init=False) @@ -77,10 +80,32 @@ class ReplState: raise RuntimeError(msg) return key - def _confirm_destructive(self) -> bool: - if self.confirm_destructive is not None: - return self.confirm_destructive - return self.config.agent_config.confirm_destructive + def effective_yolo(self) -> YoloMode: + """Resolved YOLO mode (session override or config default).""" + if self.yolo is not None: + return self.yolo + return "off" if self.config.agent_config.confirm_destructive else "on" + + def soft_confirm_enabled(self) -> bool: + """Whether destructive tools should prompt (or deny non-interactively).""" + return self.effective_yolo() == "off" + + def set_yolo(self, mode: YoloMode) -> None: + """Set YOLO mode; rebuild tool registry when soft-confirm hooks change.""" + prev = self.soft_confirm_enabled() + self.yolo = mode + if prev != self.soft_confirm_enabled(): + self.rebuild_client() + + def expire_yolo_once(self, *, quiet: bool = False) -> None: + """If mode is ``once``, drop back to ``off`` after a user turn.""" + if self.effective_yolo() != "once": + return + self.set_yolo("off") + if not quiet: + import click + + click.secho("yolo=off (once expired)", fg="bright_black", err=True) def _tool_registry(self) -> ToolRegistry | None: if not self.tools_enabled: @@ -88,7 +113,7 @@ class ReplState: from plyngent.cli.limits import prompt_confirm_tool_async from plyngent.tools.danger import classify_danger - if self._confirm_destructive(): + if self.soft_confirm_enabled(): return ToolRegistry( list(DEFAULT_TOOLS), danger=classify_danger, diff --git a/tests/test_agent/test_confirm.py b/tests/test_agent/test_confirm.py index 96d3972..4c23f69 100644 --- a/tests/test_agent/test_confirm.py +++ b/tests/test_agent/test_confirm.py @@ -56,4 +56,14 @@ async def test_confirm_deny() -> None: async def test_no_hooks_skips_confirm() -> None: registry = ToolRegistry([delete_path]) + assert registry.soft_confirm is False assert await registry.execute("delete_path", '{"path": "a.txt"}') == "deleted a.txt" + + +async def test_soft_confirm_property() -> None: + def on_confirm(name: str, args: Mapping[str, object], reason: str) -> bool: + del name, args, reason + return True + + gated = ToolRegistry([delete_path], danger=classify_danger, on_confirm=on_confirm) + assert gated.soft_confirm is True diff --git a/tests/test_cli/test_readline_setup.py b/tests/test_cli/test_readline_setup.py index 0a385ec..a42dd21 100644 --- a/tests/test_cli/test_readline_setup.py +++ b/tests/test_cli/test_readline_setup.py @@ -181,3 +181,5 @@ def test_complete_slash_args_from_registry(tmp_path: object) -> None: assert complete_slash_args(state, "/model", "a") == ["alpha"] assert complete_slash_args(state, "/export", "j") == ["json"] assert complete_slash_args(state, "/help", "st") == ["status", "stream"] + assert complete_slash_args(state, "/yolo", "") == ["on", "off", "once"] + assert complete_slash_args(state, "/yolo", "o") == ["on", "off", "once"] diff --git a/tests/test_cli/test_repl_commands.py b/tests/test_cli/test_repl_commands.py index cef9f2f..e3c618c 100644 --- a/tests/test_cli/test_repl_commands.py +++ b/tests/test_cli/test_repl_commands.py @@ -341,6 +341,63 @@ async def test_verbose_toggle(state: ReplState) -> None: assert get_verbose_tool_results() is False +async def test_yolo_toggle(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + assert state.effective_yolo() == "off" + assert state.soft_confirm_enabled() is True + assert await handle_slash(state, "/yolo") is True + assert "yolo=off" in capsys.readouterr().out + + assert await handle_slash(state, "/yolo on") is True + assert state.effective_yolo() == "on" + assert state.soft_confirm_enabled() is False + assert "yolo=on" in capsys.readouterr().out + + assert await handle_slash(state, "/yolo once") is True + assert state.effective_yolo() == "once" + assert state.soft_confirm_enabled() is False + + state.expire_yolo_once(quiet=True) + assert state.effective_yolo() == "off" + assert state.soft_confirm_enabled() is True + + assert await handle_slash(state, "/yolo once") is True + state.expire_yolo_once() + err = capsys.readouterr().err + assert "yolo=off (once expired)" in err + assert state.effective_yolo() == "off" + + +async def test_yolo_rebuilds_tool_registry(tmp_path: Path) -> None: + _ = set_workspace_root(tmp_path) + memory = await MemoryStore.open(DatabaseConfig()) + provider = OpenAIProvider(access_key_or_token="sk-test") + config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document()) + config.providers = {"local": provider} + st = ReplState( + config=config, + memory=memory, + workspace=tmp_path, + provider_name="local", + provider=provider, + model="gpt-test", + tools_enabled=True, + ) + try: + assert st.agent.tools is not None + assert st.agent.tools.soft_confirm is True + st.set_yolo("on") + assert st.agent.tools is not None + assert st.agent.tools.soft_confirm is False + st.set_yolo("once") + assert st.agent.tools is not None + assert st.agent.tools.soft_confirm is False + st.set_yolo("off") + assert st.agent.tools is not None + assert st.agent.tools.soft_confirm is True + finally: + await memory.close() + + async def test_resume(state: ReplState) -> None: sid = state.session_id assert sid is not None @@ -378,6 +435,7 @@ async def test_status_shows_context_tokens(state: ReplState, capsys: pytest.Capt assert "(est)" in out # no API usage yet assert "context_chars=" in out assert "tool_result_max=" in out + assert "yolo=off" in out assert str(state.workspace) in out state.agent.last_request_usage = TokenUsage(