from __future__ import annotations

import asyncio
import json
import os
import time
from dataclasses import dataclass
from typing import Any, Dict, Mapping, Optional
from urllib.parse import urlsplit, urlunsplit

import aiohttp
import httpx

from app.common.call_config import (
    normalize_call_config,
    resolve_agent_backend_name,
    resolve_stt_overrides,
    resolve_tts_overrides,
)
from app.livekit.adapters.azure_rest_tts import AzureRESTTTS
from app.livekit.adapters.xai_tts import OraclexAITTS
from app.livekit.azure_speech import (
    AZURE_SPEECH_TTS_SAMPLE_RATE,
    resolve_azure_speech_tts_config,
)
from app.providers.stt_fake import FakeSTT
from app.providers.tts import FakeTTS as FakeProviderTTS


STOP_STATUS_BY_RESOURCE = {
    "capacity": "stop_capacity_tia",
    "agent_runtime": "stop_agent_runtime_unavailable",
    "agent_backend": "stop_agent_backend_unavailable",
    "stt": "stop_stt_unavailable",
    "tts": "stop_tts_unavailable",
}
STOP_RESOURCE_PRECEDENCE = (
    "capacity",
    "agent_runtime",
    "agent_backend",
    "stt",
    "tts",
)
HEALTHY_CHECK_STATUSES = {"ok", "skipped"}
PROBE_TEXT = "teste de prontidao"
XAI_TTS_READINESS_MODE_ENV = "XAI_TTS_READINESS_MODE"
XAI_TTS_READINESS_MODES = {"connect", "synthesize"}
DEFAULT_FINAL_STOP_STATUS_BY_REASON = {
    "resolved": "stop_resolvido_e_finalizado",
    "unresolved": "stop_nao_resolvido",
    "other_subject": "stop_outro_assunto",
    "long_silence": "stop_silencio_longo",
}
DEFAULT_FINAL_STOP_REASON_BY_KIND = {
    "resolved": "stage_done",
    "unresolved": "nao_resolvido",
    "other_subject": "outro_assunto",
    "long_silence": "no_user_response",
}


def build_stop_message(
    *,
    status: str,
    reason: str,
    resource: str = "",
    failed_resources: Optional[list[str]] = None,
    phase: str = "",
) -> Dict[str, Any]:
    data: Dict[str, Any] = {
        "status": str(status or "").strip(),
        "reason": str(reason or "").strip(),
    }
    if resource:
        data["resource"] = str(resource).strip()
    if failed_resources:
        data["failed_resources"] = [str(item).strip() for item in failed_resources if str(item).strip()]
    if phase:
        data["phase"] = str(phase).strip()
    return {"type": "stop", "data": data}


def _final_stop_status_from_env(key: str) -> str:
    env_name = f"FINAL_STOP_STATUS_{key.upper()}"
    return (
        os.getenv(env_name, DEFAULT_FINAL_STOP_STATUS_BY_REASON[key]) or DEFAULT_FINAL_STOP_STATUS_BY_REASON[key]
    ).strip()


def _format_probe_error(exc: Exception, *, resource: str, timeout_s: float) -> str:
    """Return a useful, secret-free error summary for readiness failures."""
    error_type = type(exc).__name__
    detail = _strip_text(str(exc))
    if isinstance(exc, TimeoutError):
        return f"{resource} readiness timed out after {round(timeout_s * 1000)}ms ({error_type})"
    if detail:
        return f"{resource} readiness failed ({error_type}): {detail}"
    return f"{resource} readiness failed ({error_type})"


def _final_stop_reason_from_env(key: str) -> str:
    env_name = f"FINAL_STOP_REASON_{key.upper()}"
    return (
        os.getenv(env_name, DEFAULT_FINAL_STOP_REASON_BY_KIND[key]) or DEFAULT_FINAL_STOP_REASON_BY_KIND[key]
    ).strip()


