`de49a58`

Overhaul bash rendering and operator controls

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 3 weeks ago

SHA: de49a584a5da60982e4b3be99b443ffad60d3f01
Parents: b1c0cdf
Tree: 6fefdb2

13 changed files

Status	File	+	-
M	`src/loader/cli/main.py`	297	45
M	`src/loader/runtime/events.py`	1	0
M	`src/loader/runtime/explore.py`	5	0
M	`src/loader/runtime/finalization.py`	5	0
M	`src/loader/runtime/tool_batches.py`	5	0
M	`src/loader/runtime/workflow_lanes.py`	5	0
M	`src/loader/ui/adapter.py`	9	1
M	`src/loader/ui/app.py`	145	13
M	`src/loader/ui/widgets/approval_bar.py`	36	7
M	`src/loader/ui/widgets/input_area.py`	3	0
M	`src/loader/ui/widgets/tool_widget.py`	163	27
A	`tests/test_bash_operator_surfaces.py`	124	0
M	`tests/test_tool_batches.py`	66	1

src/loader/cli/main.pymodified

  import json
  import re
  import sys
 +from typing import Any
  import click
 -from rich.console import Console
 +from rich import box
 +from rich.console import Console, Group
  from rich.markdown import Markdown
  from rich.panel import Panel
  from rich.prompt import Confirm, Prompt
  from rich.table import Table
 +from rich.text import Text
  from ..runtime.inspection import (
      CheckStatus,
              "[bold blue]Loader[/bold blue]\n" + " | ".join(status_parts),
              border_style="blue",
          ))
 -        console.print("[dim]Type 'exit' to quit, 'clear' to reset conversation[/dim]\n")
 +        console.print(
 +            "[dim]Type 'exit' to quit, 'clear' to reset conversation, "
 +            "'jobs' to inspect bash jobs[/dim]\n"
 +        )
          await run_interactive(shell_owner, skip_confirmation=yes)
      else:
          # Launch TUI
      return ", ".join(parts)
 +_SPECIAL_TOOL_LABELS = {
 +    "bash": "Bash",
 +    "bash_jobs": "Bash Jobs",
 +    "bash_wait": "Bash Wait",
 +    "bash_kill": "Bash Kill",
 +}
++
++
 +def _tool_label(tool_name: str, phase: str | None = None) -> str:
 +    label = _SPECIAL_TOOL_LABELS.get(tool_name, tool_name)
 +    if phase == "verification":
 +        return f"Verify {label}"
 +    return label
++
++
 +def _truncate_tool_text(
 +    text: str,
 +    *,
 +    line_limit: int,
 +    char_limit: int = 6_000,
 +) -> tuple[str, bool]:
 +    lines = text.splitlines()
 +    if len(lines) <= line_limit and len(text) <= char_limit:
 +        return text, False
++
 +    preview = "\n".join(lines[:line_limit])
 +    if len(preview) > char_limit:
 +        preview = preview[:char_limit]
 +    return preview, True
++
++
 +def _render_bash_call(tool_args: dict | None, *, phase: str | None = None):
 +    command = str((tool_args or {}).get("command", "")).strip() or "(empty command)"
 +    title = _tool_label("bash", phase)
 +    border_style = "magenta" if phase == "verification" else "cyan"
 +    return Group(
 +        Text(title, style=f"bold {border_style}"),
 +        Panel(
 +            Text(command),
 +            title="Command",
 +            border_style=border_style,
 +            box=box.SQUARE,
 +            expand=True,
 +        ),
 +    )
++
++
 +def _render_bash_result(
 +    content: str,
 +    *,
 +    metadata: dict[str, Any] | None,
 +    is_error: bool,
 +    phase: str | None = None,
 +) -> Panel:
 +    metadata = metadata or {}
 +    title = _tool_label("bash", phase)
 +    lines = []
 +    status_value = str(metadata.get("status", "failed" if is_error else "completed"))
 +    lines.append(f"Status: {status_value.replace('_', ' ')}")
 +    if metadata.get("job_id"):
 +        lines.append(f"Job: {metadata['job_id']}")
 +    if metadata.get("pid"):
 +        lines.append(f"PID: {metadata['pid']}")
 +    if metadata.get("exit_code") is not None:
 +        lines.append(f"Exit: {metadata['exit_code']}")
 +    if metadata.get("background") is not None:
 +        lines.append(
 +            f"Mode: {'background' if metadata.get('background') else 'foreground'}"
 +        )
++
 +    stdout_text = str(metadata.get("stdout", "") or "")
 +    stderr_text = str(metadata.get("stderr", "") or "")
 +    show_summary_note = (
 +        (not stdout_text and not stderr_text and bool(content.strip()))
 +        or status_value not in {"completed", "running"}
 +    )
 +    body = "\n".join(lines)
 +    if show_summary_note and content.strip():
 +        preview, truncated = _truncate_tool_text(content, line_limit=20)
 +        body = f"{body}\n\n{preview}" if body else preview
 +        if truncated:
 +            body += "\n… truncated for display; full result preserved in session"
++
 +    border_style = "red" if is_error else ("magenta" if phase == "verification" else "green")
 +    return Panel(
 +        body or "(no output)",
 +        title=f"[bold {border_style}]{title}[/bold {border_style}]",
 +        border_style=border_style,
 +        box=box.SQUARE,
 +        expand=True,
 +    )
++
++
 +def _print_tool_call(tool_name: str, tool_args: dict | None, phase: str | None = None) -> None:
 +    if tool_name == "bash":
 +        console.print(_render_bash_call(tool_args, phase=phase))
 +        return
++
 +    args_str = _format_tool_args(tool_args)
 +    console.print(f"[cyan]> {_tool_label(tool_name, phase)}[/cyan]({args_str})")
