call-center/services/ai_orchestrator_service/voice.py

from __future__ import annotations

import importlib
import json
import re
from typing import Any

from fastapi import HTTPException
from sqlalchemy import select

from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.models import VoiceAIStartIn, VoiceAIStartOut, VoiceAITurnDecisionOut, VoiceAITurnIn
from services.shared.sql_models import (
    AISessionRow,
    AITurnRow,
    AsteriskCallLinkRow,
    Customer,
    CustomerExternalIdentity,
    Interaction,
    InteractionTimeline,
    VoiceAISessionRow,
    VoiceTranscriptSegmentRow,
)


def _app():
    return importlib.import_module("services.ai_orchestrator_service.app")


def _voice_max_context_segments() -> int:
    app = _app()
    return max(6, app._int_env("AI_VOICE_MAX_CONTEXT_SEGMENTS", 8))


def _voice_disclosure_prefix(language: str) -> str:
    if language == "kz":
        return "Men kompaniyanyn AI operatoriymyn. "
    return "Я AI-оператор компании. "


def _voice_greeting(language: str) -> str:
    if language == "kz":
        return (
            "Men kompaniyanyn AI operatoriymyn. Salemetsiz be. "
            "Suragynyzdy aitanyz, men birden komektesuge tyrisamyn."
        )
    return (
        "Я AI-оператор компании. Здравствуйте. "
        "Коротко расскажите, с чем помочь, и я сразу начну разбираться."
    )


def _voice_handoff_reply(language: str) -> str:
    if language == "kz":
        return (
            "Men AI operator retinde bastapky konteksti jiyap aldyм. "
            "Kazir sizdi tiry operatorga kosamyn."
        )
    return (
        "Я как AI-оператор собрал первичный контекст. "
        "Сейчас переведу вас на живого оператора."
    )


def _normalize_phone(value: str | None) -> str | None:
    raw = str(value or "").strip()
    if not raw:
        return None
    digits = "".join(ch for ch in raw if ch.isdigit())
    if not digits:
        return raw
    if raw.startswith("+"):
        return f"+{digits}"
    return digits


def _voice_identity_subjects(caller_number: str | None) -> list[str]:
    normalized = _normalize_phone(caller_number)
    values = [normalized, str(caller_number or "").strip() or None]
    seen: set[str] = set()
    result: list[str] = []
    for value in values:
        if not value or value in seen:
            continue
        seen.add(value)
        result.append(value)
    return result


def _resolve_caller_from_call(session, call_id: str) -> tuple[str | None, str | None]:
    row = session.execute(
        select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
    ).scalar_one_or_none()
    if row is None:
        return None, None
    return row.caller_number, row.caller_name


def _ensure_voice_identity(
    session,
    *,
    customer_id: str,
    caller_number: str,
    caller_name: str | None,
    now: str,
) -> None:
    for subject in _voice_identity_subjects(caller_number):
        identity = session.execute(
            select(CustomerExternalIdentity).where(
                CustomerExternalIdentity.channel == "voice",
                CustomerExternalIdentity.external_subject == subject,
            )
        ).scalar_one_or_none()
        if identity:
            identity.customer_id = customer_id
            identity.display_name_snapshot = caller_name or identity.display_name_snapshot
            identity.updated_at = now
            continue
        session.add(
            CustomerExternalIdentity(
                identity_id=new_id("cei"),
                customer_id=customer_id,
                channel="voice",
                external_subject=subject,
                display_name_snapshot=caller_name,
                created_at=now,
                updated_at=now,
            )
        )


