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, build_streaming_asr_provider
from services.ai_voice_runtime_service.runtime_tts_provider import RuntimeConfiguredTTSProvider
from services.ai_orchestrator_service import operator_persona as persona
from services.shared.ai_operator_config import (
    load_effective_ai_operator_config,
    sync_ai_operator_config_from_code,
)
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 AsteriskCallLinkRow, VoiceAISessionRow, VoiceTranscriptSegmentRow
from services.shared.voice_transcripts import add_transcript_segment, next_transcript_sequence

init_sql_schema()


LOGGER = logging.getLogger("uvicorn.error")
_VOICE_START_THREADS_LOCK = threading.Lock()
_VOICE_START_THREADS: set[threading.Thread] = set()


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 _sales_service_url() -> str:
    return os.getenv("SALES_SERVICE_URL", "http://localhost:8020").rstrip("/")


def _sales_voice_sync_tenant_id() -> str | None:
    return os.getenv("SALES_VOICE_SYNC_TENANT_ID", "").strip() or None


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


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


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", 400), _vad_frame_ms())


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


def _vad_rms_threshold() -> int:
    return max(_int_env("AI_VOICE_VAD_RMS_THRESHOLD", 250), 1)


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 _voice_v2_enabled() -> bool:
    return _bool_env("AI_VOICE_V2_ENABLED", False)


def _voice_v2_queue_codes() -> set[str]:
    raw = str(os.getenv("AI_VOICE_V2_QUEUE_CODES", "voice_lab_ai") or "voice_lab_ai").strip()
    return {item.strip() for item in raw.split(",") if item.strip()}


def _voice_v2_ack_mode() -> str:
    return str(os.getenv("AI_VOICE_V2_ACK_MODE", "immediate_short") or "immediate_short").strip().lower()


def _voice_v2_streaming_tts_enabled() -> bool:
    return _bool_env("AI_VOICE_V2_STREAMING_TTS", True)


def _voice_v2_partial_asr_enabled() -> bool:
    return _bool_env("AI_VOICE_V2_PARTIAL_ASR", True)


def _voice_v2_duplex_enabled() -> bool:
    return _bool_env("AI_VOICE_V2_DUPLEX_ENABLED", True)


def _voice_v2_streaming_asr_backend() -> str:
    return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_BACKEND", "elevenlabs_realtime") or "elevenlabs_realtime").strip().lower()


def _voice_v2_prebaked_ack_enabled() -> bool:
    return _bool_env("AI_VOICE_V2_PREBAKED_ACK_ENABLED", True)


def _voice_v2_emotive_ack_enabled() -> bool:
    return _bool_env("AI_VOICE_V2_EMOTIVE_ACK_ENABLED", True)


def _voice_v2_emotive_ack_ru_only() -> bool:
    return _bool_env("AI_VOICE_V2_EMOTIVE_ACK_RU_ONLY", True)


def _service_headers(*, tenant_id: str | None = None) -> 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",
        tenant_id=tenant_id,
        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,
    tenant_id: str | None = None,
) -> dict[str, Any]:
    with httpx.Client(timeout=timeout) as client:
        response = client.request(
            method,
            url,
            json=payload,
            headers=_service_headers(tenant_id=tenant_id),
        )
    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 _register_voice_start_thread(thread: threading.Thread) -> None:
    with _VOICE_START_THREADS_LOCK:
        _VOICE_START_THREADS.add(thread)


def _unregister_voice_start_thread(thread: threading.Thread) -> None:
    with _VOICE_START_THREADS_LOCK:
        _VOICE_START_THREADS.discard(thread)


def _run_voice_start_thread(session_id: str, start_payload: VoiceAIStartIn) -> None:
    current = threading.current_thread()
    try:
        _complete_voice_ai_session_start(session_id, start_payload)
    finally:
        _unregister_voice_start_thread(current)


