feat(midiviz): the master bus, painted behind the rain

PLN: "can we overlay behind a basic spectro? that id remove spectro VSTs
from ardour and wed have our super perf one as overlay on single win?"

So a coarse spectrum of the DEFAULT SINK's monitor -- the mix as the room
hears it, not one orbit -- painted first, under every glyph, in the window
that is already on top of everything. 40 log bands off a 2048-point FFT at
18 Hz: scenery, not an analyser. If something needs measuring, tidal-ears
is the tool and it is the right one.

Off by default and off both ways. `--spectro` or the `s` key acquires the
tap; `s` again hands it back -- the capture subprocess does not exist until
asked and does not survive being un-asked, closeEvent releases it, and the
unit deliberately carries no --spectro so login never acquires one. A live
rig does not get to grow an always-on audio consumer quietly.

Capture and FFT run in one thread whose entire contact with Qt is a single
frame swapped under a lock, depth ONE: a GUI at 5 fps reading an analyser at
18 fps shows the newest picture and discards the rest. A queue here would
only buy latency, which is what a 512-deep queue already cost the HUD once.
pw-record is asked for 100 ms latency explicitly -- a monitor client that
requests a tight buffer is how you talk the graph's quantum down and pay for
a decoration in xruns.

Measured, A/B/A, same -30 dBFS 1 kHz bed in every window, 120 s each: 0
added xruns with the tap on (2249 analysed frames, 18.7/s), 0 in either
control window, every node's counter unmoved. Nothing in the graph noticed.

Two things this cost, both recorded where they bit:

* `Popen(bufsize=0)` hands back a RAW FileIO, so `read(n)` is one os.read
  and a hop-sized read is short far more often than not. Treating that as
  EOF ended the capture on its first chunk -- blocks=0, no error recorded --
  and every pure-function test stayed green on top of a backdrop that could
  not receive a sample. Only running it against real audio found it.
  `_read_exact` now owns the distinction, with a dribbling-stream test.
* the first assertion about band placement was wrong about the code it
  tested: below ~750 Hz a 2048-point FFT has fewer bins than we have bands,
  so the edges are not the nominal log ramp, and checking 220 Hz against the
  ramp reported MISPLACED for a band that is exactly 211-234 Hz. Ask the
  edges where the band is.

And the reference was wrong before the code was: the first bed I labelled an
"80-226 Hz sweep" was 0.02*sin(2*pi*f(t)*t), whose instantaneous frequency
is f + t*df/dt -- broadband chirp garbage. The spectrum was reading it
correctly. A known 1 kHz tone lands in band 22, span 984-1148 Hz, every
other band at 0.00.

`_paint_spectrum` was checked against all 33 methods on the widget before it
was named: `_paint_chrome` had to become `_paint_controls` because a second
method quietly took the first one's name and the selftest went on passing
while a whole layer stopped being drawn.

selftest PASS (bars=40, tap released); 106 tests pass.

tools/bridge/midiviz.py

@@ -21,8 +21,10 @@ else".
     python3 midiviz.py -p 28:0          # pin a source port
     python3 midiviz.py -l               # list ports, exit
     python3 midiviz.py --selftest       # 3 s of synthetic events, headless
+    python3 midiviz.py --spectro        # + the master-bus spectrum, behind
 
-Keys: q / Esc quit · p pause · c clear · +/- resize. Drag anywhere to move.
+Keys: q / Esc quit · p pause · c clear · s spectrum · t pin · +/- resize.
+Drag anywhere to move.
 
 Design notes worth keeping
 --------------------------
@@ -237,6 +239,7 @@ IDLE_AFTER = 2.0                    # seconds of silence before the slow pulse
 # number of drawText calls, and at this size 25 fps is indistinguishable from 33
 # while costing a fifth less of the core PLN needs for audio.
 FPS_ACTIVE, FPS_IDLE = 40, 200      # timer intervals, ms
+FPS_SPECTRO = 54                    # ms; ~18 fps, the spectrum's own rate
 # Rain is decoration, and it was the single most expensive thing on screen: a
 # 96-drop cap with a trail glyph each meant up to 192 drawText per frame, 3.1 ms
 # of an 8.9 ms frame. 32 heads with a trail only on the brightest few costs ~0.6.
@@ -405,8 +408,259 @@ def _latch_close():
               file=sys.stderr)
 
 
