"""Tests for tools.wake_word — the "Hey Hermes" hotword detector.

No live audio or network: the sounddevice import is faked, engines are stubbed,
and lazy-dep availability is monkeypatched. Covers config resolution, engine
dispatch, the requirements probe, the detector fire/cooldown loop, and the
process-wide singleton lifecycle.
"""

import multiprocessing
import os
import sys
import threading
import time
import types
from pathlib import Path

import pytest

import tools.wake_word as ww


# ── Config helpers ───────────────────────────────────────────────────────


def test_config_defaults_and_clamping():
    assert ww._provider({}) == "openwakeword"
    assert ww._provider({"provider": "Porcupine"}) == "porcupine"
    assert ww._input_device({}) is None
    assert ww._input_device({"input_device": 7}) == 7
    assert ww._input_device({"input_device": " Microphone Array "}) == "Microphone Array"
    assert ww._input_device({"input_device": ""}) is None
    assert ww._input_device({"input_device": False}) is None
    assert ww._sensitivity({"sensitivity": 5}) == 1.0
    assert ww._sensitivity({"sensitivity": -1}) == 0.0
    # Invalid input falls back to the configured default, not a hardcoded 0.5.
    assert ww._sensitivity({"sensitivity": "nope"}) == ww._DEFAULTS["sensitivity"]
    assert ww._sensitivity({}) == ww._DEFAULTS["sensitivity"]
    assert ww.wake_phrase({"phrase": "hey hermes"}) == "hey hermes"
    assert ww.wake_phrase({}) == "hey hermes"


def test_wake_surface_enabled_gate():
    # Disabled → never, regardless of surface.
    assert ww.wake_surface_enabled("cli", {"enabled": False, "surface": "cli"}) is False
    # auto → every surface is eligible; ownership still admits only one.
    for s in ("cli", "tui", "gui"):
        assert ww.wake_surface_enabled(s, {"enabled": True, "surface": "auto"}) is True
    # Pinned surface → only that one.
    cfg = {"enabled": True, "surface": "tui"}
    assert ww.wake_surface_enabled("tui", cfg) is True
    assert ww.wake_surface_enabled("cli", cfg) is False
    assert ww.wake_surface_enabled("gui", cfg) is False
    # Missing/blank surface defaults to auto.
    assert ww.wake_surface_enabled("gui", {"enabled": True}) is True


def test_looks_like_path():
    assert ww._looks_like_path("models/hey_hermes.onnx")
    assert ww._looks_like_path("custom.ppn")
    assert not ww._looks_like_path("hey_jarvis")


def test_load_wake_word_config_is_a_dict_with_defaults():
    # Wired into DEFAULT_CONFIG, so a real load returns the section shape.
    cfg = ww.load_wake_word_config()
    assert isinstance(cfg, dict)
    assert cfg.get("enabled") is False
    assert cfg.get("provider") == "openwakeword"


def test_load_wake_word_config_guards_non_dict(monkeypatch):
    monkeypatch.setattr(
        "hermes_cli.config.load_config", lambda: {"wake_word": "oops"}
    )
    assert ww.load_wake_word_config() == {}


# ── Engine dispatch ──────────────────────────────────────────────────────


def test_build_engine_dispatch(monkeypatch):
    monkeypatch.setattr(ww, "_OpenWakeWordEngine", lambda cfg: "oww")
    monkeypatch.setattr(ww, "_PorcupineEngine", lambda cfg: "pv")
    assert ww._build_engine({"provider": "openwakeword"}) == "oww"
    assert ww._build_engine({"provider": "porcupine"}) == "pv"
    with pytest.raises(ValueError):
        ww._build_engine({"provider": "bogus"})


# ── Requirements probe ───────────────────────────────────────────────────


def _voice_loop_ready(monkeypatch, stt=True, tts=True):
    """Pin the STT/TTS probes so requirements tests don't depend on the
    test venv's installed voice stack."""
    monkeypatch.setattr(ww, "_stt_ready", lambda: stt)
    monkeypatch.setattr(ww, "_tts_ready", lambda: tts)


