A Window Into Her Vitals

# labs/metrics_collector.py
from collections import deque

WINDOW = 120


class MetricsCollector:
    def __init__(self, window: int = WINDOW) -> None:
        self._window: deque[tuple[str, float]] = deque(maxlen=window)

    def record_latency(self, stage: str, ms: float) -> None:
        self._window.append((stage, ms))

    def avg_latency(self, stage: str) -> float:
        readings = [ms for name, ms in self._window if name == stage]
        return sum(readings) / len(readings) if readings else 0.0


if __name__ == "__main__":
    m = MetricsCollector()
    for ms in (410.2, 389.0, 502.4):
        m.record_latency("stt", ms)
    print(f"stt avg {m.avg_latency('stt'):7.1f} ms")
    print(f"arm avg {m.avg_latency('arm'):7.1f} ms")

$ uv run python labs/metrics_collector.py
stt avg   433.9 ms
arm avg     0.0 ms

deque(maxlen=120) is the whole retention policy. Append a 121st reading and the oldest one falls off the left end without a word, so the collector never holds more than 120 tuples no matter how long she runs, and nothing anywhere has to remember to prune. One window holds every stage mixed together, so avg_latency filters by name before it divides. The if readings else 0.0 tail looks like defensive clutter and is the most important clause in the file; the failure box shows what the file does without it. arm reads 0.0 here because nothing has moved yet.

    def __init__(self, window: int = WINDOW) -> None:
        self._window: deque[tuple[str, float]] = deque(maxlen=window)
        self._counters: dict[str, int] = {}
        self._gauges: dict[str, float] = {}

    def increment(self, name: str, amount: int = 1) -> None:
        self._counters[name] = self._counters.get(name, 0) + amount

    def gauge(self, name: str, value: float) -> None:
        self._gauges[name] = value

    def get_counter(self, name: str) -> int:
        return self._counters.get(name, 0)

    def get_gauge(self, name: str) -> float:
        return self._gauges.get(name, 0.0)


if __name__ == "__main__":
    m = MetricsCollector()
    m.increment("turns")
    m.increment("turns")
    m.gauge("cpu_pct", 31.5)
    m.gauge("cpu_pct", 38.2)
    print(f"turns  {m.get_counter('turns')}")
    print(f"stalls {m.get_counter('stalls')}")
    print(f"cpu    {m.get_gauge('cpu_pct')}")

$ uv run python labs/metrics_collector.py
turns  2
stalls 0
cpu    38.2

Two calls, two different meanings, and the code makes the difference structural. increment reads the running total with .get(name, 0) and adds to it, so a counter accumulates and never goes backwards. gauge assigns, so the second CPU reading replaces the first: a gauge has no past, only a present. Keeping them in separate dictionaries with separate methods means no caller can accidentally add five to a temperature or overwrite a turn count. Note stalls, a counter nobody has touched, answering 0 instead of raising KeyError. A metric that has not happened yet has a correct value, and it is zero.

import json
import time

STAGES = ("stt", "llm", "tts", "arm")


    def snapshot(self) -> dict:
        return {
            "timestamp": time.time(),
            "window": len(self._window),
            "counters": dict(self._counters),
            "gauges": dict(self._gauges),
            "avg_ms": {s: round(self.avg_latency(s), 1) for s in STAGES},
        }


if __name__ == "__main__":
    m = MetricsCollector()
    for stage, readings in (("stt", (410.2, 389.0, 502.4)),
                            ("llm", (1204.6, 1043.2, 1293.5)),
                            ("tts", (251.9, 276.1, 277.3))):
        for ms in readings:
            m.record_latency(stage, ms)
    m.increment("turns", 3)
    m.gauge("cpu_pct", 31.5)
    m.gauge("ram_pct", 34.2)
    print(json.dumps(m.snapshot(), indent=2))

$ uv run python labs/metrics_collector.py   # bench numbers; yours will differ
{
  "timestamp": 1787694412.518,
  "window": 9,
  "counters": {
    "turns": 3
  },
  "gauges": {
    "cpu_pct": 31.5,
    "ram_pct": 34.2
  },
  "avg_ms": {
    "stt": 433.9,
    "llm": 1180.4,
    "tts": 268.4,
    "arm": 0.0
  }
}

Two decisions carry this method. dict(self._counters) copies instead of handing out the live dictionary, so a snapshot taken at 21:04 still says what it said at 21:04 after another forty turns have gone by; a caller who kept it is holding a frozen moment, not a moving reference. And avg_ms is computed over a fixed STAGES tuple, not over whatever names happen to be in the window, so a stage that stopped reporting shows up as a row of zeros instead of silently vanishing from the display. The timestamp is time.time(), the same Unix float chapter 29's sensor log and chapter 36's metrics log both stamp their entries with, so this dict drops into either store without translation.

