core/prompting: answer a question with a notification reply

`ask` now reads its answer from the wait's source as well, which is what makes
the notification's reply field mean something: the reply typed into it is the
answer the caller gets. The readline path stays whenever the source offers no
answers, so line editing and Tab completion only go away when the human asked
for a surface that can answer — and a terminal answer still wins over one that
arrives through the notification.

The read itself is the poll-based one this module already had, so the same
"no deadline, wait for a line or an answer" rule and the same Ctrl+C handling
apply; a cancelled read raises `NonInteractiveError`, as the backend's own
cancellation does.
This commit is contained in:
2026-10-21 20:02:56 +08:00
parent 35595d8b04
commit be8a0111ec
2 changed files with 63 additions and 2 deletions
+31 -2
View File
@@ -509,6 +509,34 @@ def _show_choices(
backend.echo(" (or type a custom answer)")
def _read_with_source(
prompt: str,
source: PromptAnswerSource | None,
*,
default: str | None,
completions: Sequence[str] | None = None,
) -> str:
"""Read one answer line, also accepting one offered by *source*.
The backend's readline path is kept whenever nothing can answer remotely, so
line editing (and Tab completion) only go away when the human asked for a
notification that carries the answer.
"""
backend = get_prompt_backend()
if source is None:
return backend.read_line(prompt, default=default, completions=completions).strip()
try:
answer = source.take() if not sys.stdin.isatty() else read_line_with_timeout(f"{prompt}: ", None, source=source)
except OSError, ValueError, AttributeError:
answer = source.take()
if answer is None:
if default is not None:
return default
msg = "prompt cancelled"
raise NonInteractiveError(msg)
return answer.strip()
def ask(
prompt: str,
*,
@@ -523,9 +551,10 @@ def ask(
if not backend.is_interactive() and default is None:
msg = f"non-interactive: cannot prompt for {prompt!r}"
raise NonInteractiveError(msg)
with prompt_wait("ask", prompt):
with prompt_wait("ask", prompt) as wait:
backend.secho(prompt, fg="yellow")
return backend.read_line("Answer", default=default, completions=completions).strip()
source = wait if wait is not None and wait.supports_actions else None
return _read_with_source("Answer", source, default=default, completions=completions)
def ask_secret(prompt: str) -> str:
+32
View File
@@ -382,3 +382,35 @@ def test_reset_prompting_clears_the_wait_hook() -> None:
assert get_prompt_wait_hook() is not None
reset_prompting()
assert get_prompt_wait_hook() is None
def test_ask_takes_an_answer_from_the_notification() -> None:
"""A reply typed into the notification is the answer (no readline involved)."""
from plyngent.prompting import set_prompt_wait_hook
@contextlib.contextmanager
def hook(wait: PromptWait) -> Generator[_AnswerSource]:
del wait
yield _AnswerSource("feature/x")
set_prompt_wait_hook(hook)
backend = ScriptedBackend([]) # any readline would raise
with temporary_backend(backend):
assert ask("Which branch?") == "feature/x"
def test_ask_ignores_a_source_that_offers_no_actions() -> None:
"""A notify-only surface leaves the terminal read alone."""
from plyngent.prompting import set_prompt_wait_hook
@contextlib.contextmanager
def hook(wait: PromptWait) -> Generator[_AnswerSource]:
del wait
source = _AnswerSource("from the notification")
source.supports_actions = False
yield source
set_prompt_wait_hook(hook)
backend = ScriptedBackend(["typed here"])
with temporary_backend(backend):
assert ask("Which branch?") == "typed here"