from __future__ import annotations

import base64
import io
import json
import logging
import os
import queue
import threading
import time
import urllib.parse
import uuid
import wave
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path

import httpx

from services.ai_voice_runtime_service.audiosocket import (
    pcm16le_to_wav_bytes,
    resample_pcm16le,
    resample_pcm16le_stateful,
)
from services.shared.audioop_compat import audioop
from services.shared.security import issue_app_token

logger = logging.getLogger("uvicorn.error")


def _api_base() -> str:
    return (os.getenv("AI_API_BASE", "https://api.openai.com/v1").strip() or "https://api.openai.com/v1").rstrip("/")


def _api_key() -> str:
    return os.getenv("AI_API_KEY", "").strip()


def _timeout_seconds() -> float:
    raw = os.getenv("AI_TIMEOUT_SECONDS", "20").strip()
    try:
        value = float(raw)
    except ValueError:
        value = 20.0
    return max(value, 3.0)


def _openai_asr_model() -> str:
    return os.getenv("AI_VOICE_ASR_MODEL", "gpt-4o-mini-transcribe").strip() or "gpt-4o-mini-transcribe"


def _elevenlabs_asr_api_base() -> str:
    return (
        os.getenv("AI_VOICE_ASR_ELEVENLABS_API_BASE", "https://api.elevenlabs.io").strip()
        or "https://api.elevenlabs.io"
    ).rstrip("/")


def _elevenlabs_asr_api_key() -> str:
    return (
        os.getenv("AI_VOICE_ASR_ELEVENLABS_API_KEY", "").strip()
        or os.getenv("AI_VOICE_TTS_ELEVENLABS_API_KEY", "").strip()
    )


def _elevenlabs_asr_model_id() -> str:
    return os.getenv("AI_VOICE_ASR_ELEVENLABS_MODEL_ID", "scribe_v1").strip() or "scribe_v1"


def _elevenlabs_asr_default_language() -> str:
    return os.getenv("AI_VOICE_ASR_ELEVENLABS_LANGUAGE", "ru").strip() or "ru"


def _normalize_elevenlabs_asr_language(language: str | None) -> str:
    raw = str(language or "").strip().lower().replace("_", "-")
    if not raw:
        raw = _elevenlabs_asr_default_language().strip().lower().replace("_", "-")
    mapping = {
        "ru": "rus",
        "ru-ru": "rus",
        "kk": "kaz",
        "kz": "kaz",
        "kk-kz": "kaz",
        "kz-kz": "kaz",
        "en": "eng",
        "en-us": "eng",
        "en-gb": "eng",
    }
    return mapping.get(raw, raw or "rus")


def _elevenlabs_asr_sample_rate_hz() -> int:
    raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_SAMPLE_RATE_HZ", "16000").strip()
    try:
        value = int(raw)
    except ValueError:
        value = 16000
    if value < 8000 or value > 48000:
        return 16000
    return value


def _elevenlabs_realtime_model_id() -> str:
    return (
        os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_MODEL_ID", "scribe_v2_realtime").strip()
        or "scribe_v2_realtime"
    )


def _elevenlabs_realtime_audio_format() -> str:
    return (
        os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_AUDIO_FORMAT", "pcm_16000").strip().lower()
        or "pcm_16000"
    )


def _elevenlabs_realtime_sample_rate_hz(audio_format: str | None = None) -> int:
    explicit = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_SAMPLE_RATE_HZ", "").strip()
    if explicit:
        try:
            value = int(explicit)
        except ValueError:
            value = 16000
        return value if 8000 <= value <= 48000 else 16000

    normalized = str(audio_format or _elevenlabs_realtime_audio_format()).strip().lower()
    if normalized.startswith("pcm_"):
        try:
            value = int(normalized.rsplit("_", 1)[-1])
        except ValueError:
            value = 16000
        return value if 8000 <= value <= 48000 else 16000
    return 16000


def _elevenlabs_realtime_commit_strategy() -> str:
    raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_COMMIT_STRATEGY", "manual").strip().lower()
    return raw if raw in {"manual", "vad"} else "manual"


def _elevenlabs_realtime_debug_dump_audio_enabled() -> bool:
    raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_AUDIO", "").strip().lower()
    return raw in {"1", "true", "yes", "on"}


def _elevenlabs_realtime_debug_dump_audio_dir() -> Path:
    explicit = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_DEBUG_DUMP_DIR", "").strip()
    if explicit:
        return Path(explicit)
    return Path(os.getenv("CC_DATA_DIR", ".data_local").strip() or ".data_local") / "asr_debug_dumps"


def _elevenlabs_realtime_finalize_timeout_seconds() -> float:
    raw_ms = os.getenv("AI_VOICE_V2_STREAMING_FINAL_HARD_TIMEOUT_MS", "").strip()
    if raw_ms:
        try:
            return max(float(raw_ms) / 1000.0, 0.25)
        except ValueError:
            pass
    raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS", "").strip()
    if not raw:
        return 0.7
    try:
        value = float(raw)
    except ValueError:
        value = 0.7
    return max(value, 0.25)


def _normalize_elevenlabs_realtime_language(language: str | None) -> str:
    raw = str(language or "").strip().lower().replace("_", "-")
    if not raw:
        raw = _elevenlabs_asr_default_language().strip().lower().replace("_", "-")
    mapping = {
        "rus": "ru",
        "ru-ru": "ru",
        "kaz": "kk",
        "kz": "kk",
        "kk-kz": "kk",
        "kz-kz": "kk",
        "eng": "en",
        "en-us": "en",
        "en-gb": "en",
    }
    return mapping.get(raw, raw or "ru")


def _resolve_elevenlabs_asr_language(language_code: str | None, language_hint: str | None) -> str | None:
    normalized = str(language_code or "").strip().lower()
    if normalized in {"rus", "ru", "ru-ru"}:
        return "ru"
    if normalized in {"kaz", "kk", "kz", "kk-kz", "kz-kz"}:
        return "kz"
    resolved = str(language_code or language_hint or "").strip()
    return resolved or language_hint


def _yandex_asr_api_base() -> str:
    explicit = os.getenv("AI_VOICE_ASR_YANDEX_API_BASE", "").strip()
    if explicit:
        normalized = explicit.rstrip("/")
        if "stt.api.ml.yandexcloud.kz" in normalized.lower():
            return "https://stt.api.cloud.yandex.net"
        return normalized
    return "https://stt.api.cloud.yandex.net"


def _yandex_asr_operations_base() -> str:
    explicit = os.getenv("AI_VOICE_ASR_YANDEX_OPERATIONS_BASE", "").strip()
    if explicit:
        return explicit.rstrip("/")
    return "https://operation.api.cloud.yandex.net"


def _yandex_asr_api_key() -> str:
    return (
        os.getenv("AI_VOICE_ASR_YANDEX_API_KEY", "").strip()
        or os.getenv("AI_VOICE_TTS_YANDEX_API_KEY", "").strip()
    )


def _yandex_asr_iam_token() -> str:
    return (
        os.getenv("AI_VOICE_ASR_YANDEX_IAM_TOKEN", "").strip()
        or os.getenv("AI_VOICE_TTS_YANDEX_IAM_TOKEN", "").strip()
    )


def _yandex_asr_folder_id() -> str:
    return (
        os.getenv("AI_VOICE_ASR_YANDEX_FOLDER_ID", "").strip()
        or os.getenv("AI_VOICE_TTS_YANDEX_FOLDER_ID", "").strip()
    )


def _yandex_asr_timeout_seconds() -> float:
    raw = os.getenv("AI_VOICE_ASR_YANDEX_TIMEOUT_SECONDS", "").strip()
    if not raw:
        return _timeout_seconds()
    try:
        value = float(raw)
    except ValueError:
        value = _timeout_seconds()
    return max(value, 3.0)


