"""Tests for the daemon's publish-only event stream (Phase 1)."""

import concurrent.futures
import logging
import os
import shutil
import socket
import tempfile
import threading
import time
from types import SimpleNamespace

import numpy as np
import pytest

from stt_daemon import events
from stt_daemon.bus import EventSubscriber
from stt_daemon.config import Config
from stt_daemon.daemon import Daemon
from stt_daemon.segmenter import Utterance
from stt_daemon.supervisor import Supervisor
from stt_daemon.transcriber import TranscriptResult

FRAME_SAMPLES = 480
FRAME_BYTES = FRAME_SAMPLES * 2


@pytest.fixture
def sock_path():
    directory = tempfile.mkdtemp(prefix="sttb-")
    try:
        yield os.path.join(directory, "s.sock")
    finally:
        shutil.rmtree(directory, ignore_errors=True)


def _frame(value=0):
    return np.full(FRAME_SAMPLES, value, dtype=np.int16).tobytes()


class FakeVad:
    def __init__(self, decisions):
        self._decisions = list(decisions)
        self._index = 0

    def is_speech(self, frame, sample_rate):
        if self._index < len(self._decisions):
            decision = self._decisions[self._index]
            self._index += 1
            return decision
        return False


class FakeCapture:
    def __init__(self, frames, delay=0.0, stall_after=None, healthy=True):
        self._frames = list(frames)
        self._delay = delay
        self._stall_after = stall_after
        self._healthy = healthy
        self._closed = threading.Event()
        self.frame_bytes = FRAME_BYTES

    def frames(self):
        for index, frame in enumerate(self._frames):
            if self._closed.is_set():
                return
            if self._delay:
                time.sleep(self._delay)
            yield frame
            if self._stall_after is not None and index + 1 >= self._stall_after:
                self._closed.wait()
                return

    def healthy(self):
        return self._healthy

    def close(self):
        self._closed.set()


class FakeTranscriber:
    def __init__(self, text="hello", language="en", probability=0.99):
        self.text = text
        self.language = language
        self.probability = probability

    def transcribe(self, audio):
        return TranscriptResult(
            language=self.language,
            language_probability=self.probability,
            text=self.text,
        )

    def close(self):
        pass


class FakeSink:
    def __init__(self):
        self.emitted = []
        self.transcribe_times = []

    def emit(
        self,
        result,
        duration_s,
        rms=0.0,
        reason="unknown",
        audio=None,
        transcribe_s=None,
        **kwargs,
    ):
        if not result.text:
            return None
        self.emitted.append(result.text)
        self.transcribe_times.append(transcribe_s)
        self.last_kwargs = dict(kwargs)
        return None

    def close(self):
        pass


class _Collector:
    def __init__(self, subscriber):
        self.events = []
        self._thread = threading.Thread(target=self._run, args=(subscriber,), daemon=True)
        self._thread.start()

    def _run(self, subscriber):
        try:
            for event in subscriber.events():
                self.events.append(event)
        except Exception:  # noqa: BLE001 - collector must not crash the test thread
            pass

    def types(self):
        return [e["type"] for e in self.events]

    def of(self, type_):
        return [e for e in self.events if e["type"] == type_]

    def wait_for(self, predicate, timeout=5.0):
        deadline = time.time() + timeout
        while time.time() < deadline:
            if predicate(self.events):
                return True
            time.sleep(0.01)
        return predicate(self.events)


def _base_config(sock_path):
    cfg = Config()
    cfg.events.socket = sock_path
    cfg.events.heartbeat_ms = 50
    cfg.audio.frame_ms = 30
    cfg.vad.pre_roll_ms = 60
    cfg.vad.speech_start_ms = 30
    cfg.vad.silence_ms = 60
    cfg.vad.min_utterance_ms = 30
    cfg.vad.chunk_pause_ms = 300
    cfg.vad.soft_max_s = 10.0
    cfg.vad.max_utterance_s = 30.0
    cfg.vad.rms_floor = 0.0
    cfg.daemon.queue_size = 50
    cfg.output.stdout = False
    cfg.output.jsonl_dir = None
    cfg.output.save_audio = False
    return cfg


