"""Full-duplex audio: continuous 16 kHz mic capture + interruptible playback.

Mic runs in a callback thread and feeds an asyncio queue of 32 ms frames
(512 samples @ 16 kHz — the frame size Silero VAD expects).

Playback keeps its own OutputStream; stop() cuts audio within one block
(~10 ms) for instant barge-in.
"""

from __future__ import annotations

import asyncio
import os
import subprocess
import threading
from pathlib import Path

import numpy as np
import sounddevice as sd

try:
    from pynput import keyboard as _kb
    _PYNPUT_OK = True
except ImportError:
    _PYNPUT_OK = False


# Global mute toggle, shared across all mic implementations. Muting now RELEASES
# the physical mic capture device (kills the AEC helper / closes the input
# stream) — not just gates frames — so Bluetooth earbuds drop the low-quality
# call profile (HFP/SCO) and return to high-quality music output (A2DP) while
# you're only listening to Jarvis. Unmuting re-acquires the device.
_mic_muted = threading.Event()  # not set = unmuted (normal)
_break = threading.Event()       # set = user hit the panic/break hotkey
_active_mic = None               # the running Mic instance (registers in start())
_mute_lock = threading.Lock()

MUTE_HOTKEY = os.environ.get("MUTE_KEY", "<ctrl>+<alt>+m")
# Panic key: instantly abort whatever Jarvis is doing (speaking, or driving
# the mouse/screen). Handy so a runaway control action can always be stopped.
BREAK_HOTKEY = os.environ.get("BREAK_KEY", "<ctrl>+<alt>+.")


def signal_break() -> None:
    """Request an immediate abort of the current turn/control."""
    _break.set()
    print(f"\n  BREAK — aborting current action ({BREAK_HOTKEY})")


def break_requested() -> bool:
    """True (and clears) if a break was requested since the last check."""
    if _break.is_set():
        _break.clear()
        return True
    return False


def _apply_mute(muted: bool) -> None:
    """Set the mute flag AND release / re-acquire the physical mic device."""
    with _mute_lock:
        if muted:
            _mic_muted.set()
        else:
            _mic_muted.clear()
        m = _active_mic
    if m is not None:
        try:
            m._on_mute_change(muted)  # noqa: SLF001
        except Exception as e:  # noqa: BLE001
            print(f"  [mute] mic device toggle failed: {e}")


def set_muted(muted: bool) -> None:
    """Public entry — hotkey, HUD mic button, or control file all route here."""
    if _mic_muted.is_set() == bool(muted):
        return
    _apply_mute(muted)
    print("\n  MIC MUTED — device released (earbuds back to hi-fi output)"
          if muted else
          f"\n  MIC UNMUTED — device re-acquired ({MUTE_HOTKEY})")


def _toggle_mute() -> None:
    set_muted(not _mic_muted.is_set())


def _start_mute_hotkey() -> None:
    """Global hotkey (Ctrl+Alt+M) + terminal fallback ('m'+Enter).

    The global hotkey needs macOS Input Monitoring/Accessibility for the
    terminal app (System Settings -> Privacy & Security); if that's not
    granted it silently receives nothing, so the stdin fallback matters.
    """
    if _PYNPUT_OK:
        try:
            hk = _kb.GlobalHotKeys({MUTE_HOTKEY: _toggle_mute,
                                    BREAK_HOTKEY: signal_break})
            hk.daemon = True
            hk.start()
            print(f"  mute hotkey: {MUTE_HOTKEY} · break/abort: {BREAK_HOTKEY} "
                  "(global; needs Input Monitoring) or type m/stop+Enter here")
        except Exception as e:
            print(f"  [mute] global hotkey failed ({e}); use m+Enter")
    else:
        print("  [mute] pynput missing; mute with m+Enter in this terminal")

    def stdin_loop() -> None:
        import sys
        for line in sys.stdin:
            w = line.strip().lower()
            if w in ("m", "mute", "unmute"):
                _toggle_mute()
            elif w in ("stop", "break", "abort", "x"):
                signal_break()

    t = threading.Thread(target=stdin_loop, daemon=True)
    t.start()


def is_muted() -> bool:
    return _mic_muted.is_set()


