"""Tests for outbound call manager: status callback parsing, emission, and manager lifecycle."""

from __future__ import annotations

import asyncio
import threading
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

from easycat._epoch import Epoch
from easycat.events import (
    CallAnswered,
    CallEnded,
    CallFailed,
    CallInitiated,
    CallRinging,
    EventBus,
    VoicemailDetected,
)
from easycat.telephony.outbound import (
    OutboundCallManager,
    OutboundCallManagerState,
    emit_call_status,
    parse_call_status_callback,
)


def _make_bare_manager(**overrides: object) -> OutboundCallManager:
    """Construct a manager with every private lifecycle field initialized."""
    manager = OutboundCallManager.__new__(OutboundCallManager)
    defaults: dict[str, object] = {
        "_state": OutboundCallManagerState.IDLE,
        "_active_call_sid": None,
        "_owned_call_sids": set(),
        "_pending_cleanup_call_sids": set(),
        "_reconciling_call_sids": set(),
        "_terminal_reconciliation_call_sids": set(),
        "_synthetic_failure_event_ids": set(),
        "_session_id": None,
        "_started": False,
        "_lifecycle_epoch": Epoch(None),
        "_place_call_lock": asyncio.Lock(),
    }
    defaults.update(overrides)
    for name, value in defaults.items():
        setattr(manager, name, value)
    return manager


class TestParseCallStatusCallback:
    @pytest.mark.parametrize(
        "params",
        [
            {"CallStatus": "initiated", "CallSid": "CA123"},
            {"CallStatus": "ringing", "CallSid": "CA123"},
            {"CallStatus": "in-progress", "CallSid": "CA123"},
            {"CallStatus": "completed", "CallSid": "CA123"},
            {"CallStatus": "failed", "CallSid": "CA123"},
        ],
    )
    def test_stamps_session_correlation_on_every_lifecycle_status(
        self,
        params: dict[str, str],
    ) -> None:
        result = parse_call_status_callback(params, session_id="session-1")

        assert result is not None
        assert result.session_id == "session-1"

    def test_initiated_status(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "initiated", "CallSid": "CA123", "To": "+1555", "From": "+1999"}
        )
        assert isinstance(result, CallInitiated)
        assert result.call_sid == "CA123"
        assert result.to == "+1555"
        assert result.from_ == "+1999"

    def test_ringing_status(self) -> None:
        result = parse_call_status_callback({"CallStatus": "ringing", "CallSid": "CA123"})
        assert isinstance(result, CallRinging)
        assert result.call_sid == "CA123"

    def test_answered_status(self) -> None:
        result = parse_call_status_callback({"CallStatus": "in-progress", "CallSid": "CA123"})
        assert isinstance(result, CallAnswered)
        assert result.call_sid == "CA123"

    def test_completed_status(self) -> None:
        result = parse_call_status_callback(
            {
                "CallStatus": "completed",
                "CallSid": "CA123",
                "Duration": "45",
                "To": "+1555",
                "From": "+1999",
            }
        )
        assert isinstance(result, CallEnded)
        assert result.duration_s == 45.0
        assert result.disposition == "completed"
        assert result.number == "+1999"

    def test_completed_status_reads_call_duration_alias(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "completed", "CallSid": "CA123", "CallDuration": "45"}
        )
        assert isinstance(result, CallEnded)
        assert result.duration_s == 45.0

    def test_completed_status_prefers_call_duration_alias(self) -> None:
        result = parse_call_status_callback(
            {
                "CallStatus": "completed",
                "CallSid": "CA123",
                "Duration": "1",
                "CallDuration": "45",
            }
        )
        assert isinstance(result, CallEnded)
        assert result.duration_s == 45.0

    def test_completed_status_uses_call_duration_alias_when_duration_is_malformed(self) -> None:
        result = parse_call_status_callback(
            {
                "CallStatus": "completed",
                "CallSid": "CA123",
                "Duration": "not-a-number",
                "CallDuration": "45",
            }
        )
        assert isinstance(result, CallEnded)
        assert result.duration_s == 45.0

    def test_completed_status_ignores_malformed_duration(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "completed", "CallSid": "CA123", "Duration": "not-a-number"}
        )
        assert isinstance(result, CallEnded)
        assert result.duration_s is None

    def test_completed_status_ignores_boolean_duration(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "completed", "CallSid": "CA123", "Duration": True}
        )
        assert isinstance(result, CallEnded)
        assert result.duration_s is None

    def test_busy_status(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "busy", "CallSid": "CA123", "To": "+1555", "From": "+1999"}
        )
        assert isinstance(result, CallFailed)
        assert result.reason == "busy"
        assert result.number == "+1999"

    def test_no_answer_status(self) -> None:
        result = parse_call_status_callback({"CallStatus": "no-answer", "CallSid": "CA123"})
        assert isinstance(result, CallFailed)
        assert result.reason == "no-answer"

    def test_failed_status(self) -> None:
        result = parse_call_status_callback({"CallStatus": "failed", "CallSid": "CA123"})
        assert isinstance(result, CallFailed)
        assert result.reason == "failed"

    def test_canceled_status(self) -> None:
        result = parse_call_status_callback({"CallStatus": "canceled", "CallSid": "CA123"})
        assert isinstance(result, CallFailed)
        assert result.reason == "canceled"

    def test_missing_call_status(self) -> None:
        result = parse_call_status_callback({"CallSid": "CA123"})
        assert result is None

    def test_missing_call_sid(self) -> None:
        result = parse_call_status_callback({"CallStatus": "ringing"})
        assert result is None

    def test_unknown_status(self) -> None:
        result = parse_call_status_callback({"CallStatus": "something_new", "CallSid": "CA123"})
        assert result is None

    def test_sip_response_code_607_blocked(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "failed", "CallSid": "CA123", "SipResponseCode": "607"}
        )
        assert isinstance(result, CallFailed)
        assert result.reason == "blocked_unwanted"
        assert result.sip_code == 607

    def test_sip_response_code_608_rejected(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "failed", "CallSid": "CA123", "SipResponseCode": "608"}
        )
        assert isinstance(result, CallFailed)
        assert result.reason == "blocked_rejected"
        assert result.sip_code == 608

    def test_sip_response_code_603_declined(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "failed", "CallSid": "CA123", "SipResponseCode": "603"}
        )
        assert isinstance(result, CallFailed)
        assert result.reason == "declined"
        assert result.sip_code == 603

    def test_malformed_sip_response_code_falls_back_to_status(self) -> None:
        result = parse_call_status_callback(
            {"CallStatus": "failed", "CallSid": "CA123", "SipResponseCode": "nope"}
        )
        assert isinstance(result, CallFailed)
        assert result.reason == "failed"
        assert result.sip_code is None