def _start_daemon(cfg, capture, vad, transcriber, debug_sink_factory=None):
    sink = FakeSink()
    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: capture,
        vad_factory=lambda _aggr: vad,
        transcriber_factory=lambda: transcriber,
        sink_factory=lambda: sink,
        debug_sink_factory=debug_sink_factory,
    )
    daemon._initialize()
    subscriber = EventSubscriber(cfg.events.socket, reconnect_delay=0.05)
    collector = _Collector(subscriber)
    daemon._start_workers()
    return daemon, collector, subscriber, sink


def test_transcript_and_segment_done_share_seg_id(sock_path):
    cfg = _base_config(sock_path)
    frames = [_frame() for _ in range(5)]
    vad = FakeVad([False, True, True, False, False])
    transcriber = FakeTranscriber(text="hello")
    daemon, collector, subscriber, _ = _start_daemon(cfg, FakeCapture(frames), vad, transcriber)

    collector.wait_for(lambda e: any(x["type"] == events.SEGMENT_DONE for x in e))

    starts = collector.of(events.SPEECH_START)
    stops = collector.of(events.SPEECH_STOP)
    transcripts = collector.of(events.TRANSCRIPT)
    dones = collector.of(events.SEGMENT_DONE)

    assert len(starts) == 1
    assert "seg_id" not in starts[0]  # speech_start carries no seg_id

    assert len(stops) == 1
    assert len(transcripts) == 1
    assert len(dones) == 1
    assert stops[0]["seg_id"] == transcripts[0]["seg_id"] == dones[0]["seg_id"] == 1
    assert transcripts[0]["text"] == "hello"
    assert dones[0]["emitted"] is True
    assert dones[0]["end_reason"] == "end_of_turn"

    subscriber.close()
    daemon.stop()


def test_transcribe_seconds_is_forwarded_to_sink(sock_path):
    """The worker measures wall time around transcribe() and passes it to emit()."""
    cfg = _base_config(sock_path)
    frames = [_frame() for _ in range(5)]
    vad = FakeVad([False, True, True, False, False])
    transcriber = FakeTranscriber(text="hello")
    daemon, collector, subscriber, sink = _start_daemon(
        cfg, FakeCapture(frames), vad, transcriber
    )

    collector.wait_for(lambda e: any(x["type"] == events.SEGMENT_DONE for x in e))

    assert len(sink.transcribe_times) == 1
    assert sink.transcribe_times[0] is not None
    assert sink.transcribe_times[0] >= 0.0

    subscriber.close()
    daemon.stop()


def test_gated_segment_reports_emitted_false(sock_path):
    cfg = _base_config(sock_path)
    frames = [_frame() for _ in range(5)]
    vad = FakeVad([False, True, True, False, False])
    transcriber = FakeTranscriber(text="")  # gated / no text
    daemon, collector, subscriber, _ = _start_daemon(cfg, FakeCapture(frames), vad, transcriber)

    collector.wait_for(lambda e: any(x["type"] == events.SEGMENT_DONE for x in e))

    assert collector.of(events.TRANSCRIPT) == []
    dones = collector.of(events.SEGMENT_DONE)
    assert len(dones) == 1
    assert dones[0]["emitted"] is False
    assert len(collector.of(events.SPEECH_STOP)) == 1

    subscriber.close()
    daemon.stop()


