feat: improve peek UX with scrollbar, thinking traces, and clearer hints

Add scrollbar to full-screen peek, show thinking blocks and full text
in the activity feed, display full tool args for Agent/ToolSearch,
show context token count as "ctx N", and add persistent Shift+P hint.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 2 people

src/ralphify/_console_emitter.py

@@ -144,18 +144,30 @@ def _shorten_path(path: str, max_len: int = 48) -> str:
     return "…/" + tail[-(max_len - 2) :]
 
 
+def _format_params(tool_input: dict[str, Any], keys: list[str]) -> str:
+    """Format specified tool parameters as ``key: value`` pairs."""
+    parts = []
+    for key in keys:
+        val = tool_input.get(key)
+        if val is not None:
+            parts.append(f"{key}: {val}")
+    return " · ".join(parts) if parts else ""
+
+
 _TOOL_ARG_EXTRACTORS: dict[str, Callable[[dict[str, Any]], str]] = {
     "Read": lambda i: _shorten_path(i.get("file_path", "")),
     "Write": lambda i: _shorten_path(i.get("file_path", "")),
     "Edit": lambda i: _shorten_path(i.get("file_path", "")),
     "MultiEdit": lambda i: _shorten_path(i.get("file_path", "")),
     "Glob": lambda i: i.get("pattern", ""),
     "Grep": lambda i: i.get("pattern", ""),
-    "Bash": lambda i: _truncate(i.get("command", ""), 80),
-    "Task": lambda i: _truncate(i.get("description", i.get("prompt", "")), 80),
+    "Bash": lambda i: i.get("command", ""),
+    "Task": lambda i: _format_params(i, ["description", "prompt"]),
     "WebFetch": lambda i: i.get("url", ""),
     "WebSearch": lambda i: i.get("query", ""),
     "TodoWrite": lambda i: f"{len(i.get('todos', []))} todos",
+    "Agent": lambda i: _format_params(i, ["description", "prompt"]),
+    "ToolSearch": lambda i: _format_params(i, ["query", "max_results"]),
 }
 
 
@@ -300,41 +312,31 @@ def set_peek_visible(self, visible: bool) -> None:
 
     # ── Stream-json processing ───────────────────────────────────────
 
-    def apply(self, raw: dict[str, Any]) -> str | None:
+    def apply(self, raw: dict[str, Any]) -> None:
         """Update panel state from a parsed stream-json dict.
 
-        Returns the scroll-line markup string (or ``None``).  The line is
-        also appended to the internal scroll buffer so it renders inside
-        the Live region.
+        Each handler appends its own scroll lines to the buffer so that
+        multi-line blocks (thinking, long text) can produce several rows.
         """
         event_type = raw.get("type")
 
         if event_type == "system" and raw.get("subtype") == "init":
             self._model = raw.get("model", "")
-            return None
-
-        scroll_line: str | None = None
-
-        if event_type == "assistant":
-            scroll_line = self._apply_assistant(raw)
+        elif event_type == "assistant":
+            self._apply_assistant(raw)
         elif event_type == "user":
-            scroll_line = self._apply_user(raw)
+            self._apply_user(raw)
         elif event_type == "rate_limit_event":
             info = raw.get("rate_limit_info", {})
             status = info.get("status", "")
             resets = info.get("resetsAt", "")
