diff --git a/aai_cli/code_agent/events.py b/aai_cli/code_agent/events.py index 4ee9136f..ed480bbd 100644 --- a/aai_cli/code_agent/events.py +++ b/aai_cli/code_agent/events.py @@ -21,6 +21,19 @@ class AssistantText: text: str +@dataclass(frozen=True) +class AssistantDelta: + """One streamed token of the in-progress reply, shown live then superseded by AssistantText. + + Emitted from langgraph's per-token ``messages`` stream so the front-end can render the + reply as it's generated; the authoritative full text still arrives as an AssistantText + when the step lands, so a consumer that ignores deltas (the headless renderer) loses + nothing. + """ + + text: str + + @dataclass(frozen=True) class ToolCall: """The agent's request to run a tool (announced when not gated by approval).""" @@ -44,7 +57,21 @@ class ErrorText: text: str -Event = AssistantText | ToolCall | ToolResult | ErrorText +Event = AssistantText | AssistantDelta | ToolCall | ToolResult | ErrorText + + +def assistant_delta(payload: object) -> AssistantDelta | None: + """Extract a streaming assistant-text token from a ``messages``-mode stream payload. + + langgraph's ``messages`` mode yields ``(message_chunk, metadata)``; we surface only the + AI message's text tokens (tool-call requests and tool results carry no prose, and other + message kinds aren't the assistant talking), so the live region streams just the reply. + """ + chunk = payload[0] if isinstance(payload, tuple) and payload else payload + if type(chunk).__name__ not in ("AIMessage", "AIMessageChunk"): + return None + text = _text_of(getattr(chunk, "content", "")) + return AssistantDelta(text) if text else None def _text_of(content: object) -> str: diff --git a/aai_cli/code_agent/messages.py b/aai_cli/code_agent/messages.py new file mode 100644 index 00000000..8bb1ad2d --- /dev/null +++ b/aai_cli/code_agent/messages.py @@ -0,0 +1,110 @@ +"""Mounted transcript widgets for the coding-agent TUI. + +The transcript is a ``VerticalScroll`` of these widgets rather than an append-only ``RichLog``, +which buys two things deepagents-code has: the assistant reply updates *in place* as it streams +(no separate live region), and a tool's output is a collapsible row — a clipped preview that +expands to the full output on Ctrl+O or a click. + +Dynamic content (model/tool/user strings) is wrapped in ``rich.text.Text`` so it's shown +literally — Text doesn't parse console markup, so a stray ``[`` can't raise or inject styling. +""" + +from __future__ import annotations + +from collections.abc import Mapping + +from rich.markdown import Markdown +from rich.text import Text +from textual.widgets import Static + +from aai_cli.code_agent.summarize import summarize_call, summarize_result + +_DIM = "#8a8f98" # muted gray for tool lines / notes +_ERROR = "#f04438" + + +class Note(Static): + """A dim one-line transcript aside (``cancelling…``, ``copied…``, ``voice off…``).""" + + def __init__(self, text: str) -> None: + super().__init__(Text(text, style=_DIM)) + + +class UserMessage(Static): + """The echoed user prompt, with a top margin so each turn is visually separated.""" + + DEFAULT_CSS = "UserMessage { margin-top: 1; }" + + def __init__(self, text: str) -> None: + super().__init__(Text(f"» {text}", style="bold #38bdf8")) + + +class AssistantMessage(Static): + """The assistant's reply: streams plain text token-by-token, then renders as Markdown.""" + + def __init__(self) -> None: + super().__init__() + self._tokens: list[str] = [] # accumulate tokens, not str +=, to avoid quadratic growth + + @property + def text(self) -> str: + """The reply text streamed so far (used to finalize a cancelled generation).""" + return "".join(self._tokens) + + def stream(self, delta: str) -> None: + """Append a streamed token and repaint as plain text (cheap; no per-token markdown).""" + self._tokens.append(delta) + self.update(Text(self.text)) + + def finalize(self, text: str) -> None: + """Replace the streamed text with the authoritative reply, rendered as Markdown.""" + self._tokens = [text] + self.update(Markdown(text)) + + +class ToolCallLine(Static): + """A compact tool-call line, e.g. ``→ write_file(app.py)``.""" + + def __init__(self, name: str, args: Mapping[str, object]) -> None: + super().__init__(Text(f"→ {summarize_call(name, args)}", style=_DIM)) + + +class ErrorMessage(Static): + """A failed turn, shown instead of crashing the UI.""" + + def __init__(self, text: str) -> None: + super().__init__(Text(f"✗ {text}", style=_ERROR)) + + +class ToolOutput(Static): + """A tool's output: a clipped preview that expands to the full content (Ctrl+O / click).""" + + def __init__(self, name: str, content: str) -> None: + super().__init__() + self._name = name + self._full = content.strip() + self._preview = summarize_result(content) + self._expandable = self._preview != self._full # nothing to expand when it fits already + self._expanded = False + + def on_mount(self) -> None: + self._repaint() + + def on_click(self) -> None: + self.toggle() + + def toggle(self) -> None: + """Flip between the clipped preview and the full output (no-op when it all fits).""" + if not self._expandable: + return + self._expanded = not self._expanded + self._repaint() + + def _repaint(self) -> None: + body = self._full if self._expanded else self._preview + line = Text(f" {self._name}: ", style=_DIM) + line.append(body, style=_DIM) + if self._expandable: + hint = " (Ctrl+O to collapse)" if self._expanded else " (Ctrl+O to expand)" + line.append(hint, style=f"{_DIM} italic") + self.update(line) diff --git a/aai_cli/code_agent/modals.py b/aai_cli/code_agent/modals.py new file mode 100644 index 00000000..25c54a7c --- /dev/null +++ b/aai_cli/code_agent/modals.py @@ -0,0 +1,202 @@ +"""Bottom-docked modal screens for the coding-agent TUI: tool approval and agent questions. + +Split out of `tui.py` to keep each module under the file-length gate. Both are transparent +``ModalScreen``s docked at the bottom, so the transcript stays visible above them (see the +``ModalScreen { background: transparent }`` rule in :class:`~aai_cli.code_agent.tui.CodeAgentApp`). + +In voice mode each modal is also **spoken and voice-answerable**: when constructed with a +``voice`` IO it speaks the prompt and listens for a spoken reply (approve / auto / reject, or a +free-text answer), off the UI thread. The keyboard path always stays available as a fallback. +""" + +from __future__ import annotations + +import re +import threading +from typing import TYPE_CHECKING, ClassVar + +from rich.markup import escape +from textual.app import ComposeResult +from textual.containers import Vertical +from textual.screen import ModalScreen +from textual.widgets import Input, Label + +from aai_cli.code_agent import banner, risk +from aai_cli.code_agent.summarize import describe_args, full_args +from aai_cli.core import errors + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + + from aai_cli.code_agent.voice_ui import _VoiceIO + + +def _spawn(target: Callable[[], None]) -> None: + """Run ``target`` on a daemon thread — the voice legs block, so they stay off the UI thread.""" + threading.Thread(target=target, daemon=True).start() # pragma: no mutate + + +# Spoken-answer vocabulary. "auto" wins first (it implies approval); an unclear answer falls +# back to "reject" — the same safe default as the keyboard, so a tool never runs on a guess. +_REJECT_WORDS = frozenset({"no", "reject", "deny", "stop", "cancel", "nope", "nah"}) +_APPROVE_WORDS = frozenset({"yes", "approve", "yeah", "yep", "yup", "sure", "ok", "okay"}) + + +def approval_from_speech(text: str) -> str: + """Map a spoken reply to ``"approve"`` / ``"auto"`` / ``"reject"`` (unclear → reject).""" + lowered = text.lower() + words = set(re.findall(r"[a-z]+", lowered)) + if "auto" in lowered or "always" in lowered: + return "auto" + if words & _REJECT_WORDS or "don't" in lowered or "do not" in lowered: + return "reject" + if words & _APPROVE_WORDS or "go ahead" in lowered or "do it" in lowered: + return "approve" + return "reject" + + +class ApprovalScreen(ModalScreen[str]): + """A compact, bottom-docked prompt to approve/auto-approve/reject one tool call. + + Keyboard ``y / a / n`` (and ``e`` to expand the args); in voice mode it also speaks the + prompt and accepts a spoken approve/auto/reject. The transparent background leaves the + transcript visible, and a risky call (``rm -rf``, an internal fetch) carries a warning. + """ + + DEFAULT_CSS = """ + ApprovalScreen { align: center bottom; background: transparent; } + ApprovalScreen #approvalbox { + dock: bottom; width: 1fr; height: auto; + border: round #f59e0b; background: #000000; padding: 0 1; margin: 0 1 1 1; + } + ApprovalScreen #approvalbox Label { height: auto; } + """ + BINDINGS: ClassVar = [ + ("y", "approve", "Approve"), + ("a", "auto", "Auto-approve"), + ("n", "reject", "Reject"), + ("e", "expand", "Expand"), + ] + + def __init__( + self, name: str, args: Mapping[str, object], *, voice: _VoiceIO | None = None + ) -> None: + super().__init__() + self._tool_name = name # not _name: that shadows Textual Widget's str|None attr + self._args = args + self._expanded = False # toggled by `e`; collapsed (one-line) by default + self._voice = voice # when set, the prompt is spoken and a spoken answer is accepted + self._answered = False # guards against a voice answer and a keypress both dismissing + + def compose(self) -> ComposeResult: + with Vertical(id="approvalbox"): + warning = risk.risk_warning(self._tool_name, self._args) + if warning: + yield Label(f"[b #f04438]⚠ {escape(warning)}[/]", id="approvalwarn") + yield Label(self._detail_markup(), id="approvaldetail") + yield Label( + f"[b #22c55e]y[/] approve [b {banner.BRAND_HEX}]a[/] auto-approve " + "[b #f04438]n[/] reject [b]e[/] expand" + ) + + def on_mount(self) -> None: + if (voice := self._voice) is not None: # drive the decision by voice, off the UI thread + _spawn(lambda: self._drive_by_voice(voice)) + + def _drive_by_voice(self, voice: _VoiceIO) -> None: + """Speak the prompt and accept a spoken approve/auto/reject (keyboard still works).""" + try: + voice.speak(self._spoken_prompt()) + transcript = voice.listen() + except errors.CLIError: + return # mic/STT failed: leave the keyboard hint as the way to answer + if transcript: # silence (None) must not auto-reject a tool — wait for speech or a key + self.app.call_from_thread(self._decide, approval_from_speech(transcript)) + + def _spoken_prompt(self) -> str: + """The read-aloud version of the prompt: the tool, its arg, any warning, the options.""" + parts = [f"Run {self._tool_name}."] + detail = describe_args(self._args) + if detail: + parts.append(f"{detail}.") + warning = risk.risk_warning(self._tool_name, self._args) + if warning: + parts.append(f"Warning: {warning}") + parts.append("Say approve, auto-approve, or reject.") + return " ".join(parts) + + def _decide(self, decision: str) -> None: + """Dismiss once, whether the answer came by spoken reply or keypress.""" + if self._answered: + return + self._answered = True + self.dismiss(decision) + + def _detail_markup(self) -> str: + """The 'Run tool X?' line — the compact arg, or the full args when expanded.""" + args = full_args(self._args) if self._expanded else describe_args(self._args) + return f"Run tool [b]{escape(self._tool_name)}[/b]? [dim]{escape(args)}[/dim]" + + def action_expand(self) -> None: + """Toggle between the compact identifying arg and the full args (``e``).""" + self._expanded = not self._expanded + self.query_one("#approvaldetail", Label).update(self._detail_markup()) + + def action_approve(self) -> None: + self._decide("approve") + + def action_auto(self) -> None: + self._decide("auto") + + def action_reject(self) -> None: + self._decide("reject") + + +class AskScreen(ModalScreen[str]): + """A bottom-docked prompt that relays a question from the agent and returns the answer. + + In voice mode it speaks the question and takes a spoken answer; otherwise the user types. + """ + + DEFAULT_CSS = """ + AskScreen { align: center bottom; background: transparent; } + AskScreen #askbox { + dock: bottom; width: 1fr; height: auto; + border: round #3a3f55; background: #000000; padding: 0 1; margin: 0 1 1 1; + } + """ + + def __init__(self, question: str, *, voice: _VoiceIO | None = None) -> None: + super().__init__() + self._question = question + self._voice = voice + self._answered = False + + def compose(self) -> ComposeResult: + with Vertical(id="askbox"): + yield Label(f"[b]The agent asks:[/b] {escape(self._question)}") + yield Input(id="answer", placeholder="Type your answer and press Enter…") + + def on_mount(self) -> None: + if (voice := self._voice) is not None: + _spawn(lambda: self._drive_by_voice(voice)) + + def _drive_by_voice(self, voice: _VoiceIO) -> None: + """Speak the question and submit a spoken answer (typing still works).""" + try: + voice.speak(f"The agent asks: {self._question}") + transcript = voice.listen() + except errors.CLIError: + return + if transcript: + self.app.call_from_thread(self._answer, transcript) + + def _answer(self, text: str) -> None: + """Dismiss once with the answer, whether spoken or typed.""" + if self._answered: + return + self._answered = True + self.dismiss(text) + + def on_input_submitted(self, event: Input.Submitted) -> None: + self._answer(event.value) diff --git a/aai_cli/code_agent/render.py b/aai_cli/code_agent/render.py index 499cd4e9..e0e7d639 100644 --- a/aai_cli/code_agent/render.py +++ b/aai_cli/code_agent/render.py @@ -9,20 +9,14 @@ from collections.abc import Callable +from rich.markdown import Markdown from rich.markup import escape from aai_cli.code_agent.events import AssistantText, ErrorText, Event, ToolCall, ToolResult from aai_cli.code_agent.session import Approver +from aai_cli.code_agent.summarize import summarize_call, summarize_result from aai_cli.ui import output -# Tool output can be long; clip it for the inline transcript. -_RESULT_PREVIEW = 2000 - - -def _format_args(args: dict[str, object]) -> str: - """A compact one-line view of a tool call's arguments.""" - return ", ".join(f"{key}={value!r}" for key, value in args.items()) - class RichRenderer: """An :data:`~aai_cli.code_agent.session.EventSink` that prints to the Rich console.""" @@ -31,13 +25,16 @@ def __call__(self, event: Event) -> None: # escape() dynamic content so a model/tool string with "[" can't inject Rich # markup or raise MarkupError (matches the inline-escape convention in output.py). if isinstance(event, AssistantText): - output.console.print(escape(event.text)) + # Render as Markdown so fenced code blocks are syntax-highlighted (and lists/ + # headings format) instead of showing raw ``` markers — Markdown parses its own + # syntax, not console markup, so no escape()/injection concern. + output.console.print(Markdown(event.text)) elif isinstance(event, ToolCall): output.console.print( - f"[aai.muted]→ {escape(event.name)}({escape(_format_args(event.args))})[/aai.muted]" + f"[aai.muted]→ {escape(summarize_call(event.name, event.args))}[/aai.muted]" ) elif isinstance(event, ToolResult): - preview = escape(event.content.strip()[:_RESULT_PREVIEW]) + preview = escape(summarize_result(event.content)) output.console.print(f"[aai.muted] {escape(event.name)}: {preview}[/aai.muted]") elif isinstance(event, ErrorText): output.error_console.print(output.fail(escape(event.text))) diff --git a/aai_cli/code_agent/risk.py b/aai_cli/code_agent/risk.py new file mode 100644 index 00000000..6c7b7e8e --- /dev/null +++ b/aai_cli/code_agent/risk.py @@ -0,0 +1,68 @@ +"""Heuristic risk flags for tool calls, surfaced on the approval prompt. + +The approval modal already shows *what* a tool will do; for the genuinely dangerous calls it +also shows *why to look twice* — a one-line warning, the way deepagents-code badges suspicious +shell commands and URLs. Purely advisory (the real SSRF guard lives in ``fetch_tool``); this +only nudges the human reviewing a manual approval. Pure functions so they unit-test cleanly. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping + +from aai_cli.code_agent.fetch_tool import FETCH_TOOL_NAME + +# Shell fragments that can destroy data, escalate privileges, or pipe a remote script straight +# into a shell — the classic "are you sure?" cases. Word-ish boundaries avoid matching inside +# innocuous longer tokens (e.g. ``format`` should not trip ``mkfs``). +_DANGEROUS_SHELL = ( + (re.compile(r"\brm\s+(-\w*\s+)*-\w*[rf]", re.I), "deletes files recursively/forcibly"), + (re.compile(r"\bsudo\b", re.I), "runs with elevated privileges"), + (re.compile(r"\bmkfs\b|\bdd\s+if=", re.I), "can overwrite a disk or filesystem"), + (re.compile(r":\s*\(\)\s*\{.*\|.*&\s*\}\s*;"), "looks like a fork bomb"), + ( + re.compile(r"\b(curl|wget)\b[^|]*\|\s*(sudo\s+)?(ba)?sh\b", re.I), + "pipes a download into a shell", + ), + (re.compile(r">\s*/dev/(sd|disk|nvme)", re.I), "writes directly to a block device"), +) +# URL hosts that mean a fetch is reaching a local/internal target rather than the public web. +_LOCAL_HOST = re.compile( + r"^(localhost|127\.|0\.0\.0\.0|10\.|192\.168\.|169\.254\.|172\.(1[6-9]|2\d|3[01])\.|\[?::1\]?)", + re.I, +) + + +def _shell_warning(command: str) -> str | None: + for pattern, reason in _DANGEROUS_SHELL: + if pattern.search(command): + return f"This command {reason}." + return None + + +def _url_warning(url: str) -> str | None: + stripped = url.strip() + if stripped.lower().startswith("file:"): + return "This URL reads a local file (file://)." + host = re.sub(r"^[a-z]+://", "", stripped, flags=re.I) + if _LOCAL_HOST.match(host): + return "This URL targets a local/internal address." + return None + + +def risk_warning(name: str, args: Mapping[str, object]) -> str | None: + """A one-line caution for a risky tool call, or ``None`` when nothing stands out. + + Flags destructive/privileged shell commands (``execute``) and fetches aimed at local or + ``file://`` targets; everything else returns ``None``. + """ + if name == "execute": + command = args.get("command") + if isinstance(command, str): + return _shell_warning(command) + elif name == FETCH_TOOL_NAME: + url = args.get("url") + if isinstance(url, str): + return _url_warning(url) + return None diff --git a/aai_cli/code_agent/session.py b/aai_cli/code_agent/session.py index 64b6c904..b3ce738f 100644 --- a/aai_cli/code_agent/session.py +++ b/aai_cli/code_agent/session.py @@ -18,6 +18,7 @@ from aai_cli.code_agent.events import ( ErrorText, Event, + assistant_delta, interrupt_request, message_events, new_messages, @@ -43,9 +44,18 @@ class _SupportsStream(Protocol): """ def stream( - self, graph_input: object, config: Mapping[str, object] | None, *, stream_mode: str - ) -> Iterator[dict[str, object]]: - """Yield the running state (incl. the growing ``messages``) after each super-step.""" + self, + graph_input: object, + config: Mapping[str, object] | None, + *, + stream_mode: list[str], + ) -> Iterator[tuple[str, object]]: + """Yield ``(mode, payload)`` pairs — ``"values"`` state snapshots and ``"messages"`` deltas. + + With a *list* ``stream_mode`` langgraph tags each yield with its mode, so the caller + can render off the per-super-step ``"values"`` state while still seeing the frequent + per-token ``"messages"`` deltas (used only as a fine-grained cancellation checkpoint). + """ @dataclass @@ -97,17 +107,28 @@ def send(self, text: str) -> None: def _run(self, graph_input: object, config: dict[str, object]) -> dict[str, object]: """Drive one graph segment, emitting events as each step completes; return the end state. - Streaming (``stream_mode="values"``) renders intermediate tool calls/results live and - lets :meth:`request_cancel` break the loop between steps. A double that only implements - ``invoke`` (the TUI/REPL test fakes) emits once at the end instead. + We render the finished messages from the per-super-step ``"values"`` snapshots, and + stream the ``"messages"`` (per-token) deltas alongside them for two reasons: a live + front-end shows the reply as it's generated (emitted as ``AssistantDelta``), and the + frequent deltas give :meth:`request_cancel` a checkpoint *within* a long step — a + single model generation is one super-step, so a values-only loop couldn't break until + the whole reply landed. A double that only implements ``invoke`` (the TUI/REPL test + fakes) emits once at the end instead. """ if isinstance(self.agent, _SupportsStream): last: dict[str, object] = {} - for chunk in self.agent.stream(graph_input, config, stream_mode="values"): + for mode, payload in self.agent.stream( + graph_input, config, stream_mode=["values", "messages"] + ): if self._cancel.is_set(): break - self._emit_new(chunk) - last = chunk + if mode == "values" and isinstance(payload, dict): + self._emit_new(payload) + last = payload + elif mode == "messages": + delta = assistant_delta(payload) + if delta is not None: + self.sink(delta) return last result = self.agent.invoke(graph_input, config) self._emit_new(result) diff --git a/aai_cli/code_agent/store.py b/aai_cli/code_agent/store.py index 7c0b2975..01b218da 100644 --- a/aai_cli/code_agent/store.py +++ b/aai_cli/code_agent/store.py @@ -8,6 +8,7 @@ from __future__ import annotations +import uuid from pathlib import Path from typing import TYPE_CHECKING @@ -18,6 +19,20 @@ _APP = "assemblyai" +# Length of a generated session id — short enough to read off the splash and retype as +# ``--session `` to resume, with ample uniqueness for one user's sessions. +_SESSION_ID_LEN = 12 + + +def new_session_id() -> str: + """A fresh, unique session id so each run starts a clean conversation by default. + + `assembly code` no longer reuses a fixed ``"default"`` thread (which silently resumed the + previous conversation); each run gets its own id unless ``--session NAME`` names one to + resume. Shown on the splash as ``Thread: `` so it can be resumed later. + """ + return uuid.uuid4().hex[:_SESSION_ID_LEN] + def sessions_db_path() -> Path: """Path to the SQLite file holding persisted coding sessions (dir created).""" diff --git a/aai_cli/code_agent/summarize.py b/aai_cli/code_agent/summarize.py new file mode 100644 index 00000000..ecb4a0c7 --- /dev/null +++ b/aai_cli/code_agent/summarize.py @@ -0,0 +1,96 @@ +"""Compact one-line summaries of tool activity, shared by both front-ends. + +A coding agent's tool args and output are routinely whole files or long command output. +Dumping them verbatim into the transcript buries the conversation — and, because args go +through ``repr``, renders literal ``\\n`` escapes. Both the Textual TUI (`tui.py`) and the +Rich fallback (`render.py`) route tool calls/results through these helpers so the +transcript stays scannable, mirroring how deepagents-code's collapsible tool rows show +just the identifying arg (a filename / command) and a short output preview with a +"+N more lines" tail rather than the full payload. +""" + +from __future__ import annotations + +from collections.abc import Mapping + +# Output preview budget (deepagents-code previews tool output at 4 lines / 300 chars behind +# an expand toggle; our append-only log has no expander, so we clip and tag the remainder). +_PREVIEW_LINES = 4 +_PREVIEW_CHARS = 300 +# Per-arg and arg-count caps so one giant value (a file's contents) can't flood the line. +_MAX_ARG_VALUE = 60 +_MAX_ARGS = 3 +# Per-value cap for the *expanded* approval view: values shown whole (newlines kept) but bounded +# so a multi-megabyte file can't make the modal unbounded. +_EXPANDED_VALUE = 1000 +# Args that identify a call on their own — show only this and elide bulky siblings (content). +_IDENTITY_ARGS = ("file_path", "path", "filename", "command", "url", "query", "pattern") + + +def _one_line(value: object, *, limit: int) -> str: + """Collapse ``value`` to a single clipped line (newlines → spaces, ellipsis if long).""" + text = " ".join(str(value).split()) + return text if len(text) <= limit else text[: limit - 1] + "…" + + +def describe_args(args: Mapping[str, object]) -> str: + """The compact arg view shared by the transcript line and the approval prompt. + + Prefers a single identifying arg (a path/command/URL) so a ``write_file`` reads as + ``app.py`` instead of inlining the file being written; otherwise shows up to a few + short ``key=value`` args, each clipped, with a trailing ``…`` when more were elided. + """ + for key in _IDENTITY_ARGS: + if key in args: + return _one_line(args[key], limit=_MAX_ARG_VALUE) + shown = list(args.items())[:_MAX_ARGS] + body = ", ".join(f"{key}={_one_line(value, limit=_MAX_ARG_VALUE)}" for key, value in shown) + if len(args) > _MAX_ARGS: + body = f"{body}, …" if body else "…" + return body + + +def summarize_call(name: str, args: Mapping[str, object]) -> str: + """A compact ``name(key arg)`` view of a tool call for the transcript.""" + return f"{name}({describe_args(args)})" + + +def full_args(args: Mapping[str, object]) -> str: + """The full ``key=value`` arg view shown when the approval prompt is expanded (``e``). + + Values are shown whole (newlines preserved) but each is capped at ``_EXPANDED_VALUE`` so a + huge file can't make the modal unbounded; :func:`describe_args` is the collapsed view. + """ + lines = [] + for key, value in args.items(): + text = str(value) + if len(text) > _EXPANDED_VALUE: + text = ( + f"{text[:_EXPANDED_VALUE].rstrip()} … (+{len(text) - _EXPANDED_VALUE} more chars)" + ) + lines.append(f"{key}={text}") + return "\n".join(lines) + + +def summarize_result(content: str) -> str: + """A short preview of tool output: the first few lines, clipped, with a hidden-count tail. + + Returns at most ``_PREVIEW_LINES`` lines and ``_PREVIEW_CHARS`` characters; when the + output was longer, appends ``… (+N more lines)`` (or ``… (+N more chars)`` when a single + long line was clipped) so the elision is visible rather than silent. + """ + text = content.strip() + if not text: + return "" + lines = text.splitlines() + preview_lines = lines[:_PREVIEW_LINES] + preview = "\n".join(preview_lines) + hidden_lines = len(lines) - len(preview_lines) + if len(preview) > _PREVIEW_CHARS: + kept = preview[:_PREVIEW_CHARS].rstrip() + hidden_chars = len(preview) - len(kept) + tail = f"+{hidden_lines} more lines" if hidden_lines else f"+{hidden_chars} more chars" + return f"{kept} … ({tail})" + if hidden_lines > 0: + return f"{preview} … (+{hidden_lines} more lines)" + return preview diff --git a/aai_cli/code_agent/tui.py b/aai_cli/code_agent/tui.py index cb699cf1..cc5a4010 100644 --- a/aai_cli/code_agent/tui.py +++ b/aai_cli/code_agent/tui.py @@ -13,177 +13,77 @@ import threading import time from pathlib import Path -from typing import TYPE_CHECKING, ClassVar, Protocol +from typing import TYPE_CHECKING, ClassVar from rich.markup import escape -from textual.app import App, ComposeResult -from textual.containers import Horizontal, Vertical +from textual.app import ComposeResult +from textual.containers import Horizontal, VerticalScroll from textual.screen import ModalScreen -from textual.widgets import Button, Input, Label, RichLog, Static +from textual.widgets import Input, Static from textual.worker import Worker from aai_cli.code_agent import banner from aai_cli.code_agent.agent import CompiledAgent from aai_cli.code_agent.ask_tool import AskBridge -from aai_cli.code_agent.events import AssistantText, ErrorText, Event, ToolCall, ToolResult +from aai_cli.code_agent.events import ( + AssistantDelta, + AssistantText, + ErrorText, + Event, + ToolCall, + ToolResult, +) +from aai_cli.code_agent.messages import ( + AssistantMessage, + ErrorMessage, + Note, + ToolCallLine, + ToolOutput, + UserMessage, +) +from aai_cli.code_agent.modals import ApprovalScreen, AskScreen from aai_cli.code_agent.session import CodeSession -from aai_cli.code_agent.voice import spoken_summary -from aai_cli.core import errors +from aai_cli.code_agent.tui_status import _spinner_text, _status_text +from aai_cli.code_agent.voice_ui import _VoiceIO, _VoiceLegs if TYPE_CHECKING: - from collections.abc import Callable, Mapping - from textual.timer import Timer # Glyphs cycled by the working indicator's animation (purely cosmetic). _SPIN_FRAMES = "✶✷✸✹✺" # pragma: no mutate # Seconds the Ctrl-C "press again to quit" hint stays armed (deepagents-code uses 3s too). _QUIT_HINT_SECONDS = 3 # pragma: no mutate +# Animated meter for the voice bar — a 3-cell block-char pulse (BMP, single-width, no emoji). +_VOICE_FRAMES = ("▁▃▅", "▃▅▇", "▅▇▆", "▆▇▅", "▇▅▃", "▅▃▁") # pragma: no mutate +# The three voice phases the bar distinguishes, each (label, accent color). +_VOICE_PHASES: dict[str, tuple[str, str]] = { + "listening": ("Listening — speak your request", banner.BRAND_HEX), + "thinking": ("Thinking…", "#f59e0b"), + "speaking": ("Speaking…", "#22c55e"), +} -class _VoiceIO(Protocol): - """The speak-to-it / read-back slice the TUI drives; :class:`VoiceSession` satisfies it.""" - - def listen(self) -> str | None: - """Capture one spoken turn and return its transcript (``None`` on no speech).""" - - def speak(self, text: str) -> None: - """Read ``text`` back aloud (a no-op when readback is unavailable).""" - - -def _format_args(args: Mapping[str, object]) -> str: - return ", ".join(f"{key}={value!r}" for key, value in args.items()) - - -def _spinner_text(elapsed_s: int, frame: str) -> str: - """The working-indicator line: a spinner glyph and the elapsed seconds.""" - return f"{frame} Working… ({elapsed_s}s)" - - -def _approval_decision(button_id: str | None) -> str: - """Map a pressed approval button's id to a decision, defaulting to reject if unset.""" - return button_id or "reject" - - -def _abbrev_home(path: Path) -> str: - """Render ``path`` with the home directory collapsed to ``~``.""" - try: - return f"~/{path.relative_to(Path.home())}" - except ValueError: - return str(path) - - -def _git_branch(start: Path) -> str | None: - """The current git branch for ``start`` (walking up to the repo root), or None.""" - for directory in (start, *start.parents): - head = directory / ".git" / "HEAD" - if head.is_file(): - ref = head.read_text(encoding="utf-8").strip() - return ref.removeprefix("ref: refs/heads/") if ref.startswith("ref: ") else ref[:8] - return None - - -def _status_text(cwd: Path, *, auto_approve: bool) -> str: - """The bottom status line: a mode badge, the working directory, and the git branch.""" - mode = "auto" if auto_approve else "manual" - badge = f"[black on #f59e0b] {mode} [/]" - parts = [badge, f"[dim]{_abbrev_home(cwd)}[/dim]"] - branch = _git_branch(cwd) - if branch: - parts.append(f"[dim]↗ {branch}[/dim]") - return " ".join(parts) - - -class ApprovalScreen(ModalScreen[str]): - """A compact, bottom-docked prompt to approve/auto-approve/reject one tool call. - - The transparent screen background leaves the transcript visible above (no full-screen - takeover); the decision is one of ``"approve"``, ``"auto"``, or ``"reject"``. - """ - - DEFAULT_CSS = """ - ApprovalScreen { align: center bottom; background: transparent; } - ApprovalScreen #approvalbox { - dock: bottom; width: 1fr; height: auto; - border: round #f59e0b; background: #000000; padding: 0 1; margin: 0 1 1 1; - } - ApprovalScreen #approvalbox Label { height: auto; } - ApprovalScreen #approvalbox Horizontal { height: auto; } - ApprovalScreen #approvalbox Button { margin: 0 1 0 0; } - """ - BINDINGS: ClassVar = [ - ("y", "approve", "Approve"), - ("a", "auto", "Auto-approve"), - ("n", "reject", "Reject"), - ] - - def __init__(self, name: str, args: Mapping[str, object]) -> None: - super().__init__() - self._tool_name = name # not _name: that shadows Textual Widget's str|None attr - self._args = args - - def compose(self) -> ComposeResult: - with Vertical(id="approvalbox"): - yield Label( - f"Run tool [b]{escape(self._tool_name)}[/b]? " - f"[dim]{escape(_format_args(self._args))}[/dim]" - ) - with Horizontal(): - yield Button("Approve (y)", id="approve", variant="success") - yield Button("Auto-approve (a)", id="auto", variant="primary") - yield Button("Reject (n)", id="reject", variant="error") - - def on_button_pressed(self, event: Button.Pressed) -> None: - self.dismiss(_approval_decision(event.button.id)) - - def action_approve(self) -> None: - self.dismiss("approve") - - def action_auto(self) -> None: - self.dismiss("auto") - - def action_reject(self) -> None: - self.dismiss("reject") - - -class AskScreen(ModalScreen[str]): - """A bottom-docked prompt that relays a question from the agent and returns the answer.""" - - DEFAULT_CSS = """ - AskScreen { align: center bottom; background: transparent; } - AskScreen #askbox { - dock: bottom; width: 1fr; height: auto; - border: round #3a3f55; background: #000000; padding: 0 1; margin: 0 1 1 1; - } - """ - - def __init__(self, question: str) -> None: - super().__init__() - self._question = question - - def compose(self) -> ComposeResult: - with Vertical(id="askbox"): - yield Label(f"[b]The agent asks:[/b] {escape(self._question)}") - yield Input(id="answer", placeholder="Type your answer and press Enter…") - - def on_input_submitted(self, event: Input.Submitted) -> None: - self.dismiss(event.value) - - -class CodeAgentApp(App[None]): +class CodeAgentApp(_VoiceLegs): """The coding-agent TUI: conversation transcript + prompt + approval/ask modals.""" # Flat pure-black canvas — no panel fills/gray, just the bordered prompt and a status # line, matching the deepagents-code look (wordmark in the AssemblyAI brand blue). CSS = f""" Screen {{ background: #000000; }} - #log {{ - height: 1fr; border: none; background: #000000; padding: 1 2; - scrollbar-size-vertical: 0; - }} + /* The approval/ask modals must stay see-through so the transcript shows above their + docked prompt. Their own DEFAULT_CSS sets `background: transparent`, but app CSS beats + a widget's DEFAULT_CSS — without this rule the `Screen` canvas above paints the modal + opaque black (it matches every Screen subclass) and blanks the transcript behind it. */ + ModalScreen {{ background: transparent; }} + /* The transcript is a scroll container of mounted message widgets (not a RichLog), so the + reply streams in place and tool output can expand/collapse. */ + #log {{ height: 1fr; border: none; background: #000000; padding: 1 2; }} #promptbar {{ dock: bottom; height: 3; background: #000000; border: round #3a3f55; margin: 1 1; }} #promptmark {{ width: 3; color: {banner.BRAND_HEX}; content-align: center middle; }} #prompt {{ border: none; background: #000000; padding: 0; }} + /* Shown in place of the prompt while voice capture is on (Ctrl-V brings the prompt back). */ + #voicebar {{ dock: bottom; height: 3; background: #000000; border: round {banner.BRAND_HEX}; + margin: 1 1; content-align: center middle; display: none; }} /* In normal flow below the 1fr log, so it sits just above the docked prompt bar. */ #spinner {{ height: 1; background: #000000; padding: 0 2; color: {banner.BRAND_HEX}; display: none; }} @@ -199,6 +99,8 @@ class CodeAgentApp(App[None]): ("ctrl+c", "quit_or_interrupt", "Interrupt / Quit"), ("ctrl+q", "quit", "Quit"), ("ctrl+y", "copy_last", "Copy last reply"), + ("ctrl+v", "toggle_voice", "Toggle voice"), + ("ctrl+o", "toggle_output", "Expand/collapse output"), ] def __init__( @@ -220,6 +122,12 @@ def __init__( self._initial = initial self._voice = voice # when set, spoken turns drive the prompt and replies are read back self._voice_typed = False # flips once the mic is ruled out; then input is typed only + self._voice_paused = False # user-toggled off via Ctrl-V (distinct from a mic failure) + self._voice_phase = "listening" # listening / thinking / speaking, shown in the voice bar + self._voice_frames = itertools.cycle(_VOICE_FRAMES) + self._voice_timer: Timer | None = None # animates the voice-bar meter while it's shown + self._streaming_msg: AssistantMessage | None = None # the reply widget tokens stream into + self._last_tool_output: ToolOutput | None = None # the row Ctrl+O expands/collapses self._session_name = thread_id # not _thread_id: that shadows Textual App's int self._cwd = cwd if cwd is not None else Path.cwd() self._web_note = web_note @@ -239,38 +147,63 @@ def __init__( def compose(self) -> ComposeResult: # No Header/Footer chrome — the splash is the title and the bottom status line # the only footer, so the screen stays a flat dark canvas. - yield RichLog(id="log", wrap=True, markup=True) + yield VerticalScroll(id="log") # Docked before the prompt bar, so the working indicator sits just above the input. yield Static("", id="spinner") with Horizontal(id="promptbar"): yield Static(">", id="promptmark") yield Input(id="prompt", placeholder="Ask the agent to build something…") - yield Static(_status_text(self._cwd, auto_approve=self._auto_approve), id="status") - - def _write_splash(self, log: RichLog) -> None: - for row in banner.wordmark(): - log.write(f"[bold {banner.BRAND_HEX}]{row}[/]") - log.write(f"[dim]{banner.version()}[/dim]") - log.write("") - log.write(f"[dim]Thread: {self._session_name}[/dim]") - log.write("") - log.write(f"[{banner.BRAND_HEX}]{banner.READY_LINE}[/]") - log.write(f"[dim]{banner.TIP_LINE}[/dim]") + yield Static("", id="voicebar") # filled by _render_voicebar when voice mode is shown + yield Static( + _status_text( + self._cwd, auto_approve=self._auto_approve, voice_state=self._voice_state() + ), + id="status", + ) + + def _write_splash(self) -> None: + # The whole splash is fixed copy except the session name, so this markup is safe to + # parse (only the session name — a --session value — is escaped). + rows = [f"[bold {banner.BRAND_HEX}]{row}[/]" for row in banner.wordmark()] + rows += [ + f"[dim]{banner.version()}[/dim]", + "", + f"[dim]Thread: {escape(self._session_name)}[/dim]", + "", + f"[{banner.BRAND_HEX}]{banner.READY_LINE}[/]", + f"[dim]{banner.TIP_LINE}[/dim]", + ] + self._mount("\n".join(rows)) + + def _mount(self, widget: Static | str) -> None: + """Append a transcript widget (or a markup string) and scroll it into view.""" + log = self.query_one("#log", VerticalScroll) + log.mount(Static(widget) if isinstance(widget, str) else widget) + log.scroll_end(animate=False) # pragma: no mutate — cosmetic; animate flag is unassertable + + def _note(self, text: str) -> None: + """Append a dim transcript aside (cancelling / copied / voice-off).""" + self._mount(Note(text)) def on_mount(self) -> None: # Route the agent's ask_user tool through a modal (the bridge is shared with # the tool built before this app existed). self._ask_bridge.handler = self._ask - self._write_splash(self.query_one("#log", RichLog)) + self._write_splash() if self._web_note: self.notify(self._web_note, title="Web search disabled", severity="warning", timeout=10) # Put the cursor in the prompt so the user can type immediately (RichLog would # otherwise hold focus and swallow keystrokes). self.query_one("#prompt", Input).focus() + self._sync_input_mode() # in voice mode, swap the prompt for the listening affordance if self._initial: self._submit(self._initial) else: - self._begin_listening() # in voice mode, capture the first spoken turn + # Defer the first mic open until *after* the splash has painted. Opening PortAudio + # is a GIL-holding C call; run inline on mount it races Textual's initial render and + # the banner never flushes — it stays blank until a resize/focus forces a full + # repaint. call_after_refresh runs once the screen is on-screen, so the splash wins. + self.call_after_refresh(self._begin_listening) # in voice mode, capture first turn # --- event rendering (always called on the UI thread) --------------------- @@ -279,18 +212,34 @@ def _emit_event(self, event: Event) -> None: self.call_from_thread(self._write_event, event) def _write_event(self, event: Event) -> None: - log = self.query_one("#log", RichLog) - # Escape dynamic content: a model/tool string containing "[" would otherwise be - # parsed as Rich markup and raise MarkupError (crashing the turn), or inject styling. - if isinstance(event, AssistantText): - self._last_reply = event.text - log.write(escape(event.text)) + if isinstance(event, AssistantDelta): + # Stream the token into the live reply widget (mounting one on the first token), + # updated in place until the authoritative AssistantText finalizes it below. + if self._streaming_msg is None: + self._streaming_msg = AssistantMessage() + self._mount(self._streaming_msg) + self._streaming_msg.stream(event.text) + self.query_one("#log", VerticalScroll).scroll_end(animate=False) # pragma: no mutate + elif isinstance(event, AssistantText): + self._last_reply = event.text # keep the raw text for clipboard copy + self._finalize_reply(event.text) elif isinstance(event, ToolCall): - log.write(f"[dim]→ {escape(event.name)}({escape(_format_args(event.args))})[/dim]") + self._mount(ToolCallLine(event.name, event.args)) elif isinstance(event, ToolResult): - log.write(f"[dim] {escape(event.name)}: {escape(event.content.strip()[:2000])}[/dim]") + self._last_tool_output = ToolOutput(event.name, event.content) + self._mount(self._last_tool_output) elif isinstance(event, ErrorText): - log.write(f"[#F04438]✗ {escape(event.text)}[/#F04438]") + self._mount(ErrorMessage(event.text)) + + def _finalize_reply(self, text: str) -> None: + """Commit the reply: finalize the streamed widget in place, or mount a fresh one.""" + if self._streaming_msg is not None: + self._streaming_msg.finalize(text) + self._streaming_msg = None + else: + msg = AssistantMessage() + self._mount(msg) + msg.finalize(text) def action_copy_last(self) -> None: """Copy the most recent assistant reply to the system clipboard.""" @@ -298,7 +247,12 @@ def action_copy_last(self) -> None: if self._last_reply: pyperclip.copy(self._last_reply) - self.query_one("#log", RichLog).write("[dim](copied last reply to clipboard)[/dim]") + self._note("(copied last reply to clipboard)") + + def action_toggle_output(self) -> None: + """Ctrl-O: expand/collapse the most recent tool output (a no-op if there's none).""" + if self._last_tool_output is not None: + self._last_tool_output.toggle() # --- approval / ask (called on the worker thread) ------------------------- @@ -324,7 +278,8 @@ def _approve(self, name: str, args: dict[str, object]) -> bool: """ if self._auto_approve: return True - decision = self._modal_result(ApprovalScreen(name, args), default="reject") + screen = ApprovalScreen(name, args, voice=self._modal_voice()) + decision = self._modal_result(screen, default="reject") if decision == "auto": self._enable_auto_approve() return True @@ -337,14 +292,80 @@ def _enable_auto_approve(self) -> None: self.call_from_thread(self._refresh_status) def _refresh_status(self) -> None: - """Re-render the bottom status line (e.g. after the mode flips to auto).""" + """Re-render the bottom status line (e.g. after the mode flips to auto or voice toggles).""" self.query_one("#status", Static).update( - _status_text(self._cwd, auto_approve=self._auto_approve) + _status_text( + self._cwd, auto_approve=self._auto_approve, voice_state=self._voice_state() + ) ) + def _voice_state(self) -> str | None: + """``"on"``/``"off"`` for the status badge, or ``None`` when voice isn't wired up.""" + if self._voice is None: + return None + return "on" if self._voice_active() else "off" + + def action_toggle_voice(self) -> None: + """Ctrl-V: turn spoken input/readback on or off for the session. + + A no-op notice when no voice front-end exists (e.g. a piped/typed run). Re-enabling + kicks off listening again unless a turn is mid-flight (the post-turn followup will). + """ + if self._voice is None: + self.notify("Voice isn't available in this session", severity="warning") + return + self._voice_paused = not self._voice_paused + self._refresh_status() + self._sync_input_mode() # show/hide the text box vs. the listening affordance + if self._voice_paused: + self.notify("Voice off — type your request") + elif not self._turn_running(): + self.notify("Voice on — listening") + self._begin_listening() + + def _sync_input_mode(self) -> None: + """Swap the text prompt for the 'listening' affordance while voice capture is active. + + The Input stays mounted either way (it still holds the spoken transcript and the + turn-running ``disabled`` flag); only the bars' visibility flips. The prompt regains + focus whenever it's the visible input. + """ + listening = self._voice_active() + self.query_one("#promptbar", Horizontal).display = not listening + self.query_one("#voicebar", Static).display = listening + if listening: + self._render_voicebar() + if self._voice_timer is None: # animate the meter only while the bar is shown + self._voice_timer = self.set_interval(0.3, self._tick_voice) # pragma: no mutate + else: + if self._voice_timer is not None: + self._voice_timer.stop() + self._voice_timer = None + self.query_one("#prompt", Input).focus() + + def _set_voice_phase(self, phase: str) -> None: + """Switch the voice bar between listening / thinking / speaking and repaint it.""" + self._voice_phase = phase + self._render_voicebar() + + def _render_voicebar(self) -> None: + """Paint the voice bar for the current phase: an animated meter, label, and accent.""" + label, color = _VOICE_PHASES[self._voice_phase] + meter = next(self._voice_frames) + hint = " [dim](Ctrl-V to type)[/dim]" if self._voice_phase == "listening" else "" + self.query_one("#voicebar", Static).update(f"[{color}]{meter}[/] {escape(label)}{hint}") + + def _tick_voice(self) -> None: + """Advance the voice-bar meter one frame (the animation timer's callback).""" + self._render_voicebar() + def _ask(self, question: str) -> str: """Block the worker on a modal input screen and return the user's answer.""" - return self._modal_result(AskScreen(question), default="") + return self._modal_result(AskScreen(question, voice=self._modal_voice()), default="") + + def _modal_voice(self) -> _VoiceIO | None: + """The voice IO to drive a modal by speech, or ``None`` when voice isn't active.""" + return self._voice if self._voice_active() else None # --- interrupt / quit ----------------------------------------------------- # Mirrors deepagents-code: Escape interrupts a running turn; Ctrl-C interrupts a running @@ -365,7 +386,7 @@ def _cancel_turn(self) -> bool: if not self._turn_running(): return False self._session.request_cancel() - self.query_one("#log", RichLog).write("[dim](cancelling…)[/dim]") + self._note("cancelling…") return True def action_interrupt(self) -> None: @@ -400,9 +421,9 @@ def on_input_submitted(self, event: Input.Submitted) -> None: self._submit(text) def _submit(self, text: str) -> None: - log = self.query_one("#log", RichLog) - log.write(f"[b cyan]» {escape(text)}[/b cyan]") + self._mount(UserMessage(text)) self.query_one("#prompt", Input).disabled = True + self._set_voice_phase("thinking") # voice bar reflects the turn (no-op when bar hidden) self._start_spinner() self._run_turn(text) @@ -414,8 +435,14 @@ def _run_turn(self, text: str) -> Worker[None]: # --- working indicator (spinner + elapsed) -------------------------------- def _start_spinner(self) -> None: - """Show the working indicator and animate it while the turn runs.""" + """Show the working indicator and animate it while the turn runs. + + Skipped in voice mode — the voice bar already shows a "Thinking…" state, so a second + spinner would just be redundant chrome. + """ self._turn_started = time.monotonic() + if self._voice_active(): + return self.query_one("#spinner", Static).display = True self._tick() self._spin_timer = self.set_interval(0.25, self._tick) # pragma: no mutate @@ -434,65 +461,16 @@ def _stop_spinner(self) -> None: def on_worker_state_changed(self, event: Worker.StateChanged) -> None: if event.worker.is_finished: - self._stop_spinner() - prompt = self.query_one("#prompt", Input) - prompt.disabled = False - prompt.focus() - self._voice_followup() # read a spoken summary back, then listen for the next turn - - # --- voice (speak-to-it / read-summary-back; the legs run off the UI thread) ---- - - def _voice_active(self) -> bool: - """Voice capture is on: a session exists and the mic hasn't been ruled out yet.""" - return self._voice is not None and not self._voice_typed - - def _spawn(self, target: Callable[[], None]) -> None: - """Run ``target`` on a daemon thread — voice legs block, so they stay off the UI thread.""" - threading.Thread(target=target, daemon=True).start() # pragma: no mutate - - def _begin_listening(self) -> None: - """Capture the next spoken turn on a background thread (no-op when voice is off).""" - if not self._voice_active(): - return - self._spawn(self._capture_voice_turn) - - def _voice_followup(self) -> None: - """After a turn finishes: read back a spoken summary, then listen for the next turn.""" - voice = self._voice - if voice is None: - return - self._spawn(lambda: self._speak_then_listen(voice)) - - def _speak_then_listen(self, voice: _VoiceIO) -> None: - """Read a summary of the last reply aloud (no code), then capture the next spoken turn.""" - voice.speak(spoken_summary(self._last_reply)) - self._capture_voice_turn() - - def _capture_voice_turn(self) -> None: - """Listen for one spoken turn; enter it into the prompt, or degrade to typing.""" - voice = self._voice - if voice is None or self._voice_typed: - return - try: - transcript = voice.listen() - except errors.CLIError as exc: - # A capture failure (no mic, STT error) drops voice for the rest of the session - # rather than wedging it — the user just types instead. - self._voice_typed = True - self.call_from_thread(self._notice_voice_off, exc.message) - return - if transcript: - self.call_from_thread(self._enter_and_submit, transcript) - - def _notice_voice_off(self, detail: str) -> None: - """Tell the user voice input stopped and that input is now typed (UI thread).""" - self.query_one("#log", RichLog).write( - f"[dim](voice input off: {escape(detail)}; type your request instead)[/dim]" - ) - - def _enter_and_submit(self, text: str) -> None: - """Show the spoken text in the prompt, then submit it as a turn (UI thread).""" - prompt = self.query_one("#prompt", Input) - prompt.value = text - self._submit(text) - prompt.value = "" + self._finish_turn() + + def _finish_turn(self) -> None: + """Wind down a completed turn: stop the spinner, re-enable input, resume voice.""" + self._stop_spinner() + if self._streaming_msg is not None: # a cancelled generation: keep what streamed in + self._finalize_reply(self._streaming_msg.text) + self.query_one("#prompt", Input).disabled = False + self._sync_input_mode() # focus the prompt (text mode) or show the listening bar + self._voice_followup() # read a spoken summary back, then listen for the next turn + + # The off-thread voice legs (_voice_active, _begin_listening, _capture_voice_turn, …) are + # inherited from _VoiceLegs; the render/toggle side stays above. diff --git a/aai_cli/code_agent/tui_status.py b/aai_cli/code_agent/tui_status.py new file mode 100644 index 00000000..5e385b55 --- /dev/null +++ b/aai_cli/code_agent/tui_status.py @@ -0,0 +1,51 @@ +"""Pure text helpers for the coding-agent TUI's status line and working indicator. + +Split out of `tui.py` (to keep it under the file-length gate) and free of any Textual +imports, so they unit-test as plain functions. +""" + +from __future__ import annotations + +from pathlib import Path + + +def _spinner_text(elapsed_s: int, frame: str) -> str: + """The working-indicator line: a spinner glyph and the elapsed seconds.""" + return f"{frame} Working… ({elapsed_s}s)" + + +def _abbrev_home(path: Path) -> str: + """Render ``path`` with the home directory collapsed to ``~``.""" + try: + return f"~/{path.relative_to(Path.home())}" + except ValueError: + return str(path) + + +def _git_branch(start: Path) -> str | None: + """The current git branch for ``start`` (walking up to the repo root), or None.""" + for directory in (start, *start.parents): + head = directory / ".git" / "HEAD" + if head.is_file(): + ref = head.read_text(encoding="utf-8").strip() + return ref.removeprefix("ref: refs/heads/") if ref.startswith("ref: ") else ref[:8] + return None + + +def _status_text(cwd: Path, *, auto_approve: bool, voice_state: str | None = None) -> str: + """The bottom status line: a mode badge, the working directory, git branch, and voice state. + + ``voice_state`` is ``"on"``/``"off"`` when the session has a voice front-end (so the + Ctrl-V toggle shows its effect), or ``None`` when voice isn't wired up at all. + """ + mode = "auto" if auto_approve else "manual" + badge = f"[black on #f59e0b] {mode} [/]" + parts = [badge, f"[dim]{_abbrev_home(cwd)}[/dim]"] + branch = _git_branch(cwd) + if branch: + parts.append(f"[dim]↗ {branch}[/dim]") + if voice_state is not None: + # A filled/hollow dot (BMP glyphs, like the rest of the UI — no double-width emoji). + glyph, color = ("●", "#22c55e") if voice_state == "on" else ("○", "#6b7280") + parts.append(f"[{color}]{glyph} voice {voice_state}[/]") + return " ".join(parts) diff --git a/aai_cli/code_agent/voice_ui.py b/aai_cli/code_agent/voice_ui.py new file mode 100644 index 00000000..0fcaf531 --- /dev/null +++ b/aai_cli/code_agent/voice_ui.py @@ -0,0 +1,126 @@ +"""The voice front-end legs for the coding-agent TUI, split out to keep `tui.py` small. + +These are the speak-to-it / read-back mechanics that run *off* the UI thread (mic capture and +TTS readback block), marshaling back via ``call_from_thread``. They live in a mixin that +:class:`~aai_cli.code_agent.tui.CodeAgentApp` inherits, so the app stays one ``App`` with the +voice methods folded in. The render/toggle side (the voice bar, Ctrl-V) stays in `tui.py`. +""" + +from __future__ import annotations + +import threading +from typing import TYPE_CHECKING, Protocol + +from textual.app import App +from textual.widgets import Input + +from aai_cli.code_agent.voice import spoken_summary +from aai_cli.core import errors + +if TYPE_CHECKING: + from collections.abc import Callable + + +class _VoiceIO(Protocol): + """The speak-to-it / read-back slice the TUI drives; :class:`VoiceSession` satisfies it.""" + + def listen(self) -> str | None: + """Capture one spoken turn and return its transcript (``None`` on no speech).""" + + def speak(self, text: str) -> None: + """Read ``text`` back aloud (a no-op when readback is unavailable).""" + + +class _VoiceLegs(App[None]): + """Mixin holding the off-thread voice capture/readback legs for ``CodeAgentApp``. + + Extends ``App`` so the inherited ``query_one``/``call_from_thread`` are typed; the voice + state and the few app methods it leans on (``_set_voice_phase``/``_sync_input_mode``/ + ``_submit``) are provided by the concrete app and declared here for the type checker. + """ + + if TYPE_CHECKING: # provided by CodeAgentApp (state set in __init__, methods defined there) + _voice: _VoiceIO | None + _voice_typed: bool + _voice_paused: bool + _last_reply: str + + def _set_voice_phase(self, phase: str) -> None: ... + def _sync_input_mode(self) -> None: ... + def _submit(self, text: str) -> None: ... + def _note(self, text: str) -> None: ... + + def _voice_active(self) -> bool: + """Voice capture is on: a session exists, the mic isn't ruled out, and it isn't paused.""" + return self._voice is not None and not self._voice_typed and not self._voice_paused + + def _spawn(self, target: Callable[[], None]) -> None: + """Run ``target`` on a daemon thread — voice legs block, so they stay off the UI thread.""" + thread = threading.Thread( + target=lambda: self._run_leg(target), + daemon=True, # pragma: no mutate — daemon flag only affects process exit, unassertable + ) + thread.start() + + def _run_leg(self, target: Callable[[], None]) -> None: + """Run one voice leg, dropping the callback error a torn-down app raises mid-flight. + + A leg calls back onto the UI thread (``call_from_thread``); if the app stops — a quit, + or a test's ``run_test`` block exiting — while the leg is mid-call, that callback raises + ``RuntimeError`` in this daemon thread, which would otherwise surface as an unhandled + thread exception (a flaky Windows CI failure). The spoken turn is moot once the app is + gone, so swallow it then; a genuine failure while the app is still live still propagates. + """ + try: + target() + except Exception: + if self.is_running: + raise + + def _begin_listening(self) -> None: + """Capture the next spoken turn on a background thread (no-op when voice is off).""" + if not self._voice_active(): + return + self._spawn(self._capture_voice_turn) + + def _voice_followup(self) -> None: + """After a turn finishes: read back a spoken summary, then listen for the next turn.""" + voice = self._voice + if voice is None or self._voice_paused: # paused via Ctrl-V: no readback, no listen + return + self._spawn(lambda: self._speak_then_listen(voice)) + + def _speak_then_listen(self, voice: _VoiceIO) -> None: + """Read a summary of the last reply aloud (no code), then capture the next spoken turn.""" + self.call_from_thread(self._set_voice_phase, "speaking") + voice.speak(spoken_summary(self._last_reply)) + self._capture_voice_turn() + + def _capture_voice_turn(self) -> None: + """Listen for one spoken turn; enter it into the prompt, or degrade to typing.""" + voice = self._voice + if voice is None or self._voice_typed or self._voice_paused: + return + self.call_from_thread(self._set_voice_phase, "listening") + try: + transcript = voice.listen() + except errors.CLIError as exc: + # A capture failure (no mic, STT error) drops voice for the rest of the session + # rather than wedging it — the user just types instead. + self._voice_typed = True + self.call_from_thread(self._notice_voice_off, exc.message) + return + if transcript: + self.call_from_thread(self._enter_and_submit, transcript) + + def _notice_voice_off(self, detail: str) -> None: + """Tell the user voice input stopped and that input is now typed (UI thread).""" + self._note(f"voice input off: {detail}; type your request instead") + self._sync_input_mode() # mic ruled out -> bring the text box back + + def _enter_and_submit(self, text: str) -> None: + """Show the spoken text in the prompt, then submit it as a turn (UI thread).""" + prompt = self.query_one("#prompt", Input) + prompt.value = text + self._submit(text) + prompt.value = "" diff --git a/aai_cli/commands/code/__init__.py b/aai_cli/commands/code/__init__.py index b37052d0..9d71d404 100644 --- a/aai_cli/commands/code/__init__.py +++ b/aai_cli/commands/code/__init__.py @@ -6,6 +6,7 @@ from aai_cli import command_registry, help_panels from aai_cli.app.context import run_with_options +from aai_cli.code_agent import store from aai_cli.code_agent.prompt import DEFAULT_MODEL from aai_cli.commands.code import _exec as code_exec from aai_cli.core import llm as gateway @@ -62,8 +63,10 @@ def code( memory: bool = typer.Option( True, "--memory/--no-memory", help="Load and persist the agent's long-term memory" ), - session: str = typer.Option( - "default", "--session", help="Conversation session name (reuse to resume it)" + session: str | None = typer.Option( + None, + "--session", + help="Resume a named session. Default: a new unique session each run", ), persist: bool = typer.Option( True, "--persist/--fresh", help="Persist the session to disk (--fresh: ephemeral)" @@ -98,7 +101,9 @@ def code( skills=skills, web=web, memory=memory, - session=session, + # No --session given -> a fresh unique id, so each run starts a clean conversation + # instead of silently resuming the previous one. + session=session if session is not None else store.new_session_id(), persist=persist, tui=tui, voice=voice, diff --git a/aai_cli/core/microphone.py b/aai_cli/core/microphone.py index 357ca03c..4ec65dda 100644 --- a/aai_cli/core/microphone.py +++ b/aai_cli/core/microphone.py @@ -9,6 +9,7 @@ from types import ModuleType from typing import Any, Protocol, cast +from aai_cli.core import stdio from aai_cli.core.errors import CLIError with warnings.catch_warnings(): @@ -20,6 +21,8 @@ # Used when the device's native rate can't be determined (e.g. headless CI). _FALLBACK_RATE = 48000 +# Channel count for the multichannel-input fallback: capture stereo, then downmix to mono. +_STEREO_CHANNELS = 2 class _RawInputStream(Protocol): @@ -122,7 +125,11 @@ def default_rate(kind: str, device: int | None = None) -> int: """ sd = _sounddevice() try: - raw_rate = sd.query_devices(device, kind).get("default_samplerate", _FALLBACK_RATE) + # query_devices triggers PortAudio's lazy init, which prints device-probe noise to + # the C-level stderr; suppress it so a TUI mic-open can't corrupt the rendered screen. + with stdio.suppress_native_stderr(): + devices = sd.query_devices(device, kind) + raw_rate = devices.get("default_samplerate", _FALLBACK_RATE) if not isinstance(raw_rate, str | int | float): return _FALLBACK_RATE rate = int(float(raw_rate)) @@ -144,35 +151,100 @@ def resample_pcm16(chunk: bytes, state: Any, *, src_rate: int, dst_rate: int) -> class _SoundDeviceMic: """Iterator of PCM16 byte chunks from a sounddevice raw input stream. - Yields ~100 ms blocks; closeable so MicrophoneSource can tear it down. + Yields ~100 ms blocks; closeable so MicrophoneSource can tear it down. When opened with + ``channels=2`` (the multichannel-input fallback below), each interleaved stereo block is + downmixed to mono so downstream — resampling and the STT stream — always sees one channel. """ - def __init__(self, stream: _RawInputStream, blocksize: int) -> None: + def __init__(self, stream: _RawInputStream, blocksize: int, *, channels: int = 1) -> None: self._stream = stream self._blocksize = blocksize + self._channels = channels def __iter__(self) -> Iterator[bytes]: return self def __next__(self) -> bytes: data, _overflowed = self._stream.read(self._blocksize) - return bytes(data) + pcm = bytes(data) + if self._channels == _STEREO_CHANNELS: + # Average L/R into a single channel (width=2 → int16). + pcm = audioop.tomono(pcm, 2, 0.5, 0.5) + return pcm def close(self) -> None: self._stream.stop() self._stream.close() +def _open_input_stream( + sd: _SoundDeviceModule, *, sample_rate: int, device: int | None, channels: int, blocksize: int +) -> _RawInputStream: + """Open and start a started PCM16 input stream at ``channels`` channels. + + Wrapped in ``suppress_native_stderr`` because opening/starting is PortAudio's stderr-noisy + moment — kept off the terminal so a TUI mic-open can't corrupt the rendered screen. + """ + with stdio.suppress_native_stderr(): + stream = sd.RawInputStream( + samplerate=sample_rate, + device=device, + channels=channels, + dtype="int16", + blocksize=blocksize, + ) + stream.start() + return stream + + +def _max_input_channels(sd: _SoundDeviceModule, device: int | None) -> int: + """The device's advertised input-channel count (0 when it exposes no input).""" + with stdio.suppress_native_stderr(): + info = sd.query_devices(device, "input") + raw = info.get("max_input_channels", 0) + return raw if isinstance(raw, int) else 0 + + def _default_mic_stream(*, sample_rate: int, device: int | None) -> Iterator[bytes]: - """A sounddevice-backed PCM16 mic stream (imported lazily to keep startup fast).""" - sd = _sounddevice() + """A sounddevice-backed PCM16 mono mic stream (imported lazily to keep startup fast). + Tries a mono open first. PortAudio rejects ``channels=1`` (``-9998``) when the device + exposes no usable mono input: either it has zero input channels (no mic permission, or the + default input isn't a microphone) — which no channel count can fix, so we raise an + actionable error — or it's a multichannel-only input, which we reopen at stereo and + downmix. Devices that already do mono never reach the fallback. + """ + sd = _sounddevice() blocksize = max(1, sample_rate // 10) # ~100 ms per read - stream = sd.RawInputStream( - samplerate=sample_rate, device=device, channels=1, dtype="int16", blocksize=blocksize - ) - stream.start() - return _SoundDeviceMic(stream, blocksize) + try: + return _SoundDeviceMic( + _open_input_stream( + sd, sample_rate=sample_rate, device=device, channels=1, blocksize=blocksize + ), + blocksize, + ) + except Exception: + max_in = _max_input_channels(sd, device) + if max_in < 1: + raise CLIError( + "The default microphone reports no input channels.", + error_type="mic_error", + exit_code=1, + suggestion=( + "Grant microphone access to your terminal in System Settings > Privacy & " + "Security > Microphone, or pick another input with --device." + ), + ) from None + if max_in < _STEREO_CHANNELS: + raise # a 1-channel device should accept mono; surface the real PortAudio error + stream = _open_input_stream( + sd, + sample_rate=sample_rate, + device=device, + channels=_STEREO_CHANNELS, + blocksize=blocksize, + ) + return _SoundDeviceMic(stream, blocksize, channels=_STEREO_CHANNELS) class MicrophoneSource: @@ -212,6 +284,8 @@ def __iter__(self) -> Iterator[bytes]: stream: Any = self._factory(sample_rate=self._capture_rate, device=self.device) except ImportError as exc: raise audio_missing_error() from exc + except CLIError: + raise # the factory already raised an actionable error; don't bury it in a re-wrap except Exception as exc: # "device None" reads like a bug; name the default mic in plain words. target = ( diff --git a/aai_cli/core/stdio.py b/aai_cli/core/stdio.py index 05db2ef3..f6905644 100644 --- a/aai_cli/core/stdio.py +++ b/aai_cli/core/stdio.py @@ -3,7 +3,57 @@ import contextlib import os import sys -from collections.abc import Iterator +from collections.abc import Generator, Iterator + +# The OS-level stderr descriptor. Used as a literal rather than ``sys.stderr.fileno()`` +# because inside a Textual app ``sys.stderr`` is swapped for a redirector whose ``fileno()`` +# returns an unusable fd — duping that raised EBADF and broke the very mic open we were +# trying to quiet. fd 2 is the real inherited stderr, which Textual never touches. +_STDERR_FD = 2 + + +@contextlib.contextmanager +def suppress_native_stderr() -> Generator[None]: + """Send OS-level stderr to /dev/null for the block, then restore it. + + Catches diagnostics that C extensions write straight to fd 2 via ``fprintf`` — + PortAudio/CoreAudio/ALSA print device-probe noise there on the first audio call, + *below* Python's logging, so silencing loggers can't reach it. Inside a full-screen + TUI (which draws to stdout and never repaints stderr) those raw writes scribble over + the rendered screen; the mic-open path wraps its PortAudio calls in this so they land + in the void instead. + + Safe by construction: if the descriptor can't be duplicated/redirected for any reason, + the block runs with stderr untouched rather than raising — suppression is cosmetic and + must never break the operation it wraps. Exceptions from the body propagate normally + (only the fd is redirected, not raised errors). + """ + saved_fd: int | None = None + devnull_fd: int | None = None + try: + saved_fd = os.dup(_STDERR_FD) + devnull_fd = os.open(os.devnull, os.O_WRONLY) + os.dup2(devnull_fd, _STDERR_FD) + except OSError: + # Couldn't redirect — abandon suppression, never break the caller. + _close_quietly(saved_fd) + _close_quietly(devnull_fd) + yield + return + try: + yield + finally: + with contextlib.suppress(OSError): + os.dup2(saved_fd, _STDERR_FD) # restore the real stderr + _close_quietly(saved_fd) + _close_quietly(devnull_fd) + + +def _close_quietly(fd: int | None) -> None: + """Close ``fd`` if it was opened, ignoring an already-closed/invalid descriptor.""" + if fd is not None: + with contextlib.suppress(OSError): + os.close(fd) def silence_stdout() -> None: diff --git a/tests/__snapshots__/test_snapshots_help_run.ambr b/tests/__snapshots__/test_snapshots_help_run.ambr index cdc75e01..23ab4f37 100644 --- a/tests/__snapshots__/test_snapshots_help_run.ambr +++ b/tests/__snapshots__/test_snapshots_help_run.ambr @@ -305,9 +305,8 @@ │ --memory --no-memory Load and persist the agent's │ │ long-term memory │ │ [default: memory] │ - │ --session TEXT Conversation session name (reuse to │ - │ resume it) │ - │ [default: default] │ + │ --session TEXT Resume a named session. Default: a │ + │ new unique session each run │ │ --persist --fresh Persist the session to disk │ │ (--fresh: ephemeral) │ │ [default: persist] │ diff --git a/tests/test_code_agent.py b/tests/test_code_agent.py index 739285bf..0e5d17c5 100644 --- a/tests/test_code_agent.py +++ b/tests/test_code_agent.py @@ -166,6 +166,13 @@ def test_checkpointer_in_memory_vs_sqlite(tmp_path, monkeypatch): # untyped: to saver.conn.close() +def test_new_session_id_is_unique_and_short() -> None: + a = store.new_session_id() + b = store.new_session_id() + assert a != b # each run gets its own thread id (no silent resume of a shared default) + assert len(a) == 12 and a.isalnum() # short hex, readable off the splash to resume later + + def test_cli_tool_invokes_runner_with_args() -> None: captured: list[list[str]] = [] @@ -337,49 +344,6 @@ def invoke(self, *a, **k): assert any(isinstance(e, ErrorText) and "gateway 500" in e.text for e in seen) -class StreamingAgent: - """A double exercising the streaming path: yields scripted state snapshots.""" - - def __init__(self, chunks: list[dict[str, object]]) -> None: - self._chunks = chunks - - def stream(self, graph_input, config=None, *, stream_mode="values"): - del graph_input, config, stream_mode - yield from self._chunks - - def invoke(self, *a, **k): # the streaming branch is taken, so invoke is never used - raise AssertionError("a streaming agent must not be invoked") - - -def test_send_streams_each_step_and_cancel_stops_the_loop() -> None: - from langchain_core.messages import HumanMessage - - # Three successive graph states (messages grow by one each step); a stream_mode="values" - # graph yields exactly these snapshots, so the session must emit incrementally. - chunks: list[dict[str, object]] = [ - {"messages": [HumanMessage("go")]}, - {"messages": [HumanMessage("go"), AIMessage("first")]}, - {"messages": [HumanMessage("go"), AIMessage("first"), AIMessage("second")]}, - ] - seen: list[object] = [] - session = CodeSession( - agent=StreamingAgent(chunks), sink=seen.append, approver=lambda n, a: True - ) - - def sink(event: object) -> None: - seen.append(event) - if isinstance(event, AssistantText) and event.text == "first": - session.request_cancel() # cancel mid-stream, before the "second" chunk is consumed - - session.sink = sink - session.send("go") - - texts = [e.text for e in seen if isinstance(e, AssistantText)] - # "first" streamed out as its step landed; the cancel then broke the loop, so the later - # "second" step was never emitted — proving both incremental rendering and cancellation. - assert texts == ["first"] - - def test_session_propagates_keyboard_interrupt() -> None: class Stop: def invoke(self, *a, **k): diff --git a/tests/test_code_command.py b/tests/test_code_command.py index a4384db9..b548cd6e 100644 --- a/tests/test_code_command.py +++ b/tests/test_code_command.py @@ -41,7 +41,20 @@ def test_command_parses_flags_into_options(monkeypatch): opts = captured["o"] assert opts.prompt == "build a thing" assert opts.auto is True and opts.web is False - assert opts.session == "s1" and opts.persist is False + assert opts.session == "s1" and opts.persist is False # an explicit --session is honored + + +def test_command_defaults_to_a_fresh_unique_session_each_run(monkeypatch): + # No --session: each invocation gets its own id (so a run never silently resumes the + # previous conversation), and two runs differ. + seen = [] + monkeypatch.setattr( + _exec, "run_code", lambda opts, state, *, json_mode: seen.append(opts.session) + ) + assert runner.invoke(app, ["code"]).exit_code == 0 + assert runner.invoke(app, ["code"]).exit_code == 0 + assert seen[0] != "default" # not the old shared, auto-resumed thread + assert seen[0] and seen[1] and seen[0] != seen[1] # a distinct id per run def test_run_code_dispatches_to_tui_with_voice_by_default_when_tty(monkeypatch): diff --git a/tests/test_code_messages.py b/tests/test_code_messages.py new file mode 100644 index 00000000..9a1168d4 --- /dev/null +++ b/tests/test_code_messages.py @@ -0,0 +1,149 @@ +"""Tests for the mounted-widget transcript of the `assembly code` TUI. + +Drives the real Textual app (headless) and asserts on the mounted message widgets: the reply +streams into one AssistantMessage in place and renders as Markdown, and a long tool result is +a collapsible ToolOutput (Ctrl-O / click). Split from test_code_tui.py to stay under the +file-length gate. +""" + +from __future__ import annotations + +import asyncio + +from aai_cli.code_agent.events import AssistantDelta, AssistantText, ToolResult +from aai_cli.code_agent.messages import AssistantMessage, ToolOutput +from aai_cli.code_agent.tui import CodeAgentApp + + +class FakeAgent: + """Replays scripted invoke() results so a turn can complete without a model.""" + + def __init__(self, results: list[dict[str, object]]) -> None: + self._results = results + self.calls = 0 + + def invoke(self, *args, **kwargs): + result = self._results[self.calls] + self.calls += 1 + return result + + +def _run(coro) -> None: + asyncio.run(coro) + + +def test_assistant_reply_renders_as_markdown_widget() -> None: + # The reply mounts an AssistantMessage rendered as Markdown — the fence markers are + # consumed and the code shows; the raw text is kept for clipboard copy. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + reply = "Here you go:\n\n```python\nprint('hi')\n```" + app._write_event(AssistantText(reply)) + await pilot.pause() + msg = app.query_one(AssistantMessage) + text = "\n".join(msg.render_line(y).text for y in range(msg.size.height)) + assert "```" not in text # markdown consumed the fence markers + assert "print('hi')" in text # the code itself renders + assert app._last_reply == reply # raw markdown kept for clipboard copy + + _run(go()) + + +def test_assistant_deltas_stream_in_place_then_finalize() -> None: + # Tokens stream into a single AssistantMessage in place (no separate region); the final + # AssistantText finalizes that same widget rather than mounting a second one. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._write_event(AssistantDelta("Hello, ")) + app._write_event(AssistantDelta("world!")) + await pilot.pause() + assert len(app.query(AssistantMessage)) == 1 # one widget, updated in place + assert app.query_one(AssistantMessage).text == "Hello, world!" + streaming = app._streaming_msg # local: asserting on the attr would poison the + assert streaming is not None # later `is None` check (mypy can't see the reset) + app._write_event(AssistantText("Hello, world!")) + await pilot.pause() + assert app._streaming_msg is None # finalized + assert app._last_reply == "Hello, world!" + assert len(app.query(AssistantMessage)) == 1 # finalized in place, not a 2nd widget + + _run(go()) + + +def test_finish_turn_finalizes_a_dangling_streamed_reply() -> None: + # A turn cancelled mid-generation leaves a streamed-but-unfinalized reply; finishing the + # turn commits what streamed in (so it isn't lost) and clears the streaming reference. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._write_event(AssistantDelta("partial repl")) + await pilot.pause() + streaming = app._streaming_msg # local so the later `is None` check stays reachable + assert streaming is not None + app._finish_turn() + assert app._streaming_msg is None # finalized, not left dangling + assert app.query_one(AssistantMessage).text == "partial repl" # kept what streamed + + _run(go()) + + +def test_short_tool_output_is_not_expandable() -> None: + # Output that already fits has no expand affordance and Ctrl-O is a no-op on it. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._write_event(ToolResult(name="execute", content="ok")) + await pilot.pause() + out = app.query_one(ToolOutput) + before = str(out.render()) + assert "Ctrl+O" not in before # nothing to expand -> no hint + out.toggle() + assert str(out.render()) == before # toggle is a no-op when it all fits + + _run(go()) + + +def test_tool_output_toggles_on_click_and_ctrl_o_is_safe_with_no_output() -> None: + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app.action_toggle_output() # no tool output yet -> safe no-op + app._write_event( + ToolResult(name="execute", content="\n".join(f"x{i}" for i in range(20))) + ) + await pilot.pause() + out = app.query_one(ToolOutput) + assert "x19" not in str(out.render()) + out.on_click() # clicking expands + assert "x19" in str(out.render()) + + _run(go()) + + +def test_tool_output_expands_and_collapses_on_ctrl_o() -> None: + # A long tool result mounts a collapsed ToolOutput (preview + "more lines"); Ctrl-O + # expands it to the full content and toggles back. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._write_event( + ToolResult(name="execute", content="\n".join(f"ln{i}" for i in range(20))) + ) + await pilot.pause() + out = app.query_one(ToolOutput) + collapsed = str(out.render()) + assert "ln0" in collapsed and "more lines" in collapsed and "ln19" not in collapsed + app.action_toggle_output() # Ctrl-O expands the most recent output + assert "ln19" in str(out.render()) # full content now shown + app.action_toggle_output() # toggles back to the preview + assert "ln19" not in str(out.render()) + + _run(go()) diff --git a/tests/test_code_modals.py b/tests/test_code_modals.py new file mode 100644 index 00000000..5ad276b8 --- /dev/null +++ b/tests/test_code_modals.py @@ -0,0 +1,236 @@ +"""Tests for the spoken/voice-answerable approval and ask modals. + +The pure ``approval_from_speech`` mapping is unit-tested directly; the screen wiring (speak the +prompt, listen, dismiss with the mapped decision) is driven through the real app headless with +a scripted voice double — no mic, speaker, or socket. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from textual.widgets import Input + +from aai_cli.code_agent.modals import ApprovalScreen, AskScreen, approval_from_speech +from aai_cli.code_agent.tui import CodeAgentApp +from aai_cli.core.errors import CLIError + + +class FakeAgent: + def invoke(self, *a, **k): + return {} + + +class FakeVoice: + """Scripted voice IO: speak() records, listen() replays one transcript (or raises).""" + + def __init__(self, transcript: str | None = None, *, error: CLIError | None = None) -> None: + self._transcript = transcript + self._error = error + self.spoken: list[str] = [] + + def speak(self, text: str) -> None: + self.spoken.append(text) + + def listen(self) -> str | None: + if self._error is not None: + raise self._error + return self._transcript + + +def _run(coro) -> None: + asyncio.run(coro) + + +@pytest.mark.parametrize( + ("said", "decision"), + [ + ("yes please", "approve"), + ("approve that", "approve"), + ("go ahead", "approve"), + ("auto approve", "auto"), + ("always do this", "auto"), + ("no", "reject"), + ("reject it", "reject"), + ("don't", "reject"), + ("yes but no", "reject"), # reject wins over approve when both are heard (safer) + ("uhh what", "reject"), # unclear -> safe default + ], +) +def test_approval_from_speech(said: str, decision: str) -> None: + assert approval_from_speech(said) == decision + + +async def _push_and_wait(app, pilot, screen) -> object: + box: dict[str, object] = {} + app.push_screen(screen, lambda result: box.update(value=result)) + for _ in range(300): + await pilot.pause(0.01) + if "value" in box: + break + return box.get("value", "__pending__") + + +def test_spoken_approval_speaks_prompt_and_maps_answer() -> None: + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + voice = FakeVoice(transcript="yes go for it") + result = await _push_and_wait( + app, pilot, ApprovalScreen("execute", {"command": "rm -rf build"}, voice=voice) + ) + assert result == "approve" # spoken "yes" mapped to approve + prompt = voice.spoken[0] + assert "Run execute" in prompt and "rm -rf build" in prompt + assert "Warning:" in prompt # the risky command is read aloud + assert "approve, auto-approve, or reject" in prompt + + _run(go()) + + +def test_spoken_approval_rejects_on_no() -> None: + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + result = await _push_and_wait( + app, pilot, ApprovalScreen("write_file", {"file_path": "x"}, voice=FakeVoice("no")) + ) + assert result == "reject" + + _run(go()) + + +def test_spoken_ask_speaks_question_and_returns_transcript() -> None: + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + voice = FakeVoice(transcript="use port 8080") + result = await _push_and_wait(app, pilot, AskScreen("Which port?", voice=voice)) + assert result == "use port 8080" # spoken answer returned verbatim + assert "The agent asks: Which port?" in voice.spoken[0] + + _run(go()) + + +def test_silence_does_not_auto_reject() -> None: + # No speech (listen -> None) must not auto-decide — the modal waits for speech or a keypress + # rather than rejecting a tool on a pause. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + box: dict[str, object] = {} + app.push_screen( + ApprovalScreen("execute", {"command": "ls"}, voice=FakeVoice(None)), + lambda result: box.update(value=result), + ) + for _ in range(50): + await pilot.pause(0.01) + assert "value" not in box # silence -> not dismissed + + _run(go()) + + +def test_voice_failure_falls_back_to_keyboard() -> None: + # If the mic/STT fails, the modal isn't auto-dismissed — the user can still press a key. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + voice = FakeVoice(error=CLIError("no mic", error_type="mic_missing", exit_code=2)) + box: dict[str, object] = {} + app.push_screen( + ApprovalScreen("execute", {"command": "ls"}, voice=voice), + lambda result: box.update(value=result), + ) + for _ in range(50): + await pilot.pause(0.01) + assert "value" not in box # voice failed -> not auto-dismissed + await pilot.press("n") # keyboard still works + await pilot.pause() + assert box.get("value") == "reject" + + _run(go()) + + +def test_ask_voice_failure_falls_back_to_typing() -> None: + # An ask modal whose voice fails isn't dismissed; the user types the answer instead. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + voice = FakeVoice(error=CLIError("no mic", error_type="mic_missing", exit_code=2)) + box: dict[str, object] = {} + app.push_screen(AskScreen("Which port?", voice=voice), lambda r: box.update(value=r)) + for _ in range(50): + await pilot.pause(0.01) + assert "value" not in box # voice failed -> not auto-dismissed + app.screen.query_one("#answer", Input).value = "8080" + await pilot.press("enter") + await pilot.pause() + assert box.get("value") == "8080" + + _run(go()) + + +def test_spoken_prompt_omits_detail_when_no_args() -> None: + # A tool with no identifying arg reads as just "Run . Say approve…" (no detail clause). + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + voice = FakeVoice(transcript="yes") + result = await _push_and_wait(app, pilot, ApprovalScreen("noop", {}, voice=voice)) + assert result == "approve" + assert "Run noop. Say approve" in voice.spoken[0] # straight to the options + + _run(go()) + + +def test_ask_silence_does_not_dismiss() -> None: + # No spoken answer (listen -> None) leaves the ask modal up for typing. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + box: dict[str, object] = {} + app.push_screen(AskScreen("Q?", voice=FakeVoice(None)), lambda r: box.update(value=r)) + for _ in range(50): + await pilot.pause(0.01) + assert "value" not in box # silence -> not dismissed + + _run(go()) + + +def test_decide_and_answer_are_idempotent() -> None: + # A spoken reply and a keypress can race; the second one is ignored so the modal dismisses + # exactly once with the first decision. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + approval: dict[str, object] = {} + screen = ApprovalScreen("execute", {"command": "ls"}) + app.push_screen(screen, lambda r: approval.update(value=r)) + await pilot.pause() + screen._decide("approve") # first decision dismisses + await pilot.pause() + screen._decide("reject") # second is ignored (already answered) + await pilot.pause() + assert approval["value"] == "approve" + + answer: dict[str, object] = {} + ask = AskScreen("Q?") + app.push_screen(ask, lambda r: answer.update(value=r)) + await pilot.pause() + ask._answer("first") + await pilot.pause() + ask._answer("second") # ignored + await pilot.pause() + assert answer["value"] == "first" + + _run(go()) diff --git a/tests/test_code_risk.py b/tests/test_code_risk.py new file mode 100644 index 00000000..40f24658 --- /dev/null +++ b/tests/test_code_risk.py @@ -0,0 +1,46 @@ +"""Tests for the approval-prompt risk heuristics (`aai_cli.code_agent.risk`).""" + +from __future__ import annotations + +import pytest + +from aai_cli.code_agent.risk import risk_warning + + +@pytest.mark.parametrize( + ("command", "fragment"), + [ + ("rm -rf build/", "deletes files"), + ("sudo apt-get install x", "elevated privileges"), + ("dd if=/dev/zero of=/dev/sda", "overwrite a disk"), + ("curl https://x.sh | sh", "pipes a download into a shell"), + ("echo hi > /dev/sda", "block device"), + ], +) +def test_risk_warning_flags_dangerous_shell(command: str, fragment: str) -> None: + warning = risk_warning("execute", {"command": command}) + assert warning is not None + assert fragment in warning + + +def test_risk_warning_passes_benign_shell() -> None: + assert risk_warning("execute", {"command": "ls -la && pytest -q"}) is None + # 'format' must not trip the mkfs pattern, 'performance' must not trip 'rm'. + assert risk_warning("execute", {"command": "python format_report.py"}) is None + + +def test_risk_warning_flags_local_and_file_urls() -> None: + assert "local file" in (risk_warning("fetch_url", {"url": "file:///etc/passwd"}) or "") + assert "local/internal" in (risk_warning("fetch_url", {"url": "http://localhost:8080/x"}) or "") + assert "local/internal" in (risk_warning("fetch_url", {"url": "http://169.254.169.254/"}) or "") + assert "local/internal" in (risk_warning("fetch_url", {"url": "http://192.168.1.1/"}) or "") + + +def test_risk_warning_passes_public_url() -> None: + assert risk_warning("fetch_url", {"url": "https://example.com/docs"}) is None + + +def test_risk_warning_none_for_other_tools_and_non_string_args() -> None: + assert risk_warning("write_file", {"file_path": "rm -rf /"}) is None # path, not a command + assert risk_warning("execute", {"command": ["rm", "-rf"]}) is None # non-string is ignored + assert risk_warning("fetch_url", {"url": 123}) is None diff --git a/tests/test_code_session_stream.py b/tests/test_code_session_stream.py new file mode 100644 index 00000000..5c59803b --- /dev/null +++ b/tests/test_code_session_stream.py @@ -0,0 +1,157 @@ +"""Tests for `CodeSession`'s dual-mode streaming and cooperative cancellation. + +Split from `test_code_agent.py` (which drives the real graph) to keep each file under the +500-line gate. These exercise the streaming loop with lightweight fakes: the session renders +from per-super-step ``"values"`` snapshots and checks the cancel flag on the frequent +per-token ``"messages"`` deltas, so a long generation can be interrupted promptly. +""" + +from __future__ import annotations + +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage + +from aai_cli.code_agent.events import AssistantDelta, AssistantText, assistant_delta +from aai_cli.code_agent.session import CodeSession + + +class StreamingAgent: + """A double exercising the dual-mode streaming path. + + Mirrors langgraph's ``stream_mode=["values", "messages"]`` contract: each scripted state + snapshot is yielded tagged as ``("values", snapshot)``, optionally preceded by + ``("messages", delta)`` per-token deltas (the fine-grained cancellation checkpoints). + """ + + def __init__( + self, chunks: list[dict[str, object]], *, token_deltas: tuple[str, ...] = () + ) -> None: + self._chunks = chunks + self._token_deltas = token_deltas + + def stream(self, graph_input, config=None, *, stream_mode=("values", "messages")): + del graph_input, config, stream_mode + for delta in self._token_deltas: + yield ("messages", delta) + for chunk in self._chunks: + yield ("values", chunk) + + def invoke(self, *a, **k): # the streaming branch is taken, so invoke is never used + raise AssertionError("a streaming agent must not be invoked") + + +def test_assistant_delta_is_frozen_hashable() -> None: + # frozen=True makes it immutable+hashable; a non-frozen eq dataclass sets __hash__=None, + # so hash() would raise — this keeps the event safe to dedupe/compare and pins `frozen`. + assert hash(AssistantDelta("x")) == hash(AssistantDelta("x")) + + +def test_assistant_delta_extracts_only_ai_text() -> None: + # messages-mode yields (message, metadata); only AI text becomes a delta. + assert assistant_delta((AIMessage("tok"), {"node": "agent"})) == AssistantDelta("tok") + assert assistant_delta(AIMessage("bare")) == AssistantDelta("bare") # untupled is fine too + assert assistant_delta((AIMessage(""), {})) is None # empty content (e.g. a tool-call turn) + assert assistant_delta((ToolMessage("result", tool_call_id="1"), {})) is None # not assistant + assert assistant_delta(()) is None # defensive: empty payload + + +def test_send_emits_assistant_deltas_from_messages_stream() -> None: + # The per-token messages chunks are surfaced as AssistantDelta (live preview), and the + # values snapshot still yields the authoritative AssistantText. + seen: list[object] = [] + + class TokenAgent: + def stream(self, graph_input, config=None, *, stream_mode=("values", "messages")): + del graph_input, config, stream_mode + yield ("messages", (AIMessage("Hello, "), {})) + yield ("messages", (AIMessage("world"), {})) + yield ("values", {"messages": [AIMessage("Hello, world")]}) + + def invoke(self, *a, **k): + raise AssertionError("a streaming agent must not be invoked") + + session = CodeSession(agent=TokenAgent(), sink=seen.append, approver=lambda n, a: True) + session.send("go") + + deltas = [e.text for e in seen if isinstance(e, AssistantDelta)] + finals = [e.text for e in seen if isinstance(e, AssistantText)] + assert deltas == ["Hello, ", "world"] # streamed tokens + assert finals == ["Hello, world"] # authoritative full reply from the values snapshot + + +def test_send_streams_each_step_and_cancel_stops_the_loop() -> None: + # Three successive graph states (messages grow by one each step); a stream_mode="values" + # graph yields exactly these snapshots, so the session must emit incrementally. + chunks: list[dict[str, object]] = [ + {"messages": [HumanMessage("go")]}, + {"messages": [HumanMessage("go"), AIMessage("first")]}, + {"messages": [HumanMessage("go"), AIMessage("first"), AIMessage("second")]}, + ] + seen: list[object] = [] + session = CodeSession( + agent=StreamingAgent(chunks), sink=seen.append, approver=lambda n, a: True + ) + + def sink(event: object) -> None: + seen.append(event) + if isinstance(event, AssistantText) and event.text == "first": + session.request_cancel() # cancel mid-stream, before the "second" chunk is consumed + + session.sink = sink + session.send("go") + + texts = [e.text for e in seen if isinstance(e, AssistantText)] + # "first" streamed out as its step landed; the cancel then broke the loop, so the later + # "second" step was never emitted — proving both incremental rendering and cancellation. + assert texts == ["first"] + + +def test_cancel_within_a_step_breaks_on_a_token_delta() -> None: + # A single model generation is one super-step, so a values-only loop can't break until the + # whole reply lands. Streaming the per-token "messages" deltas alongside gives a frequent + # cancel checkpoint: a Ctrl-C mid-generation breaks before the reply ("late") is ever + # rendered. Modeled by an agent that requests cancel between two token deltas. + seen: list[object] = [] + + class TokenStreamAgent: + session: CodeSession + + def stream(self, graph_input, config=None, *, stream_mode=("values", "messages")): + del graph_input, config, stream_mode + yield ("messages", "par") # first token arrives — loop sees no cancel yet + self.session.request_cancel() # user hits Ctrl-C mid-generation + yield ("messages", "tial") # next token: the loop's top-of-iteration check breaks + yield ("values", {"messages": [AIMessage("late")]}) # must never be rendered + + def invoke(self, *a, **k): + raise AssertionError("a streaming agent must not be invoked") + + agent = TokenStreamAgent() + session = CodeSession(agent=agent, sink=seen.append, approver=lambda n, a: True) + agent.session = session + session.send("go") + + texts = [e.text for e in seen if isinstance(e, AssistantText)] + assert texts == [] # the post-cancel "late" reply was dropped, not rendered + + +def test_only_values_chunks_are_rendered_not_messages_deltas() -> None: + # The dual-mode stream tags each yield by mode; only "values" snapshots are rendered (the + # "messages" deltas exist purely as cancel checkpoints). A messages delta that happens to + # be a dict must NOT be emitted — guards the `mode == "values" and ...` guard against an + # `and`->`or` slip that would render it. + seen: list[object] = [] + + class DualModeAgent: + def stream(self, graph_input, config=None, *, stream_mode=("values", "messages")): + del graph_input, config, stream_mode + yield ("messages", {"messages": [AIMessage("ghost")]}) # dict, but messages-mode + yield ("values", {"messages": [AIMessage("real")]}) + + def invoke(self, *a, **k): + raise AssertionError("a streaming agent must not be invoked") + + session = CodeSession(agent=DualModeAgent(), sink=seen.append, approver=lambda n, a: True) + session.send("go") + + texts = [e.text for e in seen if isinstance(e, AssistantText)] + assert texts == ["real"] # the messages-mode dict ("ghost") was not rendered diff --git a/tests/test_code_summarize.py b/tests/test_code_summarize.py new file mode 100644 index 00000000..ebf0eb24 --- /dev/null +++ b/tests/test_code_summarize.py @@ -0,0 +1,93 @@ +"""Tests for the shared tool-activity summarizers (`aai_cli.code_agent.summarize`). + +These keep the coding-agent transcript scannable: a tool call shows its identifying arg +(not the whole file being written), and tool output is previewed with a hidden-line tail. +""" + +from __future__ import annotations + +from aai_cli.code_agent.summarize import ( + describe_args, + full_args, + summarize_call, + summarize_result, +) + + +def test_describe_args_prefers_identity_arg_and_elides_bulk() -> None: + # write_file's content is the bulk we must NOT inline — only the path identifies the call. + body = "\n".join(f"line {i}" for i in range(50)) + assert describe_args({"file_path": "app.py", "content": body}) == "app.py" + # A shell command is the identity arg for execute. + assert describe_args({"command": "pip install flask"}) == "pip install flask" + + +def test_describe_args_clips_long_identity_value() -> None: + out = describe_args({"command": "echo " + "x" * 200}) + assert out.endswith("…") + assert len(out) == 60 # exact: clipped to the per-arg budget, ellipsis included + + +def test_describe_args_without_identity_shows_capped_key_values() -> None: + out = describe_args({"a": 1, "b": 2, "c": 3, "d": 4}) + # Only the first few args render, then an ellipsis marks the elided remainder. + assert out.startswith("a=1, b=2, c=3") + assert out.endswith(", …") + assert "d=4" not in out + + +def test_describe_args_collapses_newlines_in_values() -> None: + # A newline-bearing value must not break the one-line transcript entry. + assert "\n" not in describe_args({"x": "a\nb\nc"}) + + +def test_summarize_call_wraps_args_in_tool_name() -> None: + assert ( + summarize_call("write_file", {"file_path": "app.py", "content": "x"}) + == "write_file(app.py)" + ) + + +def test_summarize_result_previews_and_counts_hidden_lines() -> None: + out = summarize_result("\n".join(f"line {i}" for i in range(20))) + assert "line 0" in out and "line 3" in out + assert "line 4" not in out # only the first few lines are kept + assert "+16 more lines" in out # the rest are counted, not dropped silently + + +def test_summarize_result_shows_short_output_in_full() -> None: + assert summarize_result("done\n") == "done" # no tail when nothing is hidden + assert summarize_result(" ") == "" # whitespace-only collapses to empty + + +def test_full_args_shows_every_arg_whole_with_newlines() -> None: + # The expanded view keeps content (and its newlines) that describe_args elides. + out = full_args({"file_path": "app.py", "content": "a\nb\nc"}) + assert "file_path=app.py" in out + assert "content=a\nb\nc" in out # full value, newlines preserved + + +def test_full_args_caps_a_huge_value_with_char_count() -> None: + out = full_args({"content": "z" * 1500}) # over the 1000-char expanded budget + assert "+500 more chars" in out # exact: 1500 minus the 1000 budget + assert out.startswith("content=" + "z" * 1000) + + +def test_full_args_shows_a_value_at_the_budget_whole() -> None: + # Boundary: exactly the budget is shown whole (guards the cap's `>` against a `>=` slip). + out = full_args({"content": "z" * 1000}) + assert "more chars" not in out + assert out == "content=" + "z" * 1000 + + +def test_summarize_result_counts_a_single_hidden_line() -> None: + # Boundary: exactly one line over the preview budget still gets a tail (guards the + # `hidden_lines > 0` threshold against a `> 1` slip that would silently drop it). + out = summarize_result("\n".join(f"line {i}" for i in range(5))) # 4 shown, 1 hidden + assert out.endswith("(+1 more lines)") + + +def test_summarize_result_clips_one_huge_line_with_char_count() -> None: + out = summarize_result("z" * 500) # a single line longer than the char budget + assert "+200 more chars" in out # exact: 500 minus the 300-char budget = 200 hidden + assert out.startswith("z" * 300) diff --git a/tests/test_code_tui.py b/tests/test_code_tui.py index fa0aeff3..b536dec1 100644 --- a/tests/test_code_tui.py +++ b/tests/test_code_tui.py @@ -10,15 +10,15 @@ import asyncio import threading import time -from pathlib import Path import pytest from langchain_core.messages import AIMessage, HumanMessage -from textual.widgets import Input, RichLog, Static +from textual.containers import VerticalScroll +from textual.widgets import Input, Label, Static -from aai_cli.code_agent import tui from aai_cli.code_agent.events import AssistantText, ErrorText, ToolCall, ToolResult -from aai_cli.code_agent.tui import ApprovalScreen, AskScreen, CodeAgentApp +from aai_cli.code_agent.modals import ApprovalScreen, AskScreen +from aai_cli.code_agent.tui import CodeAgentApp class FakeAgent: @@ -39,39 +39,6 @@ def __init__(self, value: dict[str, object]) -> None: self.value = value -# --- pure helpers ------------------------------------------------------------- - - -def test_format_args_and_abbrev_home() -> None: - assert tui._format_args({"a": 1, "b": "x"}) == "a=1, b='x'" - assert tui._abbrev_home(Path.home() / "proj") == "~/proj" - # A path outside home renders as-is; compare to the platform-native string so this - # holds on Windows (where str(Path(...)) uses backslashes) as well as POSIX. - outside = Path("/etc/hosts") - assert tui._abbrev_home(outside) == str(outside) - - -def test_approval_decision_defaults_to_reject() -> None: - assert tui._approval_decision("approve") == "approve" - assert tui._approval_decision("auto") == "auto" - # A button with no id (Textual allows None) is treated as a rejection, not approval. - assert tui._approval_decision(None) == "reject" - assert tui._approval_decision("") == "reject" - - -def test_git_branch_and_status(tmp_path: Path) -> None: - assert tui._git_branch(tmp_path) is None # no .git - (tmp_path / ".git").mkdir() - (tmp_path / ".git" / "HEAD").write_text("ref: refs/heads/feature-x\n") - assert tui._git_branch(tmp_path) == "feature-x" - (tmp_path / ".git" / "HEAD").write_text("a1b2c3d4e5f6\n") # detached - assert tui._git_branch(tmp_path) == "a1b2c3d4" - - status = tui._status_text(tmp_path, auto_approve=True) - assert "auto" in status and "a1b2c3d4" in status - assert "manual" in tui._status_text(tmp_path, auto_approve=False) - - # --- pilot tests -------------------------------------------------------------- @@ -84,8 +51,9 @@ async def go() -> None: app = CodeAgentApp(agent=FakeAgent([]), web_note="no key", thread_id="t1") async with app.run_test(size=(100, 30)) as pilot: await pilot.pause() - log = app.query_one("#log", RichLog) - assert len(log.lines) > 6 # wordmark + tagline + log = app.query_one("#log", VerticalScroll) + assert len(log.children) >= 1 # the splash is mounted into the transcript + assert "Ready to code" in str(log.children[0].render()) # splash intro shown assert app.focused is app.query_one("#prompt", Input) _run(go()) @@ -206,23 +174,66 @@ async def go() -> None: _run(go()) -def test_approval_button_press_dismisses() -> None: - # Covers ApprovalScreen.on_button_pressed (the click path; key paths are covered - # by the approve/reject modal tests above). The bracketed name/args also guard the - # compose() escape() — without it, Label markup parsing would raise on mount. - results: list[str | None] = [] +def test_approval_prompt_renders_keyboard_hint() -> None: + # The prompt is a plain y/a/n keyboard hint, not clickable buttons — assert each + # option's copy renders so dropping one is caught. The bracketed name/args also guard + # the compose() escape(): without it, Label markup parsing would raise on mount. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app.push_screen(ApprovalScreen("exec[", {"cmd": "[ls"})) + await pilot.pause() + rendered = " ".join(str(label.render()) for label in app.screen.query(Label)) + assert "approve" in rendered + assert "auto-approve" in rendered + assert "reject" in rendered + + _run(go()) + +def test_approval_expands_args_on_e() -> None: + # Collapsed, the prompt shows only the identifying arg (the filename); pressing `e` + # expands it to the full args, revealing the file content that was elided. async def go() -> None: app = CodeAgentApp(agent=FakeAgent([])) async with app.run_test(size=(100, 30)) as pilot: await pilot.pause() - app.push_screen(ApprovalScreen("exec[", {"cmd": "[ls"}), results.append) + app.push_screen( + ApprovalScreen("write_file", {"file_path": "x.py", "content": "SECRET"}) + ) await pilot.pause() - await pilot.click("#reject") + detail = app.screen.query_one("#approvaldetail", Label) + assert "SECRET" not in str(detail.render()) # collapsed: content elided + await pilot.press("e") await pilot.pause() + assert "SECRET" in str(detail.render()) # expanded: full args shown + await pilot.press("e") # toggles back + await pilot.pause() + assert "SECRET" not in str(detail.render()) + + _run(go()) + + +def test_approval_shows_risk_warning_for_dangerous_command() -> None: + # A destructive shell command carries a one-line warning above the prompt; a benign one + # mounts no warning label at all. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app.push_screen(ApprovalScreen("execute", {"command": "rm -rf build/"})) + await pilot.pause() + warn = app.screen.query("#approvalwarn") + assert warn # warning present + assert "deletes files" in str(warn.first().render()) + app.pop_screen() + await pilot.pause() + app.push_screen(ApprovalScreen("execute", {"command": "ls -la"})) + await pilot.pause() + assert not app.screen.query("#approvalwarn") # benign: no warning mounted _run(go()) - assert results == ["reject"] def test_approval_box_is_compact_and_bottom_docked() -> None: @@ -242,6 +253,28 @@ async def go() -> None: _run(go()) +def test_modals_are_transparent_so_transcript_stays_visible() -> None: + # Regression guard: the app's `Screen { background: #000000 }` canvas rule matches every + # Screen subclass, and app CSS beats a widget's DEFAULT_CSS — so without the explicit + # `ModalScreen { background: transparent }` app rule, the modal paints opaque black and + # blanks the transcript behind it. Assert each modal resolves to a see-through background + # (alpha 0); an opaque modal (alpha 1.0) — the bug — fails here. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app.push_screen(ApprovalScreen("write_file", {"file_path": "x.py"})) + await pilot.pause() + assert app.screen.styles.background.a == 0 # approval modal is see-through + app.pop_screen() + await pilot.pause() + app.push_screen(AskScreen("which port?")) + await pilot.pause() + assert app.screen.styles.background.a == 0 # ask modal is see-through + + _run(go()) + + def test_approval_auto_approve_flips_mode_and_skips_later_prompts() -> None: # Picking "Auto-approve (a)" approves this call, flips the badge manual→auto, and # makes every later _approve return True without ever pushing a modal. @@ -344,11 +377,6 @@ async def go() -> None: _run(go()) -def test_spinner_text_formats_frame_and_elapsed() -> None: - assert tui._spinner_text(46, "✶") == "✶ Working… (46s)" - assert tui._spinner_text(0, "✷") == "✷ Working… (0s)" - - def test_spinner_starts_ticks_and_stops(monkeypatch: pytest.MonkeyPatch) -> None: async def go() -> None: app = CodeAgentApp(agent=FakeAgent([])) diff --git a/tests/test_code_tui_status.py b/tests/test_code_tui_status.py new file mode 100644 index 00000000..f261a517 --- /dev/null +++ b/tests/test_code_tui_status.py @@ -0,0 +1,49 @@ +"""Tests for the coding-agent TUI's pure status/text helpers (`tui_status`). + +Split from test_code_tui.py (which drives the Textual app) to keep each file under the +500-line gate; these need no pilot, just the plain functions. +""" + +from __future__ import annotations + +from pathlib import Path + +from aai_cli.code_agent import tui_status + + +def test_spinner_text_formats_frame_and_elapsed() -> None: + assert tui_status._spinner_text(46, "✶") == "✶ Working… (46s)" + assert tui_status._spinner_text(0, "✷") == "✷ Working… (0s)" + + +def test_abbrev_home() -> None: + assert tui_status._abbrev_home(Path.home() / "proj") == "~/proj" + # A path outside home renders as-is; compare to the platform-native string so this + # holds on Windows (where str(Path(...)) uses backslashes) as well as POSIX. + outside = Path("/etc/hosts") + assert tui_status._abbrev_home(outside) == str(outside) + + +def test_git_branch_and_status(tmp_path: Path) -> None: + assert tui_status._git_branch(tmp_path) is None # no .git + (tmp_path / ".git").mkdir() + (tmp_path / ".git" / "HEAD").write_text("ref: refs/heads/feature-x\n") + assert tui_status._git_branch(tmp_path) == "feature-x" + (tmp_path / ".git" / "HEAD").write_text("a1b2c3d4e5f6\n") # detached + assert tui_status._git_branch(tmp_path) == "a1b2c3d4" + + status = tui_status._status_text(tmp_path, auto_approve=True) + assert "auto" in status and "a1b2c3d4" in status + assert "manual" in tui_status._status_text(tmp_path, auto_approve=False) + + +def test_status_text_renders_voice_badge(tmp_path: Path) -> None: + # No voice front-end -> no voice badge (the dot glyphs are absent); on/off render the + # state so the Ctrl-V toggle shows. (Asserts on the dots, not the word — the tmp_path name + # itself can contain "voice".) + none = tui_status._status_text(tmp_path, auto_approve=False) + assert "●" not in none and "○" not in none + on = tui_status._status_text(tmp_path, auto_approve=False, voice_state="on") + off = tui_status._status_text(tmp_path, auto_approve=False, voice_state="off") + assert "voice on" in on and "●" in on # filled dot when on + assert "voice off" in off and "○" in off # hollow dot when off diff --git a/tests/test_code_tui_voice.py b/tests/test_code_tui_voice.py index 8adbaea6..e402d2d0 100644 --- a/tests/test_code_tui_voice.py +++ b/tests/test_code_tui_voice.py @@ -12,7 +12,7 @@ import pytest from langchain_core.messages import AIMessage, HumanMessage -from textual.widgets import Input +from textual.widgets import Input, Static from aai_cli.code_agent.tui import CodeAgentApp from aai_cli.core.errors import CLIError @@ -163,3 +163,202 @@ async def go() -> None: assert app._voice is None _run(go()) + + +def test_toggle_voice_pauses_and_resumes_capture() -> None: + # Ctrl-V flips voice off (no capture, no readback) and back on; the state badge tracks it. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([]), voice=FakeVoice()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + # Assert via the methods, not the `_voice_paused` attribute: mypy narrows the + # attribute and can't see action_toggle_voice() flip it back, flagging the second + # check unreachable. The method calls reflect the same state without that trap. + assert app._voice_active() + assert app._voice_state() == "on" + app.action_toggle_voice() # pause + assert not app._voice_active() + assert app._voice_state() == "off" + app.action_toggle_voice() # resume + assert app._voice_active() + assert app._voice_state() == "on" + + _run(go()) + + +def test_paused_voice_skips_followup_readback() -> None: + # While paused, the post-turn followup neither speaks a summary nor listens. + async def go() -> None: + voice = FakeVoice(transcripts=["ignored"]) + app = CodeAgentApp(agent=FakeAgent([]), voice=voice) + app._voice_paused = True # set before mount so on_mount never auto-listens + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._last_reply = "a reply" + app._voice_followup() + await pilot.pause() + assert voice.spoken == [] # paused: no readback + assert voice.listens == 0 # paused: no capture + + _run(go()) + + +def test_voice_mode_swaps_text_input_for_listening_affordance() -> None: + # While voice capture is on, the text prompt is hidden and a "listening" bar shows; + # toggling voice off (Ctrl-V) brings the text box back. (Re-query each check so mypy + # doesn't narrow a stored display bool across the toggles.) + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([]), voice=FakeVoice()) + app._voice_paused = True # start paused so on_mount doesn't race a capture thread + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + assert app.query_one("#promptbar").display is True # paused -> text box visible + assert app.query_one("#voicebar").display is False + app.action_toggle_voice() # voice on + await pilot.pause() + assert app.query_one("#promptbar").display is False # text box hidden + assert app.query_one("#voicebar").display is True # listening affordance shown + app.action_toggle_voice() # voice off + await pilot.pause() + assert app.query_one("#promptbar").display is True # text box back + assert app.query_one("#voicebar").display is False + + _run(go()) + + +def test_voice_capture_failure_restores_the_text_input() -> None: + # When the mic is ruled out mid-session, the listening bar is replaced by the text box. + async def go() -> None: + voice = FakeVoice(error=CLIError("no mic", error_type="mic_missing", exit_code=2)) + app = CodeAgentApp(agent=FakeAgent([]), voice=voice) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + assert await _wait_until(pilot, lambda: app._voice_typed) + await pilot.pause() + assert app.query_one("#promptbar").display is True # text box restored on failure + assert app.query_one("#voicebar").display is False + + _run(go()) + + +def test_voice_bar_distinguishes_phases() -> None: + # The bar shows a distinct label per phase; only the listening phase carries the type hint. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([]), voice=FakeVoice()) + app._voice_paused = True # quiet the auto-listen; drive phases directly + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._set_voice_phase("listening") + bar = str(app.query_one("#voicebar", Static).render()) + assert "Listening" in bar and "Ctrl-V to type" in bar + app._set_voice_phase("thinking") + bar = str(app.query_one("#voicebar", Static).render()) + assert "Thinking" in bar and "Ctrl-V to type" not in bar # hint is listening-only + app._set_voice_phase("speaking") + assert "Speaking" in str(app.query_one("#voicebar", Static).render()) + + _run(go()) + + +def test_spinner_suppressed_in_voice_mode() -> None: + # In voice mode the bar carries the "thinking" state, so the separate spinner stays hidden; + # pausing voice brings the spinner back. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([]), voice=FakeVoice()) + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._start_spinner() + assert app.query_one("#spinner", Static).display is False # voice active -> no spinner + app._voice_paused = True + app._start_spinner() + assert app.query_one("#spinner", Static).display is True # paused -> spinner shows + + _run(go()) + + +def test_voice_bar_animation_timer_runs_and_advances() -> None: + # The meter animation timer runs only while the bar is shown, and a tick changes the frame. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([]), voice=FakeVoice()) + app._voice_paused = True + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + # Read into fresh locals each time: asserting `is None`/`is not None` on the same + # attribute across the opaque toggle would make mypy flag the later check unreachable. + paused_timer = app._voice_timer + assert paused_timer is None # paused -> no animation + app.action_toggle_voice() # voice on -> bar shown, timer running + await pilot.pause() + running_timer = app._voice_timer + assert running_timer is not None + before = str(app.query_one("#voicebar", Static).render()) + app._tick_voice() + assert str(app.query_one("#voicebar", Static).render()) != before # meter advanced + app.action_toggle_voice() # voice off -> timer stopped + await pilot.pause() + stopped_timer = app._voice_timer + assert stopped_timer is None + + _run(go()) + + +def test_submit_sets_thinking_phase() -> None: + async def go() -> None: + agent = FakeAgent([{"messages": [HumanMessage("go"), AIMessage("done")]}]) + app = CodeAgentApp(agent=agent, voice=FakeVoice()) + app._voice_paused = True # keep the post-turn followup from flipping the phase + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app._submit("go") + assert app._voice_phase == "thinking" # set synchronously when the turn starts + await app.workers.wait_for_complete() + + _run(go()) + + +def test_run_leg_swallows_callback_error_after_the_app_stops() -> None: + # A voice leg still in flight when the app tears down calls back onto a dead UI thread; + # the resulting RuntimeError must be dropped (the spoken turn is moot), not surface as an + # unhandled thread exception. This app was never started, so is_running is False. + app = CodeAgentApp(agent=FakeAgent([]), voice=FakeVoice()) + assert app.is_running is False + ran: list[bool] = [] + + def boom() -> None: + ran.append(True) + raise RuntimeError("App is not running") + + app._run_leg(boom) # returns without raising — the teardown-race error is swallowed + assert ran == [True] # the leg body did run; only its post-teardown error was dropped + + +def test_run_leg_reraises_a_genuine_failure_while_the_app_is_live() -> None: + # While the app is running, a real exception in a leg is a bug and must propagate (so it's + # reported), not be silently swallowed like the teardown race above. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([]), voice=FakeVoice()) + app._voice_paused = True # no auto-listen thread racing this assertion + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + assert app.is_running is True + + def boom() -> None: + raise ValueError("genuine bug") + + with pytest.raises(ValueError, match="genuine bug"): + app._run_leg(boom) + + _run(go()) + + +def test_toggle_voice_without_session_notifies_and_stays_off() -> None: + # With no voice front-end the toggle is a no-op (notice only) and never marks a pause. + async def go() -> None: + app = CodeAgentApp(agent=FakeAgent([])) # no voice + async with app.run_test(size=(100, 30)) as pilot: + await pilot.pause() + app.action_toggle_voice() + assert app._voice_paused is False # nothing to pause + assert app._voice_state() is None # no badge without a session + + _run(go()) diff --git a/tests/test_microphone.py b/tests/test_microphone.py index 8199176e..ffb2468e 100644 --- a/tests/test_microphone.py +++ b/tests/test_microphone.py @@ -1,6 +1,7 @@ import signal import sys import types +from collections.abc import Callable, Mapping from typing import Any import pytest @@ -14,6 +15,8 @@ _device_default_rate, _ignore_interrupt_during_shutdown, _install_shutdown_interrupt_guard, + _max_input_channels, + _RawInputStream, _SoundDeviceMic, import_sounddevice, resample_pcm16, @@ -41,6 +44,29 @@ def close(self): self.closed = True +class _FakeSoundDevice(types.ModuleType): + """A typed `_SoundDeviceModule` double: scripted device info + a RawInputStream factory. + + Subclasses `ModuleType` so it can be slotted into `sys.modules` via `monkeypatch.setitem`, + and conforms to the protocol so it needs no escape hatches at the call sites that pass it + to the real `_max_input_channels` / `_default_mic_stream` code under test. + """ + + def __init__( + self, + info: Mapping[str, object], + raw_input_stream: Callable[..., _RawInputStream] = _FakeRawStream, + ) -> None: + super().__init__("sounddevice") + self._info = info + self.RawInputStream = raw_input_stream + + def query_devices( + self, device: int | None = None, kind: str | None = None + ) -> Mapping[str, object]: + return self._info + + def test_audio_missing_error_has_reinstall_suggestion(): from aai_cli.core.microphone import audio_missing_error @@ -292,6 +318,91 @@ def test_default_mic_stream_missing_sounddevice_raises_mic_missing(monkeypatch): assert exc.value.exit_code == 2 +class _FakeStereoStream(_FakeRawStream): + """A 2-channel input stream: one interleaved stereo frame (L=256, R=768).""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + # int16 LE: L=256 (b"\x00\x01"), R=768 (b"\x00\x03"), interleaved one frame. + self._chunks = [(b"\x00\x01\x00\x03", False)] + + +def test_sounddevice_mic_downmixes_stereo_to_mono(): + # channels=2 averages L/R per frame: (256 + 768) / 2 == 512 (b"\x00\x02"). + mic = _SoundDeviceMic(_FakeStereoStream(), blocksize=1, channels=2) + assert next(iter(mic)) == b"\x00\x02" + + +def _fake_sd_rejecting_mono(max_input_channels: int, opened: list[int]) -> _FakeSoundDevice: + """A sounddevice whose mono open fails with -9998; query reports ``max_input_channels``.""" + + def raw_input_stream(*, channels: int, **kwargs: object) -> _RawInputStream: + opened.append(channels) + if channels == 1: + raise OSError("Error opening RawInputStream: Invalid number of channels [-9998]") + return _FakeStereoStream(channels=channels, **kwargs) + + return _FakeSoundDevice({"max_input_channels": max_input_channels}, raw_input_stream) + + +def test_max_input_channels_defaults_to_zero_when_absent_or_non_int(): + # A device dict missing the key, or carrying a non-int value, must read as 0 channels (so + # the caller raises the actionable no-input error) rather than a truthy bogus count. + assert _max_input_channels(_FakeSoundDevice({}), None) == 0 # key absent -> 0, not get()'s + assert _max_input_channels(_FakeSoundDevice({"max_input_channels": None}), None) == 0 # non-int + assert _max_input_channels(_FakeSoundDevice({"max_input_channels": 2}), None) == 2 # int passes + + +def test_default_mic_stream_falls_back_to_stereo_downmix(monkeypatch): + # A multichannel-only input (mono rejected, but >=2 channels available) is reopened at + # stereo and downmixed to mono — so voice works on devices that won't open as mono. + opened: list[int] = [] + monkeypatch.setitem(sys.modules, "sounddevice", _fake_sd_rejecting_mono(2, opened)) + stream = _default_mic_stream(sample_rate=16000, device=None) + assert opened == [1, 2] # tried mono, then reopened stereo + assert next(iter(stream)) == b"\x00\x02" # yields downmixed mono + + +def test_default_mic_stream_zero_input_channels_raises_permission_error(monkeypatch): + # 0 input channels can't be salvaged (no mic permission / wrong default device): raise an + # actionable error pointing at the macOS Microphone privacy setting, not the cryptic code. + opened: list[int] = [] + monkeypatch.setitem(sys.modules, "sounddevice", _fake_sd_rejecting_mono(0, opened)) + with pytest.raises(CLIError) as exc: + _default_mic_stream(sample_rate=16000, device=None) + assert opened == [1] # only the mono attempt; no pointless stereo retry + assert exc.value.error_type == "mic_error" + assert exc.value.exit_code == 1 + assert "no input channels" in exc.value.message.lower() + assert exc.value.suggestion is not None + assert "Microphone" in exc.value.suggestion + + +def test_default_mic_stream_single_channel_failure_reraises_original(monkeypatch): + # A genuine 1-channel device should accept mono; if it still failed, the channel fallback + # can't help, so surface the real PortAudio error rather than masking it. + opened: list[int] = [] + monkeypatch.setitem(sys.modules, "sounddevice", _fake_sd_rejecting_mono(1, opened)) + with pytest.raises(OSError, match="Invalid number of channels"): + _default_mic_stream(sample_rate=16000, device=None) + assert opened == [1] # no stereo retry on a 1-channel device + + +def test_microphone_source_passes_through_factory_clierror(): + # An actionable CLIError from the factory (e.g. the zero-channel case) must propagate + # intact, not get re-wrapped into the generic "Could not open" message. + err = CLIError("no input channels", error_type="mic_error", exit_code=1, suggestion="grant it") + + def boom(**_kwargs): + raise err + + mic = MicrophoneSource(capture_rate=16000, stream_factory=boom) + with pytest.raises(CLIError) as exc: + list(mic) + assert exc.value is err # passed through unchanged + assert exc.value.suggestion == "grant it" + + def test_ignore_interrupt_during_shutdown_sets_sig_ign(): # The guard drops a second Ctrl-C during teardown so it can't raise inside # sounddevice's atexit PortAudio terminate. Save/restore the global disposition. diff --git a/tests/test_stdio.py b/tests/test_stdio.py index 9b0e2fc0..8a9a58f7 100644 --- a/tests/test_stdio.py +++ b/tests/test_stdio.py @@ -1,4 +1,5 @@ import io +import os from aai_cli.core import stdio @@ -121,3 +122,59 @@ def boom(*_a, **_k): # Raising inside the suppressed block must not propagate. monkeypatch.setattr("os.open", boom) stdio.silence_stdout() + + +def test_suppress_native_stderr_redirects_during_block_then_restores(monkeypatch): + # The fd dance: dup the real stderr (fd 2 itself — never sys.stderr.fileno(), which is + # an unusable redirector inside a TUI), point it at /dev/null for the body, then restore + # and close both temporaries. The body must run *while* redirected (between the dup2s). + events: list[object] = [] + monkeypatch.setattr("os.dup", lambda fd: events.append(("dup", fd)) or 50) + monkeypatch.setattr("os.open", lambda path, flags: events.append(("open", path)) or 99) + monkeypatch.setattr("os.dup2", lambda src, dst: events.append(("dup2", src, dst))) + monkeypatch.setattr("os.close", lambda fd: events.append(("close", fd))) + + with stdio.suppress_native_stderr(): + events.append("body") + + assert events == [ + ("dup", 2), # save the real stderr fd (literal 2) + ("open", os.devnull), # open /dev/null + ("dup2", 99, 2), # point stderr at /dev/null + "body", # the block runs while stderr is redirected + ("dup2", 50, 2), # restore the saved fd + ("close", 50), + ("close", 99), + ] + + +def test_suppress_native_stderr_runs_body_when_redirect_fails(monkeypatch): + # Safe by construction: if the fd can't be duplicated, the block still runs (suppression + # is cosmetic and must never break the wrapped mic open) and stderr is never redirected. + def boom(_fd: int) -> int: + raise OSError("cannot dup") + + redirected: list[tuple[int, int]] = [] + monkeypatch.setattr("os.dup", boom) + monkeypatch.setattr("os.dup2", lambda src, dst: redirected.append((src, dst))) + ran: list[bool] = [] + + with stdio.suppress_native_stderr(): + ran.append(True) + + assert ran == [True] # body ran despite the dup failure + assert redirected == [] # never redirected -> nothing left to restore + + +def test_suppress_native_stderr_swallows_close_failure(monkeypatch): + # A teardown close hitting an already-closed/invalid fd must not escape the block. + def boom(_fd: int) -> None: + raise OSError("already closed") + + monkeypatch.setattr("os.dup", lambda _fd: 50) + monkeypatch.setattr("os.open", lambda _path, _flags: 99) + monkeypatch.setattr("os.dup2", lambda _src, _dst: None) + monkeypatch.setattr("os.close", boom) + + with stdio.suppress_native_stderr(): + pass # exits cleanly even though both teardown closes raise