feat(obs): ask the synth whether the rig is making sound

Companion to audio-lens.py: that watches the audio DEVICE, this watches the
audio ENGINE. Together they answer, in one line, the question that cost PLN
three interruptions today — when the sound stops, did Tidal stop asking, or did
the interface stop playing?

scsynth already answers /status on the port it is listening on, and the reply
carries the live synth count and peak DSP load. SuperDirt rendering silence
sits at its idle floor with no synths; SuperDirt fed by Tidal spawns a synth
per event and the load moves. So three states, and the middle one is the point:

  synths > 0 or peak above the floor    -> Tidal is emitting
  answers but flat at the floor         -> engine fine, NOBODY IS ASKING
  no answer                             -> the server is gone

Nothing in this repo could produce that middle verdict before. During today's
stop every existing lens reported healthy — sclang's OSC loop answered, scsynth
answered, all 48 SuperCollider->Ardour links were up, the interface clocked a
steady 48 kHz — and the sound was gone anyway, because Tidal had stopped
emitting while its process stayed alive, threads scheduled, port 6010 held.

Written with the offsets bug already paid for. The first throwaway version of
this query hardcoded byte offsets into /status.reply and reported 1.68 BILLION
ugens and 0.0% CPU, and I read the CPU figure off it and told PLN the engine was
rendering silence. It was not: reading the typetag properly gives 91 synths at
30.7% peak. A number obviously wrong enough to notice sat right next to a number
wrong enough to mislead, and I used the second one. Hence osc_parse walks the
typetag, and the absurd-value story is in the docstring so the next reader
distrusts hand-rolled offsets on sight.

Passive: /status is a query, allocates nothing, never touches the default
server object, never plays a tone.

tools/rig-pulse.py