def test_chunked_utterance_yields_one_pair_per_chunk(sock_path):
    cfg = _base_config(sock_path)
    cfg.vad.soft_max_s = 0.3  # 10 frames
    cfg.vad.chunk_pause_ms = 60  # 2 frames
    cfg.vad.silence_ms = 600  # 20 frames (never reached here)
    frames = [_frame() for _ in range(47)]
    decisions = [False] + [True] * 12 + [False] * 2 + [True] * 12 + [False] * 20
    transient = FakeTranscriber(text="chunk")
    daemon, collector, subscriber, _ = _start_daemon(
        cfg, FakeCapture(frames), FakeVad(decisions), transient
    )

    collector.wait_for(
        lambda e: len([x for x in e if x["type"] == events.SEGMENT_DONE]) >= 2
    )

    stops = collector.of(events.SPEECH_STOP)
    dones = collector.of(events.SEGMENT_DONE)
    assert len(stops) == len(dones) >= 2

    stop_ids = [s["seg_id"] for s in stops]
    done_ids = [d["seg_id"] for d in dones]
    assert stop_ids == sorted(stop_ids)
    assert len(set(stop_ids)) == len(stop_ids)  # unique per chunk
    assert set(stop_ids) == set(done_ids)
    assert all(d["end_reason"] == "pause" for d in dones)

    subscriber.close()
    daemon.stop()


def test_heartbeats_advance_and_carry_health(sock_path):
    cfg = _base_config(sock_path)
    cfg.events.heartbeat_ms = 40
    frames = [_frame() for _ in range(40)]
    daemon, collector, subscriber, _ = _start_daemon(
        cfg, FakeCapture(frames, delay=0.02), FakeVad([]), FakeTranscriber()
    )

    collector.wait_for(lambda e: len([x for x in e if x["type"] == events.HEARTBEAT]) >= 3)
    heartbeats = collector.of(events.HEARTBEAT)

    assert len(heartbeats) >= 3
    seqs = [h["seq"] for h in heartbeats]
    assert seqs == sorted(seqs)
    assert len(set(seqs)) == len(seqs)

    last = heartbeats[-1]
    assert last["model_loaded"] is True
    assert last["device_ok"] is True
    assert last["capture"] == cfg.audio.device
    for field in ("in_flight", "segments_done", "worker_stall_s", "queue_depth",
                  "frames_since_last", "rms", "uptime_s", "last_transcript_age_s"):
        assert field in last

    subscriber.close()
    daemon.stop()


def test_stalled_capture_stops_heartbeats(sock_path):
    cfg = _base_config(sock_path)
    cfg.events.heartbeat_ms = 60
    frames = [_frame() for _ in range(30)]
    capture = FakeCapture(frames, delay=0.01, stall_after=10)
    daemon, collector, subscriber, _ = _start_daemon(
        cfg, capture, FakeVad([]), FakeTranscriber()
    )

    assert collector.wait_for(
        lambda e: any(x["type"] == events.HEARTBEAT for x in e), timeout=3.0
    )
    time.sleep(0.2)  # let it stall
    before = len(collector.of(events.HEARTBEAT))

    time.sleep(0.45)  # ~7 heartbeat intervals at 60ms
    after = len(collector.of(events.HEARTBEAT))
    assert after == before, f"heartbeats continued during a stall ({before} -> {after})"

    subscriber.close()
    daemon.stop()


def test_discarded_segment_still_emits_speech_stop(sock_path):
    """A turn that is discarded (blip / too quiet) must still emit speech_stop.

    Without it, consumers that track speech_open from speech_start/speech_stop
    latch true forever, which blocks readiness and chat dispatch.
    """
    cfg = _base_config(sock_path)
    cfg.vad.min_utterance_ms = 300  # a 1-frame blip is discarded
    frames = [_frame() for _ in range(4)]
    vad = FakeVad([True, False, False, False])  # confirm, then silence -> discard
    daemon, collector, subscriber, _ = _start_daemon(
        cfg, FakeCapture(frames), vad, FakeTranscriber()
    )

    collector.wait_for(lambda e: any(x["type"] == events.SPEECH_STOP for x in e))
    time.sleep(0.2)

    starts = collector.of(events.SPEECH_START)
    stops = collector.of(events.SPEECH_STOP)

    assert len(starts) == 1
    assert len(stops) == 1
    assert stops[0]["seg_id"] is None  # discarded: no segment assigned
    assert stops[0]["end_reason"] == "discarded"
    assert collector.of(events.SEGMENT_DONE) == []
    assert collector.of(events.TRANSCRIPT) == []

    subscriber.close()
    daemon.stop()


