"""Main daemon that orchestrates capture, segmentation, and transcription.""" import collections import concurrent.futures import itertools import json import logging import math import os import queue import shutil import stat import threading import time from pathlib import Path from typing import Any, Callable, Optional import numpy as np from . import events from .audio import AudioCapture, open_capture from .bus import EventPublisher from .config import Config from .output import TranscriptSink from .segmenter import FrameSegmenter, Utterance, make_vad from .speakers import SpeakerStore from .transcriber import TranscriberLike, build_transcriber class _LaneJob: """One embedding request: audio in, Outcome (or exception) out via future.""" __slots__ = ("seg_id", "audio", "dur_s", "rms", "future") def __init__( self, seg_id: int, audio: np.ndarray, dur_s: float, rms: float, future: concurrent.futures.Future, ) -> None: self.seg_id = seg_id self.audio = audio self.dur_s = dur_s self.rms = rms self.future = future _REQUEST_MAX_BYTES = 65536 class _EmbeddingLane: """Single-worker speaker-embedding lane. Confinement rule: ALL SpeakerStore access (match/process, inbox rename, prune-persisted saves) happens on this lane's one worker thread. The transcription worker only submits jobs and awaits the returned future, so the lane never touches faster-whisper and the daemon socket stays publish-only. Startup-time load()/prune() in Daemon._initialize runs single-threaded before this lane starts, so it cannot race the worker. """ def __init__( self, store: SpeakerStore, embedder: Any, publish: Callable[..., None], inbox_dir: Path, logger: logging.Logger, ) -> None: self._store = store self._embedder = embedder self._publish = publish self._inbox_dir = Path(inbox_dir) self._logger = logger self._queue: queue.Queue = queue.Queue() self._stop_event = threading.Event() self._thread: threading.Thread | None = None def start(self) -> None: """Start the single worker thread (non-daemon, like the other lanes).""" self._thread = threading.Thread(target=self._run, daemon=False) self._thread.start() def stop(self) -> None: """Signal the worker and join it.""" self._stop_event.set() if self._thread is not None: self._thread.join(timeout=5) self._thread = None def submit( self, seg_id: int, audio: np.ndarray, dur_s: float, rms: float ) -> concurrent.futures.Future: """Enqueue one embedding job; the transcription lane awaits the future.""" future: concurrent.futures.Future = concurrent.futures.Future() self._queue.put(_LaneJob(seg_id, audio, dur_s, rms, future)) return future def drain_inbox(self) -> None: """Apply pending rename requests; quarantine bad ones. Never raises.""" try: files = sorted(self._inbox_dir.glob("*.json")) except OSError as e: self._logger.warning("speaker inbox unreadable (%s)", e) return for path in files: try: self._apply_request(path) except Exception as e: # noqa: BLE001 -- boundary: drain must never raise self._logger.warning("speaker inbox request %s failed: %s", path.name, e) self._quarantine(path) def _apply_request(self, path: Path) -> None: try: st = os.lstat(path) except OSError as e: self._logger.warning("ignoring unreadable speaker request %s (%s)", path.name, e) self._quarantine(path) return if not stat.S_ISREG(st.st_mode): self._logger.warning( "ignoring non-regular speaker request %s (not a regular file)", path.name ) self._quarantine(path) return if st.st_size > _REQUEST_MAX_BYTES: self._logger.warning( "ignoring oversized speaker request %s (%d bytes)", path.name, st.st_size ) self._quarantine(path) return try: fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) except OSError as e: self._logger.warning("ignoring unreadable speaker request %s (%s)", path.name, e) self._quarantine(path) return try: chunks: list[bytes] = [] remaining = _REQUEST_MAX_BYTES + 1 while remaining > 0: try: chunk = os.read(fd, remaining) except OSError as e: self._logger.warning( "ignoring unreadable speaker request %s (%s)", path.name, e ) self._quarantine(path) return if not chunk: break chunks.append(chunk) remaining -= len(chunk) raw = b"".join(chunks) finally: try: os.close(fd) except OSError: pass if len(raw) > _REQUEST_MAX_BYTES: self._logger.warning( "ignoring oversized speaker request %s (%d bytes)", path.name, len(raw) ) self._quarantine(path) return try: payload = json.loads(raw.decode("utf-8")) except (OSError, ValueError) as e: self._logger.warning("ignoring malformed speaker request %s (%s)", path.name, e) self._quarantine(path) return if ( not isinstance(payload, dict) or not isinstance(payload.get("speaker_id"), str) or not isinstance(payload.get("name"), str) ): self._logger.warning("ignoring malformed speaker request %s (bad shape)", path.name) self._quarantine(path) return if not self._store.rename(payload["speaker_id"], payload["name"]): self._logger.warning( "ignoring speaker request %s (unknown id %r)", path.name, payload["speaker_id"] ) self._quarantine(path) return self._store.save() self._publish(events.SPEAKERS, speakers=self._store.snapshot()) try: path.unlink() except OSError as e: self._logger.warning("failed to delete applied speaker request %s (%s)", path.name, e) def _quarantine(self, path: Path) -> None: try: rejected = self._inbox_dir / "rejected" rejected.mkdir(parents=True, exist_ok=True) shutil.move(str(path), str(rejected / path.name)) except OSError as e: self._logger.warning("failed to quarantine speaker request %s (%s)", path.name, e) def _run(self) -> None: # A request file may have landed while the daemon was down; apply it # on startup before the first job so renames never wait a full tick. self.drain_inbox() while not self._stop_event.is_set(): try: job = self._queue.get(timeout=0.5) except queue.Empty: self.drain_inbox() continue try: emb = self._embedder.embed(job.audio) outcome = self._store.process(emb, job.dur_s, job.rms, job.audio) if outcome.event in ("mint", "confirm"): self._store.save() self._publish(events.SPEAKERS, speakers=self._store.snapshot()) if not job.future.done(): job.future.set_result(outcome) except Exception as e: # noqa: BLE001 -- boundary: lane must resolve every future if not job.future.done(): job.future.set_exception(e) self.drain_inbox() class Daemon: """Always-on STT daemon with worker threads.""" def __init__( self, cfg: Config, log_level: int = logging.INFO, *, capture_factory: Callable[[], Any] | None = None, vad_factory: Callable[[int], Any] | None = None, transcriber_factory: Callable[[], Any] | None = None, sink_factory: Callable[[], Any] | None = None, embedder_factory: Callable[[], Any] | None = None, debug_sink_factory: Callable[[], Any] | None = None, ): """Initialize the daemon. Args: cfg: Configuration object. log_level: Logging level (e.g., logging.INFO, logging.DEBUG). capture_factory: Override audio capture creation (used by tests). vad_factory: Override VAD creation (used by tests). transcriber_factory: Override transcriber creation (used by tests). sink_factory: Override transcript sink creation (used by tests). embedder_factory: Override speaker embedder creation (used by tests). debug_sink_factory: Override debug signal sink creation (used by tests). Distinct from sink_factory, which builds TranscriptSink. """ self.cfg = cfg self._capture_factory = capture_factory or (lambda: open_capture(self.cfg.audio)) self._vad_factory = vad_factory or make_vad self._transcriber_factory = transcriber_factory or self._build_transcriber self._sink_factory = sink_factory or self._build_sink self._embedder_factory = embedder_factory or self._build_embedder self._debug_sink_factory = debug_sink_factory self.debug_sink: Any = None # Set up logging self.logger = logging.getLogger("stt_daemon") self.logger.setLevel(log_level) if not self.logger.handlers: handler = logging.StreamHandler() handler.setFormatter(logging.Formatter("[%(levelname)s] %(message)s")) self.logger.addHandler(handler) # Initialize components self.capture: Optional[AudioCapture] = None self.segmenter: Optional[FrameSegmenter] = None self.transcriber: Optional[TranscriberLike] = None self.sink: Optional[TranscriptSink] = None self.publisher: Optional[EventPublisher] = None self.speaker_store: Optional[SpeakerStore] = None self._speaker_lane: Optional[_EmbeddingLane] = None # Threading self.segment_queue: queue.Queue = queue.Queue(maxsize=cfg.daemon.queue_size) self._stop_event = threading.Event() self._stopped = False self._backlog_warned = False self._reader_thread: Optional[threading.Thread] = None self._worker_thread: Optional[threading.Thread] = None # Event bus state (shared across reader/worker threads) self._seg_counter = itertools.count(1) self._segments_done = 0 self._last_segment_done_ts = 0.0 self._last_worker_activity_ts = 0.0 self._last_transcript_ts = 0.0 self._in_flight = False self._frames_since_heartbeat = 0 self._started_at = 0.0 rms_window = max(1, round(1000 / max(1, cfg.audio.frame_ms))) self._rms_window: collections.deque[float] = collections.deque(maxlen=rms_window) def run(self) -> None: """Run the daemon (blocks until stopped).""" try: self._initialize() self._start_workers() self._stop_event.wait() finally: self.stop() def request_stop(self) -> None: """Ask the run loop to stop. Signal-safe: cleanup happens in stop().""" self._stop_event.set() def _build_transcriber(self) -> TranscriberLike: return build_transcriber(self.cfg.whisper) def _build_sink(self) -> TranscriptSink: return TranscriptSink( stdout=self.cfg.output.stdout, jsonl_dir=self.cfg.output.jsonl_dir, jsonl=self.cfg.output.jsonl, save_audio=self.cfg.output.save_audio, audio_dir=self.cfg.output.audio_dir, sample_rate=self.cfg.audio.sample_rate, ) def _build_embedder(self) -> Any: from .embedder import build_embedder return build_embedder(self.cfg) def _debug_config(self) -> Any: """Return the [debug] section, or None when the config predates it.""" return getattr(self.cfg, "debug", None) def _build_debug_sink(self) -> Any: """Default debug sink builder (lazy import; duck-typed by callers).""" from .debug import DebugSink return DebugSink(getattr(self._debug_config(), "dir", "debug"), "daemon") def _debug_emit(self, type_: str, **fields: Any) -> None: """Best-effort debug signal line; never raises, never blocks shutdown.""" sink = self.debug_sink if sink is None: return try: sink.emit(type_, **fields) except Exception as e: # noqa: BLE001 -- debug must never break the daemon self.logger.warning("Debug sink emit failed (%s)", e) def _initialize(self) -> None: """Initialize all components and start publishing events.""" self.logger.info("Initializing audio capture...") self.capture = self._capture_factory() self.logger.info("Initializing VAD...") vad = self._vad_factory(self.cfg.vad.aggressiveness) self.logger.info("Initializing segmenter...") self.segmenter = FrameSegmenter( sample_rate=self.cfg.audio.sample_rate, frame_ms=self.cfg.audio.frame_ms, vad=vad, pre_roll_ms=self.cfg.vad.pre_roll_ms, speech_start_ms=self.cfg.vad.speech_start_ms, silence_ms=self.cfg.vad.silence_ms, min_utterance_ms=self.cfg.vad.min_utterance_ms, max_utterance_s=self.cfg.vad.max_utterance_s, chunk_pause_ms=self.cfg.vad.chunk_pause_ms, soft_max_s=self.cfg.vad.soft_max_s, rms_floor=self.cfg.vad.rms_floor, trailing_trim_ms=self.cfg.vad.trailing_trim_ms, ) self.logger.info(f"Loading Whisper model '{self.cfg.whisper.model}'...") self.transcriber = self._transcriber_factory() self.logger.info("Initializing output sink...") self.sink = self._sink_factory() self._started_at = time.time() self._start_publisher() self._start_speaker_lane() self._start_debug_sink() self.logger.info("Daemon ready. Listening...") def _start_debug_sink(self) -> None: """Construct the debug signal sink when [debug] is enabled (None otherwise).""" self.debug_sink = None debug_cfg = self._debug_config() if debug_cfg is None or not getattr(debug_cfg, "enabled", False): return try: factory = self._debug_sink_factory or self._build_debug_sink self.debug_sink = factory() except Exception as e: # noqa: BLE001 -- boundary: debug must not crash init self.logger.warning("Debug sink disabled (%s)", e) self.debug_sink = None def _start_speaker_lane(self) -> None: """Build the speaker store + embedding lane (no-op when disabled). Runs single-threaded before any lane thread starts; creates nothing on disk when the embedder resolves to None. """ try: embedder = self._embedder_factory() except Exception as e: # noqa: BLE001 -- boundary: a bad factory must not crash self.logger.warning("Speaker identification disabled (%s)", e) return if embedder is None: return speaker_cfg = self.cfg.speaker store = SpeakerStore( Path(speaker_cfg.store_dir), threshold_id=speaker_cfg.threshold_id, threshold_new=speaker_cfg.threshold_new, min_enroll_s=speaker_cfg.min_enroll_s, rms_floor=speaker_cfg.rms_floor, margin=speaker_cfg.margin, mint_cooldown_s=speaker_cfg.mint_cooldown_s, confirm_s=speaker_cfg.confirm_s, max_prototypes=speaker_cfg.max_prototypes, provisional_ttl_h=speaker_cfg.provisional_ttl_h, ) store.load() removed = store.prune() if removed: store.save() self._publish(events.SPEAKERS, speakers=store.snapshot()) self.speaker_store = store self._speaker_lane = _EmbeddingLane( store, embedder, self._publish, Path(speaker_cfg.store_dir) / "inbox", self.logger, ) def _start_publisher(self) -> None: """Start the publish-only event bus (best effort; never fatal).""" publisher = EventPublisher( path=self.cfg.events.socket or None, ring_size=self.cfg.events.ring_size, status_provider=self._status_fields, ) try: publisher.start() except OSError as e: self.logger.warning("Event bus disabled (%s)", e) self.publisher = None return self.publisher = publisher def _status_fields(self) -> dict: return { "capture": self.cfg.audio.device, "model_loaded": self.transcriber is not None, "segments_done": self._segments_done, "heartbeat_ms": self.cfg.events.heartbeat_ms, } def _publish(self, type_: str, **fields) -> None: if self.publisher is not None: self.publisher.emit(type_, **fields) def _start_workers(self) -> None: """Start reader and worker threads.""" self._reader_thread = threading.Thread(target=self._reader_loop, daemon=False) self._worker_thread = threading.Thread(target=self._worker_loop, daemon=False) self._reader_thread.start() self._worker_thread.start() 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 self._debug_emit( "capture_reopen_attempt", attempt=attempt, attempts=attempts ) 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 ) self._debug_emit( "capture_reopen_ok", attempt=attempt, attempts=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 ) self._debug_emit( "capture_reopen_fail", attempt=attempt, attempts=attempts, error=str(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 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 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" ) failed = True break next_heartbeat = time.time() + heartbeat_s continue # Shutdown: flush any in-progress utterance if self.segmenter: was_speaking = self.segmenter.speaking utt = self.segmenter.flush() if utt is not None: self._on_segment_finalized(utt) elif was_speaking: self._publish( events.SPEECH_STOP, seg_id=None, end_reason="discarded" ) self._publish_heartbeat() 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: 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._debug_emit("reader_fatal") self.request_stop() def _on_segment_finalized(self, utt: Utterance) -> None: """Assign a seg_id at finalization, announce the stop, and enqueue.""" seg_id = next(self._seg_counter) utt.seg_id = seg_id self._publish(events.SPEECH_STOP, seg_id=seg_id, end_reason=utt.reason) if not self._enqueue(utt): self._publish( events.SEGMENT_DONE, seg_id=seg_id, emitted=False, end_reason=utt.reason ) self._check_backlog() def _enqueue(self, utt: Utterance) -> bool: """Queue a segment, dropping the oldest if full. False if it could not be queued.""" try: self.segment_queue.put(utt, block=False) return True except queue.Full: self.logger.warning("Segment queue full, dropping oldest") try: dropped = self.segment_queue.get_nowait() except queue.Empty: dropped = None if dropped is not None: self._publish( events.SEGMENT_DONE, seg_id=dropped.seg_id, emitted=False, end_reason=dropped.reason, ) self._debug_emit( "queue_drop", dropped_seg_id=dropped.seg_id, end_reason=dropped.reason, queue_size=self.segment_queue.maxsize, ) try: self.segment_queue.put(utt, block=False) return True except queue.Full: return False def _track_level(self, frame: bytes) -> None: samples = np.frombuffer(frame, dtype=np.int16).astype(np.float32) / 32768.0 if samples.size: self._rms_window.append(float(np.mean(samples * samples))) def _publish_heartbeat(self) -> None: now = time.time() rms = math.sqrt(sum(self._rms_window) / len(self._rms_window)) if self._rms_window else 0.0 busy = self._in_flight or self.segment_queue.qsize() > 0 payload = { "in_flight": self._in_flight, "segments_done": self._segments_done, "worker_stall_s": ( round(now - self._last_worker_activity_ts, 3) if busy and self._last_worker_activity_ts else 0.0 ), "queue_depth": self.segment_queue.qsize(), "frames_since_last": self._frames_since_heartbeat, "rms": round(rms, 5), "capture": self.cfg.audio.device, "device_ok": self.capture.healthy() if self.capture else False, "model_loaded": self.transcriber is not None, "uptime_s": round(now - self._started_at, 3), "last_transcript_age_s": ( round(now - self._last_transcript_ts, 3) if self._last_transcript_ts else 0.0 ), } self._publish(events.HEARTBEAT, **payload) debug_cfg = self._debug_config() if debug_cfg is not None and getattr(debug_cfg, "heartbeat", True): self._debug_emit("heartbeat", **payload) self._frames_since_heartbeat = 0 def _check_backlog(self) -> None: """Warn once per overload episode; re-arm only after the queue fully drains.""" size = self.segment_queue.qsize() if size >= self.cfg.daemon.backlog_warn_at and not self._backlog_warned: self.logger.warning( "Transcription backlog reached %d of %d - worker is falling behind", size, self.segment_queue.maxsize, ) self._debug_emit( "backlog_warn", size=size, maxsize=self.segment_queue.maxsize ) self._backlog_warned = True elif size == 0 and self._backlog_warned: self._backlog_warned = False def _worker_loop(self) -> None: """Worker thread: transcribe segments and publish results.""" failed = False try: while not self._stop_event.is_set(): try: utt = self.segment_queue.get(timeout=1) except queue.Empty: continue self._process(utt) 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: 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._debug_emit("worker_fatal") self.request_stop() def _process(self, utt: Utterance) -> None: """Transcribe one segment, then always publish its segment_done.""" emitted = False self._in_flight = True self._last_worker_activity_ts = time.time() try: if self.transcriber and self.sink: speaker_future = None if self._speaker_lane is not None: speaker_future = self._speaker_lane.submit( utt.seg_id, utt.audio, utt.duration_s, utt.rms ) started = time.perf_counter() result = self.transcriber.transcribe(utt.audio) transcribe_s = time.perf_counter() - started seg_id = utt.seg_id speaker_id: str | None = None speaker_status = "none" speaker: str | None = "?" speaker_score: float | None = None if speaker_future is not None: speaker_id, speaker_status, speaker, speaker_score = ( self._resolve_speaker(speaker_future) ) try: audio_path = self.sink.emit( result, utt.duration_s, utt.rms, utt.reason, utt.audio, transcribe_s=transcribe_s, seg_id=seg_id, speaker_id=speaker_id, speaker_status=speaker_status, speaker=speaker, speaker_score=speaker_score, ) except Exception as e: # A sink failure must not drop the transcript from the bus. self.logger.error("Output error: %s", e) audio_path = None if result.text: emitted = True self._last_transcript_ts = time.time() self._publish( events.TRANSCRIPT, seg_id=seg_id, text=result.text, lang=result.language, lang_prob=result.language_probability, duration_s=utt.duration_s, rms=utt.rms, end_reason=utt.reason, audio=audio_path, speaker_id=speaker_id, speaker_status=speaker_status, speaker=speaker, speaker_score=speaker_score, ) except Exception as e: self.logger.error(f"Transcription error: {e}") finally: self._in_flight = False self._segments_done += 1 self._last_segment_done_ts = time.time() self._publish( events.SEGMENT_DONE, seg_id=utt.seg_id, emitted=emitted, end_reason=utt.reason, ) def _resolve_speaker( self, future: concurrent.futures.Future ) -> tuple[str | None, str, str | None, float | None]: """Await one embedding outcome; degrade to "none" on any failure.""" try: outcome = future.result(timeout=self.cfg.speaker.join_timeout_s) except Exception as e: # noqa: BLE001 -- TimeoutError or lane failure -> "none" self.logger.debug("Speaker lookup failed (%s)", e) self._debug_emit("speaker_fallback", error=str(e)) return None, "none", "?", None try: return self._speaker_fields(outcome) except Exception as e: # noqa: BLE001 -- mapping must never break emit self.logger.warning("Speaker mapping failed (%s)", e) return None, "none", "?", None def _speaker_fields(self, outcome) -> tuple[str | None, str, str | None, float | None]: """Map a lane outcome to (speaker_id, speaker_status, speaker, speaker_score).""" match = outcome.match status = match.status speakers = self.speaker_store.speakers if self.speaker_store is not None else {} if status == "matched": name = speakers.get(match.speaker_id, {}).get("name") if match.speaker_id else None return match.speaker_id, "matched", name or match.speaker_id, match.score if status == "uncertain": top = match.top_id name = speakers.get(top, {}).get("name") if top else None return None, "uncertain", f"{(name or top)}?", match.score if status == "unknown": minted = match.speaker_id if outcome.event == "mint" else None return minted, "unknown", "?", None return None, "none", "?", None def _drain_remaining(self) -> None: """Announce any queued-but-unprocessed segments on shutdown.""" while True: try: utt = self.segment_queue.get_nowait() except queue.Empty: break self._publish( events.SEGMENT_DONE, seg_id=utt.seg_id, emitted=False, end_reason=utt.reason, ) def stop(self) -> None: """Stop the daemon. Safe to call more than once.""" if self._stopped: return self._stopped = True self.logger.info("Stopping daemon...") self._stop_event.set() # Close capture FIRST so the reader thread's blocking read unblocks # (otherwise a non-daemon reader thread can hang process shutdown). if self.capture: self.capture.close() if self._reader_thread: self._reader_thread.join(timeout=5) if self._worker_thread: self._worker_thread.join(timeout=5) if self._speaker_lane is not None: self._speaker_lane.stop() self._speaker_lane = None if self.transcriber: self.transcriber.close() if self.sink: self.sink.close() if self.publisher: self.publisher.close() if self.debug_sink is not None: try: self.debug_sink.close() except Exception as e: # noqa: BLE001 -- shutdown path must not raise self.logger.debug("Debug sink close failed (%s)", e) finally: self.debug_sink = None self.logger.info("Daemon stopped.")