feat(obs): a lens that can see the fault between the checks

'Je jouais, d'un seul coup plus de son.' Every static check said fine:
SuperCollider up with nrestarts=0 and a clean log, Tidal's GHC alive and
holding 6010, the PipeWire graph fully linked SuperCollider -> Ardour -> UMC,
Ardour's master at -0.2 dB reaching both interface channels, the UMC reporting
48000 Hz momentary on both directions. Nothing to find, if you only look once.

The tell came from PLN's ear instead: the system volume beeps had gone 'legato
0.2' — truncated. A device that eats the head off a short sound is a device
being stopped and restarted, and no lens in this repo could see that, because
each one samples a moment and this fault lives strictly between moments.

trigger_time is what exposes it. It changes only when a PCM stream is
(re)started, so it is a restart counter that nothing else in the stack
publishes: no log line, no error, and PipeWire reports the node 'running' on
both sides of the event. Reading it twice showed the UMC's PLAYBACK stream had
restarted while the CAPTURE stream had been up for 33 minutes straight — and
since pw-top shows the capture node is the graph DRIVER, the graph keeps
clocking serenely through an output that has gone away. That asymmetry is the
whole bug, and it is invisible to every one-shot check.

So the tool watches over time and reports what a snapshot cannot: stream
restarts with the lifetime of the stream that died, effective rate from hw_ptr
deltas (RUNNING with a frozen hw_ptr is a stall, and a rate off nominal is the
'everything plays fast' flavour of the same complaint), state transitions, and
device presence across an unplug.

Two deliberate choices. It resolves the card BY NAME on every poll rather than
caching a number, because a replug can move the UMC from card1 to card2 and a
lens pinned to card1 would report 'gone' forever and miss the entire after-half
of the replug test this was built for. And it is strictly passive — opens no
PCM, plays nothing, captures nothing — because this rig has already had one
investigation where the probe's own capture caused the xruns being blamed on
the rig.

Verified: --once reads card1 with playback age 198s vs capture 2192s, the exact
asymmetry above; a 45 s watch over a quiet stretch correctly reported zero
restarts (the fault is intermittent, which is itself the finding); the unknown
device path lists the cards present instead of failing bare.

tools/audio-lens.py

