`f0c4a17`

Surface definition-of-done state in CLI and TUI

Authored by

espadonne 1 month ago

SHA: f0c4a1751c6bf030a43bfe87f13bfd84f1156fc1
Parents: 7d46142
Tree: 438a411

9 changed files

Status	File	+	-
M	`src/loader/cli/__init__.py`	9	1
M	`src/loader/cli/main.py`	22	4
A	`src/loader/cli/rendering.py`	15	0
M	`src/loader/ui/__init__.py`	9	1
M	`src/loader/ui/adapter.py`	32	8
M	`src/loader/ui/app.py`	34	12
A	`src/loader/ui/status_helpers.py`	34	0
M	`src/loader/ui/widgets/status_line.py`	42	0
A	`tests/test_status_surfaces.py`	32	0

src/loader/cli/__init__.pymodified

  """CLI interface."""
 -from .main import main
 +from typing import Any
++
++
 +def main(*args: Any, **kwargs: Any):
 +    """Lazy wrapper for the Click entrypoint."""
 +    from .main import main as click_main
++
 +    return click_main(*args, **kwargs)
++
  __all__ = ["main"]

src/loader/cli/main.pymodified

  from rich.prompt import Confirm
  from rich.table import Table
 +from .rendering import format_dod_status
++
  console = Console()
                  console.print(f" [dim]({elapsed:.1f}s)[/dim]")
                  thinking_start = None
              args_str = _format_tool_args(event.tool_args)
 -            console.print(f"[cyan]> {event.tool_name}[/cyan]({args_str})")
 +            tool_label = (
 +                f"verify {event.tool_name}"
 +                if event.phase == "verification"
 +                else event.tool_name
 +            )
 +            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]"
 -            console.print(Panel(preview, border_style="dim"))
 +            border_style = "magenta" if event.phase == "verification" else "dim"
 +            console.print(Panel(preview, border_style=border_style))
 +        elif event.type == "dod_status":
 +            console.print(f"[dim]{format_dod_status(event)}[/dim]")
          elif event.type == "recovery":
              console.print(f"[yellow]Recovering from error ({event.recovery_attempt}/3)...[/yellow]")
          elif event.type == "error":
                      console.print()  # New line after any streamed content
                      streaming_started = False
                  args_str = _format_tool_args(event.tool_args)
 -                console.print(f"[cyan]> {event.tool_name}[/cyan]({args_str})")
 +                tool_label = (
 +                    f"verify {event.tool_name}"
 +                    if event.phase == "verification"
 +                    else event.tool_name
 +                )
 +                console.print(f"[cyan]> {tool_label}[/cyan]({args_str})")
              elif event.type == "tool_result":
                  # Show compact result
                  lines = event.content.splitlines()
                      preview = event.content
                  else:
                      preview = "\n".join(lines[:3]) + f"\n[dim]... ({len(lines) - 3} more lines)[/dim]"
 -                console.print(f"[dim]{preview}[/dim]")
 +                style = "magenta" if event.phase == "verification" else "dim"
 +                console.print(f"[{style}]{preview}[/{style}]")
 +            elif event.type == "dod_status":
 +                console.print(f"\n[dim]{format_dod_status(event)}[/dim]")
              elif event.type == "recovery":
                  console.print(f"\n[yellow]Recovering from error ({event.recovery_attempt}/3)...[/yellow]")
              elif event.type == "error":

src/loader/cli/rendering.pyadded

 +"""Lightweight CLI rendering helpers."""
++
 +from __future__ import annotations
++
 +from ..runtime.events import AgentEvent
++
++
 +def format_dod_status(event: AgentEvent) -> str:
 +    """Format a definition-of-done status event for CLI output."""
 +    parts = [f"DoD: {event.dod_status or 'unknown'}"]
 +    if event.pending_items_count is not None:
 +        parts.append(f"{event.pending_items_count} pending")
 +    if event.last_verification_result:
 +        parts.append(f"last verify: {event.last_verification_result}")
 +    return " | ".join(parts)

src/loader/ui/__init__.pymodified

  """Textual TUI for Loader."""
 -from .app import LoaderApp
 +from typing import Any
