f1f059a018
ClaudeBrain now returns per-reply token usage (Reply.usage from the API response), the dashboard accumulates it (monitor.add_claude_usage), and the header shows a "클로드 토큰" stat (input+output total, with a tooltip breaking down input/output tokens and request count). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
298 lines
11 KiB
Python
298 lines
11 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.speaker = "" # who spoke (Discord display name), when known
|
|
self.guild = "" # server name, when known (for 서버별 필터)
|
|
self.channel = "" # voice channel name, when known (for 채널별 필터)
|
|
self.wall = _now_wall()
|
|
self._t0 = _now_mono()
|
|
self.heard_text = ""
|
|
self.thought_text = ""
|
|
self.reply_text = ""
|
|
self.status = "active" # active | ok | error
|
|
self.error = ""
|
|
self.total_ms = 0.0
|
|
self._steps: list[Step] = []
|
|
self._error_logged = False # count this turn's failure at most once
|
|
|
|
# -- recording API (called from the pipeline) ------------------------- #
|
|
def heard(self, text: str) -> None:
|
|
self.heard_text = text
|
|
self._touch()
|
|
|
|
def thought(self, text: str) -> None:
|
|
self.thought_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"
|
|
if not self.error:
|
|
failed = next((s for s in self._steps if s.ok is False), None)
|
|
self.error = (failed.error if failed else "") or "step failed"
|
|
else:
|
|
self.status = "ok"
|
|
# A turn that ended in error must be reflected in errors_total. That
|
|
# counter is driven by error-level log events on BOTH the server
|
|
# (Monitor.log) and the browser (dashboard SSE handler), so emit one
|
|
# log event here rather than bumping a counter the client won't mirror.
|
|
# Guarded so the repeated _touch()/finish() calls can't double-count.
|
|
if self.status == "error" and not self._error_logged:
|
|
self._error_logged = True
|
|
self._monitor.log("error", f"대화 #{self.id} 실패: {self.error}")
|
|
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,
|
|
"speaker": self.speaker,
|
|
"guild": self.guild,
|
|
"channel": self.channel,
|
|
"wall": self.wall,
|
|
"heard": self.heard_text,
|
|
"thought": self.thought_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)
|
|
# Keep a long event tail so the terminal panel can show the log from
|
|
# (voice-server) start, not just the last few lines.
|
|
self._events: deque[dict[str, Any]] = deque(maxlen=2000)
|
|
self._status: dict[str, Any] = {
|
|
"running": False,
|
|
"listening": False,
|
|
"started_wall": _now_wall(),
|
|
"components": {},
|
|
"turns_total": 0,
|
|
"errors_total": 0,
|
|
# Claude usage since server start (this bot's own consumption).
|
|
"claude_requests": 0,
|
|
"claude_input_tokens": 0,
|
|
"claude_output_tokens": 0,
|
|
}
|
|
self._lock = threading.Lock()
|
|
self._subs: list["queue.Queue[str]"] = []
|
|
self._id = 0
|
|
self._event_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 add_claude_usage(self, input_tokens: int, output_tokens: int) -> None:
|
|
"""Accumulate one Claude call's token usage for the dashboard."""
|
|
with self._lock:
|
|
self._status["claude_requests"] += 1
|
|
self._status["claude_input_tokens"] += int(input_tokens or 0)
|
|
self._status["claude_output_tokens"] += int(output_tokens or 0)
|
|
self._broadcast({"type": "status", "status": self.status_snapshot()})
|
|
|
|
def log(self, level: str, message: str) -> None:
|
|
"""A free-form lifecycle/error line (startup, disconnect, crash…)."""
|
|
with self._lock:
|
|
self._event_id += 1
|
|
evt = {"type": "log", "id": self._event_id, "level": level,
|
|
"message": message, "wall": _now_wall()}
|
|
self._events.append(evt)
|
|
if level == "error":
|
|
self._status["errors_total"] += 1
|
|
self._broadcast(evt)
|
|
|
|
def delete_event(self, event_id: int) -> bool:
|
|
"""Remove one log line by id (dashboard per-line '삭제')."""
|
|
found = False
|
|
with self._lock: # _broadcast re-locks, so must run outside this block
|
|
for e in list(self._events):
|
|
if e.get("id") == event_id:
|
|
self._events.remove(e)
|
|
found = True
|
|
break
|
|
if found:
|
|
self._broadcast({"type": "log_deleted", "id": event_id})
|
|
return found
|
|
|
|
def edit_event(self, event_id: int, message: str) -> bool:
|
|
"""Edit one log line's text by id (dashboard per-line '수정')."""
|
|
found = False
|
|
with self._lock: # _broadcast re-locks, so must run outside this block
|
|
for e in self._events:
|
|
if e.get("id") == event_id:
|
|
e["message"] = message
|
|
found = True
|
|
break
|
|
if found:
|
|
self._broadcast({"type": "log_edited", "id": event_id, "message": message})
|
|
return found
|
|
|
|
def clear_events(self) -> None:
|
|
"""Wipe the event/error log (dashboard '로그 삭제'). Broadcasts a reset so
|
|
every connected page clears its terminal panel too."""
|
|
with self._lock:
|
|
self._events.clear()
|
|
self._broadcast({"type": "logs_cleared"})
|
|
|
|
# -- 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:
|
|
# errors_total is bumped once when the turn transitions to error, inside
|
|
# Turn.finish() (via a log event), so this only streams the turn state.
|
|
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)
|