"""Animated activity indicator for the TUI status panel.
A rotating braille-spinner + phase label that runs on its own timer while the
agent is working. Because the timer is independent of the WebSocket event flow,
the spinner keeps moving during silent gaps (a long-running tool, the pause
before the first token, sub-agent reasoning) — so "still spinning" means the
agent is alive, and "frozen" means it hung. That is the whole point: make
"hung or just thinking" answerable at a glance.
"""
from __future__ import annotations
from rich.text import Text
from textual.widgets import Static
from textual.timer import Timer
from clients.terminal.tui.themes import get_active_theme
_FRAMES = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
_TICK_SECONDS = 0.1
class ActivityIndicator(Static):
"""Spinner + phase label, animated while started."""
def __init__(self) -> None:
super().__init__("")
self._frame = 0
self._label = ""
self._active = False
self._timer: Timer | None = None
def start(self, label: str) -> None:
"""Begin animating with ``label`` as the current phase."""
self._label = label
self._active = True
self._render_frame()
if self._timer is None:
self._timer = self.set_interval(_TICK_SECONDS, self._tick)
def set_label(self, label: str) -> None:
"""Update the phase label without restarting the animation."""
if not self._active:
return
self._label = label
self._render_frame()
def stop(self) -> None:
"""Stop animating and clear the line."""
self._active = False
self._label = ""
if self._timer is not None:
self._timer.stop()
self._timer = None
if self.is_mounted:
self.update("")
def _tick(self) -> None:
if not self._active:
return
self._frame = (self._frame + 1) % len(_FRAMES)
self._render_frame()
def _render_frame(self) -> None:
if not self.is_mounted:
return
theme = get_active_theme()
glyph = _FRAMES[self._frame]
self.update(Text(f"{glyph} {self._label}", style=theme.status_online.hex))