def _resolve_or_create_voice_customer_id(
    session,
    *,
    interaction: Interaction,
    call_id: str,
) -> str | None:
    app = _app()
    if app._customer_id_is_real(interaction.customer_id):
        return str(interaction.customer_id)

    caller_number, caller_name = _resolve_caller_from_call(session, call_id)
    subjects = _voice_identity_subjects(caller_number)
    now = utc_now_iso()
    for subject in subjects:
        identity = session.execute(
            select(CustomerExternalIdentity).where(
                CustomerExternalIdentity.channel == "voice",
                CustomerExternalIdentity.external_subject == subject,
            )
        ).scalar_one_or_none()
        if identity:
            interaction.customer_id = identity.customer_id
            interaction.updated_at = now
            return identity.customer_id

    if not caller_number:
        return str(interaction.customer_id or "").strip() or None

    customer = Customer(
        customer_id=new_id("cus"),
        display_name=caller_name or caller_number,
        phones_json=json.dumps([caller_number], ensure_ascii=False),
        preferred_phone=caller_number,
        tags_json=json.dumps(["voice"], ensure_ascii=False),
        created_at=now,
    )
    session.add(customer)
    interaction.customer_id = customer.customer_id
    interaction.updated_at = now
    _ensure_voice_identity(
        session,
        customer_id=customer.customer_id,
        caller_number=caller_number,
        caller_name=caller_name,
        now=now,
    )
    return customer.customer_id


def _voice_recent_segments(
    session,
    *,
    voice_session_id: str,
    limit: int,
) -> list[VoiceTranscriptSegmentRow]:
    rows = session.execute(
        select(VoiceTranscriptSegmentRow)
        .where(VoiceTranscriptSegmentRow.session_id == voice_session_id)
        .where(VoiceTranscriptSegmentRow.is_final.is_(True))
        .order_by(VoiceTranscriptSegmentRow.id.desc())
        .limit(max(limit, 1))
    ).scalars().all()
    return list(reversed(rows))


def _voice_text_key(text: str | None) -> str:
    compact = re.sub(r"[^\w\s]+", " ", str(text or "").lower(), flags=re.UNICODE)
    return re.sub(r"\s+", " ", compact).strip()


def _voice_recent_caller_texts(
    transcript_window: list[VoiceTranscriptSegmentRow],
    *,
    limit: int = 4,
) -> list[str]:
    caller_texts = [str(segment.text or "").strip() for segment in transcript_window if segment.speaker == "caller"]
    return caller_texts[-max(limit, 1) :]


def _voice_is_low_signal_caller_text(text: str | None) -> bool:
    normalized = _voice_text_key(text)
    if not normalized:
        return True
    low_signal_phrases = {
        "алло",
        "ага",
        "да",
        "добрый день",
        "здравствуйте",
        "ладно",
        "неа",
        "нет",
        "ок",
        "понял",
        "поняла",
        "привет",
        "слышу",
        "слышно",
        "угу",
        "хорошо",
        "ясно",
    }
    return normalized in low_signal_phrases


def _voice_is_confused_caller_text(text: str | None) -> bool:
    normalized = _voice_text_key(text)
    confusion_markers = (
        "не понял",
        "не поняла",
        "не понимаю",
        "неясно",
        "о каком",
        "каком запросе",
        "повторите",
        "что именно",
        "что вы имеете",
    )
    return any(marker in normalized for marker in confusion_markers)


def _voice_recent_clarification_count(transcript_window: list[VoiceTranscriptSegmentRow]) -> int:
    markers = (
        "в двух словах",
        "график работы какого",
        "какая услуга",
        "какой филиал",
        "назовите пожалуйста",
        "опишите пожалуйста",
        "подскажите",
        "скажите коротко",
        "уточните",
        "цель звонка",
        "что вам нужно",
        "что именно не работает",
        "чтобы помочь",
    )
    count = 0
    for segment in transcript_window:
        if segment.speaker != "assistant":
            continue
        normalized = _voice_text_key(segment.text)
        if any(marker in normalized for marker in markers):
            count += 1
    return count


def _voice_repeated_assistant_reply_count(transcript_window: list[VoiceTranscriptSegmentRow]) -> int:
    assistant_texts = [
        _voice_text_key(segment.text)
        for segment in transcript_window
        if segment.speaker == "assistant" and str(segment.text or "").strip()
    ]
    if not assistant_texts:
        return 0
    last_text = assistant_texts[-1]
    return sum(1 for text in assistant_texts[-3:] if text == last_text)