class TestEmitCallStatus:
    @pytest.mark.asyncio
    async def test_emits_to_bus(self) -> None:
        bus = EventBus()
        received: list[CallRinging] = []
        bus.subscribe(CallRinging, received.append)
        result = await emit_call_status({"CallStatus": "ringing", "CallSid": "CA1"}, bus)
        assert isinstance(result, CallRinging)
        assert len(received) == 1
        assert received[0].call_sid == "CA1"

    @pytest.mark.asyncio
    async def test_skips_unparseable(self) -> None:
        bus = EventBus()
        received: list[object] = []
        bus.subscribe_all(received.append)
        result = await emit_call_status({"invalid": "data"}, bus)
        assert result is None
        assert len(received) == 0

    @pytest.mark.asyncio
    async def test_emits_amd_voicemail_detected(self) -> None:
        bus = EventBus()
        received: list[VoicemailDetected] = []
        bus.subscribe(VoicemailDetected, received.append)
        result = await emit_call_status(
            {"CallStatus": "in-progress", "CallSid": "CA1", "AnsweredBy": "machine_start"},
            bus,
            session_id="session-1",
        )
        assert isinstance(result, CallAnswered)
        assert result.session_id == "session-1"
        assert len(received) == 1
        assert received[0].result == "machine"
        assert received[0].call_sid == "CA1"
        assert received[0].session_id == "session-1"

    @pytest.mark.asyncio
    async def test_initiated_callback_for_sid_evicted_call_does_not_repin_capacity(self) -> None:
        """A replayed Twilio initiated callback for a tombstoned SID is ignored."""
        from easycat.telephony.number_health import NumberHealthMonitor

        bus = EventBus()
        monitor = NumberHealthMonitor(
            bus,
            max_concurrent_per_number=3,
            max_calls_per_minute=100,
            min_inter_call_delay_s=0.0,
        )
        monitor._max_sid_tracking = 4
        from_number = "+1999"
        to_number = "+1555"
        monitor.start()
        try:
            for i in range(5):
                await emit_call_status(
                    {
                        "CallStatus": "initiated",
                        "CallSid": f"CA{i}",
                        "To": to_number,
                        "From": from_number,
                    },
                    bus,
                )
            assert monitor._concurrent[from_number] == 3
            assert not monitor.can_place_call(from_number)

            await emit_call_status(
                {
                    "CallStatus": "initiated",
                    "CallSid": "CA0",
                    "To": to_number,
                    "From": from_number,
                },
                bus,
            )
            assert monitor._concurrent[from_number] == 3
            assert "CA0" not in monitor._call_sid_to_number

            await emit_call_status(
                {
                    "CallStatus": "completed",
                    "CallSid": "CA0",
                    "Duration": "5",
                    "To": to_number,
                    "From": from_number,
                },
                bus,
            )
            assert monitor._concurrent[from_number] == 3

            for i in range(2, 5):
                await emit_call_status(
                    {
                        "CallStatus": "completed",
                        "CallSid": f"CA{i}",
                        "Duration": "5",
                        "To": to_number,
                        "From": from_number,
                    },
                    bus,
                )
            assert monitor._concurrent.get(from_number, 0) == 0
            assert monitor.can_place_call(from_number)
        finally:
            monitor.stop()

    @pytest.mark.asyncio
    async def test_terminal_callback_for_sid_evicted_call_does_not_double_release(self) -> None:
        """An outbound terminal callback for an SID-evicted call must not double-release.

        The ``completed``/``busy`` callbacks now parse ``number=From``. When a
        :class:`NumberHealthMonitor` has already evicted (and decremented) an SID
        because tracking exceeded its cap, replaying a late terminal callback for
        that SID through ``emit_call_status`` must not decrement the caller bucket
        below the true live count — the eviction tombstone short-circuits it.
        """
        from easycat.telephony.number_health import NumberHealthMonitor

        bus = EventBus()
        monitor = NumberHealthMonitor(
            bus,
            max_concurrent_per_number=100,
            max_calls_per_minute=100,
            min_inter_call_delay_s=0.0,
        )
        monitor._max_sid_tracking = 4
        from_number = "+1999"
        to_number = "+1555"
        monitor.start()
        try:
            # Five initiated callbacks on the same caller ID drive SID tracking
            # over the cap (5 > 4); the 2 oldest SIDs are evicted, leaving the
            # bucket at the true live count of 3 (CA2, CA3, CA4).
            for i in range(5):
                await emit_call_status(
                    {
                        "CallStatus": "initiated",
                        "CallSid": f"CA{i}",
                        "To": to_number,
                        "From": from_number,
                    },
                    bus,
                )
            assert monitor._concurrent[from_number] == 3

            # Late completed callback (carrying number=From) for evicted CA0 must
            # not drop the bucket below the live count.
            await emit_call_status(
                {
                    "CallStatus": "completed",
                    "CallSid": "CA0",
                    "Duration": "5",
                    "To": to_number,
                    "From": from_number,
                },
                bus,
            )
            assert monitor._concurrent[from_number] == 3
        finally:
            monitor.stop()