++
++
 +def _print_tool_result(
 +    tool_name: str,
 +    content: str,
 +    *,
 +    metadata: dict[str, Any] | None = None,
 +    is_error: bool = False,
 +    phase: str | None = None,
 +    preview_lines: int = 10,
 +) -> None:
 +    if tool_name == "bash":
 +        console.print(
 +            _render_bash_result(
 +                content,
 +                metadata=metadata,
 +                is_error=is_error,
 +                phase=phase,
 +            )
 +        )
 +        return
++
 +    preview, truncated = _truncate_tool_text(content, line_limit=preview_lines)
 +    if truncated:
 +        preview += "\n[dim]... truncated for display; full result preserved in session[/dim]"
 +    border_style = "red" if is_error else ("magenta" if phase == "verification" else "dim")
 +    console.print(Panel(preview or "(no output)", border_style=border_style))
++
++
 +def _parse_local_bash_command(user_input: str) -> tuple[str, dict[str, object]] | None:
 +    parts = user_input.strip().split()
 +    if not parts:
 +        return None
++
 +    command = parts[0].lstrip("/").lower()
 +    if command == "jobs":
 +        if len(parts) > 2:
 +            raise ValueError("Usage: jobs [limit]")
 +        tool_args: dict[str, object] = {}
 +        if len(parts) == 2:
 +            tool_args["limit"] = max(1, int(parts[1]))
 +        return "bash_jobs", tool_args
++
 +    if command == "wait":
 +        if len(parts) not in {2, 3}:
 +            raise ValueError("Usage: wait <job-id> [timeout-seconds]")
 +        tool_args = {"job_id": parts[1]}
 +        if len(parts) == 3:
 +            tool_args["timeout"] = float(parts[2])
 +        return "bash_wait", tool_args
++
 +    if command == "kill":
 +        if len(parts) not in {2, 3}:
 +            raise ValueError("Usage: kill <job-id> [force-after-ms]")
 +        tool_args = {"job_id": parts[1]}
 +        if len(parts) == 3:
 +            tool_args["force_after_ms"] = int(parts[2])
 +        return "bash_kill", tool_args
++
 +    return None
++
++
 +async def _run_local_bash_command(
 +    shell_owner: RuntimeShellOwner,
 +    tool_name: str,
 +    tool_args: dict[str, object],
 +) -> None:
 +    _print_tool_call(tool_name, tool_args, phase="local")
 +    result = await shell_owner.registry.execute(tool_name, **tool_args)
 +    _print_tool_result(
 +        tool_name,
 +        result.output,
 +        metadata=result.metadata,
 +        is_error=result.is_error,
 +        phase="local",
 +        preview_lines=8,
 +    )
++
++
 +def _get_bash_tool(shell_owner: RuntimeShellOwner):
 +    from ..tools.shell_tools import BashTool
++
 +    tool = shell_owner.registry.get("bash")
 +    return tool if isinstance(tool, BashTool) else None
++
++
 +async def _interrupt_active_foreground_bash(shell_owner: RuntimeShellOwner) -> bool:
 +    bash_tool = _get_bash_tool(shell_owner)
 +    if bash_tool is None:
 +        return False
++
 +    result = await bash_tool.manager.interrupt_active_foreground()
 +    if result is None:
 +        return False
++
 +    _print_tool_result(
 +        "bash",
 +        result.output,
 +        metadata=result.metadata,
 +        is_error=result.is_error,
 +        phase="local",
 +        preview_lines=8,
 +    )
 +    return True