def test_inbound_data_on_subscriber_socket_is_ignored(sock_path):
    cfg = _base_config(sock_path)
    cfg.events.heartbeat_ms = 40
    frames = [_frame() for _ in range(60)]
    daemon, collector, subscriber, _ = _start_daemon(
        cfg, FakeCapture(frames, delay=0.02), FakeVad([]), FakeTranscriber()
    )

    # The bus is publish-only: a client writing data must have no effect.
    raw = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    raw.connect(sock_path)
    raw.sendall(b'{"v":1,"type":"evil","seq":1,"ts":0}\n')
    time.sleep(0.1)

    assert collector.wait_for(
        lambda e: len([x for x in e if x["type"] == events.HEARTBEAT]) >= 2
    )
    assert all(x["type"] != "evil" for x in collector.events)

    raw.close()
    subscriber.close()
    daemon.stop()


class FakeDebugSink:
    """In-memory DebugSink stand-in (duck-typed emit/close, thread-safe)."""

    def __init__(self, dir="debug", proc="daemon"):
        self.dir = dir
        self.proc = proc
        self.emitted = []
        self.closed = False
        self._lock = threading.Lock()

    def emit(self, type_, **fields):
        with self._lock:
            self.emitted.append({"type": type_, **fields})

    def types(self):
        with self._lock:
            return [e["type"] for e in self.emitted]

    def of(self, type_):
        with self._lock:
            return [e for e in self.emitted if e["type"] == type_]

    def close(self):
        self.closed = True


class BoomDebugSink(FakeDebugSink):
    """Debug sink whose emit always throws (total-swallow check)."""

    def emit(self, type_, **fields):
        raise RuntimeError("disk gone")


def _enable_debug(cfg, enabled=True, heartbeat=True):
    cfg.debug = SimpleNamespace(enabled=enabled, heartbeat=heartbeat, dir="debug")


def _wait_until(predicate, timeout=5.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if predicate():
            return True
        time.sleep(0.01)
    return predicate()


def test_debug_sink_none_when_disabled(sock_path):
    cfg = _base_config(sock_path)
    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: FakeCapture([_frame()]),
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: FakeTranscriber(),
        sink_factory=lambda: FakeSink(),
    )
    daemon._initialize()
    assert daemon.debug_sink is None
    daemon.stop()


def test_debug_heartbeat_lines_carry_health_payload(sock_path):
    cfg = _base_config(sock_path)
    cfg.events.heartbeat_ms = 40
    _enable_debug(cfg)
    debug = FakeDebugSink()
    frames = [_frame() for _ in range(40)]
    daemon, collector, subscriber, _ = _start_daemon(
        cfg,
        FakeCapture(frames, delay=0.02),
        FakeVad([]),
        FakeTranscriber(),
        debug_sink_factory=lambda: debug,
    )

    assert daemon.debug_sink is debug
    assert _wait_until(lambda: bool(debug.of("heartbeat")), timeout=5.0)

    beat = debug.of("heartbeat")[0]
    assert beat["model_loaded"] is True
    for field in ("queue_depth", "device_ok", "worker_stall_s", "rms", "uptime_s"):
        assert field in beat

    subscriber.close()
    daemon.stop()
    assert debug.closed is True


def test_debug_heartbeat_gate_off_leaves_only_anomalies(sock_path):
    cfg = _base_config(sock_path)
    cfg.events.heartbeat_ms = 40
    _enable_debug(cfg, heartbeat=False)
    debug = FakeDebugSink()
    daemon, collector, subscriber, _ = _start_daemon(
        cfg,
        FakeCapture([_frame() for _ in range(20)], delay=0.02),
        FakeVad([]),
        FakeTranscriber(),
        debug_sink_factory=lambda: debug,
    )

    assert _wait_until(
        lambda: any(e["type"] == events.HEARTBEAT for e in collector.events),
        timeout=5.0,
    )
    assert daemon.debug_sink is debug
    assert debug.of("heartbeat") == []

    subscriber.close()
    daemon.stop()


