prompting: let Ctrl+C cancel poll-based prompt reads

The wait tool's prompt runs off-loop on a worker thread, but its read polls
select/msvcrt, so it can act on a cancel: the SIGINT target now sets a flag the
reader checks between polls and raises PromptCancelledError with, which the tool
reports as "cancelled by user" while the turn continues. run_prompt_async /
run_cancellable_prompt_async hand the matching hook to the CLI guard; a stale
cancel never leaks into the next prompt read.
This commit is contained in:
2026-09-22 19:28:03 +08:00
parent 6ec7b4de12
commit 27bd7b46f7
5 changed files with 204 additions and 17 deletions
+76 -8
View File
@@ -4,6 +4,7 @@ import asyncio
import contextlib import contextlib
import select import select
import sys import sys
import threading
import time import time
from contextlib import AbstractContextManager from contextlib import AbstractContextManager
from dataclasses import dataclass from dataclasses import dataclass
@@ -19,6 +20,30 @@ class NonInteractiveError(RuntimeError):
"""Raised when a prompt is required but no interactive backend is available.""" """Raised when a prompt is required but no interactive backend is available."""
class PromptCancelledError(RuntimeError):
"""Raised by a poll-based prompt read when the human cancels it (Ctrl+C).
Only poll-based reads (``read_line_with_timeout``) can observe the cancel:
they check :func:`prompt_cancelled` between polls, so the reader thread
returns and nothing is left blocked on stdin.
"""
# Set by the SIGINT target registered for a poll-based prompt read (see
# cli.interrupt.off_loop_prompt); cleared around each off-loop prompt.
_prompt_cancel = threading.Event()
def prompt_cancelled() -> bool:
"""Whether the in-flight poll-based prompt read was cancelled by SIGINT."""
return _prompt_cancel.is_set()
def cancel_prompt_read() -> None:
"""Ask the in-flight poll-based prompt read to return (thread-safe)."""
_prompt_cancel.set()
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class ChoiceOption: class ChoiceOption:
"""One selectable option for :func:`choose`.""" """One selectable option for :func:`choose`."""
@@ -117,6 +142,8 @@ def _read_line_with_timeout_posix(
Returns the line (without trailing newline), ``""`` for an empty Enter, or Returns the line (without trailing newline), ``""`` for an empty Enter, or
``None`` when the timeout elapses before any line is complete (EOF too). ``None`` when the timeout elapses before any line is complete (EOF too).
Raises :class:`PromptCancelledError` when :func:`prompt_cancelled` turns true
while waiting (the poll loop notices within one interval).
""" """
_ = sys.stdout.write(prompt) _ = sys.stdout.write(prompt)
_ = sys.stdout.flush() _ = sys.stdout.flush()
@@ -125,6 +152,8 @@ def _read_line_with_timeout_posix(
remaining = deadline - time.monotonic() remaining = deadline - time.monotonic()
if remaining <= 0: if remaining <= 0:
return None return None
if prompt_cancelled():
raise PromptCancelledError
ready, _, _ = poll([stream], [], [], min(remaining, 0.1)) ready, _, _ = poll([stream], [], [], min(remaining, 0.1))
if not ready: if not ready:
continue continue
@@ -143,6 +172,8 @@ def _read_line_with_timeout_windows(prompt: str, timeout: float) -> str | None:
deadline = time.monotonic() + timeout deadline = time.monotonic() + timeout
buf: list[str] = [] buf: list[str] = []
while time.monotonic() < deadline: while time.monotonic() < deadline:
if prompt_cancelled():
raise PromptCancelledError
if msvcrt.kbhit(): if msvcrt.kbhit():
ch = msvcrt.getwch() ch = msvcrt.getwch()
if ch in ("\r", "\n"): if ch in ("\r", "\n"):
@@ -171,7 +202,9 @@ def read_line_with_timeout(prompt: str, timeout: float) -> str | None:
Returns the typed line (without trailing newline), ``""`` for an empty Returns the typed line (without trailing newline), ``""`` for an empty
Enter, or ``None`` when the timeout elapses first (EOF too). Ctrl+C Enter, or ``None`` when the timeout elapses first (EOF too). Ctrl+C
propagates as :class:`KeyboardInterrupt`. POSIX polls stdin via ``select``; propagates as :class:`KeyboardInterrupt`; :func:`cancel_prompt_read` (the
SIGINT target of an off-loop prompt read) raises
:class:`PromptCancelledError` instead. POSIX polls stdin via ``select``;
Windows uses ``msvcrt`` char polling (best-effort). Windows uses ``msvcrt`` char polling (best-effort).
""" """
if timeout < 0: if timeout < 0:
@@ -263,16 +296,18 @@ class NonInteractiveBackend:
_backend: PromptBackend = ClickPromptBackend() _backend: PromptBackend = ClickPromptBackend()
_pause_factory: Callable[[], AbstractContextManager[None]] | None = None # CLI hook: context manager around an off-loop prompt. It receives the read's
# cancel hook (poll-based reads) or None (reads that cannot be interrupted).
_pause_factory: Callable[[Callable[[], None] | None], AbstractContextManager[None]] | None = None
_prompt_lock = asyncio.Lock() _prompt_lock = asyncio.Lock()
def configure_prompting( def configure_prompting(
*, *,
backend: PromptBackend | None = None, backend: PromptBackend | None = None,
pause_factory: Callable[[], AbstractContextManager[None]] | None = None, pause_factory: Callable[[Callable[[], None] | None], AbstractContextManager[None]] | None = None,
) -> None: ) -> None:
"""Install process-wide prompt backend and optional cancel-pause context.""" """Install process-wide prompt backend and optional SIGINT guard for prompts."""
global _backend, _pause_factory # noqa: PLW0603 global _backend, _pause_factory # noqa: PLW0603
if backend is not None: if backend is not None:
_backend = backend _backend = backend
@@ -284,10 +319,11 @@ def get_prompt_backend() -> PromptBackend:
def reset_prompting() -> None: def reset_prompting() -> None:
"""Restore default Click backend and clear pause hook (tests).""" """Restore default Click backend, clear the pause hook and cancel flag (tests)."""
global _backend, _pause_factory # noqa: PLW0603 global _backend, _pause_factory # noqa: PLW0603
_backend = ClickPromptBackend() _backend = ClickPromptBackend()
_pause_factory = None _pause_factory = None
_prompt_cancel.clear()
def _normalize_options( def _normalize_options(
@@ -464,13 +500,45 @@ def form(
backend.echo("Starting over…") backend.echo("Starting over…")
async def run_prompt_async[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R: async def _prompt_off_loop[R](
"""Run a blocking prompt off the event loop, serialized, with optional SIGINT pause.""" func: Callable[..., R],
*args: object,
cancel_hook: Callable[[], None] | None,
**kwargs: object,
) -> R:
"""Shared body: run *func* off the event loop under the CLI's SIGINT guard."""
async with _prompt_lock: async with _prompt_lock:
_prompt_cancel.clear()
try:
if _pause_factory is not None: if _pause_factory is not None:
with _pause_factory(): with _pause_factory(cancel_hook):
return await asyncio.to_thread(func, *args, **kwargs) return await asyncio.to_thread(func, *args, **kwargs)
return await asyncio.to_thread(func, *args, **kwargs) return await asyncio.to_thread(func, *args, **kwargs)
finally:
_prompt_cancel.clear()
async def run_prompt_async[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
"""Run a blocking prompt off the event loop, serialized, SIGINT-guarded.
While the prompt waits, a Ctrl+C must not cancel the enclosing turn: the CLI
guard registers the prompt as the SIGINT target. readline-based reads cannot
be interrupted without abandoning the reader thread on stdin, so they ignore
the interrupt with a one-line hint (see :func:`run_cancellable_prompt_async`
for poll-based reads).
"""
return await _prompt_off_loop(func, *args, cancel_hook=None, **kwargs)
async def run_cancellable_prompt_async[**P, R](func: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
"""Run a *poll-based* prompt read off the event loop; Ctrl+C cancels it.
Only for reads that observe :func:`prompt_cancelled`
(``read_line_with_timeout``): the SIGINT raises :class:`PromptCancelledError`
in the reader thread, so it returns to the caller instead of leaving a thread
blocked on stdin (and without cancelling the enclosing turn).
"""
return await _prompt_off_loop(func, *args, cancel_hook=cancel_prompt_read, **kwargs)
async def ask_async(prompt: str, *, default: str | None = None) -> str: async def ask_async(prompt: str, *, default: str | None = None) -> str:
+13 -4
View File
@@ -7,7 +7,12 @@ import sys
from click import style from click import style
from plyngent.agent import ToolTag, tool from plyngent.agent import ToolTag, tool
from plyngent.prompting import get_prompt_backend, read_line_with_timeout, run_prompt_async from plyngent.prompting import (
PromptCancelledError,
get_prompt_backend,
read_line_with_timeout,
run_cancellable_prompt_async,
)
from plyngent.tools.chat.shape import first_error, number_arg, string_error from plyngent.tools.chat.shape import first_error, number_arg, string_error
_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m") _ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
@@ -36,7 +41,10 @@ async def _interactive_wait(seconds: float, reason: str | None) -> str:
"""Show the two-line wait prompt and report how the wait ended.""" """Show the two-line wait prompt and report how the wait ended."""
label = f"{seconds:g}" label = f"{seconds:g}"
prompt = _wait_prompt(seconds, reason=reason) prompt = _wait_prompt(seconds, reason=reason)
line = await run_prompt_async(read_line_with_timeout, prompt, seconds) try:
line = await run_cancellable_prompt_async(read_line_with_timeout, prompt, seconds)
except PromptCancelledError:
return "cancelled by user"
if line is None: if line is None:
return f"waited {label}s" return f"waited {label}s"
text = line.strip() text = line.strip()
@@ -51,8 +59,9 @@ async def wait(duration: int, *, reason: str | None = None) -> str:
Interactive sessions show a two-line prompt: a status line, then an Interactive sessions show a two-line prompt: a status line, then an
optional-reason input — pressing Enter (optionally after typing a reason) optional-reason input — pressing Enter (optionally after typing a reason)
"disturbs" the wait so the turn continues immediately. Non-interactive runs "disturbs" the wait so the turn continues immediately, and Ctrl+C cancels
simply sleep the full duration. the wait (the turn continues). Non-interactive runs simply sleep the full
duration.
""" """
seconds, duration_error = number_arg("wait", "duration", duration, example="5") seconds, duration_error = number_arg("wait", "duration", duration, example="5")
reason_error = None if reason is None else string_error("wait", "reason", reason) reason_error = None if reason is None else string_error("wait", "reason", reason)
+38 -1
View File
@@ -21,7 +21,13 @@ from plyngent.cli.interrupt import (
) )
from plyngent.cli.limits import prompt_continue_limit from plyngent.cli.limits import prompt_continue_limit
from plyngent.cli.retry import run_cancellable, sleep_cancellable from plyngent.cli.retry import run_cancellable, sleep_cancellable
from plyngent.prompting import temporary_backend from plyngent.prompting import (
PromptCancelledError,
configure_prompting,
prompt_cancelled,
run_cancellable_prompt_async,
temporary_backend,
)
from tests.test_prompting import ScriptedBackend from tests.test_prompting import ScriptedBackend
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -224,6 +230,37 @@ async def test_off_loop_prompt_cancel_hook_is_preferred(router: None) -> None:
assert turn_calls == [] assert turn_calls == []
async def test_sigint_cancels_off_loop_prompt_read(router: None) -> None:
"""Wiring: SIGINT aborts a poll-based read (not the turn) and clears the flag."""
del router
turn_calls: list[str] = []
seen: list[str] = []
def _poll_read(prompt: str, timeout: float) -> str:
"""Stand-in for ``read_line_with_timeout``: polls the cancel flag."""
del prompt
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if prompt_cancelled():
seen.append("cancelled")
raise PromptCancelledError
time.sleep(0.01)
seen.append("timed out")
return ""
with temporary_backend(ScriptedBackend([])):
configure_prompting(pause_factory=off_loop_prompt)
with sigint_cancels(FakeTarget(turn_calls, "turn")):
reader = asyncio.create_task(run_cancellable_prompt_async(_poll_read, "p> ", 30.0))
await asyncio.sleep(0.05)
os.kill(os.getpid(), signal.SIGINT)
with pytest.raises(PromptCancelledError):
await asyncio.wait_for(reader, timeout=5)
assert seen == ["cancelled"]
assert turn_calls == []
assert not prompt_cancelled() # cleared for the next prompt
async def test_pause_task_cancel_for_prompt_swaps_sigint(router: None) -> None: async def test_pause_task_cancel_for_prompt_swaps_sigint(router: None) -> None:
"""A main-thread blocking read takes SIGINT as a KeyboardInterrupt.""" """A main-thread blocking read takes SIGINT as a KeyboardInterrupt."""
del router del router
+57
View File
@@ -1,6 +1,8 @@
from __future__ import annotations from __future__ import annotations
import contextlib
import io import io
from typing import TYPE_CHECKING
import pytest import pytest
@@ -9,15 +11,33 @@ from plyngent.prompting import (
FormField, FormField,
NonInteractiveBackend, NonInteractiveBackend,
NonInteractiveError, NonInteractiveError,
PromptCancelledError,
_read_line_with_timeout_posix, _read_line_with_timeout_posix,
ask, ask,
cancel_prompt_read,
choose, choose,
configure_prompting,
confirm, confirm,
form, form,
prompt_cancelled,
read_line_with_timeout, read_line_with_timeout,
reset_prompting,
run_cancellable_prompt_async,
run_prompt_async,
temporary_backend, temporary_backend,
) )
if TYPE_CHECKING:
from collections.abc import Generator, Iterator
@pytest.fixture(autouse=True)
def reset_prompting_state() -> Iterator[None]:
"""Prompting state (backend, SIGINT guard, cancel flag) is process-global."""
reset_prompting()
yield
reset_prompting()
def _ready_poll(r, w, x, t): def _ready_poll(r, w, x, t):
return (r, w, x) return (r, w, x)
@@ -51,6 +71,43 @@ def test_read_line_with_timeout_negative_timeout() -> None:
assert read_line_with_timeout("p> ", -1) is None assert read_line_with_timeout("p> ", -1) is None
def test_read_line_with_timeout_cancelled() -> None:
"""A cancel request (SIGINT during an off-loop prompt) ends the read."""
stream = io.StringIO()
cancel_prompt_read()
try:
assert prompt_cancelled()
with pytest.raises(PromptCancelledError):
_read_line_with_timeout_posix("p> ", 5.0, stream=stream, poll=_never_poll)
finally:
stream.close()
async def test_run_prompt_async_clears_cancel_flag() -> None:
"""A stale cancel request must not leak into the next prompt read."""
cancel_prompt_read()
with temporary_backend(NonInteractiveBackend()):
assert await run_prompt_async(lambda: "ok") == "ok"
assert not prompt_cancelled()
async def test_run_prompt_async_passes_cancel_hook_only_when_cancellable() -> None:
"""The CLI guard gets the read's cancel hook for poll-based reads only."""
seen: list[object] = []
@contextlib.contextmanager
def _guard(cancel: object) -> Generator[None]:
seen.append(cancel)
yield
with temporary_backend(NonInteractiveBackend()):
# temporary_backend resets the guard, so install it inside.
configure_prompting(pause_factory=_guard)
assert await run_cancellable_prompt_async(lambda: "a") == "a"
assert await run_prompt_async(lambda: "b") == "b"
assert seen == [cancel_prompt_read, None]
class ScriptedBackend: class ScriptedBackend:
"""Deterministic backend for unit tests.""" """Deterministic backend for unit tests."""
+17 -1
View File
@@ -3,8 +3,10 @@ from __future__ import annotations
import importlib import importlib
import json import json
import pytest
from plyngent.agent import ToolRegistry from plyngent.agent import ToolRegistry
from plyngent.prompting import NonInteractiveBackend, temporary_backend from plyngent.prompting import NonInteractiveBackend, PromptCancelledError, temporary_backend
from plyngent.tools.chat import CHAT_TOOLS, ask_user, choose_user, form_user, wait from plyngent.tools.chat import CHAT_TOOLS, ask_user, choose_user, form_user, wait
from tests.test_prompting import ScriptedBackend from tests.test_prompting import ScriptedBackend
@@ -146,6 +148,20 @@ async def test_wait_tool_times_out(monkeypatch) -> None:
assert out == "waited 3s" assert out == "waited 3s"
async def test_wait_tool_cancelled_by_user(monkeypatch: pytest.MonkeyPatch) -> None:
"""Ctrl+C during the wait prompt cancels the wait; the turn continues."""
def _cancelled(prompt: str, timeout: float) -> str:
del prompt, timeout
raise PromptCancelledError
monkeypatch.setattr(wait_module, "read_line_with_timeout", _cancelled)
with temporary_backend(ScriptedBackend([])):
registry = ToolRegistry([wait])
out = await registry.execute("wait", '{"duration": 30}')
assert out == "cancelled by user"
async def test_wait_tool_prompt_two_lines_with_reason(monkeypatch) -> None: async def test_wait_tool_prompt_two_lines_with_reason(monkeypatch) -> None:
prompts: list[str] = [] prompts: list[str] = []
monkeypatch.setattr( monkeypatch.setattr(