core/cli+prompting: let a prompt name the answers its notification offers

Every button-bearing prompt offered the same three answers, so the ones whose
terminal read takes more than yes or no lost theirs the moment the human
answered from the notification: the limit prompt could not say "stop asking this
turn" (`yyy`), the batch confirm could not say "one at a time" (`m`), a skill
write could not say "this chat" (`s`), and the limit prompt could not be
answered at all — its wait was not even consulted.

`PromptWait` now carries `choices` (a key the terminal read accepts, plus a
label a human reads), `prompt_wait` passes them through, and the notifier sends
them as buttons, appending "Go to terminal" as the way back. The limit prompt
reads both surfaces like a timed confirm does, and answers that used to be
terminal-only tokens arrive as themselves.

A prompt that follows an answer given on the notification — the "why denied"
comment, the per-call prompts after "Review one by one" — is announced at once:
the grace period and the burst throttle exist for prompts the human has not
noticed yet, and they have.

src/plyngent/cli/limits.py

@@ -69,11 +69,33 @@ _CONTINUE_PROMPT = "Raise limit and continue? [Y/n/yyy] (yyy = stop asking this
 _CONTINUE_YES_TOKENS = frozenset({"", "y", "yes"})
 _CONTINUE_AUTO_TOKEN = "yyy"
 
+# The answers each prompt accepts, as notification buttons: the key is the
+# token the terminal read takes, the label is what a human reads there.
+_LIMIT_CHOICES: tuple[tuple[str, str], ...] = (
+    ("y", "Continue"),
+    (_CONTINUE_AUTO_TOKEN, "Continue, stop asking"),
+    ("n", "Stop here"),
+)
+
 # Batch confirm answers: ``y`` allows the round's calls, ``m`` reviews them one
 # at a time through the per-call prompt.
 _BATCH_YES_TOKENS = frozenset({"y", "yes"})
 _BATCH_MANUAL_TOKENS = frozenset({"m", "manual", "each"})
 
+_BATCH_CHOICES: tuple[tuple[str, str], ...] = (
+    ("y", "Allow all"),
+    ("m", "Review one by one"),
+    ("n", "Deny"),
+)
+
+# Skill writes: ``o`` allows this change, ``s`` the whole directory for the rest
+# of the chat.
+_SKILL_CHOICES: tuple[tuple[str, str], ...] = (
+    ("o", "Allow once"),
+    ("s", "This chat"),
+    ("n", "Deny"),
+)
+
 
 @dataclass
 class _AutoContinue:
@@ -202,9 +224,17 @@ def _decide_continue_limit(reason: str) -> bool:
     backend = get_prompt_backend()
     if not backend.is_interactive():
         return False
-    with prompt_wait("limit", "limit reached — continue?", reason):
+    with prompt_wait("limit", "limit reached — continue?", reason, choices=_LIMIT_CHOICES) as wait:
         _echo_continue_limit(reason)
+        source = _answer_source(wait)
+        if source is None:
             token = backend.read_line(_CONTINUE_PROMPT, default="y").strip().lower()
+        else:
+            # Same read as the timed confirms: stdin or the notification,
+            # whichever answers first, with the terminal's default on silence.
+            token = (_read_answer(None, wait) or "").strip().lower()
+            if not token:
+                token = "y"
     if token == _CONTINUE_AUTO_TOKEN:
         _auto_continue.turn = True
         return True
