feat(spectrum): capture macOS output with a Core Audio process tap

SYNC mode on macOS needed a virtual loopback driver (BlackHole, Loopback,
Soundflower) routed through a Multi-Output Device, and shelled out to
'ffmpeg -f avfoundation' to read it. macOS 14.2 lets a process tap the
system mix directly, so do that instead: CATapDescription ->
AudioHardwareCreateProcessTap -> private aggregate device -> IOProc,
straight through ctypes. No driver, no helper binary, no ffmpeg.

The tap hands out native-rate interleaved float32, so the analyser can
now retune its window and band edges when the capture rate differs from
44.1 kHz, and mono-downmixes the frames before the FFT.

A tap that was never granted System Audio Recording permission is
created happily but only ever delivers digital silence: treat that as a
failed capture instead of flat bars, explain it in the status bar and
fall back to the stream tap. A tap that renders nothing *yet* (playback
still starting, or mpv pointed at another device) is not a permanent
failure either - the stream tap covers for a minute, then the tap is
retried.

Drop the avfoundation device enumeration, the loopback-name heuristics
and the microphone safeguards' tests, and document the new permission
and its diagnostics instead.

tests/test_macos_tap.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import pytest
+
+from somafm_tui import macos_tap
+
+
+def test_osstatus_decodes_four_character_codes() -> None:
+    # 'who?' is kAudioHardwareBadObjectError.
+    assert macos_tap._osstatus(int.from_bytes(b"who?", "big")) == "'who?'"
+    assert macos_tap._osstatus(-50) == "-50"
+
+
+def test_start_raises_without_the_native_api(monkeypatch) -> None:
+    monkeypatch.setattr(macos_tap, "available", lambda: False)
+    tap = macos_tap.SystemAudioTap()
+    with pytest.raises(macos_tap.TapError):
+        tap.start()
+    assert tap.running is False
+    tap.stop()  # tearing down a tap that never started is safe
+
+
+def test_read_returns_none_when_not_running() -> None:
+    tap = macos_tap.SystemAudioTap()
+    assert tap.read(0.0) is None
+
+
+def test_uses_no_external_helpers() -> None:
+    # The tap must never shell out: no helper binary, no ffmpeg, no driver.
+    import inspect
+
+    source = inspect.getsource(macos_tap).casefold()
+    assert "subprocess" not in source
+    assert "avfoundation" not in source

tests/test_spectrum.py

@@ -29,6 +29,55 @@ def write_pcm_script(path: Path) -> None:
     path.chmod(0o755)
 
 
+def tap_pcm(
+    freq: float = 1000,
+    frames: int = 4096,
+    sample_rate: int = 48000,
+    channels: int = 2,
+) -> bytes:
+    """Interleaved float32 stereo PCM, as the macOS tap delivers it."""
+    samples = np.arange(frames) / sample_rate
+    tone = (0.5 * np.sin(2 * np.pi * freq * samples)).astype("<f4")
+    return np.repeat(tone, channels).tobytes()
+
+
+class FakeTap:
+    """Stand-in for ``somafm_tui.macos_tap.SystemAudioTap``."""
+
+    def __init__(
+        self,
+        *,
+        silent: bool = False,
+        idle: bool = False,
+        error: Exception | None = None,
+        sample_rate: int = 48000,
+        channels: int = 2,
+    ) -> None:
+        self.sample_rate = sample_rate
+        self.channels = channels
+        self.bytes_per_frame = 4 * channels
+        self.stopped = False
+        self._idle = idle
+        self._error = error
+        self._chunk = (
+            b"\x00" * (4096 * self.bytes_per_frame)
+            if silent
+            else tap_pcm(1000, 4096, sample_rate, channels)
+        )
+
+    def start(self) -> None:
+        if self._error is not None:
+            raise self._error
+
+    def read(self, timeout: float | None = 1.0) -> bytes | None:
+        if self.stopped or self._idle:
+            return None
+        return self._chunk
+
+    def stop(self) -> None:
+        self.stopped = True
+
+
 def wait_for_spectrum(engine: SpectrumEngine, status: str, timeout: float = 6.0):
     deadline = time.monotonic() + timeout
     snap = engine.snapshot()
