From 685370d3cb4eb3aa76f5f9445a646817461df241 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Wed, 9 Sep 2026 22:44:20 +0800 Subject: [PATCH] cli/limits: yyy turn-scoped auto-continue for limit prompts --- src/plyngent/cli/limits.py | 73 ++++++++++++++++++++++++++++---- tests/test_cli/test_interrupt.py | 24 ++++++----- tests/test_cli/test_limits.py | 72 +++++++++++++++++++++++++++++-- 3 files changed, 148 insertions(+), 21 deletions(-) diff --git a/src/plyngent/cli/limits.py b/src/plyngent/cli/limits.py index 6cecda8..376a107 100644 --- a/src/plyngent/cli/limits.py +++ b/src/plyngent/cli/limits.py @@ -4,6 +4,7 @@ import contextlib import selectors import shutil import sys +from dataclasses import dataclass from typing import TYPE_CHECKING, Literal from plyngent.cli.interrupt import pause_task_cancel_for_prompt @@ -31,6 +32,43 @@ _BOX_MIN_WIDTH = 40 _BOX_MAX_WIDTH = 100 _BOX_PAD = 2 # spaces inside left/right borders +# Limit prompts accept ``yyy`` to stop asking for the rest of the turn. +_CONTINUE_PROMPT = "Raise limit and continue? [Y/n/yyy] (yyy = stop asking this turn): " +_CONTINUE_YES_TOKENS = frozenset({"", "y", "yes"}) +_CONTINUE_AUTO_TOKEN = "yyy" + + +@dataclass +class _AutoContinue: + """Auto-continue layers: process default (flag/config) + ``yyy`` turn skip.""" + + default: bool = False + turn: bool = False + + +_auto_continue = _AutoContinue() + + +def auto_continue_enabled() -> bool: + """Whether limit prompts are auto-approved (process default or ``yyy``).""" + return _auto_continue.default or _auto_continue.turn + + +def set_auto_continue_default(*, enabled: bool) -> None: + """Set the process-wide default (``--auto-continue`` / ``[agent]`` config).""" + _auto_continue.default = bool(enabled) + + +def reset_auto_continue_turn() -> None: + """Clear the ``yyy`` skip at a turn boundary.""" + _auto_continue.turn = False + + +def reset_auto_continue() -> None: + """Clear both layers (chat exit / tests).""" + _auto_continue.default = False + _auto_continue.turn = False + def _terminal_width() -> int: try: @@ -107,10 +145,33 @@ def _echo_continue_limit(reason: str) -> None: backend.echo() +def _echo_auto_continue(reason: str) -> None: + """Echo the limit reason plus the ``yyy`` notice (no prompt).""" + backend = get_prompt_backend() + backend.secho(f"[limit] {reason}", fg="yellow") + backend.secho("[limit] auto-continuing (yyy) — no more limit prompts this turn", fg="bright_black") + backend.echo() + + +def _decide_continue_limit(reason: str) -> bool: + """Shared limit decision for the sync/async prompts (may read a line).""" + if auto_continue_enabled(): + _echo_auto_continue(reason) + return True + backend = get_prompt_backend() + if not backend.is_interactive(): + return False + _echo_continue_limit(reason) + token = backend.read_line(_CONTINUE_PROMPT, default="y").strip().lower() + if token == _CONTINUE_AUTO_TOKEN: + _auto_continue.turn = True + return True + return token in _CONTINUE_YES_TOKENS + + def _prompt_continue_limit_sync(reason: str) -> bool: try: - _echo_continue_limit(reason) - return confirm("Raise limit and continue?", default=True) + return _decide_continue_limit(reason) except NonInteractiveError: return False @@ -122,15 +183,11 @@ def prompt_continue_limit(reason: str) -> bool: async def prompt_continue_limit_async(reason: str) -> bool: - """Async variant: confirm off the event loop so the turn is not cancelled.""" + """Async variant: prompt off the event loop so the turn is not cancelled.""" try: from plyngent.prompting import run_prompt_async - def _run() -> bool: - _echo_continue_limit(reason) - return confirm("Raise limit and continue?", default=True) - - return await run_prompt_async(_run) + return await run_prompt_async(lambda: _decide_continue_limit(reason)) except NonInteractiveError: return False diff --git a/tests/test_cli/test_interrupt.py b/tests/test_cli/test_interrupt.py index f550464..97a9140 100644 --- a/tests/test_cli/test_interrupt.py +++ b/tests/test_cli/test_interrupt.py @@ -16,6 +16,8 @@ from plyngent.cli.interrupt import ( ) from plyngent.cli.limits import prompt_continue_limit, prompt_continue_limit_async from plyngent.cli.retry import run_cancellable +from plyngent.prompting import temporary_backend +from tests.test_prompting import ScriptedBackend if TYPE_CHECKING: pass @@ -39,12 +41,16 @@ def test_nested_pause_depth() -> None: def test_prompt_continue_limit_under_pause(monkeypatch: pytest.MonkeyPatch) -> None: - def _confirm(*_a: object, **_k: object) -> bool: - assert allow_task_cancel() is False - return True + backend = ScriptedBackend(["y"]) - monkeypatch.setattr("click.confirm", _confirm) - assert prompt_continue_limit("too many rounds") is True + def _read_line(prompt: str, *, default: str | None = None, completions: object = None) -> str: + del prompt, default, completions + assert allow_task_cancel() is False + return "y" + + monkeypatch.setattr(backend, "read_line", _read_line) + with temporary_backend(backend): + assert prompt_continue_limit("too many rounds") is True async def test_run_in_prompt_thread_pauses_cancel() -> None: @@ -60,11 +66,9 @@ async def test_run_in_prompt_thread_pauses_cancel() -> None: async def test_prompt_continue_limit_async(monkeypatch: pytest.MonkeyPatch) -> None: - def _confirm(*_a: object, **_k: object) -> bool: - return True - - monkeypatch.setattr("click.confirm", _confirm) - assert await prompt_continue_limit_async("too many rounds") is True + del monkeypatch + with temporary_backend(ScriptedBackend(["y"])): + assert await prompt_continue_limit_async("too many rounds") is True async def test_sigint_cancels_after_prompt_pause() -> None: diff --git a/tests/test_cli/test_limits.py b/tests/test_cli/test_limits.py index 61b137c..2c33582 100644 --- a/tests/test_cli/test_limits.py +++ b/tests/test_cli/test_limits.py @@ -1,30 +1,96 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import pytest from plyngent.cli.limits import ( + auto_continue_enabled, format_tool_confirm_box, install_cli_limit_hooks, prompt_confirm_tool, prompt_continue_limit, + reset_auto_continue, + reset_auto_continue_turn, + set_auto_continue_default, ) -from plyngent.prompting import get_prompt_backend, temporary_backend +from plyngent.prompting import NonInteractiveBackend, get_prompt_backend, temporary_backend from plyngent.tools.process.pty_session import PtyManager from tests.test_prompting import ScriptedBackend +if TYPE_CHECKING: + from collections.abc import Iterator + + +@pytest.fixture +def clean_auto_continue() -> Iterator[None]: + reset_auto_continue() + yield + reset_auto_continue() + def test_prompt_continue_limit_yes() -> None: - backend = ScriptedBackend([], confirms=[True]) + backend = ScriptedBackend(["y"]) + with temporary_backend(backend): + assert prompt_continue_limit("hit a wall") is True + + +def test_prompt_continue_limit_default_yes() -> None: + backend = ScriptedBackend([]) with temporary_backend(backend): assert prompt_continue_limit("hit a wall") is True def test_prompt_continue_limit_no() -> None: - backend = ScriptedBackend([], confirms=[False]) + backend = ScriptedBackend(["n"]) with temporary_backend(backend): assert prompt_continue_limit("hit a wall") is False +def test_prompt_continue_limit_yyy_sets_turn_skip(clean_auto_continue: None) -> None: + del clean_auto_continue + backend = ScriptedBackend(["yyy"]) + with temporary_backend(backend): + assert prompt_continue_limit("hit a wall") is True + assert auto_continue_enabled() is True + # No further prompt: the script is empty and must not be consumed. + assert prompt_continue_limit("hit a wall again") is True + + +def test_reset_auto_continue_turn_clears_yyy(clean_auto_continue: None) -> None: + del clean_auto_continue + with temporary_backend(ScriptedBackend(["yyy"])): + assert prompt_continue_limit("limit") is True + reset_auto_continue_turn() + assert auto_continue_enabled() is False + with temporary_backend(ScriptedBackend(["n"])): + assert prompt_continue_limit("limit") is False + + +def test_auto_continue_default_short_circuits_prompt(clean_auto_continue: None) -> None: + del clean_auto_continue + set_auto_continue_default(enabled=True) + assert auto_continue_enabled() is True + with temporary_backend(NonInteractiveBackend()): + assert prompt_continue_limit("limit") is True + + +def test_prompt_continue_limit_noninteractive_denies() -> None: + with temporary_backend(NonInteractiveBackend()): + assert prompt_continue_limit("limit") is False + + +def test_pty_offer_raise_uses_auto_continue(clean_auto_continue: None) -> None: + del clean_auto_continue + PtyManager.set_limit_continue_hook(prompt_continue_limit) + set_auto_continue_default(enabled=True) + try: + with temporary_backend(NonInteractiveBackend()): + assert PtyManager._offer_raise("PTY session limit reached") is True + finally: + PtyManager.set_limit_continue_hook(None) + + def test_prompt_confirm_tool_default_deny() -> None: backend = ScriptedBackend([], confirms=[False]) with temporary_backend(backend):