"""Telnyx webhook and stream handlers for local mode."""

from __future__ import annotations

import asyncio
import base64
import contextlib
import json
import logging
import re
import time
from collections import deque
from urllib.parse import quote

from starlette.websockets import WebSocketDisconnect

from getpatter.observability.attributes import patter_call_scope
from getpatter.telephony.common import STREAM_TOKEN_PARAM, _validate_e164
from getpatter.utils.log_sanitize import mask_phone_number
from getpatter.stream_handler import (
    AudioSender,
    ElevenLabsConvAIStreamHandler,
    OpenAIRealtimeStreamHandler,
    PipelineStreamHandler,
    apply_call_overrides,
    create_metrics_accumulator,
    fetch_deepgram_cost,
    resolve_agent_prompt,
)

logger = logging.getLogger("getpatter")

# DTMF digits accepted by the Telnyx ``send_dtmf`` command. Duration bounds
# mirror Telnyx API constraints (100–500 ms per digit).
#
# ``w`` / ``W`` are Telnyx-specific pause characters (each inserts a 500 ms
# wait before the next digit). They are sent as-is in the ``digits`` payload
# — Telnyx interprets them server-side. Matches the Python handler with the
# TS equivalent at ``libraries/typescript/src/server.ts::TELNYX_DTMF_ALLOWED`` (Team 8).
_DTMF_ALLOWED = frozenset("0123456789*#ABCDabcdwW")
_DTMF_DEFAULT_DURATION_MS = 250

# SIP URI validator — very permissive: ``sip:`` or ``sips:`` scheme plus a
# non-empty host component, per Telnyx transfer semantics. E.164 numbers
# are checked separately via ``_validate_e164``.
_SIP_URI_RE = re.compile(r"^sips?:[^\s@]+(@[^\s]+)?$", re.IGNORECASE)


async def handle_amd_result(
    *,
    call_control_id: str,
    result: str,
    voicemail_message: str,
    telnyx_key: str,
) -> None:
    """Drop a voicemail message when AMD classifies the call as answered by machine.

    Mirrors Twilio's ``AnsweredBy == machine_end_beep/silence`` voicemail-drop
    flow. Called from the ``/webhooks/telnyx/voice`` handler when the
    ``call.machine.detection.ended`` event fires. Uses Telnyx's
    ``actions/speak`` command to play the voicemail prompt and then hangs up.

    Args:
        call_control_id: Telnyx ``call_control_id`` from the webhook payload.
        result: The AMD classification emitted by Telnyx (``machine``,
            ``human``, ``not_sure``, ``fax``). Only ``machine`` triggers drop.
        voicemail_message: Literal text to play via ``actions/speak``.
        telnyx_key: Telnyx API key for the REST action calls.
    """
    if not call_control_id or not telnyx_key or not voicemail_message:
        return
    if result not in ("machine", "machine_detected"):
        return
    import httpx as _httpx

    encoded_id = quote(call_control_id, safe="")
    # Heuristic playback-duration estimate — ~150 ms per character, capped at
    # 30 s. Avoids cutting the voicemail mid-sentence on hangup. The proper
    # fix is to subscribe to Telnyx ``call.speak.ended`` and hang up there;
    # kept as a TODO since the webhook plumbing change is broader than this
    # handler.
    estimated_ms = min(len(voicemail_message) * 150, 30_000)
    client_state = base64.b64encode(b"voicemail-drop").decode("ascii")
    try:
        async with _httpx.AsyncClient(timeout=10.0) as _http:
            # Speak the voicemail message. ``client_state`` is echoed back on
            # ``call.speak.ended`` so a future handler can hang up there
            # instead of relying on the heuristic sleep below.
            await _http.post(
                f"https://api.telnyx.com/v2/calls/{encoded_id}/actions/speak",
                headers={"Authorization": f"Bearer {telnyx_key}"},
                json={
                    "payload": voicemail_message,
                    "voice": "female",
                    "language": "en-US",
                    "client_state": client_state,
                },
            )
            await asyncio.sleep(estimated_ms / 1000)
            await _http.post(
                f"https://api.telnyx.com/v2/calls/{encoded_id}/actions/hangup",
                headers={"Authorization": f"Bearer {telnyx_key}"},
                json={},
            )
        logger.info("Voicemail dropped for Telnyx call %s", call_control_id)
    except Exception as exc:
        logger.warning("Could not drop voicemail (Telnyx): %s", exc)