def resolve_completed_stop_status(reason: str) -> str:
    normalized_reason = str(reason or "").strip().lower()
    for key in DEFAULT_FINAL_STOP_STATUS_BY_REASON:
        configured_reason = _final_stop_reason_from_env(key).lower()
        if normalized_reason and normalized_reason == configured_reason:
            return _final_stop_status_from_env(key)

    default_key = (
        os.getenv("FINAL_STOP_DEFAULT_KIND", "resolved") or "resolved"
    ).strip().lower()
    if default_key not in DEFAULT_FINAL_STOP_STATUS_BY_REASON:
        default_key = "resolved"
    return _final_stop_status_from_env(default_key)


def build_completed_stop_message(reason: str) -> Dict[str, Any]:
    resolved_reason = str(reason or "").strip() or _final_stop_reason_from_env("resolved")
    return build_stop_message(
        status=resolve_completed_stop_status(resolved_reason),
        reason=resolved_reason,
        phase="in_session",
    )


def build_capacity_stop_message() -> Dict[str, Any]:
    return build_stop_message(
        status=STOP_STATUS_BY_RESOURCE["capacity"],
        reason="capacity_exceeded",
        resource="capacity",
        failed_resources=["capacity"],
        phase="pre_ready",
    )


def build_bridge_failed_stop_message() -> Dict[str, Any]:
    return build_stop_message(
        status="stop_bridge_failed",
        reason="bridge_failed",
        resource="bridge",
        phase="in_session",
    )


def _build_resource_stop_message(failed_resources: list[str]) -> Dict[str, Any]:
    normalized = [name for name in STOP_RESOURCE_PRECEDENCE if name in failed_resources and name != "capacity"]
    if not normalized:
        raise ValueError("failed_resources must contain at least one non-capacity resource")
    primary = normalized[0]
    return build_stop_message(
        status=STOP_STATUS_BY_RESOURCE[primary],
        reason="resource_unhealthy",
        resource=primary,
        failed_resources=normalized,
        phase="pre_ready",
    )


def _env_int(name: str, default: int) -> int:
    raw = (os.getenv(name, "") or "").strip()
    if not raw:
        return default
    try:
        return int(raw)
    except ValueError:
        return default


def _env_float(name: str, default: float) -> float:
    raw = (os.getenv(name, "") or "").strip()
    if not raw:
        return default
    try:
        return float(raw)
    except ValueError:
        return default


def _env_bool(name: str, default: bool = False) -> bool:
    raw = (os.getenv(name, "") or "").strip().lower()
    if not raw:
        return default
    return raw in {"1", "true", "yes", "on"}


def _env_bool_default_true(*names: str) -> bool:
    for name in names:
        raw = (os.getenv(name, "") or "").strip().lower()
        if not raw:
            continue
        if raw in {"0", "false", "no", "off"}:
            return False
        if raw in {"1", "true", "yes", "on"}:
            return True
    return True


def _strip_text(value: Any) -> str:
    return str(value or "").strip()


def _env_first(*names: str) -> str:
    for name in names:
        value = _strip_text(os.getenv(name))
        if value:
            return value
    return ""


def _agent_env_suffix(agent_name: str) -> str:
    return "".join(
        char if char.isalnum() else "_"
        for char in _strip_text(agent_name).upper()
    ).strip("_")


