from __future__ import annotations

import asyncio
import json
import logging
import os
import threading
import time
from contextlib import asynccontextmanager
from typing import Any

import httpx
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.exc import OperationalError

from services.ai_voice_runtime_service.audiosocket import normalize_media_uuid
from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaRegistration
from services.ai_voice_runtime_service.providers.asr import build_asr_provider
from services.ai_voice_runtime_service.providers.tts import build_tts_provider
from services.shared.core import Role, new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.models import (
    HealthResponse,
    VoiceAIHandoffRequestIn,
    VoiceAIMediaBridgeEventIn,
    VoiceAISessionCreateIn,
    VoiceAISessionOut,
    VoiceAIStartIn,
    VoiceAITelephonyEventIn,
    VoiceAITurnDecisionOut,
    VoiceAITurnIn,
)
from services.shared.security import issue_app_token, require_roles
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import VoiceAISessionRow, VoiceTranscriptSegmentRow
from services.shared.voice_transcripts import add_transcript_segment, next_transcript_sequence

init_sql_schema()


LOGGER = logging.getLogger("uvicorn.error")


def _bool_env(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 _int_env(name: str, default: int) -> int:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        return int(raw.strip())
    except ValueError:
        return default


def _float_env(name: str, default: float) -> float:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        return float(raw.strip())
    except ValueError:
        return default


def _ai_orchestrator_service_url() -> str:
    return os.getenv("AI_ORCHESTRATOR_SERVICE_URL", "http://localhost:8017").rstrip("/")


def _asterisk_bridge_service_url() -> str:
    return os.getenv("ASTERISK_BRIDGE_SERVICE_URL", "http://localhost:8016").rstrip("/")


def _interaction_service_url() -> str:
    return os.getenv("INTERACTION_SERVICE_URL", "http://localhost:8004").rstrip("/")


def _asr_provider_name() -> str:
    return os.getenv("AI_VOICE_ASR_PROVIDER", "openai").strip() or "openai"


def _tts_provider_name() -> str:
    return os.getenv("AI_VOICE_TTS_PROVIDER", "openai").strip() or "openai"


def _handoff_timeout_seconds() -> float:
    return max(3.0, _float_env("AI_VOICE_HANDOFF_TIMEOUT_SECONDS", 8.0))


def _audiosocket_enabled() -> bool:
    return _bool_env("AI_VOICE_AUDIOSOCKET_ENABLED", False)


def _audiosocket_host() -> str:
    return os.getenv("AI_VOICE_AUDIOSOCKET_HOST", "0.0.0.0").strip() or "0.0.0.0"


def _audiosocket_port() -> int:
    return max(_int_env("AI_VOICE_AUDIOSOCKET_PORT", 9019), 1)


def _vad_frame_ms() -> int:
    return 20


def _vad_min_speech_ms() -> int:
    return max(_int_env("AI_VOICE_VAD_MIN_SPEECH_MS", 300), _vad_frame_ms())


def _vad_trailing_silence_ms() -> int:
    return max(_int_env("AI_VOICE_VAD_TRAILING_SILENCE_MS", 700), _vad_frame_ms())


def _turn_max_ms() -> int:
    return max(_int_env("AI_VOICE_TURN_MAX_MS", 10000), _vad_frame_ms())


def _media_idle_timeout_seconds() -> float:
    return max(_float_env("AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS", 15.0), 5.0)


def _media_registration_wait_timeout_seconds() -> float:
    # `call.started` still performs cross-service writes before the AudioSocket
    # registration event can be processed, so keep the media socket open long
    # enough for the registration to arrive.
    return max(_float_env("AI_VOICE_MEDIA_REGISTRATION_WAIT_TIMEOUT_SECONDS", 12.0), 0.0)


def _service_headers() -> dict[str, str]:
    token = issue_app_token(
        subject="svc:ai-voice-runtime",
        username="ai-voice-runtime",
        role="admin",
        auth_source="service",
        provider="ai-voice-runtime",
        ttl_seconds=300,
    )
    return {"Authorization": f"Bearer {token}"}


def _is_sqlite_locked(exc: Exception) -> bool:
    return "database is locked" in str(exc).lower()


def _retry_db_write(fn, *, attempts: int = 6, base_delay_seconds: float = 0.05):
    last_exc: Exception | None = None
    for attempt in range(attempts):
        try:
            return fn()
        except OperationalError as exc:
            last_exc = exc
            if not _is_sqlite_locked(exc) or attempt >= attempts - 1:
                raise
            time.sleep(base_delay_seconds * (attempt + 1))
    if last_exc is not None:
        raise last_exc
    raise RuntimeError("DB write retry exhausted without exception")


def _request(method: str, url: str, *, payload: dict[str, Any] | None = None, timeout: float = 10.0) -> dict[str, Any]:
    with httpx.Client(timeout=timeout) as client:
        response = client.request(
            method,
            url,
            json=payload,
            headers=_service_headers(),
        )
    response.raise_for_status()
    return response.json()


def _orchestrator_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 10.0) -> dict[str, Any]:
    return _request(method, f"{_ai_orchestrator_service_url()}{path}", payload=payload, timeout=timeout)