class TestOutboundCallManager:
    def test_twilio_sdk_import_error(self) -> None:
        bus = EventBus()
        with patch.dict("sys.modules", {"twilio": None, "twilio.rest": None}):  # noqa: SIM117 nested scopes clarify setup and cleanup
            with pytest.raises(ImportError) as exc_info:
                OutboundCallManager(bus, from_number="+1555")
        message = str(exc_info.value)
        assert "uv add 'easycat[telephony]'" in message
        assert "From the EasyCat repo, use: uv sync --extra telephony --group dev" in message
        assert "pip install easycat[telephony]" not in message

    @patch("easycat.telephony.outbound.OutboundCallManager.__init__", return_value=None)
    def test_init_stores_config(self, mock_init: MagicMock) -> None:
        manager = _make_bare_manager()
        assert manager.state == OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None

    @patch("easycat.telephony.outbound.OutboundCallManager.__init__", return_value=None)
    def test_start_stop_idempotent(self, mock_init: MagicMock) -> None:
        manager = _make_bare_manager(_event_bus=EventBus())
        initial = manager._lifecycle_epoch.capture()
        manager.start()
        started = manager._lifecycle_epoch.capture()
        assert not initial.is_current()
        manager.start()
        assert manager._started is True
        assert started.is_current()
        manager.stop()
        stopped = manager._lifecycle_epoch.capture()
        manager.stop()
        assert manager._started is False
        assert not started.is_current()
        assert not stopped.is_current()

    @patch("easycat.telephony.outbound.OutboundCallManager.__init__", return_value=None)
    def test_stop_resets_state(self, mock_init: MagicMock) -> None:
        manager = _make_bare_manager(
            _event_bus=EventBus(),
            _state=OutboundCallManagerState.ACTIVE,
            _active_call_sid="CA1",
            _owned_call_sids={"CA1"},
            _started=True,
        )
        manager.stop()
        assert manager.state == OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._started is False

    @patch("easycat.telephony.outbound.OutboundCallManager.__init__", return_value=None)
    def test_active_call_sid_is_read_only(self, mock_init: MagicMock) -> None:
        manager = OutboundCallManager.__new__(OutboundCallManager)
        manager._active_call_sid = None
        with pytest.raises(AttributeError):
            manager.active_call_sid = "CA1"  # type: ignore[misc]


