import json
import threading
import time

from fastapi.testclient import TestClient
from sqlalchemy import select

import services.ai_voice_runtime_service.app as runtime_module
from services.shared.core import utc_now_iso
from services.shared.db import get_session
from services.shared.models import VoiceAIStartIn
from services.shared.sql_models import VoiceAISessionRow, VoiceTranscriptSegmentRow, VoiceTTSSettingsRow
from services.shared.voice_tts_config import save_voice_tts_config, voice_tts_default_config


def _admin_headers() -> dict[str, str]:
    return {"X-User": "admin", "X-Role": "admin"}


def _reset_voice_tts_settings() -> None:
    session = get_session()
    try:
        row = session.execute(select(VoiceTTSSettingsRow)).scalar_one_or_none()
        if row is not None:
            session.delete(row)
            session.commit()
    finally:
        session.close()


def test_create_voice_ai_session_records_greeting_without_blocking_on_orchestrator(monkeypatch):
    allow_start = threading.Event()

    def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0):
        assert method == "POST"
        assert path.endswith("/start")
        assert allow_start.wait(timeout=2.0)
        return {
            "session_id": "ais_voice_demo",
            "language": "ru",
            "greeting_text": "Я AI-оператор компании. Здравствуйте.",
            "disclosure_required": True,
        }

    monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request)

    client = TestClient(runtime_module.app)
    response_holder: dict[str, object] = {}

    def _request() -> None:
        response_holder["response"] = client.post(
            "/internal/voice-ai/sessions",
            headers=_admin_headers(),
            json={
                "call_id": "call_voice_runtime_1",
                "linked_id": "linked_voice_runtime_1",
                "interaction_id": "int_voice_runtime_1",
                "queue_id": "que_voice_runtime_1",
                "caller_number": "+77010000031",
                "caller_name": "Runtime Caller",
                "agent_profile": "voice_support",
                "language_hint": "ru",
                "handoff_queue_id": "que_voice_runtime_1",
                "metadata": {"queue_code": "voice_lab"},
            },
        )

    request_thread = threading.Thread(target=_request, daemon=True)
    request_thread.start()
    request_thread.join(timeout=0.3)
    assert not request_thread.is_alive()

    response = response_holder["response"]
    assert response.status_code == 200
    payload = response.json()
    assert payload["ai_session_id"] is None
    assert payload["status"] == "greeting"

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1")
        ).scalar_one()
        assert voice_session.status == "greeting"
        assert voice_session.disclosure_played_at is None
        assert voice_session.last_ai_reply_at is None
        greeting_segment = session.execute(
            select(VoiceTranscriptSegmentRow)
            .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id)
            .where(VoiceTranscriptSegmentRow.speaker == "assistant")
        ).scalar_one()
        assert "AI-оператор" in greeting_segment.text
        assert greeting_segment.is_final is False
    finally:
        session.close()

    allow_start.set()
    deadline = time.time() + 2.0
    while time.time() < deadline:
        session = get_session()
        try:
            voice_session = session.execute(
                select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1")
            ).scalar_one()
            if voice_session.ai_session_id == "ais_voice_demo":
                greetings = session.execute(
                    select(VoiceTranscriptSegmentRow)
                    .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id)
                    .where(VoiceTranscriptSegmentRow.speaker == "assistant")
                ).scalars().all()
                assert len(greetings) == 1
                break
        finally:
            session.close()
        time.sleep(0.05)
    else:
        raise AssertionError("background voice-session start did not finish in time")