def test_requirements_openwakeword_available(monkeypatch):
    _voice_loop_ready(monkeypatch)
    monkeypatch.setattr(ww, "_audio_available", lambda: True)
    monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
    r = ww.check_wake_word_requirements(
        {"provider": "openwakeword", "phrase": "hey hermes"}
    )
    assert r["available"] is True
    assert r["provider"] == "openwakeword"
    assert r["phrase"] == "hey hermes"


def test_tts_ready_is_a_probe_never_an_installer(monkeypatch):
    """_tts_ready must NOT trigger lazy pip installs from a status poll.

    Regression: check_tts_requirements → _import_edge_tts → lazy_deps.ensure
    ran pip inside wake.status; a slow/failed install froze the poll and
    unmounted the desktop ear. Uninstalled-but-lazy-installable counts as
    ready WITHOUT calling ensure/check.
    """
    import types as _types

    monkeypatch.setattr(
        ww, "_tts_ready", ww.__dict__["_tts_ready"]
    )  # use the real implementation
    fake_tts = _types.SimpleNamespace(
        _get_provider=lambda cfg: "edge",
        _load_tts_config=lambda: {},
        check_tts_requirements=lambda: (_ for _ in ()).throw(
            AssertionError("check_tts_requirements must not run when deps are missing")
        ),
    )
    monkeypatch.setitem(sys.modules, "tools.tts_tool", fake_tts)

    # Deps missing + lazy installs allowed → ready (installs at first speak).
    monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: False)
    monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: True)
    assert ww._tts_ready() is True

    # Deps missing + lazy installs disabled → not ready.
    monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: False)
    assert ww._tts_ready() is False

    # Deps present → falls through to the real requirements check.
    fake_tts.check_tts_requirements = lambda: True
    monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
    assert ww._tts_ready() is True


def test_requirements_fresh_install_lazy_allowed(monkeypatch):
    """Deps missing + lazy installs allowed → available, so /wake on can
    reach the engine constructor's ``lazy_deps.ensure()`` call.

    Regression: the audio probe imports sounddevice/numpy — packages the
    lazy installer would fetch — so gating ``available`` on it made the
    lazy-install path unreachable on a fresh machine (the /wake on handler
    printed the pip hint and bailed before ensure() ever ran).
    """
    def _boom():
        raise AssertionError("audio probe must not run while deps are missing")

    _voice_loop_ready(monkeypatch)
    monkeypatch.setattr(ww, "_audio_available", _boom)
    monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: False)
    monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: True)
    r = ww.check_wake_word_requirements({"provider": "openwakeword"})
    assert r["available"] is True
    assert r["deps_available"] is False
    assert r["hint"] == ""


def test_requirements_deps_present_but_no_audio_hint(monkeypatch):
    """Once deps ARE installed, a failing audio probe blocks with a mic hint
    (lazy installs can't fix a missing audio device)."""
    _voice_loop_ready(monkeypatch)
    monkeypatch.setattr(ww, "_audio_available", lambda: False)
    monkeypatch.setattr(ww, "_local_input_device_ready", lambda: False)
    monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
    monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: True)
    r = ww.check_wake_word_requirements({"provider": "openwakeword", "capture": "local"})
    assert r["available"] is False
    assert "audio device" in r["hint"] or "microphone" in r["hint"].lower()


# ── openWakeWord engine (bundled model + base-model fetch) ───────────────


def _install_fake_openwakeword(monkeypatch):
    """Swap in a fake ``openwakeword`` so the engine builds with no network.

    Returns a ``calls`` dict recording every ``download_models`` invocation.
    """
    calls = {"download": []}

    class _FakeModel:
        def __init__(self, wakeword_models, inference_framework="onnx"):
            self.wakeword_models = list(wakeword_models)
            self.models = {"hey_hermes": object()}

        def predict(self, frame):
            return {"hey_hermes": 0.0}

        def reset(self):
            pass

    oww = types.ModuleType("openwakeword")
    oww.utils = types.SimpleNamespace(
        download_models=lambda names=[]: calls["download"].append(list(names))
    )
    model_mod = types.ModuleType("openwakeword.model")
    model_mod.Model = _FakeModel

    monkeypatch.setitem(sys.modules, "openwakeword", oww)
    monkeypatch.setitem(sys.modules, "openwakeword.model", model_mod)
    monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None)
    return calls