++
++
  async def run_once(
      shell_owner: RuntimeShellOwner,
      prompt: str,
                  elapsed = time.time() - thinking_start
                  console.print(f" [dim]({elapsed:.1f}s)[/dim]")
                  thinking_start = None
 -            args_str = _format_tool_args(event.tool_args)
 -            tool_label = (
 -                f"verify {event.tool_name}"
 -                if event.phase == "verification"
 -                else event.tool_name
 +            _print_tool_call(
 +                getattr(event, "tool_name", "") or "",
 +                getattr(event, "tool_args", None),
 +                getattr(event, "phase", None),
+             )
 -            console.print(f"[cyan]> {tool_label}[/cyan]({args_str})")
          elif event.type == "tool_result":
 -            # Show result in a compact panel
 -            lines = event.content.splitlines()
 -            preview = "\n".join(lines[:10])
 -            if len(lines) > 10:
 -                preview += f"\n[dim]... ({len(lines) - 10} more lines)[/dim]"
 -            border_style = "magenta" if event.phase == "verification" else "dim"
 -            console.print(Panel(preview, border_style=border_style))
 +            _print_tool_result(
 +                getattr(event, "tool_name", "") or "",
 +                event.content,
 +                metadata=getattr(event, "tool_metadata", None),
 +                is_error=getattr(event, "is_error", False),
 +                phase=getattr(event, "phase", None),
 +                preview_lines=10,
 +            )
          elif event.type == "dod_status":
              console.print(f"[dim]{format_dod_status(event)}[/dim]")
          elif event.type == "recovery":
          console.print("\n[red]Request timed out.[/red]")
          console.print("[dim]The model is taking too long. Try a smaller model or simpler prompt.[/dim]")
          return
 +    except KeyboardInterrupt:
 +        console.print()
 +        if not await _interrupt_active_foreground_bash(shell_owner):
 +            console.print("[yellow]Cancelled.[/yellow]")
 +        return
      except ConfirmationRequired as e:
          console.print(f"\n[yellow]Confirmation required:[/yellow] {e.message}")
          if e.details:
          if Confirm.ask("Proceed?"):
              shell_owner.registry.skip_confirmation = True
              streamed_response = False  # Reset for continuation
 -            response = await shell_owner.run(
 -                "Continue with the previous action.",
 -                on_event=on_event,
 -                on_user_question=_ask_user_question_cli,
 -            )
 -            if not streamed_response:
 -                console.print(Markdown(clean_response(response)))
 -            shell_owner.registry.skip_confirmation = skip_confirmation
 +            try:
 +                response = await shell_owner.run(
 +                    "Continue with the previous action.",
 +                    on_event=on_event,
 +                    on_user_question=_ask_user_question_cli,
 +                )
 +                if not streamed_response:
 +                    console.print(Markdown(clean_response(response)))
 +            except KeyboardInterrupt:
 +                console.print()
 +                if not await _interrupt_active_foreground_bash(shell_owner):
 +                    console.print("[yellow]Cancelled.[/yellow]")
 +            finally:
 +                shell_owner.registry.skip_confirmation = skip_confirmation
          else:
              console.print("[red]Aborted.[/red]")
      history_file = os.path.expanduser("~/.loader_history")
      session = PromptSession(history=FileHistory(history_file))
 -    console.print("[dim]Type 'exit' to quit, 'clear' to reset conversation[/dim]\n")
 +    console.print(
 +        "[dim]Type 'exit' to quit, 'clear' to reset conversation, "
 +        "'jobs' to inspect bash jobs[/dim]\n"
 +    )
      while True:
          try:
              console.print("[dim]Conversation cleared[/dim]")
              continue
 +        try:
 +            local_bash = _parse_local_bash_command(user_input)
 +        except ValueError as exc:
 +            console.print(f"[red]{exc}[/red]\n")
 +            continue
++
 +        if local_bash is not None:
 +            tool_name, tool_args = local_bash
 +            await _run_local_bash_command(shell_owner, tool_name, tool_args)
 +            console.print()
 +            continue
++
          import time
          thinking_start = None
          streaming_started = False
                  if streaming_started:
                      console.print()  # New line after any streamed content
                      streaming_started = False
 -                args_str = _format_tool_args(event.tool_args)
 -                tool_label = (
 -                    f"verify {event.tool_name}"
 -                    if event.phase == "verification"
 -                    else event.tool_name
 +                _print_tool_call(
 +                    getattr(event, "tool_name", "") or "",
 +                    getattr(event, "tool_args", None),
 +                    getattr(event, "phase", None),
+                 )
 -                console.print(f"[cyan]> {tool_label}[/cyan]({args_str})")
              elif event.type == "tool_result":
 -                # Show compact result
 -                lines = event.content.splitlines()
 -                if len(lines) <= 3:
 -                    preview = event.content
 -                else:
 -                    preview = "\n".join(lines[:3]) + f"\n[dim]... ({len(lines) - 3} more lines)[/dim]"
 -                style = "magenta" if event.phase == "verification" else "dim"
 -                console.print(f"[{style}]{preview}[/{style}]")
 +                _print_tool_result(
 +                    getattr(event, "tool_name", "") or "",
 +                    event.content,
 +                    metadata=getattr(event, "tool_metadata", None),
 +                    is_error=getattr(event, "is_error", False),
 +                    phase=getattr(event, "phase", None),
 +                    preview_lines=3,
 +                )
              elif event.type == "dod_status":
                  console.print(f"\n[dim]{format_dod_status(event)}[/dim]")
              elif event.type == "recovery":
              console.print("  • A simpler prompt")
              console.print("  • Check if Ollama is overloaded")
              continue
 +        except KeyboardInterrupt:
 +            console.print()
 +            if not await _interrupt_active_foreground_bash(shell_owner):
 +                console.print("[yellow]Cancelled.[/yellow]")
 +            console.print()
 +            continue
          except ConfirmationRequired as e:
              console.print(f"\n[yellow]Confirmation required:[/yellow] {e.message}")
              if e.details:
                      if not streamed_response:
                          console.print(Markdown(clean_response(response)))
                      console.print()
 +                except KeyboardInterrupt:
 +                    console.print()
 +                    if not await _interrupt_active_foreground_bash(shell_owner):
 +                        console.print("[yellow]Cancelled.[/yellow]")
 +                    console.print()
                  finally:
                      shell_owner.registry.skip_confirmation = skip_confirmation
              else:
      def on_event(event) -> None:
          if event.type == "tool_call":
 -            args_str = _format_tool_args(event.tool_args)
 -            console.print(f"[cyan]> {event.tool_name}[/cyan]({args_str})")
 +            _print_tool_call(
 +                getattr(event, "tool_name", "") or "",
 +                getattr(event, "tool_args", None),
 +                getattr(event, "phase", None),
 +            )
          elif event.type == "tool_result":
 -            lines = event.content.splitlines()
 -            preview = "\n".join(lines[:8])
 -            if len(lines) > 8:
 -                preview += f"\n[dim]... ({len(lines) - 8} more lines)[/dim]"
 -            console.print(Panel(preview, border_style="dim"))
 +            _print_tool_result(
 +                getattr(event, "tool_name", "") or "",
 +                event.content,
 +                metadata=getattr(event, "tool_metadata", None),
 +                is_error=getattr(event, "is_error", False),
 +                phase=getattr(event, "phase", None),
 +                preview_lines=8,
 +            )
      response = await shell_owner.run_explore(prompt, on_event=on_event, fresh=fresh)
      console.print(Markdown(clean_response(response)))

src/loader/runtime/events.pymodified

      content: str = ""
      tool_name: str | None = None
      tool_args: dict[str, Any] | None = None
 +    tool_metadata: dict[str, Any] | None = None
      phase: str | None = None
      step_info: str | None = None
      recovery_attempt: int | None = None

src/loader/runtime/explore.pymodified

                              type="tool_result",
                              content=outcome.event_content,
                              tool_name=tool_call.name,
 +                            tool_metadata=(
 +                                outcome.registry_result.metadata
 +                                if outcome.registry_result is not None
 +                                else None
 +                            ),
                              is_error=outcome.is_error,
                              phase="explore",
+                         )

src/loader/runtime/finalization.pymodified

                      type="tool_result",
                      content=outcome.event_content,
                      tool_name=verification_call.name,
 +                    tool_metadata=(
 +                        outcome.registry_result.metadata
 +                        if outcome.registry_result is not None
 +                        else None
 +                    ),
                      is_error=outcome.is_error,
                      phase="verification",
+                 )

src/loader/runtime/tool_batches.pymodified

                      type="tool_result",
                      content=outcome.event_content,
                      tool_name=tool_call.name,
 +                    tool_metadata=(
 +                        outcome.registry_result.metadata
 +                        if outcome.registry_result is not None
 +                        else None
 +                    ),
                      is_error=outcome.is_error,
                      phase="assistant",
+                 )