def _is_valid_transfer_target(target: str) -> bool:
    """Accept either a validated E.164 phone number or a SIP(s) URI."""
    if not isinstance(target, str) or not target:
        return False
    if _validate_e164(target):
        return True
    return bool(_SIP_URI_RE.match(target))


# Maximum size (bytes) of a single WebSocket message accepted from Telnyx.
# Telnyx 16 kHz PCM frames are ~640 bytes (20 ms).  1 MB defends against
# memory exhaustion from a malformed or malicious stream peer.
_MAX_WS_MESSAGE_BYTES = 1 * 1024 * 1024


def telnyx_webhook_handler(
    call_id: str,
    caller: str,
    callee: str,
    webhook_base_url: str,
    connection_id: str = "",
    stream_token: str = "",
) -> dict:
    """Generate Telnyx Call Control response for an incoming call.

    Returns a dict that should be serialised to JSON and returned with 200 OK.
    Telnyx Call Control uses a command-based model: the webhook handler responds
    with ``answer`` and then ``stream_start`` commands.

    Args:
        call_id: Telnyx ``call_control_id``.
        caller: The calling number.
        callee: The called number.
        webhook_base_url: Hostname (no scheme) of this server, e.g. "abc.ngrok.io".
        connection_id: Telnyx TeXML App / Call Control App ID (optional).
        stream_token: Per-call media-stream auth token (#204). Telnyx preserves
            the query string, so it rides as ``&patter_stream_token=...``; the
            WS handler validates it at accept before any provider session.
    """
    stream_url = (
        f"wss://{webhook_base_url}/ws/telnyx/stream/{call_id}"
        f"?caller={quote(caller)}&callee={quote(callee)}"
    )
    if stream_token:
        stream_url += f"&{STREAM_TOKEN_PARAM}={quote(stream_token, safe='')}"
    # Telnyx Call Control: answer first, then stream_start.
    # ``inbound_track`` halves WS upstream bandwidth — the bridge already
    # filters outbound media downstream, so requesting only inbound at the
    # source removes redundant frames.
    return {
        "commands": [
            {"command": "answer"},
            {
                "command": "stream_start",
                "params": {
                    "stream_url": stream_url,
                    "stream_track": "inbound_track",
                },
            },
        ]
    }


# ---------------------------------------------------------------------------
# Telnyx AudioSender — no transcoding needed (16 kHz PCM native)
# ---------------------------------------------------------------------------


class TelnyxAudioSender(AudioSender):
    """Sends audio to a Telnyx media-stream WebSocket.

    Telnyx expects outbound frames in the same codec negotiated at
    ``streaming_start`` time. The server currently negotiates
    PCMU 8 kHz bidirectional, so OpenAI Realtime is configured to emit
    ``g711_ulaw`` directly — the sender forwards the bytes as-is. For
    pipeline mode, ``input_is_mulaw_8k=False`` keeps the PCM16 16 kHz path
    (Telnyx transcodes on the RTP leg when negotiated as L16/16000).

    Wire format: ``{"event": "media", "media": {"payload": b64}}``.
    """

    def __init__(self, websocket, input_is_mulaw_8k: bool = False) -> None:
        self._ws = websocket
        self._input_is_mulaw_8k = input_is_mulaw_8k
        # Lazy import transcoding when the caller sends PCM16 16 kHz and
        # we need to match the negotiated PCMU 8 kHz bidirectional stream.
        # Uses a stateful resampler (StatefulResampler) so that audioop.ratecv
        # filter state is preserved across chunks — avoids the click/pop
        # artefact that occurs when the per-chunk stateless path restarts the
        # IIR filter every 20 ms frame.
        if not input_is_mulaw_8k:
            from getpatter.audio.transcoding import (  # type: ignore[import]
                create_resampler_16k_to_8k,
                pcm16_to_mulaw,
            )

            self._pcm16_to_mulaw = pcm16_to_mulaw
            self._resampler = create_resampler_16k_to_8k()
        else:
            self._pcm16_to_mulaw = None
            self._resampler = None

    def set_source_format(self, encoding: str, sample_rate: int) -> None:
        """Rebuild the transcode chain for the pipeline TTS's DECLARED format.

        ``mulaw`` @ 8 kHz → carrier wire codec → passthrough. Any other
        (PCM) rate → resample from that REAL rate to 8 kHz then mu-law encode.
        Replaces the previous hardcoded 16 kHz -> 8 kHz assumption that
        chipmunked an 8 kHz source. Parity with TS ``configureOutboundAudio``.
        """
        if encoding in ("mulaw", "ulaw_8000", "alaw"):
            self._input_is_mulaw_8k = True
            self._resampler = None
            self._pcm16_to_mulaw = None
            return
        from getpatter.audio.transcoding import (
            StatefulResampler,
            pcm16_to_mulaw,
        )

        self._input_is_mulaw_8k = False
        self._pcm16_to_mulaw = pcm16_to_mulaw
        self._resampler = StatefulResampler(src_rate=int(sample_rate), dst_rate=8000)

    async def send_audio(self, audio: bytes) -> None:
        """Send a PCM (or mulaw) audio chunk to the Telnyx media stream."""
        if self._input_is_mulaw_8k:
            mulaw = audio
        else:
            resampled = self._resampler.process(audio)  # type: ignore[union-attr]
            mulaw = self._pcm16_to_mulaw(resampled)
        encoded = base64.b64encode(mulaw).decode("ascii")
        await self._ws.send_text(
            json.dumps({"event": "media", "media": {"payload": encoded}})
        )

    async def flush(self) -> None:
        """Send any resampler tail bytes before closing the stream.

        StatefulResampler.flush() drains the internal carry buffer.
        Call this on the hangup / stop path to avoid clipping the last
        audio frame. No-op when input_is_mulaw_8k=True.
        """
        if self._resampler is None or self._pcm16_to_mulaw is None:
            return
        tail = self._resampler.flush()
        if tail:
            mulaw = self._pcm16_to_mulaw(tail)
            encoded = base64.b64encode(mulaw).decode("ascii")
            await self._ws.send_text(
                json.dumps({"event": "media", "media": {"payload": encoded}})
            )

    async def send_clear(self) -> None:
        """Tell Telnyx to flush any buffered outbound playback (barge-in)."""
        # Telnyx media stream clear signal — flushes any buffered playback.
        await self._ws.send_text(json.dumps({"event": "clear"}))

    async def send_mark(self, mark_name: str) -> None:
        """No-op: Telnyx media streams do not support playback marks."""
        # Telnyx media streams do not support playback marks — no-op.
        pass