def _wait_for_background_voice_start_threads(timeout_seconds: float = 5.0) -> None:
    deadline = time.monotonic() + max(timeout_seconds, 0.0)
    while True:
        with _VOICE_START_THREADS_LOCK:
            threads = [thread for thread in _VOICE_START_THREADS if thread.is_alive()]
        if not threads:
            return
        remaining = max(deadline - time.monotonic(), 0.0)
        if remaining <= 0:
            return
        for thread in threads:
            thread.join(timeout=min(0.25, remaining))


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 _sales_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 8.0) -> dict[str, Any]:
    return _request(
        method,
        f"{_sales_service_url()}{path}",
        payload=payload,
        timeout=timeout,
        tenant_id=_sales_voice_sync_tenant_id(),
    )


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 _voice_start_name_prompt(language: str | None) -> str:
    if str(language or "").strip() == "kz":
        return "Сәлеметсіз бе. Атыңызды атаңызшы."
    return "Здравствуйте. Назовите, пожалуйста, ваше имя."


def _default_voice_greeting(language: str | None, *, agent_profile: str = "voice_support", operator_config: Any | None = None) -> str:
    if str(agent_profile or "").strip() == "voice_start":
        return _voice_start_name_prompt(language)
    return persona.voice_greeting(str(language or "ru").strip() or "ru", operator_config)


_ASR_PROVIDER = build_asr_provider(_asr_provider_name())
_STREAMING_ASR_PROVIDER = build_streaming_asr_provider(_voice_v2_streaming_asr_backend())
_TTS_PROVIDER = RuntimeConfiguredTTSProvider(default_provider_name=_tts_provider_name())


def _current_tts_provider_name() -> str:
    return _TTS_PROVIDER.current_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 _load_call_link(session, call_id: str) -> AsteriskCallLinkRow | None:
    return session.execute(
        select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
    ).scalar_one_or_none()


def _latest_caller_transcript_text(session, session_id: str) -> str | None:
    row = session.execute(
        select(VoiceTranscriptSegmentRow)
        .where(VoiceTranscriptSegmentRow.session_id == session_id)
        .where(VoiceTranscriptSegmentRow.speaker == "caller")
        .where(VoiceTranscriptSegmentRow.is_final.is_(True))
        .order_by(VoiceTranscriptSegmentRow.sequence_no.desc())
        .limit(1)
    ).scalar_one_or_none()
    if row is None:
        return None
    text = str(row.text or "").strip()
    return text or None


def _sales_transcript_snapshot(session, session_id: str, *, max_chars: int = 6000) -> str | None:
    rows = session.execute(
        select(VoiceTranscriptSegmentRow)
        .where(VoiceTranscriptSegmentRow.session_id == session_id)
        .where(VoiceTranscriptSegmentRow.is_final.is_(True))
        .order_by(VoiceTranscriptSegmentRow.sequence_no.asc())
    ).scalars().all()
    lines: list[str] = []
    for row in rows:
        text = str(row.text or "").strip()
        if not text:
            continue
        speaker = "customer" if str(row.speaker or "").strip() == "caller" else str(row.speaker or "assistant").strip()
        lines.append(f"{speaker}: {text}")
    if not lines:
        return None
    selected: list[str] = []
    total = 0
    for line in reversed(lines):
        line_len = len(line) + (1 if selected else 0)
        if selected and total + line_len > max_chars:
            break
        if not selected and len(line) > max_chars:
            selected.append(line[-max_chars:])
            total = len(selected[0])
            break
        selected.append(line)
        total += line_len
    return "\n".join(reversed(selected)) or None


