feat(bridge): midistream groundwork for the live MIDI dashboard panel

MidiStream fans one aseqdump subprocess out to N web subscribers (SSE), reusing
midimon.parse_line + enrich (one parsing source of truth); reader runs only
while watched (lazy start / stop on last unsubscribe, so no MIDI port held idle).
parse_ports is pure + tested. NOT yet wired into server/UI — see resume task.

Parked mid-feature: server SSE endpoint + dashboard panel still to do.

tools/bridge/midistream.py

+"""midistream — fan live, parsed MIDI events out to web subscribers (SSE).
+
+One `aseqdump` subprocess feeds N dashboard tabs. Reuses `midimon.parse_line` +
+`enrich` (one parsing source of truth). The reader runs only while someone is
+watching: started on first subscribe, stopped when the last subscriber leaves —
+so we don't hold a MIDI port open for nothing.
+
+`parse_ports` is pure (tested without ALSA).
+"""
+from __future__ import annotations
+
+import queue
+import re
+import subprocess
+import threading
+
+import midimon
+
+_PORT = re.compile(r"^\s*(\d+:\d+)\s+(.+?)\s{2,}(.+?)\s*$")
+
+
+def parse_ports(text: str) -> list[dict]:
+    """Parse `aseqdump -l` output into [{addr, client, port}] (header skipped)."""
+    out = []
+    for line in text.splitlines():
+        m = _PORT.match(line)
+        if m and ":" in m.group(1):           # header "Port  Client name…" has no digits:digits
+            out.append({"addr": m.group(1), "client": m.group(2).strip(),
+                        "port": m.group(3).strip()})
+    return out
+
+
+class MidiStream:
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._subs: set[queue.Queue] = set()
+        self._proc = None
+        self._port = None
+
+    def list_ports(self) -> list[dict]:
+        try:
+            r = subprocess.run(["aseqdump", "-l"], capture_output=True, text=True, timeout=5)
+        except (OSError, subprocess.SubprocessError):
+            return []
+        return parse_ports(r.stdout)
+
+    def subscribe(self, port=None) -> queue.Queue:
+        q: queue.Queue = queue.Queue(maxsize=512)
+        with self._lock:
+            self._subs.add(q)
+            restart = self._proc is None or (port and port != self._port)
+        if restart:
+            self._start(port)
+        return q
+
+    def unsubscribe(self, q: queue.Queue):
+        with self._lock:
+            self._subs.discard(q)
+            stop = not self._subs
+        if stop:
+            self._stop()
+
+    # ── internals ───────────────────────────────────────────────────────
+    def _start(self, port):
+        self._stop()
+        cmd = ["aseqdump"] + (["-p", port] if port else [])
+        try:
+            self._proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, bufsize=1)
+        except OSError:
+            self._proc = None
+            return
+        self._port = port
+        threading.Thread(target=self._reader, args=(self._proc,), daemon=True).start()
+
+    def _reader(self, proc):
+        for line in proc.stdout or []:
+            ev = midimon.parse_line(line)
+            if not ev:
+                continue
+            ev = midimon.enrich(ev)
+            with self._lock:
+                for q in self._subs:
+                    try:
+                        q.put_nowait(ev)
+                    except queue.Full:
+                        pass
+
+    def _stop(self):
+        if self._proc:
+            try:
+                self._proc.terminate()
+            except OSError:
+                pass
+        self._proc = None
+        self._port = None

tools/bridge/tests/test_midistream.py

+"""Pure-parser test for midistream.parse_ports (no ALSA)."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+import midistream as MS
+
+SAMPLE = """ Port    Client name                      Port name
+  0:0    System                           Timer
+  0:1    System                           Announce
+ 14:0    Midi Through                     Midi Through Port-0
+ 28:0    nanoKONTROL2                     nanoKONTROL2 MIDI 1
+"""
+
+
+def test_parse_ports_skips_header_and_parses_rows():
+    ports = MS.parse_ports(SAMPLE)
+    assert {"addr": "28:0", "client": "nanoKONTROL2", "port": "nanoKONTROL2 MIDI 1"} in ports
+    assert all(":" in p["addr"] for p in ports)
+    assert not any(p["client"] == "Client name" for p in ports)  # header skipped
+    assert len(ports) == 4
+
+
+def test_parse_ports_empty():
+    assert MS.parse_ports("") == []