def _voice_topic_prompt(language: str, caller_texts: list[str]) -> str | None:
    context = _voice_text_key(" ".join(caller_texts))
    if not context:
        return None
    if any(marker in context for marker in ("график", "время работы", "режим работы", "часы работы", "работаете")):
        if language == "kz":
            return "Qai filialdyn, mekendyng nemese qalanyng jumys uaqyty qyzyqtyratynyn aitnyz."
        return "Подскажите, график работы какого филиала, адреса или города вас интересует?"
    if any(marker in context for marker in ("статус", "заявк", "заказ", "обращени", "запрос")):
        if language == "kz":
            return "Otinish no'mirin nemese resimdegen telefon nomerin aitnyz."
        return "Назовите, пожалуйста, номер заявки или телефон, по которому она оформлялась."
    if any(marker in context for marker in ("тариф", "стоимост", "цен", "оплат", "оплата", "услуг")):
        if language == "kz":
            return "Qai qyzmet, tarif nemese bagasy qyzyqtyratynyn naqtylańyz."
        return "Уточните, какая услуга, тариф или стоимость вас интересует."
    if any(marker in context for marker in ("адрес", "филиал", "город", "офис", "отделени", "где находит")):
        if language == "kz":
            return "Qai filial, mekenjai nemese qala qyzyqtyratynyn aitnyz."
        return "Подскажите, какой филиал, адрес или город вас интересует."
    if any(marker in context for marker in ("не работает", "ошибка", "проблем", "сбой", "интернет", "связь")):
        if language == "kz":
            return "Bir soilemmen naqty aitynyz: ne istep turmagan?"
        return "Опишите, пожалуйста, проблему одним предложением: что именно не работает?"
    return None


def _voice_generic_prompt(language: str) -> str:
    if language == "kz":
        return "Jyldam komektesu ushin eki ush sozben ne kerek ekenin aitnyz: jumys uaqyty, otinish statusy, tarif nemese operator."
    return "Чтобы помочь быстрее, скажите в двух словах, что вам нужно: график работы, статус заявки, тариф или оператор."


def _voice_confusion_prompt(language: str, caller_texts: list[str]) -> str:
    topic_prompt = _voice_topic_prompt(language, caller_texts)
    if topic_prompt:
        if language == "kz":
            return f"Qongyrau taqyrybyn naqtylap jatyrmyn. {topic_prompt}"
        return f"Сейчас уточняю цель звонка. {topic_prompt}"
    if language == "kz":
        return "Qazir qongyraudyn maqratyn anyqtap jatyrmyn. Qysqasha aitnyz: jumys uaqyty, otinish statusy, tarif nemese operator."
    return "Сейчас уточняю цель звонка. Скажите коротко, что именно нужно: график работы, статус заявки, тариф или оператор."


def _voice_loop_handoff(language: str) -> tuple[str, str, str]:
    if language == "kz":
        return (
            "Men suraqty birneshe ret naqtylap korgendim. Sizdi kuttirmey, tiri operatorga qosamyn.",
            "Voice AI birneshe naqtylau areketinen keyin suraqty anyqtay almady.",
            "AI birneshe naqtylau areketinen keyin qongyraudyn maqratyn operatorga berdi.",
        )
    return (
        "Похоже, я не смог точно уточнить вопрос в голосовом режиме. Чтобы не задерживать вас, перевожу на оператора.",
        "Voice AI не смог уточнить запрос после нескольких попыток без прогресса.",
        "AI не смог уточнить цель звонка после нескольких попыток и перевел клиента на оператора.",
    )


def _voice_recent_timeline(
    session,
    *,
    interaction_id: str,
    limit: int = 6,
) -> list[dict[str, Any]]:
    rows = session.execute(
        select(InteractionTimeline)
        .where(InteractionTimeline.interaction_id == interaction_id)
        .order_by(InteractionTimeline.id.desc())
        .limit(max(limit, 1))
    ).scalars().all()
    result: list[dict[str, Any]] = []
    for row in reversed(rows):
        try:
            metadata = json.loads(row.metadata_json or "{}")
        except Exception:
            metadata = {}
        result.append({"timestamp": row.timestamp, "action": row.action, "metadata": metadata})
    return result


def _append_timeline(interaction_id: str, action: str, metadata: dict[str, Any]) -> None:
    app = _app()
    try:
        app._interaction_request(
            "POST",
            f"/interactions/{interaction_id}/timeline",
            payload={"action": action, "metadata": metadata},
        )
    except Exception:
        session = get_session()
        try:
            app._push_timeline(session, interaction_id, action, metadata)
            session.commit()
        finally:
            session.close()