def restore_input_volume(floor: int = 70) -> None:
    """Apple's VoiceProcessingIO permanently drags the system input gain
    down (documented behavior) — a day of sessions left it at 27/100 and
    the assistant went deaf. Self-heal at startup."""
    import sys
    if sys.platform != "darwin":
        return  # no VoiceProcessingIO gain-drag off macOS; nothing to heal
    try:
        out = subprocess.run(
            ["osascript", "-e", "input volume of (get volume settings)"],
            capture_output=True, text=True, timeout=5,
        )
        current = int(out.stdout.strip())
        if current < floor:
            subprocess.run(
                ["osascript", "-e", f"set volume input volume {floor}"],
                capture_output=True, timeout=5,
            )
            print(f"  input volume was {current}/100 (VPIO gain-drag) — "
                  f"restored to {floor}")
    except Exception:
        pass

SAMPLE_RATE = 16_000
FRAME_SAMPLES = 512  # 32 ms @ 16 kHz, Silero VAD's required frame size

AEC_HELPER = Path(__file__).resolve().parent.parent / "helper" / "aecmic"
SYSAUDIO_HELPER = Path(__file__).resolve().parent.parent / "helper" / "sysaudio"


class _FrameQueue:
    """Shared drop-oldest frame delivery into the asyncio loop."""

    def __init__(self, loop: asyncio.AbstractEventLoop):
        self._loop = loop
        self.frames: asyncio.Queue[np.ndarray] = asyncio.Queue(maxsize=256)

    def _put_nowait(self, frame: np.ndarray) -> None:
        try:
            self.frames.put_nowait(frame)
        except asyncio.QueueFull:
            # consumer stalled; drop oldest so we stay realtime
            try:
                self.frames.get_nowait()
                self.frames.put_nowait(frame)
            except asyncio.QueueEmpty:
                pass

    def _deliver(self, frame: np.ndarray) -> None:
        try:
            self._loop.call_soon_threadsafe(self._put_nowait, frame)
        except RuntimeError:
            pass  # loop shutting down


def _pick_input_device() -> int | None:
    """Choose the microphone.

    MIC_DEVICE env wins (an index, or a name substring). Otherwise score all
    input devices: real microphones beat virtual ones (Steam Streaming,
    VB-Cable, Stereo Mix, Sound Mapper...), and WASAPI beats MME/DirectSound —
    Windows wireless-headset mics through MME often deliver audio too quiet
    for the VAD to ever trigger. Returns a device index, or None = default.
    """
    try:
        devs = sd.query_devices()
    except Exception:
        return None
    want = os.environ.get("MIC_DEVICE", "").strip()
    if want:
        try:
            return int(want)
        except ValueError:
            for i, d in enumerate(devs):
                if (d["max_input_channels"] > 0
                        and want.lower() in d["name"].lower()):
                    return i
    try:
        default_in = sd.default.device[0]
    except Exception:
        default_in = None
    bad = ("steam", "virtual", "cable", "loopback", "voicemeeter",
           "stereo mix", "sound mapper", "primary sound", "wave", "line in")
    best_i, best_s = None, -1e9
    for i, d in enumerate(devs):
        if d["max_input_channels"] <= 0:
            continue
        n = d["name"].lower()
        s = 0.0
        if any(b in n for b in bad):
            s -= 100
        if "microphone" in n or "mic" in n:
            s += 10
        if "headset" in n:
            s += 5
        try:
            api = sd.query_hostapis(d["hostapi"])["name"].lower()
            if "wasapi" in api:
                s += 6
            elif "wdm-ks" in api:
                s -= 5   # exclusive-mode drivers; avoid grabbing them
        except Exception:
            pass
        if i == default_in:
            s += 2
        if s > best_s:
            best_i, best_s = i, s
    return best_i


def _input_candidates() -> list[int | None]:
    """All plausible mics, best first, ending with the system default (None).

    A device can EXIST yet deliver pure zeros (a wireless dongle with the
    headset off, a dead WASAPI endpoint...), so callers should PROBE each
    candidate and take the first one that produces actual signal.
    """
    try:
        devs = sd.query_devices()
    except Exception:
        return [None]
    scored: list[tuple[float, int]] = []
    bad = ("steam", "virtual", "cable", "loopback", "voicemeeter",
           "stereo mix", "sound mapper", "primary sound", "wave", "line in")
    try:
        default_in = sd.default.device[0]
    except Exception:
        default_in = None
    for i, d in enumerate(devs):
        if d["max_input_channels"] <= 0:
            continue
        n = d["name"].lower()
        s = 0.0
        if any(b in n for b in bad):
            s -= 100
        if "microphone" in n or "mic" in n:
            s += 10
        if "headset" in n:
            s += 5
        try:
            api = sd.query_hostapis(d["hostapi"])["name"].lower()
            if "wasapi" in api:
                s += 6
            elif "wdm-ks" in api:
                s -= 20   # exclusive-mode drivers; last resort
        except Exception:
            pass
        if i == default_in:
            s += 2
        if s > -50:       # drop the clearly-virtual ones entirely
            scored.append((s, i))
    scored.sort(reverse=True)
    out: list[int | None] = [i for _s, i in scored]
    out.append(None)      # system default as the final fallback
    return out


