from __future__ import annotations

import asyncio
import json
import os
import threading
import time
from pathlib import Path
from typing import Any, Callable, Dict, Optional, Tuple

import httpx
from dotenv import load_dotenv

from livekit.agents import (
    Agent,
    AgentServer,
    AgentSession,
    JobContext,
    JobProcess,
    cli,
    tts as livekit_tts,
    vad as agents_vad,
)
from livekit.plugins import silero, elevenlabs
from livekit.plugins.turn_detector.multilingual import MultilingualModel

from app.livekit.adapters.agent_backend import AgentBackend, BackendReply
from app.livekit.adapters.azure_rest_tts import AzureRESTTTS
from app.livekit.adapters.backend_factory import build_agent_backend
from app.livekit.runtime.initial_greeting_audio_cache import InitialGreetingAudioCache
from app.livekit.adapters.fake_tts import FakeTTS as FakeSessionTTS
from app.livekit.adapters.xai_tts import OraclexAITTS
from app.livekit.compat import patch_inference_executor_is_alive
from app.livekit.call_config import (
    resolve_agent_backend_name,
    resolve_fake_agent_overrides,
    resolve_stt_overrides,
    resolve_tts_overrides,
    resolve_vad_logging_overrides,
    resolve_vad_overrides,
)
from app.livekit.azure_speech import (
    AZURE_SPEECH_TTS_SAMPLE_RATE,
    resolve_azure_speech_tts_config,
)
from app.livekit.adapters.bridge_gateway import BridgeGateway
from app.livekit.adapters.export_service import ExportService
from app.livekit.adapters.speech_service import (
    SpeechService,
    normalize_text,
)
from app.livekit.policies.finalization_policy import FinalizationPolicy
from app.livekit.policies.idle_policy import IdlePolicy, IdlePolicyConfig
from app.livekit.policies.interrupt_policy import InterruptPolicy
from app.livekit.runtime.command_executor import RuntimeCommandExecutor
from app.livekit.runtime.call_runtime import CallRuntime
from app.livekit.runtime.state import RuntimeConfig
from app.livekit.vad_dynamic_threshold import (
    DynamicVADThresholdConfig,
    DynamicVADThresholdController,
)
from app.providers.stt_internal_livekit import (
    InternalHTTPSTT,
    InternalSTTConfig,
    extract_text_from_transcript,
)
from app.providers.stt_fake import FakeSTT
from app.providers.stt_vosk import VoskSTT, DEFAULT_GRAMMAR
from app.utils.call_timeline import CallTimeline
from app.utils.logging import (
    get_call_logger,
    log_flow_event,
    setup_minimal_logging,
    set_log_session_id,
    structured_context_from_metadata,
)

logger = setup_minimal_logging()
load_dotenv()
patch_inference_executor_is_alive()

# --- ENV / AMBIENTE ---
APP_ENV = (os.getenv("APP_ENV") or os.getenv("ENV") or "dev").strip().lower()
AGENT_BASE_NAME = os.getenv("AGENT_BASE_NAME", "ws-voice-agent").strip()
AGENT_NAME = os.getenv("AGENT_NAME", f"{AGENT_BASE_NAME}-{APP_ENV}").strip()
AGENT_SERVER_PORT = int(os.getenv("AGENT_SERVER_PORT", "18081"))

server = AgentServer(
    num_idle_processes=int(os.getenv("NUM_IDLE_PROCESSES", "5")),
    port=AGENT_SERVER_PORT,
)

CALL_END_GRACE_S = float(os.getenv("CALL_END_GRACE_S", "2"))
EXPORT_SESSION_ID_ENV = os.getenv("EXPORT_SESSION_ID", "").strip()

# --- TUNING (interrupção nativa) ---
MIN_INTERRUPT_S = float(os.getenv("MIN_INTERRUPT_S", "1.6"))
MIN_ENDPOINTING_DELAY = 1.0
MAX_ENDPOINTING_DELAY = 1.2
FINAL_GRACE_S = float(os.getenv("INTERRUPT_FINAL_GRACE_S", "0.12"))
DISCARD_AUDIO_IF_UNINTERRUPTIBLE = (
    os.getenv("DISCARD_AUDIO_IF_UNINTERRUPTIBLE", "1").strip().lower()
    in {"1", "true", "yes", "on"}
)
FLOW_LOG_VAD_DECISIONS = (
    os.getenv("FLOW_LOG_VAD_DECISIONS", "0").strip().lower()
    in {"1", "true", "yes", "on"}
)
FLOW_LOG_VAD_ACTIVITY = (
    os.getenv("FLOW_LOG_VAD_ACTIVITY", "0").strip().lower()
    in {"1", "true", "yes", "on"}
)
FLOW_LOG_VAD_ACTIVITY_MIN_PROB = float(os.getenv("FLOW_LOG_VAD_ACTIVITY_MIN_PROB", "0.03"))
VAD_MIN_SPEECH_DURATION = float(os.getenv("VAD_MIN_SPEECH_DURATION", "0.15"))
VAD_MIN_SILENCE_DURATION = float(
    os.getenv("VAD_MIN_SILENCE_DURATION")
    or os.getenv("LIVEKIT_VAD_MIN_SILENCE_DURATION_S")
    or "0.35"
)
VAD_ACTIVATION_THRESHOLD = float(os.getenv("VAD_ACTIVATION_THRESHOLD", "0.35"))
VAD_DEACTIVATION_THRESHOLD = float(
    os.getenv("VAD_DEACTIVATION_THRESHOLD", str(max(VAD_ACTIVATION_THRESHOLD - 0.15, 0.01)))
)
VAD_PREFIX_PADDING_DURATION = float(os.getenv("VAD_PREFIX_PADDING_DURATION", "1.5"))
VAD_PREFIX_PADDING_MIN_DURATION = float(os.getenv("VAD_PREFIX_PADDING_MIN_DURATION", "1.5"))

DEFAULT_VAD_CONFIG = {
    "min_speech_duration": VAD_MIN_SPEECH_DURATION,
    "min_silence_duration": VAD_MIN_SILENCE_DURATION,
    "activation_threshold": VAD_ACTIVATION_THRESHOLD,
    "deactivation_threshold": VAD_DEACTIVATION_THRESHOLD,
    "prefix_padding_duration": max(VAD_PREFIX_PADDING_DURATION, VAD_PREFIX_PADDING_MIN_DURATION),
}

DEFAULT_VAD_LOGGING_CONFIG = {
    "log_decisions": FLOW_LOG_VAD_DECISIONS,
    "log_activity": FLOW_LOG_VAD_ACTIVITY,
    "activity_min_probability": FLOW_LOG_VAD_ACTIVITY_MIN_PROB,
}

