feat(ui): turn-scoped thinking indicator with live reasoning readout

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, <open response>, <active indicator>]`. 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

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,
     )

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)
 

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,15 +30,48 @@
 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:
     from shellpilot.policy.approvals import ApprovalReply, ApprovalRequest
     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,19 +86,33 @@ 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
         # `path` argument in the tool-call line is shown as its resolved,
         # 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"))