perf(bridge): the MIDI monitor had the LED bug's twin — it dropped the NEWEST event

Found by sweeping the rest of the gear for #71's bug class, which turned out to have a
sibling living in the Bridge.

`MidiStream._reader` did `q.put_nowait(ev)` and on `queue.Full` silently `pass`ed. That
keeps 512 stale events and throws away the one that just happened -- exactly backwards
for MIDI state, where the newest value IS the truth. A dashboard tab that stalled for
~1.3 s at 400 CC/s would fill its queue and then display frozen values for the rest of
the set, with no error logged anywhere. Same failure shape as the LEDs: nothing breaks,
it just quietly stops telling you the truth.

Downstream, `_sse_midi` wrote and flushed once per event, so one fader sweep cost ~400
HTTP flushes a second per open tab, each with its own json.dumps.

Fixed the way the LCXL painter was: drop-OLDEST on a full queue, and batch a frame's
worth of events into a single write. New `coalesce()` folds the batch on one rule --
CONTINUOUS controls (CC, pitchbend, aftertouch) are STATE and may be superseded by a
newer value; NOTES are EVENTS and may never be dropped. That distinction is the whole
point: a fast monitor that loses a note is strictly worse than a slow one, so the tests
assert every note survives a flood of 100 controller messages.

Order is preserved by overwriting a superseded value where it stood rather than moving
it to the end, so the monitor still reads as a timeline. The wire format is unchanged
(one `data:` line per event), so ui/index.html's onmessage/JSON.parse is untouched.

Also extracted `_fanout` so the drop policy is testable without ALSA. A green test on
`parse_line` proved nothing about the queue behaviour behind it -- verify the seam.

8 new tests, 32 green in tools/bridge.

tools/bridge/midistream.py

@@ -18,6 +18,44 @@ import midimon
 
 _PORT = re.compile(r"^\s*(\d+:\d+)\s+(.+?)\s{2,}(.+?)\s*$")
 
+# Which events are STATE and which are EVENTS. The distinction is the whole of
+# `coalesce`: a controller's older value is worthless the moment a newer one exists, but
+# a note you drop never happened. Getting this backwards would make the monitor lie
+# about what was played, which is worse than making it slow.
+_CONTINUOUS = ("control change", "pitchbend", "pitch bend", "aftertouch",
+               "channel aftertouch", "poly aftertouch", "control")
+
+
+def _coalesce_key(ev: dict):
+    """A key for events that supersede each other, or None if the event is discrete."""
+    e = (ev.get("event") or "").lower()
+    if not e.startswith(_CONTINUOUS):
+        return None
+    if e.startswith(("control change", "control")):
+        return "cc", ev.get("source"), ev.get("ch"), ev.get("controller")
+    return e.split()[0], ev.get("source"), ev.get("ch")
+
+
+def coalesce(events: list[dict]) -> list[dict]:
+    """Fold a batch: every discrete event survives, continuous ones keep only the last.
+
+    Order is preserved by overwriting in place at the first occurrence, so the monitor
+    still reads chronologically -- a superseded fader value is replaced where it stood,
+    not moved to the end.
+    """
+    out: list[dict] = []
+    at: dict = {}
+    for ev in events:
+        k = _coalesce_key(ev)
+        if k is None:
+            out.append(ev)
+        elif k in at:
+            out[at[k]] = ev
+        else:
+            at[k] = len(out)
+            out.append(ev)
+    return out
+
 
 def parse_ports(text: str) -> list[dict]:
     """Parse `aseqdump -l` output into [{addr, client, port}] (header skipped)."""
@@ -77,12 +115,24 @@ class MidiStream:
             ev = midimon.parse_line(line)
             if not ev:
                 continue
-            ev = midimon.enrich(ev)
-            with self._lock:
-                for q in self._subs:
+            self._fanout(midimon.enrich(ev))
+
+    def _fanout(self, ev: dict) -> None:
+        """Hand one event to every subscriber. Its own method so the drop policy is
+        testable without ALSA -- green tests on a parser prove nothing about the seam."""
+        with self._lock:
+            for q in self._subs:
+                try:
+                    q.put_nowait(ev)
+                except queue.Full:
+                    # Drop the OLDEST, not the newest. The original `except Full: pass`
+                    # kept 512 stale events and threw away the one that had just
+                    # happened, so a tab that stalled once showed frozen values forever
+                    # with no error anywhere. For MIDI state the newest event IS truth.
                     try:
+                        q.get_nowait()
                         q.put_nowait(ev)
-                    except queue.Full:
+                    except (queue.Empty, queue.Full):
                         pass
 
     def _stop(self):

tools/bridge/server.py

@@ -26,6 +26,7 @@ import os
 import queue
 import socket
 import sys
+import time
 from functools import partial
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
@@ -89,9 +90,21 @@ class Handler(BaseHTTPRequestHandler):
         ThreadingHTTPServer gives each connection its own thread, so blocking on
         the subscriber queue here is fine. A ': ping' every 15 s keeps the
         connection (and any proxy) from idling out.