def _yandex_asr_sample_rate_hz() -> int:
    raw = os.getenv("AI_VOICE_ASR_YANDEX_SAMPLE_RATE_HZ", "8000").strip()
    try:
        value = int(raw)
    except ValueError:
        value = 8000
    if value < 8000 or value > 48000:
        return 8000
    return value


def _yandex_asr_topic() -> str:
    return os.getenv("AI_VOICE_ASR_YANDEX_TOPIC", "general").strip() or "general"


def _yandex_asr_api_version(api_base: str) -> str:
    raw = os.getenv("AI_VOICE_ASR_YANDEX_API_VERSION", "auto").strip().lower() or "auto"
    if raw in {"v1", "v1_sync", "sync"}:
        return "v1"
    if raw in {"v3", "v3_async", "async"}:
        return "v3"
    normalized_base = str(api_base or "").lower()
    if "stt.api.cloud.yandex.net" in normalized_base or "stt.api.ml.yandexcloud.kz" in normalized_base:
        return "v3"
    return "v1"


def _yandex_asr_poll_interval_seconds() -> float:
    raw = os.getenv("AI_VOICE_ASR_YANDEX_POLL_INTERVAL_SECONDS", "0.25").strip()
    try:
        value = float(raw)
    except ValueError:
        value = 0.25
    return max(min(value, 2.0), 0.05)


def _yandex_asr_grpc_target(api_base: str) -> str:
    explicit = os.getenv("AI_VOICE_ASR_YANDEX_GRPC_TARGET", "").strip()
    if explicit:
        if "stt.api.ml.yandexcloud.kz" in explicit.lower():
            return "stt.api.cloud.yandex.net:443"
        return explicit
    return "stt.api.cloud.yandex.net:443"


def _yandex_asr_stream_queue_max_chunks() -> int:
    raw = os.getenv("AI_VOICE_ASR_YANDEX_STREAM_QUEUE_MAX_CHUNKS", "300").strip()
    try:
        value = int(raw)
    except ValueError:
        value = 300
    return max(min(value, 2000), 10)


def _yandex_asr_eou_max_pause_ms() -> int:
    raw = os.getenv("AI_VOICE_ASR_YANDEX_EOU_MAX_PAUSE_MS", "450").strip()
    try:
        value = int(raw)
    except ValueError:
        value = 450
    return max(min(value, 3000), 100)


def _yandex_asr_default_language() -> str:
    return os.getenv("AI_VOICE_ASR_YANDEX_LANGUAGE", "ru-RU").strip() or "ru-RU"


def _normalize_yandex_asr_language(language: str | None) -> str:
    raw = str(language or "").strip()
    if not raw:
        return _yandex_asr_default_language()
    lowered = raw.replace("_", "-").lower()
    if lowered in {"ru", "ru-ru"}:
        return "ru-RU"
    if lowered in {"kk", "kz", "kk-kz", "kz-kz"}:
        return "kk-KZ"
    return raw


def _streaming_asr_api_base() -> str:
    return (
        os.getenv("AI_VOICE_V2_STREAMING_ASR_BASE_URL", "http://127.0.0.1:8021").strip()
        or "http://127.0.0.1:8021"
    ).rstrip("/")


def _streaming_asr_timeout_seconds() -> float:
    raw = os.getenv("AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS", "4").strip()
    try:
        value = float(raw)
    except ValueError:
        value = 4.0
    return max(value, 0.25)


def _service_headers() -> dict[str, str]:
    token = issue_app_token(
        subject="svc:ai-voice-runtime",
        username="ai-voice-runtime",
        role="admin",
        auth_source="service",
        provider="ai-voice-runtime",
        ttl_seconds=300,
    )
    return {"Authorization": f"Bearer {token}"}


@dataclass(slots=True)
class ASRTranscription:
    text: str
    language: str | None = None
    confidence: float | None = None


@dataclass(slots=True)
class StreamingASRPartial:
    text: str
    language: str | None = None
    confidence: float | None = None
    is_final: bool = False
    is_stable: bool = False


class StreamingASRUnavailable(RuntimeError):
    pass


class ASRProvider:
    name = "stub"

    def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        del audio_bytes
        return ASRTranscription(text="", language=language_hint, confidence=None)

    def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        return self.transcribe(audio_bytes, language_hint=language_hint)


class StreamingASRProvider:
    name = "streaming-stub"
    supports_streaming = False

    def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
        del session_id, language_hint
        raise StreamingASRUnavailable("Streaming ASR backend is not configured")

    def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None:
        del stream_id, pcm_8k_chunk
        raise StreamingASRUnavailable("Streaming ASR backend is not configured")

    def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
        del stream_id
        return None

    def finalize(self, stream_id: str) -> ASRTranscription:
        del stream_id
        raise StreamingASRUnavailable("Streaming ASR backend is not configured")

    def close_stream(self, stream_id: str) -> None:
        del stream_id


class OpenAIASRProvider(ASRProvider):
    name = "openai"

    def __init__(self) -> None:
        self._api_base = _api_base()
        self._api_key = _api_key()
        self._timeout_seconds = _timeout_seconds()
        self._model = _openai_asr_model()

    def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        if not audio_bytes:
            return ASRTranscription(text="", language=language_hint, confidence=None)
        if not self._api_key:
            raise RuntimeError("AI_API_KEY is required for OpenAI ASR")

        data: dict[str, str] = {"model": self._model}
        language = str(language_hint or "").strip()
        if language:
            data["language"] = language

        with httpx.Client(timeout=self._timeout_seconds) as client:
            response = client.post(
                f"{self._api_base}/audio/transcriptions",
                headers={"Authorization": f"Bearer {self._api_key}"},
                data=data,
                files={"file": ("turn.wav", audio_bytes, "audio/wav")},
            )
        response.raise_for_status()
        payload = response.json()
        return ASRTranscription(
            text=str(payload.get("text") or "").strip(),
            language=str(payload.get("language") or language_hint or "").strip() or language_hint,
            confidence=None,
        )

    def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        return self.transcribe(audio_bytes, language_hint=language_hint)