# labs/metrics_collector.py — full file
import random
import time
from collections import deque
from contextlib import contextmanager
from pathlib import Path

import psutil

from labs.rolling_log import append_bounded

METRICS_LOG = Path("glados/data/metrics.json")
STAGES = ("stt", "llm", "tts", "arm")
WINDOW = 120


class MetricsCollector:
    def __init__(self, window: int = WINDOW) -> None:
        self._window: deque[tuple[str, float]] = deque(maxlen=window)
        self._counters: dict[str, int] = {}
        self._gauges: dict[str, float] = {}

    def record_latency(self, stage: str, ms: float) -> None:
        self._window.append((stage, ms))

    def increment(self, name: str, amount: int = 1) -> None:
        self._counters[name] = self._counters.get(name, 0) + amount

    def gauge(self, name: str, value: float) -> None:
        self._gauges[name] = value

    def get_counter(self, name: str) -> int:
        return self._counters.get(name, 0)

    def get_gauge(self, name: str) -> float:
        return self._gauges.get(name, 0.0)

    def avg_latency(self, stage: str) -> float:
        readings = [ms for name, ms in self._window if name == stage]
        return sum(readings) / len(readings) if readings else 0.0

    @contextmanager
    def measure(self, stage: str):
        start = time.perf_counter()
        try:
            yield
        finally:
            self.record_latency(stage, (time.perf_counter() - start) * 1000)

    def snapshot(self) -> dict:
        return {
            "timestamp": time.time(),
            "window": len(self._window),
            "counters": dict(self._counters),
            "gauges": dict(self._gauges),
            "avg_ms": {s: round(self.avg_latency(s), 1) for s in STAGES},
        }


def live_line(snap: dict) -> str:
    avg = snap["avg_ms"]
    return (f"turns {snap['counters'].get('turns', 0):3d}"
            f"  stt {avg['stt']:7.1f}  llm {avg['llm']:7.1f}  tts {avg['tts']:7.1f}"
            f"  cpu {snap['gauges'].get('cpu_pct', 0.0):5.1f}%"
            f"  window {snap['window']:3d}")


def stub_turn(m: MetricsCollector) -> None:
    with m.measure("stt"):
        time.sleep(random.uniform(0.38, 0.52))
    with m.measure("llm"):
        time.sleep(random.uniform(1.02, 1.31))
    with m.measure("tts"):
        time.sleep(random.uniform(0.24, 0.29))
    m.increment("turns")
    m.gauge("cpu_pct", psutil.cpu_percent())
    m.gauge("ram_pct", psutil.virtual_memory().percent)


def main() -> None:
    psutil.cpu_percent()          # prime the counter, as chapter 36 insisted
    m = MetricsCollector()
    for _ in range(3):
        stub_turn(m)
        print(live_line(m.snapshot()))
    stored = append_bounded(m.snapshot(), METRICS_LOG)
    print(f"snapshot appended to {METRICS_LOG} ({stored} entries)")


if __name__ == "__main__":
    main()

$ uv run python labs/metrics_collector.py   # bench numbers; yours will differ
turns   1  stt   410.2  llm  1204.6  tts   251.9  cpu  31.5%  window   3
turns   2  stt   399.6  llm  1123.9  tts   264.0  cpu  38.2%  window   6
turns   3  stt   433.9  llm  1180.4  tts   268.4  cpu  29.7%  window   9

$ uv run python labs/metrics_collector.py   # tail of the same run
snapshot appended to glados/data/metrics.json (42 entries)

measure is the whole instrumentation story. The @contextmanager decorator turns a generator into something usable with with: everything before yield runs on entry, everything after runs on exit, and the body of the with block runs at the yield. The finally is deliberate. A stage that raises still records its duration on the way out, so the turn where the model timed out after eleven seconds shows up in the window instead of being the one measurement you lose. Compare the averages down the three lines and you can watch the window fill: line one averages one reading per stage, line three averages three.

Inside glados/core.py, wiring this into the real loop is one more constructor argument, three with blocks and two calls. The core takes metrics alongside its three collaborators and keeps it on self. Then wrap the existing self.stt(audio_path) call in with self.metrics.measure("stt"):, do the same around self.llm(text, self.history) and self.tts(reply), wrap whatever call moves the arm in measure("arm") so the fourth row stops reading zero, then increment("turns") and one gauge for CPU at the end. The providers stay untouched. Nothing behind the contracts from chapter 44 knows it is being timed. Whether self.stt holds a bound WhisperSTT.transcribe or a two-line fake, the timing code is identical, because the core only ever knew it as something you call.