src/loader/runtime/workflow_lanes.pymodified

                  type="tool_result",
                  content=outcome.event_content,
                  tool_name=tool_call.name,
 +                tool_metadata=(
 +                    outcome.registry_result.metadata
 +                    if outcome.registry_result is not None
 +                    else None
 +                ),
                  is_error=outcome.is_error,
                  phase="plan",
+             )

src/loader/ui/adapter.pymodified

  """Event adapter bridging Agent events to Textual messages."""
  from dataclasses import dataclass
 +from typing import Any
  from typing import TYPE_CHECKING
  from textual.message import Message
      content: str
      is_error: bool = False
      phase: str | None = None
 +    metadata: dict[str, Any] | None = None
      # For edit tool diffs
      old_string: str | None = None
      new_string: str | None = None
                          content=event.content,
                          is_error=event.is_error,
                          phase=event.phase,
 +                        metadata=event.tool_metadata,
                          old_string=old_string,
                          new_string=new_string,
                          file_path=file_path,
                  # Update the todo list widget when TodoWrite succeeds
                  if tool_name == "TodoWrite" and not event.is_error:
 -                    new_todos = self._extract_todos(event.content, tool_args)
 +                    metadata_todos = (event.tool_metadata or {}).get("new_todos", [])
 +                    new_todos = (
 +                        metadata_todos
 +                        if isinstance(metadata_todos, list) and metadata_todos
 +                        else self._extract_todos(event.content, tool_args)
 +                    )
                      if new_todos:
                          self.app.post_message(TodoListUpdated(todos=new_todos))

src/loader/ui/app.pymodified

  from ..runtime.events import AgentEvent
  from ..runtime.runtime_api import RuntimeShellOwner
 +from ..tools.shell_tools import BashTool
  from .adapter import (
      ArtifactCreated,
      ClearStream,
              "Press Ctrl+C to quit, Ctrl+L to clear.[/dim]"
+         )
          self._add_message(
 -            "[dim]Commands: /help, /model, /clear, /exit[/dim]"
 +            "[dim]Commands: /help, /model, /jobs, /wait, /kill, /clear, /exit[/dim]"
+         )
      def _add_message(self, content: str, classes: str = "") -> None:
          elif cmd == "models":
              self._handle_model_command("")  # List models
 +        elif cmd == "jobs":
 +            self._handle_jobs_command(args)
++
 +        elif cmd == "wait":
 +            self._handle_wait_command(args)
++
 +        elif cmd == "kill":
 +            self._handle_kill_command(args)