++
++
 +def LoaderApp(*args: Any, **kwargs: Any):
 +    """Lazy wrapper for the Textual application."""
 +    from .app import LoaderApp as _LoaderApp
++
 +    return _LoaderApp(*args, **kwargs)
++
  __all__ = ["LoaderApp"]

src/loader/ui/adapter.pymodified

  if TYPE_CHECKING:
      from ..agent.reasoning import (
 -        TaskDecomposition,
 -        Subtask,
 -        SelfCritique,
 -        ConfidenceAssessment,
          ActionVerification,
 -        TaskCompletionCheck,
 -        RollbackPlan,
 +        ConfidenceAssessment,
          RollbackAction,
 +        RollbackPlan,
 +        SelfCritique,
 +        Subtask,
 +        TaskCompletionCheck,
 +        TaskDecomposition,
+     )
      tool_name: str
      tool_args: dict
 +    phase: str | None = None
  @dataclass
      tool_name: str
      content: str
      is_error: bool = False
 +    phase: str | None = None
      # For edit tool diffs
      old_string: str | None = None
      new_string: str | None = None
      rollback_plan: "RollbackPlan | None" = None
 +@dataclass
 +class DefinitionOfDoneUpdated(Message):
 +    """Definition-of-done status changed."""
++
 +    content: str
 +    dod_status: str
 +    pending_items_count: int = 0
 +    last_verification_result: str | None = None
++
++
  class EventAdapter:
      """Adapts Agent callback events to Textual messages."""
                      ToolCallStarted(
                          tool_name=tool_name,
                          tool_args=tool_args,
 +                        phase=event.phase,
+                     )
+                 )
+                         )
                          self._debug_log(f"  edit extracted: old={bool(old_string)} ({len(old_string) if old_string else 0} chars), new={bool(new_string)} ({len(new_string) if new_string else 0} chars), path={file_path}")
                      else:
 -                        self._debug_log(f"  edit: tool_args was empty!")
 +                        self._debug_log("  edit: tool_args was empty!")
                  elif tool_name == "write":
                      # For writes, content is the new file content
                      # Try multiple key names that models might use
+                         )
                          self._debug_log(f"  write extracted: new={bool(new_string)} ({len(new_string) if new_string else 0} chars), path={file_path}")
                      else:
 -                        self._debug_log(f"  write: tool_args was empty!")
 +                        self._debug_log("  write: tool_args was empty!")
                  self.app.post_message(
                      ToolCallCompleted(
                          tool_name=tool_name,
                          content=event.content,
                          is_error=event.is_error,
 +                        phase=event.phase,
                          old_string=old_string,
                          new_string=new_string,
                          file_path=file_path,
                      content=event.content,
                      rollback_plan=event.rollback_plan,
                  ))
++
 +            case "dod_status":
 +                self.app.post_message(
 +                    DefinitionOfDoneUpdated(
 +                        content=event.content,
 +                        dod_status=event.dod_status or "",
 +                        pending_items_count=event.pending_items_count or 0,
 +                        last_verification_result=event.last_verification_result,
 +                    )
 +                )

src/loader/ui/app.pymodified

  from pathlib import Path
  from rich.markup import escape
+-
 +from textual import work
  from textual.app import App, ComposeResult
  from textual.binding import Binding
  from textual.containers import Container, ScrollableContainer
  from textual.reactive import reactive
  from textual.widgets import Footer, Input, Static
 -from textual import work
  from textual.worker import Worker, get_current_worker
  from ..agent.loop import Agent, AgentEvent
      ConfidenceAssessed,
      CritiquePerformed,
      DecompositionCreated,
 +    DefinitionOfDoneUpdated,
      ErrorOccurred,
      EventAdapter,
      PlanCreated,
      ToolCallStarted,
      VerificationPerformed,