def test_create_voice_ai_session_records_greeting_without_blocking_on_orchestrator(monkeypatch):
    allow_start = threading.Event()

    def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0):
        assert method == "POST"
        assert path.endswith("/start")
        assert allow_start.wait(timeout=2.0)
        return {
            "session_id": "ais_voice_demo",
            "language": "ru",
            "greeting_text": "Здравствуйте. Подскажите, пожалуйста, чем помочь.",
            "disclosure_required": True,
        }

    monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request)

    client = TestClient(runtime_module.app)
    response_holder: dict[str, object] = {}

    def _request() -> None:
        response_holder["response"] = client.post(
            "/internal/voice-ai/sessions",
            headers=_admin_headers(),
            json={
                "call_id": "call_voice_runtime_1",
                "linked_id": "linked_voice_runtime_1",
                "interaction_id": "int_voice_runtime_1",
                "queue_id": "que_voice_runtime_1",
                "caller_number": "+77010000031",
                "caller_name": "Runtime Caller",
                "agent_profile": "voice_support",
                "language_hint": "ru",
                "handoff_queue_id": "que_voice_runtime_1",
                "metadata": {"queue_code": "voice_lab"},
            },
        )

    request_thread = threading.Thread(target=_request, daemon=True)
    request_thread.start()
    request_thread.join(timeout=0.3)
    assert not request_thread.is_alive()

    response = response_holder["response"]
    assert response.status_code == 200
    payload = response.json()
    assert payload["ai_session_id"] is None
    assert payload["status"] == "greeting"

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1")
        ).scalar_one()
        assert voice_session.status == "greeting"
        assert voice_session.disclosure_played_at is None
        assert voice_session.last_ai_reply_at is None
        greeting_segment = session.execute(
            select(VoiceTranscriptSegmentRow)
            .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id)
            .where(VoiceTranscriptSegmentRow.speaker == "assistant")
        ).scalar_one()
        assert "ai" not in greeting_segment.text.lower()
        assert greeting_segment.is_final is False
    finally:
        session.close()

    allow_start.set()
    deadline = time.time() + 2.0
    while time.time() < deadline:
        session = get_session()
        try:
            voice_session = session.execute(
                select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_1")
            ).scalar_one()
            if voice_session.ai_session_id == "ais_voice_demo":
                greetings = session.execute(
                    select(VoiceTranscriptSegmentRow)
                    .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id)
                    .where(VoiceTranscriptSegmentRow.speaker == "assistant")
                ).scalars().all()
                assert len(greetings) == 1
                break
        finally:
            session.close()
        time.sleep(0.05)
    else:
        raise AssertionError("background voice-session start did not finish in time")


def test_create_voice_ai_session_uses_database_selected_tts_provider(monkeypatch):
    _reset_voice_tts_settings()

    session = get_session()
    try:
        payload = voice_tts_default_config().model_dump()
        payload["provider"] = "elevenlabs"
        save_voice_tts_config(session, payload)
    finally:
        session.close()

    def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0):
        assert method == "POST"
        assert path.endswith("/start")
        return {
            "session_id": "ais_voice_tts_runtime",
            "language": "ru",
            "greeting_text": "Здравствуйте. Чем помочь?",
            "disclosure_required": False,
        }

    monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request)

    client = TestClient(runtime_module.app)
    response = client.post(
        "/internal/voice-ai/sessions",
        headers=_admin_headers(),
        json={
            "call_id": "call_voice_runtime_tts_provider",
            "linked_id": "linked_voice_runtime_tts_provider",
            "interaction_id": "int_voice_runtime_tts_provider",
            "queue_id": "que_voice_runtime_tts_provider",
            "caller_number": "+77010000032",
            "caller_name": "Runtime Caller",
            "agent_profile": "voice_support",
            "language_hint": "ru",
            "handoff_queue_id": "que_voice_runtime_tts_provider",
            "metadata": {"queue_code": "voice_lab_ai"},
        },
    )

    assert response.status_code == 200

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == "call_voice_runtime_tts_provider")
        ).scalar_one()
        assert voice_session.tts_provider == "elevenlabs"
    finally:
        session.close()
        _reset_voice_tts_settings()