def _sync_sales_voice_session(
    session_id: str,
    *,
    caller_number: str | None = None,
    caller_name: str | None = None,
    queue_code: str | None = None,
    summary: str | None = None,
    include_transcript: bool = False,
    metadata: dict[str, Any] | None = None,
) -> None:
    session = get_session()
    try:
        voice_session = _load_voice_session(session, session_id)
        link = _load_call_link(session, voice_session.call_id)
        resolved_summary = (
            str(summary or "").strip()
            or _latest_caller_transcript_text(session, session_id)
            or str(voice_session.handoff_reason or "").strip()
            or None
        )
        transcript_text = _sales_transcript_snapshot(session, session_id) if include_transcript else None
        payload = {
            "call_id": voice_session.call_id,
            "interaction_id": str(voice_session.interaction_id or getattr(link, "interaction_id", "") or "").strip() or None,
            "queue_id": str(voice_session.queue_id or getattr(link, "queue_id", "") or "").strip() or None,
            "queue_code": str(getattr(link, "queue_code", "") or queue_code or "").strip() or None,
            "caller_number": str(getattr(link, "caller_number", "") or caller_number or "").strip() or None,
            "caller_name": str(getattr(link, "caller_name", "") or caller_name or "").strip() or None,
            "voice_session_id": voice_session.session_id,
            "ai_session_id": voice_session.ai_session_id,
            "ai_state": voice_session.status,
            "handoff_reason": voice_session.handoff_reason,
            "telephony_status": str(getattr(link, "telephony_status", "") or voice_session.media_status or "").strip() or None,
            "call_status": str(getattr(link, "status", "") or voice_session.status or "").strip() or None,
            "started_at": voice_session.started_at,
            "ended_at": voice_session.ended_at,
            "summary": resolved_summary,
            "transcript_text": transcript_text,
            "metadata": {
                "linked_id": voice_session.linked_id,
                "agent_profile": voice_session.agent_profile,
                "language": voice_session.language,
                "asr_provider": voice_session.asr_provider,
                "tts_provider": voice_session.tts_provider,
                "handoff_target_queue_id": voice_session.handoff_target_queue_id,
                "media_status": voice_session.media_status,
                "media_uuid": voice_session.media_uuid,
                "media_connected_at": voice_session.media_connected_at,
                "media_ended_at": voice_session.media_ended_at,
                "last_media_frame_at": voice_session.last_media_frame_at,
                "last_user_utterance_at": voice_session.last_user_utterance_at,
                "last_ai_reply_at": voice_session.last_ai_reply_at,
                "voice_start_language": voice_session.voice_start_language,
                "customer_name_status": voice_session.customer_name_status,
                "customer_name_value": voice_session.customer_name_value,
                "customer_name_source": voice_session.customer_name_source,
                "customer_name_resolved_at": voice_session.customer_name_resolved_at,
                "claimed_by_user": getattr(link, "claimed_by_user", None),
                "claimed_at": getattr(link, "claimed_at", None),
                "operator_extension": getattr(link, "operator_extension", None),
                "channel_name": getattr(link, "channel_name", None),
                **(metadata or {}),
            },
        }
    finally:
        session.close()

    try:
        _sales_request("POST", "/internal/sales-sync/voice", payload=payload, timeout=8.0)
    except Exception:
        LOGGER.exception(
            "voice_runtime.sales_sync_failed session_id=%s call_id=%s ai_state=%s",
            payload.get("voice_session_id"),
            payload.get("call_id"),
            payload.get("ai_state"),
        )


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 _normalize_voice_text_key(text: str | None) -> str:
    compact = str(text or "").strip().lower()
    for char in ",.!?;:\"'()[]{}":
        compact = compact.replace(char, " ")
    return " ".join(compact.split())


def _is_low_signal_voice_text(text: str | None) -> bool:
    normalized = _normalize_voice_text_key(text)
    if not normalized:
        return True
    return normalized in {
        "\u0430\u043b\u043b\u043e",
        "\u0430\u0433\u0430",
        "\u0434\u0430",
        "\u0434\u043e\u0431\u0440\u044b\u0439 \u0434\u0435\u043d\u044c",
        "\u0437\u0434\u0440\u0430\u0432\u0441\u0442\u0432\u0443\u0439\u0442\u0435",
        "\u043b\u0430\u0434\u043d\u043e",
        "\u043d\u0435\u0442",
        "\u043d\u0435\u0430",
        "\u043e\u0439",
        "\u043e\u043a",
        "\u043f\u0440\u0438\u0432\u0435\u0442",
        "\u0441\u043b\u044b\u0448\u043d\u043e",
        "\u0441\u043b\u044b\u0448\u0443",
        "\u0443\u0433\u0443",
        "\u0445\u043e\u0440\u043e\u0448\u043e",
        "\u044f\u0441\u043d\u043e",
    }