class ElevenLabsASRProvider(ASRProvider):
    name = "elevenlabs"

    def __init__(
        self,
        *,
        api_base: str | None = None,
        api_key: str | None = None,
        timeout_seconds: float | None = None,
        model_id: str | None = None,
        sample_rate_hz: int | None = None,
    ) -> None:
        self._api_base = str(api_base or _elevenlabs_asr_api_base()).strip().rstrip("/")
        self._api_key = str(api_key if api_key is not None else _elevenlabs_asr_api_key()).strip()
        self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 3.0)
        self._model_id = str(model_id or _elevenlabs_asr_model_id()).strip() or _elevenlabs_asr_model_id()
        self._sample_rate_hz = int(sample_rate_hz if sample_rate_hz is not None else _elevenlabs_asr_sample_rate_hz())

    @staticmethod
    def _pcm_from_audio_bytes(audio_bytes: bytes, *, default_sample_rate_hz: int) -> tuple[bytes, int]:
        if not audio_bytes:
            return b"", default_sample_rate_hz
        try:
            with wave.open(io.BytesIO(audio_bytes), "rb") as wav_file:
                sample_width = wav_file.getsampwidth()
                channels = wav_file.getnchannels()
                sample_rate = wav_file.getframerate()
                pcm_bytes = wav_file.readframes(wav_file.getnframes())
                if sample_width != 2:
                    return audio_bytes, default_sample_rate_hz
                if channels == 2:
                    pcm_bytes = audioop.tomono(pcm_bytes, sample_width, 0.5, 0.5)  # type: ignore[name-defined]
                elif channels != 1:
                    return audio_bytes, default_sample_rate_hz
                return pcm_bytes, int(sample_rate or default_sample_rate_hz)
        except (wave.Error, EOFError):
            return audio_bytes, default_sample_rate_hz

    def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        if not audio_bytes:
            return ASRTranscription(text="", language=language_hint, confidence=None)
        if not self._api_key:
            raise RuntimeError(
                "AI_VOICE_ASR_ELEVENLABS_API_KEY or AI_VOICE_TTS_ELEVENLABS_API_KEY is required for ElevenLabs ASR"
            )

        pcm_bytes, input_sample_rate_hz = self._pcm_from_audio_bytes(
            audio_bytes,
            default_sample_rate_hz=self._sample_rate_hz,
        )
        if not pcm_bytes:
            return ASRTranscription(text="", language=language_hint, confidence=None)
        normalized_sample_rate_hz = self._sample_rate_hz
        if input_sample_rate_hz != normalized_sample_rate_hz:
            pcm_bytes = resample_pcm16le(
                pcm_bytes,
                input_rate_hz=input_sample_rate_hz,
                output_rate_hz=normalized_sample_rate_hz,
            )
        wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=normalized_sample_rate_hz)
        language_code = _normalize_elevenlabs_asr_language(language_hint)
        with httpx.Client(timeout=self._timeout_seconds) as client:
            response = client.post(
                f"{self._api_base}/v1/speech-to-text",
                headers={"xi-api-key": self._api_key},
                data={
                    "model_id": self._model_id,
                    "language_code": language_code,
                },
                files={"file": ("turn.wav", wav_bytes, "audio/wav")},
            )
        response.raise_for_status()
        payload = response.json()
        transcript_text = str(payload.get("text") or payload.get("transcript") or "").strip()
        resolved_language = _resolve_elevenlabs_asr_language(payload.get("language_code"), language_hint)
        return ASRTranscription(
            text=transcript_text,
            language=resolved_language,
            confidence=None,
        )

    def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        return self.transcribe(audio_bytes, language_hint=language_hint)


@dataclass
class _ElevenLabsRealtimeStreamState:
    stream_id: str
    session_id: str
    language: str
    websocket: object
    updates_queue: queue.Queue[StreamingASRPartial]
    thread: threading.Thread | None = None
    latest_partial: StreamingASRPartial | None = None
    stable_partial: StreamingASRPartial | None = None
    last_partial_text: str = ""
    partial_streak: int = 0
    final_transcription: ASRTranscription | None = None
    error: BaseException | None = None
    close_requested: bool = False
    final_event: threading.Event = field(default_factory=threading.Event)
    lock: threading.Lock = field(default_factory=threading.Lock)
    send_lock: threading.Lock = field(default_factory=threading.Lock)
    resample_state: object | None = None
    resample_lock: threading.Lock = field(default_factory=threading.Lock)
    committed_segments: list[str] = field(default_factory=list)
    finalize_requested: bool = False
    debug_pcm: bytearray = field(default_factory=bytearray)
    debug_lock: threading.Lock = field(default_factory=threading.Lock)