def _bridge_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 10.0) -> dict[str, Any]:
    return _request(method, f"{_asterisk_bridge_service_url()}{path}", payload=payload, timeout=timeout)


def _interaction_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 5.0) -> dict[str, Any]:
    return _request(method, f"{_interaction_service_url()}{path}", payload=payload, timeout=timeout)


def _resolved_voice_language(language_hint: str | None, fallback: str | None = None) -> str:
    return str(language_hint or fallback or "ru").strip() or "ru"


def _default_voice_greeting(language: str | None) -> str:
    if str(language or "").strip() == "kz":
        return (
            "Мен компанияның AI оператормын. Сәлеметсіз бе. "
            "Қандай сұрағыңыз бар екенін айтыңыз, мен бірден көмектесуге тырысамын."
        )
    return (
        "Я AI-оператор компании. Здравствуйте. "
        "Коротко расскажите, с чем помочь, и я сразу начну разбираться."
    )


_ASR_PROVIDER = build_asr_provider(_asr_provider_name())
_TTS_PROVIDER = build_tts_provider(_tts_provider_name())


def _load_voice_session(session, session_id: str) -> VoiceAISessionRow:
    row = session.execute(
        select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == session_id)
    ).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail="Voice AI session not found")
    return row


def _next_sequence(session, session_id: str) -> int:
    return next_transcript_sequence(session, session_id)


def _record_segment(
    session,
    *,
    voice_session: VoiceAISessionRow,
    speaker: str,
    source_type: str,
    text: str,
    sequence_no: int,
    confidence: float | None = None,
    payload: dict[str, Any] | None = None,
    is_final: bool = True,
) -> None:
    add_transcript_segment(
        session,
        session_id=voice_session.session_id,
        call_id=voice_session.call_id,
        interaction_id=voice_session.interaction_id,
        speaker=speaker,
        source_type=source_type,
        text=text,
        confidence=confidence,
        payload=payload,
        is_final=is_final,
        sequence_no=sequence_no,
    )


def _load_greeting_segment(session, session_id: str) -> VoiceTranscriptSegmentRow | None:
    rows = session.execute(
        select(VoiceTranscriptSegmentRow)
        .where(VoiceTranscriptSegmentRow.session_id == session_id)
        .where(VoiceTranscriptSegmentRow.speaker == "assistant")
        .order_by(VoiceTranscriptSegmentRow.id.asc())
    ).scalars().all()
    for row in rows:
        try:
            payload = json.loads(row.payload_json or "{}")
        except Exception:
            payload = {}
        if str(payload.get("kind") or "").strip() == "greeting":
            return row
    return None


def _ensure_greeting_segment(session, voice_session: VoiceAISessionRow, greeting_text: str) -> None:
    if not str(greeting_text or "").strip():
        return
    if _load_greeting_segment(session, voice_session.session_id) is not None:
        return
    _record_segment(
        session,
        voice_session=voice_session,
        speaker="assistant",
        source_type="tts",
        text=greeting_text,
        sequence_no=_next_sequence(session, voice_session.session_id),
        payload={"kind": "greeting", "delivery_status": "planned"},
        is_final=False,
    )


