feat(code): collapse completed tool calls into group summaries (#4373)

Collapses each assistant step's tool calls into a single collapsible
line in the code TUI — live progress while running, past tense when the
step finishes, click or Ctrl+O to expand. Errored, rejected, and skipped
tools stay visible.

The Thinking spinner now remains pinned as a stable turn-level indicator
while tool rows mount above it, and tool-group updates are coalesced to
avoid transcript flicker during streaming.

## Video


https://github.com/user-attachments/assets/8e69e331-77d7-46ed-bdc9-a8b33db9ba75

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Mason Daugherty <mason@langchain.dev>
Co-authored-by: Mason Daugherty <github@mdrxy.com>

libs/code/deepagents_code/app.py

@@ -83,10 +83,12 @@ from deepagents_code.widgets.message_store import (
 from deepagents_code.widgets.messages import (
     AppMessage,
     AssistantMessage,
+    DiffMessage,
     ErrorMessage,
     QueuedUserMessage,
     SkillMessage,
     ToolCallMessage,
+    ToolGroupSummary,
     UserMessage,
 )
 from deepagents_code.widgets.status import StatusBar
@@ -2377,6 +2379,10 @@ class DeepAgentsApp(App):
         """The `UserMessage` widget that started the in-flight turn, tracked so
         it can be dimmed if the turn is interrupted."""
 
+        self._active_tool_group: ToolGroupSummary | None = None
+        """Open tool-group summary for the current step. Tools are folded into
+        it as they stream and it is closed at the next step boundary."""
+
         self._shell_process: asyncio.subprocess.Process | None = None
         """Shell command process tracking for interruption (! commands)."""
 
@@ -5614,13 +5620,19 @@ class DeepAgentsApp(App):
         added_height = hydrated_count * estimated_height_per_message
         chat.scroll_y = old_scroll_y + added_height
 
-    async def _mount_before_queued(self, container: Container, widget: Widget) -> None:
-        """Mount a widget in the messages container, before any queued widgets.
+        # Collapse any completed tool runs brought in above the window so
+        # hydrated history matches the live transcript.
+        await self._regroup_completed_tools()
 
-        Queued-message widgets must stay at the bottom of the container so
-        they remain visually anchored below the current agent response.
-        This helper inserts `widget` just before the first queued widget,
-        or appends at the end when the queue is empty.
+    async def _mount_before_queued(self, container: Container, widget: Widget) -> None:
+        """Mount a widget in the messages container, kept above the bottom anchors.
+
+        The loading spinner and queued-message widgets must stay pinned at the
+        bottom of the container. New content mounts just above them — before the
+        spinner if it is present (so it never needs repositioning as tools
+        stream, which flickered), otherwise before the first queued widget,
+        otherwise appended at the end. The spinner itself anchors only on the
+        queued widgets so it can mount at the bottom.
 
         Args:
             container: The `#messages` container to mount into.
@@ -5628,13 +5640,24 @@ class DeepAgentsApp(App):
         """
         if not container.is_attached:
             return
-        first_queued = self._queued_widgets[0] if self._queued_widgets else None
-        if first_queued is not None and first_queued.parent is container:
+        anchor: Widget | None = None
+        spinner = self._loading_widget
+        if (
+            widget is not spinner
+            and spinner is not None
+            and spinner.parent is container
+        ):
+            anchor = spinner
+        else:
+            first_queued = self._queued_widgets[0] if self._queued_widgets else None
+            if first_queued is not None and first_queued.parent is container:
+                anchor = first_queued
+        if anchor is not None:
             try:
-                await container.mount(widget, before=first_queued)
+                await container.mount(widget, before=anchor)
             except Exception:
                 logger.warning(
-                    "Stale queued-widget reference; appending at end",
+                    "Stale mount anchor reference; appending at end",
                     exc_info=True,
                 )
             else:
@@ -5758,8 +5781,7 @@ class DeepAgentsApp(App):
         )
 
         if status is None:
-            # Hide
-            if self._loading_widget:
+            if self._loading_widget is not None:
                 await self._loading_widget.remove()
                 self._loading_widget = None
             if self._terminal_progress_enabled:
@@ -5784,8 +5806,10 @@ class DeepAgentsApp(App):
             # silently so the streaming loop doesn't crash.
             return
 
-        if self._loading_widget is None:
-            # Create new
+        if self._loading_widget is None or not self._loading_widget.is_attached:
+            # Mount once per turn. `_mount_before_queued` keeps new messages
+            # *above* the spinner, so it stays pinned at the bottom and never
+            # needs repositioning (which flickered) as tools stream in.
             self._loading_widget = LoadingWidget(status)
             await self._mount_before_queued(messages, self._loading_widget)
         else:
@@ -5794,10 +5818,9 @@ class DeepAgentsApp(App):
             # abandoned without completing the resume callback. `resume()` is a
             # no-op when the spinner is not paused.
             self._loading_widget.resume()
-            # Update existing
             self._loading_widget.set_status(status)
-            # Reposition via move_child so elapsed-time and animation state
-            # carry through; remove + re-mount would reset both.
+            # Safety fallback: messages now mount above the spinner so it should
+            # already be in place, but reposition if something left it stranded.
             if not self._is_spinner_at_correct_position(messages):
                 self._reposition_spinner(messages)
         # NOTE: Don't call anchor() here - it would re-anchor and drag user back
@@ -10244,6 +10267,16 @@ class DeepAgentsApp(App):
                 ),
                 turn_stats=turn_stats,
             )
+            # Close the final step's group once the turn ends with no trailing
+            # assistant text to trigger the boundary path. Grouping is cosmetic,
+            # so a failure here must not abort the turn — but log it, since
+            # `_mount_tool_group_summary` already handles its own mount errors and
+            # anything reaching this point is unexpected.
+            try:
+                self._close_active_tool_group()
+                await self._regroup_completed_tools()
+            except Exception:
+                logger.exception("Failed to close/regroup tool group at turn end")
         except Exception as e:  # Resilient tool rendering
             logger.exception("Agent execution failed")
             try:
@@ -10293,6 +10326,9 @@ class DeepAgentsApp(App):
             subagent_panel = self._get_subagent_panel()
             if subagent_panel is not None:
                 subagent_panel.finalize_running()
+            # Collapse the open tool group so an interrupted turn doesn't leave a
+            # summary spinning "Running…" forever (synchronous, cancel-safe).
+            self._close_active_tool_group()
             await self._cleanup_agent_task()
 
     async def _process_next_from_queue(self) -> None:
@@ -10952,20 +10988,59 @@ class DeepAgentsApp(App):
                 pass
             return
 
+        # Eagerly fold tool calls into a single live summary so they are
+        # collapsed from the moment they start, rather than rendering verbose
+        # then snapping shut. A groupable tool joins (or opens) the current
+        # step's group; a diff folds into it; anything else is a step boundary
+        # that closes the group.
+        is_groupable_tool = (
+            isinstance(widget, ToolCallMessage) and widget.tool_name != "ask_user"
+        )
+        is_diff = isinstance(widget, DiffMessage)
+
         # Store message data for virtualization
         message_data = MessageData.from_widget(widget)
         if not widget.id:
             # Keep the widget DOM id == store id so pruning can locate a
             # mounted widget (and its timestamp footer) from its MessageData.
             widget.id = message_data.id
-        self._message_store.append(message_data)
         footer = self._build_message_timestamp_footer(
             message_data, visible=self._message_timestamps_visible
         )
 
-        await self._mount_before_queued(messages, widget)
-        if footer is not None:
-            await self._mount_before_queued(messages, footer)
+        # Coalesce the whole mount-and-fold sequence into a single repaint.
+        # Otherwise mounting a groupable tool paints it at full height, then
+        # folding it into the group hides it on the next frame — bouncing the
+        # bottom-anchored transcript on every tool call.
+        with self.batch_update():
+            if not (is_groupable_tool or is_diff):
+                self._close_active_tool_group()
+                # Re-derive groups for any tools mounted outside this path
+                # (resumed history), which carry no live group.
+                await self._regroup_completed_tools()
+            elif is_groupable_tool and (
+                self._active_tool_group is None
+                or not self._active_tool_group.is_attached
+            ):
+                self._active_tool_group = ToolGroupSummary(live=True)
+                await self._mount_before_queued(messages, self._active_tool_group)
+
+            self._message_store.append(message_data)
+
+            await self._mount_before_queued(messages, widget)
+            if footer is not None:
+                await self._mount_before_queued(messages, footer)
+
+            # Fold the freshly-mounted tool/diff into the open group so it hides
+            # immediately (must run after mount so display toggles take effect).
+            if (
+                self._active_tool_group is not None
+                and self._active_tool_group.is_attached
+            ):
+                if is_groupable_tool:
+                    self._active_tool_group.add_member(widget)
+                elif is_diff:
+                    self._active_tool_group.add_collapsible(widget)
 
         # Prune old widgets if window exceeded
         await self._prune_old_messages()
@@ -11016,6 +11091,113 @@ class DeepAgentsApp(App):
 
         if pruned_ids:
             self._message_store.mark_pruned(pruned_ids)
+            # Drop any group summaries whose members were all pruned away so a
+            # stray collapsed line never lingers above the window. Only reachable
+            # when something was actually pruned this pass.
+            for summary in list(self.query(ToolGroupSummary)):
+                if not summary.has_attached_members:
+                    try:
+                        await summary.remove()
+                    except Exception:
+                        logger.debug(
+                            "Failed to remove orphaned tool group summary",
+                            exc_info=True,
+                        )
+
+    def _close_active_tool_group(self) -> None:
+        """Finalize the open tool group into its collapsed past-tense form."""
+        group = self._active_tool_group
+        self._active_tool_group = None
+        if group is not None and group.is_attached:
+            try:
+                group.close()
+            except Exception:
+                # Also runs on the interrupt/cancel finally path, so never
+                # re-raise. Log so a broken eviction (e.g. a failed tool left
+                # folded and hidden) surfaces instead of being swallowed.
+                logger.exception("Failed to close active tool group")
+
+    async def _regroup_completed_tools(self) -> None:
+        """Fold runs of completed tool calls into collapsible group summaries.
+
+        Scans the messages container for maximal runs of consecutive,
+        successfully-completed tool calls (optionally interleaved with their
+        diff previews) and inserts a `ToolGroupSummary` that collapses each run
+        into a single dim line. Footers stay transparent to the scan so a
+        timestamp row between two tools does not split a run.
+
+        Idempotent: tools already folded carry the `-grouped` class and are
+        skipped, so it is safe to call on every stream boundary and on
+        hydration. A run is only collapsed once every tool in it succeeded;
+        a run containing an error/rejection/pending tool is left expanded.
+        """
+        try:
+            messages = self.query_one("#messages", Container)
+        except NoMatches:
+            return
+
+        run_tools: list[ToolCallMessage] = []
+        run_collapsible: list[Widget] = []
+        run_anchor: Widget | None = None
+
+        async def flush() -> None:
+            nonlocal run_tools, run_collapsible, run_anchor
+            if run_tools and run_anchor is not None:
+                await self._mount_tool_group_summary(
+                    messages, run_tools, run_collapsible, run_anchor
+                )
+            run_tools = []
+            run_collapsible = []
+            run_anchor = None
+
+        # One repaint for the whole regroup — a single hydration or boundary
+        # pass can fold several runs and hide many rows at once.
+        with self.batch_update():
+            for child in list(messages.children):
+                if child.has_class(_MESSAGE_TIMESTAMP_FOOTER_CLASS):
+                    continue  # footers are transparent to grouping
+                if isinstance(child, ToolCallMessage):
+                    groupable = (
+                        child.tool_name != "ask_user"
+                        and child.is_success
+                        and not child.has_class("-grouped")
+                    )
+                    if not groupable:
+                        await flush()
+                        continue
+                    if run_anchor is None:
+                        run_anchor = child
+                    run_tools.append(child)
+                    run_collapsible.append(child)
+                    continue
+                if isinstance(child, DiffMessage):
+                    # A diff belongs to the tool above it; never starts a run.
+                    if run_anchor is not None:
+                        run_collapsible.append(child)
+                    continue
+                # Assistant text, notices, an existing summary, etc. end the run.
+                await flush()
+            await flush()
+
+    @staticmethod
+    async def _mount_tool_group_summary(
+        messages: Container,
+        tools: list[ToolCallMessage],
+        collapsible: list[Widget],
+        anchor: Widget,
+    ) -> None:
+        """Insert a `ToolGroupSummary` before `anchor` and collapse the run."""
+        if not anchor.is_attached:
+            return
+        summary = ToolGroupSummary(tools=list(tools), collapsible=list(collapsible))
+        for widget in collapsible:
+            widget.add_class("-grouped")
+        try:
+            await messages.mount(summary, before=anchor)
+        except Exception:
+            logger.warning("Failed to mount tool group summary", exc_info=True)
+            for widget in collapsible:
+                widget.remove_class("-grouped")
 
     def _set_active_message(self, message_id: str | None) -> None:
         """Set the active streaming message (won't be pruned).
@@ -11048,6 +11230,11 @@ class DeepAgentsApp(App):
         self._pending_shell_messages.clear()
         # Clear the message store first
         self._message_store.clear()
+        # Drop the open tool group; its widget is about to leave the DOM.
+        self._active_tool_group = None
+        # Drop the stale spinner ref, since remove_children() below detaches
+        # the current spinner widget.
+        self._loading_widget = None
         # Drop the tracked in-flight prompt: its widget is about to leave the
         # DOM, so the pointer must not outlive it. Keeps the "cleared screen ⇒
         # nothing to dim" invariant self-enforcing regardless of caller timing.
@@ -11818,7 +12005,7 @@ class DeepAgentsApp(App):
                     )
 
     def action_toggle_tool_output(self) -> None:
-        """Toggle expand/collapse of the most recent tool output or skill body."""
+        """Toggle the most recent collapsible unit (group, skill, or tool)."""
         # Pending ask_user takes precedence so Ctrl+O toggles the question card.
         if self._pending_ask_user_widget is not None:
             try:
@@ -11830,31 +12017,31 @@ class DeepAgentsApp(App):
                     tool_msg.toggle_args()
                     return
 
-        # Try skill messages first (most recent collapsible content)
+        # Toggle whichever collapsible unit is most recent in DOM order — a tool
+        # group, a skill body, or a standalone tool row — so content mounted
+        # after a group stays reachable instead of always hitting the last group.
+        # Grouped tool rows are folded into their summary, so skip them here.
         try:
-            skill_messages = list(self.query(SkillMessage))
+            messages = self.query_one("#messages", Container)
         except NoMatches:
-            skill_messages = []
-        for skill_msg in reversed(skill_messages):
-            if skill_msg._stripped_body.strip():
-                skill_msg.toggle_body()
+            return
+        for child in reversed(list(messages.children)):
+            if isinstance(child, ToolGroupSummary):
+                child.toggle()
                 return
-
-        # Fall back to tool messages with output or expandable args
-        try:
-            tool_messages = list(self.query(ToolCallMessage))
-        except NoMatches:
-            tool_messages = []
-        for tool_msg in reversed(tool_messages):
-            if tool_msg.has_output and tool_msg.has_expandable_output:
-                tool_msg.toggle_output()
-                return
-            if tool_msg.has_expandable_args:
-                tool_msg.toggle_args()
-                return
-            if tool_msg.has_output:
-                tool_msg.toggle_output()
+            if isinstance(child, SkillMessage) and child._stripped_body.strip():
+                child.toggle_body()
                 return
+            if isinstance(child, ToolCallMessage) and not child.has_class("-grouped"):
+                if child.has_output and child.has_expandable_output:
+                    child.toggle_output()
+                    return
+                if child.has_expandable_args:
+                    child.toggle_args()
+                    return
+                if child.has_output:
+                    child.toggle_output()
+                    return
 
     # Approval menu action handlers (delegated from App-level bindings)
     # NOTE: These only activate when approval widget is pending

libs/code/deepagents_code/textual_adapter.py

@@ -78,9 +78,6 @@ _hitl_adapter_cache: TypeAdapter | None = None
 
 _ASK_USER_UNSUPPORTED_ERROR = "ask_user not supported by this UI"
 
-_TOOL_CALLS_KEEP_THINKING_SPINNER = frozenset({"edit_file"})
-"""Tool calls whose argument/approval phase can be long enough to need feedback."""
-
 
 def _get_hitl_request_adapter(hitl_request_type: type) -> TypeAdapter:
     """Return a cached `TypeAdapter(HITLRequest)`.
@@ -1170,17 +1167,15 @@ async def execute_task_textual(
                                     buffer_name, parsed_args, buffer_id
                                 )
 
-                                keep_thinking_spinner = (
-                                    buffer_name in _TOOL_CALLS_KEEP_THINKING_SPINNER
-                                )
-
-                                # Hide spinner before showing most tool calls.
-                                # `edit_file` can spend noticeable time between
-                                # argument streaming, HITL interrupt delivery, and
-                                # approval handling, so re-anchor Thinking below
-                                # the row instead of leaving the UI visually idle.
-                                if adapter._set_spinner and not keep_thinking_spinner:
-                                    await adapter._set_spinner(None)
+                                # Keep the global "Thinking" spinner visible
+                                # across tool calls rather than hiding it per
+                                # tool: it's a stable turn-level indicator, and
+                                # the tool's own progress now shows in its
+                                # collapsed group row. Re-assert it so it stays
+                                # pinned at the bottom as the new row mounts
+                                # above it.
+                                if adapter._set_spinner:
+                                    await adapter._set_spinner("Thinking")
 
                                 # Mount tool call message
                                 logger.debug(
@@ -1191,20 +1186,11 @@ async def execute_task_textual(
                                 tool_msg = ToolCallMessage(buffer_name, parsed_args)
                                 await adapter._mount_message(tool_msg)
                                 adapter._current_tool_messages[buffer_id] = tool_msg
-                                if keep_thinking_spinner:
-                                    # The argument/approval phase uses the global
-                                    # "Thinking" spinner instead of a per-tool one.
-                                    if adapter._set_spinner:
-                                        await adapter._set_spinner("Thinking")
-                                else:
-                                    # Show a per-tool running spinner immediately so
-                                    # auto-executed tools such as grep, glob,
-                                    # read_file, and ls display activity instead of
-                                    # sitting idle until their result arrives. Every
-                                    # tool outside the frozenset hits this branch;
-                                    # those that go on to interrupt for approval are
-                                    # paused again below.
-                                    tool_msg.set_running()
+                                # Mark running so the group row reflects live
+                                # progress; the row itself is hidden inside the
+                                # group, so this drives state, not a visible
+                                # per-tool spinner.
+                                tool_msg.set_running()
 
                             tool_call_buffers.pop(buffer_key, None)
 

libs/code/deepagents_code/widgets/messages.py

@@ -59,6 +59,7 @@ if TYPE_CHECKING:
     from textual.app import ComposeResult
     from textual.events import MouseMove
     from textual.timer import Timer
+    from textual.widget import Widget
     from textual.widgets import Markdown
     from textual.widgets._markdown import MarkdownStream
 
@@ -866,6 +867,15 @@ class AssistantMessage(Vertical):
             await self._markdown.update(content)
 
 
+_ToolStatus = Literal["pending", "running", "success", "error", "rejected", "skipped"]
+"""The full set of lifecycle states a tool call can hold.
+
+Kept as a closed `Literal` so `ty` flags typos at the assignment sites and so
+the grouping predicates (`is_success`/`is_failed`/`is_pending`) partition a
+known universe.
+"""
+
+
 class ToolCallMessage(Vertical):
     """Widget displaying a tool call with collapsible output.
 
@@ -1002,7 +1012,7 @@ class ToolCallMessage(Vertical):
         super().__init__(**kwargs)
         self._tool_name = tool_name
         self._args = args or {}
-        self._status = "pending"  # Waiting for approval or auto-approve
+        self._status: _ToolStatus = "pending"  # Waiting for approval or auto-approve
         self._output: str = ""
         self._expanded: bool = False
         self._args_expanded: bool = False
@@ -1481,10 +1491,10 @@ class ToolCallMessage(Vertical):
         """Return whether search output is a terminal no-result message.
 
         These sentinels must match the empty-result strings the SDK emits
-        (`format_grep_matches` and `_format_file_paths` in
-        `deepagents.middleware.filesystem`). If those change, this silently
-        stops matching and empty searches revert to collapsing behind an expand
-        affordance rather than rendering inline.
+        (`format_grep_matches` in `deepagents.backends.utils` and
+        `_format_file_paths` in `deepagents.middleware.filesystem`). If those
+        change, this silently stops matching and empty searches revert to
+        collapsing behind an expand affordance rather than rendering inline.
         """
         if self._tool_name == "grep":
             return output.strip() == "No matches found"
@@ -2454,6 +2464,27 @@ class ToolCallMessage(Vertical):
         """Public read-only accessor for the underlying tool name."""
         return self._tool_name
 
+    @property
+    def is_success(self) -> bool:
+        """Whether the tool completed successfully."""
+        return self._status == "success"
+
+    @property
+    def is_failed(self) -> bool:
+        """Whether the tool did not succeed and should stay visible.
+
+        Covers errored, rejected, and skipped tools. `skipped` is included so a
+        reject-cascade (one tool rejected, the rest skipped) keeps the skipped
+        rows visible and out of the group's success count, matching how
+        `_regroup_completed_tools` treats a hydrated transcript.
+        """
+        return self._status in {"error", "rejected", "skipped"}
+
+    @property
+    def is_pending(self) -> bool:
+        """Whether the tool has not finished (awaiting approval or running)."""
+        return self._status in {"pending", "running"}
+
     @property
     def has_expandable_args(self) -> bool:
         """Whether the tool's args are large enough to deserve a collapsible block.
@@ -2563,6 +2594,358 @@ class ToolCallMessage(Vertical):
         return filtered
 
 
+# Maps a tool name to the summary category it aggregates under. grep/glob share
+# "search" so a mixed run folds into a single "Searched for N patterns" segment.
+_TOOL_SUMMARY_CATEGORY: dict[str, str] = {
+    "read_file": "read",
+    "write_file": "write",
+    "edit_file": "edit",
+    "delete": "delete",
+    "ls": "ls",
+    "grep": "search",
+    "glob": "search",
+    "execute": "shell",
+    "js_eval": "js",
+    "web_search": "web_search",
+    "fetch_url": "fetch",
+    "task": "task",
+    "write_todos": "todos",
+}
+
+# category -> (present verb, past verb, singular noun, plural noun).
+_TOOL_SUMMARY_PHRASES: dict[str, tuple[str, str, str, str]] = {
+    "read": ("Reading", "Read", "file", "files"),
+    "write": ("Writing", "Wrote", "file", "files"),
+    "edit": ("Editing", "Edited", "file", "files"),
+    "delete": ("Deleting", "Deleted", "file", "files"),
+    "ls": ("Listing", "Listed", "directory", "directories"),
+    "search": ("Searching for", "Searched for", "pattern", "patterns"),
+    "shell": ("Running", "Ran", "shell command", "shell commands"),
+    "js": ("Running", "Ran", "JS evaluation", "JS evaluations"),
+    "fetch": ("Fetching", "Fetched", "URL", "URLs"),
+    "task": ("Running", "Ran", "agent", "agents"),
+}
+
+_Tense = Literal["present", "past"]
+
+
+def _summary_segment(category: str, count: int, tool_name: str, tense: _Tense) -> str:
+    """Phrase a single count segment, e.g. "Read 2 files" / "Reading 2 files".
+
+    Args:
+        category: The summary category the tools were bucketed into.
+        count: How many tools fell into this category.
+        tool_name: A representative raw tool name, used to phrase categories
+            that have no dedicated entry in `_TOOL_SUMMARY_PHRASES`.
+        tense: Whether to phrase the segment in the present or past tense.
+
+    Returns:
+        The phrased segment for this category, count, and tense.
+    """
+    if category == "web_search":
+        base = "Searching the web" if tense == "present" else "Searched the web"
+        return base if count == 1 else f"{base} {count} times"
+    if category == "todos":
+        return "Updating todos" if tense == "present" else "Updated todos"
+    phrase = _TOOL_SUMMARY_PHRASES.get(category)
+    if phrase is None:
+        present, past = "Running", "Ran"
+        singular, plural = f"{tool_name} call", f"{tool_name} calls"
+    else:
+        present, past, singular, plural = phrase
+    verb = present if tense == "present" else past
+    noun = singular if count == 1 else plural
+    return f"{verb} {count} {noun}"
+
+
+def summarize_tool_group(tool_names: list[str], *, tense: _Tense = "past") -> str:
+    """Build a one-line summary of a run of tool calls.
+
+    Aggregates by category in first-appearance order and lowercases the lead
+    word of every segment after the first, e.g.
+    `["read_file", "read_file", "execute"]` -> "Read 2 files, ran 1 shell command".
+
+    Args:
+        tool_names: Raw tool names for the run, in call order.
+        tense: Whether to phrase the summary in the present or past tense.
+
+    Returns:
+        The aggregated one-line summary string in the requested tense.
+    """
+    counts: dict[str, int] = {}
+    order: list[str] = []
+    rep_name: dict[str, str] = {}
+    for name in tool_names:
+        category = _TOOL_SUMMARY_CATEGORY.get(name, name)
+        if category not in counts:
+            counts[category] = 0
+            order.append(category)
+            rep_name[category] = name
+        counts[category] += 1
+
+    segments = [
+        _summary_segment(cat, counts[cat], rep_name[cat], tense) for cat in order
+    ]
+    if not segments:
+        return "Running tools" if tense == "present" else "Ran tools"
+    first, *rest = segments
+    lowered = [f"{seg[0].lower()}{seg[1:]}" if seg else seg for seg in rest]
+    return ", ".join([first, *lowered])
+
+
+class ToolGroupSummary(Static):
+    """Collapsed one-line stand-in for an assistant step's tool calls.
+
+    Tools are hidden from the moment they start; this single line shows live
+    progress ("Running 1 shell command…") and flips to the past tense
+    ("Ran 1 shell command") once every tool finishes. Clicking the line or
+    pressing Ctrl+O expands the underlying tool rows (and their diffs).
+
+    Two modes:
+
+    - **live** (streaming): created empty, members added via `add_member` as
+      they mount, a spinner timer animates the line and re-renders present/past
+      tense, and failed tools are ejected back into view so errors stay visible.
+    - **finalized** (`live=False`, used for hydration/resume): a fixed set of
+      completed tools rendered straight to the past tense with no timer.
+
+    Purely presentational — never tracked by the message store; it is re-derived
+    from the mounted tool widgets on each stream boundary and on hydration.
+    """
+
+    DEFAULT_CSS = """
+    ToolGroupSummary {
+        height: auto;
+        padding: 0 1;
+        margin: 0 0 1 0;
+        color: $text-muted;
+        pointer: pointer;
+    }
+
+    ToolGroupSummary:hover {
+        color: $text;
+    }
+    """
+
+    _SPINNER_INTERVAL: ClassVar[float] = 0.1
+
+    _collapsed: var[bool] = var(True)
+
+    def __init__(
+        self,
+        tools: list[ToolCallMessage] | None = None,
+        collapsible: list[Widget] | None = None,
+        *,
+        live: bool = False,
+        **kwargs: Any,
+    ) -> None:
+        """Initialize the summary.
+
+        Args:
+            tools: Tool widgets the summary aggregates (drives its text). May be
+                empty for a live group that grows via `add_member`.
+            collapsible: Every widget hidden/shown with the group, including the
+                tool widgets and any interleaved diff previews.
+            live: When True, animate progress and accept new members until
+                `close`. When False, render a finalized past-tense summary.
+            **kwargs: Additional arguments passed to `Static`.
+        """
+        super().__init__("", **kwargs)
+        self._tools = list(tools or [])
+        self._collapsible = list(collapsible or [])
+        self._finalized = not live
+        self._spinner_pos = 0
+        self._timer: Timer | None = None
+        # Cached summary phrasing, rebuilt only when membership changes (not on
+        # every spinner tick). None means "recompute on next render".
+        self._present_text: str | None = None
+        self._past_text: str | None = None
+
+    def on_mount(self) -> None:
+        """Apply initial visibility, render, and arm the spinner if live."""
+        self._apply_visibility()
+        self._render_line()
+        self._sync_timer()
+
+    def add_member(self, tool: ToolCallMessage, *extra: Widget) -> None:
+        """Add a tool (and any associated widgets) to a live group."""
+        tool.add_class("-grouped")
+        self._tools.append(tool)
+        self._collapsible.append(tool)
+        for widget in extra:
+            widget.add_class("-grouped")
+            self._collapsible.append(widget)
+        self._present_text = self._past_text = None
+        self._apply_visibility()
+        self._render_line()
+        self._sync_timer()
+
+    def add_collapsible(self, widget: Widget) -> None:
+        """Attach a non-tool widget (e.g. a diff) to be folded with the group."""
+        widget.add_class("-grouped")
+        self._collapsible.append(widget)
+        if widget.is_attached:
+            widget.display = not self._collapsed
+
+    def close(self) -> None:
+        """Mark the group complete; no further members will join."""
+        self._finalized = True
+        self._evict_failed()
+        self._stop_timer()
+        if not self.is_attached:
+            return
+        if self._tools:
+            self._render_line()
+        else:
+            # Every tool failed and was ejected — nothing left to summarize.
+            self.remove()
+
+    @property
+    def has_attached_members(self) -> bool:
+        """Whether any collapsed widget is still attached to the DOM."""
+        return any(widget.is_attached for widget in self._collapsible)
+
+    def toggle(self) -> None:
+        """Toggle between collapsed and expanded."""
+        self._collapsed = not self._collapsed
+
+    def watch__collapsed(self, _collapsed: bool) -> None:
+        """Re-render and re-apply member visibility when the state changes.
+
+        Coalesced into one repaint so expanding a multi-tool group reveals every
+        row at once instead of bouncing the transcript per member.
+        """
+        if not self.is_attached:
+            self._apply_visibility()
+            self._render_line()
+            return
+        with self.app.batch_update():
+            self._apply_visibility()
+            self._render_line()
+
+    def on_click(self, event: Click) -> None:
+        """Toggle the group on click."""
+        event.stop()
+        self.toggle()
+
+    def _in_progress(self) -> bool:
+        """Whether any member tool is still pending or running.
+
+        Returns:
+            True if at least one member tool has not finished.
+        """
+        return any(tool.is_pending for tool in self._tools)
+
+    def _evict_failed(self) -> None:
+        """Un-fold errored/rejected/skipped tools so non-successes stay visible."""
+        failed = [t for t in self._tools if t.is_failed]
+        if not failed:
+            return
+        for tool in failed:
+            self._tools.remove(tool)
+            if tool in self._collapsible:
+                self._collapsible.remove(tool)
+            tool.remove_class("-grouped")
+            if tool.is_attached:
+                tool.display = True
+        self._present_text = self._past_text = None
+
+    def _sync_timer(self) -> None:
+        """Run the spinner timer only while live members are in progress."""
+        if not self._finalized and self._in_progress():
+            if self._timer is None:
+                self._timer = self.set_interval(self._SPINNER_INTERVAL, self._tick)
+        else:
+            self._stop_timer()
+
+    def _stop_timer(self) -> None:
+        if self._timer is not None:
+            self._timer.stop()
+            self._timer = None
+
+    def _tick(self) -> None:
+        """Advance the spinner, eject failures, and flip to past tense when done."""
+        try:
+            self._spinner_pos += 1
+            before = len(self._tools)
+            self._evict_failed()
+            evicted = len(self._tools) != before
+            if self._collapsed:
+                # Re-assert hidden state in case a member was shown externally
+                # (e.g. ToolCallMessage.clear_awaiting_approval after HITL).
+                self._apply_visibility()
+            if not self._tools:
+                self._stop_timer()
+                if self.is_attached:
+                    self.remove()
+                return
+            in_progress = self._in_progress()
+            if not in_progress:
+                self._stop_timer()
+            # A bare spinner advance keeps the line height; only relayout when
+            # membership changed (eviction) or the line flips to past tense.
+            self._render_line(
+                in_progress=in_progress, layout=evicted or not in_progress
+            )
+        except Exception:
+            # Fires ~10x/second, so an unhandled raise would propagate out of the
+            # interval callback and can crash the app repeatedly. The group is
+            # purely presentational; stop animating and log rather than take the
+            # transcript down.
+            logger.exception("ToolGroupSummary spinner tick failed; stopping timer")
+            self._stop_timer()
+
+    def _apply_visibility(self) -> None:
+        """Show or hide every folded widget per the collapsed state."""
+        visible = not self._collapsed
+        for widget in self._collapsible:
+            if widget.is_attached and widget.display != visible:
+                widget.display = visible
+
+    def _render_line(
+        self, *, in_progress: bool | None = None, layout: bool = True
+    ) -> None:
+        """Refresh the summary line for the current tense and collapsed state.
+
+        Args:
+            in_progress: Pre-computed progress state to avoid re-scanning members
+                on the spinner hot path; recomputed when omitted.
+            layout: Whether the update may change the line's height. The spinner
+                hot path passes False so a bare glyph swap doesn't relayout the
+                whole transcript 10x/second.
+        """
+        if not self.is_attached:
+            return
+        if not self._tools:
+            self.update(Content(""), layout=layout)
+            return
+        glyphs = get_glyphs()
+        if in_progress is None:
+            in_progress = self._in_progress()
+        if not self._finalized and in_progress:
+            if self._present_text is None:
+                self._present_text = summarize_tool_group(
+                    [tool.tool_name for tool in self._tools], tense="present"
+                )
+            frames = glyphs.spinner_frames
+            spinner = frames[self._spinner_pos % len(frames)]
+            self.update(
+                Content(f"{spinner} {self._present_text}{glyphs.ellipsis}"),
+                layout=layout,
+            )
+        else:
+            mark = (
+                glyphs.disclosure_collapsed
+                if self._collapsed
+                else glyphs.disclosure_expanded
+            )
+            if self._past_text is None:
+                self._past_text = summarize_tool_group(
+                    [tool.tool_name for tool in self._tools], tense="past"
+                )
+            self.update(Content(f"{mark} {self._past_text}"), layout=layout)
+
+
 class DiffMessage(Static):
     """Widget displaying a diff with syntax highlighting."""
 

libs/code/tests/unit_tests/test_app.py

@@ -21,10 +21,12 @@ if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable, Iterator
 
     from langchain_core.messages import HumanMessage
+    from textual.pilot import Pilot
 
     from deepagents_code.mcp_auth import McpServerSpec
     from deepagents_code.notifications import PendingNotification
     from deepagents_code.sessions import ThreadInfo
+    from deepagents_code.widgets.messages import ToolCallMessage
 
 import pytest
 from textual import events
@@ -3597,25 +3599,18 @@ class TestAskUserLifecycle:
 
     def test_ctrl_o_falls_back_to_tool_with_expandable_args(self) -> None:
         """When no ask_user is pending, Ctrl+O still expands an ask_user-like row."""
+        from deepagents_code.widgets.messages import ToolCallMessage
+
         app = DeepAgentsApp(agent=MagicMock())
         app._pending_ask_user_widget = None
-        tool = MagicMock()
+        tool = MagicMock(spec=ToolCallMessage)
+        tool.has_class.return_value = False
         tool.has_output = False
         tool.has_expandable_args = True
+        container = MagicMock()
+        container.children = [tool]
 
-        def fake_query(query_type: object) -> list[object]:
-            from deepagents_code.widgets.messages import (
-                SkillMessage,
-                ToolCallMessage,
-            )
-
-            if query_type is ToolCallMessage:
-                return [tool]
-            if query_type is SkillMessage:
-                return []
-            return []
-
-        with patch.object(app, "query", side_effect=fake_query):
+        with patch.object(app, "query_one", return_value=container):
             app.action_toggle_tool_output()
 
         tool.toggle_args.assert_called_once_with()
@@ -3628,61 +3623,97 @@ class TestAskUserLifecycle:
         which used to swallow the toggle and leave the collapsible code block
         stuck. The action must fall through to args in that case.
         """
+        from deepagents_code.widgets.messages import ToolCallMessage
+
         app = DeepAgentsApp(agent=MagicMock())
         app._pending_ask_user_widget = None
-        tool = MagicMock()
+        tool = MagicMock(spec=ToolCallMessage)
+        tool.has_class.return_value = False
         tool.has_output = True
         tool.has_expandable_output = False  # short result, nothing to expand
         tool.has_expandable_args = True  # multi-line code block
+        container = MagicMock()
+        container.children = [tool]
 
-        def fake_query(query_type: object) -> list[object]:
-            from deepagents_code.widgets.messages import (
-                SkillMessage,
-                ToolCallMessage,
-            )
-
-            if query_type is ToolCallMessage:
-                return [tool]
-            if query_type is SkillMessage:
-                return []
-            return []
-
-        with patch.object(app, "query", side_effect=fake_query):
+        with patch.object(app, "query_one", return_value=container):
             app.action_toggle_tool_output()
 
         tool.toggle_args.assert_called_once_with()
         tool.toggle_output.assert_not_called()
 
-    def test_ctrl_o_prefers_tool_with_output_over_expandable_args(self) -> None:
-        """Tool with real output wins over a later one with only expandable args."""
+    def test_ctrl_o_prefers_more_recent_tool_in_dom_order(self) -> None:
+        """The newest tool row in DOM order wins over an older one."""
+        from deepagents_code.widgets.messages import ToolCallMessage
+
         app = DeepAgentsApp(agent=MagicMock())
         app._pending_ask_user_widget = None
-        older = MagicMock()
+        older = MagicMock(spec=ToolCallMessage)
+        older.has_class.return_value = False
         older.has_output = True
         older.has_expandable_args = False
-        newer = MagicMock()
+        newer = MagicMock(spec=ToolCallMessage)
+        newer.has_class.return_value = False
         newer.has_output = False
         newer.has_expandable_args = True
+        container = MagicMock()
+        container.children = [older, newer]
 
-        def fake_query(query_type: object) -> list[object]:
-            from deepagents_code.widgets.messages import (
-                SkillMessage,
-                ToolCallMessage,
-            )
-
-            if query_type is ToolCallMessage:
-                return [older, newer]
-            if query_type is SkillMessage:
-                return []
-            return []
-
-        with patch.object(app, "query", side_effect=fake_query):
+        with patch.object(app, "query_one", return_value=container):
             app.action_toggle_tool_output()
 
-        # Iterates in reverse, so newer (expandable args) is hit first.
+        # Walks children in reverse, so the newer row is hit first.
         newer.toggle_args.assert_called_once_with()
         older.toggle_output.assert_not_called()
 
+    def test_ctrl_o_targets_content_mounted_after_a_group(self) -> None:
+        """Content mounted after a tool group stays reachable from Ctrl+O.
+
+        Regression: the handler used to toggle the last tool group whenever any
+        existed, leaving newer collapsible content (here a skill body)
+        unreachable.
+        """
+        from deepagents_code.widgets.messages import SkillMessage, ToolGroupSummary
+
+        app = DeepAgentsApp(agent=MagicMock())
+        app._pending_ask_user_widget = None
+        group = MagicMock(spec=ToolGroupSummary)
+        skill = MagicMock(spec=SkillMessage)
+        skill._stripped_body = "body"
+        container = MagicMock()
+        container.children = [group, skill]  # skill mounted after the group
+
+        with patch.object(app, "query_one", return_value=container):
+            app.action_toggle_tool_output()
+
+        skill.toggle_body.assert_called_once_with()
+        group.toggle.assert_not_called()
+
+    def test_ctrl_o_toggles_group_and_skips_its_folded_rows(self) -> None:
+        """Toggling a group skips its folded rows and leaves older content alone."""
+        from deepagents_code.widgets.messages import (
+            SkillMessage,
+            ToolCallMessage,
+            ToolGroupSummary,
+        )
+
+        app = DeepAgentsApp(agent=MagicMock())
+        app._pending_ask_user_widget = None
+        skill = MagicMock(spec=SkillMessage)
+        skill._stripped_body = "body"
+        group = MagicMock(spec=ToolGroupSummary)
+        folded = MagicMock(spec=ToolCallMessage)
+        folded.has_class.return_value = True  # folded into the group
+        # DOM: older skill, then the group summary followed by its folded row.
+        container = MagicMock()
+        container.children = [skill, group, folded]
+
+        with patch.object(app, "query_one", return_value=container):
+            app.action_toggle_tool_output()
+
+        group.toggle.assert_called_once_with()
+        folded.toggle_output.assert_not_called()
+        skill.toggle_body.assert_not_called()
+
     async def test_request_ask_user_timeout_cleans_old_widget(self) -> None:
         """Timeout cleanup should cancel then remove the previous widget."""
         app = DeepAgentsApp()
@@ -3856,6 +3887,34 @@ class TestLoadingSpinnerLifecycle:
             children = list(messages.children)
             assert children[-1] is widget
 
+    async def test_spinner_stays_pinned_and_singular_as_messages_mount(self) -> None:
+        """The spinner is reused and stays last as new messages stream in.
+
+        New content mounts above the spinner (via `_mount_before_queued`), so it
+        never needs repositioning and exactly one spinner exists — the stability
+        that replaced the per-tool hide/show flicker.
+        """
+        from deepagents_code.widgets.loading import LoadingWidget
+
+        app = DeepAgentsApp()
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await app._set_spinner("Thinking")
+            await pilot.pause()
+            widget = app._loading_widget
+            assert widget is not None
+
+            messages = app.query_one("#messages", Container)
+            for i in range(3):
+                await app._mount_message(AppMessage(f"line {i}"))
+                await app._set_spinner("Thinking")  # status update each "step"
+            await pilot.pause()
+
+            # Same instance the whole time; still exactly one; still last.
+            assert app._loading_widget is widget
+            assert len(list(app.query(LoadingWidget))) == 1
+            assert list(messages.children)[-1] is widget
+
 
 class TestTraceCommand:
     """Test /trace slash command."""
@@ -17114,6 +17173,7 @@ class TestSetSpinnerTerminalProgress:
             await app._set_spinner("Thinking")
             await app._set_spinner(None)
             await app._set_spinner("Thinking")
+            await app._set_spinner(None)
             await pilot.pause()
 
         assert calls.count("set") >= 3
@@ -21078,3 +21138,286 @@ class TestCopyFocusedInputText:
                 assert app._copy_focused_input_text() is False
 
             copy_mock.assert_not_called()
+
+
+class TestToolGroupCollapse:
+    """Integration tests for auto-collapsing completed tool runs."""
+
+    @staticmethod
+    async def _mount_tools(
+        pilot: Pilot[DeepAgentsApp],
+        container: Container,
+        specs: list[tuple[str, str, dict[str, Any], str]],
+    ) -> list[ToolCallMessage]:
+        """Mount tool widgets and apply their terminal status.
+
+        Each spec is `(id, tool_name, args, status)` where status is
+        `"success"` or `"error"`.
+        """
+        from deepagents_code.widgets.messages import ToolCallMessage
+
+        tools: list[ToolCallMessage] = []
+        for tid, name, args, _status in specs:
+            tool = ToolCallMessage(name, args)
+            tool.id = tid
+            await container.mount(tool)
+            tools.append(tool)
+        await pilot.pause()
+        for tool, (_tid, _name, _args, status) in zip(tools, specs, strict=True):
+            if status == "success":
+                tool.set_success("output")
+            else:
+                tool.set_error("boom")
+        await pilot.pause()
+        return tools
+
+    async def test_regroup_collapses_success_run(self) -> None:
+        """A run of successful tools folds into one collapsed summary."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-group")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+            tools = await self._mount_tools(
+                pilot,
+                messages,
+                [
+                    ("g1", "read_file", {"file_path": "a.py"}, "success"),
+                    ("g2", "execute", {"command": "ls"}, "success"),
+                ],
+            )
+
+            await app._regroup_completed_tools()
+            await pilot.pause()
+
+            summaries = list(app.query(ToolGroupSummary))
+            assert len(summaries) == 1
+            assert all(tool.display is False for tool in tools)
+            rendered = summaries[0].render()
+            assert isinstance(rendered, Content)
+            assert "Read 1 file, ran 1 shell command" in rendered.plain
+
+    async def test_regroup_treats_timestamp_footer_as_transparent(self) -> None:
+        """A timestamp footer between two tools does not split the run.
+
+        Production mounts a footer after every message, so a completed run
+        reaches regroup as (tool, footer, tool). The footer must be transparent
+        to grouping or every timestamped run would fragment into single-tool
+        summaries. `_mount_tools` mounts tools with no footers, so this shape is
+        otherwise never exercised.
+        """
+        from deepagents_code.widgets.message_store import MessageData, MessageType
+        from deepagents_code.widgets.messages import (
+            ToolCallMessage,
+            ToolGroupSummary,
+        )
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-footer")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+
+            t1 = ToolCallMessage("read_file", {"file_path": "a.py"})
+            t1.id = "f1"
+            t2 = ToolCallMessage("read_file", {"file_path": "b.py"})
+            t2.id = "f2"
+            # A USER footer is the simplest to build; only its footer CSS class
+            # matters to the transparency branch under test.
+            footer = app._build_message_timestamp_footer(
+                MessageData(
+                    type=MessageType.USER,
+                    content="",
+                    id="f1",
+                    timestamp=1_704_110_405.0,
+                ),
+                visible=True,
+            )
+            assert footer is not None
+            await messages.mount(t1)
+            await messages.mount(footer)
+            await messages.mount(t2)
+            await pilot.pause()
+            t1.set_success("ok")
+            t2.set_success("ok")
+            await pilot.pause()
+
+            await app._regroup_completed_tools()
+            await pilot.pause()
+
+            # Both tools fold into one summary despite the intervening footer.
+            summaries = list(app.query(ToolGroupSummary))
+            assert len(summaries) == 1
+            assert t1.display is False
+            assert t2.display is False
+            rendered = summaries[0].render()
+            assert isinstance(rendered, Content)
+            assert "Read 2 files" in rendered.plain
+
+    async def test_regroup_is_idempotent(self) -> None:
+        """Re-running regroup does not create duplicate summaries."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-idem")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+            await self._mount_tools(
+                pilot,
+                messages,
+                [("g1", "grep", {"pattern": "x"}, "success")],
+            )
+
+            await app._regroup_completed_tools()
+            await app._regroup_completed_tools()
+            await pilot.pause()
+
+            assert len(list(app.query(ToolGroupSummary))) == 1
+
+    async def test_errored_tool_stays_visible(self) -> None:
+        """An errored tool stays visible; only the successful prefix collapses."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-err")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+            ok_tool, err_tool = await self._mount_tools(
+                pilot,
+                messages,
+                [
+                    ("g1", "read_file", {"file_path": "a.py"}, "success"),
+                    ("g2", "execute", {"command": "ls"}, "error"),
+                ],
+            )
+
+            await app._regroup_completed_tools()
+            await pilot.pause()
+
+            # The success prefix folds; the errored tool is never grouped.
+            assert len(list(app.query(ToolGroupSummary))) == 1
+            assert ok_tool.display is False
+            assert err_tool.display is True
+            assert not err_tool.has_class("-grouped")
+
+    async def test_assistant_message_boundary_triggers_collapse(self) -> None:
+        """Mounting a non-tool message folds the preceding tool run."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-boundary")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+            tools = await self._mount_tools(
+                pilot,
+                messages,
+                [("g1", "read_file", {"file_path": "a.py"}, "success")],
+            )
+
+            await app._mount_message(AssistantMessage("next step"))
+            await pilot.pause()
+
+            assert len(list(app.query(ToolGroupSummary))) == 1
+            assert tools[0].display is False
+
+    async def test_separate_steps_get_separate_summaries(self) -> None:
+        """Tools split by an assistant message form two independent groups."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-steps")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+            await self._mount_tools(
+                pilot, messages, [("a1", "read_file", {"file_path": "a"}, "success")]
+            )
+            await messages.mount(AssistantMessage("step two"))
+            await self._mount_tools(
+                pilot, messages, [("b1", "execute", {"command": "ls"}, "success")]
+            )
+
+            await app._regroup_completed_tools()
+            await pilot.pause()
+
+            assert len(list(app.query(ToolGroupSummary))) == 2
+
+    async def test_mount_tool_creates_collapsed_live_group(self) -> None:
+        """Mounting a tool via _mount_message folds it immediately, no flash."""
+        from deepagents_code.widgets.messages import (
+            ToolCallMessage,
+            ToolGroupSummary,
+        )
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-live")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+
+            tool = ToolCallMessage("read_file", {"file_path": "a.py"})
+            # _mount_message folds it synchronously; don't pilot.pause() while
+            # the live spinner timer runs (it blocks the idle wait).
+            await app._mount_message(tool)
+
+            # A live group was opened and the tool hidden from the start.
+            summaries = list(app.query(ToolGroupSummary))
+            assert len(summaries) == 1
+            assert tool.display is False
+            assert app._active_tool_group is summaries[0]
+
+            # A boundary closes the group and flips it to past tense.
+            tool.set_success("done")
+            await app._mount_message(AssistantMessage("done"))
+
+            assert app._active_tool_group is None
+            rendered = summaries[0].render()
+            assert isinstance(rendered, Content)
+            assert "Read 1 file" in rendered.plain
+            assert tool.display is False
+            await pilot.pause()
+
+    async def test_group_survives_idle_after_completion(self) -> None:
+        """A folded group stays mounted across completion, idle, and a boundary.
+
+        Regression guard: the summary's finalized flag must not collide with
+        Textual's MessagePump internals, or the widget is silently pruned on the
+        next idle tick (looked like the group "disappearing" when tools finish).
+        """
+        from deepagents_code.widgets.messages import (
+            ToolCallMessage,
+            ToolGroupSummary,
+        )
+
+        app = DeepAgentsApp(agent=MagicMock(), thread_id="t-idle")
+        app._load_thread_history = AsyncMock()  # ty: ignore
+        async with app.run_test() as pilot:
+            messages = app.query_one("#messages", Container)
+            await messages.remove_children()
+
+            t1 = ToolCallMessage("execute", {"command": "ls"})
+            t2 = ToolCallMessage("read_file", {"file_path": "a.py"})
+            await app._mount_message(t1)
+            await app._mount_message(t2)
+            group = app._active_tool_group
+            assert group is not None
+
+            t1.set_success("ok")
+            t2.set_success("ok")
+            group._tick()  # flips to past tense, stops the spinner timer
+            await pilot.pause()
+            assert group.is_attached  # survives the idle tick after completion
+
+            await app._mount_message(AssistantMessage("next step"))
+            await pilot.pause()
+            summaries = list(app.query(ToolGroupSummary))
+            assert len(summaries) == 1
+            assert summaries[0].is_attached
+            rendered = summaries[0].render()
+            assert isinstance(rendered, Content)
+            assert "Ran 1 shell command, read 1 file" in rendered.plain

libs/code/tests/unit_tests/test_messages.py

@@ -3351,3 +3351,277 @@ class TestUserMessageCancelled:
             msg.set_cancelled()
             await pilot.pause()
             assert msg.has_class("-cancelled")
+
+
+class TestSummarizeToolGroup:
+    """Tests for the tool-group summary phrasing."""
+
+    @pytest.mark.parametrize(
+        ("names", "expected"),
+        [
+            (["execute"], "Ran 1 shell command"),
+            (
+                ["read_file", "read_file", "execute", "execute", "execute"],
+                "Read 2 files, ran 3 shell commands",
+            ),
+            (["grep"], "Searched for 1 pattern"),
+            (["grep", "glob", "glob"], "Searched for 3 patterns"),
+            (["read_file"], "Read 1 file"),
+            (["web_search", "web_search"], "Searched the web 2 times"),
+            (["web_search"], "Searched the web"),
+            (["write_todos"], "Updated todos"),
+            (["task", "task"], "Ran 2 agents"),
+            (
+                ["edit_file", "write_file", "read_file"],
+                "Edited 1 file, wrote 1 file, read 1 file",
+            ),
+            (["mystery", "mystery"], "Ran 2 mystery calls"),
+        ],
+    )
+    def test_summary_phrasing(self, names: list[str], expected: str) -> None:
+        """The summary aggregates by category and lowercases trailing verbs."""
+        from deepagents_code.widgets.messages import summarize_tool_group
+
+        assert summarize_tool_group(names) == expected
+
+    def test_empty_group_has_fallback(self) -> None:
+        """An empty tool list yields a generic fallback rather than crashing."""
+        from deepagents_code.widgets.messages import summarize_tool_group
+
+        assert summarize_tool_group([]) == "Ran tools"
+
+
+class _ToolGroupApp(App[None]):
+    """Minimal app mounting two completed tools plus a group summary."""
+
+    def compose(self) -> ComposeResult:
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        t1 = ToolCallMessage("read_file", {"file_path": "a.py"})
+        t1.id = "t1"
+        t2 = ToolCallMessage("execute", {"command": "ls"})
+        t2.id = "t2"
+        summary = ToolGroupSummary(tools=[t1, t2], collapsible=[t1, t2])
+        summary.id = "summary"
+        yield summary
+        yield t1
+        yield t2
+
+
+class TestToolGroupSummary:
+    """Runtime collapse/expand behavior for the group summary widget."""
+
+    async def test_collapsed_hides_members_and_renders_summary(self) -> None:
+        """On mount the summary collapses its members and shows the count line."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _ToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            t1 = pilot.app.query_one("#t1", ToolCallMessage)
+            t2 = pilot.app.query_one("#t2", ToolCallMessage)
+
+            assert summary._collapsed is True
+            assert t1.display is False
+            assert t2.display is False
+            rendered = summary.render()
+            assert isinstance(rendered, Content)
+            assert "Read 1 file, ran 1 shell command" in rendered.plain
+
+    async def test_toggle_expands_and_recollapses_members(self) -> None:
+        """Toggling flips member visibility and the disclosure glyph."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _ToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            t1 = pilot.app.query_one("#t1", ToolCallMessage)
+            t2 = pilot.app.query_one("#t2", ToolCallMessage)
+
+            summary.toggle()
+            await pilot.pause()
+            assert summary._collapsed is False
+            assert t1.display is True
+            assert t2.display is True
+
+            summary.toggle()
+            await pilot.pause()
+            assert summary._collapsed is True
+            assert t1.display is False
+            assert t2.display is False
+
+    async def test_has_attached_members_tracks_removal(self) -> None:
+        """`has_attached_members` flips to False once members are removed."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _ToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            assert summary.has_attached_members is True
+
+            await pilot.app.query_one("#t1", ToolCallMessage).remove()
+            await pilot.app.query_one("#t2", ToolCallMessage).remove()
+            await pilot.pause()
+            assert summary.has_attached_members is False
+
+
+class TestSummarizeToolGroupPresentTense:
+    """Present-tense phrasing used while a step's tools are still running."""
+
+    def test_present_tense(self) -> None:
+        from deepagents_code.widgets.messages import summarize_tool_group
+
+        assert (
+            summarize_tool_group(["execute"], tense="present")
+            == "Running 1 shell command"
+        )
+        assert (
+            summarize_tool_group(["read_file", "read_file", "grep"], tense="present")
+            == "Reading 2 files, searching for 1 pattern"
+        )
+
+
+class _LiveToolGroupApp(App[None]):
+    """Minimal app with an empty live group and two tools to add to it."""
+
+    def compose(self) -> ComposeResult:
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        summary = ToolGroupSummary(live=True)
+        summary.id = "summary"
+        t1 = ToolCallMessage("execute", {"command": "ls"})
+        t1.id = "t1"
+        t2 = ToolCallMessage("read_file", {"file_path": "a.py"})
+        t2.id = "t2"
+        yield summary
+        yield t1
+        yield t2
+
+
+class TestLiveToolGroupSummary:
+    """Eager/live group: collapsed from the start, running -> ran transition."""
+
+    async def test_present_tense_while_running_then_past_on_close(self) -> None:
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _LiveToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            t1 = pilot.app.query_one("#t1", ToolCallMessage)
+
+            # add_member renders synchronously; avoid pilot.pause() while the
+            # live spinner timer is running (it keeps the app from going idle).
+            summary.add_member(t1)
+            rendered = summary.render()
+            assert isinstance(rendered, Content)
+            assert "Running 1 shell command" in rendered.plain
+            assert t1.display is False  # collapsed from the start
+
+            t1.set_success("done")
+            summary.close()  # stops the spinner timer, flips to past tense
+
+            rendered = summary.render()
+            assert isinstance(rendered, Content)
+            assert "Ran 1 shell command" in rendered.plain
+            assert t1.display is False
+            # Survives the idle tick after close — guards against the summary's
+            # state attributes colliding with Textual's MessagePump internals
+            # (e.g. `_closed`), which would silently prune the widget.
+            await pilot.pause()
+            assert summary.is_attached
+            assert bool(pilot.app.query(ToolGroupSummary))
+
+    async def test_failed_member_is_evicted_on_close(self) -> None:
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _LiveToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            t1 = pilot.app.query_one("#t1", ToolCallMessage)
+            t2 = pilot.app.query_one("#t2", ToolCallMessage)
+
+            summary.add_member(t1)
+            summary.add_member(t2)
+            t1.set_error("boom")
+            t2.set_success("ok")
+            summary.close()
+            await pilot.pause()
+
+            # The errored tool is un-folded; the successful one stays collapsed.
+            assert t1.display is True
+            assert not t1.has_class("-grouped")
+            assert t2.display is False
+            rendered = summary.render()
+            assert isinstance(rendered, Content)
+            assert "Read 1 file" in rendered.plain
+
+    async def test_rejected_member_is_evicted_on_close(self) -> None:
+        """A rejected tool stays visible, mirroring the errored-tool path."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _LiveToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            t1 = pilot.app.query_one("#t1", ToolCallMessage)
+            t2 = pilot.app.query_one("#t2", ToolCallMessage)
+
+            summary.add_member(t1)
+            summary.add_member(t2)
+            t1.set_rejected(reason="not now")
+            t2.set_success("ok")
+            summary.close()
+            await pilot.pause()
+
+            assert t1.display is True
+            assert not t1.has_class("-grouped")
+            assert t2.display is False
+            rendered = summary.render()
+            assert isinstance(rendered, Content)
+            assert "Read 1 file" in rendered.plain
+
+    async def test_skipped_member_is_evicted_and_uncounted_on_close(self) -> None:
+        """A skipped tool stays visible and is left out of the summary count.
+
+        Regression: `skipped` once fell through `is_success`/`is_failed`/
+        `is_pending`, so a skipped tool stayed folded and inflated the count
+        (e.g. "Ran 1 shell command" for a command that never executed).
+        """
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _LiveToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            t1 = pilot.app.query_one("#t1", ToolCallMessage)  # execute
+            t2 = pilot.app.query_one("#t2", ToolCallMessage)  # read_file
+
+            summary.add_member(t1)
+            summary.add_member(t2)
+            t1.set_skipped()
+            t2.set_success("ok")
+            summary.close()
+            await pilot.pause()
+
+            # The skipped tool is un-folded and no longer part of the group.
+            assert t1.display is True
+            assert not t1.has_class("-grouped")
+            assert t2.display is False
+            rendered = summary.render()
+            assert isinstance(rendered, Content)
+            assert "Read 1 file" in rendered.plain
+            # The skipped execute must not be summarized as if it had run.
+            assert "shell command" not in rendered.plain
+
+    async def test_all_failed_members_remove_summary_on_close(self) -> None:
+        """When every member fails, the empty summary removes itself."""
+        from deepagents_code.widgets.messages import ToolGroupSummary
+
+        async with _LiveToolGroupApp().run_test() as pilot:
+            summary = pilot.app.query_one("#summary", ToolGroupSummary)
+            t1 = pilot.app.query_one("#t1", ToolCallMessage)
+            t2 = pilot.app.query_one("#t2", ToolCallMessage)
+
+            summary.add_member(t1)
+            summary.add_member(t2)
+            t1.set_error("boom")
+            t2.set_rejected(reason="no")
+            summary.close()
+            await pilot.pause()
+
+            # Nothing left to summarize: the summary detaches, both tools show.
+            assert not summary.is_attached
+            assert not pilot.app.query(ToolGroupSummary)
+            assert t1.display is True
+            assert t2.display is True

libs/code/tests/unit_tests/test_textual_adapter.py

@@ -1979,8 +1979,8 @@ def _text_message(text: str) -> SimpleNamespace:
 class TestExecuteTaskTextualParallelToolSpinner:
     """Regression tests for #1796: premature spinner with parallel tools."""
 
