import json
from datetime import datetime, timedelta, timezone

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

import services.asterisk_bridge_service.app as bridge_module
import services.asterisk_bridge_service.ivr_fastagi as bridge_ivr_fastagi
from services.shared.db import get_session
from services.shared.core import utc_now_iso
from services.shared.models import VoiceAICallStateUpdateIn, VoiceAIHandoffRequestIn
from services.shared.sql_models import (
    AISessionRow,
    AsteriskCallActionLogRow,
    AsteriskCallLinkRow,
    AsteriskEventLogRow,
    CallRecordingRow,
    VoiceAISessionRow,
    VoiceEventRow,
    VoiceTranscriptSegmentRow,
)


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


def _operator_headers(user: str = "operator_a") -> dict[str, str]:
    return {"X-User": user, "X-Role": "operator"}


@pytest.fixture(autouse=True)
def _cleanup_bridge_background_threads():
    yield
    bridge_module._shutdown()


def _wait_until_stopped(stop_event=None):
    if stop_event is not None:
        stop_event.wait(60)


def test_asterisk_status_reports_queue_mapping(monkeypatch):
    monkeypatch.setenv("ASTERISK_BRIDGE_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_AMI_HOST", "10.10.10.10")
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"sales":"que_sales","support":"que_support"}')
    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_MODE", "bearer_first")
    monkeypatch.setenv("ASTERISK_WEBRTC_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_WEBRTC_WS_URL", "wss://pbx.example.test:8089/ws")

    client = TestClient(bridge_module.app)
    response = client.get("/asterisk/status", headers=_admin_headers())

    assert response.status_code == 200
    payload = response.json()
    assert payload["status"] == "ok"
    assert payload["ami_host"] == "10.10.10.10"
    assert payload["queue_codes_loaded"] == ["sales", "support"]
    assert payload["bridge_auth_mode"] == "bearer_first"
    assert payload["callcontrol_enabled"] is False
    assert payload["webrtc_enabled"] is True
    assert payload["webrtc_ws_url"] == "wss://pbx.example.test:8089/ws"


def test_browser_softphone_config_route_returns_mapped_config(monkeypatch):
    monkeypatch.setenv("ASTERISK_WEBRTC_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_WEBRTC_WS_URL", "wss://pbx.example.test:8089/ws")
    monkeypatch.setenv("ASTERISK_AMI_HOST", "pbx.example.test")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator":"2001"}')
    monkeypatch.setenv(
        "ASTERISK_BROWSER_SIP_MAP_JSON",
        '{"operator":{"authorization_username":"2001","password":"secret-2001","display_name":"Operator Browser"}}',
    )
    monkeypatch.setenv(
        "ASTERISK_WEBRTC_ICE_SERVERS_JSON",
        '[{"urls":["stun:stun.example.test:3478"]}]',
    )

    client = TestClient(bridge_module.app)
    response = client.get("/asterisk/browser-softphone/config", headers={"X-User": "operator", "X-Role": "operator"})

    assert response.status_code == 200
    payload = response.json()
    assert payload["enabled"] is True
    assert payload["ws_url"] == "wss://pbx.example.test:8089/ws"
    assert payload["authorization_username"] == "2001"
    assert payload["password"] == "secret-2001"
    assert payload["display_name"] == "Operator Browser"
    assert payload["operator_extension"] == "2001"
    assert payload["sip_uri"] == "sip:2001@pbx.example.test"
    assert payload["ice_servers"] == [{"urls": ["stun:stun.example.test:3478"]}]


def test_browser_softphone_config_route_returns_403_when_mapping_missing(monkeypatch):
    monkeypatch.setenv("ASTERISK_WEBRTC_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_WEBRTC_WS_URL", "wss://pbx.example.test:8089/ws")
    monkeypatch.setenv("ASTERISK_BROWSER_SIP_MAP_JSON", "{}")

    client = TestClient(bridge_module.app)
    response = client.get("/asterisk/browser-softphone/config", headers={"X-User": "operator", "X-Role": "operator"})

    assert response.status_code == 403
    assert response.json()["detail"] == "Browser softphone is not configured for this user"


def test_ai_voice_config_for_queue_promotes_legacy_ivr_ai_queues_to_voice_start(monkeypatch):
    monkeypatch.setenv("AI_VOICE_ENABLED", "1")
    monkeypatch.setenv(
        "AI_VOICE_QUEUE_CONFIG_JSON",
        json.dumps(
            {
                "ivr_support_ai_ru": {
                    "mode": "ai_first",
                    "agent_profile": "voice_support",
                    "handoff_queue_code": "ivr_support",
                    "language": "ru",
                }
            }
        ),
    )
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"ivr_support":"que_ivr_support"}')

    config = bridge_module._ai_voice_config_for_queue("ivr_support_ai_ru")

    assert config is not None
    assert config["agent_profile"] == "voice_start"
    assert config["stage"] == "voice_start"
    assert config["handoff_queue_id"] == "que_ivr_support"


def test_ai_voice_config_for_queue_preserves_non_ivr_profiles(monkeypatch):
    monkeypatch.setenv("AI_VOICE_ENABLED", "1")
    monkeypatch.setenv(
        "AI_VOICE_QUEUE_CONFIG_JSON",
        json.dumps(
            {
                "voice_lab_ai": {
                    "mode": "ai_first",
                    "agent_profile": "voice_support",
                    "handoff_queue_code": "voice_lab",
                    "language": "ru",
                }
            }
        ),
    )
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"voice_lab":"que_voice_lab"}')

    config = bridge_module._ai_voice_config_for_queue("voice_lab_ai")

    assert config is not None
    assert config["agent_profile"] == "voice_support"
    assert config["stage"] is None
    assert config["handoff_queue_id"] == "que_voice_lab"


def test_post_json_bearer_first_falls_back_to_legacy(monkeypatch):
    class DummyResponse:
        def __init__(self, status_code: int, body: dict):
            self.status_code = status_code
            self._body = body

        def raise_for_status(self):
            if self.status_code >= 400:
                raise bridge_module.httpx.HTTPStatusError(
                    "error",
                    request=None,
                    response=self,
                )

        def json(self):
            return self._body

    class DummyClient:
        seen_headers = []

        def __init__(self, **kwargs):
            self.kwargs = kwargs

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def request(self, method, url, headers=None, **kwargs):
            DummyClient.seen_headers.append(headers or {})
            if headers and headers.get("Authorization"):
                return DummyResponse(401, {"detail": "unauthorized"})
            return DummyResponse(200, {"ok": True})

    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_MODE", "bearer_first")
    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_FALLBACK_LEGACY", "1")
    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_USER", "ast-bridge")
    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_ROLE", "admin")
    monkeypatch.setenv("APP_TOKEN_SECRET", "track9-test-secret")
    monkeypatch.setattr(bridge_module.httpx, "Client", DummyClient)

    result = bridge_module._post_json("http://example.internal/endpoint", {"a": 1})

    assert result == {"ok": True}
    assert len(DummyClient.seen_headers) == 2
    assert "Authorization" in DummyClient.seen_headers[0]
    assert DummyClient.seen_headers[1]["X-User"] == "ast-bridge"
    assert DummyClient.seen_headers[1]["X-Role"] == "admin"


def test_upload_recording_bearer_first_fallback_rewinds_file(monkeypatch, tmp_path):
    recording_path = tmp_path / "call.wav"
    recording_bytes = b"recording-payload"
    recording_path.write_bytes(recording_bytes)

    class DummyClient:
        def __init__(self, **kwargs):
            self.kwargs = kwargs

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

    class DummyResponse:
        def raise_for_status(self):
            return None

        def json(self):
            return {"recording_id": "rec_uploaded"}

    attempts: list[dict[str, object]] = []

    def _fake_request_with_bridge_auth(client, *, retry_reset=None, headers=None, files=None, **kwargs):
        del client, headers, kwargs
        handle = files["file"][1]
        attempts.append(
            {
                "chunk": handle.read(),
                "has_retry_reset": callable(retry_reset),
            }
        )
        assert callable(retry_reset)
        retry_reset()
        attempts.append(
            {
                "chunk": handle.read(),
                "has_retry_reset": callable(retry_reset),
            }
        )
        return DummyResponse()

    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_MODE", "bearer_first")
    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_FALLBACK_LEGACY", "1")
    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_USER", "ast-bridge")
    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_ROLE", "admin")
    monkeypatch.setenv("APP_TOKEN_SECRET", "track9-test-secret")
    monkeypatch.setattr(bridge_module.httpx, "Client", DummyClient)
    monkeypatch.setattr(bridge_module, "_request_with_bridge_auth", _fake_request_with_bridge_auth)

    result = bridge_module._upload_recording(
        local_path=recording_path,
        call_id="call_upload_fallback",
        interaction_id="int_upload_fallback",
        source_event_id="src_upload_fallback",
        file_name="call.wav",
        mime_type="audio/wav",
        duration_seconds=3,
    )

    assert result == {"recording_id": "rec_uploaded"}
    assert attempts == [
        {"chunk": recording_bytes, "has_retry_reset": True},
        {"chunk": recording_bytes, "has_retry_reset": True},
    ]


def test_assign_interaction_uses_patch(monkeypatch):
    class DummyResponse:
        status_code = 200

        def raise_for_status(self):
            return None

        def json(self):
            return {"ok": True}

    class DummyClient:
        seen: list[dict[str, str]] = []

        def __init__(self, **kwargs):
            self.kwargs = kwargs

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def request(self, method, url, headers=None, **kwargs):
            DummyClient.seen.append({"method": method, "url": url, "timeout": self.kwargs.get("timeout")})
            return DummyResponse()

    monkeypatch.setenv("ASTERISK_BRIDGE_AUTH_MODE", "legacy_headers")
    monkeypatch.setattr(bridge_module.httpx, "Client", DummyClient)

    result = bridge_module._assign_interaction(interaction_id="int_demo", assignee="operator_a")

    assert result == {"ok": True}
    assert DummyClient.seen == [
        {"method": "PATCH", "url": "http://localhost:8004/interactions/int_demo/assign", "timeout": 2.5}
    ]


def test_assign_interaction_uses_fast_callcontrol_request_policy(monkeypatch):
    captured: list[dict[str, object]] = []

    def _fake_patch_json(url, payload, **kwargs):
        captured.append({"url": url, "payload": payload, **kwargs})
        return {"ok": True}

    monkeypatch.setattr(bridge_module, "_patch_json", _fake_patch_json)

    result = bridge_module._assign_interaction(interaction_id="int_fast", assignee="operator_a")

    assert result == {"ok": True}
    assert captured == [
        {
            "url": "http://localhost:8004/interactions/int_fast/assign",
            "payload": {"assignee": "operator_a"},
            "timeout_seconds": 2.5,
            "max_attempts": 1,
            "retry_backoff_seconds": 0.0,
        }
    ]


