Fixed freeze

src/stt_daemon/audio.py

@@ -115,7 +115,12 @@ class ParecCapture(AudioCapture):
         return self._frame_bytes
 
     def frames(self) -> Iterator[bytes]:
-        """Yield frames from parec stdout."""
+        """Yield frames from parec stdout.
+
+        Raises RuntimeError on EOF or short read; the daemon reader catches
+        this, retries with a fresh capture, and requests shutdown when the
+        stream stays unrecoverable so the supervisor restarts.
+        """
         # Hold a local reference: close() may clear self._proc mid-stream.
         proc = self._proc
         if proc is None or proc.stdout is None:

src/stt_daemon/daemon.py

@@ -436,43 +436,94 @@ class Daemon:
         if self._speaker_lane is not None:
             self._speaker_lane.start()
 
+    def _reopen_capture(self, attempts: int = 3, delay_s: float = 1.0) -> bool:
+        """Close and recreate the capture backend with bounded retries.
+
+        Returns True on success (self.capture replaced), False when every
+        attempt failed or shutdown was requested. Never raises.
+        """
+        for attempt in range(1, attempts + 1):
+            if self._stop_event.is_set():
+                return False
+            try:
+                if self.capture is not None:
+                    try:
+                        self.capture.close()
+                    except Exception as e:  # noqa: BLE001 -- close must not block reopen
+                        self.logger.debug("Capture close during reopen failed (%s)", e)
+                self.capture = self._capture_factory()
+                self.logger.warning(
+                    "Capture reopened on attempt %d/%d", attempt, attempts
+                )
+                return True
+            except Exception as e:  # noqa: BLE001 -- boundary: retry, then give up
+                self.logger.warning(
+                    "Capture reopen attempt %d/%d failed (%s)", attempt, attempts, e
+                )
+                if attempt < attempts:
+                    self._stop_event.wait(delay_s)
+        return False
+
     def _reader_loop(self) -> None:
         """Reader thread: capture frames, segment them, and publish events."""
+        failed = False
         try:
             if not self.capture or not self.segmenter:
+                self.logger.error("Reader cannot start: capture or segmenter missing")
+                failed = True
                 return
 
             heartbeat_s = max(0.05, self.cfg.events.heartbeat_ms / 1000.0)
             next_heartbeat = time.time() + heartbeat_s
 
-            for frame in self.capture.frames():
-                if self._stop_event.is_set():
+            while not self._stop_event.is_set():
+                try:
+                    for frame in self.capture.frames():
+                        if self._stop_event.is_set():
+                            break
+
+                        self._track_level(frame)
+                        self._frames_since_heartbeat += 1
+
+                        was_speaking = self.segmenter.speaking
+                        utt = self.segmenter.push(frame)
+
+                        if not was_speaking and self.segmenter.speaking:
+                            self._publish(events.SPEECH_START)
+
+                        if was_speaking and not self.segmenter.speaking:
+                            # A turn ended. It may have been discarded (blip / too quiet),
+                            # in which case there is no utterance - but speech_stop must
+                            # still be published or consumers latch `speech_open` true.
+                            if utt is not None:
+                                self._on_segment_finalized(utt)
+                            else:
+                                self._publish(
+                                    events.SPEECH_STOP, seg_id=None, end_reason="discarded"
+                                )
+
+                        now = time.time()
+                        if now >= next_heartbeat:
+                            self._publish_heartbeat()
+                            next_heartbeat = now + heartbeat_s
+                    # Generator exhausted cleanly (or stop requested): leave the
+                    # retry loop and run the shutdown flush below.
                     break
-
-                self._track_level(frame)
-                self._frames_since_heartbeat += 1
-
-                was_speaking = self.segmenter.speaking
-                utt = self.segmenter.push(frame)
-
-                if not was_speaking and self.segmenter.speaking:
-                    self._publish(events.SPEECH_START)
-
-                if was_speaking and not self.segmenter.speaking:
-                    # A turn ended. It may have been discarded (blip / too quiet),
-                    # in which case there is no utterance - but speech_stop must
-                    # still be published or consumers latch `speech_open` true.
-                    if utt is not None:
-                        self._on_segment_finalized(utt)
-                    else:
-                        self._publish(
-                            events.SPEECH_STOP, seg_id=None, end_reason="discarded"
+                except Exception as e:  # noqa: BLE001 -- boundary: EOF/short-read land here
+                    if self._stop_event.is_set():
+                        self.logger.debug(f"Reader stopped: {e}")
+                        break
+                    self.logger.error("Capture failure: %s", e)
+                    if self._stop_event.is_set():
+                        break
+                    if not self._reopen_capture():
+                        self.logger.error(
+                            "Capture unrecoverable after retries; requesting shutdown"
                         )
-
-                now = time.time()
-                if now >= next_heartbeat:
-                    self._publish_heartbeat()
-                    next_heartbeat = now + heartbeat_s
+                        failed = True
+                        break
+                    next_heartbeat = time.time() + heartbeat_s
+                    continue
 
             # Shutdown: flush any in-progress utterance
             if self.segmenter:
@@ -486,14 +537,21 @@ class Daemon:
                     )
             self._publish_heartbeat()
 
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001 -- boundary: reader must never die silent
             if self._stop_event.is_set():
                 self.logger.debug(f"Reader stopped: {e}")
             else:
                 self.logger.error(f"Reader thread error: {e}")
+                failed = True
         finally:
             if self.capture:
-                self.capture.close()
+                try:
+                    self.capture.close()
+                except Exception as e:  # noqa: BLE001 -- shutdown path must not raise
+                    self.logger.debug("Capture close on reader exit failed (%s)", e)
+            if failed and not self._stop_event.is_set():
+                self.logger.error("Reader thread exited unexpectedly; requesting shutdown")
+                self.request_stop()
 
     def _on_segment_finalized(self, utt: Utterance) -> None:
         """Assign a seg_id at finalization, announce the stop, and enqueue."""
@@ -578,6 +636,7 @@ class Daemon:
 
     def _worker_loop(self) -> None:
         """Worker thread: transcribe segments and publish results."""
+        failed = False
         try:
             while not self._stop_event.is_set():
                 try:
@@ -585,10 +644,20 @@ class Daemon:
                 except queue.Empty:
                     continue
                 self._process(utt)
-        except Exception as e:
-            self.logger.error(f"Worker thread error: {e}")
+        except Exception as e:  # noqa: BLE001 -- boundary: worker must never die silent
+            if self._stop_event.is_set():
+                self.logger.debug(f"Worker stopped: {e}")
+            else:
+                self.logger.error(f"Worker thread error: {e}")
+                failed = True
         finally:
-            self._drain_remaining()
+            try:
+                self._drain_remaining()
+            except Exception as e:  # noqa: BLE001 -- shutdown path must not raise
+                self.logger.debug("Drain on worker exit failed (%s)", e)
+            if failed and not self._stop_event.is_set():
+                self.logger.error("Worker thread exited unexpectedly; requesting shutdown")
+                self.request_stop()
 
     def _process(self, utt: Utterance) -> None:
         """Transcribe one segment, then always publish its segment_done."""