+#!/usr/bin/env python3
+"""Is the rig MAKING SOUND right now? Ask the synth, not the stack.
+
+Companion to audio-lens.py. That one watches the audio DEVICE; this one watches
+the audio ENGINE. Between them they answer, in one line, the question that cost
+PLN three interruptions on 2026-08-14: when the sound stops, is it Tidal that
+stopped asking, or the interface that stopped playing?
+
+The trick is that scsynth will tell you, for free, over the same UDP port it
+already listens on. `/status` returns a `/status.reply` carrying the live synth
+count and the peak DSP load. A SuperDirt rendering silence sits at its idle
+floor with zero synths; a SuperDirt being fed by Tidal spawns a synth per event
+and the peak load moves. So:
+
+    synths > 0 or peak CPU above the floor   ->  Tidal is emitting
+    answers /status but flat at the floor    ->  engine fine, NOBODY IS ASKING
+    does not answer /status                  ->  the server itself is gone
+
+That middle line is the one no other check in this repo could produce. On
+2026-08-14 every lens said healthy — sclang's OSC loop answered, scsynth
+answered, all 48 SuperCollider->Ardour links were up, the interface was clocking
+at 48 kHz — and the sound was still gone, because Tidal had quietly stopped
+emitting while its process stayed alive. This tool names that state.
+
+It is passive and read-only: /status is a query, it allocates nothing, and it
+goes to scsynth's existing port. It never touches the default server object and
+never plays a tone.
+
+  tools/rig-pulse.py                 watch until Ctrl-C
+  tools/rig-pulse.py --watch 600     watch 10 min, then summarise
+  tools/rig-pulse.py --once          one reading
+"""
+from __future__ import annotations
+
+import argparse
+import socket
+import struct
+import sys
+import time
+from datetime import datetime
+
+SCSYNTH = ("127.0.0.1", 57110)
+SCLANG = ("127.0.0.1", 57120)
+
+# Below this peak DSP load, with no synths, the server is rendering silence.
+# SuperDirt's own idle graph (the global effects registered on every orbit) is
+# what keeps this above exactly zero.
+IDLE_PEAK_CPU = 1.5
+
+
+def pad(b: bytes) -> bytes:
+    return b + b"\0" * ((4 - len(b) % 4) % 4)
+
+
+def osc_msg(addr: str) -> bytes:
+    return pad(addr.encode() + b"\0") + pad(b",\0")
+
+
+def osc_parse(data: bytes):
+    """(address, [args]) for a reply. Reads the typetag rather than assuming
+    offsets — the first version of this hardcoded them and produced a synth
+    count of 1684275200, a number obviously wrong enough to catch, which is
+    luckier than it deserved."""
+    def take_str(b, i):
+        e = b.index(b"\0", i)
+        return b[i:e].decode("utf-8", "replace"), i + (((e - i) + 4) // 4) * 4
+    addr, i = take_str(data, 0)
+    if i >= len(data):
+        return addr, []
+    tags, i = take_str(data, i)
+    out = []
+    for t in tags[1:]:
+        if t == "i":
+            out.append(int.from_bytes(data[i:i + 4], "big", signed=True)); i += 4
+        elif t == "f":
+            out.append(struct.unpack(">f", data[i:i + 4])[0]); i += 4
+        elif t == "d":
+            out.append(struct.unpack(">d", data[i:i + 8])[0]); i += 8
+        elif t == "s":
+            s, i = take_str(data, i); out.append(s)
+    return addr, out
+
+
+class Probe:
+    def __init__(self, timeout=1.5):
+        self.s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.s.bind(("127.0.0.1", 0))
+        self.s.settimeout(timeout)
+
+    def call(self, addr: str, target, expect: str):
+        try:
+            self.s.sendto(osc_msg(addr), target)
+            deadline = time.time() + self.s.gettimeout()
+            while time.time() < deadline:
+                data, _ = self.s.recvfrom(65535)
+                a, vals = osc_parse(data)
+                if a == expect:
+                    return vals
+        except (socket.timeout, OSError):
+            pass
+        return None
+
+    def status(self):
+        """(synths, avg_cpu, peak_cpu) or None if scsynth is not answering.
+
+        /status.reply args: unused, nUGens, nSynths, nGroups, nSynthDefs,
+        avgCPU, peakCPU, sampleRate, actualSampleRate.
+        """
+        v = self.call("/status", SCSYNTH, "/status.reply")
+        if not v or len(v) < 7:
+            return None
+        return int(v[2]), float(v[5]), float(v[6])
+
+    def sclang_alive(self) -> bool:
+        return self.call("/pv/ping", SCLANG, "/pv/pong") is not None
+
+
+def verdict(st, sclang_ok) -> tuple[str, str]:
+    if st is None:
+        return "DOWN", ("scsynth is not answering /status — the audio server is gone"
+                        if sclang_ok else
+                        "neither scsynth nor sclang answering — the rig is down")
+    synths, avg, peak = st
+    if synths > 0 or peak > IDLE_PEAK_CPU:
+        return "SOUND", f"{synths} synths, peak {peak:.1f}%"
+    return "SILENT", f"engine healthy but idle ({synths} synths, peak {peak:.1f}%) — nobody is asking"
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--watch", type=float, default=0)
+    ap.add_argument("--once", action="store_true")
+    ap.add_argument("--interval", type=float, default=2.0)
+    a = ap.parse_args()
+
+    p = Probe()
+
+    if a.once:
+        st = p.status()
+        v, why = verdict(st, p.sclang_alive())
+        print(f"{v}: {why}")
+        return 0 if v != "DOWN" else 1
+
+    print("rig-pulse — passive /status polling of scsynth (no tones, no allocation)")
+    print(f"idle-floor threshold: peak CPU {IDLE_PEAK_CPU}%\n")
+
+    prev = None
+    t0 = time.time()
+    counts = {"SOUND": 0, "SILENT": 0, "DOWN": 0}
+    transitions = []
+    try:
+        while True:
+            st = p.status()
+            v, why = verdict(st, p.sclang_alive() if st is None else True)
+            counts[v] += 1
+            if v != prev:
+                line = f"  {datetime.now():%H:%M:%S}  {v:<6} {why}"
+                print(line, flush=True)
+                transitions.append(line)
+                prev = v
+            if a.watch and time.time() - t0 >= a.watch:
+                break
+            time.sleep(a.interval)
+    except KeyboardInterrupt:
+        print()
+
+    dur = time.time() - t0
+    tot = sum(counts.values()) or 1
+    print(f"\n=== {dur:.0f}s ===")
+    for k in ("SOUND", "SILENT", "DOWN"):
+        print(f"  {k:<7} {counts[k]:5d} samples  {100*counts[k]/tot:5.1f}%")
+    print(f"  transitions: {len(transitions)}")
+    if counts["SILENT"] and not counts["DOWN"]:
+        print("\n  The engine was healthy while silent for part of this window.\n"
+              "  That is Tidal not emitting — look UP the stack, not down.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())