class TestOutboundCallManagerPlaceCall:
    def _make_manager(self, bus: EventBus) -> OutboundCallManager:
        manager = _make_bare_manager(_event_bus=bus, _started=True)
        manager._from_number = "+15559876543"
        manager._amd_mode = "DetectMessageEnd"
        manager._async_amd = True
        manager._amd_timeout = 30
        manager._speech_threshold = 2400
        manager._speech_end_threshold = 1200
        manager._silence_timeout = 5000
        manager._enable_realtime_transcription = True
        manager._status_callback_url = "https://example.com/status"
        manager._twiml_url = "https://example.com/twiml"
        manager._client = MagicMock()
        manager.dnc_list = None
        manager.compliance_check = None
        manager.retry_strategy = None
        return manager

    @pytest.mark.asyncio
    async def test_place_call_emits_initiated(self) -> None:
        bus = EventBus()
        received: list[CallInitiated] = []
        bus.subscribe(CallInitiated, received.append)
        manager = self._make_manager(bus)
        manager.set_session_id("session-1")
        mock_call = MagicMock()
        mock_call.sid = "CA999"
        manager._client.calls.create.return_value = mock_call
        await manager.place_call("+15551234567")
        assert len(received) == 1
        assert received[0].call_sid == "CA999"
        assert received[0].to == "+15551234567"
        assert received[0].from_ == "+15559876543"
        assert received[0].session_id == "session-1"
        assert manager.state == OutboundCallManagerState.ACTIVE
        assert manager.active_call_sid == "CA999"

    @pytest.mark.asyncio
    async def test_initiated_dispatch_failure_completes_activated_provider_call(self) -> None:
        bus = EventBus(handler_error_policy="raise")
        failed: list[CallFailed] = []

        async def fail_initiated(_event: CallInitiated) -> None:
            raise RuntimeError("subscriber exploded")

        bus.subscribe(CallInitiated, fail_initiated)
        bus.subscribe(CallFailed, failed.append)
        manager = self._make_manager(bus)
        manager._client.calls.create.return_value = SimpleNamespace(sid="CA-DISPATCH-FAILED")
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource

        with pytest.raises(RuntimeError, match="subscriber exploded"):
            await manager.place_call("+15551234567")

        manager._client.calls.assert_called_once_with("CA-DISPATCH-FAILED")
        call_resource.update.assert_called_once_with(status="completed")
        assert [event.call_sid for event in failed] == ["CA-DISPATCH-FAILED"]
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()
        assert manager._pending_cleanup_call_sids == set()
        assert manager._reconciling_call_sids == set()

    @pytest.mark.asyncio
    async def test_cancelled_initiated_dispatch_completes_activated_provider_call(self) -> None:
        bus = EventBus()
        dispatch_started = asyncio.Event()
        failed: list[CallFailed] = []

        async def block_initiated(_event: CallInitiated) -> None:
            dispatch_started.set()
            await asyncio.Event().wait()

        bus.subscribe(CallInitiated, block_initiated)
        bus.subscribe(CallFailed, failed.append)
        manager = self._make_manager(bus)
        manager._client.calls.create.return_value = SimpleNamespace(sid="CA-CANCELLED-DISPATCH")
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource

        placement = asyncio.create_task(manager.place_call("+15551234567"))
        await dispatch_started.wait()
        placement.cancel()

        with pytest.raises(asyncio.CancelledError):
            await placement

        manager._client.calls.assert_called_once_with("CA-CANCELLED-DISPATCH")
        call_resource.update.assert_called_once_with(status="completed")
        assert [event.call_sid for event in failed] == ["CA-CANCELLED-DISPATCH"]
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()
        assert manager._pending_cleanup_call_sids == set()
        assert manager._reconciling_call_sids == set()

    @pytest.mark.asyncio
    async def test_terminal_callback_during_cancelled_initiated_dispatch_wins_over_hangup_error(
        self,
    ) -> None:
        bus = EventBus()
        dispatch_started = asyncio.Event()

        async def block_initiated(_event: CallInitiated) -> None:
            dispatch_started.set()
            await asyncio.Event().wait()

        bus.subscribe(CallInitiated, block_initiated)
        manager = self._make_manager(bus)
        manager._started = False
        manager.start()
        manager._client.calls.create.return_value = SimpleNamespace(
            sid="CA-TERMINAL-DURING-DISPATCH"
        )
        call_resource = MagicMock()
        call_resource.update.side_effect = RuntimeError("already terminal")
        manager._client.calls.return_value = call_resource

        placement = asyncio.create_task(manager.place_call("+15551234567"))
        await dispatch_started.wait()
        await bus.emit(CallEnded(call_sid="CA-TERMINAL-DURING-DISPATCH"))
        placement.cancel()

        with pytest.raises(asyncio.CancelledError):
            await placement

        call_resource.update.assert_called_once_with(status="completed")
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()
        assert manager._pending_cleanup_call_sids == set()
        assert manager._reconciling_call_sids == set()
        assert manager._terminal_reconciliation_call_sids == set()

    @pytest.mark.asyncio
    async def test_place_call_configures_amd(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        mock_call = MagicMock()
        mock_call.sid = "CA999"
        manager._client.calls.create.return_value = mock_call
        await manager.place_call("+15551234567")
        kwargs = manager._client.calls.create.call_args.kwargs
        assert kwargs["machine_detection"] == "DetectMessageEnd"
        assert kwargs["async_amd"] == "true"

    @pytest.mark.asyncio
    async def test_place_call_configures_transcription(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        mock_call = MagicMock()
        mock_call.sid = "CA999"
        manager._client.calls.create.return_value = mock_call
        await manager.place_call("+15551234567")
        kwargs = manager._client.calls.create.call_args.kwargs
        assert kwargs["transcription"] is True
        assert kwargs["transcription_track"] == "inbound_track"

    @pytest.mark.asyncio
    async def test_place_call_uses_from_number(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        mock_call = MagicMock()
        mock_call.sid = "CA999"
        manager._client.calls.create.return_value = mock_call
        await manager.place_call("+15551234567")
        kwargs = manager._client.calls.create.call_args.kwargs
        assert kwargs["from_"] == "+15559876543"

    @pytest.mark.asyncio
    async def test_place_call_returns_call_sid(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        mock_call = MagicMock()
        mock_call.sid = "CA999"
        manager._client.calls.create.return_value = mock_call
        result = await manager.place_call("+15551234567")
        assert result == "CA999"

    @pytest.mark.asyncio
    async def test_place_call_failure_emits_call_failed(self) -> None:
        bus = EventBus()
        received: list[CallFailed] = []
        bus.subscribe(CallFailed, received.append)
        manager = self._make_manager(bus)
        manager.set_session_id("session-1")
        manager._client.calls.create.side_effect = RuntimeError("network error")
        with pytest.raises(RuntimeError, match="network error"):
            await manager.place_call("+15551234567")
        assert len(received) == 1
        assert "network error" in received[0].reason
        assert received[0].session_id == "session-1"

    @pytest.mark.asyncio
    async def test_place_call_blocks_dnc_numbers(self) -> None:
        from easycat.telephony.compliance import DNCList

        bus = EventBus()
        manager = self._make_manager(bus)
        dnc = DNCList()
        dnc.add("+15551234567")
        manager.dnc_list = dnc
        with pytest.raises(ValueError, match="DNC"):
            await manager.place_call("+15551234567")
        manager._client.calls.create.assert_not_called()

    @pytest.mark.asyncio
    async def test_place_call_honors_compliance_check(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager.compliance_check = lambda _to: False
        with pytest.raises(ValueError, match="compliance_check"):
            await manager.place_call("+15551234567")
        manager._client.calls.create.assert_not_called()

    @pytest.mark.asyncio
    async def test_place_call_requires_start(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager._started = False
        with pytest.raises(RuntimeError, match="started"):
            await manager.place_call("+15551234567")
        manager._client.calls.create.assert_not_called()

    @pytest.mark.asyncio
    async def test_place_call_requires_idle(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager._state = OutboundCallManagerState.ACTIVE
        manager._active_call_sid = "CA999"
        with pytest.raises(RuntimeError, match="active call"):
            await manager.place_call("+15551234567")
        manager._client.calls.create.assert_not_called()

    @pytest.mark.asyncio
    async def test_concurrent_place_call_originates_only_one_call(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        first_create_started = threading.Event()
        release_first_create = threading.Event()
        create_lock = threading.Lock()
        create_count = 0

        def create_call(**_kwargs: object) -> SimpleNamespace:
            nonlocal create_count
            with create_lock:
                create_count += 1
                call_number = create_count
            if call_number == 1:
                first_create_started.set()
                assert release_first_create.wait(timeout=2)
            return SimpleNamespace(sid=f"CA{call_number}")

        manager._client.calls.create.side_effect = create_call
        first = asyncio.create_task(manager.place_call("+15551111111"))
        assert await asyncio.to_thread(first_create_started.wait, 1)

        second = asyncio.create_task(manager.place_call("+15552222222"))
        # Give the second task a scheduling opportunity while the first REST
        # create remains blocked. Without transaction serialization it also
        # observes IDLE and submits a second create.
        await asyncio.sleep(0.05)
        release_first_create.set()

        assert await first == "CA1"
        with pytest.raises(RuntimeError, match="active call"):
            await second
        assert create_count == 1
        assert manager.active_call_sid == "CA1"
        assert manager._owned_call_sids == {"CA1"}

    @pytest.mark.asyncio
    async def test_reentrant_place_call_handler_does_not_deadlock(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager._client.calls.create.return_value = SimpleNamespace(sid="CA1")
        reentrant_errors: list[RuntimeError] = []

        async def place_again(_event: CallInitiated) -> None:
            try:
                await manager.place_call("+15552222222")
            except RuntimeError as exc:
                reentrant_errors.append(exc)

        bus.subscribe(CallInitiated, place_again)

        assert (
            await asyncio.wait_for(
                manager.place_call("+15551111111"),
                timeout=1,
            )
            == "CA1"
        )
        assert len(reentrant_errors) == 1
        assert "active call" in str(reentrant_errors[0])
        manager._client.calls.create.assert_called_once()
        assert manager.active_call_sid == "CA1"
        assert manager._owned_call_sids == {"CA1"}

    @pytest.mark.asyncio
    async def test_cancelled_place_call_completes_created_sid_before_reraising(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        first_create_started = threading.Event()
        release_first_create = threading.Event()
        create_count = 0
        initiated: list[CallInitiated] = []
        failed: list[CallFailed] = []
        bus.subscribe(CallInitiated, initiated.append)
        bus.subscribe(CallFailed, failed.append)

        def create_call(**_kwargs: object) -> SimpleNamespace:
            nonlocal create_count
            create_count += 1
            if create_count == 1:
                first_create_started.set()
                assert release_first_create.wait(timeout=2)
            return SimpleNamespace(sid=f"CA{create_count}")

        manager._client.calls.create.side_effect = create_call
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource
        first = asyncio.create_task(manager.place_call("+15551111111"))
        assert await asyncio.to_thread(first_create_started.wait, 1)

        first.cancel()
        await asyncio.sleep(0.05)
        first.cancel()
        await asyncio.sleep(0)

        # Repeated cancellation does not abandon the uncancellable REST
        # worker or expose a provider-created SID to the caller.
        assert not first.done()
        assert create_count == 1

        release_first_create.set()
        with pytest.raises(asyncio.CancelledError):
            await first

        assert create_count == 1
        manager._client.calls.assert_called_once_with("CA1")
        call_resource.update.assert_called_once_with(status="completed")
        assert initiated == []
        assert [event.call_sid for event in failed] == ["CA1"]
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()
        assert manager._pending_cleanup_call_sids == set()
        assert manager._reconciling_call_sids == set()

        # Successful reconciliation releases placement ownership.
        assert await manager.place_call("+15552222222") == "CA2"
        assert [event.call_sid for event in initiated] == ["CA2"]

    @pytest.mark.asyncio
    async def test_stop_during_create_completes_stale_call_without_reactivation(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        create_started = threading.Event()
        release_create = threading.Event()
        initiated: list[CallInitiated] = []
        failed: list[CallFailed] = []
        bus.subscribe(CallInitiated, initiated.append)
        bus.subscribe(CallFailed, failed.append)

        def create_call(**_kwargs: object) -> SimpleNamespace:
            create_started.set()
            assert release_create.wait(timeout=2)
            return SimpleNamespace(sid="CA-STALE")

        manager._client.calls.create.side_effect = create_call
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource
        placement = asyncio.create_task(manager.place_call("+15551111111"))
        assert await asyncio.to_thread(create_started.wait, 1)

        manager.stop()
        release_create.set()

        with pytest.raises(RuntimeError, match="lifecycle change"):
            await placement
        manager._client.calls.assert_called_once_with("CA-STALE")
        call_resource.update.assert_called_once_with(status="completed")
        assert initiated == []
        assert [event.call_sid for event in failed] == ["CA-STALE"]
        assert manager._started is False
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()

    @pytest.mark.asyncio
    async def test_stop_during_initiated_dispatch_completes_stale_call(self) -> None:
        bus = EventBus()
        dispatch_started = asyncio.Event()
        release_dispatch = asyncio.Event()
        failed: list[CallFailed] = []

        async def block_initiated(_event: CallInitiated) -> None:
            dispatch_started.set()
            await release_dispatch.wait()

        bus.subscribe(CallInitiated, block_initiated)
        bus.subscribe(CallFailed, failed.append)
        manager = self._make_manager(bus)
        manager._client.calls.create.return_value = SimpleNamespace(sid="CA-STOP-DISPATCH")
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource

        placement = asyncio.create_task(manager.place_call("+15551111111"))
        await dispatch_started.wait()

        manager.stop()
        release_dispatch.set()

        with pytest.raises(RuntimeError, match="lifecycle change"):
            await placement
        manager._client.calls.assert_called_once_with("CA-STOP-DISPATCH")
        call_resource.update.assert_called_once_with(status="completed")
        assert [event.call_sid for event in failed] == ["CA-STOP-DISPATCH"]
        assert manager._started is False
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()
        assert manager._pending_cleanup_call_sids == set()
        assert manager._reconciling_call_sids == set()

    @pytest.mark.asyncio
    async def test_cancelled_stop_during_create_cleans_up_then_preserves_cancellation(
        self,
    ) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        create_started = threading.Event()
        release_create = threading.Event()
        failed: list[CallFailed] = []
        bus.subscribe(CallFailed, failed.append)

        def create_call(**_kwargs: object) -> SimpleNamespace:
            create_started.set()
            assert release_create.wait(timeout=2)
            return SimpleNamespace(sid="CA-CANCELLED-STALE")

        manager._client.calls.create.side_effect = create_call
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource
        placement = asyncio.create_task(manager.place_call("+15551111111"))
        assert await asyncio.to_thread(create_started.wait, 1)

        placement.cancel()
        manager.stop()
        release_create.set()

        with pytest.raises(asyncio.CancelledError):
            await placement
        call_resource.update.assert_called_once_with(status="completed")
        assert [event.call_sid for event in failed] == ["CA-CANCELLED-STALE"]
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager._owned_call_sids == set()

    @pytest.mark.asyncio
    async def test_stop_then_restart_does_not_adopt_old_placement(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        create_started = threading.Event()
        release_create = threading.Event()
        initiated: list[CallInitiated] = []
        bus.subscribe(CallInitiated, initiated.append)

        def create_call(**_kwargs: object) -> SimpleNamespace:
            create_started.set()
            assert release_create.wait(timeout=2)
            return SimpleNamespace(sid="CA-OLD-EPOCH")

        manager._client.calls.create.side_effect = create_call
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource
        placement = asyncio.create_task(manager.place_call("+15551111111"))
        assert await asyncio.to_thread(create_started.wait, 1)

        manager.stop()
        manager.start()
        release_create.set()

        with pytest.raises(RuntimeError, match="lifecycle change"):
            await placement
        manager._client.calls.assert_called_once_with("CA-OLD-EPOCH")
        call_resource.update.assert_called_once_with(status="completed")
        assert initiated == []
        assert manager._started is True
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()

    @pytest.mark.asyncio
    async def test_stale_call_cleanup_failure_retains_ownership_for_retry(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        create_started = threading.Event()
        release_create = threading.Event()

        def create_call(**_kwargs: object) -> SimpleNamespace:
            create_started.set()
            assert release_create.wait(timeout=2)
            return SimpleNamespace(sid="CA-NEEDS-RETRY")

        manager._client.calls.create.side_effect = create_call
        call_resource = MagicMock()
        call_resource.update.side_effect = RuntimeError("Twilio unavailable")
        manager._client.calls.return_value = call_resource
        placement = asyncio.create_task(manager.place_call("+15551111111"))
        assert await asyncio.to_thread(create_started.wait, 1)

        manager.stop()
        manager.start()
        release_create.set()

        with pytest.raises(RuntimeError, match="immediate hangup failed"):
            await placement
        assert manager._started is True
        assert manager.state is OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        assert manager._owned_call_sids == set()
        assert manager._pending_cleanup_call_sids == {"CA-NEEDS-RETRY"}
        with pytest.raises(RuntimeError, match="active call"):
            await manager.place_call("+15552222222")

        # Normal lifecycle reset must not discard retry authority for the
        # provider call that is still live.
        manager.stop()
        manager.start()
        assert manager._pending_cleanup_call_sids == {"CA-NEEDS-RETRY"}
        with pytest.raises(RuntimeError, match="pending cleanup"):
            await manager.place_call("+15553333333")

        call_resource.update.side_effect = None
        await manager.hangup_owned_call("CA-NEEDS-RETRY")
        assert call_resource.update.call_count == 2
        call_resource.update.assert_called_with(status="completed")
        assert manager._pending_cleanup_call_sids == set()

    @pytest.mark.asyncio
    async def test_cancelled_owned_hangup_waits_for_provider_success_then_clears(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager._pending_cleanup_call_sids.add("CA-PENDING")
        update_started = threading.Event()
        release_update = threading.Event()

        def update_call(*, status: str) -> None:
            assert status == "completed"
            update_started.set()
            assert release_update.wait(timeout=2)

        call_resource = MagicMock()
        call_resource.update.side_effect = update_call
        manager._client.calls.return_value = call_resource

        hanging_up = asyncio.create_task(manager.hangup_owned_call("CA-PENDING"))
        assert await asyncio.to_thread(update_started.wait, 1)
        hanging_up.cancel()
        await asyncio.sleep(0)
        assert not hanging_up.done()

        release_update.set()
        with pytest.raises(asyncio.CancelledError):
            await hanging_up

        manager._client.calls.assert_called_once_with("CA-PENDING")
        call_resource.update.assert_called_once_with(status="completed")
        assert manager._pending_cleanup_call_sids == set()

    @pytest.mark.asyncio
    async def test_terminal_callback_during_failed_reconciliation_is_authoritative(
        self,
    ) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        # Subscribe the manager's status handlers so the provider callback can
        # race the stale-placement reconciliation below.
        manager._started = False
        manager.start()
        create_started = threading.Event()
        release_create = threading.Event()
        cleanup_started = threading.Event()
        release_cleanup = threading.Event()
        create_count = 0

        def create_call(**_kwargs: object) -> SimpleNamespace:
            nonlocal create_count
            create_count += 1
            if create_count == 1:
                create_started.set()
                assert release_create.wait(timeout=2)
                return SimpleNamespace(sid="CA-TERMINAL-RACE")
            return SimpleNamespace(sid="CA-NEXT")

        def update_call(*, status: str) -> None:
            assert status == "completed"
            cleanup_started.set()
            assert release_cleanup.wait(timeout=2)
            raise RuntimeError("redundant REST cleanup failed")

        manager._client.calls.create.side_effect = create_call
        call_resource = MagicMock()
        call_resource.update.side_effect = update_call
        manager._client.calls.return_value = call_resource
        placement = asyncio.create_task(manager.place_call("+15551111111"))
        assert await asyncio.to_thread(create_started.wait, 1)

        manager.stop()
        manager.start()
        release_create.set()
        assert await asyncio.to_thread(cleanup_started.wait, 1)

        await bus.emit(CallEnded(call_sid="CA-TERMINAL-RACE"))
        release_cleanup.set()
        with pytest.raises(RuntimeError, match="immediate hangup failed"):
            await placement

        assert manager._pending_cleanup_call_sids == set()
        assert manager._reconciling_call_sids == set()
        assert manager._terminal_reconciliation_call_sids == set()
        # The terminal callback releases the call despite the redundant REST
        # failure, so a later placement is not blocked by resurrected cleanup.
        assert await manager.place_call("+15552222222") == "CA-NEXT"


class TestOutboundCallManagerStatusTracking:
    def _make_manager(self, bus: EventBus) -> OutboundCallManager:
        manager = _make_bare_manager(_event_bus=bus)
        manager._client = MagicMock()
        return manager

    @pytest.mark.asyncio
    async def test_status_events_track_active_call(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager.start()
        manager._owned_call_sids.add("CA1")
        await bus.emit(CallRinging(call_sid="CA1"))
        assert manager.state == OutboundCallManagerState.ACTIVE
        assert manager.active_call_sid == "CA1"
        await bus.emit(CallAnswered(call_sid="CA1"))
        assert manager.state == OutboundCallManagerState.ACTIVE
        assert manager.active_call_sid == "CA1"
        await bus.emit(CallEnded(call_sid="CA1"))
        assert manager.state == OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
        manager._client.calls.assert_not_called()

    @pytest.mark.asyncio
    async def test_inbound_status_events_are_not_owned_or_hung_up(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager.start()

        await bus.emit(CallAnswered(call_sid="CA-INBOUND"))
        await manager.hangup_owned_call("CA-INBOUND")
        await bus.emit(CallEnded(call_sid="CA-INBOUND", disposition="max_duration"))

        manager._client.calls.assert_not_called()
        assert manager.state == OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None

    @pytest.mark.asyncio
    async def test_failed_event_clears_active_call(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager.start()
        manager._owned_call_sids.add("CA1")
        await bus.emit(CallRinging(call_sid="CA1"))
        await bus.emit(CallFailed(call_sid="CA1", reason="busy"))
        assert manager.state == OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None

    @pytest.mark.asyncio
    async def test_ignores_terminal_event_for_different_call(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager.start()
        manager._owned_call_sids.add("CA1")
        await bus.emit(CallRinging(call_sid="CA1"))
        await bus.emit(CallEnded(call_sid="CA2"))
        assert manager.state == OutboundCallManagerState.ACTIVE
        assert manager.active_call_sid == "CA1"

    @pytest.mark.asyncio
    async def test_stop_unsubscribes_from_status_events(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager.start()
        manager.stop()
        await bus.emit(CallRinging(call_sid="CA1"))
        assert manager.state == OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None

    def test_stop_does_not_call_twilio_rest(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager._active_call_sid = "CA1"
        manager._state = OutboundCallManagerState.ACTIVE
        manager.start()
        manager._active_call_sid = "CA1"
        manager._state = OutboundCallManagerState.ACTIVE
        manager.stop()
        manager._client.calls.assert_not_called()

    @pytest.mark.asyncio
    async def test_hangup_call_uses_twilio_rest_async_api(self) -> None:
        bus = EventBus()
        manager = self._make_manager(bus)
        manager._active_call_sid = "CA1"
        manager._state = OutboundCallManagerState.ACTIVE
        call_resource = MagicMock()
        manager._client.calls.return_value = call_resource
        await manager.hangup_call()
        manager._client.calls.assert_called_once_with("CA1")
        call_resource.update.assert_called_once_with(status="completed")
        assert manager.state == OutboundCallManagerState.IDLE
        assert manager.active_call_sid is None