@@ -140,91 +189,118 @@ def test_stream_tap_pipeline(tmp_path) -> None:
         engine.stop()
 
 
-def test_avfoundation_loopback_is_preferred_on_macos(monkeypatch) -> None:
-    engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg")
-    engine._parec = None  # pretend there is no PulseAudio on this host
+def test_macos_tap_is_the_sync_monitor(monkeypatch) -> None:
+    engine = SpectrumEngine(ffmpeg_path="")
+    engine._parec = None
     engine._darwin = True
-    monkeypatch.delenv("SOMAFM_SPECTRUM_DEVICE", raising=False)
-    monkeypatch.setattr(
-        engine,
-        "_avf_audio_devices",
-        lambda: [(0, "MacBook Pro Microphone"), (1, "BlackHole 2ch")],
-    )
-    engine._avf_checked = False
-
-    assert engine._avf_loopback() == 1
-    assert engine._monitor_kind() == "avfoundation"
+    monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
+    tap = FakeTap()
+    monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
     assert engine.monitor_available is True
+    assert engine._monitor_kind() == "tap"
 
-    command = engine._monitor_command()
-    assert command is not None
-    assert command[0] == "/opt/homebrew/bin/ffmpeg"
-    assert command[command.index("-i") + 1] == ":1"
-    assert "-f" in command and "avfoundation" in command
-    assert command[-1] == "-"
-    assert "pcm_s16le" in command
-    assert command[command.index("-ar") + 1] == "44100"
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/live")
+        snap = wait_for_spectrum(engine, "monitor")
+        assert snap.status == "monitor", snap
+        assert snap.label == "SYNC"
+        top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
+        assert 36 <= top <= 42
+        assert snap.bands[top] > 0.5
+
+        engine.set_source(None)
+        time.sleep(0.3)
+        assert engine.snapshot().status == "idle"
+        assert tap.stopped is True
+    finally:
+        engine.stop()
 
 
-def test_avfoundation_never_autoselects_a_microphone(monkeypatch) -> None:
-    engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg")
+def test_silent_tap_falls_back_instead_of_flat_bars(monkeypatch) -> None:
+    # A tap that was never granted System Audio Recording permission is
+    # created happily but only ever hands out digital silence.
+    engine = SpectrumEngine(ffmpeg_path="")
     engine._parec = None
     engine._darwin = True
