diff --git a/tests/test_beat_detector.py b/tests/test_beat_detector.py
new file mode 100644
index 0000000..191cca9
--- /dev/null
+++ b/tests/test_beat_detector.py
@@ -0,0 +1,297 @@
+"""
+Tests for BeatDetector — AUD-03 (system audio capture) and AUD-04 (beat detection).
+TDD RED phase: these tests define expected behavior before implementation.
+
+Threading safety rules (D-04, D-05):
+- _audio_callback MUST NOT call asyncio APIs directly
+- Beat events MUST travel through call_soon_threadsafe
+"""
+import asyncio
+import inspect
+import threading
+import time
+import pytest
+
+
+# ─── Structural tests (no audio device required) ────────────────────────────
+
+def test_beat_detector_importable():
+    """BeatDetector must be importable from led_sync.audio."""
+    from led_sync.audio import BeatDetector
+    assert BeatDetector is not None
+
+
+def test_beat_detector_importable_from_module():
+    """BeatDetector must be importable from led_sync.audio.beat_detector."""
+    from led_sync.audio.beat_detector import BeatDetector
+    assert BeatDetector is not None
+
+
+def test_constants_present():
+    """Module constants must match latency specification."""
+    from led_sync.audio.beat_detector import SAMPLE_RATE, HOP_SIZE, BUF_SIZE
+    assert SAMPLE_RATE == 44100
+    assert HOP_SIZE == 512
+    assert BUF_SIZE == 1024
+
+
+def test_latency_spec():
+    """HOP_SIZE / SAMPLE_RATE must yield < 12ms latency (target ~11ms)."""
+    from led_sync.audio.beat_detector import SAMPLE_RATE, HOP_SIZE
+    latency_ms = HOP_SIZE / SAMPLE_RATE * 1000
+    assert latency_ms < 12.0, f"Latency {latency_ms:.2f}ms exceeds 11ms spec"
+
+
+@pytest.mark.asyncio
+async def test_initial_state():
+    """beat_count starts at 0, bpm starts at 0.0."""
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.get_running_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+    assert detector.beat_count == 0
+    assert detector.bpm == 0.0
+
+
+@pytest.mark.asyncio
+async def test_stop_before_start_is_noop():
+    """Calling stop() before start() must not raise any exception."""
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.get_running_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+    # Must not raise
+    detector.stop()
+    detector.stop()  # second call also noop
+
+
+@pytest.mark.asyncio
+async def test_double_stop_is_noop():
+    """Calling stop() twice must not raise even if stream was opened."""
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.get_running_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+    detector.stop()
+    # Shouldn't raise
+    detector.stop()
+
+
+def test_beat_count_property_is_thread_safe():
+    """beat_count property must be accessible from multiple threads (uses lock)."""
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.new_event_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+
+    errors = []
+    results = []
+
+    def read_count():
+        try:
+            results.append(detector.beat_count)
+        except Exception as e:
+            errors.append(e)
+
+    threads = [threading.Thread(target=read_count) for _ in range(10)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert not errors, f"Thread safety errors: {errors}"
+    assert all(r == 0 for r in results)
+    loop.close()
+
+
+def test_bpm_property_is_thread_safe():
+    """bpm property must be accessible from multiple threads (uses lock)."""
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.new_event_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+
+    errors = []
+    results = []
+
+    def read_bpm():
+        try:
+            results.append(detector.bpm)
+        except Exception as e:
+            errors.append(e)
+
+    threads = [threading.Thread(target=read_bpm) for _ in range(10)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert not errors, f"Thread safety errors: {errors}"
+    assert all(r == 0.0 for r in results)
+    loop.close()
+
+
+# ─── Audio callback thread-safety check ─────────────────────────────────────
+
+def test_audio_callback_uses_call_soon_threadsafe():
+    """
+    _audio_callback source code must use call_soon_threadsafe (not asyncio APIs directly).
+    This is a static analysis check enforcing D-05.
+    """
+    import inspect
+    from led_sync.audio import beat_detector as mod
+
+    source = inspect.getsource(mod)
+    # Must contain the thread-safe bridge
+    assert "call_soon_threadsafe" in source, (
+        "_audio_callback must use call_soon_threadsafe to bridge audio thread -> asyncio"
+    )
+    # Must NOT call asyncio.create_task or asyncio.ensure_future directly in callback
+    # (these would be wrong — asyncio is not thread-safe to call directly)
+    # We check by ensuring there's no bare asyncio. call in the source
+    # (call_soon_threadsafe is the correct pattern)
+    lines = source.splitlines()
+    callback_lines = []
+    in_callback = False
+    brace_depth = 0
+    for line in lines:
+        if "def _audio_callback" in line:
+            in_callback = True
+        if in_callback:
+            callback_lines.append(line)
+            # Stop at next non-indented def (crude but sufficient)
+            if len(callback_lines) > 1 and line.strip().startswith("def "):
+                break
+
+    callback_text = "\n".join(callback_lines)
+    # Must not directly call asyncio.run() or asyncio.get_event_loop() in callback
+    assert "asyncio.run(" not in callback_text
+    assert "asyncio.get_event_loop()" not in callback_text
+
+
+# ─── Simulated beat injection (no real audio device) ────────────────────────
+
+@pytest.mark.asyncio
+async def test_beat_callback_called_via_asyncio_bridge():
+    """
+    Simulate what the audio thread does: call_soon_threadsafe with on_beat coroutine.
+    Verify beat_count increments and on_beat is called in asyncio context.
+    """
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.get_running_loop()
+    received_beats = []
+    received_threads = []
+
+    async def on_beat(beat_time: float):
+        received_beats.append(beat_time)
+        received_threads.append(threading.current_thread())
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+
+    # Simulate what _audio_callback does internally when a beat is detected
+    # We manually trigger the bridge to test it works correctly
+    with detector._lock:
+        detector._beat_count += 1
+        detector._bpm = 120.0
+
+    loop.call_soon_threadsafe(
+        loop.create_task,
+        on_beat(1.0),
+    )
+
+    # Give asyncio a chance to process
+    await asyncio.sleep(0.05)
+
+    assert len(received_beats) == 1
+    assert received_beats[0] == 1.0
+    # on_beat should run in the asyncio thread (main thread in test)
+    assert received_threads[0] == threading.main_thread()
+
+
+@pytest.mark.asyncio
+async def test_beat_count_increments_correctly():
+    """beat_count reflects simulated beats correctly."""
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.get_running_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+    assert detector.beat_count == 0
+
+    # Simulate 3 beats
+    for i in range(3):
+        with detector._lock:
+            detector._beat_count += 1
+
+    assert detector.beat_count == 3
+
+
+@pytest.mark.asyncio
+async def test_bpm_updated():
+    """bpm reflects simulated BPM correctly."""
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.get_running_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+    assert detector.bpm == 0.0
+
+    with detector._lock:
+        detector._bpm = 128.5
+
+    assert detector.bpm == 128.5
+
+
+# ─── ImportError handling ────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_start_raises_runtime_error_on_missing_deps(monkeypatch):
+    """start() must raise RuntimeError if sounddevice or aubio cannot be imported."""
+    import builtins
+    from led_sync.audio.beat_detector import BeatDetector
+
+    loop = asyncio.get_running_loop()
+
+    async def on_beat(beat_time: float):
+        pass
+
+    detector = BeatDetector(loop=loop, on_beat=on_beat)
+
+    original_import = builtins.__import__
+
+    def mock_import(name, *args, **kwargs):
+        if name == "sounddevice":
+            raise ImportError("mocked missing sounddevice")
+        return original_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "__import__", mock_import)
+
+    with pytest.raises(RuntimeError, match="sounddevice"):
+        detector.start()