"""Call metrics accumulator — tracks cost and latency during a call."""

from __future__ import annotations

__all__ = ["CallMetricsAccumulator"]

import time
from typing import TYPE_CHECKING

from getpatter.models import (
    CallMetrics,
    CostBreakdown,
    LatencyBreakdown,
    TurnMetrics,
)
from getpatter.pricing import (
    calculate_llm_cost,
    calculate_realtime_cached_savings,
    calculate_realtime_cost,
    calculate_stt_cost,
    calculate_telephony_cost,
    calculate_tts_cost,
    merge_pricing,
)

if TYPE_CHECKING:
    from getpatter.observability.event_bus import EventBus


class CallMetricsAccumulator:
    """Mutable accumulator for per-call cost and latency metrics.

    Created at call start, collects data during the call, and produces
    a frozen ``CallMetrics`` via ``end_call()``.

    An optional :class:`~getpatter.observability.event_bus.EventBus` can be
    attached via :meth:`attach_event_bus`; it receives typed metric events in
    addition to the existing callback-based ``on_metrics`` path.
    """

    def __init__(
        self,
        call_id: str,
        provider_mode: str,
        telephony_provider: str,
        stt_provider: str = "",
        tts_provider: str = "",
        llm_provider: str = "",
        pricing: dict | None = None,
        report_only_initial_ttfb: bool = False,
        *,
        stt_model: str = "",
        tts_model: str = "",
        realtime_model: str = "",
    ) -> None:
        self.call_id = call_id
        self.provider_mode = provider_mode
        self.telephony_provider = telephony_provider
        self.stt_provider = stt_provider
        self.tts_provider = tts_provider
        self.llm_provider = llm_provider
        # Model identifiers for per-model rate resolution (see pricing.py).
        # Empty string means "not known" → cost calc falls back to provider
        # defaults, matching pre-2026.3 behaviour.
        self.stt_model = stt_model
        self.tts_model = tts_model
        self.realtime_model = realtime_model
        self._pricing = merge_pricing(pricing)
        self._report_only_initial_ttfb = report_only_initial_ttfb

        self._call_start = time.monotonic()
        self._turns: list[TurnMetrics] = []

        # Monotonic per-call turn counter. ``reserve_turn_index`` hands out the
        # next value when a turn OPENS (on the ``speech_stopped`` edge) so the
        # stream handler can stamp the reserved index onto each transcript line
        # the moment it is known — even before the turn completes. This keeps
        # the dashboard transcript ordering stable (user line < assistant line)
        # under drops / interrupts, where assigning ``turn_index`` only at
        # completion (``len(self._turns)``) would be unstable. Parity with TS
        # ``metrics.ts:_nextTurnIndex`` / ``reserveTurnIndex``.
        self._next_turn_index = 0

        # --- Per-turn timing state ---
        self._turn_start: float | None = None
        self._stt_complete: float | None = None
        self._llm_first_token: float | None = None  # Fix 5: LLM TTFT start
        self._llm_first_sentence: float | None = None  # Fix 3: first sentence boundary
        self._llm_complete: float | None = None
        self._tts_first_byte: float | None = None
        # Last TTS audio byte sent (for tts_total_ms). Captured by
        # ``record_tts_complete`` which now also stamps the timestamp.
        self._tts_last_byte: float | None = None
        # Endpoint signal: silence detected (VAD stop OR STT speech_final).
        # Recorded by ``record_vad_stop`` / ``record_stt_final_timestamp`` —
        # whichever fires first becomes the endpoint reference.
        self._endpoint_signal_at: float | None = None
        # Barge-in detection / completion timestamps (for bargein_ms).
        self._bargein_detected_at: float | None = None
        self._bargein_stopped_at: float | None = None
        self._turn_user_text: str = ""
        self._turn_stt_audio_seconds: float = 0.0
        # Guard against the record_turn_interrupted / record_turn_complete
        # race. A VAD-path barge-in fires ``record_turn_interrupted``
        # synchronously inside the handler while the in-flight pipeline
        # LLM stream keeps unwinding on its own task. When the LLM stream
        # eventually exits, the pipeline path falls through to
        # ``record_turn_complete``, which would push a second turn for
        # the same logical exchange (this time carrying ``user_text=""``
        # because the field was already reset). The flag is flipped by
        # ``record_turn_interrupted`` and read by ``record_turn_complete``
        # so the late ``record_turn_complete`` becomes a no-op until the
        # next ``start_turn`` re-arms the accumulator.
        self._turn_already_closed: bool = False
        # Cross-turn TTFT storage so _emit_turn_metrics can read it after reset
        self._last_turn_llm_ttft_ms: float = 0.0

        # --- Cumulative usage counters ---
        self._total_stt_audio_seconds: float = 0.0
        self._total_tts_characters: int = 0
        self._total_realtime_cost: float = 0.0
        self._total_realtime_cached_savings: float = 0.0
        # Byte counters for computing audio seconds from raw audio
        self._stt_byte_count: int = 0
        self._stt_sample_rate: int = 16000
        self._stt_bytes_per_sample: int = 2  # PCM16 = 2 bytes/sample
        # Actual provider costs (from post-call API queries)
        self._actual_telephony_cost: float | None = None
        self._actual_stt_cost: float | None = None
        # LLM token usage accumulated across all turns (pipeline mode)
        self._llm_total_input_tokens: int = 0
        self._llm_total_output_tokens: int = 0
        self._llm_total_cache_read_tokens: int = 0
        self._llm_total_cache_write_tokens: int = 0
        self._llm_provider_name: str = llm_provider
        self._llm_model: str = ""

        # Terminal error code (lowercased ErrorCode value or "other"); set by
        # ``record_error`` when the call ends abnormally. Empty for a clean call.
        self._error_code: str = ""

        # --- EventBus integration (additive; does not replace callbacks) ---
        self._event_bus: EventBus | None = None

        # --- report_only_initial_ttfb guard flags ---
        self._llm_ttfb_emitted: bool = False
        self._tts_ttfb_emitted: bool = False

        # --- EOUMetrics timestamps ---
        self._vad_stopped_at: float | None = None
        self._stt_final_at: float | None = None
        self._turn_committed_at: float | None = None
        # Monotonic stamp of LLM dispatch — paired with
        # ``_endpoint_signal_at`` to compute ``endpoint_ms``.
        self._turn_committed_mono: float | None = None
        self._on_user_turn_completed_delay_ms: float | None = None

        # --- InterruptionMetrics counters ---
        self._num_interruptions: int = 0
        self._num_backchannels: int = 0
        self._overlap_started_at: float | None = None

        # --- Barge-in anchor hygiene ---
        # Monotonic timestamp of the most recent barge-in detection. Used by
        # ``_compute_turn_latency`` to gate ``endpoint_ms`` / ``stt_ms``
        # emission on turns that started within 100 ms of the last barge-in
        # — those turns have unreliable VAD/STT anchors and would otherwise
        # pollute the p95 distribution with synthetic 6+ second spikes.
        self._last_bargein_at: float | None = None
        # Counter of turns where ``record_stt_complete`` fired but no VAD
        # ``speech_end`` had stamped ``_endpoint_signal_at`` first. Lets us
        # detect environments where PSTN packet loss is dropping VAD stops
        # (the common cause of missing endpoint signals).
        self._endpoint_signal_missing_count: int = 0

        # Per-turn LLM cost accumulated from record_llm_usage() calls.
        # Populated eagerly each time record_llm_usage() is called so the
        # observable cost matches TS metrics.ts behaviour.
        self._total_llm_cost: float = 0.0

        # --- Preemptive-generation counters (pipeline mode, opt-in) ---
        # Hits = speculative turns released on a matching final transcript;
        # misses = speculations started but discarded (mismatched final,
        # barge-in, replaced by a newer interim, buffer overflow). Surfaced
        # on the final CallMetrics. Parity with TS ``_preemptiveHits`` /
        # ``_preemptiveMisses``.
        self._preemptive_hits: int = 0
        self._preemptive_misses: int = 0

        # --- Context-size tracking (pipeline mode, built-in LLM loop) ---
        # Peak estimated prompt size (chars/4 token unit) observed across the
        # call. Recorded by ``LLMLoop.run`` once per turn before dispatch.
        # Surfaced on the final CallMetrics as ``context_tokens``. Parity with
        # TS ``_maxContextTokens``.
        self._max_context_tokens: int = 0

    # ---- EventBus attachment ----

    def attach_event_bus(self, bus: "EventBus") -> None:
        """Attach an :class:`~getpatter.observability.event_bus.EventBus`.

        The bus receives structured metric events in addition to any existing
        callback-based ``on_metrics`` path. Safe to call multiple times (last
        bus wins).
        """
        self._event_bus = bus

    def configure_stt_format(
        self, sample_rate: int = 16000, bytes_per_sample: int = 2
    ) -> None:
        """Configure audio format for STT byte → seconds conversion.

        Args:
            sample_rate: Audio sample rate in Hz (8000 for mulaw/Twilio,
                16000 for PCM/Telnyx).
            bytes_per_sample: Bytes per sample (1 for mulaw, 2 for PCM16).
        """
        self._stt_sample_rate = sample_rate
        self._stt_bytes_per_sample = bytes_per_sample

    # ---- Turn lifecycle ----

    @property
    def turn_active(self) -> bool:
        """True when ``start_turn`` was called and the turn is not yet completed."""
        return self._turn_start is not None

    def start_turn(self) -> None:
        """Begin tracking a new conversation turn."""
        self._turn_start = time.monotonic()
        self._stt_complete = None
        self._llm_first_token = None
        self._llm_first_sentence = None
        self._llm_complete = None
        self._tts_first_byte = None
        self._tts_last_byte = None
        self._endpoint_signal_at = None
        self._bargein_detected_at = None
        self._bargein_stopped_at = None
        self._turn_user_text = ""
        self._turn_stt_audio_seconds = 0.0
        self._turn_already_closed = False
        # Reset per-turn TTFB guard flags
        self._llm_ttfb_emitted = False
        self._tts_ttfb_emitted = False
        # Reset EOU timestamps for the new turn
        self._vad_stopped_at = None
        self._stt_final_at = None
        self._turn_committed_at = None
        self._turn_committed_mono = None
        self._on_user_turn_completed_delay_ms = None

        if self._event_bus is not None:
            self._event_bus.emit("turn_started", {"call_id": self.call_id})

    def start_turn_if_idle(self) -> None:
        """Start a new turn only when no turn is already open.

        Call on the first inbound audio byte / non-final transcript so that
        STT latency is measured from the start of speech rather than from the
        final-transcript callback.  No-ops when a turn is already active so
        duplicate calls (e.g. from multiple non-final callbacks) are harmless.
        """
        if self._turn_start is None:
            self.start_turn()

    def reserve_turn_index(self) -> int:
        """Reserve and return the next monotonic turn index.

        Called once per turn when the turn OPENS (Realtime ``speech_stopped``
        edge) so the handler can stamp the reserved index onto each transcript
        line as it becomes known — keeping dashboard ordering stable even when
        the user line arrives after the assistant line. The reserved value is
        later passed back to :meth:`record_turn_complete` /
        :meth:`record_turn_interrupted` as ``pre_reserved_index`` so the
        recorded ``turn_index`` matches the live transcript lines. Parity with
        TS ``metrics.ts:reserveTurnIndex``.
        """
        result = self._next_turn_index
        self._next_turn_index += 1
        return result

    def anchor_user_speech_start(self) -> None:
        """Anchor the current turn at a legitimate VAD ``speech_start``.

        Industry-standard pattern: every VAD ``speech_start`` event that fires while the
        agent is NOT in the suppressed warmup window is treated as the
        canonical start of the user's utterance for the current turn. Reset
        ``_turn_start`` and ``_endpoint_signal_at`` so that:

        * If a phantom ``speech_start`` previously stamped ``_turn_start``
          during agent TTS (e.g. echo / loopback / cough that was
          ``_can_barge_in()``-suppressed but not phantom-suppressed at the
          metrics layer in earlier 0.6.1 builds), the legitimate user-speech
          start re-anchors the timer at the right wall-clock moment.
        * The stale ``_endpoint_signal_at`` from any rejected barge-in /
          dropped final transcript on the SAME turn (before LLM dispatch)
          is cleared, so the next ``record_vad_stop`` stamps fresh.

        No-ops once the turn is committed (``_turn_committed_mono`` set):
        a new VAD ``speech_start`` after commit is the START of the next
        turn's barge-in, handled by ``record_turn_interrupted`` instead.
        """
        if self._turn_committed_mono is not None:
            return
        self._turn_start = time.monotonic()
        # A pre-commit VAD speech_start opens a NEW turn — re-arm the guard
        # exactly as start_turn() does. Without this, the guard set by the
        # previous turn's record_turn_complete persists, and every later
        # anchor-opened turn makes record_turn_complete a no-op (dropping
        # per-turn metrics AND the live SSE transcript). Safe: this method
        # no-ops after commit (guard above), so we only ever re-arm at the
        # start of a fresh, uncommitted turn — never the post-commit
        # barge-in path that _turn_already_closed protects.
        self._turn_already_closed = False
        self._endpoint_signal_at = None
        self._vad_stopped_at = None
        self._stt_final_at = None
        # Reset all the other "first signal of this turn" stamps too so a
        # mid-turn re-anchor produces a clean per-turn breakdown.
        self._stt_complete = None
        self._llm_first_token = None
        self._llm_ttfb_emitted = False

    def record_llm_first_token(self) -> None:
        """Mark when the first LLM output token arrives (TTFT).

        Call on the first streaming token yielded by the LLM. Used to compute
        llm_ttft_ms = first_token - stt_complete, which is distinct from
        llm_ms = llm_complete - stt_complete (full generation time).

        When ``report_only_initial_ttfb=True`` is set, subsequent calls within
        the same turn are silently ignored after the first emission.
        """
        if self._llm_first_token is None:
            self._llm_first_token = time.monotonic()
            if not self._report_only_initial_ttfb or not self._llm_ttfb_emitted:
                self._llm_ttfb_emitted = True
                if self._event_bus is not None and self._stt_complete is not None:
                    from getpatter.observability.metric_types import TTFBMetrics

                    self._event_bus.emit(
                        "llm_metrics",
                        TTFBMetrics(
                            processor="llm",
                            value=self._llm_first_token - self._stt_complete,
                            model=self._llm_model or None,
                        ),
                    )

    def record_llm_first_sentence(self) -> None:
        """Mark when the first sentence boundary in the LLM stream is reached.

        Call when SentenceChunker.push() first returns a non-empty list.
        Used as the TTS span start instead of llm_complete so that
        tts_ms is positive even in streaming-pipeline mode where TTS begins
        before the full LLM response is done.  Falls back to llm_complete
        when not set (realtime / non-streaming paths).
        """
        if self._llm_first_sentence is None:
            self._llm_first_sentence = time.monotonic()

    def record_stt_complete(self, text: str, audio_seconds: float = 0.0) -> None:
        """Mark STT as complete for the current turn."""
        self._stt_complete = time.monotonic()
        # Parity with TS metrics.ts: auto-stamp _stt_final_at so EOU metrics
        # work even when record_stt_final_timestamp() is never called explicitly.
        # record_stt_final_timestamp() may still override this with a more
        # precise provider-supplied timestamp.
        self._stt_final_at = time.time()
        # Don't fake _endpoint_signal_at from _stt_complete — that creates
        # dishonest endpoint_ms == stt_ms outliers (6818 ms p95 spikes observed
        # on PSTN when VAD speech_end is dropped). Honest None is better than a
        # synthetic value for SLO/alerting. The counter lets us know if this
        # happens often (PSTN packet loss dropping VAD stops is the common
        # cause).
        if self._endpoint_signal_at is None:
            self._endpoint_signal_missing_count += 1
        self._turn_user_text = text
        self._turn_stt_audio_seconds = audio_seconds
        self._total_stt_audio_seconds += audio_seconds

        if self._event_bus is not None:
            from getpatter.observability.metric_types import ProcessingMetrics

            self._event_bus.emit(
                "stt_metrics",
                ProcessingMetrics(
                    processor="stt",
                    value=(
                        (self._stt_complete - self._turn_start)
                        if self._turn_start is not None
                        else 0.0
                    ),
                ),
            )

    def record_llm_complete(self) -> None:
        """Mark LLM/on_message as complete for the current turn."""
        self._llm_complete = time.monotonic()

    def record_tts_first_byte(self) -> None:
        """Mark first TTS audio byte received for the current turn.

        When ``report_only_initial_ttfb=True`` is set, the bus emission is
        suppressed after the first byte event per turn.
        """
        if self._tts_first_byte is None:
            self._tts_first_byte = time.monotonic()
            if not self._report_only_initial_ttfb or not self._tts_ttfb_emitted:
                self._tts_ttfb_emitted = True
                if self._event_bus is not None:
                    tts_ref = (
                        self._llm_first_sentence
                        if self._llm_first_sentence is not None
                        else self._llm_complete
                    )
                    if tts_ref is not None:
                        from getpatter.observability.metric_types import TTFBMetrics

                        self._event_bus.emit(
                            "tts_metrics",
                            TTFBMetrics(
                                processor="tts",
                                value=self._tts_first_byte - tts_ref,
                            ),
                        )

    def record_tts_complete(self, text: str) -> None:
        """Mark TTS synthesis as complete, accumulating character count.

        Also captures the monotonic timestamp of the last TTS audio byte
        sent on the wire so ``tts_total_ms`` (LLM-first-token → TTS done)
        can be computed.  Idempotent within a turn — the first call wins.
        """
        self._total_tts_characters += len(text)
        if self._tts_last_byte is None:
            self._tts_last_byte = time.monotonic()

    def record_tts_complete_ts(self, ts: float | None = None) -> None:
        """Capture the timestamp when the last TTS audio byte was sent.

        Useful when the caller wants to record the timing without bumping
        the character counter (e.g. interrupted turns where the byte
        actually went out but the synthesis was truncated).
        """
        self._tts_last_byte = ts if ts is not None else time.monotonic()

    def record_bargein_detected(self, ts: float | None = None) -> None:
        """Mark the moment a user interrupt (barge-in) was detected.

        Pairs with :meth:`record_tts_stopped` to compute ``bargein_ms``.
        """
        t = ts if ts is not None else time.monotonic()
        self._bargein_detected_at = t
        # Stamp _last_bargein_at on the same monotonic clock as _turn_start so
        # the post-barge-in anchor-gating in _compute_turn_latency stays valid
        # (see the comment there for rationale).
        self._last_bargein_at = t

    def record_tts_stopped(self, ts: float | None = None) -> None:
        """Mark the moment TTS playback was actually halted after a barge-in.

        Call this *after* ``audio_sender.send_clear()`` returns. Combined with
        the matching ``record_bargein_detected`` it produces ``bargein_ms``.
        """
        self._bargein_stopped_at = ts if ts is not None else time.monotonic()

    def record_turn_complete(
        self, agent_text: str, pre_reserved_index: int | None = None
    ) -> TurnMetrics | None:
        """Finalize the current turn and return its metrics.

        Returns ``None`` when ``record_turn_interrupted`` has already
        closed the current turn — this protects against the VAD-barge-in
        / pipeline-LLM race where both paths try to finalise the same
        logical turn and the second would otherwise push a phantom entry
        with ``user_text=''``. The caller treats ``None`` as "nothing to
        emit"; ``_emit_turn_metrics`` is already null-safe.

        ``pre_reserved_index`` — when the handler reserved a turn index at
        turn-open via :meth:`reserve_turn_index`, pass it here so the recorded
        ``turn_index`` matches the live transcript lines that were stamped with
        the same value. Defaults to ``len(self._turns)`` for back-compat with
        callers that never reserve. Parity with TS
        ``recordTurnComplete(text, preReservedIndex?)``.
        """
        if self._turn_already_closed:
            return None
        latency = self._compute_turn_latency()
        turn = TurnMetrics(
            turn_index=(
                pre_reserved_index
                if pre_reserved_index is not None
                else len(self._turns)
            ),
            user_text=self._turn_user_text,
            agent_text=agent_text,
            latency=latency,
            stt_audio_seconds=self._turn_stt_audio_seconds,
            tts_characters=len(agent_text),
            timestamp=time.time(),
        )
        self._turns.append(turn)
        if self._event_bus is not None:
            self._event_bus.emit(
                "turn_ended",
                {"call_id": self.call_id, "turn": turn},
            )
            self._event_bus.emit(
                "metrics_collected",
                {"call_id": self.call_id, "turn": turn},
            )
        self._reset_turn_state()
        # Bidirectional guard: mark the turn as closed so a late
        # record_turn_interrupted (e.g. from a future refactor that
        # reorders the bargein + LLM-unwind paths) becomes a no-op
        # instead of overwriting the just-emitted turn record. Mirrors
        # the inverse guard in ``record_turn_interrupted`` and keeps
        # the two close paths symmetric.
        self._turn_already_closed = True
        return turn

    def record_turn_interrupted(
        self, pre_reserved_index: int | None = None
    ) -> TurnMetrics | None:
        """Handle a barge-in / interrupted turn.

        Returns partial ``TurnMetrics`` if a turn was in progress, else
        ``None``. Also returns ``None`` when ``record_turn_complete`` has
        already finalised the current turn — bidirectional parity with
        the guard in :meth:`record_turn_complete`. Prevents an out-of-
        order interruption (e.g. a future refactor that reorders the
        bargein + LLM-unwind paths) from overwriting a turn that the
        complete path already emitted.

        ``pre_reserved_index`` — same semantics as
        :meth:`record_turn_complete`; pass the value reserved at turn-open so
        the interrupted turn carries the same stable index as its live
        transcript lines. Defaults to ``len(self._turns)``.
        """
        if self._turn_start is None:
            return None
        if self._turn_already_closed:
            return None

        latency = self._compute_turn_latency()
        turn = TurnMetrics(
            turn_index=(
                pre_reserved_index
                if pre_reserved_index is not None
                else len(self._turns)
            ),
            user_text=self._turn_user_text,
            agent_text="[interrupted]",
            latency=latency,
            stt_audio_seconds=self._turn_stt_audio_seconds,
            tts_characters=0,
            timestamp=time.time(),
        )
        self._turns.append(turn)
        # Emit the turn record BEFORE reset so subscribers see the
        # interrupted turn with its anchors still intact. Parity with
        # record_turn_complete().
        if self._event_bus is not None:
            self._event_bus.emit(
                "turn_ended",
                {"call_id": self.call_id, "turn": turn},
            )
            self._event_bus.emit(
                "metrics_collected",
                {"call_id": self.call_id, "turn": turn},
            )
        self._reset_turn_state()
        # Mark the turn as closed so a late record_turn_complete from
        # the pipeline-LLM unwind path becomes a no-op (see
        # _turn_already_closed).
        self._turn_already_closed = True
        # Extra paranoia: explicitly null out anchors that have caused leaks
        # into subsequent turns when a barge-in is in flight. _reset_turn_state
        # already clears them, but keep this belt-and-braces line so future
        # refactors that touch _reset_turn_state don't silently regress us.
        self._turn_committed_mono = None
        self._endpoint_signal_at = None
        return turn

    # ---- EOUMetrics ----

    def record_vad_stop(
        self, ts: float | None = None, *, first_wins: bool = False
    ) -> None:
        """Record the timestamp when VAD detects end-of-speech.

        Args:
            ts: Wall-clock timestamp (seconds). Defaults to ``time.time()``.
            first_wins: When True, keep an existing stamp for this turn —
                used by the final-transcript fallback so it cannot overwrite
                the REAL VAD ``speech_end`` stamp microseconds before
                ``record_stt_final_timestamp`` (which made
                ``end_of_utterance_delay`` always ≈0 and faked the endpoint
                signal the very line before ``record_stt_complete``'s
                explicit don't-fake logic).
        """
        if first_wins and self._vad_stopped_at is not None:
            return
        self._vad_stopped_at = ts if ts is not None else time.time()
        # First endpoint signal wins for endpoint_ms calculation.
        if self._endpoint_signal_at is None:
            self._endpoint_signal_at = time.monotonic()

    def record_stt_final_timestamp(self, ts: float | None = None) -> None:
        """Record the timestamp when the STT provider returns a final transcript.

        Args:
            ts: Wall-clock timestamp (seconds). Defaults to ``time.time()``.
        """
        self._stt_final_at = ts if ts is not None else time.time()
        # First endpoint signal wins for endpoint_ms calculation.
        if self._endpoint_signal_at is None:
            self._endpoint_signal_at = time.monotonic()

    def record_turn_committed(self, ts: float | None = None) -> None:
        """Record the timestamp when the pipeline commits the turn for LLM processing.

        Args:
            ts: Wall-clock timestamp (seconds). Defaults to ``time.time()``.
        """
        self._turn_committed_at = ts if ts is not None else time.time()
        # Always stamp a monotonic reference for endpoint_ms even if the
        # caller passed an explicit wall-clock ts (they are different
        # clocks; we want monotonic for ms latency math).
        self._turn_committed_mono = time.monotonic()
        self._emit_eou_metrics()

    def record_on_user_turn_completed_delay(self, delay_ms: float) -> None:
        """Record the measured execution time of the on_user_turn_completed hook.

        Args:
            delay_ms: Hook execution time in milliseconds.
        """
        self._on_user_turn_completed_delay_ms = delay_ms

    def _emit_eou_metrics(self) -> None:
        """Emit ``EOUMetrics`` once all three EOU timestamps are available.

        Guards against emitting garbage data when only a subset of timestamps
        has been recorded (e.g. VAD skipped in non-local mode).

        Field semantics (must match TS ``emitEouMetrics``):
            ``end_of_utterance_delay`` = stt_final − vad_stopped
            ``transcription_delay``    = turn_committed − vad_stopped
        Both deltas are emitted in **milliseconds** (raw wall-clock
        ``time.time()`` timestamps are in seconds, hence the ``* 1000``).
        """
        if (
            self._vad_stopped_at is None
            or self._stt_final_at is None
            or self._turn_committed_at is None
        ):
            return

        if self._event_bus is None:
            return

        from getpatter.observability.metric_types import EOUMetrics

        eou = EOUMetrics(
            end_of_utterance_delay=max(
                0.0, (self._stt_final_at - self._vad_stopped_at) * 1000.0
            ),
            transcription_delay=max(
                0.0, (self._turn_committed_at - self._vad_stopped_at) * 1000.0
            ),
            on_user_turn_completed_delay=(self._on_user_turn_completed_delay_ms or 0.0),
        )
        self._event_bus.emit("eou_metrics", eou)

    # ---- InterruptionMetrics ----

    def record_overlap_start(self, ts: float | None = None) -> None:
        """Record when user speech begins overlapping agent playback.

        Args:
            ts: Wall-clock timestamp (seconds). Defaults to ``time.time()``.
        """
        self._overlap_started_at = ts if ts is not None else time.time()

    def record_overlap_end(
        self, was_interruption: bool = True, ts: float | None = None
    ) -> None:
        """Record when the overlap period ends and emit ``InterruptionMetrics``.

        Args:
            was_interruption: When ``True`` increments ``_num_interruptions``;
                when ``False`` increments ``_num_backchannels``.
            ts: End timestamp (seconds). Defaults to ``time.time()``.
        """
        if self._overlap_started_at is None:
            return

        end_ts = ts if ts is not None else time.time()
        # MILLISECONDS — parity with the TS emitter and with every other
        # latency field in the SDK (this event was the lone seconds outlier).
        detection_delay = (end_ts - self._overlap_started_at) * 1000.0
        self._overlap_started_at = None

        if was_interruption:
            self._num_interruptions += 1
        else:
            self._num_backchannels += 1

        if self._event_bus is not None:
            from getpatter.observability.metric_types import InterruptionMetrics

            metrics = InterruptionMetrics(
                total_duration=detection_delay,
                detection_delay=detection_delay,
                num_interruptions=self._num_interruptions,
                num_backchannels=self._num_backchannels,
            )
            self._event_bus.emit("interruption", metrics)

    # ---- Usage tracking ----

    def add_stt_audio_bytes(self, byte_count: int) -> None:
        """Accumulate raw audio bytes sent to STT (used for cost calculation)."""
        self._stt_byte_count += byte_count

    def record_realtime_usage(self, usage: dict, model: str | None = None) -> None:
        """Record OpenAI Realtime token usage from a ``response.done`` event.

        ``model`` allows the cost calc to pick the per-model rate (e.g.
        ``gpt-realtime-2``). Defaults to whatever was supplied at construction
        time (``self.realtime_model``); pass an explicit value to override
        per-call (the ``response.done`` payload carries the model used).
        """
        resolved_model = model or self.realtime_model or None
        self._total_realtime_cost += calculate_realtime_cost(
            usage, self._pricing, model=resolved_model
        )
        self._total_realtime_cached_savings += calculate_realtime_cached_savings(
            usage, self._pricing, model=resolved_model
        )

    def record_llm_usage(
        self,
        provider: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        cache_read_tokens: int = 0,
        cache_write_tokens: int = 0,
    ) -> None:
        """Accumulate LLM token usage for pipeline-mode cost calculation.

        Call once per LLM response when a usage chunk is received from the
        provider stream. Stacks across turns so end_call() can produce a
        total LLM cost line.

        Args:
            provider: Provider key in LLM_PRICING (e.g. "anthropic", "openai").
            model: Model identifier (e.g. "claude-haiku-4-5").
            input_tokens: Non-cached input tokens at the full input rate.
            output_tokens: Output tokens at the output rate.
            cache_read_tokens: Tokens served from the provider prompt cache.
            cache_write_tokens: Tokens that populated the cache this call.
        """
        self._llm_provider_name = provider
        self._llm_model = model
        self._llm_total_input_tokens += input_tokens
        self._llm_total_output_tokens += output_tokens
        self._llm_total_cache_read_tokens += cache_read_tokens
        self._llm_total_cache_write_tokens += cache_write_tokens
        # Parity with TS metrics.ts: accumulate cost per-turn eagerly so
        # get_cost_so_far() returns an up-to-date value mid-call.
        self._total_llm_cost += calculate_llm_cost(
            provider=provider,
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cache_read_tokens=cache_read_tokens,
            cache_write_tokens=cache_write_tokens,
        )

    def record_preemptive_hit(self) -> None:
        """Count a preemptive (speculative) turn RELEASED on a matching final
        transcript — the buffered LLM+TTS work became the real turn."""
        self._preemptive_hits += 1

    def record_preemptive_miss(self) -> None:
        """Count a preemptive (speculative) turn that was started but
        discarded without release (mismatched final, barge-in during
        speculation, replaced by a newer interim, or buffer overflow)."""
        self._preemptive_misses += 1

    def record_context_tokens(self, tokens: int) -> None:
        """Record the estimated prompt size for the current turn.

        Tracks the per-call PEAK (the largest single-turn prompt seen) and
        emits a ``context_tokens`` event on the bus. Called once per turn by
        ``LLMLoop.run`` before dispatch. Parity with TS
        ``recordContextTokens``.
        """
        if tokens > self._max_context_tokens:
            self._max_context_tokens = tokens
        if self._event_bus is not None:
            self._event_bus.emit(
                "context_tokens", {"call_id": self.call_id, "tokens": tokens}
            )

    @property
    def max_context_tokens(self) -> int:
        """Peak estimated prompt size (chars/4 token unit) seen this call."""
        return self._max_context_tokens

    def set_actual_telephony_cost(self, cost: float) -> None:
        """Set the actual telephony cost from the provider API (post-call).

        When set, this takes priority over the estimated cost based on
        duration and default pricing.
        """
        self._actual_telephony_cost = cost

    def set_actual_stt_cost(self, cost: float) -> None:
        """Set the actual STT cost from the provider API (post-call).

        When set, this takes priority over the estimated cost based on
        audio duration and default pricing.
        """
        self._actual_stt_cost = cost

    # ---- Finalize ----

    def end_call(self) -> CallMetrics:
        """Calculate final costs and return frozen ``CallMetrics``."""
        duration = time.monotonic() - self._call_start

        # Flush any dangling in-flight turn as interrupted so its partial state
        # doesn't evaporate into the void on abrupt hangup. ``_completed_turns``
        # drops it from percentile stats regardless.
        if self.turn_active:
            self.record_turn_interrupted()

        # Compute STT audio seconds from byte count if not already tracked
        if self._total_stt_audio_seconds == 0.0 and self._stt_byte_count > 0:
            self._total_stt_audio_seconds = self._stt_byte_count / (
                self._stt_sample_rate * self._stt_bytes_per_sample
            )

        cost = self._compute_cost(duration)
        latency_avg = self._compute_average_latency()
        latency_p50 = self._compute_percentile_latency(0.5)
        latency_p90 = self._compute_percentile_latency(0.90)
        latency_p95 = self._compute_percentile_latency(0.95)
        latency_p99 = self._compute_percentile_latency(0.99)

        result = CallMetrics(
            call_id=self.call_id,
            duration_seconds=round(duration, 2),
            turns=tuple(self._turns),
            cost=cost,
            latency_avg=latency_avg,
            latency_p50=latency_p50,
            latency_p90=latency_p90,
            latency_p95=latency_p95,
            latency_p99=latency_p99,
            provider_mode=self.provider_mode,
            stt_provider=self.stt_provider,
            tts_provider=self.tts_provider,
            llm_provider=self.llm_provider,
            telephony_provider=self.telephony_provider,
            stt_model=self.stt_model,
            tts_model=self.tts_model,
            llm_model=self._llm_model,
            error_code=self._error_code,
            preemptive_hits=self._preemptive_hits,
            preemptive_misses=self._preemptive_misses,
            context_tokens=self._max_context_tokens,
        )

        if self._event_bus is not None:
            self._event_bus.emit(
                "call_ended", {"call_id": self.call_id, "metrics": result}
            )

        return result

    def record_error(self, exc: BaseException) -> None:
        """Record the call's terminal error as a coarse, anonymous code.

        Stores ``exc.code`` (a :class:`~getpatter.exceptions.ErrorCode`) lowercased
        for a ``PatterError``; maps common stdlib timeouts/connection errors; falls
        back to ``"other"``. Never stores the message. Last write wins.
        """
        code = getattr(exc, "code", None)
        if code is not None:
            self._error_code = str(getattr(code, "value", code)).lower()
        elif isinstance(exc, TimeoutError):  # asyncio.TimeoutError aliases this (3.11+)
            self._error_code = "timeout"
        elif isinstance(exc, ConnectionError):
            self._error_code = "connection"
        else:
            self._error_code = "other"

    def get_cost_so_far(self) -> CostBreakdown:
        """Return current accumulated cost (for real-time ``on_metrics``)."""
        duration = time.monotonic() - self._call_start
        return self._compute_cost(duration)

    # ---- Internal helpers ----

    def _reset_turn_state(self) -> None:
        self._turn_start = None
        self._stt_complete = None
        self._llm_first_token = None
        self._llm_first_sentence = None
        self._llm_complete = None
        self._tts_first_byte = None
        self._tts_last_byte = None
        self._endpoint_signal_at = None
        # Parity with TS ``metrics.ts:_resetTurnState`` — without clearing
        # ``_turn_committed_mono`` here, the ``anchor_user_speech_start``
        # guard (``if self._turn_committed_mono is not None: return``)
        # falsely no-ops on a VAD ``speech_start`` arriving between
        # ``record_turn_complete`` and the next ``start_turn`` on
        # back-to-back turns.
        self._turn_committed_mono = None
        self._bargein_detected_at = None
        self._bargein_stopped_at = None
        self._turn_user_text = ""
        self._turn_stt_audio_seconds = 0.0
        self._llm_ttfb_emitted = False
        self._tts_ttfb_emitted = False

    def _compute_turn_latency(self) -> LatencyBreakdown:
        """Compute latency breakdown for the current turn."""
        stt_ms = 0.0
        llm_ms = 0.0
        llm_ttft_ms = 0.0
        tts_ms = 0.0
        total_ms = 0.0
        user_speech_duration_ms: float | None = None

        # Post-barge-in turns have unreliable anchors. Drop endpoint_ms /
        # stt_ms to avoid polluting the p95 distribution with synthetic
        # spikes. The honest None makes the metric usable for SLO/alerting;
        # without this gate, a single barge-in produces 6+ second p95
        # outliers.
        post_bargein = (
            self._last_bargein_at is not None
            and self._turn_start is not None
            and abs(self._turn_start - self._last_bargein_at) <= 0.1  # 100 ms
        )

        # ``stt_ms`` measures pure STT finalization: end-of-speech (VAD stop
        # or STT speech_final) → final transcript delivery. This is the
        # engineering metric reported as "STT latency" by the industry. When
        # the endpoint signal is unavailable (degraded provider, batch STT)
        # fall back to the legacy turn_start anchor so the field is never
        # spuriously zero.
        if self._stt_complete is not None:
            anchor = (
                self._endpoint_signal_at
                if self._endpoint_signal_at is not None
                else self._turn_start
            )
            if anchor is not None:
                stt_ms = max(0.0, (self._stt_complete - anchor) * 1000)

        if self._turn_start is not None and self._endpoint_signal_at is not None:
            user_speech_duration_ms = max(
                0.0, (self._endpoint_signal_at - self._turn_start) * 1000
            )

        # Parity with TS metrics.ts: llm_ms = TTFT when first-token is available,
        # otherwise falls back to full generation time (stt_complete → llm_complete).
        if self._stt_complete is not None and self._llm_first_token is not None:
            llm_ms = max(0.0, (self._llm_first_token - self._stt_complete) * 1000)
        elif self._stt_complete is not None and self._llm_complete is not None:
            llm_ms = max(0.0, (self._llm_complete - self._stt_complete) * 1000)

        # Fix 5: LLM TTFT = first-token time minus stt_complete.
        if self._stt_complete is not None and self._llm_first_token is not None:
            llm_ttft_ms = max(0.0, (self._llm_first_token - self._stt_complete) * 1000)

        # Fix 3: TTS span starts from the first-sentence boundary rather than
        # llm_complete.  In streaming-pipeline mode, record_tts_first_byte fires
        # on the first audio chunk of the first sentence, which is always AFTER
        # llm_first_sentence — so tts_ms = tts_first_byte - llm_first_sentence
        # is always non-negative without clamping.  Fallback to llm_complete for
        # realtime / non-streaming paths where llm_first_sentence is never set.
        tts_ref = (
            self._llm_first_sentence
            if self._llm_first_sentence is not None
            else self._llm_complete
        )
        if tts_ref is not None and self._tts_first_byte is not None:
            tts_ms = max(0.0, (self._tts_first_byte - tts_ref) * 1000)

        if self._turn_start is not None and self._tts_first_byte is not None:
            total_ms = (self._tts_first_byte - self._turn_start) * 1000

        # Persist TTFT so _emit_turn_metrics can include it after _reset_turn_state.
        self._last_turn_llm_ttft_ms = llm_ttft_ms

        # llm_total_ms — full generation duration (stt_complete → llm_complete).
        llm_total_ms: float | None = None
        if self._stt_complete is not None and self._llm_complete is not None:
            llm_total_ms = max(0.0, (self._llm_complete - self._stt_complete) * 1000)

        # endpoint_ms — silence detected (VAD or STT speech_final) → LLM dispatch.
        endpoint_ms: float | None = None
        if (
            self._endpoint_signal_at is not None
            and self._turn_committed_mono is not None
        ):
            endpoint_ms = max(
                0.0, (self._turn_committed_mono - self._endpoint_signal_at) * 1000
            )

        # bargein_ms — interrupt detected → TTS actually halted.
        bargein_ms: float | None = None
        if (
            self._bargein_detected_at is not None
            and self._bargein_stopped_at is not None
        ):
            bargein_ms = max(
                0.0, (self._bargein_stopped_at - self._bargein_detected_at) * 1000
            )

        # tts_total_ms — LLM-first-token (or first-sentence boundary) → last
        # TTS audio byte sent.  Prefer ``llm_first_token`` so this captures
        # the entire user-perceptible TTS span; fall back to first-sentence
        # boundary (streaming) and finally to llm_complete.
        tts_total_ms: float | None = None
        tts_total_ref = (
            self._llm_first_token
            if self._llm_first_token is not None
            else (
                self._llm_first_sentence
                if self._llm_first_sentence is not None
                else self._llm_complete
            )
        )
        if tts_total_ref is not None and self._tts_last_byte is not None:
            tts_total_ms = max(0.0, (self._tts_last_byte - tts_total_ref) * 1000)

        # agent_response_ms — the user-perceived latency. Sum of the three
        # system-controlled segments (silence detection + LLM TTFT + TTS
        # first-byte). Undefined when any prerequisite signal is missing —
        # we deliberately do NOT fall back to total_ms so dashboards can
        # distinguish "metric available" vs "metric missing".
        agent_response_ms: float | None = None
        # Gate on the actual first-token SIGNAL: ``llm_ttft_ms`` is
        # initialised to 0.0, so the old ``is not None`` check was always
        # true — when record_llm_first_token never fired (non-streaming
        # on_message, custom handlers) the flagship SLO metric silently
        # EXCLUDED the entire LLM segment. TS already leaves it undefined.
        if endpoint_ms is not None and self._llm_first_token is not None and tts_ms > 0:
            agent_response_ms = round(endpoint_ms + llm_ttft_ms + tts_ms, 1)

        # Post-barge-in anchor hygiene: when the current turn began within
        # 100 ms of the last detected barge-in, the VAD/STT anchors are
        # unreliable. Drop the polluted endpoint_ms and stt_ms so percentile
        # aggregations ignore them (stt_ms = 0 is excluded by the non-zero
        # filter in _compute_percentile_latency).
        if post_bargein:
            stt_ms = 0.0
            endpoint_ms = None

        # Note: in Realtime mode OpenAI handles STT+LLM+TTS as a single opaque
        # pipeline, so stt_ms / llm_ms / tts_ms stay 0 and only total_ms is
        # meaningful. Dashboards should prefer total_ms as the end-to-end
        # proxy and treat the component buckets as "unknown / bundled by
        # provider" when total_ms > 0 but all three are 0.
        return LatencyBreakdown(
            stt_ms=round(stt_ms, 1),
            llm_ms=round(llm_ms, 1),
            tts_ms=round(tts_ms, 1),
            total_ms=round(total_ms, 1),
            user_speech_duration_ms=(
                round(user_speech_duration_ms, 1)
                if user_speech_duration_ms is not None
                else None
            ),
            llm_ttft_ms=round(llm_ttft_ms, 1) if llm_ttft_ms else None,
            llm_total_ms=round(llm_total_ms, 1) if llm_total_ms is not None else None,
            endpoint_ms=round(endpoint_ms, 1) if endpoint_ms is not None else None,
            bargein_ms=round(bargein_ms, 1) if bargein_ms is not None else None,
            tts_total_ms=round(tts_total_ms, 1) if tts_total_ms is not None else None,
            agent_response_ms=agent_response_ms,
        )

    @property
    def last_turn_llm_ttft_ms(self) -> float:
        """LLM TTFT (first-token latency, ms) from the most recently completed turn.

        Available after ``record_turn_complete`` or ``record_turn_interrupted``
        returns.  Zero when the LLM first-token was not recorded (e.g. Realtime
        / non-streaming paths).
        """
        return self._last_turn_llm_ttft_ms

    def _compute_cost(self, duration_seconds: float) -> CostBreakdown:
        """Compute cost breakdown from accumulated usage data."""
        if self.provider_mode in ("openai_realtime", "openai_realtime_2"):
            # OpenAI Realtime (v1-beta AND GA): STT+LLM+TTS cost comes from
            # token usage. Matching only the exact "openai_realtime" string
            # made every GA-engine call fall into the pipeline branch and
            # report $0 AI cost (while still emitting cached savings).
            stt_cost = 0.0
            tts_cost = 0.0
            llm_cost = self._total_realtime_cost
        elif self.provider_mode == "elevenlabs_convai":
            # ElevenLabs ConvAI: bundled pricing, estimate from duration
            stt_cost = 0.0
            tts_cost = 0.0
            llm_cost = 0.0  # ElevenLabs doesn't expose per-token pricing
        else:
            # Pipeline mode: separate providers
            # Prefer actual STT cost from provider API over estimate
            if self._actual_stt_cost is not None:
                stt_cost = self._actual_stt_cost
            else:
                stt_cost = calculate_stt_cost(
                    self.stt_provider,
                    self._total_stt_audio_seconds,
                    self._pricing,
                    model=self.stt_model or None,
                )
            tts_cost = calculate_tts_cost(
                self.tts_provider,
                self._total_tts_characters,
                self._pricing,
                model=self.tts_model or None,
            )
            # Pipeline LLM cost: already accumulated per-turn in
            # record_llm_usage(); _total_llm_cost is 0 when on_message
            # is a custom handler that never calls record_llm_usage().
            llm_cost = self._total_llm_cost

        # Prefer actual telephony cost from provider API over estimate
        if self._actual_telephony_cost is not None:
            telephony_cost = self._actual_telephony_cost
        else:
            telephony_cost = calculate_telephony_cost(
                self.telephony_provider, duration_seconds, self._pricing
            )

        total = stt_cost + tts_cost + llm_cost + telephony_cost

        return CostBreakdown(
            stt=round(stt_cost, 6),
            tts=round(tts_cost, 6),
            llm=round(llm_cost, 6),
            telephony=round(telephony_cost, 6),
            total=round(total, 6),
            llm_cached_savings=round(self._total_realtime_cached_savings, 6),
        )

    def _completed_turns(self) -> list:
        """Turns eligible for latency statistics.

        Excludes turns marked ``[interrupted]`` (barge-in, cancelled
        replacements) because their recorded latency either reflects partial
        state or zero — including them would drag every p95/avg bucket toward
        meaningless numbers.
        """
        return [
            t
            for t in self._turns
            if t.agent_text != "[interrupted]" and t.latency.total_ms > 0
        ]

    def _compute_average_latency(self) -> LatencyBreakdown:
        """Compute average latency across completed turns."""
        turns = self._completed_turns()
        if not turns:
            return LatencyBreakdown()

        n = len(turns)

        def _opt_avg(attr: str) -> float | None:
            # Filter zeros like TS optAvg (and like both SDKs' percentile
            # paths) — keeping 0.0 samples made the two SDKs' averages drift
            # for the same call whenever a missing-signal turn recorded 0.
            vals = [
                v
                for t in turns
                if (v := getattr(t.latency, attr)) is not None and v > 0
            ]
            return round(sum(vals) / len(vals), 1) if vals else None

        ttft_avg = _opt_avg("llm_ttft_ms")
        return LatencyBreakdown(
            stt_ms=round(sum(t.latency.stt_ms for t in turns) / n, 1),
            llm_ms=round(sum(t.latency.llm_ms for t in turns) / n, 1),
            tts_ms=round(sum(t.latency.tts_ms for t in turns) / n, 1),
            total_ms=round(sum(t.latency.total_ms for t in turns) / n, 1),
            llm_ttft_ms=ttft_avg,
            llm_total_ms=_opt_avg("llm_total_ms"),
            endpoint_ms=_opt_avg("endpoint_ms"),
            bargein_ms=_opt_avg("bargein_ms"),
            tts_total_ms=_opt_avg("tts_total_ms"),
            user_speech_duration_ms=_opt_avg("user_speech_duration_ms"),
            agent_response_ms=_opt_avg("agent_response_ms"),
        )

    def _compute_percentile_latency(self, p: float) -> LatencyBreakdown:
        """Compute an arbitrary percentile latency across completed turns.

        Uses linear interpolation between order statistics (Hyndman-Fan type
        7, same as numpy.percentile default). Previous ``floor()`` variant
        returned the sample max for any n < 21, making p95/p99 on short calls
        indistinguishable from max. Linear interpolation is meaningful even
        on 2-3 sample sets.

        Fix 4: Per-component, zero-valued entries are excluded before computing
        percentiles.  Realtime turns and firstMessage turns never record
        component-level latencies (stt_ms/llm_ms/tts_ms stay 0) and including
        them would drag every component percentile bucket toward zero.  When all
        values for a component are zero the result is returned as 0.0.
        total_ms is NOT filtered because it is always populated for completed turns.
        """
        turns = self._completed_turns()
        if not turns:
            return LatencyBreakdown()

        def pct(values: list[float]) -> float:
            # Filter out zeros so Realtime / firstMessage turns don't drag
            # component-level buckets (stt, llm, tts) toward zero.
            non_zero = [v for v in values if v > 0]
            if not non_zero:
                return 0.0
            sorted_v = sorted(non_zero)
            if len(sorted_v) == 1:
                return sorted_v[0]
            rank = p * (len(sorted_v) - 1)
            lo = int(rank)
            hi = min(lo + 1, len(sorted_v) - 1)
            if lo == hi:
                return sorted_v[lo]
            frac = rank - lo
            return sorted_v[lo] + (sorted_v[hi] - sorted_v[lo]) * frac

        def pct_all(values: list[float]) -> float:
            """Percentile over all values, including zeros (used for total_ms)."""
            if not values:
                return 0.0
            sorted_v = sorted(values)
            if len(sorted_v) == 1:
                return sorted_v[0]
            rank = p * (len(sorted_v) - 1)
            lo = int(rank)
            hi = min(lo + 1, len(sorted_v) - 1)
            if lo == hi:
                return sorted_v[lo]
            frac = rank - lo
            return sorted_v[lo] + (sorted_v[hi] - sorted_v[lo]) * frac

        def _opt_pct(attr: str) -> float | None:
            vals = [
                getattr(t.latency, attr)
                for t in turns
                if getattr(t.latency, attr) is not None
            ]
            v = pct(vals)
            return round(v, 1) if v else None

        ttft_pct_val = pct(
            [t.latency.llm_ttft_ms for t in turns if t.latency.llm_ttft_ms is not None]
        )
        return LatencyBreakdown(
            stt_ms=round(pct([t.latency.stt_ms for t in turns]), 1),
            llm_ms=round(pct([t.latency.llm_ms for t in turns]), 1),
            tts_ms=round(pct([t.latency.tts_ms for t in turns]), 1),
            total_ms=round(pct_all([t.latency.total_ms for t in turns]), 1),
            llm_ttft_ms=round(ttft_pct_val, 1) if ttft_pct_val else None,
            llm_total_ms=_opt_pct("llm_total_ms"),
            endpoint_ms=_opt_pct("endpoint_ms"),
            bargein_ms=_opt_pct("bargein_ms"),
            tts_total_ms=_opt_pct("tts_total_ms"),
            user_speech_duration_ms=_opt_pct("user_speech_duration_ms"),
            agent_response_ms=_opt_pct("agent_response_ms"),
        )