def test_mark_reply_delivered_finalizes_only_matching_pending_assistant_segment():
    now = utc_now_iso()
    session = get_session()
    try:
        session.add(
            VoiceAISessionRow(
                session_id="avs_runtime_delivery",
                call_id="call_runtime_delivery",
                linked_id="linked_runtime_delivery",
                interaction_id="int_runtime_delivery",
                customer_id=None,
                queue_id="que_runtime_delivery",
                ai_session_id="ais_runtime_delivery",
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="speaking",
                handoff_reason=None,
                handoff_target_queue_id="que_runtime_delivery",
                disclosure_played_at=None,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=now,
                updated_at=now,
                ended_at=None,
            )
        )
        session.add(
            VoiceTranscriptSegmentRow(
                segment_id="vts_runtime_delivery_greeting",
                session_id="avs_runtime_delivery",
                call_id="call_runtime_delivery",
                interaction_id="int_runtime_delivery",
                speaker="assistant",
                source_type="tts",
                sequence_no=1,
                text="greeting",
                confidence=None,
                is_final=False,
                barge_in_interrupted=False,
                payload_json=json.dumps({"kind": "greeting", "delivery_status": "planned"}),
                created_at=now,
            )
        )
        session.add(
            VoiceTranscriptSegmentRow(
                segment_id="vts_runtime_delivery_pending",
                session_id="avs_runtime_delivery",
                call_id="call_runtime_delivery",
                interaction_id="int_runtime_delivery",
                speaker="assistant",
                source_type="tts",
                sequence_no=2,
                text="reply that was spoken",
                confidence=None,
                is_final=False,
                barge_in_interrupted=False,
                payload_json=json.dumps({"delivery_status": "planned"}),
                created_at=now,
            )
        )
        session.commit()
    finally:
        session.close()

    runtime_module._mark_reply_delivered(
        "avs_runtime_delivery",
        "reply that was spoken",
        False,
    )

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_runtime_delivery")
        ).scalar_one()
        segments = session.execute(
            select(VoiceTranscriptSegmentRow)
            .where(VoiceTranscriptSegmentRow.session_id == "avs_runtime_delivery")
            .order_by(VoiceTranscriptSegmentRow.sequence_no.asc())
        ).scalars().all()
        assert voice_session.last_ai_reply_at is not None
        assert [segment.is_final for segment in segments] == [False, True]
        payloads = [json.loads(segment.payload_json or "{}") for segment in segments]
        assert payloads[0]["delivery_status"] == "planned"
        assert payloads[1]["delivery_status"] == "delivered"
    finally:
        session.close()


