"""Meeting-mode tests — WAV recorder, multilingual STT selection/fallback,
transcript ordering, and the background transcription worker + backlog/finalize
logic. No real models, no audio devices, no Claude session (fakes throughout).

Run:  .venv/bin/python -m pytest test_meeting.py -q
"""
from __future__ import annotations

import asyncio
import os
import wave
from pathlib import Path

import numpy as np

os.environ["RESUME"] = "0"

import voice.meeting as vm
from voice.meeting import MeetingSession, WavRecorder

import main as appmod
from main import VoiceApp


# ---------- WAV recorder ----------

def test_wav_recorder_roundtrip(tmp_path):
    p = tmp_path / "clip.wav"
    rec = WavRecorder(p)
    a = np.sin(np.linspace(0, 20, 1600, dtype=np.float32)) * 0.5   # 0.1s
    b = np.full(400, 0.25, dtype=np.float32)
    rec.write(a)
    rec.write(b)
    assert rec.samples == 2000
    assert abs(rec.seconds - 2000 / 16000) < 1e-6
    rec.close()

    with wave.open(str(p), "rb") as w:
        assert w.getnchannels() == 1
        assert w.getsampwidth() == 2        # int16
        assert w.getframerate() == 16000
        assert w.getnframes() == 2000
        pcm = np.frombuffer(w.readframes(2000), dtype="<i2").astype(np.float32) / 32767.0
    assert np.allclose(pcm[:1600], a, atol=1e-3)
    assert np.allclose(pcm[1600:], b, atol=1e-3)


def test_wav_recorder_clips(tmp_path):
    p = tmp_path / "hot.wav"
    rec = WavRecorder(p)
    rec.write(np.array([2.0, -2.0, 0.0, 1.5], dtype=np.float32))  # out of range
    rec.close()
    with wave.open(str(p), "rb") as w:
        pcm = np.frombuffer(w.readframes(4), dtype="<i2")
    assert pcm[0] == 32767 and pcm[1] == -32767  # clipped to full scale, no wrap
    assert pcm[2] == 0


# ---------- multilingual STT model selection ----------

def test_meeting_stt_defaults():
    assert vm.MEETING_STT_MODEL == "mlx-community/whisper-large-v3-mlx"
    assert vm.MEETING_STT_LANG is None    # "" env -> auto-detect
    assert vm._STT_FALLBACKS[0] == vm.MEETING_STT_MODEL
    # degrades to turbo then a small multilingual model
    assert any("turbo" in r for r in vm._STT_FALLBACKS)
    assert any("small" in r for r in vm._STT_FALLBACKS)


def test_load_meeting_stt_fallback(monkeypatch):
    """large-v3 + turbo fail to load -> loader degrades to the small model."""
    attempts = []

    class FakeMeetingSTT:
        def __init__(self, repo):
            attempts.append(repo)
            if "small" not in repo:      # only the small model "loads"
                raise RuntimeError(f"{repo} not in cache")
            self.repo = repo

    monkeypatch.setattr(vm, "MeetingSTT", FakeMeetingSTT)
    stt, name = vm.load_meeting_stt()
    assert "small" in name
    assert attempts == [
        "mlx-community/whisper-large-v3-mlx",
        "mlx-community/whisper-large-v3-turbo",
        "mlx-community/whisper-small-mlx",
    ]


def test_load_meeting_stt_all_fail(monkeypatch):
    class Boom:
        def __init__(self, repo):
            raise RuntimeError("no model")
    monkeypatch.setattr(vm, "MeetingSTT", Boom)
    try:
        vm.load_meeting_stt()
        assert False, "should raise when every model fails"
    except RuntimeError:
        pass


# ---------- session: capture-time ordering + shared stem ----------

def test_meeting_session_ordering_and_paths():
    m = MeetingSession(mode="online", title="Q3 sync")
    m.add("them", "second", ts=100.0)
    m.add("Ahmed", "first", ts=50.0)     # added later, earlier capture time
    lines = m.full_text().splitlines()[2:]   # skip header + blank
    assert lines[0].startswith("Ahmed: first")   # sorted by capture ts
    assert lines[1].startswith("them: second")
    # transcript + both WAVs share one stem
    assert m.room_wav_path().name == f"{m.stem}_room.wav"
    assert m.them_wav_path().name == f"{m.stem}_them.wav"
    assert m.stem.startswith("meeting_")


# ---------- background transcription worker ----------

class FakeMeetingSTT:
    """Transcribe returns a label derived from the audio's constant fill so a
    test can prove WHICH utterance produced WHICH line."""
    def transcribe(self, audio):
        return f"said {int(round(float(audio[0]) * 100))}"


