"""Continuous, channel-independent PCM controls conversation turns."""

import asyncio
import struct

import pytest

from voice_assistant.agent.mock import MockAgentAdapter
from voice_assistant.application.events import TurnCompleted, TurnOpened, TurnInterrupted, SpeechChunk
from voice_assistant.application.voice_session import VoiceSession, VoiceSessionError
from voice_assistant.domain.session import SessionState
from voice_assistant.stt.mock import MockSTTAdapter
from voice_assistant.tts.mock import MockTTSAdapter
from voice_assistant.turn_manager import TurnManager
from voice_assistant.vad.adapter import VADAdapter
from voice_assistant.vad.models import SpeechStarted, SpeechEnded
from voice_assistant.voice_pipeline import VoicePipeline


@pytest.fixture
def anyio_backend():
    return "asyncio"


class ScriptedVAD(VADAdapter):
    def __init__(self, script=(), tail=()):
        self.script = list(script)
        self.tail = tail
        self.received = []
        self.closed = False

    async def process(self, audio):
        self.received.append(audio)
        return self.script.pop(0) if self.script else ()

    async def flush(self):
        return self.tail

    async def reset(self):
        pass

    async def aclose(self):
        self.closed = True


class RecordingSTT(MockSTTAdapter):
    def __init__(self):
        super().__init__()
        self.received = []
        self.aborted = False

    async def send_audio(self, audio):
        await super().send_audio(audio)
        self.received.append(audio)

    async def stop_session(self):
        await self.emit_final("Olá")
        await super().stop_session()

    async def abort_session(self):
        self.aborted = True
        await super().abort_session()


def build(vad, factory=RecordingSTT, event_sink=None, **options):
    providers = []
    events = []
    completed = asyncio.Event()

    def create_stt():
        provider = factory()
        providers.append(provider)
        return provider

    async def emit(event):
        events.append(event)
        if event_sink is not None:
            await event_sink(event)
        if isinstance(event, TurnCompleted):
            completed.set()

    session = VoiceSession(
        pipeline=VoicePipeline(
            session=SessionState(), turn_manager=TurnManager(),
            agent=MockAgentAdapter(), tts=MockTTSAdapter(),
        ),
        create_stt=create_stt, emit=emit, vad=vad, **options,
    )
    return session, providers, events, completed


def pcm(value, samples=256):
    return struct.pack(f'<{samples}h', *([value] * samples))


@pytest.mark.anyio
async def test_delayed_start_replays_pre_roll_without_duplicating_audio():
    vad = ScriptedVAD([(), (), (SpeechStarted(400),), ()])
    session, providers, events, _ = build(vad, pre_roll_ms=10)
    chunks = [pcm(i) for i in [1, 2, 3, 4]]
    try:
        await session.receive_audio(chunks[0])
        await session.receive_audio(chunks[1])
        assert providers == []
        await session.receive_audio(chunks[2])
        await session.receive_audio(chunks[3])
        assert len(providers) == 1
        assert b''.join(providers[0].received) == b''.join(chunks)[240 * 2:]
        assert isinstance(events[0], TurnOpened)
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_speech_end_finalizes_in_background_while_silence_continues():
    vad = ScriptedVAD([(SpeechStarted(0),), (SpeechEnded(400, 'silence'),), ()])
    session, providers, events, completed = build(vad, pre_roll_ms=0)
    try:
        await session.receive_audio(pcm(1))
        await session.receive_audio(pcm(2))
        await session.receive_audio(pcm(3))
        await asyncio.wait_for(completed.wait(), timeout=1.0)
        assert b''.join(providers[0].received) == pcm(1) + pcm(2)[:144 * 2]
        assert isinstance(events[-1], TurnCompleted)
        assert session.active_turn_id is None
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_multiple_speech_intervals_in_one_chunk_keep_audio_separate():
    vad = ScriptedVAD([(
        SpeechStarted(0), SpeechEnded(160, 'silence'),
        SpeechStarted(240), SpeechEnded(400, 'silence'),
    )])
    session, providers, events, completed = build(vad, pre_roll_ms=0)
    audio = struct.pack('<512h', *range(512))
    try:
        await session.receive_audio(audio)
        await asyncio.wait_for(completed.wait(), timeout=1.0)
        assert len(providers) == 2
        assert b''.join(providers[0].received) == audio[:320]
        assert b''.join(providers[1].received) == audio[480:800]
        assert any(isinstance(event, TurnInterrupted) and event.turn_id == 1 for event in events)
        assert [event.turn_id for event in events if isinstance(event, TurnOpened)] == [1, 2]
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_new_speech_cancels_stt_finalization_and_opens_fresh_turn():
    started = asyncio.Event()
    cancelled = asyncio.Event()

    class WaitingSTT(RecordingSTT):
        async def stop_session(self):
            started.set()
            try:
                await asyncio.Event().wait()
            except asyncio.CancelledError:
                cancelled.set()
                raise

    vad = ScriptedVAD([
        (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), (SpeechStarted(600),),
    ])
    session, providers, events, _ = build(vad, factory=WaitingSTT, pre_roll_ms=0)
    try:
        await session.receive_audio(pcm(1))
        await session.receive_audio(pcm(2))
        await asyncio.wait_for(started.wait(), timeout=1.0)
        await session.receive_audio(pcm(3))
        assert cancelled.is_set()
        assert providers[0].aborted
        assert len(providers) == 2
        assert session.active_turn_id == 2
        assert b''.join(providers[1].received) == pcm(3)[88 * 2:]
        assert any(isinstance(event, TurnInterrupted) for event in events)
    finally:
        await session.aclose()


