from __future__ import annotations

import asyncio
import json
from types import SimpleNamespace
from unittest import mock

from fastapi.testclient import TestClient

from app.ws_gateway import main as main_module
from app.ws_gateway.readiness import (
    BridgeReadinessManager,
    CapacityReservation,
    ReadinessReport,
    ResourceCheck,
    ServicesHealthReport,
    build_bridge_failed_stop_message,
    build_capacity_stop_message,
    build_completed_stop_message,
)


def _build_start_payload() -> dict:
    return {
        "type": "start",
        "data": {
            "agent": "conta",
            "ani": "5511999999999",
            "gsm": "5511999999999",
            "session_id": "550e8400-e29b-41d4-a716-446655440200",
            "routerCallKeyDay": "20260409",
            "routerCallKey": "RCK-001",
            "callIdGed": "GED-001",
            "agentData": {
                "idFatura": "fat-123",
            },
        },
    }


class _FakeLease:
    def __init__(self) -> None:
        self.release_calls = 0

    async def release(self) -> None:
        self.release_calls += 1


class _FakeReadinessManager:
    def __init__(
        self,
        *,
        reservation: CapacityReservation | None = None,
        report: ReadinessReport | None = None,
    ) -> None:
        self.lease = reservation.lease if reservation is not None else _FakeLease()
        self.reservation = reservation or CapacityReservation(
            allowed=True,
            lease=self.lease,
            active_connections=1,
            max_connections=5,
        )
        self.report = report or ReadinessReport(
            checks={
                "agent_runtime": ResourceCheck(status="ok"),
                "agent_backend": ResourceCheck(status="ok"),
                "stt": ResourceCheck(status="ok"),
                "tts": ResourceCheck(status="ok"),
            },
            failed_resources=[],
            active_connections=1,
            max_connections=5,
            cached=False,
        )
        self.services_report = ServicesHealthReport(
            checks={
                "agent_runtime": {
                    "htp_cod_status": 200,
                    "hhtp_cod_desc": "SUCCESS",
                },
                "agent_backend": {
                    "contas": {
                        "htp_cod_status": 200,
                        "hhtp_cod_desc": "SUCCESS",
                    },
                    "oferta": {
                        "htp_cod_status": 200,
                        "hhtp_cod_desc": "SUCCESS",
                    },
                },
                "stt": {
                    "sofya": {
                        "htp_cod_status": 200,
                        "hhtp_cod_desc": "SUCCESS",
                    },
                },
                "tts": {
                    "xAI": {
                        "htp_cod_status": 200,
                        "hhtp_cod_desc": "SUCCESS",
                    },
                },
            },
            healthy=True,
        )
        self.evaluate_calls: list[tuple[str, dict]] = []
        self.default_calls = 0
        self.services_calls = 0

    async def reserve_capacity(self) -> CapacityReservation:
        return self.reservation

    async def evaluate_resources(self, *, agent_name: str, call_config: dict | None) -> ReadinessReport:
        self.evaluate_calls.append((agent_name, dict(call_config or {})))
        return self.report

    async def evaluate_default_resources(self) -> ReadinessReport:
        self.default_calls += 1
        return self.report

    async def evaluate_services(self) -> ServicesHealthReport:
        self.services_calls += 1
        return self.services_report


class _FakeTimeline:
    def __init__(self) -> None:
        self.events: list[tuple[str, dict]] = []
        self.path = "/tmp/timeline.jsonl"

    def emit(self, event: str, **fields) -> None:
        self.events.append((event, fields))


def test_stop_message_helpers_follow_contract() -> None:
    assert build_capacity_stop_message() == {
        "type": "stop",
        "data": {
            "status": "stop_capacity_tia",
            "reason": "capacity_exceeded",
            "resource": "capacity",
            "failed_resources": ["capacity"],
            "phase": "pre_ready",
        },
    }
    assert build_completed_stop_message("finished") == {
        "type": "stop",
        "data": {
            "status": "stop_resolvido_e_finalizado",
            "reason": "finished",
            "phase": "in_session",
        },
    }
    assert build_bridge_failed_stop_message() == {
        "type": "stop",
        "data": {
            "status": "stop_bridge_failed",
            "reason": "bridge_failed",
            "resource": "bridge",
            "phase": "in_session",
        },
    }