def _is_intent_bearing_turn(transcript_text: str | None, intent: str | None) -> bool:
    normalized_intent = str(intent or "").strip().lower()
    if normalized_intent in {
        "schedule",
        "address",
        "price",
        "status",
        "problem",
        "operator_request",
        "handoff_request",
        "kb_answer",
    }:
        return True
    normalized_text = _normalize_voice_text_key(transcript_text)
    if not normalized_text or _is_low_signal_voice_text(normalized_text):
        return False
    markers = (
        "\u0433\u0440\u0430\u0444\u0438\u043a",
        "\u0432\u0440\u0435\u043c\u044f \u0440\u0430\u0431\u043e\u0442\u044b",
        "\u0440\u0435\u0436\u0438\u043c \u0440\u0430\u0431\u043e\u0442\u044b",
        "\u0430\u0434\u0440\u0435\u0441",
        "\u0444\u0438\u043b\u0438\u0430\u043b",
        "\u0433\u043e\u0440\u043e\u0434",
        "\u0437\u0430\u044f\u0432\u043a",
        "\u0441\u0442\u0430\u0442\u0443\u0441",
        "\u0442\u0430\u0440\u0438\u0444",
        "\u0446\u0435\u043d\u0430",
        "\u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u044c",
        "\u043e\u043f\u0435\u0440\u0430\u0442\u043e\u0440",
        "\u043c\u0435\u043d\u0435\u0434\u0436\u0435\u0440",
        "\u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442",
        "\u043e\u0448\u0438\u0431\u043a",
        "\u043f\u0440\u043e\u0431\u043b\u0435\u043c",
    )
    return any(marker in normalized_text for marker in markers)


def _annotate_latest_caller_segment(
    session,
    *,
    session_id: str,
    provider_name: str,
    transcript_text: str,
    intent: str | None,
    metadata: dict[str, Any] | None,
) -> None:
    row = session.execute(
        select(VoiceTranscriptSegmentRow)
        .where(VoiceTranscriptSegmentRow.session_id == session_id)
        .where(VoiceTranscriptSegmentRow.speaker == "caller")
        .order_by(VoiceTranscriptSegmentRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()
    if row is None:
        return
    try:
        payload = json.loads(row.payload_json or "{}")
    except Exception:
        payload = {}
    payload["provider"] = str(provider_name or "").strip() or payload.get("provider")
    payload["intent_bearing"] = _is_intent_bearing_turn(transcript_text, intent)
    if intent:
        payload["decision_intent"] = str(intent).strip()
    if isinstance(metadata, dict) and metadata:
        merged_metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
        merged_metadata.update(metadata)
        payload["metadata"] = merged_metadata
    row.payload_json = json.dumps(payload, ensure_ascii=False)


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 _load_latest_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.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 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",
            "provider": _current_tts_provider_name(),
            "delivery_status": "planned",
            "delivery_state": "planned",
        },
        is_final=False,
    )


def _upsert_greeting_segment(session, voice_session: VoiceAISessionRow, greeting_text: str) -> None:
    normalized = str(greeting_text or "").strip()
    if not normalized:
        return
    row = _load_latest_greeting_segment(session, voice_session.session_id)
    if row is None:
        _ensure_greeting_segment(session, voice_session, normalized)
        return
    row.text = normalized
    try:
        payload = json.loads(row.payload_json or "{}")
    except Exception:
        payload = {}
    payload["kind"] = "greeting"
    payload.setdefault("delivery_status", "planned")
    payload.setdefault("delivery_state", str(payload.get("delivery_status") or "planned"))
    payload.setdefault("provider", _current_tts_provider_name())
    row.payload_json = json.dumps(payload, ensure_ascii=False)
    row.is_final = False


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