++
          else:
              self._add_message(
                  f"[red]Unknown command: /{cmd}[/red]\n"
  [cyan]/clear[/cyan], [cyan]/c[/cyan]         Clear the conversation
  [cyan]/model[/cyan], [cyan]/models[/cyan]    Open model selector (fzf-style)
  [cyan]/model[/cyan] [dim]<name>[/dim]     Switch to a specific model
 +[cyan]/jobs[/cyan] [dim][limit][/dim]     List active and recent bash jobs
 +[cyan]/wait[/cyan] [dim]<job-id> [timeout][/dim] Wait for a bash job to finish
 +[cyan]/kill[/cyan] [dim]<job-id> [ms][/dim] Stop a tracked bash job
  [bold]Shortcuts:[/bold]
  [dim]Ctrl+C[/dim]            Exit
 -[dim]Ctrl+L[/dim]            Clear conversation"""
 +[dim]Ctrl+L[/dim]            Clear conversation
 +[dim]Esc[/dim]               Interrupt foreground bash or cancel the turn"""
          self._add_message(help_text)
 +    def _get_bash_tool(self) -> BashTool | None:
 +        tool = self.shell_owner.registry.get("bash")
 +        return tool if isinstance(tool, BashTool) else None
++
 +    def _launch_local_tool(self, tool_name: str, tool_args: dict[str, object]) -> None:
 +        asyncio.create_task(self._execute_local_tool(tool_name, tool_args))
++
 +    async def _execute_local_tool(
 +        self,
 +        tool_name: str,
 +        tool_args: dict[str, object],
 +    ) -> None:
 +        self.post_message(
 +            ToolCallStarted(
 +                tool_name=tool_name,
 +                tool_args=tool_args,
 +                phase="local",
 +            )
 +        )
 +        try:
 +            result = await self.shell_owner.registry.execute(tool_name, **tool_args)
 +        except Exception as exc:
 +            self.post_message(
 +                ToolCallCompleted(
 +                    tool_name=tool_name,
 +                    content=f"Tool execution error: {exc}",
 +                    is_error=True,
 +                    phase="local",
 +                )
 +            )
 +            return
 +        self.post_message(
 +            ToolCallCompleted(
 +                tool_name=tool_name,
 +                content=result.output,
 +                is_error=result.is_error,
 +                phase="local",
 +                metadata=result.metadata,
 +            )
 +        )
++
 +    def _handle_jobs_command(self, args: str) -> None:
 +        limit = 20
 +        if args.strip():
 +            try:
 +                limit = max(1, int(args.strip()))
 +            except ValueError:
 +                self._add_message("[red]Usage: /jobs [limit][/red]")
 +                return
 +        self._launch_local_tool("bash_jobs", {"limit": limit})
++
 +    def _handle_wait_command(self, args: str) -> None:
 +        parts = args.split()
 +        if not parts:
 +            self._add_message("[red]Usage: /wait <job-id> [timeout-seconds][/red]")
 +            return
 +        tool_args: dict[str, object] = {"job_id": parts[0]}
 +        if len(parts) > 1:
 +            try:
 +                tool_args["timeout"] = float(parts[1])
 +            except ValueError:
 +                self._add_message("[red]Usage: /wait <job-id> [timeout-seconds][/red]")
 +                return
 +        self._launch_local_tool("bash_wait", tool_args)
++
 +    def _handle_kill_command(self, args: str) -> None:
 +        parts = args.split()
 +        if not parts:
 +            self._add_message("[red]Usage: /kill <job-id> [force-after-ms][/red]")
 +            return
 +        tool_args: dict[str, object] = {"job_id": parts[0]}
 +        if len(parts) > 1:
 +            try:
 +                tool_args["force_after_ms"] = int(parts[1])
 +            except ValueError:
 +                self._add_message("[red]Usage: /kill <job-id> [force-after-ms][/red]")
 +                return
 +        self._launch_local_tool("bash_kill", tool_args)
++
 +    async def _interrupt_active_bash_job(self) -> None:
 +        bash_tool = self._get_bash_tool()
 +        if bash_tool is None:
 +            return
 +        await bash_tool.manager.interrupt_active_foreground()
++
 +    def _terminate_all_bash_jobs(self) -> list[str]:
 +        bash_tool = self._get_bash_tool()
 +        if bash_tool is None:
 +            return []
 +        return bash_tool.manager.terminate_all_now()
++
      def _handle_model_command(self, args: str) -> None:
          """Handle /model command - switch or show selector."""
          if not args:
          # Create tool widget
          widget = ToolCallWidget(
 -            tool_name=(
 -                f"verify {message.tool_name}"
 -                if message.phase == "verification"
 -                else message.tool_name
 -            ),
 +            tool_name=message.tool_name,
              tool_args=message.tool_args,
 +            phase=message.phase,
+         )
          msg_area.mount(widget)
          widget.set_running()  # Must be after mount() so children exist
          tool_widget = None
          if self._tool_widget_queue:
              for i, w in enumerate(self._tool_widget_queue):
 -                # Match on tool name (strip "verify " prefix for verification phase)
 -                widget_name = w.tool_name.removeprefix("verify ")
 -                if widget_name == message.tool_name:
 +                if w.tool_name == message.tool_name and w.phase == message.phase:
                      tool_widget = self._tool_widget_queue.pop(i)
                      break
              else:
 -                # No name match — fall back to FIFO
 -                tool_widget = self._tool_widget_queue.pop(0)
 +                for i, w in enumerate(self._tool_widget_queue):
 +                    if w.tool_name == message.tool_name:
 +                        tool_widget = self._tool_widget_queue.pop(i)
 +                        break
 +                else:
 +                    # No name match — fall back to FIFO
 +                    tool_widget = self._tool_widget_queue.pop(0)
          # Check if this is an edit tool with diff info
          # Note: old_string can be empty string (inserting), so check `is not None`
              # Update existing tool widget with result
              self._debug_log("  -> showing regular tool widget result")
              tool_widget.set_result(
 -                message.content, is_error=message.is_error
 +                message.content,
 +                is_error=message.is_error,
 +                metadata=message.metadata,
+             )
          msg_area.scroll_end(animate=False)
      # Actions
      def action_clear_messages(self) -> None:
          """Clear all messages."""
 +        killed_jobs = self._terminate_all_bash_jobs()
 +        self.workers.cancel_all()
 +        self.is_generating = False
 +        self._stop_timer()
 +        self.query_one(StatusLine).set_generating(False)
          msg_area = self.query_one("#message-area", ScrollableContainer)
          msg_area.remove_children()
 +        self._current_streaming = None
 +        self._streamed_content = False
 +        self._tool_widget_queue.clear()
          self.shell_owner.clear_history()
          self.query_one(StatusLine).clear_definition_of_done()
          self.query_one(StatusLine).update_session_id(self.shell_owner.session.session_id)
+         )
          self.query_one(StatusLine).update_workflow_mode("execute")
          self._add_message("[dim]Conversation cleared.[/dim]")
 +        if killed_jobs:
 +            self._add_message(
 +                f"[yellow]Stopped bash jobs:[/yellow] {', '.join(killed_jobs)}"
 +            )
      def action_cancel(self) -> None:
          """Cancel current operation."""
 +        bash_tool = self._get_bash_tool()
 +        if (
 +            bash_tool is not None
 +            and bash_tool.manager.active_foreground_job_id is not None
 +        ):
 +            asyncio.create_task(self._interrupt_active_bash_job())
 +            return
          # Cancel any running workers
          self.workers.cancel_all()
          self.is_generating = False
          self._stop_timer()
          self.query_one(StatusLine).set_generating(False)
 +    def on_unmount(self) -> None:
 +        """Clean up any tracked bash jobs when the TUI exits."""
 +        killed_jobs = self._terminate_all_bash_jobs()
 +        if killed_jobs:
 +            self._debug_log(f"on_unmount: stopped bash jobs {killed_jobs}")
++
  def _definition_of_done_verification_attempt(dod) -> str | None:
      """Render one compact verification-attempt label from DoD state."""