+# ── the spectrum backdrop ──────────────────────────────────────────────────
+# PLN, 2026-09-05: "can we overlay behind a basic spectro? that id remove
+# spectro VSTs from ardour and wed have our super perf one as overlay on
+# single win?"
+#
+# So: a coarse spectrum of the MASTER BUS, painted BEHIND the glyph rain, in
+# one window that is already on top of everything. Four things are
+# load-bearing and each of them is a lesson already paid for:
+#
+#   * **OFF BY DEFAULT, both ways.** This machine performs live. A new
+#     always-on audio consumer is not something the rig gets to acquire
+#     silently, so the capture subprocess does not exist until either
+#     `--spectro` or the runtime `s` key asks for it, and `s` again takes it
+#     away (process killed, not merely hidden).
+#   * **Never on the GUI thread.** Capture and FFT live in one thread whose
+#     only contact with Qt is a single `_latest` tuple swapped under a lock.
+#     paintEvent reads that tuple; it can never wait on audio.
+#   * **Drop-oldest, depth one.** There is no queue. Each analysed frame
+#     OVERWRITES the last, so a GUI at 5 fps reading an analyser at 18 fps
+#     shows the newest picture and discards the rest. (A 512-deep SSE queue
+#     was itself the HUD's 1 s lag; depth is latency, and a backdrop that is
+#     one frame late is invisible while a backdrop that is a second late is a
+#     lie.)
+#   * **Decimated on purpose.** 18 Hz, 40 log bands off a 2048-point FFT.
+#     This is scenery, not an analyser: the glyphs stay temporally and
+#     visually primary and the spectrum is a dim wash under them. If PLN ever
+#     wants to *measure* something, `tidal-ears` is the tool and it is the
+#     right one.
+#
+# The tap is the DEFAULT SINK's monitor, re-resolved on every (re)launch
+# rather than captured once -- the same "never trust a resolved binding past
+# the moment it might have gone stale" rule as the MIDI port above. The sink
+# is what the audience hears after Ardour's master, which is the whole point:
+# it is the mix, not one orbit. `pw-record`'s default 100 ms latency is asked
+# for EXPLICITLY here: a monitor client requesting a tight buffer is how you
+# talk the graph's quantum down and buy xruns for a decoration.
+SPEC_BANDS = 40                     # log-spaced bars across the full width
+SPEC_FFT = 2048                     # 23.4 Hz bins at 48k -- coarse on purpose
+SPEC_HOP = 2560                     # 18.75 analysis frames a second
+SPEC_RATE = 48000
+SPEC_FMIN, SPEC_FMAX = 32.0, 16000.0
+SPEC_DB_FLOOR, SPEC_DB_CEIL = -80.0, -6.0
+SPEC_ATTACK, SPEC_RELEASE = 0.60, 0.12   # rise fast, fall slow
+SPEC_RELAUNCH_S = 2.0               # backoff after the capture ends unasked
+SPEC_LEVELS = 20                    # colour LUT depth for the wash
+
+
+def _spec_band_edges(bands: int, fft_n: int, rate: int):
+    """rfft bin index boundaries for `bands` log-spaced bands.
+
+    Below ~750 Hz a 2048-point FFT has fewer bins than we have bands, so the
+    low edges collapse onto consecutive bins. That is fine and deliberate --
+    the alternative (empty bands, painted as silence) would read as a hole in
+    the bass, which is exactly the misreading the "mapped-but-idle must not be
+    black" note above is about.
+    """
+    import numpy as np
+    nyq = rate * 0.5
+    fmax = min(SPEC_FMAX, nyq * 0.98)
+    ratio = np.arange(bands + 1) / float(bands)
+    freqs = SPEC_FMIN * (fmax / SPEC_FMIN) ** ratio
+    idx = np.rint(freqs / rate * fft_n).astype(int)
+    top = fft_n // 2
+    idx = np.clip(idx, 1, top)
+    for i in range(1, idx.size):            # strictly increasing, so no band is empty
+        if idx[i] <= idx[i - 1]:
+            idx[i] = idx[i - 1] + 1
+    if idx[-1] > top:                       # ran out of bins: slide the whole ramp down
+        idx -= (idx[-1] - top)
+        idx = np.clip(idx, 1, top)
+    return idx
+
+
+def _read_exact(stream, n: int) -> "bytes | None":
+    """Exactly `n` bytes, or None at end of stream.
+
+    2026-09-05, and this cost a whole verification round: `Popen(bufsize=0)`
+    hands back a RAW `FileIO`, whose `read(n)` is one `os.read` and therefore
+    returns whatever the pipe happens to hold. The first version read one hop
+    with `stdout.read(need)` and treated a short read as EOF -- so the very
+    first partial chunk ended the capture, `blocks=0`, `err=''`, and a green
+    selftest sat happily on top of a backdrop that could never receive a
+    sample. The pure half proved nothing about the plumbing; only running it
+    against real audio did.
+    """
+    chunks, got = [], 0
+    while got < n:
+        b = stream.read(n - got)
+        if not b:
+            return None
+        chunks.append(b)
+        got += len(b)
+    return b"".join(chunks) if len(chunks) > 1 else chunks[0]
+
+
+def spec_default_target() -> str | None:
+    """The default sink's node name, or None if PulseAudio/PipeWire is absent."""
+    try:
+        out = subprocess.run(["pactl", "get-default-sink"], capture_output=True,
+                             text=True, timeout=3)
+    except (OSError, subprocess.SubprocessError):
+        return None
+    name = (out.stdout or "").strip()
+    return name or None
+
+
+class SpectrumSource:
+    """`pw-record` on the default sink's monitor → one coarse band frame.
+
+    `push()` is the seam: it is the whole analysis path and it does not care
+    where its samples came from, so `--selftest` drives the real FFT and the
+    real paint with synthetic audio on a box with no sound at all. Everything
+    the subprocess adds is in `_run`, and nothing in `_run` computes anything.
+    """
+
+    def __init__(self, target: str | None = None, bands: int = SPEC_BANDS,
+                 rate: int = SPEC_RATE, fft_n: int = SPEC_FFT,
+                 hop: int = SPEC_HOP):
+        import numpy as np
+        self._np = np
+        self.target = target
+        self.bands = bands
+        self.rate = rate
+        self.fft_n = fft_n
+        self.hop = hop
+        self.edges = _spec_band_edges(bands, fft_n, rate)
+        self._win = np.hanning(fft_n).astype(np.float32)
+        self._winsum = float(self._win.sum())
+        self._buf = np.zeros(fft_n, dtype=np.float32)
+        self._env = np.zeros(bands, dtype=np.float32)
+        self._lock = threading.Lock()
+        self._latest: tuple[float, ...] | None = None
+        self.frames = 0                 # analysed frames
+        self.blocks = 0                 # blocks read off the pipe
+        self.err = ""                   # last thing that went wrong, for the header
+        self._stop = threading.Event()
+        self._th: threading.Thread | None = None
+        self._proc = None
+
+    # ── analysis (thread-agnostic; the tested half) ────────────────────────
+    def push(self, block) -> tuple[float, ...]:
+        """One hop of mono float samples → the published band frame."""
+        np = self._np
+        b = np.asarray(block, dtype=np.float32).ravel()
+        if b.size >= self.fft_n:
+            self._buf[:] = b[-self.fft_n:]
+        elif b.size:
+            self._buf[:-b.size] = self._buf[b.size:]
+            self._buf[-b.size:] = b
+        spec = np.abs(np.fft.rfft(self._buf * self._win)) * (2.0 / self._winsum)
+        power = spec * spec
+        e = self.edges
+        # One reduceat over the whole spectrum instead of `bands` slices.
+        # power is truncated at the LAST edge: reduceat's final segment runs to
+        # the end of its input, so an untruncated array would dump everything
+        # from 16 kHz to Nyquist into the top band.
+        sums = np.add.reduceat(power[:e[-1]], e[:-1])
+        widths = np.maximum(1, e[1:] - e[:-1])
+        db = 10.0 * np.log10(sums / widths + 1e-20)
+        lvl = (db - SPEC_DB_FLOOR) / (SPEC_DB_CEIL - SPEC_DB_FLOOR)
+        np.clip(lvl, 0.0, 1.0, out=lvl)
+        rising = lvl > self._env
+        a = np.where(rising, SPEC_ATTACK, SPEC_RELEASE).astype(np.float32)
+        self._env += a * (lvl.astype(np.float32) - self._env)
+        frame = tuple(float(v) for v in self._env)
+        with self._lock:
+            self._latest = frame
+            self.frames += 1
+        return frame
+
+    def latest(self) -> "tuple[float, ...] | None":
+        with self._lock:
+            return self._latest
+
+    # ── capture (the subprocess half) ──────────────────────────────────────
+    def _cmd(self, target: str) -> list[str]:
+        return ["pw-record", "--target", target, "--rate", str(self.rate),
+                "--channels", "1", "--format", "f32", "--latency", "100ms",
+                "--raw", "-"]
+
+    def start(self) -> "SpectrumSource":
+        if self._th is None:
+            self._th = threading.Thread(target=self._run, name="spectro",
+                                        daemon=True)
+            self._th.start()
+        return self
+
+    def _run(self) -> None:
+        need = self.hop * 4                      # float32 mono
+        while not self._stop.is_set():
+            target = self.target or spec_default_target()
+            if not target:
+                self.err = "no sink"
+                if self._stop.wait(SPEC_RELAUNCH_S):
+                    return
+                continue
+            try:
+                self._proc = subprocess.Popen(self._cmd(target),
+                                              stdout=subprocess.PIPE,
+                                              stderr=subprocess.DEVNULL,
+                                              bufsize=0)
+            except OSError as exc:               # no pw-record on this box
+                self.err = type(exc).__name__
+                return
+            try:
+                out = self._proc.stdout
+                while not self._stop.is_set():
+                    raw = _read_exact(out, need)
+                    if raw is None:
+                        self.err = "eof"
+                        break
+                    self.blocks += 1
+                    self.err = ""
+                    self.push(self._np.frombuffer(raw, dtype="<f4"))
+            except (OSError, ValueError) as exc:
+                self.err = type(exc).__name__
+            finally:
+                self._kill()
+            # EOF unasked = the sink went away (device switch, replug). Re-resolve
+            # from scratch rather than keeping a binding a hotplug just killed.
+            if self._stop.wait(SPEC_RELAUNCH_S):
+                return
+
+    def _kill(self) -> None:
+        proc, self._proc = self._proc, None
+        if proc is None:
+            return
+        try:
+            proc.terminate()
+            proc.wait(timeout=1.0)
+        except (OSError, subprocess.TimeoutExpired):
+            try:
+                proc.kill()
+            except OSError:
+                pass
+        for stream in (proc.stdout,):
+            try:
+                if stream:
+                    stream.close()
+            except OSError:
+                pass
+
+    def close(self) -> None:
+        self._stop.set()
+        self._kill()
+        th, self._th = self._th, None
+        if th is not None:
+            th.join(timeout=1.5)
+
+
 def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