def _record_voice_ai_turn(
    session,
    *,
    ai_session_id: str,
    interaction_id: str,
    role: str,
    source_type: str,
    text: str,
    payload: dict[str, Any],
    model: str | None = None,
    latency_ms: int | None = None,
) -> None:
    session.add(
        AITurnRow(
            turn_id=new_id("ait"),
            session_id=ai_session_id,
            thread_id=None,
            interaction_id=interaction_id,
            role=role,
            source_type=source_type,
            text=text,
            payload_json=json.dumps(payload, ensure_ascii=False),
            model=model,
            finish_reason="stop",
            latency_ms=latency_ms,
            created_at=utc_now_iso(),
        )
    )


def _ensure_voice_ai_session(
    session,
    *,
    voice_session: VoiceAISessionRow,
    interaction: Interaction,
    customer_id: str | None,
    language: str,
    agent_profile: str,
) -> tuple[AISessionRow, bool]:
    existing = None
    if str(voice_session.ai_session_id or "").strip():
        existing = session.execute(
            select(AISessionRow).where(AISessionRow.session_id == voice_session.ai_session_id)
        ).scalar_one_or_none()
    if existing and existing.status not in {"closed", "error"}:
        existing.call_id = voice_session.call_id
        existing.interaction_id = interaction.interaction_id
        existing.customer_id = customer_id
        existing.language = language
        existing.agent_profile = agent_profile
        existing.updated_at = utc_now_iso()
        return existing, False

    now = utc_now_iso()
    ai_session = AISessionRow(
        session_id=new_id("ais"),
        channel="voice",
        call_id=voice_session.call_id,
        thread_id=None,
        interaction_id=interaction.interaction_id,
        customer_id=customer_id,
        agent_profile=agent_profile,
        language=language,
        status="active",
        summary_text="",
        last_user_message_id=None,
        last_ai_message_id=None,
        handoff_reason=None,
        created_at=now,
        updated_at=now,
        closed_at=None,
    )
    session.add(ai_session)
    voice_session.ai_session_id = ai_session.session_id
    return ai_session, True


def _voice_decision_legacy(
    *,
    language: str,
    customer: Customer | None,
    interaction: Interaction,
    transcript_text: str,
    transcript_window: list[VoiceTranscriptSegmentRow],
    kb_results: list[Any],
    disclosure_required: bool,
) -> dict[str, Any]:
    app = _app()
    normalized = str(transcript_text or "").strip()
    lower_text = normalized.lower()
    needs_handoff = app._looks_like_human_request(lower_text) or app._is_sensitive_request(lower_text)
    model = app._ai_model()
    if needs_handoff:
        reply_text = _voice_handoff_reply(language)
        if disclosure_required and not reply_text.startswith(_voice_disclosure_prefix(language)):
            reply_text = f"{_voice_disclosure_prefix(language)}{reply_text}"
        return {
            "language": language,
            "intent": "handoff_request",
            "reply_text": reply_text,
            "confidence": 0.25,
            "needs_handoff": True,
            "handoff_reason": "Запрос требует участия живого оператора.",
            "case_action": "keep_open",
            "kb_refs": [],
            "summary_text": "AI собрал первичный контекст и запросил живого оператора.",
            "model": model,
            "latency_ms": 1,
        }

    customer_name = customer.display_name if customer else "клиент"
    if kb_results:
        article = kb_results[0]
        snippet = app._article_snippet(article, limit=220)
        reply_text = f"По базе знаний вижу следующее: {snippet}"
        summary_text = f"AI дал первичный ответ по базе знаний для {customer_name}."
        kb_refs = [article.article_id]
        confidence = 0.82
        intent = "kb_answer"
    else:
        transcript_context = " ".join(segment.text for segment in transcript_window[-3:] if segment.speaker == "caller")
        reply_text = (
            "Я услышал запрос и уже собрал основной контекст. "
            "Пожалуйста, уточните самый важный результат, который вы хотите получить."
        )
        if transcript_context and transcript_context != normalized:
            reply_text += " Если правильно понял, речь идет об этом вопросе из разговора."
        summary_text = f"AI уточняет цель звонка и собирает контекст для {customer_name}."
        kb_refs = []
        confidence = 0.68
        intent = "clarification"

    if disclosure_required and not reply_text.startswith(_voice_disclosure_prefix(language)):
        reply_text = f"{_voice_disclosure_prefix(language)}{reply_text}"
    return {
        "language": language,
        "intent": intent,
        "reply_text": reply_text,
        "confidence": confidence,
        "needs_handoff": False,
        "handoff_reason": None,
        "case_action": "keep_open",
        "kb_refs": kb_refs,
        "summary_text": summary_text,
        "model": model,
        "latency_ms": 1,
    }


