Index

@abstractmethod
async def on_run_started(self, thread_id: str, run_id: str) -> None:
    """Handle run started event.

    Args:
        thread_id: Unique identifier for the conversation thread.
        run_id: Unique identifier for this agent run.
    """

@abstractmethod
async def on_text_start(self, message_id: str, role: str) -> None:
    """Handle start of a text message.

    Args:
        message_id: Unique identifier for the message.
        role: Message role (usually "assistant").
    """

@abstractmethod
async def on_text_content(self, message_id: str, delta: str) -> None:
    """Handle streaming text content.

    Args:
        message_id: Identifier of the message being streamed.
        delta: New text chunk to append.
    """

@abstractmethod
async def on_text_end(self, message_id: str) -> None:
    """Handle end of a text message.

    Args:
        message_id: Identifier of the completed message.
    """

@abstractmethod
async def on_tool_start(self, tool_call_id: str, tool_name: str) -> None:
    """Handle start of tool execution.

    Args:
        tool_call_id: Unique identifier for this tool call.
        tool_name: Name of the tool being called.
    """

@abstractmethod
async def on_tool_args(self, tool_call_id: str, args_delta: str) -> None:
    """Handle streaming tool arguments.

    Args:
        tool_call_id: Identifier of the tool call.
        args_delta: JSON string chunk of tool arguments.
    """

@abstractmethod
async def on_tool_result(self, tool_call_id: str, result: Any) -> None:
    """Handle tool execution result.

    Args:
        tool_call_id: Identifier of the completed tool call.
        result: Result returned by the tool.
    """

@abstractmethod
async def on_tool_end(self, tool_call_id: str) -> None:
    """Handle end of tool execution.

    Args:
        tool_call_id: Identifier of the completed tool call.
    """

@abstractmethod
async def on_state_snapshot(self, snapshot: dict[str, Any]) -> None:
    """Handle state snapshot event.

    Args:
        snapshot: Complete state snapshot dictionary.
    """

@abstractmethod
async def on_run_finished(self, thread_id: str, run_id: str) -> None:
    """Handle run finished event.

    Args:
        thread_id: Identifier of the conversation thread.
        run_id: Identifier of the completed run.
    """

@abstractmethod
async def on_error(self, message: str, code: str | None = None) -> None:
    """Handle error event.

    Args:
        message: Error message description.
        code: Optional error code.
    """

async def start(self) -> None:
    """Initialize the consumer (optional).

    Override to perform setup before events start streaming.
    Called before the first event is received.
    """

async def stop(self) -> None:
    """Cleanup the consumer (optional).

    Override to perform cleanup after events stop streaming.
    Called after the last event is received or on error.
    """

async def start(self) -> None:
    """Initialize the consumer (optional).

    Override to perform setup before events start streaming.
    Called before the first event is received.
    """

async def stop(self) -> None:
    """Cleanup the consumer (optional).

    Override to perform cleanup after events stop streaming.
    Called after the last event is received or on error.
    """

async def on_run_started(self, thread_id: str, run_id: str) -> None:
    """No output."""

async def on_text_start(self, message_id: str, role: str) -> None:
    """No output."""

async def on_text_content(self, message_id: str, delta: str) -> None:
    """No output."""

async def on_text_end(self, message_id: str) -> None:
    """No output."""

async def on_tool_start(self, tool_call_id: str, tool_name: str) -> None:
    """No output."""

async def on_tool_args(self, tool_call_id: str, args_delta: str) -> None:
    """No output."""

async def on_tool_result(self, tool_call_id: str, result: Any) -> None:
    """No output."""

async def on_tool_end(self, tool_call_id: str) -> None:
    """No output."""

async def on_state_snapshot(self, snapshot: dict[str, Any]) -> None:
    """No output."""

async def on_run_finished(self, thread_id: str, run_id: str) -> None:
    """No output."""

async def on_error(self, message: str, code: str | None = None) -> None:
    """Print error to stderr."""
    error_text = f"[{code}] {message}" if code else message
    self._stderr.print(f"Error: {error_text}")

async def start(self) -> None:
    """Start the dashboard display."""
    self.state.reset()
    self._live = Live(
        self._build_display(),
        console=self.console,
        refresh_per_second=2,
        transient=True,
        vertical_overflow="ellipsis",
    )
    bind_live_renderable(self._live, self._build_display)
    self._live.start()