def _voice_start_metadata(payload: dict[str, Any] | None) -> dict[str, str | None]:
    metadata = payload if isinstance(payload, dict) else {}
    return {
        "voice_start_language": str(
            metadata.get("voice_start_language") or metadata.get("language") or ""
        ).strip()
        or None,
        "customer_name_status": str(metadata.get("customer_name_status") or "").strip() or None,
        "customer_name_value": str(metadata.get("customer_name_value") or "").strip() or None,
        "customer_name_source": str(metadata.get("customer_name_source") or "").strip() or None,
        "customer_name_resolved_at": str(metadata.get("customer_name_resolved_at") or "").strip() or None,
        "downstream_queue_code": str(metadata.get("downstream_queue_code") or "").strip() or None,
    }


def _apply_voice_start_metadata(voice_session: VoiceAISessionRow, metadata: dict[str, Any] | None) -> None:
    payload = metadata if isinstance(metadata, dict) else {}
    values = _voice_start_metadata(payload)
    if "voice_start_language" in payload or "language" in payload:
        voice_session.voice_start_language = values["voice_start_language"]
    if "customer_name_status" in payload:
        voice_session.customer_name_status = values["customer_name_status"]
    if "customer_name_value" in payload:
        voice_session.customer_name_value = values["customer_name_value"]
    if "customer_name_source" in payload:
        voice_session.customer_name_source = values["customer_name_source"]
    if "customer_name_resolved_at" in payload:
        voice_session.customer_name_resolved_at = values["customer_name_resolved_at"]


def _voice_start_metadata_from_start_response(started: dict[str, Any] | None) -> dict[str, Any]:
    payload = dict((started or {}).get("metadata") or {})
    result = (started or {}).get("start_result")
    if isinstance(result, dict):
        if result.get("language") and not payload.get("voice_start_language"):
            payload["voice_start_language"] = result.get("language")
        if "customer_name_status" in result and "customer_name_status" not in payload:
            payload["customer_name_status"] = result.get("customer_name_status")
        if "customer_name_value" in result and "customer_name_value" not in payload:
            payload["customer_name_value"] = result.get("customer_name_value")
        if "customer_name_source" in result and "customer_name_source" not in payload:
            payload["customer_name_source"] = result.get("customer_name_source")
        if result.get("resolved_at") and "customer_name_resolved_at" not in payload:
            payload["customer_name_resolved_at"] = result.get("resolved_at")
        if result.get("downstream_queue_code") and "downstream_queue_code" not in payload:
            payload["downstream_queue_code"] = result.get("downstream_queue_code")
    return payload


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"
        payload["delivery_state"] = "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, *, queue_code: str | None = None) -> MediaRegistration:
    normalized_queue_code = str(queue_code or "").strip() or None
    voice_v2_for_session = bool(
        _voice_v2_enabled() and normalized_queue_code and normalized_queue_code in _voice_v2_queue_codes()
    )
    streaming_backend = _voice_v2_streaming_asr_backend() if voice_v2_for_session else None
    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,
        queue_code=normalized_queue_code,
        queue_id=str(row.queue_id or "").strip() or None,
        agent_profile=str(row.agent_profile or "").strip() or None,
        voice_v2_enabled=voice_v2_for_session,
        voice_v2_ack_mode=_voice_v2_ack_mode() if voice_v2_for_session else "disabled",
        voice_v2_streaming_tts=bool(voice_v2_for_session and _voice_v2_streaming_tts_enabled()),
        voice_v2_partial_asr=bool(voice_v2_for_session and _voice_v2_partial_asr_enabled()),
        voice_v2_duplex=bool(voice_v2_for_session and _voice_v2_duplex_enabled()),
        voice_v2_streaming_asr_backend=streaming_backend,
        voice_v2_prebaked_ack=bool(voice_v2_for_session and _voice_v2_prebaked_ack_enabled()),
        voice_v2_emotive_ack=bool(voice_v2_for_session and _voice_v2_emotive_ack_enabled()),
        voice_v2_emotive_ack_ru_only=bool(_voice_v2_emotive_ack_ru_only()),
    )


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
        call_link = session.execute(
            select(AsteriskCallLinkRow)
            .where(AsteriskCallLinkRow.call_id == row.call_id)
            .order_by(AsteriskCallLinkRow.id.desc())
            .limit(1)
        ).scalar_one_or_none()
        queue_code = str(call_link.queue_code or "").strip() or None if call_link is not None else None
        return _media_registration_from_row(row, queue_code=queue_code)
    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 _update_reply_delivery_status(
    session,
    *,
    session_id: str,
    text: str,
    is_greeting: bool,
    phase: str | None,
    status: str,
) -> VoiceAISessionRow:
    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")
        .order_by(VoiceTranscriptSegmentRow.id.desc())
    ).scalars().all()
    matched_row = None
    normalized_text = str(text or "").strip()
    expected_phase = str(phase or "").strip() or None
    for row in pending_rows:
        try:
            payload = json.loads(row.payload_json or "{}")
        except Exception:
            payload = {}
        row_kind = str(payload.get("kind") or "").strip()
        row_phase = str(payload.get("metadata", {}).get("phase") or "").strip() or None
        if is_greeting:
            if row_kind != "greeting":
                continue
        elif str(row.text or "").strip() != normalized_text:
            continue
        if expected_phase and row_phase not in {expected_phase, None}:
            continue
        matched_row = row
        payload["delivery_status"] = status
        payload["delivery_state"] = status
        payload.setdefault("provider", _current_tts_provider_name())
        row.payload_json = json.dumps(payload, ensure_ascii=False)
        break
    if matched_row is not None:
        matched_row.is_final = status == "delivered"
        if status in {"interrupted", "discarded"}:
            matched_row.barge_in_interrupted = status == "interrupted"
    if is_greeting and status == "delivered" and not voice_session.disclosure_played_at:
        voice_session.disclosure_played_at = now
    if status == "delivered":
        voice_session.last_ai_reply_at = now
    voice_session.updated_at = now
    return voice_session


