rig_health: Bluetooth is a sound card, a blind probe is amber, a dead recorder is not an alarm

A Bluetooth headset (bluez_output.*) now counts as a hardware clock, so
monitoring on it no longer raises a permanent critical. When pw-top goes quiet
the clock check is WARN, never NA: cannot-measure is a third state. 'Recovered'
is only said for a real recovery (not when the rig closed or the probe went
blind). A health.json left by a gig-log that died uncleanly reads as not
observed instead of aging into a red stale dot on every day it is not running.

tools/rig_health.py

@@ -64,6 +64,10 @@ EVIDENCE_COOLDOWN_S = 120.0
 EVIDENCE_KEEP = 10
 
 RIG_APPS = ("ardour", "SuperCollider")          # pw-top node names that make a rig
+# A driver that is a real sound device. A Bluetooth headset (bluez_output.<MAC>.1)
+# is one: treating only alsa_* as hardware would raise a permanent false FAIL the
+# moment the rig is monitored on the Bose.
+HW_CLOCK_PREFIXES = ("alsa_", "bluez_")
 STATES = {"R", "I", "S", "C", "D", "E", "!"}
 
 
@@ -112,7 +116,7 @@ def clock_verdict(rows: list[dict]) -> tuple[str, str]:
             if app["quant"] > 0:
                 continue
             bad.append(f"{app['name']} has no clock")
-        elif not d["name"].startswith("alsa_"):
+        elif not d["name"].startswith(HW_CLOCK_PREFIXES):
             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")
@@ -126,8 +130,11 @@ 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."""
 
+    BLIND_AFTER_S = 15.0    # longer than XrunReader's 3 s respawn, so a blip is not a blip
+
     def __init__(self, clock=time.monotonic):
         self._clock = clock
+        self._born = clock()
         self._rows: list[dict] = []
         self._seen_header = False
         self._lock = threading.Lock()
@@ -148,12 +155,20 @@ class ClockWatch:
             with self._lock:
                 self._rows.append(row)
 
-    def verdict(self, max_age: float = 5.0) -> tuple[str, str]:
+    def verdict(self) -> tuple[str, str]:
+        """The latest table's verdict. When pw-top has gone quiet the answer is WARN,
+        never NA: "cannot measure" is a third state, and a blind probe reading green
+        is the exact failure this module exists to prevent."""
         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"
+            at = self._at
+        now = self._clock()
+        if at is None:
+            if now - self._born < self.BLIND_AFTER_S:
+                return NA, "warming up"
+            return WARN, "pw-top has produced no table: the clock check is blind"
+        if now - at > self.BLIND_AFTER_S:
+            return WARN, "pw-top has gone quiet: the clock check is blind"
         return state, detail
 
 
@@ -314,7 +329,9 @@ class Health:
               "to": now_state, "detail": detail}
         if now_state == FAIL:
             self._on_fail(name, detail, t)
-        elif was == FAIL:
+        elif was == FAIL and now_state == OK:
+            # Only a real recovery says so. FAIL -> na (the rig closed) or FAIL -> warn
+            # (the probe went blind) is not "recovered", and saying it would be a lie.
             self._maybe_notify(name, f"{name} recovered", detail, urgent=False, t=t)
         return ev
 
@@ -363,7 +380,17 @@ class Health:
             pass
 
 
-def read_health(path: Path | None = None, now=time.time) -> dict:
+def observer_alive(pid) -> bool:
+    """Is that pid still a gig-log? Forkless: /proc only. Without it a recorder that
+    died uncleanly (a crash, a reboot) leaves a file that ages into a red "observer
+    stale" dot on every day gig-log is simply not running."""
+    try:
+        return "gig-log" in Path(f"/proc/{int(pid)}/cmdline").read_text(errors="replace")
+    except (OSError, ValueError, TypeError):
+        return False
+
+
+def read_health(path: Path | None = None, now=time.time, alive=observer_alive) -> dict:
     """The reader's view (tray, Bridge, gate): never raises, and a missing or
     stale file is reported as such, never as 'ok'."""
     try:
@@ -371,7 +398,10 @@ def read_health(path: Path | None = None, now=time.time) -> dict:
     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":
+    if doc.get("pid") is not None and not alive(doc["pid"]):
+        doc["detail"] = "gig-log is not running (the file is left over from a previous run)"
+        doc["overall"] = "unknown"
+    elif doc.get("overall") == "stopped":
         doc["detail"] = "gig-log stopped"
         doc["overall"] = "unknown"
     elif age > STALE_AFTER_S:

tools/tests/test_rig_health.py

@@ -104,15 +104,37 @@ def test_clock_watch_judges_the_latest_complete_table():
     assert cw.verdict()[0] == rh.OK
 
 
-def test_clock_watch_goes_quiet_when_pw_top_stops_not_green():
+def test_clock_watch_goes_amber_when_pw_top_stops_never_green():
+    """"Cannot measure" is a third state. A blind probe must not read as healthy."""
     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] = 10.0                                  # a respawn-sized gap is not a blip to flag
+    assert cw.verdict()[0] == rh.OK
     t[0] = 60.0
     state, detail = cw.verdict()
-    assert state == rh.NA and "not reporting" in detail
+    assert state == rh.WARN and "blind" in detail
+
+
+def test_a_clock_watch_that_never_gets_a_table_warms_up_then_warns():
+    t = [0.0]
+    cw = rh.ClockWatch(clock=lambda: t[0])
+    t[0] = 5.0
+    assert cw.verdict() == (rh.NA, "warming up")
+    t[0] = 20.0
+    assert cw.verdict()[0] == rh.WARN
+
+
+def test_a_bluetooth_headset_is_a_sound_card_not_a_stall():
+    """Monitoring on the Bose must not raise a permanent critical alert."""
+    bt = "bluez_output.70_26_05_D0_D5_90.1"
+    table = [HDR,
+             f"R  361   1024  48000   4.7ms  10.7us  0.22  0.00   83    S32LE 2 48000 {bt}",
+             "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"]
+    assert rh.clock_verdict(rows(table))[0] == rh.OK
 
 
 # --------------------------------------------------------------------------- #
@@ -265,3 +287,45 @@ def test_a_dead_observer_is_loud_but_an_absent_one_is_not():
 def test_degraded_names_the_warning_checks():
     s = rh.tray_summary(_doc("degraded", superdirt=(rh.WARN, "slow")))
     assert s["badge"] == "warn" and "superdirt" in s["row"]
+
+
+def test_only_a_real_recovery_says_recovered(tmp_path):
+    for to_state in (rh.WARN, rh.NA):            # went blind / the rig closed
+        r = Rig(tmp_path)
+        r.tick(rh.OK)
+        r.tick(rh.FAIL, n=6)
+        r.notes.clear()
+        r.tick(to_state, n=4)
+        assert not [n for n in r.notes if n[0].endswith("recovered")], to_state
+    r = Rig(tmp_path)
+    r.tick(rh.OK)
+    r.tick(rh.FAIL, n=6)
+    r.tick(rh.OK, n=3)
+    assert r.notes[-1][0].endswith("recovered")
+
+
+# --------------------------------------------------------------------------- #
+# the reader and a recorder that is gone
+# --------------------------------------------------------------------------- #
+def test_a_leftover_file_from_a_dead_recorder_is_not_an_alarm(tmp_path):
+    """After a crash or reboot the file ages. That must read as 'not observed' (no
+    badge), not as a red 'observer stale' dot on every day gig-log is not running."""
+    p = tmp_path / "health.json"
+    p.write_text(json.dumps({"v": 1, "t": 100.0, "pid": 424242, "overall": "ok", "checks": {}}))
+    h = rh.read_health(p, now=lambda: 99999.0, alive=lambda pid: False)
+    assert h["overall"] == "unknown" and "left over" in h["detail"]
+    assert rh.tray_summary(h)["badge"] is None
+
+
+def test_a_live_but_silent_recorder_is_stale_and_loud(tmp_path):
+    p = tmp_path / "health.json"
+    p.write_text(json.dumps({"v": 1, "t": 100.0, "pid": 1, "overall": "ok", "checks": {}}))
+    assert rh.read_health(p, now=lambda: 500.0, alive=lambda pid: True)["overall"] == "stale"
+    assert rh.read_health(p, now=lambda: 105.0, alive=lambda pid: True)["overall"] == "ok"
+
+
+def test_observer_alive_is_forkless_and_checks_the_command():
+    import os
+    assert rh.observer_alive(10**9) is False             # no such pid
+    assert rh.observer_alive(os.getpid()) is False       # alive, but not a gig-log
+    assert rh.observer_alive("not-a-pid") is False