-                 pinned_port: str | None = None, watch: bool = False):
+                 pinned_port: str | None = None, watch: bool = False,
+                 spectro: bool = False):
     QtCore, QtGui, QtWidgets = _qt()
     Qt = QtCore.Qt
 
@@ -453,6 +707,15 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
             self._bucket_t = self.t0
             self._bucket_n = 0
 
+            # The backdrop. `spec_src` is None until somebody asks: OFF means
+            # no thread and no subprocess, not a hidden one.
+            self.spectro = False
+            self.spec_src: "SpectrumSource | None" = None
+            self.spec_err = ""
+            self._spec_frame: tuple[float, ...] | None = None
+            self._spec_peak: list[float] = []
+            self._spec_hot_t = 0.0        # last time the mix was audibly awake
+
             self._build_palette()
             self._apply_scale()
 
@@ -465,6 +728,11 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
             # by RECONNECT_MS. `watch` is off for --selftest (headless, fed
             # synthetic events by hand) so the suite never shells out to
             # `aconnect`/`aseqdump` on its own clock.
+            # Asked for on the command line: acquire it once the widget is
+            # otherwise fully built, through the same one door as the `s` key.
+            if spectro:
+                self.set_spectro(True)
+
             self.rebind_timer = None
             if watch:
                 self.rebind_timer = QtCore.QTimer(self)