+
+        Batched, for the same reason the LCXL painter is (#71/#72): one write+flush per
+        event meant a fader sweep cost ~400 HTTP flushes a second per open tab. Here we
+        block for the FIRST event, then sweep up everything already waiting, coalesce
+        away superseded controller values, and emit the lot in one write. The wire
+        format is unchanged -- still one `data:` line per event -- so the UI's
+        `onmessage`/`JSON.parse` is untouched.
+
+        The rate limit is a floor on the gap BETWEEN frames, not a fixed tick, so an
+        idle stream still delivers a single note the instant it arrives.
         """
         port = (q.get("port") or [None])[0] or None
         sub = _MIDI.subscribe(port)
+        frame = 1.0 / 30.0
+        next_ok = 0.0
         try:
             self.send_response(200)
             self.send_header("Content-Type", "text/event-stream")
@@ -102,11 +115,24 @@ class Handler(BaseHTTPRequestHandler):
             self.wfile.flush()
             while True:
                 try:
-                    ev = sub.get(timeout=15)
-                    self.wfile.write(("data: " + json.dumps(ev) + "\n\n").encode())
+                    batch = [sub.get(timeout=15)]
                 except queue.Empty:
                     self.wfile.write(b": ping\n\n")
+                    self.wfile.flush()
+                    continue
+                gap = next_ok - time.monotonic()
+                if gap > 0:
+                    time.sleep(gap)          # the coalescing window
+                while len(batch) < 1024:
+                    try:
+                        batch.append(sub.get_nowait())
+                    except queue.Empty:
+                        break
+                payload = "".join("data: " + json.dumps(e) + "\n\n"
+                                  for e in MS.coalesce(batch))
+                self.wfile.write(payload.encode())
                 self.wfile.flush()
+                next_ok = time.monotonic() + frame
         except (BrokenPipeError, ConnectionResetError, OSError):
             pass  # client navigated away / closed the tab — normal
         finally:

tools/bridge/tests/test_midistream.py

-"""Pure-parser test for midistream.parse_ports (no ALSA)."""
+"""Pure tests for midistream: port parsing, and the coalescer (no ALSA).
+
+The coalescer's one rule is the only thing standing between "the monitor is fast" and
+"the monitor lies": continuous controls are STATE and may be superseded, notes are
+EVENTS and may never be dropped. Get that backwards and a fast monitor silently loses
+what was played, which is strictly worse than a slow one.
+"""
+import queue
 import sys
 from pathlib import Path
 
@@ -23,3 +30,71 @@ def test_parse_ports_skips_header_and_parses_rows():
 
 def test_parse_ports_empty():
     assert MS.parse_ports("") == []
+
+
+# ------------------------------------------------------------------ coalesce
+
+def cc(controller, value, ch=0, source="24:0"):
+    return {"source": source, "event": "Control change", "ch": ch,
+            "controller": controller, "value": value, "data": ""}
+
+
+def note(n, vel=100, on=True, ch=0, source="24:0"):
+    return {"source": source, "event": "Note on" if on else "Note off", "ch": ch,
+            "note": n, "velocity": vel, "data": ""}
+
+
+def test_a_fader_sweep_collapses_to_its_final_value():
+    out = MS.coalesce([cc(77, v) for v in range(128)])
+    assert len(out) == 1 and out[0]["value"] == 127
+
+
+def test_every_note_survives_even_a_flood_of_controllers():
+    """Notes are what was PLAYED. Losing one to a fader sweep is unforgivable."""
+    batch = [note(60), *(cc(77, v) for v in range(100)), note(64), note(60, on=False)]
+    out = MS.coalesce(batch)
+    assert [e for e in out if "note" in e] == [note(60), note(64), note(60, on=False)]
+
+
+def test_distinct_controllers_do_not_collapse_into_each_other():
+    out = MS.coalesce([cc(77, 10), cc(78, 20), cc(77, 30)])
+    assert len(out) == 2
+    assert {e["controller"]: e["value"] for e in out} == {77: 30, 78: 20}
+
+
+def test_same_controller_on_a_different_channel_or_device_is_a_different_control():
+    out = MS.coalesce([cc(77, 1, ch=0), cc(77, 2, ch=1), cc(77, 3, source="28:0")])
+    assert len(out) == 3
+
+
+def test_chronological_order_is_preserved():
+    """A superseded value is replaced WHERE IT STOOD, so the monitor still reads as a
+    timeline rather than reshuffling history to the end."""
+    out = MS.coalesce([cc(77, 1), note(60), cc(77, 2), note(62)])
+    assert [e.get("controller", e.get("note")) for e in out] == [77, 60, 62]
+    assert out[0]["value"] == 2
+
+
+def test_coalesce_is_a_no_op_on_an_already_sparse_batch():
+    batch = [note(60), cc(77, 5), note(60, on=False)]
+    assert MS.coalesce(batch) == batch
+
+
+def test_unknown_event_kinds_are_treated_as_discrete():
+    """Unrecognised == keep. A coalescer that guesses wrong should lose nothing."""
+    weird = {"source": "24:0", "event": "System exclusive", "ch": None, "data": "F0"}
+    assert MS.coalesce([weird, weird]) == [weird, weird]
+
+
+# ---------------------------------------------------- the subscriber queue
+
+def test_a_full_queue_drops_the_OLDEST_event_not_the_newest():
+    """The original `except queue.Full: pass` kept 512 stale events and threw away the
+    one that had just happened, so a tab that stalled once froze forever."""
+    stream = MS.MidiStream()
+    q: queue.Queue = queue.Queue(maxsize=4)
+    stream._subs.add(q)
+    for v in range(10):
+        stream._fanout(cc(77, v))
+    got = [q.get_nowait()["value"] for _ in range(q.qsize())]
+    assert got == [6, 7, 8, 9], f"kept {got}; the newest value must survive"