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.
This commit is contained in:
2026-07-17 14:55:34 +08:00
parent 6a1160f1e3
commit 2761877460
8 changed files with 175 additions and 17 deletions
+5
View File
@@ -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)
+9 -3
View File
@@ -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",
+11 -3
View File
@@ -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()
+47 -3
View File
@@ -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
+33 -8
View File
@@ -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,
+10
View File
@@ -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
+2
View File
@@ -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"]
+58
View File
@@ -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(