Watching the System Over Time

# labs/metrics.py
import time

import psutil

def collect_metrics() -> dict:
    mem = psutil.virtual_memory()
    return {
        "timestamp": time.time(),
        "cpu_pct": psutil.cpu_percent(interval=0.1),
        "ram_mb": round(mem.used / 1024 ** 2, 1),
        "ram_pct": mem.percent,
    }

if __name__ == "__main__":
    print(collect_metrics())

$ uv add psutil && uv run python labs/metrics.py
{'timestamp': 1787492103.882, 'cpu_pct': 7.4, 'ram_mb': 5218.3, 'ram_pct': 33.1}

Measured on the bench, so your numbers will land elsewhere; the keys are what stays fixed. Two decisions are already made here. The timestamp is time.time(), the same Unix float the sensor log stamps its readings with, because the two logs will be read side by side one day and one sort key beats two. And virtual_memory() is called once into mem: call it twice and the used-megabytes figure and the percentage come from two different moments, a small lie that surfaces later as a snapshot whose own fields disagree. interval=0.1 is the load-bearing argument in the file, and the failure box explains what it buys.

# labs/timing.py — two lines added to chapter 28's decorator
STAGE_TIMINGS: dict[str, float] = {}

            elapsed = (time.perf_counter() - start) * 1000
            STAGE_TIMINGS[func.__name__] = elapsed
            logger.debug(f"{func.__name__} completed in {elapsed:.1f}ms")

# labs/metrics.py
from labs.timing import STAGE_TIMINGS

def collect_metrics(turns: int = 0) -> dict:
    mem = psutil.virtual_memory()
    return {
        "timestamp": time.time(),
        "cpu_pct": psutil.cpu_percent(interval=0.1),
        "ram_mb": round(mem.used / 1024 ** 2, 1),
        "ram_pct": mem.percent,
        "turns": turns,
        "stage_ms": {name: round(ms, 1) for name, ms in STAGE_TIMINGS.items()},
    }

$ uv run python labs/metrics.py   # after two decorated stubs have run
stage_ms: {'transcribe_stub': 10.2, 'llm_stub': 20.1}

The decorator was already computing this number and throwing it away after printing it. Now it keeps the most recent duration per stage in a module-level dict, and the collector copies that dict into the snapshot. The copy matters: writing "stage_ms": STAGE_TIMINGS stores a reference to a dict that keeps mutating, so every snapshot in memory would show the last turn's numbers. The comprehension freezes the values at collection time. Nesting them under stage_ms also keeps a stage called respond from colliding with a machine field.

from datetime import datetime

def print_dashboard(m: dict) -> None:
    stamp = datetime.fromtimestamp(m["timestamp"]).strftime("%H:%M:%S")
    print(f"=== GLaDOS system: {stamp} ===")
    print(f"  CPU    {m['cpu_pct']:5.1f} %")
    print(f"  RAM    {m['ram_mb']:7.1f} MB  ({m['ram_pct']:.1f} %)")
    print(f"  turns  {m['turns']}")
    for name, ms in sorted(m["stage_ms"].items(), key=lambda kv: -kv[1]):
        print(f"  {name:<16} {ms:7.1f} ms")

if __name__ == "__main__":
    print_dashboard(collect_metrics(turns=2))

$ uv run python labs/metrics.py
=== GLaDOS system: 21:15:41 ===
  CPU      7.4 %
  RAM     5218.3 MB  (33.1 %)
  turns  2
  llm_stub            20.1 ms
  transcribe_stub     10.2 ms

The renderer does three jobs the collector refuses to do. It turns the Unix float into a clock time a person can match against their memory of the evening. It aligns the numbers with format specs (:7.1f pads to seven characters, :<16 left-aligns the name in a sixteen-wide column) so the numbers can be scanned vertically. And it sorts the stages slowest first, putting the answer to "what is costing me the most" on the top line every time. None of that belongs in the snapshot, which stays a plain dict of numbers for the log and every future consumer.

# labs/rolling_log.py
import json
from pathlib import Path

def append_bounded(entry: dict, log_path: Path,
                   max_entries: int = 500) -> int:
    log_path.parent.mkdir(parents=True, exist_ok=True)
    entries = []
    if log_path.exists():
        with open(log_path) as f:
            entries = json.load(f)
    entries.append(entry)
    entries = entries[-max_entries:]
    with open(log_path, "w") as f:
        json.dump(entries, f, indent=2)
    return len(entries)

# labs/metrics.py
from labs.rolling_log import append_bounded

METRICS_LOG = Path("glados/data/metrics.json")

if __name__ == "__main__":
    for _ in range(12):
        stored = append_bounded(collect_metrics(), METRICS_LOG, max_entries=5)
    print(f"writes: 12, snapshots stored: {stored}")

$ uv run python labs/metrics.py
writes: 12, snapshots stored: 5

Not one line of that writer is new. It is chapter 29's sensor logger with the word sensor taken out of the name, and the fact that nothing in the body had to change is the evidence it belonged one level down all along: it never looked inside the dict it was given. labs/sensor_log.py now calls it too, so log_sensor_reading becomes a single delegating line and the retention rule lives in one place. The default cap of 500, at one snapshot per turn, is several long evenings of conversation, and entries[-max_entries:] still evicts the oldest on every write, so the file has a maximum size you can state in advance.

# glados/core.py — the tail of run_turn
        self.history = self.history[-2 * MAX_TURNS:]
        snapshot = collect_metrics(turns=len(self.history) // 2)
        append_bounded(snapshot, METRICS_LOG)
        return True

# labs/metrics.py
def load_metrics(log_path: Path) -> list[dict]:
    if not log_path.exists():
        return []
    with open(log_path) as f:
        entries = json.load(f)
    entries.sort(key=lambda e: e["timestamp"])
    return entries

def slowest_run(entries: list[dict], stage: str) -> dict | None:
    runs = [e for e in entries if stage in e.get("stage_ms", {})]
    return max(runs, key=lambda e: e["stage_ms"][stage]) if runs else None

if __name__ == "__main__":
    entries = load_metrics(METRICS_LOG)
    span = entries[-1]["timestamp"] - entries[0]["timestamp"]
    mean_cpu = sum(e["cpu_pct"] for e in entries) / len(entries)
    worst = slowest_run(entries, "respond")
    when = datetime.fromtimestamp(worst["timestamp"]).strftime("%H:%M:%S")
    print(f"{len(entries)} snapshots over {span:.0f}s, mean CPU {mean_cpu:.1f}%")
    print(f"slowest respond: {worst['stage_ms']['respond']:.1f} ms at {when}, "
          f"CPU {worst['cpu_pct']:.1f}%")

$ uv run python labs/metrics.py   # after a real session, bench numbers
41 snapshots over 1264s, mean CPU 22.6%
slowest respond: 4820.3 ms at 21:58:12, CPU 96.4%

Two lines inside run_turn and the whole session becomes queryable. The sort in load_metrics is the same guarantee chapter 29 argued for: the writer appends in order today, the reader promises order forever. Read the bench output as a diagnosis, because it is one. Her slowest reply of the evening happened at 96 percent CPU, which points the investigation at whatever else was running instead of at the model. Without the machine numbers beside the timing, that same 4.8 seconds looks like her fault.