-    monkeypatch.delenv("SOMAFM_SPECTRUM_DEVICE", raising=False)
+    monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
+    tap = FakeTap(silent=True)
+    monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
+
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/live")
+        snap = wait_for_spectrum(engine, "sim")
+        assert snap.status == "sim", snap
+        assert engine.monitor_available is False
+        assert tap.stopped is True
+        notice = engine.take_notice()
+        assert notice is not None and "System Audio Recording" in notice
+        assert engine.take_notice() is None
+    finally:
+        engine.stop()
+
+
+def test_tap_start_failure_falls_back_and_explains(monkeypatch) -> None:
+    engine = SpectrumEngine(ffmpeg_path="")
+    engine._parec = None
+    engine._darwin = True
+    monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
     monkeypatch.setattr(
-        engine, "_avf_audio_devices", lambda: [(0, "MacBook Pro Microphone")]
-    )
-    engine._avf_checked = False
-    assert engine._avf_loopback() is None
-    assert engine._monitor_kind() is None
-    assert engine._monitor_command() is None
-
-
-def test_avfoundation_device_override_by_name_and_index(monkeypatch) -> None:
-    devices = [(0, "MacBook Pro Microphone"), (1, "BlackHole 2ch"), (2, "Loopback Audio")]
-
-    engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg")
-    engine._parec = None
-    engine._darwin = True
-    monkeypatch.setattr(engine, "_avf_audio_devices", lambda: devices)
-    monkeypatch.setenv("SOMAFM_SPECTRUM_DEVICE", "loopback audio")
-    engine._avf_checked = False
-    assert engine._avf_loopback() == 2
-
-    engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg")
-    engine._parec = None
-    engine._darwin = True
-    monkeypatch.setattr(engine, "_avf_audio_devices", lambda: devices)
-    monkeypatch.setenv("SOMAFM_SPECTRUM_DEVICE", "0")
-    engine._avf_checked = False
-    assert engine._avf_loopback() == 0
-
-
-def test_avfoundation_listing_parser(monkeypatch) -> None:
-    listing = (
-        "[AVFoundation indev @ 0x1] AVFoundation video devices:\n"
-        "[AVFoundation indev @ 0x1] [0] Capture screen 0\n"
-        "[AVFoundation indev @ 0x1] AVFoundation audio devices:\n"
-        "[AVFoundation indev @ 0x1] [0] MacBook Pro Microphone\n"
-        "[AVFoundation indev @ 0x1] [1] BlackHole 2ch\n"
+        spectrum_module.macos_tap,
+        "SystemAudioTap",
+        lambda: FakeTap(error=spectrum_module.macos_tap.TapError("boom")),
     )
 
-    class Result:
-        stdout = listing
-        returncode = 1
-
-    def fake_run(*args, **kwargs):
-        return Result()
-
-    engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg")
-    monkeypatch.setattr(spectrum_module.subprocess, "run", fake_run)
-    assert engine._avf_audio_devices() == [
-        (0, "MacBook Pro Microphone"),
-        (1, "BlackHole 2ch"),
-    ]
+    engine.start()
+    try:
+        engine.set_source("https://example.invalid/live")
+        snap = wait_for_spectrum(engine, "sim")
+        assert snap.status == "sim", snap
+        assert engine.monitor_available is False
+        notice = engine.take_notice()
+        assert notice is not None and "macOS audio capture failed" in notice
+    finally:
+        engine.stop()
 
 
-def test_dead_loopback_falls_back_instead_of_flat_bars(tmp_path) -> None:
+def test_idle_tap_retries_instead_of_latching(monkeypatch) -> None:
+    # If the output device renders nothing yet (slow stream, or mpv pointed at
+    # another device) the stream tap covers us and the tap is retried later.
+    monkeypatch.setattr(spectrum_module, "TAP_IDLE_SECONDS", 0.05)
+    engine = SpectrumEngine(ffmpeg_path="")
+    engine._parec = None
+    engine._darwin = True
+    monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
+    tap = FakeTap(idle=True)
+    monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
+
+    engine.start()
+    try:
+        engine.set_source("/tmp/tone.wav")
+        snap = wait_for_spectrum(engine, "sim")
+        assert snap.status == "sim", snap
+        assert engine.monitor_available is True  # not a permanent failure
+        assert engine._tap_retry_at > time.monotonic()
+        notice = engine.take_notice()
+        assert notice is not None and "default output device" in notice
+    finally:
+        engine.stop()
+
+
+def test_tap_audio_is_downmixed_and_retuned() -> None:
+    engine = SpectrumEngine(ffmpeg_path="")
+    engine._retune(48000)
+    engine.feed_audio(tap_pcm(1000, sample_rate=48000), channels=2)
+
+    snap = engine.snapshot()
+    assert snap.status == "monitor"
+    top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
+    assert 36 <= top <= 42
+    assert snap.bands[top] > 0.8
+
+
+def test_dead_monitor_falls_back_instead_of_flat_bars(tmp_path) -> None:
     script = tmp_path / "silent-parec"
     script.write_text(
         "#!/usr/bin/env python3\n"