async def telnyx_stream_bridge(
    websocket,
    agent,
    openai_key: str,
    on_call_start=None,
    on_call_end=None,
    on_transcript=None,
    on_message=None,
    deepgram_key: str = "",
    elevenlabs_key: str = "",
    telnyx_key: str = "",
    recording: bool = False,
    local_recorder_factory=None,
    on_metrics=None,
    on_transcript_line=None,
    pricing: dict | None = None,
    report_only_initial_ttfb: bool = False,
    patter_side: str = "uut",
    pop_prewarm_audio=None,
    pop_prewarmed_connections=None,
    speech_events=None,
) -> None:
    """Bridge a Telnyx WebSocket media stream to the configured AI provider.

    Supports two provider modes depending on ``agent.provider``:

    * ``"openai_realtime"`` (default) — streams 16 kHz PCM directly to the
      OpenAI Realtime API (no transcoding needed on Telnyx).
    * ``"pipeline"`` — uses Deepgram for STT (16 kHz PCM), calls ``on_message``
      with the transcript, then synthesises the response with ElevenLabs TTS
      and sends it back to Telnyx.

    Args:
        websocket: A Starlette/FastAPI WebSocket instance.
        agent: An ``Agent`` dataclass with prompt, voice, model, tools, etc.
        openai_key: OpenAI API key for the Realtime API (openai_realtime mode).
        on_call_start: Optional async callable(dict) — fired when streaming starts.
        on_call_end: Optional async callable(dict) — fired when streaming ends.
        on_transcript: Optional async callable(dict) — fired for each user utterance.
        on_message: Optional async callable(dict) -> str — called with the user's
            text in pipeline mode; return value is synthesised and played back.
        deepgram_key: Deepgram API key (pipeline mode).
        elevenlabs_key: ElevenLabs API key (pipeline mode).
    """
    await websocket.accept()

    caller: str = websocket.query_params.get("caller", "")
    callee: str = websocket.query_params.get("callee", "")

    call_id_actual: str = ""
    transcript_entries: deque[dict] = deque(maxlen=200)
    stream_started = False

    handler: (
        OpenAIRealtimeStreamHandler
        | ElevenLabsConvAIStreamHandler
        | PipelineStreamHandler
        | None
    ) = None
    audio_sender: TelnyxAudioSender | None = None
    metrics = None
    # Carrier-neutral local recorder for this call (None = off). Tracked
    # bridge-side (not just on the handler) so the on_call_end payload can
    # surface ``recording_path`` without poking handler internals.
    local_recorder = None

    # Wall-clock duration tracking for patter.cost.telephony_minutes. Set on
    # the ``start`` event so we measure only the bridged audio period, not
    # the time spent waiting for the first frame.
    _call_start_monotonic: float | None = None

    # ExitStack lets us enter ``patter_call_scope`` *after* the start frame
    # arrives (when call_id is known) while still keeping the scope active
    # for the entire WebSocket loop AND the finally cleanup block.
    _scope_stack = contextlib.ExitStack()

    try:
        while True:
            raw = await websocket.receive_text()
            if len(raw) > _MAX_WS_MESSAGE_BYTES:
                logger.warning(
                    "Oversized Telnyx WebSocket message dropped (%d bytes)", len(raw)
                )
                continue
            data = json.loads(raw)
            # Telnyx media-stream WebSocket uses ``event`` (not
            # ``event_type``, which is a Call Control REST notification
            # field).
            event_type_telnyx = data.get("event", "")

            if event_type_telnyx == "connected":
                # First frame after WS open — Telnyx ping. Nothing to do.
                continue

            if event_type_telnyx == "start" and not stream_started:
                stream_started = True
                _call_start_monotonic = time.monotonic()
                start_info = data.get("start", {}) or {}
                call_id_actual = start_info.get("call_control_id", "")
                caller = start_info.get("from", "") or caller
                callee = start_info.get("to", "") or callee

                # Single INFO line per call-start — full context in one place.
                _mode = (
                    f"engine={getattr(agent.engine, 'kind', 'unknown')}"
                    if getattr(agent, "engine", None) is not None
                    else "pipeline"
                    if (
                        getattr(agent, "stt", None) is not None
                        and getattr(agent, "tts", None) is not None
                    )
                    else f"engine={getattr(agent, 'provider', 'unknown')}"
                )
                logger.info(
                    "Call started: %s (Telnyx, %s, %s → %s)",
                    call_id_actual,
                    _mode,
                    mask_phone_number(caller) or "?",
                    mask_phone_number(callee) or "?",
                )

                # Fire on_call_start callback — may return per-call config overrides
                _call_overrides = None
                if on_call_start:
                    _call_overrides = await on_call_start(
                        {
                            "call_id": call_id_actual,
                            "caller": caller,
                            "callee": callee,
                            "direction": "inbound",
                            "telephony_provider": "telnyx",
                        }
                    )
                    if not isinstance(_call_overrides, dict):
                        _call_overrides = None

                # Apply per-call overrides (dynamic agent config)
                if _call_overrides:
                    agent = apply_call_overrides(agent, _call_overrides)

                # Resolve dynamic variables in system prompt
                resolved_prompt = resolve_agent_prompt(agent)
                provider = getattr(agent, "provider", "openai_realtime")

                # Initialize metrics
                metrics = create_metrics_accumulator(
                    call_id=call_id_actual,
                    provider=provider,
                    telephony_provider="telnyx",
                    agent=agent,
                    deepgram_key=deepgram_key,
                    elevenlabs_key=elevenlabs_key,
                    pricing=pricing,
                    report_only_initial_ttfb=report_only_initial_ttfb,
                )
                # Telnyx uses PCM 16kHz (2 bytes/sample)
                metrics.configure_stt_format(sample_rate=16000, bytes_per_sample=2)

                # Create audio sender. OpenAI Realtime negotiates g711_ulaw
                # 8 kHz to match the `streaming_start` PCMU bidirectional
                # stream — forward bytes as-is. Pipeline and ConvAI still
                # produce PCM16 that Telnyx accepts when L16 is negotiated.
                _input_is_mulaw = getattr(agent, "provider", "openai_realtime") in (
                    "openai_realtime",
                    "openai_realtime_2",
                )
                audio_sender = TelnyxAudioSender(
                    websocket, input_is_mulaw_8k=_input_is_mulaw
                )

                # --- Telnyx-specific call control helpers ---
                async def _telnyx_transfer(
                    number,
                    *,
                    mode: str = "cold",
                    summary: str = "",
                    client_state: str | None = None,
                ):
                    """Blind-transfer the call via the Telnyx Call Control API.

                    Accepts either an E.164 phone number or a SIP URI
                    (``sip:user@host`` / ``sips:user@host``).

                    ``mode="warm"`` is NOT yet implemented on Telnyx — the
                    Call Control conference flow requires a second outbound
                    leg (connection_id + answer-webhook coordination) that
                    the bridge does not plumb today. A clear error envelope
                    is returned so the agent keeps the call instead of
                    silently degrading to a blind redirect. ``summary`` is
                    accepted for cross-carrier signature parity and unused
                    in cold mode.

                    ``client_state`` (optional) is a caller-supplied string
                    that Telnyx will echo on every subsequent webhook for
                    this call leg. Base64-encoded per Telnyx contract.
                    """
                    if mode == "warm":
                        logger.warning(
                            "warm transfer requested but not yet supported on telnyx"
                        )
                        return {"error": "warm transfer not yet supported on telnyx"}
                    if not _is_valid_transfer_target(number):
                        logger.warning(
                            "Telnyx transfer rejected: invalid target %s",
                            mask_phone_number(number),
                        )
                        return
                    if telnyx_key and call_id_actual:
                        import httpx as _httpx

                        body: dict = {"to": number}
                        if client_state:
                            import base64 as _b64

                            body["client_state"] = _b64.b64encode(
                                client_state.encode("utf-8")
                            ).decode("ascii")
                        async with _httpx.AsyncClient() as _http:
                            await _http.post(
                                f"https://api.telnyx.com/v2/calls/{quote(call_id_actual, safe='')}/actions/transfer",
                                headers={"Authorization": f"Bearer {telnyx_key}"},
                                json=body,
                                timeout=10.0,
                            )
                        logger.debug(
                            "Telnyx call transferred to %s", mask_phone_number(number)
                        )

                async def _telnyx_hangup():
                    if telnyx_key and call_id_actual:
                        import httpx as _httpx

                        async with _httpx.AsyncClient() as _http:
                            await _http.post(
                                f"https://api.telnyx.com/v2/calls/{quote(call_id_actual, safe='')}/actions/hangup",
                                headers={"Authorization": f"Bearer {telnyx_key}"},
                                json={},
                                timeout=10.0,
                            )
                        logger.debug("Telnyx call hung up")

                async def _telnyx_send_dtmf(digits: str, delay_ms: int = 300) -> None:
                    """Send DTMF digits via the Telnyx Call Control API.

                    Emits one ``send_dtmf`` command per digit, sleeping
                    ``delay_ms`` milliseconds between digits. Telnyx's
                    ``duration_millis`` is clamped to 100–500 ms per digit.
                    """
                    if not digits:
                        logger.warning("Telnyx send_dtmf called with empty digits")
                        return
                    if not (telnyx_key and call_id_actual):
                        logger.warning(
                            "Telnyx send_dtmf skipped: telnyx_key or call_id missing"
                        )
                        return

                    filtered = [d for d in digits if d in _DTMF_ALLOWED]
                    if not filtered:
                        logger.warning(
                            "Telnyx send_dtmf: no valid digits in %r", digits
                        )
                        return

                    duration = max(100, min(500, _DTMF_DEFAULT_DURATION_MS))
                    import httpx as _httpx

                    async with _httpx.AsyncClient() as _http:
                        for idx, digit in enumerate(filtered):
                            await _http.post(
                                f"https://api.telnyx.com/v2/calls/{quote(call_id_actual, safe='')}/actions/send_dtmf",
                                headers={"Authorization": f"Bearer {telnyx_key}"},
                                json={
                                    "digits": digit,
                                    "duration_millis": duration,
                                },
                                timeout=10.0,
                            )
                            if idx < len(filtered) - 1 and delay_ms > 0:
                                await asyncio.sleep(delay_ms / 1000)
                    logger.debug(
                        "Telnyx DTMF sent (%d digits, delay=%dms)",
                        len(filtered),
                        delay_ms,
                    )

                async def _telnyx_start_recording() -> None:
                    """Start recording the call via Telnyx Call Control API."""
                    if not (telnyx_key and call_id_actual):
                        return
                    import httpx as _httpx

                    async with _httpx.AsyncClient() as _http:
                        resp = await _http.post(
                            f"https://api.telnyx.com/v2/calls/{quote(call_id_actual, safe='')}/actions/record_start",
                            headers={"Authorization": f"Bearer {telnyx_key}"},
                            json={"format": "mp3", "channels": "single"},
                            timeout=10.0,
                        )
                        if resp.status_code >= 400:
                            logger.warning(
                                "Telnyx record_start failed (%d): %s",
                                resp.status_code,
                                resp.text[:200],
                            )
                        else:
                            logger.debug("Telnyx recording started")

                async def _telnyx_stop_recording() -> None:
                    """Stop recording the call via Telnyx Call Control API."""
                    if not (telnyx_key and call_id_actual):
                        return
                    import httpx as _httpx

                    async with _httpx.AsyncClient() as _http:
                        resp = await _http.post(
                            f"https://api.telnyx.com/v2/calls/{quote(call_id_actual, safe='')}/actions/record_stop",
                            headers={"Authorization": f"Bearer {telnyx_key}"},
                            json={},
                            timeout=10.0,
                        )
                        if resp.status_code >= 400:
                            logger.warning(
                                "Telnyx record_stop failed (%d): %s",
                                resp.status_code,
                                resp.text[:200],
                            )
                        else:
                            logger.debug("Telnyx recording stopped")

                # Kick off recording if requested
                if recording:
                    try:
                        await _telnyx_start_recording()
                    except Exception as _exc:
                        logger.warning("Could not start recording: %s", _exc)

                # Create the appropriate stream handler
                if provider == "pipeline":
                    handler = PipelineStreamHandler(
                        agent=agent,
                        audio_sender=audio_sender,
                        call_id=call_id_actual,
                        caller=caller,
                        callee=callee,
                        resolved_prompt=resolved_prompt,
                        metrics=metrics,
                        openai_key=openai_key,
                        deepgram_key=deepgram_key,
                        elevenlabs_key=elevenlabs_key,
                        for_twilio=False,
                        # Telnyx bidirectional PCMU 8 kHz: inbound caller
                        # audio arrives as mulaw 8 kHz and must be
                        # transcoded to PCM16 16 kHz before STT. Outbound
                        # PCM16 16 kHz from the TTS must be transcoded
                        # back to mulaw 8 kHz in the audio_sender.
                        input_is_mulaw_8k=True,
                        output_is_mulaw_8k=False,  # TTS produces PCM16; sender transcodes
                        transfer_fn=_telnyx_transfer,
                        hangup_fn=_telnyx_hangup,
                        send_dtmf_fn=_telnyx_send_dtmf,
                        on_transcript=on_transcript,
                        on_message=on_message,
                        on_metrics=on_metrics,
                        transcript_entries=transcript_entries,
                        pop_prewarm_audio=pop_prewarm_audio,
                        pop_prewarmed_connections=pop_prewarmed_connections,
                        speech_events=speech_events,
                    )
                elif provider == "elevenlabs_convai":
                    handler = ElevenLabsConvAIStreamHandler(
                        agent=agent,
                        audio_sender=audio_sender,
                        call_id=call_id_actual,
                        caller=caller,
                        callee=callee,
                        resolved_prompt=resolved_prompt,
                        metrics=metrics,
                        elevenlabs_key=elevenlabs_key,
                        # Telnyx negotiates PCMU @ 8 kHz on streaming_start
                        # (see the webhook handler) — same wire format as
                        # Twilio. Without this flag the handler fed the
                        # caller's mulaw bytes to ConvAI as if they were
                        # PCM16 @ 16 kHz: garbled audio in both directions.
                        for_twilio=True,
                        on_transcript=on_transcript,
                        on_metrics=on_metrics,
                        on_transcript_line=on_transcript_line,
                        transcript_entries=transcript_entries,
                        speech_events=speech_events,
                    )
                else:
                    handler = OpenAIRealtimeStreamHandler(
                        agent=agent,
                        audio_sender=audio_sender,
                        call_id=call_id_actual,
                        caller=caller,
                        callee=callee,
                        resolved_prompt=resolved_prompt,
                        metrics=metrics,
                        openai_key=openai_key,
                        transfer_fn=_telnyx_transfer,
                        hangup_fn=_telnyx_hangup,
                        on_transcript=on_transcript,
                        on_metrics=on_metrics,
                        on_transcript_line=on_transcript_line,
                        transcript_entries=transcript_entries,
                        # Telnyx Call Control media streams deliver PCMU
                        # 8 kHz (g711 mulaw) in both directions when
                        # streaming_start negotiates PCMU bidirectional.
                        # Verified 2026-04-21: inbound chunks are 160 B /
                        # 20 ms → PCMU 8 kHz. OpenAI Realtime with this
                        # codec forwards bytes pass-through on both legs.
                        audio_format="g711_ulaw",
                        pop_prewarmed_connections=pop_prewarmed_connections,
                        speech_events=speech_events,
                    )

                # Inherit patter.side from the parent Patter instance so all
                # spans emitted during the call lifetime carry the right side.
                try:
                    handler._patter_side = patter_side
                except Exception:  # pragma: no cover — defense in depth
                    logger.debug("Failed to set handler._patter_side", exc_info=True)

                # Attach the carrier-neutral local recorder BEFORE
                # handler.start() so the firstMessage TTS is captured. The
                # factory returns None when local recording is off / failed.
                if local_recorder_factory is not None:
                    try:
                        local_recorder = local_recorder_factory(call_id_actual)
                        handler.local_recorder = local_recorder
                    except Exception as _exc:  # noqa: BLE001 - best-effort
                        logger.warning("Local recorder setup failed: %s", _exc)

                # Enter patter_call_scope NOW that call_id is known. The
                # ExitStack keeps the scope active until the finally cleanup
                # block runs.
                try:
                    if call_id_actual:
                        _scope_stack.enter_context(
                            patter_call_scope(call_id=call_id_actual, side=patter_side)
                        )
                except Exception:  # pragma: no cover — defense in depth
                    logger.debug("patter_call_scope entry failed", exc_info=True)

                await handler.start()

            elif event_type_telnyx == "media":
                media = data.get("media", {}) or {}
                # Telnyx with ``stream_track=both_tracks`` emits media for
                # both the caller leg (``track=inbound``) and the leg we
                # are injecting audio into (``track=outbound``). Forwarding
                # the ``outbound`` echo to OpenAI Realtime would feed the
                # agent its own voice and break turn detection — only the
                # inbound track is actual caller audio.
                track = media.get("track", "inbound")
                if track != "inbound":
                    continue
                audio_chunk_b64 = media.get("payload", "")
                if not audio_chunk_b64:
                    continue

                pcm_audio = base64.b64decode(audio_chunk_b64)
                if handler is not None:
                    await handler.on_audio_received(pcm_audio)

            elif event_type_telnyx == "dtmf":
                dtmf_info = data.get("dtmf", {}) or {}
                digit = str(dtmf_info.get("digit", "")).strip()
                if digit:
                    logger.debug("Telnyx DTMF received: %s", digit)
                    if handler is not None:
                        try:
                            on_dtmf = getattr(handler, "on_dtmf", None)
                            if callable(on_dtmf):
                                await on_dtmf(digit)
                        except Exception as _exc:
                            logger.debug("on_dtmf handler error: %s", _exc)
                    if on_transcript:
                        try:
                            await on_transcript(
                                {
                                    "role": "user",
                                    "text": f"[DTMF: {digit}]",
                                    "call_id": call_id_actual,
                                }
                            )
                        except Exception as _exc:
                            logger.debug("on_transcript DTMF dispatch error: %s", _exc)

            elif event_type_telnyx == "error":
                logger.warning("Telnyx stream error: %s", data.get("payload") or data)

            elif event_type_telnyx == "stop":
                break

    except WebSocketDisconnect:
        # Carrier-side teardown without a ``stop`` frame is a normal-ish
        # hangup, not a failure — don't pollute error telemetry with it.
        logger.info("Carrier WebSocket disconnected without stop frame")
    except Exception as exc:
        logger.exception("Stream error: %s", exc)
        # Record the terminal error code on the metrics so call telemetry and the
        # dashboard can attribute the failure (code only, never the message).
        if metrics is not None:
            try:
                metrics.record_error(exc)
            except Exception:
                pass
    finally:
        # Best-effort recording stop — only if recording was requested and
        # the call is still active. Telnyx auto-stops on hangup so errors
        # are non-fatal.
        if recording and telnyx_key and call_id_actual:
            try:
                import httpx as _httpx

                async with _httpx.AsyncClient() as _http:
                    await _http.post(
                        f"https://api.telnyx.com/v2/calls/{quote(call_id_actual, safe='')}/actions/record_stop",
                        headers={"Authorization": f"Bearer {telnyx_key}"},
                        json={},
                        timeout=5.0,
                    )
            except Exception as _exc:
                logger.debug("Telnyx record_stop best-effort failed: %s", _exc)

        # Flush resampler tail before tearing down — drains any carry bytes so
        # the last audio frame isn't clipped on graceful shutdown.
        if audio_sender is not None:
            try:
                await audio_sender.flush()
            except Exception as _exc:
                logger.debug("Telnyx audio_sender flush failed: %s", _exc)

        if handler is not None:
            try:
                await handler.cleanup()
            except Exception as _exc:  # noqa: BLE001 - teardown must complete
                # cleanup() awaits adapter/STT/TTS closes; one raise here used
                # to skip the rest of the finally (no metrics finalize, no
                # on_call_end, dashboard row stuck active forever).
                logger.exception("handler.cleanup failed: %s", _exc)

        # --- Observability: emit patter.cost.telephony_minutes ---
        # Wired here so the span inherits patter.call_id / patter.side
        # from the active patter_call_scope. Bridge is the inbound
        # webhook endpoint, so direction is always "inbound" today.
        if _call_start_monotonic is not None and telnyx_key:
            try:
                from getpatter.providers.telnyx_adapter import TelnyxAdapter

                _duration = time.monotonic() - _call_start_monotonic
                TelnyxAdapter(api_key=telnyx_key).record_call_end_cost(
                    duration_seconds=_duration, direction="inbound"
                )
            except Exception as exc:
                logger.debug("record_call_end_cost failed: %s", exc)

        # --- Metrics: query actual telephony cost from Telnyx ---
        if metrics is not None and telnyx_key and call_id_actual:
            try:
                import httpx as _httpx

                async with _httpx.AsyncClient() as _http:
                    resp = await _http.get(
                        f"https://api.telnyx.com/v2/calls/{quote(call_id_actual, safe='')}",
                        headers={"Authorization": f"Bearer {telnyx_key}"},
                        timeout=5.0,
                    )
                    if resp.status_code == 200:
                        call_data = resp.json().get("data", {})
                        cost = call_data.get("cost", {})
                        total_cost = cost.get("amount")
                        if total_cost is not None:
                            metrics.set_actual_telephony_cost(abs(float(total_cost)))
                            logger.debug(
                                "Telnyx actual cost: $%s", abs(float(total_cost))
                            )
            except Exception as exc:
                logger.debug("Could not fetch Telnyx call cost: %s", exc)

        # --- Metrics: query actual STT cost from Deepgram ---
        stt = getattr(handler, "stt", None) if handler is not None else None
        await fetch_deepgram_cost(metrics, stt, deepgram_key)

        # --- Metrics: finalize ---
        call_metrics = None
        if metrics is not None:
            try:
                call_metrics = metrics.end_call()
            except Exception as exc:
                logger.warning("Metrics finalization error: %s", exc)
        if on_call_end:
            try:
                _end_payload = {
                    "call_id": call_id_actual,
                    "caller": caller,
                    "callee": callee,
                    "ended_at": time.time(),
                    "transcript": list(transcript_entries),
                    # The Telnyx bridge doesn't keep its own history deque
                    # (Twilio/Plivo do) — read the handler's so the
                    # on_call_end payload shape matches across carriers.
                    "conversation_history": list(
                        getattr(handler, "conversation_history", None) or []
                    )
                    if handler is not None
                    else [],
                    "metrics": call_metrics,
                }
                # Surface the local recording path when active. The handler's
                # cleanup() above finalized the WAV; ``close()`` is idempotent
                # and returns the path (or None when the recorder broke).
                if local_recorder is not None:
                    _end_payload["recording_path"] = local_recorder.close()
                await on_call_end(_end_payload)
            except Exception as exc:
                logger.exception("on_call_end error: %s", exc)

        # Single INFO line per call-end — duration, turns, cost, latency.
        # "p95 wait" = agent_response_ms (user-perceived wait after they stop
        # speaking). Matches the dashboard "p95 wait" tile. Fallback to
        # total_ms for legacy / short calls where agent_response_ms is unset.
        if call_metrics is not None:
            _dur = getattr(call_metrics, "duration_seconds", 0) or 0
            _turns = len(getattr(call_metrics, "turns", []) or [])
            _cost = getattr(getattr(call_metrics, "cost", None), "total", 0) or 0
            _p95_obj = getattr(call_metrics, "latency_p95", None)
            _p95 = (
                getattr(_p95_obj, "agent_response_ms", None)
                or getattr(_p95_obj, "total_ms", 0)
                or 0
            )
            logger.info(
                "Call ended: %s (%.1fs, %d turns, cost=$%.4f, p95 wait=%dms)",
                call_id_actual,
                _dur,
                _turns,
                _cost,
                round(_p95),
            )
        else:
            logger.info("Call ended: %s", call_id_actual)

        # Close the patter_call_scope (if entered) — done last so all
        # cleanup-emitted spans inherit patter.call_id / patter.side.
        try:
            _scope_stack.close()
        except Exception:  # pragma: no cover — defense in depth
            logger.debug("ExitStack close failed", exc_info=True)