@@ -526,6 +794,19 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
                     col.append(QtGui.QColor.fromHsvF(hue / 360.0, sat, val))
                 self.lut.append(col)
             self.chrome = self.lut[FAM_OTHER]
+            # The backdrop's own ramp, and it is deliberately NOT one of the
+            # families: the eight families MEAN role, and a wash that borrowed
+            # one would read as "orbit N did something". A cold blue-violet
+            # that no control ever uses, at alphas low enough that the dimmest
+            # glyph still wins the foreground. Precomputed like everything
+            # else, so a spectrum frame constructs no QColor.
+            self.spec_lut = [
+                QtGui.QColor.fromHsvF(0.68 - 0.055 * (lv / (SPEC_LEVELS - 1)),
+                                      0.80, 0.30 + 0.70 * (lv / (SPEC_LEVELS - 1)),
+                                      (12 + 84 * (lv / (SPEC_LEVELS - 1))) / 255.0)
+                for lv in range(SPEC_LEVELS)
+            ]
+            self.spec_peak_col = QtGui.QColor(0xc8, 0xa8, 0xff, 90)
 
         def _font(self, px: int):
             f = QtGui.QFont()
@@ -680,14 +961,87 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
                               if (d.__setitem__(1, d[1] + dt * 9.0) or True)
                               and d[1] < rows + 2 and now - d[2] < 4.0]
 
+            self._spec_tick(now)
+
             idle = (now - self.last_event_t) > IDLE_AFTER and not self.drops
             want = FPS_IDLE if idle else FPS_ACTIVE
+            # A third rate, and only when earned. The two-rate rule exists so a
+            # silent window is not a busy loop; a MOVING backdrop at 5 fps is
+            # visibly a slideshow, so while the mix is actually making sound the
+            # idle rate becomes the analyser's own 18 Hz -- never the 25 fps
+            # active rate, and never at all once the audio goes quiet too.
+            if idle and self.spectro and (now - self._spec_hot_t) <= IDLE_AFTER:
+                want = FPS_SPECTRO
             if want != self._interval:
                 self._interval = want
                 self.timer.setInterval(want)
             self.frames += 1
             self.update()
 
+        # ── the backdrop's switch and its clock ────────────────────────────
+        def set_spectro(self, on: bool, source: "SpectrumSource | None" = None):
+            """Acquire or release the audio tap. Idempotent.
+
+            `source` is an injection point for the tests: pass one and no
+            subprocess is ever spawned. Off genuinely tears the capture down --
+            "hidden but still consuming" is the state this whole feature is
+            not allowed to have.
+            """
+            on = bool(on)
+            if on == self.spectro:
+                return
+            self.spectro = on
+            if not on:
+                src, self.spec_src = self.spec_src, None
+                if src is not None:
+                    src.close()
+                self._spec_frame = None
+                self._spec_peak = []
+                self.update()
+                return
+            if source is not None:
+                self.spec_src = source
+            else:
+                try:
+                    self.spec_src = SpectrumSource().start()
+                except ImportError as exc:       # numpy absent: say so, stay off
+                    self.spectro = False
+                    self.spec_err = "numpy?"
+                    print("midiviz: no spectrum backdrop (%s)" % (exc,),
+                          file=sys.stderr)
+                    return
+            self.spec_err = ""
+            self._spec_hot_t = time.monotonic()
+            self.update()
+
+        def _spec_tick(self, now: float) -> None:
+            """Swap in the newest analysed frame. Depth one, never blocking."""
+            src = self.spec_src
+            if src is None:
+                return
+            frame = src.latest()
+            if frame is None:
+                # No frame yet. If the capture already gave a reason, say it
+                # ONCE on stderr: a backdrop that silently never appears (no
+                # `pw-record`, no sink) is indistinguishable from one that is
+                # merely switched off, and the window itself has no words to
+                # tell them apart with.
+                if src.err and src.err != self.spec_err:
+                    self.spec_err = src.err
+                    print("midiviz: spectrum backdrop is not getting audio (%s)"
+                          % (src.err,), file=sys.stderr)
+                return
+            self.spec_err = src.err
+            self._spec_frame = frame
+            if len(self._spec_peak) != len(frame):
+                self._spec_peak = list(frame)
+            else:
+                pk = self._spec_peak
+                for i, v in enumerate(frame):
+                    pk[i] = v if v > pk[i] else pk[i] * 0.93
+            if max(frame) > 0.04:
+                self._spec_hot_t = now
+
         def _lv(self, age: float, floor: int = 0) -> int:
             """Age in seconds → LUT level. The only decay curve in the file."""
             if age < 0:
@@ -706,6 +1060,10 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
                 now = time.monotonic()
                 # the slow dim pulse: the only thing that moves when it is silent
                 pulse = 0.5 + 0.5 * math.sin((now - self.t0) * 0.55)
+                # FIRST, over the background and under everything else. The
+                # ordering IS the feature: PLN asked for it "behind".
+                if self.spectro:
+                    self._paint_spectrum(p, w, h)
                 self._paint_chrome(p, w, h, pulse)
                 self._paint_header(p, now, pulse)
                 self._paint_controls(p)
@@ -717,6 +1075,43 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
             finally:
                 p.end()
 