async def stop(self) -> None:
    """Stop the dashboard and print final summary."""
    if self._live:
        self._live.stop()
        self._live = None

    elapsed = self.state.get_elapsed_seconds()
    total = len(self.state.completed_tools)
    errors = self.state.tool_errors
    self.console.print(
        f"[dim]  {total} tools in {self._format_time(elapsed)}  ✓ {total - errors}  ✗ {errors}[/dim]",
    )

async def on_run_started(self, thread_id: str, run_id: str) -> None:
    """Handle run started."""
    self.state.thread_id = thread_id
    self.state.run_id = run_id
    self.state.start_time = time.monotonic()

async def on_text_start(self, message_id: str, role: str) -> None:
    """Handle text start — no display action."""
    self.state.active_message_id = message_id

async def on_text_content(self, message_id: str, delta: str) -> None:
    """Handle text delta — silently accumulate (not displayed)."""
    self.state.text_buffer += delta

async def on_text_end(self, message_id: str) -> None:
    """Handle text end — increment turn counter."""
    self.state.active_message_id = None
    self.state.turn_count += 1
    self.state.text_buffer = ""

async def on_tool_start(self, tool_call_id: str, tool_name: str) -> None:
    """Handle tool start — show in active spinner."""
    self.state.active_tool = ToolCallState(
        tool_call_id=tool_call_id,
        tool_name=tool_name,
    )
    self.state.tool_start_time = time.monotonic()
    self._detect_phase(tool_name)

async def on_tool_args(self, tool_call_id: str, args_delta: str) -> None:
    """Handle tool args — silently accumulate."""
    if self.state.active_tool and self.state.active_tool.tool_call_id == tool_call_id:
        self.state.active_tool.args_buffer += args_delta

async def on_tool_result(self, tool_call_id: str, result: Any) -> None:
    """Handle tool result — check for errors and extract discoveries."""
    if self.state.active_tool and self.state.active_tool.tool_call_id == tool_call_id:
        self.state.active_tool.result = result
        if (isinstance(result, str) and "error" in result.lower()) or (
            isinstance(result, dict) and result.get("error")
        ):
            self.state.tool_errors += 1
        discovery = self._extract_discovery(self.state.active_tool.tool_name, result)
        if discovery:
            self.state.add_discovery(discovery)

async def on_tool_end(self, tool_call_id: str) -> None:
    """Handle tool end — move to completed list."""
    if self.state.active_tool and self.state.active_tool.tool_call_id == tool_call_id:
        self.state.complete_active_tool()

async def on_state_snapshot(self, snapshot: dict[str, Any]) -> None:
    """Handle state snapshot."""
    self.state.state_snapshot = snapshot
    if "phase" in snapshot:
        self.state.current_phase = str(snapshot["phase"])

async def on_run_finished(self, thread_id: str, run_id: str) -> None:
    """Handle run finished."""
    self.state.completed = True

async def on_error(self, message: str, code: str | None = None) -> None:
    """Handle error."""
    error_text = f"[{code}] {message}" if code else message
    self.state.error = error_text

def clear_text_buffer(self) -> str:
    """Clear and return the text buffer.

    Returns:
        The text buffer contents before clearing.
    """
    text = self.text_buffer
    self.text_buffer = ""
    return text

def complete_active_tool(self, result: Any = None) -> None:
    """Mark the active tool as completed and move to history.

    Args:
        result: Optional result to store with the tool call.
    """
    if self.active_tool:
        self.active_tool.status = "completed"
        self.active_tool.result = result
        self.completed_tools.append(self.active_tool)
        self.active_tool = None

def record_tool(self, *, tool_id: str, tool_name: str, status: str, result: str) -> None:
    """Record a tool call's start or completion for the dashboard.

    Called by ``ToolDisplayHook`` on both the before- and after-tool-call
    events. A ``running`` status opens a new active tool (and advances the
    phase resolved from the tool name); any other status closes it and moves
    it to history, counting errors.

    Args:
        tool_id: The tool-use identifier from the event.
        tool_name: Name of the tool being executed.
        status: ``"running"`` to open the call, ``"completed"``/``"error"`` to close it.
        result: Truncated result string to store when closing the call.
    """
    if status == "running":
        self.active_tool = ToolCallState(tool_call_id=tool_id, tool_name=tool_name, status="running")
        self.tool_start_time = time.monotonic()
        phase = self.phase_spec.resolve(tool_name)
        if phase is not None:
            self.advance_phase(phase)
        return

    if self.active_tool is not None:
        self.active_tool.status = status
        self.active_tool.result = result
        self.completed_tools.append(self.active_tool)
        self.active_tool = None
    else:
        self.completed_tools.append(
            ToolCallState(tool_call_id=tool_id, tool_name=tool_name, status=status, result=result)
        )
    if status == "error":
        self.tool_errors += 1
    self.tool_start_time = None