def test_ivr_fastagi_runtime_completes_and_sets_transfer_target(monkeypatch):
    class DummyBridge:
        def _queue_map(self):
            return {"voice_lab_ivr": "q_ivr", "support": "q_support", "voice_lab": "q_voice"}

        def _ivr_dtmf_timeout_seconds(self):
            return 5

        def _ivr_max_no_input_retries(self):
            return 2

        def _ivr_max_invalid_retries(self):
            return 2

    class DummyChannel:
        def __init__(self):
            self.variables: dict[str, str | None] = {}
            self.playbacks: list[str] = []
            self.waits: list[int] = []
            self.verbose_messages: list[str] = []

        def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None:
            self.playbacks.append(prompt_audio_key)
            return None

        def wait_for_digit(self, timeout_ms: int) -> str | None:
            self.waits.append(timeout_ms)
            return "1"

        def set_variable(self, name: str, value: str | None) -> None:
            self.variables[name] = value

        def verbose(self, message: str, *, level: int = 1) -> None:
            self.verbose_messages.append(f"{level}:{message}")

    calls: list[tuple[str, object]] = []

    monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge())
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_wait_for_call_link_context",
        lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"},
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_resolve_transfer_extension",
        lambda queue_code: {"support": "7010", "voice_lab": "7000"}.get(queue_code),
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_start_session",
        lambda call_context, interaction_id, queue_id: calls.append(("start", queue_id)) or {
            "session": {"session_id": "ivs_fastagi"},
            "current_node": {
                "node_id": "root",
                "prompt_audio_key": "ivr/demo-root",
                "options": [{"digit": "1", "target_node_id": "support"}],
            },
            "completed": False,
        },
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_send_dtmf",
        lambda session_id, digit: calls.append(("dtmf", digit)) or {
            "session": {
                "session_id": session_id,
                "status": "completed",
                "resolved_queue_id": "q_support",
                "resolved_queue_code": "support",
            },
            "current_node": {"node_id": "support", "is_terminal": True},
            "completed": True,
        },
    )
    monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: calls.append(("abandon", session_id)))

    channel = DummyChannel()
    request = bridge_ivr_fastagi.AgiRequest(
        env={},
        args=["call_fastagi", "linked_fastagi", "voice_lab_ivr", "voice_lab", "7200"],
    )

    bridge_ivr_fastagi.run_ivr_call(channel, request)

    assert calls == [("start", "q_ivr"), ("dtmf", "1")]
    assert channel.playbacks == ["ivr/demo-root"]
    assert channel.waits == [5000]
    assert channel.variables["MVPCC_IVR_SESSION_ID"] == "ivs_fastagi"
    assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "support"
    assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7010"
    assert channel.variables["MVPCC_IVR_RESULT"] == "completed"


def test_ivr_fastagi_runtime_falls_back_after_no_input_retries(monkeypatch):
    class DummyBridge:
        def _queue_map(self):
            return {"voice_lab_ivr": "q_ivr", "voice_lab": "q_voice"}

        def _ivr_dtmf_timeout_seconds(self):
            return 5

        def _ivr_max_no_input_retries(self):
            return 2

        def _ivr_max_invalid_retries(self):
            return 2

    class DummyChannel:
        def __init__(self):
            self.variables: dict[str, str | None] = {}
            self.wait_calls = 0

        def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None:
            return None

        def wait_for_digit(self, timeout_ms: int) -> str | None:
            self.wait_calls += 1
            return None

        def set_variable(self, name: str, value: str | None) -> None:
            self.variables[name] = value

        def verbose(self, message: str, *, level: int = 1) -> None:
            return None

    no_input_calls: list[str] = []
    abandoned: list[str] = []

    monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge())
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_wait_for_call_link_context",
        lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"},
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_resolve_transfer_extension",
        lambda queue_code: {"voice_lab": "7000"}.get(queue_code),
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_start_session",
        lambda call_context, interaction_id, queue_id: {
            "session": {"session_id": "ivs_no_input"},
            "current_node": {
                "node_id": "root",
                "prompt_audio_key": "ivr/demo-root",
                "options": [{"digit": "1", "target_node_id": "support"}],
            },
            "completed": False,
        },
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_send_no_input",
        lambda session_id: no_input_calls.append(session_id) or {
            "session": {"session_id": session_id, "status": "active"},
            "current_node": {
                "node_id": "root",
                "prompt_audio_key": "ivr/demo-root",
                "options": [{"digit": "1", "target_node_id": "support"}],
            },
            "completed": False,
        },
    )
    monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: abandoned.append(session_id))

    channel = DummyChannel()
    request = bridge_ivr_fastagi.AgiRequest(
        env={},
        args=["call_no_input", "linked_no_input", "voice_lab_ivr", "voice_lab", "7200"],
    )

    bridge_ivr_fastagi.run_ivr_call(channel, request)

    assert no_input_calls == ["ivs_no_input", "ivs_no_input", "ivs_no_input"]
    assert abandoned == ["ivs_no_input"]
    assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "voice_lab"
    assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7000"
    assert channel.variables["MVPCC_IVR_RESULT"] == "fallback_no_input"


def test_ivr_fastagi_runtime_falls_back_after_invalid_digit_retries(monkeypatch):
    class DummyBridge:
        def _queue_map(self):
            return {"voice_lab_ivr": "q_ivr", "voice_lab": "q_voice"}

        def _ivr_dtmf_timeout_seconds(self):
            return 5

        def _ivr_max_no_input_retries(self):
            return 2

        def _ivr_max_invalid_retries(self):
            return 2

    class DummyChannel:
        def __init__(self):
            self.variables: dict[str, str | None] = {}
            self.digits = ["9", "9", "9"]

        def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None:
            return None

        def wait_for_digit(self, timeout_ms: int) -> str | None:
            return self.digits.pop(0)

        def set_variable(self, name: str, value: str | None) -> None:
            self.variables[name] = value

        def verbose(self, message: str, *, level: int = 1) -> None:
            return None

    dtmf_calls: list[str] = []
    abandoned: list[str] = []

    monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge())
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_wait_for_call_link_context",
        lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"},
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_resolve_transfer_extension",
        lambda queue_code: {"voice_lab": "7000"}.get(queue_code),
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_start_session",
        lambda call_context, interaction_id, queue_id: {
            "session": {"session_id": "ivs_invalid"},
            "current_node": {
                "node_id": "root",
                "prompt_audio_key": "ivr/demo-root",
                "options": [{"digit": "1", "target_node_id": "support"}],
            },
            "completed": False,
        },
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_send_dtmf",
        lambda session_id, digit: dtmf_calls.append(digit) or {
            "session": {"session_id": session_id, "status": "active"},
            "current_node": {
                "node_id": "root",
                "prompt_audio_key": "ivr/demo-root",
                "options": [{"digit": "1", "target_node_id": "support"}],
            },
            "completed": False,
        },
    )
    monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: abandoned.append(session_id))

    channel = DummyChannel()
    request = bridge_ivr_fastagi.AgiRequest(
        env={},
        args=["call_invalid", "linked_invalid", "voice_lab_ivr", "voice_lab", "7200"],
    )

    bridge_ivr_fastagi.run_ivr_call(channel, request)

    assert dtmf_calls == ["9", "9", "9"]
    assert abandoned == ["ivs_invalid"]
    assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "voice_lab"
    assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7000"
    assert channel.variables["MVPCC_IVR_RESULT"] == "fallback_invalid"


def test_ivr_fastagi_runtime_abandons_on_hangup(monkeypatch):
    class DummyBridge:
        def _queue_map(self):
            return {"voice_lab_ivr": "q_ivr"}

        def _ivr_dtmf_timeout_seconds(self):
            return 5

        def _ivr_max_no_input_retries(self):
            return 2

        def _ivr_max_invalid_retries(self):
            return 2

    class DummyChannel:
        def __init__(self):
            self.variables: dict[str, str | None] = {}

        def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None:
            return None

        def wait_for_digit(self, timeout_ms: int) -> str | None:
            raise bridge_ivr_fastagi.AgiHangup("caller hung up")

        def set_variable(self, name: str, value: str | None) -> None:
            self.variables[name] = value

        def verbose(self, message: str, *, level: int = 1) -> None:
            return None

    abandoned: list[str] = []

    monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge())
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_wait_for_call_link_context",
        lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"},
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_start_session",
        lambda call_context, interaction_id, queue_id: {
            "session": {"session_id": "ivs_hangup"},
            "current_node": {"node_id": "root", "options": []},
            "completed": False,
        },
    )
    monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: abandoned.append(session_id))

    channel = DummyChannel()
    request = bridge_ivr_fastagi.AgiRequest(
        env={},
        args=["call_hangup", "linked_hangup", "voice_lab_ivr", "voice_lab", "7200"],
    )

    bridge_ivr_fastagi.run_ivr_call(channel, request)

    assert abandoned == ["ivs_hangup"]


def test_ivr_fastagi_runtime_uses_digit_captured_during_prompt(monkeypatch):
    class DummyBridge:
        def _queue_map(self):
            return {"voice_lab_ivr": "q_ivr", "support": "q_support", "voice_lab": "q_voice"}

        def _ivr_dtmf_timeout_seconds(self):
            return 5

        def _ivr_max_no_input_retries(self):
            return 2

        def _ivr_max_invalid_retries(self):
            return 2

    class DummyChannel:
        def __init__(self):
            self.variables: dict[str, str | None] = {}
            self.waits: list[int] = []
            self.playbacks: list[str] = []

        def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None:
            self.playbacks.append(prompt_audio_key)
            return "2"

        def wait_for_digit(self, timeout_ms: int) -> str | None:
            self.waits.append(timeout_ms)
            return None

        def set_variable(self, name: str, value: str | None) -> None:
            self.variables[name] = value

        def verbose(self, message: str, *, level: int = 1) -> None:
            return None

    dtmf_calls: list[str] = []

    monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge())
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_wait_for_call_link_context",
        lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"},
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_resolve_transfer_extension",
        lambda queue_code: {"support": "7010", "voice_lab": "7000"}.get(queue_code),
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_start_session",
        lambda call_context, interaction_id, queue_id: {
            "session": {"session_id": "ivs_prompt_digit"},
            "current_node": {
                "node_id": "root",
                "prompt_audio_key": "ivr/demo-root",
                "options": [{"digit": "2", "target_node_id": "support"}],
            },
            "completed": False,
        },
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_send_dtmf",
        lambda session_id, digit: dtmf_calls.append(digit) or {
            "session": {
                "session_id": session_id,
                "status": "completed",
                "resolved_queue_id": "q_support",
                "resolved_queue_code": "support",
            },
            "current_node": {"node_id": "support", "is_terminal": True},
            "completed": True,
        },
    )
    monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: None)

    channel = DummyChannel()
    request = bridge_ivr_fastagi.AgiRequest(
        env={},
        args=["call_prompt_digit", "linked_prompt_digit", "voice_lab_ivr", "voice_lab", "7200"],
    )

    bridge_ivr_fastagi.run_ivr_call(channel, request)

    assert channel.playbacks == ["ivr/demo-root"]
    assert channel.waits == []
    assert dtmf_calls == ["2"]
    assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "support"
    assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7010"


def test_ivr_fastagi_runtime_plays_prompt_sequence_before_wait(monkeypatch):
    class DummyBridge:
        def _queue_map(self):
            return {"voice_lab_ivr": "q_ivr", "support": "q_support", "voice_lab": "q_voice"}

        def _ivr_dtmf_timeout_seconds(self):
            return 5

        def _ivr_max_no_input_retries(self):
            return 2

        def _ivr_max_invalid_retries(self):
            return 2

    class DummyChannel:
        def __init__(self):
            self.variables: dict[str, str | None] = {}
            self.playbacks: list[tuple[str, int]] = []
            self.waits: list[int] = []

        def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None:
            self.playbacks.append((prompt_audio_key, timeout_ms))
            if prompt_audio_key == "ivr/demo-language-kz":
                return "2"
            return None

        def wait_for_digit(self, timeout_ms: int) -> str | None:
            self.waits.append(timeout_ms)
            return None

        def set_variable(self, name: str, value: str | None) -> None:
            self.variables[name] = value

        def verbose(self, message: str, *, level: int = 1) -> None:
            return None

    dtmf_calls: list[str] = []

    monkeypatch.setattr(bridge_ivr_fastagi, "_bridge_app", lambda: DummyBridge())
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_wait_for_call_link_context",
        lambda call_id, stop_event=None: {"interaction_id": "int_ivr", "queue_id": "q_ivr", "queue_code": "voice_lab_ivr"},
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_resolve_transfer_extension",
        lambda queue_code: {"support": "7010", "voice_lab": "7000"}.get(queue_code),
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_start_session",
        lambda call_context, interaction_id, queue_id: {
            "session": {"session_id": "ivs_prompt_sequence"},
            "current_node": {
                "node_id": "root",
                "prompt_text": "Combined bilingual prompt",
                "prompt_sequence": [
                    {"prompt_audio_key": "ivr/demo-language-ru", "prompt_text": "Russian segment", "language": "ru"},
                    {"prompt_audio_key": "ivr/demo-language-kz", "prompt_text": "Kazakh segment", "language": "kz"},
                ],
                "options": [{"digit": "2", "target_node_id": "support"}],
            },
            "completed": False,
        },
    )
    monkeypatch.setattr(
        bridge_ivr_fastagi,
        "_send_dtmf",
        lambda session_id, digit: dtmf_calls.append(digit) or {
            "session": {
                "session_id": session_id,
                "status": "completed",
                "resolved_queue_id": "q_support",
                "resolved_queue_code": "support",
            },
            "current_node": {"node_id": "support", "is_terminal": True},
            "completed": True,
        },
    )
    monkeypatch.setattr(bridge_ivr_fastagi, "_abandon_session", lambda session_id: None)

    channel = DummyChannel()
    request = bridge_ivr_fastagi.AgiRequest(
        env={},
        args=["call_prompt_sequence", "linked_prompt_sequence", "voice_lab_ivr", "voice_lab", "7200"],
    )

    bridge_ivr_fastagi.run_ivr_call(channel, request)

    assert channel.playbacks == [
        ("ivr/demo-language-ru", 1),
        ("ivr/demo-language-kz", 1),
    ]
    assert channel.waits == []
    assert dtmf_calls == ["2"]
    assert channel.variables["MVPCC_IVR_TARGET_QUEUE_CODE"] == "support"
    assert channel.variables["MVPCC_IVR_TARGET_EXTENSION"] == "7010"