-    async def test_spinner_not_shown_until_all_parallel_tools_complete(self) -> None:
-        """With two parallel tools, Thinking appears only at start and after last."""
+    async def test_spinner_stays_up_across_parallel_tools(self) -> None:
+        """With two parallel tools, the spinner stays "Thinking" and is never hidden."""
         statuses: list[str | None] = []
 
         async def record_spinner(status: str | None) -> None:
@@ -2041,11 +2041,9 @@ class TestExecuteTaskTextualParallelToolSpinner:
         )
 
         assert statuses[0] == "Thinking"
-        thinking_count = sum(1 for s in statuses if s == "Thinking")
-        assert thinking_count == 2, (
-            "Expected exactly 2 Thinking calls (start + after last tool); "
-            f"got {thinking_count}: {statuses}"
-        )
+        assert statuses[-1] == "Thinking"
+        # Stable turn-level indicator: never hidden while parallel tools run.
+        assert None not in statuses
 
     async def test_on_tool_complete_fires_per_tool_message(self) -> None:
         """`on_tool_complete` should fire once per `ToolMessage`, even in parallel."""
@@ -2228,12 +2226,12 @@ class TestExecuteTaskTextualParallelToolSpinner:
         tool_msg = adapter._current_tool_messages["tool-1"]
         assert tool_msg._status == "running"
 