def test_debug_reopen_attempt_success_and_fail_lines(sock_path):
    cfg = _base_config(sock_path)
    _enable_debug(cfg)
    debug = FakeDebugSink()
    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: FakeCapture([_frame()]),
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: FakeTranscriber(),
        sink_factory=lambda: FakeSink(),
        debug_sink_factory=lambda: debug,
    )
    daemon._initialize()

    calls = []

    def flaky_factory():
        calls.append(1)
        if len(calls) == 1:
            raise RuntimeError("no device")
        return FakeCapture([_frame()])

    daemon._capture_factory = flaky_factory
    assert daemon._reopen_capture(attempts=2, delay_s=0.0) is True
    assert debug.types().count("capture_reopen_attempt") == 2
    fails = debug.of("capture_reopen_fail")
    assert len(fails) == 1
    assert "no device" in fails[0]["error"]
    assert len(debug.of("capture_reopen_ok")) == 1

    def dead_factory():
        raise RuntimeError("still gone")

    daemon._capture_factory = dead_factory
    assert daemon._reopen_capture(attempts=2, delay_s=0.0) is False
    assert debug.types().count("capture_reopen_attempt") == 4
    assert len(debug.of("capture_reopen_fail")) == 3

    daemon.stop()


def test_debug_capture_failure_emits_fatal_and_stops(sock_path):
    cfg = _base_config(sock_path)
    _enable_debug(cfg)
    debug = FakeDebugSink()

    class BoomCapture(FakeCapture):
        def frames(self):
            raise RuntimeError("device lost")
            yield

    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: BoomCapture([]),
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: FakeTranscriber(),
        sink_factory=lambda: FakeSink(),
        debug_sink_factory=lambda: debug,
    )
    daemon._initialize()

    def dead_factory():
        raise RuntimeError("device lost")

    daemon._capture_factory = dead_factory
    daemon._start_workers()

    assert daemon._stop_event.wait(timeout=10.0)
    daemon.stop()
    assert "capture_reopen_attempt" in debug.types()
    assert "reader_fatal" in debug.types()


def test_debug_clean_exhaustion_emits_no_fatal(sock_path):
    cfg = _base_config(sock_path)
    _enable_debug(cfg)
    debug = FakeDebugSink()
    daemon, collector, subscriber, _ = _start_daemon(
        cfg,
        FakeCapture([_frame()]),
        FakeVad([False]),
        FakeTranscriber(),
        debug_sink_factory=lambda: debug,
    )

    reader = daemon._reader_thread
    assert reader is not None
    assert _wait_until(lambda: not reader.is_alive(), timeout=5.0)
    assert not daemon._stop_event.is_set()
    assert "reader_fatal" not in debug.types()
    assert "worker_fatal" not in debug.types()

    subscriber.close()
    daemon.stop()


def test_debug_queue_drop_and_backlog_warn(sock_path):
    cfg = _base_config(sock_path)
    cfg.daemon.queue_size = 2
    cfg.daemon.backlog_warn_at = 1
    _enable_debug(cfg)
    debug = FakeDebugSink()
    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: FakeCapture([_frame()]),
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: FakeTranscriber(),
        sink_factory=lambda: FakeSink(),
        debug_sink_factory=lambda: debug,
    )
    daemon._initialize()

    def _utt(i):
        return Utterance(
            audio=np.zeros(160, dtype=np.float32),
            duration_s=0.01,
            started_at=time.time(),
            rms=0.0,
            reason="end_of_turn",
            seg_id=i,
        )

    assert daemon._enqueue(_utt(1)) is True
    assert daemon._enqueue(_utt(2)) is True
    assert daemon._enqueue(_utt(3)) is True  # drops seg 1
    daemon._check_backlog()

    drops = debug.of("queue_drop")
    assert len(drops) == 1
    assert drops[0]["dropped_seg_id"] == 1
    warns = debug.of("backlog_warn")
    assert len(warns) == 1
    assert warns[0]["size"] == 2

    daemon.stop()