def test_openwakeword_ensures_base_models_for_custom_path(monkeypatch):
    # Regression: a custom ``.onnx`` path used to skip download_models entirely,
    # so a fresh install crashed at load time on a missing melspectrogram.onnx.
    # The base feature models must be ensured for a custom path too.
    calls = _install_fake_openwakeword(monkeypatch)
    eng = ww._OpenWakeWordEngine(
        {"provider": "openwakeword", "openwakeword": {"model": "/models/hey_hermes.onnx"}}
    )
    assert calls["download"] == [["/models/hey_hermes.onnx"]]
    assert eng._labels == ["hey_hermes"]


def test_bundled_hey_hermes_model_ships_on_disk():
    # The "hey hermes" wake word works out of the box only if the model is
    # actually bundled. Both framework artifacts must exist and be non-trivial.
    for framework in ("onnx", "tflite"):
        path = ww._bundled_wakeword_path(framework)
        assert os.path.exists(path), path
        assert os.path.getsize(path) > 1024, path


# ── platform-aware backend selection (openWakeWord onnx is broken on macOS ARM64,
#    upstream dscripka/openWakeWord#336) ────────────────────────────────────────

def test_default_framework_tracks_the_macos_arm64_probe():
    """``default_inference_framework()`` is exactly the ``_is_macos_arm64()``
    branch — tflite there, onnx everywhere else.

    Stated as an invariant between the probe and its consumer so it holds on
    every host, including the macOS runner (where both sides are real) and an
    Intel Mac (where ONNX is fine and both sides say so).
    """
    expected = "tflite" if ww._is_macos_arm64() else "onnx"
    assert ww.default_inference_framework() == expected


@pytest.mark.macos_only
def test_macos_arm64_prefers_tflite_on_this_host():
    """On a real ARM64 Mac the default must be tflite (upstream #336).

    Runs on the macOS CI job, where ``platform.machine()`` and
    ``sys.platform`` are the genuine article rather than a patched pair.
    """
    if not ww._is_macos_arm64():
        pytest.skip("Intel Mac — ONNX works here, nothing to assert")
    assert ww.default_inference_framework() == "tflite"
    assert ww.resolve_inference_framework({}) == "tflite"
    assert ww.resolve_inference_framework({"openwakeword": {"inference_framework": ""}}) == "tflite"
    # The one explicit value we override: pinned onnx is provably dead here.
    assert ww.resolve_inference_framework(
        {"openwakeword": {"inference_framework": "onnx"}}
    ) == "tflite"


def test_explicit_framework_kept_where_onnx_works(monkeypatch):
    # An operator who pins a backend keeps it everywhere ONNX actually works.
    calls = _install_fake_openwakeword(monkeypatch)
    monkeypatch.setattr(ww, "_is_macos_arm64", lambda: False)
    ww._OpenWakeWordEngine(
        {"provider": "openwakeword", "openwakeword": {"inference_framework": "onnx"}}
    )
    (downloaded,) = calls["download"]
    assert downloaded == [ww._bundled_wakeword_path("onnx")]


def test_empty_framework_falls_back_to_platform_default(monkeypatch):
    """Empty/missing config defers to ``default_inference_framework()``.

    The macOS-ARM64 side of the fallback is asserted for real in
    ``test_macos_arm64_prefers_tflite_on_this_host``; here we pin the
    delegation itself by swapping the platform probe (a seam in our own
    module) rather than lying to the interpreter about which OS it is on.
    """
    monkeypatch.setattr(ww, "_is_macos_arm64", lambda: True)
    assert ww.resolve_inference_framework({}) == "tflite"
    assert ww.resolve_inference_framework({"openwakeword": {"inference_framework": ""}}) == "tflite"
    monkeypatch.setattr(ww, "_is_macos_arm64", lambda: False)
    assert ww.resolve_inference_framework({}) == "onnx"


# ── ambient-speech rejection: consecutive-frame confirmation ──────────────────

