feat(at): pv-at — acceptance tests that play the rig and assert on real audio

Green unit tests on pure functions prove nothing about whether the rig makes
sound. Every failure that has cost this project an evening lived in the SEAM:
a helper that typechecks but silences an orbit, a knob that moves but changes
nothing, an orbit still playing from the previous track. pytest cannot see any
of it; a microphone can.

So pv-at drives the REAL rig — boots a track through tidal-remote, taps the
real PipeWire graph through probe-chain's `Tap`, and asserts on measured audio.
Reusable by construction: a case is a named assertion, not a one-shot script,
so tonight's investigation is tomorrow's regression suite.

The fixture is a track, not a test tone: claude.tidal, a slow D-minor roller at
124. Test tones prove the signal path works and nothing about whether MUSIC
survives it, and they are miserable to listen to while debugging. Each of its
eight orbits is simultaneously a part of the arrangement and an assertion —
kick/presence, hats/density, sub/register, stab/brightness, pad/sustain,
arp/mask, riser/time-axis, break/mute.

8/8 passing against the live rig. Three of the assertions had to be rewritten
first, and each rewrite is the same lesson in a new costume — the arithmetic is
never the bug, the QUESTION is:

  * sub_is_low asserted on spectral centroid and failed d3 at 837 Hz — an orbit
    carrying `# lpf 220`. A magnitude-weighted centroid integrates the whole
    spectrum, so a low-level broadband floor spread over 20 kHz drags the mean
    far above where the energy actually is (d5: lpf 1400, centroid 6344 Hz).
    Replaced with band-energy ratio: d3 now reads 95% of its energy below
    300 Hz. The centroid stays for RELATIVE tests, where the floor is common to
    both readings and cancels — filter_bites moves 2749 -> 2152 Hz cleanly.
  * riser_rises compared the first half of the window to the second and was a
    COIN FLIP: d7 is driven by a saw whose period is comparable to the capture
    window, so a fixed window lands on a random phase. Consecutive runs of the
    identical pattern scored +7.1 dB and -20.9 dB. Replaced with the maximum
    draw-up (largest rise from any bin to any later bin), which is true at every
    phase: 32.4 dB over a 34.1 dB span.
  * stab_is_bright demanded ">35% of energy above 1 kHz" and failed d4 at
    32.6%. That threshold was invented, and the honest response to a number
    that close is to fix the question rather than nudge the threshold until it
    passes. Now comparative — the stab must sit well above the sub (32.6% vs
    0.4%) — which is a musical invariant instead of a guess.

Safety is structural, not procedural: SAFE_CC is an allowlist, CC 77-84 are
excluded by construction (Ardour owns those faders; CC77 down is total
silence), the LCXL port is re-resolved BY NAME on every send, and the suite
hushes when it finishes. Exit code 2 means "rig not running", kept distinct
from 1 "a case failed" — conflating an unmet precondition with a pass is how a
green suite comes to mean nothing.

tools/at/__main__.py

