6b0755e1ff
Add a stdlib-only observability site so you can open a browser and watch, step by step: whether it is listening, what it heard, what the brain thought and answered, how long each stage took, and whether anything errored. - wsai/monitor.py: thread-safe telemetry hub (per-turn timed steps, status header, error log) with a pub/sub for live push. - wsai/dashboard.py: stdlib http.server serving a self-contained page plus an SSE (/events) live stream; /api/state snapshot fallback. - Pipeline emits step-by-step turn telemetry (화면 맥락 → 두뇌 → 응답) and listening/running status; optional monitor, so existing paths are untouched. - `python -m wsai --dashboard` starts the site (0.0.0.0:8787, WSAI_DASHBOARD_PORT) and loops the mock voice demo so there is always live activity to watch. - Tests cover turn recording, per-step timing, error marking, and live push. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
227 lines
7.3 KiB
Python
227 lines
7.3 KiB
Python
"""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)
|