def _openwakeword_engine_with_scores(monkeypatch, cfg_wake, scores):
    """Build a real _OpenWakeWordEngine whose predict() replays ``scores``."""
    seq = iter(scores)

    class _ScriptedModel:
        def __init__(self, wakeword_models, inference_framework="onnx"):
            self.models = {"hey_hermes": object()}

        def predict(self, frame):
            return {"hey_hermes": next(seq)}

        def reset(self):
            pass

    oww = types.ModuleType("openwakeword")
    oww.utils = types.SimpleNamespace(download_models=lambda names=[]: None)
    model_mod = types.ModuleType("openwakeword.model")
    model_mod.Model = _ScriptedModel
    monkeypatch.setitem(sys.modules, "openwakeword", oww)
    monkeypatch.setitem(sys.modules, "openwakeword.model", model_mod)
    monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None)
    monkeypatch.setattr(ww, "ensure_tflite_runtime", lambda: True)
    return ww._OpenWakeWordEngine({"provider": "openwakeword", **cfg_wake})


# ── sherpa-onnx open-vocabulary engine ───────────────────────────────────


def _install_fake_sherpa(monkeypatch, tmp_path):
    """Fake sherpa_onnx + a fake model dir so the engine builds offline."""
    calls = {"text2token": [], "spotter": [], "results": []}

    model_dir = tmp_path / "kws-model"
    model_dir.mkdir()
    for name in (
        "tokens.txt",
        "bpe.model",
        "encoder-epoch-12-avg-2-chunk-16-left-64.onnx",
        "decoder-epoch-12-avg-2-chunk-16-left-64.onnx",
        "joiner-epoch-12-avg-2-chunk-16-left-64.onnx",
    ):
        (model_dir / name).write_bytes(b"x")

    class _FakeStream:
        def accept_waveform(self, sample_rate, samples):
            pass

    class _FakeSpotter:
        def __init__(self, **kwargs):
            calls["spotter"].append(kwargs)

        def create_stream(self):
            return _FakeStream()

        def is_ready(self, stream):
            return bool(calls["results"])

        def decode_stream(self, stream):
            pass

        def get_result(self, stream):
            return calls["results"].pop(0) if calls["results"] else ""

        def reset_stream(self, stream):
            pass

    def _fake_text2token(phrases, tokens, tokens_type, bpe_model):
        calls["text2token"].append(list(phrases))
        return [p.split() for p in phrases]

    sherpa = types.ModuleType("sherpa_onnx")
    sherpa.KeywordSpotter = _FakeSpotter
    sherpa.text2token = _fake_text2token
    monkeypatch.setitem(sys.modules, "sherpa_onnx", sherpa)
    monkeypatch.setattr("tools.lazy_deps.ensure", lambda *a, **k: None)

    # numpy is an optional voice-extra dep, lazy-installed at runtime — CI's
    # hermetic slices don't have it. process() only calls asarray(...)/32768,
    # so a minimal stub keeps these tests runnable without the real package.
    if "numpy" not in sys.modules:
        class _FakeArr(list):
            def __truediv__(self, other):
                return self

        np_stub = types.ModuleType("numpy")
        np_stub.float32 = "float32"
        np_stub.asarray = lambda x, dtype=None: _FakeArr(x)
        monkeypatch.setitem(sys.modules, "numpy", np_stub)
    return calls, model_dir


# ── Multi-profile phrase routing ─────────────────────────────────────────


# ── Detector loop ────────────────────────────────────────────────────────


class _FakeStream:
    """Always-readable input stream that yields trivial frames."""

    def __init__(self, **_kw):
        self.closed = False

    def start(self):
        pass

    def read(self, n):
        time.sleep(0.01)
        return [0] * n, False

    def stop(self):
        pass

    def close(self):
        self.closed = True


class _FakeEngine:
    frame_length = 4

    def __init__(self, fire=True):
        self._fire = fire
        self.closed = False
        self.resets = 0

    def process(self, frame):
        return self._fire

    def reset(self):
        self.resets += 1

    def close(self):
        self.closed = True