def test_bridge_process_call_started_creates_interaction_and_call_link(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}')
    interaction_id = f"int_ast_{tmp_path.name}"
    emitted: dict[str, object] = {}

    monkeypatch.setattr(
        bridge_module,
        "_create_interaction",
        lambda **kwargs: {"interaction_id": interaction_id},
    )
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.update(kwargs) or {"event_id": "vev_ast"},
    )

    call_id = f"call_ast_{tmp_path.name}"
    session = get_session()
    try:
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_demo",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_demo",
                "CallerNumber": "+77010000001",
                "CallerName": "Lab Caller",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )

        bridge_module._process_bridge_row(session, row)
        session.commit()

        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one_or_none()

        assert row.forward_status == "forwarded"
        assert row.interaction_id == interaction_id
        assert link is not None
        assert link.queue_code == "lab"
        assert link.queue_id == "que_lab"
        assert link.interaction_id == interaction_id
        assert emitted["event_type"] == "call.started"
        assert emitted["interaction_id"] == interaction_id
        assert emitted["payload"]["source"] == "asterisk"
    finally:
        session.close()

    client = TestClient(bridge_module.app)
    response = client.get(
        f"/asterisk/live-calls/{call_id}/ai-summary",
        headers=_admin_headers(),
    )

    assert response.status_code == 200
    payload = response.json()


def test_call_started_uses_api_fallback_after_create_failure(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}')
    interaction_id = f"int_ast_api_{tmp_path.name}"
    emitted: dict[str, object] = {}

    def _raise_create_error(**kwargs):
        raise RuntimeError("interaction create failed")

    monkeypatch.setattr(bridge_module, "_create_interaction", _raise_create_error)
    monkeypatch.setattr(
        bridge_module,
        "_find_interaction_by_call_id",
        lambda call_id: {"interaction_id": interaction_id, "subject": f"Inbound call [{call_id}]"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.update(kwargs) or {"event_id": "vev_ast_api"},
    )

    call_id = f"call_ast_api_{tmp_path.name}"
    session = get_session()
    try:
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_demo_api",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_demo_api",
                "CallerNumber": "+77010000021",
                "CallerName": "API Fallback Caller",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )

        bridge_module._process_bridge_row(session, row)
        session.commit()

        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one_or_none()

        assert row.forward_status == "forwarded"
        assert row.interaction_id == interaction_id
        assert link is not None
        assert link.interaction_id == interaction_id
        assert emitted["event_type"] == "call.started"
        assert emitted["interaction_id"] == interaction_id
    finally:
        session.close()


def test_voice_ai_summary_exposes_customer_name_state_for_operator(tmp_path):
    now = utc_now_iso()
    call_id = f"call_voice_summary_{tmp_path.name}"
    session_id = f"avs_voice_summary_{tmp_path.name}"
    ai_session_id = f"ais_voice_summary_{tmp_path.name}"
    interaction_id = f"int_voice_summary_{tmp_path.name}"

    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=f"linked_voice_summary_{tmp_path.name}",
                queue_code="voice_support",
                queue_id="que_voice_support",
                interaction_id=interaction_id,
                caller_number="+77010001122",
                caller_name="Summary Caller",
                status="active",
                telephony_status="connected",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name="PJSIP/1001-000099",
                started_at=now,
                connected_at=now,
                ended_at=None,
                updated_at=now,
                voice_session_id=session_id,
                ai_session_id=ai_session_id,
                ai_state="handoff_required",
                ai_handoff_reason="customer requested operator",
                ai_last_model_at=now,
                voice_start_language="ru",
                customer_name_status="name_followup_required",
                customer_name_value="Айдос",
                customer_name_source="voice_start",
                customer_name_resolved_at=now,
            )
        )
        session.add(
            AISessionRow(
                session_id=ai_session_id,
                channel="voice",
                call_id=call_id,
                thread_id=None,
                interaction_id=interaction_id,
                customer_id="cus_voice_summary",
                agent_profile="voice_support",
                language="ru",
                status="handoff_required",
                summary_text="AI collected context and asked for operator handoff.",
                last_user_message_id=None,
                last_ai_message_id=None,
                handoff_reason="customer requested operator",
                created_at=now,
                updated_at=now,
                closed_at=None,
            )
        )
        session.add(
            VoiceAISessionRow(
                session_id=session_id,
                call_id=call_id,
                linked_id=f"linked_voice_summary_{tmp_path.name}",
                interaction_id=interaction_id,
                customer_id="cus_voice_summary",
                queue_id="que_voice_support",
                ai_session_id=ai_session_id,
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="yandex",
                status="handoff_requested",
                handoff_reason="customer requested operator",
                handoff_target_queue_id="que_voice_support",
                disclosure_played_at=now,
                last_user_utterance_at=now,
                last_ai_reply_at=now,
                started_at=now,
                updated_at=now,
                ended_at=None,
                voice_start_language="ru",
                customer_name_status="name_followup_required",
                customer_name_value="Айдос",
                customer_name_source="voice_start",
                customer_name_resolved_at=now,
            )
        )
        session.add_all(
            [
                VoiceTranscriptSegmentRow(
                    segment_id=f"{session_id}_seg_1",
                    session_id=session_id,
                    call_id=call_id,
                    interaction_id=interaction_id,
                    sequence_no=1,
                    speaker="caller",
                    source_type="voice_asr",
                    text="Соедините с оператором",
                    confidence=0.96,
                    is_final=True,
                    barge_in_interrupted=False,
                    payload_json='{"intent_bearing": true, "provider": "elevenlabs"}',
                    created_at=now,
                ),
                VoiceTranscriptSegmentRow(
                    segment_id=f"{session_id}_seg_2",
                    session_id=session_id,
                    call_id=call_id,
                    interaction_id=interaction_id,
                    sequence_no=2,
                    speaker="assistant",
                    source_type="voice_policy",
                    text="Сейчас переведу вас на оператора.",
                    confidence=None,
                    is_final=True,
                    barge_in_interrupted=False,
                    payload_json='{"delivery_state": "delivered", "provider": "elevenlabs"}',
                    created_at=now,
                ),
            ]
        )
        session.commit()
    finally:
        session.close()

    summary = bridge_module._voice_ai_summary_for_call(call_id)

    assert summary is not None
    assert summary.customer_name_status == "name_followup_required"
    assert summary.customer_name_value == "Айдос"
    assert summary.customer_name_source == "voice_start"
    assert summary.voice_start_language == "ru"
    assert summary.customer_request_text == "Соедините с оператором"
    assert summary.ai_outcome_text == "Сейчас переведу вас на оператора."
    assert [segment.text for segment in summary.transcript_segments] == [
        "Соедините с оператором",
        "Сейчас переведу вас на оператора.",
    ]


def test_voice_ai_summary_uses_intent_bearing_caller_turns_and_delivered_assistant_segments(tmp_path):
    now = utc_now_iso()
    call_id = f"call_voice_truth_{tmp_path.name}"
    session_id = f"avs_voice_truth_{tmp_path.name}"
    ai_session_id = f"ais_voice_truth_{tmp_path.name}"
    interaction_id = f"int_voice_truth_{tmp_path.name}"

    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=f"linked_voice_truth_{tmp_path.name}",
                queue_code="voice_lab_ai",
                queue_id="que_voice_lab",
                interaction_id=interaction_id,
                caller_number="+77010002233",
                caller_name="Truth Caller",
                status="active",
                telephony_status="connected",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name="PJSIP/1001-000111",
                started_at=now,
                connected_at=now,
                ended_at=None,
                updated_at=now,
                voice_session_id=session_id,
                ai_session_id=ai_session_id,
                ai_state="active",
                ai_handoff_reason=None,
                ai_last_model_at=now,
                voice_start_language="ru",
            )
        )
        session.add(
            AISessionRow(
                session_id=ai_session_id,
                channel="voice",
                call_id=call_id,
                thread_id=None,
                interaction_id=interaction_id,
                customer_id="cus_voice_truth",
                agent_profile="voice_support",
                language="ru",
                status="active",
                summary_text="",
                last_user_message_id=None,
                last_ai_message_id=None,
                handoff_reason=None,
                created_at=now,
                updated_at=now,
                closed_at=None,
            )
        )
        session.add(
            VoiceAISessionRow(
                session_id=session_id,
                call_id=call_id,
                linked_id=f"linked_voice_truth_{tmp_path.name}",
                interaction_id=interaction_id,
                customer_id="cus_voice_truth",
                queue_id="que_voice_lab",
                ai_session_id=ai_session_id,
                agent_profile="voice_support",
                language="ru",
                asr_provider="elevenlabs",
                tts_provider="elevenlabs",
                status="active",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_lab",
                disclosure_played_at=now,
                last_user_utterance_at=now,
                last_ai_reply_at=now,
                started_at=now,
                updated_at=now,
                ended_at=None,
                voice_start_language="ru",
            )
        )
        session.add_all(
            [
                VoiceTranscriptSegmentRow(
                    segment_id=f"{session_id}_seg_1",
                    session_id=session_id,
                    call_id=call_id,
                    interaction_id=interaction_id,
                    sequence_no=1,
                    speaker="caller",
                    source_type="voice_asr",
                    text="Мне надо узнать график работы.",
                    confidence=0.96,
                    is_final=True,
                    barge_in_interrupted=False,
                    payload_json='{"intent_bearing": true, "provider": "elevenlabs"}',
                    created_at=now,
                ),
                VoiceTranscriptSegmentRow(
                    segment_id=f"{session_id}_seg_2",
                    session_id=session_id,
                    call_id=call_id,
                    interaction_id=interaction_id,
                    sequence_no=2,
                    speaker="assistant",
                    source_type="voice_policy",
                    text="Подскажите, пожалуйста, город или филиал.",
                    confidence=None,
                    is_final=True,
                    barge_in_interrupted=False,
                    payload_json='{"delivery_state": "delivered", "provider": "elevenlabs"}',
                    created_at=now,
                ),
                VoiceTranscriptSegmentRow(
                    segment_id=f"{session_id}_seg_3",
                    session_id=session_id,
                    call_id=call_id,
                    interaction_id=interaction_id,
                    sequence_no=3,
                    speaker="caller",
                    source_type="voice_asr",
                    text="Алло, приветствую.",
                    confidence=0.8,
                    is_final=True,
                    barge_in_interrupted=False,
                    payload_json='{"intent_bearing": false, "provider": "elevenlabs"}',
                    created_at=now,
                ),
                VoiceTranscriptSegmentRow(
                    segment_id=f"{session_id}_seg_4",
                    session_id=session_id,
                    call_id=call_id,
                    interaction_id=interaction_id,
                    sequence_no=4,
                    speaker="assistant",
                    source_type="voice_policy",
                    text="Это прерванный ответ.",
                    confidence=None,
                    is_final=True,
                    barge_in_interrupted=True,
                    payload_json='{"delivery_state": "interrupted", "provider": "elevenlabs"}',
                    created_at=now,
                ),
            ]
        )
        session.commit()
    finally:
        session.close()

    summary = bridge_module._voice_ai_summary_for_call(call_id)

    assert summary is not None
    assert summary.customer_request_text == "Мне надо узнать график работы."
    assert summary.ai_outcome_text == "Подскажите, пожалуйста, город или филиал."
    assert [(segment.speaker, segment.text) for segment in summary.transcript_segments] == [
        ("caller", "Мне надо узнать график работы."),
        ("assistant", "Подскажите, пожалуйста, город или филиал."),
        ("caller", "Алло, приветствую."),
    ]