@@ -453,7 +483,7 @@ def prompt_skill_write_confirm(
     try:
         with (
             pause_task_cancel_for_prompt(),
-            prompt_wait("skills", f"allow {action} in {skill_dir.name!r}?", detail) as wait,
+            prompt_wait("skills", f"allow {action} in {skill_dir.name!r}?", detail, choices=_SKILL_CHOICES) as wait,
         ):
             backend = get_prompt_backend()
             if not backend.is_interactive():
@@ -591,7 +621,7 @@ def _prompt_confirm_batch_sync(requests: Sequence[ToolConfirmRequest]) -> BatchC
         # Non-interactive: deny, like the default of the single prompt.
         return BatchConfirmDecision(allowed=False)
     names = ", ".join(request.name for request in requests)
-    with prompt_wait("batch", f"{len(requests)} tool calls need confirmation", names) as wait:
+    with prompt_wait("batch", f"{len(requests)} tool calls need confirmation", names, choices=_BATCH_CHOICES) as wait:
         backend.echo()
         backend.secho(format_batch_confirm_box(requests), fg="yellow")
         backend.secho("m = review them one call at a time", fg="bright_black")

src/plyngent/cli/notify.py

@@ -3,10 +3,12 @@
 A prompt blocks the turn, so someone who switched to another window learns about
 it only when they come back. The notifier announces a prompt once it has waited
 ``grace_seconds`` — answering at the terminal stays silent — and, where the
-platform allows it, offers the answer itself: allow / deny / review buttons for
+platform allows it, offers the answer itself: the prompt's own answers as
-a confirmation, a reply field for a question. A prompt that can be answered from
+buttons (allow / deny / review for a confirmation), a reply field for a
-the notification reads from both surfaces from its first poll, since the
+question. A prompt that can be answered from the notification reads from both
-announcement only appears once the grace period is over.
+surfaces from its first poll, since the announcement only appears once the
+grace period is over. A prompt that follows an answer given on the notification
+is announced at once: the human is already looking at it.
 
 Two channels. The desktop one is the optional ``desktop-notifier`` extra
 (``pip install plyngent[notify]``); the terminal one writes an OSC escape
@@ -48,6 +50,7 @@ _ANSWER_CAPABILITY: dict[str, str] = {
     "policy": "buttons",
     "access": "buttons",
     "skills": "buttons",
+    "limit": "buttons",
     "ask": "reply",
 }
 
@@ -339,12 +342,16 @@ class WaitHandle:
     #: one from the first moment it blocks — the notification itself is only
     #: shown after the grace period.
     supports_actions: bool
+    #: Whether this prompt continues an exchange the human answered on the
+    #: notification: it is then announced at once, and not throttled.
+    handoff: bool
 
     def __init__(self, notifier: WaitNotifier, wait: PromptWait) -> None:
         self.notifier = notifier
         self.wait = wait
         self.identifier = None
         self.supports_actions = notifier.answer_capability(wait.kind) is not None
+        self.handoff = notifier.take_handoff()
         self._answers: SimpleQueue[str] = SimpleQueue()
         self._timer: threading.Timer | None = None
         self._closed: bool = False
@@ -364,10 +371,14 @@ class WaitHandle:
             # "review" means "I will answer in the terminal": withdraw and wait.
             self.notifier.dismiss(self)
             return
+        # The human is at the notification now, so a prompt that follows this
+        # answer is announced at once instead of waiting out the grace period.
+        self.notifier.handoff()
         self._answers.put(key)
 
     def __enter__(self) -> Self:
-        timer = threading.Timer(self.notifier.grace_seconds, self._notify)
+        delay = 0.0 if self.handoff else self.notifier.grace_seconds
+        timer = threading.Timer(delay, self._notify)
         timer.daemon = True
         self._timer = timer
         timer.start()
@@ -386,7 +397,7 @@ class WaitHandle:
         self.notifier.dismiss(self)
 
     def _notify(self) -> None:
-        if self._closed or not self.notifier.may_notify():
+        if self._closed or not self.notifier.may_notify(self):
             return
         if not self.notifier.notify_desktop(self):
             self.notifier.notify_terminal(self.wait)
@@ -407,6 +418,7 @@ class WaitNotifier:
     _env: Mapping[str, str]
     _clock: Callable[[], float]
     _last_sent: float
+    _handoff: bool
 
     def __init__(
         self,
@@ -429,6 +441,7 @@ class WaitNotifier:
         self._env = os.environ if env is None else env
         self._clock = clock
         self._last_sent = float("-inf")
+        self._handoff = False
 
     # -- hook ------------------------------------------------------------- #
 
@@ -439,16 +452,27 @@ class WaitNotifier:
     def grace_seconds(self) -> float:
         return self._config.grace_seconds
 
+    def handoff(self) -> None:
+        """Note that the human is answering on the notification right now."""
+        self._handoff = True
+
+    def take_handoff(self) -> bool:
+        """Consume the handoff: this prompt is part of the same exchange."""
+        handoff = self._handoff
+        self._handoff = False
+        return handoff
+
     def _wants(self, channel: str) -> bool:
         channels = self._config.channels
         return channel in channels or "auto" in channels
 
-    def may_notify(self) -> bool:
+    def may_notify(self, handle: WaitHandle) -> bool:
-        """Whether a notification is worth sending right now."""
+        """Whether this prompt is worth announcing right now."""
         if not self._config.enabled or not get_prompt_backend().is_interactive():
             return False
         now = self._clock()
-        if now - self._last_sent < self._config.min_interval_seconds:
+        # A prompt the human reached through the notification is not a burst.
+        if not handle.handoff and now - self._last_sent < self._config.min_interval_seconds:
             return False
         self._last_sent = now
         return True
@@ -459,6 +483,12 @@ class WaitNotifier:
             return f"{title}\n{_clean(wait.detail, _MAX_MESSAGE)}"
         return title
 
+    def _buttons(self, wait: PromptWait) -> tuple[tuple[str, str], ...]:
+        """The answers *wait* accepts, and always a way back to the terminal."""
+        if not wait.choices:
+            return _BUTTONS
+        return (*wait.choices, (REVIEW, "Go to terminal"))
+
     def desktop_channel(self) -> DesktopChannel | None:
         """The desktop surface, resolved once (importing the extra on demand)."""
         if self._resolve_desktop:
@@ -509,7 +539,7 @@ class WaitNotifier:
                 DesktopRequest(
                     title=_clean(wait.title, _MAX_TITLE),
                     message=self._message(wait),
-                    buttons=_BUTTONS if capability == "buttons" else (),
+                    buttons=self._buttons(wait) if capability == "buttons" else (),
                     reply_field=capability == "reply",
                     on_answer=handle.answer,
                     on_reply=handle.answer,

src/plyngent/config/models.py

@@ -284,7 +284,8 @@ class NotifyConfig(Struct, omit_defaults=True):
     """Desktop notifications shown while a prompt waits on the human.
 
     A prompt is announced only after ``grace_seconds`` of waiting, so answering
-    at the terminal stays silent; ``min_interval_seconds`` throttles bursts.
+    at the terminal stays silent; ``min_interval_seconds`` throttles bursts, and
+    a prompt that follows an answer given on the notification skips both.
     ``channels`` picks the surfaces: ``auto`` tries the desktop first and falls
     back to the terminal, ``desktop`` / ``terminal`` pin one of them.
 

src/plyngent/prompting.py

@@ -126,12 +126,16 @@ class PromptWait:
     ``choose``, ``policy``, ``access``, ``skills``, ``limit``) so the host can
     tailor what it shows and which answers it offers; ``title`` is short and
     safe to surface anywhere, while ``detail`` carries one optional line the
-    caller has already redacted.
+    caller has already redacted. ``choices`` are the answers this particular
+    prompt accepts, as ``(key, label)`` pairs — a host that offers them sends
+    the key back, so the label can stay human and the key stay the token the
+    terminal read would accept. Empty means the host decides.
     """
 
     kind: str
     title: str
     detail: str = ""
+    choices: tuple[tuple[str, str], ...] = ()
 
 
 @runtime_checkable
@@ -176,9 +180,17 @@ def get_prompt_wait_hook() -> PromptWaitHook | None:
 
 
 @contextlib.contextmanager
-def prompt_wait(kind: str, title: str, detail: str = "") -> Generator[PromptAnswerSource | None]:
+def prompt_wait(
+    kind: str,
+    title: str,
+    detail: str = "",
+    choices: tuple[tuple[str, str], ...] = (),
+) -> Generator[PromptAnswerSource | None]:
     """Announce a blocking prompt to the host; yields its answer source (or None).
 
+    ``choices`` says which answers this prompt accepts (see :class:`PromptWait`);
+    a host that shows them sends the key back as the answer.
+
     Only the outermost prompt of a nested read is announced: a confirm that asks
     a follow-up question while it is already on screen must not notify — and
     interrupt — a second time.
@@ -189,7 +201,7 @@ def prompt_wait(kind: str, title: str, detail: str = "") -> Generator[PromptAnsw
         return
     token = _wait_depth.set(_wait_depth.get() + 1)
     try:
-        with hook(PromptWait(kind, title, detail)) as source:
+        with hook(PromptWait(kind, title, detail, choices)) as source:
             yield source
     finally:
         _wait_depth.reset(token)