def _voice_decision(
    *,
    language: str,
    customer: Customer | None,
    interaction: Interaction,
    transcript_text: str,
    transcript_window: list[VoiceTranscriptSegmentRow],
    kb_results: list[Any],
    disclosure_required: bool,
) -> dict[str, Any]:
    app = _app()
    normalized = str(transcript_text or "").strip()
    lower_text = normalized.lower()
    caller_texts = _voice_recent_caller_texts(transcript_window)
    needs_handoff = app._looks_like_human_request(lower_text) or app._is_sensitive_request(lower_text)
    model = app._ai_model()
    if needs_handoff:
        reply_text = _voice_handoff_reply(language)
        if disclosure_required and not reply_text.startswith(_voice_disclosure_prefix(language)):
            reply_text = f"{_voice_disclosure_prefix(language)}{reply_text}"
        return {
            "language": language,
            "intent": "handoff_request",
            "reply_text": reply_text,
            "confidence": 0.25,
            "needs_handoff": True,
            "handoff_reason": "Запрос требует участия живого оператора.",
            "case_action": "keep_open",
            "kb_refs": [],
            "summary_text": "AI собрал первичный контекст и запросил живого оператора.",
            "model": model,
            "latency_ms": 1,
        }

    customer_name = customer.display_name if customer else "клиент"
    if kb_results:
        article = kb_results[0]
        snippet = app._article_snippet(article, limit=220)
        reply_text = f"По базе знаний вижу следующее: {snippet}"
        summary_text = f"AI дал первичный ответ по базе знаний для {customer_name}."
        kb_refs = [article.article_id]
        confidence = 0.82
        intent = "kb_answer"
    else:
        clarification_count = _voice_recent_clarification_count(transcript_window)
        repeated_reply_count = _voice_repeated_assistant_reply_count(transcript_window)
        recent_caller_window = caller_texts[-3:]
        low_signal_count = sum(1 for text in recent_caller_window if _voice_is_low_signal_caller_text(text))
        caller_confused = _voice_is_confused_caller_text(normalized)

        if clarification_count >= 4 or (
            clarification_count >= 3 and (caller_confused or low_signal_count >= 2 or repeated_reply_count >= 2)
        ):
            reply_text, handoff_reason, summary_text = _voice_loop_handoff(language)
            if disclosure_required and not reply_text.startswith(_voice_disclosure_prefix(language)):
                reply_text = f"{_voice_disclosure_prefix(language)}{reply_text}"
            return {
                "language": language,
                "intent": "handoff_request",
                "reply_text": reply_text,
                "confidence": 0.34,
                "needs_handoff": True,
                "handoff_reason": handoff_reason,
                "case_action": "keep_open",
                "kb_refs": [],
                "summary_text": summary_text,
                "model": model,
                "latency_ms": 1,
            }

        if caller_confused:
            reply_text = _voice_confusion_prompt(language, caller_texts)
        else:
            topic_prompt = _voice_topic_prompt(language, caller_texts)
            if clarification_count >= 2 and not topic_prompt:
                reply_text = _voice_confusion_prompt(language, caller_texts)
            else:
                reply_text = topic_prompt or _voice_generic_prompt(language)
        summary_text = f"AI уточняет цель звонка и собирает контекст для {customer_name}."
        kb_refs = []
        confidence = 0.68
        intent = "clarification"

    if disclosure_required and not reply_text.startswith(_voice_disclosure_prefix(language)):
        reply_text = f"{_voice_disclosure_prefix(language)}{reply_text}"
    return {
        "language": language,
        "intent": intent,
        "reply_text": reply_text,
        "confidence": confidence,
        "needs_handoff": False,
        "handoff_reason": None,
        "case_action": "keep_open",
        "kb_refs": kb_refs,
        "summary_text": summary_text,
        "model": model,
        "latency_ms": 1,
    }


