12 Commits

Author SHA1 Message Date
NCBM 0e8d404642 release: plyngent 0.14.0 2026-10-24 04:43:15 +08:00
NCBM 19707c3bd6 docs: drop "Go to terminal" and explain the bell
The answer tables lose the third button (two answers, or the prompt's own),
the popup's close and a body click both withdraw it, and the desktop section
says why every popup rings the terminal: the human is at another window by
construction — the announcement waits out the grace period — and no desktop
lets a program raise one.
2026-10-24 04:41:01 +08:00
NCBM ae20f5131f test: cover the bell at announcement and the two-button answer set
A confirmation offers Allow once / Deny and a prompt's own ``choices`` are
the whole button set; a popup that goes out rings the bell once, and
withdrawing it afterwards — by click or close — adds nothing and never
answers the prompt. A reply field's text is an answer whatever it spells.
2026-10-24 04:40:50 +08:00
NCBM a7abc50aa5 core/cli: drop the go-to-terminal button and ring the bell when a popup goes out
"Go to terminal" promised something no desktop can do — raise another
application's window — and the popup's own close button does as much, so it
was a third button that answered nothing. `REVIEW` leaves the answer
vocabulary; a click on the body (the platform's default action) or a close
still withdraws the notification and leaves the prompt waiting.

The bell now rings as the notification is shown. A prompt is only announced
after `grace_seconds` of nobody answering it, so the human is at some other
window — possibly another virtual desktop, where a popup is easy to miss — and
the terminal's bell is the only portable cue that says where to look. It rings
once per announcement; withdrawing the popup afterwards stays silent.
2026-10-24 04:40:39 +08:00
NCBM 7452224793 docs: name the workspace line, the popup deadline and the bell
The body table gains the workspace (last line, `~` for home, following a
rebind), the desktop section says which popups expire and why, and "Go to
terminal" is described as what it does: withdraw, ring, leave the prompt
waiting — no window is raised, because no desktop allows it.
2026-10-24 04:35:26 +08:00
NCBM 474bf57093 test: cover the workspace line, the popup deadline and the bell
That a home path under `~` names the chat and follows a rebind, that a
timed prompt hands its deadline to the popup while an untimed one keeps it,
and that going to the terminal — by button or by clicking the body —
withdraws the notification, rings the bell and leaves the prompt waiting.
2026-10-24 04:35:15 +08:00
NCBM ea0827ddfb core/cli: name the workspace, let the popup expire, ring the bell
A notification said which prompt was waiting but not which chat, and two
windows look alike. The body's last line is now the workspace the chat runs
in, with the home directory shown as `~`, and it follows a session resumed
into — or rebound to — another directory: `ReplState` retargets the installed
notifier instead of freezing the directory the process started in.

A timed prompt has a deadline, the popup did not, so it could stay on screen
announcing an answer window that had already closed. The deadline is now the
popup's own timeout (rounded up to whole seconds); a prompt without one keeps
the old "until dismissed" behaviour.

"Go to terminal" (and a click on the body) withdrew the notification and left
the prompt waiting, but nothing showed the human *where*. No desktop lets a
program raise another window — Wayland has no such call and the X11 ones need
a helper per WM — so the terminal rings its bell instead.
2026-10-24 04:35:04 +08:00
NCBM 39258b2670 docs: say what a notification now carries and when silence answers
The body is no longer "the reason or nothing": it is the question, the
reason behind `include_reason`, the object of the prompt and the deadline,
with the desktop keeping the lines apart and the terminal flattening them.
The tables gain the answers a question offers (a reply field and the way
back), the channels' click-to-terminal behaviour, and the secret prompt that
is announced nowhere at all.
2026-10-24 04:22:42 +08:00
NCBM f511c08afb test: cover the body a prompt notification carries
What a prompt declares (question, object, deadline), how the two channels
render it (lines apart on the desktop, one line in a terminal), that a long
reason goes before the deadline does, that a question keeps its reply field
*and* the way back to the terminal, that a body click leaves the prompt
waiting, that a reply spelling "review" is still an answer, and that a secret
is never announced.
2026-10-24 04:22:00 +08:00
NCBM 04d34b34f3 core/cli: name what a confirm is about, and when it denies itself
A tool confirm's notification said which tool asked and nothing else, and
the three timed prompts (denylist, directory access, skill write) hid the
clock: each of them denies on timeout, so a human reading the notification
late was answering a prompt that had already given up.

`_tool_facts` lifts the object out of the call — the program's name, or the
file's — and never an argument, a URL or a piece of code: those stay in the
reason, which only travels with `include_reason`. The timed prompts pass
their timeout as `expires_in`, and a batch names the tools it asks about the
same way instead of keeping them behind the reason.
2026-10-24 04:21:37 +08:00
NCBM 9ed4620579 core/cli: render a prompt's body from what it declares
The body was the reason line or nothing, so with the default
`include_reason = false` every notification was a bare title. It is now the
question, then the reason when the config allows it, then what the prompt is
about, and last the deadline — filled in that order until the budget is gone,
so a long reason drops out before "no answer in 30s = deny" does. A desktop
popup wraps and keeps the lines apart; a terminal escape sequence carries them
on one line.

Two answers were missing from the surface: a reply prompt had no button at all
(a question could not be sent back to the terminal), and the notification body
was not clickable. Clicking it is the freedesktop default action, so it now
withdraws the notification and leaves the prompt waiting — the same as the
"Go to terminal" button. A typed reply is recorded by `reply()`, which keeps a
reply that spells "review" an answer; a secret is never announced at all.
2026-10-24 04:21:25 +08:00
NCBM 5f7a70bf84 core/prompting: let a prompt say what it asks and when silence answers
`PromptWait` only had a short title and the reason line behind
`include_reason`, so a host that keeps the reason out had a bare title to
show — and a question, whose text *is* what is asked, was cut to the title's
few words. It now carries `body` (the question itself), `facts` (the object:
a program or a file name, never an argument) and `expires_in` (when the
default answer takes over). `ask` and `choose` put their question in the body
and a choice names its options; a secret says in its docstring why its prompt
is announced but never answered.
2026-10-24 04:21:14 +08:00
13 changed files with 616 additions and 115 deletions
+1 -1
View File
@@ -117,7 +117,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.
Shared interactive I/O: `ask` / `choose` / `form` / `confirm` with pluggable backend; non-TTY uses defaults or errors. CLI limit/confirm hooks and chat tools both use this. Async helpers serialize prompts (`run_prompt_async`).
Three process-wide hooks let a host take over a prompt: `set_prompt_output_hook` (break an open pretty line before the prompt draws), `configure_prompting(pause_factory=…)` (the SIGINT guard around an off-loop read), and `set_prompt_wait_hook` — a prompt that blocks announces itself through `prompt_wait(kind, title, detail, choices)`, which yields the `PromptAnswerSource` the read may consult (`take()` returns `allow` / `deny` / reply text, or any key the prompt named in `choices`, without blocking) or `None` for notify-only. Only the outermost prompt of a nested read is announced; with no hook installed nothing changes. A source that *will* offer an answer says so through `supports_actions` before it offers it, so the read polls from its first moment instead of deciding once. `read_line_with_timeout` accepts a `source` and a `None` deadline, and stdin always wins over a host answer. The CLI's implementation is `cli/notify.py` (`[notify]`, desktop via the optional `desktop-notifier` extra, terminal via OSC 777/9 to `/dev/tty`): a prompt is announced only after `grace_seconds` — except one that follows an answer given on the notification, which goes out at once and past the burst throttle — `choices` become the notification's buttons (plus "Go to terminal"), the answer is treated like a typed one (never a session grant), the notification is sent without expiry (`timeout=0`, urgency at the library default) so it stays on screen until it is dismissed, and every failure falls through to the next channel instead of raising.
Three process-wide hooks let a host take over a prompt: `set_prompt_output_hook` (break an open pretty line before the prompt draws), `configure_prompting(pause_factory=…)` (the SIGINT guard around an off-loop read), and `set_prompt_wait_hook` — a prompt that blocks announces itself through `prompt_wait(kind, title, detail, choices, *, body, facts, expires_in)`: the short `title`, the `body` it asks (a question, whole), the gated `detail` (a reason that can name a command or a URL), `facts` (the object — a program or file name, never an argument), and the `expires_in` after which its default answer takes over. It yields the `PromptAnswerSource` the read may consult (`take()` returns `allow` / `deny` / reply text, or any key the prompt named in `choices`, without blocking) or `None` for notify-only. Only the outermost prompt of a nested read is announced; with no hook installed nothing changes. A source that *will* offer an answer says so through `supports_actions` before it offers it, so the read polls from its first moment instead of deciding once. `read_line_with_timeout` accepts a `source` and a `None` deadline, and stdin always wins over a host answer. The CLI's implementation is `cli/notify.py` (`[notify]`, desktop via the optional `desktop-notifier` extra, terminal via OSC 777/9 to `/dev/tty`): a prompt is announced only after `grace_seconds` — except one that follows an answer given on the notification, which goes out at once and past the burst throttle — `choices` become the notification's buttons, the answer is treated like a typed one (never a session grant), the notification is sent without expiry (`timeout=0`, urgency at the library default) so it stays on screen until it is dismissed — a prompt with a deadline gives it that deadline instead — and every failure falls through to the next channel instead of raising. A notification that went to the desktop rings the terminal's bell: a popup may be on another virtual desktop, and no desktop lets a program raise another window. The body is filled in reading order (question, reason, object) until the budget is gone, with the deadline and the workspace — the directory the chat runs in, `~` for home — kept for last, so a long reason drops out before they do; the desktop keeps the lines apart and the terminal flattens them. A popup is sent with no expiry, except for a prompt that gives up on its own: its deadline becomes the popup's timeout. There is no "go to terminal" button — no desktop lets a program raise another window, and the popup's own close does as much; instead every notification that went to the desktop also rings the terminal's bell, and a click on the body (the platform's default action) or a close just withdraws it, leaving the prompt waiting. A `secret` prompt is never announced.
### CLI (`cli/`)
+2 -2
View File
@@ -178,7 +178,7 @@ max_context_tokens = 200000
# [notify]
# grace_seconds = 5.0 # announce only after this long
# channels = ["auto"] # auto | desktop | terminal
# actions = true # allow / deny / review buttons + reply field
# actions = true # allow / deny buttons + a reply field
# include_reason = false # off: the reason may name a command or URL
# Deletes: `delete_path` moves the path to the system trash. See doc/trash.md.
@@ -206,7 +206,7 @@ Third-party **plugins**: install a package that declares `project.entry-points."
**Skills** (instruction directories with a `SKILL.md`): discovered under `[skills].paths`, our own user skills directory, and other harnesses' skill directories (`~/.claude/skills`, plus `<workspace>/.claude/skills` when `include_project_roots`). A session-start notice (a host notice, never a system-prompt edit) tells the model that the skill tools exist and how to use them — `skill_list` to see what exists, `skill_read` for a `SKILL.md`, `skill_search` for content — and deliberately lists nothing; `skill_create` / `skill_edit` write one behind a timed `once`/`session`/`no` grant (`[skills].allow_write = true` for a standing one; the path denylist still wins and `--yes`/`/yolo` never skips it). Reading needs no grant: skill roots are read-only roots, so `read_file` / `regex_files` see them too while writes stay denied. Details: [doc/skills.md](doc/skills.md).
**Notifications while a prompt waits**: a turn stops at a confirm, a directory-access request or an `ask_user_*` question, so if you walked away you only find out on your return. After `[notify].grace_seconds` (5s) the prompt is announced — with the answers that prompt accepts as buttons where the platform renders them (a confirmation: **Allow once / Deny / Go to terminal**; a limit prompt: **Continue / Continue, stop asking / Stop here**; a skill write: **Allow once / This chat / Deny**) and a reply field for a question — and the answer is taken as if typed at the terminal (never as a session grant). The desktop channel needs the optional extra (`pip install 'plyngent[notify]'`); without it the CLI writes an OSC 777 / OSC 9 escape sequence to `/dev/tty`, which every modern terminal turns into a native notification with no dependency at all. Answering at the terminal inside the grace period stays silent, and a prompt that follows an answer given on the notification is announced at once; `[notify].include_reason` keeps the reason line (a command, argv or URL) out by default; the notification is kept on screen (no expiry) until the prompt is answered instead of collapsing into the tray; and no channel can break a prompt: a missing extra or a refusing desktop falls through to the next one. Details: [doc/notifications.md](doc/notifications.md).
**Notifications while a prompt waits**: a turn stops at a confirm, a directory-access request or an `ask_user_*` question, so if you walked away you only find out on your return. After `[notify].grace_seconds` (5s) the prompt is announced — with the answers that prompt accepts as buttons where the platform renders them (a confirmation: **Allow once / Deny**; a limit prompt: **Continue / Continue, stop asking / Stop here**; a skill write: **Allow once / This chat / Deny**) and a reply field for a question — and the answer is taken as if typed at the terminal (never as a session grant). The announcement says what is asked: a question travels whole in the body, a confirmation names the program or the file it is about (`program: rm`, `target: todo.md` — never the arguments), a prompt that denies itself on timeout says so (`no answer in 30s = deny`) and hands that deadline to the popup, which then disappears with it, and the last line is the workspace the chat runs in (`~/Projects/plyngent`). It also rings the terminal's bell, since a popup can sit on another virtual desktop and no desktop lets a program raise another window; closing the popup or clicking its body withdraws it and leaves the prompt waiting. A secret is never announced. The desktop channel needs the optional extra (`pip install 'plyngent[notify]'`); without it the CLI writes an OSC 777 / OSC 9 escape sequence to `/dev/tty`, which every modern terminal turns into a native notification with no dependency at all. Answering at the terminal inside the grace period stays silent, and a prompt that follows an answer given on the notification is announced at once; `[notify].include_reason` keeps the reason line (a command, argv or URL) out by default; the notification is kept on screen (no expiry) until the prompt is answered instead of collapsing into the tray; and no channel can break a prompt: a missing extra or a refusing desktop falls through to the next one. Details: [doc/notifications.md](doc/notifications.md).
**Deletes go to the trash**: `delete_path` moves the path into the desktop's own trash — the freedesktop trash on Linux, the Finder trash on macOS, the Recycle Bin on Windows — so a delete the agent made on its own can be undone from the file manager you already use. That matters because the tool is YOLO-tagged: under `--yes` / `/yolo on` it runs without a confirmation. Ask for the old behaviour with `permanent=true`, or turn it off for every call with `[trash] enabled = false`. Paths under the system temp directory are scratch and are deleted outright, a trashing that fails (a volume with no usable trash directory, a broken install) leaves the file alone and says so instead of deleting it, and a shell `rm` never uses the trash. Details: [doc/trash.md](doc/trash.md).
+53 -23
View File
@@ -5,12 +5,25 @@ window, the wait is invisible until you come back — so the CLI announces it.
A prompt is announced only after `grace_seconds` of waiting (5s by default), so
answering at the terminal stays silent. Bursts are throttled by
`min_interval_seconds`, and the announcement says what is being asked, not what
it would run: the one-line reason can name a command, argv or URL, and a
notification outlives the prompt (and may be mirrored to a lock screen), so it
travels only with `include_reason = true`. The title is the prompt line and the
body is nothing but that reason — a channel gives the title a field of its own,
so repeating it in the body would only say the same thing twice.
`min_interval_seconds`.
A notification says what is asked, not what would be run. The title is a short
label; the body carries, in this order:
| Line | When |
|---|---|
| the question itself | an `ask_user_line` question, an `ask_user_choice` prompt |
| the reason line | only with `include_reason = true`: it can name a command, argv or URL, and a notification outlives the prompt (and may be mirrored to a lock screen) |
| the object | a confirmation or a batch: the program's name (`program: rm`), the file's (`target: todo.md`), the tools of a batch, `kind: directory` for an access request, the skill directory — never an argument, a URL or a piece of code |
| `no answer in 30s = deny` | a prompt that takes a default answer on timeout (command denylist, directory access, skill write) |
| the workspace | always: the directory the chat runs in, with the home directory shown as `~` (`~/Projects/plyngent`), so several waiting chats are told apart — it follows a session resumed in another workspace |
The lines are filled in that order until the body is full and each one is
capped, so a long reason drops out before the deadline and the workspace do.
The deadline is the one line a human must not miss, because not answering *is*
an answer there; the workspace comes last and is kept for the same reason. A
desktop popup wraps and keeps the lines apart; the terminal channel carries the
same body on a single line.
## Configuration
@@ -34,25 +47,39 @@ a prompt answered inside the grace period is never announced at all.
| Prompt | Notification offers |
|---|---|
| tool confirm (single) | Allow once / Deny / Go to terminal |
| a round's batch of confirms | Allow all / Review one by one / Deny / Go to terminal |
| command denylist, `fetch_local` private host | Allow once / Deny / Go to terminal |
| directory-access request | same three (the level asked for, not a lower one) |
| skill write | Allow once / This chat / Deny / Go to terminal |
| limit-continue | Continue / Continue, stop asking / Stop here / Go to terminal |
| tool confirm (single) | Allow once / Deny |
| a round's batch of confirms | Allow all / Review one by one / Deny |
| command denylist, `fetch_local` private host | Allow once / Deny |
| directory-access request | same two (the level asked for, not a lower one) |
| skill write | Allow once / This chat / Deny |
| limit-continue | Continue / Continue, stop asking / Stop here |
| `ask_user_line` | a reply field — the answer *is* the text you type |
| `ask_user_choice`, `ask_user_form`, PTY secrets | nothing (terminal only) |
| `ask_user_choice`, `ask_user_form` | nothing (terminal only): the question and its options are in the body |
| PTY secrets | nothing at all: a secret prompt is never announced |
Each prompt says which answers it accepts (`prompt_wait(..., choices=…)`), and
those are the buttons; one that names none gets Allow once / Deny / Go to
terminal. "Go to terminal" is always the last button: it withdraws the
notification and leaves the prompt waiting.
those are the buttons; one that names none gets Allow once / Deny. There is no
"go to terminal" button: it could not do what it says — no desktop lets a
program raise another application's window (Wayland has no such call, the X11
ones need a helper per window manager) — and the popup's own close button does
as much. Clicking the body (the platform's default action) or closing the popup
withdraws the notification and leaves the prompt waiting; the prompt is not
announced a second time.
What says *where* to look is the bell: every notification that went to the
desktop also rings the terminal's bell. The announcement only happens when
nobody answered inside the grace period, so the human is at some other window —
possibly another virtual desktop, where a popup is easy to miss — and the bell
is the only portable cue. It is the terminal's own bell, so a terminal with its
bell switched off stays silent.
An answer read from the notification is treated exactly like a typed one —
`allow` is the `y` the prompt accepts, `deny` is `n` — and it never records a
session grant: the button means "this once", which is why a prompt with an
extra answer (this chat, stop asking) offers it as a button of its own. A secret
is never asked for in a notification, and never echoed into one.
extra answer (this chat, stop asking) offers it as a button of its own. Text
typed into a reply field is the answer, whatever it spells. A secret prompt is
announced nowhere and echoable from nothing: its value would have to travel
back through a surface that outlives it.
A prompt that can be answered from the notification reads both surfaces while it
blocks, from the first moment — the notification itself only shows up after
@@ -88,11 +115,14 @@ The notification stays on screen: it is sent with no expiry (`timeout = 0`, the
freedesktop value for "until dismissed"), so the server does not slide it into
the tray a few seconds later. That is the point of it — the turn is blocked
until the prompt is answered, and a collapsed notification is one the human
would find only on coming back anyway. Urgency stays at the library default: on
KDE Plasma, `critical` keeps the notification up just the same but paints the
popup as urgent, and on every platform it asks the Focus mode to let it through
— which a prompt has no business doing. Once the prompt is answered, the
notification is dismissed.
would find only on coming back anyway. A prompt that gives up on its own is the
exception: the popup is sent with its deadline as its timeout (30 seconds for a
prompt that denies after 30s), so the notification disappears when the answer
window closes instead of announcing a prompt that is already gone. Urgency stays
at the library default: on KDE Plasma, `critical` keeps the notification up just
the same but paints the popup as urgent, and on every platform it asks the Focus
mode to let it through — which a prompt has no business doing. Once the prompt
is answered, the notification is dismissed.
## Terminal channel
+14 -3
View File
@@ -52,10 +52,21 @@
# `pip install plyngent[notify]`) and falls back to a terminal escape sequence;
# "desktop" / "terminal" pin one of them.
# `actions` offers the answers the prompt accepts as notification buttons (allow /
# deny / review when it names none); where the platform has none the notification
# stays text-only.
# deny when it names none); where the platform has none the notification stays
# text-only. Every notification that went to the desktop also rings the terminal's
# bell — the popup may be on another virtual desktop, and no desktop lets a program
# raise another application's window. Closing the popup (or clicking its body)
# withdraws it and leaves the prompt waiting.
# The body says what is asked: a question or a choice travels whole (the title is
# only a short label), a confirmation names the program or file it is about, and a
# timed prompt says when silence answers ("no answer in 30s = deny", which is a
# deny for the denylist, directory-access and skill-write prompts). The last line
# is the workspace the chat runs in (`~` for home), so several waiting chats are
# told apart; a prompt that has a deadline hands it to the popup, which then
# disappears with it.
# `include_reason` adds the one-line reason — it can name a command or a URL, so
# it stays off by default (a notification may be mirrored to a lock screen).
# it stays off by default (a notification may be mirrored to a lock screen). A
# secret is never announced.
# [notify]
# enabled = true
# grace_seconds = 5.0
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "plyngent"
version = "0.13.0"
version = "0.14.0"
description = "LLM chat and agent toolkit"
authors = [
{ name = "HivertMoZara", email = "worldmozara@163.com" },
+16 -4
View File
@@ -129,20 +129,27 @@ def _read_prompt_text(prompt: str | None, *, stdin_isatty: bool) -> str | None:
return text or None
def _setup_hooks(*, interactive: bool, auto_continue: bool, notify: NotifyConfig) -> None:
def _setup_hooks(
*,
interactive: bool,
auto_continue: bool,
notify: NotifyConfig,
workspace: Path | None = None,
) -> None:
"""Install limit hooks (workspace policy is set on ReplState).
``auto_continue`` keeps the PTY/limit hook installed for one-shot runs so
``--auto-continue`` / ``[agent] auto_continue_limits`` can raise limits
without a prompt; otherwise non-interactive runs keep today's deny. The
prompt-wait notifier only makes sense for a human at a terminal, so a
non-interactive run clears it whatever ``[notify]`` says.
non-interactive run clears it whatever ``[notify]`` says. *workspace* is
what a notification names, so two chats are told apart.
"""
from plyngent.cli.notify import install_notifier
from plyngent.prompting import set_prompt_wait_hook
if interactive:
_ = install_notifier(notify)
_ = install_notifier(notify, workspace=workspace)
else:
set_prompt_wait_hook(None)
if interactive or auto_continue:
@@ -258,7 +265,12 @@ async def _run_chat( # noqa: C901, PLR0915 — chat orchestration
_warn_recoverable_providers(store.recoverable_providers)
auto_effective = auto_continue or store.agent_config.auto_continue_limits
_setup_hooks(interactive=interactive, auto_continue=auto_effective, notify=store.notify_config)
_setup_hooks(
interactive=interactive,
auto_continue=auto_effective,
notify=store.notify_config,
workspace=workspace,
)
# --yes forces sticky YOLO; else derive from config.confirm_destructive.
from plyngent.cli.state import YoloMode
+54 -12
View File
@@ -6,6 +6,7 @@ import shutil
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Literal
from plyngent.agent import BatchConfirmDecision, ToolConfirmRequest
@@ -32,7 +33,6 @@ from plyngent.tools.workspace import AccessMode, parse_access_mode
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from pathlib import Path
type WorkspaceMismatchChoice = Literal["keep", "rebind", "abort"]
@@ -269,11 +269,34 @@ def _confirm_with(wait: PromptAnswerSource | None, prompt: str, *, default: bool
return confirm(prompt, default=default, source=_answer_source(wait))
def _tool_facts(args: Mapping[str, object]) -> tuple[str, ...]:
"""What a call is about, short enough to travel in a notification.
Only the *object* — the program or the file — never an argument, a URL or a
piece of code: those stay in the reason, which a notification carries only
when ``include_reason`` allows it.
"""
facts: list[str] = []
argv = args.get("argv")
if isinstance(argv, list) and argv and isinstance(argv[0], str):
facts.append(f"program: {Path(argv[0]).name}")
for key in ("path", "dst", "src"):
value = args.get(key)
if isinstance(value, str) and value:
facts.append(f"target: {Path(value).name}")
break
return tuple(facts)
def _prompt_confirm_tool_sync(name: str, args: Mapping[str, object], reason: str) -> bool | str:
"""True allow; False deny; non-empty str = deny with comment for the model."""
del args
try:
with prompt_wait("confirm", f"tool {name!r} needs confirmation", reason) as wait:
with prompt_wait(
"confirm",
f"tool {name!r} needs confirmation",
reason,
facts=_tool_facts(args),
) as wait:
_echo_tool_confirm(name, reason)
allowed = _confirm_with(wait, "Allow this tool call?", default=False)
except NonInteractiveError:
@@ -387,7 +410,7 @@ def prompt_policy_command_confirm(
try:
with (
pause_task_cancel_for_prompt(),
prompt_wait("policy", f"allow command {basename!r}?", reason) as wait,
prompt_wait("policy", f"allow command {basename!r}?", reason, expires_in=timeout_seconds) as wait,
):
backend = get_prompt_backend()
if not backend.is_interactive():
@@ -434,7 +457,7 @@ def prompt_policy_fetch_confirm(
try:
with (
pause_task_cancel_for_prompt(),
prompt_wait("policy", f"allow fetch_local {host!r}?", reason) as wait,
prompt_wait("policy", f"allow fetch_local {host!r}?", reason, expires_in=timeout_seconds) as wait,
):
backend = get_prompt_backend()
if not backend.is_interactive():
@@ -483,7 +506,14 @@ def prompt_skill_write_confirm(
try:
with (
pause_task_cancel_for_prompt(),
prompt_wait("skills", f"allow {action} in {skill_dir.name!r}?", detail, choices=_SKILL_CHOICES) as wait,
prompt_wait(
"skills",
f"allow {action} in {skill_dir.name!r}?",
detail,
choices=_SKILL_CHOICES,
facts=(f"path: {skill_dir}",),
expires_in=timeout_seconds,
) as wait,
):
backend = get_prompt_backend()
if not backend.is_interactive():
@@ -524,9 +554,10 @@ def prompt_directory_access_confirm(
that level instead. The host's hook decides YOLO/``--yes`` before calling
this. Timeout, cancel, or non-interactive → ``None`` (deny).
"""
kind = "directory" if path.is_dir() else "file"
lines = [
f"request: {mode.name.lower()} access to {path}",
f"kind: {'directory' if path.is_dir() else 'file'}",
f"kind: {kind}",
]
if reason:
lines.append(f"why: {reason}")
@@ -534,7 +565,13 @@ def prompt_directory_access_confirm(
try:
with (
pause_task_cancel_for_prompt(),
prompt_wait("access", f"allow {mode.name.lower()} access to {path}?", reason) as wait,
prompt_wait(
"access",
f"allow {mode.name.lower()} access to {path}?",
reason,
facts=(f"kind: {kind}",),
expires_in=timeout_seconds,
) as wait,
):
backend = get_prompt_backend()
if not backend.is_interactive():
@@ -573,11 +610,11 @@ def prompt_directory_access_confirm(
async def prompt_confirm_tool_async(name: str, args: Mapping[str, object], reason: str) -> bool | str:
"""Async confirm: True allow, False deny, str = deny with user comment."""
del args
facts = _tool_facts(args)
try:
# Box is printed inside the worker thread via confirm's backend.
def _run() -> bool:
with prompt_wait("confirm", f"tool {name!r} needs confirmation", reason) as wait:
with prompt_wait("confirm", f"tool {name!r} needs confirmation", reason, facts=facts) as wait:
_echo_tool_confirm(name, reason)
return _confirm_with(wait, "Allow this tool call?", default=False)
@@ -620,8 +657,13 @@ def _prompt_confirm_batch_sync(requests: Sequence[ToolConfirmRequest]) -> BatchC
if not backend.is_interactive():
# 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, choices=_BATCH_CHOICES) as wait:
tools = tuple(dict.fromkeys(request.name for request in requests))
with prompt_wait(
"batch",
f"{len(requests)} tool calls need confirmation",
choices=_BATCH_CHOICES,
facts=tools,
) 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")
+175 -37
View File
@@ -4,16 +4,21 @@ 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: the prompt's own answers as
buttons (allow / deny / review for a confirmation), a reply field for a
question. A prompt that can be answered from the notification reads from both
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.
buttons (allow / deny for a confirmation), a reply field for a question. It says
what is asked: a question travels in the body where it is read whole, the object
of a confirmation (a program or a file name) rides along with it, a prompt that
takes a default answer says when not answering gives it, and the workspace it
belongs to comes last. A prompt that can be answered from the notification reads
from both 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
sequence to the controlling terminal, which every modern terminal turns into a
native notification without any dependency at all.
native notification without any dependency at all. A notification that went to
the desktop also rings the terminal's bell: the popup may be on another virtual
desktop, and no desktop lets a program raise another application's window.
"""
from __future__ import annotations
@@ -21,6 +26,7 @@ from __future__ import annotations
import asyncio
import contextlib
import importlib.util
import math
import os
import sys
import threading
@@ -40,7 +46,6 @@ if TYPE_CHECKING:
ALLOW = "allow"
DENY = "deny"
REVIEW = "review"
#: Kinds whose notification carries an answer, and the capability it needs: a
#: confirmation gets buttons, a question gets a reply field.
@@ -54,14 +59,21 @@ _ANSWER_CAPABILITY: dict[str, str] = {
"ask": "reply",
}
#: The answers a confirmation offers when the prompt names none of its own.
_BUTTONS: tuple[tuple[str, str], ...] = (
(ALLOW, "Allow once"),
(DENY, "Deny"),
(REVIEW, "Go to terminal"),
)
_MAX_TITLE = 80
_MAX_MESSAGE = 200
#: One body line is capped so a single long line cannot starve the others.
_MAX_BODY_LINE = 100
#: Prompt families a notification never announces: the value being asked for
#: would have to travel back through a surface that outlives the prompt.
_QUIET_KINDS = frozenset({"secret"})
#: What not answering means, for a prompt that has a timed default.
_DEADLINE = "no answer in {seconds:g}s = deny"
#: Seconds a desktop notification may stay up; ``0`` is the freedesktop value
#: for "until dismissed" (``-1`` would hand the notification server's default
@@ -79,6 +91,70 @@ def _clean(text: str, limit: int) -> str:
return stripped if len(stripped) <= limit else stripped[: limit - 1] + "…"
def _deadline(wait: PromptWait) -> str:
"""What not answering means, when the prompt takes a default answer."""
if wait.expires_in is None:
return ""
return _clean(_DEADLINE.format(seconds=wait.expires_in), _MAX_BODY_LINE)
def _expiry(wait: PromptWait) -> int | None:
"""How long the popup may stay up: as long as the prompt waits for an answer.
``None`` keeps it until it is dismissed — a prompt with no deadline has no
moment at which the notification is stale.
"""
if wait.expires_in is None or wait.expires_in <= 0:
return None
return max(1, math.ceil(wait.expires_in))
def _shorten_home(path: Path) -> str:
"""The path as the human types it: ``~`` for the home directory."""
home = Path.home()
if path == home:
return "~"
with contextlib.suppress(ValueError):
return f"~/{path.relative_to(home).as_posix()}"
return str(path)
def _body_parts(wait: PromptWait, *, include_reason: bool) -> list[str]:
"""The lines the prompt asks for, in reading order.
The question first, then the reason when the config allows it, then what the
prompt is about: a host that hides the reason still says what is being asked
and what it concerns.
"""
parts: list[str] = []
if wait.body:
parts.append(wait.body)
if include_reason and wait.detail:
parts.append(wait.detail)
if wait.facts:
parts.append(" · ".join(wait.facts))
return [_clean(part, _MAX_BODY_LINE) for part in parts]
def _body(wait: PromptWait, *, include_reason: bool, multiline: bool, source: str = "") -> str:
"""The notification body, with the deadline and the source kept for last.
The lines are filled in reading order until the budget is gone — a long
reason drops out before the deadline does — and each one is capped, so no
single line can starve the rest. A desktop popup wraps, so its body keeps
the lines apart; a terminal escape sequence carries the body on one line.
"""
tail = [line for line in (_deadline(wait), _clean(source, _MAX_BODY_LINE)) if line]
budget = _MAX_MESSAGE - sum(len(line) + 1 for line in tail)
kept: list[str] = []
for line in _body_parts(wait, include_reason=include_reason):
if not line or len(line) + 1 > budget:
continue
kept.append(line)
budget -= len(line) + 1
return "\n".join([*kept, *tail]) if multiline else " ".join([*kept, *tail])
# --------------------------------------------------------------------------- #
# Desktop channel (optional `desktop-notifier` extra)
# --------------------------------------------------------------------------- #
@@ -94,6 +170,9 @@ class DesktopRequest:
reply_field: bool = False
on_answer: Callable[[str], None] | None = None
on_reply: Callable[[str], None] | None = None
on_click: Callable[[], None] | None = None
#: Seconds the popup may stay up; ``None`` keeps it until it is dismissed.
timeout: int | None = None
@runtime_checkable
@@ -184,11 +263,14 @@ class DesktopNotifierChannel:
message=request.message,
# A prompt holds the turn until it is answered, so the
# notification must still be on screen when the human looks
# back. Dismissing is the prompt's own job once it is
# answered, so nothing lingers.
timeout=_NO_EXPIRY,
# back: it stays up until it is dismissed — or, for a prompt
# that gives up on its own, until that deadline.
timeout=_NO_EXPIRY if request.timeout is None else request.timeout,
buttons=buttons,
reply_field=reply,
# Clicking the body is the default action: it takes the
# human back to the terminal instead of answering.
on_clicked=request.on_click,
)
)
)
@@ -367,18 +449,31 @@ class WaitHandle:
return None
def answer(self, key: str) -> None:
"""Record one answer from the notification (a button, or a reply)."""
"""Record one answer pressed on the notification (a button)."""
if self._closed:
return
if key == REVIEW and self.wait.kind != "ask":
# "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 reply(self, text: str) -> None:
"""Record one answer typed into the notification's reply field."""
if self._closed:
return
self.notifier.handoff()
self._answers.put(text)
def withdraw(self) -> None:
"""Take the notification down; the prompt keeps waiting at the terminal.
This is what a click on the body does, and the popup's own close button
means the same. Nothing can raise another application's window (no
desktop exposes that portably), which is why the bell — rung as the
notification went out — is what said where the prompt is.
"""
self.notifier.dismiss(self)
def __enter__(self) -> Self:
delay = 0.0 if self.handoff else self.notifier.grace_seconds
timer = threading.Timer(delay, self._notify)
@@ -402,8 +497,13 @@ class WaitHandle:
def _notify(self) -> None:
if self._closed or not self.notifier.may_notify(self):
return
if not self.notifier.notify_desktop(self):
self.notifier.notify_terminal(self.wait)
if self.notifier.notify_desktop(self):
# The popup went to the desktop, so the human is looking at some
# other window — possibly another virtual desktop, where a popup is
# easy to miss. The terminal rings to say which window waits.
self.notifier.ring()
return
self.notifier.notify_terminal(self.wait)
class WaitNotifier:
@@ -422,6 +522,7 @@ class WaitNotifier:
_clock: Callable[[], float]
_last_sent: float
_handoff: bool
_source: str
def __init__(
self,
@@ -431,6 +532,7 @@ class WaitNotifier:
write: Callable[[str], None] | None = None,
env: Mapping[str, str] | None = None,
clock: Callable[[], float] = time.monotonic,
workspace: Path | str | None = None,
) -> None:
self._config = config
if isinstance(desktop, _Resolve):
@@ -445,6 +547,9 @@ class WaitNotifier:
self._clock = clock
self._last_sent = float("-inf")
self._handoff = False
self._source = ""
if workspace is not None:
self.set_workspace(workspace)
# -- hook ------------------------------------------------------------- #
@@ -455,6 +560,19 @@ class WaitNotifier:
def grace_seconds(self) -> float:
return self._config.grace_seconds
def set_workspace(self, workspace: Path | str | None) -> None:
"""Name the directory the prompt belongs to (``~`` for a home path).
A chat can be rebound to another workspace, so the label is kept on the
notifier rather than baked into a notification: several chats look
alike, and which one is waiting is the first thing a human needs.
"""
self._source = "" if workspace is None else _shorten_home(Path(workspace))
def ring(self) -> None:
"""Ring the terminal's bell: the prompt is waiting *there*."""
self._write(_BELL)
def handoff(self) -> None:
"""Note that the human is answering on the notification right now."""
self._handoff = True
@@ -473,6 +591,10 @@ class WaitNotifier:
"""Whether this prompt is worth announcing right now."""
if not self._config.enabled or not get_prompt_backend().is_interactive():
return False
# A secret is never announced: its value would have to come back
# through a surface that outlives the prompt.
if handle.wait.kind in _QUIET_KINDS:
return False
now = self._clock()
# 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:
@@ -481,20 +603,30 @@ class WaitNotifier:
return True
def _message(self, wait: PromptWait) -> str:
"""The body: the reason line, when the config allows it, else nothing.
"""The one-line body the terminal channel carries."""
return _body(
wait,
include_reason=self._config.include_reason,
multiline=False,
source=self._source,
)
Both channels show the title as their own field, so repeating it in the
body only makes the notification say the same thing twice.
"""
if self._config.include_reason and wait.detail:
return _clean(wait.detail, _MAX_MESSAGE)
return ""
def _desktop_message(self, wait: PromptWait) -> str:
"""The body the desktop channel carries: the same lines, kept apart."""
return _body(
wait,
include_reason=self._config.include_reason,
multiline=True,
source=self._source,
)
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 _buttons(self, wait: PromptWait, capability: str) -> tuple[tuple[str, str], ...]:
"""The answers *wait* accepts; a reply prompt has none to offer."""
if capability == "reply":
# The reply field carries the answer; a button beside it would only
# be a way out, and the popup's own close is that.
return ()
return wait.choices or _BUTTONS
def desktop_channel(self) -> DesktopChannel | None:
"""The desktop surface, resolved once (importing the extra on demand)."""
@@ -540,16 +672,18 @@ class WaitNotifier:
if channel is None:
return False
wait = handle.wait
capability = self.answer_capability(wait.kind)
capability = self.answer_capability(wait.kind) or ""
try:
handle.identifier = channel.send(
DesktopRequest(
title=_clean(wait.title, _MAX_TITLE),
message=self._message(wait),
buttons=self._buttons(wait) if capability == "buttons" else (),
message=self._desktop_message(wait),
buttons=self._buttons(wait, capability) if capability else (),
reply_field=capability == "reply",
on_answer=handle.answer,
on_reply=handle.answer,
on_reply=handle.reply,
on_click=handle.withdraw,
timeout=_expiry(wait),
)
)
except Exception: # noqa: BLE001 - a notification never breaks a prompt
@@ -579,12 +713,16 @@ class WaitNotifier:
channel.clear(identifier)
def install_notifier(config: NotifyConfig) -> WaitNotifier | None:
"""Install the prompt-wait hook from ``[notify]`` (None clears it)."""
def install_notifier(config: NotifyConfig, *, workspace: Path | str | None = None) -> WaitNotifier | None:
"""Install the prompt-wait hook from ``[notify]`` (None clears it).
*workspace* is the directory the chat runs in: a notification names it, so
two waiting chats are told apart.
"""
if not config.enabled:
set_prompt_wait_hook(None)
return None
notifier = WaitNotifier(config)
notifier = WaitNotifier(config, workspace=workspace)
notifier.warm()
set_prompt_wait_hook(notifier)
return notifier
+14
View File
@@ -108,6 +108,7 @@ class ReplState:
self.workspace = Path(self.workspace).expanduser().resolve()
self.instance_state.workspace_root = self.workspace
self.instance_state.workspace.root = self.workspace
self._retarget_notifier(self.workspace)
self._install_config_access()
self._install_skills()
from plyngent.cli.limits import set_auto_continue_default
@@ -711,11 +712,24 @@ class ReplState:
self.workspace = resolved
self.instance_state.workspace_root = resolved
self.instance_state.workspace.root = resolved
# A prompt announced from here names this directory, not the one the
# process started in.
self._retarget_notifier(resolved)
# Project-scope skill roots follow the workspace.
self._install_skills()
if hasattr(self, "agent") and self.agent.tools is not None:
self.agent.tools.set_instance_state(self.instance_state)
@staticmethod
def _retarget_notifier(workspace: Path) -> None:
"""Point an installed prompt notifier at *workspace* (if there is one)."""
from plyngent.cli.notify import WaitNotifier
from plyngent.prompting import get_prompt_wait_hook
hook = get_prompt_wait_hook()
if isinstance(hook, WaitNotifier):
hook.set_workspace(workspace)
def _apply_session_workspace(self, row: SessionRow) -> None:
"""Bind tools/REPL workspace to the session's directory when set."""
if not row.workspace:
+41 -10
View File
@@ -125,17 +125,27 @@ class PromptWait:
``kind`` names the prompt family (``confirm``, ``ask``, ``secret``,
``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. ``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.
safe to surface anywhere. ``body`` is the prompt itself — a question, which
the host shows as it stands — while ``detail`` carries one optional reason
line the caller has already redacted: it can name a command or a URL, so a
host shows it only when the human allowed that. ``facts`` are short items
naming the *object* of the prompt (a program or a file name, never an
argument): they travel unconditionally, so a host that hides the reason
still says what the prompt is about. ``expires_in`` is how long the prompt
waits before taking its default answer, which is usually "deny" — a host
says so, because not answering is then an answer. ``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], ...] = ()
body: str = ""
facts: tuple[str, ...] = ()
expires_in: float | None = None
@runtime_checkable
@@ -185,11 +195,17 @@ def prompt_wait(
title: str,
detail: str = "",
choices: tuple[tuple[str, str], ...] = (),
*,
body: str = "",
facts: tuple[str, ...] = (),
expires_in: float | None = None,
) -> 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.
``choices`` says which answers this prompt accepts, ``body`` what it asks,
``facts`` what it is about and ``expires_in`` when its default answer takes
over (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
@@ -201,7 +217,8 @@ def prompt_wait(
return
token = _wait_depth.set(_wait_depth.get() + 1)
try:
with hook(PromptWait(kind, title, detail, choices)) as source:
wait = PromptWait(kind, title, detail, choices, body, facts, expires_in)
with hook(wait) as source:
yield source
finally:
_wait_depth.reset(token)
@@ -554,6 +571,15 @@ def _read_with_source(
return answer.strip()
#: Titles for the prompts whose own text *is* the question: a host gives the
#: title a line of its own and cuts it short, so the question itself goes into
#: the body, where it is read whole.
_ASK_TITLE = "a question is waiting"
_CHOOSE_TITLE = "a choice is waiting"
#: How many option labels a choice prompt names alongside its question.
_MAX_CHOICE_FACTS = 4
def ask(
prompt: str,
*,
@@ -568,7 +594,7 @@ 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) as wait:
with prompt_wait("ask", _ASK_TITLE, body=prompt) as wait:
backend.secho(prompt, fg="yellow")
source = wait if wait is not None and wait.supports_actions else None
return _read_with_source("Answer", source, default=default, completions=completions)
@@ -577,6 +603,10 @@ def ask(
def ask_secret(prompt: str) -> str:
"""Prompt for a secret line (no echo). Never use for values that return to the model.
The ``secret`` family is announced, never answered: the prompt is on screen
at the terminal only, since a value typed into a notification would have to
travel to get back to the process.
Cancel (Ctrl+C / EOF) raises :class:`NonInteractiveError`.
"""
backend = get_prompt_backend()
@@ -616,8 +646,9 @@ def choose(
msg = f"non-interactive: cannot prompt for {prompt!r}"
raise NonInteractiveError(msg)
facts = tuple(dict.fromkeys(option.label for option in choices[:_MAX_CHOICE_FACTS]))
while True:
with prompt_wait("choose", prompt):
with prompt_wait("choose", _CHOOSE_TITLE, body=prompt, facts=facts):
raw = backend.read_line(
"Choice",
default=default_display,
+47
View File
@@ -19,9 +19,11 @@ from plyngent.cli.limits import (
)
from plyngent.prompting import (
NonInteractiveBackend,
PromptWait,
get_prompt_backend,
get_prompt_output_hook,
get_prompt_wait_hook,
set_prompt_wait_hook,
temporary_backend,
)
from plyngent.tools.process.pty_session import PtyManager
@@ -406,6 +408,51 @@ def test_a_skill_write_can_be_allowed_for_the_chat_from_the_notification() -> No
assert decision.session is True
def test_a_tool_confirm_names_the_object_it_is_about() -> None:
"""A notification body carries the program or file, never the argument."""
seen: list[PromptWait] = []
@contextlib.contextmanager
def hook(wait: PromptWait) -> Generator[None]:
seen.append(wait)
yield None
previous = get_prompt_wait_hook()
set_prompt_wait_hook(hook)
try:
with temporary_backend(ScriptedBackend([], confirms=[True, True])):
_ = prompt_confirm_tool("run_argv", {"argv": ["/usr/bin/rm", "-rf", "build"]}, "command 'rm'")
_ = prompt_confirm_tool("write_file", {"path": "notes/todo.md", "content": "x"}, "overwrite")
finally:
set_prompt_wait_hook(previous)
assert [wait.facts for wait in seen] == [("program: rm",), ("target: todo.md",)]
def test_a_timed_prompt_says_when_silence_answers() -> None:
"""Timeout is a deny, so the notification must not hide the clock."""
from pathlib import Path
from plyngent.cli import limits
from plyngent.tools.workspace import AccessMode
seen: list[PromptWait] = []
@contextlib.contextmanager
def hook(wait: PromptWait) -> Generator[None]:
seen.append(wait)
yield None
previous = get_prompt_wait_hook()
set_prompt_wait_hook(hook)
try:
with temporary_backend(ScriptedBackend(["y"], confirms=[True])):
_ = limits.prompt_directory_access_confirm(Path("/tmp"), AccessMode.READ, "why", 30.0)
finally:
set_prompt_wait_hook(previous)
assert seen[0].expires_in == 30.0
assert seen[0].facts == ("kind: directory",)
def test_every_confirm_announces_itself() -> None:
"""Each blocking confirm reports its kind, so a host can pick a surface."""
from pathlib import Path
+153 -21
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import sys
import time
import types
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar
import pytest
@@ -10,7 +11,6 @@ import pytest
from plyngent.cli.notify import (
ALLOW,
DENY,
REVIEW,
DesktopNotifierChannel,
DesktopRequest,
WaitNotifier,
@@ -98,6 +98,7 @@ def _notifier(
desktop: _FakeDesktop | None = None,
grace_seconds: float = 60.0,
min_interval_seconds: float = 0.0,
workspace: str | None = None,
**kwargs: object,
) -> tuple[WaitNotifier, list[str]]:
"""A notifier with a captured terminal writer and a fake desktop channel."""
@@ -112,6 +113,7 @@ def _notifier(
desktop=desktop,
write=written.append,
env={"TERM_PROGRAM": "WezTerm"},
workspace=workspace,
)
return notifier, written
@@ -214,7 +216,9 @@ def test_desktop_channel_can_be_pinned() -> None:
with notifier(PromptWait("confirm", "Allow?")) as handle:
handle._notify()
assert len(desktop.requests) == 1
assert written == []
# No terminal notification (it was pinned), but the bell still says which
# window holds the prompt.
assert written == ["\a"]
def test_terminal_fallback_when_the_desktop_refuses() -> None:
@@ -255,14 +259,14 @@ def test_disabled_notifications_send_nothing() -> None:
# --------------------------------------------------------------------------- #
def test_a_confirmation_offers_three_buttons() -> None:
def test_a_confirmation_offers_allow_and_deny() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
with notifier(PromptWait("confirm", "Allow?")) as handle:
handle._notify()
assert handle.supports_actions is True
request = desktop.requests[0]
assert [key for key, _ in request.buttons] == [ALLOW, DENY, REVIEW]
assert [key for key, _ in request.buttons] == [ALLOW, DENY]
assert request.reply_field is False
@@ -294,16 +298,136 @@ def test_no_answer_is_expected_without_the_capability() -> None:
assert handle.supports_actions is False
def test_an_ask_offers_a_reply_field_instead_of_buttons() -> None:
def test_a_question_offers_a_reply_field_without_the_answers() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
with notifier(PromptWait("ask", "Which branch?")) as handle:
with notifier(PromptWait("ask", "a question is waiting", body="Which branch?")) as handle:
handle._notify()
request = desktop.requests[0]
assert request.reply_field is True
assert request.buttons == ()
def test_a_question_carries_the_question_in_its_body() -> None:
"""A question *is* what is asked, so the title cannot be the whole story."""
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
question = "Which branch should the release be cut from?"
with notifier(PromptWait("ask", "a question is waiting", body=question)) as handle:
handle._notify()
assert desktop.requests[0].message == question
def test_the_body_names_the_object_and_when_not_answering_answers() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
wait = PromptWait("policy", "allow command 'rm'?", facts=("program: rm",), expires_in=30.0)
with notifier(wait) as handle:
handle._notify()
# A popup wraps, so the desktop body keeps its lines apart.
assert desktop.requests[0].message.split("\n") == ["program: rm", "no answer in 30s = deny"]
def test_the_terminal_body_stays_on_one_line() -> None:
notifier, _ = _notifier()
wait = PromptWait("confirm", "Allow?", facts=("program: rm",), expires_in=30.0)
assert notifier._message(wait) == "program: rm no answer in 30s = deny"
def test_a_long_reason_goes_before_the_deadline_does() -> None:
"""The deadline is short and is the one line a human must not miss."""
notifier, _ = _notifier(include_reason=True)
wait = PromptWait("confirm", "Allow?", "x" * 400, facts=("program: rm",), expires_in=30.0)
message = notifier._message(wait)
assert message.endswith("no answer in 30s = deny")
assert len(message) <= 200
def test_a_click_on_the_notification_leaves_the_prompt_waiting() -> None:
"""The body click is the default action: the notification goes, the wait stays."""
desktop = _FakeDesktop()
notifier, written = _notifier(desktop=desktop)
with notifier(PromptWait("confirm", "Allow?")) as handle:
handle._notify()
request = desktop.requests[0]
assert request.on_click is not None
request.on_click()
assert handle.take() is None
assert desktop.cleared == ["id-1"]
# One bell, rung when the notification went out — not again for the click.
assert written == ["\a"]
def test_the_bell_rings_when_the_popup_goes_out() -> None:
"""A popup can sit on another virtual desktop; the bell says where to look."""
desktop = _FakeDesktop()
notifier, written = _notifier(desktop=desktop)
with notifier(PromptWait("confirm", "Allow?")) as handle:
handle._notify()
assert written == ["\a"]
assert desktop.requests[0].buttons == ((ALLOW, "Allow once"), (DENY, "Deny"))
def test_the_body_names_the_workspace_under_a_tilde(monkeypatch: pytest.MonkeyPatch) -> None:
"""Two chats look alike; which directory waits is what tells them apart."""
monkeypatch.setenv("HOME", "/home/tester")
notifier, _ = _notifier(workspace="/home/tester/work/plyngent")
assert notifier._message(PromptWait("confirm", "Allow?")) == "~/work/plyngent"
notifier.set_workspace("/srv/other")
assert notifier._message(PromptWait("confirm", "Allow?")) == "/srv/other"
def test_a_rebound_workspace_moves_the_label(monkeypatch: pytest.MonkeyPatch) -> None:
"""A resumed session runs in its own directory; the label follows it."""
monkeypatch.setenv("HOME", "/home/tester")
from plyngent.cli.state import ReplState
notifier = install_notifier(NotifyConfig(), workspace="/home/tester/one")
assert notifier is not None
ReplState._retarget_notifier(Path("/home/tester/two"))
assert notifier._message(PromptWait("confirm", "Allow?")) == "~/two"
def test_a_timed_prompt_lets_the_popup_expire_with_it() -> None:
"""The popup must not outlive the answer window it announces."""
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
with notifier(PromptWait("policy", "allow command 'rm'?", expires_in=30.0)) as handle:
handle._notify()
assert desktop.requests[0].timeout == 30
def test_a_prompt_without_a_deadline_keeps_its_popup() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
with notifier(PromptWait("confirm", "Allow?")) as handle:
handle._notify()
assert desktop.requests[0].timeout is None
def test_a_reply_is_an_answer_whatever_it_spells() -> None:
"""A reply field's text is the answer, even when it reads like a button key."""
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
with notifier(PromptWait("ask", "a question is waiting", body="What now?")) as handle:
handle._notify()
request = desktop.requests[0]
assert request.on_reply is not None
request.on_reply("review")
assert handle.take() == "review"
def test_a_secret_prompt_is_never_announced() -> None:
"""The value would have to come back through a surface that outlives it."""
desktop = _FakeDesktop()
notifier, written = _notifier(desktop=desktop, grace_seconds=0.0)
with notifier(PromptWait("secret", "API key?")) as handle:
assert notifier.may_notify(handle) is False
handle._notify()
assert desktop.requests == []
assert written == []
def test_a_limit_prompt_can_be_answered() -> None:
"""A limit prompt blocks the turn too, so it offers buttons like a confirm."""
desktop = _FakeDesktop()
@@ -359,18 +483,6 @@ def test_a_reply_reaches_the_prompt() -> None:
assert handle.take() == "feature/x"
def test_review_keeps_waiting_and_dismisses() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
with notifier(PromptWait("confirm", "Allow?")) as handle:
handle._notify()
request = desktop.requests[0]
assert request.on_answer is not None
request.on_answer(REVIEW)
assert handle.take() is None
assert desktop.cleared == ["id-1"]
def test_an_answer_after_the_prompt_is_ignored() -> None:
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
@@ -469,15 +581,36 @@ def test_a_desktop_notification_is_kept_on_screen() -> None:
assert sent["buttons"] and sent["reply_field"] is not None
@pytest.mark.usefixtures("_stub_extra")
def test_the_desktop_call_carries_the_click_handler() -> None:
"""Clicking the body is the default action, so it must reach the prompt."""
channel = DesktopNotifierChannel()
clicked: list[str] = []
def _click() -> None:
clicked.append("click")
_ = channel.send(DesktopRequest(title="Allow?", message="", on_click=_click))
assert _RecordingNotifier.sent[-1]["on_clicked"] is _click
@pytest.mark.usefixtures("_stub_extra")
def test_the_popup_can_be_told_when_to_expire() -> None:
"""A prompt that gives up on its own takes its popup with it."""
channel = DesktopNotifierChannel()
_ = channel.send(DesktopRequest(title="Allow?", message="", timeout=30))
assert _RecordingNotifier.sent[-1]["timeout"] == 30
def test_a_prompt_offers_its_own_answers() -> None:
"""The prompt's ``choices`` become the buttons, plus a way to the terminal."""
"""The prompt's ``choices`` become the buttons."""
desktop = _FakeDesktop()
notifier, _ = _notifier(desktop=desktop)
choices = (("y", "Continue"), ("yyy", "Continue, stop asking"), ("n", "Stop here"))
with notifier(PromptWait("limit", "limit reached — continue?", "why", choices)) as handle:
handle._notify()
request = desktop.requests[0]
assert request.buttons == (*choices, (REVIEW, "Go to terminal"))
assert request.buttons == choices
assert request.on_answer is not None
request.on_answer("yyy")
assert handle.take() == "yyy"
@@ -492,7 +625,6 @@ def test_a_prompt_without_choices_gets_the_usual_buttons() -> None:
assert request.buttons == (
(ALLOW, "Allow once"),
(DENY, "Deny"),
(REVIEW, "Go to terminal"),
)
# Neither channel repeats the title in the body.
assert request.message == ""
+45 -1
View File
@@ -356,6 +356,34 @@ def test_prompt_wait_without_hook_yields_none() -> None:
assert source is None
def test_prompt_wait_carries_what_the_caller_declares() -> None:
"""Everything a host needs to say what is asked and when silence answers."""
seen: list[PromptWait] = []
@contextlib.contextmanager
def hook(wait: PromptWait) -> Generator[None]:
seen.append(wait)
yield None
set_prompt_wait_hook(hook)
choices = (("y", "Continue"), ("n", "Stop"))
with prompt_wait(
"policy",
"allow command 'rm'?",
"argv: rm -rf x",
choices,
body="",
facts=("program: rm",),
expires_in=30.0,
):
pass
assert seen[0].body == ""
assert seen[0].detail == "argv: rm -rf x"
assert seen[0].choices == choices
assert seen[0].facts == ("program: rm",)
assert seen[0].expires_in == 30.0
def test_prompt_wait_hook_wraps_ask() -> None:
"""The CLI hook sees one PromptWait per prompt and is exited afterwards."""
seen: list[PromptWait] = []
@@ -374,10 +402,26 @@ def test_prompt_wait_hook_wraps_ask() -> None:
backend = ScriptedBackend(["hi"])
with temporary_backend(backend):
assert ask("Name?") == "hi"
assert [(wait.kind, wait.title) for wait in seen] == [("ask", "Name?")]
assert [(wait.kind, wait.title, wait.body) for wait in seen] == [("ask", "a question is waiting", "Name?")]
assert entered == ["in", "out"]
def test_a_choice_prompt_names_its_question_and_options() -> None:
"""The prompt a host shows carries the question and what can be picked."""
seen: list[PromptWait] = []
@contextlib.contextmanager
def hook(wait: PromptWait) -> Generator[None]:
seen.append(wait)
yield None
set_prompt_wait_hook(hook)
with temporary_backend(ScriptedBackend(["1"])):
assert choose("Pick one", ["main", "release", "next", "other", "fifth"]) == "main"
assert seen[0].body == "Pick one"
assert seen[0].facts == ("main", "release", "next", "other")
def test_prompt_wait_hook_reports_the_prompt_family() -> None:
"""Each entry point announces its own kind so a host can pick a surface."""
kinds: list[str] = []