STT-Assistant/src/stt_daemon/daemon.py

"""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.")