def test_debug_speaker_fallback_on_lane_failure(sock_path):
    cfg = _base_config(sock_path)
    cfg.speaker.join_timeout_s = 0.05
    _enable_debug(cfg)
    debug = FakeDebugSink()
    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: FakeCapture([_frame()]),
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: FakeTranscriber(),
        sink_factory=lambda: FakeSink(),
        debug_sink_factory=lambda: debug,
    )
    daemon._initialize()

    failed: concurrent.futures.Future = concurrent.futures.Future()
    failed.set_exception(RuntimeError("lane died"))
    assert daemon._resolve_speaker(failed) == (None, "none", "?", None)

    pending: concurrent.futures.Future = concurrent.futures.Future()
    assert daemon._resolve_speaker(pending) == (None, "none", "?", None)

    fallbacks = debug.of("speaker_fallback")
    assert len(fallbacks) == 2
    assert "lane died" in fallbacks[0]["error"]

    daemon.stop()


def test_debug_sink_closed_last_in_stop(sock_path):
    cfg = _base_config(sock_path)
    _enable_debug(cfg)
    order = []
    transcriber = FakeTranscriber()
    transcriber.close = lambda: order.append("transcriber")
    sink = FakeSink()
    sink.close = lambda: order.append("sink")
    debug = FakeDebugSink()
    orig_debug_close = debug.close

    def tracked_debug_close() -> None:
        order.append("debug")
        orig_debug_close()

    debug.close = tracked_debug_close
    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: FakeCapture([_frame()]),
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: transcriber,
        sink_factory=lambda: sink,
        debug_sink_factory=lambda: debug,
    )
    daemon._initialize()
    assert daemon.publisher is not None
    orig_pub_close = daemon.publisher.close

    def tracked_pub_close() -> None:
        order.append("publisher")
        orig_pub_close()

    daemon.publisher.close = tracked_pub_close

    daemon.stop()

    assert order == ["transcriber", "sink", "publisher", "debug"]
    assert debug.closed is True
    assert daemon.debug_sink is None


def test_debug_factory_throw_disables_sink(sock_path):
    cfg = _base_config(sock_path)
    _enable_debug(cfg)

    def boom_factory():
        raise RuntimeError("no dir")

    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: FakeCapture([_frame()]),
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: FakeTranscriber(),
        sink_factory=lambda: FakeSink(),
        debug_sink_factory=boom_factory,
    )
    daemon._initialize()
    assert daemon.debug_sink is None
    daemon.stop()


class FakeProc:
    """Minimal Popen stand-in for supervisor debug tests."""

    _pid_counter = 20000

    def __init__(self, exit_on_poll=None, stubborn=False):
        self.pid = FakeProc._pid_counter
        FakeProc._pid_counter += 1
        self._exit_on_poll = exit_on_poll
        self._stubborn = stubborn
        self.returncode = None
        self.terminated = False
        self.killed = False

    def poll(self):
        if self.returncode is None and self._exit_on_poll is not None:
            self.returncode = self._exit_on_poll
        return self.returncode

    def terminate(self):
        self.terminated = True
        if not self._stubborn and self.returncode is None:
            self.returncode = -15

    def kill(self):
        self.killed = True
        if self.returncode is None:
            self.returncode = -9

    def wait(self, timeout=None):
        if self.returncode is None:
            self.returncode = 0
        return self.returncode


def _supervisor_spawn(procs):
    def fake_spawn(argv, *, redirect_output=False):
        joined = " ".join(argv)
        if "stt_daemon.assistant" in joined:
            return procs["assistant"]
        return procs["daemon"]

    return fake_spawn