class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
    name = "elevenlabs-realtime"
    supports_streaming = True

    def __init__(
        self,
        *,
        api_base: str | None = None,
        api_key: str | None = None,
        timeout_seconds: float | None = None,
        finalize_timeout_seconds: float | None = None,
        model_id: str | None = None,
        audio_format: str | None = None,
        commit_strategy: str | None = None,
        websocket_factory: Callable[..., object] | None = None,
        debug_dump_audio: bool | None = None,
        debug_dump_dir: str | Path | None = None,
    ) -> None:
        self._api_base = str(api_base or _elevenlabs_asr_api_base()).strip().rstrip("/")
        self._api_key = str(api_key if api_key is not None else _elevenlabs_asr_api_key()).strip()
        self._timeout_seconds = max(
            float(timeout_seconds if timeout_seconds is not None else _streaming_asr_timeout_seconds()),
            0.25,
        )
        self._finalize_timeout_seconds = max(
            float(
                finalize_timeout_seconds
                if finalize_timeout_seconds is not None
                else _elevenlabs_realtime_finalize_timeout_seconds()
            ),
            0.25,
        )
        self._model_id = str(model_id or _elevenlabs_realtime_model_id()).strip() or "scribe_v2_realtime"
        self._audio_format = str(audio_format or _elevenlabs_realtime_audio_format()).strip().lower() or "pcm_16000"
        self._sample_rate_hz = _elevenlabs_realtime_sample_rate_hz(self._audio_format)
        self._commit_strategy = str(commit_strategy or _elevenlabs_realtime_commit_strategy()).strip().lower()
        if self._commit_strategy not in {"manual", "vad"}:
            self._commit_strategy = "manual"
        self._websocket_factory = websocket_factory
        self._streams: dict[str, _ElevenLabsRealtimeStreamState] = {}
        self._lock = threading.Lock()
        self._debug_dump_audio = (
            debug_dump_audio if debug_dump_audio is not None else _elevenlabs_realtime_debug_dump_audio_enabled()
        )
        self._debug_dump_dir = Path(debug_dump_dir) if debug_dump_dir is not None else _elevenlabs_realtime_debug_dump_audio_dir()

    def _websocket_url(self, *, language: str) -> str:
        base = self._api_base
        if base.startswith("https://"):
            websocket_base = "wss://" + base[len("https://") :]
        elif base.startswith("http://"):
            websocket_base = "ws://" + base[len("http://") :]
        else:
            websocket_base = base
        params = {
            "model_id": self._model_id,
            "audio_format": self._audio_format,
            "language_code": language,
            "commit_strategy": self._commit_strategy,
            "include_timestamps": "false",
            "include_language_detection": "false",
        }
        return f"{websocket_base}/v1/speech-to-text/realtime?{urllib.parse.urlencode(params)}"

    def _create_websocket(self, url: str) -> object:
        if not self._api_key:
            raise StreamingASRUnavailable(
                "AI_VOICE_ASR_ELEVENLABS_API_KEY or AI_VOICE_TTS_ELEVENLABS_API_KEY is required for ElevenLabs realtime ASR"
            )
        factory = self._websocket_factory
        if factory is None:
            try:
                import websocket  # type: ignore[import-not-found]
            except ImportError as exc:
                raise StreamingASRUnavailable("websocket-client is required for ElevenLabs realtime ASR") from exc
            factory = websocket.create_connection
        try:
            return factory(
                url,
                header=[f"xi-api-key: {self._api_key}"],
                timeout=self._timeout_seconds,
            )
        except BaseException as exc:
            raise StreamingASRUnavailable(str(exc)[:500] or "ElevenLabs realtime ASR is unavailable") from exc

    @staticmethod
    def _message_type(payload: object) -> str:
        if not isinstance(payload, dict):
            return ""
        return str(payload.get("message_type") or payload.get("type") or "").strip()

    @staticmethod
    def _payload_text(payload: object) -> str:
        if not isinstance(payload, dict):
            return ""
        return str(payload.get("text") or payload.get("transcript") or "").strip()

    @staticmethod
    def _payload_language(payload: object, fallback: str) -> str:
        if not isinstance(payload, dict):
            return fallback
        return str(payload.get("language_code") or payload.get("language") or fallback).strip() or fallback

    @staticmethod
    def _payload_error_message(payload: object) -> str:
        if not isinstance(payload, dict):
            return "ElevenLabs realtime ASR error"
        for key in ("message", "error", "reason", "detail"):
            value = payload.get(key)
            if value:
                return str(value)[:500]
        return json.dumps(payload, ensure_ascii=True)[:500]

    def _send_json(self, state: _ElevenLabsRealtimeStreamState, payload: dict[str, object]) -> None:
        with state.lock:
            if state.error is not None:
                raise StreamingASRUnavailable(str(state.error)[:500])
            if state.close_requested:
                raise StreamingASRUnavailable("ElevenLabs realtime ASR stream is closed")
        try:
            with state.send_lock:
                state.websocket.send(json.dumps(payload, separators=(",", ":")))  # type: ignore[attr-defined]
        except BaseException as exc:
            with state.lock:
                state.error = exc
                state.final_event.set()
            raise StreamingASRUnavailable(str(exc)[:500] or "ElevenLabs realtime ASR send failed") from exc

    def _drain_updates(self, state: _ElevenLabsRealtimeStreamState) -> None:
        while True:
            try:
                update = state.updates_queue.get_nowait()
            except queue.Empty:
                return
            with state.lock:
                normalized_text = " ".join(str(update.text or "").strip().lower().split())
                if normalized_text and normalized_text == state.last_partial_text:
                    state.partial_streak += 1
                else:
                    state.last_partial_text = normalized_text
                    state.partial_streak = 1 if normalized_text else 0
                if update.is_final:
                    # A committed segment from the server. Under commit_strategy=vad
                    # the server can autonomously commit more than one segment across
                    # a single long utterance while the caller keeps talking, so
                    # accumulate every segment instead of treating the first one as
                    # the whole turn's answer — only our own finalize() call (which
                    # sets finalize_requested) marks the turn as actually complete.
                    segment_text = str(update.text or "").strip()
                    if segment_text:
                        state.committed_segments.append(segment_text)
                    accumulated_text = " ".join(state.committed_segments).strip()
                    accumulated = StreamingASRPartial(
                        text=accumulated_text,
                        language=update.language or state.language,
                        confidence=update.confidence,
                        is_final=state.finalize_requested,
                        is_stable=True,
                    )
                    state.stable_partial = accumulated
                    state.latest_partial = accumulated
                    if state.finalize_requested:
                        state.final_transcription = ASRTranscription(
                            text=accumulated_text,
                            language=update.language or state.language,
                            confidence=update.confidence,
                        )
                        state.final_event.set()
                    continue
                if update.is_stable or state.partial_streak >= 2:
                    state.stable_partial = StreamingASRPartial(
                        text=update.text,
                        language=update.language or state.language,
                        confidence=update.confidence,
                        is_final=False,
                        is_stable=True,
                    )
                    update = state.stable_partial
                state.latest_partial = update

    @staticmethod
    def _transcription_from_partial(
        partial: StreamingASRPartial,
        *,
        fallback_language: str,
    ) -> ASRTranscription:
        return ASRTranscription(
            text=partial.text,
            language=partial.language or fallback_language,
            confidence=partial.confidence,
        )

    def _best_partial_transcription(self, state: _ElevenLabsRealtimeStreamState) -> ASRTranscription | None:
        with state.lock:
            partial = state.stable_partial or state.latest_partial
            if partial is None:
                return None
            return self._transcription_from_partial(partial, fallback_language=state.language)

    def _run_reader(self, state: _ElevenLabsRealtimeStreamState) -> None:
        try:
            while True:
                with state.lock:
                    if state.close_requested:
                        return
                raw_message = state.websocket.recv()  # type: ignore[attr-defined]
                if raw_message is None:
                    continue
                if isinstance(raw_message, bytes):
                    raw_message = raw_message.decode("utf-8", errors="replace")
                try:
                    payload = json.loads(str(raw_message))
                except json.JSONDecodeError:
                    continue
                message_type = self._message_type(payload)
                if message_type == "partial_transcript":
                    text = self._payload_text(payload)
                    if text:
                        logger.info(
                            "asr.elevenlabs_realtime.partial stream_id=%s text=%s",
                            state.stream_id,
                            text[:160],
                        )
                        state.updates_queue.put(
                            StreamingASRPartial(
                                text=text,
                                language=self._payload_language(payload, state.language),
                                confidence=None,
                                is_final=False,
                                is_stable=False,
                            )
                        )
                    continue
                if message_type in {"committed_transcript", "committed_transcript_with_timestamps"}:
                    text = self._payload_text(payload)
                    language = self._payload_language(payload, state.language)
                    if text:
                        logger.info(
                            "asr.elevenlabs_realtime.committed stream_id=%s language=%s text=%s",
                            state.stream_id,
                            language,
                            text[:160],
                        )
                    # Route empty commits through the same queue/accumulation path as
                    # non-empty ones (rather than setting final_event directly here) so
                    # an autonomous empty vad-mode commit mid-utterance doesn't end the
                    # turn early — only our own finalize_requested gate does that.
                    state.updates_queue.put(
                        StreamingASRPartial(
                            text=text,
                            language=language,
                            confidence=None,
                            is_final=True,
                            is_stable=True,
                        )
                    )
                    continue
                if message_type.startswith("scribe") and "error" in message_type.lower():
                    raise StreamingASRUnavailable(self._payload_error_message(payload))
        except BaseException as exc:
            with state.lock:
                if not state.close_requested:
                    state.error = exc
                state.final_event.set()
            if not state.close_requested:
                logger.warning(
                    "asr.elevenlabs_realtime.reader_error stream_id=%s error=%s",
                    state.stream_id,
                    str(exc)[:500],
                )

    def _state(self, stream_id: str) -> _ElevenLabsRealtimeStreamState:
        with self._lock:
            state = self._streams.get(stream_id)
        if state is None:
            raise StreamingASRUnavailable("ElevenLabs realtime ASR stream is not active")
        return state

    def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
        stream_id = f"elrt_{uuid.uuid4().hex}"
        language = _normalize_elevenlabs_realtime_language(language_hint)
        websocket = self._create_websocket(self._websocket_url(language=language))
        state = _ElevenLabsRealtimeStreamState(
            stream_id=stream_id,
            session_id=str(session_id or "").strip(),
            language=language,
            websocket=websocket,
            updates_queue=queue.Queue(),
        )
        thread = threading.Thread(
            target=self._run_reader,
            args=(state,),
            name=f"elevenlabs-realtime-asr-{stream_id[:16]}",
            daemon=True,
        )
        state.thread = thread
        with self._lock:
            self._streams[stream_id] = state
        thread.start()
        logger.info(
            "asr.elevenlabs_realtime.stream_open stream_id=%s session_id=%s language=%s model=%s "
            "audio_format=%s sample_rate=%s commit_strategy=%s",
            stream_id,
            state.session_id,
            language,
            self._model_id,
            self._audio_format,
            self._sample_rate_hz,
            self._commit_strategy,
        )
        return stream_id

    def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None:
        if not pcm_8k_chunk:
            return
        state = self._state(stream_id)
        pcm_bytes = pcm_8k_chunk
        if self._sample_rate_hz != 8000:
            with state.resample_lock:
                pcm_bytes, state.resample_state = resample_pcm16le_stateful(
                    pcm_8k_chunk,
                    input_rate_hz=8000,
                    output_rate_hz=self._sample_rate_hz,
                    state=state.resample_state,
                )
        if self._debug_dump_audio:
            with state.debug_lock:
                state.debug_pcm.extend(pcm_bytes)
        self._send_json(
            state,
            {
                "message_type": "input_audio_chunk",
                "audio_base_64": base64.b64encode(pcm_bytes).decode("ascii"),
                "commit": False,
                "sample_rate": self._sample_rate_hz,
            },
        )

    def _write_debug_dump(self, state: _ElevenLabsRealtimeStreamState) -> None:
        with state.debug_lock:
            pcm_bytes = bytes(state.debug_pcm)
        if not pcm_bytes:
            return
        try:
            self._debug_dump_dir.mkdir(parents=True, exist_ok=True)
            path = self._debug_dump_dir / f"{state.stream_id}.wav"
            path.write_bytes(pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=self._sample_rate_hz))
        except OSError as exc:
            logger.warning(
                "asr.elevenlabs_realtime.debug_dump_failed stream_id=%s error=%s",
                state.stream_id,
                str(exc)[:300],
            )
            return
        logger.info(
            "asr.elevenlabs_realtime.debug_dump_written stream_id=%s path=%s bytes=%s",
            state.stream_id,
            path,
            len(pcm_bytes),
        )

    def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
        state = self._state(stream_id)
        self._drain_updates(state)
        with state.lock:
            if state.error is not None and state.latest_partial is None:
                raise StreamingASRUnavailable(str(state.error)[:500])
            return state.stable_partial or state.latest_partial

    def finalize(self, stream_id: str) -> ASRTranscription:
        state = self._state(stream_id)
        start_monotonic = time.monotonic()

        def _log_and_return(transcription: ASRTranscription, note: str) -> ASRTranscription:
            logger.info(
                "asr.elevenlabs_realtime.finalize stream_id=%s note=%s elapsed_ms=%s text=%s",
                stream_id,
                note,
                int((time.monotonic() - start_monotonic) * 1000),
                transcription.text[:160],
            )
            return transcription

        # Drain anything already queued (e.g. an autonomous vad-mode commit that
        # arrived mid-utterance) *before* marking finalize_requested, so those
        # segments get accumulated as ongoing speech rather than mistaken for the
        # response to the commit we're about to send.
        self._drain_updates(state)
        with state.lock:
            state.finalize_requested = True

        silence = b"\x00\x00" * int(self._sample_rate_hz * 0.1)
        try:
            self._send_json(
                state,
                {
                    "message_type": "input_audio_chunk",
                    "audio_base_64": base64.b64encode(silence).decode("ascii"),
                    "commit": True,
                    "sample_rate": self._sample_rate_hz,
                },
            )
        except StreamingASRUnavailable:
            partial_transcription = self._best_partial_transcription(state)
            if partial_transcription is not None and partial_transcription.text.strip():
                return _log_and_return(partial_transcription, "partial_after_send_failure")
            raise
        deadline = time.monotonic() + self._finalize_timeout_seconds
        while True:
            self._drain_updates(state)
            with state.lock:
                if state.final_transcription is not None:
                    return _log_and_return(state.final_transcription, "final")
                latest_partial = state.stable_partial or state.latest_partial
                error = state.error
                event_is_set = state.final_event.is_set()
            if event_is_set or time.monotonic() >= deadline:
                if latest_partial is not None:
                    return _log_and_return(
                        self._transcription_from_partial(latest_partial, fallback_language=state.language),
                        "partial_on_event_or_deadline",
                    )
                if error is not None:
                    raise StreamingASRUnavailable(str(error)[:500])
                if time.monotonic() >= deadline:
                    raise StreamingASRUnavailable("ElevenLabs realtime ASR finalize timed out")
                return _log_and_return(ASRTranscription(text="", language=state.language, confidence=None), "empty")
            state.final_event.wait(timeout=0.05)

    def close_stream(self, stream_id: str) -> None:
        try:
            state = self._state(stream_id)
        except StreamingASRUnavailable:
            return
        with state.lock:
            state.close_requested = True
            state.final_event.set()
        close = getattr(state.websocket, "close", None)
        if callable(close):
            try:
                close()
            except BaseException:
                pass
        thread = state.thread
        if thread is not None and thread.is_alive():
            thread.join(timeout=0.5)
        if self._debug_dump_audio:
            self._write_debug_dump(state)
        with self._lock:
            self._streams.pop(stream_id, None)
        logger.info("asr.elevenlabs_realtime.stream_close stream_id=%s", stream_id)


