gig-log: publish rig health (audio clock, surface driver, sound engine)

Four faults on 2026-10-11 fell between the gate and the per-unit watchdogs, each
leaving every unit active. rig_health judges the signal path from what gig-log
already reads: the pw-top stream (are Ardour/SuperCollider clocked by a sound
card or by the MIDI bridge), /proc/asound (raw LCXL3 present but no driver
port), and the gear sample (sclang up, scsynth dead). Hysteresis, atomic
health.json with a heartbeat, notify + evidence snapshot only on a failure
transition. Steady state adds no fork.

tools/gig-log.py

@@ -139,6 +139,7 @@ import perf  # noqa: E402  (rootless Thermals/_read live here already — DRY wi
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 import ardour_session  # noqa: E402  (the argv>cwd>fds ladder, shared with check-mix.py)
+import rig_health  # noqa: E402  (is the signal path alive NOW: clock, surface, engine)
 
 # The v3 DAW -> v2 corpus CC map, so a night that binds the board's own port is
 # still logged in the numbering the corpus speaks. Optional import: a log with
@@ -416,6 +417,9 @@ class XrunReader(threading.Thread):
         self.lock = threading.Lock()
         self.alive = False
         self.lines = 0
+        # Same stream, second reader: which node is the graph's clock. Costs no
+        # process and no extra read, see rig_health.ClockWatch.
+        self.clock = rig_health.ClockWatch()
         self._proc: subprocess.Popen | None = None
         self._stop = threading.Event()
 
@@ -479,6 +483,7 @@ class XrunReader(threading.Thread):
             for line in self._proc.stdout:            # blocking read, own thread
                 if self._stop.is_set():
                     break
+                self.clock.feed(line)
                 hit = self.parse_line(line)
                 if hit is None:
                     continue
@@ -1369,7 +1374,8 @@ def _package_throttle_sum() -> int:
 class Recorder:
     def __init__(self, out_dir: Path = LOG_DIR, hz: float = 1.0,
                  xruns: bool = True, midi: bool = True,
-                 midi_port: str | None = None, rec: bool = True):
+                 midi_port: str | None = None, rec: bool = True,
+                 health: bool | None = None):
         self.dir = out_dir
         self.period = 1.0 / max(0.05, hz)
         self.th = perf.Thermals()
@@ -1387,6 +1393,12 @@ class Recorder:
         self.rec_enabled = rec
         self.files = FilesLens() if rec else None
         self.osc = OscLens() if rec else None
+        # The health publisher writes a real file under ~/.cache, so only the
+        # production recorder owns it: tests and --selftest pass a scratch dir.
+        if health is None:
+            health = out_dir == LOG_DIR
+        self.health = rig_health.Health(
+            clock_watch=self.xr.clock if self.xr else None) if health else None
         self.path = self._open()
         self.samples = 0
         self._stop = threading.Event()
@@ -1539,8 +1551,12 @@ class Recorder:
         try:
             while not self._stop.is_set():
                 next_t += self.period
-                self._write(self.sample())
+                rec = self.sample()
+                self._write(rec)
                 self.samples += 1
+                if self.health:
+                    for ev in self.health.tick(rec.get("gear")):
+                        self._write(ev)
                 tr = self.poll_track()
                 if tr:
                     self._write(tr)
@@ -1570,6 +1586,8 @@ class Recorder:
             for t in (self.xr, self.mid, self.osc):
                 if t:
                     t.stop()
+            if self.health:
+                self.health.close()
             self.f.close()
         print(f"gig-log: {self.samples} samples -> {self.path}", file=sys.stderr)
         return 0
@@ -2324,6 +2342,14 @@ def cmd_status(d: Path = LOG_DIR) -> int:
     print(f"all logs             : {len(list(d.glob('gig-*.jsonl')))} files, "
           f"{sum(f.stat().st_size for f in d.glob('gig-*.jsonl'))/1e6:.2f} MB")
 