def test_supervisor_debug_spawn_exit_shutdown_lines():
    cfg = Config()
    _enable_debug(cfg)
    debug = FakeDebugSink(proc="supervisor")
    procs = {"daemon": FakeProc(exit_on_poll=3), "assistant": FakeProc()}
    supervisor = Supervisor(
        cfg,
        with_tui=False,
        spawn=_supervisor_spawn(procs),
        sleep=lambda s: None,
        debug_sink_factory=lambda: debug,
    )

    assert supervisor.run() == 3
    assert debug.types() == ["spawn", "spawn", "child_exit", "shutdown"]
    assert debug.emitted[2]["code"] == 3
    assert debug.emitted[3]["codes"] == [3, -15]


def test_supervisor_debug_kill_escalation_line():
    cfg = Config()
    _enable_debug(cfg)
    debug = FakeDebugSink(proc="supervisor")
    daemon_proc = FakeProc(stubborn=True)
    procs = {"daemon": daemon_proc, "assistant": FakeProc(exit_on_poll=1)}
    supervisor = Supervisor(
        cfg,
        with_tui=False,
        spawn=_supervisor_spawn(procs),
        sleep=lambda s: None,
        grace_s=0,
        debug_sink_factory=lambda: debug,
    )

    assert supervisor.run() == 1
    assert daemon_proc.killed is True
    kills = debug.of("kill_escalation")
    assert len(kills) == 1
    assert kills[0]["pids"] == [daemon_proc.pid]


def test_supervisor_debug_socket_timeout_line(tmp_path):
    cfg = Config()
    cfg.assistant.control_socket = str(tmp_path / "missing.sock")
    _enable_debug(cfg)
    debug = FakeDebugSink(proc="supervisor")
    procs = {"daemon": FakeProc(), "assistant": FakeProc()}
    supervisor = Supervisor(
        cfg,
        with_tui=True,
        spawn=_supervisor_spawn(procs),
        sleep=lambda s: None,
        wait_timeout_s=0,
        debug_sink_factory=lambda: debug,
    )

    assert supervisor.run() == 1
    assert len(debug.of("socket_timeout")) == 1
    assert debug.of("shutdown")


def test_supervisor_debug_disabled_never_builds_sink():
    cfg = Config()
    calls = []
    procs = {"daemon": FakeProc(exit_on_poll=2), "assistant": FakeProc()}
    supervisor = Supervisor(
        cfg,
        with_tui=False,
        spawn=_supervisor_spawn(procs),
        sleep=lambda s: None,
        debug_sink_factory=lambda: calls.append(1),
    )

    assert supervisor.run() == 2
    assert calls == []


def test_supervisor_debug_emit_throw_never_breaks_shutdown():
    cfg = Config()
    _enable_debug(cfg)
    procs = {"daemon": FakeProc(exit_on_poll=4), "assistant": FakeProc()}
    supervisor = Supervisor(
        cfg,
        with_tui=False,
        spawn=_supervisor_spawn(procs),
        sleep=lambda s: None,
        debug_sink_factory=lambda: BoomDebugSink(proc="supervisor"),
    )

    assert supervisor.run() == 4


def test_request_stop_shuts_down_run_loop(tmp_path):
    """request_stop() (the signal-handler path) must unwind run() cleanly."""
    sock_path = str(tmp_path / "events.sock")
    cfg = _base_config(sock_path)
    capture = FakeCapture([_frame(0)] * 5, stall_after=1)
    sink = FakeSink()
    daemon = Daemon(
        cfg,
        log_level=logging.WARNING,
        capture_factory=lambda: capture,
        vad_factory=lambda _aggr: FakeVad([]),
        transcriber_factory=lambda: FakeTranscriber(),
        sink_factory=lambda: sink,
    )

    thread = threading.Thread(target=daemon.run, daemon=True)
    thread.start()
    time.sleep(0.2)
    daemon.request_stop()
    thread.join(timeout=5)

    assert not thread.is_alive()
    assert capture._closed.is_set()