def _fake_audio(monkeypatch):
    fake_sd = types.SimpleNamespace(InputStream=lambda **kw: _FakeStream(**kw))
    monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))


class _Frame(list):
    """List with numpy-ish abs()/max() so the silence probe sees real peaks."""

    def __abs__(self):
        return _Frame(abs(x) for x in self)

    def max(self):
        return max(self) if self else 0


class _SilentStream(_FakeStream):
    """Stream that always yields near-zero frames (dead macOS mic)."""

    def read(self, n):
        time.sleep(0.005)
        return _Frame([0] * n), False


class _LoudStream(_FakeStream):
    """Stream that yields audible frames."""

    def read(self, n):
        time.sleep(0.005)
        return _Frame([500] * n), False


def test_detector_opens_configured_input_device_and_reports_backend(monkeypatch):
    opened = []
    reads = []
    processed = []

    class _NativeRateStream(_LoudStream):
        def read(self, n):
            reads.append(n)
            return super().read(n)

    class _RecordingEngine(_FakeEngine):
        def process(self, frame):
            processed.append(frame)
            return False

    def _stream(**kwargs):
        opened.append(kwargs)
        return _NativeRateStream(**kwargs)

    fake_sd = types.SimpleNamespace(
        InputStream=_stream,
        query_devices=lambda selector, kind: {
            "name": "Microphone Array",
            "hostapi": 2,
            "max_input_channels": 2,
            "default_samplerate": 48000.0,
        },
        query_hostapis=lambda index: {"name": "Windows WASAPI"},
    )
    np = pytest.importorskip("numpy")
    monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, np))

    det = ww.WakeWordDetector(
        _RecordingEngine(fire=False),
        lambda: None,
        input_device="Microphone Array",
    )
    det.start()
    try:
        assert opened[0]["device"] == "Microphone Array"
        assert opened[0]["samplerate"] == 48000
        assert opened[0]["blocksize"] == 12
        deadline = time.monotonic() + 2.0
        while not processed and time.monotonic() < deadline:
            time.sleep(0.01)
        assert reads[0] == 12
        assert len(processed[0]) == 4
        assert processed[0].tolist() == [500] * 4
        assert det.input_device_details == {
            "selector": "Microphone Array",
            "name": "Microphone Array",
            "hostapi_index": 2,
            "hostapi": "Windows WASAPI",
            "max_input_channels": 2,
            "default_samplerate": 48000.0,
        }
    finally:
        det.stop()


@pytest.mark.windows_only
def test_windows_silent_hint_names_selected_device():
    hint = ww.silent_audio_hint(
        {
            "selector": 3,
            "name": "Microphone Array",
            "hostapi": "Windows WASAPI",
        }
    )
    assert "Microphone Array (Windows WASAPI)" in hint
    assert "wake_word.input_device" in hint
    assert "macOS" not in hint


@pytest.mark.macos_only
def test_macos_silent_hint_points_at_privacy_settings():
    """On macOS a silent stream is almost always the TCC mic permission, so the
    hint names System Settings rather than the device."""
    hint = ww.silent_audio_hint(
        {"selector": 1, "name": "MacBook Pro Microphone", "hostapi": "Core Audio"}
    )
    assert "Privacy & Security" in hint
    assert "Microphone" in hint


@pytest.mark.linux_only
def test_linux_silent_hint_names_selected_device():
    hint = ww.silent_audio_hint(
        {"selector": 2, "name": "HD Audio Capture", "hostapi": "ALSA"}
    )
    assert "HD Audio Capture (ALSA)" in hint
    assert "Privacy & Security" not in hint