def test_call_started_re_raises_original_create_error_when_all_fallbacks_fail(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}')

    def _raise_create_error(**kwargs):
        raise RuntimeError("interaction create failed")

    monkeypatch.setattr(bridge_module, "_create_interaction", _raise_create_error)
    monkeypatch.setattr(bridge_module, "_find_interaction_by_call_id", lambda call_id: None)

    call_id = f"call_ast_fail_{tmp_path.name}"
    session = get_session()
    try:
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_demo_fail",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_demo_fail",
                "CallerNumber": "+77010000022",
                "CallerName": "Failed Fallback Caller",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )

        with pytest.raises(RuntimeError, match="interaction create failed"):
            bridge_module._process_bridge_row(session, row)
    finally:
        session.close()


def test_reconcile_keeps_live_call_active_when_ami_still_sees_channels(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1")

    call_id = f"call_reconcile_live_{tmp_path.name}"
    stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat()

    session = get_session()
    try:
        link = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_reconcile_live",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_reconcile_live_{tmp_path.name}",
            caller_number="+77010000004",
            caller_name="Track11 Reconcile Live",
            status="active",
            telephony_status="connected",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-stale",
            started_at=stale_started,
            connected_at=stale_started,
            ended_at=None,
            updated_at=stale_started,
        )
        session.add(link)
        session.commit()
    finally:
        session.close()

    emitted: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_live"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_list_channels_via_coreshowchannels",
        lambda **kwargs: ["PJSIP/2001-00009999"],
    )

    bridge_module._reconcile_stale_calls_once()

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.status == "active"
        assert link.telephony_status == "connected"
        assert link.ended_at is None
        assert link.channel_name == "PJSIP/2001-00009999"
        assert [item for item in emitted if item["call_id"] == call_id] == []
    finally:
        session.close()


def test_reconcile_skips_synthetic_end_when_channel_lookup_is_unavailable(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1")

    call_id = f"call_reconcile_unavailable_{tmp_path.name}"
    stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat()

    session = get_session()
    try:
        link = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_reconcile_unavailable",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_reconcile_unavailable_{tmp_path.name}",
            caller_number="+77010000024",
            caller_name="Track11 Reconcile Unavailable",
            status="active",
            telephony_status="connected",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-00009998",
            started_at=stale_started,
            connected_at=stale_started,
            ended_at=None,
            updated_at=stale_started,
        )
        session.add(link)
        session.commit()
    finally:
        session.close()

    emitted: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_unavailable"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_list_channels_via_coreshowchannels",
        lambda **kwargs: None,
    )

    bridge_module._reconcile_stale_calls_once()

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.status == "active"
        assert link.telephony_status == "connected"
        assert link.ended_at is None
        assert [item for item in emitted if item["call_id"] == call_id] == []
    finally:
        session.close()


def test_reconcile_emits_synthetic_end_and_closes_call(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1")

    call_id = f"call_reconcile_soft_{tmp_path.name}"
    stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat()

    session = get_session()
    try:
        link = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_reconcile_soft",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_reconcile_soft_{tmp_path.name}",
            caller_number="+77010000005",
            caller_name="Track11 Reconcile Soft",
            status="active",
            telephony_status="connected",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-00000010",
            started_at=stale_started,
            connected_at=stale_started,
            ended_at=None,
            updated_at=stale_started,
        )
        session.add(link)
        session.commit()
    finally:
        session.close()

    emitted: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_soft"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_list_channels_via_coreshowchannels",
        lambda **kwargs: [],
    )

    bridge_module._reconcile_stale_calls_once()

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.status == "ended"
        assert link.telephony_status == "ended"
        assert link.ended_at is not None
        matching = [item for item in emitted if item["call_id"] == call_id]
        assert len(matching) == 1
        assert matching[0]["event_type"] == "call.ended"
        assert matching[0]["payload"]["reconciled"] is True
    finally:
        session.close()


def test_reconcile_keeps_recent_active_ai_call_open(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1")

    call_id = f"call_reconcile_ai_live_{tmp_path.name}"
    session_id = f"avs_reconcile_ai_live_{tmp_path.name}"
    stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat()
    recent_media = (datetime.now(timezone.utc) - timedelta(seconds=2)).isoformat()

    session = get_session()
    try:
        link = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_reconcile_ai_live",
            queue_code="voice_lab_ai",
            queue_id="que_voice_lab",
            interaction_id=f"int_reconcile_ai_live_{tmp_path.name}",
            caller_number="+77010000055",
            caller_name="Track16 Reconcile AI",
            status="active",
            telephony_status="connected",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension=None,
            channel_name="PJSIP/1001-0000AI55",
            voice_session_id=session_id,
            ai_session_id=f"ais_reconcile_ai_live_{tmp_path.name}",
            ai_state="speaking",
            ai_handoff_reason=None,
            ai_last_model_at=recent_media,
            started_at=stale_started,
            connected_at=stale_started,
            ended_at=None,
            updated_at=stale_started,
        )
        session.add(link)
        session.add(
            VoiceAISessionRow(
                session_id=session_id,
                call_id=call_id,
                linked_id="linked_reconcile_ai_live",
                interaction_id=link.interaction_id,
                customer_id=None,
                queue_id="que_voice_lab",
                ai_session_id=f"ais_reconcile_ai_live_{tmp_path.name}",
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="yandex",
                status="active",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_lab",
                media_uuid="11111111-1111-1111-1111-111111111111",
                media_status="connected",
                media_connected_at=stale_started,
                media_ended_at=None,
                last_media_frame_at=recent_media,
                disclosure_played_at=stale_started,
                last_user_utterance_at=recent_media,
                last_ai_reply_at=recent_media,
                started_at=stale_started,
                updated_at=recent_media,
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    emitted: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_ai_live"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_list_channels_via_coreshowchannels",
        lambda **kwargs: [],
    )

    bridge_module._reconcile_stale_calls_once()

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.status == "active"
        assert link.telephony_status == "connected"
        assert link.ended_at is None
        assert [item for item in emitted if item["call_id"] == call_id] == []
    finally:
        session.close()


def test_reconcile_closes_active_call_without_duplicate_end_event(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1")

    call_id = f"call_reconcile_existing_end_{tmp_path.name}"
    stale_started = (datetime.now(timezone.utc) - timedelta(seconds=60)).isoformat()

    session = get_session()
    try:
        link = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_reconcile_existing_end",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_reconcile_existing_end_{tmp_path.name}",
            caller_number="+77010000023",
            caller_name="Track11 Reconcile Existing End",
            status="active",
            telephony_status="connected",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-00000011",
            started_at=stale_started,
            connected_at=stale_started,
            ended_at=None,
            updated_at=stale_started,
        )
        session.add(link)
        session.add(
            VoiceEventRow(
                event_id=f"vev_reconcile_existing_end_{tmp_path.name}",
                event_type="call.ended",
                call_id=call_id,
                interaction_id=link.interaction_id,
                payload_json=json.dumps({"hangup_cause": "normal_clearing"}),
                created_at=stale_started,
            )
        )
        session.commit()
    finally:
        session.close()

    emitted: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_reconcile_existing_end"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_list_channels_via_coreshowchannels",
        lambda **kwargs: [],
    )

    bridge_module._reconcile_stale_calls_once()

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.status == "ended"
        assert link.telephony_status == "ended"
        assert link.ended_at is not None
        assert [item for item in emitted if item["call_id"] == call_id] == []
    finally:
        session.close()


def test_reconcile_prioritizes_oldest_unresolved_ended_call(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_BRIDGE_RECONCILE_SCAN_LIMIT", "1")

    older_iso = (datetime.now(timezone.utc) - timedelta(seconds=120)).isoformat()
    newer_iso = (datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat()
    older_call_id = f"call_reconcile_oldest_{tmp_path.name}"
    newer_call_id = f"call_reconcile_newer_with_recording_{tmp_path.name}"
    recording_path = tmp_path / "fixtures" / "reconcile-oldest.wav"
    recording_path.parent.mkdir(parents=True, exist_ok=True)
    recording_path.write_bytes(b"RIFF" + b"\x00" * 32)

    session = get_session()
    try:
        session.query(CallRecordingRow).delete()
        session.query(AsteriskCallLinkRow).delete()
        session.commit()

        older_link = AsteriskCallLinkRow(
            call_id=older_call_id,
            linked_id="linked_reconcile_oldest",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_reconcile_oldest_{tmp_path.name}",
            caller_number="+77010000025",
            caller_name="Track11 Reconcile Oldest",
            status="ended",
            telephony_status="ended",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-oldest",
            started_at=older_iso,
            connected_at=older_iso,
            ended_at=older_iso,
            updated_at=older_iso,
        )
        newer_link = AsteriskCallLinkRow(
            call_id=newer_call_id,
            linked_id="linked_reconcile_newer",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_reconcile_newer_{tmp_path.name}",
            caller_number="+77010000026",
            caller_name="Track11 Reconcile Newer",
            status="ended",
            telephony_status="ended",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-newer",
            started_at=newer_iso,
            connected_at=newer_iso,
            ended_at=newer_iso,
            updated_at=newer_iso,
        )
        session.add_all([older_link, newer_link])
        session.flush()
        session.add(
            CallRecordingRow(
                recording_id=f"rec_reconcile_newer_{tmp_path.name}",
                channel="voice",
                call_id=newer_call_id,
                interaction_id=newer_link.interaction_id,
                source_event_id=f"src_reconcile_newer_{tmp_path.name}",
                file_name="newer.wav",
                storage_backend="local_fs",
                storage_path=str(recording_path),
                mime_type="audio/wav",
                size_bytes=recording_path.stat().st_size,
                duration_seconds=4,
                checksum_sha256="abc123",
                status="ready",
                recorded_at=newer_iso,
                created_at=newer_iso,
                updated_at=newer_iso,
                archived_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    uploaded: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_find_remote_recording_for_call",
        lambda call_id: (
            ("/remote/oldest.wav", "oldest.wav", "audio/wav", datetime.now(timezone.utc) - timedelta(seconds=60))
            if call_id == older_call_id
            else None
        ),
    )
    monkeypatch.setattr(
        bridge_module,
        "_fetch_recording_file",
        lambda remote_path, file_name: (recording_path, False),
    )
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: {"event_id": f"vev_{kwargs['call_id']}"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_upload_recording",
        lambda **kwargs: uploaded.append(kwargs) or {"recording_id": f"rec_uploaded_{kwargs['call_id']}"},
    )

    bridge_module._reconcile_stale_calls_once()

    assert [entry["call_id"] for entry in uploaded] == [older_call_id]


def test_recent_calls_route_returns_recent_transferred_call(tmp_path):
    client = TestClient(bridge_module.app)
    now = utc_now_iso()
    call_id = f"call_recent_transfer_{tmp_path.name}"

    session = get_session()
    try:
        link = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_recent_transfer",
            queue_code="voice_lab",
            queue_id="que_recent_transfer",
            interaction_id=f"int_recent_transfer_{tmp_path.name}",
            caller_number="+77010000006",
            caller_name="Recent Transfer",
            status="ended",
            telephony_status="ended",
            claimed_by_user="operator",
            claimed_at=now,
            operator_extension="2001",
            channel_name="PJSIP/2001-00010000",
            started_at=now,
            connected_at=now,
            ended_at=now,
            updated_at=now,
        )
        session.add(link)
        session.add(
            AsteriskCallActionLogRow(
                action_id=f"aca_recent_transfer_{tmp_path.name}",
                call_id=call_id,
                interaction_id=link.interaction_id,
                action_type="blind-transfer",
                actor_user="operator",
                actor_role="operator",
                request_json=json.dumps({"target_value": "2002", "resolved_extension": "2002"}),
                result_status="ok",
                ami_action_id="ami_recent_transfer",
                error=None,
                created_at=now,
            )
        )
        session.add(
            VoiceEventRow(
                event_id=f"vev_recent_transfer_{tmp_path.name}",
                event_type="call.ended",
                call_id=call_id,
                interaction_id=link.interaction_id,
                payload_json=json.dumps({"hangup_cause": "normal_clearing"}),
                created_at=now,
            )
        )
        session.commit()
    finally:
        session.close()

    response = client.get("/asterisk/recent-calls", headers=_admin_headers())

    assert response.status_code == 200
    payload = response.json()
    item = next(entry for entry in payload if entry["call_id"] == call_id)
    assert item["terminal_action"] == "blind-transfer"
    assert item["terminal_target"] == "2002"
    assert item["hangup_cause"] == "normal_clearing"
    assert item["last_transition_at"] == now


def test_recent_calls_route_skips_stale_calls(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_RECENT_CALLS_WINDOW_SECONDS", "120")
    client = TestClient(bridge_module.app)
    old_iso = (datetime.now(timezone.utc) - timedelta(seconds=600)).isoformat()
    call_id = f"call_recent_stale_{tmp_path.name}"

    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id="linked_recent_stale",
                queue_code="voice_lab",
                queue_id="que_recent_stale",
                interaction_id=f"int_recent_stale_{tmp_path.name}",
                caller_number="+77010000007",
                caller_name="Recent Stale",
                status="ended",
                telephony_status="ended",
                claimed_by_user="operator",
                claimed_at=old_iso,
                operator_extension="2001",
                channel_name="PJSIP/2001-00010001",
                started_at=old_iso,
                connected_at=old_iso,
                ended_at=old_iso,
                updated_at=old_iso,
            )
        )
        session.commit()
    finally:
        session.close()

    response = client.get("/asterisk/recent-calls", headers=_admin_headers())

    assert response.status_code == 200
    payload = response.json()
    assert all(entry["call_id"] != call_id for entry in payload)


def test_retry_endpoint_replays_failed_bridge_event(monkeypatch, tmp_path):
    call_id = f"call_retry_{tmp_path.name}"
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", "{}")

    session = get_session()
    try:
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_retry",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_retry",
                "CallerNumber": "+77010000002",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )
        try:
            bridge_module._process_bridge_row(session, row)
        except Exception as exc:  # noqa: BLE001
            bridge_module._mark_log_failed(session, row, str(exc))
        session.commit()
        bridge_event_id = row.bridge_event_id
        assert row.forward_status == "failed"
    finally:
        session.close()

    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_retry"}')
    monkeypatch.setattr(
        bridge_module,
        "_create_interaction",
        lambda **kwargs: {"interaction_id": f"int_retry_{tmp_path.name}"},
    )
    emitted: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_retry"},
    )

    client = TestClient(bridge_module.app)
    retried = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers())
    repeated = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers())

    assert retried.status_code == 200
    assert repeated.status_code == 200
    payload = retried.json()
    assert payload["forward_status"] == "forwarded"
    assert payload["interaction_id"].startswith("int_retry_")
    assert repeated.json()["forward_status"] == "forwarded"
    assert len(emitted) == 1


