fix(obs): detect activity by MOVEMENT, not by level — two wrong versions later

rig-pulse answers one question when the sound stops: did Tidal stop asking, or
did the engine stop playing? It got that wrong twice, in two different ways,
and both are the same mistake wearing different clothes.

v1 assumed a universal idle threshold: SOUND if synths > 0 or peak CPU > 1.5%.
But this rig idles at 71 synths / 3531 ugens / 17.7% peak — start_and_midi.scd
registers the Mutable-Instruments global effects on all 14 orbits and those
synths never die. So the tool reported SOUND continuously, straight through a
silence PLN was actively complaining about. It could not have fired. I shipped
a check that cannot fail one commit after writing a message about how a check
that never fires is indistinguishable from a check that passes.

v2 learned the floor as a running minimum. Also wrong, and only live data
showed it: the floor RISES as more of the graph gets instantiated — 71 synths
at 15:02, 103 at 15:05. A baseline that only ratchets down cannot track a rig
that ratchets up, so everything above the all-time minimum looks like activity
forever.

What separates the states is stillness, and the data said so plainly: an idle
SuperDirt returned ugens=3531 five times over eight seconds — exactly, not
approximately — while a played rig moved 4144 -> 4300 in two seconds. So
compare the last few samples to each other and ignore the level entirely. That
is immune to the floor changing between boots or during one.

Known limit, in the header rather than hidden: a perfectly static drone with a
constant voice count would read SILENT. Tidal events are discrete and the count
moves, so it holds for how this rig is played.

Verified against both states: frozen at 103 synths / 4144 ugens -> SILENT;
moving 4066-4222 -> SOUND.

tools/rig-pulse.py

@@ -42,10 +42,70 @@ 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
+# There is no universal idle threshold, and assuming one is how the first
+# version of this tool shipped broken. It called SOUND on `synths > 0 or peak >
+# 1.5%` — but THIS rig idles at 71 synths / 3531 ugens / 17.7% peak, because
+# start_and_midi.scd registers the Mutable-Instruments global effects on every
+# one of 14 orbits and those synths live forever. So the tool read SOUND
+# continuously, including while PLN sat in silence, and would never once have
+# fired. A check that cannot fail is not a check.
+#
+# The floor is a property of the boot, not a constant, so LEARN it: track the
+# running minimum and treat anything above it as activity. Tidal events add
+# synths and ugens on top of the permanent graph; an idle rig sits perfectly
+# still (five samples over 8 s gave ugens=3531 every single time, zero
+# variance). That stillness is the signal.
+# Activity is detected by VARIANCE, not by level — and this took two wrong
+# versions to get right.
+#
+# v1 assumed a universal idle threshold ("synths > 0 or peak > 1.5%"). But this
+# rig idles at 71 synths / 3531 ugens / 17.7% peak, because start_and_midi.scd
+# registers the Mutable-Instruments global effects on all 14 orbits and those
+# synths live forever. The tool read SOUND continuously, including through a
+# silence PLN was actively complaining about, and could never once have fired.
+#
+# v2 learned the floor as a running minimum. Also wrong: the floor RISES as more
+# of the graph gets instantiated — measured 71 synths at 15:02 and 103 at 15:05
+# — so an all-time minimum leaves everything above it looking like activity
+# forever. A baseline that only ever ratchets down cannot track a rig that
+# ratchets up.
+#
+# What actually separates the two states is stillness. An idle SuperDirt is
+# EXACTLY constant: five samples over 8 s returned ugens=3531 every single time,
+# not approximately. Tidal events spawn and free synths continuously, so a rig
+# being played jitters sample to sample (4300 vs 4144 across two seconds). So:
+# no change across the window means nobody is asking, whatever the level is.
+#
+# Known limit, stated rather than hidden: a perfectly static drone with a
+# constant voice count would read SILENT. Tidal events are discrete and the
+# count moves, so this holds for how this rig is actually played.
+WINDOW = 5          # samples compared for movement
+
+
+class Activity:
+    """Is the synth graph MOVING? Level-independent, so it survives a rig whose
+    idle baseline changes between boots — or during one."""
+
+    def __init__(self, window: int = WINDOW):
+        self.window = window
+        self.recent: list[tuple[int, int]] = []
+
+    def update(self, ugens: int, synths: int):
+        self.recent.append((ugens, synths))
+        if len(self.recent) > self.window:
+            self.recent.pop(0)
+
+    def moving(self) -> bool | None:
+        """True/False, or None while still filling the window."""
+        if len(self.recent) < min(3, self.window):
+            return None
+        return len(set(self.recent)) > 1
+
+    def describe(self) -> str:
+        if not self.recent:
+            return "no samples"
+        u = [r[0] for r in self.recent]
+        return f"ugens {min(u)}-{max(u)} over last {len(u)} samples"
 
 
 def pad(b: bytes) -> bytes:
@@ -101,7 +161,7 @@ class Probe:
         return None
 
     def status(self):
-        """(synths, avg_cpu, peak_cpu) or None if scsynth is not answering.
+        """(ugens, synths, avg_cpu, peak_cpu) or None if scsynth is not answering.
 
         /status.reply args: unused, nUGens, nSynths, nGroups, nSynthDefs,
         avgCPU, peakCPU, sampleRate, actualSampleRate.
@@ -109,21 +169,26 @@ class Probe:
         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])
+        return int(v[1]), 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]:
+def verdict(st, sclang_ok, act: Activity) -> 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"
+    ugens, synths, avg, peak = st
+    act.update(ugens, synths)
+    m = act.moving()
+    if m is None:
+        return "WARMUP", f"{synths} synths / {ugens} ugens — filling the window"
+    if m:
+        return "SOUND", f"{synths} synths / {ugens} ugens, peak {peak:.1f}%"
+    return "SILENT", (f"graph frozen at {synths} synths / {ugens} ugens "
+                      f"(peak {peak:.1f}%) — engine healthy, nobody is asking")
 
 
 def main():
@@ -135,23 +200,30 @@ def main():
 
     p = Probe()
 
+    act = Activity()
     if a.once:
+        # One reading cannot show movement, so --once takes the window it needs.
+        for _ in range(WINDOW):
             st = p.status()
-        v, why = verdict(st, p.sclang_alive())
+            if st is None:
+                break
+            act.update(st[0], st[1])
+            time.sleep(0.4)
+        v, why = verdict(st, p.sclang_alive(), act)
         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")
+    print("activity = MOVEMENT in the synth graph, level-independent — see the header\n")
 
     prev = None
     t0 = time.time()
-    counts = {"SOUND": 0, "SILENT": 0, "DOWN": 0}
+    counts = {"SOUND": 0, "SILENT": 0, "DOWN": 0, "WARMUP": 0}
     transitions = []
     try:
         while True:
             st = p.status()
-            v, why = verdict(st, p.sclang_alive() if st is None else True)
+            v, why = verdict(st, p.sclang_alive() if st is None else True, act)
             counts[v] += 1
             if v != prev:
                 line = f"  {datetime.now():%H:%M:%S}  {v:<6} {why}"
@@ -167,9 +239,10 @@ def main():
     dur = time.time() - t0
     tot = sum(counts.values()) or 1
     print(f"\n=== {dur:.0f}s ===")
-    for k in ("SOUND", "SILENT", "DOWN"):
+    for k in ("SOUND", "SILENT", "DOWN", "WARMUP"):
         print(f"  {k:<7} {counts[k]:5d} samples  {100*counts[k]/tot:5.1f}%")
     print(f"  transitions: {len(transitions)}")
+    print(f"  window: {act.describe()}")
     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.")