+ )
 -from .widgets import ApprovalBar, ConfirmationModal, DiffWidget, InputArea, StatusLine, StreamingText, ToolCallWidget
 +from .widgets import (
 +    ApprovalBar,
 +    DiffWidget,
 +    InputArea,
 +    StatusLine,
 +    StreamingText,
 +    ToolCallWidget,
 +)
  class LoaderApp(App):
          # Show generating status immediately (before async work starts)
          self.is_generating = True
          self._start_timer()
 -        self.query_one(StatusLine).set_generating(True)
 +        status = self.query_one(StatusLine)
 +        status.clear_definition_of_done()
 +        status.set_generating(True)
          # Start agent task
          self.run_agent(user_input)
          details: str,
      ) -> bool:
          """Show approval bar and wait for user response."""
 -        import threading
          # Create a future to wait on
          loop = asyncio.get_event_loop()
              try:
                  approval_bar.show_approval(tool_name, message, details)
                  with open("/tmp/loader_debug.log", "a") as f:
 -                    f.write(f"[approval] Bar shown, waiting for user input\n")
 +                    f.write("[approval] Bar shown, waiting for user input\n")
              except Exception as e:
                  with open("/tmp/loader_debug.log", "a") as f:
                      f.write(f"[approval] Error showing bar: {e}\n")
              except Exception:
                  pass
              return result
 -        except asyncio.TimeoutError:
 +        except TimeoutError:
              try:
                  with open("/tmp/loader_debug.log", "a") as f:
 -                    f.write(f"[approval] Timeout waiting for user\n")
 +                    f.write("[approval] Timeout waiting for user\n")
              except Exception:
                  pass
              return False
          """Handle approval from the bar."""
          try:
              with open("/tmp/loader_debug.log", "a") as f:
 -                f.write(f"[approval] Approved handler called\n")
 +                f.write("[approval] Approved handler called\n")
          except Exception:
              pass
          if self._pending_confirmation and not self._pending_confirmation.done():
          """Handle rejection from the bar."""
          try:
              with open("/tmp/loader_debug.log", "a") as f:
 -                f.write(f"[approval] Rejected handler called\n")
 +                f.write("[approval] Rejected handler called\n")
          except Exception:
              pass
          if self._pending_confirmation and not self._pending_confirmation.done():
          """Handle edit request - put command in input for editing."""
          try:
              with open("/tmp/loader_debug.log", "a") as f:
 -                f.write(f"[approval] Edit handler called\n")
 +                f.write("[approval] Edit handler called\n")
          except Exception:
              pass
          if self._pending_confirmation and not self._pending_confirmation.done():
      async def run_agent(self, user_input: str) -> str:
          """Run the agent asynchronously."""
          import asyncio
++
          import httpx
          worker = get_current_worker()
          # Create tool widget
          widget = ToolCallWidget(
 -            tool_name=message.tool_name,
 +            tool_name=(
 +                f"verify {message.tool_name}"
 +                if message.phase == "verification"
 +                else message.tool_name
 +            ),
              tool_args=message.tool_args,
+         )
          msg_area.mount(widget)
          if not self._streamed_content and message.content.strip():
              self._add_message(message.content)
 +    def on_definition_of_done_updated(self, message: DefinitionOfDoneUpdated) -> None:
 +        """Handle definition-of-done status changes."""
 +        self.query_one(StatusLine).update_definition_of_done(
 +            message.dod_status,
 +            message.pending_items_count,
 +            message.last_verification_result,
 +        )
++
      def on_steering_received(self, message: SteeringReceived) -> None:
          """Handle steering message being processed by agent."""
          # Don't display anything - auto-steering is internal
          msg_area = self.query_one("#message-area", ScrollableContainer)
          msg_area.remove_children()
          self.agent.clear_history()
 +        self.query_one(StatusLine).clear_definition_of_done()
          self._add_message("[dim]Conversation cleared.[/dim]")
      def action_cancel(self) -> None:

src/loader/ui/status_helpers.pyadded

 +"""Formatting helpers for user-visible runtime status."""
++
 +from __future__ import annotations
++
++
 +def format_definition_of_done_parts(
 +    status: str,
 +    pending_items_count: int,
 +    last_verification_result: str,
 +) -> list[str]:
 +    """Format definition-of-done state for the status line."""
 +    if not status:
 +        return []
