# ws_bridge.py
import asyncio
import json
import os
import time
import uuid
import audioop
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, Mapping, Optional

from dotenv import load_dotenv
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, JSONResponse

from livekit import rtc
from livekit.api import AccessToken, VideoGrants
from livekit import api as lkapi

from app.common.call_config import (
    resolve_agent_backend_name,
    resolve_stt_overrides,
    resolve_tts_overrides,
    resolve_ws_overrides,
)
from app.ws_gateway.fake_remote_agent import build_fake_remote_agent_response
from app.ws_gateway.session_audio import (
    enable_client_audio,
    notify_client_audio_enabled,
    stream_agent_audio_when_ready,
    watch_mock_stop_after_first_audio,
)
from app.ws_gateway.session_bootstrap import build_bridge_session_bootstrap
from app.ws_gateway.session_lifecycle import (
    RoomLifecycleState,
    close_websocket,
    register_room_lifecycle_handlers,
    watch_call_done,
)
from app.ws_gateway.readiness import (
    BridgeReadinessManager,
    build_bridge_failed_stop_message,
    build_capacity_stop_message,
)
from app.ws_gateway.session_start import recv_start_message
from app.utils.call_timeline import CallTimeline
from app.utils.audio_backlog import ShedResult, rms_threshold_from_dbfs, shed_queue_backlog
from app.utils.background import AudioActivity, BridgeOutputStats, ws_out_loop
from app.utils.full_call_recording import create_entire_call_recorder_from_env
from app.utils.pcm import pcm16_dbfs
from app.utils.logging import log_flow_event, setup_minimal_logging

logger = setup_minimal_logging()
app = FastAPI()
READINESS_GATE = BridgeReadinessManager()

SAMPLE_RATE = 16000
CHANNELS = 1
FRAME_MS = 20
SAMPLES_PER_FRAME = int(SAMPLE_RATE * FRAME_MS / 1000)  # 320
BYTES_PER_FRAME = SAMPLES_PER_FRAME * 2 * CHANNELS  # 640

HOLD_SILENCE_DBFS = float(os.getenv("HOLD_SILENCE_DBFS", "-50"))
RMS_TH = int(32768 * (10 ** (HOLD_SILENCE_DBFS / 20.0)))
SILENCE_CHECK_EVERY = int(os.getenv("SILENCE_CHECK_EVERY", "5"))  # checa a cada 5 frames (100ms)

# --- ENV / AMBIENTE ---
APP_ENV = (os.getenv("APP_ENV") or os.getenv("ENV") or "dev").strip().lower()
DOTENV_PATH = os.getenv("DOTENV_PATH", f".env.{APP_ENV}")

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()

LK_CATCHUP_ENABLED = os.getenv("LK_CATCHUP_ENABLED", "1") == "1"
LK_CATCHUP_KEEP_MS = int(os.getenv("LK_CATCHUP_KEEP_MS", "1000"))
LK_CATCHUP_MAX_FAST_FRAMES = int(os.getenv("LK_CATCHUP_MAX_FAST_FRAMES", "4000"))

# --- Marcador de "fala ativa" do ÁUDIO DO AGENTE (só telemetria/idle; NÃO filtra áudio) ---
AGENT_OUT_SPEECH_DBFS = float(os.getenv("AGENT_OUT_SPEECH_DBFS", "-52"))   # acima disso = fala ativa
AGENT_OUT_SPEECH_RMS_TH = int(32768 * (10 ** (AGENT_OUT_SPEECH_DBFS / 20.0)))
AGENT_OUT_RMS_EVERY = int(os.getenv("AGENT_OUT_RMS_EVERY", "1"))          # 1 = checa todo frame
FLOW_AUDIO_BURST_GAP_S = float(os.getenv("FLOW_AUDIO_BURST_GAP_S", "0.8"))
FLOW_LOG_AUDIO_BURST_END = (
    os.getenv("FLOW_LOG_AUDIO_BURST_END", "0").strip().lower()
    in {"1", "true", "yes", "on"}
)
VOICE_CLIENT_HTML = Path(__file__).with_name("voice_client.html")


def _load_runtime_env() -> None:
    if DOTENV_PATH:
        load_dotenv(DOTENV_PATH, override=False)
    load_dotenv(override=False)


def _env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def _env_int(name: str, default: int) -> int:
    try:
        return int(os.getenv(name, str(default)))
    except ValueError:
        return default


def _env_float(name: str, default: float) -> float:
    try:
        return float(os.getenv(name, str(default)))
    except ValueError:
        return default