def get_recent_tools(self, count: int = 5) -> list[ToolCallState]:
    """Get the most recent completed tools.

    Args:
        count: Maximum number of tools to return.

    Returns:
        List of recently completed tool states.
    """
    return self.completed_tools[-count:]

def get_tool_stats(self) -> dict[str, int]:
    """Get tool call counts grouped by prefix.

    Groups tools by their name prefix (e.g., ``lsp_*``, ``git_*``) for
    display in the TUI dashboard.

    Returns:
        Dictionary mapping tool prefixes to call counts.
    """
    stats: dict[str, int] = {}
    for tool in self.completed_tools:
        parts = tool.tool_name.split("_")
        prefix = parts[0] + "_*" if len(parts) > 1 else tool.tool_name
        stats[prefix] = stats.get(prefix, 0) + 1
    return stats

def get_elapsed_seconds(self) -> float:
    """Get elapsed time since the run started.

    Returns:
        Elapsed seconds, or 0.0 if not started.
    """
    if self.start_time is None:
        return 0.0
    return time.monotonic() - self.start_time

def get_tool_elapsed_seconds(self) -> float:
    """Get elapsed time for the current tool.

    Returns:
        Elapsed seconds for the active tool, or 0.0 if no active tool.
    """
    if self.tool_start_time is None:
        return 0.0
    return time.monotonic() - self.tool_start_time

def get_success_count(self) -> int:
    """Get count of successful tool calls.

    Returns:
        Number of completed tools minus errors.
    """
    return len(self.completed_tools) - self.tool_errors

def advance_phase(self, phase: int) -> None:
    """Advance to a new phase if it is higher than the current one.

    Phase name and description are resolved from the injected
    ``phase_spec``.

    Args:
        phase: 1-based phase index to advance to.
    """
    # Don't regress to a lower phase
    if self.phases and phase <= self.phases[-1].phase:
        return

    now = time.monotonic()
    if self.phases and not self.phases[-1].is_complete:
        self.phases[-1].completed_at = now

    self.phases.append(
        PhaseState(
            phase=phase,
            name=self.phase_spec.name(phase),
            description=self.phase_spec.description(phase),
            started_at=now,
        ),
    )
    self.current_phase_index = len(self.phases) - 1

def add_discovery(self, event: DiscoveryEvent) -> None:
    """Add a discovery event, evicting oldest if at capacity.

    Args:
        event: DiscoveryEvent to add.
    """
    self.discoveries.append(event)
    while len(self.discoveries) > self.max_discoveries:
        self.discoveries.pop(0)

def start_team(self, team_id: str, mandate: str, agent_count: int = 0) -> None:
    """Register a newly dispatched team.

    Args:
        team_id: Unique team identifier.
        mandate: One-line description of the team's focus.
        agent_count: Number of agents in the team.
    """
    self.teams.append(
        TeamDispatchState(
            team_id=team_id,
            mandate=mandate,
            agent_count=agent_count,
            status="running",
            started_at=time.monotonic(),
        ),
    )

def complete_team(self, team_id: str, status: str = "done") -> None:
    """Mark a team as completed.

    Args:
        team_id: Team identifier to mark complete.
        status: Final status ("done" or "error").
    """
    for team in self.teams:
        if team.team_id == team_id:
            team.status = status
            if team.started_at is not None:
                team.duration_seconds = time.monotonic() - team.started_at
            break

def reset(self) -> None:
    """Reset state for a new run (preserves the injected ``phase_spec``)."""
    self.current_phase = ""
    self.text_buffer = ""
    self.active_message_id = None
    self.active_tool = None
    self.completed_tools = []
    self.state_snapshot = {}
    self.error = None
    self.completed = False
    self.start_time = None
    self.turn_count = 0
    self.tool_errors = 0
    self.tool_start_time = None
    self.phases = []
    self.current_phase_index = -1
    self.discoveries = []
    self.teams = []