def _probe_device(dev: int | None, seconds: float = 0.6) -> float:
    """Record briefly from `dev`; return the peak. A LIVE mic always shows
    ambient noise > 0 — an exact 0.0 means the endpoint is dead/silent."""
    try:
        rec = sd.rec(int(seconds * SAMPLE_RATE), samplerate=SAMPLE_RATE,
                     channels=1, dtype="float32", device=dev)
        sd.wait()
        return float(np.abs(rec).max())
    except Exception:
        return -1.0   # can't open at 16 kHz at all


class MicStream(_FrameQueue):
    """Raw mic capture (sounddevice) with device auto-pick + auto-gain.

    Auto-gain: wireless/USB headset mics on Windows often capture speech at
    ~0.03 peak, far below what the VAD reads as voice. We track the rolling
    peak and scale frames toward a working level before the VAD sees them.
    MIC_GAIN env forces a fixed gain instead (e.g. MIC_GAIN=6).
    """

    _TARGET = 0.30      # desired speech peak after gain (headroom, no clipping)
    _MAX_GAIN = 8.0
    _FLOOR = 0.003      # below this = silence; don't chase noise

    def __init__(self, loop: asyncio.AbstractEventLoop):
        super().__init__(loop)
        self._stream: sd.InputStream | None = None
        self._device: int | None = None  # remembered for reopen after mute
        g = os.environ.get("MIC_GAIN", "").strip().lower()
        self._fixed_gain = float(g) if g and g != "auto" else None
        self._gain = self._fixed_gain or 1.0
        self._roll_peak = 0.0

    def _agc(self, frame: np.ndarray) -> np.ndarray:
        if self._fixed_gain is not None:
            return np.clip(frame * self._fixed_gain, -1.0, 1.0)
        peak = float(np.abs(frame).max())
        # rolling peak: fast attack, slow decay (~4s to halve)
        self._roll_peak = max(peak, self._roll_peak * 0.992)
        if self._roll_peak > self._FLOOR:
            target = min(self._MAX_GAIN, self._TARGET / self._roll_peak)
            # smooth gain moves so speech isn't pumped mid-word
            self._gain += 0.1 * (target - self._gain)
        out = frame * self._gain
        # Anti-clip: if this frame would exceed headroom, pull gain DOWN fast
        # so we never feed a full-scale (distorted) frame to the transcriber.
        outpeak = float(np.abs(out).max())
        if outpeak > 0.85:
            self._gain *= 0.85 / outpeak
            out = frame * self._gain
        return np.clip(out, -1.0, 1.0)

    def _callback(self, indata, _frames, _time, status):
        if _mic_muted.is_set():
            return
        self._deliver(self._agc(indata[:, 0]).astype(np.float32))

    def _on_mute_change(self, muted: bool) -> None:
        # Close the input stream on mute (releases the device → Bluetooth
        # returns to hi-fi output); reopen it on unmute with the same device.
        if muted:
            s, self._stream = self._stream, None
            if s is not None:
                try:
                    s.stop(); s.close()
                except Exception:  # noqa: BLE001
                    pass
        elif self._stream is None:
            try:
                self._stream = sd.InputStream(
                    device=self._device, samplerate=SAMPLE_RATE, channels=1,
                    dtype="float32", blocksize=FRAME_SAMPLES,
                    callback=self._callback,
                )
                self._stream.start()
            except Exception as e:  # noqa: BLE001
                print(f"  [mic] reopen after unmute failed: {e}")

    def start(self) -> None:
        forced = os.environ.get("MIC_DEVICE", "").strip()
        if forced and not forced.isdigit():
            # A name like "C920" can match several endpoints (MME/WASAPI/WDM);
            # probe all of them and take the first that shows real signal, so
            # a dead duplicate endpoint doesn't leave us deaf.
            want = forced.lower()
            try:
                matches = [i for i, d in enumerate(sd.query_devices())
                           if d["max_input_channels"] > 0
                           and want in d["name"].lower()]
            except Exception:  # noqa: BLE001
                matches = []
            candidates = []
            for dev in matches:
                if _probe_device(dev) > 0.0:
                    candidates.append(dev)
            candidates += matches            # then any matching (even if quiet)
            if not candidates:
                candidates = [_pick_input_device()]
        elif forced:
            candidates: list[int | None] = [int(forced)]
        else:
            # Probe candidates and take the FIRST one that shows real signal
            # (existing-but-silent endpoints are common on Windows: a wireless
            # dongle with the headset off, a dead WASAPI channel, ...).
            candidates = []
            silent: list[int | None] = []
            for dev in _input_candidates():
                peak = _probe_device(dev)
                name = self._dev_name(dev)
                if peak > 0.0:
                    print(f"  mic probe: {name} — live (peak {peak:.4f})")
                    candidates.append(dev)
                    break
                print(f"  mic probe: {name} — "
                      f"{'silent' if peak == 0.0 else 'unavailable'}")
                if peak == 0.0:
                    silent.append(dev)
            # nothing live: fall back to silent-but-open devices anyway
            candidates.extend(silent)
            if not candidates:
                candidates = [None]
        last_err: Exception | None = None
        for dev in candidates:
            try:
                self._stream = sd.InputStream(
                    device=dev,
                    samplerate=SAMPLE_RATE,
                    channels=1,
                    dtype="float32",
                    blocksize=FRAME_SAMPLES,
                    callback=self._callback,
                )
                self._stream.start()
                self._device = dev
                global _active_mic
                _active_mic = self
                self.device_name = self._dev_name(dev)
                gain = (f"fixed x{self._fixed_gain}" if self._fixed_gain
                        else "auto")
                print(f"  mic device: {self.device_name} (gain: {gain}; "
                      f"MIC_DEVICE / MIC_GAIN to override)")
                return
            except Exception as e:
                last_err = e
                print(f"  mic open failed on {self._dev_name(dev)}: {e}")
                self._stream = None
        raise RuntimeError(f"no usable microphone ({last_err})")

    @staticmethod
    def _dev_name(dev: int | None) -> str:
        try:
            idx = dev if dev is not None else sd.default.device[0]
            return sd.query_devices(idx)["name"]
        except Exception:
            return "default"

    def stop(self) -> None:
        if self._stream is not None:
            self._stream.stop()
            self._stream.close()
            self._stream = None