def start_voice_session(session_id: str, payload: VoiceAIStartIn) -> VoiceAIStartOut:
    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == session_id)
        ).scalar_one_or_none()
        if voice_session is None:
            raise HTTPException(status_code=404, detail="Voice AI session not found")
        interaction = session.execute(
            select(Interaction).where(Interaction.interaction_id == payload.interaction_id)
        ).scalar_one_or_none()
        if interaction is None:
            raise HTTPException(status_code=404, detail="Interaction not found")

        language = str(payload.language_hint or voice_session.language or "ru").strip() or "ru"
        customer_id = payload.customer_id or _resolve_or_create_voice_customer_id(
            session,
            interaction=interaction,
            call_id=payload.call_id,
        )
        customer = None
        if customer_id:
            customer = session.execute(
                select(Customer).where(Customer.customer_id == customer_id)
            ).scalar_one_or_none()
        ai_session, created = _ensure_voice_ai_session(
            session,
            voice_session=voice_session,
            interaction=interaction,
            customer_id=customer_id,
            language=language,
            agent_profile=payload.agent_profile or voice_session.agent_profile,
        )
        now = utc_now_iso()
        voice_session.language = language
        voice_session.status = "greeting"
        voice_session.updated_at = now
        if customer and not interaction.customer_id:
            interaction.customer_id = customer.customer_id
            interaction.updated_at = now
        session.commit()
        if created:
            _append_timeline(
                interaction.interaction_id,
                "ai.session_started",
                {
                    "call_id": payload.call_id,
                    "voice_session_id": voice_session.session_id,
                    "ai_session_id": ai_session.session_id,
                    "language": language,
                },
            )
        return VoiceAIStartOut(
            session_id=ai_session.session_id,
            language=language,
            greeting_text=_voice_greeting(language),
            disclosure_required=voice_session.disclosure_played_at is None,
        )
    finally:
        session.close()


