import asyncio
import time
import uuid

from services.ai_voice_runtime_service.audiosocket import EnergyVAD
from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaActor, MediaRegistration
from services.ai_voice_runtime_service.providers.asr import (
    ASRProvider,
    ASRTranscription,
    StreamingASRPartial,
    StreamingASRProvider,
    StreamingASRUnavailable,
)
from services.ai_voice_runtime_service.providers.tts import TTSSynthesis, TTSProvider
from services.shared.models import VoiceAITurnDecisionOut


class _StubTTSProvider(TTSProvider):
    name = "stub-tts"

    def synthesize(
        self,
        text: str,
        *,
        language: str | None = None,
        style_hints: dict[str, object] | None = None,
    ) -> TTSSynthesis:
        del language, style_hints
        assert text
        return TTSSynthesis(
            text=text,
            audio_bytes=(b"\x10\x00" * 960),
            sample_rate_hz=24000,
        )


def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr_stable():
    reply_starts: list[tuple[str, str | None, float]] = []
    turns: list[tuple[str, str]] = []

    class _BatchASRProvider(ASRProvider):
        name = "batch-asr"

        def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
            raise AssertionError("batch ASR should not be used when streaming sidecar is active")

    class _StreamingProvider(StreamingASRProvider):
        name = "streaming-sidecar"
        supports_streaming = True

        def __init__(self) -> None:
            self.events: list[tuple[str, float]] = []
            self.chunk_count = 0

        def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
            assert session_id
            assert language_hint == "ru"
            self.events.append(("open", time.monotonic()))
            return "stream-1"

        def push_pcm(self, stream_id: str, pcm_bytes: bytes) -> None:
            assert stream_id == "stream-1"
            assert pcm_bytes
            self.chunk_count += 1
            self.events.append(("push", time.monotonic()))

        def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
            assert stream_id == "stream-1"
            if self.chunk_count < 2:
                return None
            self.events.append(("partial", time.monotonic()))
            return StreamingASRPartial(
                text="need work schedule",
                language="ru",
                confidence=0.84,
                is_final=False,
                is_stable=True,
            )

        def finalize(self, stream_id: str) -> ASRTranscription:
            assert stream_id == "stream-1"
            self.events.append(("finalize", time.monotonic()))
            return ASRTranscription(text="need work schedule", language="ru", confidence=0.91)

        def close_stream(self, stream_id: str) -> None:
            assert stream_id == "stream-1"
            self.events.append(("close", time.monotonic()))

    streaming_provider = _StreamingProvider()
    registration = MediaRegistration(
        voice_session_id="avs_voice_v2_streaming",
        call_id="call_voice_v2_streaming",
        interaction_id="int_voice_v2_streaming",
        ai_session_id="ais_voice_v2_streaming",
        language="ru",
        media_uuid=str(uuid.uuid4()),
        queue_code="voice_lab_ai",
        queue_id="que_voice_lab_ai",
        agent_profile="voice_support",
        voice_v2_enabled=True,
        voice_v2_ack_mode="immediate_short",
        voice_v2_streaming_tts=True,
        voice_v2_partial_asr=True,
        voice_v2_duplex=True,
        voice_v2_streaming_asr_backend="local_sidecar",
        voice_v2_prebaked_ack=False,
    )

    runtime = AudioSocketMediaRuntime(
        enabled=True,
        host="127.0.0.1",
        port=0,
        frame_ms=20,
        idle_timeout_seconds=2.0,
        registration_wait_timeout_seconds=0.5,
        min_speech_ms=40,
        trailing_silence_ms=40,
        max_turn_ms=2000,
        asr_provider=_BatchASRProvider(),
        streaming_asr_provider=streaming_provider,
        tts_provider=_StubTTSProvider(),
        load_registration_by_media_uuid=lambda value: None,
        mark_media_connected=lambda session_id, value: None,
        mark_media_ended=lambda session_id, reason: None,
        touch_media_frame=lambda session_id: None,
        set_state=lambda session_id, state, handoff_reason, metadata: None,
        get_pending_greeting=lambda session_id: None,
        mark_reply_started=lambda session_id, text, is_greeting, reply_phase=None: None,
        mark_reply_delivered=lambda session_id, text, is_greeting, reply_phase=None: None,
        mark_reply_discarded=lambda session_id, text, reply_phase, reason: None,
        plan_reply=lambda session_id, text, metadata, kind: None,
        record_latency=lambda session_id, metric, latency_ms: None,
        process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
            turns.append((str((metadata or {}).get("reply_phase") or "final"), transcript_text))
            or VoiceAITurnDecisionOut(
                language=language or "ru",
                intent="schedule",
                reply_text="Please name the city.",
                confidence=0.92,
                needs_handoff=False,
                handoff_reason=None,
                case_action="keep_open",
                kb_refs=[],
                summary_text="reply ready",
                model="stub-voice",
                latency_ms=1,
                status="active",
            )
        ),
        request_handoff=lambda session_id, customer_request_text, decision: None,
        handle_media_error=lambda session_id, message, metadata: None,
    )

    async def _fake_speak_reply(
        actor: MediaActor,
        text: str,
        *,
        is_greeting: bool,
        style_hints: dict[str, object] | None = None,
        reply_phase: str | None = "main",
    ) -> None:
        del actor, is_greeting, style_hints
        reply_starts.append((text, reply_phase, time.monotonic()))

    runtime._speak_reply = _fake_speak_reply  # type: ignore[method-assign]

    async def _scenario() -> None:
        actor = MediaActor(
            registration=registration,
            reader=asyncio.StreamReader(),
            writer=None,  # type: ignore[arg-type]
            vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000),
            frame_ms=20,
            frame_bytes=320,
        )
        runtime._reset_live_turn_state(actor)
        await runtime._ensure_streaming_asr(actor)
        assert actor.asr_stream_id == "stream-1"

        pcm_chunk = (1000).to_bytes(2, "little", signed=True) * 160
        await asyncio.to_thread(streaming_provider.push_pcm, actor.asr_stream_id, pcm_chunk)
        await asyncio.to_thread(streaming_provider.push_pcm, actor.asr_stream_id, pcm_chunk)
        actor.asr_poll_due_monotonic = 0.0
        await runtime._poll_streaming_partial(actor)

        actor.speech_started_monotonic = time.monotonic()
        actor.speech_ended_monotonic = actor.speech_started_monotonic
        await runtime._process_utterance(actor, pcm_chunk * 14, False)

    asyncio.run(_scenario())

    phases = [phase for _, phase, _ in reply_starts]
    finalize_started_at = next(ts for name, ts in streaming_provider.events if name == "finalize")
    assert [text for phase, text in turns if phase == "final"] == ["need work schedule"]
    assert "open" in [name for name, _ in streaming_provider.events]
    assert "close" in [name for name, _ in streaming_provider.events]
    assert phases[:2] == ["ack", "main"]
    assert reply_starts[0][2] <= finalize_started_at
    assert any(text == "Please name the city." and phase == "main" for text, phase, _ in reply_starts)