+        # ── the backdrop ───────────────────────────────────────────────────
+        # Named `_paint_spectrum` and checked against every other method on
+        # this class before being written: `_paint_chrome` had to become
+        # `_paint_controls` because a second method quietly took the first
+        # one's name and the selftest went on passing while a whole layer
+        # stopped being drawn. A shadowed painter is invisible until a gig.
+        #
+        # Bars, bottom-anchored, spanning the FULL width -- low left to high
+        # right. Not a widget in a corner: a wash the glyphs sit on. `bands`
+        # fillRects and `bands` more for the peak hold is the entire cost, and
+        # every colour is a LUT hit.
+        def _paint_spectrum(self, p, w, h) -> int:
+            frame = self._spec_frame
+            if not frame:
+                return 0
+            n = len(frame)
+            peak = self._spec_peak
+            top = self.hdr_y + self.hdr_h        # never under the header text
+            span = max(8, h - top)
+            bw = w / float(n)
+            drawn = 0
+            for i, v in enumerate(frame):
+                x0 = int(i * bw)
+                x1 = int((i + 1) * bw)
+                bar = int(v * span * 0.92)
+                if bar >= 1:
+                    lv = int(v * (SPEC_LEVELS - 1) + 0.5)
+                    p.fillRect(x0, h - bar, max(1, x1 - x0 - 1), bar,
+                               self.spec_lut[lv])
+                    drawn += 1
+                if i < len(peak):
+                    pk = int(peak[i] * span * 0.92)
+                    if pk >= 2:
+                        p.fillRect(x0, h - pk, max(1, x1 - x0 - 1), 1,
+                                   self.spec_peak_col)
+            return drawn
+
         def _paint_chrome(self, p, w, h, pulse):
             p.setPen(self.chrome[2 + int(3 * pulse)])
             p.drawRect(0, 0, w - 1, h - 1)
@@ -988,6 +1383,11 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
                 self._set_on_top(not self.on_top)
             elif k == Qt.Key.Key_P:
                 self.paused = not self.paused
+            elif k == Qt.Key.Key_S:
+                # The runtime half of "off by default": `s` both acquires the
+                # tap and gives it back, so nothing about the audio graph is
+                # permanent and nothing needs a restart to undo.
+                self.set_spectro(not self.spectro)
             elif k == Qt.Key.Key_C:
                 self.cc.clear()
                 self.stream.clear()