def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDecisionOut:
    app = _app()
    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == session_id)
        ).scalar_one_or_none()
        if voice_session is None:
            raise HTTPException(status_code=404, detail="Voice AI session not found")
        interaction = session.execute(
            select(Interaction).where(Interaction.interaction_id == payload.interaction_id)
        ).scalar_one_or_none()
        if interaction is None:
            raise HTTPException(status_code=404, detail="Interaction not found")

        customer_id = _resolve_or_create_voice_customer_id(
            session,
            interaction=interaction,
            call_id=payload.call_id,
        )
        customer = None
        if customer_id:
            customer = session.execute(
                select(Customer).where(Customer.customer_id == customer_id)
            ).scalar_one_or_none()

        ai_session, _ = _ensure_voice_ai_session(
            session,
            voice_session=voice_session,
            interaction=interaction,
            customer_id=customer_id,
            language=str(payload.language or voice_session.language or "ru").strip() or "ru",
            agent_profile=voice_session.agent_profile,
        )
        now = utc_now_iso()
        voice_session.last_user_utterance_at = now
        voice_session.status = "thinking"
        voice_session.updated_at = now
        ai_session.updated_at = now
        ai_session.call_id = payload.call_id
        ai_session.interaction_id = interaction.interaction_id
        ai_session.customer_id = customer_id
        ai_session.language = str(payload.language or voice_session.language or "ru").strip() or "ru"

        _record_voice_ai_turn(
            session,
            ai_session_id=ai_session.session_id,
            interaction_id=interaction.interaction_id,
            role="user",
            source_type="voice_asr",
            text=payload.transcript_text,
            payload={
                "voice_session_id": payload.voice_session_id,
                "call_id": payload.call_id,
                "sequence_no": payload.sequence_no,
                "barge_in": payload.barge_in,
                "metadata": payload.metadata,
            },
        )
        if payload.barge_in:
            last_assistant_segment = session.execute(
                select(VoiceTranscriptSegmentRow)
                .where(VoiceTranscriptSegmentRow.session_id == voice_session.session_id)
                .where(VoiceTranscriptSegmentRow.speaker == "assistant")
                .order_by(VoiceTranscriptSegmentRow.id.desc())
                .limit(1)
            ).scalar_one_or_none()
            if last_assistant_segment:
                last_assistant_segment.barge_in_interrupted = True
        transcript_window = _voice_recent_segments(
            session,
            voice_session_id=voice_session.session_id,
            limit=_voice_max_context_segments(),
        )
        timeline_window = _voice_recent_timeline(
            session,
            interaction_id=interaction.interaction_id,
        )
        kb_results = app._kb_search(
            session,
            payload.transcript_text,
            language=ai_session.language,
        )
        disclosure_required = voice_session.disclosure_played_at is None
        decision = _voice_decision(
            language=ai_session.language or "ru",
            customer=customer,
            interaction=interaction,
            transcript_text=payload.transcript_text,
            transcript_window=transcript_window,
            kb_results=kb_results,
            disclosure_required=disclosure_required,
        )
        _record_voice_ai_turn(
            session,
            ai_session_id=ai_session.session_id,
            interaction_id=interaction.interaction_id,
            role="assistant",
            source_type="voice_policy",
            text=decision["reply_text"] or decision["summary_text"] or decision["handoff_reason"],
            payload={
                "voice_session_id": payload.voice_session_id,
                "call_id": payload.call_id,
                "kb_refs": decision["kb_refs"],
                "timeline_window": timeline_window,
                "context_segments": [
                    {
                        "speaker": segment.speaker,
                        "text": segment.text,
                        "sequence_no": segment.sequence_no,
                    }
                    for segment in transcript_window
                ],
                "decision": decision,
            },
            model=decision["model"],
            latency_ms=decision["latency_ms"],
        )
        voice_session.language = decision["language"]
        voice_session.updated_at = now
        voice_session.status = "handoff_requested" if decision["needs_handoff"] else "active"
        voice_session.handoff_reason = decision["handoff_reason"]
        ai_session.status = "handoff_required" if decision["needs_handoff"] else "active"
        ai_session.handoff_reason = decision["handoff_reason"]
        ai_session.summary_text = decision["summary_text"] or decision["reply_text"] or ai_session.summary_text
        ai_session.updated_at = now
        session.commit()
        _append_timeline(
            interaction.interaction_id,
            "ai.handoff_requested" if decision["needs_handoff"] else "ai.reply_generated",
            {
                "call_id": payload.call_id,
                "voice_session_id": payload.voice_session_id,
                "ai_session_id": ai_session.session_id,
                "sequence_no": payload.sequence_no,
                "confidence": decision["confidence"],
                "kb_refs": decision["kb_refs"],
                "handoff_reason": decision["handoff_reason"],
            },
        )
        return VoiceAITurnDecisionOut(
            language=decision["language"],
            intent=decision["intent"],
            reply_text=decision["reply_text"],
            confidence=decision["confidence"],
            needs_handoff=decision["needs_handoff"],
            handoff_reason=decision["handoff_reason"],
            case_action=decision["case_action"],
            kb_refs=decision["kb_refs"],
            summary_text=decision["summary_text"],
            model=decision["model"],
            latency_ms=decision["latency_ms"],
            status="handoff_requested" if decision["needs_handoff"] else "active",
        )
    except HTTPException:
        raise
    except Exception as exc:
        try:
            session.rollback()
        finally:
            pass
        raise HTTPException(status_code=502, detail=f"Voice AI turn failed: {exc}") from exc
    finally:
        session.close()


def close_voice_session(session_id: str) -> dict[str, Any]:
    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == session_id)
        ).scalar_one_or_none()
        if voice_session is None:
            raise HTTPException(status_code=404, detail="Voice AI session not found")
        now = utc_now_iso()
        voice_session.status = "completed" if voice_session.status != "error" else "error"
        voice_session.ended_at = voice_session.ended_at or now
        voice_session.updated_at = now
        if str(voice_session.ai_session_id or "").strip():
            ai_session = session.execute(
                select(AISessionRow).where(AISessionRow.session_id == voice_session.ai_session_id)
            ).scalar_one_or_none()
            if ai_session:
                ai_session.status = "closed" if ai_session.status != "error" else "error"
                ai_session.closed_at = now
                ai_session.updated_at = now
        session.commit()
        return {
            "ok": True,
            "voice_session_id": voice_session.session_id,
            "ai_session_id": voice_session.ai_session_id,
            "status": voice_session.status,
        }
    finally:
        session.close()