def _mark_reply_started(session_id: str, text: str, is_greeting: bool, phase: str | None = None) -> None:
    def _write() -> None:
        session = get_session()
        try:
            _update_reply_delivery_status(
                session,
                session_id=session_id,
                text=text,
                is_greeting=is_greeting,
                phase=phase,
                status="started",
            )
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _mark_reply_delivered(session_id: str, text: str, is_greeting: bool, phase: str | None = None) -> None:
    def _write() -> None:
        session = get_session()
        try:
            _update_reply_delivery_status(
                session,
                text=text,
                session_id=session_id,
                is_greeting=is_greeting,
                phase=phase,
                status="delivered",
            )
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _record_latency_metric(session_id: str, metric: str, latency_ms: int) -> None:
    normalized_metric = str(metric or "").strip()
    if not normalized_metric:
        return

    def _write() -> None:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            _record_segment(
                session,
                voice_session=voice_session,
                speaker="system",
                source_type="runtime",
                text=normalized_metric,
                sequence_no=_next_sequence(session, voice_session.session_id),
                payload={
                    "kind": "latency_metric",
                    "metric": normalized_metric,
                    "latency_ms": int(latency_ms),
                },
                is_final=True,
            )
            voice_session.updated_at = utc_now_iso()
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _mark_reply_discarded(session_id: str, text: str, phase: str | None = None, status: str = "discarded") -> None:
    normalized_status = str(status or "").strip().lower() or "discarded"

    def _write() -> None:
        session = get_session()
        try:
            _update_reply_delivery_status(
                session,
                session_id=session_id,
                text=text,
                is_greeting=False,
                phase=phase,
                status=normalized_status,
            )
            session.commit()
        finally:
            session.close()

    _retry_db_write(_write)