def test_detector_flags_silent_stream_and_recovers(monkeypatch):
    """A stream of zeros sets audio_silent; audible input clears it."""
    monkeypatch.setattr(ww, "_SILENCE_ALERT_SECONDS", 0.001)  # trip on the first frame
    stream_cls = {"cls": _SilentStream}
    fake_sd = types.SimpleNamespace(InputStream=lambda **kw: stream_cls["cls"](**kw))
    monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))

    det = ww.WakeWordDetector(_FakeEngine(fire=False), lambda: None)
    det.start()
    try:
        deadline = time.monotonic() + 2.0
        while not det.audio_silent and time.monotonic() < deadline:
            time.sleep(0.01)
        assert det.audio_silent is True
        assert ww.audio_is_silent() is False  # module accessor needs the singleton

        monkeypatch.setattr(ww, "_detector", det)
        assert ww.audio_is_silent() is True

        # Audio returns (permission granted / real mic) — flag clears.
        det.pause()
        stream_cls["cls"] = _LoudStream
        det.resume()
        deadline = time.monotonic() + 2.0
        while det.audio_silent and time.monotonic() < deadline:
            time.sleep(0.01)
        assert det.audio_silent is False
    finally:
        monkeypatch.setattr(ww, "_detector", None)
        det.stop()


# ── Singleton lifecycle ──────────────────────────────────────────────────


def test_detection_callback_can_pause_and_close_stream(monkeypatch, tmp_path):
    streams = []

    def _stream(**kw):
        stream = _FakeStream(**kw)
        streams.append(stream)
        return stream

    fake_sd = types.SimpleNamespace(InputStream=_stream)
    monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))
    monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=True))
    monkeypatch.setattr(ww, "_lock_path", lambda: tmp_path / "wake.lock")
    owner = object()
    paused = threading.Event()

    def _on_wake():
        if ww.pause_listening(owner=owner):
            paused.set()

    ww.start_listening(_on_wake, owner=owner, config={})
    assert paused.wait(2)
    assert ww.is_listening() is False
    assert streams[0].closed is True
    assert ww.stop_listening(owner=owner) is True


def test_startup_failure_releases_owner_and_machine_lock(monkeypatch, tmp_path):
    class _BrokenSoundDevice:
        @staticmethod
        def InputStream(**_kw):
            raise OSError("no microphone")

    lock_path = tmp_path / "wake.lock"
    monkeypatch.setattr(ww, "_import_audio", lambda: (_BrokenSoundDevice, None))
    monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=False))
    monkeypatch.setattr(ww, "_lock_path", lambda: lock_path)
    owner = object()

    with pytest.raises(RuntimeError, match="Failed to open"):
        ww.start_listening(lambda: None, owner=owner, config={})

    assert ww.owns_listener(owner) is False
    handle = ww._acquire_machine_lock(lock_path)
    ww._release_machine_lock(handle)


def test_stream_failure_releases_owner_and_machine_lock(monkeypatch, tmp_path):
    class _FailingStream(_FakeStream):
        def read(self, _n):
            raise OSError("device disconnected")

    fake_sd = types.SimpleNamespace(InputStream=lambda **kw: _FailingStream(**kw))
    engine = _FakeEngine(fire=False)
    lock_path = tmp_path / "wake.lock"
    monkeypatch.setattr(ww, "_import_audio", lambda: (fake_sd, None))
    monkeypatch.setattr(ww, "_build_engine", lambda cfg: engine)
    monkeypatch.setattr(ww, "_lock_path", lambda: lock_path)
    owner = object()

    ww.start_listening(lambda: None, owner=owner, config={})
    deadline = time.time() + 2
    while ww.owns_listener(owner) and time.time() < deadline:
        time.sleep(0.01)

    assert ww.owns_listener(owner) is False
    assert engine.closed is True
    handle = ww._acquire_machine_lock(lock_path)
    ww._release_machine_lock(handle)


def _hold_machine_lock(path: str, ready, release) -> None:
    from tools import wake_word

    handle = wake_word._acquire_machine_lock(Path(path))
    ready.set()
    release.wait(10)
    assert handle is not None


def test_machine_lock_is_released_when_owner_process_exits(tmp_path):
    lock_path = tmp_path / "wake.lock"
    ctx = multiprocessing.get_context("spawn")
    ready = ctx.Event()
    release = ctx.Event()
    process = ctx.Process(
        target=_hold_machine_lock,
        args=(str(lock_path), ready, release),
    )
    process.start()
    try:
        assert ready.wait(10)
        with pytest.raises(ww.WakeWordInUse):
            ww._acquire_machine_lock(lock_path)
        release.set()
        process.join(10)
        assert process.exitcode == 0
        handle = ww._acquire_machine_lock(lock_path)
        ww._release_machine_lock(handle)
    finally:
        release.set()
        if process.is_alive():
            process.terminate()
        process.join(10)