def test_process_voice_ai_turn_triggers_bridge_handoff(monkeypatch):
    handoffs: list[dict] = []

    def _fake_bridge_request(method: str, path: str, *, payload=None, timeout=10.0):
        handoffs.append({"method": method, "path": path, "payload": payload, "timeout": timeout})
        return {"ok": True}

    def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0):
        assert method == "POST"
        assert path.endswith("/turns")
        return {
            "language": "ru",
            "intent": "handoff_request",
            "reply_text": "Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора.",
            "confidence": 0.2,
            "needs_handoff": True,
            "handoff_reason": "Нужен живой оператор.",
            "case_action": "keep_open",
            "kb_refs": [],
            "summary_text": "AI собрал контекст и запросил человека.",
            "model": "stub-voice",
            "latency_ms": 1,
            "status": "handoff_requested",
            "metadata": {
                "voice_start_language": "ru",
                "customer_name_status": "name_obtained",
                "customer_name_value": "Айдос",
                "customer_name_source": "voice_followup",
            },
        }

    monkeypatch.setattr(runtime_module, "_bridge_request", _fake_bridge_request)
    monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request)

    session = get_session()
    try:
        session.add(
            VoiceAISessionRow(
                session_id="avs_runtime_handoff",
                call_id="call_voice_runtime_handoff",
                linked_id="linked_voice_runtime_handoff",
                interaction_id="int_voice_runtime_handoff",
                customer_id=None,
                queue_id="que_voice_runtime_handoff",
                ai_session_id="ais_runtime_handoff",
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="listening",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_runtime_handoff",
                disclosure_played_at=utc_now_iso(),
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=utc_now_iso(),
                updated_at=utc_now_iso(),
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    client = TestClient(runtime_module.app)
    response = client.post(
        "/internal/voice-ai/sessions/avs_runtime_handoff/turns",
        headers=_admin_headers(),
        json={
            "voice_session_id": "avs_runtime_handoff",
            "call_id": "call_voice_runtime_handoff",
            "interaction_id": "int_voice_runtime_handoff",
            "transcript_text": "Соедините с оператором",
            "language": "ru",
            "sequence_no": 1,
            "barge_in": False,
            "metadata": {"turn_duration_ms": 3200},
        },
    )

    assert response.status_code == 200
    assert response.json()["needs_handoff"] is True
    assert handoffs
    assert handoffs[0]["path"] == "/internal/voice-ai/calls/call_voice_runtime_handoff/handoff"
    assert handoffs[0]["payload"]["summary"]["customer_name_status"] == "name_obtained"
    assert handoffs[0]["payload"]["summary"]["customer_name_value"] == "Айдос"
    assert handoffs[0]["payload"]["summary"]["customer_name_source"] == "voice_followup"
    assert handoffs[0]["payload"]["reason"] == "Нужен живой оператор."

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_runtime_handoff")
        ).scalar_one()
        assert voice_session.status == "handoff_requested"
        segments = session.execute(
            select(VoiceTranscriptSegmentRow)
            .where(VoiceTranscriptSegmentRow.session_id == "avs_runtime_handoff")
            .order_by(VoiceTranscriptSegmentRow.sequence_no.asc())
        ).scalars().all()
        assert [item.speaker for item in segments] == ["caller", "assistant"]
        assert [item.is_final for item in segments] == [True, False]
    finally:
        session.close()


def test_register_media_bridge_updates_voice_session(monkeypatch):
    closed: list[tuple[str, str]] = []
    monkeypatch.setattr(
        runtime_module._MEDIA_RUNTIME,
        "close_session_sync",
        lambda session_id, *, reason: closed.append((session_id, reason)),
    )

    session = get_session()
    try:
        session.add(
            VoiceAISessionRow(
                session_id="avs_media_bridge_runtime",
                call_id="call_media_bridge_runtime",
                linked_id="linked_media_bridge_runtime",
                interaction_id="int_media_bridge_runtime",
                customer_id=None,
                queue_id="que_media_bridge_runtime",
                ai_session_id="ais_media_bridge_runtime",
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="greeting",
                handoff_reason=None,
                handoff_target_queue_id="que_media_bridge_runtime",
                media_uuid=None,
                media_status=None,
                media_connected_at=None,
                media_ended_at=None,
                last_media_frame_at=None,
                disclosure_played_at=None,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=utc_now_iso(),
                updated_at=utc_now_iso(),
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()


def test_complete_voice_ai_session_start_persists_voice_start_result_and_requests_handoff(monkeypatch):
    bridge_calls: list[dict] = []

    def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0):
        assert method == "POST"
        assert path.endswith("/start")
        return {
            "session_id": "ais_voice_start_done",
            "language": "kz",
            "greeting_text": "",
            "disclosure_required": False,
            "needs_handoff": True,
            "handoff_reason": "voice_start_completed",
            "summary_text": "Voice start completed with status name_followup_required and candidate name Айдос.",
            "start_result": {
                "language": "kz",
                "customer_id": "cus_voice_start",
                "customer_name_status": "name_followup_required",
                "customer_name_value": "Айдос",
                "customer_name_source": "voice_start",
                "downstream_queue_id": "que_voice_support_kz",
                "downstream_queue_code": "voice_support_kz",
                "resolved_at": "2040-01-01T10:00:00+00:00",
            },
        }

    def _fake_bridge_request(method: str, path: str, *, payload=None, timeout=10.0):
        bridge_calls.append({"method": method, "path": path, "payload": payload, "timeout": timeout})
        return {"ok": True}

    monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request)
    monkeypatch.setattr(runtime_module, "_bridge_request", _fake_bridge_request)

    now = utc_now_iso()
    session = get_session()
    try:
        session.add(
            VoiceAISessionRow(
                session_id="avs_voice_start_complete",
                call_id="call_voice_start_complete",
                linked_id="linked_voice_start_complete",
                interaction_id="int_voice_start_complete",
                customer_id=None,
                queue_id="que_voice_start_kz",
                ai_session_id=None,
                agent_profile="voice_start",
                language="kz",
                asr_provider="openai",
                tts_provider="openai",
                status="greeting",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_support_kz",
                disclosure_played_at=None,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=now,
                updated_at=now,
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    runtime_module._complete_voice_ai_session_start(
        "avs_voice_start_complete",
        VoiceAIStartIn(
            voice_session_id="avs_voice_start_complete",
            call_id="call_voice_start_complete",
            interaction_id="int_voice_start_complete",
            customer_id=None,
            language_hint="kz",
            agent_profile="voice_start",
            metadata={"stage": "voice_start", "next_queue_code": "voice_support_kz"},
        ),
    )

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_voice_start_complete")
        ).scalar_one()
        assert voice_session.ai_session_id == "ais_voice_start_done"
        assert voice_session.status == "handoff_requested"
        assert voice_session.voice_start_language == "kz"
        assert voice_session.customer_name_status == "name_followup_required"
        assert voice_session.customer_name_value == "Айдос"
        assert voice_session.customer_name_source == "voice_start"
    finally:
        session.close()

    relevant_calls = [call for call in bridge_calls if "call_voice_start_complete" in call["path"]]
    assert [call["path"] for call in relevant_calls] == [
        "/internal/voice-ai/calls/call_voice_start_complete/state",
        "/internal/voice-ai/calls/call_voice_start_complete/handoff",
    ]
    assert relevant_calls[1]["payload"]["metadata"]["customer_name_status"] == "name_followup_required"
    assert relevant_calls[1]["payload"]["metadata"]["customer_name_value"] == "Айдос"

 

def test_complete_voice_ai_session_start_handles_disabled_name_collection_without_greeting(monkeypatch):
    bridge_calls: list[dict] = []

    def _fake_orchestrator_request(method: str, path: str, *, payload=None, timeout=10.0):
        assert method == "POST"
        assert path.endswith("/start")
        return {
            "session_id": "ais_voice_start_disabled",
            "language": "ru",
            "greeting_text": "",
            "disclosure_required": False,
            "needs_handoff": True,
            "handoff_reason": "voice_start_completed",
            "summary_text": "Voice start name collection is disabled and handed off without a name.",
            "start_result": {
                "language": "ru",
                "customer_id": None,
                "customer_name_status": "name_not_obtained",
                "customer_name_value": None,
                "customer_name_source": "none",
                "downstream_queue_id": "que_voice_support_disabled",
                "downstream_queue_code": "voice_support_disabled",
                "resolved_at": "2040-01-01T11:00:00+00:00",
            },
        }

    def _fake_bridge_request(method: str, path: str, *, payload=None, timeout=10.0):
        bridge_calls.append({"method": method, "path": path, "payload": payload, "timeout": timeout})
        return {"ok": True}

    monkeypatch.setattr(runtime_module, "_orchestrator_request", _fake_orchestrator_request)
    monkeypatch.setattr(runtime_module, "_bridge_request", _fake_bridge_request)

    now = utc_now_iso()
    session = get_session()
    try:
        session.add(
            VoiceAISessionRow(
                session_id="avs_voice_start_disabled",
                call_id="call_voice_start_disabled",
                linked_id="linked_voice_start_disabled",
                interaction_id="int_voice_start_disabled",
                customer_id=None,
                queue_id="que_voice_start_disabled",
                ai_session_id=None,
                agent_profile="voice_start",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="greeting",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_support_disabled",
                disclosure_played_at=None,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=now,
                updated_at=now,
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    runtime_module._complete_voice_ai_session_start(
        "avs_voice_start_disabled",
        VoiceAIStartIn(
            voice_session_id="avs_voice_start_disabled",
            call_id="call_voice_start_disabled",
            interaction_id="int_voice_start_disabled",
            customer_id=None,
            language_hint="ru",
            agent_profile="voice_start",
            metadata={"stage": "voice_start", "next_queue_code": "voice_support_disabled"},
        ),
    )

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_voice_start_disabled")
        ).scalar_one()
        assert voice_session.ai_session_id == "ais_voice_start_disabled"
        assert voice_session.status == "handoff_requested"
        assert voice_session.customer_name_status == "name_not_obtained"
        segments = session.execute(
            select(VoiceTranscriptSegmentRow)
            .where(VoiceTranscriptSegmentRow.session_id == "avs_voice_start_disabled")
            .where(VoiceTranscriptSegmentRow.speaker == "assistant")
        ).scalars().all()
        assert segments == []
    finally:
        session.close()

    relevant_calls = [call for call in bridge_calls if "call_voice_start_disabled" in call["path"]]
    assert [call["path"] for call in relevant_calls] == [
        "/internal/voice-ai/calls/call_voice_start_disabled/state",
        "/internal/voice-ai/calls/call_voice_start_disabled/handoff",
    ]


def test_register_media_bridge_updates_voice_session_round_trip(monkeypatch):
    closed: list[tuple[str, str]] = []
    monkeypatch.setattr(
        runtime_module._MEDIA_RUNTIME,
        "close_session_sync",
        lambda session_id, *, reason: closed.append((session_id, reason)),
    )

    session = get_session()
    try:
        session.add(
            VoiceAISessionRow(
                session_id="avs_media_bridge_runtime_rt",
                call_id="call_media_bridge_runtime_rt",
                linked_id="linked_media_bridge_runtime_rt",
                interaction_id="int_media_bridge_runtime_rt",
                customer_id=None,
                queue_id="que_media_bridge_runtime_rt",
                ai_session_id="ais_media_bridge_runtime_rt",
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="greeting",
                handoff_reason=None,
                handoff_target_queue_id="que_media_bridge_runtime_rt",
                media_uuid=None,
                media_status=None,
                media_connected_at=None,
                media_ended_at=None,
                last_media_frame_at=None,
                disclosure_played_at=None,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=utc_now_iso(),
                updated_at=utc_now_iso(),
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    client = TestClient(runtime_module.app)
    requested = client.post(
        "/internal/voice-ai/sessions/avs_media_bridge_runtime_rt/media-bridge",
        headers=_admin_headers(),
        json={
            "event_type": "requested",
            "media_uuid": "75f4d61f-f674-4bb4-91c1-8ddfcf2fc2b4",
            "call_id": "call_media_bridge_runtime_rt",
            "linked_id": "linked_media_bridge_runtime_rt",
            "channel": "PJSIP/1001-00000001",
            "service_address": "127.0.0.1:9019",
        },
    )
    assert requested.status_code == 200

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == "avs_media_bridge_runtime_rt")
        ).scalar_one()
        assert voice_session.media_uuid == "75f4d61f-f674-4bb4-91c1-8ddfcf2fc2b4"
        assert voice_session.media_status == "requested"
    finally:
        session.close()

    ended = client.post(
        "/internal/voice-ai/sessions/avs_media_bridge_runtime_rt/media-bridge",
        headers=_admin_headers(),
        json={
            "event_type": "ended",
            "media_uuid": "75f4d61f-f674-4bb4-91c1-8ddfcf2fc2b4",
            "call_id": "call_media_bridge_runtime_rt",
            "linked_id": "linked_media_bridge_runtime_rt",
            "channel": "PJSIP/1001-00000001",
            "service_address": "127.0.0.1:9019",
            "reason": "audiosocket_closed",
        },
    )
    assert ended.status_code == 200
    assert closed == [("avs_media_bridge_runtime_rt", "audiosocket_closed")]


def test_push_voice_ai_telephony_event_call_ended_returns_detached_safe_payload(monkeypatch):
    closed: list[tuple[str, str]] = []
    orchestrator_calls: list[tuple[str, str]] = []
    monkeypatch.setattr(
        runtime_module._MEDIA_RUNTIME,
        "close_session_sync",
        lambda session_id, *, reason: closed.append((session_id, reason)),
    )
    monkeypatch.setattr(
        runtime_module,
        "_orchestrator_request",
        lambda method, path, *, payload=None, timeout=10.0: orchestrator_calls.append((method, path)) or {"ok": True},
    )

    session = get_session()
    try:
        session.add(
            VoiceAISessionRow(
                session_id="avs_runtime_call_ended",
                call_id="call_voice_runtime_call_ended",
                linked_id="linked_voice_runtime_call_ended",
                interaction_id="int_voice_runtime_call_ended",
                customer_id=None,
                queue_id="que_voice_runtime_call_ended",
                ai_session_id="ais_runtime_call_ended",
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="listening",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_runtime_call_ended",
                media_uuid="2fc59167-fdc7-43a0-b4e5-c4948f7c24d8",
                media_status="connected",
                media_connected_at=utc_now_iso(),
                media_ended_at=None,
                last_media_frame_at=utc_now_iso(),
                disclosure_played_at=utc_now_iso(),
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=utc_now_iso(),
                updated_at=utc_now_iso(),
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    client = TestClient(runtime_module.app)
    response = client.post(
        "/internal/voice-ai/sessions/avs_runtime_call_ended/telephony-events",
        headers=_admin_headers(),
        json={
            "event_type": "call.ended",
            "payload": {"hangup_cause": "16"},
        },
    )

    assert response.status_code == 200
    assert response.json() == {
        "voice_session_id": "avs_runtime_call_ended",
        "ai_session_id": "ais_runtime_call_ended",
        "status": "completed",
    }
    assert closed == [("avs_runtime_call_ended", "call_ended")]
    assert orchestrator_calls == [("POST", "/ai/voice/sessions/avs_runtime_call_ended/close")]