def test_retry_endpoint_does_not_double_process_row_already_claimed_by_background_worker(monkeypatch, tmp_path):
    call_id = f"call_retry_claimed_{tmp_path.name}"
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_retry_claimed"}')

    session = get_session()
    try:
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_retry_claimed",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_retry_claimed",
                "CallerNumber": "+77010000027",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )
        bridge_module._mark_log_failed(session, row, "retry me")
        session.commit()
        bridge_event_id = row.bridge_event_id
    finally:
        session.close()

    emitted: list[dict[str, object]] = []
    monkeypatch.setattr(
        bridge_module,
        "_create_interaction",
        lambda **kwargs: {"interaction_id": f"int_retry_claimed_{tmp_path.name}"},
    )
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: emitted.append(kwargs) or {"event_id": "vev_retry_claimed"},
    )

    assert bridge_module._claim_bridge_event_for_processing(bridge_event_id) is True

    client = TestClient(bridge_module.app)
    blocked = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers())

    assert blocked.status_code == 200
    assert blocked.json()["forward_status"] == "processing"
    assert emitted == []

    processed = bridge_module._process_claimed_bridge_event(bridge_event_id)
    assert processed is not None
    assert processed.forward_status == "forwarded"
    assert len(emitted) == 1


def test_retry_recovers_after_voice_event_was_sent_before_db_failure(monkeypatch, tmp_path):
    call_id = f"call_retry_voice_commit_{tmp_path.name}"
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_retry_voice_commit"}')

    emitted_by_source: dict[str, dict[str, str]] = {}
    emitted_calls: list[str] = []

    def _emit_once_per_source(**kwargs):
        source_event_id = str(kwargs["source_event_id"])
        emitted_calls.append(source_event_id)
        return emitted_by_source.setdefault(
            source_event_id,
            {"event_id": f"vev_retry_voice_{len(emitted_by_source) + 1}"},
        )

    original_mark_log_forwarded = bridge_module._mark_log_forwarded
    should_fail_once = {"value": True}

    def _mark_log_forwarded_once(session, row, **kwargs):
        original_mark_log_forwarded(session, row, **kwargs)
        if should_fail_once["value"]:
            should_fail_once["value"] = False
            raise RuntimeError("db write failed after voice event")

    monkeypatch.setattr(
        bridge_module,
        "_create_interaction",
        lambda **kwargs: {"interaction_id": f"int_retry_voice_commit_{tmp_path.name}"},
    )
    monkeypatch.setattr(bridge_module, "_emit_voice_event", _emit_once_per_source)
    monkeypatch.setattr(bridge_module, "_mark_log_forwarded", _mark_log_forwarded_once)

    bridge_module._record_ami_payload(
        {
            "UserEvent": "MVPCCCallStarted",
            "CallID": call_id,
            "LinkedID": "linked_retry_voice_commit",
            "CallerNumber": "+77010000028",
            "CallerName": "Retry Voice Commit",
            "QueueCode": "lab",
            "Extension": "7000",
            "Context": "mvpcc-inbound",
            "Direction": "inbound",
        }
    )

    session = get_session()
    try:
        row = session.execute(
            select(AsteriskEventLogRow).where(AsteriskEventLogRow.call_id == call_id)
        ).scalar_one()
        bridge_event_id = row.bridge_event_id
        assert row.forward_status == "failed"
    finally:
        session.close()

    client = TestClient(bridge_module.app)
    retried = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers())

    assert retried.status_code == 200
    assert retried.json()["forward_status"] == "forwarded"
    assert len(emitted_calls) == 2
    assert len(emitted_by_source) == 1


def test_retry_recovers_after_recording_upload_before_db_failure(monkeypatch, tmp_path):
    call_id = f"call_retry_recording_commit_{tmp_path.name}"
    ready_iso = (datetime.now(timezone.utc) - timedelta(seconds=10)).isoformat()
    recording_fixture = tmp_path / "fixtures" / "retry-recording.wav"
    recording_fixture.parent.mkdir(parents=True, exist_ok=True)
    recording_fixture.write_bytes(b"RIFF" + b"\x00" * 64)

    session = get_session()
    try:
        link = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_retry_recording_commit",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_retry_recording_commit_{tmp_path.name}",
            caller_number="+77010000029",
            caller_name="Retry Recording Commit",
            status="ended",
            telephony_status="ended",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-recording",
            started_at=ready_iso,
            connected_at=ready_iso,
            ended_at=ready_iso,
            updated_at=ready_iso,
        )
        session.add(link)
        session.add(
            VoiceEventRow(
                event_id=f"vev_retry_recording_ended_{tmp_path.name}",
                event_type="call.ended",
                call_id=call_id,
                interaction_id=link.interaction_id,
                source_event_id=f"src_retry_recording_ended_{tmp_path.name}",
                payload_json=json.dumps({"hangup_cause": "normal_clearing"}),
                created_at=ready_iso,
            )
        )
        session.add(
            VoiceEventRow(
                event_id=f"vev_retry_recording_ready_{tmp_path.name}",
                event_type="recording.ready",
                call_id=call_id,
                interaction_id=link.interaction_id,
                source_event_id=f"src_retry_recording_ready_{tmp_path.name}",
                payload_json=json.dumps({"remote_path": "/remote/retry-recording.wav"}),
                created_at=ready_iso,
            )
        )
        session.commit()
    finally:
        session.close()

    uploaded_by_source: dict[str, dict[str, str]] = {}
    upload_calls: list[str] = []

    def _upload_once_per_source(**kwargs):
        source_event_id = str(kwargs["source_event_id"])
        upload_calls.append(source_event_id)
        return uploaded_by_source.setdefault(
            source_event_id,
            {"recording_id": f"rec_retry_recording_{len(uploaded_by_source) + 1}"},
        )

    original_mark_log_forwarded = bridge_module._mark_log_forwarded
    should_fail_once = {"value": True}

    def _mark_log_forwarded_once(session, row, **kwargs):
        original_mark_log_forwarded(session, row, **kwargs)
        if should_fail_once["value"]:
            should_fail_once["value"] = False
            raise RuntimeError("db write failed after recording upload")

    monkeypatch.setattr(
        bridge_module,
        "_fetch_recording_file",
        lambda remote_path, file_name: (recording_fixture, False),
    )
    monkeypatch.setattr(bridge_module, "_upload_recording", _upload_once_per_source)
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: {"event_id": f"vev_{kwargs['event_type']}"},
    )
    monkeypatch.setattr(bridge_module, "_mark_log_forwarded", _mark_log_forwarded_once)

    bridge_module._record_ami_payload(
        {
            "UserEvent": "MVPCCRecordingReady",
            "CallID": call_id,
            "LinkedID": "linked_retry_recording_commit",
            "RemotePath": "/remote/retry-recording.wav",
            "FileName": "retry-recording.wav",
            "MimeType": "audio/wav",
            "DurationSeconds": "7",
        }
    )

    session = get_session()
    try:
        row = session.execute(
            select(AsteriskEventLogRow)
            .where(AsteriskEventLogRow.call_id == call_id)
            .where(AsteriskEventLogRow.ami_event_name == "MVPCCRecordingReady")
        ).scalar_one()
        bridge_event_id = row.bridge_event_id
        assert row.forward_status == "failed"
    finally:
        session.close()

    client = TestClient(bridge_module.app)
    retried = client.post(f"/asterisk/events/{bridge_event_id}/retry", headers=_admin_headers())

    assert retried.status_code == 200
    assert retried.json()["forward_status"] == "forwarded"
    assert len(upload_calls) == 2
    assert len(uploaded_by_source) == 1


def test_startup_fails_fast_when_singleton_guard_is_already_held(monkeypatch):
    monkeypatch.setenv("ASTERISK_BRIDGE_ENABLED", "1")
    monkeypatch.setattr(bridge_module, "_ami_loop", _wait_until_stopped)
    monkeypatch.setattr(bridge_module, "_failed_retry_loop", _wait_until_stopped)
    monkeypatch.setattr(bridge_module, "_try_acquire_bridge_singleton_guard", lambda: False)

    with pytest.raises(RuntimeError, match="replicas=1, workers=1"):
        bridge_module._startup()


def test_disabled_bridge_startup_does_not_require_singleton_guard(monkeypatch):
    monkeypatch.delenv("ASTERISK_BRIDGE_ENABLED", raising=False)
    monkeypatch.setattr(bridge_module, "_ami_loop", _wait_until_stopped)
    monkeypatch.setattr(bridge_module, "_failed_retry_loop", _wait_until_stopped)
    monkeypatch.setattr(
        bridge_module,
        "_try_acquire_bridge_singleton_guard",
        lambda: (_ for _ in ()).throw(AssertionError("singleton guard should not be acquired")),
    )

    bridge_module._startup()
    assert len(bridge_module._background_threads_alive()) == 2