-    async def test_edit_file_does_not_get_per_tool_spinner_at_mount(self) -> None:
-        """`edit_file` relies on the global Thinking spinner, not a per-tool one.
+    async def test_edit_file_marks_running_at_mount(self) -> None:
+        """All tool rows are marked running at mount, including `edit_file`.
 
-        Negative counterpart to the auto-executed case: tools in
-        `_TOOL_CALLS_KEEP_THINKING_SPINNER` must NOT be flipped to "running" at
-        mount, or they would show a duplicate spinner alongside "Thinking".
+        The row is hidden inside its collapsed group, so "running" drives the
+        group's live progress state rather than showing a duplicate spinner
+        alongside the global "Thinking" indicator.
         """
         chunks = [
             (
@@ -2269,10 +2267,10 @@ class TestExecuteTaskTextualParallelToolSpinner:
         )
 
         tool_msg = adapter._current_tool_messages["tool-1"]
-        assert tool_msg._status != "running"
+        assert tool_msg._status == "running"
 
     async def test_spinner_with_three_parallel_tools_out_of_order(self) -> None:
-        """Three parallel tools completed out of order; Thinking after all."""
+        """Three parallel tools complete out of order; spinner stays up throughout."""
         statuses: list[str | None] = []
 
         async def record_spinner(status: str | None) -> None:
@@ -2335,11 +2333,9 @@ class TestExecuteTaskTextualParallelToolSpinner:
             adapter=adapter,
         )
 
-        thinking_count = sum(1 for s in statuses if s == "Thinking")
-        assert thinking_count == 2, (
-            "Expected exactly 2 Thinking calls (start + after last tool); "
-            f"got {thinking_count}: {statuses}"
-        )
+        assert statuses[-1] == "Thinking"
+        # Stable turn-level indicator: never hidden while parallel tools run.
+        assert None not in statuses
 
     async def test_spinner_recovers_with_untracked_tool_id(self) -> None:
         """Spinner still shows Thinking with an untracked tool_call_id."""
@@ -2444,24 +2440,14 @@ class TestExecuteTaskTextualTextThenToolSpinner:
                 adapter=adapter,
             )
 
-        # Expected sequence:
-        #   1. "Thinking" before astream
-        #   2. "Thinking" after mounting the streaming AssistantMessage
-        #      (re-anchor the spinner below the message so the user still
-        #      sees activity if the model pauses before the tool call)
-        #   3. None when the tool call mounts
-        #   4. "Thinking" after the tool result
+        # The spinner is a stable turn-level indicator: it shows "Thinking"
+        # before the stream, stays up while text streams and while the tool
+        # runs (the tool's own progress shows in its collapsed group row), and
+        # is never hidden mid-turn — so it no longer flickers off for each tool.
         assert statuses[0] == "Thinking"
-        assert statuses[1] == "Thinking"
-        assert None in statuses
         assert statuses[-1] == "Thinking"
-
-        # The spinner must never be hidden before the tool call arrives.
-        first_none = statuses.index(None)
-        text_thinking_seen = statuses[:first_none].count("Thinking") >= 2
-        assert text_thinking_seen, (
-            f"Spinner was hidden during text streaming before tool call: {statuses}"
-        )
+        assert None not in statuses, f"Spinner was hidden mid-turn: {statuses}"
+        assert all(s == "Thinking" for s in statuses)
 
     async def test_spinner_reanchors_for_text_after_tool_cycle(self) -> None:
         """Text -> tool_call -> tool_result -> text must re-anchor the spinner.