class YandexSpeechKitASRProvider(ASRProvider):
    name = "yandex"

    def __init__(
        self,
        *,
        api_base: str | None = None,
        api_key: str | None = None,
        iam_token: str | None = None,
        folder_id: str | None = None,
        timeout_seconds: float | None = None,
        sample_rate_hz: int | None = None,
        topic: str | None = None,
    ) -> None:
        self._api_base = str(api_base or _yandex_asr_api_base()).strip().rstrip("/")
        self._api_key = str(api_key if api_key is not None else _yandex_asr_api_key()).strip()
        self._iam_token = str(iam_token if iam_token is not None else _yandex_asr_iam_token()).strip()
        self._folder_id = str(folder_id if folder_id is not None else _yandex_asr_folder_id()).strip()
        self._operations_base = _yandex_asr_operations_base()
        self._timeout_seconds = max(
            float(timeout_seconds if timeout_seconds is not None else _yandex_asr_timeout_seconds()),
            3.0,
        )
        self._sample_rate_hz = int(sample_rate_hz if sample_rate_hz is not None else _yandex_asr_sample_rate_hz())
        self._topic = str(topic if topic is not None else _yandex_asr_topic()).strip()
        self._api_version = _yandex_asr_api_version(self._api_base)
        self._poll_interval_seconds = _yandex_asr_poll_interval_seconds()

    @staticmethod
    def _lpcm_from_audio_bytes(audio_bytes: bytes, *, default_sample_rate_hz: int) -> tuple[bytes, int]:
        if not audio_bytes:
            return b"", default_sample_rate_hz
        try:
            with wave.open(io.BytesIO(audio_bytes), "rb") as wav_file:
                sample_width = wav_file.getsampwidth()
                channels = wav_file.getnchannels()
                sample_rate = wav_file.getframerate()
                if sample_width != 2 or channels != 1:
                    return audio_bytes, default_sample_rate_hz
                return wav_file.readframes(wav_file.getnframes()), int(sample_rate or default_sample_rate_hz)
        except (wave.Error, EOFError):
            return audio_bytes, default_sample_rate_hz

    def _headers(self, *, content_type: str = "application/octet-stream") -> dict[str, str]:
        if not self._api_key and not self._iam_token:
            raise RuntimeError(
                "AI_VOICE_ASR_YANDEX_API_KEY or AI_VOICE_ASR_YANDEX_IAM_TOKEN is required for Yandex ASR"
            )
        headers = {"Content-Type": content_type}
        if self._api_key:
            headers["Authorization"] = f"Api-Key {self._api_key}"
        else:
            headers["Authorization"] = f"Bearer {self._iam_token}"
        if self._folder_id and self._api_version == "v3":
            headers["x-folder-id"] = self._folder_id
        return headers

    def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        pcm_bytes, sample_rate_hz = self._lpcm_from_audio_bytes(
            audio_bytes,
            default_sample_rate_hz=self._sample_rate_hz,
        )
        if not pcm_bytes:
            return ASRTranscription(text="", language=language_hint, confidence=None)

        language = _normalize_yandex_asr_language(language_hint)
        if self._api_version == "v3":
            return self._transcribe_v3_async(pcm_bytes, sample_rate_hz=sample_rate_hz, language=language)
        return self._transcribe_v1_sync(pcm_bytes, sample_rate_hz=sample_rate_hz, language=language)

    def _transcribe_v1_sync(self, pcm_bytes: bytes, *, sample_rate_hz: int, language: str) -> ASRTranscription:
        params: dict[str, str] = {
            "lang": language,
            "format": "lpcm",
            "sampleRateHertz": str(sample_rate_hz),
        }
        if self._topic:
            params["topic"] = self._topic
        if self._folder_id:
            params["folderId"] = self._folder_id

        with httpx.Client(timeout=self._timeout_seconds) as client:
            response = client.post(
                f"{self._api_base}/speech/v1/stt:recognize",
                headers=self._headers(),
                params=params,
                content=pcm_bytes,
            )
        response.raise_for_status()
        payload = response.json()
        return ASRTranscription(
            text=str(payload.get("result") or "").strip(),
            language=language,
            confidence=None,
        )

    def _transcribe_v3_async(self, pcm_bytes: bytes, *, sample_rate_hz: int, language: str) -> ASRTranscription:
        payload = {
            "content": base64.b64encode(pcm_bytes).decode("ascii"),
            "recognitionModel": {
                "model": self._topic or "general",
                "audioFormat": {
                    "rawAudio": {
                        "audioEncoding": "LINEAR16_PCM",
                        "sampleRateHertz": str(sample_rate_hz),
                        "audioChannelCount": "1",
                    }
                },
                "textNormalization": {
                    "textNormalization": "TEXT_NORMALIZATION_ENABLED",
                    "profanityFilter": False,
                    "literatureText": False,
                    "phoneFormattingMode": "PHONE_FORMATTING_MODE_DISABLED",
                },
                "languageRestriction": {
                    "restrictionType": "WHITELIST",
                    "languageCode": [language],
                },
                "audioProcessingType": "FULL_DATA",
            },
        }
        headers = self._headers(content_type="application/json")
        deadline = time.monotonic() + self._timeout_seconds
        with httpx.Client(timeout=self._timeout_seconds) as client:
            response = client.post(
                f"{self._api_base}/stt/v3/recognizeFileAsync",
                headers=headers,
                json=payload,
            )
            response.raise_for_status()
            operation_payload = response.json()
            operation_id = str(operation_payload.get("id") or "").strip()
            if not operation_id:
                raise RuntimeError("Yandex ASR v3 response is missing operation id")

            while not bool(operation_payload.get("done")):
                if time.monotonic() >= deadline:
                    raise TimeoutError("Yandex ASR v3 recognition timed out")
                time.sleep(self._poll_interval_seconds)
                operation_response = client.get(
                    f"{self._operations_base}/operations/{operation_id}",
                    headers=headers,
                )
                operation_response.raise_for_status()
                operation_payload = operation_response.json()

            error_payload = operation_payload.get("error")
            if isinstance(error_payload, dict) and error_payload:
                raise RuntimeError(str(error_payload.get("message") or error_payload)[:500])

            result_response = client.get(
                f"{self._api_base}/stt/v3/getRecognition",
                headers=headers,
                params={"operationId": operation_id},
            )
            result_response.raise_for_status()
            result_payload = result_response.json()

        return ASRTranscription(
            text=self._extract_v3_text(result_payload),
            language=language,
            confidence=None,
        )

    @classmethod
    def _extract_v3_text(cls, payload: object) -> str:
        if isinstance(payload, list):
            texts = [cls._extract_v3_text(item) for item in payload]
            return " ".join(text for text in texts if text).strip()
        if not isinstance(payload, dict):
            return ""

        final_refinement = payload.get("finalRefinement")
        if isinstance(final_refinement, dict):
            normalized = final_refinement.get("normalizedText")
            text = cls._extract_alternatives_text(normalized)
            if text:
                return text

        for key in ("final", "partial"):
            text = cls._extract_alternatives_text(payload.get(key))
            if text:
                return text

        response = payload.get("response")
        if isinstance(response, (dict, list)):
            return cls._extract_v3_text(response)
        return ""

    @staticmethod
    def _extract_alternatives_text(payload: object) -> str:
        if not isinstance(payload, dict):
            return ""
        alternatives = payload.get("alternatives")
        if not isinstance(alternatives, list):
            return ""
        texts: list[str] = []
        for alternative in alternatives:
            if not isinstance(alternative, dict):
                continue
            text = str(alternative.get("text") or "").strip()
            if text:
                texts.append(text)
        return " ".join(texts).strip()

    def transcribe_partial(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
        return self.transcribe(audio_bytes, language_hint=language_hint)


class LocalSidecarStreamingASRProvider(StreamingASRProvider):
    name = "local-sidecar"
    supports_streaming = True

    def __init__(
        self,
        *,
        api_base: str | None = None,
        timeout_seconds: float | None = None,
    ) -> None:
        self._api_base = str(api_base or _streaming_asr_api_base()).strip().rstrip("/")
        self._timeout_seconds = max(float(timeout_seconds or _streaming_asr_timeout_seconds()), 0.25)

    def _request(
        self,
        method: str,
        path: str,
        *,
        payload: dict[str, object] | None = None,
    ) -> dict[str, object]:
        if not self._api_base:
            raise StreamingASRUnavailable("Streaming ASR API base is not configured")
        try:
            with httpx.Client(timeout=self._timeout_seconds) as client:
                response = client.request(
                    method,
                    f"{self._api_base}{path}",
                    json=payload,
                    headers=_service_headers(),
                )
            response.raise_for_status()
        except httpx.HTTPError as exc:
            raise StreamingASRUnavailable(str(exc)[:500] or "Streaming ASR sidecar is unavailable") from exc
        body = response.json() if response.content else {}
        return body if isinstance(body, dict) else {}

    def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
        payload = {
            "session_id": str(session_id or "").strip(),
            "language_hint": str(language_hint or "").strip() or None,
            "sample_rate_hz": 8000,
            "encoding": "pcm_s16le",
        }
        body = self._request("POST", "/internal/asr/streams", payload=payload)
        stream_id = str(body.get("stream_id") or "").strip()
        if not stream_id:
            raise StreamingASRUnavailable("Streaming ASR sidecar did not return stream_id")
        return stream_id

    def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None:
        if not pcm_8k_chunk:
            return
        self._request(
            "POST",
            f"/internal/asr/streams/{stream_id}/chunks",
            payload={
                "pcm_b64": base64.b64encode(pcm_8k_chunk).decode("ascii"),
                "sample_rate_hz": 8000,
                "encoding": "pcm_s16le",
            },
        )

    def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
        body = self._request("GET", f"/internal/asr/streams/{stream_id}/partial")
        text = str(body.get("text") or "").strip()
        if not text:
            return None
        language = str(body.get("language") or "").strip() or None
        confidence_raw = body.get("confidence")
        confidence = float(confidence_raw) if isinstance(confidence_raw, (int, float)) else None
        return StreamingASRPartial(
            text=text,
            language=language,
            confidence=confidence,
            is_final=bool(body.get("is_final")),
            is_stable=bool(body.get("is_stable")),
        )

    def finalize(self, stream_id: str) -> ASRTranscription:
        body = self._request("POST", f"/internal/asr/streams/{stream_id}/finalize")
        return ASRTranscription(
            text=str(body.get("text") or "").strip(),
            language=str(body.get("language") or "").strip() or None,
            confidence=float(body["confidence"]) if isinstance(body.get("confidence"), (int, float)) else None,
        )

    def close_stream(self, stream_id: str) -> None:
        try:
            self._request("DELETE", f"/internal/asr/streams/{stream_id}")
        except StreamingASRUnavailable:
            return


@dataclass
class _YandexGrpcStreamState:
    stream_id: str
    session_id: str
    language: str
    input_queue: queue.Queue[bytes | None]
    updates_queue: queue.Queue[StreamingASRPartial]
    thread: threading.Thread | None = None
    latest_partial: StreamingASRPartial | None = None
    final_transcription: ASRTranscription | None = None
    error: BaseException | None = None
    finish_requested: bool = False
    lock: threading.Lock = field(default_factory=threading.Lock)


class YandexSpeechKitGrpcStreamingASRProvider(StreamingASRProvider):
    name = "yandex-speechkit-grpc"
    supports_streaming = True

    def __init__(
        self,
        *,
        api_base: str | None = None,
        api_key: str | None = None,
        iam_token: str | None = None,
        folder_id: str | None = None,
        timeout_seconds: float | None = None,
        sample_rate_hz: int | None = None,
        topic: str | None = None,
        grpc_target: str | None = None,
        queue_max_chunks: int | None = None,
        eou_max_pause_ms: int | None = None,
        grpc_module: object | None = None,
        stt_pb2_module: object | None = None,
        stt_service_pb2_grpc_module: object | None = None,
    ) -> None:
        self._api_base = str(api_base or _yandex_asr_api_base()).strip().rstrip("/")
        self._api_key = str(api_key if api_key is not None else _yandex_asr_api_key()).strip()
        self._iam_token = str(iam_token if iam_token is not None else _yandex_asr_iam_token()).strip()
        self._folder_id = str(folder_id if folder_id is not None else _yandex_asr_folder_id()).strip()
        self._timeout_seconds = max(
            float(timeout_seconds if timeout_seconds is not None else _yandex_asr_timeout_seconds()),
            3.0,
        )
        self._sample_rate_hz = int(sample_rate_hz if sample_rate_hz is not None else _yandex_asr_sample_rate_hz())
        self._topic = str(topic if topic is not None else _yandex_asr_topic()).strip() or "general"
        self._grpc_target = str(grpc_target or _yandex_asr_grpc_target(self._api_base)).strip()
        self._queue_max_chunks = int(queue_max_chunks or _yandex_asr_stream_queue_max_chunks())
        self._eou_max_pause_ms = int(eou_max_pause_ms or _yandex_asr_eou_max_pause_ms())
        self._grpc_module = grpc_module
        self._stt_pb2_module = stt_pb2_module
        self._stt_service_pb2_grpc_module = stt_service_pb2_grpc_module
        self._streams: dict[str, _YandexGrpcStreamState] = {}
        self._lock = threading.Lock()

    def _load_grpc_modules(self) -> tuple[object, object, object]:
        if (
            self._grpc_module is not None
            and self._stt_pb2_module is not None
            and self._stt_service_pb2_grpc_module is not None
        ):
            return self._grpc_module, self._stt_pb2_module, self._stt_service_pb2_grpc_module
        try:
            import grpc  # type: ignore[import-not-found]
            from yandex.cloud.ai.stt.v3 import stt_pb2  # type: ignore[import-not-found]
            from yandex.cloud.ai.stt.v3 import stt_service_pb2_grpc  # type: ignore[import-not-found]
        except ImportError as exc:
            raise StreamingASRUnavailable("Yandex SpeechKit gRPC dependencies are not installed") from exc
        return grpc, stt_pb2, stt_service_pb2_grpc

    def _metadata(self) -> tuple[tuple[str, str], ...]:
        if not self._api_key and not self._iam_token:
            raise StreamingASRUnavailable(
                "AI_VOICE_ASR_YANDEX_API_KEY or AI_VOICE_ASR_YANDEX_IAM_TOKEN is required for Yandex streaming ASR"
            )
        metadata: list[tuple[str, str]] = []
        if self._api_key:
            metadata.append(("authorization", f"Api-Key {self._api_key}"))
        else:
            metadata.append(("authorization", f"Bearer {self._iam_token}"))
        if self._folder_id:
            metadata.append(("x-folder-id", self._folder_id))
        return tuple(metadata)

    @staticmethod
    def _enum_value(owner: object, name: str, default: int) -> int:
        return int(getattr(owner, name, default))

    def _build_session_options(self, stt_pb2: object, *, language: str) -> object:
        raw_audio = stt_pb2.RawAudio(  # type: ignore[attr-defined]
            audio_encoding=self._enum_value(stt_pb2.RawAudio, "LINEAR16_PCM", 1),  # type: ignore[attr-defined]
            sample_rate_hertz=self._sample_rate_hz,
            audio_channel_count=1,
        )
        recognition_model = stt_pb2.RecognitionModelOptions(  # type: ignore[attr-defined]
            model=self._topic,
            audio_format=stt_pb2.AudioFormatOptions(raw_audio=raw_audio),  # type: ignore[attr-defined]
            text_normalization=stt_pb2.TextNormalizationOptions(  # type: ignore[attr-defined]
                text_normalization=self._enum_value(
                    stt_pb2.TextNormalizationOptions,  # type: ignore[attr-defined]
                    "TEXT_NORMALIZATION_ENABLED",
                    1,
                ),
                profanity_filter=False,
                literature_text=False,
                phone_formatting_mode=self._enum_value(
                    stt_pb2.TextNormalizationOptions,  # type: ignore[attr-defined]
                    "PHONE_FORMATTING_MODE_DISABLED",
                    1,
                ),
            ),
            language_restriction=stt_pb2.LanguageRestrictionOptions(  # type: ignore[attr-defined]
                restriction_type=self._enum_value(stt_pb2.LanguageRestrictionOptions, "WHITELIST", 1),  # type: ignore[attr-defined]
                language_code=[language],
            ),
            audio_processing_type=self._enum_value(stt_pb2.RecognitionModelOptions, "REAL_TIME", 1),  # type: ignore[attr-defined]
        )
        eou_classifier = stt_pb2.EouClassifierOptions(  # type: ignore[attr-defined]
            default_classifier=stt_pb2.DefaultEouClassifier(  # type: ignore[attr-defined]
                type=self._enum_value(stt_pb2.DefaultEouClassifier, "HIGH", 2),  # type: ignore[attr-defined]
                max_pause_between_words_hint_ms=self._eou_max_pause_ms,
            )
        )
        return stt_pb2.StreamingOptions(  # type: ignore[attr-defined]
            recognition_model=recognition_model,
            eou_classifier=eou_classifier,
        )

    def _request_iterator(self, state: _YandexGrpcStreamState, stt_pb2: object):
        yield stt_pb2.StreamingRequest(  # type: ignore[attr-defined]
            session_options=self._build_session_options(stt_pb2, language=state.language)
        )
        while True:
            item = state.input_queue.get()
            try:
                if item is None:
                    yield stt_pb2.StreamingRequest(eou=stt_pb2.Eou())  # type: ignore[attr-defined]
                    return
                if item:
                    yield stt_pb2.StreamingRequest(chunk=stt_pb2.AudioChunk(data=item))  # type: ignore[attr-defined]
            finally:
                state.input_queue.task_done()

    @staticmethod
    def _has_field(message: object, field_name: str) -> bool:
        try:
            return bool(message.HasField(field_name))  # type: ignore[attr-defined]
        except (AttributeError, ValueError):
            return getattr(message, field_name, None) is not None

    @classmethod
    def _partial_from_response(cls, response: object) -> StreamingASRPartial | None:
        for field_name, is_final in (
            ("final_refinement", True),
            ("final", True),
            ("partial", False),
        ):
            if not cls._has_field(response, field_name):
                continue
            payload = getattr(response, field_name, None)
            if field_name == "final_refinement" and payload is not None:
                payload = getattr(payload, "normalized_text", None)
            text, confidence, language = cls._extract_alternatives(payload)
            if text:
                return StreamingASRPartial(
                    text=text,
                    language=language,
                    confidence=confidence,
                    is_final=is_final,
                    is_stable=is_final,
                )
        return None

    @staticmethod
    def _extract_alternatives(payload: object) -> tuple[str, float | None, str | None]:
        alternatives = getattr(payload, "alternatives", None)
        if not alternatives:
            return "", None, None
        texts: list[str] = []
        confidence: float | None = None
        language: str | None = None
        for alternative in alternatives:
            text = str(getattr(alternative, "text", "") or "").strip()
            if text:
                texts.append(text)
            if confidence is None:
                raw_confidence = getattr(alternative, "confidence", None)
                if isinstance(raw_confidence, (int, float)):
                    confidence = float(raw_confidence)
            if language is None:
                languages = getattr(alternative, "languages", None)
                if languages:
                    first_language = languages[0]
                    language_code = str(getattr(first_language, "language_code", "") or "").strip()
                    if language_code:
                        language = language_code
        return " ".join(texts).strip(), confidence, language

    def _drain_updates(self, state: _YandexGrpcStreamState) -> None:
        while True:
            try:
                update = state.updates_queue.get_nowait()
            except queue.Empty:
                return
            with state.lock:
                state.latest_partial = update
                if update.is_final:
                    state.final_transcription = ASRTranscription(
                        text=update.text,
                        language=update.language or state.language,
                        confidence=update.confidence,
                    )

    def _run_stream(self, state: _YandexGrpcStreamState) -> None:
        channel = None
        try:
            grpc, stt_pb2, stt_service_pb2_grpc = self._load_grpc_modules()
            channel = grpc.secure_channel(  # type: ignore[attr-defined]
                self._grpc_target,
                grpc.ssl_channel_credentials(),  # type: ignore[attr-defined]
            )
            stub = stt_service_pb2_grpc.RecognizerStub(channel)  # type: ignore[attr-defined]
            responses = stub.RecognizeStreaming(
                self._request_iterator(state, stt_pb2),
                metadata=self._metadata(),
                timeout=self._timeout_seconds,
            )
            for response in responses:
                partial = self._partial_from_response(response)
                if partial is not None:
                    state.updates_queue.put(partial)
        except BaseException as exc:
            with state.lock:
                state.error = exc
        finally:
            if channel is not None:
                close = getattr(channel, "close", None)
                if callable(close):
                    close()

    def _state(self, stream_id: str) -> _YandexGrpcStreamState:
        with self._lock:
            state = self._streams.get(stream_id)
        if state is None:
            raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream is not active")
        return state

    def _signal_finish(self, state: _YandexGrpcStreamState) -> None:
        with state.lock:
            if state.finish_requested:
                return
            state.finish_requested = True
        try:
            state.input_queue.put_nowait(None)
        except queue.Full:
            try:
                state.input_queue.get_nowait()
                state.input_queue.task_done()
            except queue.Empty:
                pass
            state.input_queue.put_nowait(None)

    def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
        self._metadata()
        self._load_grpc_modules()
        stream_id = f"yasr_grpc_{uuid.uuid4().hex}"
        language = _normalize_yandex_asr_language(language_hint)
        state = _YandexGrpcStreamState(
            stream_id=stream_id,
            session_id=str(session_id or "").strip(),
            language=language,
            input_queue=queue.Queue(maxsize=self._queue_max_chunks),
            updates_queue=queue.Queue(),
        )
        thread = threading.Thread(
            target=self._run_stream,
            args=(state,),
            name=f"yandex-speechkit-asr-{stream_id[:16]}",
            daemon=True,
        )
        state.thread = thread
        with self._lock:
            self._streams[stream_id] = state
        thread.start()
        return stream_id

    def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None:
        if not pcm_8k_chunk:
            return
        state = self._state(stream_id)
        with state.lock:
            if state.error is not None:
                raise StreamingASRUnavailable(str(state.error)[:500])
            if state.finish_requested:
                raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream is already finalizing")
        try:
            state.input_queue.put(pcm_8k_chunk, timeout=0.1)
        except queue.Full as exc:
            raise StreamingASRUnavailable("Yandex SpeechKit gRPC input queue is full") from exc

    def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
        state = self._state(stream_id)
        self._drain_updates(state)
        with state.lock:
            if state.error is not None and state.latest_partial is None:
                raise StreamingASRUnavailable(str(state.error)[:500])
            return state.latest_partial

    def finalize(self, stream_id: str) -> ASRTranscription:
        state = self._state(stream_id)
        self._signal_finish(state)
        deadline = time.monotonic() + self._timeout_seconds
        while True:
            self._drain_updates(state)
            thread = state.thread
            if thread is None or not thread.is_alive():
                break
            if time.monotonic() >= deadline:
                raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream finalize timed out")
            thread.join(timeout=0.05)
        self._drain_updates(state)
        with state.lock:
            if state.final_transcription is not None:
                return state.final_transcription
            if state.latest_partial is not None:
                return ASRTranscription(
                    text=state.latest_partial.text,
                    language=state.latest_partial.language or state.language,
                    confidence=state.latest_partial.confidence,
                )
            if state.error is not None:
                raise StreamingASRUnavailable(str(state.error)[:500])
        return ASRTranscription(text="", language=state.language, confidence=None)

    def close_stream(self, stream_id: str) -> None:
        try:
            state = self._state(stream_id)
        except StreamingASRUnavailable:
            return
        self._signal_finish(state)
        thread = state.thread
        if thread is not None and thread.is_alive():
            thread.join(timeout=0.5)
        with self._lock:
            self._streams.pop(stream_id, None)


class YandexSpeechKitBufferedStreamingASRProvider(StreamingASRProvider):
    name = "yandex-speechkit-buffered"
    supports_streaming = True

    def __init__(self, *, asr_provider: YandexSpeechKitASRProvider | None = None) -> None:
        self._asr_provider = asr_provider or YandexSpeechKitASRProvider()
        self._streams: dict[str, dict[str, object]] = {}
        self._lock = threading.Lock()

    def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
        stream_id = f"yasr_{uuid.uuid4().hex}"
        with self._lock:
            self._streams[stream_id] = {
                "session_id": str(session_id or "").strip(),
                "language_hint": str(language_hint or "").strip() or None,
                "pcm": bytearray(),
            }
        return stream_id

    def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None:
        if not pcm_8k_chunk:
            return
        with self._lock:
            stream = self._streams.get(stream_id)
            if stream is None:
                raise StreamingASRUnavailable("Yandex ASR stream is not active")
            pcm = stream.get("pcm")
            if isinstance(pcm, bytearray):
                pcm.extend(pcm_8k_chunk)

    def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
        del stream_id
        return None

    def finalize(self, stream_id: str) -> ASRTranscription:
        with self._lock:
            stream = self._streams.get(stream_id)
            if stream is None:
                raise StreamingASRUnavailable("Yandex ASR stream is not active")
            pcm = bytes(stream.get("pcm") or b"")
            language_hint = str(stream.get("language_hint") or "").strip() or None
        return self._asr_provider.transcribe(pcm, language_hint=language_hint)

    def close_stream(self, stream_id: str) -> None:
        with self._lock:
            self._streams.pop(stream_id, None)


def build_asr_provider(name: str) -> ASRProvider:
    normalized = str(name or "stub").strip().lower()
    if normalized == "elevenlabs":
        return ElevenLabsASRProvider()
    if normalized == "openai":
        return OpenAIASRProvider()
    if normalized in {"yandex", "yandex_speechkit", "speechkit"}:
        return YandexSpeechKitASRProvider()
    return ASRProvider()


def build_streaming_asr_provider(name: str) -> StreamingASRProvider:
    normalized = str(name or "disabled").strip().lower()
    if normalized in {
        "elevenlabs",
        "elevenlabs_realtime",
        "elevenlabs-realtime",
        "scribe_realtime",
        "scribe-v2-realtime",
        "scribe_v2_realtime",
    }:
        return ElevenLabsRealtimeStreamingASRProvider()
    if normalized in {"local_sidecar", "local-sidecar", "sidecar"}:
        return LocalSidecarStreamingASRProvider()
    if normalized in {"yandex", "yandex_speechkit", "yandex-speechkit", "speechkit", "yandex_grpc"}:
        return YandexSpeechKitGrpcStreamingASRProvider()
    if normalized in {"yandex_buffered", "yandex_speechkit_buffered", "yandex-speechkit-buffered"}:
        return YandexSpeechKitBufferedStreamingASRProvider()
    return StreamingASRProvider()