def _record_runtime_reply_planned(
    session_id: str,
    text: str,
    metadata: dict[str, Any] | None = None,
    kind: str = "reply",
) -> None:
    normalized_text = str(text or "").strip()
    if not normalized_text:
        return

    def _write() -> None:
        session = get_session()
        try:
            voice_session = _load_voice_session(session, session_id)
            _record_segment(
                session,
                voice_session=voice_session,
                speaker="assistant",
                source_type="tts",
                text=normalized_text,
                sequence_no=_next_sequence(session, voice_session.session_id),
                payload={
                    "provider": _current_tts_provider_name(),
                    "delivery_status": "planned",
                    "delivery_state": "planned",
                    "kind": kind,
                    "metadata": metadata or {},
                },
                is_final=False,
            )
            voice_session.updated_at = utc_now_iso()
            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
            _apply_voice_start_metadata(voice_session, metadata)
            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,
    )
    _sync_sales_voice_session(
        voice_session.session_id,
        include_transcript=ai_state in {"handoff_requested", "human_owned", "completed", "closed", "error"},
        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_payload(
    session_id: str,
    *,
    reason: str,
    summary: dict[str, Any] | None = None,
    metadata: dict[str, Any] | None = None,
) -> 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=reason,
            summary=summary or {},
            metadata=metadata or {},
        )
        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,
            reason[: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 _request_runtime_handoff(
    session_id: str,
    customer_request_text: str,
    decision: VoiceAITurnDecisionOut,
) -> None:
    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": "Continue the call manually and confirm the collected context.",
    }
    if decision.metadata:
        for key in (
            "customer_name_status",
            "customer_name_value",
            "customer_name_source",
            "voice_start_language",
        ):
            value = decision.metadata.get(key)
            if value:
                handoff_summary[key] = value
    return _request_runtime_handoff_payload(
        session_id,
        reason=decision.handoff_reason or "AI requested human handoff.",
        summary=handoff_summary,
        metadata=decision.metadata,
    )


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
        _sync_sales_voice_session(
            voice_session.session_id,
            include_transcript=True,
            metadata=metadata,
        )
    finally:
        session.close()


def _process_voice_ai_turn_sync(
    session_id: str,
    payload: VoiceAITurnIn,
    *,
    auto_handoff: bool,
) -> VoiceAITurnDecisionOut:
    session = get_session()
    voice_session = None
    payload_metadata = payload.metadata if isinstance(payload.metadata, dict) else {}
    reply_phase = str(payload_metadata.get("reply_phase") or "").strip().lower()
    early_plan_only = reply_phase == "early_plan"
    try:
        voice_session = _load_voice_session(session, session_id)
        if not early_plan_only and 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()
        if not early_plan_only:
            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={
                    "provider": _ASR_PROVIDER.name,
                    "metadata": payload_metadata,
                    "barge_in": payload.barge_in,
                    "intent_bearing": False,
                },
            )
            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)
        if not early_plan_only:
            voice_session.language = decision.language or voice_session.language
            voice_session.handoff_reason = decision.handoff_reason
            _apply_voice_start_metadata(voice_session, decision.metadata)
            _annotate_latest_caller_segment(
                session,
                session_id=voice_session.session_id,
                provider_name=_ASR_PROVIDER.name,
                transcript_text=payload.transcript_text,
                intent=decision.intent,
                metadata=payload_metadata,
            )
            voice_session.status = decision.status or ("handoff_requested" if decision.needs_handoff else "active")
            voice_session.updated_at = utc_now_iso()
            if decision.reply_text and not bool(payload_metadata.get("runtime_defer_reply_planned")):
                _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,
                        "provider": _current_tts_provider_name(),
                        "delivery_status": "planned",
                        "delivery_state": "planned",
                        "metadata": decision.metadata,
                    },
                    is_final=False,
                )
            session.commit()
        if auto_handoff and decision.needs_handoff and not early_plan_only:
            _request_runtime_handoff(session_id, payload.transcript_text, decision)
        if not early_plan_only:
            _sync_sales_voice_session(
                session_id,
                include_transcript=True,
                summary=decision.summary_text or decision.reply_text or payload.transcript_text,
                metadata=decision.metadata,
            )
        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()
        handoff_metadata = _voice_start_metadata_from_start_response(started)
        needs_handoff = bool(started.get("needs_handoff"))
        handoff_reason = str(started.get("handoff_reason") or "").strip() or None
        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)
        _apply_voice_start_metadata(voice_session, handoff_metadata)
        if greeting_text:
            _upsert_greeting_segment(session, voice_session, greeting_text)
        if needs_handoff:
            voice_session.status = "handoff_requested"
        elif 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"
        else:
            voice_session.status = "active"
        voice_session.handoff_reason = handoff_reason
        voice_session.updated_at = now
        session.commit()
        session.refresh(voice_session)
    finally:
        session.close()
    _push_bridge_call_state(
        voice_session,
        ai_state=voice_session.status,
        handoff_reason=voice_session.handoff_reason,
        metadata=handoff_metadata,
    )
    _sync_sales_voice_session(
        voice_session.session_id,
        include_transcript=False,
        metadata=handoff_metadata,
    )
    if needs_handoff:
        handoff_summary = {
            "customer_request_text": "voice_start",
            "ai_outcome_text": str(started.get("summary_text") or "").strip(),
            "recommended_next_step": "Continue the next voice stage using the language and customer name handoff.",
        }
        for key in (
            "customer_name_status",
            "customer_name_value",
            "customer_name_source",
            "voice_start_language",
        ):
            value = handoff_metadata.get(key)
            if value:
                handoff_summary[key] = value
        _request_runtime_handoff_payload(
            session_id,
            reason=handoff_reason or "voice_start_completed",
            summary=handoff_summary,
            metadata=handoff_metadata,
        )