def test_media_runtime_voice_v2_falls_back_when_streaming_sidecar_is_unavailable_stable():
    turns: list[str] = []

    class _BatchASRProvider(ASRProvider):
        name = "batch-asr"

        def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
            assert audio_bytes
            return ASRTranscription(text="need operator", language=language_hint or "ru", confidence=0.88)

    class _UnavailableStreamingProvider(StreamingASRProvider):
        name = "missing-sidecar"
        supports_streaming = True

        def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
            del session_id, language_hint
            raise StreamingASRUnavailable("sidecar down")

    registration = MediaRegistration(
        voice_session_id="avs_voice_v2_fallback",
        call_id="call_voice_v2_fallback",
        interaction_id="int_voice_v2_fallback",
        ai_session_id="ais_voice_v2_fallback",
        language="ru",
        media_uuid=str(uuid.uuid4()),
        queue_code="voice_lab_ai",
        queue_id="que_voice_lab_ai",
        agent_profile="voice_support",
        voice_v2_enabled=True,
        voice_v2_ack_mode="immediate_short",
        voice_v2_streaming_tts=True,
        voice_v2_partial_asr=True,
        voice_v2_duplex=True,
        voice_v2_streaming_asr_backend="local_sidecar",
        voice_v2_prebaked_ack=False,
    )

    runtime = AudioSocketMediaRuntime(
        enabled=True,
        host="127.0.0.1",
        port=0,
        frame_ms=20,
        idle_timeout_seconds=2.0,
        registration_wait_timeout_seconds=0.5,
        min_speech_ms=40,
        trailing_silence_ms=40,
        max_turn_ms=2000,
        asr_provider=_BatchASRProvider(),
        streaming_asr_provider=_UnavailableStreamingProvider(),
        tts_provider=_StubTTSProvider(),
        load_registration_by_media_uuid=lambda value: None,
        mark_media_connected=lambda session_id, value: None,
        mark_media_ended=lambda session_id, reason: None,
        touch_media_frame=lambda session_id: None,
        set_state=lambda session_id, state, handoff_reason, metadata: None,
        get_pending_greeting=lambda session_id: None,
        mark_reply_started=lambda session_id, text, is_greeting, reply_phase=None: None,
        mark_reply_delivered=lambda session_id, text, is_greeting, reply_phase=None: None,
        mark_reply_discarded=lambda session_id, text, reply_phase, reason: None,
        plan_reply=lambda session_id, text, metadata, kind: None,
        record_latency=lambda session_id, metric, latency_ms: None,
        process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
            turns.append(transcript_text)
            or VoiceAITurnDecisionOut(
                language=language or "ru",
                intent="handoff",
                reply_text="Connecting you to an operator.",
                confidence=0.9,
                needs_handoff=False,
                handoff_reason=None,
                case_action="keep_open",
                kb_refs=[],
                summary_text="reply ready",
                model="stub-voice",
                latency_ms=1,
                status="active",
            )
        ),
        request_handoff=lambda session_id, customer_request_text, decision: None,
        handle_media_error=lambda session_id, message, metadata: None,
    )

    async def _fake_speak_reply(
        actor: MediaActor,
        text: str,
        *,
        is_greeting: bool,
        style_hints: dict[str, object] | None = None,
        reply_phase: str | None = "main",
    ) -> None:
        del actor, text, is_greeting, style_hints, reply_phase

    runtime._speak_reply = _fake_speak_reply  # type: ignore[method-assign]

    async def _scenario() -> None:
        actor = MediaActor(
            registration=registration,
            reader=asyncio.StreamReader(),
            writer=None,  # type: ignore[arg-type]
            vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000),
            frame_ms=20,
            frame_bytes=320,
        )
        runtime._reset_live_turn_state(actor)
        await runtime._ensure_streaming_asr(actor)
        assert actor.asr_streaming_enabled is False

        pcm_chunk = (1000).to_bytes(2, "little", signed=True) * 160
        actor.speech_started_monotonic = time.monotonic()
        actor.speech_ended_monotonic = actor.speech_started_monotonic
        await runtime._process_utterance(actor, pcm_chunk * 4, False)

    asyncio.run(_scenario())

    assert turns == ["need operator"]
    assert registration.voice_v2_duplex is True
    assert registration.voice_v2_partial_asr is True