@pytest.mark.anyio
@pytest.mark.parametrize('audio', [b'', b'\x00', bytes(65538)])
async def test_invalid_input_does_not_reach_vad_or_advance_history(audio):
    vad = ScriptedVAD([(SpeechStarted(0),)])
    session, providers, _, _ = build(vad, pre_roll_ms=0)
    try:
        with pytest.raises(ValueError):
            await session.receive_audio(audio)
        assert vad.received == []
        await session.receive_audio(pcm(1))
        assert b''.join(providers[0].received) == pcm(1)
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_manual_commands_cannot_compete_with_vad_turn_control():
    session, _, _, _ = build(ScriptedVAD())
    try:
        for operation in [session.start_turn, lambda: session.send_audio(1, 0, pcm(1))]:
            with pytest.raises(VoiceSessionError) as error:
                await operation()
            assert error.value.code == 'TURN_CONTROLLED_BY_VAD'
        with pytest.raises(VoiceSessionError):
            session.stop_turn(1)
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_input_error_prevents_unsafe_retry_and_cleanup_closes_vad():
    class FailingVAD(ScriptedVAD):
        async def process(self, audio):
            raise RuntimeError('Inference failed')

    vad = FailingVAD()
    session, _, _, _ = build(vad)
    try:
        with pytest.raises(VoiceSessionError, match='Inference failed'):
            await session.receive_audio(pcm(1))
        with pytest.raises(VoiceSessionError, match='new session'):
            await session.receive_audio(pcm(1))
    finally:
        await session.aclose()
    assert vad.closed