def _health_desc_value(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    if isinstance(value, (int, float, bool)):
        return str(value)
    try:
        return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
    except TypeError:
        return str(value).strip()


def _health_response_desc(response: httpx.Response) -> str:
    try:
        payload = response.json()
    except ValueError:
        payload = None

    if isinstance(payload, Mapping):
        for key in (
            "http_cod_desc",
            "hhtp_cod_desc",
            "cod_desc",
            "description",
            "message",
            "detail",
            "error",
            "reason",
        ):
            desc = _health_desc_value(payload.get(key))
            if desc:
                return desc
    elif payload is None:
        text = _strip_text(response.text)
        if text:
            return text

    return _strip_text(getattr(response, "reason_phrase", "")) or (
        "OK" if 200 <= response.status_code < 300 else "FAIL"
    )


def _with_path(url: str, *, path: str) -> str:
    parsed = urlsplit(_strip_text(url))
    if not parsed.scheme or not parsed.netloc:
        return ""
    return urlunsplit((parsed.scheme, parsed.netloc, path, "", "")).rstrip("/")


def _derive_http_health_url(url: str) -> str:
    return _with_path(url, path="/health")


def _derive_websocket_health_url(url: str) -> str:
    parsed = urlsplit(_strip_text(url))
    if not parsed.scheme or not parsed.netloc:
        return ""

    scheme = parsed.scheme.lower()
    if scheme == "ws":
        http_scheme = "http"
    elif scheme == "wss":
        http_scheme = "https"
    else:
        return ""

    return urlunsplit((http_scheme, parsed.netloc, "/health", "", "")).rstrip("/")


def _remote_agent_tls_verify(agent_name: str) -> bool:
    suffix = _agent_env_suffix(agent_name)
    names: list[str] = []
    if suffix:
        names.extend(
            [
                f"REMOTE_AGENT_SSE_TLS_VERIFY_{suffix}",
                f"REMOTE_AGENT_TLS_VERIFY_{suffix}",
                f"REMOTE_AGENT_SSE_VERIFY_TLS_{suffix}",
                f"REMOTE_AGENT_VERIFY_TLS_{suffix}",
            ]
        )
    names.extend(
        [
            "REMOTE_AGENT_SSE_TLS_VERIFY",
            "REMOTE_AGENT_TLS_VERIFY",
            "REMOTE_AGENT_SSE_VERIFY_TLS",
            "REMOTE_AGENT_VERIFY_TLS",
        ]
    )
    return _env_bool_default_true(*names)


@dataclass(slots=True)
class ResourceCheck:
    status: str
    message: str = ""
    latency_ms: Optional[int] = None
    url: str = ""
    http_status_code: Optional[int] = None
    http_status_desc: str = ""

    def clone(self) -> "ResourceCheck":
        return ResourceCheck(
            status=self.status,
            message=self.message,
            latency_ms=self.latency_ms,
            url=self.url,
            http_status_code=self.http_status_code,
            http_status_desc=self.http_status_desc,
        )

    def is_healthy(self) -> bool:
        return self.status in HEALTHY_CHECK_STATUSES

    def as_dict(self) -> Dict[str, Any]:
        payload: Dict[str, Any] = {"status": self.status}
        if self.message:
            payload["message"] = self.message
        if self.latency_ms is not None:
            payload["latency_ms"] = self.latency_ms
        if self.url:
            payload["url"] = self.url
        return payload


@dataclass(slots=True)
class ReadinessReport:
    checks: Dict[str, ResourceCheck]
    failed_resources: list[str]
    active_connections: int
    max_connections: int
    cached: bool = False

    @property
    def status(self) -> str:
        return "ok" if not self.failed_resources else "fail"

    def is_healthy(self) -> bool:
        return not self.failed_resources

    def clone(self, *, active_connections: Optional[int] = None, max_connections: Optional[int] = None, cached: Optional[bool] = None) -> "ReadinessReport":
        return ReadinessReport(
            checks={name: check.clone() for name, check in self.checks.items()},
            failed_resources=list(self.failed_resources),
            active_connections=self.active_connections if active_connections is None else active_connections,
            max_connections=self.max_connections if max_connections is None else max_connections,
            cached=self.cached if cached is None else cached,
        )

    def as_dict(self) -> Dict[str, Any]:
        return {
            "status": self.status,
            "cached": self.cached,
            "active_connections": self.active_connections,
            "max_connections": self.max_connections,
            "failed_resources": list(self.failed_resources),
            "checks": {name: check.as_dict() for name, check in self.checks.items()},
        }

    def build_stop_message(self) -> Dict[str, Any]:
        return _build_resource_stop_message(self.failed_resources)


@dataclass(slots=True)
class ServicesHealthReport:
    checks: Dict[str, Any]
    healthy: bool

    @property
    def status(self) -> str:
        return "ok" if self.healthy else "fail"

    def is_healthy(self) -> bool:
        return self.healthy

    def as_dict(self) -> Dict[str, Any]:
        return {
            "status": self.status,
            "checks": self.checks,
        }


def _service_health_payload(check: ResourceCheck) -> Dict[str, Any]:
    fallback_status = 200 if check.is_healthy() else 500
    fallback_desc = "SUCCESS" if check.is_healthy() else "FAIL"
    return {
        "htp_cod_status": (
            check.http_status_code
            if check.http_status_code is not None
            else fallback_status
        ),
        "hhtp_cod_desc": check.http_status_desc or check.message or fallback_desc,
    }


class CapacityLease:
    def __init__(self, manager: "BridgeReadinessManager") -> None:
        self._manager = manager
        self._released = False

    async def release(self) -> None:
        if self._released:
            return
        self._released = True
        await self._manager._release_capacity()


@dataclass(slots=True)
class CapacityReservation:
    allowed: bool
    lease: Optional[CapacityLease]
    active_connections: int
    max_connections: int


class BridgeReadinessManager:
    def __init__(self) -> None:
        self._capacity_lock = asyncio.Lock()
        self._active_connections = 0
        self._cache_lock = asyncio.Lock()
        self._cache: Dict[str, tuple[float, ReadinessReport]] = {}

    def _max_connections(self) -> int:
        return max(0, _env_int("TIA_WS_MAX_CONNECTIONS", 0))

    def _ttl_s(self) -> float:
        return max(0.0, _env_float("TIA_RESOURCE_HEALTH_TTL_S", 5.0))

    def _timeout_s(self) -> float:
        return max(0.1, _env_float("TIA_RESOURCE_HEALTH_TIMEOUT_S", 3.0))

    async def reserve_capacity(self) -> CapacityReservation:
        async with self._capacity_lock:
            max_connections = self._max_connections()
            if max_connections > 0 and self._active_connections >= max_connections:
                return CapacityReservation(
                    allowed=False,
                    lease=None,
                    active_connections=self._active_connections,
                    max_connections=max_connections,
                )

            self._active_connections += 1
            return CapacityReservation(
                allowed=True,
                lease=CapacityLease(self),
                active_connections=self._active_connections,
                max_connections=max_connections,
            )

    async def _release_capacity(self) -> None:
        async with self._capacity_lock:
            if self._active_connections > 0:
                self._active_connections -= 1

    async def capacity_snapshot(self) -> tuple[int, int]:
        async with self._capacity_lock:
            return self._active_connections, self._max_connections()

    def _cache_key(self, *, agent_name: str, call_config: Mapping[str, Any] | None) -> str:
        normalized = normalize_call_config(call_config)
        backend_name = resolve_agent_backend_name(normalized, os.getenv("AGENT_BACKEND", "remote_ws"))
        stt_provider = (resolve_stt_overrides(normalized).get("provider") or os.getenv("STT_PROVIDER", "internal_http")).strip().lower()
        tts_provider = (resolve_tts_overrides(normalized).get("provider") or os.getenv("TTS_PROVIDER", "elevenlabs")).strip().lower()
        return "|".join(
            [
                _strip_text(agent_name).lower() or "conta",
                _strip_text(backend_name).lower() or "remote_ws",
                stt_provider or "internal_http",
                tts_provider or "elevenlabs",
            ]
        )

    async def _get_cached_report(self, key: str) -> Optional[ReadinessReport]:
        ttl_s = self._ttl_s()
        if ttl_s <= 0:
            return None

        async with self._cache_lock:
            cached = self._cache.get(key)
            if cached is None:
                return None

            stored_at, report = cached
            if (time.monotonic() - stored_at) > ttl_s:
                self._cache.pop(key, None)
                return None

            active_connections, max_connections = await self.capacity_snapshot()
            return report.clone(
                active_connections=active_connections,
                max_connections=max_connections,
                cached=True,
            )

    async def _store_cached_report(self, key: str, report: ReadinessReport) -> None:
        async with self._cache_lock:
            self._cache[key] = (
                time.monotonic(),
                report.clone(cached=False),
            )

    async def evaluate_resources(
        self,
        *,
        agent_name: str,
        call_config: Mapping[str, Any] | None,
    ) -> ReadinessReport:
        key = self._cache_key(agent_name=agent_name, call_config=call_config)
        cached = await self._get_cached_report(key)
        if cached is not None:
            return cached

        normalized = normalize_call_config(call_config)
        agent_runtime, agent_backend, stt, tts = await asyncio.gather(
            self.probe_agent_runtime(),
            self.probe_agent_backend(
                agent_name=agent_name,
                call_config=normalized,
            ),
            self.probe_stt(normalized),
            self.probe_tts(normalized),
        )
        checks = {
            "agent_runtime": agent_runtime,
            "agent_backend": agent_backend,
            "stt": stt,
            "tts": tts,
        }

        failed_resources = [
            resource_name
            for resource_name in STOP_RESOURCE_PRECEDENCE
            if resource_name in checks and not checks[resource_name].is_healthy()
        ]

        active_connections, max_connections = await self.capacity_snapshot()
        report = ReadinessReport(
            checks=checks,
            failed_resources=failed_resources,
            active_connections=active_connections,
            max_connections=max_connections,
            cached=False,
        )
        await self._store_cached_report(key, report)
        return report

    async def evaluate_default_resources(self) -> ReadinessReport:
        return await self.evaluate_resources(agent_name="conta", call_config={})

    async def evaluate_services(self) -> ServicesHealthReport:
        (
            agent_runtime,
            contas_backend,
            oferta_backend,
            sofya_stt,
            xai_tts,
        ) = await asyncio.gather(
            self.probe_agent_runtime(),
            self.probe_configured_agent_backend_service(agent_name="conta"),
            self.probe_configured_agent_backend_service(agent_name="oferta"),
            self.probe_configured_stt_service(),
            self.probe_configured_xai_tts_service(),
        )
        service_checks = [agent_runtime, contas_backend, oferta_backend, sofya_stt, xai_tts]
        return ServicesHealthReport(
            healthy=all(check.is_healthy() for check in service_checks),
            checks={
                "agent_runtime": _service_health_payload(agent_runtime),
                "agent_backend": {
                    "contas": _service_health_payload(contas_backend),
                    "oferta": _service_health_payload(oferta_backend),
                },
                "stt": {
                    "sofya": _service_health_payload(sofya_stt),
                },
                "tts": {
                    "xAI": _service_health_payload(xai_tts),
                },
            },
        )

    async def probe_configured_agent_backend_service(self, *, agent_name: str) -> ResourceCheck:
        normalized_agent = {
            "contas": "conta",
            "ofertas": "oferta",
        }.get(_strip_text(agent_name).lower(), _strip_text(agent_name).lower())

        if normalized_agent == "conta":
            health_url = _env_first(
                "REMOTE_AGENT_HEALTH_URL_CONTA",
                "REMOTE_AGENT_HEALTH_URL_CONTAS",
                "REMOTE_AGENT_HEALTH_URL",
            )
            if not health_url:
                health_url = _derive_http_health_url(
                    _env_first("REMOTE_AGENT_SSE_URL_CONTA", "REMOTE_AGENT_SSE_URL_CONTAS")
                )
            if not health_url:
                health_url = _derive_websocket_health_url(
                    _env_first("REMOTE_AGENT_WS_URL_CONTA", "REMOTE_AGENT_WS_URL_CONTAS", "REMOTE_AGENT_WS_URL")
                )
        elif normalized_agent == "oferta":
            health_url = _env_first(
                "REMOTE_AGENT_HEALTH_URL_OFERTA",
                "REMOTE_AGENT_HEALTH_URL_OFERTAS",
            )
            if not health_url:
                health_url = _derive_http_health_url(
                    _env_first("REMOTE_AGENT_SSE_URL_OFERTA", "REMOTE_AGENT_SSE_URL_OFERTAS")
                )
            if not health_url:
                health_url = _derive_websocket_health_url(
                    _env_first("REMOTE_AGENT_WS_URL_OFERTA", "REMOTE_AGENT_WS_URL_OFERTAS")
                )
        else:
            return ResourceCheck(status="fail", message=f"unsupported service agent backend: {agent_name}")

        if not health_url:
            return ResourceCheck(
                status="fail",
                message=f"{normalized_agent} backend health URL is not configured",
            )

        return await self._probe_http_health(
            "agent_backend",
            health_url,
            verify=_remote_agent_tls_verify(normalized_agent),
        )

    async def probe_configured_stt_service(self) -> ResourceCheck:
        health_url = _env_first("STT_HEALTH_URL")
        if not health_url:
            health_url = _derive_http_health_url(_env_first("STT_URL"))

        if not health_url:
            return ResourceCheck(status="fail", message="STT health URL is not configured")

        return await self._probe_http_health("stt", health_url)

    async def probe_configured_xai_tts_service(self) -> ResourceCheck:
        return await self.probe_tts({"tts": {"provider": "xai"}})

    async def probe_agent_runtime(self) -> ResourceCheck:
        url = f"http://127.0.0.1:{_env_int('AGENT_SERVER_PORT', 18081)}/"
        return await self._probe_http_health("agent_runtime", url)

    async def probe_agent_backend(
        self,
        *,
        agent_name: str,
        call_config: Mapping[str, Any] | None,
    ) -> ResourceCheck:
        raw_agent = _strip_text(agent_name).lower()
        normalized_agent = {
            "contas": "conta",
            "ofert": "oferta",
            "ofertas": "oferta",
            "cobra": "cobranca",
            "cobrancas": "cobranca",
            "cobrança": "cobranca",
            "cobranças": "cobranca",
        }.get(raw_agent, raw_agent or "conta")

        backend_name = resolve_agent_backend_name(call_config, os.getenv("AGENT_BACKEND", "remote_ws")).strip().lower()
        if backend_name in {"remote_ws_fake", "fake_remote_ws", "ws_fake"}:
            return ResourceCheck(status="ok", message="fake backend skips external health probe")

        if backend_name not in {"", "remote_ws", "ws", "websocket", "remote_sse", "sse"}:
            return ResourceCheck(status="fail", message=f"unsupported agent backend: {backend_name or 'unknown'}")

        agent_health_envs = {
            "conta": ("REMOTE_AGENT_HEALTH_URL_CONTA", "REMOTE_AGENT_HEALTH_URL_CONTAS"),
            "oferta": ("REMOTE_AGENT_HEALTH_URL_OFERTA", "REMOTE_AGENT_HEALTH_URL_OFERTAS"),
            "cobranca": (
                "REMOTE_AGENT_HEALTH_URL_COBRANCA",
                "REMOTE_AGENT_HEALTH_URL_COBRANCAS",
                "REMOTE_AGENT_HEALTH_URL_COBRA",
            ),
        }.get(normalized_agent, ())
        health_url = ""
        for env_name in agent_health_envs:
            health_url = _strip_text(os.getenv(env_name))
            if health_url:
                break

        if normalized_agent == "oferta" and backend_name in {"remote_sse", "sse"} and not health_url:
            return ResourceCheck(
                status="skipped",
                message="oferta backend health probe disabled until health URL is provided",
            )

        if not health_url:
            health_url = _strip_text(os.getenv("REMOTE_AGENT_HEALTH_URL"))

        if not health_url:
            if backend_name in {"remote_sse", "sse"}:
                agent_sse_envs = {
                    "conta": ("REMOTE_AGENT_SSE_URL_CONTA", "REMOTE_AGENT_SSE_URL_CONTAS"),
                    "cobranca": (
                        "REMOTE_AGENT_SSE_URL_COBRANCA",
                        "REMOTE_AGENT_SSE_URL_COBRANCAS",
                        "REMOTE_AGENT_SSE_URL_COBRA",
                    ),
                }.get(normalized_agent, ())
                sse_url = ""
                for env_name in agent_sse_envs:
                    sse_url = _strip_text(os.getenv(env_name))
                    if sse_url:
                        break
                health_url = _derive_http_health_url(sse_url)
            else:
                agent_ws_envs = {
                    "conta": ("REMOTE_AGENT_WS_URL_CONTA", "REMOTE_AGENT_WS_URL_CONTAS"),
                    "oferta": ("REMOTE_AGENT_WS_URL_OFERTA", "REMOTE_AGENT_WS_URL_OFERTAS"),
                    "cobranca": (
                        "REMOTE_AGENT_WS_URL_COBRANCA",
                        "REMOTE_AGENT_WS_URL_COBRANCAS",
                        "REMOTE_AGENT_WS_URL_COBRA",
                    ),
                }.get(normalized_agent, ())
                ws_url = ""
                for env_name in (*agent_ws_envs, "REMOTE_AGENT_WS_URL"):
                    ws_url = _strip_text(os.getenv(env_name))
                    if ws_url:
                        break
                health_url = _derive_websocket_health_url(ws_url)

        if not health_url:
            return ResourceCheck(status="fail", message="remote agent health URL is not configured")

        return await self._probe_http_health("agent_backend", health_url)

    async def probe_stt(self, call_config: Mapping[str, Any] | None) -> ResourceCheck:
        if _env_bool("TIA_SKIP_STT_READINESS", False):
            return ResourceCheck(status="skipped", message="STT readiness disabled by env")

        overrides = resolve_stt_overrides(call_config)
        provider = (overrides.get("provider") or os.getenv("STT_PROVIDER", "internal_http")).strip().lower()

        if provider == "fake":
            FakeSTT()
            return ResourceCheck(status="ok", message="fake STT skips external health probe")

        if provider not in {"", "internal_http"}:
            return ResourceCheck(status="fail", message=f"unsupported STT provider: {provider}")

        health_url = _strip_text(os.getenv("STT_HEALTH_URL"))
        if not health_url:
            health_url = _derive_http_health_url(os.getenv("STT_URL", ""))

        if not health_url:
            return ResourceCheck(status="fail", message="STT health URL is not configured")

        return await self._probe_http_health("stt", health_url)

    async def probe_tts(self, call_config: Mapping[str, Any] | None) -> ResourceCheck:
        overrides = resolve_tts_overrides(call_config)
        provider = (overrides.get("provider") or os.getenv("TTS_PROVIDER", "elevenlabs")).strip().lower()
        timeout_s = self._timeout_s()
        started_at = time.monotonic()

        try:
            if provider == "fake":
                tts = FakeProviderTTS()
                audio = tts.synthesize_pcm16k(PROBE_TEXT)
                self._ensure_audio_bytes(audio)
                return ResourceCheck(status="ok", latency_ms=round((time.monotonic() - started_at) * 1000))

            if provider in {"", "elevenlabs"}:
                from livekit.plugins import elevenlabs

                async with aiohttp.ClientSession() as http_session:
                    tts = elevenlabs.TTS(
                        model=overrides.get("model_id") or os.getenv("ELEVENLABS_MODEL_ID", ""),
                        voice_id=overrides.get("voice_id") or os.getenv("ELEVENLABS_VOICE_ID", ""),
                        api_key=os.getenv("ELEVENLABS_API_KEY", ""),
                        voice_settings=elevenlabs.VoiceSettings(
                            stability=0.35,
                            speed=1.02,
                            similarity_boost=0.75,
                            use_speaker_boost=True,
                            style=0.7,
                        ),
                        http_session=http_session,
                        language="pt",
                    )
                    try:
                        audio = await asyncio.wait_for(self._collect_livekit_tts_bytes(tts), timeout=timeout_s)
                    finally:
                        await tts.aclose()
                self._ensure_audio_bytes(audio)
                return ResourceCheck(status="ok", latency_ms=round((time.monotonic() - started_at) * 1000))

            if provider == "xai":
                auth_method = _strip_text(os.getenv("XAI_TTS_AUTH_METHOD", "API_KEY")).upper() or "API_KEY"
                xai_auth_options: dict[str, str] = {"auth_method": auth_method}
                if auth_method == "API_KEY":
                    api_key = _strip_text(os.getenv("XAI_API_KEY"))
                    if not api_key:
                        return ResourceCheck(status="fail", message="missing xAI TTS config: XAI_API_KEY")
                    xai_auth_options["api_key"] = api_key
                readiness_mode = (
                    _strip_text(os.getenv(XAI_TTS_READINESS_MODE_ENV, "synthesize")).lower()
                    or "synthesize"
                )
                if readiness_mode not in XAI_TTS_READINESS_MODES:
                    return ResourceCheck(
                        status="fail",
                        message=f"unsupported {XAI_TTS_READINESS_MODE_ENV}: {readiness_mode}",
                    )

                async with aiohttp.ClientSession() as http_session:
                    tts = OraclexAITTS(
                        voice=overrides.get("voice_id") or os.getenv("XAI_TTS_VOICE", ""),
                        language=overrides.get("language") or os.getenv("XAI_TTS_LANGUAGE", "pt-BR"),
                        http_session=http_session,
                        **xai_auth_options,
                    )
                    try:
                        if readiness_mode == "connect":
                            await tts.connect(timeout_s)
                            return ResourceCheck(
                                status="ok",
                                message="xAI TTS websocket connected",
                                latency_ms=round((time.monotonic() - started_at) * 1000),
                            )
                        audio = await asyncio.wait_for(self._collect_livekit_tts_bytes(tts), timeout=timeout_s)
                    finally:
                        await tts.aclose()
                self._ensure_audio_bytes(audio)
                return ResourceCheck(status="ok", latency_ms=round((time.monotonic() - started_at) * 1000))

            if provider == "azure":
                azure_tts_config, missing_keys = resolve_azure_speech_tts_config(overrides)
                if missing_keys:
                    return ResourceCheck(
                        status="fail",
                        message=f"missing Azure TTS config: {', '.join(missing_keys)}",
                    )

                azure_impl = (os.getenv("AZURE_TTS_IMPLEMENTATION", "plugin") or "plugin").strip().lower()
                if azure_impl == "rest":
                    tts = AzureRESTTTS(
                        sample_rate=AZURE_SPEECH_TTS_SAMPLE_RATE,
                        **azure_tts_config,
                    )
                    try:
                        audio = await asyncio.wait_for(asyncio.to_thread(tts.synthesize_pcm, PROBE_TEXT), timeout=timeout_s)
                    finally:
                        await tts.aclose()
                elif azure_impl == "plugin":
                    from livekit.plugins import azure as livekit_azure

                    async with aiohttp.ClientSession() as http_session:
                        tts = livekit_azure.TTS(
                            sample_rate=AZURE_SPEECH_TTS_SAMPLE_RATE,
                            http_session=http_session,
                            **azure_tts_config,
                        )
                        try:
                            audio = await asyncio.wait_for(self._collect_livekit_tts_bytes(tts), timeout=timeout_s)
                        finally:
                            await tts.aclose()
                else:
                    return ResourceCheck(status="fail", message=f"unsupported AZURE_TTS_IMPLEMENTATION: {azure_impl}")

                self._ensure_audio_bytes(audio)
                return ResourceCheck(status="ok", latency_ms=round((time.monotonic() - started_at) * 1000))

            return ResourceCheck(status="fail", message=f"unsupported TTS provider: {provider}")
        except Exception as exc:
            return ResourceCheck(
                status="fail",
                message=_format_probe_error(exc, resource=f"{provider or 'default'} TTS", timeout_s=timeout_s),
                latency_ms=round((time.monotonic() - started_at) * 1000),
            )

    async def _collect_livekit_tts_bytes(self, tts: Any) -> bytes:
        frame = await tts.synthesize(PROBE_TEXT).collect()
        return bytes(getattr(frame, "data", b""))

    def _ensure_audio_bytes(self, audio: bytes) -> None:
        if not audio:
            raise RuntimeError("TTS returned empty audio")

    async def _probe_http_health(self, check_name: str, url: str, *, verify: bool = True) -> ResourceCheck:
        started_at = time.monotonic()
        timeout_s = self._timeout_s()
        try:
            async with httpx.AsyncClient(follow_redirects=True, verify=verify) as client:
                response = await client.get(url, timeout=timeout_s)
            latency_ms = round((time.monotonic() - started_at) * 1000)
            http_status_desc = _health_response_desc(response)
            if 200 <= response.status_code < 300:
                return ResourceCheck(
                    status="ok",
                    latency_ms=latency_ms,
                    url=url,
                    http_status_code=response.status_code,
                    http_status_desc=http_status_desc,
                )
            return ResourceCheck(
                status="fail",
                message=(
                    f"{check_name} returned status {response.status_code}: {http_status_desc}"
                    if http_status_desc
                    else f"{check_name} returned status {response.status_code}"
                ),
                latency_ms=latency_ms,
                url=url,
                http_status_code=response.status_code,
                http_status_desc=http_status_desc,
            )
        except Exception as exc:
            return ResourceCheck(
                status="fail",
                message=_format_probe_error(exc, resource=check_name, timeout_s=timeout_s),
                latency_ms=round((time.monotonic() - started_at) * 1000),
                url=url,
            )