def _mark_last_assistant_segment_interrupted(session, session_id: str) -> None:
    row = session.execute(
        select(VoiceTranscriptSegmentRow)
        .where(VoiceTranscriptSegmentRow.session_id == session_id)
        .where(VoiceTranscriptSegmentRow.speaker == "assistant")
        .where(VoiceTranscriptSegmentRow.is_final.is_(False))
        .order_by(VoiceTranscriptSegmentRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()
    if row is None:
        row = session.execute(
            select(VoiceTranscriptSegmentRow)
            .where(VoiceTranscriptSegmentRow.session_id == session_id)
            .where(VoiceTranscriptSegmentRow.speaker == "assistant")
            .order_by(VoiceTranscriptSegmentRow.id.desc())
            .limit(1)
        ).scalar_one_or_none()
    if row is not None:
        row.barge_in_interrupted = True
        try:
            payload = json.loads(row.payload_json or "{}")
        except Exception:
            payload = {}
        payload["delivery_status"] = "interrupted"
        row.payload_json = json.dumps(payload, ensure_ascii=False)


def _append_interaction_timeline(interaction_id: str | None, action: str, metadata: dict[str, Any] | None = None) -> None:
    if not str(interaction_id or "").strip():
        return
    _interaction_request(
        "POST",
        f"/interactions/{interaction_id}/timeline",
        payload={"action": action, "metadata": metadata or {}},
        timeout=3.0,
    )


def _push_bridge_call_state(
    voice_session: VoiceAISessionRow,
    *,
    ai_state: str,
    handoff_reason: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    _bridge_request(
        "POST",
        f"/internal/voice-ai/calls/{voice_session.call_id}/state",
        payload={
            "voice_session_id": voice_session.session_id,
            "ai_session_id": voice_session.ai_session_id,
            "ai_state": ai_state,
            "handoff_reason": handoff_reason,
            "metadata": metadata or {},
        },
        timeout=3.0,
    )


def _media_registration_from_row(row: VoiceAISessionRow) -> MediaRegistration:
    return MediaRegistration(
        voice_session_id=row.session_id,
        call_id=row.call_id,
        interaction_id=str(row.interaction_id or "").strip(),
        ai_session_id=str(row.ai_session_id or "").strip() or None,
        language=str(row.language or "").strip() or None,
        media_uuid=str(row.media_uuid or "").strip() or None,
    )


def _load_media_registration_by_uuid(media_uuid: str) -> MediaRegistration | None:
    session = get_session()
    try:
        normalized = normalize_media_uuid(media_uuid)
        row = session.execute(
            select(VoiceAISessionRow)
            .where(VoiceAISessionRow.media_uuid == normalized)
            .order_by(VoiceAISessionRow.id.desc())
            .limit(1)
        ).scalar_one_or_none()
        if row is None:
            return None
        return _media_registration_from_row(row)
    finally:
        session.close()


def _persist_media_bridge_event(session_id: str, payload: VoiceAIMediaBridgeEventIn) -> VoiceAISessionRow:
    def _write() -> VoiceAISessionRow:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            if str(voice_session.call_id or "").strip() != str(payload.call_id or "").strip():
                raise HTTPException(status_code=409, detail="call_id mismatch")
            media_uuid = normalize_media_uuid(payload.media_uuid)
            now = utc_now_iso()
            if payload.event_type == "requested":
                voice_session.media_uuid = media_uuid
                voice_session.media_status = "requested"
                voice_session.media_connected_at = None
                voice_session.media_ended_at = None
            else:
                voice_session.media_uuid = voice_session.media_uuid or media_uuid
                voice_session.media_status = "ended"
                voice_session.media_ended_at = now
            voice_session.updated_at = now
            session.commit()
            session.refresh(voice_session)
            return voice_session
        finally:
            session.close()

    return _retry_db_write(_write)


def _mark_media_connected(session_id: str, media_uuid: str) -> None:
    def _write() -> None:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            now = utc_now_iso()
            voice_session.media_uuid = normalize_media_uuid(media_uuid)
            voice_session.media_status = "connected"
            voice_session.media_connected_at = now
            voice_session.updated_at = now
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _mark_media_ended(session_id: str, reason: str) -> None:
    def _write() -> None:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            now = utc_now_iso()
            voice_session.media_status = "ended"
            voice_session.media_ended_at = now
            voice_session.updated_at = now
            if voice_session.status in {"queued", "greeting", "listening", "thinking", "speaking", "active"}:
                voice_session.status = "completed"
                voice_session.ended_at = voice_session.ended_at or now
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _touch_media_frame(session_id: str) -> None:
    def _write() -> None:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            now = utc_now_iso()
            voice_session.last_media_frame_at = now
            voice_session.updated_at = now
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _load_pending_greeting_text(session_id: str) -> str | None:
    session = get_session()
    try:
        voice_session = _load_voice_session(session, session_id)
        if voice_session.disclosure_played_at:
            return None
        rows = session.execute(
            select(VoiceTranscriptSegmentRow)
            .where(VoiceTranscriptSegmentRow.session_id == session_id)
            .where(VoiceTranscriptSegmentRow.speaker == "assistant")
            .order_by(VoiceTranscriptSegmentRow.id.desc())
        ).scalars().all()
        for row in rows:
            try:
                payload = json.loads(row.payload_json or "{}")
            except Exception:
                payload = {}
            if str(payload.get("kind") or "").strip() == "greeting":
                return str(row.text or "").strip() or None
        return None
    finally:
        session.close()


def _mark_reply_delivered(session_id: str, text: str, is_greeting: bool) -> None:
    def _write() -> None:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            now = utc_now_iso()
            pending_rows = session.execute(
                select(VoiceTranscriptSegmentRow)
                .where(VoiceTranscriptSegmentRow.session_id == session_id)
                .where(VoiceTranscriptSegmentRow.speaker == "assistant")
                .where(VoiceTranscriptSegmentRow.is_final.is_(False))
                .order_by(VoiceTranscriptSegmentRow.id.desc())
            ).scalars().all()
            matched_row = None
            normalized_text = str(text or "").strip()
            for row in pending_rows:
                try:
                    payload = json.loads(row.payload_json or "{}")
                except Exception:
                    payload = {}
                row_kind = str(payload.get("kind") or "").strip()
                if is_greeting:
                    if row_kind == "greeting":
                        matched_row = row
                        payload["delivery_status"] = "delivered"
                        row.payload_json = json.dumps(payload, ensure_ascii=False)
                        break
                    continue
                if str(row.text or "").strip() != normalized_text:
                    continue
                matched_row = row
                payload["delivery_status"] = "delivered"
                row.payload_json = json.dumps(payload, ensure_ascii=False)
                break
            if matched_row is not None:
                matched_row.is_final = True
            if is_greeting and not voice_session.disclosure_played_at:
                voice_session.disclosure_played_at = now
            voice_session.last_ai_reply_at = now
            voice_session.updated_at = now
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _set_voice_session_state(
    session_id: str,
    ai_state: str,
    handoff_reason: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    def _write() -> VoiceAISessionRow:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            now = utc_now_iso()
            voice_session.status = ai_state
            if handoff_reason is not None:
                voice_session.handoff_reason = str(handoff_reason).strip() or None
            voice_session.updated_at = now
            if ai_state in {"completed", "closed"}:
                voice_session.ended_at = voice_session.ended_at or now
            session.commit()
            session.refresh(voice_session)
            return voice_session
        finally:
            session.close()

    voice_session = _retry_db_write(_write)
    _push_bridge_call_state(
        voice_session,
        ai_state=ai_state,
        handoff_reason=handoff_reason,
        metadata=metadata,
    )
    if ai_state == "error":
        _append_interaction_timeline(
            voice_session.interaction_id,
            "ai.error",
            {
                "call_id": voice_session.call_id,
                "voice_session_id": voice_session.session_id,
                "ai_session_id": voice_session.ai_session_id,
                "error": str(handoff_reason or "").strip()[:500],
                **(metadata or {}),
            },
        )


def _request_runtime_handoff(
    session_id: str,
    customer_request_text: str,
    decision: VoiceAITurnDecisionOut,
) -> None:
    session = get_session()
    try:
        voice_session = _load_voice_session(session, session_id)
        if not str(voice_session.interaction_id or "").strip():
            raise RuntimeError("Voice AI session is missing interaction_id")
        handoff_payload = VoiceAIHandoffRequestIn(
            voice_session_id=voice_session.session_id,
            ai_session_id=voice_session.ai_session_id,
            interaction_id=voice_session.interaction_id,
            target_queue_id=voice_session.handoff_target_queue_id,
            reason=decision.handoff_reason or "AI requested human handoff.",
            summary={
                "customer_request_text": customer_request_text,
                "ai_outcome_text": decision.summary_text or decision.reply_text or decision.handoff_reason or "",
                "recommended_next_step": "Продолжить звонок вручную и проверить контекст обращения.",
            },
        )
        LOGGER.warning(
            "voice_runtime.handoff_request call_id=%s session_id=%s ai_session_id=%s interaction_id=%s target_queue_id=%s reason=%s",
            voice_session.call_id,
            voice_session.session_id,
            voice_session.ai_session_id,
            voice_session.interaction_id,
            voice_session.handoff_target_queue_id,
            (decision.handoff_reason or "AI requested human handoff.")[:300],
        )
        try:
            _bridge_request(
                "POST",
                f"/internal/voice-ai/calls/{voice_session.call_id}/handoff",
                payload=handoff_payload.model_dump(),
                timeout=_handoff_timeout_seconds(),
            )
        except Exception as exc:
            LOGGER.warning(
                "voice_runtime.handoff_failed call_id=%s session_id=%s ai_session_id=%s error=%s",
                voice_session.call_id,
                voice_session.session_id,
                voice_session.ai_session_id,
                str(exc)[:500],
            )
            raise
    finally:
        session.close()


def _handle_media_error(session_id: str, message: str, metadata: dict[str, Any] | None = None) -> None:
    session = get_session()
    try:
        voice_session = _load_voice_session(session, session_id)
        previous_status = str(voice_session.status or "").strip()
        now = utc_now_iso()
        voice_session.status = "error"
        voice_session.handoff_reason = str(message or "media_error")[:1000]
        voice_session.media_status = "error"
        voice_session.media_ended_at = now
        voice_session.updated_at = now
        session.commit()
        _push_bridge_call_state(
            voice_session,
            ai_state="error",
            handoff_reason=voice_session.handoff_reason,
            metadata=metadata,
        )
        _append_interaction_timeline(
            voice_session.interaction_id,
            "ai.error",
            {
                "call_id": voice_session.call_id,
                "voice_session_id": voice_session.session_id,
                "ai_session_id": voice_session.ai_session_id,
                "error": voice_session.handoff_reason,
                **(metadata or {}),
            },
        )
        if previous_status not in {"human_owned", "completed", "closed", "handoff_requested", "handoff_required"} and str(voice_session.interaction_id or "").strip():
            handoff_payload = VoiceAIHandoffRequestIn(
                voice_session_id=voice_session.session_id,
                ai_session_id=voice_session.ai_session_id,
                interaction_id=voice_session.interaction_id,
                target_queue_id=voice_session.handoff_target_queue_id,
                reason=voice_session.handoff_reason or "Voice AI media error.",
                summary={
                    "customer_request_text": "Не удалось завершить AI-аудио-сценарий.",
                    "ai_outcome_text": voice_session.handoff_reason or "Voice AI media error.",
                    "recommended_next_step": "Продолжить звонок вручную и проверить состояние Voice AI runtime.",
                },
            )
            try:
                _bridge_request(
                    "POST",
                    f"/internal/voice-ai/calls/{voice_session.call_id}/handoff",
                    payload=handoff_payload.model_dump(),
                    timeout=_handoff_timeout_seconds(),
                )
            except Exception:
                pass
    finally:
        session.close()


def _process_voice_ai_turn_sync(
    session_id: str,
    payload: VoiceAITurnIn,
    *,
    auto_handoff: bool,
) -> VoiceAITurnDecisionOut:
    session = get_session()
    voice_session = None
    try:
        voice_session = _load_voice_session(session, session_id)
        if voice_session.status in {"human_owned", "completed", "error"}:
            raise HTTPException(status_code=409, detail="Voice AI session is no longer active")
        now = utc_now_iso()
        voice_session.status = "thinking"
        voice_session.last_user_utterance_at = now
        voice_session.updated_at = now
        if payload.barge_in:
            _mark_last_assistant_segment_interrupted(session, voice_session.session_id)
        _record_segment(
            session,
            voice_session=voice_session,
            speaker="caller",
            source_type="asr",
            text=payload.transcript_text,
            sequence_no=max(payload.sequence_no, _next_sequence(session, voice_session.session_id)),
            payload={"metadata": payload.metadata, "barge_in": payload.barge_in},
        )
        session.commit()

        decision_payload = _orchestrator_request(
            "POST",
            f"/ai/voice/sessions/{session_id}/turns",
            payload=payload.model_dump(),
            timeout=12.0,
        )
        decision = VoiceAITurnDecisionOut.model_validate(decision_payload)
        voice_session.language = decision.language or voice_session.language
        voice_session.handoff_reason = decision.handoff_reason
        voice_session.status = "handoff_requested" if decision.needs_handoff else "active"
        voice_session.updated_at = utc_now_iso()
        if decision.reply_text:
            _record_segment(
                session,
                voice_session=voice_session,
                speaker="assistant",
                source_type="tts",
                text=decision.reply_text,
                sequence_no=_next_sequence(session, voice_session.session_id),
                payload={
                    "intent": decision.intent,
                    "kb_refs": decision.kb_refs,
                    "model": decision.model,
                    "delivery_status": "planned",
                },
                is_final=False,
            )
        session.commit()
        if auto_handoff and decision.needs_handoff:
            _request_runtime_handoff(session_id, payload.transcript_text, decision)
        return decision
    except HTTPException:
        raise
    except Exception as exc:
        if voice_session is not None:
            voice_session.status = "error"
            voice_session.handoff_reason = str(exc)[:1000]
            voice_session.updated_at = utc_now_iso()
            session.commit()
        raise HTTPException(status_code=502, detail=f"Voice AI runtime turn failed: {exc}") from exc
    finally:
        session.close()


def _media_process_turn(
    session_id: str,
    transcript_text: str,
    language: str | None,
    barge_in: bool,
    metadata: dict[str, Any] | None,
) -> VoiceAITurnDecisionOut:
    session = get_session()
    try:
        voice_session = _load_voice_session(session, session_id)
        interaction_id = str(voice_session.interaction_id or "").strip()
        call_id = str(voice_session.call_id or "").strip()
    finally:
        session.close()
    if not interaction_id or not call_id:
        raise RuntimeError("Voice AI session is missing call linkage")

    payload = VoiceAITurnIn(
        voice_session_id=session_id,
        call_id=call_id,
        interaction_id=interaction_id,
        transcript_text=transcript_text,
        language=language,
        sequence_no=1,
        barge_in=barge_in,
        metadata=metadata or {},
    )
    return _process_voice_ai_turn_sync(session_id, payload, auto_handoff=False)


def _complete_voice_ai_session_start(session_id: str, start_payload: VoiceAIStartIn) -> None:
    try:
        started = _orchestrator_request(
            "POST",
            f"/ai/voice/sessions/{session_id}/start",
            payload=start_payload.model_dump(),
            timeout=10.0,
        )
    except Exception as exc:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            voice_session.handoff_reason = str(exc)[:1000]
            voice_session.updated_at = utc_now_iso()
            session.commit()
        finally:
            session.close()
        return

    session = get_session()
    try:
        voice_session = _load_voice_session(session, session_id)
        now = utc_now_iso()
        greeting_text = str(started.get("greeting_text") or "").strip()
        voice_session.ai_session_id = str(started.get("session_id") or voice_session.ai_session_id or "").strip() or None
        voice_session.language = _resolved_voice_language(started.get("language"), voice_session.language)
        if greeting_text or _load_greeting_segment(session, session_id) is not None:
            voice_session.status = "greeting"
        elif not str(voice_session.status or "").strip():
            voice_session.status = "active"
        voice_session.handoff_reason = None
        voice_session.updated_at = now
        if greeting_text:
            _ensure_greeting_segment(session, voice_session, greeting_text)
        session.commit()
    finally:
        session.close()


_MEDIA_RUNTIME = AudioSocketMediaRuntime(
    enabled=_audiosocket_enabled(),
    host=_audiosocket_host(),
    port=_audiosocket_port(),
    frame_ms=_vad_frame_ms(),
    idle_timeout_seconds=_media_idle_timeout_seconds(),
    registration_wait_timeout_seconds=_media_registration_wait_timeout_seconds(),
    min_speech_ms=_vad_min_speech_ms(),
    trailing_silence_ms=_vad_trailing_silence_ms(),
    max_turn_ms=_turn_max_ms(),
    asr_provider=_ASR_PROVIDER,
    tts_provider=_TTS_PROVIDER,
    load_registration_by_media_uuid=_load_media_registration_by_uuid,
    mark_media_connected=_mark_media_connected,
    mark_media_ended=_mark_media_ended,
    touch_media_frame=_touch_media_frame,
    set_state=_set_voice_session_state,
    get_pending_greeting=_load_pending_greeting_text,
    mark_reply_delivered=_mark_reply_delivered,
    process_turn=_media_process_turn,
    request_handoff=_request_runtime_handoff,
    handle_media_error=_handle_media_error,
)


@asynccontextmanager
async def _lifespan(_: FastAPI):
    await _MEDIA_RUNTIME.start()
    try:
        yield
    finally:
        await _MEDIA_RUNTIME.stop()


app = FastAPI(title="ai-voice-runtime-service", version="1.0.0", lifespan=_lifespan)


@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
    return HealthResponse(status="ok", service="ai-voice-runtime-service", version="v1")


@app.post("/internal/voice-ai/sessions", response_model=VoiceAISessionOut)
def create_voice_ai_session(
    payload: VoiceAISessionCreateIn,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> VoiceAISessionOut:
    session = get_session()
    voice_session = None
    should_start_async = False
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == payload.call_id)
        ).scalar_one_or_none()
        now = utc_now_iso()
        language = _resolved_voice_language(
            payload.language_hint,
            voice_session.language if voice_session is not None else None,
        )
        if voice_session is None:
            voice_session = VoiceAISessionRow(
                session_id=new_id("avs"),
                call_id=payload.call_id,
                linked_id=payload.linked_id,
                interaction_id=payload.interaction_id,
                customer_id=None,
                queue_id=payload.queue_id,
                ai_session_id=None,
                agent_profile=payload.agent_profile,
                language=language,
                asr_provider=_ASR_PROVIDER.name,
                tts_provider=_TTS_PROVIDER.name,
                status="greeting",
                handoff_reason=None,
                handoff_target_queue_id=payload.handoff_queue_id or payload.queue_id,
                media_uuid=None,
                media_status=None,
                media_connected_at=None,
                media_ended_at=None,
                last_media_frame_at=None,
                disclosure_played_at=None,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=now,
                updated_at=now,
                ended_at=None,
            )
            session.add(voice_session)
            should_start_async = True
        else:
            voice_session.linked_id = payload.linked_id
            voice_session.interaction_id = payload.interaction_id
            voice_session.queue_id = payload.queue_id
            voice_session.agent_profile = payload.agent_profile
            voice_session.language = language
            voice_session.status = "greeting"
            voice_session.handoff_reason = None
            voice_session.handoff_target_queue_id = payload.handoff_queue_id or voice_session.handoff_target_queue_id or payload.queue_id
            voice_session.updated_at = now
            should_start_async = not str(voice_session.ai_session_id or "").strip()
        _ensure_greeting_segment(session, voice_session, _default_voice_greeting(language))
        session.commit()

        start_payload = VoiceAIStartIn(
            voice_session_id=voice_session.session_id,
            call_id=payload.call_id,
            interaction_id=payload.interaction_id,
            customer_id=None,
            language_hint=language,
            agent_profile=payload.agent_profile,
        )
        if should_start_async:
            threading.Thread(
                target=_complete_voice_ai_session_start,
                args=(voice_session.session_id, start_payload),
                name=f"voice-start-{voice_session.session_id}",
                daemon=True,
            ).start()
        return VoiceAISessionOut(
            voice_session_id=voice_session.session_id,
            ai_session_id=voice_session.ai_session_id,
            status=voice_session.status,  # type: ignore[arg-type]
        )
    except HTTPException:
        raise
    except Exception as exc:
        if voice_session is not None:
            voice_session.status = "error"
            voice_session.handoff_reason = str(exc)[:1000]
            voice_session.updated_at = utc_now_iso()
            session.commit()
        raise HTTPException(status_code=502, detail=f"Voice AI session start failed: {exc}") from exc
    finally:
        session.close()


@app.post("/internal/voice-ai/sessions/{session_id}/media-bridge", response_model=VoiceAISessionOut)
def register_voice_ai_media_bridge(
    session_id: str,
    payload: VoiceAIMediaBridgeEventIn,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> VoiceAISessionOut:
    voice_session = _persist_media_bridge_event(session_id, payload)
    if payload.event_type == "ended":
        _MEDIA_RUNTIME.close_session_sync(session_id, reason=payload.reason or "media_bridge_ended")
    return VoiceAISessionOut(
        voice_session_id=voice_session.session_id,
        ai_session_id=voice_session.ai_session_id,
        status=voice_session.status,  # type: ignore[arg-type]
    )


@app.post("/internal/voice-ai/sessions/{session_id}/telephony-events", response_model=VoiceAISessionOut)
def push_voice_ai_telephony_event(
    session_id: str,
    payload: VoiceAITelephonyEventIn,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> VoiceAISessionOut:
    session = get_session()
    try:
        voice_session = _load_voice_session(session, session_id)
        now = utc_now_iso()
        if payload.event_type == "call.connected":
            if voice_session.status in {"queued", "greeting", "active", "thinking", "speaking"}:
                voice_session.status = "listening"
        elif payload.event_type == "operator.connected":
            voice_session.status = "human_owned"
        elif payload.event_type == "call.ended":
            voice_session.status = "completed" if voice_session.status != "error" else "error"
            voice_session.ended_at = voice_session.ended_at or now
            voice_session.media_status = voice_session.media_status or "ended"
            voice_session.media_ended_at = voice_session.media_ended_at or now
        voice_session.updated_at = now
        session.commit()
        ai_session_id = voice_session.ai_session_id
        status = str(voice_session.status or "queued")
    finally:
        session.close()

    if payload.event_type == "call.ended":
        _MEDIA_RUNTIME.close_session_sync(session_id, reason="call_ended")
        try:
            _orchestrator_request(
                "POST",
                f"/ai/voice/sessions/{session_id}/close",
                payload={},
                timeout=5.0,
            )
        except Exception:
            pass

    return VoiceAISessionOut(
        voice_session_id=session_id,
        ai_session_id=ai_session_id,
        status=status,  # type: ignore[arg-type]
    )


@app.post("/internal/voice-ai/sessions/{session_id}/turns", response_model=VoiceAITurnDecisionOut)
def process_voice_ai_turn(
    session_id: str,
    payload: VoiceAITurnIn,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> VoiceAITurnDecisionOut:
    return _process_voice_ai_turn_sync(session_id, payload, auto_handoff=True)