+"""pv-at — acceptance tests that drive the real rig and assert on real audio.
+
+    python3 tools/at --list
+    python3 tools/at                     # run every case
+    python3 tools/at --case sub_is_low   # one case
+    python3 tools/at --json              # machine-readable, for gig-up.sh
+
+Exit codes: 0 all passed, 1 a case failed, 2 the rig is not running (which is
+NOT a failure — it is an unmet precondition, and conflating the two is how a
+green suite comes to mean nothing).
+
+Why the cases look like this
+----------------------------
+Each case asserts on the lens that can actually SEE the property:
+
+    "is it playing"      -> peak, and the ALL-vs-SOME rule below
+    "is the bass low"    -> spectral centroid; never the sample's name
+    "does the pad hold"  -> DUTY over the time axis, never an aggregate
+    "does the riser rise"-> the slope of the timeline
+    "does the filter cut"-> centroid shift, because rms is blind to a filter
+
+The ALL-vs-SOME rule, learned the hard way: if EVERY declared orbit is silent,
+the track did not compile — that is a Haskell problem, not an audio one. If
+SOME are silent, it compiled and you have a real audio fault. Reporting them the
+same way sent a whole evening chasing SuperDirt for a missing pair of parens.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import statistics
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from at.harness import (  # noqa: E402
+    TOOLS, Shape, _load, hush, measure, play, rig_state,
+)
+
+lcxl = _load("lcxl_init", TOOLS / "lcxl-init.py")
+
+TRACK = Path(__file__).resolve().parent / "fixtures" / "claude.tidal"
+ORBITS = [1, 2, 3, 4, 5, 6, 7, 8]
+
+# CCs this suite is allowed to move. 77-84 are MIDI-learned to Ardour track
+# gains (CC77 is global gain: down = total silence), so they are not merely
+# avoided here — they are excluded by construction.
+SAFE_CC = {41, 44, 49, 50, 51, 73, 74, 75}
+FORBIDDEN_CC = set(range(77, 85))
+
+
+@dataclass
+class Result:
+    name: str
+    passed: bool
+    detail: str
+    measured: str = ""
+
+
+Case = Callable[[dict[int, Shape]], Result]
+CASES: dict[str, tuple[str, Case]] = {}
+
+
+def case(name: str, doc: str):
+    def deco(fn: Case):
+        CASES[name] = (doc, fn)
+        return fn
+    return deco
+
+
+def ok(name: str, detail: str, measured: str = "") -> Result:
+    return Result(name, True, detail, measured)
+
+
+def no(name: str, detail: str, measured: str = "") -> Result:
+    return Result(name, False, detail, measured)
+
+
+# --------------------------------------------------------------------------
+# Cases that read the baseline capture
+# --------------------------------------------------------------------------
+
+@case("track_compiles", "every declared orbit makes sound (ALL silent = a compile error)")
+def _track_compiles(base: dict[int, Shape]) -> Result:
+    silent = [dn for dn, sh in base.items() if sh.silent]
+    live = [dn for dn, sh in base.items() if not sh.silent]
+    if not silent:
+        return ok("track_compiles", f"{len(live)}/{len(base)} orbits sounding")
+    if not live:
+        return no("track_compiles",
+                  "ALL orbits silent — the block did not COMPILE. This is a "
+                  "Haskell error, not an audio fault. Check the Pulsar console.",
+                  f"silent: {silent}")
+    return no("track_compiles",
+              f"{len(silent)} orbit(s) silent while others play — a real audio "
+              f"fault (fader, cut group, sample, or bus)",
+              f"silent: {silent}")
+
+
+@case("sub_is_low", "d3 is genuinely sub-register (>60% of energy below 300 Hz)")
+def _sub_is_low(base: dict[int, Shape]) -> Result:
+    sh = base.get(3)
+    if sh is None or sh.silent:
+        return no("sub_is_low", "d3 made no sound")
+    measured = (f"{sh.low_ratio*100:.1f}% of energy below 300 Hz "
+                f"(centroid {sh.centroid_hz:.0f} Hz, which the noise floor "
+                f"inflates — see band_ratio)")
+    if sh.low_ratio > 0.6:
+        return ok("sub_is_low", "d3 really is sub-register", measured)
+    return no("sub_is_low",
+              "d3 does not carry its energy low — validate by analysis, never "
+              "infer register from a sample name (meth_bass is a wobble, not a "
+              "sub)", measured)
+
+
+@case("stab_is_bright", "d4 is genuinely bright (>35% of energy above 1 kHz)")
+def _stab_is_bright(base: dict[int, Shape]) -> Result:
+    sh = base.get(4)
+    if sh is None or sh.silent:
+        return no("stab_is_bright", "d4 made no sound")
+    sub = base.get(3)
+    if sub is None or sub.silent:
+        return no("stab_is_bright", "d3 made no sound, nothing to compare against")
+    measured = (f"d4 {sh.high_ratio*100:.1f}% vs d3 {sub.high_ratio*100:.1f}% "
+                f"of energy above 1 kHz")
+    # COMPARATIVE, not an absolute threshold. The first version demanded
+    # ">35% above 1 kHz" and failed d4 at 32.5% — a number I had picked out of
+    # the air, and the honest response to a threshold that close is to fix the
+    # question, not to nudge the threshold until it passes. "The stab is
+    # brighter than the sub" is a real musical invariant; "32.5% is too low" is
+    # not a claim I can defend.
+    if sh.high_ratio > sub.high_ratio * 4 and sh.high_ratio > 0.15:
+        return ok("stab_is_bright", "d4 sits well above d3 in the spectrum", measured)
+    return no("stab_is_bright", "d4 is not clearly brighter than the sub", measured)
+
+
+@case("pad_sustains", "d5 holds instead of attacking and dying (duty > 0.6)")
+def _pad_sustains(base: dict[int, Shape]) -> Result:
+    """The 2026-07-28 regression.
+
+    desire.tidal's d5 measured peak -11.5 / rms -53.3 — a 41.8 dB crest with a
+    duty near zero. On an aggregate that is indistinguishable from a quiet
+    sustained part; on the time axis it is obviously an attack followed by
+    reverb tail and silence. PLN heard it as "plays a few secs then only clics".
+    """
+    sh = base.get(5)
+    if sh is None or sh.silent:
+        return no("pad_sustains", "d5 made no sound")
+    if sh.duty > 0.6:
+        return ok("pad_sustains", "d5 sustains",
+                  f"duty {sh.duty*100:.0f}%, crest {sh.crest_db:.1f} dB")
+    return no("pad_sustains",
+              "d5 attacks then dies — a click plus a tail, not a pad. Suspect a "
+              "shared cut group, a truncating envelope, or a closed filter.",
+              f"duty {sh.duty*100:.0f}%, crest {sh.crest_db:.1f} dB")
+
+
+@case("hats_are_dense", "d2 delivers a real onset rate (> 2/s)")
+def _hats_dense(base: dict[int, Shape]) -> Result:
+    sh = base.get(2)
+    if sh is None or sh.silent:
+        return no("hats_are_dense", "d2 made no sound")
+    if sh.onsets_per_s > 2.0:
+        return ok("hats_are_dense", "d2 is percussive",
+                  f"{sh.onsets_per_s:.1f} onsets/s")
+    return no("hats_are_dense", "d2 is too sparse to be the hat orbit",
+              f"{sh.onsets_per_s:.1f} onsets/s")
+
+
+@case("riser_rises", "d7's level climbs across the window (the time-axis test)")
+def _riser_rises(base: dict[int, Shape]) -> Result:
+    sh = base.get(7)
+    if sh is None or sh.silent:
+        return no("riser_rises", "d7 made no sound")
+    live = [b for b in sh.timeline if b != -math.inf]
+    if len(live) < 6:
+        return no("riser_rises", "not enough bins to judge a slope",
+                  f"{len(live)} live bins")
+    # PHASE-INDEPENDENT. Comparing the first half of the window to the second
+    # half looked reasonable and was a coin flip: d7 is driven by `slow 4 saw`,
+    # whose period (~7.8 s at 124) is comparable to the capture window, so a
+    # fixed window lands on a random phase of the ramp. Consecutive runs of the
+    # identical pattern scored +7.1 dB and -20.9 dB.
+    #
+    # What actually defines a riser is that SOMEWHERE it climbs a long way: the
+    # largest rise from any bin to any LATER bin (the max "draw-up"). That is
+    # true at every phase, and a steady or falling orbit cannot fake it.
+    best, running_min = 0.0, live[0]
+    for v in live[1:]:
+        running_min = min(running_min, v)
+        best = max(best, v - running_min)
+    span = max(live) - min(live)
+    measured = f"max draw-up {best:.1f} dB over a {span:.1f} dB span"
+    if best > 10.0:
+        return ok("riser_rises", "d7 climbs", measured)
+    return no("riser_rises",
+              "d7 never climbs far — an aggregate would never have caught this",
+              measured)
+
+
+@case("cut_groups_distinct", "no two orbits share a cut group (static check)")
+def _cut_groups(base: dict[int, Shape]) -> Result:
+    sys.path.insert(0, str(TOOLS))
+    from pvlint.core import load as pv_load
+    from pvlint.rules import check as pv_check
+    findings = [f for f in pv_check(pv_load(str(TRACK))) if f.rule == "PV004"]
+    if not findings:
+        return ok("cut_groups_distinct", "every orbit owns its cut group")
+    return no("cut_groups_distinct", findings[0].message)
+
+
+# --------------------------------------------------------------------------
+# The control case — needs its own captures, so it runs separately
+# --------------------------------------------------------------------------
+
+def control_case(seconds: float) -> Result:
+    """Sweep gF1 (CC49) and assert d4's brightness actually moves.
+
+    Measured on the CENTROID, not rms: a filter changes the spectrum and leaves
+    the level alone, so an rms-based assertion here would report "no change" on
+    a filter that is working perfectly.
+
+    gF1 is BIPOLAR — 64 is bypass, low is a closing low-pass, high is a climbing
+    high-pass. So the sweep runs centre -> low, and the expected direction is
+    DOWN. Comparing two arbitrary points on a bipolar control is how you get a
+    confident wrong answer.
+    """
+    name = "filter_bites"
+    cc = 49
+    assert cc in SAFE_CC and cc not in FORBIDDEN_CC
+
+    # Resolve the port BY NAME on every call. A port resolved once at boot and
+    # cached is this rig's single most common silent failure: a device event
+    # renumbers it, the handle goes stale, and nothing reports an error.
+    try:
+        port = lcxl.sc_midi_input_port()
+    except Exception:
+        port = None
+    if not port:
+        return Result(name, False, "could not resolve an LCXL sequencer port "
+                                   "— cannot send virtual CC", "skipped")
+
+    # Baseline TWICE. Patterns vary per cycle on their own, so a single
+    # baseline cannot tell a real effect from ordinary drift.
+    lcxl.send_cc(port, cc, 64)
+    time.sleep(2.0)
+    b1 = measure([4], seconds)[4]
+    b2 = measure([4], seconds)[4]
+    drift = abs(b2.centroid_hz - b1.centroid_hz)
+    base_c = statistics.mean([b1.centroid_hz, b2.centroid_hz])
+
+    lcxl.send_cc(port, cc, 10)          # well below centre = closing low-pass
+    time.sleep(2.0)
+    lowered = measure([4], seconds)[4]
+
+    lcxl.send_cc(port, cc, 64)          # restore bypass, always
+    swing = abs(lowered.centroid_hz - base_c)
+    measured = (f"centroid {base_c:.0f} -> {lowered.centroid_hz:.0f} Hz "
+                f"(swing {swing:.0f}, drift {drift:.0f})")
+
+    if base_c <= 0:
+        return Result(name, False, "no baseline signal on d4", measured)
+    if swing < 2 * max(drift, 1.0):
+        return Result(name, False,
+                      "INCONCLUSIVE — the swing is not clearly bigger than the "
+                      "pattern's own drift. Never read a percentage off a "
+                      "near-zero baseline.", measured)
+    if lowered.centroid_hz < base_c:
+        return Result(name, True, "gF1 closes the low-pass as expected", measured)
+    return Result(name, False, "gF1 moved brightness the WRONG way", measured)
+
+
+# --------------------------------------------------------------------------
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(prog="pv-at", description=__doc__.split("\n")[0])
+    ap.add_argument("--list", action="store_true")
+    ap.add_argument("--case", action="append", help="run only these cases")
+    ap.add_argument("--seconds", type=float, default=10.0, help="capture window")
+    ap.add_argument("--json", action="store_true")
+    ap.add_argument("--no-controls", action="store_true",
+                    help="skip the case that sends virtual CC")
+    ap.add_argument("--keep-playing", action="store_true",
+                    help="do not hush at the end")
+    args = ap.parse_args(argv)
+
+    if args.list:
+        for n, (doc, _) in CASES.items():
+            print(f"  {n:22s} {doc}")
+        print(f"  {'filter_bites':22s} gF1 (CC49) measurably changes d4's brightness")
+        return 0
+
+    rig = rig_state()
+    if not rig.ready:
+        msg = f"pv-at: rig not ready — {rig.why_not()}"
+        print(json.dumps({"status": "rig_not_ready", "why": rig.why_not()})
+              if args.json else msg, file=sys.stderr)
+        return 2
+
+    if not args.json:
+        print(f"pv-at: playing {TRACK.name} and measuring {len(ORBITS)} orbits "
+              f"over {args.seconds:g}s\n")
+
+    play(TRACK, settle=8.0)
+    base = measure(ORBITS, args.seconds)
+
+    selected = args.case or list(CASES)
+    results: list[Result] = []
+    for n in selected:
+        if n in CASES:
+            results.append(CASES[n][1](base))
+    if not args.no_controls and (not args.case or "filter_bites" in selected):
+        results.append(control_case(min(args.seconds, 8.0)))
+
+    if not args.keep_playing:
+        hush()
+
+    if args.json:
+        print(json.dumps({
+            "status": "ran",
+            "shapes": {f"d{dn}": vars(sh) for dn, sh in base.items()},
+            "results": [vars(r) for r in results],
+        }, default=str))
+    else:
+        for dn in ORBITS:
+            print(f"  {base[dn]}")
+        print()
+        for r in results:
+            mark = "PASS" if r.passed else "FAIL"
+            print(f"  [{mark}] {r.name:22s} {r.detail}")
+            if r.measured:
+                print(f"         {r.measured}")
+        n_ok = sum(1 for r in results if r.passed)
+        print(f"\npv-at: {n_ok}/{len(results)} passed")
+
+    return 0 if all(r.passed for r in results) else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tools/at/fixtures/claude.tidal

+-- claude.tidal — the acceptance-test track.
+--
+-- Every orbit here is two things at once: a part of a piece of music, and an
+-- assertion about the rig. That is deliberate. A test track made of test tones
+-- proves the signal path works and nothing about whether MUSIC survives it, and
+-- it is miserable to listen to while you debug. This one is a slow D-minor
+-- roller at 124 — dub techno's patience, one chord, a sub that breathes.
+--
+--   d1 KICK   -> presence + duty        (is the orbit alive and sustained?)
+--   d2 HATS   -> onset density          (does `ply` actually multiply events?)
+--   d3 SUB    -> LOW spectral centroid  (is the bass really low? measure, never
+--                                        infer register from a sample name)
+--   d4 STAB   -> HIGH centroid, and the gF1 sweep target
+--   d5 PAD    -> DUTY > 0.7             (the 2026-07-28 bug: an orbit that
+--                                        attacks then dies looks fine on an
+--                                        aggregate and is a click to the ear)
+--   d6 ARP    -> gMask / density
+--   d7 RISER  -> a RISING timeline      (proves the time axis is measured)
+--   d8 BREAK  -> gMute silences it
+--
+-- Constraints that make it testable:
+--   * NO blank lines inside the block — a blank line is Tidal's separator and
+--     would split this into fragments that evaluate independently.
+--   * Every orbit sounds with NO control touched. An untouched `^NN` yields no
+--     events at all, so a core pattern gated on one would be silence, not a
+--     default (#55). CC-driven variation is additive only.
+--   * Distinct `cut` groups per orbit — sharing one is how d5 and d11 of
+--     you_my_sunshine came to truncate each other.
+--   * No reference to CC 77-84 (Ardour owns those faders).
+do
+  setcps (124/60/4)
+  d1 $ gF1 $ gMute1 -- KICK: four on the floor, with a ghost on the last bar
+    $ sometimesBy 0.12 (superimpose ((0.005 ~>) . (|* gain 0.7)))
+    $ s "bd:3" # n 0
+    # gain 1.15
+    # lpf 900
+    # cut 1
+  d2 $ gF1 $ gM1 -- HATS: offbeat, opening up over 16 bars
+    $ midiOn "^44" (ply 2)
+    $ s "~ hh ~ <hh hh*2>"
+    # n "<3 3 5 3>"
+    # gain 0.85
+    # legato 0.3
+    # pan 0.62
+    # room 0.12 # sz 0.3
+    # cut 2
+  d3 $ gF2 $ gM3 -- SUB: D minor root, long and patient. MUST measure LOW.
+    $ note (scale "aeolian" "0 ~ <0 -3> ~" + 2 - 24)
+    # s "moog" # n 2
+    # lpf 220 # resonance 0.1
+    # gain 1.1
+    # sustain 0.9
+    # cut 3
+  d4 $ gF1 $ gM2 -- STAB: the gF1 sweep target. MUST measure BRIGHT.
+    $ off 0.125 ((|+ note 12) . (|* gain 0.55))
+    $ note (scale "aeolian" "~ 4 ~ [7 9]" + 2)
+    # s "arpy" # n 4
+    # hpf 400
+    # gain 0.95
+    # room 0.3 # sz 0.6
+    # delay 0.35 # delayt 0.375 # delayfb 0.35
+    # cut 4
+  d5 $ gF3 $ gM3 -- PAD: the SUSTAIN test. duty must stay high, not attack-and-die.
+    $ note (scale "aeolian" "[0,3,7]" + 2 - 12)
+    # s "moog" # n 5
+    # lpf 1400 # resonance 0.05
+    # gain 0.8
+    # sustain 3.4
+    # legato 1
+    # room 0.45 # sz 0.85
+    # cut 5
+  d6 $ gF2 $ gM1 -- ARP: density under gMask
+    $ arp "<up updown>"
+    $ note (scale "aeolian" "<[0,3,7] [-3,0,4]>" + 2)
+    # s "arpy" # n 7
+    # gain 0.7
+    # legato 0.5
+    # pan 0.38
+    # room 0.25
+    # cut 6
+  d7 $ gF1 $ gM2 -- RISER: must measure as RISING over the window
+    $ s "hh*16"
+    # n 9
+    # hpf (slow 4 $ range 300 6000 saw)
+    # gain (slow 4 $ range 0.25 0.95 saw)
+    # legato 0.4
+    # room 0.4 # sz 0.7
+    # cut 7
+  d8 $ gF1 $ gM1 -- BREAK: the gMute target
+    $ chop 8
+    $ s "jvbass:2"
+    # gain 0.75
+    # lpf 2200
+    # cut 8

tools/at/harness.py

+"""Acceptance-test harness for the ParVagues rig.
+
+Why this exists (2026-07-28, J-7 to OPAL)
+-----------------------------------------
+Unit tests on pure functions prove nothing about whether the rig makes sound.
+Every failure that has actually cost this project an evening lived in the SEAM:
+a helper that typechecks but silences an orbit, a knob that moves but changes
+nothing, a sample folder whose index is one short, an orbit still playing from
+the previous track. None of those are visible to pytest, and all of them are
+measurable.
+
+So this harness drives the REAL rig — boots a track through tidal-remote, taps
+the real PipeWire graph through probe-chain's `Tap`, and asserts on measured
+audio. It is deliberately reusable: a case is a small object with a name and an
+assertion, not a one-shot script, so tonight's investigation becomes tomorrow's
+regression suite.
+
+The right lens per control kind
+-------------------------------
+An assertion is only as good as the quantity it measures, and the single most
+expensive lesson of this week is that rms is BLIND to a filter or a crusher:
+they change the spectrum, not the level. So:
+
+    gain / mute      -> rms dBFS
+    filter / crush   -> spectral centroid
+    ply / mask       -> onset density
+    fade vs sparse   -> the TIME axis, never an aggregate
+
+Safety
+------
+Nothing here sends CC. LED work goes through the mock surface (see mock_lcxl),
+never the live device, and no case ever touches CC 77-84.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import math
+import subprocess
+import sys
+import tempfile
+import time
+import wave
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import numpy as np
+
+TOOLS = Path(__file__).resolve().parent.parent
+REPO = TOOLS.parent
+
+
+def _load(name: str, path: Path):
+    """Import a hyphenated tool file as a module (filenames aren't identifiers)."""
+    spec = importlib.util.spec_from_file_location(name, path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+probe = _load("probe_chain", TOOLS / "probe-chain.py")
+
+
+# --------------------------------------------------------------------------
+# Rig readiness — a failing AT must distinguish "broken" from "not running"
+# --------------------------------------------------------------------------
+
+@dataclass
+class RigState:
+    sclang: int = 0
+    scsynth: int = 0
+    ardour: int = 0
+    sc_ports: int = 0
+
+    @property
+    def ready(self) -> bool:
+        return self.scsynth >= 1 and self.sc_ports > 0
+
+    def why_not(self) -> str:
+        if self.scsynth < 1:
+            return "scsynth is not running"
+        if self.sc_ports == 0:
+            return "SuperCollider has no output ports"
+        return ""
+
+
+def rig_state() -> RigState:
+    def count(pattern: str) -> int:
+        try:
+            out = subprocess.run(["pgrep", "-fc", pattern], capture_output=True,
+                                 text=True, timeout=5).stdout.strip()
+            return int(out or 0)
+        except Exception:
+            return 0
+
+    ports = probe.existing_ports()
+    return RigState(
+        sclang=count("sclang"),
+        scsynth=count("scsynth"),
+        ardour=count("ardour-[0-9]"),
+        sc_ports=sum(1 for p in ports if p.startswith(f"{probe.SC_NODE}:out_")),
+    )
+
+
+# --------------------------------------------------------------------------
+# Measurement
+# --------------------------------------------------------------------------
+
+@dataclass
+class Shape:
+    """What one orbit's audio looked like over one capture window."""
+
+    orbit: int
+    peak_db: float = -math.inf
+    rms_db: float = -math.inf
+    centroid_hz: float = 0.0
+    low_ratio: float = 0.0     # energy below 300 Hz, as a fraction of total
+    high_ratio: float = 0.0    # energy above 1 kHz, as a fraction of total
+    onsets_per_s: float = 0.0
+    duty: float = 0.0          # fraction of bins within 25 dB of the loudest
+    timeline: list[float] = field(default_factory=list)
+
+    @property
+    def crest_db(self) -> float:
+        if self.peak_db == -math.inf or self.rms_db == -math.inf:
+            return 0.0
+        return self.peak_db - self.rms_db
+
+    @property
+    def silent(self) -> bool:
+        return self.peak_db < -60.0
+
+    def __str__(self) -> str:
+        return (f"d{self.orbit}: peak {self.peak_db:6.1f} rms {self.rms_db:6.1f} "
+                f"crest {self.crest_db:5.1f} centroid {self.centroid_hz:7.1f}Hz "
+                f"low {self.low_ratio*100:4.1f}% high {self.high_ratio*100:4.1f}% "
+                f"onsets {self.onsets_per_s:5.2f}/s duty {self.duty*100:5.1f}%")
+
+
+def _read(path: Path) -> tuple[np.ndarray, int]:
+    with wave.open(str(path)) as w:
+        sr, n, ch = w.getframerate(), w.getnframes(), w.getnchannels()
+        a = np.frombuffer(w.readframes(n), dtype=np.int16).astype(np.float32) / 32768
+    if ch > 1:
+        a = a[: (len(a) // ch) * ch].reshape(-1, ch).mean(1)
+    return a, sr
+
+
+def _db(x: float) -> float:
+    return 20 * math.log10(x) if x > 1e-9 else -math.inf
+
+
+def spectral_centroid(sig: np.ndarray, sr: int) -> float:
+    """Brightness. The ONLY lens that sees a filter or a crusher.
+
+    Computed over frames that actually contain signal — averaging silence into
+    the centroid drags every reading toward the same meaningless number.
+    """
+    n = 2048
+    hop = 1024
+    if len(sig) < n:
+        return 0.0
+    freqs = np.fft.rfftfreq(n, 1.0 / sr)
+    win = np.hanning(n)
+    vals, weights = [], []
+    for i in range(0, len(sig) - n, hop):
+        frame = sig[i:i + n]
+        rms = float(np.sqrt((frame ** 2).mean()))
+        if rms < 1e-4:
+            continue
+        mag = np.abs(np.fft.rfft(frame * win))
+        s = mag.sum()
+        if s <= 0:
+            continue
+        vals.append(float((freqs * mag).sum() / s))
+        weights.append(rms)
+    if not vals:
+        return 0.0
+    return float(np.average(vals, weights=weights))
+
+
+def band_ratio(sig: np.ndarray, sr: int, f_lo: float, f_hi: float) -> float:
+    """Fraction of total energy inside [f_lo, f_hi).
+
+    Why this exists, and why the centroid is NOT used for register claims:
+    the first version of `sub_is_low` asserted on the spectral centroid and
+    failed d3 at 837 Hz — an orbit carrying `# lpf 220`. A magnitude-weighted
+    centroid integrates the whole spectrum, so a low-level broadband noise floor
+    spread across 20 kHz drags the mean far above where the actual ENERGY sits.
+    d5 showed the same distortion: `lpf 1400`, centroid 6344 Hz.
+
+    The centroid is still the right lens for a RELATIVE test — `filter_bites`
+    moved 2731 -> 2164 Hz cleanly, because the noise floor is common to both
+    readings and cancels. It is the wrong lens for an ABSOLUTE one. Band-energy
+    ratio answers "where does the energy live" directly and is unbothered by a
+    floor that carries almost none of it.
+    """
+    n, hop = 4096, 2048
+    if len(sig) < n:
+        return 0.0
+    freqs = np.fft.rfftfreq(n, 1.0 / sr)
+    sel = (freqs >= f_lo) & (freqs < f_hi)
+    win = np.hanning(n)
+    inside = total = 0.0
+    for i in range(0, len(sig) - n, hop):
+        frame = sig[i:i + n]
+        if float(np.sqrt((frame ** 2).mean())) < 1e-4:
+            continue
+        power = np.abs(np.fft.rfft(frame * win)) ** 2
+        inside += float(power[sel].sum())
+        total += float(power.sum())
+    return inside / total if total > 0 else 0.0
+
+
+def onset_rate(sig: np.ndarray, sr: int) -> float:
+    """Onsets per second via spectral flux.
+
+    Envelope-ratio detection returns a flat 0.0 on a sustained bass, and a zero
+    baseline turns any comparison into a division by zero that reads as "+inf %,
+    EFFECTIVE". Flux does not have that failure mode.
+    """
+    n, hop = 1024, 512
+    if len(sig) < n * 4:
+        return 0.0
+    win = np.hanning(n)
+    prev = None
+    flux = []
+    for i in range(0, len(sig) - n, hop):
+        mag = np.abs(np.fft.rfft(sig[i:i + n] * win))
+        if prev is not None:
+            flux.append(float(np.sum(np.maximum(mag - prev, 0))))
+        prev = mag
+    if not flux:
+        return 0.0
+    f = np.array(flux)
+    med = np.median(f)
+    mad = np.median(np.abs(f - med)) or 1e-9
+    thresh = med + 2.5 * mad
+    refractory = int(0.05 * sr / hop)
+    count, last = 0, -999
+    for i, v in enumerate(f):
+        if v > thresh and i - last > refractory:
+            count += 1
+            last = i
+    return count / (len(sig) / sr)
+
+
+def measure(orbits: list[int], seconds: float = 8.0,
+            bin_s: float = 0.5) -> dict[int, Shape]:
+    """Tap the live graph and describe each orbit with every lens at once.
+
+    One capture, all lenses: a second capture of "the same" state is not the
+    same state (patterns vary per cycle), so measuring gain and brightness in
+    separate runs invites a comparison between two different moments.
+    """
+    tmp = Path(tempfile.mkdtemp(prefix="pv-at-"))
+    taps = {dn: probe.Tap(f"d{dn}_sc", probe.sc_ports_for(dn), tmp) for dn in orbits}
+
+    try:
+        for dn, tap in taps.items():
+            tap.start(f"at_d{dn}")
+        time.sleep(1.2)
+        for dn, tap in taps.items():
+            tap.link(f"at_d{dn}")
+        time.sleep(seconds)
+        contaminated = {dn for dn, tap in taps.items() if not tap.verify()[0]}
+    finally:
+        for tap in taps.values():
+            tap.stop()
+
+    out: dict[int, Shape] = {}
+    for dn, tap in taps.items():
+        sh = Shape(orbit=dn)
+        if dn in contaminated:
+            # probe-chain's own lesson: a contaminated row must be DISCARDED,
+            # not interpreted. Leaving it silent-looking would read as a failure.
+            sh.peak_db = math.nan
+            out[dn] = sh
+            continue
+        try:
+            a, sr = _read(tap.path)
+        except Exception:
+            out[dn] = sh
+            continue
+        if len(a) == 0:
+            out[dn] = sh
+            continue
+        sh.peak_db = _db(float(np.max(np.abs(a))))
+        sh.rms_db = _db(float(np.sqrt((a ** 2).mean())))
+        sh.centroid_hz = spectral_centroid(a, sr)
+        sh.low_ratio = band_ratio(a, sr, 0.0, 300.0)
+        sh.high_ratio = band_ratio(a, sr, 1000.0, sr / 2)
+        sh.onsets_per_s = onset_rate(a, sr)
+        hop = max(1, int(sr * bin_s))
+        bins = [_db(float(np.sqrt((a[i:i + hop] ** 2).mean())))
+                for i in range(0, len(a) - hop + 1, hop)]
+        sh.timeline = bins
+        live = [b for b in bins if b != -math.inf]
+        if live:
+            top = max(live)
+            sh.duty = sum(1 for b in live if b > top - 25.0) / len(bins)
+        out[dn] = sh
+    return out
+
+
+# --------------------------------------------------------------------------
+# Driving the rig
+# --------------------------------------------------------------------------
+
+def remote(*args: str) -> bool:
+    """Fire a tidal-remote command. Fire-and-forget: success is MEASURED, never claimed."""
+    try:
+        subprocess.run([sys.executable, str(TOOLS / "tidal-remote.py"), *args],
+                       capture_output=True, timeout=30)
+        return True
+    except Exception:
+        return False
+
+
+def play(track: Path, settle: float = 6.0) -> None:
+    remote("eval-file", str(track))
+    time.sleep(settle)
+
+
+def hush() -> None:
+    remote("hush")