def _frames_from_ms(value_ms: int, *, minimum: int = 1) -> int:
    value_ms = max(0, int(value_ms))
    return max(minimum, (value_ms + FRAME_MS - 1) // FRAME_MS)


@dataclass(frozen=True)
class AudioInputBacklogConfig:
    shed_enabled: bool
    shed_threshold_ms: int
    shed_keep_ms: int
    latency_metrics_enabled: bool
    latency_alert_ms: int
    latency_log_interval_s: float
    livekit_source_queue_size_ms: int
    livekit_source_clear_on_shed: bool
    config_source: str = "env"
    call_config_overrides: tuple[str, ...] = ()
    # Descarte guiado por energia: no regime de drift (excesso pequeno) so
    # descarta silencio; so cai no descarte cego quando o excesso passa de
    # energy_shed_max_excess_ms (regime de rajada).
    energy_shed_enabled: bool = True
    energy_shed_max_excess_ms: int = 1000
    silence_dbfs: float = -50.0

    @property
    def shed_threshold_frames(self) -> int:
        return _frames_from_ms(self.shed_threshold_ms)

    @property
    def shed_keep_frames(self) -> int:
        return min(
            self.shed_threshold_frames,
            _frames_from_ms(self.shed_keep_ms),
        )

    @property
    def silence_rms_threshold(self) -> int:
        return rms_threshold_from_dbfs(self.silence_dbfs)


def _call_config_has_value(overrides: Mapping[str, Any], key: str) -> bool:
    return overrides.get(key) not in (None, "")


def _config_bool(
    overrides: Mapping[str, Any],
    key: str,
    env_name: str,
    default: bool,
) -> bool:
    env_value = _env_bool(env_name, default)
    if not _call_config_has_value(overrides, key):
        return env_value
    return str(overrides.get(key)).strip().lower() in {"1", "true", "yes", "on"}


def _config_int(
    overrides: Mapping[str, Any],
    key: str,
    env_name: str,
    default: int,
) -> int:
    env_value = _env_int(env_name, default)
    if not _call_config_has_value(overrides, key):
        return env_value
    try:
        return int(str(overrides.get(key)).strip())
    except (TypeError, ValueError):
        return env_value


def _config_float(
    overrides: Mapping[str, Any],
    key: str,
    env_name: str,
    default: float,
) -> float:
    env_value = _env_float(env_name, default)
    if not _call_config_has_value(overrides, key):
        return env_value
    try:
        return float(str(overrides.get(key)).strip())
    except (TypeError, ValueError):
        return env_value


def audio_input_backlog_config_from_env(
    call_config: Optional[Mapping[str, Any]] = None,
) -> AudioInputBacklogConfig:
    ws_overrides = resolve_ws_overrides(call_config)
    override_fields = tuple(
        key
        for key in (
            "audio_in_backlog_shed_enabled",
            "audio_in_backlog_shed_threshold_ms",
            "audio_in_backlog_shed_keep_ms",
            "audio_in_latency_metrics_enabled",
            "audio_in_latency_alert_ms",
            "audio_in_latency_log_interval_s",
            "livekit_audio_source_queue_size_ms",
            "livekit_audio_source_clear_on_shed",
            "audio_in_backlog_energy_shed_enabled",
            "audio_in_backlog_energy_shed_max_excess_ms",
            "audio_in_backlog_silence_dbfs",
        )
        if _call_config_has_value(ws_overrides, key)
    )
    shed_enabled = _config_bool(
        ws_overrides,
        "audio_in_backlog_shed_enabled",
        "AUDIO_IN_BACKLOG_SHED_ENABLED",
        True,
    )
    default_source_queue_ms = 500 if shed_enabled else 5000
    return AudioInputBacklogConfig(
        shed_enabled=shed_enabled,
        shed_threshold_ms=max(
            FRAME_MS,
            _config_int(
                ws_overrides,
                "audio_in_backlog_shed_threshold_ms",
                "AUDIO_IN_BACKLOG_SHED_THRESHOLD_MS",
                500,
            ),
        ),
        shed_keep_ms=max(
            FRAME_MS,
            _config_int(
                ws_overrides,
                "audio_in_backlog_shed_keep_ms",
                "AUDIO_IN_BACKLOG_SHED_KEEP_MS",
                300,
            ),
        ),
        latency_metrics_enabled=_config_bool(
            ws_overrides,
            "audio_in_latency_metrics_enabled",
            "AUDIO_IN_LATENCY_METRICS_ENABLED",
            True,
        ),
        latency_alert_ms=max(
            0,
            _config_int(
                ws_overrides,
                "audio_in_latency_alert_ms",
                "AUDIO_IN_LATENCY_ALERT_MS",
                1000,
            ),
        ),
        latency_log_interval_s=max(
            0.1,
            _config_float(
                ws_overrides,
                "audio_in_latency_log_interval_s",
                "AUDIO_IN_LATENCY_LOG_INTERVAL_S",
                15.0,
            ),
        ),
        livekit_source_queue_size_ms=max(
            FRAME_MS,
            _config_int(
                ws_overrides,
                "livekit_audio_source_queue_size_ms",
                "LIVEKIT_AUDIO_SOURCE_QUEUE_SIZE_MS",
                default_source_queue_ms,
            ),
        ),
        livekit_source_clear_on_shed=_config_bool(
            ws_overrides,
            "livekit_audio_source_clear_on_shed",
            "LIVEKIT_AUDIO_SOURCE_CLEAR_ON_SHED",
            shed_enabled,
        ),
        energy_shed_enabled=_config_bool(
            ws_overrides,
            "audio_in_backlog_energy_shed_enabled",
            "AUDIO_IN_BACKLOG_ENERGY_SHED_ENABLED",
            True,
        ),
        energy_shed_max_excess_ms=max(
            0,
            _config_int(
                ws_overrides,
                "audio_in_backlog_energy_shed_max_excess_ms",
                "AUDIO_IN_BACKLOG_ENERGY_SHED_MAX_EXCESS_MS",
                1000,
            ),
        ),
        silence_dbfs=_config_float(
            ws_overrides,
            "audio_in_backlog_silence_dbfs",
            "AUDIO_IN_BACKLOG_SILENCE_DBFS",
            -60.0,
        ),
        config_source="call_config" if override_fields else "env",
        call_config_overrides=override_fields,
    )


class AudioInputLatencyTracker:
    def __init__(
        self,
        config: AudioInputBacklogConfig,
        *,
        timeline: Optional[CallTimeline] = None,
        call_context: Optional[Mapping[str, Any]] = None,
        debug_event_publisher: Optional[Callable[[str, Mapping[str, Any]], None]] = None,
    ) -> None:
        self.config = config
        self.timeline = timeline
        self.call_context = dict(call_context or {})
        self.debug_event_publisher = debug_event_publisher
        self.enabled_at: float | None = None
        self.last_latency_log_at = 0.0
        self.total_dropped_frames = 0
        self.peak_excess_ms = 0
        self.peak_raw_excess_ms = 0

    def mark_enabled(self, now: Optional[float] = None) -> None:
        if self.enabled_at is None:
            self.enabled_at = time.monotonic() if now is None else now
            self.last_latency_log_at = 0.0

    def snapshot(
        self,
        *,
        frame_count: int,
        now: Optional[float] = None,
        queue_size_frames: int = 0,
    ) -> Dict[str, int]:
        if self.enabled_at is None:
            self.mark_enabled(now)
        now = time.monotonic() if now is None else now
        enabled_at = self.enabled_at if self.enabled_at is not None else now
        elapsed_ms = max(0, round((now - enabled_at) * 1000))
        frames_since_enable = max(0, int(frame_count))
        received_audio_ms = frames_since_enable * FRAME_MS
        effective_frames = max(0, frames_since_enable - self.total_dropped_frames)
        effective_audio_ms = effective_frames * FRAME_MS
        raw_excess_ms = received_audio_ms - elapsed_ms
        excess_ms = effective_audio_ms - elapsed_ms
        raw_lag_ms = max(0, raw_excess_ms)
        lag_ms = max(0, excess_ms)
        self.peak_excess_ms = max(self.peak_excess_ms, excess_ms)
        self.peak_raw_excess_ms = max(self.peak_raw_excess_ms, raw_excess_ms)
        return {
            "frames_since_enable": frames_since_enable,
            "received_audio_ms": received_audio_ms,
            "effective_frames_since_enable": effective_frames,
            "effective_audio_ms": effective_audio_ms,
            "elapsed_since_enable_ms": elapsed_ms,
            "excess_ms": excess_ms,
            "lag_ms": lag_ms,
            "peak_excess_ms": self.peak_excess_ms,
            "raw_excess_ms": raw_excess_ms,
            "raw_lag_ms": raw_lag_ms,
            "peak_raw_excess_ms": self.peak_raw_excess_ms,
            "queue_frames": max(0, int(queue_size_frames)),
            "queue_ms": max(0, int(queue_size_frames)) * FRAME_MS,
            "total_dropped_frames": self.total_dropped_frames,
            "total_dropped_ms": self.total_dropped_frames * FRAME_MS,
            "saved_latency_ms": self.total_dropped_frames * FRAME_MS,
        }

    def maybe_log_latency(
        self,
        *,
        frame_count: int,
        now: Optional[float] = None,
        queue_size_frames: int = 0,
        reason: str,
        force: bool = False,
    ) -> None:
        if not self.config.latency_metrics_enabled:
            return
        now = time.monotonic() if now is None else now
        fields = self.snapshot(
            frame_count=frame_count,
            now=now,
            queue_size_frames=queue_size_frames,
        )
        should_log = (
            force
            or fields["lag_ms"] >= self.config.latency_alert_ms
            or fields["queue_ms"] >= self.config.latency_alert_ms
            or (now - self.last_latency_log_at) >= self.config.latency_log_interval_s
        )
        if not should_log:
            return
        self.last_latency_log_at = now
        payload = {
            **self.call_context,
            **fields,
            "reason": reason,
            "alert_ms": self.config.latency_alert_ms,
        }
        log_flow_event(logger, "audio_in_latency", **payload)
        if self.timeline is not None:
            self.timeline.emit("audio_in_latency", **payload)

    def record_shed(
        self,
        *,
        source: str,
        dropped_frames: int,
        queue_before_frames: int,
        queue_after_frames: int,
        frame_count: int,
        mode: str = "blind",
        dropped_silent: int = 0,
        dropped_voiced: int = 0,
        now: Optional[float] = None,
    ) -> None:
        if dropped_frames <= 0:
            return
        now = time.monotonic() if now is None else now
        self.total_dropped_frames += int(dropped_frames)
        fields = self.snapshot(
            frame_count=frame_count,
            now=now,
            queue_size_frames=queue_after_frames,
        )
        payload = {
            **self.call_context,
            **fields,
            "source": source,
            "mode": mode,
            "content_policy": "silence_only" if mode == "energy" else "oldest_frames",
            "sync_action": "compress_silence" if mode == "energy" else "drop_audio_to_catch_up",
            "dropped_frames": int(dropped_frames),
            "dropped_ms": int(dropped_frames) * FRAME_MS,
            "dropped_silent_frames": int(dropped_silent),
            "dropped_silent_ms": int(dropped_silent) * FRAME_MS,
            "dropped_voiced_frames": int(dropped_voiced),
            "dropped_voiced_ms": int(dropped_voiced) * FRAME_MS,
            "saved_ms": int(dropped_frames) * FRAME_MS,
            "queue_before_frames": int(queue_before_frames),
            "queue_before_ms": int(queue_before_frames) * FRAME_MS,
            "queue_after_frames": int(queue_after_frames),
            "queue_after_ms": int(queue_after_frames) * FRAME_MS,
            "shed_threshold_ms": self.config.shed_threshold_ms,
            "shed_keep_ms": self.config.shed_keep_ms,
        }
        log_flow_event(logger, "audio_in_latency_shed", **payload)
        if self.timeline is not None:
            self.timeline.emit("audio_in_latency_shed", **payload)
        if self.debug_event_publisher is not None:
            self.debug_event_publisher("bridge.audio_in.shed", payload)

    def record_livekit_source_clear(
        self,
        *,
        source: str,
        queued_ms: int,
        frame_count: int,
        queue_size_frames: int,
        now: Optional[float] = None,
    ) -> None:
        now = time.monotonic() if now is None else now
        fields = self.snapshot(
            frame_count=frame_count,
            now=now,
            queue_size_frames=queue_size_frames,
        )
        payload = {
            **self.call_context,
            **fields,
            "source": source,
            "livekit_queued_ms": max(0, int(queued_ms)),
        }
        log_flow_event(logger, "audio_in_livekit_source_cleared", **payload)
        if self.timeline is not None:
            self.timeline.emit("audio_in_livekit_source_cleared", **payload)


def _shed_audio_queue_backlog(
    audio_q: asyncio.Queue[bytes],
    *,
    config: AudioInputBacklogConfig,
    tracker: Optional[AudioInputLatencyTracker],
    source: str,
    frame_count: int,
    now: Optional[float] = None,
) -> ShedResult:
    if not config.shed_enabled:
        return ShedResult()
    queue_before = audio_q.qsize()
    if queue_before <= config.shed_threshold_frames:
        return ShedResult()

    keep_frames = config.shed_keep_frames
    over_keep_ms = max(0, queue_before - keep_frames) * FRAME_MS
    # Regime de drift (excesso pequeno): so descarta silencio, preservando fala.
    # Regime de rajada (excesso grande): descarte cego para recuperar latencia.
    blind = (
        not config.energy_shed_enabled
        or over_keep_ms > config.energy_shed_max_excess_ms
    )
    result = shed_queue_backlog(
        audio_q,
        keep_frames=keep_frames,
        rms_threshold=config.silence_rms_threshold,
        blind=blind,
    )

    if result.dropped and tracker is not None:
        tracker.record_shed(
            source=source,
            dropped_frames=result.dropped,
            dropped_silent=result.dropped_silent,
            dropped_voiced=result.dropped_voiced,
            mode=result.mode,
            queue_before_frames=queue_before,
            queue_after_frames=audio_q.qsize(),
            frame_count=frame_count,
            now=now,
        )
    return result


def _audio_input_latency_burst_fields(
    tracker: Optional[AudioInputLatencyTracker],
    *,
    frame_count: int,
    now: float,
    queue_size_frames: int,
) -> Dict[str, Any]:
    if tracker is None or not tracker.config.latency_metrics_enabled:
        return {}
    fields = tracker.snapshot(
        frame_count=frame_count,
        now=now,
        queue_size_frames=queue_size_frames,
    )
    return {
        "input_excess_ms": fields["excess_ms"],
        "input_lag_ms": fields["lag_ms"],
        "input_raw_excess_ms": fields["raw_excess_ms"],
        "input_raw_lag_ms": fields["raw_lag_ms"],
        "input_queue_ms": fields["queue_ms"],
        "input_total_dropped_ms": fields["total_dropped_ms"],
    }


def _maybe_clear_livekit_source_queue(
    source: rtc.AudioSource,
    *,
    tracker: Optional[AudioInputLatencyTracker],
    reason: str,
    frame_count: int,
    queue_size_frames: int,
    now: Optional[float] = None,
) -> int:
    if tracker is None or not tracker.config.livekit_source_clear_on_shed:
        return 0
    clear_queue = getattr(source, "clear_queue", None)
    if not callable(clear_queue):
        return 0
    try:
        queued_ms = max(0, round(float(getattr(source, "queued_duration", 0.0)) * 1000))
    except Exception:
        queued_ms = 0
    if queued_ms <= 0:
        return 0
    try:
        clear_queue()
    except Exception:
        logger.debug("LIVEKIT_AUDIO_SOURCE_CLEAR_FAIL", exc_info=True)
        return 0
    tracker.record_livekit_source_clear(
        source=reason,
        queued_ms=queued_ms,
        frame_count=frame_count,
        queue_size_frames=queue_size_frames,
        now=now,
    )
    return queued_ms


def _is_new_audio_burst(now: float, last_audio_at: float) -> bool:
    return last_audio_at <= 0.0 or (now - last_audio_at) >= FLOW_AUDIO_BURST_GAP_S


class FlowAudioBurstLogger:
    def __init__(self, *, start_step: str, end_step: str) -> None:
        self.start_step = start_step
        self.end_step = end_step
        self.seq = 0
        self.active = False
        self.first_frame = 0
        self.last_frame = 0
        self.last_audio_at = 0.0
        self.active_frames = 0
        self.peak_rms = 0
        self.peak_dbfs = -120.0

    def maybe_end(self, *, now: float, reason: str) -> None:
        if self.active and _is_new_audio_burst(now, self.last_audio_at):
            self._finish(reason=reason)

    def observe(
        self,
        *,
        rms: int,
        dbfs: float,
        frame: int,
        now: float,
        start_fields: Optional[Dict[str, Any]] = None,
    ) -> Optional[int]:
        self.maybe_end(now=now, reason="silence_gap")
        if rms <= RMS_TH:
            return None

        new_burst = not self.active
        if new_burst:
            self.seq += 1
            self.active = True
            self.first_frame = frame
            self.active_frames = 0
            self.peak_rms = 0
            self.peak_dbfs = -120.0

        self.last_frame = frame
        self.last_audio_at = now
        self.active_frames += 1
        if rms >= self.peak_rms:
            self.peak_rms = rms
            self.peak_dbfs = dbfs

        if new_burst:
            log_flow_event(
                logger,
                self.start_step,
                seq=self.seq,
                rms=rms,
                dbfs=f"{dbfs:.2f}",
                frame=frame,
                **(start_fields or {}),
            )
            return self.seq

        return None

    def _finish(self, *, reason: str) -> None:
        if FLOW_LOG_AUDIO_BURST_END:
            duration_ms = max(FRAME_MS, (self.last_frame - self.first_frame + 1) * FRAME_MS)
            log_flow_event(
                logger,
                self.end_step,
                seq=self.seq,
                reason=reason,
                duration_ms=duration_ms,
                active_frames=self.active_frames,
                first_frame=self.first_frame,
                last_frame=self.last_frame,
                peak_rms=self.peak_rms,
                peak_dbfs=f"{self.peak_dbfs:.2f}",
            )
        self.active = False


class BridgeDebugState:
    """
    Debug timeline (somente DEBUG).
    Mantive pra você poder ligar nível DEBUG quando precisar investigar latência,
    sem poluir INFO.
    """

    def __init__(self, *, base_t: float, timeline: Optional[CallTimeline] = None):
        self.base_t = base_t
        self.timeline = timeline
        self.first_ws_in: Optional[float] = None
        self.first_lk_out: Optional[float] = None
        self.in_frames: int = 0
        self.out_frames: int = 0

    def tlog(self, msg: str, **kv):
        logger.debug("[T+%.3fs] %s | %s", time.monotonic() - self.base_t, msg, kv)

    def mark_ws_in(self):
        if self.first_ws_in is None:
            self.first_ws_in = time.monotonic()
            self.tlog("WS->BRIDGE first_frame", in_frames=self.in_frames)
            if self.timeline is not None:
                self.timeline.emit(
                    "client_audio_first_frame_received",
                    in_frames=self.in_frames,
                )

    def mark_lk_out(self, backlog: int):
        if self.first_lk_out is None:
            self.first_lk_out = time.monotonic()
            self.tlog("BRIDGE->LK first_capture", out_frames=self.out_frames, backlog=backlog)
            if self.timeline is not None:
                self.timeline.emit(
                    "client_audio_first_frame_published",
                    out_frames=self.out_frames,
                    backlog_frames=backlog,
                )


def make_token(identity: str, room_name: str) -> str:
    grants = VideoGrants(room_join=True, room=room_name)
    return (
        AccessToken(
            api_key=os.environ["LIVEKIT_API_KEY"],
            api_secret=os.environ["LIVEKIT_API_SECRET"],
        )
        .with_identity(identity)
        .with_grants(grants)
        .to_jwt()
    )


def is_agent(p: rtc.Participant) -> bool:
    try:
        return int(p.kind) == 4
    except Exception:
        ident = str(getattr(p, "identity", "") or "")
        return ident == "agent" or ident.startswith("agent")


@app.get("/voice-client")
def voice_client() -> FileResponse:
    return FileResponse(VOICE_CLIENT_HTML)


@app.websocket("/fake-agent/ws")
async def fake_remote_agent_ws(ws: WebSocket):
    await ws.accept()
    logger.info("FAKE_REMOTE_AGENT_CONNECT")

    try:
        while True:
            raw = await ws.receive_text()
            payload = json.loads(raw)
            response = build_fake_remote_agent_response(payload)
            logger.info(
                "FAKE_REMOTE_AGENT_REPLY | agent=%s | stage=%s",
                str(response.get("agent") or payload.get("agent") or payload.get("action") or "-"),
                str(response.get("stage") or "-"),
            )
            await ws.send_text(json.dumps(response, ensure_ascii=False))
    except WebSocketDisconnect:
        logger.info("FAKE_REMOTE_AGENT_DISCONNECT")
    except Exception:
        logger.exception("FAKE_REMOTE_AGENT_FAIL")
        try:
            await ws.send_text(json.dumps({"error": "fake_remote_agent_failed"}))
        except Exception:
            pass
        try:
            await close_websocket(ws, code=1011, reason="fake_agent_failed")
        except Exception:
            pass


async def dispatch_agent(lk_url: str, room_name: str, metadata: Dict[str, Any]) -> None:
    lk = lkapi.LiveKitAPI(
        url=lk_url,
        api_key=os.environ["LIVEKIT_API_KEY"],
        api_secret=os.environ["LIVEKIT_API_SECRET"],
    )
    try:
        meta_str = json.dumps(metadata, ensure_ascii=False)
        await lk.agent_dispatch.create_dispatch(
            lkapi.CreateAgentDispatchRequest(
                agent_name=AGENT_NAME,
                room=room_name,
                metadata=meta_str,
            )
        )
    finally:
        await lk.aclose()

async def ws_audio_receiver(
    ws: WebSocket,
    out_q: asyncio.Queue[bytes],
    dbg: Optional[BridgeDebugState] = None,
    *,
    audio_enabled: Optional[asyncio.Event] = None,
    recording_sink: Any = None,
    input_backlog_config: Optional[AudioInputBacklogConfig] = None,
    input_latency_tracker: Optional[AudioInputLatencyTracker] = None,
) -> None:
    """
    Pipe puro:
    - sempre lê do WS (evita backpressure/buffer)
    - só enfileira frames quando audio_enabled estiver setado
    - não faz lógica de "barge-in" aqui
    """
    buf = bytearray()
    frame_idx = 0

    enabled_prev = False
    dropped_bytes = 0
    client_bursts = FlowAudioBurstLogger(
        start_step="audio_in",
        end_step="audio_in_end",
    )

    while True:
        msg = await ws.receive()

        if msg.get("type") == "websocket.disconnect":
            raise WebSocketDisconnect(code=msg.get("code", 1000))

        data = msg.get("bytes")
        if not data:
            continue

        enabled = True if audio_enabled is None else audio_enabled.is_set()

        # enquanto não estiver liberado, consome e descarta
        if not enabled:
            dropped_bytes += len(data)
            buf.clear()
            continue

        # transição: bridge liberou captura do cliente
        if enabled and not enabled_prev:
            enabled_prev = True
            buf.clear()
            frame_idx = 0
            now_enabled = time.monotonic()
            if input_latency_tracker is not None:
                input_latency_tracker.mark_enabled(now_enabled)
                input_latency_tracker.maybe_log_latency(
                    frame_count=0,
                    now=now_enabled,
                    queue_size_frames=out_q.qsize(),
                    reason="receiver_enabled",
                    force=True,
                )
            if dropped_bytes:
                logger.info("CLIENT_AUDIO_ENABLED | dropped_bytes_before_enable=%s", dropped_bytes)

        buf.extend(data)
        while len(buf) >= BYTES_PER_FRAME:
            frame_bytes = bytes(buf[:BYTES_PER_FRAME])
            del buf[:BYTES_PER_FRAME]

            if dbg is not None:
                dbg.in_frames += 1
                if dbg.in_frames == 1:
                    dbg.mark_ws_in()
            frame_count = dbg.in_frames if dbg is not None else frame_idx + 1

            if recording_sink is not None:
                try:
                    recording_sink.record_client_frame(frame_bytes)
                except Exception:
                    pass

            # mantém apenas atividade / medição (não interfere em áudio do agente)
            now = time.monotonic()
            if frame_idx % SILENCE_CHECK_EVERY == 0:
                try:
                    rms = audioop.rms(frame_bytes, 2)
                    dbfs = pcm16_dbfs(frame_bytes) if rms > RMS_TH else -120.0
                    seq = client_bursts.observe(
                        rms=rms,
                        dbfs=dbfs,
                        frame=frame_count,
                        now=now,
                        start_fields=_audio_input_latency_burst_fields(
                            input_latency_tracker,
                            frame_count=frame_count,
                            now=now,
                            queue_size_frames=out_q.qsize(),
                        ),
                    )
                    if seq is not None:
                        if dbg is not None and dbg.timeline is not None:
                            dbg.timeline.emit(
                                "client_audio_activity_detected",
                                seq=seq,
                                rms=rms,
                                dbfs=round(dbfs, 2),
                            )
                    elif rms <= RMS_TH:
                        client_bursts.maybe_end(now=now, reason="silence_gap")
                except Exception:
                    pass
            frame_idx += 1

            try:
                out_q.put_nowait(frame_bytes)
            except asyncio.QueueFull:
                queue_before = out_q.qsize()
                selective = ShedResult()
                if input_backlog_config is not None and input_backlog_config.energy_shed_enabled:
                    selective = shed_queue_backlog(
                        out_q,
                        keep_frames=max(0, queue_before - 1),
                        rms_threshold=input_backlog_config.silence_rms_threshold,
                        blind=False,
                    )
                    if selective.dropped and input_latency_tracker is not None:
                        input_latency_tracker.record_shed(
                            source="receiver_queue_full",
                            dropped_frames=selective.dropped,
                            mode=selective.mode,
                            dropped_silent=selective.dropped_silent,
                            dropped_voiced=selective.dropped_voiced,
                            queue_before_frames=queue_before,
                            queue_after_frames=out_q.qsize(),
                            frame_count=frame_count,
                            now=now,
                        )
                try:
                    out_q.put_nowait(frame_bytes)
                except asyncio.QueueFull:
                    try:
                        _ = out_q.get_nowait()
                    except asyncio.QueueEmpty:
                        pass
                    else:
                        if input_latency_tracker is not None:
                            input_latency_tracker.record_shed(
                                source="receiver_queue_full",
                                dropped_frames=1,
                                mode="queue_full_blind",
                                dropped_voiced=1,
                                queue_before_frames=queue_before,
                                queue_after_frames=max(0, out_q.qsize() - 1),
                                frame_count=frame_count,
                                now=now,
                            )
                    out_q.put_nowait(frame_bytes)

            if input_backlog_config is not None:
                _shed_audio_queue_backlog(
                    out_q,
                    config=input_backlog_config,
                    tracker=input_latency_tracker,
                    source="receiver",
                    frame_count=frame_count,
                    now=now,
                )

            if input_latency_tracker is not None:
                input_latency_tracker.maybe_log_latency(
                    frame_count=frame_count,
                    now=now,
                    queue_size_frames=out_q.qsize(),
                    reason="receiver_audio",
                )


async def connect_publish_livekit(
    room: rtc.Room,
    lk_url: str,
    token: str,
    track: rtc.LocalAudioTrack,
    opts: rtc.TrackPublishOptions,
    track_published: asyncio.Event,
    track_subscribed: asyncio.Event,
    dbg: Optional[BridgeDebugState] = None,
    timeline: Optional[CallTimeline] = None,
) -> None:
    try:
        await room.connect(lk_url, token)
        if timeline is not None:
            timeline.emit("livekit_room_connected")
        publication = await room.local_participant.publish_track(track, opts)

        track_published.set()
        if dbg:
            dbg.tlog("LK connect+publish OK")
        if timeline is not None:
            timeline.emit("livekit_track_published")

        await publication.wait_for_subscription()
        track_subscribed.set()
        if dbg:
            dbg.tlog("LK track subscribed")
        if timeline is not None:
            timeline.emit("livekit_track_subscribed")
    except Exception:
        logger.exception("LIVEKIT_SETUP_FAIL | connect/publish")
        if timeline is not None:
            timeline.emit("livekit_setup_failed")
        raise


async def publish_queue_to_livekit(
    source: rtc.AudioSource,
    in_q: asyncio.Queue[bytes],
    track_subscribed: asyncio.Event,
    dbg: Optional[BridgeDebugState] = None,
    *,
    audio_enabled: Optional[asyncio.Event] = None,
    input_backlog_config: Optional[AudioInputBacklogConfig] = None,
    input_latency_tracker: Optional[AudioInputLatencyTracker] = None,
) -> None:
    if dbg:
        dbg.tlog("BRIDGE waiting track_subscribed")

    await track_subscribed.wait()

    # só começa a publicar áudio do cliente depois da liberação do bridge
    if audio_enabled is not None:
        if dbg:
            dbg.tlog("BRIDGE waiting audio_enabled")
        await audio_enabled.wait()

    now = time.monotonic()
    if input_latency_tracker is not None:
        input_latency_tracker.mark_enabled(now)

    frame_count = dbg.in_frames if dbg is not None else 0
    if input_backlog_config is not None:
        _shed_audio_queue_backlog(
            in_q,
            config=input_backlog_config,
            tracker=input_latency_tracker,
            source="publisher_start",
            frame_count=frame_count,
            now=now,
        )
    if input_latency_tracker is not None:
        input_latency_tracker.maybe_log_latency(
            frame_count=frame_count,
            now=now,
            queue_size_frames=in_q.qsize(),
            reason="publisher_start",
            force=True,
        )

    backlog0 = in_q.qsize()
    publish_frame_idx = 0
    publish_bursts = FlowAudioBurstLogger(
        start_step="audio_to_livekit",
        end_step="audio_to_livekit_end",
    )
    if dbg:
        backlog_frames = in_q.qsize()
        backlog_s = backlog_frames * (FRAME_MS / 1000.0)
        dbg.tlog("BRIDGE start publish", backlog_frames=backlog_frames, backlog_s=round(backlog_s, 3))

    if LK_CATCHUP_ENABLED and backlog0 > 0 and not (
        input_backlog_config is not None and input_backlog_config.shed_enabled
    ):
        keep_frames = max(1, int(LK_CATCHUP_KEEP_MS / FRAME_MS))
        if backlog0 > keep_frames:
            if dbg:
                dbg.tlog("BRIDGE catchup begin", backlog=backlog0, keep_frames=keep_frames, keep_ms=LK_CATCHUP_KEEP_MS)

            drained = 0
            while in_q.qsize() > keep_frames and drained < LK_CATCHUP_MAX_FAST_FRAMES:
                try:
                    frame_bytes = in_q.get_nowait()
                except asyncio.QueueEmpty:
                    break

                if dbg is not None:
                    dbg.out_frames += 1
                    if dbg.out_frames == 1:
                        dbg.mark_lk_out(in_q.qsize())

                frame = rtc.AudioFrame.create(
                    sample_rate=SAMPLE_RATE,
                    num_channels=CHANNELS,
                    samples_per_channel=SAMPLES_PER_FRAME,
                )
                mv = frame.data.cast("B")
                mv[:BYTES_PER_FRAME] = frame_bytes

                await source.capture_frame(frame)
                drained += 1

            if dbg:
                dbg.tlog("BRIDGE catchup end", drained=drained, backlog_now=in_q.qsize())

    next_t = time.monotonic()
    while True:
        now = time.monotonic()
        frame_count = dbg.in_frames if dbg is not None else publish_frame_idx
        if input_backlog_config is not None:
            shed_result = _shed_audio_queue_backlog(
                in_q,
                config=input_backlog_config,
                tracker=input_latency_tracker,
                source="publisher_loop",
                frame_count=frame_count,
                now=now,
            )
            if shed_result.dropped and shed_result.mode == "blind":
                _maybe_clear_livekit_source_queue(
                    source,
                    tracker=input_latency_tracker,
                    reason="publisher_loop_shed",
                    frame_count=frame_count,
                    queue_size_frames=in_q.qsize(),
                    now=now,
                )
                next_t = time.monotonic()

        frame_bytes = await in_q.get()

        if dbg is not None:
            dbg.out_frames += 1
            if dbg.out_frames == 1:
                dbg.mark_lk_out(in_q.qsize())

        now = time.monotonic()
        if publish_frame_idx % SILENCE_CHECK_EVERY == 0:
            try:
                rms = audioop.rms(frame_bytes, 2)
                dbfs = pcm16_dbfs(frame_bytes) if rms > RMS_TH else -120.0
                if rms > RMS_TH:
                    publish_bursts.observe(
                        rms=rms,
                        dbfs=dbfs,
                        frame=dbg.out_frames if dbg is not None else publish_frame_idx,
                        now=now,
                        start_fields={
                            "backlog_frames": in_q.qsize(),
                            "livekit_queued_ms": max(0, round(source.queued_duration * 1000)),
                            **_audio_input_latency_burst_fields(
                                input_latency_tracker,
                                frame_count=dbg.in_frames if dbg is not None else publish_frame_idx + 1,
                                now=now,
                                queue_size_frames=in_q.qsize(),
                            ),
                        },
                    )
                else:
                    publish_bursts.maybe_end(now=now, reason="silence_gap")
            except Exception:
                pass
        publish_frame_idx += 1

        if now - next_t > 0.2:
            next_t = now

        if now < next_t:
            await asyncio.sleep(next_t - now)

        frame = rtc.AudioFrame.create(
            sample_rate=SAMPLE_RATE,
            num_channels=CHANNELS,
            samples_per_channel=SAMPLES_PER_FRAME,
        )
        mv = frame.data.cast("B")
        mv[:BYTES_PER_FRAME] = frame_bytes

        await source.capture_frame(frame)
        next_t += (FRAME_MS / 1000.0)


async def stream_agent_audio_to_queue(
    agent_q: asyncio.Queue[bytes],
    agent_participant: rtc.Participant,
    activity: AudioActivity,
    timeline: Optional[CallTimeline] = None,
    call_context: Optional[Mapping[str, Any]] = None,
):
    flow_context = dict(call_context or {})
    stream = rtc.AudioStream.from_participant(
        participant=agent_participant,
        track_source=rtc.TrackSource.SOURCE_MICROPHONE,
        sample_rate=SAMPLE_RATE,
        num_channels=CHANNELS,
        frame_size_ms=FRAME_MS,
    )

    buf = bytearray()
    frame_idx = 0

    first_frame_sent = False
    last_agent_audio_at = 0.0
    agent_audio_burst_seq = 0
    dropped_queue_frames = 0

    async for ev in stream:
        b = bytes(ev.frame.data)
        buf.extend(b)

        while len(buf) >= BYTES_PER_FRAME:
            frame = bytes(buf[:BYTES_PER_FRAME])
            del buf[:BYTES_PER_FRAME]

            now = time.monotonic()

            # mede RMS só para telemetria/atividade — nunca para descartar frame
            rms = None
            if frame_idx % AGENT_OUT_RMS_EVERY == 0:
                try:
                    rms = audioop.rms(frame, 2)
                except Exception:
                    rms = None
            frame_idx += 1

            # todo frame do agente é encaminhado bit a bit; a energia só marca
            # "fala ativa" (usado por idle/hold e mock-stop), não filtra o áudio.
            if rms is not None and rms > AGENT_OUT_SPEECH_RMS_TH:
                new_burst = _is_new_audio_burst(now, last_agent_audio_at)
                last_agent_audio_at = now
                activity.last_agent_out = now
                if new_burst:
                    agent_audio_burst_seq += 1
                    log_flow_event(
                        logger,
                        "audio_from_agent",
                        **flow_context,
                        seq=agent_audio_burst_seq,
                        participant=str(getattr(agent_participant, "identity", "") or ""),
                        rms=rms if rms is not None else "",
                        frame=frame_idx,
                    )
                    if timeline is not None:
                        timeline.emit(
                            "agent_audio_activity_detected",
                            **flow_context,
                            seq=agent_audio_burst_seq,
                            participant_identity=str(getattr(agent_participant, "identity", "") or ""),
                            rms=rms,
                        )
                if not first_frame_sent:
                    first_frame_sent = True
                    if timeline is not None:
                        timeline.emit(
                            "agent_audio_first_frame_received",
                            **flow_context,
                            participant_identity=str(getattr(agent_participant, "identity", "") or ""),
                        )

            try:
                agent_q.put_nowait(frame)
            except asyncio.QueueFull:
                dropped_queue_frames += 1
                if dropped_queue_frames == 1 or dropped_queue_frames % 50 == 0:
                    log_flow_event(
                        logger,
                        "audio_from_agent_queue_overflow",
                        **flow_context,
                        participant=str(getattr(agent_participant, "identity", "") or ""),
                        dropped_agent_audio_frames=dropped_queue_frames,
                        dropped_voiced_frames=1,
                        queue_size=agent_q.qsize(),
                        content_policy="oldest_frame",
                        sync_action="drop_audio_to_catch_up",
                    )
                    if timeline is not None:
                        timeline.emit(
                            "agent_audio_queue_overflow",
                            **flow_context,
                            participant_identity=str(getattr(agent_participant, "identity", "") or ""),
                            dropped_agent_audio_frames=dropped_queue_frames,
                            dropped_voiced_frames=1,
                            queue_size=agent_q.qsize(),
                            content_policy="oldest_frame",
                            sync_action="drop_audio_to_catch_up",
                        )
                _ = agent_q.get_nowait()
                agent_q.put_nowait(frame)


@app.on_event("startup")
async def _startup():
    _load_runtime_env()


@app.websocket("/ws/agent")
async def ws_agent(ws: WebSocket):
    _load_runtime_env()
    call_started_unix_ms = round(time.time() * 1000)
    await ws.accept()

    lk_url = os.environ["LIVEKIT_URL"]
    call_t0 = time.monotonic()
    capacity_lease = None
    ws_close_code = 1000
    ws_close_reason = "ws_agent_cleanup"
    ws_client_disconnected = False
    bridge_output_stats: Optional[BridgeOutputStats] = None
    bridge_log_context: Dict[str, Any] = {}
    entire_call_recorder = None
    readiness_task: asyncio.Task[Any] | None = None
    try:
        start_ctx = await recv_start_message(ws, logger=logger)
        reservation = await READINESS_GATE.reserve_capacity()
        if not reservation.allowed:
            logger.warning(
                "WS_AGENT_PRE_READY_STOP | reason=capacity_exceeded | active_connections=%s | max_connections=%s",
                reservation.active_connections,
                reservation.max_connections,
            )
            await ws.send_text(json.dumps(build_capacity_stop_message(), ensure_ascii=False))
            try:
                await close_websocket(ws, code=1013, reason="pre_ready:capacity_exceeded")
            except Exception:
                pass
            return
        capacity_lease = reservation.lease

        agent_name = str(start_ctx.data.get("agent") or "").strip().lower() or "conta"
        readiness_task = asyncio.create_task(
            READINESS_GATE.evaluate_resources(agent_name=agent_name, call_config=start_ctx.call_config),
            name="evaluate_call_readiness",
        )

        bootstrap = build_bridge_session_bootstrap(
            start_ctx=start_ctx,
            app_env=APP_ENV,
            livekit_room=os.getenv("LIVEKIT_ROOM", ""),
            default_protocol=os.getenv("DEFAULT_PROTOCOL", ""),
            token_factory=make_token,
            logger=logger,
            timeline_origin_unix_ms=call_started_unix_ms,
        )
        call_id_ged = bootstrap.call_id_ged
        call_config = bootstrap.call_config
        remote_agent_context = bootstrap.remote_agent_context
        room_name = bootstrap.room_name
        identity = bootstrap.identity
        token = bootstrap.token
        protocol = bootstrap.protocol
        session_id = str(bootstrap.dispatch_metadata.get("session_id") or "")
        phone_number = bootstrap.phone_number
        timeline = bootstrap.timeline

        logger.info(
            "CALL_START | room=%s | protocol=%s | session_id=%s | call_id_ged=%s | identity=%s",
            room_name,
            protocol,
            session_id or "-",
            call_id_ged or "-",
            identity,
        )
        log_flow_event(
            logger,
            "call_start",
            room=room_name,
            protocol=protocol,
            session_id=session_id,
            call_id_ged=call_id_ged,
            phone_number=phone_number,
            bridge=identity,
        )
        timeline.emit(
            "call_start",
            call_id_ged=call_id_ged,
            bridge_identity=identity,
            remote_agent=remote_agent_context.get("agent", ""),
            agent_backend=resolve_agent_backend_name(call_config, os.getenv("AGENT_BACKEND", "remote_ws")),
            stt_provider=(
                resolve_stt_overrides(call_config).get("provider")
                or os.getenv("STT_PROVIDER", "internal_http")
            ),
            tts_provider=(
                resolve_tts_overrides(call_config).get("provider")
                or os.getenv("TTS_PROVIDER", "elevenlabs")
            ),
            timeline_file=str(timeline.path),
        )
        timeline.emit("readiness_started", mode="parallel_post_ready")
        timeline.emit("bridge_start_received")


        # READY imediato
        await ws.send_text(
            json.dumps(
                {
                    "type": "ready",
                    "room": room_name,
                    "session_id": bootstrap.dispatch_metadata.get("session_id", ""),
                    "sample_rate": SAMPLE_RATE,
                    "channels": CHANNELS,
                    "frame_ms": FRAME_MS,
                    "bytes_per_frame": BYTES_PER_FRAME,
                    "debug_events_enabled": start_ctx.debug_events_enabled,
                    "stress_test": start_ctx.stress_test,
                }
            )
        )
        timeline.emit("ready_sent")

        now = time.monotonic()
        activity = AudioActivity(last_agent_out=now)

        client_q: asyncio.Queue[bytes] = asyncio.Queue(maxsize=1500)  # ~30s
        agent_q: asyncio.Queue[bytes] = asyncio.Queue(maxsize=600)    # mais folga pra áudio do agente

        dbg = BridgeDebugState(base_t=time.monotonic(), timeline=timeline)
        bridge_output_stats = BridgeOutputStats()
        bridge_log_context = {
            "room": room_name,
            "protocol": protocol,
            "session_id": session_id,
            "call_id_ged": call_id_ged,
            "phone_number": phone_number,
            "bridge": identity,
        }
        lifecycle = RoomLifecycleState()

        def _publish_bridge_debug_event(
            event: str, payload: Mapping[str, Any]
        ) -> None:
            if not start_ctx.debug_events_enabled:
                return

            async def _send() -> None:
                try:
                    await ws.send_text(
                        json.dumps(
                            {
                                "type": "debug_event",
                                "stress_test": True,
                                "event": event,
                                "data": dict(payload),
                            },
                            ensure_ascii=False,
                        )
                    )
                except Exception:
                    logger.debug("bridge debug event forwarding failed", exc_info=True)

            task = asyncio.create_task(
                _send(), name=f"forward_{event.replace('.', '_')}"
            )
            lifecycle.debug_forward_tasks.add(task)
            task.add_done_callback(lifecycle.debug_forward_tasks.discard)

        input_backlog_config = audio_input_backlog_config_from_env(call_config)
        input_latency_tracker = AudioInputLatencyTracker(
            input_backlog_config,
            timeline=timeline,
            call_context=bridge_log_context,
            debug_event_publisher=_publish_bridge_debug_event,
        )
        input_backlog_config_payload = {
            **bridge_log_context,
            "shed_enabled": input_backlog_config.shed_enabled,
            "shed_threshold_ms": input_backlog_config.shed_threshold_ms,
            "shed_keep_ms": input_backlog_config.shed_keep_ms,
            "latency_metrics_enabled": input_backlog_config.latency_metrics_enabled,
            "latency_alert_ms": input_backlog_config.latency_alert_ms,
            "latency_log_interval_s": input_backlog_config.latency_log_interval_s,
            "livekit_source_queue_size_ms": input_backlog_config.livekit_source_queue_size_ms,
            "livekit_source_clear_on_shed": input_backlog_config.livekit_source_clear_on_shed,
            "config_source": input_backlog_config.config_source,
            "call_config_overrides": list(input_backlog_config.call_config_overrides),
        }
        log_flow_event(logger, "audio_in_backlog_config", **input_backlog_config_payload)
        timeline.emit("audio_in_backlog_config", **input_backlog_config_payload)
        entire_call_recorder = create_entire_call_recorder_from_env(
            session_id=session_id,
            sample_rate=SAMPLE_RATE,
            channels=CHANNELS,
            sample_width=2,
            frame_ms=FRAME_MS,
            bytes_per_frame=BYTES_PER_FRAME,
            logger_override=logger,
            timeline=timeline,
            metadata=bridge_log_context,
        )
        first_agent_audio_sent = asyncio.Event()
        mock_stop_after_first_audio_enabled = (
            os.getenv("MOCK_STOP_AFTER_FIRST_AUDIO_ENABLED", "0").strip().lower()
            in {"1", "true", "yes", "on"}
        )
        try:
            mock_stop_after_first_audio_silence_s = max(
                0.0,
                float(os.getenv("MOCK_STOP_AFTER_FIRST_AUDIO_SILENCE_S", "0.35")),
            )
        except ValueError:
            mock_stop_after_first_audio_silence_s = 0.35
        mock_stop_after_first_audio_reason = (
            os.getenv("MOCK_STOP_AFTER_FIRST_AUDIO_REASON", "DONE").strip()
            or "DONE"
        )

        # gate: o bridge só aceita áudio do cliente depois de liberar a sessão
        client_audio_enabled = asyncio.Event()  # começa bloqueado

        # receiver sempre lendo (pra não estourar), mas só enfileira depois do gate
        t_rx = asyncio.create_task(
            ws_audio_receiver(
                ws,
                client_q,
                dbg=dbg,
                audio_enabled=client_audio_enabled,
                recording_sink=entire_call_recorder,
                input_backlog_config=input_backlog_config,
                input_latency_tracker=input_latency_tracker,
            ),
            name="ws_audio_receiver",
        )

        room = rtc.Room()
        agent_ready = asyncio.Event()
        track_published = asyncio.Event()
        track_subscribed = asyncio.Event()

        call_done = asyncio.Event()
        agent_reconnect_enabled = _env_bool("AGENT_RECONNECT_ENABLED", True)
        agent_reconnect_max_attempts = max(0, _env_int("AGENT_RECONNECT_MAX_ATTEMPTS", 1))
        agent_reconnect_timeout_s = max(0.1, _env_float("AGENT_RECONNECT_TIMEOUT_S", 10.0))
        agent_reconnect_poll_s = 0.05
        agent_reconnect_lock = asyncio.Lock()

        async def _dispatch_agent_with_timeline(*, reason: str, attempt: int = 0) -> None:
            fields: Dict[str, Any] = {"agent_name": AGENT_NAME, "reason": reason}
            if attempt:
                fields["attempt"] = attempt
            timeline.emit("dispatch_started", **fields)
            log_flow_event(
                logger,
                "agent_dispatch",
                agent_name=AGENT_NAME,
                room=room_name,
                reason=reason,
                attempt=attempt,
            )
            try:
                await dispatch_agent(lk_url, room_name, bootstrap.dispatch_metadata)
            except Exception:
                timeline.emit("dispatch_failed", **fields)
                raise
            timeline.emit("dispatch_completed", **fields)

        async def _wait_for_agent_reconnect(previous_generation: int) -> bool:
            while not call_done.is_set():
                if (
                    lifecycle.agent_participant is not None
                    and lifecycle.agent_generation > previous_generation
                ):
                    return True
                await asyncio.sleep(agent_reconnect_poll_s)
            return False

        def _stop_for_agent_disconnect(agent_identity: str) -> None:
            if call_done.is_set():
                return
            lifecycle.done_payload = {
                "stage": "DONE",
                "status": "stop_agent_runtime_unavailable",
                "reason": "agent_disconnected",
                "resource": "agent_runtime",
                "failed_resources": ["agent_runtime"],
                "phase": "in_session",
            }
            logger.error(
                "AGENT_RECONNECT_EXHAUSTED | room=%s | protocol=%s | agent_identity=%s",
                room_name,
                protocol,
                agent_identity or "-",
            )
            timeline.emit(
                "agent_reconnect_exhausted",
                agent_identity=agent_identity or "",
                max_attempts=agent_reconnect_max_attempts,
                timeout_ms=round(agent_reconnect_timeout_s * 1000),
            )
            call_done.set()

        async def _recover_agent_disconnect(agent_identity: str, generation: int) -> None:
            if call_done.is_set():
                return
            async with agent_reconnect_lock:
                if call_done.is_set():
                    return
                if lifecycle.agent_participant is not None and lifecycle.agent_generation > generation:
                    return
                if not agent_reconnect_enabled or agent_reconnect_max_attempts <= 0:
                    _stop_for_agent_disconnect(agent_identity)
                    return

                for attempt in range(1, agent_reconnect_max_attempts + 1):
                    if call_done.is_set():
                        return
                    logger.warning(
                        "AGENT_RECONNECT_ATTEMPT | room=%s | protocol=%s | agent_identity=%s | attempt=%s/%s | timeout_s=%.3f",
                        room_name,
                        protocol,
                        agent_identity or "-",
                        attempt,
                        agent_reconnect_max_attempts,
                        agent_reconnect_timeout_s,
                    )
                    timeline.emit(
                        "agent_reconnect_attempt",
                        agent_identity=agent_identity or "",
                        attempt=attempt,
                        max_attempts=agent_reconnect_max_attempts,
                        timeout_ms=round(agent_reconnect_timeout_s * 1000),
                    )
                    try:
                        await _dispatch_agent_with_timeline(reason="agent_reconnect", attempt=attempt)
                    except Exception:
                        logger.exception(
                            "AGENT_RECONNECT_DISPATCH_FAIL | room=%s | protocol=%s | attempt=%s/%s",
                            room_name,
                            protocol,
                            attempt,
                            agent_reconnect_max_attempts,
                        )
                        timeline.emit(
                            "agent_reconnect_dispatch_failed",
                            agent_identity=agent_identity or "",
                            attempt=attempt,
                            max_attempts=agent_reconnect_max_attempts,
                        )

                    try:
                        reconnected = await asyncio.wait_for(
                            _wait_for_agent_reconnect(generation),
                            timeout=agent_reconnect_timeout_s,
                        )
                    except asyncio.TimeoutError:
                        reconnected = False

                    if call_done.is_set():
                        return

                    if reconnected:
                        new_identity = str(getattr(lifecycle.agent_participant, "identity", "") or "")
                        logger.info(
                            "AGENT_RECONNECT_OK | room=%s | protocol=%s | old_agent_identity=%s | new_agent_identity=%s | attempt=%s",
                            room_name,
                            protocol,
                            agent_identity or "-",
                            new_identity or "-",
                            attempt,
                        )
                        timeline.emit(
                            "agent_reconnect_completed",
                            old_agent_identity=agent_identity or "",
                            new_agent_identity=new_identity,
                            attempt=attempt,
                        )
                        return

                    timeline.emit(
                        "agent_reconnect_timeout",
                        agent_identity=agent_identity or "",
                        attempt=attempt,
                        max_attempts=agent_reconnect_max_attempts,
                    )

                _stop_for_agent_disconnect(agent_identity)

        t_dispatch0 = time.monotonic()
        register_room_lifecycle_handlers(
            room=room,
            logger=logger,
            timeline=timeline,
            room_name=room_name,
            protocol=protocol,
            dispatch_started_at=t_dispatch0,
            agent_ready=agent_ready,
            call_done=call_done,
            state=lifecycle,
            is_agent=is_agent,
            on_agent_disconnect=_recover_agent_disconnect,
            ws=ws,
            debug_events_enabled=start_ctx.debug_events_enabled,
        )
        t_done = asyncio.create_task(
            watch_call_done(
                call_done=call_done,
                state=lifecycle,
                ws=ws,
                logger=logger,
                room_name=room_name,
                protocol=protocol,
                timeline=timeline,
            ),
            name="watch_call_done",
        )

        async def _watch_parallel_readiness() -> None:
            try:
                readiness = await readiness_task
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                logger.exception("WS_AGENT_READINESS_FAIL | agent=%s", agent_name)
                timeline.emit("readiness_failed", error=type(exc).__name__)
                if not call_done.is_set():
                    lifecycle.done_payload = build_bridge_failed_stop_message()["data"]
                    call_done.set()
                return

            if readiness.is_healthy():
                timeline.emit(
                    "readiness_ok",
                    cached=readiness.cached,
                    active_connections=readiness.active_connections,
                    max_connections=readiness.max_connections,
                )
                return

            logger.warning(
                "WS_AGENT_READINESS_FAIL | agent=%s | failed_resources=%s | report=%s",
                agent_name, ",".join(readiness.failed_resources), json.dumps(readiness.as_dict(), ensure_ascii=False),
            )
            timeline.emit("readiness_failed", failed_resources=readiness.failed_resources)
            if not call_done.is_set():
                stop_message = readiness.build_stop_message()
                stop_message["data"]["phase"] = "in_session"
                lifecycle.done_payload = stop_message["data"]
                call_done.set()

        t_readiness = asyncio.create_task(_watch_parallel_readiness(), name="watch_call_readiness")

        try:
            source = rtc.AudioSource(
                SAMPLE_RATE,
                CHANNELS,
                queue_size_ms=input_backlog_config.livekit_source_queue_size_ms,
            )
        except TypeError:
            source = rtc.AudioSource(SAMPLE_RATE, CHANNELS)
        livekit_source_config_payload = {
            **bridge_log_context,
            "queue_size_ms": input_backlog_config.livekit_source_queue_size_ms,
            "clear_on_shed": input_backlog_config.livekit_source_clear_on_shed,
        }
        log_flow_event(logger, "livekit_audio_source_config", **livekit_source_config_payload)
        timeline.emit("livekit_audio_source_config", **livekit_source_config_payload)

        track = rtc.LocalAudioTrack.create_audio_track("ws-mic", source)
        opts = rtc.TrackPublishOptions()
        opts.source = rtc.TrackSource.SOURCE_MICROPHONE

        # dispatch do agent
        t_dispatch = asyncio.create_task(
            _dispatch_agent_with_timeline(reason="initial"),
            name="dispatch_agent",
        )

        # conecta + publica track + espera subscribe
        t_lk = asyncio.create_task(
            connect_publish_livekit(
                room=room,
                lk_url=lk_url,
                token=token,
                track=track,
                opts=opts,
                track_published=track_published,
                track_subscribed=track_subscribed,
                dbg=dbg,
                timeline=timeline,
            ),
            name="connect_publish_livekit",
        )

        # publish do áudio do cliente -> LK (só depois do gate)
        t_in = asyncio.create_task(
            publish_queue_to_livekit(
                source,
                client_q,
                track_subscribed,
                dbg=dbg,
                audio_enabled=client_audio_enabled,
                input_backlog_config=input_backlog_config,
                input_latency_tracker=input_latency_tracker,
            ),
            name="publish_queue_to_livekit",
        )

        input_latency_tracker.mark_enabled(time.monotonic())
        await enable_client_audio(
            logger=logger,
            timeline=timeline,
            client_audio_enabled=client_audio_enabled,
            room_name=room_name,
            protocol=protocol,
        )
        t_control = asyncio.create_task(
            notify_client_audio_enabled(
                room=room,
                logger=logger,
                timeline=timeline,
                room_name=room_name,
                protocol=protocol,
                track_published=track_published,
                agent_ready=agent_ready,
                lifecycle=lifecycle,
                repeat=True,
            ),
            name="notify_client_audio_enabled",
        )

        ws_overrides = resolve_ws_overrides(call_config)

        # saída: agente -> cliente (via ws_out_loop)
        t_out_loop = asyncio.create_task(
            ws_out_loop(
                ws,
                agent_q,
                frame_ms=FRAME_MS,
                bytes_per_frame=BYTES_PER_FRAME,
                first_agent_audio_sent=first_agent_audio_sent,
                flow_logger=logger,
                timeline=timeline,
                call_context=bridge_log_context,
                output_stats=bridge_output_stats,
                output_gain=ws_overrides.get("output_gain"),
                recording_sink=entire_call_recorder,
                debug_event_publisher=_publish_bridge_debug_event,
            ),
            name="ws_out_loop",
        )

        t_agent_to_q = asyncio.create_task(
            stream_agent_audio_when_ready(
                agent_ready=agent_ready,
                lifecycle=lifecycle,
                agent_q=agent_q,
                activity=activity,
                timeline=timeline,
                stream_agent_audio=(
                    lambda agent_q, agent_participant, activity, timeline=None: stream_agent_audio_to_queue(
                        agent_q,
                        agent_participant,
                        activity,
                        timeline=timeline,
                        call_context=bridge_log_context,
                    )
                ),
                repeat=True,
            ),
            name="stream_agent_audio_when_ready",
        )
        t_mock_stop = None
        if mock_stop_after_first_audio_enabled:
            logger.info(
                "MOCK_STOP_AFTER_FIRST_AUDIO_ENABLED | room=%s | protocol=%s | silence_s=%.3f | reason=%s",
                room_name,
                protocol,
                mock_stop_after_first_audio_silence_s,
                mock_stop_after_first_audio_reason,
            )
            timeline.emit(
                "mock_stop_after_first_audio_enabled",
                silence_ms=round(mock_stop_after_first_audio_silence_s * 1000),
                reason=mock_stop_after_first_audio_reason,
            )
            t_mock_stop = asyncio.create_task(
                watch_mock_stop_after_first_audio(
                    ws=ws,
                    logger=logger,
                    timeline=timeline,
                    room_name=room_name,
                    protocol=protocol,
                    first_agent_audio_sent=first_agent_audio_sent,
                    activity=activity,
                    agent_q=agent_q,
                    silence_s=mock_stop_after_first_audio_silence_s,
                    reason=mock_stop_after_first_audio_reason,
                ),
                name="mock_stop_after_first_audio",
            )

        try:
            tasks = {t_rx, t_out_loop, t_in, t_agent_to_q, t_lk, t_dispatch, t_done, t_readiness, t_control}
            if t_mock_stop is not None:
                tasks.add(t_mock_stop)
            pending = set(tasks)
            while pending:
                done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
                if t_done in done:
                    break
                for d in done:
                    exc = d.exception()
                    if exc:
                        logger.error(
                            "WS_AGENT_TASK_FAIL | room=%s | protocol=%s | task=%s",
                            room_name,
                            protocol,
                            d.get_name(),
                            exc_info=exc,
                        )
                        raise exc

        except WebSocketDisconnect as exc:
            ws_client_disconnected = True
            logger.info(
                "WS_DISCONNECT | room=%s | protocol=%s | code=%s",
                room_name,
                protocol,
                getattr(exc, "code", ""),
            )
            timeline.emit("ws_disconnect")
        except Exception as exc:
            ws_close_code = 1011
            ws_close_reason = "bridge_failed"
            logger.exception("WS_AGENT_FAIL | room=%s | protocol=%s", room_name, protocol)
            timeline.emit("bridge_failed")
            try:
                await ws.send_text(json.dumps(build_bridge_failed_stop_message(), ensure_ascii=False))
            except Exception:
                pass
        finally:
            for t in (t_rx, t_out_loop, t_in, t_agent_to_q, t_lk, t_dispatch, t_done, t_readiness, t_control, t_mock_stop):
                try:
                    t.cancel()
                except Exception:
                    pass

            try:
                await room.disconnect()
            except Exception:
                pass

            if not ws_client_disconnected:
                try:
                    await close_websocket(ws, code=ws_close_code, reason=ws_close_reason)
                except Exception:
                    pass

            if entire_call_recorder is not None:
                try:
                    await entire_call_recorder.finalize()
                except Exception:
                    logger.exception(
                        "ENTIRE_CALL_RECORDING_FINALIZE_FAIL | room=%s | protocol=%s | session_id=%s",
                        room_name,
                        protocol,
                        session_id or "-",
                    )

            dur = time.monotonic() - call_t0
            audio_to_tia_summary: Dict[str, Any] = {}
            if bridge_output_stats is not None:
                audio_to_tia_summary = {
                    "destination": "ws_client",
                    "ws_close_reason": ws_close_reason,
                    "ws_client_disconnected": ws_client_disconnected,
                    "duration_ms": round(dur * 1000),
                    "total_ws_frames": bridge_output_stats.total_ws_frames,
                    "agent_audio_frames": bridge_output_stats.agent_audio_frames,
                    "silence_frames": bridge_output_stats.silence_frames,
                    "agent_audio_bursts": bridge_output_stats.agent_audio_bursts,
                    "first_agent_audio_sent": bridge_output_stats.first_agent_audio_sent,
                    "queued_agent_frames": agent_q.qsize(),
                }
                log_flow_event(
                    logger,
                    "audio_to_tia_summary",
                    **bridge_log_context,
                    **audio_to_tia_summary,
                )
                timeline.emit(
                    "agent_audio_to_tia_summary",
                    **bridge_log_context,
                    **audio_to_tia_summary,
                )
            logger.info(
                "CALL_END | room=%s | protocol=%s | session_id=%s | duration_s=%.1f",
                room_name,
                protocol,
                session_id or "-",
                dur,
            )
            timeline.emit(
                "call_end",
                duration_ms=round(dur * 1000),
                client_in_frames=dbg.in_frames,
                bridge_out_frames=dbg.out_frames,
                audio_to_tia_total_ws_frames=audio_to_tia_summary.get("total_ws_frames", 0),
                audio_to_tia_agent_audio_frames=audio_to_tia_summary.get("agent_audio_frames", 0),
                audio_to_tia_agent_audio_bursts=audio_to_tia_summary.get("agent_audio_bursts", 0),
                audio_to_tia_first_agent_audio_sent=audio_to_tia_summary.get("first_agent_audio_sent", False),
            )
    finally:
        if capacity_lease is not None:
            await capacity_lease.release()
        if readiness_task is not None and not readiness_task.done():
            readiness_task.cancel()


@app.websocket("/ws/text")
async def ws_text(ws: WebSocket):
    from app.services.text_pipeline import TextSessionPipeline

    await ws.accept()
    session_id = str(uuid.uuid4())
    t0 = time.monotonic()
    logger.info("TEXT_START | session_id=%s", session_id)
    try:
        start_ctx = await recv_start_message(ws, logger=logger)
        pipeline = TextSessionPipeline(
            ws,
            start_ctx.payload,
            session_id,
            session_data=start_ctx.session_data,
            intro=start_ctx.intro,
        )
        await pipeline.run()
    except WebSocketDisconnect:
        logger.info("TEXT_END | session_id=%s | reason=disconnect | duration_s=%.1f", session_id, time.monotonic() - t0)
    except Exception as e:
        logger.exception("TEXT_FAIL | session_id=%s", session_id)
        try:
            await ws.send_text(json.dumps({"type": "error", "message": str(e)}))
        except Exception:
            pass
        try:
            await close_websocket(ws, code=1011, reason="text_pipeline_failed")
        except Exception:
            pass


@app.websocket("/ws/text_stream")
async def ws_text_stream(ws: WebSocket):
    from app.services.text_pipeline_stream import TextSessionPipelineStream

    await ws.accept()
    session_id = str(uuid.uuid4())
    t0 = time.monotonic()
    logger.info("TEXT_STREAM_START | session_id=%s", session_id)
    try:
        start_ctx = await recv_start_message(ws, logger=logger)
        pipeline = TextSessionPipelineStream(
            ws,
            start_ctx.payload,
            session_id,
            session_data=start_ctx.session_data,
            intro=start_ctx.intro,
        )
        await pipeline.run()
    except WebSocketDisconnect:
        logger.info(
            "TEXT_STREAM_END | session_id=%s | reason=disconnect | duration_s=%.1f",
            session_id,
            time.monotonic() - t0,
        )
    except Exception as e:
        logger.exception("TEXT_STREAM_FAIL | session_id=%s", session_id)
        try:
            await ws.send_text(json.dumps({"type": "error", "message": str(e)}))
        except Exception:
            pass
        try:
            await close_websocket(ws, code=1011, reason="text_stream_pipeline_failed")
        except Exception:
            pass

@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/health/resources")
async def health_resources():
    report = await READINESS_GATE.evaluate_default_resources()
    status_code = 200 if report.is_healthy() else 503
    return JSONResponse(status_code=status_code, content=report.as_dict())


@app.get("/health/services")
async def health_services():
    report = await READINESS_GATE.evaluate_services()
    status_code = 200 if report.is_healthy() else 503
    return JSONResponse(status_code=status_code, content=report.as_dict())

if __name__ == "__main__":
    import uvicorn
    import argparse

    parser = argparse.ArgumentParser(description="LiveKit WebSocket Gateway (PCM16 16k <-> LiveKit)")
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument("--host", type=str, default="0.0.0.0")
    parser.add_argument("--log-level", type=str, default="info")
    parser.add_argument("--reload", action="store_true")
    args = parser.parse_args()

    uvicorn.run(
        app,
        host=args.host,
        port=args.port,
        ws_ping_interval=float(os.getenv("UVICORN_WS_PING_INTERVAL_S", "20")),
        ws_ping_timeout=float(os.getenv("UVICORN_WS_PING_TIMEOUT_S", "20")),
        ws_max_size=50 * 1024 * 1024,
        log_level="info" if args.log_level is None else args.log_level,
        reload=args.reload,
    )