class AECMicStream(_FrameQueue):
    """Echo-cancelled mic via helper/aecmic (Apple VoiceProcessingIO).

    Audio played by this Mac — including our own TTS — is subtracted
    from the mic signal in hardware-adjacent DSP, so barge-in detection
    never hears the assistant's own voice. Silence between utterances is
    gated to literal zeros by Apple's noise suppression; that's normal.
    """

    def __init__(self, loop: asyncio.AbstractEventLoop):
        super().__init__(loop)
        self._proc: subprocess.Popen | None = None
        self._thread: threading.Thread | None = None
        self._stopping = threading.Event()
        self._wake = threading.Event()  # pulsed on mute-state change
        # No boost by default. VoiceProcessingIO already AGC-normalizes
        # close speech to ~0.45 peak; any multiplier clips loud speech to
        # full-scale, which (a) makes the VAD never see end-of-utterance
        # silence and (b) distorts the audio so the transcriber returns
        # empty. Per-utterance normalization (in main.run) handles quiet
        # input instead, without clipping at capture.
        self._gain = float(os.environ.get("AEC_GAIN", "1.0"))

    @staticmethod
    def available() -> bool:
        return AEC_HELPER.is_file()

    def start(self) -> None:
        global _active_mic
        _active_mic = self
        if not _mic_muted.is_set():
            self._spawn()
        self._thread = threading.Thread(target=self._reader, daemon=True)
        self._thread.start()

    def _spawn(self) -> None:
        """Launch the aecmic helper (acquires the mic / VoiceProcessingIO)."""
        if self._proc is not None:
            return
        args = [str(AEC_HELPER)]
        if os.environ.get("AEC_BYPASS") == "1":  # debug: no cancellation
            args.append("--bypass")
        try:
            self._proc = subprocess.Popen(
                args, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                bufsize=0,
            )
        except Exception as e:  # noqa: BLE001
            print(f"  [aecmic] spawn failed: {e}")
            self._proc = None

    def _kill(self) -> None:
        """Terminate the helper — RELEASES the mic device so Bluetooth
        earbuds drop the call profile and go back to hi-fi output."""
        p, self._proc = self._proc, None
        if p is not None:
            try:
                p.terminate()
            except Exception:  # noqa: BLE001
                pass

    def _on_mute_change(self, muted: bool) -> None:
        # Kill the helper immediately on mute so the blocking read below
        # unblocks at once and the OS releases the mic. The reader respawns
        # on unmute. _wake pulses the reader out of its idle wait.
        if muted:
            self._kill()
        self._wake.set()

    def _reader(self) -> None:
        frame_bytes = FRAME_SAMPLES * 2  # int16 from the helper
        frame_dur = FRAME_SAMPLES / SAMPLE_RATE
        silence = np.zeros(FRAME_SAMPLES, dtype=np.float32)
        buf = b""
        while not self._stopping.is_set():
            if _mic_muted.is_set():
                # Device released; feed the pipeline silence at frame cadence
                # (VAD never triggers) without holding the mic open.
                self._kill()
                buf = b""
                self._deliver(silence)
                self._wake.wait(frame_dur)
                self._wake.clear()
                continue
            proc = self._proc
            if proc is None:                       # unmuted: re-acquire device
                self._spawn()
                buf = b""
                proc = self._proc
                if proc is None:
                    self._wake.wait(0.2)           # spawn failed; back off
                    continue
            try:
                data = proc.stdout.read(frame_bytes)
            except Exception:                      # noqa: BLE001 — killed mid-read
                self._kill()
                continue
            if not data:                           # helper exited (e.g. muted)
                self._kill()
                continue
            buf += data
            while len(buf) >= frame_bytes:
                pcm16 = np.frombuffer(buf[:frame_bytes], dtype=np.int16)
                buf = buf[frame_bytes:]
                if _mic_muted.is_set():
                    break
                frame = pcm16.astype(np.float32) * (self._gain / 32768.0)
                self._deliver(np.clip(frame, -1.0, 1.0))

    def stop(self) -> None:
        global _active_mic
        if _active_mic is self:
            _active_mic = None
        self._stopping.set()
        self._wake.set()
        self._kill()


