fix: soundcard capture detects a dead arecord instead of silently recording nothing

ALSAStream._read_loop just returned when arecord died or hit EOF, and
__exit__ never checked the exit status. SoundcardRecorder.record's inner
loop only slept and flushed, so nothing ever noticed: the recorder stayed
"running", kept rolling over to a new split file on schedule, and wrote
zero bytes into every one of them. The max_retries/reconnect path below
it was unreachable. Worst kind of failure for an unattended recorder —
files keep appearing, at the right times, with the right names.

ALSAStream now tracks when audio last arrived and exposes check_alive(),
which raises if the process exited or nothing has been captured for
SILENCE_TIMEOUT (15 s). The record loop calls it every 0.5 s tick, so a
dead device unwinds into the existing retry handler.

retry_count resets only when the stream actually delivered audio —
resetting on every healthy tick would let a device that opens but never
captures retry forever, which is the same silent failure by another
route.

Also: join the stderr thread in __exit__, and break _read_loop on EOF
rather than looping until poll() catches up.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.gitignore

@@ -7,3 +7,4 @@ __pycache__/
 .pytest_cache/
 .claude/
 nul
+.claude-notes/

README.md

@@ -208,6 +208,8 @@ Analysis, clips, cut, and delete return `409` for files that are still being rec
 
 Both recorder types run in separate threads and retry independently up to `max_retries`.
 
+Soundcard capture is health-checked twice a second: if `arecord` exits, or delivers no audio for 15 seconds while still running (device unplugged, a `dsnoop` device that went away), the source is torn down and reopened as a normal retry. Only a stream that actually delivered audio resets the retry counter, so a device that opens but never captures stops after `max_retries` instead of producing an endless series of empty split files.
+
 ---
 
 ## Docker notes

isr.py

@@ -162,6 +162,11 @@ class ALSABackend(AudioBackend):
 class ALSAStream:
     """Context manager for ALSA recording using arecord (raw PCM output)."""
 
+    # No audio for this long means capture is dead even though the process is
+    # still up (device yanked, dsnoop shm gone). Generous enough that normal
+    # scheduling hiccups never trip it.
+    SILENCE_TIMEOUT = 15.0
+
     def __init__(self, device: AudioDevice, sample_rate: int, channels: int,
                  callback: Callable[[bytes], None], logger: logging.Logger):
         self.device = device
@@ -169,12 +174,16 @@ class ALSAStream:
         self.channels = channels
         self.callback = callback
         self.logger = logger
+        self.received_audio = False
         self._running = False
         self._thread = None
+        self._stderr_thread = None
         self._process = None
+        self._last_data = 0.0
 
     def __enter__(self):
         self._running = True
+        self._last_data = time.monotonic()
         cmd = [
             'arecord',
             '-D', self.device.id,
@@ -207,8 +216,25 @@ class ALSAStream:
                 self._process.wait(timeout=2)
             except subprocess.TimeoutExpired:
                 self._process.kill()
-        if self._thread:
-            self._thread.join(timeout=1)
+        for thread in (self._thread, self._stderr_thread):
+            if thread:
+                thread.join(timeout=1)
+
+    def check_alive(self):
+        """Raise if capture has stopped delivering audio.
+
+        Nothing else notices a dead arecord: _read_loop just returns, so
+        without this the recorder would keep rolling over to new split files
+        on schedule and write nothing into any of them — a failure that is
+        invisible from the outside because the files still appear on time.
+        """
+        rc = self._process.poll()
+        if rc is not None:
+            raise RuntimeError(f"arecord exited with code {rc}")
+        idle = time.monotonic() - self._last_data
+        if idle > self.SILENCE_TIMEOUT:
+            raise RuntimeError(
+                f"no audio from '{self.device.id}' for {idle:.0f}s")
 
     def _log_stderr(self):
         for line in self._process.stderr:
@@ -221,8 +247,11 @@ class ALSAStream:
         while self._running and self._process.poll() is None:
             try:
                 data = self._process.stdout.read(chunk_size)
-                if data:
-                    self.callback(data)
+                if not data:
+                    break        # EOF: arecord is on its way out, don't spin
+                self._last_data = time.monotonic()
+                self.received_audio = True
+                self.callback(data)
             except Exception as e:
                 self.logger.error(f"ALSA read error: {e}")
                 break
@@ -755,13 +784,21 @@ class SoundcardRecorder(BaseRecorder):
 
                 with backend.open_stream(
                     self.device, self.sample_rate, self.channels, self._audio_callback
-                ):
+                ) as stream:
                     self.logger.info(f"[{self.name}] Recording started")