# --- IDLE NUDGE / FECHAMENTO ---
IDLE_NUDGE_ENABLED = (os.getenv("IDLE_NUDGE_ENABLED", "0").strip().lower() in {"1", "true", "yes", "on"})
IDLE_NUDGE_DELAY_S = float(os.getenv("IDLE_NUDGE_DELAY_S", "30"))
IDLE_NUDGE_JOIN_DELAY_S = float(os.getenv("IDLE_NUDGE_JOIN_DELAY_S", str(IDLE_NUDGE_DELAY_S)))
IDLE_NUDGE_CLOSE_DELAY_S = float(os.getenv("IDLE_NUDGE_CLOSE_DELAY_S", str(IDLE_NUDGE_DELAY_S)))
IDLE_NUDGE_TEXT_ENV = (os.getenv("IDLE_NUDGE_TEXT", "Alô, você está aí?") or "").strip()
IDLE_NUDGE_MAX_TRIES = int(os.getenv("IDLE_NUDGE_MAX_TRIES", "3"))
IDLE_NUDGE_END_REASON = (os.getenv("IDLE_NUDGE_END_REASON", "no_user_response") or "").strip()
AGENT_WAIT_TIMEOUT_RETRY_VAD_THRESHOLD_ENABLED = (
    os.getenv("AGENT_WAIT_TIMEOUT_RETRY_VAD_THRESHOLD_ENABLED", "0").strip().lower()
    in {"1", "true", "yes", "on"}
)
AGENT_WAIT_TIMEOUT_RETRY_VAD_ACTIVATION_THRESHOLD = float(
    os.getenv("AGENT_WAIT_TIMEOUT_RETRY_VAD_ACTIVATION_THRESHOLD", "0.20")
)
_AGENT_WAIT_TIMEOUT_RETRY_VAD_DEACTIVATION_THRESHOLD_RAW = (
    os.getenv("AGENT_WAIT_TIMEOUT_RETRY_VAD_DEACTIVATION_THRESHOLD", "") or ""
).strip()
AGENT_WAIT_TIMEOUT_RETRY_VAD_DEACTIVATION_THRESHOLD = (
    float(_AGENT_WAIT_TIMEOUT_RETRY_VAD_DEACTIVATION_THRESHOLD_RAW)
    if _AGENT_WAIT_TIMEOUT_RETRY_VAD_DEACTIVATION_THRESHOLD_RAW
    else None
)
REMOTE_AGENT_INFLIGHT_WAIT_INTERVAL_S = float(os.getenv("REMOTE_AGENT_INFLIGHT_WAIT_INTERVAL_S", "12"))
REMOTE_AGENT_INFLIGHT_WAIT_TIMEOUT_S = float(os.getenv("REMOTE_AGENT_INFLIGHT_WAIT_TIMEOUT_S", "180"))
# Zero means unlimited notices. The total wait is bounded exclusively by the
# backend timeout above, preventing a silent gap between a notice-count limit
# and termination.
REMOTE_AGENT_INFLIGHT_WAIT_MAX_NOTICES = int(os.getenv("REMOTE_AGENT_INFLIGHT_WAIT_MAX_NOTICES", "0"))
REMOTE_AGENT_INFLIGHT_WAIT_TEXT = (
    os.getenv("REMOTE_AGENT_INFLIGHT_WAIT_TEXT", "Um momento, ainda estou consultando para te ajudar.")
    or ""
).strip()
PRE_BACKEND_WAIT_NOTICE_FAST_ON_VAD_PAUSE = (
    os.getenv("PRE_BACKEND_WAIT_NOTICE_FAST_ON_VAD_PAUSE", "0").strip().lower()
    in {"1", "true", "yes", "on"}
)
LIVEKIT_APP_DIR = Path(__file__).resolve().parent
DEFERRED_INTERRUPTION_MIN_AUDIO_MS = max(
    0,
    int(os.getenv("DEFERRED_INTERRUPTION_MIN_AUDIO_MS", "1000")),
)
DEFERRED_INTERRUPTION_ENABLED = (
    os.getenv("DEFERRED_INTERRUPTION_ENABLED", "1").strip().lower()
    in {"1", "true", "yes", "on"}
)
DEFERRED_INTERRUPTION_STT_SETTLE_TIMEOUT_S = max(
    0.0,
    float(os.getenv("DEFERRED_INTERRUPTION_STT_SETTLE_TIMEOUT_S", "3.0")),
)
DEFERRED_INTERRUPTION_USER_TURN_TIMEOUT_S = max(
    0.0,
    float(os.getenv("DEFERRED_INTERRUPTION_USER_TURN_TIMEOUT_S", "10.0")),
)
REMOTE_AGENT_INFLIGHT_WAIT_AUDIO_BASE_DIR = LIVEKIT_APP_DIR / "assets" / "comfort"
REMOTE_AGENT_INFLIGHT_WAIT_SHORT_AUDIO_DIR = (
    os.getenv(
        "REMOTE_AGENT_INFLIGHT_WAIT_SHORT_AUDIO_DIR",
        str(REMOTE_AGENT_INFLIGHT_WAIT_AUDIO_BASE_DIR / "short"),
    )
    or ""
).strip()
REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_DIR = (
    os.getenv(
        "REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_DIR",
        str(REMOTE_AGENT_INFLIGHT_WAIT_AUDIO_BASE_DIR / "long"),
    )
    or ""
).strip()
REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_PATH = (
    os.getenv(
        "REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_PATH",
        "",
    )
    or ""
).strip()
REMOTE_AGENT_INFLIGHT_WAIT_AUDIO_ENABLED = bool(
    REMOTE_AGENT_INFLIGHT_WAIT_SHORT_AUDIO_DIR
    or REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_DIR
    or REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_PATH
)


def _float_override(
    value: Any,
    default: float,
    *,
    min_value: Optional[float] = None,
    max_value: Optional[float] = None,
) -> float:
    if value in (None, ""):
        return default
    try:
        resolved = float(value)
    except (TypeError, ValueError):
        return default
    if min_value is not None and resolved < min_value:
        return default
    if max_value is not None and resolved > max_value:
        return default
    return resolved


def _bool_override(value: Any, default: bool) -> bool:
    if value in (None, ""):
        return default
    raw = str(value).strip().lower()
    if raw in {"1", "true", "yes", "on"}:
        return True
    if raw in {"0", "false", "no", "off"}:
        return False
    return default


def _has_any_override(overrides: Dict[str, Any]) -> bool:
    return any(value not in (None, "") for value in overrides.values())


def resolve_vad_runtime_config(call_config: Optional[Dict[str, Any]]) -> Tuple[Dict[str, float], bool]:
    overrides = resolve_vad_overrides(call_config)
    vad_tuning_overrides = {
        key: overrides.get(key)
        for key in (
            "min_speech_duration",
            "min_silence_duration",
            "activation_threshold",
            "deactivation_threshold",
            "prefix_padding_duration",
        )
    }
    return (
        {
            "min_speech_duration": _float_override(
                overrides.get("min_speech_duration"),
                DEFAULT_VAD_CONFIG["min_speech_duration"],
                min_value=0.0,
            ),
            "min_silence_duration": _float_override(
                overrides.get("min_silence_duration"),
                DEFAULT_VAD_CONFIG["min_silence_duration"],
                min_value=0.0,
            ),
            "activation_threshold": _float_override(
                overrides.get("activation_threshold"),
                DEFAULT_VAD_CONFIG["activation_threshold"],
                min_value=0.0,
                max_value=1.0,
            ),
            "deactivation_threshold": _float_override(
                overrides.get("deactivation_threshold"),
                DEFAULT_VAD_CONFIG["deactivation_threshold"],
                min_value=0.0,
                max_value=1.0,
            ),
            "prefix_padding_duration": _float_override(
                overrides.get("prefix_padding_duration"),
                DEFAULT_VAD_CONFIG["prefix_padding_duration"],
                min_value=VAD_PREFIX_PADDING_MIN_DURATION,
            ),
        },
        _has_any_override(vad_tuning_overrides),
    )


