from __future__ import annotations

import re
import unicodedata
import asyncio
import copy
import threading
import os
import io
import json
import uuid
import wave
import time
import logging
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional, Tuple

import httpx
from livekit import rtc
from livekit.agents.stt.stt import (
    STT,
    STTCapabilities,
    SpeechEvent,
    SpeechEventType,
    SpeechData,
    APIConnectOptions,
)

from app.utils.pcm import pcm_duration_ms, dbfs_pcm16le
from app.utils.dump import _dump_audio_for_debug
from app.utils.stt_audio_upload import enqueue_stt_vad_audio_upload
from app.providers.stt_config import override_config
from app.providers.stt_vosk import VoskSTT
from app.providers.stt_word_dictionary import (
    SHORT_WORD_ALLOWLIST,
    SHORT_WORD_ALLOWLIST_MIN_PROB,
)
from app.common.timed import timed
from app.utils.logging import (
    EVENT_RECEBIMENTO_MSG,
    event_latency_ms,
    log_flow_event,
    log_structured_event,
    setup_minimal_logging,
)
from app.utils.turn_ids import (
    clear_started_turn_message_id,
    next_turn_message_id,
    register_started_turn_message_id,
    register_transcribed_turn,
)

logger = setup_minimal_logging()

_HTTP_500_MAX_RETRIES = 1

# --- single-word rewrite (apenas para raw_json) ---
_SINGLE_WORD_REWRITE_MAP = {
    "bom": "Não",
    "bem": "Sim",
    "falou": "Alô",
}

_FINAL_PUNCT_RE = re.compile(r"[\s]*[.!?…,:;]+[\s]*$")

def _env_non_negative_int(name: str, default: int) -> int:
    try:
        return max(0, int(os.getenv(name, str(default))))
    except ValueError:
        return max(0, default)


def _json_for_log(payload: Any) -> str:
    try:
        return json.dumps(payload, ensure_ascii=False, default=str)
    except Exception:
        return str(payload)


def _prepend_pcm16le_silence(
    pcm: bytes,
    *,
    sample_rate: int,
    channels: int,
    padding_ms: int,
) -> tuple[bytes, int]:
    if not pcm or padding_ms <= 0:
        return pcm, 0
    sample_frames = round(sample_rate * (padding_ms / 1000.0))
    silence = b"\x00" * max(0, sample_frames) * max(1, channels) * 2
    if not silence:
        return pcm, 0
    actual_padding_ms = round(pcm_duration_ms(silence, sample_rate, channels))
    return silence + pcm, actual_padding_ms


def strip_final_punctuation(text: str) -> str:
    return _FINAL_PUNCT_RE.sub("", (text or "").strip())

def _normalize_token(s: str) -> str:
    s = strip_final_punctuation(s)  # remove pontuação final
    s = (s or "").strip().lower()
    s = unicodedata.normalize("NFD", s)
    s = "".join(ch for ch in s if unicodedata.category(ch) != "Mn")  # remove acentos
    return s


_NORMALIZED_SHORT_WORD_ALLOWLIST = frozenset(
    _normalize_token(word)
    for word in SHORT_WORD_ALLOWLIST
    if _normalize_token(word)
)


def _is_short_word_allowlisted(text: str) -> bool:
    return _normalize_token(text) in _NORMALIZED_SHORT_WORD_ALLOWLIST


def _short_word_allowlist_accepts(text: str, _probability: float) -> bool:
    return _is_short_word_allowlisted(text)


def apply_single_word_rewrite_for_raw_json(norm_payload: Dict[str, Any]) -> Dict[str, Any]:
    """
    Se o payload (já normalizado em {"data":{"text","words"}}) tiver 1 palavra,
    troca data.text conforme _SINGLE_WORD_REWRITE_MAP.
    """
    try:
        data = norm_payload.get("data") if isinstance(norm_payload.get("data"), dict) else {}
        text = (data.get("text") or "").strip()
        words = data.get("words") if isinstance(data.get("words"), list) else []

        # "retorno for 1": prioriza words.length, senão cai pra contagem de tokens do text
        if words:
            is_single = (len(words) == 1)
        else:
            is_single = (len([t for t in _normalize_token(text).split() if t]) == 1)

        if not is_single or not text:
            return norm_payload

        key = _normalize_token(text)
        replacement = _SINGLE_WORD_REWRITE_MAP.get(key)
        if replacement:
            data["text"] = replacement
            norm_payload["data"] = data

            # Se quiser manter consistência com words[0]["word"], descomente:
            # if isinstance(words, list) and len(words) == 1 and isinstance(words[0], dict):
            #     words[0]["word"] = f" {replacement}"
            #     data["words"] = words

        return norm_payload
    except Exception:
        return norm_payload