def test_build_completed_stop_message_maps_reason_to_configured_status(monkeypatch) -> None:
    monkeypatch.setenv("FINAL_STOP_STATUS_RESOLVED", "stop_custom_resolvido")
    monkeypatch.setenv("FINAL_STOP_STATUS_UNRESOLVED", "stop_custom_nao_resolvido")
    monkeypatch.setenv("FINAL_STOP_STATUS_OTHER_SUBJECT", "stop_custom_outro_assunto")
    monkeypatch.setenv("FINAL_STOP_STATUS_LONG_SILENCE", "stop_custom_silencio")
    monkeypatch.setenv("FINAL_STOP_REASON_RESOLVED", "resolvido")
    monkeypatch.setenv("FINAL_STOP_REASON_UNRESOLVED", "nao_resolvido")
    monkeypatch.setenv("FINAL_STOP_REASON_OTHER_SUBJECT", "outro_assunto")
    monkeypatch.setenv("FINAL_STOP_REASON_LONG_SILENCE", "silencio_longo")

    assert build_completed_stop_message("nao_resolvido") == {
        "type": "stop",
        "data": {
            "status": "stop_custom_nao_resolvido",
            "reason": "nao_resolvido",
            "phase": "in_session",
        },
    }
    assert build_completed_stop_message("outro_assunto") == {
        "type": "stop",
        "data": {
            "status": "stop_custom_outro_assunto",
            "reason": "outro_assunto",
            "phase": "in_session",
        },
    }
    assert build_completed_stop_message("silencio_longo") == {
        "type": "stop",
        "data": {
            "status": "stop_custom_silencio",
            "reason": "silencio_longo",
            "phase": "in_session",
        },
    }
    assert build_completed_stop_message("resolvido") == {
        "type": "stop",
        "data": {
            "status": "stop_custom_resolvido",
            "reason": "resolvido",
            "phase": "in_session",
        },
    }