def test_shutdown_releases_singleton_guard(monkeypatch):
    monkeypatch.setenv("ASTERISK_BRIDGE_ENABLED", "1")
    monkeypatch.setattr(bridge_module, "_ami_loop", _wait_until_stopped)
    monkeypatch.setattr(bridge_module, "_failed_retry_loop", _wait_until_stopped)

    bridge_module._startup()
    try:
        assert bridge_module._BRIDGE_SINGLETON_FILE_HANDLE is not None
    finally:
        bridge_module._shutdown()

    assert bridge_module._BRIDGE_SINGLETON_FILE_HANDLE is None
    assert bridge_module._try_acquire_bridge_singleton_guard() is True
    bridge_module._release_bridge_singleton_guard()


def test_claim_endpoint_updates_call_link_and_action_log(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}')
    monkeypatch.setattr(bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_claim_{tmp_path.name}"})
    monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_claim"})
    monkeypatch.setattr(
        bridge_module,
        "_resolve_channel_name",
        lambda session, link, prefer_operator=False, refresh=False, snapshot=None: "PJSIP/1001-0000001",
    )
    monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_claim"})
    monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"})

    call_id = f"call_claim_{tmp_path.name}"
    session = get_session()
    try:
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_claim",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_claim",
                "CallerNumber": "+77010000003",
                "CallerName": "Track11 Claim",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
                "Channel": "PJSIP/1001-0000001",
            },
        )
        bridge_module._process_bridge_row(session, row)
        session.commit()
    finally:
        session.close()

    client = TestClient(bridge_module.app)
    claimed = client.post(
        f"/asterisk/live-calls/{call_id}/claim",
        headers=_operator_headers("operator_a"),
        json={},
    )
    assert claimed.status_code == 200
    payload = claimed.json()
    assert payload["claimed_by_user"] == "operator_a"
    assert payload["operator_extension"] == "2001"
    assert payload["telephony_status"] == "claimed"

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.claimed_by_user == "operator_a"
        assert link.operator_extension == "2001"
        action = session.execute(
            select(AsteriskCallActionLogRow)
            .where(AsteriskCallActionLogRow.call_id == call_id)
            .where(AsteriskCallActionLogRow.action_type == "claim")
            .order_by(AsteriskCallActionLogRow.id.desc())
        ).scalar_one()
        assert action.result_status == "ok"
        assert action.ami_action_id == "ami_claim"
    finally:
        session.close()


