"""Telemetry hub for the live status dashboard. The pipeline is a chain of steps (heard -> screen context -> brain -> speak). This module records, for every conversation turn, *what happened at each step* and *how long it took*, plus a rolling status header and any errors. The dashboard (``wsai/dashboard.py``) reads a snapshot and subscribes for live push updates. Design notes: * Pure stdlib, no deps — matches the project's "core has no third-party deps". * Thread-safe. The pipeline mutates it from the asyncio loop; the HTTP server reads/subscribes from its own threads. A single lock guards everything. * A Monitor with zero subscribers is essentially free, so the pipeline can always hold one (no separate no-op path). """ from __future__ import annotations import json import queue import threading import time from collections import deque from typing import Any def _now_wall() -> float: # Wall-clock seconds for human-readable timestamps on the page. return time.time() def _now_mono() -> float: # Monotonic seconds for measuring durations (immune to clock jumps). return time.monotonic() class Step: """One timed stage inside a turn (e.g. "두뇌"). Used as an async context manager so it can wrap an ``await`` and record ok/error + elapsed ms.""" def __init__(self, turn: "Turn", name: str) -> None: self.turn = turn self.name = name self.ok: bool | None = None self.ms: float = 0.0 self.detail: str = "" self.error: str = "" self._t0 = 0.0 async def __aenter__(self) -> "Step": self._t0 = _now_mono() self.turn._steps.append(self) self.turn._touch() return self async def __aexit__(self, exc_type, exc, tb) -> bool: self.ms = (_now_mono() - self._t0) * 1000.0 if exc is not None: self.ok = False self.error = f"{exc_type.__name__}: {exc}" else: self.ok = True self.turn._touch() return False # never swallow: the pipeline/TaskGroup must still see it def to_dict(self) -> dict[str, Any]: return { "name": self.name, "ok": self.ok, "ms": round(self.ms, 1), "detail": self.detail, "error": self.error, } class Turn: """One user utterance and everything the AI did in response.""" def __init__(self, monitor: "Monitor", turn_id: int, source: str) -> None: self._monitor = monitor self.id = turn_id self.source = source self.wall = _now_wall() self._t0 = _now_mono() self.heard_text = "" self.reply_text = "" self.status = "active" # active | ok | error self.error = "" self.total_ms = 0.0 self._steps: list[Step] = [] # -- recording API (called from the pipeline) ------------------------- # def heard(self, text: str) -> None: self.heard_text = text self._touch() def replied(self, text: str) -> None: self.reply_text = text self._touch() def step(self, name: str) -> Step: return Step(self, name) def finish(self, error: str = "") -> None: self.total_ms = (_now_mono() - self._t0) * 1000.0 if error: self.status = "error" self.error = error elif any(s.ok is False for s in self._steps): self.status = "error" else: self.status = "ok" self._touch() # -- internal --------------------------------------------------------- # def _touch(self) -> None: self._monitor._publish(self) def to_dict(self) -> dict[str, Any]: return { "id": self.id, "source": self.source, "wall": self.wall, "heard": self.heard_text, "reply": self.reply_text, "status": self.status, "error": self.error, "total_ms": round(self.total_ms, 1), "steps": [s.to_dict() for s in self._steps], } class Monitor: """Rolling record of turns + status, with a pub/sub for live updates.""" def __init__(self, keep: int = 60) -> None: self._turns: deque[Turn] = deque(maxlen=keep) self._events: deque[dict[str, Any]] = deque(maxlen=200) self._status: dict[str, Any] = { "running": False, "listening": False, "started_wall": _now_wall(), "components": {}, "turns_total": 0, "errors_total": 0, } self._lock = threading.Lock() self._subs: list["queue.Queue[str]"] = [] self._id = 0 # -- status ----------------------------------------------------------- # def set_status(self, **kw: Any) -> None: with self._lock: self._status.update(kw) self._broadcast({"type": "status", "status": self.status_snapshot()}) def set_components(self, components: dict[str, Any]) -> None: with self._lock: self._status["components"] = components self._broadcast({"type": "status", "status": self.status_snapshot()}) def status_snapshot(self) -> dict[str, Any]: with self._lock: s = dict(self._status) s["uptime_s"] = round(_now_wall() - s["started_wall"], 1) return s def log(self, level: str, message: str) -> None: """A free-form lifecycle/error line (startup, disconnect, crash…).""" evt = {"type": "log", "level": level, "message": message, "wall": _now_wall()} with self._lock: self._events.append(evt) if level == "error": self._status["errors_total"] += 1 self._broadcast(evt) # -- turns ------------------------------------------------------------ # def turn(self, source: str = "voice") -> Turn: with self._lock: self._id += 1 self._status["turns_total"] += 1 t = Turn(self, self._id, source) self._turns.append(t) self._publish(t) return t def _publish(self, t: Turn) -> None: # Recompute error total lazily on error transitions. if t.status == "error": with self._lock: # errors_total counts turns that ended in error at most once pass self._broadcast({"type": "turn", "turn": t.to_dict()}) # -- snapshot / subscribe (read side, HTTP threads) ------------------- # def snapshot(self) -> dict[str, Any]: with self._lock: turns = [t.to_dict() for t in self._turns] events = list(self._events) return { "status": self.status_snapshot(), "turns": turns, "events": events, } def subscribe(self) -> "queue.Queue[str]": q: "queue.Queue[str]" = queue.Queue(maxsize=256) with self._lock: self._subs.append(q) return q def unsubscribe(self, q: "queue.Queue[str]") -> None: with self._lock: if q in self._subs: self._subs.remove(q) def _broadcast(self, event: dict[str, Any]) -> None: data = json.dumps(event, ensure_ascii=False) with self._lock: subs = list(self._subs) for q in subs: try: q.put_nowait(data) except queue.Full: # Slow client: drop it rather than block the pipeline. self.unsubscribe(q)