diff --git a/src/plyngent/prompting.py b/src/plyngent/prompting.py index a88e5e2..aa4a346 100644 --- a/src/plyngent/prompting.py +++ b/src/plyngent/prompting.py @@ -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: diff --git a/tests/test_prompting.py b/tests/test_prompting.py index 1ab95a5..b674f11 100644 --- a/tests/test_prompting.py +++ b/tests/test_prompting.py @@ -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"