from __future__ import annotations

import asyncio
import inspect
import json
import os
import random
import re
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Optional

from livekit.agents import UserInputTranscribedEvent, room_io

from app.livekit.adapters.agent_backend import BackendReply
from app.livekit.policies.agent_finalization import AgentFinalStop, final_stop_from_agent_result
from app.livekit.runtime.commands import (
    EndServiceOnce,
    ExportSession,
    ExtractSpokenText,
    InterruptSpeech,
    InjectIdleNudge,
    NotifyBridgeDone,
    NotifyBridgeStop,
    RunPipelineInput,
    SetPendingInterrupt,
    SetPipelineInterruption,
    SetPipelineProcessingInterruption,
    StartSession,
    StartSpeech,
    WaitForSpeechPlayout,
)
from app.livekit.runtime.scheduler import TimerScheduler
from app.livekit.runtime.state import (
    CallState,
    DeferredVADUtterance,
    PendingUserTurn,
    RuntimeConfig,
)
from app.livekit.runtime.wav_audio import wav_audio_frames, wav_duration_ms
from app.utils.logging import (
    EVENT_ENVIO_MSG,
    EVENT_RECEBIMENTO_MSG,
    error_message_from_resource,
    event_latency_ms,
    log_flow_event,
    log_structured_event,
)
from app.utils.turn_ids import (
    clear_started_turn_message_id,
    consume_transcribed_turn,
    next_turn_message_id,
    peek_started_turn_message_id,
)

POST_USER_FINAL_SPEECH_GRACE_S = 0.35
PRE_BACKEND_WAIT_NOTICE_GUARD_S = 0.15
PROTECTED_SPEECH_POST_PLAYOUT_DROP_GRACE_S = 0.250
USER_AUDIO_INPUT_SETUP_TIMEOUT_S = 15.0
AGENT_WAIT_TIMEOUT_STOP_STATUS = "stop_silencio_longo"
TTS_FAILURE_DETAIL = "falha de reproducao do TTS"
IN_SESSION_STOP_STATUS_BY_RESOURCE = {
    "agent_runtime": "stop_agent_runtime_unavailable",
    "agent_backend": "stop_agent_backend_unavailable",
    "stt": "stop_stt_unavailable",
    "tts": "stop_tts_unavailable",
    "bridge": "stop_bridge_failed",
}
INFLIGHT_BACKEND_WAIT_STAGE = "AGENT_BACKEND_WAIT"
AGENT_WAIT_TIMEOUT_RETRY_VAD_STAGES = {"AGENT_WAIT_TIMEOUT_RETRY"}
EMPTY_STT_FINAL_DEDUP_WINDOW_S = 1.0
PENDING_VAD_STT_MAX_AGE_S = 15.0
PENDING_VAD_STT_MAX_UTTERANCES = 32
DEFERRED_INTERRUPTION_COMFORT_TEXT = "Um instante"
DEFERRED_INTERRUPTION_COMFORT_AUDIO_PATH = (
    Path(__file__).resolve().parents[1]
    / "assets"
    / "comfort"
    / "interruption"
    / "01.wav"
)


class InflightBackendWaitTimedOut(Exception):
    pass


@dataclass(frozen=True, slots=True)
class InflightBackendWaitNoticeResult:
    started: bool
    interrupted_by_user: bool = False

    def __bool__(self) -> bool:
        return self.started and not self.interrupted_by_user


@dataclass(frozen=True, slots=True)
class AgentWaitTimeoutConfig:
    timeout_s: float
    retry_messages: tuple[str, ...]


@dataclass(frozen=True, slots=True)
class _TTSPartialFailure:
    detail: str
    provider_synthesis_ms: int | None
    pcm_duration_ms: int | None


def _http_status_details_from_exception(exc: Exception) -> tuple[int | str | None, str | None]:
    response = getattr(exc, "response", None)
    status = getattr(exc, "status_code", None)
    if status is None:
        status = getattr(exc, "status", None)
    if status is None and response is not None:
        status = getattr(response, "status_code", None)
    if status in ("", "NA"):
        status = None

    desc = None
    if response is not None:
        desc = str(getattr(response, "reason_phrase", "") or "").strip() or None
    if desc is None:
        desc = str(getattr(exc, "message", "") or "").strip() or None
    if desc is None and status is not None:
        desc = str(exc).strip() or type(exc).__name__
    return status, desc