def test_capacity_reservation_blocks_and_releases(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("TIA_WS_MAX_CONNECTIONS", "1")

        first = await manager.reserve_capacity()
        second = await manager.reserve_capacity()
        assert first.allowed is True
        assert second.allowed is False

        await first.lease.release()

        third = await manager.reserve_capacity()
        assert third.allowed is True
        await third.lease.release()

    asyncio.run(_run())


def test_probe_agent_backend_prefers_explicit_health_url(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("REMOTE_AGENT_HEALTH_URL_CONTA", "https://agent.example/health")
        monkeypatch.setenv("AGENT_BACKEND", "remote_ws")

        async def _fake_probe(check_name: str, url: str) -> ResourceCheck:
            assert check_name == "agent_backend"
            assert url == "https://agent.example/health"
            return ResourceCheck(status="ok", url=url)

        with mock.patch.object(manager, "_probe_http_health", side_effect=_fake_probe):
            check = await manager.probe_agent_backend(agent_name="conta", call_config={})

        assert check.status == "ok"
        assert check.url == "https://agent.example/health"

    asyncio.run(_run())


def test_probe_agent_backend_derives_health_url_from_websocket(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL_CONTA", raising=False)
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL", raising=False)
        monkeypatch.setenv("REMOTE_AGENT_WS_URL", "wss://agent.internal/agent/ws")
        monkeypatch.setenv("AGENT_BACKEND", "remote_ws")

        async def _fake_probe(check_name: str, url: str) -> ResourceCheck:
            assert check_name == "agent_backend"
            assert url == "https://agent.internal/health"
            return ResourceCheck(status="ok", url=url)

        with mock.patch.object(manager, "_probe_http_health", side_effect=_fake_probe):
            check = await manager.probe_agent_backend(agent_name="conta", call_config={})

        assert check.status == "ok"
        assert check.url == "https://agent.internal/health"

    asyncio.run(_run())


def test_probe_agent_backend_derives_health_url_from_sse(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL_CONTA", raising=False)
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL_CONTAS", raising=False)
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL", raising=False)
        monkeypatch.setenv("REMOTE_AGENT_SSE_URL_CONTA", "http://agent.internal/agent/sse")
        monkeypatch.setenv("AGENT_BACKEND", "remote_sse")

        async def _fake_probe(check_name: str, url: str) -> ResourceCheck:
            assert check_name == "agent_backend"
            assert url == "http://agent.internal/health"
            return ResourceCheck(status="ok", url=url)

        with mock.patch.object(manager, "_probe_http_health", side_effect=_fake_probe):
            check = await manager.probe_agent_backend(agent_name="conta", call_config={})

        assert check.status == "ok"
        assert check.url == "http://agent.internal/health"

    asyncio.run(_run())


def test_probe_agent_backend_uses_oferta_health_url(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("REMOTE_AGENT_HEALTH_URL_OFERTA", "https://oferta.example/health")
        monkeypatch.setenv("AGENT_BACKEND", "remote_sse")

        async def _fake_probe(check_name: str, url: str) -> ResourceCheck:
            assert check_name == "agent_backend"
            assert url == "https://oferta.example/health"
            return ResourceCheck(status="ok", url=url)

        with mock.patch.object(manager, "_probe_http_health", side_effect=_fake_probe):
            check = await manager.probe_agent_backend(agent_name="oferta", call_config={})

        assert check.status == "ok"
        assert check.url == "https://oferta.example/health"

    asyncio.run(_run())


def test_probe_agent_backend_skips_oferta_health_probe_without_explicit_route(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL_OFERTA", raising=False)
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL_OFERTAS", raising=False)
        monkeypatch.setenv("REMOTE_AGENT_HEALTH_URL", "http://conta.internal/health")
        monkeypatch.setenv("REMOTE_AGENT_SSE_URL_OFERTA", "http://oferta.internal/agent/execute")
        monkeypatch.setenv("AGENT_BACKEND", "remote_sse")

        with mock.patch.object(manager, "_probe_http_health") as mocked_probe:
            check = await manager.probe_agent_backend(agent_name="ofertas", call_config={})

        mocked_probe.assert_not_called()
        assert check.status == "skipped"
        assert check.url == ""
        assert "oferta backend health probe disabled" in check.message

    asyncio.run(_run())


def test_probe_stt_derives_health_url_from_stt_url(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.delenv("STT_HEALTH_URL", raising=False)
        monkeypatch.setenv("STT_PROVIDER", "internal_http")
        monkeypatch.setenv("STT_URL", "http://stt.internal/api/transcriber")

        async def _fake_probe(check_name: str, url: str) -> ResourceCheck:
            assert check_name == "stt"
            assert url == "http://stt.internal/health"
            return ResourceCheck(status="ok", url=url)

        with mock.patch.object(manager, "_probe_http_health", side_effect=_fake_probe):
            check = await manager.probe_stt({})

        assert check.status == "ok"
        assert check.url == "http://stt.internal/health"

    asyncio.run(_run())


def test_probe_stt_can_be_temporarily_skipped_by_env(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("TIA_SKIP_STT_READINESS", "1")
        monkeypatch.setenv("STT_PROVIDER", "internal_http")
        monkeypatch.setenv("STT_URL", "http://stt.internal/api/transcriber")

        with mock.patch.object(manager, "_probe_http_health") as mocked_probe:
            check = await manager.probe_stt({})

        assert check.status == "skipped"
        assert check.message == "STT readiness disabled by env"
        mocked_probe.assert_not_called()

    asyncio.run(_run())


def test_probe_http_health_preserves_returned_status_and_description() -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()

        class _FakeResponse:
            status_code = 503
            reason_phrase = "Service Unavailable"
            text = '{"message":"db down"}'

            def json(self) -> dict:
                return {"message": "db down"}

        class _FakeAsyncClient:
            def __init__(self, *args, **kwargs) -> None:
                pass

            async def __aenter__(self) -> "_FakeAsyncClient":
                return self

            async def __aexit__(self, *args) -> None:
                return None

            async def get(self, url: str, timeout: float):
                assert url == "https://service.example/health"
                assert timeout > 0
                return _FakeResponse()

        with mock.patch("app.ws_gateway.readiness.httpx.AsyncClient", _FakeAsyncClient):
            check = await manager._probe_http_health("agent_backend", "https://service.example/health")

        assert check.status == "fail"
        assert check.message == "agent_backend returned status 503: db down"
        assert check.http_status_code == 503
        assert check.http_status_desc == "db down"

    asyncio.run(_run())


def test_evaluate_resources_uses_cache(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("TIA_RESOURCE_HEALTH_TTL_S", "60")
        counters = {
            "agent_runtime": 0,
            "agent_backend": 0,
            "stt": 0,
            "tts": 0,
        }

        async def _ok(name: str) -> ResourceCheck:
            counters[name] += 1
            return ResourceCheck(status="ok")

        async def _probe_agent_runtime() -> ResourceCheck:
            return await _ok("agent_runtime")

        async def _probe_agent_backend(**_: dict) -> ResourceCheck:
            return await _ok("agent_backend")

        async def _probe_stt(*_: dict) -> ResourceCheck:
            return await _ok("stt")

        async def _probe_tts(*_: dict) -> ResourceCheck:
            return await _ok("tts")

        with (
            mock.patch.object(manager, "probe_agent_runtime", side_effect=_probe_agent_runtime),
            mock.patch.object(manager, "probe_agent_backend", side_effect=_probe_agent_backend),
            mock.patch.object(manager, "probe_stt", side_effect=_probe_stt),
            mock.patch.object(manager, "probe_tts", side_effect=_probe_tts),
        ):
            first = await manager.evaluate_resources(agent_name="conta", call_config={})
            second = await manager.evaluate_resources(agent_name="conta", call_config={})

        assert first.cached is False
        assert second.cached is True
        assert counters == {
            "agent_runtime": 1,
            "agent_backend": 1,
            "stt": 1,
            "tts": 1,
        }

    asyncio.run(_run())


def test_evaluate_resources_runs_resource_probes_in_parallel(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("TIA_RESOURCE_HEALTH_TTL_S", "0")
        started: set[str] = set()
        all_started = asyncio.Event()

        async def _probe(name: str) -> ResourceCheck:
            started.add(name)
            if len(started) == 4:
                all_started.set()
            await asyncio.wait_for(all_started.wait(), timeout=0.2)
            return ResourceCheck(status="ok")

        async def _probe_agent_runtime() -> ResourceCheck:
            return await _probe("agent_runtime")

        async def _probe_agent_backend(**_: dict) -> ResourceCheck:
            return await _probe("agent_backend")

        async def _probe_stt(*_: dict) -> ResourceCheck:
            return await _probe("stt")

        async def _probe_tts(*_: dict) -> ResourceCheck:
            return await _probe("tts")

        with (
            mock.patch.object(manager, "probe_agent_runtime", side_effect=_probe_agent_runtime),
            mock.patch.object(manager, "probe_agent_backend", side_effect=_probe_agent_backend),
            mock.patch.object(manager, "probe_stt", side_effect=_probe_stt),
            mock.patch.object(manager, "probe_tts", side_effect=_probe_tts),
        ):
            report = await manager.evaluate_resources(agent_name="conta", call_config={})

        assert report.is_healthy() is True
        assert started == {"agent_runtime", "agent_backend", "stt", "tts"}

    asyncio.run(_run())


def test_probe_tts_fake_returns_ok() -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        check = await manager.probe_tts({"tts": {"provider": "fake"}})
        assert check.status == "ok"

    asyncio.run(_run())


def test_probe_tts_elevenlabs_can_be_mocked(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("ELEVENLABS_API_KEY", "key-123")
        monkeypatch.setenv("ELEVENLABS_VOICE_ID", "voice-123")
        monkeypatch.setenv("ELEVENLABS_MODEL_ID", "model-123")

        fake_stream = mock.Mock()
        fake_stream.collect = mock.AsyncMock(return_value=mock.Mock(data=b"\x00\x01"))
        fake_tts = mock.Mock()
        fake_tts.synthesize.return_value = fake_stream
        fake_tts.aclose = mock.AsyncMock()

        with mock.patch("livekit.plugins.elevenlabs.TTS", return_value=fake_tts) as mocked_tts:
            check = await manager.probe_tts({"tts": {"provider": "elevenlabs"}})

        assert check.status == "ok"
        assert mocked_tts.call_args.kwargs["http_session"] is not None
        fake_tts.aclose.assert_awaited_once()

    asyncio.run(_run())


def test_probe_tts_xai_can_be_mocked(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("XAI_API_KEY", "key-123")

        fake_stream = mock.Mock()
        fake_stream.collect = mock.AsyncMock(return_value=mock.Mock(data=b"\x00\x01"))
        fake_tts = mock.Mock()
        fake_tts.synthesize.return_value = fake_stream
        fake_tts.aclose = mock.AsyncMock()

        with mock.patch("app.ws_gateway.readiness.OraclexAITTS", return_value=fake_tts) as mocked_tts:
            check = await manager.probe_tts(
                {"tts": {"provider": "xai", "voiceId": "voice-123", "language": "pt-BR"}}
            )

        assert check.status == "ok"
        assert mocked_tts.call_args.kwargs["api_key"] == "key-123"
        assert mocked_tts.call_args.kwargs["voice"] == "voice-123"
        assert mocked_tts.call_args.kwargs["language"] == "pt-BR"
        assert mocked_tts.call_args.kwargs["http_session"] is not None
        fake_tts.aclose.assert_awaited_once()

    asyncio.run(_run())


def test_probe_tts_xai_can_only_connect(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("XAI_API_KEY", "key-123")
        monkeypatch.setenv("XAI_TTS_READINESS_MODE", "connect")

        fake_tts = mock.Mock()
        fake_tts.connect = mock.AsyncMock()
        fake_tts.aclose = mock.AsyncMock()

        with mock.patch("app.ws_gateway.readiness.OraclexAITTS", return_value=fake_tts):
            check = await manager.probe_tts({"tts": {"provider": "xai"}})

        assert check.status == "ok"
        assert check.message == "xAI TTS websocket connected"
        fake_tts.connect.assert_awaited_once()
        fake_tts.synthesize.assert_not_called()
        fake_tts.aclose.assert_awaited_once()

    asyncio.run(_run())


def test_probe_http_health_timeout_has_diagnostic_message(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("TIA_RESOURCE_HEALTH_TIMEOUT_S", "0.1")

        class _FailingAsyncClient:
            def __init__(self, *args, **kwargs) -> None:
                pass

            async def __aenter__(self) -> "_FailingAsyncClient":
                return self

            async def __aexit__(self, *args) -> None:
                return None

            async def get(self, url: str, timeout: float):
                raise TimeoutError()

        with mock.patch("app.ws_gateway.readiness.httpx.AsyncClient", _FailingAsyncClient):
            check = await manager._probe_http_health("stt", "https://service.example/health")

        assert check.status == "fail"
        assert check.message == "stt readiness timed out after 100ms (TimeoutError)"

    asyncio.run(_run())


def test_probe_tts_xai_timeout_has_diagnostic_message(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("XAI_API_KEY", "key-123")
        monkeypatch.setenv("TIA_RESOURCE_HEALTH_TIMEOUT_S", "0.1")

        fake_tts = mock.Mock()
        fake_tts.connect = mock.AsyncMock(side_effect=TimeoutError())
        fake_tts.aclose = mock.AsyncMock()

        with mock.patch("app.ws_gateway.readiness.OraclexAITTS", return_value=fake_tts):
            monkeypatch.setenv("XAI_TTS_READINESS_MODE", "connect")
            check = await manager.probe_tts({"tts": {"provider": "xai"}})

        assert check.status == "fail"
        assert check.message == "xai TTS readiness timed out after 100ms (TimeoutError)"
        fake_tts.aclose.assert_awaited_once()

    asyncio.run(_run())


def test_probe_tts_xai_iam_does_not_require_legacy_api_key(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("XAI_TTS_AUTH_METHOD", "INSTANCE_PRINCIPAL")
        monkeypatch.setenv("OCI_COMPARTMENT_ID", "ocid1.compartment.oc1..example")
        monkeypatch.setenv("XAI_TTS_READINESS_MODE", "connect")
        monkeypatch.delenv("XAI_API_KEY", raising=False)

        fake_tts = mock.Mock()
        fake_tts.connect = mock.AsyncMock()
        fake_tts.aclose = mock.AsyncMock()

        with mock.patch("app.ws_gateway.readiness.OraclexAITTS", return_value=fake_tts) as mocked_tts:
            check = await manager.probe_tts({"tts": {"provider": "xai"}})

        assert check.status == "ok"
        assert mocked_tts.call_args.kwargs["auth_method"] == "INSTANCE_PRINCIPAL"
        assert "api_key" not in mocked_tts.call_args.kwargs
        fake_tts.connect.assert_awaited_once()
        fake_tts.aclose.assert_awaited_once()

    asyncio.run(_run())


def test_probe_tts_xai_requires_api_key(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.delenv("XAI_API_KEY", raising=False)

        check = await manager.probe_tts({"tts": {"provider": "xai"}})

        assert check.status == "fail"
        assert check.message == "missing xAI TTS config: XAI_API_KEY"

    asyncio.run(_run())


def test_probe_tts_azure_rest_can_be_mocked(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("TTS_PROVIDER", "azure")
        monkeypatch.setenv("AZURE_TTS_IMPLEMENTATION", "rest")
        monkeypatch.setenv("AZURE_SPEECH_KEY", "key-123")
        monkeypatch.setenv("AZURE_SPEECH_REGION", "brazilsouth")
        monkeypatch.setenv("AZURE_SPEECH_VOICE", "pt-BR-FranciscaNeural")

        with mock.patch("app.ws_gateway.readiness.AzureRESTTTS.synthesize_pcm", return_value=b"\x00\x01"):
            check = await manager.probe_tts({"tts": {"provider": "azure"}})

        assert check.status == "ok"

    asyncio.run(_run())


def test_probe_tts_azure_plugin_can_be_mocked(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("TTS_PROVIDER", "azure")
        monkeypatch.setenv("AZURE_TTS_IMPLEMENTATION", "plugin")
        monkeypatch.setenv("AZURE_SPEECH_KEY", "key-123")
        monkeypatch.setenv("AZURE_SPEECH_REGION", "brazilsouth")
        monkeypatch.setenv("AZURE_SPEECH_VOICE", "pt-BR-FranciscaNeural")

        fake_stream = mock.Mock()
        fake_stream.collect = mock.AsyncMock(return_value=mock.Mock(data=b"\x00\x01"))
        fake_tts = mock.Mock()
        fake_tts.synthesize.return_value = fake_stream
        fake_tts.aclose = mock.AsyncMock()

        with mock.patch("livekit.plugins.azure.TTS", return_value=fake_tts) as mocked_tts:
            check = await manager.probe_tts({"tts": {"provider": "azure"}})

        assert check.status == "ok"
        assert mocked_tts.call_args.kwargs["http_session"] is not None
        fake_tts.aclose.assert_awaited_once()

    asyncio.run(_run())


def test_evaluate_services_returns_requested_shape() -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        backend_calls: list[tuple[str, dict]] = []

        async def _probe_agent_runtime() -> ResourceCheck:
            return ResourceCheck(
                status="fail",
                message="down",
                http_status_code=503,
                http_status_desc="Service Unavailable",
            )

        async def _probe_agent_backend(*, agent_name: str) -> ResourceCheck:
            backend_calls.append((agent_name, {}))
            if agent_name == "conta":
                return ResourceCheck(
                    status="ok",
                    http_status_code=204,
                    http_status_desc="No Content",
                )
            return ResourceCheck(
                status="ok",
                http_status_code=200,
                http_status_desc="oferta healthy",
            )

        async def _probe_stt() -> ResourceCheck:
            return ResourceCheck(
                status="ok",
                http_status_code=200,
                http_status_desc="sofya ready",
            )

        async def _probe_tts() -> ResourceCheck:
            return ResourceCheck(status="ok", message="xAI TTS websocket connected")

        with (
            mock.patch.object(manager, "probe_agent_runtime", side_effect=_probe_agent_runtime),
            mock.patch.object(manager, "probe_configured_agent_backend_service", side_effect=_probe_agent_backend),
            mock.patch.object(manager, "probe_configured_stt_service", side_effect=_probe_stt),
            mock.patch.object(manager, "probe_configured_xai_tts_service", side_effect=_probe_tts),
        ):
            report = await manager.evaluate_services()

        assert report.is_healthy() is False
        assert report.as_dict() == {
            "status": "fail",
            "checks": {
                "agent_runtime": {
                    "htp_cod_status": 503,
                    "hhtp_cod_desc": "Service Unavailable",
                },
                "agent_backend": {
                    "contas": {
                        "htp_cod_status": 204,
                        "hhtp_cod_desc": "No Content",
                    },
                    "oferta": {
                        "htp_cod_status": 200,
                        "hhtp_cod_desc": "oferta healthy",
                    },
                },
                "stt": {
                    "sofya": {
                        "htp_cod_status": 200,
                        "hhtp_cod_desc": "sofya ready",
                    },
                },
                "tts": {
                    "xAI": {
                        "htp_cod_status": 200,
                        "hhtp_cod_desc": "xAI TTS websocket connected",
                    },
                },
            },
        }
        assert backend_calls == [("conta", {}), ("oferta", {})]

    asyncio.run(_run())


def test_configured_agent_backend_service_uses_real_env_urls_when_backend_is_fake(monkeypatch) -> None:
    async def _run() -> None:
        manager = BridgeReadinessManager()
        monkeypatch.setenv("AGENT_BACKEND", "remote_ws_fake")
        monkeypatch.setenv("REMOTE_AGENT_HEALTH_URL_CONTA", "http://contas.example/health")
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL_OFERTA", raising=False)
        monkeypatch.delenv("REMOTE_AGENT_HEALTH_URL_OFERTAS", raising=False)
        monkeypatch.setenv("REMOTE_AGENT_SSE_URL_OFERTA", "https://oferta.example/agent/execute")
        monkeypatch.setenv("REMOTE_AGENT_SSE_TLS_VERIFY_OFERTA", "0")
        monkeypatch.delenv("REMOTE_AGENT_SSE_TLS_VERIFY", raising=False)
        calls: list[tuple[str, str, bool]] = []

        async def _fake_probe(check_name: str, url: str, *, verify: bool = True) -> ResourceCheck:
            calls.append((check_name, url, verify))
            return ResourceCheck(status="ok", http_status_code=200, http_status_desc="OK")

        with mock.patch.object(manager, "_probe_http_health", side_effect=_fake_probe):
            contas = await manager.probe_configured_agent_backend_service(agent_name="conta")
            oferta = await manager.probe_configured_agent_backend_service(agent_name="oferta")

        assert contas.status == "ok"
        assert oferta.status == "ok"
        assert calls == [
            ("agent_backend", "http://contas.example/health", True),
            ("agent_backend", "https://oferta.example/health", False),
        ]

    asyncio.run(_run())


def test_health_resources_route_returns_503(monkeypatch) -> None:
    report = ReadinessReport(
        checks={
            "agent_runtime": ResourceCheck(status="fail", message="down"),
            "agent_backend": ResourceCheck(status="ok"),
            "stt": ResourceCheck(status="ok"),
            "tts": ResourceCheck(status="ok"),
        },
        failed_resources=["agent_runtime"],
        active_connections=2,
        max_connections=5,
        cached=False,
    )
    fake_manager = _FakeReadinessManager(report=report)
    monkeypatch.setattr(main_module, "READINESS_GATE", fake_manager)

    with TestClient(main_module.app) as client:
        response = client.get("/health/resources")

    assert response.status_code == 503
    assert response.json()["failed_resources"] == ["agent_runtime"]
    assert fake_manager.default_calls == 1


def test_health_services_route_returns_503(monkeypatch) -> None:
    services_report = ServicesHealthReport(
        checks={
            "agent_runtime": {
                "htp_cod_status": 500,
                "hhtp_cod_desc": "FAIL",
            },
            "agent_backend": {
                "contas": {
                    "htp_cod_status": 200,
                    "hhtp_cod_desc": "SUCCESS",
                },
                "oferta": {
                    "htp_cod_status": 200,
                    "hhtp_cod_desc": "SUCCESS",
                },
            },
            "stt": {
                "sofya": {
                    "htp_cod_status": 200,
                    "hhtp_cod_desc": "SUCCESS",
                },
            },
            "tts": {
                "xAI": {
                    "htp_cod_status": 200,
                    "hhtp_cod_desc": "SUCCESS",
                },
            },
        },
        healthy=False,
    )
    fake_manager = _FakeReadinessManager()
    fake_manager.services_report = services_report
    monkeypatch.setattr(main_module, "READINESS_GATE", fake_manager)

    with TestClient(main_module.app) as client:
        response = client.get("/health/services")

    assert response.status_code == 503
    assert response.json() == services_report.as_dict()
    assert fake_manager.services_calls == 1


def test_ws_agent_rejects_on_capacity_with_stop_data(monkeypatch) -> None:
    lease = _FakeLease()
    fake_manager = _FakeReadinessManager(
        reservation=CapacityReservation(
            allowed=False,
            lease=lease,
            active_connections=1,
            max_connections=1,
        )
    )
    monkeypatch.setattr(main_module, "READINESS_GATE", fake_manager)
    monkeypatch.setenv("LIVEKIT_URL", "ws://127.0.0.1:7880")

    with TestClient(main_module.app) as client:
        with client.websocket_connect("/ws/agent") as websocket:
            websocket.send_text(json.dumps(_build_start_payload()))
            payload = websocket.receive_json()

    assert payload == build_capacity_stop_message()
    assert fake_manager.evaluate_calls == []
    assert lease.release_calls == 0


def test_ws_agent_rejects_on_readiness_failure_and_releases_capacity(monkeypatch) -> None:
    lease = _FakeLease()
    report = ReadinessReport(
        checks={
            "agent_runtime": ResourceCheck(status="ok"),
            "agent_backend": ResourceCheck(status="ok"),
            "stt": ResourceCheck(status="fail", message="down"),
            "tts": ResourceCheck(status="ok"),
        },
        failed_resources=["stt"],
        active_connections=1,
        max_connections=2,
        cached=False,
    )
    fake_manager = _FakeReadinessManager(
        reservation=CapacityReservation(
            allowed=True,
            lease=lease,
            active_connections=1,
            max_connections=2,
        ),
        report=report,
    )
    monkeypatch.setattr(main_module, "READINESS_GATE", fake_manager)
    monkeypatch.setenv("LIVEKIT_URL", "ws://127.0.0.1:7880")

    with TestClient(main_module.app) as client:
        with client.websocket_connect("/ws/agent") as websocket:
            websocket.send_text(json.dumps(_build_start_payload()))
            ready_payload = websocket.receive_json()
            while True:
                message = websocket.receive()
                if message.get("text") is not None:
                    payload = json.loads(message["text"])
                    break
    assert ready_payload["type"] == "ready"

    assert payload == {
        "type": "stop",
        "data": {
            "status": "stop_stt_unavailable",
            "reason": "resource_unhealthy",
            "resource": "stt",
            "failed_resources": ["stt"],
            "phase": "in_session",
        },
    }
    assert fake_manager.evaluate_calls == [("conta", {})]
    assert lease.release_calls == 1


def test_ws_agent_sends_terminal_stop_on_in_session_bridge_failure(monkeypatch) -> None:
    lease = _FakeLease()
    fake_manager = _FakeReadinessManager(
        reservation=CapacityReservation(
            allowed=True,
            lease=lease,
            active_connections=1,
            max_connections=2,
        )
    )
    timeline = _FakeTimeline()
    monkeypatch.setattr(main_module, "READINESS_GATE", fake_manager)
    monkeypatch.setenv("LIVEKIT_URL", "ws://127.0.0.1:7880")

    class _FakeRoom:
        def __init__(self) -> None:
            self.handlers = {}
            self.disconnected = 0

        def on(self, event_name: str):
            def _decorator(fn):
                self.handlers[event_name] = fn
                return fn

            return _decorator

        async def disconnect(self) -> None:
            self.disconnected += 1

    class _FakeAudioSource:
        def __init__(self, *args, **kwargs) -> None:
            self.args = args
            self.kwargs = kwargs

    class _FakeLocalAudioTrack:
        @staticmethod
        def create_audio_track(name: str, source: object) -> object:
            return {"name": name, "source": source}

    class _FakeTrackPublishOptions:
        def __init__(self) -> None:
            self.source = None

    class _FakeTrackSource:
        SOURCE_MICROPHONE = "microphone"

    async def _pending(*args, **kwargs) -> None:
        await asyncio.Event().wait()

    async def _raise_dispatch(*args, **kwargs) -> None:
        raise RuntimeError("bridge exploded")

    finalize_calls: list[bool] = []

    class _FakeRecorder:
        async def finalize(self) -> None:
            finalize_calls.append(True)

    bootstrap = SimpleNamespace(
        call_id_ged="GED-001",
        call_config={},
        remote_agent_context={"agent": "conta"},
        room_name="dev-room-123",
        identity="ws-bridge-dev-001",
        token="token-001",
        protocol="PRT-001",
        phone_number="5511999999999",
        timeline=timeline,
        dispatch_metadata={},
    )

    monkeypatch.setattr(main_module, "build_bridge_session_bootstrap", lambda **kwargs: bootstrap)
    monkeypatch.setattr(
        main_module,
        "create_entire_call_recorder_from_env",
        lambda **kwargs: _FakeRecorder(),
    )
    monkeypatch.setattr(main_module, "dispatch_agent", _raise_dispatch)
    monkeypatch.setattr(main_module, "connect_publish_livekit", _pending)
    monkeypatch.setattr(main_module, "publish_queue_to_livekit", _pending)
    monkeypatch.setattr(main_module, "ws_out_loop", _pending)
    monkeypatch.setattr(main_module, "notify_client_audio_enabled", _pending)
    monkeypatch.setattr(main_module.rtc, "Room", _FakeRoom)
    monkeypatch.setattr(main_module.rtc, "AudioSource", _FakeAudioSource)
    monkeypatch.setattr(main_module.rtc, "LocalAudioTrack", _FakeLocalAudioTrack)
    monkeypatch.setattr(main_module.rtc, "TrackPublishOptions", _FakeTrackPublishOptions)
    monkeypatch.setattr(main_module.rtc, "TrackSource", _FakeTrackSource)

    with TestClient(main_module.app) as client:
        with client.websocket_connect("/ws/agent") as websocket:
            websocket.send_text(json.dumps(_build_start_payload()))
            ready_payload = websocket.receive_json()
            stop_payload = websocket.receive_json()

    assert ready_payload["type"] == "ready"
    assert stop_payload == build_bridge_failed_stop_message()
    assert fake_manager.evaluate_calls == [("conta", {})]
    assert lease.release_calls == 1
    assert finalize_calls == [True]
