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",
    )
    # Rich auto-refresh calls get_renderable — point it at our builder
    # so the dashboard always shows fresh state (timer, tool elapsed, etc.)
    self._live.get_renderable = self._build_display  # type: ignore[assignment]
    self._live.start()

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

    # Print a static final summary after Live is gone
    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
        # Extract discovery events
        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 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_*, code_graph_*) for
    display in the TUI dashboard.

    Returns:
        Dictionary mapping tool prefixes to call counts.
    """
    stats: dict[str, int] = {}
    for tool in self.completed_tools:
        # Group by first part of tool name (before underscore)
        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 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 current tool.

    Returns:
        Elapsed seconds for 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: Any) -> None:
    """Advance to a new phase if it's higher than the current one.

    Args:
        phase: AnalysisPhase enum value to advance to.
    """
    from .phases import PHASE_DESCRIPTIONS, PHASE_NAMES, PhaseState

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

    # Complete the previous phase
    now = time.monotonic()
    if self.phases and not self.phases[-1].is_complete:
        self.phases[-1].completed_at = now

    # Start the new phase
    self.phases.append(
        PhaseState(
            phase=phase,
            name=PHASE_NAMES.get(phase, f"Phase {phase}"),
            description=PHASE_DESCRIPTIONS.get(phase, ""),
            started_at=now,
        ),
    )
    self.current_phase_index = len(self.phases) - 1

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

    Args:
        event: DiscoveryEvent to add.
    """
    self.discoveries.append(event)
    # Capped list — evict oldest when over capacity
    while len(self.discoveries) > self.max_discoveries:
        self.discoveries.pop(0)

def reset(self) -> None:
    """Reset state for a new run."""
    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
    # Reset metrics
    self.start_time = None
    self.turn_count = 0
    self.tool_errors = 0
    self.tool_start_time = None
    # Reset phase tracking (v7)
    self.phases = []
    self.current_phase_index = -1
    self.discoveries = []