def test_claim_resolves_live_channels_once_on_happy_path(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_claim_fast_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_claim_fast",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_claim_fast_{tmp_path.name}",
            caller_number="+77010000013",
            caller_name="Track12 Claim Fast",
            status="active",
            telephony_status="ringing",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension=None,
            channel_name="PJSIP/1001-0000013",
            started_at=utc_now_iso(),
            connected_at=None,
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    channel_calls = []

    def _fake_channels(**kwargs):
        channel_calls.append(kwargs)
        return ["PJSIP/1001-0000013"]

    monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", _fake_channels)
    monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"})
    monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_claim_fast"})

    client = TestClient(bridge_module.app)
    claimed = client.post(
        f"/asterisk/live-calls/{call_id}/claim",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert claimed.status_code == 200
    assert len(channel_calls) == 1


def test_claim_skips_redirect_when_operator_leg_already_exists(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_claim_existing_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_claim_existing",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_claim_existing_{tmp_path.name}",
            caller_number="+77010000007",
            caller_name="Track11 Existing Leg",
            status="active",
            telephony_status="ringing",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension=None,
            channel_name="PJSIP/1001-0000004",
            started_at=utc_now_iso(),
            connected_at=None,
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    monkeypatch.setattr(
        bridge_module,
        "_list_channels_via_coreshowchannels",
        lambda **kwargs: ["PJSIP/1001-0000004", "PJSIP/2001-0000005"],
    )
    monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"})

    def _unexpected_ami(*args, **kwargs):
        raise AssertionError("Redirect should be skipped when operator leg already exists")

    monkeypatch.setattr(bridge_module, "_ami_action", _unexpected_ami)

    client = TestClient(bridge_module.app)
    claimed = client.post(
        f"/asterisk/live-calls/{call_id}/claim",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert claimed.status_code == 200
    payload = claimed.json()
    assert payload["claimed_by_user"] == "operator_a"
    assert payload["operator_extension"] == "2001"
    assert payload["telephony_status"] == "claimed"


def test_claim_connected_call_becomes_idempotent_success(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_claim_connected_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_claim_connected",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_claim_connected_{tmp_path.name}",
            caller_number="+77010000010",
            caller_name="Track11 Connected Claim",
            status="active",
            telephony_status="connected",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-0000009",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: [])
    monkeypatch.setattr(
        bridge_module,
        "_latest_event_channel",
        lambda session, *, call_id, event_name: "PJSIP/2001-0000009"
        if event_name.endswith("OperatorConnected")
        else "PJSIP/1001-0000008",
    )
    monkeypatch.setattr(bridge_module, "_assign_interaction", lambda **kwargs: {"status": "ok"})

    def _unexpected_ami(*args, **kwargs):
        raise AssertionError("Redirect should be skipped when call is already connected")

    monkeypatch.setattr(bridge_module, "_ami_action", _unexpected_ami)

    client = TestClient(bridge_module.app)
    claimed = client.post(
        f"/asterisk/live-calls/{call_id}/claim",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert claimed.status_code == 200
    payload = claimed.json()
    assert payload["claimed_by_user"] == "operator_a"
    assert payload["operator_extension"] == "2001"
    assert payload["telephony_status"] == "connected"


def test_operator_cannot_hangup_foreign_claim(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")

    call_id = f"call_foreign_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_foreign",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_foreign_{tmp_path.name}",
            caller_number="+77010000004",
            caller_name="Track11 Foreign",
            status="active",
            telephony_status="claimed",
            claimed_by_user="operator_b",
            claimed_at=utc_now_iso(),
            operator_extension="2002",
            channel_name="PJSIP/1001-0000002",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    client = TestClient(bridge_module.app)
    denied = client.post(
        f"/asterisk/live-calls/{call_id}/hangup",
        headers=_operator_headers("operator_a"),
        json={},
    )
    assert denied.status_code == 403


def test_operator_can_hangup_own_extension_without_claimed_user(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_own_ext_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_own_ext",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_own_ext_{tmp_path.name}",
            caller_number="+77010000008",
            caller_name="Track11 Own Ext",
            status="active",
            telephony_status="connected",
            claimed_by_user=None,
            claimed_at=None,
            operator_extension="2001",
            channel_name="PJSIP/2001-0000006",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    monkeypatch.setattr(
        bridge_module,
        "_resolve_channel_name",
        lambda session, link, prefer_operator=False, refresh=False, snapshot=None: "PJSIP/2001-0000006",
    )
    monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_hangup"})

    client = TestClient(bridge_module.app)
    response = client.post(
        f"/asterisk/live-calls/{call_id}/hangup",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert response.status_code == 200
    assert response.json()["telephony_status"] == "ended"


def test_hangup_falls_back_to_another_live_leg(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_hangup_fallback_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_hangup_fallback",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_hangup_fallback_{tmp_path.name}",
            caller_number="+77010000011",
            caller_name="Track11 Hangup Fallback",
            status="active",
            telephony_status="connected",
            claimed_by_user="operator_a",
            claimed_at=utc_now_iso(),
            operator_extension="2001",
            channel_name="PJSIP/2001-0000010",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    monkeypatch.setattr(
        bridge_module,
        "_candidate_channels_for_call",
        lambda session, link, operator_extension=None, prefer_operator=False, snapshot=None: [
            "PJSIP/2001-0000010",
            "PJSIP/1001-0000011",
        ],
    )

    calls = []

    def _ami_action(action, params):
        calls.append(params["Channel"])
        if params["Channel"] == "PJSIP/2001-0000010":
            raise RuntimeError("No such channel")
        return {"action_id": "ami_hangup_fallback"}

    monkeypatch.setattr(bridge_module, "_ami_action", _ami_action)

    client = TestClient(bridge_module.app)
    response = client.post(
        f"/asterisk/live-calls/{call_id}/hangup",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert response.status_code == 200
    assert response.json()["telephony_status"] == "ended"
    assert calls == ["PJSIP/2001-0000010", "PJSIP/1001-0000011"]


def test_hangup_becomes_idempotent_success_when_channel_already_closed(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_hangup_closed_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_hangup_closed",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_hangup_closed_{tmp_path.name}",
            caller_number="+77010000012",
            caller_name="Track11 Hangup Closed",
            status="active",
            telephony_status="connected",
            claimed_by_user="operator_a",
            claimed_at=utc_now_iso(),
            operator_extension="2001",
            channel_name="PJSIP/2001-0000012",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    monkeypatch.setattr(
        bridge_module,
        "_candidate_channels_for_call",
        lambda session, link, operator_extension=None, prefer_operator=False, snapshot=None: ["PJSIP/2001-0000012"],
    )
    monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", lambda **kwargs: [])

    def _ami_action(action, params):
        raise RuntimeError("No such channel")

    monkeypatch.setattr(bridge_module, "_ami_action", _ami_action)

    client = TestClient(bridge_module.app)
    response = client.post(
        f"/asterisk/live-calls/{call_id}/hangup",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert response.status_code == 200
    assert response.json()["telephony_status"] == "ended"

    session = get_session()
    try:
        action = session.execute(
            select(AsteriskCallActionLogRow)
            .where(AsteriskCallActionLogRow.call_id == call_id)
            .where(AsteriskCallActionLogRow.action_type == "hangup")
            .order_by(AsteriskCallActionLogRow.id.desc())
        ).scalar_one()
        assert action.result_status == "ok"
        assert action.error == "channel_already_closed"
    finally:
        session.close()


def test_hangup_resolves_live_channels_once_on_happy_path(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_hangup_fast_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_hangup_fast",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_hangup_fast_{tmp_path.name}",
            caller_number="+77010000014",
            caller_name="Track12 Hangup Fast",
            status="active",
            telephony_status="connected",
            claimed_by_user="operator_a",
            claimed_at=utc_now_iso(),
            operator_extension="2001",
            channel_name="PJSIP/2001-0000014",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    channel_calls = []

    def _fake_channels(**kwargs):
        channel_calls.append(kwargs)
        return ["PJSIP/2001-0000014"]

    monkeypatch.setattr(bridge_module, "_list_channels_via_coreshowchannels", _fake_channels)
    monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_hangup_fast"})

    client = TestClient(bridge_module.app)
    response = client.post(
        f"/asterisk/live-calls/{call_id}/hangup",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert response.status_code == 200
    assert len(channel_calls) == 1


def test_hangup_hangs_up_all_live_legs_on_success(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')

    call_id = f"call_hangup_all_legs_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_hangup_all_legs",
            queue_code="lab",
            queue_id="que_lab",
            interaction_id=f"int_hangup_all_legs_{tmp_path.name}",
            caller_number="+77010000015",
            caller_name="Track14 Hangup All Legs",
            status="active",
            telephony_status="connected",
            claimed_by_user="operator_a",
            claimed_at=utc_now_iso(),
            operator_extension="2001",
            channel_name="PJSIP/2001-0000015",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    monkeypatch.setattr(
        bridge_module,
        "_candidate_channels_for_call",
        lambda session, link, operator_extension=None, prefer_operator=False, snapshot=None: [
            "PJSIP/2001-0000015",
            "PJSIP/1001-0000016",
        ],
    )

    calls = []

    def _ami_action(action, params):
        calls.append(params["Channel"])
        return {"action_id": f"ami_{len(calls)}"}

    monkeypatch.setattr(bridge_module, "_ami_action", _ami_action)

    client = TestClient(bridge_module.app)
    response = client.post(
        f"/asterisk/live-calls/{call_id}/hangup",
        headers=_operator_headers("operator_a"),
        json={},
    )

    assert response.status_code == 200
    assert response.json()["telephony_status"] == "ended"
    assert calls == ["PJSIP/2001-0000015", "PJSIP/1001-0000016"]


def test_blind_transfer_emits_transferred_voice_event(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_CALLCONTROL_ENABLED", "1")
    monkeypatch.setenv("ASTERISK_TRANSFER_TARGET_MAP_JSON", '{"voice_lab":"2001"}')
    events: list[dict] = []

    call_id = f"call_transfer_{tmp_path.name}"
    session = get_session()
    try:
        row = AsteriskCallLinkRow(
            call_id=call_id,
            linked_id="linked_transfer",
            queue_code="voice_lab",
            queue_id="que_lab",
            interaction_id=f"int_transfer_{tmp_path.name}",
            caller_number="+77010000005",
            caller_name="Track11 Transfer",
            status="active",
            telephony_status="connected",
            claimed_by_user="operator_a",
            claimed_at=utc_now_iso(),
            operator_extension="2001",
            channel_name="PJSIP/1001-0000003",
            started_at=utc_now_iso(),
            connected_at=utc_now_iso(),
            ended_at=None,
            updated_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
    finally:
        session.close()

    monkeypatch.setattr(
        bridge_module,
        "_resolve_channel_name",
        lambda session, link, prefer_operator=False, refresh=False, snapshot=None: "PJSIP/1001-0000003",
    )
    monkeypatch.setattr(bridge_module, "_ami_action", lambda action, params: {"action_id": "ami_transfer"})
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: events.append(kwargs) or {"event_id": "vev_transfer"},
    )

    client = TestClient(bridge_module.app)
    moved = client.post(
        f"/asterisk/live-calls/{call_id}/blind-transfer",
        headers=_operator_headers("operator_a"),
        json={"target_type": "queue_code", "target_value": "voice_lab"},
    )
    assert moved.status_code == 200
    assert moved.json()["call_id"] == call_id
    assert events
    assert events[-1]["event_type"] == "call.transferred"


def test_operator_connected_event_sets_connected_status(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}')
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator_a":"2001"}')
    monkeypatch.setattr(bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_conn_{tmp_path.name}"})
    captured: list[dict] = []
    monkeypatch.setattr(
        bridge_module,
        "_emit_voice_event",
        lambda **kwargs: captured.append(kwargs) or {"event_id": "vev_conn"},
    )

    call_id = f"call_conn_{tmp_path.name}"
    session = get_session()
    try:
        started = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_conn",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_conn",
                "CallerNumber": "+77010000006",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )
        bridge_module._process_bridge_row(session, started)
        session.commit()

        connected = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCOperatorConnected",
            call_id=call_id,
            linked_id="linked_conn",
            payload={
                "UserEvent": "MVPCCOperatorConnected",
                "CallID": call_id,
                "LinkedID": "linked_conn",
                "OperatorExtension": "2001",
                "ClaimedByUser": "s",
                "Channel": "PJSIP/1001-0000010",
            },
        )
        bridge_module._process_bridge_row(session, connected)
        session.commit()

        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.telephony_status == "connected"
        assert link.operator_extension == "2001"
        assert link.claimed_by_user == "operator_a"
    finally:
        session.close()

    assert any(item.get("event_type") == "call.connected" for item in captured)


def test_operator_connected_does_not_guess_claimed_user_for_ambiguous_extension(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"lab":"que_lab"}')
    monkeypatch.setenv("ASTERISK_OPERATOR_EXTENSION_MAP_JSON", '{"operator":"2001","admin":"2001"}')
    monkeypatch.setattr(bridge_module, "_create_interaction", lambda **kwargs: {"interaction_id": f"int_conn_amb_{tmp_path.name}"})
    monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_conn_amb"})

    call_id = f"call_conn_amb_{tmp_path.name}"
    session = get_session()
    try:
        started = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id="linked_conn_amb",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": "linked_conn_amb",
                "CallerNumber": "+77010000009",
                "QueueCode": "lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )
        bridge_module._process_bridge_row(session, started)
        session.commit()

        connected = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCOperatorConnected",
            call_id=call_id,
            linked_id="linked_conn_amb",
            payload={
                "UserEvent": "MVPCCOperatorConnected",
                "CallID": call_id,
                "LinkedID": "linked_conn_amb",
                "OperatorExtension": "2001",
                "ClaimedByUser": "s",
                "Channel": "PJSIP/1001-0000011",
            },
        )
        bridge_module._process_bridge_row(session, connected)
        session.commit()

        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.telephony_status == "connected"
        assert link.operator_extension == "2001"
        assert link.claimed_by_user is None
    finally:
        session.close()


def test_app_module_exposes_refactor_compatibility_surface():
    assert bridge_module.app is not None
    assert callable(bridge_module._startup)
    assert callable(bridge_module._shutdown)
    assert callable(bridge_module._release_bridge_singleton_guard)
    assert bridge_module._process_bridge_row.__module__ == "services.asterisk_bridge_service.bridge_processing"
    assert bridge_module._live_call_to_out.__module__ == "services.asterisk_bridge_service.presenters"
    assert bridge_module._find_remote_recording_for_call.__module__ == "services.asterisk_bridge_service.recording_io"


def test_private_shims_remain_monkeypatchable(monkeypatch):
    monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "shim"})
    result = bridge_module._emit_voice_event(
        event_type="call.started",
        call_id="call_shim",
        interaction_id=None,
        payload={},
    )
    assert result == {"event_id": "shim"}


def test_call_started_with_ai_queue_creates_voice_ai_session(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"voice_lab":"que_voice_lab"}')
    monkeypatch.setenv("AI_VOICE_ENABLED", "1")
    monkeypatch.setenv(
        "AI_VOICE_QUEUE_CONFIG_JSON",
        '{"voice_lab":{"mode":"ai_first","agent_profile":"voice_support","handoff_queue_code":"voice_lab","language":"ru"}}',
    )
    monkeypatch.setattr(
        bridge_module,
        "_create_interaction",
        lambda **kwargs: {"interaction_id": f"int_voice_ai_{tmp_path.name}"},
    )
    monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_voice_ai"})
    monkeypatch.setattr(
        bridge_module,
        "_start_voice_ai_session",
        lambda **kwargs: {
            "voice_session_id": f"avs_voice_ai_{tmp_path.name}",
            "ai_session_id": f"ais_voice_ai_{tmp_path.name}",
            "status": "greeting",
        },
    )

    call_id = f"call_voice_ai_{tmp_path.name}"
    session = get_session()
    try:
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id=f"linked_voice_ai_{tmp_path.name}",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": f"linked_voice_ai_{tmp_path.name}",
                "CallerNumber": "+77010000041",
                "CallerName": "Voice AI Caller",
                "QueueCode": "voice_lab",
                "Extension": "7000",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )
        bridge_module._process_bridge_row(session, row)
        session.commit()

        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.voice_session_id == f"avs_voice_ai_{tmp_path.name}"
        assert link.ai_session_id == f"ais_voice_ai_{tmp_path.name}"
        assert link.ai_state == "greeting"
    finally:
        session.close()


def test_call_started_ai_does_not_flush_db_before_start_request(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"voice_lab":"que_voice_lab"}')
    monkeypatch.setenv("AI_VOICE_ENABLED", "1")
    monkeypatch.setenv(
        "AI_VOICE_QUEUE_CONFIG_JSON",
        '{"voice_lab":{"mode":"ai_first","agent_profile":"voice_support","handoff_queue_code":"voice_lab","language":"ru"}}',
    )
    monkeypatch.setattr(
        bridge_module,
        "_create_interaction",
        lambda **kwargs: {"interaction_id": f"int_voice_ai_flush_{tmp_path.name}"},
    )
    monkeypatch.setattr(bridge_module, "_emit_voice_event", lambda **kwargs: {"event_id": "vev_voice_ai_flush"})

    flush_count = {"value": 0}
    session = get_session()
    original_flush = type(session).flush

    def _tracked_flush(*args, **kwargs):
        flush_count["value"] += 1
        return original_flush(session, *args, **kwargs)

    call_id = f"call_voice_ai_flush_{tmp_path.name}"
    try:
        monkeypatch.setattr(session, "flush", _tracked_flush)
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCCallStarted",
            call_id=call_id,
            linked_id=f"linked_voice_ai_flush_{tmp_path.name}",
            payload={
                "UserEvent": "MVPCCCallStarted",
                "CallID": call_id,
                "LinkedID": f"linked_voice_ai_flush_{tmp_path.name}",
                "CallerNumber": "+77010000042",
                "CallerName": "Voice AI Flush Caller",
                "QueueCode": "voice_lab",
                "Extension": "7100",
                "Context": "mvpcc-inbound",
                "Direction": "inbound",
            },
        )
        flush_count["value"] = 0

        def _start_voice_ai_session(**kwargs):
            del kwargs
            assert flush_count["value"] == 0
            return {
                "voice_session_id": f"avs_voice_ai_flush_{tmp_path.name}",
                "ai_session_id": f"ais_voice_ai_flush_{tmp_path.name}",
                "status": "greeting",
            }

        monkeypatch.setattr(bridge_module, "_start_voice_ai_session", _start_voice_ai_session)

        bridge_module._process_bridge_row(session, row)
        session.commit()
    finally:
        session.close()


def test_audio_bridge_requested_registers_media_runtime(monkeypatch, tmp_path):
    recorded: list[dict] = []
    monkeypatch.setattr(
        bridge_module,
        "_register_voice_ai_media_bridge",
        lambda **kwargs: recorded.append(kwargs) or {"ok": True},
    )

    call_id = f"call_audio_bridge_{tmp_path.name}"
    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=f"linked_audio_bridge_{tmp_path.name}",
                queue_code="voice_lab",
                queue_id="que_voice_lab",
                interaction_id=f"int_audio_bridge_{tmp_path.name}",
                caller_number="+77010000051",
                caller_name="Audio Bridge Caller",
                status="active",
                telephony_status="ringing",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name=None,
                voice_session_id=f"avs_audio_bridge_{tmp_path.name}",
                ai_session_id=f"ais_audio_bridge_{tmp_path.name}",
                ai_state="queued",
                ai_handoff_reason=None,
                ai_last_model_at=None,
                started_at=utc_now_iso(),
                connected_at=None,
                ended_at=None,
                updated_at=utc_now_iso(),
            )
        )
        session.commit()
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCAIAudioBridgeRequested",
            call_id=call_id,
            linked_id=f"linked_audio_bridge_{tmp_path.name}",
            payload={
                "UserEvent": "MVPCCAIAudioBridgeRequested",
                "CallID": call_id,
                "LinkedID": f"linked_audio_bridge_{tmp_path.name}",
                "MediaUUID": "d64fa2f7-0063-4fe1-a36f-e1b372980ec1",
                "AudioSocketService": "127.0.0.1:9019",
                "Channel": "PJSIP/1001-00000011",
            },
        )
        bridge_module._process_bridge_row(session, row)
        session.commit()

        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.ai_state == "greeting"
    finally:
        session.close()

    assert recorded == [
        {
            "voice_session_id": f"avs_audio_bridge_{tmp_path.name}",
            "event_type": "requested",
            "media_uuid": "d64fa2f7-0063-4fe1-a36f-e1b372980ec1",
            "call_id": call_id,
            "linked_id": f"linked_audio_bridge_{tmp_path.name}",
            "channel": "PJSIP/1001-00000011",
            "service_address": "127.0.0.1:9019",
        }
    ]


def test_audio_bridge_requested_recovers_voice_session_from_db(monkeypatch, tmp_path):
    recorded: list[dict] = []
    monkeypatch.setattr(
        bridge_module,
        "_register_voice_ai_media_bridge",
        lambda **kwargs: recorded.append(kwargs) or {"ok": True},
    )

    call_id = f"call_audio_bridge_recover_{tmp_path.name}"
    linked_id = f"linked_audio_bridge_recover_{tmp_path.name}"
    voice_session_id = f"avs_audio_bridge_recover_{tmp_path.name}"
    ai_session_id = f"ais_audio_bridge_recover_{tmp_path.name}"
    interaction_id = f"int_audio_bridge_recover_{tmp_path.name}"
    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=linked_id,
                queue_code="voice_lab_ai",
                queue_id="que_voice_lab",
                interaction_id=interaction_id,
                caller_number="+77010000061",
                caller_name="Recovered Audio Bridge Caller",
                status="active",
                telephony_status="ringing",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name=None,
                voice_session_id=None,
                ai_session_id=None,
                ai_state="queued",
                ai_handoff_reason=None,
                ai_last_model_at=None,
                started_at=utc_now_iso(),
                connected_at=None,
                ended_at=None,
                updated_at=utc_now_iso(),
            )
        )
        session.add(
            VoiceAISessionRow(
                session_id=voice_session_id,
                call_id=call_id,
                linked_id=linked_id,
                interaction_id=interaction_id,
                customer_id=None,
                queue_id="que_voice_lab",
                ai_session_id=ai_session_id,
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="greeting",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_lab",
                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()
        row = bridge_module._create_bridge_log(
            session,
            ami_event_name="MVPCCAIAudioBridgeRequested",
            call_id=call_id,
            linked_id=linked_id,
            payload={
                "UserEvent": "MVPCCAIAudioBridgeRequested",
                "CallID": call_id,
                "LinkedID": linked_id,
                "MediaUUID": "9f23dd6f-3cd7-4aaf-8a4d-71154a3626d5",
                "AudioSocketService": "127.0.0.1:9019",
                "Channel": "PJSIP/1001-00000021",
            },
        )
        bridge_module._process_bridge_row(session, row)
        session.commit()

        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.voice_session_id == voice_session_id
        assert link.ai_session_id == ai_session_id
        assert link.ai_state == "greeting"
    finally:
        session.close()

    assert recorded == [
        {
            "voice_session_id": voice_session_id,
            "event_type": "requested",
            "media_uuid": "9f23dd6f-3cd7-4aaf-8a4d-71154a3626d5",
            "call_id": call_id,
            "linked_id": linked_id,
            "channel": "PJSIP/1001-00000021",
            "service_address": "127.0.0.1:9019",
        }
    ]


def test_voice_ai_update_call_state_updates_live_link(tmp_path):
    call_id = f"call_voice_state_{tmp_path.name}"
    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=f"linked_voice_state_{tmp_path.name}",
                queue_code="voice_lab",
                queue_id="que_voice_lab",
                interaction_id=f"int_voice_state_{tmp_path.name}",
                caller_number="+77010000052",
                caller_name="State Caller",
                status="active",
                telephony_status="connected",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name="PJSIP/1001-00000012",
                voice_session_id=f"avs_voice_state_{tmp_path.name}",
                ai_session_id=f"ais_voice_state_{tmp_path.name}",
                ai_state="greeting",
                ai_handoff_reason=None,
                ai_last_model_at=None,
                started_at=utc_now_iso(),
                connected_at=utc_now_iso(),
                ended_at=None,
                updated_at=utc_now_iso(),
            )
        )
        session.commit()
    finally:
        session.close()

    result = bridge_module._update_voice_ai_call_state(
        call_id,
        VoiceAICallStateUpdateIn(
            voice_session_id=f"avs_voice_state_{tmp_path.name}",
            ai_session_id=f"ais_voice_state_{tmp_path.name}",
            ai_state="speaking",
            handoff_reason=None,
            metadata={"media_uuid": "11111111-1111-1111-1111-111111111111"},
        ),
        actor={"auth_source": "service", "sub": "svc:ai-voice-runtime", "user": "ai-voice-runtime", "role": "admin"},
    )

    assert result.ai_state == "speaking"

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.ai_state == "speaking"
        assert link.ai_last_model_at is not None
    finally:
        session.close()


def test_live_call_ai_summary_route_returns_handoff_context(tmp_path):
    call_id = f"call_voice_summary_{tmp_path.name}"
    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=f"linked_voice_summary_{tmp_path.name}",
                queue_code="voice_lab",
                queue_id="que_voice_lab",
                interaction_id=f"int_voice_summary_{tmp_path.name}",
                caller_number="+77010000042",
                caller_name="Summary Caller",
                status="active",
                telephony_status="ringing",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name="PJSIP/1001-0000042",
                voice_session_id=f"avs_voice_summary_{tmp_path.name}",
                ai_session_id=f"ais_voice_summary_{tmp_path.name}",
                ai_state="handoff_required",
                ai_handoff_reason="Нужен живой оператор.",
                ai_last_model_at=utc_now_iso(),
                started_at=utc_now_iso(),
                connected_at=None,
                ended_at=None,
                updated_at=utc_now_iso(),
            )
        )
        session.add(
            VoiceAISessionRow(
                session_id=f"avs_voice_summary_{tmp_path.name}",
                call_id=call_id,
                linked_id=f"linked_voice_summary_{tmp_path.name}",
                interaction_id=f"int_voice_summary_{tmp_path.name}",
                customer_id=None,
                queue_id="que_voice_lab",
                ai_session_id=f"ais_voice_summary_{tmp_path.name}",
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="openai",
                status="handoff_requested",
                handoff_reason="Нужен живой оператор.",
                handoff_target_queue_id="que_voice_lab",
                disclosure_played_at=utc_now_iso(),
                last_user_utterance_at=utc_now_iso(),
                last_ai_reply_at=utc_now_iso(),
                started_at=utc_now_iso(),
                updated_at=utc_now_iso(),
                ended_at=None,
            )
        )
        session.add(
            AISessionRow(
                session_id=f"ais_voice_summary_{tmp_path.name}",
                channel="voice",
                call_id=call_id,
                thread_id=None,
                interaction_id=f"int_voice_summary_{tmp_path.name}",
                customer_id=None,
                agent_profile="voice_support",
                language="ru",
                status="handoff_required",
                summary_text="AI собрал контекст и передал звонок оператору.",
                last_user_message_id=None,
                last_ai_message_id=None,
                handoff_reason="Нужен живой оператор.",
                created_at=utc_now_iso(),
                updated_at=utc_now_iso(),
                closed_at=None,
            )
        )
        session.add(
            VoiceTranscriptSegmentRow(
                segment_id=f"vts_voice_summary_customer_{tmp_path.name}",
                session_id=f"avs_voice_summary_{tmp_path.name}",
                call_id=call_id,
                interaction_id=f"int_voice_summary_{tmp_path.name}",
                speaker="caller",
                source_type="asr",
                sequence_no=1,
                text="Соедините меня с оператором",
                confidence=0.91,
                is_final=True,
                barge_in_interrupted=False,
                payload_json="{}",
                created_at=utc_now_iso(),
            )
        )
        session.add(
            VoiceTranscriptSegmentRow(
                segment_id=f"vts_voice_summary_ai_{tmp_path.name}",
                session_id=f"avs_voice_summary_{tmp_path.name}",
                call_id=call_id,
                interaction_id=f"int_voice_summary_{tmp_path.name}",
                speaker="assistant",
                source_type="tts",
                sequence_no=2,
                text="Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора.",
                confidence=None,
                is_final=True,
                barge_in_interrupted=False,
                payload_json="{}",
                created_at=utc_now_iso(),
            )
        )
        session.add(
            VoiceTranscriptSegmentRow(
                segment_id=f"vts_voice_summary_ai_pending_{tmp_path.name}",
                session_id=f"avs_voice_summary_{tmp_path.name}",
                call_id=call_id,
                interaction_id=f"int_voice_summary_{tmp_path.name}",
                speaker="assistant",
                source_type="tts",
                sequence_no=3,
                text="Плановая недоставленная реплика",
                confidence=None,
                is_final=False,
                barge_in_interrupted=True,
                payload_json="{}",
                created_at=utc_now_iso(),
            )
        )
        session.commit()
    finally:
        session.close()


def test_voice_ai_handoff_revives_false_ended_call_when_channel_is_resolvable(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTERISK_QUEUE_MAP_JSON", '{"ivr_support":"que_ivr_support","voice_lab_ai":"que_voice_lab"}')
    monkeypatch.setenv("ASTERISK_TRANSFER_TARGET_MAP_JSON", '{"ivr_support":"2001","voice_lab_ai":"2001"}')

    call_id = f"call_ai_handoff_revive_{tmp_path.name}"
    interaction_id = f"int_ai_handoff_revive_{tmp_path.name}"
    session_id = f"avs_ai_handoff_revive_{tmp_path.name}"
    ai_session_id = f"ais_ai_handoff_revive_{tmp_path.name}"
    now = utc_now_iso()

    session = get_session()
    try:
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=f"linked_ai_handoff_revive_{tmp_path.name}",
                queue_code="voice_lab_ai",
                queue_id="que_voice_lab",
                interaction_id=interaction_id,
                caller_number="+77010000077",
                caller_name="Track16 Handoff Revive",
                status="ended",
                telephony_status="ended",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name="PJSIP/1001-0000077",
                voice_session_id=session_id,
                ai_session_id=ai_session_id,
                ai_state="handoff_requested",
                ai_handoff_reason="Нужен живой оператор.",
                ai_last_model_at=now,
                started_at=now,
                connected_at=None,
                ended_at=now,
                updated_at=now,
            )
        )
        session.add(
            VoiceAISessionRow(
                session_id=session_id,
                call_id=call_id,
                linked_id=f"linked_ai_handoff_revive_{tmp_path.name}",
                interaction_id=interaction_id,
                customer_id=None,
                queue_id="que_voice_lab",
                ai_session_id=ai_session_id,
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="yandex",
                status="handoff_requested",
                handoff_reason="Нужен живой оператор.",
                handoff_target_queue_id="que_ivr_support",
                media_uuid="22222222-2222-2222-2222-222222222222",
                media_status="connected",
                media_connected_at=now,
                media_ended_at=None,
                last_media_frame_at=now,
                disclosure_played_at=now,
                last_user_utterance_at=now,
                last_ai_reply_at=now,
                started_at=now,
                updated_at=now,
                ended_at=None,
            )
        )
        session.commit()
    finally:
        session.close()

    ami_calls: list[tuple[str, dict]] = []
    monkeypatch.setattr(
        bridge_module,
        "_list_channels_via_coreshowchannels",
        lambda **kwargs: ["PJSIP/1001-0000077"],
    )
    monkeypatch.setattr(
        bridge_module,
        "_ami_action",
        lambda action, params: ami_calls.append((action, params)) or {"action_id": "ami_handoff_revive"},
    )

    result = bridge_module._request_voice_ai_handoff(
        call_id,
        VoiceAIHandoffRequestIn(
            voice_session_id=session_id,
            ai_session_id=ai_session_id,
            interaction_id=interaction_id,
            target_queue_id="que_ivr_support",
            reason="Нужен живой оператор.",
            summary={"customer_request_text": "Соедините меня с оператором"},
        ),
        actor={"auth_source": "service", "sub": "svc:ai-voice-runtime", "user": "ai-voice-runtime", "role": "admin"},
    )

    assert result.call_id == call_id
    assert ami_calls
    assert [call[0] for call in ami_calls[:4]] == ["Setvar", "Setvar", "Setvar", "Setvar"]
    assert ami_calls[0][1]["Variable"] == "MVPCC_START_LANGUAGE"
    assert ami_calls[3][1]["Variable"] == "MVPCC_CUSTOMER_NAME_SOURCE"
    assert ami_calls[-1][0] == "Redirect"
    assert ami_calls[-1][1]["Channel"] == "PJSIP/1001-0000077"
    assert ami_calls[-1][1]["Exten"] == "2001"
    payload = {
        "status_label": "AI передал звонок оператору",
        "customer_request_text": "Соедините меня с оператором",
        "ai_outcome_text": "Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора.",
        "handoff_reason": "Нужен живой оператор.",
    }

    session = get_session()
    try:
        link = session.execute(
            select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
        ).scalar_one()
        assert link.status == "active"
        assert link.telephony_status == "connected"
        assert link.ended_at is None
        assert link.ai_state == "handoff_required"
    finally:
        session.close()

    assert payload["status_label"] == "AI передал звонок оператору"
    assert payload["customer_request_text"] == "Соедините меня с оператором"
    assert payload["ai_outcome_text"] == "Я как AI-оператор собрал первичный контекст. Сейчас переведу вас на живого оператора."
    assert payload["handoff_reason"] == "Нужен живой оператор."