class SystemAudioStream(_FrameQueue):
    """System-audio (loopback) capture via helper/sysaudio (ScreenCaptureKit).

    Taps what the Mac is PLAYING — the remote voices of an online meeting come
    out of the speakers — cleanly and at full quality. This is the "them" side
    of a meeting: the AEC mic deliberately erases speaker output, so the far end
    is only audible here. Emits the same 16 kHz int16 mono frames as AECMicStream.

    Deliberately simple: NOT part of the mute-hotkey logic (a meeting recording
    must not be muted by the mic hotkey) and no AGC (digital audio is already
    leveled) — just start/stop and clip to [-1,1]. On SCK failure (no
    Screen-Recording grant when spawned from a plain terminal) the helper exits
    with a reason on stderr and NO frames arrive; last_error surfaces it so the
    caller can tell Ahmed instead of silently recording his side only.
    """

    def __init__(self, loop: asyncio.AbstractEventLoop):
        super().__init__(loop)
        self._proc: subprocess.Popen | None = None
        self._thread: threading.Thread | None = None
        self._errthread: threading.Thread | None = None
        self._stopping = threading.Event()
        self.last_error = ""

    @staticmethod
    def available() -> bool:
        return SYSAUDIO_HELPER.is_file()

    def start(self) -> None:
        try:
            self._proc = subprocess.Popen(
                [str(SYSAUDIO_HELPER)], stdout=subprocess.PIPE,
                stderr=subprocess.PIPE, bufsize=0,
            )
        except Exception as e:  # noqa: BLE001
            self.last_error = str(e)
            print(f"  [sysaudio] spawn failed: {e}")
            self._proc = None
            return
        self._thread = threading.Thread(target=self._reader, daemon=True)
        self._thread.start()
        # Drain stderr so the helper's failure reason (e.g. permission) is
        # visible to the caller via last_error, and the pipe never blocks.
        self._errthread = threading.Thread(target=self._drain_err, daemon=True)
        self._errthread.start()

    def _drain_err(self) -> None:
        proc = self._proc
        if proc is None or proc.stderr is None:
            return
        for raw in iter(proc.stderr.readline, b""):
            line = raw.decode("utf-8", "replace").strip()
            if not line:
                continue
            # "sysaudio: running…" is the healthy line; anything else is a fault.
            if "running" not in line:
                self.last_error = line.replace("sysaudio:", "").strip()

    def _reader(self) -> None:
        frame_bytes = FRAME_SAMPLES * 2  # int16 from the helper
        proc = self._proc
        if proc is None or proc.stdout is None:
            return
        buf = b""
        while not self._stopping.is_set():
            try:
                data = proc.stdout.read(frame_bytes)
            except Exception:  # noqa: BLE001 — killed mid-read
                break
            if not data:  # helper exited (permission/error) or was stopped
                break
            buf += data
            while len(buf) >= frame_bytes:
                pcm16 = np.frombuffer(buf[:frame_bytes], dtype=np.int16)
                buf = buf[frame_bytes:]
                frame = pcm16.astype(np.float32) / 32768.0
                self._deliver(np.clip(frame, -1.0, 1.0))

    def stop(self) -> None:
        self._stopping.set()
        p, self._proc = self._proc, None
        if p is not None:
            try:
                p.terminate()
            except Exception:  # noqa: BLE001
                pass


