From ee9958e0e43f0c174c5a8ef9431cd393ec987004 Mon Sep 17 00:00:00 2001 From: Lavindeep Dhillon Date: Sun, 28 Jun 2026 04:29:31 -0400 Subject: [PATCH] feat(ui): approval focus-swap for the full-screen app The worker thread blocks on a concurrent.futures.Future while the loop thread renders the approval prompt into the pane and the dock becomes the approval input. ApprovalGate carries the three-way y/e/n + HIGH typed-run contract; [e]dit steers via steer_text re-entering the classifier; Ctrl-C/ EOF during an approval declines this action and the turn continues. Adds DESIGN section 31.16. --- docs/DESIGN.md | 12 ++ shellpilot/cli/app.py | 28 +++ shellpilot/cli/app_approval.py | 240 ++++++++++++++++++++++++ shellpilot/cli/app_main.py | 3 + shellpilot/cli/app_turn.py | 39 ++-- shellpilot/cli/app_ui.py | 36 +++- shellpilot/cli/terminal.py | 10 +- tests/test_app.py | 124 +++++++++++++ tests/test_app_approval.py | 324 +++++++++++++++++++++++++++++++++ tests/test_app_ui.py | 46 ++++- 10 files changed, 832 insertions(+), 30 deletions(-) create mode 100644 shellpilot/cli/app_approval.py create mode 100644 tests/test_app_approval.py diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 7ffaeb0..1916b21 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -2867,6 +2867,18 @@ Ctrl-C aborts a model turn mid-stream — the motivating case is a long invisibl **Out of scope (branch 6b — subprocess killpg).** A cancel requested while a tool *command* is running does NOT kill that subprocess; the command finishes/times out, then the next model call's stream-read sees the set event and aborts the turn. Killing the running subprocess on cancel is deferred. +### 31.16 Approval focus-swap (v2) + +Approvals are the exception to the fire-and-forget marshaling of §31.13: `ask_approval` / `ask_plan_approval` RETURN a value, so they cannot be enqueued and forgotten. The full-screen app resolves them with a **focus-swap handshake** (`cli/app_approval.py`, `ApprovalGate`): the worker thread blocks on a `concurrent.futures.Future` while the loop thread renders the prompt into the pane and the dock becomes the approval input. + +**Worker blocks, loop thread answers.** `ThreadedUI.ask_approval` delegates to `gate.ask_command(request)` (and `ask_plan_approval` → `gate.ask_plan`). The gate schedules a loop-thread setup thunk (via the same `TurnRunner.schedule` marshaler) and then calls `future.result()`, which blocks the worker. The thunk renders the prompt block — `AppUI.show_approval` (capped diff panel via `render_diff(..., max_rows=DiffReveal.WINDOW_ROWS)`, the risk/reason info line, the CWD line) or `AppUI.show_plan_approval` (the plan panel + the sanitized artifact path) — and arms `gate._pending`. Both `Future` operations are thread-safe stdlib primitives, so no lock is needed; `_pending` is touched only on the loop thread (the setup thunk arms it; the dock keybindings read+clear it). + +**Three-way outcome, unchanged contract (§14.6).** The dock's submit keybinding routes the typed line to `gate.submit(line)` when `gate.active`, BEFORE the `/exit` check (a mid-approval `/exit` is an approval answer, not a quit). The pure helpers `parse_command_choice` / `parse_plan_choice` carry the contract: a HIGH-risk **command** executes only on the literal `run` (the typed-`run` gate); every other request takes y/e/n; an unrecognized plan token re-prompts. `[e]dit` enters a steer phase — the next line becomes `ApprovalReply.steer_text` (empty = plain decline), which re-enters the deterministic classifier as a re-proposal exactly as in the REPL; the un-approved action never runs. The accepted line is echoed into the pane (sanitized, via `show_status` — never `show_user_message`, which would reset the live turn indicator). + +**Ctrl-C / EOF decline THIS action.** During an approval the worker is blocked on the Future, not in a model-stream read, so the §31.15 model-cancel path would not fire. The `c-c` and `c-d` handlers therefore check `gate.active` FIRST and call `gate.cancel()` (resolve as decline — `DECLINE` for a command, `("n", "")` for a plan), mirroring `TerminalUI.ask_approval`'s `except KeyboardInterrupt: return DECLINE`. The turn continues; turn-level cancel is available again once the prompt returns. + +**Shutdown ceiling.** A pending approval at app exit leaves the daemon worker blocked on `future.result()`; harmless for this opt-in dev entry (process exit reaps it). On promotion to the shipping default, resolve pending approvals as DECLINE on exit. + ## 32. Model Selection And Preload ### 32.1 Boot Model Picker diff --git a/shellpilot/cli/app.py b/shellpilot/cli/app.py index caa4bf6..50f1c3c 100644 --- a/shellpilot/cli/app.py +++ b/shellpilot/cli/app.py @@ -23,6 +23,7 @@ from collections.abc import Callable, Sequence from dataclasses import dataclass from pathlib import Path +from typing import TYPE_CHECKING from prompt_toolkit.application import Application, get_app from prompt_toolkit.buffer import Buffer @@ -48,6 +49,9 @@ Glyphs, ) +if TYPE_CHECKING: + from shellpilot.cli.app_approval import ApprovalGate + # The dock grows to fit multi-line input up to this many rows, then scrolls # internally. NOTE: a fixed cap — a per-terminal-height fraction would be nicer # on a very short terminal, but a flat cap is enough until the live wiring @@ -151,6 +155,7 @@ def build_app( ui: AppUI | None = None, on_submit: Callable[[str], None] | None = None, on_interrupt: Callable[[], bool] | None = None, + approval_gate: ApprovalGate | None = None, ) -> Application[None]: """Build the full-screen app shell. @@ -296,6 +301,14 @@ def _bar() -> Window: @kb.add("enter", filter=dock_focused) @kb.add("c-j", filter=dock_focused) def _submit(event: KeyPressEvent) -> None: + # During an approval the dock IS the approval input: route the line to the + # gate (which resolves the worker's Future) BEFORE the /exit check, so a + # mid-approval "/exit" is an approval answer, not a quit (§31.16). + if approval_gate is not None and approval_gate.active: + line = dock_buffer.text + dock_buffer.reset() + approval_gate.submit(line) + return text = dock_buffer.text if text.strip() == "/exit": event.app.exit() @@ -326,6 +339,14 @@ def _page_down(event: KeyPressEvent) -> None: @kb.add("c-c") def _interrupt(event: KeyPressEvent) -> None: + # During an approval the worker is blocked on the gate's Future, not in a + # model-stream read, so on_interrupt (model cancel) would not fire. Ctrl-C + # must resolve the approval as a decline of THIS action; the turn continues + # and turn-level cancel is available again after the prompt returns (mirrors + # TerminalUI.ask_approval's KeyboardInterrupt → DECLINE) (§31.16). + if approval_gate is not None and approval_gate.active: + approval_gate.cancel() + return # Raw mode disables ISIG, so Ctrl-C is a normal key press, not a SIGINT # (branch 6, §31.15). When a turn is in flight, on_interrupt cancels it and # returns True (the worker aborts the stream and renders the marker), so we @@ -334,6 +355,13 @@ def _interrupt(event: KeyPressEvent) -> None: return _ui.show_status(_IDLE_HINT) + @kb.add("c-d", filter=dock_focused) + def _eof(event: KeyPressEvent) -> None: + # EOF during an approval declines THIS action (same as Ctrl-C). Otherwise a + # no-op: the dock owns c-d so a stray press never tears down the app. + if approval_gate is not None and approval_gate.active: + approval_gate.cancel() + # NOTE: keyboard PageUp/PageDown drive the pane via the cursor-line model # above (auto-follow + scroll-back). Mouse-wheel scroll-back is deferred to # branch 9: the wheel nudges the window's own vertical_scroll, which the diff --git a/shellpilot/cli/app_approval.py b/shellpilot/cli/app_approval.py new file mode 100644 index 0000000..321a77f --- /dev/null +++ b/shellpilot/cli/app_approval.py @@ -0,0 +1,240 @@ +"""Approval focus-swap rendezvous for the full-screen app (design section 31.16). + +The full-screen app runs one conversation turn on a worker thread while the +prompt_toolkit event loop owns the main thread (see ``app_turn.py``). Every +fire-and-forget UI call is marshaled loop-ward, but ``ask_approval`` / +``ask_plan_approval`` RETURN a value, so they cannot be fire-and-forget. This is +the focus-swap handshake: the worker blocks on a ``concurrent.futures.Future`` +while the loop thread renders the prompt into the pane, reads the user's +keystrokes in the dock, and resolves the future. + +The pure parse helpers (:func:`parse_command_choice` / :func:`parse_plan_choice`) +carry the three-way y/e/n + HIGH typed-"run" contract and are unit-tested +directly, with no threading. +""" + +from __future__ import annotations + +import functools +from collections.abc import Callable +from concurrent.futures import Future +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from shellpilot.cli.render import _sanitize_line +from shellpilot.cli.theme import UNICODE_GLYPHS, Glyphs +from shellpilot.policy.approvals import APPROVE, DECLINE, ApprovalReply +from shellpilot.policy.risk import RiskLevel + +if TYPE_CHECKING: + from shellpilot.cli.app_turn import Schedule + from shellpilot.cli.app_ui import AppUI + from shellpilot.policy.approvals import ApprovalRequest + from shellpilot.runtime.planner import TaskPlan + + +class _NeedSteer: + """Sentinel: the user chose [e]dit at the choice phase; read steering next.""" + + +NEED_STEER = _NeedSteer() + + +def parse_command_choice(request: ApprovalRequest, answer: str) -> ApprovalReply | _NeedSteer: + """Map a command/tool approval keystroke to its three-way outcome. + + A HIGH-risk COMMAND requires the literal "run" to execute (the typed-"run" + gate, design section 14.6); [e]dit steers without running and anything else + declines. Every other request takes the y/e/n path. + """ + low = answer.strip().lower() + if request.risk is RiskLevel.HIGH and request.kind == "command": + if low == "run": + return APPROVE + if low in ("e", "edit"): + return NEED_STEER + return DECLINE + if low in ("y", "yes"): + return APPROVE + if low in ("e", "edit"): + return NEED_STEER + return DECLINE + + +def parse_plan_choice(answer: str) -> str | None: + """Map a plan-approval keystroke to 'y' / 'e' / 'n', or None to re-prompt. + + Mirrors :meth:`TerminalUI.ask_plan_approval`: y/e/n plus empty→n; an + unrecognized non-empty token re-prompts (its loop), modeled here as None. + """ + low = answer.strip().lower() + if low in ("y", "yes"): + return "y" + if low in ("e", "edit"): + return "e" + if low in ("n", "no", ""): + return "n" + return None + + +@dataclass +class _Pending: + """The in-flight prompt. Loop-thread-only (see :class:`ApprovalGate`).""" + + future: Future[object] + # feed(line) -> True when the future was resolved (prompt done), False when + # another input line is wanted (steer/revision phase, or a re-prompt). + feed: Callable[[str], bool] + # Resolve as decline (Ctrl-C/EOF during approval cancels THIS action only). + on_cancel: Callable[[], None] + + +class ApprovalGate: + """Thread-safe approval rendezvous (design section 31.16). + + ``_pending`` is touched ONLY on the loop thread: :meth:`_enter_command` / + :meth:`_enter_plan` set it (scheduled loop-ward by the worker), and + :meth:`submit` / :meth:`cancel` read+clear it (called from keybindings, which + run on the loop thread). The worker thread only ever touches the ``Future`` + (``result()`` blocks it; the loop thread's ``set_result`` unblocks it) — both + are thread-safe stdlib primitives, so no lock is needed. + + NOTE: a pending approval at app exit leaves the daemon worker blocked on + result(); harmless (process exit reaps it). On promotion to the shipping + default, resolve pending approvals as DECLINE on exit. + """ + + def __init__(self, *, ui: AppUI, schedule: Schedule, glyphs: Glyphs = UNICODE_GLYPHS) -> None: + self._ui = ui + self._schedule = schedule + self._glyphs = glyphs + self._pending: _Pending | None = None + + # ------------------------------------------------------------------ + # Worker-thread entry points — block on the future. + # ------------------------------------------------------------------ + + def ask_command(self, request: ApprovalRequest) -> ApprovalReply: + future: Future[object] = Future() + self._schedule(functools.partial(self._enter_command, request, future)) + result = future.result() # BLOCKS the worker thread + assert isinstance(result, ApprovalReply) + return result + + def ask_plan(self, plan: TaskPlan, path: str) -> tuple[str, str]: + future: Future[object] = Future() + self._schedule(functools.partial(self._enter_plan, plan, path, future)) + result = future.result() # BLOCKS the worker thread + assert isinstance(result, tuple) + return result + + # ------------------------------------------------------------------ + # Loop-thread prompt setup (scheduled by the worker entry points). + # ------------------------------------------------------------------ + + def _echo(self, line: str) -> None: + # Keep the accepted input visible after the dock clears. show_status + # re-sanitizes, so user-controlled text never reaches the pane raw; it + # also (unlike show_user_message) does NOT reset the live turn indicator. + self._ui.show_status(f" {self._glyphs.chevron} {_sanitize_line(line)}") + + def _enter_command(self, request: ApprovalRequest, future: Future[object]) -> None: + # Fail closed: a render sink raising on the loop thread must resolve the + # worker's Future (via set_exception) instead of leaving it blocked on + # result() forever — the turn then ends through TurnRunner._run's except + # ("Turn failed") rather than wedging the app. Never approves by accident. + try: + self._ui.show_approval(request) + if request.risk is RiskLevel.HIGH and request.kind == "command": + hint = ' Type "run" to execute, [e]dit to steer, or Enter to cancel' + else: + hint = " Approve? [y]es / [e]dit / [n]o" + self._ui.show_status(hint) + except Exception as exc: # noqa: BLE001 - never leave the worker hung + self._pending = None + if not future.done(): + future.set_exception(exc) + return + phase = {"steer": False} + + def feed(line: str) -> bool: + if phase["steer"]: + # Empty steer = plain decline (matches TerminalUI._read_steer). + self._echo(line) + future.set_result(ApprovalReply(approved=False, steer_text=line.strip() or None)) + return True + decision = parse_command_choice(request, line) + if isinstance(decision, _NeedSteer): + phase["steer"] = True + self._ui.show_status(" Tell the model what to do instead:") + return False + self._echo(line) + future.set_result(decision) + return True + + self._pending = _Pending(future, feed, lambda: future.set_result(DECLINE)) + + def _enter_plan(self, plan: TaskPlan, path: str, future: Future[object]) -> None: + hint = " Approve plan? [y]es / [e]dit / [n]o" + # Fail closed (see _enter_command): a render sink raising here resolves + # the worker's Future rather than wedging it on result(). + try: + self._ui.show_plan_approval(plan, path) + self._ui.show_status(hint) + except Exception as exc: # noqa: BLE001 - never leave the worker hung + self._pending = None + if not future.done(): + future.set_exception(exc) + return + phase = {"revision": False} + + def feed(line: str) -> bool: + if phase["revision"]: + self._echo(line) + future.set_result(("e", line.strip())) + return True + choice = parse_plan_choice(line) + if choice is None: + self._ui.show_status(hint) # re-prompt (matches the TerminalUI loop) + return False + if choice == "e": + phase["revision"] = True + self._ui.show_status(" Describe the changes you want:") + return False + self._echo(line) + future.set_result((choice, "")) + return True + + self._pending = _Pending(future, feed, lambda: future.set_result(("n", ""))) + + # ------------------------------------------------------------------ + # Loop-thread public surface — driven by the dock keybindings. + # ------------------------------------------------------------------ + + @property + def active(self) -> bool: + return self._pending is not None + + def submit(self, line: str) -> None: + pending = self._pending + if pending is None: + return + # Fail closed: feed() echoes via show_status BEFORE resolving the Future, + # so a sink raising there must still resolve it — never leave the worker + # blocked on result(). + try: + done = pending.feed(line) + except Exception as exc: # noqa: BLE001 - never leave the worker hung + self._pending = None + if not pending.future.done(): + pending.future.set_exception(exc) + return + if done: + self._pending = None + + def cancel(self) -> None: + pending = self._pending + if pending is None: + return + self._pending = None + pending.on_cancel() diff --git a/shellpilot/cli/app_main.py b/shellpilot/cli/app_main.py index 45827d1..edff8cc 100644 --- a/shellpilot/cli/app_main.py +++ b/shellpilot/cli/app_main.py @@ -21,6 +21,7 @@ from shellpilot.cli.app_turn import TurnRunner if TYPE_CHECKING: + from shellpilot.cli.app_approval import ApprovalGate from shellpilot.cli.app_ui import AppUI from shellpilot.cli.theme import Glyphs from shellpilot.runtime.conversation import ConversationRuntime @@ -38,6 +39,7 @@ def run_app( commands: Sequence[str], is_cloud: bool = False, ctx_pct: int = 0, + approval_gate: ApprovalGate | None = None, ) -> int: """Build the full-screen app around an already-wired conversation and run it. @@ -71,6 +73,7 @@ def run_app( ui=app_ui, on_submit=runner.start, on_interrupt=runner.request_cancel, + approval_gate=approval_gate, ) runner.app = app runner.conversation = runtime diff --git a/shellpilot/cli/app_turn.py b/shellpilot/cli/app_turn.py index 9c1bca8..fc6634f 100644 --- a/shellpilot/cli/app_turn.py +++ b/shellpilot/cli/app_turn.py @@ -14,8 +14,9 @@ * :class:`ThreadedUI` — a ``RuntimeUI`` that wraps the real ``AppUI`` and enqueues every fire-and-forget content call onto the loop thread instead of running it inline. The blocking approval methods cannot be fire-and-forget - (they return a value), so they delegate straight to the inner UI and run on - the worker thread. + (they return a value): with an :class:`ApprovalGate` wired they hand off to it + (the worker blocks on the gate's Future); with no gate they raise + ``NotImplementedError`` so an approval can never silently default. * :class:`TurnRunner` — owns the single worker thread for one turn and a ``busy`` flag that is only ever touched on the loop thread (no lock needed). """ @@ -32,10 +33,11 @@ if TYPE_CHECKING: from prompt_toolkit.application import Application + from shellpilot.cli.app_approval import ApprovalGate from shellpilot.cli.app_ui import AppUI from shellpilot.policy.approvals import ApprovalReply, ApprovalRequest from shellpilot.runtime.conversation import ConversationRuntime - from shellpilot.runtime.events import RuntimeUI, TurnStats + from shellpilot.runtime.events import TurnStats from shellpilot.runtime.planner import TaskPlan # A zero-arg callback scheduled to run on the loop thread. @@ -59,9 +61,19 @@ class ThreadedUI: ``"b"``, not ``"b"``, ``"b"``. """ - def __init__(self, *, inner: RuntimeUI, schedule: Schedule) -> None: + def __init__( + self, + *, + inner: AppUI, + schedule: Schedule, + approval_gate: ApprovalGate | None = None, + ) -> None: self._inner = inner self._schedule = schedule + # Branch 7 (§31.16): the focus-swap handshake for the two blocking + # approval methods. None falls back to the inner UI (back-compat — the + # inner raises NotImplementedError until a gate is wired). + self._approval_gate = approval_gate # ------------------------------------------------------------------ # Fire-and-forget content methods — enqueued, never run inline. @@ -110,16 +122,21 @@ def show_plan_progress(self, plan: TaskPlan) -> None: # They delegate straight to the inner UI and run on the worker thread. # ------------------------------------------------------------------ - # NOTE: branch 7 replaces these with a thread-safe Future focus-swap - # handshake (marshal the prompt to the loop thread, block the worker on a - # Future for the reply). Until then the inner AppUI raises NotImplementedError - # here — no approval may silently default — and the worker's try/except - # surfaces that as a pane error (see TurnRunner._run). + # Branch 7 (§31.16): with a gate wired, the worker blocks on the gate's + # Future while the loop thread renders the prompt and reads the dock. With no + # gate (CI back-compat only — the real app always wires one), they raise so an + # approval can never silently default; the worker's try/except surfaces it as + # a pane error (see TurnRunner._run). The inner AppUI is content-only and has + # no approval methods, hence the direct raise rather than an inner delegation. def ask_approval(self, request: ApprovalRequest) -> ApprovalReply: - return self._inner.ask_approval(request) + if self._approval_gate is not None: + return self._approval_gate.ask_command(request) + raise NotImplementedError("approval requires an ApprovalGate (no gate wired)") def ask_plan_approval(self, plan: TaskPlan, path: str) -> tuple[str, str]: - return self._inner.ask_plan_approval(plan, path) + if self._approval_gate is not None: + return self._approval_gate.ask_plan(plan, path) + raise NotImplementedError("plan approval requires an ApprovalGate (no gate wired)") class TurnRunner: diff --git a/shellpilot/cli/app_ui.py b/shellpilot/cli/app_ui.py index 628f33c..3bcbb1f 100644 --- a/shellpilot/cli/app_ui.py +++ b/shellpilot/cli/app_ui.py @@ -27,17 +27,24 @@ from rich.padding import Padding from rich.text import Text -from shellpilot.cli.render import _sanitize_line, plan_step_line +from shellpilot.cli.render import ( + _sanitize_line, + approval_cwd, + approval_info, + plan_panel, + plan_step_line, + render_diff, +) from shellpilot.cli.render import tool_call as render_tool_call from shellpilot.cli.render import tool_result as render_tool_result -from shellpilot.cli.streaming import phase_for_elapsed +from shellpilot.cli.streaming import DiffReveal, phase_for_elapsed from shellpilot.cli.theme import SHELLPILOT_THEME, UNICODE_GLYPHS, Glyphs from shellpilot.memory.redaction import redact_structure from shellpilot.runtime.budget import CHARS_PER_TOKEN from shellpilot.tools.base import workspace_display if TYPE_CHECKING: - from shellpilot.policy.approvals import ApprovalReply, ApprovalRequest + from shellpilot.policy.approvals import ApprovalRequest from shellpilot.runtime.events import TurnStats from shellpilot.runtime.planner import TaskPlan @@ -337,11 +344,24 @@ def show_plan_progress(self, plan: TaskPlan) -> None: self._cache = None # ------------------------------------------------------------------ - # Approval stubs — raise until branch 7 wires the focus-swap + # Approval prompt content (design section 31.16). The blocking handshake + # lives in ApprovalGate; these only append the prompt block to the pane, + # mirroring the render block of TerminalUI.ask_approval / ask_plan_approval + # MINUS the DiffReveal animation — the pane scroll replaces the Rich Live. # ------------------------------------------------------------------ - def ask_approval(self, request: ApprovalRequest) -> ApprovalReply: - raise NotImplementedError("approval focus-swap is wired in branch 7") + def show_approval(self, request: ApprovalRequest) -> None: + if request.diff: + self._add_renderable( + Padding( + render_diff(request.diff, self._glyphs, max_rows=DiffReveal.WINDOW_ROWS), + (0, 0, 0, 2), + ) + ) + self._add_renderable(approval_info(request, plain_badge=False)) + self._add_renderable(approval_cwd(request)) - def ask_plan_approval(self, plan: TaskPlan, path: str) -> tuple[str, str]: - raise NotImplementedError("approval focus-swap is wired in branch 7") + def show_plan_approval(self, plan: TaskPlan, path: str) -> None: + self._add_renderable(plan_panel(plan, self._glyphs)) + # The path is user-visible state reaching the pane → sanitize it. + self._add_renderable(Text(_sanitize_line(path), style="sp.faint")) diff --git a/shellpilot/cli/terminal.py b/shellpilot/cli/terminal.py index 8c3975c..080b046 100644 --- a/shellpilot/cli/terminal.py +++ b/shellpilot/cli/terminal.py @@ -15,6 +15,7 @@ from rich.padding import Padding from rich.text import Text +from shellpilot.cli.app_approval import ApprovalGate from shellpilot.cli.app_main import run_app from shellpilot.cli.app_turn import ThreadedUI, TurnRunner from shellpilot.cli.app_ui import AppUI @@ -609,6 +610,7 @@ def _preload(model_name: str) -> None: app_mode = env.get("SHELLPILOT_UI") == "app" app_ui: AppUI | None = None app_runner: TurnRunner | None = None + approval_gate: ApprovalGate | None = None ui: RuntimeUI if app_mode: app_ui = AppUI( @@ -618,7 +620,10 @@ def _preload(model_name: str) -> None: show_reasoning=settings.ui.show_reasoning_summary, ) app_runner = TurnRunner(inner_ui=app_ui) - ui = ThreadedUI(inner=app_ui, schedule=app_runner.schedule) + # The focus-swap gate handles the two blocking approval methods (§31.16): + # the worker blocks on a Future while the dock reads the user's answer. + approval_gate = ApprovalGate(ui=app_ui, schedule=app_runner.schedule, glyphs=glyphs) + ui = ThreadedUI(inner=app_ui, schedule=app_runner.schedule, approval_gate=approval_gate) else: ui = TerminalUI(console, glyphs=glyphs, spinner=settings.ui.spinner, workspace=workspace) runtime = ConversationRuntime( @@ -648,7 +653,7 @@ def _preload(model_name: str) -> None: # fully-configured runtime; the default REPL below is reached only when # app_mode is False, so it stays byte-identical. if app_mode: - assert app_runner is not None and app_ui is not None + assert app_runner is not None and app_ui is not None and approval_gate is not None return run_app( runtime, app_runner, @@ -663,6 +668,7 @@ def _preload(model_name: str) -> None: runtime.status().estimated_prompt_tokens, runtime.status().budget.model_context_tokens, ), + approval_gate=approval_gate, ) attachments = AttachmentQueue() dispatcher = SlashDispatcher( diff --git a/tests/test_app.py b/tests/test_app.py index 38fdc14..640d581 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -316,3 +316,127 @@ def test_c_c_shows_idle_hint_when_on_interrupt_none(tmp_path: Path) -> None: # Back-compat: with no on_interrupt (the inert shell), Ctrl-C shows the hint. ui = _run_with_interrupt(tmp_path, on_interrupt=None) assert "idle" in ui._render_ansi() + + +# --- Branch-7 approval focus-swap routing (§31.16) ---------------------------- + + +class _FakeGate: + """Records the keybinding handshake without the real Future plumbing. + + One ``submit``/``cancel`` resolves the fake prompt (``active`` flips False), + mirroring the real gate clearing ``_pending`` once the future resolves. + """ + + def __init__(self, *, active: bool = True) -> None: + self._active = active + self.submitted: list[str] = [] + self.cancelled = 0 + + @property + def active(self) -> bool: + return self._active + + def submit(self, line: str) -> None: + self.submitted.append(line) + self._active = False + + def cancel(self) -> None: + self.cancelled += 1 + self._active = False + + +def _build_with_gate( + tmp_path: Path, + inp: object, + gate: _FakeGate, + *, + on_interrupt: object | None = None, +) -> tuple[Application[None], AppUI]: + ui = AppUI(glyphs=UNICODE_GLYPHS, width_fn=lambda: 80) + app = build_app( + workspace=tmp_path, + model="gemma4:e4b", + profile="balanced", + glyphs=UNICODE_GLYPHS, + commands=command_words(), + input=inp, # type: ignore[arg-type] + output=DummyOutput(), + ui=ui, + on_interrupt=on_interrupt, # type: ignore[arg-type] + approval_gate=gate, # type: ignore[arg-type] + ) + return app, ui + + +def test_approval_active_submit_routes_to_gate(tmp_path: Path) -> None: + gate = _FakeGate(active=True) + with create_pipe_input() as inp: + app, ui = _build_with_gate(tmp_path, inp, gate) + inp.send_text("y\n") # gate active → gate.submit("y"), gate deactivates + inp.send_text("/exit\n") # gate now inactive → real /exit quits + app.run() + assert gate.submitted == ["y"] + assert gate.cancelled == 0 + # The routed answer did NOT fall through to the inert show_status echo path + # (that path runs only when no gate intercepts the submit). + assert ui._renderables == [] + + +def test_approval_active_intercepts_exit(tmp_path: Path) -> None: + # A mid-approval "/exit" is an approval answer (gate.submit), NOT a quit — the + # gate check sits BEFORE the /exit check. + gate = _FakeGate(active=True) + with create_pipe_input() as inp: + app, ui = _build_with_gate(tmp_path, inp, gate) + inp.send_text("/exit\n") # routed to the gate, deactivates it + inp.send_text("/exit\n") # now inactive → quits + app.run() + assert gate.submitted == ["/exit"] + assert "/exit" not in ui._render_ansi() + + +def test_approval_active_ctrl_c_cancels_gate_not_interrupt(tmp_path: Path) -> None: + gate = _FakeGate(active=True) + interrupts: list[int] = [] + + def on_interrupt() -> bool: + interrupts.append(1) + return True + + with create_pipe_input() as inp: + app, _ = _build_with_gate(tmp_path, inp, gate, on_interrupt=on_interrupt) + inp.send_text("\x03") # Ctrl-C while gate active → gate.cancel() + inp.send_text("/exit\n") # gate now inactive → quits + app.run() + assert gate.cancelled == 1 + assert interrupts == [] # model cancel did NOT fire during the approval + + +def test_ctrl_c_falls_through_to_interrupt_when_gate_inactive(tmp_path: Path) -> None: + # Existing behavior preserved: with no active approval, Ctrl-C reaches + # on_interrupt (the turn-level cancel), not the gate. + gate = _FakeGate(active=False) + interrupts: list[int] = [] + + def on_interrupt() -> bool: + interrupts.append(1) + return True + + with create_pipe_input() as inp: + app, _ = _build_with_gate(tmp_path, inp, gate, on_interrupt=on_interrupt) + inp.send_text("\x03") + inp.send_text("/exit\n") + app.run() + assert interrupts == [1] + assert gate.cancelled == 0 + + +def test_eof_cancels_active_gate(tmp_path: Path) -> None: + gate = _FakeGate(active=True) + with create_pipe_input() as inp: + app, _ = _build_with_gate(tmp_path, inp, gate) + inp.send_text("\x04") # Ctrl-D (EOF) while gate active → gate.cancel() + inp.send_text("/exit\n") # gate now inactive → quits + app.run() + assert gate.cancelled == 1 diff --git a/tests/test_app_approval.py b/tests/test_app_approval.py new file mode 100644 index 0000000..7ea4b6b --- /dev/null +++ b/tests/test_app_approval.py @@ -0,0 +1,324 @@ +"""Tests for the approval focus-swap gate (design section 31.16). + +The pure parse helpers carry the three-way y/e/n + HIGH typed-"run" contract and +are tested directly. The gate handshake is exercised across a real worker thread +(the worker blocks on a ``concurrent.futures.Future`` while the test thread acts +as the loop thread: it drains the scheduled ``_enter`` thunk, then submits keys). +""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path + +from shellpilot.cli.app_approval import ( + NEED_STEER, + ApprovalGate, + parse_command_choice, + parse_plan_choice, +) +from shellpilot.cli.app_ui import AppUI +from shellpilot.cli.theme import UNICODE_GLYPHS +from shellpilot.policy.approvals import APPROVE, DECLINE, ApprovalReply, ApprovalRequest +from shellpilot.policy.risk import RiskLevel +from shellpilot.runtime.planner import PlanStep, TaskPlan + + +def high_command() -> ApprovalRequest: + return ApprovalRequest( + kind="command", + display="rm -rf build", + risk=RiskLevel.HIGH, + reasons=("recursive delete",), + cwd=Path("/tmp/ws"), + ) + + +def normal_tool() -> ApprovalRequest: + return ApprovalRequest( + kind="tool", + display="patch_file x.py", + risk=RiskLevel.MEDIUM, + reasons=("writes inside workspace",), + cwd=Path("/tmp/ws"), + ) + + +def sensitive_read() -> ApprovalRequest: + return ApprovalRequest( + kind="tool", + display="read_file ~/.ssh/id", + risk=RiskLevel.HIGH, + reasons=("sensitive read",), + cwd=Path("/tmp/ws"), + ) + + +def make_plan() -> TaskPlan: + return TaskPlan( + task_id="20260611-040000-demo", + goal="Demo goal", + user_intent="demo", + workspace=Path("/tmp/ws"), + profile="balanced", + steps=[PlanStep(title="First"), PlanStep(title="Second")], + ) + + +# --- Pure parse helpers (no threading) --------------------------------------- + + +def test_parse_command_choice_high_command() -> None: + req = high_command() + assert parse_command_choice(req, "run") == APPROVE + assert parse_command_choice(req, " RUN ") == APPROVE # trimmed + case-folded + assert parse_command_choice(req, "e") is NEED_STEER + assert parse_command_choice(req, "edit") is NEED_STEER + # Only the literal "run" executes a HIGH command; y/yes do NOT. + assert parse_command_choice(req, "y") == DECLINE + assert parse_command_choice(req, "yes") == DECLINE + assert parse_command_choice(req, "x") == DECLINE + assert parse_command_choice(req, "") == DECLINE + + +def test_parse_command_choice_normal() -> None: + req = normal_tool() + assert parse_command_choice(req, "y") == APPROVE + assert parse_command_choice(req, "yes") == APPROVE + assert parse_command_choice(req, "e") is NEED_STEER + assert parse_command_choice(req, "edit") is NEED_STEER + assert parse_command_choice(req, "n") == DECLINE + assert parse_command_choice(req, "x") == DECLINE + assert parse_command_choice(req, "") == DECLINE + + +def test_parse_command_choice_high_tool_uses_yen_path() -> None: + # A HIGH-risk TOOL is a sensitive read, not a command — it takes the y/e/n + # path, not the typed-"run" path. + req = sensitive_read() + assert parse_command_choice(req, "y") == APPROVE + assert parse_command_choice(req, "run") == DECLINE # "run" is not y/yes here + assert parse_command_choice(req, "e") is NEED_STEER + + +def test_parse_plan_choice() -> None: + assert parse_plan_choice("y") == "y" + assert parse_plan_choice("yes") == "y" + assert parse_plan_choice("e") == "e" + assert parse_plan_choice("edit") == "e" + assert parse_plan_choice("n") == "n" + assert parse_plan_choice("no") == "n" + assert parse_plan_choice("") == "n" # empty → decline, mirrors TerminalUI + assert parse_plan_choice("x") is None # unrecognized → re-prompt + + +# --- Gate handshake across a real worker thread ------------------------------ + + +def make_gate() -> tuple[ApprovalGate, list[object]]: + sink: list[object] = [] + ui = AppUI(glyphs=UNICODE_GLYPHS, width_fn=lambda: 80) + gate = ApprovalGate(ui=ui, schedule=sink.append, glyphs=UNICODE_GLYPHS) + return gate, sink + + +def drain_enter(sink: list[object], timeout: float = 2.0) -> None: + """Busy-wait until the worker scheduled its _enter thunk, then run it here. + + Running the thunk on the test thread plays the role of the loop thread: it + renders the prompt and arms gate._pending. + """ + deadline = time.monotonic() + timeout + while not sink and time.monotonic() < deadline: + time.sleep(0.001) + assert sink, "worker never scheduled the enter thunk" + thunk = sink.pop(0) + assert callable(thunk) + thunk() + + +def run_command( + gate: ApprovalGate, request: ApprovalRequest +) -> tuple[threading.Thread, list[object]]: + holder: list[object] = [] + # daemon=True so a regression that fails to resolve the Future fails the test + # FAST (join times out → assert not is_alive) instead of wedging the suite at + # process exit on a blocked non-daemon worker. + thread = threading.Thread(target=lambda: holder.append(gate.ask_command(request)), daemon=True) + thread.start() + return thread, holder + + +def run_plan( + gate: ApprovalGate, plan: TaskPlan, path: str +) -> tuple[threading.Thread, list[object]]: + holder: list[object] = [] + thread = threading.Thread(target=lambda: holder.append(gate.ask_plan(plan, path)), daemon=True) + thread.start() + return thread, holder + + +def test_gate_command_run_approves() -> None: + gate, sink = make_gate() + thread, holder = run_command(gate, high_command()) + drain_enter(sink) + assert gate.active + gate.submit("run") + thread.join(2.0) + assert not thread.is_alive() + assert holder == [APPROVE] + assert not gate.active + + +def test_gate_command_decline() -> None: + gate, sink = make_gate() + thread, holder = run_command(gate, high_command()) + drain_enter(sink) + gate.submit("") # Enter → cancel a HIGH command + thread.join(2.0) + assert holder == [DECLINE] + + +def test_gate_command_edit_then_steer() -> None: + gate, sink = make_gate() + thread, holder = run_command(gate, high_command()) + drain_enter(sink) + gate.submit("e") # enter steer phase — does NOT resolve + assert gate.active + assert holder == [] + gate.submit("do X instead") + thread.join(2.0) + assert holder == [ApprovalReply(approved=False, steer_text="do X instead")] + + +def test_gate_command_empty_steer_is_plain_decline() -> None: + gate, sink = make_gate() + thread, holder = run_command(gate, high_command()) + drain_enter(sink) + gate.submit("e") + gate.submit(" ") # whitespace-only steer → plain decline (steer_text None) + thread.join(2.0) + assert holder == [ApprovalReply(approved=False, steer_text=None)] + + +def test_gate_command_cancel_declines() -> None: + gate, sink = make_gate() + thread, holder = run_command(gate, high_command()) + drain_enter(sink) + gate.cancel() # Ctrl-C during approval → decline THIS action + thread.join(2.0) + assert holder == [DECLINE] + assert not gate.active + + +def test_gate_plan_yes() -> None: + gate, sink = make_gate() + thread, holder = run_plan(gate, make_plan(), "/tmp/ws/PLAN.md") + drain_enter(sink) + gate.submit("y") + thread.join(2.0) + assert holder == [("y", "")] + + +def test_gate_plan_edit_then_revision() -> None: + gate, sink = make_gate() + thread, holder = run_plan(gate, make_plan(), "/tmp/ws/PLAN.md") + drain_enter(sink) + gate.submit("e") + assert gate.active + gate.submit("make it shorter") + thread.join(2.0) + assert holder == [("e", "make it shorter")] + + +def test_gate_plan_unrecognized_reprompts_then_resolves() -> None: + gate, sink = make_gate() + thread, holder = run_plan(gate, make_plan(), "/tmp/ws/PLAN.md") + drain_enter(sink) + gate.submit("x") # unrecognized → re-prompt, stays active + assert gate.active + assert holder == [] + gate.submit("y") + thread.join(2.0) + assert holder == [("y", "")] + + +def test_gate_plan_cancel_declines() -> None: + gate, sink = make_gate() + thread, holder = run_plan(gate, make_plan(), "/tmp/ws/PLAN.md") + drain_enter(sink) + gate.cancel() + thread.join(2.0) + assert holder == [("n", "")] + + +# --- Fail-closed: a render/echo sink raising must NOT wedge the worker --------- + + +class _BoomUI: + """AppUI stand-in whose chosen sink raises, to prove the gate fails closed. + + ``boom_on="approval"`` raises in the setup render (``_enter_*``); ``"echo"`` + lets setup succeed (hint is status call #1) then raises on the echo (call #2, + inside ``feed`` before the Future is resolved). + """ + + def __init__(self, *, boom_on: str) -> None: + self.boom_on = boom_on + self._status_calls = 0 + + def show_approval(self, request: object) -> None: + if self.boom_on == "approval": + raise RuntimeError("boom-approval") + + def show_plan_approval(self, plan: object, path: str) -> None: + if self.boom_on == "approval": + raise RuntimeError("boom-approval") + + def show_status(self, text: str) -> None: + self._status_calls += 1 + if self.boom_on == "echo" and self._status_calls >= 2: + raise RuntimeError("boom-echo") + + +def _run_command_capturing(gate: ApprovalGate) -> tuple[threading.Thread, list[BaseException]]: + errs: list[BaseException] = [] + + def worker() -> None: + try: + gate.ask_command(high_command()) + except BaseException as exc: # noqa: BLE001 - capture the forwarded failure + errs.append(exc) + + thread = threading.Thread(target=worker, daemon=True) + thread.start() + return thread, errs + + +def test_gate_setup_exception_resolves_future_not_hang() -> None: + # show_approval raises during _enter_command → the Future is resolved with the + # error (set_exception) and the worker unblocks, instead of hanging on result(). + sink: list[object] = [] + gate = ApprovalGate(ui=_BoomUI(boom_on="approval"), schedule=sink.append, glyphs=UNICODE_GLYPHS) # type: ignore[arg-type] + thread, errs = _run_command_capturing(gate) + drain_enter(sink) + thread.join(2.0) + assert not thread.is_alive() # did NOT wedge + assert errs and "boom-approval" in str(errs[0]) + assert not gate.active + + +def test_gate_feed_exception_resolves_future_not_hang() -> None: + # The echo sink raises inside feed (before set_result) → submit forwards the + # error to the Future; the worker unblocks rather than hanging. + sink: list[object] = [] + gate = ApprovalGate(ui=_BoomUI(boom_on="echo"), schedule=sink.append, glyphs=UNICODE_GLYPHS) # type: ignore[arg-type] + thread, errs = _run_command_capturing(gate) + drain_enter(sink) + assert gate.active + gate.submit("run") + thread.join(2.0) + assert not thread.is_alive() + assert errs and "boom-echo" in str(errs[0]) + assert not gate.active diff --git a/tests/test_app_ui.py b/tests/test_app_ui.py index 30de687..937908e 100644 --- a/tests/test_app_ui.py +++ b/tests/test_app_ui.py @@ -323,12 +323,29 @@ def test_cache_invalidated_on_new_content() -> None: # --------------------------------------------------------------------------- -# Approval stubs — must raise, never silently default +# Approval prompt content (§31.16) — the pane block for the focus-swap gate # --------------------------------------------------------------------------- -def test_ask_approval_raises_not_implemented() -> None: - """ask_approval must raise NotImplementedError until branch 7 wires the focus-swap.""" +def test_show_approval_renders_diff_info_and_cwd() -> None: + ui = make_ui(workspace=Path("/tmp/ws")) + request = ApprovalRequest( + kind="command", + display="patch x.py", + risk=RiskLevel.HIGH, + reasons=("recursive delete",), + cwd=Path("/tmp/ws"), + diff="--- a/x.py\n+++ b/x.py\n@@ -1 +1 @@\n-oldline\n+newline\n", + ) + ui.show_approval(request) + text = plain(ui) + assert "HIGH" in text # risk badge + assert "recursive delete" in text # classifier reason + assert "CWD: /tmp/ws" in text # working directory + assert "oldline" in text and "newline" in text # the diff content rendered + + +def test_show_approval_without_diff_omits_panel() -> None: ui = make_ui() request = ApprovalRequest( kind="tool", @@ -337,15 +354,26 @@ def test_ask_approval_raises_not_implemented() -> None: reasons=("writes inside workspace",), cwd=Path("/tmp/ws"), ) - with pytest.raises(NotImplementedError): - ui.ask_approval(request) + ui.show_approval(request) + text = plain(ui) + assert "MEDIUM" in text + assert "writes inside workspace" in text + + +def test_show_plan_approval_renders_panel_and_path() -> None: + ui = make_ui() + ui.show_plan_approval(make_plan(), "/tmp/ws/PLAN.md") + text = plain(ui) + assert "Demo goal" in text # plan panel goal + assert "/tmp/ws/PLAN.md" in text # the artifact path -def test_ask_plan_approval_raises_not_implemented() -> None: - """ask_plan_approval must raise NotImplementedError until branch 7.""" +def test_show_plan_approval_sanitizes_path() -> None: ui = make_ui() - with pytest.raises(NotImplementedError): - ui.ask_plan_approval(make_plan(), "/tmp/ws/PLAN.md") + ui.show_plan_approval(make_plan(), "/tmp/ws/PL\x07AN.md") # embedded bell + text = plain(ui) + assert "\x07" not in text + assert "/tmp/ws/PLAN.md" in text # ---------------------------------------------------------------------------