def extract_text_from_transcript(transcript: str) -> str:
    """
    Aceita:
      - texto puro: "oi tudo bem"
      - JSON string novo: {"data":{"text":"...","words":[...]}}
      - JSON string legado: {"text":"..."} ou {"data":{"text":"..."}}

    Retorna SEMPRE texto limpo (ou "" se não achar texto).
    """
    s = (transcript or "").strip()
    if not s:
        return ""

    if s[:1] in "{[":
        try:
            obj = json.loads(s)

            if isinstance(obj, dict):
                data = obj.get("data")
                if isinstance(data, dict):
                    t = data.get("text")
                    if isinstance(t, str) and t.strip():
                        return t.strip()

                t = obj.get("text")
                if isinstance(t, str) and t.strip():
                    return t.strip()

                return ""

            if isinstance(obj, list):
                for it in obj:
                    if isinstance(it, dict):
                        data = it.get("data")
                        if isinstance(data, dict):
                            t = data.get("text")
                            if isinstance(t, str) and t.strip():
                                return t.strip()
                        t = it.get("text")
                        if isinstance(t, str) and t.strip():
                            return t.strip()
                return ""
        except Exception:
            pass

    return s


def normalize_to_data_words_format(payload: Dict[str, Any]) -> Dict[str, Any]:
    """
    Normaliza diferentes formatos para:
      {"data": {"text": "...", "words": [{"word","start","end","probability"}]}}

    Suporta:
      - {"data": {"text": "...", "words":[{"probability":...}]}}
      - {"text": "...", "words":[...]}  (move pra dentro de data)
      - {"text": "...", "result":[{"word","start","end","conf"}]} (Vosk padrão)
      - {"text": "...", "Probabilites"/"Probabilities":[{"words":[...]}]} (legado)
    """
    if not isinstance(payload, dict):
        return {"data": {"text": "", "words": []}}

    # já é novo
    if isinstance(payload.get("data"), dict):
        data = payload["data"]
        text = (data.get("text") or "").strip()
        words = data.get("words") if isinstance(data.get("words"), list) else []
        return {"data": {"text": text, "words": words}}

    root = payload
    text = (root.get("text") or "").strip()

    # words direto
    if isinstance(root.get("words"), list):
        return {"data": {"text": text, "words": root["words"]}}

    # vosk "result"
    if isinstance(root.get("result"), list):
        out_words = []
        for w in root["result"]:
            if not isinstance(w, dict):
                continue
            out_words.append({
                "word": w.get("word", ""),
                "start": float(w.get("start", 0.0) or 0.0),
                "end": float(w.get("end", 0.0) or 0.0),
                "probability": float(w.get("conf", w.get("probability", 0.0)) or 0.0),
            })
        return {"data": {"text": text, "words": out_words}}

    # legado Probabilites/Probabilities
    probs = (
        root.get("Probabilites")
        or root.get("Probabilities")
        or root.get("probabilites")
        or root.get("probabilities")
        or []
    )
    if isinstance(probs, list) and probs:
        first = probs[0] if isinstance(probs[0], dict) else {}
        words = first.get("words") if isinstance(first.get("words"), list) else []
        seg_text = (first.get("text") or text or "").strip()
        return {"data": {"text": seg_text, "words": words}}

    return {"data": {"text": text, "words": []}}


def stt_text_with_single_word_threshold(
    payload: Dict[str, Any],
    min_prob_single_word: float = 0.10,
) -> Optional[str]:
    """
    Regra:
      - allowlist passa independente da probabilidade
      - 1 palavra: só aceita se prob > min_prob_single_word
      - >1 palavra: aceita
      - sem words: aceita text
    """
    try:
        norm = normalize_to_data_words_format(payload)
        data = norm.get("data") if isinstance(norm.get("data"), dict) else {}
        text = (data.get("text") or "").strip()
        if not text:
            return None

        words = data.get("words") if isinstance(data.get("words"), list) else []
        if not words:
            return text

        if len(words) == 1:
            prob = words[0].get("probability", 0.0)
            try:
                prob = float(prob)
            except Exception:
                prob = 0.0
            if prob > min_prob_single_word:
                return text
            if _short_word_allowlist_accepts(text, prob):
                return text
            return None

        return text
    except Exception:
        return None


@dataclass
class InternalSTTConfig:
    url: str
    api_key: str
    language: str = "pt"
    sample_rate: int = 16000
    channels: int = 1
    timeout_s: float = 30.0
    min_prob_single_word: float = 0.10
    initial_prompt: str = ""
    config_override: str = ""
    output_mode: str = "threshold_text"  # "threshold_text" | "raw_json" | "api_text"


def _normalize_override_config_json(raw_config: str) -> str:
    text = (raw_config or "").strip()
    if not text:
        return ""

    payload = json.loads(text)
    if not isinstance(payload, dict):
        raise ValueError("STT config override must be a JSON object")
    return json.dumps(payload, ensure_ascii=False)