++
 +    dod_color = {
 +        "draft": "cyan",
 +        "in_progress": "cyan",
 +        "verifying": "yellow",
 +        "fixing": "magenta",
 +        "done": "green",
 +        "failed": "red",
 +    }.get(status, "white")
 +    parts = [f"[{dod_color}]DoD: {status}[/{dod_color}]"]
 +    parts.append(f"[dim]{pending_items_count} pending[/dim]")
++
 +    if last_verification_result:
 +        verify_color = "green" if last_verification_result == "passed" else "red"
 +        if last_verification_result == "skipped":
 +            verify_color = "dim"
 +        parts.append(
 +            f"[{verify_color}]verify {last_verification_result}[/{verify_color}]"
 +        )
++
 +    return parts

src/loader/ui/widgets/status_line.pymodified

  from textual.reactive import reactive
  from textual.widgets import Static
 +from ..status_helpers import format_definition_of_done_parts
++
  class StatusLine(Static):
      """Status bar showing model info, activity, elapsed time, and tokens."""
      activity: reactive[str] = reactive("")
      elapsed: reactive[float] = reactive(0.0)
      tokens: reactive[int] = reactive(0)
 +    dod_status: reactive[str] = reactive("")
 +    pending_items_count: reactive[int] = reactive(0)
 +    last_verification_result: reactive[str] = reactive("")
      def render(self) -> str:
          """Render the status line."""
          if self.tokens > 0:
              parts.append(f"[dim]{self.tokens} tokens[/dim]")
 +        parts.extend(
 +            format_definition_of_done_parts(
 +                self.dod_status,
 +                self.pending_items_count,
 +                self.last_verification_result,
 +            )
 +        )
++
          # Model info
          if self.model:
              parts.append(f"[blue]{self.model}[/blue]")
          """React to token count changes."""
          self.refresh()
 +    def watch_dod_status(self, dod_status: str) -> None:
 +        """React to DoD status changes."""
 +        self.refresh()
++
 +    def watch_pending_items_count(self, pending_items_count: int) -> None:
 +        """React to DoD pending item changes."""
 +        self.refresh()
++
 +    def watch_last_verification_result(self, last_verification_result: str) -> None:
 +        """React to verification result changes."""
 +        self.refresh()
++
      def set_generating(self, is_generating: bool) -> None:
          """Set generating state."""
          if is_generating:
      def update_tokens(self, tokens: int) -> None:
          """Update token count."""
          self.tokens = tokens
++
 +    def update_definition_of_done(
 +        self,
 +        status: str,
 +        pending_items_count: int,
 +        last_verification_result: str | None,
 +    ) -> None:
 +        """Update definition-of-done status."""
 +        self.dod_status = status
 +        self.pending_items_count = pending_items_count
 +        self.last_verification_result = last_verification_result or ""
++
 +    def clear_definition_of_done(self) -> None:
 +        """Clear definition-of-done status."""
 +        self.dod_status = ""
 +        self.pending_items_count = 0
 +        self.last_verification_result = ""

tests/test_status_surfaces.pyadded

 +"""Tests for user-visible definition-of-done status formatting."""
++
 +from loader.cli.rendering import format_dod_status
 +from loader.runtime.events import AgentEvent
 +from loader.ui.status_helpers import format_definition_of_done_parts
++
++
 +def test_status_helper_formats_definition_of_done_parts() -> None:
 +    parts = format_definition_of_done_parts("verifying", 1, "failed")
++
 +    assert parts == [
 +        "[yellow]DoD: verifying[/yellow]",
 +        "[dim]1 pending[/dim]",
 +        "[red]verify failed[/red]",
 +    ]
++
++
 +def test_status_helper_omits_definition_of_done_when_absent() -> None:
 +    assert format_definition_of_done_parts("", 0, "") == []
++
++
 +def test_cli_dod_status_format_includes_pending_and_verification() -> None:
 +    event = AgentEvent(
 +        type="dod_status",
 +        dod_status="fixing",
 +        pending_items_count=2,
 +        last_verification_result="failed",
 +    )
++
 +    formatted = format_dod_status(event)
++
 +    assert formatted == "DoD: fixing | 2 pending | last verify: failed"