+    # The same file the tray and the Bridge read (rig_health): one truth, shown
+    # here in words. A stale or missing file prints as such, never as ok.
+    h = rig_health.read_health()
+    extra = f" — {h['detail']}" if h.get("detail") else ""
+    print(f"rig health           : {h['overall']}{extra}")
+    for name, c in sorted(h.get("checks", {}).items()):
+        print(f"                       {name:<12} {c['state']:<5} {c['detail']}")
+
     # WHAT IS THE MIDI LENS ACTUALLY BOUND TO (added 2026-09-25)
     # `status` used to answer "is it writing", which was true and useless for
     # the one question that mattered on 2026-09-24: the recorder was LIVE, the

tools/rig_health.py

+"""rig_health — is the signal path alive RIGHT NOW?
+
+The gate (check-gig) proves the rig at launch and the per-service watchdogs
+restart what dies, but nothing asked whether the whole path still carried a
+signal. On 2026-10-11 four faults fell between the layers, each leaving every
+unit `active`:
+
+  * LCXL3 on the bus, lcxl3-driver never started       -> no fader reached Ardour
+  * PipeWire graph clocked by the MIDI bridge          -> no sound, no MIDI relay
+  * a sink node re-created, Ardour Master left dangling -> silence
+  * a gate that cried NO-GO on every launch            -> the real fault went unread
+
+This module is the OBSERVER half. It lives inside gig-log's process on purpose
+(one recorder, one set of long-lived readers) and obeys gig-log's law: the observer
+must not perturb. Steady state costs no fork and no subprocess: it reads /proc
+files and the `pw-top -b` stream gig-log already drains. Subprocesses happen only
+on a FAILURE transition, when something is already broken (evidence snapshot,
+desktop notification).
+
+Owners repair, this observes: autoroute re-links Master, midi-autoconnect starts
+the surface driver. Nothing here writes to the graph or restarts a unit.
+
+THE CLOCK PREDICATE (the one that would have caught the silent stall)
+  A pw-top table lists DRIVER rows (no `+`) with their followers (`+` rows) beneath.
+  The rig is hardware-clocked when Ardour / SuperCollider follow a driver that is
+  an alsa_* node with a non-zero QUANT. When they follow Midi-Bridge, Dummy-Driver,
+  or nothing, the graph has no hardware clock: sound and the MIDI relay are dead
+  while every unit still reads active.
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+OK, WARN, FAIL, NA = "ok", "warn", "fail", "na"
+
+
+
+def health_file() -> Path:
+    """Resolved at CALL time, not import time: a test (or a second instance) that
+    sets PARVAGUES_HEALTH_FILE after import must actually be redirected, or it
+    writes the real file the tray is reading."""
+    return Path(os.environ.get(
+        "PARVAGUES_HEALTH_FILE", os.path.expanduser("~/.cache/parvagues/health.json")))
+
+
+def evidence_dir() -> Path:
+    return Path(os.environ.get(
+        "PARVAGUES_HEALTH_EVIDENCE",
+        os.path.expanduser("~/.local/share/parvagues/health-evidence")))
+
+
+FAIL_AFTER = 5          # consecutive bad ticks before a check is declared failed
+OK_AFTER = 3            # consecutive good ticks before it is declared recovered
+HEARTBEAT_S = 5.0       # health.json is rewritten at least this often (stale = dead)
+STALE_AFTER_S = 30.0    # readers treat an older file as "the observer died"
+NOTIFY_COOLDOWN_S = 60.0
+EVIDENCE_COOLDOWN_S = 120.0
+EVIDENCE_KEEP = 10
+
+RIG_APPS = ("ardour", "SuperCollider")          # pw-top node names that make a rig
+STATES = {"R", "I", "S", "C", "D", "E", "!"}
+
+
+# --------------------------------------------------------------------------- #
+# the clock predicate, from the pw-top stream
+# --------------------------------------------------------------------------- #
+def parse_row(line: str) -> dict | None:
+    """One pw-top data row -> {state, id, quant, rate, name, follower}, else None.
+
+    Layout: S ID QUANT RATE WAIT BUSY W/Q B/Q ERR [FORMAT...] [+] NAME. FORMAT is
+    variable-width, so index from both ends. A follower carries a '+' token just
+    before its name.
+    """
+    f = line.split()
+    if len(f) < 10 or f[0] not in STATES or not f[1].isdigit():
+        return None
+    try:
+        quant, rate = int(f[2]), int(f[3])
+    except ValueError:
+        return None
+    return {"state": f[0], "id": int(f[1]), "quant": quant, "rate": rate,
+            "name": f[-1], "follower": f[-2] == "+"}
+
+
+def is_header(line: str) -> bool:
+    f = line.split()
+    return len(f) >= 3 and f[0] == "S" and f[1] == "ID"
+
+
+def clock_verdict(rows: list[dict]) -> tuple[str, str]:
+    """Judge one complete pw-top table."""
+    apps = [r for r in rows if r["name"] in RIG_APPS and r["state"] != "S"]
+    if not apps:
+        return NA, "no Ardour or SuperCollider in the graph"
+    driver = None
+    group: dict[int, dict] = {}
+    for r in rows:
+        if not r["follower"]:
+            driver = r
+        elif driver is not None:
+            group[r["id"]] = driver
+    bad = []
+    for app in apps:
+        d = group.get(app["id"])
+        if d is None:                       # not following anyone: its own driver
+            if app["quant"] > 0:
+                continue
+            bad.append(f"{app['name']} has no clock")
+        elif not d["name"].startswith("alsa_"):
+            bad.append(f"{app['name']} follows {d['name']}, not a sound card")
+        elif d["quant"] <= 0:
+            bad.append(f"{app['name']} follows an idle sound card")
+    if bad:
+        return FAIL, "; ".join(bad) + " (no hardware clock: sound and MIDI relay are dead)"
+    d = group.get(apps[0]["id"]) or apps[0]
+    return OK, f"{apps[0]['name']} clocked by {d['name'][-28:]} at {d['quant']}/{d['rate']}"
+
+
+class ClockWatch:
+    """Accumulates pw-top lines (fed from gig-log's existing reader thread) into
+    tables, and keeps the verdict for the latest complete one."""
+
+    def __init__(self, clock=time.monotonic):
+        self._clock = clock
+        self._rows: list[dict] = []
+        self._seen_header = False
+        self._lock = threading.Lock()
+        self._verdict: tuple[str, str] = (NA, "no pw-top table yet")
+        self._at: float | None = None
+
+    def feed(self, line: str) -> None:
+        if is_header(line):
+            with self._lock:
+                if self._seen_header and self._rows:
+                    self._verdict = clock_verdict(self._rows)
+                    self._at = self._clock()
+                self._rows = []
+                self._seen_header = True
+            return
+        row = parse_row(line)
+        if row is not None and self._seen_header:
+            with self._lock:
+                self._rows.append(row)
+
+    def verdict(self, max_age: float = 5.0) -> tuple[str, str]:
+        with self._lock:
+            state, detail = self._verdict
+            age = None if self._at is None else self._clock() - self._at
+        if age is None or age > max_age:
+            return NA, "pw-top is not reporting"
+        return state, detail
+
+
+# --------------------------------------------------------------------------- #
+# the other probes: procfs files and the gear sample, no forks
+# --------------------------------------------------------------------------- #
+def _read(path: str) -> str:
+    try:
+        return Path(path).read_text(errors="replace")
+    except OSError:
+        return ""
+
+
+def surface_check(cards: str | None = None, seq: str | None = None) -> tuple[str, str]:
+    """Raw LCXL3 on the USB bus => the driver's virtual port must exist.
+
+    The driver publishes 'ParVagues LCXL3'; the raw board is a kernel client. The
+    gate's probe asks systemd; this asks the ALSA sequencer, which is the thing
+    that actually carries the faders, so a driver that is `active` but never bound
+    its port still fails."""
+    cards = _read("/proc/asound/cards") if cards is None else cards
+    seq = _read("/proc/asound/seq/clients") if seq is None else seq
+    if "LCXL3" not in cards:
+        return NA, "no LCXL3 on the bus"
+    if "ParVagues LCXL3" in seq:
+        return OK, "LCXL3 present, driver port published"
+    return FAIL, "LCXL3 on the bus but lcxl3-driver publishes no port: no faders reach Ardour"
+
+
+def superdirt_check(gear: dict | None) -> tuple[str, str]:
+    """sclang alive with scsynth dead is the 2026-08-01 failure (systemd stays
+    active because sclang is the MainPID)."""
+    gear = gear or {}
+    sclang = bool(gear.get("sclang", {}).get("up"))
+    scsynth = bool(gear.get("scsynth", {}).get("up"))
+    if sclang and scsynth:
+        return OK, "sclang and scsynth up"
+    if sclang and not scsynth:
+        return FAIL, "sclang is up but scsynth is dead: no sound engine"
+    return NA, "SuperDirt not running"
+
+
+# --------------------------------------------------------------------------- #
+# publishing: hysteresis, atomic file, notify + evidence on failure transitions
+# --------------------------------------------------------------------------- #
+EVIDENCE_CMDS = (
+    ("pw-link -l", ["pw-link", "-l"]),
+    ("pw-top (2 frames)", ["pw-top", "-b", "-n", "2"]),
+    ("aconnect -l", ["aconnect", "-l"]),
+    ("units", ["systemctl", "--user", "--no-pager", "status", "lcxl3-driver",
+               "midi-autoconnect", "tidal-ardour-autoroute", "parvagues-sc"]),
+    ("pipewire/wireplumber journal", ["journalctl", "--user", "-u", "pipewire",
+                                      "-u", "wireplumber", "--since", "-5min",
+                                      "--no-pager"]),
+)
+
+
+def capture_evidence(check: str, detail: str, out_dir: Path | None = None) -> Path | None:
+    """State first, repair second: what the graph looked like at the moment a check
+    failed, so the next stall has a cause on file instead of a theory."""
+    try:
+        out_dir = out_dir or evidence_dir()
+        out_dir.mkdir(parents=True, exist_ok=True)
+        path = out_dir / f"{time.strftime('%Y%m%d-%H%M%S')}-{check}.txt"
+        parts = [f"check: {check}\ndetail: {detail}\nat: {time.ctime()}\n"]
+        for label, cmd in EVIDENCE_CMDS:
+            if shutil.which(cmd[0]) is None:
+                continue
+            try:
+                r = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+                parts.append(f"\n===== {label} =====\n{r.stdout}{r.stderr}")
+            except Exception as e:                       # noqa: BLE001
+                parts.append(f"\n===== {label} =====\n(failed: {e})\n")
+        path.write_text("".join(parts))
+        path.chmod(0o600)
+        for old in sorted(out_dir.glob("*.txt"))[:-EVIDENCE_KEEP]:
+            old.unlink(missing_ok=True)
+        return path
+    except OSError:
+        return None
+
+
+def desktop_notify(title: str, body: str, urgent: bool) -> None:
+    if shutil.which("notify-send") is None:
+        return
+    try:
+        subprocess.Popen(["notify-send", "-a", "ParVagues",
+                          "-u", "critical" if urgent else "normal", "-t", "0" if urgent else "6000",
+                          title, body], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    except OSError:
+        pass
+
+
+class Health:
+    """Owns the effective state of each check and the published health.json."""
+
+    def __init__(self, path: Path | None = None, clock_watch: ClockWatch | None = None,
+                 notify=desktop_notify, evidence=capture_evidence,
+                 now=time.time, async_evidence: bool = True):
+        self.path = Path(path) if path else health_file()
+        self.clock = clock_watch
+        self._notify, self._evidence, self._now = notify, evidence, now
+        self._async = async_evidence
+        self.state: dict[str, dict] = {}          # name -> {state, detail, since}
+        self._bad: dict[str, int] = {}
+        self._good: dict[str, int] = {}
+        self._last_write = 0.0
+        self._last_notify: dict[str, float] = {}
+        self._last_evidence = 0.0
+
+    # -- the checks that feed one tick ------------------------------------ #
+    def probe(self, gear: dict | None) -> dict[str, tuple[str, str]]:
+        checks = {"surface": surface_check(), "superdirt": superdirt_check(gear)}
+        if self.clock is not None:
+            checks["audio_clock"] = self.clock.verdict()
+        return checks
+
+    def tick(self, gear: dict | None) -> list[dict]:
+        return self.update(self.probe(gear))
+
+    # -- hysteresis + publication ----------------------------------------- #
+    def update(self, checks: dict[str, tuple[str, str]]) -> list[dict]:
+        now = self._now()
+        events: list[dict] = []
+        changed = False
+        for name, (raw, detail) in checks.items():
+            cur = self.state.get(name)
+            if raw == FAIL:
+                self._bad[name] = self._bad.get(name, 0) + 1
+                self._good[name] = 0
+            else:
+                self._good[name] = self._good.get(name, 0) + 1
+                self._bad[name] = 0
+            if cur is None:
+                eff = raw if raw != FAIL else NA        # never open with a verdict of FAIL
+                self.state[name] = {"state": eff, "detail": detail, "since": now}
+                changed = True
+                continue
+            eff = cur["state"]
+            if raw == FAIL and self._bad[name] >= FAIL_AFTER and eff != FAIL:
+                eff = FAIL
+            elif raw != FAIL and eff == FAIL and self._good[name] >= OK_AFTER:
+                eff = raw
+            elif raw != FAIL and eff != FAIL and raw != eff:
+                eff = raw                               # ok <-> na moves freely
+            if eff != cur["state"]:
+                events.append(self._transition(name, cur["state"], eff, detail, now))
+                self.state[name] = {"state": eff, "detail": detail, "since": now}
+                changed = True
+            elif detail != cur["detail"] and eff != FAIL:
+                cur["detail"] = detail                  # wording only: not a change
+        if changed or now - self._last_write >= HEARTBEAT_S:
+            self._publish(now)
+        return events
+
+    def _transition(self, name: str, was: str, now_state: str, detail: str, t: float) -> dict:
+        ev = {"t": round(t, 3), "k": "health", "check": name, "from": was,
+              "to": now_state, "detail": detail}
+        if now_state == FAIL:
+            self._on_fail(name, detail, t)
+        elif was == FAIL:
+            self._maybe_notify(name, f"{name} recovered", detail, urgent=False, t=t)
+        return ev
+
+    def _on_fail(self, name: str, detail: str, t: float) -> None:
+        self._maybe_notify(name, f"Rig: {name} is down", detail, urgent=True, t=t)
+        if t - self._last_evidence >= EVIDENCE_COOLDOWN_S:
+            self._last_evidence = t
+            if self._async:
+                threading.Thread(target=self._evidence, args=(name, detail), daemon=True).start()
+            else:
+                self._evidence(name, detail)
+
+    def _maybe_notify(self, name: str, title: str, body: str, urgent: bool, t: float) -> None:
+        if t - self._last_notify.get(name, 0.0) < NOTIFY_COOLDOWN_S and urgent:
+            return
+        self._last_notify[name] = t
+        self._notify(title, body, urgent)
+
+    def overall(self) -> str:
+        states = [c["state"] for c in self.state.values()]
+        if FAIL in states:
+            return "down"
+        if WARN in states:
+            return "degraded"
+        return "ok"
+
+    def _publish(self, now: float) -> None:
+        doc = {"v": 1, "t": round(now, 3), "pid": os.getpid(),
+               "overall": self.overall(), "checks": self.state}
+        try:
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            tmp = self.path.with_suffix(".tmp")
+            tmp.write_text(json.dumps(doc, separators=(",", ":")))
+            os.replace(tmp, self.path)
+            self._last_write = now
+        except OSError:
+            pass
+
+    def close(self) -> None:
+        """Leave an honest tombstone: a dead observer must not read as health."""
+        try:
+            self.path.write_text(json.dumps({"v": 1, "t": round(self._now(), 3),
+                                             "pid": os.getpid(), "overall": "stopped",
+                                             "checks": {}}))
+        except OSError:
+            pass
+
+
+def read_health(path: Path | None = None, now=time.time) -> dict:
+    """The reader's view (tray, Bridge, gate): never raises, and a missing or
+    stale file is reported as such, never as 'ok'."""
+    try:
+        doc = json.loads(Path(path or health_file()).read_text())
+    except (OSError, ValueError):
+        return {"overall": "unknown", "detail": "no health file: gig-log is not running", "checks": {}}
+    age = now() - float(doc.get("t", 0))
+    if doc.get("overall") == "stopped":
+        doc["detail"] = "gig-log stopped"
+        doc["overall"] = "unknown"
+    elif age > STALE_AFTER_S:
+        doc["overall"] = "stale"
+        doc["detail"] = f"health.json is {age:.0f}s old: the observer is dead or wedged"
+    doc["age_s"] = round(age, 1)
+    return doc

tools/tests/test_gig_log.py

@@ -1491,3 +1491,20 @@ def test_poll_rec_a_mute_osc_lens_never_conflicts_and_never_flaps_src(tmp_path):
         assert "conflict" not in got, "a mute lens must never manufacture a conflict"
         assert got["src"] == "files", "src must not flap to files+osc just because osc is alive"
     r.f.close()
+
+
+# --------------------------------------------------------------------------- #
+# rig health rides inside the recorder (rig_health.py)
+# --------------------------------------------------------------------------- #
+def test_only_the_production_recorder_publishes_health(tmp_path):
+    """health.json is a real file under ~/.cache. A scratch-dir recorder (tests,
+    --selftest) must never write it, or a test run would repaint the tray."""
+    r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
+    assert r.health is None
+
+
+def test_health_can_be_forced_on_and_tolerates_no_pw_top(tmp_path, monkeypatch):
+    monkeypatch.setenv("PARVAGUES_HEALTH_FILE", str(tmp_path / "h.json"))
+    r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False, health=True)
+    assert r.health is not None and r.health.clock is None     # no xruns => no clock probe
+    r.health.tick({"sclang": {"up": False}, "scsynth": {"up": False}})

tools/tests/test_rig_health.py

+"""rig_health: the observer that would have caught 2026-10-11 from the tray.
+
+Fixtures are real `pw-top -b` rows captured on this machine, healthy and stalled.
+"""
+from __future__ import annotations
+
+import json
+import pathlib
+import sys
+
+import pytest
+
+TOOLS = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(TOOLS))
+import rig_health as rh  # noqa: E402
+
+HDR = "S   ID  QUANT   RATE    WAIT    BUSY   W/Q   B/Q  ERR FORMAT           NAME"
+SINK = "alsa_output.pci-0000_00_1f.3-platform-sof_sdw.HiFi__hw_sofsoundwire__"
+
+# SuperCollider and Ardour follow a sound card (the healthy shape)
+HEALTHY = [
+    HDR,
+    "I   29      0      0   0.0us   0.0us  ???   ???     0                  Dummy-Driver",
+    f"R  361   1024  48000   4.7ms  10.7us  0.22  0.00   83    S32LE 2 48000 {SINK}",
+    "R   37      0      0  27.7us  33.7us  0.00  0.00    8                   + Midi-Bridge",
+    "R  112      0      0 124.9us   2.1ms  0.01  0.10    2                   + SuperCollider",
+    "R  137      0      0 102.9us   2.2ms  0.00  0.10  356                   + ardour",
+]
+# the 13:46 signature: the MIDI bridge is the clock, the sound cards drive nothing
+STALLED = [
+    HDR,
+    "I   29      0      0   0.0us   0.0us  ???   ???     0                  Dummy-Driver",
+    "R   37    512  48000   3.4ms   9.3us  0.32  0.00    1                  Midi-Bridge",
+    "R  112      0      0  39.8us   2.1ms  0.00  0.20    0                   + SuperCollider",
+    "R  137      0      0  38.3us   1.2ms  0.00  0.11  168                   + ardour",
+    f"R  362      0      0   0.0us   0.0us  ???   ???     0    S32LE 2 48000 {SINK}",
+]
+# the pre-toggle signature: nothing is scheduled at all
+FROZEN = [
+    HDR,
+    "R   37      0      0   0.0us   0.0us  ???   ???     0                  Midi-Bridge",
+    "R  112      0      0   0.0us   0.0us  ???   ???     0                  SuperCollider",
+    "R  137      0      0   0.0us   0.0us  ???   ???     0                  ardour",
+    f"R  341      0      0   0.0us   0.0us  ???   ???     0    S32LE 2 48000 {SINK}",
+]
+NO_RIG = [HDR, "I   29      0      0   0.0us   0.0us  ???   ???     0                  Dummy-Driver"]
+
+
+def rows(lines):
+    return [r for r in map(rh.parse_row, lines) if r]
+
+
+# --------------------------------------------------------------------------- #
+# the clock predicate
+# --------------------------------------------------------------------------- #
+def test_parse_row_reads_drivers_followers_and_formatted_followers():
+    d = rh.parse_row(HEALTHY[2])
+    assert (d["id"], d["quant"], d["rate"], d["follower"]) == (361, 1024, 48000, False)
+    f = rh.parse_row(HEALTHY[5])
+    assert (f["name"], f["follower"]) == ("ardour", True)
+    ff = rh.parse_row("R  341      0      0  84.9us  17.6us  0.00  0.00   91    S32LE 2 48000  + alsa_output.x")
+    assert (ff["name"], ff["follower"]) == ("alsa_output.x", True)
+
+
+def test_headers_and_junk_are_not_rows():
+    assert rh.parse_row(HDR) is None
+    assert rh.parse_row("") is None
+    assert rh.is_header(HDR)
+
+
+def test_hardware_clocked_rig_is_ok():
+    state, detail = rh.clock_verdict(rows(HEALTHY))
+    assert state == rh.OK and "1024/48000" in detail
+
+
+def test_following_the_midi_bridge_is_the_stall():
+    state, detail = rh.clock_verdict(rows(STALLED))
+    assert state == rh.FAIL
+    assert "Midi-Bridge" in detail and "no hardware clock" in detail
+
+
+def test_nothing_scheduled_is_the_stall_too():
+    state, detail = rh.clock_verdict(rows(FROZEN))
+    assert state == rh.FAIL and "has no clock" in detail
+
+
+def test_no_rig_in_the_graph_is_not_a_failure():
+    assert rh.clock_verdict(rows(NO_RIG))[0] == rh.NA
+
+
+def test_a_suspended_app_is_not_running_so_not_judged():
+    lines = [HDR, "S  137      0      0    ---     ---   ---   ---     0                  ardour"]
+    assert rh.clock_verdict(rows(lines))[0] == rh.NA
+
+
+def test_clock_watch_judges_the_latest_complete_table():
+    t = [0.0]
+    cw = rh.ClockWatch(clock=lambda: t[0])
+    for ln in STALLED + [HDR]:                  # the next header completes the table
+        cw.feed(ln)
+    assert cw.verdict()[0] == rh.FAIL
+    for ln in HEALTHY[1:] + [HDR]:
+        cw.feed(ln)
+    assert cw.verdict()[0] == rh.OK
+
+
+def test_clock_watch_goes_quiet_when_pw_top_stops_not_green():
+    t = [0.0]
+    cw = rh.ClockWatch(clock=lambda: t[0])
+    for ln in HEALTHY + [HDR]:
+        cw.feed(ln)
+    assert cw.verdict()[0] == rh.OK
+    t[0] = 60.0
+    state, detail = cw.verdict()
+    assert state == rh.NA and "not reporting" in detail
+
+
+# --------------------------------------------------------------------------- #
+# the cheap probes
+# --------------------------------------------------------------------------- #
+def test_surface_ok_when_board_and_driver_port_both_present():
+    assert rh.surface_check(" 2 [L1] : USB-Audio - LCXL3 1", 'Port 0 : "ParVagues LCXL3"')[0] == rh.OK
+
+
+def test_surface_fails_when_board_present_and_driver_port_missing():
+    state, detail = rh.surface_check(" 2 [L1] : USB-Audio - LCXL3 1", 'Client 24 : "LCXL3 1"')
+    assert state == rh.FAIL and "no faders reach Ardour" in detail
+
+
+def test_surface_is_na_without_a_board():
+    assert rh.surface_check("0 [NVidia]", "")[0] == rh.NA
+
+
+def test_superdirt_dead_engine_under_a_live_sclang_fails():
+    assert rh.superdirt_check({"sclang": {"up": True}, "scsynth": {"up": False}})[0] == rh.FAIL
+    assert rh.superdirt_check({"sclang": {"up": True}, "scsynth": {"up": True}})[0] == rh.OK
+    assert rh.superdirt_check({"sclang": {"up": False}, "scsynth": {"up": False}})[0] == rh.NA
+    assert rh.superdirt_check(None)[0] == rh.NA
+
+
+# --------------------------------------------------------------------------- #
+# publishing
+# --------------------------------------------------------------------------- #
+class Rig:
+    def __init__(self, tmp_path):
+        self.t = 1000.0
+        self.notes, self.evidence = [], []
+        self.h = rh.Health(path=tmp_path / "health.json", notify=lambda *a: self.notes.append(a),
+                           evidence=lambda n, d: self.evidence.append((n, d)),
+                           now=lambda: self.t, async_evidence=False)
+
+    def tick(self, state, detail="d", check="audio_clock", n=1, dt=1.0):
+        out = []
+        for _ in range(n):
+            self.t += dt
+            out += self.h.update({check: (state, detail)})
+        return out
+
+
+def test_four_bad_ticks_are_a_blip_five_are_a_failure(tmp_path):
+    r = Rig(tmp_path)
+    r.tick(rh.OK)
+    assert r.tick(rh.FAIL, n=4) == []
+    assert r.h.state["audio_clock"]["state"] == rh.OK
+    ev = r.tick(rh.FAIL)
+    assert [(e["from"], e["to"]) for e in ev] == [(rh.OK, rh.FAIL)]
+    assert r.h.overall() == "down"
+
+
+def test_a_failure_notifies_once_and_captures_evidence_once(tmp_path):
+    r = Rig(tmp_path)
+    r.tick(rh.OK)
+    r.tick(rh.FAIL, "no hardware clock", n=30)
+    assert len(r.notes) == 1 and r.notes[0][2] is True
+    assert len(r.evidence) == 1 and r.evidence[0][1] == "no hardware clock"
+
+
+def test_recovery_needs_three_good_ticks_and_says_so(tmp_path):
+    r = Rig(tmp_path)
+    r.tick(rh.OK)
+    r.tick(rh.FAIL, n=6)
+    assert r.tick(rh.OK, n=2) == []
+    ev = r.tick(rh.OK)
+    assert [(e["from"], e["to"]) for e in ev] == [(rh.FAIL, rh.OK)]
+    assert r.notes[-1][0].endswith("recovered") and r.notes[-1][2] is False
+
+
+def test_a_flapping_check_does_not_spam_notifications(tmp_path):
+    r = Rig(tmp_path)
+    r.tick(rh.OK)
+    for _ in range(4):
+        r.tick(rh.FAIL, n=6)
+        r.tick(rh.OK, n=4)
+    assert len([n for n in r.notes if n[2] is True]) == 1       # inside the cooldown
+
+
+def test_never_opens_with_a_failure_verdict(tmp_path):
+    """The first sample after start can be mid-boot: it must not shout."""
+    r = Rig(tmp_path)
+    assert r.tick(rh.FAIL) == []
+    assert r.h.state["audio_clock"]["state"] == rh.NA
+
+
+def test_published_file_is_atomic_json_with_a_heartbeat(tmp_path):
+    r = Rig(tmp_path)
+    r.tick(rh.OK, "fine")
+    doc = json.loads((tmp_path / "health.json").read_text())
+    assert doc["overall"] == "ok" and doc["checks"]["audio_clock"]["state"] == rh.OK
+    t0 = doc["t"]
+    r.tick(rh.OK, "fine", n=7)                       # no change, but the heartbeat moves
+    assert json.loads((tmp_path / "health.json").read_text())["t"] > t0
+
+
+def test_reader_never_calls_missing_or_stale_health_ok(tmp_path):
+    p = tmp_path / "health.json"
+    assert rh.read_health(p)["overall"] == "unknown"
+    p.write_text(json.dumps({"v": 1, "t": 100.0, "overall": "ok", "checks": {}}))
+    assert rh.read_health(p, now=lambda: 105.0)["overall"] == "ok"
+    stale = rh.read_health(p, now=lambda: 500.0)
+    assert stale["overall"] == "stale" and "dead or wedged" in stale["detail"]
+
+
+def test_a_stopped_recorder_leaves_a_tombstone_not_a_green_file(tmp_path):
+    r = Rig(tmp_path)
+    r.tick(rh.OK)
+    r.h.close()
+    assert rh.read_health(tmp_path / "health.json", now=lambda: r.t)["overall"] == "unknown"
+
+
+def test_evidence_capture_writes_a_private_file_and_keeps_only_a_few(tmp_path):
+    for i in range(rh.EVIDENCE_KEEP + 3):
+        (tmp_path / f"2026010{i % 9}-{i:06d}-x.txt").write_text("old")
+    p = rh.capture_evidence("audio_clock", "stall", out_dir=tmp_path)
+    assert p and p.read_text().startswith("check: audio_clock")
+    assert oct(p.stat().st_mode & 0o777) == "0o600"
+    assert len(list(tmp_path.glob("*.txt"))) <= rh.EVIDENCE_KEEP