@@ -1063,6 +1463,13 @@ def build_widget(port_label: str, reader: "Reader | None", scale: float = 1.0,
                 self.rebind_timer.stop()
             if self.reader:
                 self.reader.close()
+            # Hand the audio tap back before anything else. A closed window
+            # that left a `pw-record` on the master bus would be exactly the
+            # "always-on consumer" this feature promised not to become, and it
+            # would outlive the only UI that could have switched it off.
+            if self.spec_src is not None:
+                self.spec_src.close()
+                self.spec_src = None
             # "if i close i wanna close it". A clean exit is not a failure, so
             # systemd will not restart us -- but rig_units.ensure() starts
             # every non-manual unit that is not active, and gig-up and the
@@ -1108,6 +1515,73 @@ def _synthetic_lines() -> list[str]:
     return out
 
 
+def _synthetic_audio(n: int, t0: float, rate: int = SPEC_RATE):
+    """A sweeping tone plus noise: enough spectral structure that a picture
+    made of it cannot be a flat wash, at an amplitude (-20 dBFS) chosen so the
+    bars land mid-scale instead of pinned. Synthetic on purpose -- the selftest
+    must be able to prove the ANALYSIS and the PAINT on a box with no audio
+    server at all. What it cannot prove is `pw-record`; that half is verified
+    by running the real thing.
+    """
+    import numpy as np
+    t = (t0 + np.arange(n, dtype=np.float64) / rate)
+    f = 90.0 * (2.0 ** (2.6 * (0.5 + 0.5 * math.sin(t0 * 0.7))))
+    sig = 0.09 * np.sin(2 * math.pi * f * t)
+    sig += 0.012 * np.random.default_rng(int(t0 * 1000) & 0xFFFF).standard_normal(n)
+    return sig.astype(np.float32)
+
+
+def selftest_spectrum(bands: int = SPEC_BANDS) -> tuple[bool, str]:
+    """The backdrop's own checks: real FFT, real bands, real numbers.
+
+    Three properties, each of which a plausible bug breaks:
+      * a tone lands in ONE part of the spectrum, not everywhere (band edges
+        and the reduceat truncation are both wrong in ways that smear it);
+      * digital silence reads as silence (an absent epsilon or a log of zero
+        gives NaN, which paints as a full-height bar -- silence must never
+        look like signal, the inverse of the "mapped-but-idle" rule);
+      * the envelope RISES faster than it FALLS, or the wash lags the music.
+    """
+    import numpy as np
+    src = SpectrumSource(target="__never__", bands=bands)
+    rate = src.rate
+    t = np.arange(src.fft_n, dtype=np.float64) / rate
+    notes = []
+
+    for _ in range(24):                       # let the envelope settle
+        loud = src.push(0.25 * np.sin(2 * math.pi * 220.0 * t))
+    peak_band = int(np.argmax(loud))
+    # Ask the band EDGES where the band is, not the nominal log ramp. Below
+    # ~750 Hz the ramp is not what `_spec_band_edges` returns -- bins run out
+    # and the low edges become consecutive -- so checking 220 Hz against the
+    # ramp reported MISPLACED for a band that is in fact exactly 211-234 Hz.
+    # The first version of this assertion was wrong about the code it tested.
+    bin_hz = rate / float(src.fft_n)
+    lo = float(src.edges[peak_band]) * bin_hz
+    hi = float(src.edges[peak_band + 1]) * bin_hz
+    placed = lo <= 220.0 <= hi
+    notes.append("tone_220Hz->band%d[%.0f-%.0fHz]:%s"
+                 % (peak_band, lo, hi, "ok" if placed else "MISPLACED"))
+    energy = float(np.sum(loud))
+    concentrated = bool(loud[peak_band] > 0.2 and energy < loud[peak_band] * 12)
+    notes.append("concentration:%s" % ("ok" if concentrated else "SMEARED"))
+
+    for _ in range(80):                       # release is slow; give it room
+        quiet = src.push(np.zeros(src.fft_n, dtype=np.float32))
+    silent = bool(np.all(np.isfinite(quiet)) and max(quiet) < 0.02)
+    notes.append("silence->%.3f:%s" % (max(quiet), "ok" if silent else "NOT SILENT"))
+
+    rise = src.push(0.25 * np.sin(2 * math.pi * 220.0 * t))[peak_band]
+    fall = src.push(np.zeros(src.fft_n, dtype=np.float32))[peak_band]
+    asym = bool(rise > 0.05 and fall > rise * 0.5)
+    notes.append("attack%.3f>release(kept %.3f):%s"
+                 % (rise, fall, "ok" if asym else "BACKWARDS"))
+
+    src.close()                               # no thread was started; must be safe
+    ok = placed and concentrated and silent and asym
+    return ok, " ".join(notes)
+
+
 def selftest(seconds: float = 3.0, show: bool = False) -> int:
     if not show:
         os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
@@ -1118,6 +1592,14 @@ def selftest(seconds: float = 3.0, show: bool = False) -> int:
     w.show()
     app.processEvents()
 
+    # The backdrop, fed by hand. `target` is a name no node can have, so even
+    # if `start()` were called nothing would be captured -- but it is not
+    # called: `push()` is the seam, and the selftest is the other caller of it.
+    spec_ok, spec_notes = selftest_spectrum()
+    spec_src = SpectrumSource(target="__never__")
+    w.set_spectro(True, spec_src)
+    spec_t = 0.0
+
     parsed = [midimon.enrich(ev) for ev in
               (midimon.parse_line(ln) for ln in _synthetic_lines()) if ev]
     if not parsed:
@@ -1131,6 +1613,8 @@ def selftest(seconds: float = 3.0, show: bool = False) -> int:
         for _ in range(3):                       # a plausible live event rate
             w.ingest(parsed[i % len(parsed)])
             i += 1
+        spec_src.push(_synthetic_audio(spec_src.hop, spec_t))
+        spec_t += spec_src.hop / float(spec_src.rate)
         w.tick()
         app.processEvents()
         pm = w.grab()                            # forces the real paint path
@@ -1154,12 +1638,28 @@ def selftest(seconds: float = 3.0, show: bool = False) -> int:
     app.processEvents()
     w.grab()
     painted = w.painted
+    # Prove the LAYER, not just the maths: count the bars the painter actually
+    # emitted on a real frame, then prove `s` gives the tap back.
+    _pm = QtGui.QPixmap(w.width(), w.height())
+    _pm.fill(QtGui.QColor(*BG))
+    _pp = QtGui.QPainter(_pm)
+    try:
+        bars = w._paint_spectrum(_pp, w.width(), w.height())
+    finally:
+        _pp.end()
+    spec_frames = spec_src.frames
+    w.set_spectro(False)
+    released = w.spec_src is None and not w.spectro
     w.close()
 
-    ok = grabs >= 30 and len(colours) >= 8 and painted >= grabs
+    ok = (grabs >= 30 and len(colours) >= 8 and painted >= grabs
+          and spec_ok and bars >= SPEC_BANDS // 2 and spec_frames >= 20
+          and released)
     print("midiviz selftest: parsed=%d ingested=%d frames=%d paints=%d grabs=%d "
-          "distinct_sampled_colours=%d platform=%s -> %s"
+          "distinct_sampled_colours=%d spectro[frames=%d bars=%d released=%s %s] "
+          "platform=%s -> %s"
           % (len(parsed), w.ingested, w.frames, painted, grabs, len(colours),
+             spec_frames, bars, released, spec_notes,
              os.environ.get("QT_QPA_PLATFORM", "native"), "PASS" if ok else "FAIL"))
     return 0 if ok else 1
 
@@ -1170,6 +1670,11 @@ def main(argv=None) -> int:
     ap.add_argument("-l", "--list", action="store_true", help="list ports and exit")
     ap.add_argument("-s", "--scale", type=float, default=1.0,
                     help="initial scale, 0.6-2.4 (also +/- at runtime)")
+    ap.add_argument("--spectro", action="store_true",
+                    help="spectrum backdrop from the default sink's monitor "
+                         "(OFF by default; 's' toggles it at runtime)")
+    ap.add_argument("--spectro-target",
+                    help="capture this node instead of the default sink")
     ap.add_argument("--selftest", action="store_true",
                     help="render synthetic events for --seconds, then exit")
     ap.add_argument("--seconds", type=float, default=3.0)
@@ -1212,12 +1717,19 @@ def main(argv=None) -> int:
     app.setDesktopFileName("midiviz")
     w = build_widget(port or "--", reader, scale=max(0.6, min(2.4, a.scale)),
                      pinned_port=pinned, watch=True)
+    if a.spectro:
+        src = None
+        if a.spectro_target:
+            src = SpectrumSource(target=a.spectro_target).start()
+        w.set_spectro(True, src)
     w.show()
     w.raise_()
     w.activateWindow()
     rc = app.exec()
     if w.reader is not None:
         w.reader.close()
+    if getattr(w, "spec_src", None) is not None:
+        w.spec_src.close()
     # A deliberate close reports itself, so the unit does not undo it. Only
     # when the exit was otherwise clean: a real failure keeps its own code and
     # stays restartable.

tools/bridge/tests/test_midiviz.py

@@ -304,3 +304,234 @@ def test_user_close_exit_code_matches_the_unit():
     assert str(mv.USER_CLOSE_EXIT) in ok.group(1).split(), (
         f"SuccessExitStatus={ok.group(1)!r} does not include "
         f"{mv.USER_CLOSE_EXIT}")
+
+
+# ── the spectrum backdrop ──────────────────────────────────────────────────
+#
+# PLN asked for a spectro "behind" the rain so the Ardour spectro VSTs could
+# go. The whole risk of the feature is that it is an AUDIO consumer on a box
+# that performs live, so what is asserted here is mostly about restraint: it
+# must not exist until asked, it must genuinely go away when un-asked, and
+# its numbers must not be able to paint silence as signal.
+
+def test_the_spectro_does_not_exist_until_it_is_asked_for(monkeypatch):
+    """Off by default means NO PROCESS, not a hidden one.
+
+    A backdrop that spawned `pw-record` at startup and merely declined to
+    paint would be exactly the always-on consumer this is not allowed to be,
+    and nothing on screen would reveal it.
+    """
+    import os
+    import pytest
+    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+    mv, w = _widget(1.0, 900)
+    if mv is None:
+        pytest.skip("no Qt available")
+    spawned = []
+    monkeypatch.setattr(mv.subprocess, "Popen",
+                        lambda *a, **k: spawned.append(a) or (_ for _ in ()).throw(
+                            AssertionError("the backdrop spawned a capture unasked")))
+    assert w.spectro is False
+    assert w.spec_src is None
+    w.tick()                              # a full frame with nobody asking
+    w.grab()
+    assert spawned == []
+    w.close()
+
+
+def test_toggling_the_spectro_acquires_and_releases_the_tap():
+    """`s` on, `s` off — and off must hand the capture back, not hide it."""
+    import os
+    import pytest
+    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+    mv, w = _widget(1.0, 900)
+    if mv is None:
+        pytest.skip("no Qt available")
+    closed = []
+
+    class FakeSource:
+        hop, rate, err, frames = 256, 48000, "", 0
+
+        def latest(self):
+            return tuple(0.5 for _ in range(mv.SPEC_BANDS))
+
+        def close(self):
+            closed.append(True)
+
+    w.set_spectro(True, FakeSource())
+    assert w.spectro and w.spec_src is not None
+    w.tick()
+    assert w._spec_frame is not None, "a published frame never reached the paint state"
+    w.set_spectro(False)
+    assert not w.spectro and w.spec_src is None
+    assert closed == [True], "switching the backdrop off left the capture running"
+    assert w._spec_frame is None
+    w.close()
+
+
+def test_closing_the_window_releases_the_audio_tap():
+    """A closed window must not outlive its own switch.
+
+    The X writes the close latch and systemd will not restart us, so nothing
+    would ever come back to turn a leaked `pw-record` off again.
+    """
+    import os
+    import pytest
+    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+    mv, w = _widget(1.0, 900)
+    if mv is None:
+        pytest.skip("no Qt available")
+    closed = []
+
+    class FakeSource:
+        err, frames = "", 0
+
+        def latest(self):
+            return None
+
+        def close(self):
+            closed.append(True)
+
+    w.set_spectro(True, FakeSource())
+    w.close()
+    assert closed == [True], "closeEvent left the capture on the master bus"
+
+
+def test_band_edges_never_leave_a_band_empty():
+    """Strictly increasing, in range, at every width the window can be.
+
+    A 2048-point FFT has fewer bins than we have bands below ~750 Hz, so the
+    naive log ramp produces duplicate edges — and a duplicated edge is a band
+    that sums zero bins, which paints as a black hole in the bass. "Dark = not
+    mapped" is the reading a cockpit trains you into (the LED work settled
+    that), so an empty band is worse than a coarse one.
+    """
+    import importlib.util
+    import sys as _sys
+    from pathlib import Path as _Path
+    root = _Path(__file__).resolve().parents[3]
+    spec = importlib.util.spec_from_file_location("_mv_bands",
+                                                  root / "tools/bridge/midiviz.py")
+    mv = importlib.util.module_from_spec(spec)
+    _sys.modules["_mv_bands"] = mv
+    spec.loader.exec_module(mv)
+    for bands in (12, 24, mv.SPEC_BANDS, 64, 96):
+        e = mv._spec_band_edges(bands, mv.SPEC_FFT, mv.SPEC_RATE)
+        assert len(e) == bands + 1
+        assert all(e[i] < e[i + 1] for i in range(bands)), \
+            f"an empty band at bands={bands}: {list(e)}"
+        assert e[0] >= 1 and e[-1] <= mv.SPEC_FFT // 2, \
+            f"edges ran off the spectrum at bands={bands}: {e[0]}..{e[-1]}"
+
+
+def test_the_analysis_seam_is_the_only_thing_the_paint_path_needs():
+    """`push()` carries the whole feature; the subprocess carries none of it.
+
+    This is what lets the backdrop be tested at all on a box with no audio
+    server, and it is also the honest boundary: green here says nothing about
+    `pw-record`, only that everything downstream of a block of samples works.
+    """
+    import math
+    import importlib.util
+    import sys as _sys
+    from pathlib import Path as _Path
+    import pytest
+    np = pytest.importorskip("numpy")
+    root = _Path(__file__).resolve().parents[3]
+    spec = importlib.util.spec_from_file_location("_mv_push",
+                                                  root / "tools/bridge/midiviz.py")
+    mv = importlib.util.module_from_spec(spec)
+    _sys.modules["_mv_push"] = mv
+    spec.loader.exec_module(mv)
+
+    src = mv.SpectrumSource(target="__never__")
+    t = np.arange(src.fft_n) / float(src.rate)
+    for _ in range(30):
+        frame = src.push(0.25 * np.sin(2 * math.pi * 3000.0 * t))
+    assert len(frame) == mv.SPEC_BANDS
+    assert all(0.0 <= v <= 1.0 for v in frame), "a band escaped 0..1"
+    hi = int(np.argmax(frame))
+    bin_hz = src.rate / float(src.fft_n)
+    assert src.edges[hi] * bin_hz <= 3000.0 <= src.edges[hi + 1] * bin_hz, \
+        "a 3 kHz tone did not land in the band that contains 3 kHz"
+    for _ in range(120):
+        quiet = src.push(np.zeros(src.fft_n, dtype=np.float32))
+    assert all(v == v for v in quiet), "silence produced NaN, which paints full-height"
+    assert max(quiet) < 0.02, f"silence read as signal: {max(quiet):.3f}"
+    src.close()                          # never started; must still be safe
+
+
+def test_the_backdrop_is_painted_behind_everything_else():
+    """Ordering IS the feature — "overlay behind a basic spectro"."""
+    import os
+    import pytest
+    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+    mv, w = _widget(1.0, 900)
+    if mv is None:
+        pytest.skip("no Qt available")
+    order = []
+    for name in ("_paint_spectrum", "_paint_chrome", "_paint_header",
+                 "_paint_matrix", "_paint_gutter", "_paint_stream"):
+        real = getattr(w, name)
+
+        def probe(*a, _n=name, _r=real, **k):
+            order.append(_n)
+            return _r(*a, **k)
+        setattr(w, name, probe)
+
+    class FakeSource:
+        err, frames = "", 0
+
+        def latest(self):
+            return tuple(0.6 for _ in range(mv.SPEC_BANDS))
+
+        def close(self):
+            pass
+
+    w.set_spectro(True, FakeSource())
+    w.tick()
+    w.grab()
+    assert order and order[0] == "_paint_spectrum", \
+        f"the spectrum is not the first layer painted: {order}"
+    w.set_spectro(False)
+    w.close()
+
+
+def test_a_short_pipe_read_is_not_mistaken_for_end_of_stream():
+    """The bug that made the whole backdrop dead while the selftest passed.
+
+    `Popen(bufsize=0)` hands back a raw `FileIO`: `read(n)` is one `os.read`
+    and returns whatever the pipe holds, so a hop-sized read is short far more
+    often than not. Treating that as EOF ended the capture on its first chunk
+    (`blocks=0`, no error recorded) — and every pure-function test still went
+    green, because none of them touched the pipe.
+    """
+    import io
+    import importlib.util
+    import sys as _sys
+    from pathlib import Path as _Path
+    root = _Path(__file__).resolve().parents[3]
+    spec = importlib.util.spec_from_file_location("_mv_read",
+                                                  root / "tools/bridge/midiviz.py")
+    mv = importlib.util.module_from_spec(spec)
+    _sys.modules["_mv_read"] = mv
+    spec.loader.exec_module(mv)
+
+    class Dribble(io.RawIOBase):
+        """A pipe that never gives you as much as you asked for."""
+
+        def __init__(self, data, most=7):
+            self.buf, self.most = bytearray(data), most
+
+        def read(self, n=-1):
+            take = min(n if n and n > 0 else 1, self.most, len(self.buf))
+            out, self.buf = bytes(self.buf[:take]), self.buf[take:]
+            return out
+
+    payload = bytes(range(256)) * 8              # 2048 bytes
+    got = mv._read_exact(Dribble(payload), len(payload))
+    assert got == payload, "a dribbling stream lost or truncated bytes"
+    # and the real end of stream still reports itself as one
+    assert mv._read_exact(Dribble(b""), 16) is None
+    assert mv._read_exact(Dribble(payload[:100]), len(payload)) is None, \
+        "a stream that ends mid-block must read as EOF, not a partial frame"

tools/midiviz.service

@@ -20,6 +20,11 @@ StartLimitIntervalSec=0
 # Inherits the Wayland/Plasma session env from the user manager.
 Environment=PYTHONUNBUFFERED=1
 ExecStart=/usr/bin/python3 %h/Work/Sound/Tidal/tools/bridge/midiviz.py
+# No --spectro here ON PURPOSE. The spectrum backdrop is an AUDIO consumer,
+# and this unit is Restart=always and starts with the session -- putting the
+# flag here would make the rig acquire a permanent tap on the master bus at
+# login, which is precisely what the feature is not allowed to be. It is a
+# runtime choice: `s` in the window, or a hand-launched --spectro.
 # 2026-09-05: an unplugged LCXL used to make midiviz.py exit CLEANLY (the
 # `aseqdump` it shelled out to hit EOF and nothing kept the window open), and
 # `Restart=on-failure` never fires on a clean exit -- PLN's window was gone