# ── Client capture (remote desktop mic → wake.feed) ──────────────────────


def test_resolve_capture_mode_auto_and_prefer_client(monkeypatch):
    monkeypatch.setattr(ww, "_local_input_device_ready", lambda: False)
    # auto without prefer_client stays local (CLI/TUI/status semantics)
    assert ww.resolve_capture_mode({"capture": "auto"}) == "local"
    assert ww.resolve_capture_mode({"capture": "auto"}, prefer_client=True) == "client"
    assert ww.resolve_capture_mode({"capture": "local"}, prefer_client=True) == "local"
    assert ww.resolve_capture_mode({"capture": "client"}) == "client"
    assert ww.resolve_capture_mode({"capture": "auto"}, force_local=True) == "local"
    monkeypatch.setattr(ww, "_local_input_device_ready", lambda: True)
    assert ww.resolve_capture_mode({"capture": "auto"}) == "local"
    # A working backend mic wins under auto even for a preferring surface, so
    # local desktops keep PortAudio + wake_word.input_device selection.
    assert ww.resolve_capture_mode({"capture": "auto"}, prefer_client=True) == "local"
    # Explicit client still forces streaming (backend mic exists but is wrong).
    assert ww.resolve_capture_mode({"capture": "client"}, prefer_client=True) == "client"


def test_requirements_client_capture_without_local_mic(monkeypatch):
    monkeypatch.setattr(ww, "_audio_available", lambda: False)
    monkeypatch.setattr(ww, "_local_input_device_ready", lambda: False)
    monkeypatch.setattr(ww, "_stt_ready", lambda: True)
    monkeypatch.setattr(ww, "_tts_ready", lambda: True)

    class _LD:
        @staticmethod
        def is_available(feature):
            return True

        @staticmethod
        def _allow_lazy_installs():
            return False

        @staticmethod
        def feature_install_command(feature):
            return ""

    monkeypatch.setattr(ww, "lazy_deps", _LD, raising=False)
    import tools.lazy_deps as real_ld
    monkeypatch.setattr("tools.lazy_deps.is_available", lambda f: True)
    monkeypatch.setattr("tools.lazy_deps._allow_lazy_installs", lambda: False)

    reqs = ww.check_wake_word_requirements({"capture": "client", "provider": "openwakeword"})
    assert reqs["available"] is True
    assert reqs["capture"] == "client"


def test_client_capture_feed_fires(monkeypatch, tmp_path):
    np = pytest.importorskip("numpy")

    monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=True))
    monkeypatch.setattr(ww, "_lock_path", lambda: tmp_path / "wake.lock")
    # External mode must not import sounddevice
    monkeypatch.setattr(
        ww,
        "_import_audio",
        lambda: (_ for _ in ()).throw(OSError("no local mic")),
    )
    owner = object()
    fired = threading.Event()

    def _on_wake():
        fired.set()

    ww.start_listening(_on_wake, owner=owner, config={}, external_audio=True)
    assert ww.is_listening() is True
    info = ww.detector_frame_info()
    fl = int(info["frame_length"])
    # Non-silent frame so silence flag does not dominate
    pcm = (np.ones(fl, dtype=np.int16) * 5000).tobytes()
    assert ww.feed_audio(owner=owner, pcm_int16=pcm) is True
    assert fired.wait(2.0)
    assert ww.stop_listening(owner=owner) is True


def test_feed_audio_rejects_wrong_owner(monkeypatch, tmp_path):
    monkeypatch.setattr(ww, "_build_engine", lambda cfg: _FakeEngine(fire=False))
    monkeypatch.setattr(ww, "_lock_path", lambda: tmp_path / "wake.lock")
    owner = object()
    ww.start_listening(lambda: None, owner=owner, config={}, external_audio=True)
    assert ww.feed_audio(owner=object(), pcm_int16=b"\x00\x00") is False
    assert ww.stop_listening(owner=owner) is True