_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(),
    vad_rms_threshold=_vad_rms_threshold(),
    asr_provider=_ASR_PROVIDER,
    streaming_asr_provider=_STREAMING_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_started=_mark_reply_started,
    mark_reply_delivered=_mark_reply_delivered,
    mark_reply_discarded=_mark_reply_discarded,
    plan_reply=_record_runtime_reply_planned,
    record_latency=_record_latency_metric,
    process_turn=_media_process_turn,
    request_handoff=_request_runtime_handoff,
    handle_media_error=_handle_media_error,
)


@asynccontextmanager
async def _lifespan(_: FastAPI):
    _sync_session = get_session()
    try:
        sync_ai_operator_config_from_code(_sync_session)
    finally:
        _sync_session.close()
    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,
        )
        payload_metadata = payload.metadata if isinstance(payload.metadata, dict) else {}
        effective_tts_provider_name = _current_tts_provider_name()
        is_voice_start = (
            str(payload.agent_profile or "").strip() == "voice_start"
            or str(payload_metadata.get("stage") or "").strip() == "voice_start"
        )
        operator_config = load_effective_ai_operator_config(session)
        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=effective_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:
            stage_changed = (
                str(voice_session.queue_id or "").strip() != str(payload.queue_id or "").strip()
                or str(voice_session.agent_profile or "").strip() != str(payload.agent_profile or "").strip()
            )
            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.tts_provider = effective_tts_provider_name
            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
            if stage_changed:
                voice_session.ai_session_id = None
                voice_session.disclosure_played_at = None
                voice_session.ended_at = None
            should_start_async = stage_changed or not str(voice_session.ai_session_id or "").strip()
            voice_session.updated_at = now
        _apply_voice_start_metadata(voice_session, payload_metadata)
        greeting_text = _default_voice_greeting(
            language,
            agent_profile=payload.agent_profile,
            operator_config=operator_config,
        )
        if not is_voice_start:
            if voice_session.ai_session_id is None or should_start_async:
                _queue_new_greeting_segment(session, voice_session, greeting_text)
            else:
                _upsert_greeting_segment(session, voice_session, greeting_text)
        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,
            metadata=payload_metadata,
        )
        if should_start_async:
            voice_start_thread = threading.Thread(
                target=_run_voice_start_thread,
                args=(voice_session.session_id, start_payload),
                name=f"voice-start-{voice_session.session_id}",
                daemon=True,
            )
            _register_voice_start_thread(voice_start_thread)
            voice_start_thread.start()
        _sync_sales_voice_session(
            voice_session.session_id,
            caller_number=payload.caller_number,
            caller_name=payload.caller_name,
            queue_code=str(payload_metadata.get("queue_code") or "").strip() or None,
            include_transcript=False,
            metadata=payload_metadata,
        )
        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
    _sync_sales_voice_session(
        session_id,
        include_transcript=payload.event_type in {"operator.connected", "call.ended"},
        metadata={"telephony_event_type": payload.event_type, **(payload.payload or {})},
    )

    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)