-            scroll_line = (
+            self.add_scroll_line(
                 f"[bold {_brand.PEACH}]⚠ rate limit:[/]"
                 f" [dim]{escape_markup(str(status))}"
                 f", resets {escape_markup(str(resets))}[/]"
             )
 
-        if scroll_line is not None:
-            self.add_scroll_line(scroll_line)
-
-        return scroll_line
-
-    def _apply_assistant(self, raw: dict[str, Any]) -> str | None:
+    def _apply_assistant(self, raw: dict[str, Any]) -> None:
         msg = raw.get("message", {})
 
         # Update token counts from usage
@@ -348,21 +350,30 @@ def _apply_assistant(self, raw: dict[str, Any]) -> str | None:
 
         content = msg.get("content", [])
         if not isinstance(content, list):
-            return None
+            return
 
-        scroll_line: str | None = None
         for block in content:
             if not isinstance(block, dict):
                 continue
             block_type = block.get("type")
 
-            if block_type == "text":
+            if block_type == "thinking":
+                text = block.get("thinking", "")
+                if text:
+                    for tline in text.split("\n"):
+                        self.add_scroll_line(
+                            f"[dim italic]{escape_markup(tline)}[/]"
+                        )
+
+            elif block_type == "text":
                 text = block.get("text", "")
-                preview = _truncate(text.replace("\n", " "), 100)
-                if preview:
-                    scroll_line = (
-                        f"[italic {_brand.LAVENDER}]“{escape_markup(preview)}”[/]"
-                    )
+                if text:
+                    for tline in text.split("\n"):
+                        if tline.strip():
+                            self.add_scroll_line(
+                                f"[italic {_brand.LAVENDER}]"
+                                f"\"{escape_markup(tline)}\"[/]"
+                            )
 
             elif block_type == "tool_use":
                 name = block.get("name", "?")
@@ -387,40 +398,40 @@ def _apply_assistant(self, raw: dict[str, Any]) -> str | None:
                 else:
                     name_col = f"{name}  "
                 if arg:
-                    scroll_line = (
+                    self.add_scroll_line(
                         f"[bold {color}]{escape_markup(name_col)}[/]"
                         f"[dim]{escape_markup(arg)}[/]"
                     )
                 else:
-                    scroll_line = f"[bold {color}]{escape_markup(name)}[/]"
-
-        return scroll_line
+                    self.add_scroll_line(
+                        f"[bold {color}]{escape_markup(name)}[/]"
+                    )
 
-    def _apply_user(self, raw: dict[str, Any]) -> str | None:
+    def _apply_user(self, raw: dict[str, Any]) -> None:
         msg = raw.get("message", {})
         content = msg.get("content", [])
         if not isinstance(content, list):
-            return None
+            return
         for block in content:
             if not isinstance(block, dict):
                 continue
             if block.get("type") == "tool_result" and block.get("is_error"):
                 snippet = _truncate(str(block.get("content", "")), 100)
-                return (
+                self.add_scroll_line(
                     f"[bold {_brand.DEEP_ORANGE}]{_ICON_FAILURE} tool error:[/]"
                     f" [dim]{escape_markup(snippet)}[/]"
                 )
-        return None
+                return
 
     def _format_tokens(self) -> str:
-        """Format token counts as compact ↑in ↓out string."""
+        """Format token counts as compact ctx/out string."""
         parts: list[str] = []
         total_in = self._input_tokens
         if total_in > 0:
-            parts.append(f"↑{self._format_count(total_in)}")
+            parts.append(f"ctx {self._format_count(total_in)}")
         if self._output_tokens > 0:
-            parts.append(f"↓{self._format_count(self._output_tokens)}")
-        return " ".join(parts)
+            parts.append(f"out {self._format_count(self._output_tokens)}")
+        return " · ".join(parts)
 
     @staticmethod
     def _format_count(n: int) -> str:
@@ -466,7 +477,7 @@ def _build_subtitle(self) -> Text | None:
         return sub
 
     def _build_footer(self) -> Table:
-        """Bottom row of the panel: spinner + tool counts."""
+        """Bottom row of the panel: spinner + tool counts + peek hint."""
         summary = Text(no_wrap=True, overflow="ellipsis")
         if self._tool_count > 0:
             summary.append(
@@ -480,10 +491,13 @@ def _build_footer(self) -> Table:
         else:
             summary.append("waiting for first tool call…", style="dim italic")
 
+        hint = Text("Shift+P full screen", style="dim", no_wrap=True)
+
         grid = Table.grid(expand=True)
         grid.add_column(width=2, no_wrap=True)
         grid.add_column(ratio=1, no_wrap=True, overflow="ellipsis")
-        grid.add_row(self._spinner, summary)
+        grid.add_column(no_wrap=True, justify="right")
+        grid.add_row(self._spinner, summary, hint)
         return grid
 
     def _build_body(self) -> Group:
@@ -601,7 +615,8 @@ def _build_header(self, total: int, visible: int) -> Text:
     def _build_footer(self) -> Text:
         hint = Text(no_wrap=True, overflow="ellipsis")
         hint.append(
-            " ↑/k up · ↓/j down · b/space page · g/G top/bottom · q/", style="dim"
+            " ↑/k up · ↓/j down · b page up · space page down · g/G top/bottom · q/",
+            style="dim",
         )
         hint.append(FULLSCREEN_PEEK_KEY, style=f"bold {_brand.PURPLE}")
         hint.append(" exit ", style="dim")
@@ -627,21 +642,52 @@ def __rich_console__(
         start = max(0, end - visible)
         window = lines[start:end]
 
+        # Scrollbar metrics
+        show_scrollbar = total > visible
+        thumb_start = 0
+        thumb_size = visible
+        if show_scrollbar:
+            thumb_size = max(1, visible * visible // total)
+            max_off_val = max(total - visible, 1)
+            frac = 1.0 - (self._offset / max_off_val)
+            track_space = visible - thumb_size
+            thumb_start = int(frac * track_space)
+
         rows: list[Any] = []
         rows.append(self._build_header(total, visible))
         rows.append(Text(""))
-        if window:
-            for line in window:
+
+        # Content area with optional scrollbar column
+        content = Table.grid(expand=True)
+        content.add_column(ratio=1, no_wrap=True, overflow="ellipsis")
+        if show_scrollbar:
+            content.add_column(width=1, no_wrap=True)
+
+        for i in range(visible):
+            if i < len(window):
+                line = window[i]
                 line.no_wrap = True
                 line.overflow = "ellipsis"
-                rows.append(line)
-        else:
+            else:
+                line = Text("")
+            if show_scrollbar:
+                in_thumb = thumb_start <= i < thumb_start + thumb_size
+                bar = Text(
+                    "█" if in_thumb else "│",
+                    style=_brand.PURPLE if in_thumb else "dim",
+                )
+                content.add_row(line, bar)
+            else:
+                content.add_row(line)
+
+        if not window and not show_scrollbar:
+            # Replace the empty grid with a waiting message
             rows.append(Text("  (waiting for activity…)", style="dim italic"))
-        # Pad the body so the footer hugs the bottom border even when
-        # the buffer is shorter than the viewport.
-        padding = visible - len(window)
-        for _ in range(max(0, padding)):
-            rows.append(Text(""))
+            for _ in range(max(0, visible - 1)):
+                rows.append(Text(""))
+        else:
+            rows.append(content)
+
         rows.append(Text(""))
         rows.append(self._build_footer())
 
@@ -1145,10 +1191,14 @@ def _build_footer(self) -> Table:
             summary.append(" of agent output", style="dim")
         else:
             summary.append("waiting for agent output…", style="dim italic")
+
+        hint = Text("Shift+P full screen", style="dim", no_wrap=True)
+
         grid = Table.grid(expand=True)
         grid.add_column(width=2, no_wrap=True)
         grid.add_column(ratio=1, no_wrap=True, overflow="ellipsis")
-        grid.add_row(self._spinner, summary)
+        grid.add_column(no_wrap=True, justify="right")
+        grid.add_row(self._spinner, summary, hint)
         return grid
 
     def _build_body(self) -> Group:

tests/test_console_emitter.py

@@ -587,8 +587,8 @@ def test_assistant_text_scroll_line(self):
         assert any("fix the bug" in line.plain for line in panel._scroll_lines)
         emitter._stop_live()
 
-    def test_thinking_does_not_scroll(self):
-        """Thinking events update the panel status but don't produce scroll output."""
+    def test_thinking_produces_scroll_lines(self):
+        """Thinking blocks appear in the scroll buffer as dim italic text."""
         emitter, console = self._make_structured_emitter()
         emitter.emit(_make_event(EventType.ITERATION_STARTED, iteration=1))
         emitter.emit(
@@ -605,7 +605,8 @@ def test_thinking_does_not_scroll(self):
         )
         panel = emitter._iteration_panel
         assert panel is not None
-        assert len(panel._scroll_lines) == 0
+        assert len(panel._scroll_lines) == 1
+        assert "let me think..." in panel._scroll_lines[0].plain
         emitter._stop_live()
 
     def test_rate_limit_scroll_line(self):
@@ -1348,7 +1349,7 @@ def test_panel_renders_elapsed(self):
 
     def test_apply_tool_use_updates_counters(self):
         panel = _IterationPanel()
-        result = panel.apply(
+        panel.apply(
             {
                 "type": "assistant",
                 "message": {
@@ -1362,17 +1363,16 @@ def test_apply_tool_use_updates_counters(self):
                 },
             }
         )
-        assert result is not None
-        assert "Read" in result
         assert panel._tool_count == 1
         assert panel._tool_categories.get("read") == 1
+        assert len(panel._scroll_lines) == 1
+        assert "Read" in panel._scroll_lines[0].plain
 
     def test_apply_system_init_sets_model(self):
         panel = _IterationPanel()
-        result = panel.apply(
+        panel.apply(
             {"type": "system", "subtype": "init", "model": "claude-opus-4-6"}
         )
-        assert result is None
         assert panel._model == "claude-opus-4-6"
 
     def test_apply_usage_updates_tokens(self):
@@ -1401,15 +1401,15 @@ def test_apply_unknown_type_returns_none(self):
     def test_format_tokens_does_not_double_count_cached_input(self):
         """The Anthropic API's input_tokens already includes cache_read_input_tokens
         as a subset.  _format_tokens must not add them again — that would inflate
-        the displayed total (e.g. 1000 input + 800 cached → wrong ↑1.8k instead
-        of correct ↑1.0k)."""
+        the displayed total (e.g. 1000 input + 800 cached → wrong ctx 1.8k instead
+        of correct ctx 1.0k)."""
         panel = _IterationPanel()
         panel._input_tokens = 1000
         panel._cache_read_tokens = 800
         panel._output_tokens = 200
         result = panel._format_tokens()
-        assert "↑1.0k" in result, (
-            f"Expected ↑1.0k (input_tokens already includes cache), got: {result!r}"
+        assert "ctx 1.0k" in result, (
+            f"Expected ctx 1.0k (input_tokens already includes cache), got: {result!r}"
         )
 
     def test_format_count(self):