+#!/usr/bin/env python3
+"""Watch the audio interface the way it actually fails: passively, over time.
+
+Written 2026-08-14, after "je jouais d'un seul coup plus de son" — sound gone
+mid-play, with SuperCollider healthy, Tidal alive, the PipeWire graph fully
+linked and Ardour's master at -0.2 dB. Every one-shot check said FINE. The tell
+was PLN's ear, not the checks: system volume beeps had gone "legato 0.2" —
+truncated. A device that cuts the first N ms off a short sound is a device that
+is being STOPPED AND RESTARTED, and none of the static lenses can see that,
+because each one samples a moment and the fault lives between moments.
+
+What this reads, and why each one:
+
+  hw_ptr        The only honest proof that audio is MOVING. `state: RUNNING`
+                is what a wedged device says too — it means the stream is set
+                up, not that frames are flowing. Two reads and a subtraction
+                give the effective rate, which is also how a clock problem
+                shows itself (a device running 44100 into a 48000 graph plays
+                everything ~9% fast; that IS the "legato" symptom).
+  trigger_time  Changes ONLY when the stream is (re)started. It is therefore a
+                restart counter that nothing else in the stack exposes: no log
+                line is printed, no error is raised, PipeWire reports the node
+                `running` on both sides of it. This is the field that found it.
+  state         XRUN / SETUP / SUSPENDED as distinct from RUNNING.
+  card number   Resolved BY NAME on every poll, never cached. A replug can move
+                the device from card1 to card2, and a lens hardcoded to card1
+                would report "gone" forever and miss the whole after-half of an
+                unplug/replug test — which is the exact test this was built for.
+
+It is passive: it opens no PCM, plays nothing, captures nothing, and takes no
+audio path. That matters — a probe that opens the device perturbs the thing it
+measures, and this rig has already had one investigation where the capture
+itself caused the xruns being investigated.
+
+  tools/audio-lens.py                     watch until Ctrl-C
+  tools/audio-lens.py --watch 60          watch 60 s, then summarise
+  tools/audio-lens.py --once              one snapshot, no watching
+  tools/audio-lens.py --device 'Launch'   match a different card by name
+"""
+from __future__ import annotations
+
+import argparse
+import re
+import subprocess
+import sys
+import time
+from datetime import datetime
+from pathlib import Path
+
+PROC = Path("/proc/asound")
+DEFAULT_DEVICE = "UMC202HD"
+
+# Kernel lines worth counting across a run. The clock-source one is specific to
+# this interface: 60+ of them arrived in a burst when the UMC re-enumerated on
+# 2026-08-14, and a device whose clock source the driver rejected is a device
+# whose rate you cannot trust.
+KERNEL_PATTERNS = {
+    "clock-source rejected": r"clock source .* not valid",
+    "usb disconnect": r"USB disconnect",
+    "usb enumerate": r"New USB device found",
+    "xhci error": r"xhci_hcd.*(error|timeout|halt)",
+}
+
+
+def now() -> str:
+    return datetime.now().strftime("%H:%M:%S")
+
+
+def find_card(name_match: str) -> tuple[int, str] | tuple[None, None]:
+    """(card_number, description) for the first card whose entry mentions
+    `name_match`. Resolved fresh every call — see the docstring."""
+    try:
+        txt = (PROC / "cards").read_text()
+    except OSError:
+        return None, None
+    for line in txt.splitlines():
+        m = re.match(r"\s*(\d+)\s*\[([^\]]+)\]\s*:\s*(.*)", line)
+        if m and name_match.lower() in line.lower():
+            return int(m.group(1)), m.group(3).strip()
+    return None, None
+
+
+def read_sub(card: int, direction: str) -> dict | None:
+    """Parse one substream's status. direction is 'p' (playback) or 'c'."""
+    p = PROC / f"card{card}" / f"pcm0{direction}" / "sub0" / "status"
+    try:
+        txt = p.read_text()
+    except OSError:
+        return None
+    if txt.startswith("closed"):
+        return {"state": "closed"}
+    out = {"state": "?"}
+    m = re.search(r"state:\s*(\w+)", txt)
+    if m:
+        out["state"] = m.group(1)
+    for key in ("hw_ptr", "appl_ptr", "trigger_time", "tstamp", "delay", "avail"):
+        m = re.search(rf"{key}\s*:\s*([0-9.]+)", txt)
+        if m:
+            out[key] = float(m.group(1))
+    return out
+
+
+def nominal_rate(card: int) -> int | None:
+    """Momentary frequency the hardware reports, from stream0."""
+    try:
+        txt = (PROC / f"card{card}" / "stream0").read_text()
+    except OSError:
+        return None
+    m = re.search(r"Momentary freq = (\d+) Hz", txt)
+    return int(m.group(1)) if m else None
+
+
+def kernel_counts() -> dict[str, int]:
+    """Count matching kernel lines in the recent journal.
+
+    Deliberately -n (a fixed tail) rather than --since: a time-bounded journal
+    scan is pathologically slow on this box. Counts are only ever compared to
+    another count from the same command, so the window being fixed is fine.
+    """
+    counts = {k: 0 for k in KERNEL_PATTERNS}
+    try:
+        txt = subprocess.run(["journalctl", "-k", "-n", "2000", "--no-pager", "-o", "cat"],
+                             capture_output=True, text=True, timeout=20).stdout
+    except (OSError, subprocess.SubprocessError):
+        return counts
+    for label, pat in KERNEL_PATTERNS.items():
+        counts[label] = len(re.findall(pat, txt, re.I))
+    return counts
+
+
+class Watcher:
+    def __init__(self, device: str, verbose: bool):
+        self.device = device
+        self.verbose = verbose
+        self.prev: dict[str, dict] = {}
+        self.card: int | None = None
+        self.events: list[str] = []
+        self.restarts = {"playback": 0, "capture": 0}
+        self.rates: dict[str, list[float]] = {"playback": [], "capture": []}
+        self.gone_since: float | None = None
+
+    def log(self, msg: str, keep: bool = True):
+        line = f"  {now()}  {msg}"
+        print(line, flush=True)
+        if keep:
+            self.events.append(line)
+
+    def poll(self):
+        card, desc = find_card(self.device)
+
+        # --- presence, which a replug test turns on and off by design
+        if card is None:
+            if self.card is not None or self.gone_since is None:
+                self.log(f"DEVICE GONE — no card matching {self.device!r}")
+                self.gone_since = time.time()
+                self.card = None
+                self.prev.clear()
+            return
+        if self.card != card:
+            if self.gone_since is not None:
+                gap = time.time() - self.gone_since
+                self.log(f"DEVICE BACK as card{card} after {gap:.1f}s — {desc}")
+                self.gone_since = None
+            else:
+                # Not a gap: either the first poll, or the card RENUMBERED
+                # without us seeing it vanish (fast replug between polls).
+                what = "found" if self.card is None else f"MOVED from card{self.card}"
+                self.log(f"device {what}: card{card} — {desc}")
+            self.card = card
+
+        for direction, key in (("p", "playback"), ("c", "capture")):
+            cur = read_sub(card, direction)
+            if cur is None:
+                continue
+            old = self.prev.get(key)
+            self.prev[key] = cur
+
+            if old is None:
+                self.log(f"{key:<8} {cur['state']}", keep=False)
+                continue
+
+            if cur["state"] != old["state"]:
+                self.log(f"{key:<8} state {old['state']} -> {cur['state']}")
+
+            # The restart detector. trigger_time moving means the stream was
+            # stopped and started again — silence, and a truncated head on
+            # whatever plays next.
+            if cur.get("trigger_time") and old.get("trigger_time") and \
+                    cur["trigger_time"] != old["trigger_time"]:
+                lived = cur["trigger_time"] - old["trigger_time"]
+                self.restarts[key] += 1
+                self.log(f"{key:<8} !! STREAM RESTARTED "
+                         f"(previous stream lived {lived:.1f}s)")
+
+            # Effective rate: the arithmetic that separates "set up" from
+            # "moving", and a clock mismatch from a healthy device.
+            dh = cur.get("hw_ptr", 0) - old.get("hw_ptr", 0)
+            dt = cur.get("tstamp", 0) - old.get("tstamp", 0)
+            if dt > 0.5 and dh >= 0 and cur["state"] == "RUNNING":
+                eff = dh / dt
+                self.rates[key].append(eff)
+                nom = nominal_rate(card) or 48000
+                if dh == 0:
+                    self.log(f"{key:<8} !! STALLED — RUNNING but hw_ptr frozen "
+                             f"for {dt:.1f}s")
+                elif abs(eff - nom) / nom > 0.02:
+                    self.log(f"{key:<8} !! RATE OFF — {eff:.0f} Hz effective vs "
+                             f"{nom} nominal ({100*(eff-nom)/nom:+.1f}%)")
+                elif self.verbose:
+                    self.log(f"{key:<8} {eff:.0f} Hz", keep=False)
+
+    def summary(self, dur: float, kern_before: dict, kern_after: dict):
+        print(f"\n=== {dur:.0f}s on {self.device} " + "=" * 34)
+        for key in ("playback", "capture"):
+            r = self.rates[key]
+            if r:
+                print(f"  {key:<9} {len(r)} samples  "
+                      f"rate min/med/max {min(r):.0f}/{sorted(r)[len(r)//2]:.0f}/{max(r):.0f} Hz"
+                      f"   restarts={self.restarts[key]}")
+            else:
+                print(f"  {key:<9} no RUNNING samples   restarts={self.restarts[key]}")
+        deltas = {k: kern_after[k] - kern_before[k] for k in kern_before}
+        hot = {k: v for k, v in deltas.items() if v}
+        print("  kernel   " + (", ".join(f"{k} +{v}" for k, v in hot.items())
+                               if hot else "nothing new"))
+        if self.events:
+            print(f"  events   {len(self.events)}")
+        else:
+            print("  events   none — device stayed up, streams never restarted")
+        # The point of the whole tool, said out loud.
+        total = sum(self.restarts.values())
+        if total:
+            print(f"\n  VERDICT: {total} stream restart(s). Each one is a gap in the "
+                  f"sound\n           and a truncated head on the next note. This is "
+                  f"the fault.")
+        else:
+            print("\n  VERDICT: no restarts observed in this window. If the symptom "
+                  "happened\n           here, it was not the interface stopping.")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--device", default=DEFAULT_DEVICE,
+                    help=f"substring matched against /proc/asound/cards (default {DEFAULT_DEVICE})")
+    ap.add_argument("--watch", type=float, default=0,
+                    help="seconds to watch (default: until Ctrl-C)")
+    ap.add_argument("--once", action="store_true", help="one snapshot and exit")
+    ap.add_argument("--interval", type=float, default=0.25)
+    ap.add_argument("--verbose", action="store_true", help="print every rate sample")
+    a = ap.parse_args()
+
+    card, desc = find_card(a.device)
+    if card is None:
+        print(f"No sound card matching {a.device!r} in /proc/asound/cards.")
+        print("Cards present:")
+        try:
+            print("  " + (PROC / "cards").read_text().replace("\n", "\n  "))
+        except OSError:
+            pass
+        return 1
+
+    if a.once:
+        print(f"card{card} — {desc}   nominal {nominal_rate(card)} Hz")
+        for direction, key in (("p", "playback"), ("c", "capture")):
+            s = read_sub(card, direction) or {}
+            age = (s.get("tstamp", 0) - s.get("trigger_time", 0)) if s.get("trigger_time") else None
+            print(f"  {key:<9} state={s.get('state','?'):<9} "
+                  f"stream age {age:.1f}s" if age is not None
+                  else f"  {key:<9} state={s.get('state','?')}")
+        print("\nA snapshot cannot see a restart — that needs two reads. "
+              "Use --watch N.")
+        return 0
+
+    w = Watcher(a.device, a.verbose)
+    kern_before = kernel_counts()
+    print(f"audio-lens on card{card} — {desc}")
+    print(f"nominal {nominal_rate(card)} Hz · polling {a.interval}s · "
+          f"{'watching ' + str(int(a.watch)) + 's' if a.watch else 'Ctrl-C to stop'}")
+    print("passive: opens no PCM, plays nothing\n")
+
+    t0 = time.time()
+    try:
+        while True:
+            w.poll()
+            if a.watch and time.time() - t0 >= a.watch:
+                break
+            time.sleep(a.interval)
+    except KeyboardInterrupt:
+        print()
+    w.summary(time.time() - t0, kern_before, kernel_counts())
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())