class InternalHTTPSTT(STT):
    """
    STT batch (não-streaming).
    O LiveKit Agents vai usar VAD/turn_detection para segmentar e chamar recognize() por trecho.
    """

    def __init__(
        self,
        cfg: InternalSTTConfig,
        *,
        client: httpx.AsyncClient,
        vosk: Optional[VoskSTT] = None,
        vosk_lock: Optional[threading.Lock] = None,
        timeline: Any | None = None,
        structured_log_context: Any | None = None,
        structured_logger: Any | None = None,
        structured_interruption_flag: Callable[[], bool] | None = None,
        empty_transcript_handler: Callable[[], None] | None = None,
        metrics_handler: Callable[[dict[str, Any]], None] | None = None,
    ):
        super().__init__(capabilities=STTCapabilities(streaming=False, interim_results=False))
        self._cfg = cfg
        self._client = client
        self._vosk = vosk
        self._vosk_lock = vosk_lock
        self._timeline = timeline
        self._structured_log_context = structured_log_context
        self._structured_logger = structured_logger
        self._structured_interruption_flag = structured_interruption_flag
        self._empty_transcript_handler = empty_transcript_handler
        self._metrics_handler = metrics_handler
        self._resampler: Optional[rtc.AudioResampler] = None

    def set_empty_transcript_handler(self, handler: Callable[[], None] | None) -> None:
        self._empty_transcript_handler = handler

    def set_metrics_handler(
        self, handler: Callable[[dict[str, Any]], None] | None
    ) -> None:
        self._metrics_handler = handler

    def _notify_metrics(self, **fields: Any) -> None:
        handler = self._metrics_handler
        if handler is None:
            return
        try:
            handler(fields)
        except Exception:
            logger.exception("[stt][metrics_handler_error]")

    def _notify_empty_transcript(self, *, request_id: str, source: str) -> None:
        handler = self._empty_transcript_handler
        if handler is None:
            return
        try:
            handler()
        except Exception:
            logger.exception(
                "[stt][empty_transcript_handler_error] req_id=%s source=%s",
                request_id,
                source,
            )

    def _structured_interruption_value(self) -> int:
        if self._structured_interruption_flag is None:
            return 0
        try:
            return 1 if bool(self._structured_interruption_flag()) else 0
        except Exception:
            return 0

    def _log_recebimento_msg(
        self,
        *,
        request_id: str,
        started_ns: int,
        message_id: str = "",
        finished_ns: int | None = None,
        erro_msg: str | None = None,
        erro_detalhe: str | None = None,
        http_cod_status: int | None = None,
        http_cod_desc: str | None = None,
    ) -> str:
        resolved_message_id = str(message_id or "").strip() or next_turn_message_id(
            self._structured_log_context
        )
        register_started_turn_message_id(self._structured_log_context, resolved_message_id)
        if self._structured_logger is None:
            return resolved_message_id

        ended_ns = int(finished_ns if finished_ns is not None else time.time_ns())
        total_ms = event_latency_ms(started_ns, ended_ns)
        log_structured_event(
            self._structured_logger,
            self._structured_log_context,
            tipo_evento=EVENT_RECEBIMENTO_MSG,
            message_id=resolved_message_id,
            inicio_ns=started_ns,
            fim_ns=ended_ns,
            latencia_total_ms=total_ms,
            latencia_tffb_ms=total_ms,
            interrupcao=self._structured_interruption_value(),
            erro_msg=erro_msg,
            erro_detalhe=erro_detalhe,
            http_cod_status=http_cod_status,
            http_cod_desc=http_cod_desc,
        )
        return resolved_message_id

    def _log_detected_text(
        self,
        *,
        request_id: str,
        output_text: str,
        engine: str,
        mode: str | None = None,
        duration_ms: float | int | None = None,
        audio_duration_ms: float | int | None = None,
        level_dbfs: float | None = None,
    ) -> None:
        detected_text = extract_text_from_transcript(output_text)
        log_flow_event(
            self._structured_logger or logger,
            "stt_done",
            request_id=request_id,
            engine=engine,
            mode=mode or self._stt_output_mode(),
            returned=True,
            duration_ms=round(float(duration_ms)) if duration_ms is not None else None,
            audio_ms=round(float(audio_duration_ms)) if audio_duration_ms is not None else None,
            dbfs=round(float(level_dbfs), 2) if level_dbfs is not None else None,
            text_len=len(detected_text),
            text=detected_text,
        )

    def _register_transcribed_turn(self, *, message_id: str, transcription: str) -> None:
        detected_text = extract_text_from_transcript(transcription)
        if not detected_text:
            return

        register_transcribed_turn(
            self._structured_log_context,
            message_id=message_id,
            transcription=transcription,
            text=detected_text,
        )

    def _stt_output_mode(self) -> str:
        return (os.getenv("STT_OUTPUT_MODE", self._cfg.output_mode) or "threshold_text").strip().lower()

    def _format_stt_output(self, payload: Dict[str, Any], *, request_id: str = "") -> str:
        """
        Converte o payload para uma string, dependendo do modo:

        - raw_json: devolve JSON string no formato {"data":{...}}
        - api_text: devolve apenas data.text
        - threshold_text: aplica regra de 1 palavra com prob
        """
        mode = self._stt_output_mode()

        norm = normalize_to_data_words_format(payload)
        data = norm.get("data") if isinstance(norm.get("data"), dict) else {}
        api_text = (data.get("text") or "").strip()
        words = data.get("words") if isinstance(data.get("words"), list) else []
        single_word_prob = None
        if len(words) == 1 and isinstance(words[0], dict):
            try:
                single_word_prob = float(words[0].get("probability", 0.0) or 0.0)
            except Exception:
                single_word_prob = 0.0

        if mode == "raw_json":
            logger.info("[stt][raw_json_before_rewrite] %s", norm)
            norm = apply_single_word_rewrite_for_raw_json(norm)
            norm["STT"] = "Sofya"
            logger.info("[stt][raw_json_after_rewrite] %s", norm)
            return json.dumps(norm, ensure_ascii=False)


        if mode == "api_text":
            return (data.get("text") or "").strip()

        out_text = stt_text_with_single_word_threshold(norm, self._cfg.min_prob_single_word) or ""
        filter_reason = "accepted"
        if (
            out_text
            and len(words) == 1
            and single_word_prob is not None
            and single_word_prob <= self._cfg.min_prob_single_word
            and _short_word_allowlist_accepts(api_text, single_word_prob)
        ):
            filter_reason = "single_word_allowlist_low_confidence"
        if not out_text:
            if not api_text:
                filter_reason = "api_text_empty"
            elif len(words) == 1 and single_word_prob is not None:
                filter_reason = "single_word_below_threshold"
            else:
                filter_reason = "threshold_text_empty"

        log_flow_event(
            self._structured_logger or logger,
            "stt_payload",
            request_id=request_id or None,
            mode=mode,
            api_text=api_text,
            api_text_len=len(api_text),
            words_count=len(words),
            min_prob_single_word=self._cfg.min_prob_single_word,
            single_word_prob=(
                f"{single_word_prob:.3f}" if single_word_prob is not None else None
            ),
            allowlist_min_prob=SHORT_WORD_ALLOWLIST_MIN_PROB,
            filter_reason=filter_reason,
        )
        return out_text

    def _build_override_config_json(self) -> str:
        forced_override = (os.getenv("STT_FORCE_OVERRIDE_CONFIG", "") or "").strip()
        if forced_override:
            return _normalize_override_config_json(forced_override)

        inline_override = (self._cfg.config_override or "").strip()
        if inline_override:
            return _normalize_override_config_json(inline_override)

        base = override_config
        ovr = copy.deepcopy(base)

        proc = ovr.setdefault("processor", {}).setdefault("config", {}).setdefault("extra_params", {})

        forced = (os.getenv("STT_FORCE_INITIAL_PROMPT", "").strip() or "").strip()
        prompt = forced or (self._cfg.initial_prompt or "").strip() or "Aguarda confimação de números do cpf"
        proc["initial_prompt"] = prompt

        return json.dumps(ovr, ensure_ascii=False)

    async def _try_vosk(self, pcm: bytes, req_id: str) -> Tuple[str, Dict[str, Any]]:
        """
        Roda Vosk em thread e serializa com lock.
        Retorna: (texto, payload_padronizado_em_data)
        """
        if self._vosk is None:
            return "", {}
        if self._timeline is not None:
            self._timeline.emit(
                "stt_vosk_started",
                request_id=req_id,
                pcm_bytes=len(pcm),
            )

        lock = self._vosk_lock

        def _run() -> Tuple[str, Dict[str, Any]]:
            if lock is not None:
                with lock:
                    raw = self._vosk.transcribe_pcm16le(pcm)
            else:
                raw = self._vosk.transcribe_pcm16le(pcm)

            # raw já vem padronizado se seu VoskSTT estiver atualizado,
            # mas garantimos aqui também:
            txt, payload = VoskSTT.result_to_text_and_payload(raw)
            payload = normalize_to_data_words_format(payload)
            return (txt or "").strip(), payload

        try:
            text, payload = await asyncio.to_thread(_run)
            if self._timeline is not None:
                self._timeline.emit(
                    "stt_vosk_completed",
                    request_id=req_id,
                    text=text,
                    text_len=len(text),
                    has_payload=bool(payload),
                )
            return text, payload
        except Exception:
            logger.exception("[stt][vosk_error] req_id=%s", req_id)
            if self._timeline is not None:
                self._timeline.emit("stt_vosk_failed", request_id=req_id)
            return "", {}

    async def _post_internal_http(
        self,
        pcm: bytes,
        wav_bytes: bytes,
        req_id: str,
        started_ns: int,
        language: Optional[str],
        message_id: str = "",
        audio_duration_ms: float | int | None = None,
        original_audio_duration_ms: float | int | None = None,
        input_padding_ms: float | int | None = None,
        level_dbfs: float | None = None,
    ) -> SpeechEvent:
        files = {"file": ("audio.wav", wav_bytes, "audio/wav")}
        headers = {"x-api-key": self._cfg.api_key, "Connection": "close"}
        data = {"override_config": self._build_override_config_json()}
        timeout = httpx.Timeout(self._cfg.timeout_s)

        if self._timeline is not None:
            self._timeline.emit(
                "stt_http_started",
                request_id=req_id,
                url=self._cfg.url,
            )
        log_flow_event(
            self._structured_logger or logger,
            "stt_dispatch",
            request_id=req_id,
            engine="http",
            audio_ms=round(float(audio_duration_ms)) if audio_duration_ms is not None else None,
            dbfs=round(float(level_dbfs), 2) if level_dbfs is not None else None,
        )
        attempt = 0
        try:
            while True:
                t0 = time.perf_counter()
                r = await self._client.post(
                    self._cfg.url,
                    files=files,
                    data=data,
                    headers=headers,
                    timeout=timeout,
                )
                dt_ms = (time.perf_counter() - t0) * 1000.0
                try:
                    r.raise_for_status()
                    finished_ns = time.time_ns()
                    break
                except httpx.HTTPStatusError as e:
                    http_status = getattr(e.response, "status_code", None)
                    if http_status == 500 and attempt < _HTTP_500_MAX_RETRIES:
                        attempt += 1
                        logger.warning(
                            "[stt][http_500_retry] req_id=%s attempt=%s/%s took=%.0fms url=%s",
                            req_id,
                            attempt,
                            _HTTP_500_MAX_RETRIES,
                            dt_ms,
                            str(getattr(e.request, "url", self._cfg.url)),
                        )
                        log_flow_event(
                            self._structured_logger or logger,
                            "stt_http_retry",
                            request_id=req_id,
                            engine="http",
                            status=http_status,
                            attempt=attempt,
                            max_retries=_HTTP_500_MAX_RETRIES,
                            duration_ms=round(dt_ms),
                        )
                        if self._timeline is not None:
                            self._timeline.emit(
                                "stt_http_retry",
                                request_id=req_id,
                                status=http_status,
                                attempt=attempt,
                                max_retries=_HTTP_500_MAX_RETRIES,
                                duration_ms=round(dt_ms),
                            )
                        continue
                    raise
        except httpx.HTTPStatusError as e:
            dt_ms = (time.perf_counter() - t0) * 1000.0
            http_status = getattr(e.response, "status_code", None)
            http_desc = str(getattr(e.response, "reason_phrase", "") or "").strip() or None
            body_preview = ""
            try:
                body_preview = (e.response.text or "")[:800]
            except Exception:
                pass
            logger.error(
                "[stt][http_error] req_id=%s status=%s took=%.0fms url=%s body(800)=%s",
                req_id,
                getattr(e.response, "status_code", "NA"),
                dt_ms,
                str(getattr(e.request, "url", self._cfg.url)),
                body_preview,
            )
            log_flow_event(
                self._structured_logger or logger,
                "stt_error",
                request_id=req_id,
                engine="http",
                status=http_status or "NA",
                duration_ms=round(dt_ms),
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "stt_http_failed",
                    request_id=req_id,
                    status=getattr(e.response, "status_code", "NA"),
                    duration_ms=round(dt_ms),
                )
            self._log_recebimento_msg(
                request_id=req_id,
                started_ns=started_ns,
                message_id=message_id,
                erro_msg="Falha STT",
                erro_detalhe=(
                    f"http_status={http_status if http_status is not None else 'NA'} "
                    f"| url={str(getattr(e.request, 'url', self._cfg.url))} "
                    f"| body={body_preview}"
                ),
                http_cod_status=http_status,
                http_cod_desc=http_desc,
            )
            self._notify_metrics(
                event="failed",
                request_id=req_id,
                provider="internal_http",
                duration_ms=round(dt_ms),
                audio_duration_ms=round(float(audio_duration_ms)) if audio_duration_ms is not None else None,
                original_audio_duration_ms=(
                    round(float(original_audio_duration_ms))
                    if original_audio_duration_ms is not None
                    else None
                ),
                input_padding_ms=round(float(input_padding_ms)) if input_padding_ms is not None else None,
                input_dbfs=round(float(level_dbfs), 2) if level_dbfs is not None else None,
                retry_count=attempt,
                http_status=http_status,
                error="http_error",
            )
            if http_status == 500:
                log_flow_event(
                    self._structured_logger or logger,
                    "stt_error_nonfatal",
                    request_id=req_id,
                    engine="http",
                    status=http_status,
                    action="return_empty_transcript",
                    max_retries=_HTTP_500_MAX_RETRIES,
                )
                if self._timeline is not None:
                    self._timeline.emit(
                        "stt_http_failed_nonfatal",
                        request_id=req_id,
                        status=http_status,
                        max_retries=_HTTP_500_MAX_RETRIES,
                    )
                return SpeechEvent(
                    type=SpeechEventType.FINAL_TRANSCRIPT,
                    request_id=req_id,
                    alternatives=[
                        SpeechData(
                            language=language or self._cfg.language,
                            text="",
                            confidence=0.0,
                        )
                    ],
                )
            raise
        except Exception:
            dt_ms = (time.perf_counter() - t0) * 1000.0
            logger.exception("[stt][error] req_id=%s took=%.0fms", req_id, dt_ms)
            log_flow_event(
                self._structured_logger or logger,
                "stt_error",
                request_id=req_id,
                engine="http",
                status="exception",
                duration_ms=round(dt_ms),
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "stt_http_failed",
                    request_id=req_id,
                    status="exception",
                    duration_ms=round(dt_ms),
                )
            self._log_recebimento_msg(
                request_id=req_id,
                started_ns=started_ns,
                message_id=message_id,
                erro_msg="Falha STT",
                erro_detalhe=f"exception | took={round(dt_ms)}ms",
            )
            self._notify_metrics(
                event="failed",
                request_id=req_id,
                provider="internal_http",
                duration_ms=round(dt_ms),
                audio_duration_ms=round(float(audio_duration_ms)) if audio_duration_ms is not None else None,
                original_audio_duration_ms=(
                    round(float(original_audio_duration_ms))
                    if original_audio_duration_ms is not None
                    else None
                ),
                input_padding_ms=round(float(input_padding_ms)) if input_padding_ms is not None else None,
                input_dbfs=round(float(level_dbfs), 2) if level_dbfs is not None else None,
                retry_count=attempt,
                http_status=None,
                error="exception",
            )
            raise

        payload = r.json()
        payload_json = _json_for_log(payload)
        logger.info(
            "[stt][response_json] req_id=%s status=%s took=%.0fms json=%s",
            req_id,
            r.status_code,
            dt_ms,
            payload_json,
        )
        out_text = self._format_stt_output(payload, request_id=req_id)
        if self._timeline is not None:
            self._timeline.emit(
                "stt_http_completed",
                request_id=req_id,
                duration_ms=round(dt_ms),
                text=out_text,
                text_len=len(out_text),
            )
        empty_transcript = not bool(extract_text_from_transcript(out_text))
        self._notify_metrics(
            event="completed",
            request_id=req_id,
            provider="internal_http",
            mode=self._stt_output_mode(),
            duration_ms=round(dt_ms),
            audio_duration_ms=round(float(audio_duration_ms)) if audio_duration_ms is not None else None,
            original_audio_duration_ms=(
                round(float(original_audio_duration_ms))
                if original_audio_duration_ms is not None
                else None
            ),
            input_padding_ms=round(float(input_padding_ms)) if input_padding_ms is not None else None,
            input_dbfs=round(float(level_dbfs), 2) if level_dbfs is not None else None,
            retry_count=attempt,
            http_status=r.status_code,
            empty_transcript=empty_transcript,
            text_length=len(extract_text_from_transcript(out_text)),
        )
        if empty_transcript:
            # Um resultado sem texto nao constitui turno de conversa. Nao publique
            # recebimento msg nem deixe o identificador tecnico do upload ocupar o
            # estado de message_id ativo, pois uma fala proativa do agente poderia
            # herda-lo antes da chegada de user_input_transcribed.
            clear_started_turn_message_id(
                self._structured_log_context,
                message_id=message_id,
            )
        else:
            message_id = self._log_recebimento_msg(
                request_id=req_id,
                started_ns=started_ns,
                message_id=message_id,
                finished_ns=finished_ns,
                erro_detalhe=None,
                http_cod_status=r.status_code,
                http_cod_desc=str(getattr(r, "reason_phrase", "") or "").strip() or None,
            )
            self._register_transcribed_turn(message_id=message_id, transcription=out_text)
        self._log_detected_text(
            request_id=req_id,
            output_text=out_text,
            engine="http",
            duration_ms=dt_ms,
            audio_duration_ms=audio_duration_ms,
            level_dbfs=level_dbfs,
        )
        if empty_transcript:
            self._notify_empty_transcript(request_id=req_id, source="http_success")

        # log reduzido
        preview = out_text if len(out_text) < 250 else out_text[:250] + "..."

        return SpeechEvent(
            type=SpeechEventType.FINAL_TRANSCRIPT,
            request_id=req_id,
            alternatives=[SpeechData(language=language or self._cfg.language, text=out_text, confidence=0.0)],
        )

    async def aclose(self) -> None:
        return

    def _frames_to_pcm16le_16k(self, buffer) -> bytes:
        """
        Junta frames e garante PCM16 mono 16k (re-amostra se precisar).
        """
        if isinstance(buffer, rtc.AudioFrame):
            frames = [buffer]
        else:
            frames = list(buffer)

        if not frames:
            return b""

        in_sr = frames[0].sample_rate
        in_ch = frames[0].num_channels

        target_sr = self._cfg.sample_rate
        target_ch = self._cfg.channels

        # resample se necessário
        if in_sr != target_sr:
            if self._resampler is None:
                self._resampler = rtc.AudioResampler(in_sr, target_sr, quality=rtc.AudioResamplerQuality.HIGH)
            out_frames = []
            for f in frames:
                out_frames.extend(self._resampler.push(f))
            out_frames.extend(self._resampler.flush())
            frames = out_frames

        # canais (se precisar no futuro, faz stereo->mono aqui)
        if in_ch != target_ch:
            pass

        return b"".join(bytes(f.data) for f in frames)

    @timed("STT")
    async def _recognize_impl(
        self,
        buffer,
        *,
        language: str | None,
        conn_options: APIConnectOptions,
    ) -> SpeechEvent:
        started_ns = time.time_ns()
        pcm = self._frames_to_pcm16le_16k(buffer)
        req_id = uuid.uuid4().hex

        min_ms = int(os.getenv("STT_MIN_AUDIO_MS", "160"))
        min_dbfs = float(os.getenv("STT_MIN_DBFS", "-50.0"))

        dur_ms = pcm_duration_ms(pcm, self._cfg.sample_rate, self._cfg.channels)
        level_dbfs = dbfs_pcm16le(pcm)
        if self._timeline is not None:
            self._timeline.emit(
                "stt_recognize_started",
                request_id=req_id,
                duration_ms=round(dur_ms),
                level_dbfs=round(level_dbfs, 2),
            )
        if not pcm or dur_ms < min_ms or level_dbfs < min_dbfs:
            if (os.getenv("FLOW_LOG_STT_SKIPS", "0") or "0").strip() == "1":
                log_flow_event(
                    self._structured_logger or logger,
                    "stt_skip",
                    request_id=req_id,
                    reason="too_short_or_silent",
                    audio_ms=round(dur_ms),
                    dbfs=round(level_dbfs, 2),
                )
            if self._timeline is not None:
                self._timeline.emit(
                    "stt_recognize_skipped",
                    request_id=req_id,
                    reason="too_short_or_silent",
                    duration_ms=round(dur_ms),
                    level_dbfs=round(level_dbfs, 2),
                )
            return SpeechEvent(
                type=SpeechEventType.FINAL_TRANSCRIPT,
                request_id=req_id,
                alternatives=[SpeechData(language=language or self._cfg.language, text="", confidence=0.0)],
            )
        log_flow_event(
            self._structured_logger or logger,
            "stt_start",
            request_id=req_id,
            audio_ms=round(dur_ms),
            dbfs=round(level_dbfs, 2),
        )

        # tenta Vosk para trechos curtos
        vosk_max_ms = int(os.getenv("STT_VOSK_MAX_MS", "2000"))
        if dur_ms < vosk_max_ms:
            vosk_started = time.perf_counter()
            txt_vosk, payload_vosk = await self._try_vosk(pcm, req_id=req_id)
            vosk_duration_ms = (time.perf_counter() - vosk_started) * 1000.0

            if payload_vosk:
                mode = self._stt_output_mode()

                if mode == "raw_json":
                    payload_vosk = apply_single_word_rewrite_for_raw_json(payload_vosk)
                    payload_vosk["STT"] = "Vosk"
                    out_text = json.dumps(payload_vosk, ensure_ascii=False)
                    if self._timeline is not None:
                        self._timeline.emit(
                            "stt_recognize_completed",
                            request_id=req_id,
                            engine="vosk",
                            mode=mode,
                            text=out_text,
                            text_len=len(out_text),
                        )
                    message_id = self._log_recebimento_msg(
                        request_id=req_id,
                        started_ns=started_ns,
                    )
                    self._register_transcribed_turn(message_id=message_id, transcription=out_text)
                    self._log_detected_text(
                        request_id=req_id,
                        output_text=out_text,
                        engine="vosk",
                        mode=mode,
                        duration_ms=vosk_duration_ms,
                        audio_duration_ms=dur_ms,
                        level_dbfs=level_dbfs,
                    )
                    return SpeechEvent(
                        type=SpeechEventType.FINAL_TRANSCRIPT,
                        request_id=req_id,
                        alternatives=[SpeechData(language=language or self._cfg.language, text=out_text, confidence=0.0)],
                    )

                if mode == "api_text":
                    data = payload_vosk.get("data") if isinstance(payload_vosk.get("data"), dict) else {}
                    out_text = (data.get("text") or "").strip()
                    if out_text:
                        if self._timeline is not None:
                            self._timeline.emit(
                                "stt_recognize_completed",
                                request_id=req_id,
                                engine="vosk",
                                mode=mode,
                                text=out_text,
                                text_len=len(out_text),
                            )
                        message_id = self._log_recebimento_msg(
                            request_id=req_id,
                            started_ns=started_ns,
                        )
                        self._register_transcribed_turn(message_id=message_id, transcription=out_text)
                        self._log_detected_text(
                            request_id=req_id,
                            output_text=out_text,
                            engine="vosk",
                            mode=mode,
                            duration_ms=vosk_duration_ms,
                            audio_duration_ms=dur_ms,
                            level_dbfs=level_dbfs,
                        )
                        return SpeechEvent(
                            type=SpeechEventType.FINAL_TRANSCRIPT,
                            request_id=req_id,
                            alternatives=[SpeechData(language=language or self._cfg.language, text=out_text, confidence=0.0)],
                        )

                # threshold_text
                filtered = stt_text_with_single_word_threshold(payload_vosk, self._cfg.min_prob_single_word)
                if filtered:
                    if self._timeline is not None:
                        self._timeline.emit(
                            "stt_recognize_completed",
                            request_id=req_id,
                            engine="vosk",
                            mode="threshold_text",
                            text=filtered,
                            text_len=len(filtered),
                        )
                    message_id = self._log_recebimento_msg(
                        request_id=req_id,
                        started_ns=started_ns,
                    )
                    self._register_transcribed_turn(message_id=message_id, transcription=filtered)
                    self._log_detected_text(
                        request_id=req_id,
                        output_text=filtered,
                        engine="vosk",
                        mode="threshold_text",
                        duration_ms=vosk_duration_ms,
                        audio_duration_ms=dur_ms,
                        level_dbfs=level_dbfs,
                    )
                    return SpeechEvent(
                        type=SpeechEventType.FINAL_TRANSCRIPT,
                        request_id=req_id,
                        alternatives=[SpeechData(language=language or self._cfg.language, text=filtered, confidence=0.0)],
                    )

        input_padding_ms = _env_non_negative_int("STT_INPUT_PREFIX_PADDING_MS", 250)
        http_pcm, applied_input_padding_ms = _prepend_pcm16le_silence(
            pcm,
            sample_rate=self._cfg.sample_rate,
            channels=self._cfg.channels,
            padding_ms=input_padding_ms,
        )
        http_dur_ms = pcm_duration_ms(http_pcm, self._cfg.sample_rate, self._cfg.channels)
        http_level_dbfs = dbfs_pcm16le(http_pcm)
        if applied_input_padding_ms:
            log_flow_event(
                self._structured_logger or logger,
                "stt_input_padding",
                request_id=req_id,
                padding_ms=applied_input_padding_ms,
                original_audio_ms=round(dur_ms),
                sent_audio_ms=round(http_dur_ms),
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "stt_input_padding",
                    request_id=req_id,
                    padding_ms=applied_input_padding_ms,
                    original_audio_ms=round(dur_ms),
                    sent_audio_ms=round(http_dur_ms),
                )

        wav_buf = io.BytesIO()
        with wave.open(wav_buf, "wb") as wf:
            wf.setnchannels(self._cfg.channels)
            wf.setsampwidth(2)
            wf.setframerate(self._cfg.sample_rate)
            wf.writeframes(http_pcm)
        wav_bytes = wav_buf.getvalue()

        _dump_audio_for_debug(
            req_id=req_id,
            pcm=http_pcm,
            wav_bytes=wav_bytes,
            sample_rate=self._cfg.sample_rate,
            channels=self._cfg.channels,
        )

        message_id = next_turn_message_id(self._structured_log_context)
        enqueue_stt_vad_audio_upload(
            wav_bytes=wav_bytes,
            req_id=req_id,
            message_id=message_id,
            structured_log_context=self._structured_log_context,
            metadata={
                "sample_rate": self._cfg.sample_rate,
                "channels": self._cfg.channels,
                "original_audio_ms": round(dur_ms),
                "sent_audio_ms": round(http_dur_ms),
                "original_dbfs": round(level_dbfs, 2),
                "sent_dbfs": round(http_level_dbfs, 2),
                "input_padding_ms": applied_input_padding_ms,
            },
            logger_override=self._structured_logger or logger,
            timeline=self._timeline,
        )

        # fallback: API interna
        return await self._post_internal_http(
            http_pcm,
            wav_bytes,
            req_id=req_id,
            started_ns=started_ns,
            language=language,
            message_id=message_id,
            audio_duration_ms=http_dur_ms,
            original_audio_duration_ms=dur_ms,
            input_padding_ms=applied_input_padding_ms,
            level_dbfs=http_level_dbfs,
        )