@@ -2516,14 +2502,13 @@ class TestExecuteTaskTextualTextThenToolSpinner:
             f"each text mount; got {thinking_count}: {statuses}"
         )
 
-    async def test_spinner_reanchor_skipped_while_tools_pending(self) -> None:
-        """The re-anchor must be gated on `not _current_tool_messages`.
+    async def test_spinner_stays_up_when_text_arrives_mid_tool(self) -> None:
+        """A text chunk arriving while a tool is in flight keeps the spinner up.
 
         Contrived sequence: a tool call mounts (populating
         `_current_tool_messages`), then a text chunk arrives before the tool
-        result. The new re-anchor logic must NOT call `_set_spinner("Thinking")`
-        in that window — the tool-call widget is the dominant progress
-        indicator.
+        result. The spinner stays "Thinking" throughout and is never hidden —
+        the text re-anchor stays gated on `not _current_tool_messages`.
         """
         statuses: list[str | None] = []
 
@@ -2557,15 +2542,11 @@ class TestExecuteTaskTextualTextThenToolSpinner:
                 adapter=adapter,
             )
 
-        # Thinking calls should be:
-        #   1. Before astream
-        #   2. After tool result (guard is back to empty)
-        # The re-anchor must NOT fire while the tool is in flight.
-        thinking_count = sum(1 for s in statuses if s == "Thinking")
-        assert thinking_count == 2, (
-            f"Expected 2 Thinking calls (start + after tool); got "
-            f"{thinking_count}: {statuses}"
-        )
+        # The spinner stays "Thinking" and is never hidden while the tool is in
+        # flight; the text re-anchor stays gated on no pending tools.
+        assert statuses[0] == "Thinking"
+        assert statuses[-1] == "Thinking"
+        assert None not in statuses
 
 
 class TestExecuteTaskTextualRubricRevisionStreaming: