watch_sceen_ai/wsai/monitor.py

"""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] = []
        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 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,
            "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:
        # 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)