-                    retry_count = 0
 
                     while self.running:
                         time.sleep(0.5)
                         self._flush_buffer_to_file()
+                        # Raises if capture died; the retry logic below then
+                        # reopens the device instead of silently recording
+                        # empty files forever.
+                        stream.check_alive()
+                        # Only a stream that actually delivered audio counts as
+                        # progress. Resetting on every healthy tick would let a
+                        # device that opens but never captures retry forever.
+                        if stream.received_audio:
+                            retry_count = 0
 
                         if self._clock() >= next_split:
                             self._open_output_file()

tests/test_isr.py

@@ -6,6 +6,7 @@ Run with:  pytest tests/
 
 import logging
 import struct
+import time
 import wave
 from datetime import datetime
 from pathlib import Path
@@ -423,6 +424,77 @@ class TestAudioFileWriter:
         assert info.samplerate == 44100
 
 
+# ===========================================================================
+# ALSAStream — capture liveness
+# ===========================================================================
+
+class TestALSAStreamLiveness:
+    """check_alive() is what stops a dead arecord from being silently ignored."""
+
+    def _stream(self, poll_result=None) -> isr.ALSAStream:
+        device = isr.AudioDevice(
+            id="hw:0,0", name="Test", channels=2, sample_rate=44100, backend="alsa",
+        )
+        s = isr.ALSAStream(device, 44100, 2, lambda data: None, make_logger())
+        s._process = MagicMock()
+        s._process.poll.return_value = poll_result
+        s._last_data = time.monotonic()
+        return s
+
+    def test_healthy_stream_does_not_raise(self):
+        self._stream().check_alive()
+
+    def test_raises_when_arecord_exited(self):
+        s = self._stream(poll_result=1)
+        with pytest.raises(RuntimeError, match="exited with code 1"):
+            s.check_alive()
+
+    def test_raises_when_no_audio_within_timeout(self):
+        # Process still up, but nothing has arrived — device yanked, or the
+        # dsnoop shared memory went away underneath us.
+        s = self._stream()
+        s._last_data = time.monotonic() - (isr.ALSAStream.SILENCE_TIMEOUT + 1)
+        with pytest.raises(RuntimeError, match="no audio"):
+            s.check_alive()
+
+    def test_read_loop_records_arrival_of_audio(self):
+        s = self._stream()
+        s._running = True
+        s._process.stdout.read.side_effect = [b"\x00" * 64, b""]  # one chunk, then EOF
+        chunks = []
+        s.callback = chunks.append
+
+        s._read_loop()
+
+        assert chunks == [b"\x00" * 64]
+        assert s.received_audio is True
+
+    def test_read_loop_stops_at_eof_without_spinning(self):
+        s = self._stream()
+        s._running = True
+        s._process.stdout.read.return_value = b""
+
+        s._read_loop()
+
+        assert s._process.stdout.read.call_count == 1
+        assert s.received_audio is False
+
+
+class _DeadStream:
+    """Stand-in for a capture stream that opens but never delivers audio."""
+
+    received_audio = False
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        return False
+
+    def check_alive(self):
+        raise RuntimeError("arecord exited with code 1")
+
+
 # ===========================================================================
 # RecorderManager — config loading
 # ===========================================================================
@@ -631,6 +703,33 @@ class TestSoundcardRecorder:
             with pytest.raises(ImportError, match="soundfile"):
                 rec._open_output_file()
 
+    def test_dead_capture_gives_up_instead_of_recording_nothing(self, tmp_path):
+        # A capture that opens but never delivers audio must exhaust
+        # max_retries and stop. Before check_alive() existed, record() sat in
+        # its inner loop forever, rolling over to a new split file every
+        # split_minutes and writing zero bytes into each one.
+        device = self._make_device()
+        audio_system = self._make_audio_system(device)
+        backend = MagicMock()
+        backend.open_stream.return_value = _DeadStream()
+        audio_system.get_backend.return_value = backend
+
+        cfg = {
+            "device": "default",
+            "sample_rate": 44100,
+            "channels": 2,
+            "format": "wav",
+            "output_directory": str(tmp_path),
+            "split_minutes": 60,
+            "max_retries": 2,
+            "retry_delay_seconds": 0,
+        }
+        rec = isr.SoundcardRecorder("sc_dead", cfg, make_logger(), audio_system=audio_system)
+        with patch.object(isr.time, "sleep"):   # don't pay the 0.5 s poll interval
+            rec.record()
+
+        assert backend.open_stream.call_count == cfg["max_retries"]
+
     @pytest.mark.skipif(not isr.SOUNDFILE_AVAILABLE, reason="soundfile not installed")
     def test_flac_file_written_correctly(self, tmp_path):
         import soundfile as sf