class CallRuntime:
    def __init__(
        self,
        *,
        ctx: Any,
        session: Any,
        agent: Any,
        command_executor: Any,
        call_logger: Any,
        protocol: str,
        session_id: str,
        bridge_identity: str,
        agent_starts_conversation: bool,
        nudge_text: str,
        call_t0: float,
        extract_text_from_transcript,
        interrupt_policy: Any,
        idle_policy: Any,
        finalization_policy: Any,
        create_task_logged: Callable[..., asyncio.Task],
        config: RuntimeConfig,
        logger: Any,
        vad_threshold_controller: Any | None = None,
        initial_greeting_tts: Any | None = None,
        timeline: Any | None = None,
        structured_log_context: Any | None = None,
        debug_event_publisher: Callable[..., Any] | None = None,
    ) -> None:
        self._ctx = ctx
        self._session = session
        self._agent = agent
        self._command_executor = command_executor
        self._call_logger = call_logger
        self._protocol = protocol
        self._session_id = session_id
        self._bridge_identity = bridge_identity
        self._agent_starts_conversation = bool(agent_starts_conversation)
        self._nudge_text = nudge_text
        self._call_t0 = call_t0
        self._extract_text_from_transcript = extract_text_from_transcript
        self._interrupt_policy = interrupt_policy
        self._idle_policy = idle_policy
        self._finalization_policy = finalization_policy
        self._create_task_logged = create_task_logged
        self._initial_greeting_tts = initial_greeting_tts
        self._vad_threshold_controller = vad_threshold_controller
        self._config = config
        self._logger = logger
        self._timeline = timeline
        self._structured_log_context = structured_log_context
        self._debug_event_publisher = debug_event_publisher
        self._scheduler = TimerScheduler(create_task_logged=create_task_logged, logger=logger)

        self.state = CallState()
        self._initial_agent_turn_requested = False
        self._user_audio_input_enabled = True
        self._user_audio_input_gate_released = False
        self._say_stage_seq = 0
        self._last_vad_speech_end_at = 0.0
        self._last_vad_speech_duration_ms: int | None = None

        self.speaking = asyncio.Event()
        self.say_lock = asyncio.Lock()
        self.pending_lock = asyncio.Lock()
        self.finalize_lock = asyncio.Lock()
        self.finalized = asyncio.Event()
        self._backend_push_task: asyncio.Task[Any] | None = None
        self._terminal_stop_lock = asyncio.Lock()
        self._terminal_stop_sent = False
        self._last_interrupt_task: asyncio.Task[Any] | None = None
        self._inflight_backend_activity_at = 0.0
        self._pre_backend_wait_notice_task: asyncio.Task[Any] | None = None
        self._pre_backend_wait_notice_generation = 0
        self._pre_backend_wait_notice_reserved = False
        self._agent_wait_timeout_config: AgentWaitTimeoutConfig | None = None
        self._last_agent_wait_timeout_config: AgentWaitTimeoutConfig | None = None
        self._last_agent_wait_timeout_user_seq = 0
        self._last_agent_wait_timeout_source = ""
        self._last_agent_wait_timeout_message_id = ""
        self._last_agent_message_id = ""
        self._empty_stt_recovery_pending = False
        self._agent_state_name = ""
        self._agent_speaking_seq = 0
        self._agent_speaking_at = 0.0
        self._agent_speaking_event = asyncio.Event()
        self._user_state_name = "listening"
        self._user_not_speaking = asyncio.Event()
        self._user_not_speaking.set()
        self._tts_tffb_ms_by_speech_handle_id: dict[str, int] = {}
        self._tts_duration_ms_by_speech_handle_id: dict[str, int] = {}
        self._tts_audio_duration_ms_by_speech_handle_id: dict[str, int] = {}
        self._tts_turn_health_by_segment_id: dict[str, tuple[int, int, int, int]] = {}
        self._tts_turn_health_by_speech_handle_id: dict[str, tuple[int, int, int, int]] = {}
        self._tts_metric_events_by_speech_handle_id: dict[str, asyncio.Event] = {}
        self._retryable_tts_errors_by_speech_handle_id: dict[str, str] = {}
        self._tts_partial_failures_by_speech_handle_id: dict[str, _TTSPartialFailure] = {}
        self._structured_error_event_keys: set[tuple[str, str, str, str, str, str]] = set()
        self._structured_interruption_pending = False
        self._deferred_long_stt_settled = asyncio.Event()
        self._deferred_long_stt_settled.set()
        self._deferred_empty_final_latched_at = 0.0
        self._drop_next_user_final_context: dict[str, Any] | None = None
        self._backend_processing_drop_context: dict[str, Any] | None = None
        self._post_playout_drop_context: dict[str, Any] | None = None
        self._post_playout_drop_expires_at = 0.0
        self._inflight_backend_wait_audio_candidates_cache = (
            self._inflight_backend_wait_audio_candidates_from_config(config)
        )
        self._inflight_backend_wait_audio_duration_cache, self._inflight_backend_wait_audio_failures = (
            self._preload_wait_audio_paths(
                self._unique_wait_audio_candidates(
                    self._inflight_backend_wait_audio_candidates_cache
                ),
                logger=logger,
            )
        )
        self._inflight_backend_wait_audio_text_cache: dict[tuple[Path, str], str] = {}
        self._preload_inflight_backend_wait_audio_texts()
        subscribe_tts_event = getattr(self._initial_greeting_tts, "on", None)
        if callable(subscribe_tts_event):
            subscribe_tts_event("xai_tts_turn_timing", self._record_xai_tts_turn_timing)
            subscribe_tts_event("xai_tts_turn_failed", self._record_xai_tts_turn_failed)

    async def execute(self, command: Any) -> Any:
        return await self._command_executor.execute(command)

    def _emit_debug_event(self, event: str, **data: Any) -> None:
        if self._debug_event_publisher is None:
            return
        try:
            self._create_task_logged(
                self._publish_debug_event(event, **data),
                name=f"debug_event_{str(event).replace('.', '_')}",
            )
        except Exception:
            self._logger.debug("failed scheduling debug event %s", event, exc_info=True)

    async def _publish_debug_event(self, event: str, **data: Any) -> None:
        if self._debug_event_publisher is None:
            return
        try:
            result = self._debug_event_publisher(event, **data)
            if inspect.isawaitable(result):
                await result
        except Exception:
            self._logger.debug("failed publishing debug event %s", event, exc_info=True)

    def publish_stt_provider_metrics(self, fields: dict[str, Any]) -> None:
        data = dict(fields)
        outcome = str(data.pop("event", "completed") or "completed").strip().lower()
        self._emit_debug_event(f"stt.sofya.{outcome}", **data)

    def execute_now(self, command: Any) -> Any:
        return self._command_executor.execute_now(command)

    def _activate_agent_wait_timeout_retry_vad_threshold(
        self,
        *,
        attempt: int,
        reason: str = "agent_wait_timeout_retry",
    ) -> None:
        controller = self._vad_threshold_controller
        activate = getattr(controller, "activate_agent_wait_timeout_retry", None)
        if not callable(activate):
            return
        try:
            activate(attempt=attempt, reason=reason)
        except Exception:
            self._logger.exception("[vad] failed activating agent wait timeout retry threshold")

    def _restore_agent_wait_timeout_retry_vad_threshold(self, *, reason: str) -> None:
        controller = self._vad_threshold_controller
        restore = getattr(controller, "restore", None)
        if not callable(restore):
            return
        try:
            restore(reason=reason)
        except Exception:
            self._logger.exception("[vad] failed restoring agent wait timeout retry threshold")

    @staticmethod
    def _speech_handle_id(handle: Any) -> str:
        return str(getattr(handle, "id", "") or getattr(handle, "_id", "") or "").strip()

    @staticmethod
    def _seconds_to_ms(value: Any) -> int | None:
        try:
            seconds = float(value)
        except (TypeError, ValueError):
            return None
        if seconds < 0:
            return None
        return max(0, round(seconds * 1000))

    @classmethod
    def _positive_seconds_to_ms(cls, value: Any) -> int | None:
        ms = cls._seconds_to_ms(value)
        if ms is None or ms <= 0:
            return None
        return ms

    @staticmethod
    def _nonnegative_int(value: Any) -> int | None:
        try:
            parsed = int(value)
        except (TypeError, ValueError):
            return None
        return max(0, parsed)

    def _record_xai_tts_turn_timing(
        self, event: Any, *, publish_debug: bool = True
    ) -> None:
        """Keep provider-stream health data until its LiveKit metric arrives."""
        if not isinstance(event, dict):
            return
        segment_id = str(event.get("segment_id") or "").strip()
        gap_ms = self._nonnegative_int(event.get("max_audio_delta_gap_ms"))
        underrun_ms = self._nonnegative_int(event.get("max_playout_underrun_0ms"))
        if not segment_id or gap_ms is None or underrun_ms is None:
            return
        underflow_count = self._nonnegative_int(event.get("xai_micro_underflows")) or 0
        avg_underflow_ms = self._nonnegative_int(event.get("xai_avg_underrun_ms")) or 0
        (
            previous_gap_ms,
            previous_underrun_ms,
            previous_underflow_count,
            previous_underflow_total_ms,
        ) = self._tts_turn_health_by_segment_id.get(segment_id, (0, 0, 0, 0))
        self._tts_turn_health_by_segment_id[segment_id] = (
            max(previous_gap_ms, gap_ms),
            max(previous_underrun_ms, underrun_ms),
            previous_underflow_count + underflow_count,
            previous_underflow_total_ms + underflow_count * avg_underflow_ms,
        )
        if publish_debug:
            self._emit_debug_event("tts.xai.completed", **event)

    def _record_xai_tts_turn_failed(self, event: Any) -> None:
        """Bind a provider partial failure to the active speech immediately.

        LiveKit does not emit ``metrics_collected`` when a streamed response
        fails after PCM was released. This event is the authoritative signal
        for the structured error path in that case.
        """
        if not isinstance(event, dict):
            return
        segment_id = str(event.get("segment_id") or "").strip()
        reason = str(event.get("reason") or "provider_partial_failure").strip()
        gap_ms = self._nonnegative_int(event.get("max_audio_delta_gap_ms"))
        underrun_ms = self._nonnegative_int(event.get("max_playout_underrun_0ms"))
        if not segment_id or gap_ms is None or underrun_ms is None:
            return

        self._record_xai_tts_turn_timing(event, publish_debug=False)
        self._emit_debug_event("tts.xai.failed", **event)
        handle = self.state.current_speech.handle
        speech_handle_id = self._speech_handle_id(handle)
        if not speech_handle_id:
            log_flow_event(
                self._call_logger,
                "tts_partial_failure_without_active_handle",
                segment_id=segment_id,
                reason=reason,
            )
            return

        health = self._tts_turn_health_by_segment_id.get(segment_id)
        if health is not None:
            self._tts_turn_health_by_speech_handle_id[speech_handle_id] = health
        underflow_count = self._nonnegative_int(event.get("xai_micro_underflows")) or 0
        avg_underflow_ms = self._nonnegative_int(event.get("xai_avg_underrun_ms")) or 0
        provider_synthesis_ms = self._nonnegative_int(event.get("provider_synthesis_ms"))
        pcm_duration_ms = self._nonnegative_int(event.get("pcm_duration_ms"))
        detail = (
            f"xAI TTS partial audio failure: {reason}; "
            f"xai_underrun_estimado_ms={underrun_ms}; "
            f"xai_micro_underflows={underflow_count}; "
            f"xai_avg_underrun_ms={avg_underflow_ms}; "
            f"max_audio_delta_gap_ms={gap_ms}"
        )
        self._tts_partial_failures_by_speech_handle_id[speech_handle_id] = (
            _TTSPartialFailure(
                detail=detail,
                provider_synthesis_ms=provider_synthesis_ms,
                pcm_duration_ms=pcm_duration_ms,
            )
        )
        log_flow_event(
            self._call_logger,
            "tts_partial_failure_observed",
            stage=self.state.current_speech.stage or self.state.current_stage,
            speech_id=self.state.current_speech.speech_id,
            speech_handle_id=speech_handle_id,
            segment_id=segment_id,
            reason=reason,
            max_gap_ms=gap_ms,
            max_underrun_0ms=underrun_ms,
            underflow_count=underflow_count,
            avg_underflow_ms=avg_underflow_ms,
        )

    def _pop_tts_partial_failure(self, handle: Any) -> _TTSPartialFailure | None:
        speech_handle_id = self._speech_handle_id(handle)
        if not speech_handle_id:
            return None
        return self._tts_partial_failures_by_speech_handle_id.pop(
            speech_handle_id, None
        )

    def _resolve_tts_turn_health_metrics(
        self, handle: Any
    ) -> tuple[int | None, int | None, int | None, int | None]:
        speech_handle_id = self._speech_handle_id(handle)
        if not speech_handle_id:
            return None, None, None, None
        health = self._tts_turn_health_by_speech_handle_id.get(speech_handle_id)
        if health is None:
            return None, None, None, None
        gap_ms, max_underrun_ms, underflow_count, underflow_total_ms = health
        return (
            gap_ms,
            max_underrun_ms,
            underflow_count,
            round(underflow_total_ms / underflow_count) if underflow_count else 0,
        )

    def _record_tts_metric(self, ev: Any) -> None:
        metrics = getattr(ev, "metrics", ev)
        if str(getattr(metrics, "type", "") or "") != "tts_metrics":
            return

        speech_handle_id = str(getattr(metrics, "speech_id", "") or "").strip()
        if not speech_handle_id:
            speech_handle_id = self._speech_handle_id(self.state.current_speech.handle)
        if not speech_handle_id:
            return

        segment_id = str(getattr(metrics, "segment_id", "") or "").strip()
        segment_health = self._tts_turn_health_by_segment_id.get(segment_id)
        if segment_health is not None:
            self._tts_turn_health_by_speech_handle_id.setdefault(
                speech_handle_id, segment_health
            )
        duration_ms = self._seconds_to_ms(getattr(metrics, "duration", None))
        if duration_ms is not None and duration_ms > 0:
            self._tts_duration_ms_by_speech_handle_id.setdefault(
                speech_handle_id,
                duration_ms,
            )

        audio_ms = self._positive_seconds_to_ms(getattr(metrics, "audio_duration", None))
        if audio_ms is not None:
            self._tts_audio_duration_ms_by_speech_handle_id.setdefault(
                speech_handle_id,
                audio_ms,
            )

        ttfb_ms = self._positive_seconds_to_ms(getattr(metrics, "ttfb", None))
        if ttfb_ms is None:
            metric_event = self._tts_metric_events_by_speech_handle_id.get(speech_handle_id)
            if metric_event is not None:
                metric_event.set()
            log_flow_event(
                self._call_logger,
                "tts_metric_invalid",
                speech_handle_id=speech_handle_id,
                tffb_seconds=getattr(metrics, "ttfb", None),
                duration_ms=duration_ms,
                audio_ms=audio_ms,
                provider=str(getattr(metrics, "label", "") or ""),
            )
            return

        self._tts_tffb_ms_by_speech_handle_id.setdefault(speech_handle_id, ttfb_ms)
        metric_event = self._tts_metric_events_by_speech_handle_id.get(speech_handle_id)
        if metric_event is not None:
            metric_event.set()
        log_flow_event(
            self._call_logger,
            "tts_metric",
            speech_handle_id=speech_handle_id,
            tffb_ms=ttfb_ms,
            duration_ms=duration_ms,
            audio_ms=audio_ms,
            provider=str(getattr(metrics, "label", "") or ""),
        )
        self._emit_debug_event(
            "tts.completed",
            stage=self.state.current_speech.stage or self.state.current_stage,
            speech_id=speech_handle_id,
            duration_ms=duration_ms,
            ttfb_ms=ttfb_ms,
            audio_duration_ms=audio_ms,
            provider=str(getattr(metrics, "label", "") or ""),
            max_audio_delta_gap_ms=(
                segment_health[0] if segment_health is not None else None
            ),
            max_playout_underrun_ms=(
                segment_health[1] if segment_health is not None else None
            ),
            underflow_count=(
                segment_health[2] if segment_health is not None else None
            ),
            avg_underflow_ms=(
                round(segment_health[3] / segment_health[2])
                if segment_health is not None and segment_health[2]
                else (0 if segment_health is not None else None)
            ),
        )

    async def _wait_for_tts_metric_event(self, speech_handle_id: str) -> None:
        wait_s = self._tts_metric_wait_s()
        if wait_s <= 0:
            return

        metric_event = self._tts_metric_events_by_speech_handle_id.setdefault(
            speech_handle_id,
            asyncio.Event(),
        )
        try:
            await asyncio.wait_for(metric_event.wait(), timeout=wait_s)
        except asyncio.TimeoutError:
            pass
        finally:
            if self._tts_metric_events_by_speech_handle_id.get(speech_handle_id) is metric_event:
                self._tts_metric_events_by_speech_handle_id.pop(speech_handle_id, None)

    def _has_tts_metric_for_speech_handle_id(self, speech_handle_id: str) -> bool:
        return any(
            metrics.get(speech_handle_id) is not None
            for metrics in (
                self._tts_tffb_ms_by_speech_handle_id,
                self._tts_duration_ms_by_speech_handle_id,
                self._tts_audio_duration_ms_by_speech_handle_id,
            )
        )

    @staticmethod
    def _tts_metric_wait_s() -> float:
        try:
            wait_ms = int(str(os.getenv("TTS_TTFB_METRIC_WAIT_MS", "250") or "250"))
        except ValueError:
            wait_ms = 250
        return max(0.0, wait_ms / 1000.0)

    @staticmethod
    def _float_env_s(env_name: str, default: float) -> float:
        try:
            timeout_s = float(str(os.getenv(env_name, str(default)) or str(default)))
        except ValueError:
            timeout_s = default
        return max(0.0, timeout_s)

    @classmethod
    def _tts_first_frame_timeout_s(cls) -> float:
        if os.getenv("TTS_FIRST_FRAME_TIMEOUT_S") is not None:
            return cls._float_env_s("TTS_FIRST_FRAME_TIMEOUT_S", 3.0)
        return cls._float_env_s("TTS_EMPTY_FRAME_RETRY_TIMEOUT_S", 3.0)

    @classmethod
    def _tts_connection_start_grace_s(cls) -> float:
        return cls._float_env_s("TTS_CONNECTION_START_GRACE_TIMEOUT_S", 3.0)

    @classmethod
    def _tts_playout_start_timeout_s(cls) -> float:
        if os.getenv("TTS_PLAYOUT_START_TIMEOUT_S") is not None:
            return cls._float_env_s("TTS_PLAYOUT_START_TIMEOUT_S", 6.0)
        return cls._tts_first_frame_timeout_s() + cls._tts_connection_start_grace_s()

    @classmethod
    def _tts_empty_frame_retry_timeout_s(cls) -> float:
        return cls._tts_playout_start_timeout_s()

    @staticmethod
    def _exception_chain_text(exc: Exception) -> str:
        parts: list[str] = []
        current: BaseException | None = exc
        seen: set[int] = set()
        while current is not None and id(current) not in seen:
            seen.add(id(current))
            parts.append(type(current).__name__)
            parts.append(str(current))
            current = current.__cause__ or current.__context__
        return " ".join(parts).lower()

    @staticmethod
    def _tts_failure_metric(detail: str | None, key: str) -> int | None:
        match = re.search(rf"(?:^|;\s*){re.escape(key)}=(\d+)(?:;|$)", str(detail or ""))
        return int(match.group(1)) if match else None

    @classmethod
    def _is_retryable_tts_playout_error(cls, exc: Exception) -> bool:
        return cls._is_retryable_tts_error_text(cls._exception_chain_text(exc))

    @staticmethod
    def _retryable_tts_error_markers() -> tuple[str, ...]:
        return (
            "audio frame gap",
            "no audio frames",
            "empty frame",
            "apitimeouterror",
            "request timed out",
            "xai tts partial audio failure",
            "xai tts total timeout",
            "xai tts failed before audio",
            "failed to connect to xai tts",
        )

    @classmethod
    def _is_retryable_tts_error_text(cls, details: str) -> bool:
        details = str(details or "").lower()
        return any(marker in details for marker in cls._retryable_tts_error_markers())

    @classmethod
    def _retryable_tts_error_detail(cls, error: Any) -> str | None:
        underlying = getattr(error, "error", error)
        if isinstance(underlying, BaseException):
            details = cls._exception_chain_text(underlying)
        else:
            details = f"{type(underlying).__name__} {underlying}".lower()
        if not cls._is_retryable_tts_error_text(details):
            return None
        return details[:500]

    def _record_retryable_tts_session_error(self, error: Any) -> None:
        error_type = str(getattr(error, "type", "") or "").strip().lower()
        if error_type != "tts_error":
            return

        detail = self._retryable_tts_error_detail(error)
        if detail is None:
            return

        speech_handle_id = self._speech_handle_id(self.state.current_speech.handle)
        if not speech_handle_id:
            return

        self._retryable_tts_errors_by_speech_handle_id[speech_handle_id] = detail
        log_flow_event(
            self._call_logger,
            "tts_retryable_error_observed",
            stage=self.state.current_speech.stage or self.state.current_stage,
            speech_id=self.state.current_speech.speech_id,
            speech_handle_id=speech_handle_id,
            recoverable=bool(getattr(error, "recoverable", False)),
            detail=detail,
        )

    def _pop_retryable_tts_error_detail(self, handle: Any) -> str | None:
        speech_handle_id = self._speech_handle_id(handle)
        if not speech_handle_id:
            return None
        return self._retryable_tts_errors_by_speech_handle_id.pop(speech_handle_id, None)

    def _agent_speaking_seen_after(
        self,
        *,
        started_monotonic: float,
        seq_at_start: int,
    ) -> bool:
        session_state = str(getattr(self._session, "agent_state", "") or "").lower()
        session_state_name = session_state.rsplit(".", 1)[-1]
        if session_state_name == "speaking" and self._agent_speaking_at >= started_monotonic:
            return True
        return (
            self._agent_speaking_seq > seq_at_start
            and self._agent_speaking_at >= started_monotonic
        )

    async def _wait_for_agent_speaking_after(
        self,
        *,
        started_monotonic: float,
        seq_at_start: int,
        timeout_s: float,
    ) -> bool:
        deadline = time.monotonic() + timeout_s
        while True:
            if self._agent_speaking_seen_after(
                started_monotonic=started_monotonic,
                seq_at_start=seq_at_start,
            ):
                return True

            remaining_s = deadline - time.monotonic()
            if remaining_s <= 0:
                return False

            self._agent_speaking_event.clear()
            try:
                await asyncio.wait_for(
                    self._agent_speaking_event.wait(),
                    timeout=min(remaining_s, 0.05),
                )
            except asyncio.TimeoutError:
                pass

    async def _wait_for_tts_playout_or_empty_frame_retry(
        self,
        handle: Any,
        *,
        started_monotonic: float,
        agent_speaking_seq_at_start: int,
        timeout_s: float,
        stage: str,
        message_id: str,
    ) -> bool:
        if timeout_s <= 0:
            await self.execute(WaitForSpeechPlayout(handle))
            return False

        playout_task = asyncio.create_task(self.execute(WaitForSpeechPlayout(handle)))
        speaking_task = asyncio.create_task(
            self._wait_for_agent_speaking_after(
                started_monotonic=started_monotonic,
                seq_at_start=agent_speaking_seq_at_start,
                timeout_s=timeout_s,
            )
        )
        try:
            done, _ = await asyncio.wait(
                {playout_task, speaking_task},
                return_when=asyncio.FIRST_COMPLETED,
            )
            if speaking_task in done:
                first_audio_seen = await speaking_task
                if first_audio_seen:
                    await playout_task
                    return False

                if playout_task.done() and playout_task.exception() is not None:
                    await playout_task

            elif playout_task in done:
                await playout_task
                return False

            speech_handle_id = self._speech_handle_id(handle)
            log_flow_event(
                self._call_logger,
                "tts_empty_frame_retry",
                stage=stage,
                message_id=message_id,
                speech_handle_id=speech_handle_id,
                timeout_ms=round(timeout_s * 1000),
                timeout_kind="playout_start",
                first_frame_timeout_ms=round(self._tts_first_frame_timeout_s() * 1000),
                connection_grace_ms=round(self._tts_connection_start_grace_s() * 1000),
                action="resend_text",
            )
            try:
                await self.execute(InterruptSpeech(handle, force=True))
            except Exception as exc:
                log_flow_event(
                    self._call_logger,
                    "tts_empty_frame_retry_interrupt_error",
                    stage=stage,
                    message_id=message_id,
                    speech_handle_id=speech_handle_id,
                    error=type(exc).__name__,
                )

            playout_task.cancel()
            await asyncio.gather(playout_task, return_exceptions=True)
            return True
        finally:
            if not speaking_task.done():
                speaking_task.cancel()
                await asyncio.gather(speaking_task, return_exceptions=True)

    async def _resolve_tts_tffb_ms(
        self,
        handle: Any,
    ) -> int | None:
        speech_handle_id = self._speech_handle_id(handle)
        if not speech_handle_id:
            return None

        ttfb_ms = self._tts_tffb_ms_by_speech_handle_id.get(speech_handle_id)
        if ttfb_ms is not None and ttfb_ms > 0:
            return ttfb_ms

        if self._has_tts_metric_for_speech_handle_id(speech_handle_id):
            return None

        await self._wait_for_tts_metric_event(speech_handle_id)
        ttfb_ms = self._tts_tffb_ms_by_speech_handle_id.get(speech_handle_id)
        if ttfb_ms is not None and ttfb_ms > 0:
            return ttfb_ms

        return None

    def _resolve_tts_provider_duration_ms(
        self,
        handle: Any,
    ) -> int | None:
        speech_handle_id = self._speech_handle_id(handle)
        if not speech_handle_id:
            return None

        duration_ms = self._tts_duration_ms_by_speech_handle_id.get(speech_handle_id)
        if duration_ms is not None and duration_ms > 0:
            return duration_ms

        return None

    def _resolve_tts_structured_metrics(
        self,
        handle: Any,
    ) -> tuple[int | None, int | None, int | None, int | None, str]:
        speech_handle_id = self._speech_handle_id(handle)
        if not speech_handle_id:
            return None, None, None, None, "livekit_tts_metrics_missing"

        ttfb_ms = self._tts_tffb_ms_by_speech_handle_id.get(speech_handle_id)
        metric_duration_ms = self._tts_duration_ms_by_speech_handle_id.get(speech_handle_id)
        audio_duration_ms = self._tts_audio_duration_ms_by_speech_handle_id.get(speech_handle_id)
        has_livekit_metric = any(
            value is not None and value > 0
            for value in (ttfb_ms, metric_duration_ms, audio_duration_ms)
        )
        metrics_source = (
            "livekit_tts_metrics" if has_livekit_metric else "livekit_tts_metrics_missing"
        )
        total_ms = metric_duration_ms
        return total_ms, ttfb_ms, audio_duration_ms, metric_duration_ms, metrics_source

    def _resolve_tts_audio_duration_ms(
        self,
        handle: Any,
    ) -> int | None:
        speech_handle_id = self._speech_handle_id(handle)
        if speech_handle_id:
            audio_ms = self._tts_audio_duration_ms_by_speech_handle_id.get(speech_handle_id)
            if audio_ms is not None and audio_ms > 0:
                return audio_ms

        return None

    def _mark_structured_interruption(self, *, reason: str = "") -> None:
        self._structured_interruption_pending = True
        log_flow_event(
            self._call_logger,
            "structured_interruption_marked",
            reason=reason,
            stage=self.state.current_speech.stage or self.state.current_stage,
        )

    def _clear_structured_interruption(self) -> None:
        self._structured_interruption_pending = False

    def structured_recebimento_interruption_flag(self) -> bool:
        return bool(
            self._structured_interruption_pending
            or (self.speaking.is_set() and self._current_speech_allows_interruption())
        )

    def reset_gap_state(self) -> None:
        self.state.gap.active = False
        self.state.gap.guard_seq = 0
        self.state.gap.stage = ""
        self.state.gap.text = ""
        self.state.gap.speech_id = ""
        self.state.gap.cancelled = False

    @staticmethod
    def _metadata_bool(metadata: Any, key: str) -> Optional[bool]:
        if not isinstance(metadata, Mapping) or key not in metadata:
            return None

        value = metadata.get(key)
        if isinstance(value, bool):
            return value
        if isinstance(value, str):
            normalized = value.strip().lower()
            if normalized in {"1", "true", "yes", "on"}:
                return True
            if normalized in {"0", "false", "no", "off"}:
                return False
        return None

    @staticmethod
    def _metadata_str(metadata: Any, key: str) -> str:
        if not isinstance(metadata, Mapping):
            return ""
        return str(metadata.get(key) or "").strip()

    @classmethod
    def _reply_expects_user_response(cls, reply: BackendReply) -> Optional[bool]:
        return cls._metadata_bool(reply.metadata, "expects_user_response")

    @classmethod
    def _reply_drops_user_input_while_speaking(cls, reply: BackendReply) -> bool:
        return bool(cls._metadata_bool(reply.metadata, "drop_user_input_while_speaking"))

    @classmethod
    def _reply_agent_message_type(cls, reply: BackendReply) -> str:
        return cls._metadata_str(reply.metadata, "agent_message_type")

    @classmethod
    def _reply_agent_result_type(cls, reply: BackendReply) -> str:
        return cls._metadata_str(reply.metadata, "agent_result_type")

    def _protected_speech_drop_context_from_values(
        self,
        *,
        reason: str,
        stage: str,
        allow_interruptions: bool,
        drop_user_input_while_speaking: bool,
        agent_message_type: str,
        agent_result_type: str,
    ) -> dict[str, Any] | None:
        stage = stage or self.state.current_stage
        protected_by_metadata = bool(drop_user_input_while_speaking)
        protected_presentation = (
            str(stage or "").strip().upper() == "PRESENTATION"
            and not allow_interruptions
        )
        if not protected_by_metadata and not protected_presentation:
            return None

        agent_message_type = str(agent_message_type or "").strip()
        if not agent_message_type and protected_presentation:
            agent_message_type = "ready" if self.state.user_final_seq == 0 else "presentation"

        return {
            "reason": reason,
            "stage": stage,
            "agent_message_type": agent_message_type,
            "agent_result_type": str(agent_result_type or "").strip(),
        }

    def _current_speech_drop_context(self, *, reason: str) -> dict[str, Any] | None:
        current = self.state.current_speech
        if not self.speaking.is_set():
            return None
        return self._protected_speech_drop_context_from_values(
            reason=reason,
            stage=current.stage or self.state.current_stage,
            allow_interruptions=current.allow_interruptions,
            drop_user_input_while_speaking=current.drop_user_input_while_speaking,
            agent_message_type=current.agent_message_type,
            agent_result_type=current.agent_result_type,
        )

    def _mark_drop_next_user_final(self, context: Mapping[str, Any]) -> bool:
        self._drop_next_user_final_context = dict(context)
        self._call_logger.info(
            "USER_INPUT_DROP_ARMED | reason=%s | stage=%s | agent_message_type=%s | agent_result_type=%s",
            context.get("reason") or "-",
            context.get("stage") or "-",
            context.get("agent_message_type") or "-",
            context.get("agent_result_type") or "-",
        )
        log_flow_event(
            self._call_logger,
            "user_input_drop_armed",
            reason=context.get("reason") or "",
            stage=context.get("stage") or "",
            agent_message_type=context.get("agent_message_type") or "",
            agent_result_type=context.get("agent_result_type") or "",
        )
        if self._timeline is not None:
            self._timeline.emit("user_input_drop_armed", **context)
        return True

    def _mark_drop_next_user_final_from_current_speech(self, *, reason: str) -> bool:
        context = self._current_speech_drop_context(reason=reason)
        if context is None:
            return False
        return self._mark_drop_next_user_final(context)

    def _active_post_playout_drop_context(self, *, reason: str) -> dict[str, Any] | None:
        context = self._post_playout_drop_context
        if context is None:
            return None
        if time.monotonic() >= self._post_playout_drop_expires_at:
            self._clear_post_playout_drop_window(reason="expired")
            return None
        active_context = dict(context)
        active_context["reason"] = reason
        return active_context

    def _mark_drop_next_user_final_from_post_playout_window(self, *, reason: str) -> bool:
        context = self._active_post_playout_drop_context(reason=reason)
        if context is None:
            return False
        return self._mark_drop_next_user_final(context)

    def _arm_post_playout_drop_window(self, context: Mapping[str, Any]) -> None:
        grace_s = PROTECTED_SPEECH_POST_PLAYOUT_DROP_GRACE_S
        if grace_s <= 0:
            return
        self._post_playout_drop_context = dict(context)
        self._post_playout_drop_expires_at = time.monotonic() + grace_s
        self._call_logger.info(
            "USER_INPUT_DROP_GRACE_ARMED | reason=%s | stage=%s | agent_message_type=%s | agent_result_type=%s | grace_ms=%s",
            context.get("reason") or "-",
            context.get("stage") or "-",
            context.get("agent_message_type") or "-",
            context.get("agent_result_type") or "-",
            round(grace_s * 1000),
        )
        log_flow_event(
            self._call_logger,
            "user_input_drop_grace_armed",
            reason=context.get("reason") or "",
            stage=context.get("stage") or "",
            agent_message_type=context.get("agent_message_type") or "",
            agent_result_type=context.get("agent_result_type") or "",
            grace_ms=round(grace_s * 1000),
        )
        if self._timeline is not None:
            self._timeline.emit(
                "user_input_drop_grace_armed",
                **dict(context),
                grace_ms=round(grace_s * 1000),
            )

    def _clear_post_playout_drop_window(self, *, reason: str = "clear") -> None:
        if self._post_playout_drop_context is None:
            return
        self._post_playout_drop_context = None
        self._post_playout_drop_expires_at = 0.0
        log_flow_event(
            self._call_logger,
            "user_input_drop_grace_cleared",
            reason=reason,
        )

    def _clear_pending_user_input_drop(self) -> None:
        self._drop_next_user_final_context = None

    def _backend_processing_drop_context_from_reply(
        self,
        reply: BackendReply,
        *,
        reason: str,
    ) -> dict[str, Any]:
        return {
            "reason": reason,
            "stage": reply.stage or self.state.current_stage,
            "agent_message_type": self._reply_agent_message_type(reply),
            "agent_result_type": self._reply_agent_result_type(reply),
        }

    def _update_backend_processing_drop_from_reply(
        self,
        reply: BackendReply,
        *,
        source: str,
    ) -> None:
        if (
            self._reply_agent_message_type(reply) == "feedback"
            or self._reply_agent_result_type(reply) == "feedback"
        ):
            self._backend_processing_drop_context = self._backend_processing_drop_context_from_reply(
                reply,
                reason=f"backend_processing_feedback:{source}",
            )
            return

        self._backend_processing_drop_context = None

    def _drop_context_for_user_final(self) -> dict[str, Any] | None:
        if self.finalized.is_set():
            return {
                "reason": "finalized",
                "stage": self.state.current_stage,
                "agent_message_type": "",
                "agent_result_type": "",
            }

        context = self._current_speech_drop_context(reason="protected_speech_active")
        if context is not None:
            self._clear_pending_user_input_drop()
            return context

        if self._backend_processing_drop_context is not None:
            return dict(self._backend_processing_drop_context)

        if self._drop_next_user_final_context is not None:
            context = dict(self._drop_next_user_final_context)
            self._clear_pending_user_input_drop()
            return context

        context = self._active_post_playout_drop_context(
            reason="protected_speech_post_playout_grace"
        )
        if context is not None:
            self._clear_post_playout_drop_window(reason="user_final_dropped")
            return context

        return None

    def _log_user_input_dropped(
        self,
        *,
        context: Mapping[str, Any],
        transcription: str,
        text: str,
    ) -> None:
        reason = str(context.get("reason") or "protected_speech").strip()
        stage = str(context.get("stage") or self.state.current_stage or "").strip()
        agent_message_type = str(context.get("agent_message_type") or "").strip()
        agent_result_type = str(context.get("agent_result_type") or "").strip()
        self._call_logger.info(
            "USER_INPUT_DROPPED | reason=%s | stage=%s | agent_message_type=%s | agent_result_type=%s | text=%r",
            reason or "-",
            stage or "-",
            agent_message_type or "-",
            agent_result_type or "-",
            text,
        )
        log_flow_event(
            self._call_logger,
            "user_input_dropped",
            reason=reason,
            stage=stage,
            agent_message_type=agent_message_type,
            agent_result_type=agent_result_type,
            text=text,
            transcript_len=len(transcription or ""),
        )
        if self._timeline is not None:
            self._timeline.emit(
                "user_input_dropped",
                reason=reason,
                stage=stage,
                agent_message_type=agent_message_type,
                agent_result_type=agent_result_type,
                text=text,
                transcript_len=len(transcription or ""),
            )

    @staticmethod
    def _speech_id_from_reply(reply: BackendReply) -> str:
        metadata = reply.metadata
        if not isinstance(metadata, Mapping):
            return ""
        return str(metadata.get("speech_id") or "").strip()

    @staticmethod
    def _message_id_from_reply(reply: BackendReply) -> str:
        metadata = reply.metadata
        if not isinstance(metadata, Mapping):
            return ""

        for key in ("message_id", "messageId"):
            value = metadata.get(key)
            if value not in (None, ""):
                return str(value).strip()

        payload = metadata.get("payload")
        if isinstance(payload, Mapping):
            for key in ("message_id", "messageId"):
                value = payload.get(key)
                if value not in (None, ""):
                    return str(value).strip()

        return ""

    @staticmethod
    def _backend_reply_with_message_id(reply: BackendReply, message_id: str) -> BackendReply:
        message_id = str(message_id or "").strip()
        if not message_id:
            return reply

        metadata = dict(reply.metadata) if isinstance(reply.metadata, Mapping) else {}
        metadata["message_id"] = message_id
        return BackendReply(
            stage=reply.stage,
            text=reply.text,
            done=reply.done,
            export_payload=reply.export_payload,
            metadata=metadata,
        )

    @staticmethod
    def _payload_with_message_id(payload: Any, message_id: str, *, text: str = "") -> Any:
        message_id = str(message_id or "").strip()
        if not message_id:
            return payload

        if isinstance(payload, Mapping):
            enriched = dict(payload)
            enriched["message_id"] = message_id
            if text and not any(
                key in enriched
                for key in ("text", "transcript", "utterance", "message", "content")
            ):
                enriched["text"] = text
            return enriched

        return {
            "text": str(text or payload or "").strip(),
            "message_id": message_id,
        }

    def _consume_turn_message_id(self, transcription: str, text: str) -> str:
        if not text:
            return ""

        turn = consume_transcribed_turn(
            self._structured_log_context,
            transcription=transcription,
            text=text,
        )
        if turn is None:
            return ""
        clear_started_turn_message_id(
            self._structured_log_context,
            message_id=turn.message_id,
        )
        return turn.message_id

    @staticmethod
    def _combined_interruption_transcription(turns: list[PendingUserTurn]) -> tuple[str, str]:
        combined_text = ". ".join(
            turn.text.strip().rstrip(".") for turn in turns if turn.text.strip()
        ).strip()
        if not turns:
            return "", combined_text
        try:
            payload = json.loads(turns[0].transcription)
        except Exception:
            return combined_text, combined_text
        if not isinstance(payload, dict):
            return combined_text, combined_text
        data = payload.get("data")
        if isinstance(data, dict):
            payload = dict(payload)
            payload["data"] = dict(data)
            payload["data"]["text"] = combined_text
            payload["data"].pop("words", None)
        elif "text" in payload:
            payload = dict(payload)
            payload["text"] = combined_text
        else:
            payload = dict(payload)
            payload["text"] = combined_text
        return json.dumps(payload, ensure_ascii=False), combined_text

    def _prune_pending_vad_utterances(self) -> None:
        deferred = self.state.deferred_interruption
        now = time.monotonic()
        queues = (
            ("pre_backend", deferred.pre_backend_vad_utterances),
            ("backend", deferred.pending_vad_utterances),
        )
        for queue_name, queue in queues:
            expired = 0
            capacity_dropped = 0
            while (
                queue
                and queue[0].ended_at
                and now - queue[0].ended_at > PENDING_VAD_STT_MAX_AGE_S
            ):
                utterance = queue.popleft()
                expired += 1
                if queue_name == "backend" and utterance.is_long:
                    deferred.pending_long_stt_finals = max(
                        0, deferred.pending_long_stt_finals - 1
                    )
            while len(queue) > PENDING_VAD_STT_MAX_UTTERANCES:
                utterance = queue.popleft()
                capacity_dropped += 1
                if queue_name == "backend" and utterance.is_long:
                    deferred.pending_long_stt_finals = max(
                        0, deferred.pending_long_stt_finals - 1
                    )
            if expired:
                log_flow_event(
                    self._call_logger,
                    "deferred_interruption_vad_expired",
                    queue=queue_name,
                    discarded_utterances=expired,
                    max_age_s=PENDING_VAD_STT_MAX_AGE_S,
                )
            if capacity_dropped:
                log_flow_event(
                    self._call_logger,
                    "deferred_interruption_vad_capacity_shed",
                    queue=queue_name,
                    discarded_utterances=capacity_dropped,
                    max_utterances=PENDING_VAD_STT_MAX_UTTERANCES,
                )
        if not deferred.pending_long_stt_finals:
            self._deferred_long_stt_settled.set()

    def _promote_pre_backend_vad_utterances(self) -> None:
        deferred = self.state.deferred_interruption
        self._prune_pending_vad_utterances()
        if not deferred.pre_backend_vad_utterances:
            return
        promoted = list(deferred.pre_backend_vad_utterances)
        deferred.pre_backend_vad_utterances.clear()
        deferred.pending_vad_utterances.extend(promoted)
        long_count = sum(1 for utterance in promoted if utterance.is_long)
        if not long_count:
            return
        if long_count:
            deferred.pending_long_stt_finals += long_count
            self._deferred_long_stt_settled.clear()
            if not deferred.special_comfort_sent:
                deferred.special_comfort_sent = True
                self._record_inflight_backend_activity()
                self._play_deferred_interruption_comfort()
        log_flow_event(
            self._call_logger,
            "deferred_interruption_vad_promoted",
            promoted_utterances=len(promoted),
            promoted_long_stt_finals=long_count,
        )

    def _consume_vad_for_stt_final(self) -> DeferredVADUtterance | None:
        deferred = self.state.deferred_interruption
        self._prune_pending_vad_utterances()
        queue = (
            deferred.pending_vad_utterances
            if deferred.backend_in_flight or deferred.replay_in_flight
            else deferred.pre_backend_vad_utterances
        )
        if not queue:
            return None
        utterance = queue.popleft()
        if deferred.backend_in_flight and utterance.is_long:
            deferred.pending_long_stt_finals = max(0, deferred.pending_long_stt_finals - 1)
            if not deferred.pending_long_stt_finals:
                self._deferred_long_stt_settled.set()
        return utterance

    def _reset_deferred_interruption(self) -> None:
        deferred = self.state.deferred_interruption
        deferred.backend_in_flight = False
        deferred.pre_backend_vad_utterances.clear()
        deferred.pending_vad_utterances.clear()
        deferred.pending_long_stt_finals = 0
        deferred.long_turns.clear()
        deferred.special_comfort_sent = False
        deferred.replay_in_flight = False
        deferred.cycle_seq += 1
        self._deferred_empty_final_latched_at = 0.0
        self._deferred_long_stt_settled.set()

    def _release_deferred_cycle(self, *, is_deferred_replay: bool) -> None:
        """Fecha o ciclo diferido ao encerrar um turno do pipeline.

        R2 sempre limpa: e o fim do reenvio unico. R1 so limpa se chegou a abrir
        o ciclo, e nunca quando acabou de despachar R2 -- ai a marca de reenvio
        pertence ao turno seguinte, nao a este.
        """
        deferred = self.state.deferred_interruption
        if is_deferred_replay or (
            deferred.special_comfort_sent and not deferred.replay_in_flight
        ):
            self._reset_deferred_interruption()

    def _play_deferred_interruption_comfort(self, *, message_id: str = "") -> None:
        deferred = self.state.deferred_interruption
        cycle_seq = deferred.cycle_seq

        async def _play() -> None:
            # O audio so entra no silencio do cliente. A espera tem teto porque
            # um estado de fala travado deixaria a tarefa pendurada ate o fim da
            # ligacao.
            if not await self._wait_event_or_finalized(
                self._user_not_speaking,
                timeout=max(
                    0.0,
                    float(self._config.deferred_interruption_user_turn_timeout_s),
                ),
            ):
                log_flow_event(
                    self._call_logger,
                    "deferred_interruption_comfort_skipped",
                    reason="user_still_speaking",
                )
                return
            deferred = self.state.deferred_interruption
            if (
                self.finalized.is_set()
                or deferred.cycle_seq != cycle_seq
                or deferred.replay_in_flight
                or not deferred.special_comfort_sent
            ):
                return
            base_message_id = str(message_id or "").strip() or self._current_base_message_id()
            comfort_message_id = self._interruption_comfort_message_id(
                message_id=base_message_id,
            )
            audio_duration_ms = wav_duration_ms(str(DEFERRED_INTERRUPTION_COMFORT_AUDIO_PATH))
            await self.say_stage(
                DEFERRED_INTERRUPTION_COMFORT_TEXT,
                "INTERRUPTION_COMFORT",
                # Comeca protegido contra o STT final da fala que originou este
                # conforto e volta a ser interrompivel assim que esse final chega.
                allow_interruptions=False,
                rearm_interruptions_on_next_user_speech=True,
                add_to_chat_ctx=False,
                schedule_idle_after=False,
                require_user_not_speaking=True,
                message_id=comfort_message_id,
                audio=wav_audio_frames(str(DEFERRED_INTERRUPTION_COMFORT_AUDIO_PATH)),
                audio_duration_ms=audio_duration_ms,
            )

        self._create_task_logged(
            _play(),
            name="deferred_interruption_comfort",
        )

    def _play_deferred_short_comfort(self, *, source: str) -> None:
        deferred = self.state.deferred_interruption
        cycle_seq = deferred.cycle_seq
        replay_in_flight = deferred.replay_in_flight
        log_flow_event(
            self._call_logger,
            "deferred_short_comfort_scheduled",
            source=source,
            replay_in_flight=replay_in_flight,
        )

        async def _play() -> None:
            current = self.state.deferred_interruption
            if (
                self.finalized.is_set()
                or current.cycle_seq != cycle_seq
                or current.replay_in_flight != replay_in_flight
            ):
                return
            await self._play_inflight_backend_wait_notice_audio(
                user_seq=None,
                attempt=1,
                require_user_not_speaking=True,
            )

        self._create_task_logged(
            _play(),
            name=f"deferred_short_comfort_{source}",
        )

    async def _dispatch_deferred_interruption(self, reply: BackendReply) -> bool:
        deferred = self.state.deferred_interruption
        if deferred.replay_in_flight or not deferred.long_turns:
            return False
        turns = list(deferred.long_turns)
        transcription, text = self._combined_interruption_transcription(turns)
        turn = turns[-1]
        speech_id = self._speech_id_from_reply(reply)
        # O reenvio e unico: a marca precisa estar ativa antes de R2 ser
        # agendado para que nenhum evento concorrente do cliente abra R3.
        deferred.pending_vad_utterances.clear()
        deferred.pending_long_stt_finals = 0
        deferred.long_turns.clear()
        deferred.replay_in_flight = True
        self._deferred_long_stt_settled.set()
        await self.execute(
            SetPendingInterrupt(listened_text="", skipped=True, speech_id=speech_id)
        )
        log_flow_event(
            self._call_logger,
            "deferred_interruption_dispatched",
            user_seq=turn.seq,
            speech_id=speech_id,
            accumulated_turns=len(turns),
            text=text,
        )
        self._create_task_logged(
            self.run_pipeline(
                transcription,
                text,
                is_deferred_replay=True,
                user_seq=turn.seq,
                message_id=turn.message_id,
                inflight_initial_notice_consumed=True,
            ),
            name=f"run_pipeline_deferred_interruption_{turn.seq}",
        )
        return True

    def note_vad_speech_end(self, speech_duration_ms: int) -> None:
        duration_ms = max(0, int(speech_duration_ms or 0))
        self._last_vad_speech_end_at = time.monotonic()
        self._last_vad_speech_duration_ms = duration_ms
        self._emit_debug_event(
            "user.speech.finished",
            speech_duration_ms=duration_ms,
        )
        deferred = self.state.deferred_interruption
        minimum_audio_ms = max(0, int(self._config.deferred_interruption_min_audio_ms))
        is_long = duration_ms >= minimum_audio_ms
        self._prune_pending_vad_utterances()

        if deferred.replay_in_flight:
            log_flow_event(
                self._call_logger,
                "deferred_replay_vad_discarded",
                speech_duration_ms=duration_ms,
                is_long=is_long,
                minimum_audio_ms=minimum_audio_ms,
                short_comfort_scheduled=False,
            )
            return

        utterance = DeferredVADUtterance(
            duration_ms=duration_ms,
            is_long=is_long,
            ended_at=time.monotonic(),
        )
        if not deferred.backend_in_flight:
            if not self._config.deferred_interruption_enabled:
                return
            # O STT pode terminar depois que R1 for aberto. Guardamos o fim de
            # fala para preservar a associacao VAD/STT nessa corrida.
            deferred.pre_backend_vad_utterances.append(utterance)
            log_flow_event(
                self._call_logger,
                "deferred_interruption_vad_pending_pre_backend",
                speech_duration_ms=duration_ms,
                is_long=is_long,
                minimum_audio_ms=minimum_audio_ms,
            )
            return

        if not self._config.deferred_interruption_enabled:
            log_flow_event(
                self._call_logger,
                "backend_processing_vad_discarded",
                speech_duration_ms=duration_ms,
                minimum_audio_ms=minimum_audio_ms,
                mode="feedback_compat",
            )
            return

        deferred.pending_vad_utterances.append(utterance)
        if is_long:
            # Only utterances accepted by the business duration rule start a
            # new interaction cycle. Short utterances may temporarily suppress
            # playout while the customer is speaking, but must not postpone the
            # periodic notice for the request already in flight.
            self._record_inflight_backend_activity()
            deferred.pending_long_stt_finals += 1
            self._deferred_long_stt_settled.clear()
            if not deferred.special_comfort_sent:
                deferred.special_comfort_sent = True
                self._play_deferred_interruption_comfort()
            else:
                self._play_deferred_short_comfort(source="repeated_interruption")

        log_flow_event(
            self._call_logger,
            "deferred_interruption_vad_end",
            speech_duration_ms=duration_ms,
            is_long=is_long,
            minimum_audio_ms=minimum_audio_ms,
        )

    def _close_deferred_vad_window(self) -> None:
        """Descarta as falas VAD que nao receberam final dentro do ciclo.

        A associacao VAD/STT e por ordem de chegada. Uma fala sem final que fique
        na fila seria consumida pelo final de outra fala no turno seguinte, e a
        partir dai todo o pareamento anda deslocado.
        """
        deferred = self.state.deferred_interruption
        deferred.backend_in_flight = False
        discarded_backend = len(deferred.pending_vad_utterances)
        discarded_pre_backend = len(deferred.pre_backend_vad_utterances)
        if discarded_backend or discarded_pre_backend:
            log_flow_event(
                self._call_logger,
                "deferred_interruption_vad_window_closed",
                discarded_utterances=discarded_backend + discarded_pre_backend,
                discarded_backend_utterances=discarded_backend,
                discarded_pre_backend_utterances=discarded_pre_backend,
                pending_long_stt_finals=deferred.pending_long_stt_finals,
            )
        deferred.pre_backend_vad_utterances.clear()
        deferred.pending_vad_utterances.clear()
        deferred.pending_long_stt_finals = 0
        self._deferred_long_stt_settled.set()

    async def _wait_event_or_finalized(self, event: asyncio.Event, *, timeout: float) -> bool:
        """Espera o evento, mas desiste se a chamada terminar antes.

        Sem a corrida com `finalized` um desligamento no meio da espera ficaria
        parado ate o teto do timeout.
        """
        if event.is_set():
            return True
        waiters = [
            asyncio.ensure_future(event.wait()),
            asyncio.ensure_future(self.finalized.wait()),
        ]
        try:
            await asyncio.wait(
                waiters, timeout=timeout, return_when=asyncio.FIRST_COMPLETED
            )
        finally:
            for waiter in waiters:
                if not waiter.done():
                    waiter.cancel()
            await asyncio.gather(*waiters, return_exceptions=True)
        return event.is_set()

    async def _wait_deferred_interruption_settlement(self, *, user_seq: int) -> None:
        """Segura a resposta de R1 ate o turno do cliente fechar.

        Com o VAD aberto nao ha o que decidir: a fala em curso ainda pode virar
        mais uma transcricao do acumulado. Depois do END_OF_SPEECH resta esperar
        o STT correspondente, e essa espera e curta e com teto -- a associacao
        VAD/STT e por ordem de chegada, entao um final que nunca vem (falha do
        provider, dois segmentos de VAD para um final so) prenderia o run_lock
        pelo resto da ligacao.
        """
        deferred = self.state.deferred_interruption
        user_turn_timeout_s = max(
            0.0, float(self._config.deferred_interruption_user_turn_timeout_s)
        )
        stt_settle_timeout_s = max(
            0.0, float(self._config.deferred_interruption_stt_settle_timeout_s)
        )
        user_turn_deadline = time.monotonic() + user_turn_timeout_s
        waited = False

        while not self.finalized.is_set():
            if not self._user_not_speaking.is_set():
                remaining = user_turn_deadline - time.monotonic()
                if remaining <= 0.0:
                    log_flow_event(
                        self._call_logger,
                        "deferred_interruption_settle_timeout",
                        reason="user_turn",
                        user_seq=user_seq,
                        timeout_s=user_turn_timeout_s,
                    )
                    break
                waited = True
                if not await self._wait_event_or_finalized(
                    self._user_not_speaking, timeout=remaining
                ):
                    log_flow_event(
                        self._call_logger,
                        "deferred_interruption_settle_timeout",
                        reason="user_turn",
                        user_seq=user_seq,
                        timeout_s=user_turn_timeout_s,
                    )
                    break
                # O END_OF_SPEECH do VAD e a mudanca de estado do usuario nao tem
                # ordem garantida; a folga evita liberar R1 no instante entre os
                # dois e perder a ultima fala.
                await asyncio.sleep(POST_USER_FINAL_SPEECH_GRACE_S)
                continue

            if not deferred.pending_long_stt_finals:
                break

            waited = True
            if not await self._wait_event_or_finalized(
                self._deferred_long_stt_settled, timeout=stt_settle_timeout_s
            ):
                log_flow_event(
                    self._call_logger,
                    "deferred_interruption_settle_timeout",
                    reason="stt_final",
                    user_seq=user_seq,
                    timeout_s=stt_settle_timeout_s,
                    pending_long_stt_finals=deferred.pending_long_stt_finals,
                )
                # A fila esta dessincronizada: um final atrasado consumiria a
                # fala errada mais adiante.
                deferred.pending_vad_utterances.clear()
                deferred.pending_long_stt_finals = 0
                self._deferred_long_stt_settled.set()
                break

        if waited:
            log_flow_event(
                self._call_logger,
                "deferred_interruption_settled",
                user_seq=user_seq,
                accumulated_turns=len(deferred.long_turns),
            )

    def _peek_started_turn_message_id(self) -> str:
        return peek_started_turn_message_id(self._structured_log_context)

    def _clear_started_turn_message_id(self, message_id: str = "") -> None:
        clear_started_turn_message_id(
            self._structured_log_context,
            message_id=message_id,
        )

    @staticmethod
    def _base_message_id(message_id: str) -> str:
        normalized_message_id = str(message_id or "").strip()
        if not normalized_message_id:
            return ""

        for marker in (
            "_conforto_",
            "_inatividade_",
            "_idle_nudge_",
            "_interruption_confort_",
            "_feedback_",
            "_tts_error_",
            "_transferencia_",
            "_erro_terminal_",
            "_erro_recurso_",
        ):
            base_message_id, separator, suffix = normalized_message_id.rpartition(marker)
            if separator and base_message_id and suffix:
                return base_message_id
        return normalized_message_id


    @staticmethod
    def _feedback_message_id(*, message_id: str = "", speech_id: str = "") -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        feedback_id = str(speech_id or "").strip() or uuid.uuid4().hex
        return f"{base}_feedback_{feedback_id}"

    @staticmethod
    def _idle_nudge_message_id(*, message_id: str = "", attempt: int) -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        return f"{base}_idle_nudge_{max(1, int(attempt or 1))}"

    @staticmethod
    def _interruption_comfort_message_id(*, message_id: str = "") -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        return f"{base}_interruption_confort_{uuid.uuid4().hex}"

    @staticmethod
    def _tts_error_message_id(*, message_id: str = "") -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        return f"{base}_tts_error_{uuid.uuid4().hex}"

    @staticmethod
    def _transfer_message_id(*, message_id: str = "") -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        return f"{base}_transferencia_{uuid.uuid4().hex}"

    @staticmethod
    def _resource_error_message_id(
        *,
        message_id: str = "",
        terminal: bool,
        tipo_evento: str,
    ) -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        family = "erro_terminal" if terminal else "erro_recurso"
        direction = "recebimento" if tipo_evento == EVENT_RECEBIMENTO_MSG else "envio"
        return f"{base}_{family}_{direction}_{uuid.uuid4().hex}"

    def _remember_agent_base_message_id(self, message_id: str) -> None:
        base_message_id = self._base_message_id(message_id)
        if base_message_id:
            self._last_agent_message_id = base_message_id

    def _current_base_message_id(self) -> str:
        return self._peek_started_turn_message_id() or self._last_agent_message_id

    def _resolve_agent_message_id(self, message_id: str = "", *, generate: bool = False) -> str:
        resolved_message_id = str(message_id or "").strip() or self._current_base_message_id()
        if not resolved_message_id and generate:
            resolved_message_id = next_turn_message_id(self._structured_log_context)
        self._remember_agent_base_message_id(resolved_message_id)
        return resolved_message_id

    @classmethod
    def _allow_interruptions_for_reply(
        cls,
        reply: BackendReply,
        default: bool,
    ) -> bool:
        explicit = cls._metadata_bool(reply.metadata, "is_interruptible")
        if explicit is None:
            return default
        return explicit

    def _current_speech_allows_interruption(self) -> bool:
        if not self.speaking.is_set():
            return False
        return bool(self.state.current_speech.allow_interruptions)

    def _rearm_current_speech_interruptions_on_new_user_speech(self) -> None:
        current = self.state.current_speech
        if (
            not self.speaking.is_set()
            or not current.rearm_interruptions_on_next_user_speech
        ):
            return

        handle = current.handle
        stage = current.stage
        if handle is not None and bool(getattr(handle, "interrupted", False)):
            return
        try:
            if handle is not None:
                handle.allow_interruptions = True
        except Exception:
            self._logger.exception(
                "[audio] failed rearming comfort interruptions | stage=%s",
                stage,
            )
            return

        current.rearm_interruptions_on_next_user_speech = False
        current.allow_interruptions = True
        log_flow_event(
            self._call_logger,
            "comfort_interruptions_rearmed",
            stage=stage,
            reason="new_user_speech",
        )

    @staticmethod
    def _status_for_resource(resource: str) -> str:
        normalized = str(resource or "").strip()
        return IN_SESSION_STOP_STATUS_BY_RESOURCE.get(normalized, "stop_bridge_failed")

    @staticmethod
    def _resource_from_session_error(error: Any) -> str:
        error_type = str(getattr(error, "type", "") or "").strip().lower()
        if error_type == "stt_error":
            return "stt"
        if error_type == "tts_error":
            return "tts"
        return "agent_runtime"

    @staticmethod
    def _payload_requests_transfer(payload: Any) -> bool:
        if isinstance(payload, Mapping):
            for key in ("type", "status", "result_type", "resultType"):
                value = str(payload.get(key) or "").strip().lower()
                if value in {"transferred", "transferido"}:
                    return True
            result = payload.get("result")
            if result is not payload and CallRuntime._payload_requests_transfer(result):
                return True
            return False

        if isinstance(payload, (list, tuple)):
            return any(CallRuntime._payload_requests_transfer(item) for item in payload)

        return False

    @classmethod
    def _error_message_from_backend_reply(cls, reply: BackendReply) -> str | None:
        if cls._payload_requests_transfer(reply.export_payload):
            return error_message_from_resource(
                status="transferred",
                tipo_evento=EVENT_ENVIO_MSG,
            )
        return None

    def _log_backend_reply_error_event(
        self,
        reply: BackendReply,
        *,
        source: str,
        message_id: str = "",
    ) -> None:
        transfer_status = ""
        if isinstance(reply.export_payload, Mapping):
            transfer_status = str(reply.export_payload.get("status") or "").strip()

        self._log_structured_error_event(
            tipo_evento=EVENT_ENVIO_MSG,
            erro_msg=self._error_message_from_backend_reply(reply),
            message_id=self._transfer_message_id(
                message_id=message_id or self._message_id_from_reply(reply)
            ),
            status=transfer_status,
            reason="agent_transfer",
            source=source,
        )

    def _log_structured_error_event(
        self,
        *,
        tipo_evento: str,
        erro_msg: str | None,
        erro_detalhe: str | None = None,
        message_id: str = "",
        status: str = "",
        reason: str = "",
        resource: str = "",
        source: str = "",
        http_cod_status: int | str | None = None,
        http_cod_desc: str | None = None,
    ) -> None:
        erro_msg = str(erro_msg or "").strip()
        if not erro_msg:
            return

        dedupe_key = (
            str(tipo_evento or ""),
            erro_msg,
            str(status or ""),
            str(reason or ""),
            str(resource or ""),
            str(message_id or ""),
        )
        if dedupe_key in self._structured_error_event_keys:
            return
        self._structured_error_event_keys.add(dedupe_key)

        now_ns = time.time_ns()
        log_structured_event(
            self._call_logger,
            self._structured_log_context,
            tipo_evento=tipo_evento,
            message_id=message_id,
            inicio_ns=now_ns,
            fim_ns=now_ns,
            latencia_total_ms=0,
            latencia_tffb_ms=0,
            erro_msg=erro_msg,
            erro_detalhe=erro_detalhe,
            http_cod_status=http_cod_status,
            http_cod_desc=http_cod_desc,
            finalizacao=status or None,
        )
        log_flow_event(
            self._call_logger,
            "structured_error",
            tipo_evento=tipo_evento,
            erro_msg=erro_msg,
            status=status,
            reason=reason,
            resource=resource,
            source=source,
        )

    def _log_resource_error_events(
        self,
        *,
        status: str = "",
        reason: str = "",
        resource: str = "",
        source: str = "",
        message_id: str = "",
        exc: Exception | None = None,
        terminal: bool = False,
    ) -> None:
        message_id = str(message_id or "").strip() or self._current_base_message_id()
        detail = f"{type(exc).__name__}: {exc}" if exc is not None else None
        for tipo_evento in (EVENT_RECEBIMENTO_MSG, EVENT_ENVIO_MSG):
            erro_msg = error_message_from_resource(
                resource=resource,
                status=status,
                reason=reason,
                tipo_evento=tipo_evento,
            )
            self._log_structured_error_event(
                tipo_evento=tipo_evento,
                erro_msg=erro_msg,
                erro_detalhe=detail,
                status=status,
                reason=reason,
                resource=resource,
                source=source,
                message_id=self._resource_error_message_id(
                    message_id=message_id,
                    terminal=terminal,
                    tipo_evento=tipo_evento,
                ),
            )

    async def notify_bridge_stop(
        self,
        *,
        status: str,
        reason: str,
        resource: str = "",
        failed_resources: tuple[str, ...] = (),
        phase: str = "in_session",
    ) -> None:
        try:
            await self.execute(
                NotifyBridgeStop(
                    status=status,
                    reason=reason,
                    resource=resource,
                    failed_resources=failed_resources,
                    phase=phase,
                )
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "bridge_stop_requested",
                    status=status,
                    reason=reason,
                    resource=resource,
                    failed_resources=list(failed_resources),
                    phase=phase,
                )
        except Exception as exc:
            self._log_resource_error_events(
                status="stop_bridge_failed",
                reason="bridge_notify_stop_failed",
                resource="bridge",
                source="notify_bridge_stop",
                exc=exc,
            )
            self._logger.exception("[bridge] Falha enviando STOP")

    async def terminate_with_resource_stop(
        self,
        *,
        resource: str,
        source: str,
        reason: str = "resource_unhealthy",
    ) -> None:
        async with self._terminal_stop_lock:
            if self._terminal_stop_sent:
                return
            self._terminal_stop_sent = True

        status = self._status_for_resource(resource)
        if self._timeline is not None:
            self._timeline.emit(
                "terminal_stop_started",
                status=status,
                reason=reason,
                resource=resource,
                source=source,
            )

        await self.notify_bridge_stop(
            status=status,
            reason=reason,
            resource=resource,
            failed_resources=(resource,) if resource else (),
            phase="in_session",
        )
        self._log_resource_error_events(
            status=status,
            reason=reason,
            resource=resource,
            source=source,
            terminal=True,
        )
        await self.finalize(status)

    async def _wait_post_user_final_grace(self, *, stage: str) -> None:
        last_user_final_at = float(self.state.last_user_final_at or 0.0)
        if last_user_final_at <= 0.0:
            return

        remaining_s = POST_USER_FINAL_SPEECH_GRACE_S - (time.monotonic() - last_user_final_at)
        if remaining_s <= 0.0:
            return

        self._call_logger.info(
            "BOT_SAY_WAIT_POST_FINAL | stage=%s | wait_ms=%s",
            stage,
            round(remaining_s * 1000),
        )
        if self._timeline is not None:
            self._timeline.emit(
                "tts_stage_wait_post_final",
                stage=stage,
                wait_ms=round(remaining_s * 1000),
            )
        await asyncio.sleep(remaining_s)

    def _has_remote_participants(self) -> bool:
        participants = getattr(getattr(self._ctx, "room", None), "remote_participants", None)
        return bool(participants)

    @staticmethod
    def _decode_data_packet(data_packet: Any) -> dict[str, Any] | None:
        raw = getattr(data_packet, "data", "")
        if isinstance(raw, (bytes, bytearray)):
            raw = raw.decode("utf-8", "ignore")
        if not isinstance(raw, str):
            return None

        message = raw.strip()
        if not message:
            return None

        try:
            payload = json.loads(message)
        except Exception:
            return None

        if not isinstance(payload, dict):
            return None
        return payload

    @classmethod
    def _user_audio_input_setup_timeout_s(cls) -> float:
        return cls._float_env_s(
            "USER_AUDIO_INPUT_SETUP_TIMEOUT_S",
            USER_AUDIO_INPUT_SETUP_TIMEOUT_S,
        )

    def _set_session_audio_input_enabled(self, enabled: bool) -> bool:
        session_input = getattr(self._session, "input", None)
        set_audio_enabled = getattr(session_input, "set_audio_enabled", None)
        if not callable(set_audio_enabled):
            return False

        try:
            set_audio_enabled(enabled)
        except Exception:
            self._logger.exception("[audio] failed toggling session audio input")
            return False
        return True

    def hold_user_audio_input(self, *, reason: str) -> None:
        """Descarta o audio do usuario na origem ate a primeira fala do agente terminar."""
        if self._user_audio_input_gate_released or not self._user_audio_input_enabled:
            return
        if not self._set_session_audio_input_enabled(False):
            return

        self._user_audio_input_enabled = False
        self._call_logger.info("USER_AUDIO_INPUT_HELD | reason=%s", reason)
        log_flow_event(
            self._call_logger,
            "user_audio_input_held",
            reason=reason,
            stage=self.state.current_stage,
        )
        if self._timeline is not None:
            self._timeline.emit("user_audio_input_held", reason=reason)

    def release_user_audio_input(self, *, reason: str) -> None:
        if self._user_audio_input_gate_released:
            return

        self._user_audio_input_gate_released = True
        self._scheduler.cancel(
            "user_audio_input_gate",
            reason=reason,
            log_name="USER_AUDIO_INPUT_GATE_CANCEL",
        )
        if self._user_audio_input_enabled:
            return

        self._set_session_audio_input_enabled(True)
        self._user_audio_input_enabled = True
        self._call_logger.info("USER_AUDIO_INPUT_RELEASED | reason=%s", reason)
        log_flow_event(
            self._call_logger,
            "user_audio_input_released",
            reason=reason,
            stage=self.state.current_stage,
        )
        if self._timeline is not None:
            self._timeline.emit("user_audio_input_released", reason=reason)

    def _arm_user_audio_input_gate_timeout(self) -> None:
        """Rede de seguranca: sem o turno inicial, o gate nunca seria liberado."""
        delay_s = self._user_audio_input_setup_timeout_s()
        if delay_s <= 0:
            return

        token = 0

        async def _timer() -> None:
            try:
                await asyncio.sleep(delay_s)
            except asyncio.CancelledError:
                return

            if not self._scheduler.is_current("user_audio_input_gate", token):
                return
            if self._initial_agent_turn_requested:
                return

            self._call_logger.warning(
                "USER_AUDIO_INPUT_GATE_TIMEOUT | timeout_s=%.1f | reason=initial_agent_turn_not_requested",
                delay_s,
            )
            self.release_user_audio_input(reason="setup_timeout")

        token = self._scheduler.arm(
            "user_audio_input_gate",
            task_name="user_audio_input_gate_timeout",
            coro=_timer(),
        )

    def request_initial_agent_turn(self, *, reason: str) -> None:
        if not self._agent_starts_conversation:
            return
        if self._initial_agent_turn_requested or self.finalized.is_set():
            return
        if self.state.user_final_seq != 0:
            return
        if str(self.state.current_stage or "INTRO").upper() not in {"", "INTRO"}:
            return

        self._initial_agent_turn_requested = True
        self._scheduler.cancel(
            "user_audio_input_gate",
            reason="initial_agent_turn_requested",
            log_name="USER_AUDIO_INPUT_GATE_CANCEL",
        )
        self._call_logger.info("INITIAL_AGENT_TURN | requested | reason=%s", reason)
        if self._timeline is not None:
            self._timeline.emit("initial_agent_turn_requested", reason=reason)

        async def _run_initial_turn() -> None:
            try:
                await self.run_pipeline("", "", user_seq=0)
            except Exception:
                self._logger.exception("[pipeline] initial agent turn failed")
                if self._timeline is not None:
                    self._timeline.emit("initial_agent_turn_failed", reason=reason)
            finally:
                self.release_user_audio_input(reason="first_agent_message_played")

        self._create_task_logged(
            _run_initial_turn(),
            name="run_pipeline_initial",
        )

    def _supports_backend_push(self) -> bool:
        pipeline = self._agent.pipeline
        if pipeline is None:
            return False

        supports = getattr(pipeline, "supports_server_push", None)
        if callable(supports):
            try:
                return bool(supports())
            except Exception:
                return False

        return callable(getattr(pipeline, "wait_for_server_push", None))

    def _supports_inflight_backend_push(self) -> bool:
        pipeline = self._agent.pipeline
        if pipeline is None:
            return False

        supports = getattr(pipeline, "supports_inflight_backend_push", None)
        if not callable(supports):
            return False

        try:
            return bool(supports())
        except Exception:
            return False

    def _inflight_backend_wait_enabled(self) -> bool:
        return (
            self._supports_inflight_backend_push()
            and float(self._config.inflight_backend_wait_timeout_s or 0.0) > 0.0
        )

    def _inflight_backend_wait_notice_enabled(self) -> bool:
        return (
            self._inflight_backend_wait_enabled()
            and self._has_inflight_backend_wait_audio()
            and float(self._config.inflight_backend_wait_interval_s or 0.0) > 0.0
        )

    def _record_inflight_backend_activity(self) -> None:
        self._inflight_backend_activity_at = time.monotonic()

    def _should_continue_inflight_backend_wait(self) -> bool:
        """Keep supervising the backend task for its whole lifetime.

        A valid deferred interruption increments ``user_final_seq`` while the
        original backend request remains in flight. Tying this supervisor to
        that sequence used to disable both periodic notices and the timeout,
        leaving the call waiting silently for the backend. The coroutine that
        owns the backend task already defines the request lifetime; only call
        finalization should stop its supervision early.
        """
        return not self.finalized.is_set()

    @staticmethod
    def _resolve_wait_audio_path(raw_path: str) -> Path:
        path = Path(raw_path).expanduser()
        if not path.is_absolute():
            path = Path.cwd() / path
        return path

    @classmethod
    def _wait_audio_candidates_from_value(cls, value: str) -> tuple[Path, ...]:
        raw_path = str(value or "").strip()
        if not raw_path:
            return ()

        path = cls._resolve_wait_audio_path(raw_path)
        if path.is_file() and path.suffix.lower() == ".wav":
            return (path,)
        if not path.is_dir():
            return ()

        return tuple(
            sorted(
                child
                for child in path.iterdir()
                if child.is_file() and child.suffix.lower() == ".wav"
            )
        )

    @classmethod
    def _inflight_backend_wait_audio_candidates_from_config(
        cls,
        config: RuntimeConfig,
    ) -> dict[str, tuple[Path, ...]]:
        long_candidates: list[Path] = []
        long_candidates.extend(
            cls._wait_audio_candidates_from_value(
                config.inflight_backend_wait_long_audio_dir
            )
        )
        long_candidates.extend(
            cls._wait_audio_candidates_from_value(
                config.inflight_backend_wait_long_audio_path
            )
        )
        return {
            "short": cls._wait_audio_candidates_from_value(
                config.inflight_backend_wait_short_audio_dir
            ),
            "long": tuple(dict.fromkeys(long_candidates)),
        }

    @staticmethod
    def _unique_wait_audio_candidates(
        candidates_by_kind: Mapping[str, Iterable[Path]],
    ) -> tuple[Path, ...]:
        paths: list[Path] = []
        for candidates in candidates_by_kind.values():
            paths.extend(candidates)
        return tuple(dict.fromkeys(paths))

    @staticmethod
    def _safe_log_warning(logger: Any, msg: str, *args: Any) -> None:
        log_fn = getattr(logger, "warning", None) or getattr(logger, "info", None)
        if not callable(log_fn):
            return
        try:
            log_fn(msg, *args)
        except Exception:
            return

    @classmethod
    def _preload_wait_audio_paths(
        cls,
        paths: Iterable[Path],
        *,
        logger: Any = None,
    ) -> tuple[dict[Path, int], int]:
        durations: dict[Path, int] = {}
        failures = 0
        for audio_path in paths:
            try:
                durations[audio_path] = wav_duration_ms(str(audio_path))
            except Exception as exc:
                failures += 1
                cls._safe_log_warning(
                    logger,
                    "[audio] failed preloading wait audio path=%s error=%s: %s",
                    audio_path,
                    type(exc).__name__,
                    exc,
                )
        return durations, failures

    @classmethod
    def prewarm_wait_audio_cache(
        cls,
        *,
        short_audio_dir: str = "",
        long_audio_dir: str = "",
        long_audio_path: str = "",
        logger: Any = None,
    ) -> tuple[int, int]:
        config = RuntimeConfig(
            call_end_grace_s=0.0,
            final_grace_s=0.0,
            idle_nudge_delay_s=0.0,
            idle_nudge_join_delay_s=0.0,
            idle_nudge_close_delay_s=0.0,
            idle_nudge_max_tries=0,
            idle_nudge_end_reason="",
            inflight_backend_wait_short_audio_dir=short_audio_dir,
            inflight_backend_wait_long_audio_dir=long_audio_dir,
            inflight_backend_wait_long_audio_path=long_audio_path,
        )
        candidates_by_kind = cls._inflight_backend_wait_audio_candidates_from_config(config)
        durations, failures = cls._preload_wait_audio_paths(
            cls._unique_wait_audio_candidates(candidates_by_kind),
            logger=logger,
        )
        return len(durations), failures

    def _inflight_backend_wait_audio_candidates(self, kind: str) -> tuple[Path, ...]:
        normalized = str(kind or "").strip().lower()
        if normalized == "short":
            return self._inflight_backend_wait_audio_candidates_cache.get("short", ())
        if normalized == "long":
            return self._inflight_backend_wait_audio_candidates_cache.get("long", ())
        return ()

    def _has_inflight_backend_wait_audio(self) -> bool:
        return bool(
            self._inflight_backend_wait_audio_candidates("short")
            or self._inflight_backend_wait_audio_candidates("long")
        )

    @staticmethod
    def _preferred_inflight_backend_wait_audio_kind(attempt: int) -> str:
        return "short" if int(attempt or 0) <= 1 else "long"

    def _select_inflight_backend_wait_audio(
        self,
        *,
        attempt: int,
    ) -> tuple[Path | None, str]:
        preferred_kind = self._preferred_inflight_backend_wait_audio_kind(attempt)
        candidates = self._inflight_backend_wait_audio_candidates(preferred_kind)
        selected_kind = preferred_kind
        if not candidates:
            selected_kind = "long" if preferred_kind == "short" else "short"
            candidates = self._inflight_backend_wait_audio_candidates(selected_kind)

        if not candidates:
            return None, selected_kind
        return random.choice(candidates), selected_kind

    def _inflight_backend_wait_audio_path(self, *, attempt: int = 2) -> Path | None:
        audio_path, _kind = self._select_inflight_backend_wait_audio(attempt=attempt)
        return audio_path

    def _default_inflight_backend_wait_text(self) -> str:
        text = str(self._config.inflight_backend_wait_text or "").strip()
        return text or "Um momento, ainda estou consultando para te ajudar."

    @staticmethod
    def _inflight_backend_wait_audio_text(audio_path: Path, *, fallback_text: str) -> str:
        text_path = audio_path.with_suffix(".txt")
        if not text_path.is_file():
            return fallback_text

        for encoding in ("utf-8-sig", "utf-8", "latin-1"):
            try:
                text = " ".join(text_path.read_text(encoding=encoding).split())
            except UnicodeDecodeError:
                continue
            except OSError:
                return fallback_text
            return text or fallback_text

        return fallback_text

    def _cached_inflight_backend_wait_audio_text(
        self,
        audio_path: Path,
        *,
        fallback_text: str,
    ) -> str:
        key = (audio_path, fallback_text)
        cached = self._inflight_backend_wait_audio_text_cache.get(key)
        if cached is not None:
            return cached
        text = self._inflight_backend_wait_audio_text(
            audio_path,
            fallback_text=fallback_text,
        )
        self._inflight_backend_wait_audio_text_cache[key] = text
        return text

    def _preload_inflight_backend_wait_audio_texts(self) -> None:
        fallback_text = self._default_inflight_backend_wait_text()
        for audio_path in self._unique_wait_audio_candidates(
            self._inflight_backend_wait_audio_candidates_cache
        ):
            self._cached_inflight_backend_wait_audio_text(
                audio_path,
                fallback_text=fallback_text,
            )

    def _inflight_backend_wait_audio_duration_ms(self, audio_path: Path) -> int | None:
        if audio_path in self._inflight_backend_wait_audio_duration_cache:
            return self._inflight_backend_wait_audio_duration_cache[audio_path]
        try:
            duration_ms = wav_duration_ms(str(audio_path))
        except Exception:
            return None
        self._inflight_backend_wait_audio_duration_cache[audio_path] = duration_ms
        return duration_ms

    @staticmethod
    def _inflight_backend_wait_message_id(*, message_id: str = "", attempt: int) -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        comfort_id = uuid.uuid4().hex
        return f"{base}_conforto_{comfort_id}"

    @staticmethod
    def _agent_wait_timeout_inactivity_message_id(*, message_id: str = "", attempt: int) -> str:
        base = str(message_id or "").strip() or str(uuid.uuid4())
        return f"{base}_inatividade_{max(1, int(attempt or 1))}"

    async def _play_inflight_backend_wait_notice_audio(
        self,
        *,
        user_seq: int | None,
        attempt: int,
        message_id: str = "",
        require_user_not_speaking: bool = True,
        abort_if_agent_spoke_since: int | None = None,
    ) -> InflightBackendWaitNoticeResult:
        if user_seq is not None and not self._should_continue_inflight_backend_wait():
            return InflightBackendWaitNoticeResult(started=False)

        processing_notice = user_seq is not None
        fallback_text = self._default_inflight_backend_wait_text()
        audio_path, notice_kind = self._select_inflight_backend_wait_audio(attempt=attempt)
        if audio_path is None or not self._has_remote_participants():
            return InflightBackendWaitNoticeResult(started=False)
        if require_user_not_speaking and not self._user_not_speaking.is_set():
            return InflightBackendWaitNoticeResult(
                started=False,
                interrupted_by_user=processing_notice,
            )
        text = self._cached_inflight_backend_wait_audio_text(
            audio_path,
            fallback_text=fallback_text,
        )
        duration_ms = self._inflight_backend_wait_audio_duration_ms(audio_path)

        base_message_id = str(message_id or "").strip() or self._current_base_message_id()
        comfort_message_id = self._inflight_backend_wait_message_id(
            message_id=base_message_id,
            attempt=attempt,
        )
        self._call_logger.info(
            "INFLIGHT_BACKEND_WAIT_AUDIO | message_id=%s | original_message_id=%s | user_seq=%s | attempt=%s | kind=%s | path=%s | text=%r",
            comfort_message_id,
            base_message_id,
            user_seq,
            attempt,
            notice_kind,
            audio_path,
            text,
        )
        if self._timeline is not None:
            self._timeline.emit(
                "inflight_backend_wait_audio",
                message_id=comfort_message_id,
                user_seq=user_seq,
                attempt=attempt,
                kind=notice_kind,
                path=str(audio_path),
                duration_ms=duration_ms,
                text=text,
            )

        interrupted_by_user = False

        def _capture_playout_result(
            *,
            interrupted: bool,
            user_spoke_during: bool,
        ) -> None:
            nonlocal interrupted_by_user
            interrupted_by_user = processing_notice and (
                interrupted or user_spoke_during
            )

        try:
            started = await self.say_stage(
                text,
                INFLIGHT_BACKEND_WAIT_STAGE,
                add_to_chat_ctx=False,
                # O aviso curto comeca antes do STT final do mesmo turno. Ele fica
                # protegido somente ate esse final chegar; depois uma nova fala
                # real do cliente pode interrompe-lo normalmente.
                allow_interruptions=notice_kind != "short",
                rearm_interruptions_on_next_user_speech=notice_kind == "short",
                schedule_idle_after=False,
                wait_post_final_grace=False,
                require_user_not_speaking=require_user_not_speaking,
                abort_if_agent_spoke_since=abort_if_agent_spoke_since,
                message_id=comfort_message_id,
                audio=wav_audio_frames(str(audio_path)),
                audio_duration_ms=duration_ms,
                playout_result_callback=_capture_playout_result,
            )
            result = InflightBackendWaitNoticeResult(
                started=bool(started),
                interrupted_by_user=interrupted_by_user,
            )
            if result:
                self._record_inflight_backend_activity()
            return result
        except Exception:
            self._logger.exception("[audio] failed playing inflight backend wait audio")
            if self._timeline is not None:
                self._timeline.emit(
                    "audio_stage_failed",
                    stage=INFLIGHT_BACKEND_WAIT_STAGE,
                    source="inflight_backend_wait_audio",
                    path=str(audio_path),
                )
            await self.terminate_with_resource_stop(
                resource="tts",
                source="inflight_backend_wait_audio",
            )
            raise InflightBackendWaitTimedOut()

    def _cancel_pre_backend_wait_notice(self, *, reason: str) -> None:
        self._pre_backend_wait_notice_generation += 1
        self._pre_backend_wait_notice_reserved = False
        task = self._pre_backend_wait_notice_task
        self._pre_backend_wait_notice_task = None
        if task is not None and task is not asyncio.current_task() and not task.done():
            task.cancel()
            if self._timeline is not None:
                self._timeline.emit("pre_backend_wait_notice_cancelled", reason=reason)

    def _reserve_pre_backend_wait_notice(self) -> bool:
        if (
            self.finalized.is_set()
            or not self._inflight_backend_wait_notice_enabled()
            or not self._has_remote_participants()
        ):
            return False

        self._pre_backend_wait_notice_reserved = True
        self._record_inflight_backend_activity()
        return True

    def _consume_pre_backend_wait_notice(self) -> bool:
        if not self._pre_backend_wait_notice_reserved:
            return False

        self._pre_backend_wait_notice_reserved = False
        return True

    def _pre_backend_wait_notice_is_current_speech(self) -> bool:
        return (
            self._pre_backend_wait_notice_reserved
            and self.state.current_speech.stage == INFLIGHT_BACKEND_WAIT_STAGE
        )

    def _inflight_backend_wait_notice_is_speaking(self) -> bool:
        return (
            self.speaking.is_set()
            and self.state.current_speech.stage == INFLIGHT_BACKEND_WAIT_STAGE
        )

    def _pre_backend_wait_notice_fast_on_vad_pause(self, *, reason: str) -> bool:
        return (
            bool(self._config.pre_backend_wait_notice_fast_on_vad_pause)
            and str(reason or "").strip() == "vad_end_of_speech"
        )

    def _log_pre_backend_wait_notice_skipped(self, *, reason: str, skipped_reason: str) -> None:
        log_flow_event(
            self._call_logger,
            "pre_backend_wait_notice_skipped",
            reason=reason,
            skipped_reason=skipped_reason,
        )
        if self._timeline is not None:
            self._timeline.emit(
                "pre_backend_wait_notice_skipped",
                reason=reason,
                skipped_reason=skipped_reason,
            )

    def _schedule_pre_backend_wait_notice(self, *, reason: str) -> None:
        if not self._inflight_backend_wait_notice_enabled() or self.finalized.is_set():
            return
        post_playout_drop_context = self._active_post_playout_drop_context(
            reason="protected_speech_post_playout_grace"
        )
        if (
            self._drop_next_user_final_context is not None
            or self._backend_processing_drop_context is not None
            or post_playout_drop_context is not None
        ):
            self._log_pre_backend_wait_notice_skipped(
                reason=reason,
                skipped_reason="user_input_drop_pending",
            )
            return
        if self.speaking.is_set():
            # Nao ha silencio para mascarar: o aviso so entraria na fila do
            # say_lock e sairia colado no fim da fala do agente.
            self._log_pre_backend_wait_notice_skipped(
                reason=reason,
                skipped_reason="agent_speaking",
            )
            return
        if self._agent._run_lock.locked():
            # O fim de fala aconteceu enquanto o backend ainda processava o
            # turno anterior. O STT pode transformar essa fala em uma
            # interrupcao diferida, que possui seu proprio conforto dedicado.
            # Tocar o aviso especulativo agora produziria dois confortos em
            # sequencia: AGENT_BACKEND_WAIT e, logo depois,
            # INTERRUPTION_COMFORT.
            self._log_pre_backend_wait_notice_skipped(
                reason=reason,
                skipped_reason="backend_processing",
            )
            return
        if self._pre_backend_wait_notice_reserved:
            return
        if (
            self._pre_backend_wait_notice_task is not None
            and not self._pre_backend_wait_notice_task.done()
        ):
            return

        self._cancel_pre_backend_wait_notice(reason=f"restart:{reason}")
        self._pre_backend_wait_notice_generation += 1
        generation = self._pre_backend_wait_notice_generation
        fast_on_vad_pause = self._pre_backend_wait_notice_fast_on_vad_pause(
            reason=reason
        )
        say_stage_seq_at_schedule = self._say_stage_seq

        async def _run() -> None:
            try:
                if not fast_on_vad_pause:
                    await asyncio.sleep(PRE_BACKEND_WAIT_NOTICE_GUARD_S)
                if generation != self._pre_backend_wait_notice_generation:
                    return
                if not fast_on_vad_pause and not self._user_not_speaking.is_set():
                    self._log_pre_backend_wait_notice_skipped(
                        reason=reason,
                        skipped_reason="user_speaking",
                    )
                    return
                if self.speaking.is_set():
                    self._log_pre_backend_wait_notice_skipped(
                        reason=reason,
                        skipped_reason="agent_speaking",
                    )
                    return
                if not self._reserve_pre_backend_wait_notice():
                    return

                if self._timeline is not None:
                    self._timeline.emit(
                        "pre_backend_wait_notice_started",
                        reason=reason,
                        fast_on_vad_pause=fast_on_vad_pause,
                    )

                played = await self._play_inflight_backend_wait_notice_audio(
                    user_seq=None,
                    attempt=1,
                    message_id=self._peek_started_turn_message_id(),
                    require_user_not_speaking=not fast_on_vad_pause,
                    abort_if_agent_spoke_since=say_stage_seq_at_schedule,
                )
                if not played:
                    # Sem audio reproduzido nao ha aviso a consumir no pipeline:
                    # devolve a reserva para nao bloquear o proximo agendamento.
                    self._cancel_pre_backend_wait_notice(reason="notice_not_played")
            except asyncio.CancelledError:
                return
            except InflightBackendWaitTimedOut:
                return
            finally:
                if self._pre_backend_wait_notice_task is asyncio.current_task():
                    self._pre_backend_wait_notice_task = None

        self._pre_backend_wait_notice_task = self._create_task_logged(
            _run(),
            name="pre_backend_wait_notice",
        )

    def schedule_pre_backend_wait_notice(self, *, reason: str) -> None:
        self._schedule_pre_backend_wait_notice(reason=reason)

    async def _execute_pipeline_with_inflight_backend_wait(
        self,
        stt_payload: Any,
        *,
        user_seq: int,
        message_id: str = "",
        initial_notice_consumed: bool = False,
    ) -> BackendReply:
        if not self._inflight_backend_wait_enabled():
            return await self.execute(RunPipelineInput(stt_payload))

        timeout_s = max(0.0, float(self._config.inflight_backend_wait_timeout_s or 0.0))
        interval_s = max(0.0, float(self._config.inflight_backend_wait_interval_s or 0.0))
        max_notices = max(0, int(self._config.inflight_backend_wait_max_notices or 0))
        can_notice = self._inflight_backend_wait_notice_enabled()

        def _notice_limit_reached(notices: int) -> bool:
            # A non-positive limit explicitly means unlimited notices. The
            # absolute backend timeout remains the terminal boundary.
            return max_notices > 0 and notices >= max_notices

        started_at = time.monotonic()
        backend_task = self._create_task_logged(
            self.execute(RunPipelineInput(stt_payload)),
            name=f"run_pipeline_backend_{user_seq}",
        )
        notices_sent = 1 if initial_notice_consumed else 0
        pre_notice_consumed = initial_notice_consumed
        max_notices_logged = False
        user_silence_task: asyncio.Task[bool] | None = None
        if initial_notice_consumed:
            # O INTERRUPTION_COMFORT ja confirmou a fala do usuario. Conta como
            # primeiro aviso para que o pipeline substituto nao toque um short
            # logo em seguida; o proximo aviso, se necessario, sera long e
            # respeitara o intervalo normal.
            self._inflight_backend_activity_at = started_at

        def _notice_activity_at() -> float:
            if pre_notice_consumed:
                return self._inflight_backend_activity_at or started_at
            return max(started_at, self._inflight_backend_activity_at)

        def _defer_notice_until_user_silence() -> None:
            nonlocal user_silence_task
            # Any real user activity, including a short utterance that business
            # rules later discard, must rearm the notice interval. Otherwise an
            # already-due notice is retried immediately after every silence and
            # several comfort audios appear back-to-back.
            self._record_inflight_backend_activity()
            if user_silence_task is None or user_silence_task.done():
                user_silence_task = asyncio.create_task(
                    self._user_not_speaking.wait(),
                    name="inflight_backend_wait_user_silence",
                )

        def _log_max_notices_reached() -> None:
            nonlocal max_notices_logged
            if max_notices_logged or not can_notice or not _notice_limit_reached(notices_sent):
                return
            max_notices_logged = True
            elapsed_s = max(0.0, time.monotonic() - started_at)
            remaining_timeout_s = max(0.0, timeout_s - elapsed_s)
            log_flow_event(
                self._call_logger,
                "inflight_backend_wait_max_notices_reached",
                user_seq=user_seq,
                max_notices=max_notices,
                notices_sent=notices_sent,
                wait_timeout_s=timeout_s,
                remaining_timeout_s=round(remaining_timeout_s, 3),
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "inflight_backend_wait_max_notices_reached",
                    user_seq=user_seq,
                    max_notices=max_notices,
                    notices_sent=notices_sent,
                    wait_timeout_s=timeout_s,
                    remaining_timeout_s=round(remaining_timeout_s, 3),
                )
            self._log_resource_error_events(
                status=self._status_for_resource("agent_backend"),
                reason="inflight_backend_wait_max_notices_reached",
                resource="agent_backend",
                source="inflight_backend_wait_max_notices_reached",
                message_id=message_id,
            )

        try:
            if can_notice:
                if initial_notice_consumed:
                    pass
                elif self._consume_pre_backend_wait_notice():
                    notices_sent = 1
                    pre_notice_consumed = True
                else:
                    self._inflight_backend_activity_at = started_at
                    await asyncio.sleep(0)
                    if backend_task.done():
                        return backend_task.result()

                    notices_sent += 1
                    notice_result = await self._play_inflight_backend_wait_notice_audio(
                        user_seq=user_seq,
                        attempt=notices_sent,
                        message_id=message_id,
                    )
                    if not notice_result:
                        notices_sent -= 1
                        if notice_result.interrupted_by_user:
                            _defer_notice_until_user_silence()
                        else:
                            self._record_inflight_backend_activity()
                    else:
                        _log_max_notices_reached()
            else:
                self._inflight_backend_activity_at = started_at

            _log_max_notices_reached()

            while True:
                if not self._should_continue_inflight_backend_wait():
                    return await backend_task

                now = time.monotonic()
                timeout_at = started_at + timeout_s
                wait_until = timeout_at
                pre_notice_task = self._pre_backend_wait_notice_task
                pre_notice_running = (
                    notices_sent > 0
                    and pre_notice_task is not None
                    and not pre_notice_task.done()
                )
                waiting_for_user_silence = (
                    user_silence_task is not None
                    and not user_silence_task.done()
                )
                if (
                    can_notice
                    and not _notice_limit_reached(notices_sent)
                    and not pre_notice_running
                    and not waiting_for_user_silence
                ):
                    last_activity_at = _notice_activity_at()
                    wait_until = min(wait_until, last_activity_at + interval_s)

                wait_s = max(0.0, wait_until - now)
                wait_tasks = {backend_task}
                if pre_notice_running:
                    wait_tasks.add(pre_notice_task)
                if waiting_for_user_silence and user_silence_task is not None:
                    wait_tasks.add(user_silence_task)

                done, _pending = await asyncio.wait(
                    wait_tasks,
                    timeout=wait_s,
                    return_when=asyncio.FIRST_COMPLETED,
                )
                if backend_task in done:
                    return backend_task.result()
                if pre_notice_task is not None and pre_notice_task in done:
                    await asyncio.gather(pre_notice_task, return_exceptions=True)
                    continue
                if user_silence_task is not None and user_silence_task in done:
                    await asyncio.gather(user_silence_task, return_exceptions=True)
                    user_silence_task = None
                    continue

                now = time.monotonic()
                if timeout_s > 0.0 and now >= timeout_at:
                    if self._timeline is not None:
                        self._timeline.emit(
                            "inflight_backend_wait_timeout",
                            user_seq=user_seq,
                            wait_timeout_s=timeout_s,
                            notices_sent=notices_sent,
                        )
                    backend_task.cancel()
                    await asyncio.gather(backend_task, return_exceptions=True)
                    await self.terminate_with_resource_stop(
                        resource="agent_backend",
                        source="inflight_backend_wait_timeout",
                    )
                    raise InflightBackendWaitTimedOut()

                if can_notice and not _notice_limit_reached(notices_sent):
                    last_activity_at = _notice_activity_at()
                    if now - last_activity_at >= interval_s:
                        notices_sent += 1
                        notice_result = await self._play_inflight_backend_wait_notice_audio(
                            user_seq=user_seq,
                            attempt=notices_sent,
                            message_id=message_id,
                        )
                        if not notice_result:
                            notices_sent -= 1
                            if notice_result.interrupted_by_user:
                                _defer_notice_until_user_silence()
                            else:
                                self._record_inflight_backend_activity()
                        else:
                            _log_max_notices_reached()
        except InflightBackendWaitTimedOut:
            if not backend_task.done():
                backend_task.cancel()
                await asyncio.gather(backend_task, return_exceptions=True)
            raise
        except asyncio.CancelledError:
            backend_task.cancel()
            await asyncio.gather(backend_task, return_exceptions=True)
            raise
        finally:
            if user_silence_task is not None and not user_silence_task.done():
                user_silence_task.cancel()
                await asyncio.gather(user_silence_task, return_exceptions=True)

    def _cancel_backend_push_listener(self, *, reason: str, force: bool = False) -> None:
        task = self._backend_push_task
        if not force and self.speaking.is_set():
            return
        self._backend_push_task = None
        if task is not None and not task.done():
            task.cancel()
            if self._timeline is not None:
                self._timeline.emit("backend_push_cancelled", reason=reason)

    def _schedule_backend_push_listener(
        self,
        *,
        user_seq: int,
        reason: str,
        inflight: bool = False,
    ) -> None:
        if self.finalized.is_set():
            return

        supported = (
            self._supports_inflight_backend_push()
            if inflight
            else self._supports_backend_push()
        )
        if not supported:
            if not inflight:
                self._cancel_backend_push_listener(reason=f"unsupported:{reason}", force=True)
            return

        self._cancel_backend_push_listener(reason=f"restart:{reason}", force=True)

        async def _listen() -> None:
            try:
                while not self.finalized.is_set():
                    if user_seq != self.state.user_final_seq:
                        return

                    pipeline = self._agent.pipeline
                    if pipeline is None:
                        return

                    wait_for_push = getattr(pipeline, "wait_for_server_push", None)
                    if not callable(wait_for_push):
                        return

                    raw_reply = await wait_for_push()
                    if raw_reply is None or self.finalized.is_set():
                        return
                    if user_seq != self.state.user_final_seq:
                        return

                    self._record_inflight_backend_activity()
                    reply = self._coerce_backend_reply(
                        raw_reply,
                        fallback_stage=self.state.current_stage,
                    )
                    self._update_backend_processing_drop_from_reply(reply, source="backend_push")
                    self.state.current_stage = reply.stage
                    if not reply.text:
                        self._log_backend_reply_error_event(reply, source="backend_push")

                    if reply.text:
                        if not await self._wait_for_stable_user_silence_before_terminal_reply(
                            reply,
                            user_seq=user_seq,
                            source="backend_push",
                        ):
                            return
                        try:
                            await self._speak_backend_reply(
                                reply,
                                source="backend_push",
                                add_to_chat_ctx=True,
                            )
                        except Exception:
                            self._logger.exception("[tts] failed speaking backend push reply")
                            if self._timeline is not None:
                                self._timeline.emit(
                                    "tts_stage_failed",
                                    stage=self.state.current_stage,
                                    source="backend_push",
                                )
                            await self.terminate_with_resource_stop(
                                resource="tts",
                                source="backend_push_tts",
                            )
                            return

                    self.arm_agent_wait_timeout_from_reply(
                        reply,
                        user_seq=user_seq,
                        source="backend_push",
                    )

                    terminal_stop = self._terminal_stop_from_reply(reply)
                    if terminal_stop is not None:
                        await self._finish_with_agent_final_stop(
                            terminal_stop,
                            source="backend_push",
                        )
                        return

                    if reply.done or self._finalization_policy.is_done_stage(reply.stage):
                        self._create_task_logged(
                            self.notify_bridge_done("stage_done"),
                            name="bridge_done_push",
                        )
                        self._create_task_logged(
                            self.finalize("stage_done"),
                            name="finalize_push",
                        )
                        return
            except asyncio.CancelledError:
                return
            except Exception:
                self._logger.exception("[pipeline] backend push listener failed")
                if self._timeline is not None:
                    self._timeline.emit("backend_push_failed", reason=reason)
                await self.terminate_with_resource_stop(
                    resource="agent_backend",
                    source="backend_push_listener",
                )

        self._backend_push_task = self._create_task_logged(
            _listen(),
            name=f"backend_push_{user_seq}",
        )
        if self._timeline is not None:
            self._timeline.emit(
                "backend_push_started",
                reason=reason,
                user_seq=user_seq,
                inflight=inflight,
            )

    def _coerce_backend_reply(self, raw_reply: Any, *, fallback_stage: str) -> BackendReply:
        if isinstance(raw_reply, BackendReply):
            stage = (raw_reply.stage or fallback_stage or "UNKNOWN").strip().upper()
            done = bool(raw_reply.done) or stage == "DONE"
            return BackendReply(
                stage=stage,
                text=(raw_reply.text or "").strip(),
                done=done,
                export_payload=raw_reply.export_payload,
                metadata=raw_reply.metadata,
            )

        if isinstance(raw_reply, tuple) and len(raw_reply) == 2:
            stage_raw, text_raw = raw_reply
            stage = str(stage_raw or fallback_stage or "UNKNOWN").strip().upper()
            return BackendReply(
                stage=stage,
                text=str(text_raw or "").strip(),
                done=stage == "DONE",
                export_payload=None,
            )

        if isinstance(raw_reply, dict):
            stage = str(raw_reply.get("stage") or fallback_stage or "UNKNOWN").strip().upper()
            text = str(raw_reply.get("text") or "").strip()
            done = bool(raw_reply.get("done")) or stage == "DONE"
            export_payload = raw_reply.get("export_payload")
            if export_payload is None and "result" in raw_reply:
                export_payload = raw_reply.get("result")
            return BackendReply(
                stage=stage,
                text=text,
                done=done,
                export_payload=export_payload,
                metadata=raw_reply.get("metadata"),
            )

        stage = str(fallback_stage or self.state.current_stage or "UNKNOWN").strip().upper()
        return BackendReply(
            stage=stage,
            text="",
            done=stage == "DONE",
            export_payload=raw_reply,
        )

    @staticmethod
    def _wait_timeout_from_reply(reply: BackendReply) -> float:
        metadata = reply.metadata
        if not isinstance(metadata, Mapping):
            return 0.0

        raw_timeout = metadata.get("wait_timeout_seconds")
        try:
            timeout_s = float(raw_timeout)
        except (TypeError, ValueError):
            return 0.0

        if timeout_s <= 0:
            return 0.0
        return timeout_s

    @staticmethod
    def _reply_has_wait_timeout_metadata(reply: BackendReply) -> bool:
        metadata = reply.metadata
        return isinstance(metadata, Mapping) and "wait_timeout_seconds" in metadata

    @staticmethod
    def _wait_retry_messages_from_reply(reply: BackendReply) -> tuple[str, ...]:
        metadata = reply.metadata
        if not isinstance(metadata, Mapping):
            return ()

        raw_messages = metadata.get("wait_retry_messages")
        if not isinstance(raw_messages, (list, tuple)):
            return ()

        messages: list[str] = []
        for raw_message in raw_messages:
            message = str(raw_message or "").strip()
            if message:
                messages.append(message)
        return tuple(messages)

    @classmethod
    def _wait_timeout_config_from_reply(
        cls,
        reply: BackendReply,
    ) -> AgentWaitTimeoutConfig | None:
        timeout_s = cls._wait_timeout_from_reply(reply)
        if timeout_s <= 0:
            return None
        return AgentWaitTimeoutConfig(
            timeout_s=timeout_s,
            retry_messages=cls._wait_retry_messages_from_reply(reply),
        )

    @staticmethod
    def _terminal_stop_from_reply(reply: BackendReply) -> AgentFinalStop | None:
        return final_stop_from_agent_result(reply.export_payload)

    async def _wait_for_stable_user_silence_before_terminal_reply(
        self,
        reply: BackendReply,
        *,
        user_seq: int,
        source: str,
    ) -> bool:
        """Do not let a terminal reply start on top of an active user utterance.

        User state changes to ``listening`` slightly before the final transcript is
        delivered.  Requiring a short stable-silence window keeps the terminal
        message from starting on top of the user, but a late transcript must not
        invalidate a reply that closes the conversation.
        """
        terminal = self._terminal_stop_from_reply(reply) is not None or (
            reply.done or self._finalization_policy.is_done_stage(reply.stage)
        )
        if not terminal:
            return True

        waited = False
        while not self.finalized.is_set():
            if not self._user_not_speaking.is_set():
                if not waited:
                    waited = True
                    log_flow_event(
                        self._call_logger,
                        "terminal_tts_wait_user_silence",
                        source=source,
                        stage=reply.stage,
                        user_seq=user_seq,
                    )
                await self._user_not_speaking.wait()

            await asyncio.sleep(POST_USER_FINAL_SPEECH_GRACE_S)
            if self._user_not_speaking.is_set():
                break

        allowed = not self.finalized.is_set()
        if not allowed:
            log_flow_event(
                self._call_logger,
                "terminal_tts_skipped",
                reason="already_finalized",
                source=source,
                stage=reply.stage,
                user_seq=user_seq,
                current_seq=self.state.user_final_seq,
            )
        elif waited:
            log_flow_event(
                self._call_logger,
                "terminal_tts_user_silence_confirmed",
                source=source,
                stage=reply.stage,
                user_seq=user_seq,
            )
        return allowed

    async def _finish_with_agent_final_stop(
        self,
        stop: AgentFinalStop,
        *,
        source: str,
    ) -> None:
        async with self._terminal_stop_lock:
            if self._terminal_stop_sent:
                return
            self._terminal_stop_sent = True

        if self._timeline is not None:
            self._timeline.emit(
                "agent_final_stop_started",
                status=stop.status,
                reason=stop.reason,
                result_type=stop.result_type,
                source=source,
            )

        if source == "backend_push":
            self._backend_push_task = None
        else:
            self._cancel_backend_push_listener(reason=f"agent_final_stop:{stop.reason}", force=True)
        self.cancel_idle_timer("agent_final_stop")
        self.cancel_idle_close("agent_final_stop")
        self.cancel_agent_wait_timeout("agent_final_stop")

        self._log_resource_error_events(
            status=stop.status,
            reason=stop.reason,
            source=source,
            terminal=True,
        )
        await self.notify_bridge_stop(
            status=stop.status,
            reason=stop.reason,
            phase=stop.phase,
        )
        self.finalized.set()
        if self._timeline is not None:
            self._timeline.emit(
                "agent_final_stop_completed",
                status=stop.status,
                reason=stop.reason,
                result_type=stop.result_type,
                source=source,
            )

    async def _speak_backend_reply(
        self,
        reply: BackendReply,
        *,
        initial_turn: bool = False,
        source: str,
        add_to_chat_ctx: bool,
    ) -> None:
        structured_erro_msg = self._error_message_from_backend_reply(reply)
        message_id = self._message_id_from_reply(reply)
        agent_message_type = self._reply_agent_message_type(reply)
        agent_result_type = self._reply_agent_result_type(reply)
        if (
            agent_message_type == "feedback"
            or agent_result_type == "feedback"
            or self._metadata_str(reply.metadata, "event") == "feedback"
        ):
            message_id = self._feedback_message_id(
                message_id=message_id,
                speech_id=self._speech_id_from_reply(reply),
            )
        self._update_backend_processing_drop_from_reply(reply, source=source)
        if not reply.text:
            self._log_backend_reply_error_event(reply, source=source, message_id=message_id)
            return
        self._call_logger.debug(
            "BACKEND_REPLY_TEXT | source=%s | stage=%s | done=%s | text=%r",
            source,
            reply.stage,
            reply.done,
            reply.text,
        )
        if self.finalized.is_set():
            self._call_logger.debug(
                "BACKEND_TTS_SKIPPED | source=%s | stage=%s | reason=already_finalized",
                source,
                reply.stage,
            )
            log_flow_event(
                self._call_logger,
                "tts_skip",
                source=source,
                stage=reply.stage,
                reason="already_finalized",
            )
            return
        if not self._has_remote_participants():
            self._call_logger.debug(
                "BACKEND_TTS_SKIPPED | source=%s | stage=%s | reason=no_remote_participants",
                source,
                reply.stage,
            )
            log_flow_event(
                self._call_logger,
                "tts_skip",
                source=source,
                stage=reply.stage,
                reason="no_remote_participants",
            )
            return

        default_allow_interruptions = self._interrupt_policy.allow_stage(reply.stage)
        allow_interruptions = self._allow_interruptions_for_reply(
            reply,
            default_allow_interruptions,
        )
        if self._terminal_stop_from_reply(reply) is not None or (
            reply.done or self._finalization_policy.is_done_stage(reply.stage)
        ):
            # A resposta terminal e a confirmacao audivel do desfecho da chamada.
            # Ela pode esperar o usuario terminar, mas nunca pode ser cortada.
            allow_interruptions = False
        if self.state.deferred_interruption.replay_in_flight:
            # A resposta do reenvio nao pode ser interrompida. Descartar a fala do
            # cliente no runtime nao basta: o barge-in do proprio LiveKit corta o
            # playout por atividade de audio, e R2 sairia pela metade.
            allow_interruptions = False
        expects_user_response = self._reply_expects_user_response(reply)
        initial_audio = None
        if initial_turn:
            cache_lookup = getattr(self._initial_greeting_tts, "initial_greeting_audio", None)
            if callable(cache_lookup):
                try:
                    initial_audio = cache_lookup(reply.text)
                except Exception:
                    self._logger.exception("INITIAL_GREETING_AUDIO_CACHE_LOOKUP_FAIL")
            if self._timeline is not None:
                self._timeline.emit(
                    "initial_greeting_audio_cache",
                    outcome="hit" if initial_audio is not None else "miss_or_disabled",
                )
        await self.say_stage(
            reply.text,
            reply.stage,
            add_to_chat_ctx=add_to_chat_ctx,
            allow_interruptions=allow_interruptions,
            speech_id=self._speech_id_from_reply(reply),
            message_id=message_id,
            schedule_idle_after=True if expects_user_response is None else expects_user_response,
            drop_user_input_while_speaking=self._reply_drops_user_input_while_speaking(reply),
            audio=initial_audio,
            agent_message_type=agent_message_type,
            agent_result_type=agent_result_type,
            erro_msg=structured_erro_msg,
        )

    def cancel_agent_wait_timeout(self, reason: str = "cancel") -> None:
        self._scheduler.cancel(
            "agent_wait_timeout",
            reason=reason,
            log_name="AGENT_WAIT_TIMEOUT_CANCEL",
        )

    def _remember_agent_wait_timeout(
        self,
        config: AgentWaitTimeoutConfig,
        *,
        user_seq: int,
        source: str,
        message_id: str,
    ) -> None:
        self._last_agent_wait_timeout_config = config
        self._last_agent_wait_timeout_user_seq = user_seq
        self._last_agent_wait_timeout_source = source
        self._last_agent_wait_timeout_message_id = message_id

    def _clear_agent_wait_timeout_memory(self) -> None:
        self._last_agent_wait_timeout_config = None
        self._last_agent_wait_timeout_user_seq = 0
        self._last_agent_wait_timeout_source = ""
        self._last_agent_wait_timeout_message_id = ""

    def _clear_agent_wait_timeout_config(self) -> None:
        self._agent_wait_timeout_config = None

    def _restore_agent_wait_timeout_after_empty_stt(self) -> bool:
        config = self._last_agent_wait_timeout_config
        if config is None:
            return False
        if self._last_agent_wait_timeout_user_seq != self.state.user_final_seq:
            return False
        if self.finalized.is_set() or (self.state.current_stage or "").upper() == "DONE":
            return False

        previous_source = self._last_agent_wait_timeout_source
        previous_message_id = self._last_agent_wait_timeout_message_id
        self._arm_agent_wait_timeout(
            config,
            user_seq=self.state.user_final_seq,
            source="empty_stt_final",
            message_id=previous_message_id,
        )
        log_flow_event(
            self._call_logger,
            "stt_final_empty_recovery",
            action="rearm_agent_wait_timeout",
            previous_source=previous_source,
            message_id=previous_message_id,
            user_seq=self.state.user_final_seq,
            stage=self.state.current_stage,
        )
        return True

    def _recover_after_empty_stt_final(self) -> None:
        if self.finalized.is_set():
            self._empty_stt_recovery_pending = False
            return
        if self.speaking.is_set() or self.state.gap.active:
            if not self._pre_backend_wait_notice_is_current_speech():
                self._cancel_pre_backend_wait_notice(reason="empty_stt_final")
            self._empty_stt_recovery_pending = True
            log_flow_event(
                self._call_logger,
                "stt_final_empty_recovery",
                action="defer_while_agent_active",
                stage=self.state.current_stage,
                speaking=self.speaking.is_set(),
                gap_active=self.state.gap.active,
            )
            return
        self._cancel_pre_backend_wait_notice(reason="empty_stt_final")
        self._empty_stt_recovery_pending = False
        if self._restore_agent_wait_timeout_after_empty_stt():
            return

        self.arm_idle_timer(reason="empty_stt_final")
        log_flow_event(
            self._call_logger,
            "stt_final_empty_recovery",
            action="arm_idle_timer",
            stage=self.state.current_stage,
        )

    def handle_empty_stt_final(self, *, transcript_len: int = 0, source: str = "stt_final") -> None:
        deferred = self.state.deferred_interruption
        if source != "user_input_transcribed":
            # O provider avisa o final vazio antes de devolver o SpeechEvent. Em
            # raw_json o transcript cru nao e vazio, o LiveKit repassa o evento e
            # a mesma fala chegaria duas vezes ate a fila VAD.
            self._deferred_empty_final_latched_at = time.monotonic()
        if deferred.replay_in_flight:
            log_flow_event(self._call_logger, "deferred_replay_stt_discarded", text="")
            return
        if self._handle_deferred_stt_final(transcription="", text=""):
            return

        log_flow_event(
            self._call_logger,
            "stt_final_empty",
            reason="empty_text",
            source=source,
            stage=self.state.current_stage,
            speaking=self.speaking.is_set(),
            transcript_len=transcript_len,
            pre_backend_wait_reserved=self._pre_backend_wait_notice_reserved,
        )
        if self._timeline is not None:
            self._timeline.emit(
                "user_transcript_final_empty",
                current_stage=self.state.current_stage,
                speaking=self.speaking.is_set(),
                transcript_len=transcript_len,
                pre_backend_wait_reserved=self._pre_backend_wait_notice_reserved,
            )
        self._recover_after_empty_stt_final()

    async def _stop_due_to_agent_wait_timeout(
        self,
        *,
        reason: str,
        source: str,
        message_id: str = "",
    ) -> None:
        async with self._terminal_stop_lock:
            if self._terminal_stop_sent:
                return
            self._terminal_stop_sent = True

        self.cancel_idle_timer("agent_wait_timeout")
        self.cancel_idle_close("agent_wait_timeout")
        if self._timeline is not None:
            self._timeline.emit(
                "agent_wait_timeout_fired",
                status=AGENT_WAIT_TIMEOUT_STOP_STATUS,
                reason=reason,
                source=source,
                message_id=message_id,
            )

        self._restore_agent_wait_timeout_retry_vad_threshold(
            reason=f"agent_wait_timeout:{reason}"
        )
        await self.notify_bridge_stop(
            status=AGENT_WAIT_TIMEOUT_STOP_STATUS,
            reason=reason,
            phase="in_session",
        )
        self._log_resource_error_events(
            status=AGENT_WAIT_TIMEOUT_STOP_STATUS,
            reason=reason,
            source=source,
            message_id=message_id,
            terminal=True,
        )
        self.finalized.set()

    def _arm_agent_wait_timeout(
        self,
        config: AgentWaitTimeoutConfig,
        *,
        user_seq: int,
        source: str,
        message_id: str = "",
    ) -> None:
        timeout_s = config.timeout_s
        if timeout_s <= 0:
            self.cancel_agent_wait_timeout(f"clear:{source}")
            self._clear_agent_wait_timeout_memory()
            return

        timeout_message_id = self._resolve_agent_message_id(message_id, generate=True)
        self._remember_agent_wait_timeout(
            config,
            user_seq=user_seq,
            source=source,
            message_id=timeout_message_id,
        )
        retry_messages = config.retry_messages
        self.cancel_idle_timer("agent_wait_timeout")
        self.cancel_idle_close("agent_wait_timeout")
        self.cancel_agent_wait_timeout(f"rearm:{source}")
        reason = self._idle_policy.end_reason
        token = 0

        def _should_continue() -> bool:
            if not self._scheduler.is_current("agent_wait_timeout", token):
                return False
            if user_seq != self.state.user_final_seq:
                return False
            if self.finalized.is_set():
                return False
            return True

        async def _fire() -> None:
            for attempt, retry_text in enumerate(retry_messages, start=1):
                try:
                    await asyncio.sleep(timeout_s)
                except asyncio.CancelledError:
                    return

                if not _should_continue():
                    return

                self.cancel_idle_timer("agent_wait_timeout_retry")
                self.cancel_idle_close("agent_wait_timeout_retry")
                inactivity_message_id = self._agent_wait_timeout_inactivity_message_id(
                    message_id=timeout_message_id,
                    attempt=attempt,
                )
                self._call_logger.info(
                    "AGENT_WAIT_TIMEOUT_RETRY | attempt=%s/%s | wait_s=%.1f | message_id=%s | text=%r",
                    attempt,
                    len(retry_messages),
                    timeout_s,
                    inactivity_message_id,
                    retry_text,
                )
                if self._timeline is not None:
                    self._timeline.emit(
                        "agent_wait_timeout_retry",
                        wait_timeout_s=timeout_s,
                        attempt=attempt,
                        max_attempts=len(retry_messages),
                        reason=reason,
                        source=source,
                        user_seq=user_seq,
                        text=retry_text,
                        message_id=inactivity_message_id,
                    )

                self._activate_agent_wait_timeout_retry_vad_threshold(
                    attempt=attempt,
                    reason="agent_wait_timeout_retry",
                )
                try:
                    if self._agent.pipeline is not None:
                        await self.execute(InjectIdleNudge(retry_text))
                except Exception:
                    self._logger.debug("IDLE_NUDGE_PIPELINE_HOOK_FAIL", exc_info=True)

                try:
                    await self.say_stage(
                        retry_text,
                        "AGENT_WAIT_TIMEOUT_RETRY",
                        add_to_chat_ctx=False,
                        schedule_idle_after=False,
                        message_id=inactivity_message_id,
                    )
                except Exception:
                    self._logger.exception("[tts] failed speaking agent wait retry")
                    if self._timeline is not None:
                        self._timeline.emit(
                            "tts_stage_failed",
                            stage="AGENT_WAIT_TIMEOUT_RETRY",
                            source="agent_wait_timeout",
                            message_id=inactivity_message_id,
                        )
                    await self.terminate_with_resource_stop(
                        resource="tts",
                        source="agent_wait_timeout_retry_tts",
                    )
                    return

                if not _should_continue():
                    return

            try:
                await asyncio.sleep(timeout_s)
            except asyncio.CancelledError:
                return

            if not _should_continue():
                return
            await self._stop_due_to_agent_wait_timeout(
                reason=reason,
                source=source,
                message_id=timeout_message_id,
            )

        token = self._scheduler.arm(
            "agent_wait_timeout",
            task_name="agent_wait_timeout_timer",
            coro=_fire(),
        )
        if self._timeline is not None:
            self._timeline.emit(
                "agent_wait_timeout_armed",
                wait_timeout_s=timeout_s,
                retry_messages=len(retry_messages),
                reason=reason,
                source=source,
                user_seq=user_seq,
            )

    def arm_agent_wait_timeout_from_reply(
        self,
        reply: BackendReply,
        *,
        user_seq: int,
        source: str,
    ) -> None:
        stage = str(reply.stage or self.state.current_stage or "").strip().upper()
        if reply.done or self._finalization_policy.is_done_stage(stage):
            self.cancel_agent_wait_timeout(f"done:{source}")
            self._clear_agent_wait_timeout_memory()
            self._clear_agent_wait_timeout_config()
            return

        expects_user_response = self._reply_expects_user_response(reply)
        if expects_user_response is False:
            self.cancel_agent_wait_timeout(f"not_expected:{source}")
            self._clear_agent_wait_timeout_memory()
            return

        incoming_config = self._wait_timeout_config_from_reply(reply)
        has_wait_timeout_metadata = self._reply_has_wait_timeout_metadata(reply)
        if incoming_config is not None:
            self._agent_wait_timeout_config = incoming_config
        elif has_wait_timeout_metadata:
            self.cancel_agent_wait_timeout(f"disable:{source}")
            self._clear_agent_wait_timeout_memory()
            self._clear_agent_wait_timeout_config()
            return
        elif not reply.text:
            self._clear_agent_wait_timeout_memory()
            return

        config = self._agent_wait_timeout_config
        if config is None:
            self._clear_agent_wait_timeout_memory()
            return

        self._arm_agent_wait_timeout(
            config,
            user_seq=user_seq,
            source=source,
            message_id=self._message_id_from_reply(reply),
        )

    async def notify_bridge_done(self, reason: str = "stage_done") -> None:
        try:
            await self.execute(NotifyBridgeDone(reason))
            if self._timeline is not None:
                self._timeline.emit("bridge_done_requested", reason=reason)
        except Exception as exc:
            self._log_resource_error_events(
                status="stop_bridge_failed",
                reason="bridge_notify_done_failed",
                resource="bridge",
                source="notify_bridge_done",
                exc=exc,
            )
            self._logger.exception("[bridge] Falha enviando DONE")

    async def finalize(self, reason: str) -> None:
        if self._finalization_policy.should_skip(finalized=self.finalized.is_set()):
            return

        async with self.finalize_lock:
            if self._finalization_policy.should_skip(finalized=self.finalized.is_set()):
                return
            self._restore_agent_wait_timeout_retry_vad_threshold(
                reason=f"finalize:{reason}"
            )
            if self._timeline is not None:
                self._timeline.emit("finalize_started", reason=reason)

            self._cancel_backend_push_listener(reason=f"finalize:{reason}", force=True)
            self.cancel_idle_timer("finalize")
            self.cancel_idle_close("finalize")
            self.cancel_agent_wait_timeout("finalize")
            self._reset_deferred_interruption()
            self.release_user_audio_input(reason=f"finalize:{reason}")

            dur = time.monotonic() - self._call_t0
            self._call_logger.info(
                "CALL_END | room=%s | protocol=%s | session_id=%s | reason=%s | duration_s=%.1f",
                self._ctx.room.name,
                self._protocol,
                self._session_id,
                reason,
                dur,
            )

            try:
                await asyncio.wait_for(self._agent._ready.wait(), timeout=5.0)
            except Exception:
                pass

            try:
                raw_reply = await self.execute(EndServiceOnce())
                reply = self._coerce_backend_reply(
                    raw_reply,
                    fallback_stage=self.state.current_stage,
                )
                self.state.current_stage = reply.stage
                await self._speak_backend_reply(
                    reply,
                    source="end_service_once",
                    add_to_chat_ctx=False,
                )
                export_payload = reply.export_payload if reply.export_payload is not None else []
                await self.execute(ExportSession(output=export_payload, session_id=self._session_id))
            except Exception:
                self._logger.exception(
                    "[finalize] Falha ao finalizar/exportar session_id=%s",
                    self._session_id,
                )

            self.finalized.set()
            if self._timeline is not None:
                self._timeline.emit("finalize_completed", reason=reason)

    def cancel_idle_timer(self, reason: str = "cancel") -> None:
        self._scheduler.cancel("idle", reason=reason, log_name="IDLE_TIMER_CANCEL")

    def cancel_idle_close(self, reason: str = "cancel") -> None:
        self._scheduler.cancel("idle_close", reason=reason, log_name="IDLE_CLOSE_CANCEL")

    def note_user_speech_activity(self, *, reason: str) -> None:
        self.cancel_idle_timer(reason)
        self.cancel_idle_close(reason)
        self.cancel_agent_wait_timeout(reason)
        self.state.idle_nudge_count = 0
        if self._timeline is not None:
            self._timeline.emit(
                "user_speech_activity_detected",
                reason=reason,
                speaking=self.speaking.is_set(),
                current_stage=self.state.current_stage,
            )

    async def _end_due_to_no_response(self) -> None:
        end_reason = self._idle_policy.end_reason
        self._restore_agent_wait_timeout_retry_vad_threshold(reason="idle_no_response")
        self._log_resource_error_events(
            status=AGENT_WAIT_TIMEOUT_STOP_STATUS,
            reason=end_reason,
            source="idle_no_response",
            terminal=True,
        )
        self._create_task_logged(
            self.notify_bridge_done(end_reason),
            name="bridge_done_noresp",
        )
        self._create_task_logged(
            self.finalize(end_reason),
            name="finalize_noresp",
        )

    def arm_idle_close_timer(self, *, reason: str) -> None:
        if self._idle_policy.close_delay_s <= 0:
            self._create_task_logged(
                self._end_due_to_no_response(),
                name="idle_close_immediate",
            )
            return

        self.cancel_idle_timer("close_arm")
        self.cancel_idle_close(f"rearm:{reason}")
        seq_snapshot = self.state.user_final_seq
        token = 0

        async def _close() -> None:
            try:
                await asyncio.sleep(self._idle_policy.close_delay_s)
            except asyncio.CancelledError:
                return

            if not self._idle_policy.should_fire_close(
                token_is_current=self._scheduler.is_current("idle_close", token),
                seq_matches=self.state.user_final_seq == seq_snapshot,
                finalized=self.finalized.is_set(),
                current_stage=self.state.current_stage,
            ):
                return

            t0 = time.monotonic()
            while (
                (self.speaking.is_set() or self.state.gap.active)
                and not self.finalized.is_set()
                and (time.monotonic() - t0) < 3.0
            ):
                await asyncio.sleep(0.25)
                if not self._scheduler.is_current("idle_close", token):
                    return
                if self.state.user_final_seq != seq_snapshot:
                    return

            self._call_logger.info(
                "IDLE_CLOSE_FIRE | after_max_tries=%s | wait_s=%.1f | reason=%s",
                self._idle_policy.max_tries,
                self._idle_policy.close_delay_s,
                self._idle_policy.end_reason,
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "idle_close_fire",
                    reason=self._idle_policy.end_reason,
                    max_tries=self._idle_policy.max_tries,
                )
            await self._end_due_to_no_response()

        token = self._scheduler.arm(
            "idle_close",
            task_name="idle_close_timer",
            coro=_close(),
        )

    def arm_idle_timer(self, *, reason: str, delay_s: Optional[float] = None) -> None:
        delay = self._idle_policy.resolve_delay(delay_s)
        if not self._idle_policy.should_arm_nudge(delay_s=delay, nudge_text=self._nudge_text):
            return

        self.cancel_idle_timer(f"rearm:{reason}")
        seq_snapshot = self.state.user_final_seq
        token = 0

        async def _timer() -> None:
            try:
                await asyncio.sleep(delay)
            except asyncio.CancelledError:
                return

            if not self._idle_policy.should_fire_nudge(
                token_is_current=self._scheduler.is_current("idle", token),
                seq_matches=self.state.user_final_seq == seq_snapshot,
                speaking=self.speaking.is_set(),
                gap_active=self.state.gap.active,
                finalized=self.finalized.is_set(),
                current_stage=self.state.current_stage,
            ):
                return

            if self._idle_policy.should_arm_close_before_fire(nudge_count=self.state.idle_nudge_count):
                self._call_logger.info(
                    "IDLE_NUDGE_MAX_REACHED | tries=%s | will_close_in_s=%.1f",
                    self.state.idle_nudge_count,
                    self._idle_policy.close_delay_s,
                )
                self.arm_idle_close_timer(reason="max_reached_before_fire")
                return

            self.state.idle_nudge_count += 1
            self._call_logger.info(
                "IDLE_NUDGE_FIRE | attempt=%s/%s | delay_s=%.1f | text=%r",
                self.state.idle_nudge_count,
                self._idle_policy.max_tries,
                delay,
                self._nudge_text,
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "idle_nudge_fire",
                    attempt=self.state.idle_nudge_count,
                    max_tries=self._idle_policy.max_tries,
                    text=self._nudge_text,
                )

            try:
                if self._agent.pipeline is not None:
                    await self.execute(InjectIdleNudge(self._nudge_text))
            except Exception:
                self._logger.debug("IDLE_NUDGE_PIPELINE_HOOK_FAIL", exc_info=True)

            inactivity_message_id = self._idle_nudge_message_id(
                message_id=self._current_base_message_id(),
                attempt=self.state.idle_nudge_count,
            )
            await self.say_stage(
                self._nudge_text,
                "IDLE_NUDGE",
                add_to_chat_ctx=False,
                message_id=inactivity_message_id,
            )

        token = self._scheduler.arm(
            "idle",
            task_name="idle_timer",
            coro=_timer(),
        )

    def should_ignore_backchannel(self, text: str) -> bool:
        return False

    def _remember_interrupt_task(self, task: asyncio.Task[Any]) -> asyncio.Task[Any]:
        self.state.current_speech.interruption_task = task
        self._last_interrupt_task = task
        return task

    def request_current_speech_interruption(
        self,
        *,
        reason: str,
        force: bool = False,
    ) -> asyncio.Task[Any] | None:
        if not force and not self._current_speech_allows_interruption():
            return None

        self._mark_structured_interruption(reason=reason)
        existing = self.state.current_speech.interruption_task
        if self.state.current_speech.interruption_requested and existing is not None:
            return existing

        self.state.current_speech.interruption_requested = True

        handle = self.state.current_speech.handle
        speech_id = self.state.current_speech.speech_id
        stage = self.state.current_speech.stage

        self._call_logger.info(
            "BARGE_IN_INTERRUPT_REQUEST | reason=%s | stage=%s | speech_id=%s | has_handle=%s",
            reason,
            stage or "-",
            speech_id or "-",
            handle is not None,
        )
        if self._timeline is not None:
            self._timeline.emit(
                "barge_in_interrupt_requested",
                reason=reason,
                stage=stage or "",
                speech_id=speech_id,
                has_handle=handle is not None,
            )

        async def _interrupt() -> None:
            if handle is None:
                return
            await self.execute(InterruptSpeech(handle, force=force))

        return self._remember_interrupt_task(
            self._create_task_logged(
                _interrupt(),
                name=f"interrupt_current_speech_{reason}",
            )
        )

    @staticmethod
    def _track_publication_fields(publication: Any) -> dict[str, Any]:
        source = getattr(publication, "source", "")
        source_name = getattr(source, "name", None) or str(source)
        track = getattr(publication, "track", None)
        return {
            "track_sid": str(
                getattr(publication, "sid", "")
                or getattr(track, "sid", "")
                or ""
            ),
            "track_name": str(getattr(publication, "name", "") or getattr(track, "name", "") or ""),
            "source": source_name,
            "kind": str(getattr(publication, "kind", "") or getattr(track, "kind", "") or ""),
            "muted": bool(getattr(publication, "muted", False)),
            "subscribed": bool(getattr(publication, "subscribed", False)),
            "has_track": track is not None,
        }

    def log_room_input_snapshot(self, *, reason: str) -> None:
        room = getattr(self._ctx, "room", None)
        participants = getattr(room, "remote_participants", {}) or {}
        bridge_identity = (self._bridge_identity or "").strip()
        participant = participants.get(bridge_identity) if bridge_identity else None

        if participant is None and len(participants) == 1:
            participant = next(iter(participants.values()))

        if participant is None:
            log_flow_event(
                self._call_logger,
                "agent_input_snapshot",
                reason=reason,
                bridge_identity=bridge_identity,
                participant_found=False,
                remote_participants=len(participants),
            )
            return

        publications = list((getattr(participant, "track_publications", {}) or {}).values())
        audio_publications = [
            publication
            for publication in publications
            if "audio" in str(getattr(publication, "kind", "") or "").lower()
            or "microphone" in str(getattr(publication, "source", "") or "").lower()
        ]
        selected = audio_publications[0] if audio_publications else (publications[0] if publications else None)
        fields = self._track_publication_fields(selected) if selected is not None else {}
        log_flow_event(
            self._call_logger,
            "agent_input_snapshot",
            reason=reason,
            bridge_identity=bridge_identity,
            participant_found=True,
            participant=str(getattr(participant, "identity", "") or ""),
            remote_participants=len(participants),
            publication_count=len(publications),
            audio_publication_count=len(audio_publications),
            **fields,
        )

    def mark_interrupt_from_handle_if_any(self, *, reason: str) -> asyncio.Task[Any]:
        handle = self.state.current_speech.handle
        speech_id = self.state.current_speech.speech_id
        if handle is None:
            task = self._create_task_logged(
                self.execute(
                    SetPendingInterrupt(
                        listened_text="",
                        skipped=True,
                        speech_id=speech_id,
                    )
                ),
                name="pending_interrupt_nohandle",
            )
            self._call_logger.info(
                "INTERRUPT_MARK | reason=%s | skipped=True | speech_id=%s | spoken=''",
                reason,
                speech_id or "-",
            )
            log_flow_event(
                self._call_logger,
                "interrupt_ignored",
                reason="no_current_speech_handle",
                source=reason,
                stage=self.state.current_speech.stage or self.state.current_stage,
                skipped=True,
                speech_id=speech_id,
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "interrupt_marked",
                    reason=reason,
                    skipped=True,
                    spoken="",
                    speech_id=speech_id,
                )
            self.state.current_speech.interruption_requested = True
            return self._remember_interrupt_task(task)

        spoken = self.execute_now(ExtractSpokenText(handle))
        skipped = not bool((spoken or "").strip())
        task = self._create_task_logged(
            self.execute(
                SetPendingInterrupt(
                    listened_text=spoken,
                    skipped=skipped,
                    speech_id=speech_id,
                )
            ),
            name="pending_interrupt_handle",
        )
        self._call_logger.info(
            "INTERRUPT_MARK | reason=%s | stage=%s | skipped=%s | speech_id=%s | spoken=%r",
            reason,
            self.state.current_speech.stage or "-",
            skipped,
            speech_id or "-",
            spoken,
        )
        log_flow_event(
            self._call_logger,
            "interrupt_marked",
            reason=reason,
            stage=self.state.current_speech.stage or self.state.current_stage,
            skipped=skipped,
            text=spoken,
            speech_id=speech_id,
        )
        if skipped:
            log_flow_event(
                self._call_logger,
                "interrupt_ignored",
                reason="empty_spoken_text",
                source=reason,
                stage=self.state.current_speech.stage or self.state.current_stage,
                skipped=True,
                speech_id=speech_id,
            )
        if self._timeline is not None:
            self._timeline.emit(
                "interrupt_marked",
                reason=reason,
                stage=self.state.current_speech.stage or "",
                skipped=skipped,
                spoken=spoken,
                speech_id=speech_id,
            )
        self.state.current_speech.interruption_requested = True
        return self._remember_interrupt_task(task)

    async def say_stage(
        self,
        text: str,
        stage: str,
        *,
        add_to_chat_ctx: bool = True,
        allow_interruptions: Optional[bool] = None,
        rearm_interruptions_on_next_user_speech: bool = False,
        speech_id: str = "",
        message_id: str = "",
        schedule_idle_after: bool = True,
        wait_post_final_grace: bool = True,
        require_user_not_speaking: bool = False,
        abort_if_agent_spoke_since: int | None = None,
        audio: Any = None,
        audio_duration_ms: int | None = None,
        drop_user_input_while_speaking: bool = False,
        agent_message_type: str = "",
        agent_result_type: str = "",
        erro_msg: str | None = None,
        playout_result_callback: Optional[Callable[..., None]] = None,
    ) -> bool:
        text = (text or "").strip()
        if not text:
            return False
        message_id = str(message_id or "").strip()
        stage_key = str(stage or "").strip().upper()
        if stage_key not in AGENT_WAIT_TIMEOUT_RETRY_VAD_STAGES:
            self._restore_agent_wait_timeout_retry_vad_threshold(
                reason=f"agent_stage:{stage_key or 'UNKNOWN'}"
            )
        if require_user_not_speaking and not self._user_not_speaking.is_set():
            log_flow_event(
                self._call_logger,
                "tts_skipped",
                stage=stage,
                reason="user_speaking",
                message_id=message_id or self._current_base_message_id(),
                text=text,
            )
            return False

        self._clear_post_playout_drop_window(reason="new_tts")
        allow_int = (
            self._interrupt_policy.allow_stage(stage)
            if allow_interruptions is None
            else bool(allow_interruptions)
        )
        speech_id = str(speech_id or "").strip()
        message_id = self._resolve_agent_message_id(message_id, generate=True)

        self.cancel_idle_timer("bot_speaking")
        self.cancel_idle_close("bot_speaking")

        if self.state.gap.active and self.state.gap.cancelled:
            self._call_logger.info("BOT_SAY_SKIPPED | stage=%s | reason=gap_cancelled", stage)
            self.reset_gap_state()
            return False

        if self.state.gap.active and self.state.user_final_seq != self.state.gap.guard_seq:
            await self.execute(
                SetPendingInterrupt(
                    listened_text="",
                    skipped=True,
                    speech_id=self.state.gap.speech_id,
                )
            )
            self._call_logger.info("BOT_SAY_SKIPPED | stage=%s | reason=user_spoke_in_gap", stage)
            self.reset_gap_state()
            return False

        async with self.say_lock:
            if require_user_not_speaking and not self._user_not_speaking.is_set():
                log_flow_event(
                    self._call_logger,
                    "tts_skipped",
                    stage=stage,
                    reason="user_speaking_after_lock",
                    message_id=message_id,
                    text=text,
                )
                return False
            if (
                abort_if_agent_spoke_since is not None
                and self._say_stage_seq != abort_if_agent_spoke_since
            ):
                # Outra fala ocupou o say_lock enquanto este audio esperava na
                # fila: o silencio que ele mascararia ja foi preenchido.
                log_flow_event(
                    self._call_logger,
                    "tts_skipped",
                    stage=stage,
                    reason="agent_spoke_after_schedule",
                    message_id=message_id,
                    text=text,
                )
                return False
            self._say_stage_seq += 1
            self.state.current_speech.stage = stage
            self.state.current_speech.text = text
            self.state.current_speech.speech_id = speech_id
            self.state.current_speech.drop_user_input_while_speaking = bool(
                drop_user_input_while_speaking
            )
            self.state.current_speech.agent_message_type = str(agent_message_type or "").strip()
            self.state.current_speech.agent_result_type = str(agent_result_type or "").strip()
            self.state.current_speech.allow_interruptions = allow_int
            self.state.current_speech.rearm_interruptions_on_next_user_speech = bool(
                rearm_interruptions_on_next_user_speech
            )
            self.state.current_speech.interruption_requested = False
            self.state.current_speech.interruption_task = None
            self.speaking.set()
            log_flow_event(
                self._call_logger,
                "tts_start",
                stage=stage,
                allow_interruptions=allow_int,
                message_id=message_id,
                text=text,
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "tts_stage_started",
                    stage=stage,
                    allow_interruptions=allow_int,
                    text=text,
                )
            self._emit_debug_event(
                "agent.speech.started",
                stage=stage,
                message_id=message_id,
                text=text,
            )
            seq_at_start = self.state.user_final_seq
            handle = None
            tts_started_ns = None
            tts_started_monotonic = 0.0
            agent_speaking_seq_at_start = 0
            tts_tffb_ms = None
            tts_audio_duration_ms = None
            tts_playout_completed = False

            def _tts_was_interrupted() -> bool:
                return bool(getattr(handle, "interrupted", False)) or (
                    self.state.user_final_seq != seq_at_start
                )

            try:
                self.reset_gap_state()
                if wait_post_final_grace:
                    await self._wait_post_user_final_grace(stage=stage)

                async def _start_speech_attempt():
                    started_ns = time.time_ns()
                    started_monotonic = time.monotonic()
                    speaking_seq_at_start = self._agent_speaking_seq
                    started_handle = await self.execute(
                        StartSpeech(
                            text=text,
                            allow_interruptions=allow_int,
                            add_to_chat_ctx=add_to_chat_ctx,
                            audio=audio,
                        )
                    )
                    self.state.current_speech.handle = started_handle

                    self._call_logger.info(
                        "BOT_SAY_HANDLE | stage=%s | speech_handle_id=%s | seq_at_start=%s | current_seq=%s",
                        stage,
                        self._speech_handle_id(started_handle) or "-",
                        seq_at_start,
                        self.state.user_final_seq,
                    )
                    return (
                        started_handle,
                        started_ns,
                        started_monotonic,
                        speaking_seq_at_start,
                    )

                (
                    handle,
                    tts_started_ns,
                    tts_started_monotonic,
                    agent_speaking_seq_at_start,
                ) = await _start_speech_attempt()

                try:
                    should_abort_tts = False
                    abort_reason = "empty_frame_timeout"
                    retry_error: Exception | None = None
                    retryable_error_detail: str | None = None
                    partial_failure: _TTSPartialFailure | None = None
                    if audio is None:
                        playout_start_timeout_s = self._tts_playout_start_timeout_s()
                        try:
                            should_abort_tts = await self._wait_for_tts_playout_or_empty_frame_retry(
                                handle,
                                started_monotonic=tts_started_monotonic,
                                agent_speaking_seq_at_start=agent_speaking_seq_at_start,
                                timeout_s=playout_start_timeout_s,
                                stage=stage,
                                message_id=message_id,
                            )
                        except Exception as exc:
                            if not self._is_retryable_tts_playout_error(exc):
                                raise
                            retry_error = exc
                            abort_reason = "playout_error"
                            should_abort_tts = True
                        if not should_abort_tts:
                            partial_failure = self._pop_tts_partial_failure(handle)
                            if partial_failure is not None:
                                abort_reason = "provider_partial_failure"
                                retryable_error_detail = partial_failure.detail
                                should_abort_tts = True
                            else:
                                retryable_error_detail = self._pop_retryable_tts_error_detail(handle)
                                observed_audio_duration_ms = self._resolve_tts_audio_duration_ms(
                                    handle
                                )
                                if (
                                    retryable_error_detail
                                    and (
                                        observed_audio_duration_ms is None
                                        or "xai tts partial audio failure" in retryable_error_detail
                                    )
                                    and not bool(getattr(handle, "interrupted", False))
                                    and self.state.user_final_seq == seq_at_start
                                ):
                                    abort_reason = "provider_error_no_audio_metric"
                                    should_abort_tts = True
                                    log_flow_event(
                                        self._call_logger,
                                        "tts_playout_missing_audio_after_error",
                                        stage=stage,
                                        message_id=message_id,
                                        speech_handle_id=self._speech_handle_id(handle),
                                        detail=retryable_error_detail,
                                    )
                    else:
                        await self.execute(WaitForSpeechPlayout(handle))

                    if should_abort_tts:
                        if not bool(getattr(handle, "interrupted", False)):
                            try:
                                await self.execute(InterruptSpeech(handle, force=True))
                            except Exception as exc:
                                log_flow_event(
                                    self._call_logger,
                                    "tts_failure_interrupt_error",
                                    stage=stage,
                                    message_id=message_id,
                                    speech_handle_id=self._speech_handle_id(handle),
                                    error=type(exc).__name__,
                                )
                        tts_total_ms = self._resolve_tts_provider_duration_ms(handle)
                        tts_audio_duration_ms = self._resolve_tts_audio_duration_ms(handle)
                        if partial_failure is not None:
                            tts_total_ms = (
                                partial_failure.provider_synthesis_ms or tts_total_ms
                            )
                            tts_audio_duration_ms = (
                                partial_failure.pcm_duration_ms
                                or tts_audio_duration_ms
                            )
                        if retry_error is not None:
                            retry_error_detail = f"{type(retry_error).__name__}: {retry_error}"
                            http_cod_status, http_cod_desc = _http_status_details_from_exception(
                                retry_error
                            )
                        elif retryable_error_detail:
                            retry_error_detail = retryable_error_detail
                            http_cod_status, http_cod_desc = None, None
                        else:
                            retry_error_detail = (
                                "TimeoutError: nenhum frame de audio do TTS foi observado "
                                f"em {round(playout_start_timeout_s * 1000)} ms"
                            )
                            http_cod_status, http_cod_desc = None, None
                        tts_error_message_id = self._tts_error_message_id(
                            message_id=message_id
                        )
                        tts_max_gap_ms = self._tts_failure_metric(retry_error_detail, "max_audio_delta_gap_ms")
                        tts_max_underrun_0ms = self._tts_failure_metric(retry_error_detail, "xai_underrun_estimado_ms")
                        tts_underflow_count = self._tts_failure_metric(retry_error_detail, "xai_micro_underflows")
                        tts_avg_underflow_ms = self._tts_failure_metric(retry_error_detail, "xai_avg_underrun_ms")
                        log_structured_event(
                            self._call_logger,
                            self._structured_log_context,
                            tipo_evento=EVENT_ENVIO_MSG,
                            message_id=tts_error_message_id,
                            inicio_ns=tts_started_ns,
                            fim_ns=time.time_ns(),
                            latencia_total_ms=tts_total_ms,
                            duracao_audio_ms=tts_audio_duration_ms,
                            tts_max_gap_ms=tts_max_gap_ms,
                            tts_max_underrun_0ms=tts_max_underrun_0ms,
                            tts_underflow_count=tts_underflow_count,
                            tts_avg_underflow_ms=tts_avg_underflow_ms,
                            interrupcao=True if _tts_was_interrupted() else None,
                            erro_msg="Falha TTS",
                            erro_detalhe=retry_error_detail,
                            http_cod_status=http_cod_status,
                            http_cod_desc=http_cod_desc,
                        )
                        log_flow_event(
                            self._call_logger,
                            "tts_failure",
                            stage=stage,
                            message_id=tts_error_message_id,
                            reason=abort_reason,
                            erro_msg="Falha TTS",
                            erro_detalhe=retry_error_detail,
                            detail=retry_error_detail,
                            duration_ms=tts_total_ms,
                            audio_ms=tts_audio_duration_ms,
                            timeout_ms=round(playout_start_timeout_s * 1000),
                            timeout_kind="playout_start",
                            first_frame_timeout_ms=round(self._tts_first_frame_timeout_s() * 1000),
                            connection_grace_ms=round(self._tts_connection_start_grace_s() * 1000),
                            error=type(retry_error).__name__ if retry_error is not None else "",
                        )
                        self._call_logger.warning(
                            "BOT_TTS_FAILURE | stage=%s | message_id=%s | reason=%s",
                            stage,
                            message_id or "-",
                            TTS_FAILURE_DETAIL,
                        )
                        log_flow_event(
                            self._call_logger,
                            "tts_failure_return_to_listening",
                            stage=stage,
                            message_id=message_id,
                            reason=abort_reason,
                            socket_action="reconnect_on_next_synthesis",
                            action="wait_for_client_repetition",
                        )
                        # The adapter has discarded the websocket after a partial failure.
                        # Do not replay a consumed LiveKit stream or add a recovery prompt.
                        return False

                    tts_finished_ns = time.time_ns()
                    await self._resolve_tts_tffb_ms(handle)
                    (
                        tts_total_ms,
                        tts_tffb_ms,
                        tts_audio_duration_ms,
                        tts_metric_duration_ms,
                        tts_metrics_source,
                    ) = self._resolve_tts_structured_metrics(handle)
                    tts_interrupted = _tts_was_interrupted()
                    (
                        tts_max_gap_ms,
                        tts_max_underrun_0ms,
                        tts_underflow_count,
                        tts_avg_underflow_ms,
                    ) = self._resolve_tts_turn_health_metrics(handle)
                    log_structured_event(
                        self._call_logger,
                        self._structured_log_context,
                        tipo_evento=EVENT_ENVIO_MSG,
                        message_id=message_id,
                        inicio_ns=tts_started_ns,
                        fim_ns=tts_finished_ns,
                        latencia_total_ms=tts_total_ms,
                        latencia_tffb_ms=tts_tffb_ms,
                        duracao_audio_ms=tts_audio_duration_ms,
                        tts_max_gap_ms=tts_max_gap_ms,
                        tts_max_underrun_0ms=tts_max_underrun_0ms,
                        tts_underflow_count=tts_underflow_count,
                        tts_avg_underflow_ms=tts_avg_underflow_ms,
                        interrupcao=True if tts_interrupted else None,
                        erro_msg=erro_msg,
                        http_cod_status=200,
                        http_cod_desc="OK",
                    )
                    log_flow_event(
                        self._call_logger,
                        "tts_done",
                        stage=stage,
                        message_id=message_id,
                        vocalized=True,
                        interrupted=bool(getattr(handle, "interrupted", False)),
                        interruption=tts_interrupted,
                        duration_ms=tts_total_ms,
                        provider_duration_ms=tts_metric_duration_ms,
                        metrics_source=tts_metrics_source,
                        tffb_ms=tts_tffb_ms,
                        audio_ms=tts_audio_duration_ms,
                        max_gap_ms=tts_max_gap_ms,
                        max_underrun_0ms=tts_max_underrun_0ms,
                        underflow_count=tts_underflow_count,
                        avg_underflow_ms=tts_avg_underflow_ms,
                        regenerated=False,
                    )
                    self.log_room_input_snapshot(reason=f"tts_done:{stage}")
                    if self._timeline is not None:
                        self._timeline.emit(
                            "tts_stage_playout_done",
                            stage=stage,
                            interrupted=bool(getattr(handle, "interrupted", False)),
                        )
                    await self._publish_debug_event(
                        "agent.speech.finished",
                        stage=stage,
                        message_id=message_id,
                        text=text,
                        interrupted=bool(getattr(handle, "interrupted", False)),
                        duration_ms=tts_total_ms,
                        ttfb_ms=tts_tffb_ms,
                        audio_duration_ms=tts_audio_duration_ms,
                    )
                    tts_playout_completed = True
                except Exception as exc:
                    tts_finished_ns = time.time_ns()
                    await self._resolve_tts_tffb_ms(handle)
                    (
                        tts_total_ms,
                        tts_tffb_ms,
                        tts_audio_duration_ms,
                        tts_metric_duration_ms,
                        tts_metrics_source,
                    ) = self._resolve_tts_structured_metrics(handle)
                    http_cod_status, http_cod_desc = _http_status_details_from_exception(exc)
                    tts_interrupted = _tts_was_interrupted()
                    tts_error_message_id = self._tts_error_message_id(
                        message_id=message_id
                    )
                    log_structured_event(
                        self._call_logger,
                        self._structured_log_context,
                        tipo_evento=EVENT_ENVIO_MSG,
                        message_id=tts_error_message_id,
                        inicio_ns=tts_started_ns,
                        fim_ns=tts_finished_ns,
                        latencia_total_ms=tts_total_ms,
                        latencia_tffb_ms=tts_tffb_ms,
                        duracao_audio_ms=tts_audio_duration_ms,
                        interrupcao=True if tts_interrupted else None,
                        erro_msg="Falha TTS",
                        erro_detalhe=f"{type(exc).__name__}: {exc}",
                        http_cod_status=http_cod_status,
                        http_cod_desc=http_cod_desc,
                    )
                    log_flow_event(
                        self._call_logger,
                        "tts_error",
                        stage=stage,
                        message_id=tts_error_message_id,
                        source_message_id=message_id,
                        duration_ms=tts_total_ms,
                        provider_duration_ms=tts_metric_duration_ms,
                        metrics_source=tts_metrics_source,
                        interruption=tts_interrupted,
                        tffb_ms=tts_tffb_ms,
                        audio_ms=tts_audio_duration_ms,
                        error=type(exc).__name__,
                    )
                    self._call_logger.exception(
                        "BOT_PLAYOUT_ERROR | stage=%s | duration_ms=%s | tffb_ms=%s",
                        stage,
                        tts_total_ms,
                        tts_tffb_ms if tts_tffb_ms is not None else "-",
                    )

                user_spoke_during = self.state.user_final_seq != seq_at_start
                handle_interrupted = bool(getattr(handle, "interrupted", False))
                if playout_result_callback is not None:
                    playout_result_callback(
                        interrupted=handle_interrupted,
                        user_spoke_during=user_spoke_during,
                    )

                self._call_logger.info(
                    "BOT_SAY_RESULT | stage=%s | interrupted=%s | user_spoke_during=%s",
                    stage,
                    handle_interrupted,
                    user_spoke_during,
                )
                if self._timeline is not None:
                    self._timeline.emit(
                        "tts_stage_result",
                        stage=stage,
                        interrupted=handle_interrupted,
                        user_spoke_during=user_spoke_during,
                    )

                if (
                    tts_playout_completed
                    and not handle_interrupted
                    and not user_spoke_during
                    and str(stage or "").strip().upper() == "PRESENTATION"
                    and not allow_int
                ):
                    post_playout_drop_context = self._protected_speech_drop_context_from_values(
                        reason="protected_speech_post_playout_grace",
                        stage=stage,
                        allow_interruptions=allow_int,
                        drop_user_input_while_speaking=True,
                        agent_message_type=agent_message_type,
                        agent_result_type=agent_result_type,
                    )
                    if post_playout_drop_context is not None:
                        self._arm_post_playout_drop_window(post_playout_drop_context)

                if user_spoke_during or handle_interrupted:
                    if self._config.final_grace_s > 0:
                        await asyncio.sleep(self._config.final_grace_s)
                    spoken = self.execute_now(ExtractSpokenText(handle))
                    skipped = not bool((spoken or "").strip())
                    await self.execute(
                        SetPendingInterrupt(
                            listened_text=spoken,
                            skipped=skipped,
                            speech_id=speech_id,
                        )
                    )

                    self._call_logger.info(
                        "BOT_INTERRUPTED_AFTER_START | stage=%s | skipped=%s | speech_id=%s | spoken=%r",
                        stage,
                        skipped,
                        speech_id or "-",
                        spoken,
                    )
                    log_flow_event(
                        self._call_logger,
                        "interrupt_marked",
                        reason="tts_interrupted_after_start",
                        stage=stage,
                        skipped=skipped,
                        text=spoken,
                    )
                    if skipped:
                        log_flow_event(
                            self._call_logger,
                            "interrupt_ignored",
                            reason="empty_spoken_text_after_tts",
                            stage=stage,
                            skipped=True,
                        )
                    if self._timeline is not None:
                        self._timeline.emit(
                            "tts_stage_interrupted",
                            stage=stage,
                            skipped=skipped,
                            spoken=spoken,
                            speech_id=speech_id,
                        )
            except Exception as exc:
                if tts_started_ns is None:
                    tts_started_ns = time.time_ns()
                tts_failed_ns = time.time_ns()
                await self._resolve_tts_tffb_ms(handle)
                (
                    tts_total_ms,
                    tts_tffb_ms,
                    tts_audio_duration_ms,
                    tts_metric_duration_ms,
                    tts_metrics_source,
                ) = self._resolve_tts_structured_metrics(handle)
                http_cod_status, http_cod_desc = _http_status_details_from_exception(exc)
                tts_interrupted = _tts_was_interrupted()
                tts_error_message_id = self._tts_error_message_id(
                    message_id=message_id
                )
                log_structured_event(
                    self._call_logger,
                    self._structured_log_context,
                    tipo_evento=EVENT_ENVIO_MSG,
                    message_id=tts_error_message_id,
                    inicio_ns=tts_started_ns,
                    fim_ns=tts_failed_ns,
                    latencia_total_ms=tts_total_ms,
                    latencia_tffb_ms=tts_tffb_ms,
                    duracao_audio_ms=tts_audio_duration_ms,
                    interrupcao=True if tts_interrupted else None,
                    erro_msg="Falha TTS",
                    erro_detalhe=f"{type(exc).__name__}: {exc}",
                    http_cod_status=http_cod_status,
                    http_cod_desc=http_cod_desc,
                )
                log_flow_event(
                    self._call_logger,
                    "tts_error",
                    stage=stage,
                    message_id=tts_error_message_id,
                    source_message_id=message_id,
                    duration_ms=tts_total_ms,
                    provider_duration_ms=tts_metric_duration_ms,
                    metrics_source=tts_metrics_source,
                    interruption=tts_interrupted,
                    tffb_ms=tts_tffb_ms,
                    audio_ms=tts_audio_duration_ms,
                    error=type(exc).__name__,
                )
                self._call_logger.exception(
                    "BOT_SAY_ERROR | stage=%s | duration_ms=%s | tffb_ms=%s",
                    stage,
                    tts_total_ms,
                    tts_tffb_ms if tts_tffb_ms is not None else "-",
                )
                raise
            finally:
                self.state.current_speech.handle = None
                self.state.current_speech.stage = ""
                self.state.current_speech.text = ""
                self.state.current_speech.speech_id = ""
                self.state.current_speech.allow_interruptions = False
                self.state.current_speech.rearm_interruptions_on_next_user_speech = False
                self.state.current_speech.drop_user_input_while_speaking = False
                self.state.current_speech.agent_message_type = ""
                self.state.current_speech.agent_result_type = ""
                self.state.current_speech.interruption_requested = False
                self.state.current_speech.interruption_task = None
                self.speaking.clear()
                self.reset_gap_state()
                if self._empty_stt_recovery_pending:
                    self._recover_after_empty_stt_final()

        if (
            schedule_idle_after
            and self.state.user_final_seq == seq_at_start
            and (stage or "").upper() != "DONE"
        ):
            if (
                self._idle_policy.should_arm_close_after_nudge(
                    stage=stage,
                    nudge_count=self.state.idle_nudge_count,
                )
            ):
                self._call_logger.info(
                    "IDLE_NUDGE_MAX_REACHED_AFTER_SAY | tries=%s | will_close_in_s=%.1f",
                    self.state.idle_nudge_count,
                    self._idle_policy.close_delay_s,
                )
                self.arm_idle_close_timer(reason="max_reached_after_nudge")
                return True

            self.arm_idle_timer(reason=f"bot_end:{(stage or '').upper()}")

        if not allow_int and not drop_user_input_while_speaking:
            async with self.pending_lock:
                pending_turn = self.state.pending_user_final
                self.state.pending_user_final = None
            if pending_turn:
                self._create_task_logged(
                    self.run_pipeline(
                        pending_turn.transcription,
                        pending_turn.text,
                        user_seq=pending_turn.seq,
                        message_id=pending_turn.message_id,
                    ),
                    name=f"run_pipeline_pending_{pending_turn.seq}",
                )

        return True

    async def finalize_when_room_empty(self) -> None:
        while not self.finalized.is_set():
            await asyncio.sleep(0.5)
            try:
                if self._finalization_policy.should_finalize_room_empty(
                    remote_participants=len(self._ctx.room.remote_participants),
                    finalized=self.finalized.is_set(),
                ):
                    await asyncio.sleep(self._finalization_policy.call_end_grace_s)
                    if self._finalization_policy.should_finalize_room_empty(
                        remote_participants=len(self._ctx.room.remote_participants),
                        finalized=self.finalized.is_set(),
                    ):
                        await self.finalize("room_empty")
                        return
            except Exception:
                continue

    async def run_pipeline(
        self,
        transcription: str,
        text: str,
        *,
        user_seq: int,
        message_id: str = "",
        inflight_initial_notice_consumed: bool = False,
        is_deferred_replay: bool = False,
    ) -> None:
        text = (text or "").strip()
        transcription = transcription or ""
        message_id = str(message_id or "").strip()
        if not message_id:
            message_id = self._consume_turn_message_id(transcription, text)
        elif text:
            self._clear_started_turn_message_id(message_id)

        if self.should_ignore_backchannel(text):
            log_flow_event(
                self._call_logger,
                "interrupt_ignored",
                reason="backchannel",
                message_id=message_id,
                user_seq=user_seq,
                stage=self.state.current_stage,
                speaking=self.speaking.is_set(),
                text=text,
            )
            if is_deferred_replay:
                self._reset_deferred_interruption()
            return

        await self._agent._ready.wait()
        if self._agent.pipeline is None:
            self._logger.exception("[pipeline] pipeline=None (ready)")
            if is_deferred_replay:
                self._reset_deferred_interruption()
            return

        async with self._agent._run_lock:
            if self.finalized.is_set() or self._finalization_policy.is_done_stage(
                self.state.current_stage
            ):
                log_flow_event(
                    self._call_logger,
                    "pipeline_run_skipped_terminal",
                    user_seq=user_seq,
                    stage=self.state.current_stage,
                )
                if is_deferred_replay:
                    self._reset_deferred_interruption()
                return

            if user_seq != self.state.user_final_seq:
                self._call_logger.info(
                    "RUN_PIPELINE_SKIP_STALE | user_seq=%s | current_seq=%s | text=%r",
                    user_seq,
                    self.state.user_final_seq,
                    text,
                )
                if self._timeline is not None:
                    self._timeline.emit(
                        "pipeline_run_skipped_stale",
                        user_seq=user_seq,
                        current_seq=self.state.user_final_seq,
                    )
                if is_deferred_replay:
                    self._reset_deferred_interruption()
                return

            pending_interrupt = await self._agent.consume_pending_interrupt()
            if len(pending_interrupt) == 3:
                listened_text, speech_id, skipped = pending_interrupt
            else:
                listened_text, skipped = pending_interrupt
                speech_id = ""
            try:
                if listened_text is not None:
                    if is_deferred_replay:
                        await self.execute(
                            SetPipelineProcessingInterruption(
                                listened_text=listened_text,
                                skipped=skipped,
                                speech_id=speech_id,
                            )
                        )
                    else:
                        await self.execute(
                            SetPipelineInterruption(
                                interrupted=True,
                                listened_text=listened_text,
                                skipped=skipped,
                                speech_id=speech_id,
                            )
                        )
                else:
                    await self.execute(SetPipelineInterruption(interrupted=False))
            except Exception:
                self._logger.exception("PIPELINE_INTERRUPT | erro chamando set_interruption")

            try:
                stt_payload = json.loads(transcription)
            except Exception:
                stt_payload = text
            stt_payload = self._payload_with_message_id(stt_payload, message_id, text=text)

            log_flow_event(
                self._call_logger,
                "agent_start",
                message_id=message_id,
                user_seq=user_seq,
                stage=self.state.current_stage,
                text=text,
                payload=stt_payload,
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "pipeline_run_started",
                    user_seq=user_seq,
                    input_text=text,
                    current_stage=self.state.current_stage,
                )
            if text and self._supports_inflight_backend_push():
                self._schedule_backend_push_listener(
                    user_seq=user_seq,
                    reason="run_pipeline_inflight",
                    inflight=True,
                )
            backend_started_ns = time.time_ns()
            deferred = self.state.deferred_interruption
            # A janela do ciclo cobre o backend e a espera de liquidacao. Depois
            # dela a resposta ja esta liberada e a fala do cliente volta a ser
            # barge-in comum.
            deferred.backend_in_flight = not is_deferred_replay
            if deferred.backend_in_flight:
                self._promote_pre_backend_vad_utterances()
            try:
                raw_reply = await self._execute_pipeline_with_inflight_backend_wait(
                    stt_payload,
                    user_seq=user_seq,
                    message_id=message_id,
                    initial_notice_consumed=inflight_initial_notice_consumed,
                )
            except InflightBackendWaitTimedOut:
                self._close_deferred_vad_window()
                self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
                return
            except Exception:
                self._close_deferred_vad_window()
                backend_duration_ms = event_latency_ms(backend_started_ns)
                log_backend_error = getattr(self._call_logger, "error", self._call_logger.exception)
                log_backend_error(
                    "AGENT_BACKEND_ERROR | user_seq=%s | stage=%s | duration_ms=%s",
                    user_seq,
                    self.state.current_stage,
                    backend_duration_ms,
                )
                log_flow_event(
                    self._call_logger,
                    "agent_error",
                    user_seq=user_seq,
                    stage=self.state.current_stage,
                    duration_ms=backend_duration_ms,
                )
                self._logger.exception("[pipeline] backend run failed")
                if self._timeline is not None:
                    self._timeline.emit(
                        "pipeline_run_failed",
                        user_seq=user_seq,
                        resource="agent_backend",
                    )
                self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
                await self.terminate_with_resource_stop(
                    resource="agent_backend",
                    source="run_pipeline",
                )
                return

            try:
                backend_duration_ms = event_latency_ms(backend_started_ns)
                reply = self._coerce_backend_reply(
                    raw_reply,
                    fallback_stage=self.state.current_stage,
                )
                reply = self._backend_reply_with_message_id(reply, message_id)
                if not is_deferred_replay:
                    await self._wait_deferred_interruption_settlement(user_seq=user_seq)
            finally:
                self._close_deferred_vad_window()

            terminal_reply = self._terminal_stop_from_reply(reply) is not None or (
                reply.done or self._finalization_policy.is_done_stage(reply.stage)
            )
            if user_seq != self.state.user_final_seq and not terminal_reply:
                self._call_logger.info(
                    "RUN_PIPELINE_DROP_AFTER_RUN | user_seq=%s | current_seq=%s | stage=%s",
                    user_seq,
                    self.state.user_final_seq,
                    reply.stage,
                )
                if self._timeline is not None:
                    self._timeline.emit(
                        "pipeline_run_dropped_after_run",
                        user_seq=user_seq,
                        current_seq=self.state.user_final_seq,
                        stage=reply.stage,
                    )
                if not is_deferred_replay and await self._dispatch_deferred_interruption(reply):
                    log_flow_event(
                        self._call_logger,
                        "pipeline_to_tts_skipped",
                        stage=reply.stage,
                        reason="deferred_interruption",
                        speech_id=self._speech_id_from_reply(reply),
                    )
                self.reset_gap_state()
                self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
                return

            self.state.current_stage = reply.stage
            self._update_backend_processing_drop_from_reply(reply, source="run_pipeline")
            out = reply.text
            log_flow_event(
                self._call_logger,
                "agent_done",
                message_id=message_id,
                user_seq=user_seq,
                stage=self.state.current_stage,
                done=reply.done,
                duration_ms=backend_duration_ms,
                text=out,
                result=reply.export_payload,
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "pipeline_run_completed",
                    user_seq=user_seq,
                    stage=self.state.current_stage,
                    output_len=len(out),
                )

            self._emit_debug_event(
                "agent.completed",
                user_seq=user_seq,
                message_id=message_id,
                stage=self.state.current_stage,
                duration_ms=backend_duration_ms,
                done=reply.done,
            )

            if not out:
                self._log_backend_reply_error_event(
                    reply,
                    source="run_pipeline",
                    message_id=message_id,
                )

                self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
            if out:
                self.state.gap.active = True
                self.state.gap.cancelled = False
                self.state.gap.guard_seq = user_seq
                self.state.gap.stage = self.state.current_stage
                self.state.gap.text = out
                self.state.gap.speech_id = self._speech_id_from_reply(reply)

                if (
                    self.state.user_final_seq != self.state.gap.guard_seq
                    and not terminal_reply
                ):
                    await self.execute(
                        SetPendingInterrupt(
                            listened_text="",
                            skipped=True,
                            speech_id=self.state.gap.speech_id,
                        )
                    )
                    self.state.gap.cancelled = True
                    self._call_logger.info(
                        "SKIPPED_BETWEEN_RUN_AND_TTS | stage=%s | reason=user_spoke_during_run",
                        self.state.current_stage,
                    )
                    log_flow_event(
                        self._call_logger,
                        "interrupt_ignored",
                        reason="user_spoke_during_agent_run",
                        user_seq=user_seq,
                        stage=self.state.current_stage,
                        skipped=True,
                    )
                    if self._timeline is not None:
                        self._timeline.emit(
                            "pipeline_to_tts_skipped",
                            stage=self.state.current_stage,
                            reason="user_spoke_during_run",
                        )
                    self.reset_gap_state()
                    self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
                    return

                if not await self._wait_for_stable_user_silence_before_terminal_reply(
                    reply,
                    user_seq=user_seq,
                    source="run_pipeline",
                ):
                    self.reset_gap_state()
                    self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
                    return

                if terminal_reply:
                    # Um STT que chega na janela entre backend e TTS marca o gap
                    # como cancelado. Para respostas terminais esse marcador serve
                    # apenas para esperar o usuario terminar; nao pode suprimir a
                    # confirmacao final.
                    self.state.gap.cancelled = False
                    self.state.gap.guard_seq = self.state.user_final_seq

                try:
                    await self._speak_backend_reply(
                        reply,
                        source="run_pipeline",
                        add_to_chat_ctx=True,
                        initial_turn=user_seq == 0,
                    )
                except Exception:
                    self._logger.exception("[tts] failed speaking backend reply")
                    if self._timeline is not None:
                        self._timeline.emit(
                            "tts_stage_failed",
                            stage=self.state.current_stage,
                            source="run_pipeline",
                        )
                    self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
                    await self.terminate_with_resource_stop(
                        resource="tts",
                        source="run_pipeline_tts",
                    )
                    return
                self.reset_gap_state()
                self._release_deferred_cycle(is_deferred_replay=is_deferred_replay)
            self.arm_agent_wait_timeout_from_reply(
                reply,
                user_seq=user_seq,
                source="run_pipeline",
            )

            terminal_stop = self._terminal_stop_from_reply(reply)
            if terminal_stop is not None:
                await self._finish_with_agent_final_stop(
                    terminal_stop,
                    source="run_pipeline",
                )
                return

            if reply.done or self._finalization_policy.is_done_stage(reply.stage):
                self._create_task_logged(
                    self.notify_bridge_done("stage_done"),
                    name="bridge_done",
                )
                self._create_task_logged(
                    self.finalize("stage_done"),
                    name="finalize",
                )
            elif user_seq == self.state.user_final_seq:
                self._schedule_backend_push_listener(
                    user_seq=user_seq,
                    reason="run_pipeline_completed",
                )

    async def stash_pending_user_turn(
        self,
        *,
        seq: int,
        transcription: str,
        text: str,
        message_id: str = "",
    ) -> None:
        async with self.pending_lock:
            self.state.pending_user_final = PendingUserTurn(
                seq=seq,
                transcription=transcription,
                text=text,
                message_id=message_id,
            )

    def _consume_deferred_empty_final_latch(self) -> bool:
        latched_at = self._deferred_empty_final_latched_at
        if not latched_at:
            return False
        self._deferred_empty_final_latched_at = 0.0
        # A marca vale so para o evento imediatamente seguinte: se o LiveKit
        # filtrou a copia, ela nao pode sobrar e engolir o final vazio de outra
        # fala mais adiante.
        return (time.monotonic() - latched_at) <= EMPTY_STT_FINAL_DEDUP_WINDOW_S

    def _handle_deferred_stt_final(self, *, transcription: str, text: str) -> bool:
        if not self.state.deferred_interruption.backend_in_flight:
            return False
        utterance = self._consume_vad_for_stt_final()
        if utterance is None:
            return False

        message_id = self._consume_turn_message_id(transcription, text) if text else ""
        if not message_id:
            self._clear_started_turn_message_id()

        if not utterance.is_long:
            log_flow_event(
                self._call_logger,
                "interrupt_ignored",
                reason="deferred_interruption_too_short",
                message_id=message_id,
                stage=self.state.current_stage,
                audio_ms=utterance.duration_ms,
                minimum_audio_ms=self._config.deferred_interruption_min_audio_ms,
                text=text,
            )
            return True

        if not text:
            log_flow_event(
                self._call_logger,
                "deferred_interruption_stt_empty",
                stage=self.state.current_stage,
                audio_ms=utterance.duration_ms,
            )
            return True

        self.cancel_idle_timer("deferred_interruption")
        self.cancel_idle_close("deferred_interruption")
        self.cancel_agent_wait_timeout("deferred_interruption")
        self._cancel_backend_push_listener(reason="deferred_interruption")
        self.state.user_final_seq += 1
        event_seq = self.state.user_final_seq
        self.state.last_user_final_at = time.monotonic()
        deferred = self.state.deferred_interruption
        deferred.long_turns.append(
            PendingUserTurn(
                seq=event_seq,
                transcription=transcription,
                text=text,
                message_id=message_id,
            )
        )
        log_flow_event(
            self._call_logger,
            "deferred_interruption_accumulated",
            message_id=message_id,
            user_seq=event_seq,
            stage=self.state.current_stage,
            audio_ms=utterance.duration_ms,
            accumulated_turns=len(deferred.long_turns),
            text=text,
        )
        return True

    def on_user_input_transcribed(self, ev: UserInputTranscribedEvent) -> None:
        transcription = ev.transcript or ""
        text = (self._extract_text_from_transcript(transcription) or "").strip()

        if not ev.is_final:
            return

        deferred = self.state.deferred_interruption
        if deferred.replay_in_flight:
            self._consume_vad_for_stt_final()
            dropped_message_id = self._consume_turn_message_id(transcription, text) if text else ""
            if not dropped_message_id:
                self._clear_started_turn_message_id()
            log_flow_event(
                self._call_logger,
                "deferred_replay_stt_discarded",
                message_id=dropped_message_id,
                text=text,
            )
            self._clear_structured_interruption()
            return

        if deferred.backend_in_flight and not self._config.deferred_interruption_enabled:
            dropped_message_id = (
                self._consume_turn_message_id(transcription, text) if text else ""
            )
            if not dropped_message_id:
                self._clear_started_turn_message_id()
            self._log_user_input_dropped(
                context={
                    "reason": "backend_processing_feedback_compat",
                    "stage": self.state.current_stage,
                    "agent_message_type": "feedback",
                    "agent_result_type": "feedback",
                },
                transcription=transcription,
                text=text,
            )
            self._clear_structured_interruption()
            return

        # Finais normais consomem o VAD pré-R1 correspondente. Se R1 já
        # estiver em voo, _handle_deferred_stt_final faz esse consumo e
        # acumula o turno longo para o replay R2.
        if not deferred.backend_in_flight:
            self._consume_vad_for_stt_final()

        if not text and self._consume_deferred_empty_final_latch():
            # O provider ja contabilizou este final vazio; este evento e a copia
            # que o LiveKit repassa.
            self._clear_structured_interruption()
            return

        self._restore_agent_wait_timeout_retry_vad_threshold(reason="user_final")
        drop_context = self._drop_context_for_user_final()
        if drop_context is not None:
            # A fala protegida tem prioridade sobre o ciclo diferido, mas a fala
            # VAD correspondente precisa sair da fila junto: senao o proximo
            # final consumiria a fala errada.
            if deferred.backend_in_flight:
                self._consume_vad_for_stt_final()
            dropped_message_id = (
                self._consume_turn_message_id(transcription, text)
                if text
                else ""
            )
            if not dropped_message_id:
                self._clear_started_turn_message_id()
            self._log_user_input_dropped(
                context=drop_context,
                transcription=transcription,
                text=text,
            )
            self._clear_structured_interruption()
            return

        if self._handle_deferred_stt_final(transcription=transcription, text=text):
            self._clear_structured_interruption()
            return

        if not text:
            self._clear_started_turn_message_id()
            self.handle_empty_stt_final(
                transcript_len=len(transcription),
                source="user_input_transcribed",
            )
            self._clear_structured_interruption()
            return

        if deferred.backend_in_flight:
            # Dentro da janela do ciclo todo final precisa de uma fala VAD
            # correspondente; sem ela a associacao por ordem ja esta
            # dessincronizada e reenviar este texto misturaria os turnos. Fora da
            # janela o final segue o fluxo normal de interrupcao.
            message_id = self._consume_turn_message_id(transcription, text)
            log_flow_event(
                self._call_logger,
                "interrupt_ignored",
                reason="deferred_interruption_missing_vad",
                message_id=message_id,
                stage=self.state.current_stage,
                text=text,
            )
            self._clear_structured_interruption()
            return

        self.cancel_idle_timer("user_final")
        message_id = self._consume_turn_message_id(transcription, text)
        self.cancel_idle_close("user_final")
        self.cancel_agent_wait_timeout("user_final")
        self._clear_agent_wait_timeout_memory()
        self._cancel_backend_push_listener(reason="user_final")
        self.state.idle_nudge_count = 0

        self.state.user_final_seq += 1
        event_seq = self.state.user_final_seq
        self.state.last_user_final_at = time.monotonic()
        was_speaking = self.speaking.is_set()
        keep_pre_backend_notice = (
            was_speaking and self._pre_backend_wait_notice_is_current_speech()
        )
        interruptible = (
            self._current_speech_allows_interruption()
            if was_speaking
            else self._interrupt_policy.allow_stage(self.state.current_stage)
        )
        if keep_pre_backend_notice:
            interruptible = False
        log_flow_event(
            self._call_logger,
            "stt_final",
            message_id=message_id,
            user_seq=event_seq,
            stage=self.state.current_stage,
            speaking=was_speaking,
            interruptible=interruptible,
            text=text,
        )
        if self._timeline is not None:
            self._timeline.emit(
                "user_transcript_final",
                user_seq=event_seq,
                message_id=message_id,
                text=text,
                speaking=self.speaking.is_set(),
                current_stage=self.state.current_stage,
            )
        stt_duration_ms = (
            round((time.monotonic() - self._last_vad_speech_end_at) * 1000)
            if self._last_vad_speech_end_at > 0
            else None
        )
        self._emit_debug_event(
            "stt.completed",
            user_seq=event_seq,
            message_id=message_id,
            duration_ms=stt_duration_ms,
            audio_duration_ms=self._last_vad_speech_duration_ms,
            text=text,
        )

        interrupt_tasks: list[asyncio.Task[Any]] = []
        if keep_pre_backend_notice:
            log_flow_event(
                self._call_logger,
                "interrupt_ignored",
                reason="pre_backend_wait_notice_same_turn",
                action="keep_speaking",
                message_id=message_id,
                user_seq=event_seq,
                stage=self.state.current_speech.stage,
                text=text,
            )

        if self.state.gap.active and not self.state.gap.cancelled:
            self.state.gap.cancelled = True
            interrupt_tasks.append(self._create_task_logged(
                self.execute(
                    SetPendingInterrupt(
                        listened_text="",
                        skipped=True,
                        speech_id=self.state.gap.speech_id,
                    )
                ),
                name="pending_interrupt_gap",
            ))
            self._call_logger.info(
                "GAP_INTERRUPTION | stage=%s | skipped=True | speech_id=%s | reason=user_spoke_before_tts",
                self.state.gap.stage or self.state.current_stage,
                self.state.gap.speech_id or "-",
            )
            log_flow_event(
                self._call_logger,
                "interrupt_marked",
                reason="user_spoke_before_tts",
                message_id=message_id,
                user_seq=event_seq,
                stage=self.state.gap.stage or self.state.current_stage,
                skipped=True,
                text=text,
            )

        if was_speaking and interruptible:
            log_flow_event(
                self._call_logger,
                "interrupt_detected",
                reason="user_final_while_speaking",
                message_id=message_id,
                user_seq=event_seq,
                stage=self.state.current_stage,
                text=text,
            )
            stop_task = self.request_current_speech_interruption(reason="user_final_while_speaking")
            if stop_task is not None:
                interrupt_tasks.append(stop_task)
            interrupt_tasks.append(
                self.mark_interrupt_from_handle_if_any(reason="user_final_while_speaking")
            )
        elif self._last_interrupt_task is not None and not self._last_interrupt_task.done():
            if not keep_pre_backend_notice:
                interrupt_tasks.append(self._last_interrupt_task)

        if was_speaking and not interruptible and not keep_pre_backend_notice:
            log_flow_event(
                self._call_logger,
                "interrupt_ignored",
                reason="stage_not_interruptible",
                action="stash_pending",
                message_id=message_id,
                user_seq=event_seq,
                stage=self.state.current_stage,
                text=text,
            )
            self._create_task_logged(
                self.stash_pending_user_turn(
                    seq=event_seq,
                    transcription=transcription,
                    text=text,
                    message_id=message_id,
                ),
                name=f"stash_pending_final_{event_seq}",
            )
            self._clear_structured_interruption()
            return

        async def _run_after_interrupt_mark() -> None:
            if interrupt_tasks:
                try:
                    seen: set[asyncio.Task[Any]] = set()
                    unique_tasks: list[asyncio.Task[Any]] = []
                    for task in interrupt_tasks:
                        if task not in seen:
                            seen.add(task)
                            unique_tasks.append(task)
                    await asyncio.gather(*unique_tasks)
                except Exception:
                    self._logger.exception("PIPELINE_INTERRUPT | erro marcando interrupcao")
            await self.run_pipeline(transcription, text, user_seq=event_seq, message_id=message_id)

        self._create_task_logged(
            _run_after_interrupt_mark(),
            name=f"run_pipeline_{event_seq}",
        )
        self._clear_structured_interruption()

    async def on_shutdown(self) -> None:
        await self.finalize("shutdown_callback")

    def register_callbacks(self) -> None:
        @self._session.on("user_input_transcribed")
        def _on_user_input_transcribed(ev: UserInputTranscribedEvent) -> None:
            self.on_user_input_transcribed(ev)

        @self._session.on("error")
        def _on_session_error(ev: Any) -> None:
            error = getattr(ev, "error", None)
            self._record_retryable_tts_session_error(error)
            if self._timeline is not None:
                self._timeline.emit(
                    "session_error",
                    error_type=str(getattr(error, "type", "") or ""),
                    recoverable=bool(getattr(error, "recoverable", False)),
                )

        @self._session.on("close")
        def _on_session_close(ev: Any) -> None:
            error = getattr(ev, "error", None)
            raw_reason = getattr(ev, "reason", "")
            reason = str(getattr(raw_reason, "value", raw_reason) or "")
            if self._timeline is not None:
                self._timeline.emit(
                    "session_closed",
                    reason=reason,
                    error_type=str(getattr(error, "type", "") or ""),
                )
            if reason != "error" or error is None:
                return

            resource = self._resource_from_session_error(error)
            self._create_task_logged(
                self.terminate_with_resource_stop(
                    resource=resource,
                    source="session_close",
                ),
                name=f"terminal_stop_session_close_{resource}",
            )

        @self._session.on("metrics_collected")
        def _on_metrics_collected(ev: Any) -> None:
            self._record_tts_metric(ev)

        @self._session.on("agent_state_changed")
        def _on_agent_state_changed(ev: Any) -> None:
            new_state = str(getattr(ev, "new_state", "") or "").lower()
            state_name = new_state.rsplit(".", 1)[-1]
            self._agent_state_name = state_name or new_state
            if state_name == "speaking":
                self._agent_speaking_seq += 1
                self._agent_speaking_at = time.monotonic()
                self._agent_speaking_event.set()
            if self._timeline is not None:
                self._timeline.emit(
                    "agent_state_changed",
                    old_state=str(getattr(ev, "old_state", "") or ""),
                    new_state=str(getattr(ev, "new_state", "") or ""),
                )

        @self._session.on("user_state_changed")
        def _on_user_state_changed(ev) -> None:
            new_state = str(getattr(ev, "new_state", "") or "").lower()
            state_name = new_state.rsplit(".", 1)[-1]
            self._user_state_name = state_name or new_state
            if state_name == "speaking":
                self._user_not_speaking.clear()
            else:
                self._user_not_speaking.set()
            self._call_logger.info(
                "USER_STATE | old=%s | new=%s",
                getattr(ev, "old_state", ""),
                getattr(ev, "new_state", ""),
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "user_state_changed",
                    old_state=str(getattr(ev, "old_state", "") or ""),
                    new_state=str(getattr(ev, "new_state", "") or ""),
                )
            if state_name == "speaking":
                if self.state.deferred_interruption.replay_in_flight:
                    log_flow_event(
                        self._call_logger,
                        "deferred_replay_user_speech_discarded",
                        source="user_state_changed",
                    )
                    return
                if (
                    self.state.deferred_interruption.backend_in_flight
                    and not self._config.deferred_interruption_enabled
                ):
                    log_flow_event(
                        self._call_logger,
                        "backend_processing_user_speech_discarded",
                        source="user_state_changed",
                        mode="feedback_compat",
                    )
                    return
                # O STT final pode chegar enquanto o conforto iniciado pelo
                # mesmo turno ainda toca. So uma nova transicao real do VAD
                # para speaking pode retirar essa protecao.
                self._rearm_current_speech_interruptions_on_new_user_speech()
                if self._mark_drop_next_user_final_from_current_speech(
                    reason="user_state_speaking_protected_speech"
                ):
                    return
                if self._mark_drop_next_user_final_from_post_playout_window(
                    reason="user_state_speaking_protected_post_playout_grace"
                ):
                    return
                self._clear_pending_user_input_drop()
                if self._inflight_backend_wait_notice_is_speaking():
                    log_flow_event(
                        self._call_logger,
                        "interrupt_detected",
                        reason="user_state_speaking",
                        action="interrupt_inflight_backend_wait_notice",
                        stage=self.state.current_speech.stage,
                    )
                    self.note_user_speech_activity(reason="user_state_speaking")
                    self.request_current_speech_interruption(
                        reason="user_state_speaking_inflight_backend_wait",
                        force=True,
                    )
                    return
                self._cancel_pre_backend_wait_notice(reason="user_state_speaking")
                self.note_user_speech_activity(reason="user_state_speaking")
                self.request_current_speech_interruption(reason="user_state_speaking")

        @self._session.on("overlapping_speech")
        def _on_overlapping_speech(ev) -> None:
            is_interruption = bool(getattr(ev, "is_interruption", False))
            self._call_logger.info(
                "OVERLAPPING_SPEECH | interruption=%s | delay_s=%.3f",
                is_interruption,
                float(getattr(ev, "detection_delay", 0.0) or 0.0),
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "overlapping_speech",
                    is_interruption=is_interruption,
                    detection_delay_ms=round(float(getattr(ev, "detection_delay", 0.0) or 0.0) * 1000),
                )
            if self.state.deferred_interruption.replay_in_flight:
                log_flow_event(
                    self._call_logger,
                    "deferred_replay_user_speech_discarded",
                    source="overlapping_speech",
                )
                return
            if (
                self.state.deferred_interruption.backend_in_flight
                and not self._config.deferred_interruption_enabled
            ):
                log_flow_event(
                    self._call_logger,
                    "backend_processing_user_speech_discarded",
                    source="overlapping_speech",
                    mode="feedback_compat",
                )
                return
            if is_interruption:
                if self._mark_drop_next_user_final_from_current_speech(
                    reason="overlapping_speech_protected_speech"
                ):
                    return
                self.request_current_speech_interruption(reason="overlapping_speech")

        room_on = getattr(self._ctx.room, "on", None)
        if callable(room_on):
            @room_on("participant_connected")
            def _on_room_participant_connected(participant: Any) -> None:
                log_flow_event(
                    self._call_logger,
                    "agent_room_participant_connected",
                    participant=str(getattr(participant, "identity", "") or ""),
                    kind=str(getattr(participant, "kind", "") or ""),
                    bridge_identity=self._bridge_identity or "",
                )
                self.log_room_input_snapshot(reason="participant_connected")

            @room_on("participant_disconnected")
            def _on_room_participant_disconnected(participant: Any) -> None:
                log_flow_event(
                    self._call_logger,
                    "agent_room_participant_disconnected",
                    participant=str(getattr(participant, "identity", "") or ""),
                    reason=str(getattr(participant, "disconnect_reason", "") or ""),
                    bridge_identity=self._bridge_identity or "",
                )
                self.log_room_input_snapshot(reason="participant_disconnected")

            @room_on("track_published")
            def _on_room_track_published(publication: Any, participant: Any) -> None:
                log_flow_event(
                    self._call_logger,
                    "agent_room_track_published",
                    participant=str(getattr(participant, "identity", "") or ""),
                    bridge_identity=self._bridge_identity or "",
                    **self._track_publication_fields(publication),
                )
                self.log_room_input_snapshot(reason="track_published")

            @room_on("track_subscribed")
            def _on_room_track_subscribed(track: Any, publication: Any, participant: Any) -> None:
                fields = self._track_publication_fields(publication)
                fields["track_sid"] = fields.get("track_sid") or str(getattr(track, "sid", "") or "")
                log_flow_event(
                    self._call_logger,
                    "agent_room_track_subscribed",
                    participant=str(getattr(participant, "identity", "") or ""),
                    bridge_identity=self._bridge_identity or "",
                    **fields,
                )
                self.log_room_input_snapshot(reason="track_subscribed")

            @room_on("track_unsubscribed")
            def _on_room_track_unsubscribed(track: Any, publication: Any, participant: Any) -> None:
                fields = self._track_publication_fields(publication)
                fields["track_sid"] = fields.get("track_sid") or str(getattr(track, "sid", "") or "")
                log_flow_event(
                    self._call_logger,
                    "agent_room_track_unsubscribed",
                    participant=str(getattr(participant, "identity", "") or ""),
                    bridge_identity=self._bridge_identity or "",
                    **fields,
                )
                self.log_room_input_snapshot(reason="track_unsubscribed")

            @room_on("track_muted")
            def _on_room_track_muted(participant: Any, publication: Any) -> None:
                log_flow_event(
                    self._call_logger,
                    "agent_room_track_muted",
                    participant=str(getattr(participant, "identity", "") or ""),
                    bridge_identity=self._bridge_identity or "",
                    **self._track_publication_fields(publication),
                )
                self.log_room_input_snapshot(reason="track_muted")

            @room_on("track_unmuted")
            def _on_room_track_unmuted(participant: Any, publication: Any) -> None:
                log_flow_event(
                    self._call_logger,
                    "agent_room_track_unmuted",
                    participant=str(getattr(participant, "identity", "") or ""),
                    bridge_identity=self._bridge_identity or "",
                    **self._track_publication_fields(publication),
                )
                self.log_room_input_snapshot(reason="track_unmuted")

            @room_on("data_received")
            def _on_room_data_received(data_packet: Any) -> None:
                if getattr(data_packet, "topic", "") != "bridge.control":
                    return

                message = self._decode_data_packet(data_packet)
                if message is None:
                    return

                control_type = str(message.get("type") or "").strip()
                if self._timeline is not None:
                    self._timeline.emit("bridge_control_received", control_type=control_type)

                if control_type == "client_audio_enabled":
                    self.request_initial_agent_turn(reason="client_audio_enabled")

        self._ctx.add_shutdown_callback(self.on_shutdown)

    async def run(self) -> None:
        self.register_callbacks()
        self._create_task_logged(
            self.finalize_when_room_empty(),
            name="finalize_when_room_empty",
        )

        if self._agent_starts_conversation:
            # Precisa vir antes do StartSession: o RoomIO respeita o estado do
            # input ao anexar a track, entao o audio ja nasce descartado.
            self.hold_user_audio_input(reason="room_setup")
            self._arm_user_audio_input_gate_timeout()

        room_opts_kwargs = dict(
            audio_input=room_io.AudioInputOptions(
                sample_rate=16000,
                num_channels=1,
                frame_size_ms=20,
                pre_connect_audio=False,
            ),
            audio_output=room_io.AudioOutputOptions(
                sample_rate=16000,
                num_channels=1,
            ),
        )

        if self._bridge_identity:
            room_opts_kwargs["participant_identity"] = self._bridge_identity

        self._call_logger.debug(
            "ROOM_ENTER | room=%s | participant_identity=%s",
            self._ctx.room.name,
            self._bridge_identity or "-",
        )
        if self._timeline is not None:
            self._timeline.emit(
                "room_enter",
                participant_identity=self._bridge_identity or "",
            )

        async def _start_session() -> None:
            await self.execute(
                StartSession(
                    room=self._ctx.room,
                    room_options=room_io.RoomOptions(**room_opts_kwargs),
                )
            )

        t_session = self._create_task_logged(_start_session(), name="session_start")
        if self._timeline is not None:
            self._timeline.emit("session_start_requested")
        self.arm_idle_timer(reason="client_join", delay_s=self._idle_policy.join_delay_s)

        try:
            await t_session
            if self._timeline is not None:
                self._timeline.emit("session_started")
        except Exception as exc:
            self._log_resource_error_events(
                status="stop_agent_runtime_unavailable",
                reason="livekit_session_start_failed",
                resource="agent_runtime",
                source="session_start",
                exc=exc,
            )
            raise
        finally:
            self._call_logger.debug("ROOM_EXIT | room=%s", self._ctx.room.name)
            if self._timeline is not None:
                self._timeline.emit("room_exit")