class Speaker:
    """Plays queued PCM chunks; stop() silences within ~10 ms."""

    def __init__(self, samplerate: int):
        self.samplerate = samplerate
        self._buffer = np.zeros(0, dtype=np.float32)
        self._lock = threading.Lock()
        self._stream: sd.OutputStream | None = None
        self._active = False  # audio queued or still draining

    def _callback(self, outdata, frames, _time, _status):
        with self._lock:
            n = min(frames, len(self._buffer))
            outdata[:n, 0] = self._buffer[:n]
            outdata[n:, 0] = 0.0
            self._buffer = self._buffer[n:]
            if len(self._buffer) == 0:
                self._active = False

    def _open(self, device=None) -> None:
        self._stream = sd.OutputStream(
            samplerate=self.samplerate,
            channels=1,
            dtype="float32",
            blocksize=0,        # let CoreAudio pick a small block
            latency="low",
            device=device,
            callback=self._callback,
        )
        self._stream.start()

    def start(self) -> None:
        """Open the output stream. The default output can be a flaky Bluetooth
        device (e.g. disconnected CMF Buds) that fails with a PortAudio error —
        that used to CRASH the whole engine at startup. So on failure we switch
        the system default output to the built-in speakers and retry, and if
        even that fails we degrade to mute rather than die."""
        import sys
        try:
            self._open()
            return
        except Exception as e:  # noqa: BLE001
            print(f"  speaker: default output failed ({str(e)[:60]}) — "
                  "trying the built-in speakers")
        if sys.platform == "darwin":
            try:
                import shutil
                sas = (shutil.which("SwitchAudioSource")
                       or "/opt/homebrew/bin/SwitchAudioSource")
                if os.path.exists(sas):
                    allo = subprocess.run([sas, "-a", "-t", "output"],
                                          capture_output=True, text=True,
                                          timeout=5).stdout.splitlines()
                    builtin = next((o.strip() for o in allo
                                    if "macbook" in o.lower()
                                    or "built-in" in o.lower()), None)
                    if builtin:
                        subprocess.run([sas, "-t", "output", "-s", builtin],
                                       capture_output=True, timeout=5)
                        print(f"  speaker: switched output to '{builtin}'")
            except Exception:  # noqa: BLE001
                pass
        for attempt in range(3):
            try:
                self._open()
                return
            except Exception as e:  # noqa: BLE001
                if attempt == 2:
                    print(f"  speaker: could not open any output ({str(e)[:60]})"
                          " — Jarvis will be MUTE until an output device works")
                    self._stream = None
                    return
                import time as _t
                _t.sleep(0.5)

    def stop_stream(self) -> None:
        if self._stream is not None:
            self._stream.stop()
            self._stream.close()
            self._stream = None

    def play(self, pcm: np.ndarray) -> None:
        """Queue audio (float32 mono at self.samplerate)."""
        with self._lock:
            self._buffer = np.concatenate([self._buffer, pcm.astype(np.float32)])
            self._active = True

    def cut(self) -> None:
        """Barge-in: drop everything queued, silence immediately."""
        with self._lock:
            self._buffer = np.zeros(0, dtype=np.float32)
            self._active = False

    @property
    def speaking(self) -> bool:
        return self._active

    async def wait_done(self, poll: float = 0.05, timeout: float = 120.0) -> None:
        """Wait for playback to drain. The timeout is a fail-safe: if the
        output device wedges (never drains), cut instead of hanging the
        whole conversation loop in 'speaking' forever."""
        waited = 0.0
        while self.speaking:
            await asyncio.sleep(poll)
            waited += poll
            if waited >= timeout:
                self.cut()
                print("       [speaker: playback wedged — cut after timeout]")
                break