def resolve_vad_logging_runtime_config(call_config: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    overrides = resolve_vad_logging_overrides(call_config)
    return {
        "log_decisions": _bool_override(
            overrides.get("log_decisions"),
            DEFAULT_VAD_LOGGING_CONFIG["log_decisions"],
        ),
        "log_activity": _bool_override(
            overrides.get("log_activity"),
            DEFAULT_VAD_LOGGING_CONFIG["log_activity"],
        ),
        "activity_min_probability": _float_override(
            overrides.get("activity_min_probability"),
            DEFAULT_VAD_LOGGING_CONFIG["activity_min_probability"],
            min_value=0.0,
            max_value=1.0,
        ),
    }


def resolve_agent_runtime_config(call_config: Optional[Dict[str, Any]]) -> Tuple[Dict[str, Any], bool]:
    overrides = resolve_vad_overrides(call_config)
    fast_on_vad_pause_override = overrides.get("pre_backend_wait_notice_fast_on_vad_pause")
    deferred_interruption_min_audio_ms_override = overrides.get(
        "deferred_interruption_min_audio_ms"
    )
    deferred_interruption_enabled_override = overrides.get(
        "deferred_interruption_enabled"
    )
    return (
        {
            "pre_backend_wait_notice_fast_on_vad_pause": _bool_override(
                fast_on_vad_pause_override,
                PRE_BACKEND_WAIT_NOTICE_FAST_ON_VAD_PAUSE,
            ),
            "deferred_interruption_min_audio_ms": int(
                _float_override(
                    deferred_interruption_min_audio_ms_override,
                    float(DEFERRED_INTERRUPTION_MIN_AUDIO_MS),
                    min_value=0.0,
                )
            ),
            "deferred_interruption_enabled": _bool_override(
                deferred_interruption_enabled_override,
                DEFERRED_INTERRUPTION_ENABLED,
            ),
        },
        any(
            value not in (None, "")
            for value in (
                fast_on_vad_pause_override,
                deferred_interruption_min_audio_ms_override,
                deferred_interruption_enabled_override,
            )
        ),
    )


def load_silero_vad(vad_config: Dict[str, float]) -> agents_vad.VAD:
    return silero.VAD.load(
        min_speech_duration=vad_config["min_speech_duration"],
        min_silence_duration=vad_config["min_silence_duration"],
        activation_threshold=vad_config["activation_threshold"],
        deactivation_threshold=vad_config["deactivation_threshold"],
        prefix_padding_duration=vad_config["prefix_padding_duration"],
    )


def vad_config_cache_key(vad_config: Dict[str, float]) -> Tuple[float, float, float, float, float]:
    return (
        vad_config["min_speech_duration"],
        vad_config["min_silence_duration"],
        vad_config["activation_threshold"],
        vad_config["deactivation_threshold"],
        vad_config["prefix_padding_duration"],
    )


def require_env(name: str, *, context: str, timeline: Optional[CallTimeline] = None) -> str:
    value = (os.getenv(name, "") or "").strip()
    if value:
        return value

    if timeline is not None:
        timeline.emit(
            "config_missing",
            key=name,
            context=context,
        )

    raise RuntimeError(
        f"Missing required env var {name} for {context}. "
        f"Add {name} to .env.dev before starting the agent."
    )


def create_task_logged(coro, *, name: str):
    task = asyncio.create_task(coro, name=name)

    def _done(t: asyncio.Task):
        try:
            exc = t.exception()
        except asyncio.CancelledError:
            return
        except Exception:
            logger.exception("[task:%s] erro lendo exception()", t.get_name())
            return
        if exc:
            logger.exception("[task:%s] exception: %r", t.get_name(), exc)

    task.add_done_callback(_done)
    return task


class FlowLoggingVAD(agents_vad.VAD):
    def __init__(
        self,
        inner: agents_vad.VAD,
        *,
        call_logger: Any,
        min_interrupt_s: float,
        vad_config: Dict[str, float],
        log_decisions: bool,
        log_activity: bool,
        activity_min_prob: float,
        threshold_provider: Optional[Callable[[], Dict[str, Any]]] = None,
        on_pre_backend_wait_pause: Optional[Callable[[str], None]] = None,
        on_speech_end: Optional[Callable[[int], None]] = None,
    ) -> None:
        super().__init__(capabilities=inner.capabilities)
        self._inner = inner
        self._call_logger = call_logger
        self._min_interrupt_s = min_interrupt_s
        self._vad_config = dict(vad_config)
        self._log_decisions = log_decisions
        self._log_activity = log_activity
        self._activity_min_prob = activity_min_prob
        self._threshold_provider = threshold_provider
        self._on_pre_backend_wait_pause = on_pre_backend_wait_pause
        self._on_speech_end = on_speech_end
        self._stream_lock = threading.Lock()
        self._next_stream_id = 0
        self._logging_stream_id: Optional[int] = None

    def _claim_logging_stream(self) -> Tuple[int, bool]:
        with self._stream_lock:
            stream_id = self._next_stream_id
            self._next_stream_id += 1
            if self._logging_stream_id is None:
                self._logging_stream_id = stream_id
                return stream_id, True
            return stream_id, False

    def _release_logging_stream(self, stream_id: int) -> None:
        with self._stream_lock:
            if self._logging_stream_id == stream_id:
                self._logging_stream_id = None

    @property
    def model(self) -> str:
        return self._inner.model

    @property
    def provider(self) -> str:
        return self._inner.provider

    def stream(self):
        stream_id, should_log = self._claim_logging_stream()
        return FlowLoggingVADStream(
            self._inner.stream(),
            stream_id=stream_id,
            should_log=should_log,
            release_logging_stream=self._release_logging_stream,
            call_logger=self._call_logger,
            min_interrupt_s=self._min_interrupt_s,
            vad_config=self._vad_config,
            log_decisions=self._log_decisions,
            log_activity=self._log_activity,
            activity_min_prob=self._activity_min_prob,
            threshold_provider=self._threshold_provider,
            on_pre_backend_wait_pause=self._on_pre_backend_wait_pause,
            on_speech_end=self._on_speech_end,
        )


class FlowLoggingVADStream:
    def __init__(
        self,
        inner: Any,
        *,
        stream_id: int,
        should_log: bool,
        release_logging_stream: Callable[[int], None],
        call_logger: Any,
        min_interrupt_s: float,
        vad_config: Dict[str, float],
        log_decisions: bool,
        log_activity: bool,
        activity_min_prob: float,
        threshold_provider: Optional[Callable[[], Dict[str, Any]]] = None,
        on_pre_backend_wait_pause: Optional[Callable[[str], None]] = None,
        on_speech_end: Optional[Callable[[int], None]] = None,
    ) -> None:
        self._inner = inner
        self._stream_id = stream_id
        self._should_log = should_log
        self._release_logging_stream = release_logging_stream
        self._call_logger = call_logger
        self._min_interrupt_s = min_interrupt_s
        self._vad_config = dict(vad_config)
        self._log_decisions = log_decisions
        self._log_activity = log_activity
        self._activity_min_prob = activity_min_prob
        self._threshold_provider = threshold_provider
        self._on_pre_backend_wait_pause = on_pre_backend_wait_pause
        self._on_speech_end = on_speech_end
        self._last_activity_log_at = 0.0
        self._reset_speech_log_state()

    def push_frame(self, frame):
        return self._inner.push_frame(frame)

    def flush(self) -> None:
        return self._inner.flush()

    def end_input(self) -> None:
        return self._inner.end_input()

    async def aclose(self) -> None:
        try:
            return await self._inner.aclose()
        finally:
            self._release_logging_stream(self._stream_id)

    def __aiter__(self):
        return self

    async def __anext__(self):
        ev = await self._inner.__anext__()
        self._log_vad_decision(ev)
        return ev

    def _reset_speech_log_state(self) -> None:
        self._threshold_logged = False
        self._activity_logged = False
        self._pause_logged = False
        self._pre_backend_wait_pause_notified = False
        self._speech_seen = False
        self._max_speech_ms = 0
        self._max_silence_ms = 0

    @staticmethod
    def _duration_ms(value: Any) -> int:
        return round(float(value or 0.0) * 1000)

    def _pre_backend_wait_min_speech_ms(self) -> int:
        return max(1, round(float(self._vad_config["min_speech_duration"]) * 1000))

    def _current_thresholds(self) -> Dict[str, Any]:
        provider = self._threshold_provider
        if provider is not None:
            try:
                thresholds = provider()
            except Exception:
                logger.exception("[vad] failed reading dynamic threshold state")
            else:
                if isinstance(thresholds, dict):
                    return {
                        "mode": thresholds.get("mode") or "dynamic",
                        "activation_threshold": float(
                            thresholds.get("activation_threshold", self._vad_config["activation_threshold"])
                        ),
                        "deactivation_threshold": float(
                            thresholds.get(
                                "deactivation_threshold",
                                self._vad_config["deactivation_threshold"],
                            )
                        ),
                    }
        return {
            "mode": "baseline",
            "activation_threshold": self._vad_config["activation_threshold"],
            "deactivation_threshold": self._vad_config["deactivation_threshold"],
        }

    def _observe_speech_ms(self, *, speech_ms: int, raw_speech_ms: int) -> int:
        self._max_speech_ms = max(self._max_speech_ms, speech_ms, raw_speech_ms)
        return self._max_speech_ms

    def _mark_speech_seen_if_meaningful(
        self,
        *,
        speaking: bool,
        speech_ms: int,
        raw_speech_ms: int,
    ) -> None:
        observed_speech_ms = self._observe_speech_ms(
            speech_ms=speech_ms,
            raw_speech_ms=raw_speech_ms,
        )
        if observed_speech_ms >= self._pre_backend_wait_min_speech_ms():
            self._speech_seen = True

    def _observed_silence_ms(self, *, raw_silence_ms: int, silence_ms: int) -> int:
        self._max_silence_ms = max(self._max_silence_ms, raw_silence_ms, silence_ms)
        return self._max_silence_ms

    def _log_user_pause(
        self,
        *,
        decision: str,
        silence_ms: int,
        pause_min_ms: int,
        speech_ms: int,
        min_ms: int,
        eligible: bool,
        probability: float,
        raw_speech_ms: int,
        raw_silence_ms: int,
    ) -> None:
        self._pause_logged = True
        log_flow_event(
            self._call_logger,
            "vad_user_pause",
            decision=decision,
            silence_ms=silence_ms,
            pause_min_ms=pause_min_ms,
            speech_duration_ms=speech_ms,
            min_interrupt_ms=min_ms,
            eligible=eligible,
            probability=f"{probability:.3f}",
            raw_speech_ms=raw_speech_ms,
            raw_silence_ms=raw_silence_ms,
        )

    def _maybe_notify_pre_backend_wait_end_of_speech(
        self,
        *,
        silence_ms: int,
        speech_ms: int,
        pause_min_ms: int,
    ) -> None:
        callback = self._on_pre_backend_wait_pause
        if (
            callback is None
            or self._pre_backend_wait_pause_notified
            or not self._speech_seen
            or self._max_speech_ms < self._pre_backend_wait_min_speech_ms()
        ):
            return

        self._pre_backend_wait_pause_notified = True
        log_flow_event(
            self._call_logger,
            "pre_backend_wait_end_of_speech",
            silence_ms=silence_ms,
            pause_min_ms=pause_min_ms,
            speech_duration_ms=speech_ms,
        )
        try:
            callback("vad_end_of_speech")
        except Exception:
            logger.exception("[vad] failed notifying pre backend wait pause")

    def _log_vad_decision(self, ev: agents_vad.VADEvent) -> None:
        if not self._should_log:
            return

        min_ms = round(self._min_interrupt_s * 1000)
        speech_ms = self._duration_ms(getattr(ev, "speech_duration", 0.0))
        eligible = speech_ms >= min_ms
        probability = float(getattr(ev, "probability", 0.0) or 0.0)
        raw_speech_ms = self._duration_ms(getattr(ev, "raw_accumulated_speech", 0.0))
        silence_ms = self._duration_ms(getattr(ev, "silence_duration", 0.0))
        raw_silence_ms = self._duration_ms(getattr(ev, "raw_accumulated_silence", 0.0))
        current_silence_ms = max(raw_silence_ms, silence_ms)
        pause_min_ms = round(float(self._vad_config["min_silence_duration"]) * 1000)

        if ev.type == agents_vad.VADEventType.START_OF_SPEECH:
            self._reset_speech_log_state()
            self._mark_speech_seen_if_meaningful(
                speaking=True,
                speech_ms=speech_ms,
                raw_speech_ms=raw_speech_ms,
            )
            if self._log_decisions:
                log_flow_event(
                    self._call_logger,
                    "vad_speech_start",
                    speech_duration_ms=speech_ms,
                    min_interrupt_ms=min_ms,
                )
            return

        if ev.type == agents_vad.VADEventType.INFERENCE_DONE:
            speaking = bool(getattr(ev, "speaking", False))
            self._mark_speech_seen_if_meaningful(
                speaking=speaking,
                speech_ms=speech_ms,
                raw_speech_ms=raw_speech_ms,
            )
            if current_silence_ms < 200:
                self._pause_logged = False
                self._max_silence_ms = 0
                if not self._speech_seen:
                    self._max_speech_ms = 0
            observed_silence_ms = self._observed_silence_ms(
                raw_silence_ms=raw_silence_ms,
                silence_ms=silence_ms,
            )
            if self._log_activity:
                activity_detected = speaking or raw_speech_ms > 0 or probability >= self._activity_min_prob
                should_log_activity = activity_detected and (
                    not self._activity_logged
                    or eligible
                    or (time.monotonic() - self._last_activity_log_at) >= 0.25
                )
                if should_log_activity:
                    thresholds = self._current_thresholds()
                    self._activity_logged = True
                    self._last_activity_log_at = time.monotonic()
                    log_flow_event(
                        self._call_logger,
                        "vad_activity",
                        speaking=speaking,
                        speech_duration_ms=speech_ms,
                        min_interrupt_ms=min_ms,
                        eligible=eligible,
                        probability=f"{probability:.3f}",
                        threshold_mode=thresholds["mode"],
                        activation_threshold=f"{thresholds['activation_threshold']:.3f}",
                        deactivation_threshold=f"{thresholds['deactivation_threshold']:.3f}",
                        activity_log_min_prob=f"{self._activity_min_prob:.3f}",
                        silence_ms=observed_silence_ms,
                        pause_min_ms=pause_min_ms,
                        silence_remaining_ms=max(0, pause_min_ms - observed_silence_ms),
                        raw_speech_ms=raw_speech_ms,
                        raw_silence_ms=raw_silence_ms,
                    )
                elif not activity_detected and raw_silence_ms >= 200:
                    self._activity_logged = False

            if (
                self._speech_seen
                and observed_silence_ms >= pause_min_ms
            ):
                if not self._pause_logged:
                    self._log_user_pause(
                        decision="pause_threshold_reached",
                        silence_ms=observed_silence_ms,
                        pause_min_ms=pause_min_ms,
                        speech_ms=speech_ms,
                        min_ms=min_ms,
                        eligible=eligible,
                        probability=probability,
                        raw_speech_ms=raw_speech_ms,
                        raw_silence_ms=raw_silence_ms,
                    )
                self._maybe_notify_pre_backend_wait_end_of_speech(
                    silence_ms=observed_silence_ms,
                    speech_ms=speech_ms,
                    pause_min_ms=pause_min_ms,
                )

            if self._log_decisions and speaking and eligible and not self._threshold_logged:
                self._threshold_logged = True
                log_flow_event(
                    self._call_logger,
                    "vad_interrupt_check",
                    decision="eligible_by_duration",
                    speech_duration_ms=speech_ms,
                    min_interrupt_ms=min_ms,
                    probability=f"{probability:.3f}",
                    raw_speech_ms=raw_speech_ms,
                    raw_silence_ms=raw_silence_ms,
                )
            return

        if ev.type == agents_vad.VADEventType.END_OF_SPEECH:
            self._observe_speech_ms(
                speech_ms=speech_ms,
                raw_speech_ms=raw_speech_ms,
            )
            if self._on_speech_end is not None:
                try:
                    # Silero's final speech_duration excludes the terminal
                    # silence used to detect the endpoint. Do not use the
                    # maximum INFERENCE_DONE duration (which still includes
                    # that silence) or ev.frames (which include VAD padding).
                    self._on_speech_end(speech_ms)
                except Exception:
                    logger.exception("[vad] failed notifying speech duration")
            observed_silence_ms = self._observed_silence_ms(
                raw_silence_ms=raw_silence_ms,
                silence_ms=silence_ms,
            )
            self._maybe_notify_pre_backend_wait_end_of_speech(
                silence_ms=observed_silence_ms,
                speech_ms=speech_ms,
                pause_min_ms=pause_min_ms,
            )
            if self._speech_seen and not self._pause_logged:
                self._log_user_pause(
                    decision="end_of_speech",
                    silence_ms=observed_silence_ms,
                    pause_min_ms=pause_min_ms,
                    speech_ms=speech_ms,
                    min_ms=min_ms,
                    eligible=eligible,
                    probability=probability,
                    raw_speech_ms=raw_speech_ms,
                    raw_silence_ms=raw_silence_ms,
                )
            if self._log_decisions:
                log_flow_event(
                    self._call_logger,
                    "vad_speech_end",
                    decision="eligible_by_duration" if eligible else "too_short",
                    speech_duration_ms=speech_ms,
                    min_interrupt_ms=min_ms,
                    eligible=eligible,
                    silence_ms=observed_silence_ms,
                    raw_silence_ms=raw_silence_ms,
                    pause_min_ms=pause_min_ms,
                )
            self._reset_speech_log_state()


class PreBackendWaitNoticeBridge:
    def __init__(self) -> None:
        self._runtime: Optional[CallRuntime] = None

    def bind(self, runtime: CallRuntime) -> None:
        self._runtime = runtime

    def notify(self, reason: str) -> None:
        runtime = self._runtime
        if runtime is None:
            return
        runtime.schedule_pre_backend_wait_notice(reason=reason)

    def notify_speech_end(self, speech_duration_ms: int) -> None:
        runtime = self._runtime
        if runtime is None:
            return
        runtime.note_vad_speech_end(speech_duration_ms)


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = load_silero_vad(DEFAULT_VAD_CONFIG)
    proc.userdata["vad_cache"] = {
        vad_config_cache_key(DEFAULT_VAD_CONFIG): proc.userdata["vad"],
    }
    proc.userdata["initial_greeting_audio_cache"] = InitialGreetingAudioCache()
    wait_audio_loaded, wait_audio_failed = CallRuntime.prewarm_wait_audio_cache(
        short_audio_dir=REMOTE_AGENT_INFLIGHT_WAIT_SHORT_AUDIO_DIR,
        long_audio_dir=REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_DIR,
        long_audio_path=REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_PATH,
        logger=logger,
    )
    if wait_audio_loaded or wait_audio_failed:
        logger.info(
            "WAIT_AUDIO_PREWARM | loaded=%s | failed=%s",
            wait_audio_loaded,
            wait_audio_failed,
        )

    proc.userdata["http"] = httpx.AsyncClient(
        limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
        timeout=httpx.Timeout(30.0),
    )

    vosk_model_path = os.getenv("VOSK_MODEL_PATH", "").strip()
    if vosk_model_path:
        use_grammar = os.getenv("VOSK_USE_GRAMMAR", "1") == "1"
        grammar = DEFAULT_GRAMMAR if use_grammar else None

        proc.userdata["vosk"] = VoskSTT(
            model_path=vosk_model_path,
            sample_rate=int(os.getenv("VOSK_SAMPLE_RATE", "16000")),
            grammar_json=grammar,
        )
        proc.userdata["vosk_lock"] = threading.Lock()
    else:
        proc.userdata["vosk"] = None
        proc.userdata["vosk_lock"] = None
        logger.debug("[prewarm] VOSK_MODEL_PATH vazio; vosk desativado")


server.setup_fnc = prewarm

class PipelineVoiceAgent(Agent):
    def __init__(
        self,
        *,
        elegibility: bool,
        protocol: str,
        intro: str,
        backend_name: str = "",
        remote_agent_context: Optional[Dict[str, Any]] = None,
        timeline: Optional[CallTimeline] = None,
    ) -> None:
        super().__init__(instructions=(
            "Você é um agente de voz em português (pt-BR). "
            "Responda curto, direto e educado. "
            "Não use emojis, markdown ou caracteres especiais."
        ))
        self._elegibility = elegibility
        self._protocol = protocol
        self._intro = intro
        self._backend_name = backend_name
        self._remote_agent_context = remote_agent_context or {}
        self._timeline = timeline

        self.pipeline: Optional[AgentBackend] = None
        self._ready = asyncio.Event()
        self._prepare_task: Optional[asyncio.Task] = None
        self._run_lock = asyncio.Lock()

        self._end_lock = asyncio.Lock()
        self._ended = False
        self._end_reply: Optional[BackendReply] = None

        self._pending_interrupt: Optional[Tuple[str, str, bool]] = None
        self._pending_lock = asyncio.Lock()

    async def set_pending_interrupt(
        self,
        *,
        listened_text: str = "",
        skipped: bool = False,
        speech_id: str = "",
    ) -> None:
        listened_text = normalize_text(listened_text or "")
        speech_id = str(speech_id or "").strip()
        async with self._pending_lock:
            self._pending_interrupt = (speech_id, listened_text, bool(skipped))

    async def consume_pending_interrupt(self) -> Tuple[Optional[str], str, bool]:
        async with self._pending_lock:
            v = self._pending_interrupt
            self._pending_interrupt = None
        if not v:
            return None, "", False
        speech_id, listened_text, skipped = v
        return (normalize_text(listened_text or ""), speech_id, bool(skipped))

    async def on_enter(self):
        if self._timeline is not None:
            self._timeline.emit(
                "agent_backend_build_started",
                backend=self._backend_name or "remote_ws",
            )
        self.pipeline = build_agent_backend(
            intro=self._intro,
            backend_name=self._backend_name,
            remote_agent_context=self._remote_agent_context,
            timeline=self._timeline,
            streaming=False,
        )

        self._prepare_task = asyncio.create_task(
            self.pipeline.prepare(self._elegibility, self._protocol)
        )

        prepare_ok = False
        try:
            await self._prepare_task
            prepare_ok = True
        except Exception:
            logger.exception("[pipeline] prepare() falhou")
            if self._timeline is not None:
                self._timeline.emit(
                    "agent_backend_prepare_failed",
                    backend=self._backend_name or "remote_ws",
                )
        finally:
            if self._timeline is not None and prepare_ok:
                self._timeline.emit(
                    "agent_backend_ready",
                    backend=self._backend_name or "remote_ws",
                )
            self._ready.set()

    async def end_service_once(self) -> BackendReply:
        async with self._end_lock:
            if self._ended:
                return self._end_reply or BackendReply(stage="DONE", done=True, export_payload=[])

            self._ended = True
            if self.pipeline is None:
                self._end_reply = BackendReply(stage="DONE", done=True, export_payload=[])
                return self._end_reply

            try:
                self._end_reply = await self.pipeline.end_service_once()
            except Exception:
                logger.exception("[pipeline] end_service() falhou")
                self._end_reply = BackendReply(stage="DONE", done=True, export_payload=[])

            return self._end_reply


@server.rtc_session(agent_name=AGENT_NAME)
async def entrypoint(ctx: JobContext):
    ctx.log_context_fields = {"room": ctx.room.name}

    http_client: httpx.AsyncClient = ctx.proc.userdata["http"]
    default_vad = ctx.proc.userdata["vad"]
    vad = default_vad
    vosk = ctx.proc.userdata.get("vosk")
    initial_greeting_audio_cache = ctx.proc.userdata["initial_greeting_audio_cache"]
    vosk_lock = ctx.proc.userdata.get("vosk_lock")

    try:
        md = json.loads(ctx.job.metadata or "{}")
    except Exception:
        md = {}

    session_data = md.get("session_data") or {}
    elegibility = bool(md.get("elegibility", True))
    protocol = str(md.get("protocol", os.getenv("DEFAULT_PROTOCOL", "PRT-2025-0000123456")))
    intro = md.get("intro", "")
    agent_starts_conversation = bool(md.get("agent_starts_conversation", False))
    bridge_identity = str(md.get("bridge_identity", "")).strip()
    call_config = md.get("call_config") or {}
    remote_agent_context = dict(md.get("remote_agent") or {})
    fake_agent_overrides = resolve_fake_agent_overrides(call_config)
    if fake_agent_overrides.get("delay_ms") is not None:
        remote_agent_context["_fake_agent_delay_ms"] = fake_agent_overrides["delay_ms"]
    if fake_agent_overrides.get("responses"):
        remote_agent_context["_fake_agent_responses"] = fake_agent_overrides["responses"]
    debug_events_enabled = bool(md.get("debug_events_enabled", False))
    timeline_origin_ms = int(md.get("timeline_origin_ms") or round(time.time() * 1000))
    structured_context = structured_context_from_metadata(md)
    session_id = str(
        md.get("session_id")
        or session_data.get("session_id")
        or session_data.get("sessionId")
        or remote_agent_context.get("session_id")
        or remote_agent_context.get("sessionId")
        or ""
    ).strip()
    set_log_session_id(session_id)
    ctx.log_context_fields = {"room": ctx.room.name, "session_id": session_id}

    phone_number = (
        session_data.get("gsm")
        or session_data.get("msisdn")
        or session_data.get("phone")
        or remote_agent_context.get("GSM")
        or ""
    )

    call_logger = get_call_logger(
        phone_number=phone_number,
        session_id=session_id,
    )
    timeline = CallTimeline(
        logger=call_logger,
        component="agent",
        timeline_id=str(md.get("timeline_id") or ctx.room.name),
        protocol=protocol,
        room=ctx.room.name,
        session_id=session_id,
        phone_number=phone_number,
        origin_unix_ms=timeline_origin_ms,
    )

    vad_config, has_vad_overrides = resolve_vad_runtime_config(call_config)
    vad_logging_config = resolve_vad_logging_runtime_config(call_config)
    agent_runtime_config, has_agent_runtime_overrides = resolve_agent_runtime_config(call_config)
    vad_config_source = "call_config" if has_vad_overrides else "env"
    pre_backend_wait_notice_fast_source = (
        "call_config" if has_agent_runtime_overrides else "env"
    )
    if has_vad_overrides:
        try:
            vad_cache = ctx.proc.userdata.setdefault("vad_cache", {})
            cache_key = vad_config_cache_key(vad_config)
            vad = vad_cache.get(cache_key)
            if vad is None:
                vad = load_silero_vad(vad_config)
                vad_cache[cache_key] = vad
            timeline.emit(
                "vad_config_override_applied",
                min_speech_duration=vad_config["min_speech_duration"],
                min_silence_duration=vad_config["min_silence_duration"],
                activation_threshold=vad_config["activation_threshold"],
                deactivation_threshold=vad_config["deactivation_threshold"],
                prefix_padding_duration=vad_config["prefix_padding_duration"],
            )
        except Exception:
            logger.exception("[vad] failed loading call_config override; falling back to env VAD")
            call_logger.exception("[vad] failed loading call_config override; falling back to env VAD")
            vad = default_vad
            vad_config = dict(DEFAULT_VAD_CONFIG)
            vad_config_source = "env_fallback"
            timeline.emit("vad_config_override_failed")

    agent_wait_timeout_retry_vad_threshold_enabled = (
        AGENT_WAIT_TIMEOUT_RETRY_VAD_THRESHOLD_ENABLED
        and AGENT_WAIT_TIMEOUT_RETRY_VAD_ACTIVATION_THRESHOLD
        < vad_config["activation_threshold"]
    )
    agent_wait_timeout_retry_vad_deactivation_threshold = min(
        AGENT_WAIT_TIMEOUT_RETRY_VAD_ACTIVATION_THRESHOLD,
        max(
            0.01,
            (
                AGENT_WAIT_TIMEOUT_RETRY_VAD_DEACTIVATION_THRESHOLD
                if AGENT_WAIT_TIMEOUT_RETRY_VAD_DEACTIVATION_THRESHOLD is not None
                else vad_config["deactivation_threshold"]
            ),
        ),
    )

    if agent_wait_timeout_retry_vad_threshold_enabled:
        try:
            vad = load_silero_vad(vad_config)
            vad_config_source = f"{vad_config_source}_dedicated"
            timeline.emit(
                "vad_dynamic_threshold_dedicated_instance",
                activation_threshold=vad_config["activation_threshold"],
                deactivation_threshold=vad_config["deactivation_threshold"],
            )
        except Exception:
            logger.exception("[vad] failed loading dedicated VAD for dynamic threshold")
            call_logger.exception("[vad] failed loading dedicated VAD for dynamic threshold")
            agent_wait_timeout_retry_vad_threshold_enabled = False
            timeline.emit("vad_dynamic_threshold_dedicated_instance_failed")

    nudge_text = (str(md.get("nudge") or IDLE_NUDGE_TEXT_ENV or "Alô, você está aí?") or "").strip()

    call_t0 = time.monotonic()

    await ctx.connect()
    call_logger.info(
        "CALL_START | room=%s | protocol=%s | session_id=%s | bridge=%s",
        ctx.room.name,
        protocol,
        session_id,
        bridge_identity or "-",
    )
    log_flow_event(
        call_logger,
        "call_start",
        room=ctx.room.name,
        protocol=protocol,
        session_id=session_id,
        bridge=bridge_identity or "-",
    )
    log_flow_event(
        call_logger,
        "agent_runtime_config",
        min_interrupt_s=MIN_INTERRUPT_S,
        discard_audio_if_uninterruptible=DISCARD_AUDIO_IF_UNINTERRUPTIBLE,
        min_endpointing_delay=MIN_ENDPOINTING_DELAY,
        max_endpointing_delay=MAX_ENDPOINTING_DELAY,
        stt_min_audio_ms=os.getenv("STT_MIN_AUDIO_MS", "160"),
        stt_min_dbfs=os.getenv("STT_MIN_DBFS", "-50.0"),
        vad_config_source=vad_config_source,
        vad_decision_logging=vad_logging_config["log_decisions"],
        vad_activity_logging=vad_logging_config["log_activity"],
        vad_activity_min_prob=vad_logging_config["activity_min_probability"],
        vad_min_speech_duration=vad_config["min_speech_duration"],
        vad_min_silence_duration=vad_config["min_silence_duration"],
        vad_activation_threshold=vad_config["activation_threshold"],
        vad_deactivation_threshold=vad_config["deactivation_threshold"],
        vad_prefix_padding_duration=vad_config["prefix_padding_duration"],
        agent_wait_timeout_retry_vad_threshold_enabled=(
            agent_wait_timeout_retry_vad_threshold_enabled
        ),
        agent_wait_timeout_retry_vad_activation_threshold=(
            AGENT_WAIT_TIMEOUT_RETRY_VAD_ACTIVATION_THRESHOLD
        ),
        agent_wait_timeout_retry_vad_deactivation_threshold=(
            agent_wait_timeout_retry_vad_deactivation_threshold
        ),
        inflight_wait_interval_s=REMOTE_AGENT_INFLIGHT_WAIT_INTERVAL_S,
        inflight_wait_timeout_s=REMOTE_AGENT_INFLIGHT_WAIT_TIMEOUT_S,
        inflight_wait_max_notices=REMOTE_AGENT_INFLIGHT_WAIT_MAX_NOTICES,
        pre_backend_wait_notice_fast_on_vad_pause=(
            agent_runtime_config["pre_backend_wait_notice_fast_on_vad_pause"]
        ),
        pre_backend_wait_notice_fast_on_vad_pause_source=(
            pre_backend_wait_notice_fast_source
        ),
        inflight_wait_short_audio_dir=REMOTE_AGENT_INFLIGHT_WAIT_SHORT_AUDIO_DIR,
        inflight_wait_long_audio_dir=REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_DIR,
        inflight_wait_long_audio_path=REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_PATH,
    )
    timeline.emit(
        "call_start",
        backend=resolve_agent_backend_name(call_config, os.getenv("AGENT_BACKEND", "remote_ws")),
        bridge_identity=bridge_identity,
        remote_agent=remote_agent_context.get("agent", ""),
        timeline_file=str(timeline.path),
    )
    bridge_gateway = BridgeGateway(
        room=ctx.room,
        bridge_identity=bridge_identity,
        protocol=protocol,
        timeline=timeline,
        stress_test=bool(fake_agent_overrides.get("responses")),
    )
    export_service = ExportService()
    runtime_ref: dict[str, CallRuntime | None] = {"runtime": None}

    def _structured_interruption_flag() -> bool:
        runtime = runtime_ref["runtime"]
        return bool(runtime and runtime.structured_recebimento_interruption_flag())

    stt_overrides = resolve_stt_overrides(call_config)
    stt_provider = (stt_overrides.get("provider") or os.getenv("STT_PROVIDER", "internal_http")).strip().lower()
    if stt_provider not in {"", "internal_http", "fake"}:
        raise RuntimeError(f"Unsupported STT provider: {stt_provider}")

    if stt_provider == "fake":
        stt = FakeSTT(
            language=stt_overrides.get("language") or os.getenv("STT_LANG", "pt-BR"),
            timeline=timeline,
            structured_log_context=structured_context,
            structured_logger=call_logger,
            structured_interruption_flag=_structured_interruption_flag,
        )
    else:
        disable_vosk = str(stt_overrides.get("disable_vosk") or "").strip().lower() in {
            "1",
            "true",
            "yes",
            "on",
        }
        stt_cfg = InternalSTTConfig(
            url=require_env("STT_URL", context="STT provider internal_http", timeline=timeline),
            api_key=(
                stt_overrides.get("api_key")
                or stt_overrides.get("STT_KEY")
                or os.getenv("STT_KEY", "")
                or "unknow"
            ),
            language=stt_overrides.get("language",  "portuguese") or os.getenv("STT_LANG", ""),
            min_prob_single_word=float(
                stt_overrides.get("min_prob_single_word") or os.getenv("STT_MIN_PROB_SINGLE_WORD", "0.10")
            ),
            initial_prompt=stt_overrides.get("initial_prompt") or os.getenv("STT_INITIAL_PROMPT", ""),
            config_override=stt_overrides.get("config_override") or os.getenv("STT_CONFIG_OVERRIDE", ""),
        )
        stt = InternalHTTPSTT(
            stt_cfg,
            client=http_client,
            vosk=None if disable_vosk else vosk,
            vosk_lock=None if disable_vosk else vosk_lock,
            timeline=timeline,
            structured_log_context=structured_context,
            structured_logger=call_logger,
            structured_interruption_flag=_structured_interruption_flag,
        )

    tts_overrides = resolve_tts_overrides(call_config)
    tts_provider = (tts_overrides.get("provider") or os.getenv("TTS_PROVIDER", "elevenlabs")).strip().lower()
    if tts_provider not in {"", "elevenlabs", "azure", "fake", "xai"}:
        raise RuntimeError(f"Unsupported TTS provider: {tts_provider}")

    if tts_provider == "fake":
        tts = livekit_tts.StreamAdapter(tts=FakeSessionTTS())
    elif tts_provider in {"", "elevenlabs"}:
        tts = elevenlabs.TTS(
            model=tts_overrides.get("model_id") or os.getenv("ELEVENLABS_MODEL_ID", ""),
            voice_id=tts_overrides.get("voice_id") or os.getenv("ELEVENLABS_VOICE_ID", ""),
            api_key=os.getenv("ELEVENLABS_API_KEY", ""),
            voice_settings=elevenlabs.VoiceSettings(
                stability=0.35,
                speed=1.02,
                similarity_boost=0.75,
                use_speaker_boost=True,
                style=0.7,
            ),
            language="pt",
        )
    elif tts_provider == "xai":
        xai_auth_method = (os.getenv("XAI_TTS_AUTH_METHOD", "API_KEY") or "API_KEY").strip().upper()
        xai_auth_options: dict[str, str] = {"auth_method": xai_auth_method}
        if xai_auth_method == "API_KEY":
            xai_auth_options["api_key"] = require_env(
                "XAI_API_KEY", context="TTS provider xai", timeline=timeline
            )
        tts = OraclexAITTS(
            voice=tts_overrides.get("voice_id") or os.getenv("XAI_TTS_VOICE", ""),
            language=tts_overrides.get("language") or os.getenv("XAI_TTS_LANGUAGE", "pt-BR"),
            initial_greeting_audio_cache=initial_greeting_audio_cache,
            initial_greeting_agent=str(remote_agent_context.get("agent") or AGENT_NAME),
            **xai_auth_options,
        )
    else:
        azure_tts_config, missing_keys = resolve_azure_speech_tts_config(tts_overrides)
        for key in missing_keys:
            if timeline is not None:
                timeline.emit(
                    "config_missing",
                    key=key,
                    context="TTS provider azure",
                )
        if missing_keys:
            missing = ", ".join(missing_keys)
            raise RuntimeError(
                f"Missing required Azure Speech TTS config: {missing}. "
                "Add the Azure Speech variables to .env.dev before starting the agent."
            )

        azure_impl = (os.getenv("AZURE_TTS_IMPLEMENTATION", "plugin") or "plugin").strip().lower()
        if azure_impl not in {"plugin", "rest"}:
            raise RuntimeError(
                f"Unsupported AZURE_TTS_IMPLEMENTATION: {azure_impl}. "
                "Use 'plugin' or 'rest'."
            )

        if azure_impl == "rest":
            tts = AzureRESTTTS(
                sample_rate=AZURE_SPEECH_TTS_SAMPLE_RATE,
                **azure_tts_config,
            )
        else:
            try:
                from livekit.plugins import azure as livekit_azure
            except ModuleNotFoundError as exc:
                raise RuntimeError(
                    "Azure Speech plugin is not installed. "
                    "Run make setup or reinstall requirements.txt before using TTS_PROVIDER=azure."
                ) from exc

            tts = livekit_azure.TTS(
                sample_rate=AZURE_SPEECH_TTS_SAMPLE_RATE,
                **azure_tts_config,
            )

    if isinstance(tts, OraclexAITTS):
        async def _prewarm_initial_tts_connection() -> None:
            timeout_s = max(0.1, float(os.getenv("XAI_TTS_PREWARM_TIMEOUT_S", "3")))
            if timeline is not None:
                timeline.emit("initial_tts_prewarm_started", timeout_ms=round(timeout_s * 1000))
            try:
                await tts.prewarm_connection(timeout=timeout_s)
            except Exception as exc:
                logger.info("INITIAL_TTS_PREWARM_FAILED | error=%s", type(exc).__name__)
                if timeline is not None:
                    timeline.emit("initial_tts_prewarm_failed", error=type(exc).__name__)
            else:
                if timeline is not None:
                    timeline.emit("initial_tts_prewarm_completed")

        create_task_logged(_prewarm_initial_tts_connection(), name="initial_tts_prewarm")

    vad_threshold_controller = (
        DynamicVADThresholdController(
            vad,
            config=DynamicVADThresholdConfig(
                enabled=agent_wait_timeout_retry_vad_threshold_enabled,
                baseline_activation_threshold=vad_config["activation_threshold"],
                baseline_deactivation_threshold=vad_config["deactivation_threshold"],
                retry_activation_threshold=AGENT_WAIT_TIMEOUT_RETRY_VAD_ACTIVATION_THRESHOLD,
                retry_deactivation_threshold=agent_wait_timeout_retry_vad_deactivation_threshold,
            ),
            call_logger=call_logger,
            timeline=timeline,
        )
        if AGENT_WAIT_TIMEOUT_RETRY_VAD_THRESHOLD_ENABLED
        else None
    )

    pre_backend_wait_notice_bridge = PreBackendWaitNoticeBridge()
    # The deferred-interruption threshold is decided from END_OF_SPEECH events,
    # so this wrapper is required even when verbose VAD logging is disabled.
    wrap_vad = True
    session_vad = (
        FlowLoggingVAD(
            vad,
            call_logger=call_logger,
            min_interrupt_s=MIN_INTERRUPT_S,
            vad_config=vad_config,
            log_decisions=vad_logging_config["log_decisions"],
            log_activity=vad_logging_config["log_activity"],
            activity_min_prob=vad_logging_config["activity_min_probability"],
            threshold_provider=(
                vad_threshold_controller.current_thresholds
                if vad_threshold_controller is not None
                else None
            ),
            on_pre_backend_wait_pause=pre_backend_wait_notice_bridge.notify,
            on_speech_end=pre_backend_wait_notice_bridge.notify_speech_end,
        )
        if wrap_vad
        else vad
    )

    session = AgentSession(
        stt=stt,
        tts=tts,
        vad=session_vad,
        discard_audio_if_uninterruptible=DISCARD_AUDIO_IF_UNINTERRUPTIBLE,
        min_interruption_duration=MIN_INTERRUPT_S,
        min_endpointing_delay=MIN_ENDPOINTING_DELAY,
        max_endpointing_delay=MAX_ENDPOINTING_DELAY,
        use_tts_aligned_transcript=True,
        user_away_timeout=None,
    )
    speech_service = SpeechService(session)

    agent = PipelineVoiceAgent(
        elegibility=elegibility,
        protocol=protocol,
        intro=intro,
        backend_name=resolve_agent_backend_name(call_config, os.getenv("AGENT_BACKEND", "remote_ws")),
        remote_agent_context=remote_agent_context,
        timeline=timeline,
    )
    command_executor = RuntimeCommandExecutor(
        agent=agent,
        bridge_gateway=bridge_gateway,
        export_service=export_service,
        session=session,
        speech_service=speech_service,
    )
    interrupt_policy = InterruptPolicy()
    idle_policy = IdlePolicy(
        IdlePolicyConfig(
            enabled=IDLE_NUDGE_ENABLED,
            delay_s=IDLE_NUDGE_DELAY_S,
            join_delay_s=IDLE_NUDGE_JOIN_DELAY_S,
            close_delay_s=IDLE_NUDGE_CLOSE_DELAY_S,
            max_tries=IDLE_NUDGE_MAX_TRIES,
            end_reason=IDLE_NUDGE_END_REASON,
        )
    )
    finalization_policy = FinalizationPolicy(call_end_grace_s=CALL_END_GRACE_S)
    runtime = CallRuntime(
        ctx=ctx,
        session=session,
        agent=agent,
        command_executor=command_executor,
        call_logger=call_logger,
        protocol=protocol,
        session_id=session_id,
        bridge_identity=bridge_identity,
        agent_starts_conversation=agent_starts_conversation,
        nudge_text=nudge_text,
        call_t0=call_t0,
        extract_text_from_transcript=extract_text_from_transcript,
        interrupt_policy=interrupt_policy,
        idle_policy=idle_policy,
        finalization_policy=finalization_policy,
        create_task_logged=create_task_logged,
        vad_threshold_controller=vad_threshold_controller,
        initial_greeting_tts=tts,
        config=RuntimeConfig(
            call_end_grace_s=CALL_END_GRACE_S,
            final_grace_s=FINAL_GRACE_S,
            idle_nudge_delay_s=IDLE_NUDGE_DELAY_S,
            idle_nudge_join_delay_s=IDLE_NUDGE_JOIN_DELAY_S,
            idle_nudge_close_delay_s=IDLE_NUDGE_CLOSE_DELAY_S,
            idle_nudge_max_tries=IDLE_NUDGE_MAX_TRIES,
            idle_nudge_end_reason=IDLE_NUDGE_END_REASON,
            inflight_backend_wait_interval_s=REMOTE_AGENT_INFLIGHT_WAIT_INTERVAL_S,
            inflight_backend_wait_timeout_s=REMOTE_AGENT_INFLIGHT_WAIT_TIMEOUT_S,
            inflight_backend_wait_max_notices=REMOTE_AGENT_INFLIGHT_WAIT_MAX_NOTICES,
            inflight_backend_wait_text=REMOTE_AGENT_INFLIGHT_WAIT_TEXT,
            inflight_backend_wait_short_audio_dir=REMOTE_AGENT_INFLIGHT_WAIT_SHORT_AUDIO_DIR,
            inflight_backend_wait_long_audio_dir=REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_DIR,
            deferred_interruption_min_audio_ms=agent_runtime_config[
                "deferred_interruption_min_audio_ms"
            ],
            deferred_interruption_enabled=agent_runtime_config[
                "deferred_interruption_enabled"
            ],
            deferred_interruption_stt_settle_timeout_s=DEFERRED_INTERRUPTION_STT_SETTLE_TIMEOUT_S,
            deferred_interruption_user_turn_timeout_s=DEFERRED_INTERRUPTION_USER_TURN_TIMEOUT_S,
            inflight_backend_wait_long_audio_path=REMOTE_AGENT_INFLIGHT_WAIT_LONG_AUDIO_PATH,
            pre_backend_wait_notice_fast_on_vad_pause=(
                agent_runtime_config["pre_backend_wait_notice_fast_on_vad_pause"]
            ),
        ),
        logger=logger,
        timeline=timeline,
        structured_log_context=structured_context,
        debug_event_publisher=(
            bridge_gateway.publish_debug_event if debug_events_enabled else None
        ),
    )
    runtime_ref["runtime"] = runtime
    empty_transcript_handler = getattr(stt, "set_empty_transcript_handler", None)
    if callable(empty_transcript_handler):
        empty_transcript_handler(
            lambda: runtime.handle_empty_stt_final(source="stt_provider")
        )
    stt_metrics_handler = getattr(stt, "set_metrics_handler", None)
    if callable(stt_metrics_handler):
        stt_metrics_handler(runtime.publish_stt_provider_metrics)
    pre_backend_wait_notice_bridge.bind(runtime)
    await runtime.run()


if __name__ == "__main__":
    cli.run_app(server)