@pytest.mark.anyio
async def test_speech_older_than_history_is_reported_instead_of_silently_truncated():
    vad = ScriptedVAD([(), (SpeechStarted(0),)])
    session, providers, _, _ = build(vad, pre_roll_ms=0, history_ms=1)
    try:
        await session.receive_audio(pcm(1))
        with pytest.raises(VoiceSessionError, match='increase history_ms'):
            await session.receive_audio(pcm(2))
        assert providers == []
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_normal_end_of_input_flushes_active_speech():
    vad = ScriptedVAD([(SpeechStarted(0),)], tail=(SpeechEnded(256, 'end_of_input'),))
    session, _, _, completed = build(vad, pre_roll_ms=0)
    try:
        await session.receive_audio(pcm(1))
        await session.finish_audio_input()
        await asyncio.wait_for(completed.wait(), timeout=1.0)
        with pytest.raises(VoiceSessionError, match='closed'):
            await session.receive_audio(pcm(2))
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_session_end_during_vad_processing_does_not_open_late_turn():
    started = asyncio.Event()
    release = asyncio.Event()

    class WaitingVAD(ScriptedVAD):
        async def process(self, audio):
            started.set()
            await release.wait()
            return (SpeechStarted(0),)

    vad = WaitingVAD()
    session, providers, events, _ = build(vad)
    receiving = asyncio.create_task(session.receive_audio(pcm(1)))
    ending = None
    try:
        await asyncio.wait_for(started.wait(), timeout=1.0)
        ending = asyncio.create_task(session.end())
        await asyncio.sleep(0)
        release.set()
        with pytest.raises(VoiceSessionError, match='shutdown'):
            await receiving
        await asyncio.wait_for(ending, timeout=1.0)
        assert session.closed
        assert providers == []
        assert events == []
    finally:
        release.set()
        await asyncio.gather(receiving, return_exceptions=True)
        if ending:
            await asyncio.gather(ending, return_exceptions=True)
        await session.aclose()


@pytest.mark.parametrize('options', [
    {'pre_roll_ms': -1}, {'pre_roll_ms': True}, {'history_ms': 0},
    {'history_ms': 100, 'pre_roll_ms': 200}, {'max_input_chunk_bytes': 0},
])
def test_invalid_audio_history_policy_is_rejected(options):
    with pytest.raises(ValueError):
        build(ScriptedVAD(), **options)


@pytest.mark.anyio
async def test_empty_stt_result_releases_turn_without_stopping_continuous_input():
    interrupted = asyncio.Event()

    async def sink(event):
        if isinstance(event, TurnInterrupted):
            interrupted.set()

    vad = ScriptedVAD([
        (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), (SpeechStarted(600),),
    ])
    session, providers, _, _ = build(vad, factory=MockSTTAdapter, event_sink=sink)
    try:
        await session.receive_audio(pcm(1))
        await session.receive_audio(pcm(2))
        await asyncio.wait_for(interrupted.wait(), timeout=1.0)
        assert session.active_turn_id is None
        await session.receive_audio(pcm(3))
        assert session.active_turn_id == 2
        assert len(providers) == 2
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_new_speech_cancels_response_delivery_before_opening_next_turn():
    delivering = asyncio.Event()
    cancelled = asyncio.Event()

    async def sink(event):
        if isinstance(event, SpeechChunk) and event.turn_id == 1:
            delivering.set()
            try:
                await asyncio.Event().wait()
            except asyncio.CancelledError:
                cancelled.set()
                raise

    vad = ScriptedVAD([
        (SpeechStarted(0),), (SpeechEnded(400, 'silence'),), (SpeechStarted(600),),
    ])
    session, providers, events, _ = build(vad, event_sink=sink, pre_roll_ms=0)
    try:
        await session.receive_audio(pcm(1))
        await session.receive_audio(pcm(2))
        await asyncio.wait_for(delivering.wait(), timeout=1.0)
        await session.receive_audio(pcm(3))
        assert cancelled.is_set()
        assert providers[0].aborted
        assert session.active_turn_id == 2
        opened = next(i for i, event in enumerate(events) if isinstance(event, TurnOpened) and event.turn_id == 2)
        assert not any(isinstance(event, SpeechChunk) and event.turn_id == 1 for event in events[opened:])
    finally:
        await session.aclose()


@pytest.mark.anyio
async def test_pre_roll_never_replays_previous_utterance_into_next_stt():
    vad = ScriptedVAD([(
        SpeechStarted(0), SpeechEnded(160, 'silence'),
        SpeechStarted(240), SpeechEnded(400, 'silence'),
    )])
    session, providers, _, completed = build(vad, pre_roll_ms=200)
    audio = struct.pack('<512h', *range(512))
    try:
        await session.receive_audio(audio)
        await asyncio.wait_for(completed.wait(), timeout=1.0)
        assert b''.join(providers[0].received) == audio[:320]
        assert b''.join(providers[1].received) == audio[320:800]
    finally:
        await session.aclose()