class FakeTTS:
    def __init__(self):
        self.said = []
    def synth(self, text):
        self.said.append(text)
        return np.zeros(8, dtype=np.int16)


class FakeSpeaker:
    def __init__(self):
        self.played = 0
        self.speaking = False
    def play(self, pcm):
        self.played += 1


def _app():
    app = VoiceApp()
    app.tts = FakeTTS()
    app.speaker = FakeSpeaker()
    return app


def test_worker_transcribes_labels_and_orders(monkeypatch):
    monkeypatch.setattr(vm, "load_meeting_stt", lambda: (FakeMeetingSTT(), "fake"))

    async def go():
        app = _app()
        session = MeetingSession(mode="online")
        q = asyncio.Queue(maxsize=10)
        app._meeting_stt_queue = q
        worker = asyncio.create_task(app._meeting_transcribe_worker(session, q))
        # enqueue OUT of capture order (later ts first) with distinct labels
        app._enqueue_meeting_audio("them", np.full(512, 0.90, np.float32), ts=200.0)
        app._enqueue_meeting_audio("Ahmed", np.full(512, 0.10, np.float32), ts=100.0)
        q.put_nowait(None)                        # end-of-input sentinel
        await asyncio.wait_for(worker, timeout=5)
        assert worker.done()
        body = session.full_text().splitlines()[2:]
        # ts-sorted: Ahmed(100) before them(200); text carries the right audio
        assert body[0] == "Ahmed: said 10"
        assert body[1] == "them: said 90"
    asyncio.run(go())


def test_enqueue_backlog_never_raises_and_is_wav_safe():
    async def go():
        app = _app()
        app._meeting_stt_queue = asyncio.Queue(maxsize=3)
        for _ in range(10):   # overflow the live queue — must NOT raise/block
            app._enqueue_meeting_audio("them", np.full(512, 0.2, np.float32), ts=1.0)
        assert app._meeting_stt_queue.qsize() == 3   # capped; rest rely on the WAV
    asyncio.run(go())


# ---------- stop: live-vs-batch decision ----------

def _prep_stopping_app():
    app = _app()
    app.meeting = MeetingSession(mode="in_person")
    app._room_wav = None
    app._them_wav = None
    app._remote_capture_task = None
    app._sysaudio = None
    app._meeting_worker_task = None      # finalize is stubbed; no real worker
    captured = {}
    async def fake_finalize(m, worker, was_backlog):
        captured["was_backlog"] = was_backlog
    app._finalize_meeting = fake_finalize
    return app, captured


def test_stop_no_backlog_says_stopped():
    async def go():
        app, captured = _prep_stopping_app()
        app._meeting_stt_queue = asyncio.Queue(maxsize=120)   # empty
        await app._stop_meeting()
        await asyncio.sleep(0.01)   # let the background finalize task run
        assert captured["was_backlog"] is False
        assert any("Stopped" in s for s in app.tts.said)
        assert not any("processing" in s.lower() for s in app.tts.said)
    asyncio.run(go())


def test_stop_with_backlog_says_still_processing():
    async def go():
        app, captured = _prep_stopping_app()
        q = asyncio.Queue(maxsize=120)
        for _ in range(4):    # pending >= MEETING_BACKLOG_THRESHOLD
            q.put_nowait(("them", np.zeros(512, np.float32), 1.0))
        app._meeting_stt_queue = q
        await app._stop_meeting()
        await asyncio.sleep(0.01)   # let the background finalize task run
        assert captured["was_backlog"] is True
        assert any("processing" in s.lower() for s in app.tts.said)
        # a sentinel was appended so the (real) worker would drain then exit
        drained = [q.get_nowait() for _ in range(q.qsize())]
        assert None in drained
    asyncio.run(go())


def test_finalize_saves_and_notifies_inbox(monkeypatch):
    async def go():
        app = _app()
        m = MeetingSession(mode="in_person", title="finalize test")
        m.add("them", "hello", ts=1.0)
        saved = {}
        monkeypatch.setattr(m, "save", lambda: saved.setdefault("p", "x.md") or "x.md")
        async def no_summary(mm):
            saved["summarized"] = True
        app._summarize_meeting = no_summary
        await app._finalize_meeting(m, None, was_backlog=True)
        assert saved.get("summarized") is True
        # readiness announced through the proactive inbox rail
        assert not app.inbox.empty()
        msg = app.inbox.get_nowait()
        assert "ready" in msg.lower()
    asyncio.run(go())