src/loader/ui/widgets/approval_bar.pymodified

  """Approval bar widget for command confirmation (Claude Code style)."""
 +from rich import box
 +from rich.console import Group
 +from rich.panel import Panel
 +from rich.text import Text
  from textual.app import ComposeResult
  from textual.binding import Binding
  from textual.message import Message
      DEFAULT_CSS = """
      ApprovalBar {
          height: auto;
 -        max-height: 4;
 +        max-height: 12;
          display: none;
          padding: 0 1;
          background: $warning 15%;
          self._full_command = details
          preview = details if details else message
 -        if len(preview) > 70:
 -            preview = preview[:67] + "..."
          content = self.query_one("#approval-content", Static)
 -        content.update(
 -            f"[bold $warning]\\[{tool_name}][/] {preview}  "
 -            f"[bold green]\\[Y][/]es  [bold red]\\[n][/]o  [bold]\\[e][/]dit"
 -        )
 +        if tool_name == "bash":
 +            header = Text("Bash", style="bold yellow")
 +            command = Text(preview or "(empty command)")
 +            controls = Text.assemble(
 +                ("[Y]", "bold green"),
 +                ("es  ",),
 +                ("[n]", "bold red"),
 +                ("o  ",),
 +                ("[e]", "bold"),
 +                ("dit",),
 +            )
 +            content.update(
 +                Group(
 +                    header,
 +                    Panel(
 +                        command,
 +                        title="Command",
 +                        border_style="yellow",
 +                        box=box.SQUARE,
 +                        expand=True,
 +                    ),
 +                    controls,
 +                )
 +            )
 +        else:
 +            if len(preview) > 70:
 +                preview = preview[:67] + "..."
 +            content.update(
 +                f"[bold $warning]\\[{tool_name}][/] {preview}  "
 +                f"[bold green]\\[Y][/]es  [bold red]\\[n][/]o  [bold]\\[e][/]dit"
 +            )
          # Show the bar
          self.add_class("visible")

src/loader/ui/widgets/input_area.pymodified

      "/help",
      "/model",
      "/models",
 +    "/jobs",
 +    "/wait",
 +    "/kill",
      "/clear",
      "/exit",
+ ]

src/loader/ui/widgets/tool_widget.pymodified

 -"""Tool call widget with inline truncation (claw-code style)."""
 +"""Tool call widget with bash-specific rich rendering."""
 +from typing import Any
++
 +from rich import box
 +from rich.console import Group
  from rich.markup import escape
 +from rich.panel import Panel
  from rich.text import Text
  from textual.app import ComposeResult
  from textual.containers import Vertical
  class ToolCallWidget(Vertical):
      """Widget for tool calls with inline content display."""
 +    TOOL_LABELS = {
 +        "bash": "Bash",
 +        "bash_jobs": "Bash Jobs",
 +        "bash_wait": "Bash Wait",
 +        "bash_kill": "Bash Kill",
 +    }
++
      TOOL_BULLETS = {
          "pending": "[yellow]○[/yellow]",
          "running": "[yellow]◐[/yellow]",
          self,
          tool_name: str,
          tool_args: dict | None = None,
 +        phase: str | None = None,
          **kwargs,
      ) -> None:
          super().__init__(**kwargs)
          self.tool_name = tool_name
          self.tool_args = tool_args or {}
 +        self.phase = phase
          self._result: str = ""
          self._is_error: bool = False
 +        self._metadata: dict[str, Any] = {}
      def compose(self) -> ComposeResult:
 -        args_str = self._format_args()
+-
          yield Static(
 -            f"{self.TOOL_BULLETS['pending']} [bold cyan]{self.tool_name}[/bold cyan]({args_str})",
 +            self._header_markup(),
              id="tool-header",
              classes="tool-header",
+         )
 +        yield Static(self._build_initial_summary(), id="tool-summary", classes="tool-summary")
++
 +    def _format_args(self) -> str:
 +        """Format tool arguments for display."""
 +        if self._is_bash_command_tool():
 +            return ""
 +        if not self.tool_args:
 +            return ""
 +        parts = []
 +        for k, v in self.tool_args.items():
 +            if isinstance(v, str):
 +                limit = 200 if k in ("file_path", "path") else (80 if k == "content" else 40)
 +                if len(v) > limit:
 +                    v = v[: limit - 3] + "..."
 +                parts.append(f'{k}="[dim]{escape(v)}[/dim]"')
 +            else:
 +                parts.append(f"{k}={escape(repr(v))}")
 +        return ", ".join(parts)
++
 +    def _display_name(self) -> str:
 +        base = self.TOOL_LABELS.get(self.tool_name, self.tool_name)
 +        if self.phase == "verification":
 +            return f"Verify {base}"
 +        return base
++
 +    def _is_bash_command_tool(self) -> bool:
 +        return self.tool_name == "bash"
++
 +    def _header_markup(self) -> str:
 +        args_str = self._format_args()
 +        bullet = self.TOOL_BULLETS.get(self.state, self.TOOL_BULLETS["pending"])
 +        color = "red" if self._is_error else "cyan"
 +        label = self._display_name()
 +        suffix = f"({args_str})" if args_str else ""
 +        return f"{bullet} [bold {color}]{label}[/bold {color}]{suffix}"
++
 +    def _build_initial_summary(self):
 +        if self._is_bash_command_tool():
 +            return Group(self._render_bash_command_panel())
          # For write/edit tools, show content as pre-approval preview
          initial_summary = Text()
                          f"({_TRUNCATION_NOTICE})\n",
                          style="dim",
+                     )
 -        yield Static(initial_summary, id="tool-summary", classes="tool-summary")
 +        return initial_summary
 -    def _format_args(self) -> str:
 -        """Format tool arguments for display."""
 -        if not self.tool_args:
 -            return ""
 -        parts = []
 -        for k, v in self.tool_args.items():
 -            if isinstance(v, str):
 -                limit = 200 if k in ("file_path", "path") else (80 if k == "content" else 40)
 -                if len(v) > limit:
 -                    v = v[: limit - 3] + "..."
 -                parts.append(f'{k}="[dim]{escape(v)}[/dim]"')
 -            else:
 -                parts.append(f"{k}={escape(repr(v))}")
 -        return ", ".join(parts)
 +    def _render_bash_command_panel(self) -> Panel:
 +        command = str(self.tool_args.get("command", "")).strip() or "(empty command)"
 +        return Panel(
 +            Text(command),
 +            title="Command",
 +            border_style="cyan",
 +            box=box.SQUARE,
 +            expand=True,
 +        )
++
 +    def _truncate_result(self, result: str, *, line_limit: int) -> tuple[str, bool]:
 +        lines = result.splitlines()
 +        if len(lines) <= line_limit and len(result) <= TOOL_RESULT_MAX_CHARS:
 +            return result, False
++
 +        display = lines[:line_limit]
 +        text = "\n".join(display)
 +        if len(text) > TOOL_RESULT_MAX_CHARS:
 +            text = text[:TOOL_RESULT_MAX_CHARS]
 +        return text, True
++
 +    def _build_bash_result(self, result: str):
 +        metadata = self._metadata
 +        renderables = [self._render_bash_command_panel()]
 +        status = Text()
 +        status.append(
 +            "✗ Failed\n" if self._is_error else "✓ Success\n",
 +            style="bold red" if self._is_error else "bold green",
 +        )
++
 +        detail_lines = []
 +        status_value = str(metadata.get("status", "failed" if self._is_error else "completed"))
 +        detail_lines.append(f"Status: {status_value.replace('_', ' ')}")
 +        job_id = metadata.get("job_id")
 +        if job_id:
 +            detail_lines.append(f"Job: {job_id}")
 +        pid = metadata.get("pid")
 +        if pid:
 +            detail_lines.append(f"PID: {pid}")
 +        if metadata.get("exit_code") is not None:
 +            detail_lines.append(f"Exit: {metadata['exit_code']}")
 +        if metadata.get("background") is not None:
 +            detail_lines.append(
 +                f"Mode: {'background' if metadata.get('background') else 'foreground'}"
 +            )
 +        if detail_lines:
 +            status.append("\n".join(detail_lines))
++
 +        stdout_text = str(metadata.get("stdout", "") or "")
 +        stderr_text = str(metadata.get("stderr", "") or "")
 +        show_summary_note = (
 +            (not stdout_text and not stderr_text and bool(result.strip()))
 +            or status_value not in {"completed", "running"}
 +        )
 +        if show_summary_note and result.strip():
 +            if status.plain:
 +                status.append("\n\n")
 +            preview, truncated = self._truncate_result(result, line_limit=24)
 +            status.append(preview)
 +            if truncated:
 +                status.append(f"\n… {_TRUNCATION_NOTICE}", style="dim")
++
 +        renderables.append(
 +            Panel(
 +                status,
 +                title="Status",
 +                border_style="red" if self._is_error else "green",
 +                box=box.SQUARE,
 +                expand=True,
 +            )
 +        )
++
 +        for stream_name, stream_text, truncated in (
 +            ("stdout", stdout_text, bool(metadata.get("stdout_truncated"))),
 +            ("stderr", stderr_text, bool(metadata.get("stderr_truncated"))),
 +        ):
 +            if not stream_text:
 +                continue
 +            preview, preview_truncated = self._truncate_result(stream_text, line_limit=40)
 +            stream_panel_text = Text(preview)
 +            if truncated or preview_truncated:
 +                stream_panel_text.append(f"\n… {_TRUNCATION_NOTICE}", style="dim")
 +            renderables.append(
 +                Panel(
 +                    stream_panel_text,
 +                    title=stream_name,
 +                    border_style="red" if stream_name == "stderr" else "dim",
 +                    box=box.SQUARE,
 +                    expand=True,
 +                )
 +            )
++
 +        return Group(*renderables)
      def set_running(self) -> None:
          """Mark as running."""
          self.state = "running"
          self._update_header()
 -    def set_result(self, result: str, is_error: bool = False) -> None:
 +    def set_result(
 +        self,
 +        result: str,
 +        is_error: bool = False,
 +        metadata: dict[str, Any] | None = None,
 +    ) -> None:
          """Update widget with tool result using inline truncation."""
          self._result = result
          self._is_error = is_error
 +        self._metadata = metadata or {}
          self.state = "error" if is_error else "success"
 -        self.remove_class("pending", "error", "success")
 +        self.remove_class("pending", "running", "error", "success")
          self.add_class(self.state)
          self._update_header()
 +        if self._is_bash_command_tool():
 +            self.query_one("#tool-summary", Static).update(self._build_bash_result(result))
 +            return
++
          summary = Text()
          if is_error:
              summary.append("✗ Failed\n", style="bold red")
      def _update_header(self) -> None:
          """Update the header with current state."""
 -        args_str = self._format_args()
 -        bullet = self.TOOL_BULLETS.get(self.state, self.TOOL_BULLETS["pending"])
 -        color = "red" if self._is_error else "cyan"
 -        self.query_one("#tool-header", Static).update(
 -            f"{bullet} [bold {color}]{self.tool_name}[/bold {color}]({args_str})"
 -        )
 +        self.query_one("#tool-header", Static).update(self._header_markup())
      def watch_state(self, state: str) -> None:
          """React to state changes."""

tests/test_bash_operator_surfaces.pyadded

 +"""Tests for bash job metadata and operator-facing surfaces."""
++
 +from __future__ import annotations
++
 +from types import SimpleNamespace
++
 +import pytest
 +from rich.console import Console
++
 +import loader.cli.main as cli_main_module
 +from loader.runtime.events import AgentEvent
 +from loader.tools import BashTool
 +from loader.ui.adapter import EventAdapter, ToolCallCompleted
 +from loader.ui.widgets.tool_widget import ToolCallWidget
++
++
 +class _FakeApp:
 +    def __init__(self) -> None:
 +        self.messages: list[object] = []
++
 +    def post_message(self, message: object) -> None:
 +        self.messages.append(message)
++
++
 +def _render_text(renderable, *, width: int = 100) -> str:
 +    console = Console(record=True, width=width)
 +    console.print(renderable)
 +    return console.export_text(styles=False)
++
++
 +def test_event_adapter_preserves_tool_metadata_on_completion() -> None:
 +    app = _FakeApp()
 +    adapter = EventAdapter(app)  # type: ignore[arg-type]
 +    metadata = {"job_id": "bash-3", "status": "running", "background": True}
++
 +    adapter.handle_event(
 +        AgentEvent(
 +            type="tool_call",
 +            tool_name="bash",
 +            tool_args={"command": "python -m http.server 8000", "background": True},
 +            phase="assistant",
 +        )
 +    )
 +    adapter.handle_event(
 +        AgentEvent(
 +            type="tool_result",
 +            tool_name="bash",
 +            content="Started bash job bash-3",
 +            tool_metadata=metadata,
 +            phase="assistant",
 +        )
 +    )
++
 +    completed = next(message for message in app.messages if isinstance(message, ToolCallCompleted))
 +    assert completed.metadata == metadata
++
++
 +def test_tool_call_widget_renders_full_bash_command_in_box() -> None:
 +    command = "python -m http.server 8000 --directory /tmp/preview-pages"
 +    widget = ToolCallWidget("bash", {"command": command})
++
 +    header = widget._header_markup()
 +    rendered = _render_text(widget._build_initial_summary(), width=120)
++
 +    assert "Bash" in header
 +    assert "command=" not in header
 +    assert "Command" in rendered
 +    assert command in rendered
++
++
 +def test_cli_print_tool_call_renders_bash_panel_without_truncating(monkeypatch: pytest.MonkeyPatch) -> None:
 +    console = Console(record=True, width=120)
 +    monkeypatch.setattr(cli_main_module, "console", console)
++
 +    command = "python -m http.server 8000 --directory /tmp/preview-pages"
 +    cli_main_module._print_tool_call("bash", {"command": command})
++
 +    rendered = console.export_text(styles=False)
 +    assert "Bash" in rendered
 +    assert "Command" in rendered
 +    assert command in rendered
 +    assert "command=" not in rendered
++
++
 +def test_cli_parse_local_bash_commands_supports_slash_aliases() -> None:
 +    assert cli_main_module._parse_local_bash_command("/jobs 5") == ("bash_jobs", {"limit": 5})
 +    assert cli_main_module._parse_local_bash_command("/wait bash-7 2.5") == (
 +        "bash_wait",
 +        {"job_id": "bash-7", "timeout": 2.5},
 +    )
 +    assert cli_main_module._parse_local_bash_command("kill bash-2 50") == (
 +        "bash_kill",
 +        {"job_id": "bash-2", "force_after_ms": 50},
 +    )
++
++
 +@pytest.mark.asyncio
 +async def test_cli_interrupt_active_foreground_bash_prints_interrupted_result(
 +    monkeypatch: pytest.MonkeyPatch,
 +) -> None:
 +    console = Console(record=True, width=120)
 +    monkeypatch.setattr(cli_main_module, "console", console)
++
 +    bash_tool = BashTool(timeout=10.0)
 +    job = await bash_tool.manager.start(
 +        command='python -c "import time; time.sleep(30)"',
 +        cwd=None,
 +        timeout=10.0,
 +        background=False,
 +        mutability="workspace-write",
 +    )
 +    owner = SimpleNamespace(
 +        registry=SimpleNamespace(get=lambda name: bash_tool if name == "bash" else None)
 +    )
++
 +    try:
 +        interrupted = await cli_main_module._interrupt_active_foreground_bash(owner)
 +        assert interrupted is True
 +        assert bash_tool.manager.active_foreground_job_id is None
 +        rendered = console.export_text(styles=False)
 +        assert "Status: interrupted" in rendered
 +    finally:
 +        if job.is_running:
 +            await bash_tool.manager.kill_job(job.job_id, interrupted=True)

tests/test_tool_batches.pymodified

      tool_call: ToolCall,
      output: str,
      is_error: bool,
 +    metadata: dict[str, object] | None = None,
  ) -> ToolExecutionOutcome:
      return ToolExecutionOutcome(
          tool_call=tool_call,
          event_content=output,
          is_error=is_error,
          result_output=output,
 -        registry_result=RegistryToolResult(output=output, is_error=is_error),
 +        registry_result=RegistryToolResult(
 +            output=output,
 +            is_error=is_error,
 +            metadata=metadata or {},
 +        ),
+     )
      assert any(event.type == "recovery" for event in events)
 +@pytest.mark.asyncio
 +async def test_tool_batch_runner_emits_tool_metadata(temp_dir: Path) -> None:
 +    async def assess_confidence(tool_name: str, tool_args: dict, context: str) -> ConfidenceAssessment:
 +        raise AssertionError("Confidence scoring should be disabled in this scenario")
++
 +    async def verify_action(tool_name: str, tool_args: dict, result: str, expected: str = "") -> ActionVerification:
 +        raise AssertionError("Verification should not run for this scenario")
++
 +    context = build_context(
 +        temp_dir=temp_dir,
 +        messages=[],
 +        safeguards=FakeSafeguards(),
 +        assess_confidence=assess_confidence,
 +        verify_action=verify_action,
 +        auto_recover=False,
 +    )
 +    runner = ToolBatchRunner(context, DefinitionOfDoneStore(temp_dir))
 +    tool_call = ToolCall(
 +        id="bash-1",
 +        name="bash",
 +        arguments={"command": "python -m http.server 8000", "background": True},
 +    )
 +    metadata = {
 +        "job_id": "bash-1",
 +        "status": "running",
 +        "background": True,
 +    }
 +    executor = FakeExecutor(
 +        [
 +            tool_outcome(
 +                tool_call=tool_call,
 +                output="Started bash job bash-1",
 +                is_error=False,
 +                metadata=metadata,
 +            )
 +        ]
 +    )
 +    events: list[AgentEvent] = []
++
 +    async def emit(event: AgentEvent) -> None:
 +        events.append(event)
++
 +    await runner.execute_batch(
 +        tool_calls=[tool_call],
 +        tool_source="assistant",
 +        pending_tool_calls_seen=set(),
 +        emit=emit,
 +        summary=TurnSummary(final_response=""),
 +        dod=create_definition_of_done("Launch a preview server"),
 +        executor=executor,  # type: ignore[arg-type]
 +        on_confirmation=None,
 +        on_user_question=None,
 +        emit_confirmation=None,
 +        consecutive_errors=0,
 +    )
++
 +    tool_result = next(event for event in events if event.type == "tool_result")
 +    assert tool_result.tool_metadata == metadata
++
++
  @pytest.mark.asyncio
  async def test_tool_batch_runner_verifies_with_context_services(temp_dir: Path) -> None:
      verification_calls: list[str] = []