From cf58f61cf967d0676cf05a14261e8b876f38344c Mon Sep 17 00:00:00 2001 From: Lavindeep Dhillon Date: Sun, 28 Jun 2026 02:52:21 -0400 Subject: [PATCH] feat(ui): turn-scoped thinking indicator with live reasoning readout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Surface what a model turn is doing — the feature the full-screen rework was for. On submit the app echoes the user's message, then a live 'frontier' line renders as the last transcript row: ✈·· cruising… 38s · 1.8k reasoning The plane glides, the phrase advances through flight phases, the timer ticks, and the reasoning-token estimate climbs while the model is thinking. Completed tool calls and responses append above the line, so it moves down the transcript as work finishes, and the auto-following pane keeps it in view. At turn end it freezes to a permanent record: ✓ done · 38s · 1.8k reasoning · 2.4k total The timer and phrase span the WHOLE turn — started once at the first model call and never reset per tool call (unlike the old per-call spinner). Only turn_finished freezes it. The reasoning figure is a chars/4 estimate (no tokenizer); the done line's total is the exact summed output-token count from the turn stats. The readout is gated on [ui] show_reasoning_summary (default on; off → plane/phrase/timer only), which until now was dead config. Animation rides the Application's refresh_interval; the indicator render bypasses the width cache while active. A dangling indicator from an errored turn is discarded at the next turn's start so it never poisons the following timer (full turn-failure UX is a later branch). Only AppUI gains this; the default TerminalUI REPL is unchanged. DESIGN.md §31.14 added. --- docs/DESIGN.md | 14 +++ shellpilot/cli/app.py | 13 ++ shellpilot/cli/app_turn.py | 10 +- shellpilot/cli/app_ui.py | 146 ++++++++++++++++++++-- shellpilot/cli/terminal.py | 1 + tests/test_app_turn.py | 5 +- tests/test_app_ui.py | 241 +++++++++++++++++++++++++++++++++++-- 7 files changed, 400 insertions(+), 30 deletions(-) diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 4ed5458..f6e27ac 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -2835,6 +2835,20 @@ The interactive REPL is migrating from a per-turn `prompt_toolkit` `prompt` to a The opt-in runnable entry is `run_app` (`cli/app_main.py`), reached ONLY via `SHELLPILOT_UI=app` from `run_interactive` so a default session (TTY and non-TTY) stays byte-identical. The construction cycle — `schedule` needs `app`, `app` needs `runner.start`, `runner` needs the conversation, the conversation needs the `ThreadedUI`/`schedule` — is broken by build order and deferred attribute assignment: `AppUI` is built first (its `width_fn` reads `get_app()` lazily), then `TurnRunner` (its `schedule` reads `runner.app` lazily), then `ThreadedUI` over them; the conversation is constructed **with that `ThreadedUI` as its UI** (so its plan tools capture the marshaling bound methods — not repointed after the fact); then `build_app(on_submit=runner.start, ui=app_ui, …)`; then `runner.app`/`runner.conversation` are set after the app exists and before `app.run()`, by which time they are only read once a turn runs. The UI choice is made in `run_interactive` where the conversation is built (a `ThreadedUI` for app mode, the same `TerminalUI` as before otherwise); the default branch is unchanged. As an opt-in dev/live-test entry it has two known gaps to close when the app is promoted to the shipping default: `run_app` does not write the `session_end` audit event the default REPL emits, and a turn still in flight when `/exit` clears `app.loop` drops its final `busy`-clear in the already-dead app (harmless while the process is exiting). +### 31.14 Turn-scoped thinking indicator (v2) + +The full-screen pane shows a **live frontier line** while a turn runs, so a long (especially cloud-reasoning) turn is visibly alive instead of a blind spinner. The indicator is **turn-scoped**: it spans the whole tool loop, from the turn's first model call to the end of the turn, and never resets on a per-tool-call boundary. + +**Placement (it moves down).** The active line always renders LAST: `AppUI._render_ansi` builds the render list as `[*renderables, , ]`. Completed content (responses, tool calls) is appended to `_renderables`, so it lands ABOVE the live line, which therefore "moves down" the transcript as the turn progresses and ends at the bottom. Because the pane auto-follows the bottom (§31.12), the frontier line stays visible the whole turn, then freezes in place. + +**State machine.** `AppUI` holds a `_TurnIndicator | None` (`start`, `reasoning_chars`) and an injected clock `time_fn` (defaults to `time.monotonic`; tests pass a fake). `begin_response` starts the indicator **only if one is not already active** (records `start = time_fn()` once); a later `begin_response` in the same turn — the next tool-loop iteration — is a no-op for the indicator, so the timer, phrase, and reasoning count carry across tool calls (this is the deliberate departure from the v0.10.1 AviationSpinner, which `start`/`stop`ed per model call and reset every tool call). `end_response` does NOT touch the indicator — it only closes the open response, so the indicator keeps running across the tool call into the next model call. `stream_thinking(text)` adds `len(text)` to `reasoning_chars` while active (so the count climbs only while the model thinks, freezing when it stops and resuming if it thinks again). ONLY `turn_finished` freezes the indicator: it appends a permanent `✓ done` line and clears `_indicator`. + +**Lines.** Live: `{plane} {phrase}… {N}s · {fmt(reasoning)} reasoning`, where the plane is `Glyphs.spinner_frames` advanced one cell every `FRAME_SECONDS` (≈0.15s) of elapsed time, the phrase is a DETERMINISTIC pick from the current flight phase's pool (reusing `cli/streaming.py` `phase_for_elapsed`, indexed by 10-second buckets — not random, so it is testable), `N = int(elapsed)` with `elapsed = time_fn() - start`, and `reasoning` is the chars→tokens estimate `ceil(reasoning_chars / CHARS_PER_TOKEN)` (consistent with `budget.estimate_tokens` — an estimate, not exact). Done: `✓ done · {N}s · {fmt(reasoning)} reasoning · {fmt(total)} total`, where `N = int(stats.elapsed_s)` (the runtime's authoritative elapsed, not the UI clock) and `total = stats.output_tokens` (the exact summed `eval_count`). `fmt` is a k/m abbreviation (`999`→`"999"`, `1800`→`"1.8k"`, `2_400_000`→`"2.4m"`). + +**Reasoning gate.** The reasoning/total readout is gated on `settings.ui.show_reasoning_summary` (default true; previously a dead flag — this is its consumer), threaded into `AppUI(show_reasoning=…)` at both construction sites (`build_app`, and `terminal.py` app mode passing the setting). When false, the live line is `{plane} {phrase}… {N}s` and the done line is `✓ done · {N}s` (plane/phrase/timer only). This is display-only; audit capture of thinking is unaffected. + +**Animation and caching.** `build_app` sets the `Application`'s `refresh_interval=0.1` so the timer ticks and the plane glides even between thinking chunks. While an indicator is active, `_render_ansi` bypasses the width cache entirely (elapsed changes every render, so it never reads or writes the cache); when idle, the per-tick render is a width-cache hit (a cheap no-op). `AppUI` stays pure state+render — the clock is injected and the repaint is the app's job, so it never calls `get_app()`/`invalidate()`. A turn that errors before `turn_finished` (or, later, a Ctrl-C cancel) leaves the indicator active; the full turn-failure/cancel robustness is a later branch. As a minimal cross-turn guard, `show_user_message` discards any such dangling indicator at the start of the next turn, so an errored turn never poisons the following turn's timer (without it, the next `begin_response` would be a no-op against the stale indicator and count from the old start). + ## 32. Model Selection And Preload ### 32.1 Boot Model Picker diff --git a/shellpilot/cli/app.py b/shellpilot/cli/app.py index 47a1929..fc6e57c 100644 --- a/shellpilot/cli/app.py +++ b/shellpilot/cli/app.py @@ -144,6 +144,7 @@ def build_app( commands: Sequence[str], is_cloud: bool = False, ctx_pct: int = 0, + show_reasoning: bool = True, input: Input | None = None, output: Output | None = None, ui: AppUI | None = None, @@ -174,6 +175,7 @@ def build_app( glyphs=glyphs, workspace=workspace, width_fn=lambda: get_app().output.get_size().columns, + show_reasoning=show_reasoning, ) # Explicit AppUI annotation (not AppUI | None) so the closures below capture a # non-optional type; ui is already narrowed to AppUI by the guard above. @@ -298,6 +300,11 @@ def _submit(event: KeyPressEvent) -> None: return if text.strip(): if on_submit is not None: + # Echo the typed line into the pane on the loop thread, then run + # the turn. The live indicator (started by the turn's first + # begin_response) renders just below this echo and moves down as + # content appends above it (design section 31.14). + _ui.show_user_message(text) on_submit(text) else: _ui.show_status(text) @@ -329,6 +336,12 @@ def _interrupt(event: KeyPressEvent) -> None: key_bindings=kb, full_screen=True, mouse_support=True, + # Periodic repaint so the live thinking indicator's timer ticks and the + # plane glides even between thinking chunks (design section 31.14). When + # idle the per-tick AppUI._render_ansi is a width-cache hit (cheap no-op). + # NOTE: gating the refresh on an active turn is the upgrade path if + # profiling ever shows the idle tick costs anything (branch 5). + refresh_interval=0.1, input=input, output=output, ) diff --git a/shellpilot/cli/app_turn.py b/shellpilot/cli/app_turn.py index 5db6073..394359c 100644 --- a/shellpilot/cli/app_turn.py +++ b/shellpilot/cli/app_turn.py @@ -67,14 +67,14 @@ def __init__(self, *, inner: RuntimeUI, schedule: Schedule) -> None: def stream_token(self, token: str) -> None: self._schedule(functools.partial(self._inner.stream_token, token)) - # NOTE: stream_thinking is marshaled even though AppUI's is a no-op today; - # branch 5 gives it a real reasoning-readout consumer. + # stream_thinking drives the live reasoning readout in AppUI (§31.14); it is + # marshaled onto the loop thread like every other content call. def stream_thinking(self, text: str) -> None: self._schedule(functools.partial(self._inner.stream_thinking, text)) - # NOTE: begin_response is marshaled even though AppUI's is a no-op today; the - # waiting/thinking indicator is wired in branch 5. No args → the bound method - # is already a zero-arg Scheduled, so no partial is needed. + # begin_response starts AppUI's turn-scoped thinking indicator (§31.14). No + # args → the bound method is already a zero-arg Scheduled, so no partial is + # needed. def begin_response(self) -> None: self._schedule(self._inner.begin_response) diff --git a/shellpilot/cli/app_ui.py b/shellpilot/cli/app_ui.py index 6b3ede0..032356b 100644 --- a/shellpilot/cli/app_ui.py +++ b/shellpilot/cli/app_ui.py @@ -8,14 +8,17 @@ W while prompt_toolkit handles layout and scrolling. ``AppUI`` is a pure state-and-render object: it never calls ``get_app()`` or -``invalidate()`` — those are branch-4 concerns — so it is fully testable without a -running app. +``invalidate()`` — periodic repaint is the app's job (``refresh_interval``) and the +clock is injected (``time_fn``) — so it is fully testable without a running app. """ from __future__ import annotations import io +import math +import time from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING @@ -27,8 +30,10 @@ from shellpilot.cli.render import _sanitize_line, plan_step_line from shellpilot.cli.render import tool_call as render_tool_call from shellpilot.cli.render import tool_result as render_tool_result +from shellpilot.cli.streaming import phase_for_elapsed from shellpilot.cli.theme import SHELLPILOT_THEME, UNICODE_GLYPHS, Glyphs from shellpilot.memory.redaction import redact_structure +from shellpilot.runtime.budget import CHARS_PER_TOKEN from shellpilot.tools.base import workspace_display if TYPE_CHECKING: @@ -36,6 +41,37 @@ from shellpilot.runtime.events import TurnStats from shellpilot.runtime.planner import TaskPlan +# The plane glyph advances one track cell every FRAME_SECONDS of elapsed time, so +# the glide is a pure function of the (injected) clock — no animation thread. +FRAME_SECONDS = 0.15 + + +def _fmt_count(n: int) -> str: + """k/m abbreviation for token counts (design section 31.14). + + ``999`` → ``"999"``; ``1800`` → ``"1.8k"``; ``2_400_000`` → ``"2.4m"``. The + 1m check precedes the 1k check so a seven-figure count never renders as ``k``. + """ + if n >= 1_000_000: + return f"{n / 1_000_000:.1f}m" + if n >= 1000: + return f"{n / 1000:.1f}k" + return str(n) + + +@dataclass +class _TurnIndicator: + """Turn-scoped live-frontier state (design section 31.14). + + ``start`` is the injected-clock timestamp of the turn's first model call; + ``reasoning_chars`` accumulates the length of every thinking chunk. The token + estimate is derived at render time (``ceil(chars / CHARS_PER_TOKEN)``, + consistent with ``budget.estimate_tokens``) — chars are stored, not tokens. + """ + + start: float + reasoning_chars: int = 0 + class AppUI: """RuntimeUI implementation that routes content into the full-screen app pane. @@ -50,6 +86,8 @@ def __init__( glyphs: Glyphs = UNICODE_GLYPHS, workspace: Path | None = None, width_fn: Callable[[], int], + show_reasoning: bool = True, + time_fn: Callable[[], float] = time.monotonic, ) -> None: self._glyphs = glyphs # Workspace for display-integrity (design section 14.5): when set, a @@ -57,12 +95,24 @@ def __init__( # workspace-relative target — the SAME resolution the tool acts on. self._workspace = workspace self._width_fn = width_fn + # Gate for the reasoning-token readout (settings.ui.show_reasoning_summary, + # design section 31.14): when False, the live/done lines show plane+phrase+ + # timer only — no reasoning estimate, no total. Display-only; audit capture + # of thinking is unaffected. + self._show_reasoning = show_reasoning + # Injected clock so the indicator's elapsed/animation is testable with a + # fixed time_fn rather than the wall clock. + self._time_fn = time_fn # Source of truth: every committed renderable in the transcript. self._renderables: list[RenderableType] = [] # Accumulated token text for the current open response. # None means no response is open; an open response is the last renderable # in spirit but lives here until end_response() finalizes it. self._open_response: str | None = None + # Turn-scoped live indicator (design section 31.14): None when idle, a + # _TurnIndicator while a turn runs. begin_response starts it; turn_finished + # freezes it to a permanent done line and clears it. + self._indicator: _TurnIndicator | None = None # Width-keyed ANSI cache: (width, ansi_string), or None when stale. self._cache: tuple[int, str] | None = None @@ -96,7 +146,13 @@ def _render_ansi(self) -> str: caller supplies the width source via ``width_fn`` at construction time. """ width = self._width_fn() - if self._cache is not None and self._cache[0] == width: + active = self._indicator is not None + # An active indicator's elapsed/animation changes every render, so it + # bypasses the width cache entirely (never read, never write). Idle + # renders stay cached by width — a refresh tick with no active turn is a + # cheap cache hit. NOTE: gating refresh on an active turn is the upgrade + # path (branch 5) if profiling ever shows the idle tick costs anything. + if not active and self._cache is not None and self._cache[0] == width: return self._cache[1] buf = io.StringIO() console = Console( @@ -112,12 +168,40 @@ def _render_ansi(self) -> str: # model text at the sink (mirrors ResponseStream, streaming.py:171) — # _sanitize_line keeps LF, so markdown structure survives. renderables.append(Markdown(_sanitize_line(self._open_response))) + if active: + # The live frontier line always renders LAST, below all completed + # content, so it "moves down" as tool calls/responses append above it. + renderables.append(self._indicator_line()) for renderable in renderables: console.print(renderable) ansi = buf.getvalue() - self._cache = (width, ansi) + if not active: + self._cache = (width, ansi) return ansi + def _indicator_line(self) -> Text: + """The live frontier line for the active turn (design section 31.14). + + ``{plane} {phrase}… {N}s · {fmt(reasoning)} reasoning`` — the plane glides + a track cell every ``FRAME_SECONDS`` and the phrase is a DETERMINISTIC pick + from the current flight phase's pool (10-second buckets, not random), so + the whole line is reproducible under a fixed ``time_fn``. + """ + indicator = self._indicator + assert indicator is not None # only called from _render_ansi when active + elapsed = self._time_fn() - indicator.start + frames = self._glyphs.spinner_frames + plane = frames[int(elapsed / FRAME_SECONDS) % len(frames)] + pool = phase_for_elapsed(elapsed).pool + phrase = pool[int(elapsed / 10) % len(pool)] + line = Text() + line.append(plane, style="sp.accent") + line.append(f" {phrase}{self._glyphs.ellipsis} {int(elapsed)}s", style="sp.dim") + if self._show_reasoning: + reasoning = math.ceil(indicator.reasoning_chars / CHARS_PER_TOKEN) + line.append(f" · {_fmt_count(reasoning)} reasoning", style="sp.dim") + return line + # ------------------------------------------------------------------ # RuntimeUI content methods — mirroring TerminalUI exactly # ------------------------------------------------------------------ @@ -134,17 +218,59 @@ def end_response(self) -> None: """Close the open response so the next stream_token starts a fresh one.""" self._close_open_response() - # NOTE: begin_response is a no-op here; the waiting indicator is wired in branch 4/5. def begin_response(self) -> None: - return + # Turn-scoped: the FIRST model call of a turn starts the live indicator; + # later calls within the same tool loop do NOT restart it (the elapsed + # timer and reasoning count span the whole turn, not each model call). + if self._indicator is None: + self._indicator = _TurnIndicator(start=self._time_fn()) + self._cache = None - # NOTE: turn_finished is a no-op here; post-turn stats live in the status bar (branch 4). def turn_finished(self, stats: TurnStats) -> None: - return + # Freeze the live frontier into a permanent done line and clear the active + # indicator. Elapsed comes from the runtime's authoritative stats.elapsed_s + # (not the UI clock); reasoning is frozen from the accumulated chars; total + # is the exact summed output-token count. + # NOTE: only the success path freezes — a turn that errors before + # turn_finished (or a Ctrl-C cancel) leaves the indicator active; that + # turn-failure/cancel robustness is branch 6's. + reasoning_chars = self._indicator.reasoning_chars if self._indicator is not None else 0 + self._indicator = None + line = Text() + line.append(self._glyphs.check, style="sp.accent") + line.append(" done", style="sp.dim") + line.append(f" · {int(stats.elapsed_s)}s", style="sp.dim") + if self._show_reasoning: + reasoning = math.ceil(reasoning_chars / CHARS_PER_TOKEN) + line.append(f" · {_fmt_count(reasoning)} reasoning", style="sp.dim") + line.append(f" · {_fmt_count(stats.output_tokens)} total", style="sp.dim") + self._add_renderable(line) - # NOTE: stream_thinking is a no-op here; the reasoning readout is wired in branch 5. def stream_thinking(self, text: str) -> None: - return + # The reasoning count climbs ONLY while the model is thinking, so it + # freezes naturally when thinking stops and resumes if it thinks again. No + # reasoning TEXT is ever rendered — only its char count, as a token + # estimate. A no-op when no turn is active (defensive; begin_response runs + # first in the runtime's tool loop). + if self._indicator is not None: + self._indicator.reasoning_chars += len(text) + self._cache = None + + def show_user_message(self, text: str) -> None: + # Echo the submitted user message into the transcript. App-side (NOT a + # RuntimeUI protocol method) — the full-screen analogue of the REPL + # leaving the typed line in scrollback. The user-controlled text is + # control-char sanitized (display-integrity) before it reaches the pane. + # + # A new turn begins here, so discard any indicator left dangling by a + # PRIOR turn that errored before turn_finished (the error was already + # surfaced via show_error). Without this, the next begin_response would be + # a no-op against the stale indicator and the new turn's timer would count + # from the old turn's start. Full turn-failure/cancel UX is branch 6's; this + # is the minimal guard so an error never poisons the following turn. + self._indicator = None + echo = f"{self._glyphs.chevron} {_sanitize_line(text)}" + self._add_renderable(Text(echo, style="sp.accent")) def show_status(self, text: str) -> None: self._add_renderable(Text(_sanitize_line(text), style="sp.dim")) diff --git a/shellpilot/cli/terminal.py b/shellpilot/cli/terminal.py index 1473da2..8c3975c 100644 --- a/shellpilot/cli/terminal.py +++ b/shellpilot/cli/terminal.py @@ -615,6 +615,7 @@ def _preload(model_name: str) -> None: glyphs=glyphs, workspace=workspace, width_fn=lambda: get_app().output.get_size().columns, + show_reasoning=settings.ui.show_reasoning_summary, ) app_runner = TurnRunner(inner_ui=app_ui) ui = ThreadedUI(inner=app_ui, schedule=app_runner.schedule) diff --git a/tests/test_app_turn.py b/tests/test_app_turn.py index 1b1ae89..7543b3e 100644 --- a/tests/test_app_turn.py +++ b/tests/test_app_turn.py @@ -361,8 +361,9 @@ def test_build_app_on_submit_receives_text(tmp_path: Path) -> None: inp.send_text("/exit\n") app.run() assert received == ["hello there"] - # on_submit handled it; the inert branch-3 echo did NOT fire into the pane. - assert "hello there" not in ui._render_ansi() + # The submit echoes the user message into the pane (branch 5, §31.14) BEFORE + # routing the text to on_submit, so the typed line is now visible. + assert "hello there" in ui._render_ansi() def test_build_app_on_submit_none_falls_back_to_echo(tmp_path: Path) -> None: diff --git a/tests/test_app_ui.py b/tests/test_app_ui.py index 425bdba..89380f6 100644 --- a/tests/test_app_ui.py +++ b/tests/test_app_ui.py @@ -3,6 +3,7 @@ from __future__ import annotations import re +from collections.abc import Callable from pathlib import Path import pytest @@ -12,6 +13,7 @@ from shellpilot.memory.redaction import REDACTED from shellpilot.policy.approvals import ApprovalRequest from shellpilot.policy.risk import RiskLevel +from shellpilot.runtime.events import TurnStats from shellpilot.runtime.planner import PlanStep, TaskPlan GLYPHS = UNICODE_GLYPHS @@ -347,27 +349,240 @@ def test_ask_plan_approval_raises_not_implemented() -> None: # --------------------------------------------------------------------------- -# No-op stubs — must not raise or produce output +# stream_thinking with no active turn renders nothing (only the count is ever +# surfaced, and only while a turn runs) # --------------------------------------------------------------------------- -def test_begin_response_is_noop() -> None: +def test_stream_thinking_without_active_turn_renders_nothing() -> None: ui = make_ui() - ui.begin_response() # must not raise + ui.stream_thinking("deep thoughts") # no begin_response → no indicator + assert "deep thoughts" not in plain(ui) assert plain(ui) == "" -def test_stream_thinking_is_noop() -> None: - ui = make_ui() - ui.stream_thinking("deep thoughts") - assert "deep thoughts" not in plain(ui) +# --------------------------------------------------------------------------- +# §31.14 — turn-scoped live thinking indicator +# --------------------------------------------------------------------------- -def test_turn_finished_is_noop() -> None: - from shellpilot.runtime.events import TurnStats +def _clock(values: list[float]) -> Callable[[], float]: + """A fake monotonic clock returning successive values, holding the last one.""" + state = {"i": 0} - ui = make_ui() - ui.turn_finished( - TurnStats(elapsed_s=1.5, context_tokens=500, context_pct=6, warn=False, output_tokens=80) + def now() -> float: + i = min(state["i"], len(values) - 1) + state["i"] += 1 + return values[i] + + return now + + +def make_stats(*, elapsed_s: float = 2.0, output_tokens: int = 80) -> TurnStats: + return TurnStats( + elapsed_s=elapsed_s, + context_tokens=500, + context_pct=6, + warn=False, + output_tokens=output_tokens, ) - assert plain(ui) == "" + + +@pytest.mark.parametrize( + "n,expected", + [ + (0, "0"), + (999, "999"), + (1000, "1.0k"), + (1800, "1.8k"), + (999_999, "1000.0k"), + (1_000_000, "1.0m"), + (2_400_000, "2.4m"), + ], +) +def test_fmt_count_boundaries(n: int, expected: str) -> None: + from shellpilot.cli.app_ui import _fmt_count + + assert _fmt_count(n) == expected + + +def test_begin_response_starts_active_indicator() -> None: + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=lambda: 0.0) + ui.begin_response() + # The plane glyph (first spinner frame) renders → the indicator is active. + assert GLYPHS.spinner_frames[0] in ansi_text(ui) + # The flight phrase + an elapsed timer are on the live line. + assert "taxiing" in plain(ui) + assert "0s" in plain(ui) + + +def test_second_begin_response_does_not_restart_indicator() -> None: + # First begin_response at t=0 starts the indicator; a second at t=30 must NOT + # restart it — the elapsed timer keeps climbing from the original start. + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=_clock([0.0, 30.0, 30.0])) + ui.begin_response() # consumes 0.0 → start + ui.begin_response() # consumes 30.0 → must be ignored (no restart) + out = plain(ui) # consumes 30.0 → render at elapsed 30 + assert "30s" in out # elapsed measured from the ORIGINAL start, not the second call + + +def test_indicator_runs_continuously_across_tool_calls() -> None: + # LOAD-BEARING regression (§31.14): the timer + phrase + reasoning count must + # span the WHOLE turn, NOT reset per model call. This mirrors the real tool + # loop: begin_response → think → end_response → (tool call) → begin_response + # again. The second begin_response (next tool-loop iteration) must be a NO-OP + # for the indicator — start, phrase, and reasoning count all carry over. + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=_clock([3.0, 45.0])) + ui.begin_response() # turn start recorded at t=3.0 (first model call) + start_before = ui._indicator.start # type: ignore[union-attr] + ui.stream_thinking("a" * 2000) # 500 tokens of reasoning in the first call + ui.end_response() # model call ended — must NOT stop/reset the indicator + ui.show_tool_call("read_file", {"path": "x"}) # a tool call between model calls + ui.begin_response() # next tool-loop iteration — must NOT restart the indicator + # The turn-start timestamp is unchanged (no restart). + assert ui._indicator is not None + assert ui._indicator.start == start_before == 3.0 + # The reasoning count carried over (NOT reset to 0 by end_response or the + # second begin_response). + assert ui._indicator.reasoning_chars == 2000 + # Elapsed counts up monotonically from the single turn-start: render at t=45 + # shows 42s (45 - 3), and the frozen 500-token reasoning estimate persists. + out = plain(ui) + assert "42s" in out + assert "500 reasoning" in out + + +def test_end_response_does_not_touch_the_indicator() -> None: + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=lambda: 0.0) + ui.begin_response() + ui.stream_thinking("b" * 1200) # 300 tokens + ui.end_response() + # end_response only closes the open response; the indicator stays active with + # its reasoning count intact. + assert ui._indicator is not None + assert ui._indicator.reasoning_chars == 1200 + assert "300 reasoning" in plain(ui) + + +def test_new_turn_after_error_starts_fresh_indicator() -> None: + # A turn that errors before turn_finished leaves the indicator dangling. The + # NEXT turn's show_user_message must discard it so the new begin_response starts + # fresh — otherwise the no-op-while-active begin_response would keep the OLD + # start and the new turn's timer would read inflated (the cross-turn poisoning + # the §31.14 guard prevents; full turn-failure UX is branch 6's). + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=_clock([3.0, 100.0])) + ui.begin_response() # turn 1 starts at t=3 + ui.stream_thinking("x" * 2000) # 500 tokens of reasoning + assert ui._indicator is not None # ... then turn 1 errors: no turn_finished + ui.show_user_message("next question") # turn 2 begins → discard the stale one + assert ui._indicator is None + ui.begin_response() # turn 2 — a FRESH indicator at t=100, not the stale t=3 + assert ui._indicator is not None + assert ui._indicator.start == 100.0 + assert ui._indicator.reasoning_chars == 0 + + +def test_stream_thinking_climbs_reasoning_estimate() -> None: + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=lambda: 0.0) + ui.begin_response() + assert "0 reasoning" in plain(ui) + # 4400 chars / CHARS_PER_TOKEN(4) = 1100 tokens → "1.1k reasoning". + ui.stream_thinking("x" * 4400) + assert "1.1k reasoning" in plain(ui) + # Thinking stops → the count freezes (more renders do not change it). + assert "1.1k reasoning" in plain(ui) + + +def test_live_line_phrase_and_elapsed_with_fixed_clock() -> None: + # At elapsed 23s the phase is "cruise" (start=20); phrase index = int(23/10)%len + # = 2 → the 3rd cruise phrase ("scanning the instruments"). Reasoning 400 chars + # / 4 = 100 tokens. + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=_clock([0.0, 23.0])) + ui.begin_response() # start at 0.0 + ui.stream_thinking("y" * 400) + out = plain(ui) # render at 23.0 + assert "scanning the instruments" in out + assert "23s" in out + assert "100 reasoning" in out + + +def test_turn_finished_freezes_done_line_and_clears_indicator() -> None: + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=lambda: 0.0) + ui.begin_response() + ui.stream_thinking("z" * 2000) # 500 tokens + ui.turn_finished(make_stats(elapsed_s=7.0, output_tokens=1800)) + out = plain(ui) + # The frozen done line: check glyph, authoritative elapsed, frozen reasoning, + # exact total (k-formatted). + assert GLYPHS.check in out + assert "done" in out + assert "7s" in out + assert "500 reasoning" in out + assert "1.8k total" in out + # The indicator is cleared: no live plane frame remains in the transcript. + assert ui._indicator is None + # A fresh begin_response starts a brand-new indicator (count back to 0). + ui.begin_response() + assert "0 reasoning" in plain(ui) + + +def test_show_reasoning_false_omits_reasoning_and_total() -> None: + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, show_reasoning=False, time_fn=lambda: 0.0) + ui.begin_response() + ui.stream_thinking("q" * 4000) + live = plain(ui) + assert "reasoning" not in live # live line is plane/phrase/timer only + assert "taxiing" in live and "0s" in live + ui.turn_finished(make_stats(elapsed_s=3.0, output_tokens=1234)) + done = plain(ui) + assert "done" in done and "3s" in done + assert "reasoning" not in done + assert "total" not in done + + +def test_active_indicator_render_bypasses_width_cache() -> None: + # Two renders at the SAME width with the clock advanced must differ (live + # elapsed) — the active-turn render never reads or writes the width cache. + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=_clock([0.0, 5.0, 12.0])) + ui.begin_response() # start at 0.0 + first = ui._render_ansi() # elapsed 5 + second = ui._render_ansi() # elapsed 12 + assert first is not second + assert "5s" in strip_ansi(first) + assert "12s" in strip_ansi(second) + # No cache was written while active. + assert ui._cache is None + + +def test_show_user_message_echoes_with_chevron() -> None: + ui = make_ui() + ui.show_user_message("run the tests") + out = plain(ui) + assert GLYPHS.chevron in out + assert "run the tests" in out + + +def test_show_user_message_sanitizes_control_chars() -> None: + ui = make_ui() + ui.show_user_message("clean\x1b[2J\x00\x07injected") + raw = ansi_text(ui) + assert "\x1b[2J" not in raw + assert "\x00" not in raw + assert "\x07" not in raw + assert "clean" in plain(ui) + assert "injected" in plain(ui) + + +def test_frontier_ordering_user_then_content_then_done() -> None: + # user message echoed, then a tool call renders ABOVE the live indicator, then + # turn_finished freezes the done line LAST: order is `❯ msg`, tool call, `✓ done`. + ui = AppUI(glyphs=GLYPHS, width_fn=lambda: 80, time_fn=lambda: 0.0) + ui.show_user_message("do the thing") + ui.begin_response() + ui.show_tool_call("read_file", {"path": "notes.txt"}) + ui.turn_finished(make_stats()) + out = plain(ui) + i_msg = out.index("do the thing") + i_tool = out.index("read_file") + i_done = out.index("done") + assert i_msg < i_tool < i_done