import json
import time
import importlib
from types import SimpleNamespace

import httpx
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

ai_module = importlib.import_module("services.ai_orchestrator_service.app")
ai_app = ai_module.app
voice_module = importlib.import_module("services.ai_orchestrator_service.voice")
voice_config_module = importlib.import_module("services.ai_orchestrator_service.voice_name_config")
ai_operator_config_module = importlib.import_module("services.shared.ai_operator_config")
voice_tts_config_module = importlib.import_module("services.shared.voice_tts_config")
from services.interaction_service.app import app as interaction_app
from services.shared.core import new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.models import VoiceAIStartIn, VoiceAITurnIn
from services.shared.sql_models import (
    AIJobRow,
    AIOperatorSettingsRow,
    AISessionRow,
    AITurnRow,
    AsteriskCallLinkRow,
    Customer,
    CustomerExternalIdentity,
    Interaction,
    InteractionTimeline,
    KBArticleRow,
    KBCategoryRow,
    TelegramMessageRow,
    TelegramThreadRow,
    VoiceNameCollectionSettingsRow,
    VoiceTTSSettingsRow,
    VoiceAISessionRow,
    VoiceTranscriptSegmentRow,
    WhatsAppThreadRow,
)
from services.telegram_adapter_service import app as telegram_module
from services.telegram_adapter_service.app import app as telegram_app


def admin_headers():
    return {"X-User": "admin", "X-Role": "admin"}


def operator_headers(user="operator"):
    return {"X-User": user, "X-Role": "operator"}


@pytest.fixture(autouse=True)
def reset_voice_name_collection_settings():
    session = get_session()
    try:
        row = session.execute(select(VoiceNameCollectionSettingsRow)).scalar_one_or_none()
        if row is not None:
            session.delete(row)
            session.commit()
    finally:
        session.close()
    yield
    session = get_session()
    try:
        row = session.execute(select(VoiceNameCollectionSettingsRow)).scalar_one_or_none()
        if row is not None:
            session.delete(row)
            session.commit()
    finally:
        session.close()


@pytest.fixture(autouse=True)
def reset_voice_tts_settings():
    session = get_session()
    try:
        row = session.execute(select(VoiceTTSSettingsRow)).scalar_one_or_none()
        if row is not None:
            session.delete(row)
            session.commit()
    finally:
        session.close()
    yield
    session = get_session()
    try:
        row = session.execute(select(VoiceTTSSettingsRow)).scalar_one_or_none()
        if row is not None:
            session.delete(row)
            session.commit()
    finally:
        session.close()


@pytest.fixture(autouse=True)
def reset_ai_operator_settings():
    session = get_session()
    try:
        row = session.execute(select(AIOperatorSettingsRow)).scalar_one_or_none()
        if row is not None:
            session.delete(row)
            session.commit()
    finally:
        session.close()
    yield
    session = get_session()
    try:
        row = session.execute(select(AIOperatorSettingsRow)).scalar_one_or_none()
        if row is not None:
            session.delete(row)
            session.commit()
    finally:
        session.close()


def _u(value: str) -> str:
    return value.encode("ascii").decode("unicode_escape")


def seed_voice_downstream_session(
    *,
    marker: str,
    name_status: str,
    name_value: str | None = None,
    name_source: str = "none",
    customer_display_name: str | None = None,
    caller_number: str | None = None,
    caller_name: str = "Voice Caller",
) -> dict[str, str]:
    session = get_session()
    try:
        now = utc_now_iso()
        resolved_caller_number = caller_number or f"+7{abs(hash(marker)) % 10_000_000_000:010d}"
        interaction_id = f"{marker}_int"
        customer_id = f"{marker}_cus"
        call_id = f"{marker}_call"
        session_id = f"{marker}_avs"
        linked_id = f"{marker}_linked"
        session.add(
            Customer(
                customer_id=customer_id,
                display_name=customer_display_name or resolved_caller_number,
                phones_json=f'["{resolved_caller_number}"]',
                preferred_phone=resolved_caller_number,
                tags_json='["voice"]',
                created_at=now,
            )
        )
        session.add(
            CustomerExternalIdentity(
                identity_id=f"{marker}_cei",
                customer_id=customer_id,
                channel="voice",
                external_subject=resolved_caller_number,
                display_name_snapshot=caller_name,
                created_at=now,
                updated_at=now,
            )
        )
        session.add(
            Interaction(
                interaction_id=interaction_id,
                channel="voice",
                subject=f"Voice downstream {marker}",
                customer_id=customer_id,
                queue_id="que_voice_support",
                priority=3,
                status="open",
                assigned_to=None,
                created_at=now,
                updated_at=now,
            )
        )
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=linked_id,
                queue_code="voice_support",
                queue_id="que_voice_support",
                interaction_id=interaction_id,
                caller_number=resolved_caller_number,
                caller_name=caller_name,
                status="active",
                telephony_status="connected",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name="PJSIP/1001-000001",
                started_at=now,
                connected_at=now,
                ended_at=None,
                updated_at=now,
                voice_start_language="ru",
                customer_name_status=name_status,
                customer_name_value=name_value,
                customer_name_source=name_source,
                customer_name_resolved_at=now,
                voice_session_id=session_id,
            )
        )
        session.add(
            VoiceAISessionRow(
                session_id=session_id,
                call_id=call_id,
                linked_id=linked_id,
                interaction_id=interaction_id,
                customer_id=customer_id,
                queue_id="que_voice_support",
                ai_session_id=None,
                agent_profile="voice_support",
                language="ru",
                asr_provider="openai",
                tts_provider="yandex",
                status="active",
                handoff_reason=None,
                handoff_target_queue_id="que_voice_support",
                disclosure_played_at=now,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=now,
                updated_at=now,
                ended_at=None,
                voice_start_language="ru",
                customer_name_status=name_status,
                customer_name_value=name_value,
                customer_name_source=name_source,
                customer_name_resolved_at=now,
            )
        )
        session.commit()
        return {
            "interaction_id": interaction_id,
            "customer_id": customer_id,
            "call_id": call_id,
            "session_id": session_id,
            "caller_number": resolved_caller_number,
        }
    finally:
        session.close()


def seed_voice_start_session(
    *,
    marker: str,
    language: str = "ru",
    customer_display_name: str | None = None,
    caller_number: str | None = None,
    caller_name: str = "Voice Caller",
    next_queue_code: str = "voice_support",
    next_queue_id: str = "que_voice_support",
) -> dict[str, str]:
    session = get_session()
    try:
        now = utc_now_iso()
        resolved_caller_number = caller_number or f"+7{abs(hash(f'{marker}_start')) % 10_000_000_000:010d}"
        interaction_id = f"{marker}_int"
        customer_id = f"{marker}_cus"
        call_id = f"{marker}_call"
        session_id = f"{marker}_avs"
        linked_id = f"{marker}_linked"

        if customer_display_name is not None:
            session.add(
                Customer(
                    customer_id=customer_id,
                    display_name=customer_display_name,
                    phones_json=f'["{resolved_caller_number}"]',
                    preferred_phone=resolved_caller_number,
                    tags_json='["voice"]',
                    created_at=now,
                )
            )
            session.add(
                CustomerExternalIdentity(
                    identity_id=f"{marker}_cei",
                    customer_id=customer_id,
                    channel="voice",
                    external_subject=resolved_caller_number,
                    display_name_snapshot=caller_name,
                    created_at=now,
                    updated_at=now,
                )
            )

        session.add(
            Interaction(
                interaction_id=interaction_id,
                channel="voice",
                subject=f"Voice start {marker}",
                customer_id=customer_id if customer_display_name is not None else None,
                queue_id=f"que_voice_start_{language}",
                priority=3,
                status="open",
                assigned_to=None,
                created_at=now,
                updated_at=now,
            )
        )
        session.add(
            AsteriskCallLinkRow(
                call_id=call_id,
                linked_id=linked_id,
                queue_code=f"voice_start_{language}",
                queue_id=f"que_voice_start_{language}",
                interaction_id=interaction_id,
                caller_number=resolved_caller_number,
                caller_name=caller_name,
                status="active",
                telephony_status="connected",
                claimed_by_user=None,
                claimed_at=None,
                operator_extension=None,
                channel_name="PJSIP/1002-000002",
                started_at=now,
                connected_at=now,
                ended_at=None,
                updated_at=now,
                voice_session_id=session_id,
            )
        )
        session.add(
            VoiceAISessionRow(
                session_id=session_id,
                call_id=call_id,
                linked_id=linked_id,
                interaction_id=interaction_id,
                customer_id=customer_id if customer_display_name is not None else None,
                queue_id=f"que_voice_start_{language}",
                ai_session_id=None,
                agent_profile="voice_start",
                language=language,
                asr_provider="openai",
                tts_provider="yandex",
                status="active",
                handoff_reason=None,
                handoff_target_queue_id=next_queue_id,
                disclosure_played_at=None,
                last_user_utterance_at=None,
                last_ai_reply_at=None,
                started_at=now,
                updated_at=now,
                ended_at=None,
                voice_start_language=language,
                customer_name_status=None,
                customer_name_value=None,
                customer_name_source=None,
                customer_name_resolved_at=None,
            )
        )
        session.commit()
        return {
            "interaction_id": interaction_id,
            "customer_id": customer_id,
            "call_id": call_id,
            "session_id": session_id,
            "linked_id": linked_id,
            "language": language,
            "next_queue_code": next_queue_code,
            "next_queue_id": next_queue_id,
        }
    finally:
        session.close()


def patch_interaction_request(monkeypatch):
    def fake_request(method: str, path: str, *, payload: dict | None = None) -> dict:
        session = get_session()
        try:
            interaction_id = path.split("/")[2]
            interaction = session.execute(
                select(Interaction).where(Interaction.interaction_id == interaction_id)
            ).scalar_one()
            if path.endswith("/assign"):
                interaction.assigned_to = payload["assignee"]
                interaction.status = "in_progress"
            elif path.endswith("/status"):
                interaction.status = payload["status"]
            elif path.endswith("/escalate"):
                interaction.status = "escalated"
                interaction.queue_id = payload["target_queue_id"]
            interaction.updated_at = telegram_module.utc_now_iso()
            session.commit()
            return {
                "interaction_id": interaction.interaction_id,
                "status": interaction.status,
                "assigned_to": interaction.assigned_to,
                "queue_id": interaction.queue_id,
            }
        finally:
            session.close()

    monkeypatch.setattr(telegram_module, "_interaction_request", fake_request)


def patch_ai_internal_calls(monkeypatch, telegram_client: TestClient, interaction_client: TestClient):
    def fake_telegram_request(method: str, path: str, *, payload: dict | None = None) -> dict:
        response = telegram_client.request(method, path, json=payload, headers=admin_headers())
        response.raise_for_status()
        return response.json()

    def fake_interaction_request(method: str, path: str, *, payload: dict | None = None) -> dict:
        response = interaction_client.request(method, path, json=payload, headers=admin_headers())
        response.raise_for_status()
        return response.json()

    monkeypatch.setattr(ai_module, "_telegram_request", fake_telegram_request)
    monkeypatch.setattr(ai_module, "_interaction_request", fake_interaction_request)


def seed_kb_article(
    title: str,
    body: str,
    tag: str,
    *,
    language: str = "ru",
    article_group_id: str | None = None,
) -> dict[str, str]:
    session = get_session()
    try:
        now = utc_now_iso()
        category_id = new_id("kbc")
        article_id = new_id("kba")
        resolved_group_id = article_group_id or article_id
        session.add(
            KBCategoryRow(
                category_id=category_id,
                name="Telegram AI",
                description="AI test category",
                created_at=now,
            )
        )
        session.add(
            KBArticleRow(
                article_id=article_id,
                category_id=category_id,
                article_group_id=resolved_group_id,
                language=language,
                title=title,
                body=body,
                tags_json=f'["{tag}"]',
                created_at=now,
                updated_at=now,
            )
        )
        session.commit()
        return {"article_id": article_id, "article_group_id": resolved_group_id}
    finally:
        session.close()


def create_inbound_thread(telegram_client: TestClient, chat_id: str, text: str) -> dict:
    response = telegram_client.post(
        "/integrations/telegram/webhook",
        json={
            "chat_id": chat_id,
            "text": text,
            "payload": {"telegram_user_id": f"user-{chat_id}", "username": f"user_{chat_id}"},
        },
    )
    assert response.status_code == 200
    return response.json()


def fetch_ai_summary(telegram_client: TestClient, thread_id: str, headers: dict | None = None):
    return telegram_client.get(
        f"/integrations/telegram/threads/{thread_id}/ai-summary",
        headers=headers or admin_headers(),
    )


def deliver_latest_pending_message(monkeypatch, telegram_client: TestClient, thread_id: str, external_id: int = 7001) -> None:
    messages = telegram_client.get(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    message_id = messages.json()[-1]["message_id"]
    monkeypatch.setattr(
        telegram_module,
        "_send_telegram_message",
        lambda chat_id, text: {"ok": True, "result": {"message_id": external_id, "chat": {"id": chat_id}, "text": text}},
    )
    telegram_module._deliver_pending_telegram_reply(message_id)


def seed_ai_analytics_dataset(marker: str) -> dict[str, str]:
    session = get_session()
    try:
        window_from = "2040-01-01T00:00:00+00:00"
        window_to = "2040-01-03T00:00:00+00:00"
        queue_tg = f"{marker}_queue_tg"
        queue_wa = f"{marker}_queue_wa"
        queue_thread = f"{marker}_queue_thread"

        interaction_tg = f"{marker}_int_tg"
        interaction_wa = f"{marker}_int_wa"
        interaction_tg_human = f"{marker}_int_tg_human"
        interaction_thread = f"{marker}_int_thread"

        thread_tg = f"{marker}_thread_tg"
        thread_wa = f"{marker}_thread_wa"
        thread_tg_human = f"{marker}_thread_tg_human"
        thread_wa_thread_queue = f"{marker}_thread_wa_thread_queue"

        session_tg = f"{marker}_sess_tg"
        session_wa = f"{marker}_sess_wa"
        session_tg_human = f"{marker}_sess_tg_human"
        session_thread_queue = f"{marker}_sess_thread_queue"

        session.add_all(
            [
                Interaction(
                    interaction_id=interaction_tg,
                    channel="telegram",
                    subject=f"{marker} contained telegram",
                    customer_id=f"{marker}_cust_1",
                    queue_id=queue_tg,
                    priority=3,
                    status="closed",
                    assigned_to=None,
                    created_at="2040-01-01T09:00:00+00:00",
                    updated_at="2040-01-01T09:40:00+00:00",
                ),
                Interaction(
                    interaction_id=interaction_wa,
                    channel="whatsapp",
                    subject=f"{marker} whatsapp handoff",
                    customer_id=f"{marker}_cust_2",
                    queue_id=queue_wa,
                    priority=3,
                    status="in_progress",
                    assigned_to="agent_whatsapp",
                    created_at="2040-01-01T13:00:00+00:00",
                    updated_at="2040-01-01T13:30:00+00:00",
                ),
                Interaction(
                    interaction_id=interaction_tg_human,
                    channel="telegram",
                    subject=f"{marker} telegram with operator",
                    customer_id=f"{marker}_cust_3",
                    queue_id=queue_tg,
                    priority=3,
                    status="closed",
                    assigned_to="agent_telegram",
                    created_at="2040-01-02T10:00:00+00:00",
                    updated_at="2040-01-02T10:40:00+00:00",
                ),
                Interaction(
                    interaction_id=interaction_thread,
                    channel="whatsapp",
                    subject=f"{marker} thread queue fallback",
                    customer_id=f"{marker}_cust_4",
                    queue_id=None,
                    priority=3,
                    status="closed",
                    assigned_to=None,
                    created_at="2040-01-02T15:00:00+00:00",
                    updated_at="2040-01-02T15:15:00+00:00",
                ),
                TelegramThreadRow(
                    thread_id=thread_tg,
                    chat_id=f"{marker}_chat_tg",
                    interaction_id=interaction_tg,
                    telegram_user_id=f"{marker}_tg_user",
                    username=f"{marker}_tg",
                    display_name="AI Telegram",
                    queue_id=queue_tg,
                    status="closed",
                    claimed_by_user=None,
                    claimed_at=None,
                    ai_session_id=session_tg,
                    ai_state="closed",
                    ai_handoff_reason=None,
                    ai_last_model_at="2040-01-01T09:15:00+00:00",
                    last_message_at="2040-01-01T09:16:00+00:00",
                    last_message_preview="contained",
                    created_at="2040-01-01T09:00:00+00:00",
                    updated_at="2040-01-01T09:16:00+00:00",
                ),
                WhatsAppThreadRow(
                    thread_id=thread_wa,
                    chat_id=f"{marker}_chat_wa",
                    interaction_id=interaction_wa,
                    whatsapp_user_id=f"{marker}_wa_user",
                    phone_number="+77000000001",
                    display_name="AI WhatsApp",
                    queue_id=queue_wa,
                    is_group=False,
                    status="in_progress",
                    claimed_by_user="supervisor_wa",
                    claimed_at="2040-01-01T13:12:00+00:00",
                    ai_session_id=session_wa,
                    ai_state="human_owned",
                    ai_handoff_reason="requested by customer",
                    ai_last_model_at="2040-01-01T13:10:00+00:00",
                    last_message_at="2040-01-01T13:11:00+00:00",
                    last_message_preview="handoff",
                    created_at="2040-01-01T13:00:00+00:00",
                    updated_at="2040-01-01T13:12:00+00:00",
                ),
                TelegramThreadRow(
                    thread_id=thread_tg_human,
                    chat_id=f"{marker}_chat_tg_human",
                    interaction_id=interaction_tg_human,
                    telegram_user_id=f"{marker}_tg_user_human",
                    username=f"{marker}_tg_human",
                    display_name="AI Telegram Human",
                    queue_id=queue_tg,
                    status="closed",
                    claimed_by_user=None,
                    claimed_at=None,
                    ai_session_id=session_tg_human,
                    ai_state="closed",
                    ai_handoff_reason=None,
                    ai_last_model_at="2040-01-02T10:10:00+00:00",
                    last_message_at="2040-01-02T10:11:00+00:00",
                    last_message_preview="closed with operator",
                    created_at="2040-01-02T10:00:00+00:00",
                    updated_at="2040-01-02T10:11:00+00:00",
                ),
                WhatsAppThreadRow(
                    thread_id=thread_wa_thread_queue,
                    chat_id=f"{marker}_chat_wa_thread",
                    interaction_id=interaction_thread,
                    whatsapp_user_id=f"{marker}_wa_thread_user",
                    phone_number="+77000000002",
                    display_name="AI WhatsApp Thread Queue",
                    queue_id=queue_thread,
                    is_group=False,
                    status="closed",
                    claimed_by_user=None,
                    claimed_at=None,
                    ai_session_id=session_thread_queue,
                    ai_state="closed",
                    ai_handoff_reason=None,
                    ai_last_model_at="2040-01-02T15:10:00+00:00",
                    last_message_at="2040-01-02T15:11:00+00:00",
                    last_message_preview="thread fallback",
                    created_at="2040-01-02T15:00:00+00:00",
                    updated_at="2040-01-02T15:11:00+00:00",
                ),
                AISessionRow(
                    session_id=session_tg,
                    channel="telegram",
                    thread_id=thread_tg,
                    interaction_id=interaction_tg,
                    customer_id=f"{marker}_cust_1",
                    agent_profile="telegram_support",
                    language="ru",
                    status="closed",
                    summary_text="",
                    last_user_message_id=None,
                    last_ai_message_id=None,
                    handoff_reason=None,
                    created_at="2040-01-01T09:00:00+00:00",
                    updated_at="2040-01-01T09:18:00+00:00",
                    closed_at="2040-01-01T09:18:00+00:00",
                ),
                AISessionRow(
                    session_id=session_wa,
                    channel="whatsapp",
                    thread_id=thread_wa,
                    interaction_id=interaction_wa,
                    customer_id=f"{marker}_cust_2",
                    agent_profile="whatsapp_support",
                    language="ru",
                    status="human_owned",
                    summary_text="",
                    last_user_message_id=None,
                    last_ai_message_id=None,
                    handoff_reason="requested by customer",
                    created_at="2040-01-01T13:00:00+00:00",
                    updated_at="2040-01-01T13:12:00+00:00",
                    closed_at=None,
                ),
                AISessionRow(
                    session_id=session_tg_human,
                    channel="telegram",
                    thread_id=thread_tg_human,
                    interaction_id=interaction_tg_human,
                    customer_id=f"{marker}_cust_3",
                    agent_profile="telegram_support",
                    language="ru",
                    status="closed",
                    summary_text="",
                    last_user_message_id=None,
                    last_ai_message_id=None,
                    handoff_reason=None,
                    created_at="2040-01-02T10:00:00+00:00",
                    updated_at="2040-01-02T10:12:00+00:00",
                    closed_at="2040-01-02T10:12:00+00:00",
                ),
                AISessionRow(
                    session_id=session_thread_queue,
                    channel="whatsapp",
                    thread_id=thread_wa_thread_queue,
                    interaction_id=interaction_thread,
                    customer_id=f"{marker}_cust_4",
                    agent_profile="whatsapp_support",
                    language="ru",
                    status="closed",
                    summary_text="",
                    last_user_message_id=None,
                    last_ai_message_id=None,
                    handoff_reason=None,
                    created_at="2040-01-02T15:00:00+00:00",
                    updated_at="2040-01-02T15:12:00+00:00",
                    closed_at="2040-01-02T15:12:00+00:00",
                ),
                AITurnRow(
                    turn_id=f"{marker}_turn_1",
                    session_id=session_tg,
                    thread_id=thread_tg,
                    interaction_id=interaction_tg,
                    role="assistant",
                    source_type="model",
                    text="reply 1",
                    payload_json="{}",
                    model="stub",
                    finish_reason="stop",
                    latency_ms=100,
                    created_at="2040-01-01T09:05:00+00:00",
                ),
                AITurnRow(
                    turn_id=f"{marker}_turn_2",
                    session_id=session_tg,
                    thread_id=thread_tg,
                    interaction_id=interaction_tg,
                    role="assistant",
                    source_type="model",
                    text="reply 2",
                    payload_json="{}",
                    model="stub",
                    finish_reason="stop",
                    latency_ms=200,
                    created_at="2040-01-01T09:10:00+00:00",
                ),
                AITurnRow(
                    turn_id=f"{marker}_turn_3",
                    session_id=session_wa,
                    thread_id=thread_wa,
                    interaction_id=interaction_wa,
                    role="assistant",
                    source_type="model",
                    text="reply 3",
                    payload_json="{}",
                    model="stub",
                    finish_reason="stop",
                    latency_ms=900,
                    created_at="2040-01-01T13:05:00+00:00",
                ),
                AITurnRow(
                    turn_id=f"{marker}_turn_4",
                    session_id=session_wa,
                    thread_id=thread_wa,
                    interaction_id=interaction_wa,
                    role="assistant",
                    source_type="kb",
                    text="ignored",
                    payload_json="{}",
                    model="stub",
                    finish_reason="stop",
                    latency_ms=50,
                    created_at="2040-01-01T13:06:00+00:00",
                ),
                AITurnRow(
                    turn_id=f"{marker}_turn_5",
                    session_id=session_thread_queue,
                    thread_id=thread_wa_thread_queue,
                    interaction_id=interaction_thread,
                    role="assistant",
                    source_type="model",
                    text="reply 5",
                    payload_json="{}",
                    model="stub",
                    finish_reason="stop",
                    latency_ms=400,
                    created_at="2040-01-02T15:05:00+00:00",
                ),
                AITurnRow(
                    turn_id=f"{marker}_turn_6",
                    session_id=session_thread_queue,
                    thread_id=thread_wa_thread_queue,
                    interaction_id=interaction_thread,
                    role="user",
                    source_type="customer",
                    text="ignored user turn",
                    payload_json="{}",
                    model=None,
                    finish_reason=None,
                    latency_ms=999,
                    created_at="2040-01-02T15:04:00+00:00",
                ),
            ]
        )
        session.commit()
        return {
            "from_ts": window_from,
            "to_ts": window_to,
            "queue_tg": queue_tg,
            "queue_wa": queue_wa,
            "queue_thread": queue_thread,
        }
    finally:
        session.close()


def cleanup_ai_analytics_dataset(marker: str) -> None:
    session = get_session()
    try:
        session.query(AITurnRow).filter(AITurnRow.turn_id.like(f"{marker}_turn_%")).delete(synchronize_session=False)
        session.query(AISessionRow).filter(AISessionRow.session_id.like(f"{marker}_sess_%")).delete(synchronize_session=False)
        session.query(TelegramThreadRow).filter(TelegramThreadRow.thread_id.like(f"{marker}_thread_%")).delete(synchronize_session=False)
        session.query(WhatsAppThreadRow).filter(WhatsAppThreadRow.thread_id.like(f"{marker}_thread_%")).delete(synchronize_session=False)
        session.query(Interaction).filter(Interaction.interaction_id.like(f"{marker}_int_%")).delete(synchronize_session=False)
        session.commit()
    finally:
        session.close()


def seed_voice_name_flow_analytics_dataset(marker: str) -> dict[str, str]:
    session = get_session()
    try:
        window_from = "2041-02-01T00:00:00+00:00"
        window_to = "2041-02-03T00:00:00+00:00"
        queue_ru = f"{marker}_queue_ru"
        queue_kz = f"{marker}_queue_kz"
        queue_support = f"{marker}_queue_support"

        sessions = [
            {
                "voice_session_id": f"{marker}_voice_start",
                "ai_session_id": f"{marker}_ai_start",
                "call_id": f"{marker}_call_start",
                "interaction_id": f"{marker}_int_start",
                "started_at": "2041-02-01T09:00:00+00:00",
                "queue_id": queue_ru,
                "interaction_queue_id": queue_ru,
                "language": "ru",
                "voice_start_language": "ru",
                "status": "closed",
                "handoff_reason": None,
                "call_queue_id": queue_ru,
                "call_voice_start_language": "ru",
                "claimed_by_user": None,
                "operator_extension": None,
                "customer_name_source": "voice_start",
                "turns": [
                    {
                        "turn_id": f"{marker}_turn_start_1",
                        "created_at": "2041-02-01T09:02:00+00:00",
                        "decision": {
                            "customer_name_status": "name_obtained",
                            "customer_name_value": "Алия",
                            "customer_name_source": "voice_start",
                        },
                    },
                ],
            },
            {
                "voice_session_id": f"{marker}_voice_downstream",
                "ai_session_id": f"{marker}_ai_downstream",
                "call_id": f"{marker}_call_downstream",
                "interaction_id": f"{marker}_int_downstream",
                "started_at": "2041-02-01T10:00:00+00:00",
                "queue_id": None,
                "interaction_queue_id": None,
                "language": "kz",
                "voice_start_language": None,
                "status": "human_owned",
                "handoff_reason": "requested_human",
                "call_queue_id": queue_kz,
                "call_voice_start_language": "kz",
                "claimed_by_user": "operator_kz",
                "operator_extension": "2101",
                "customer_name_source": "voice_followup",
                "turns": [
                    {
                        "turn_id": f"{marker}_turn_down_1",
                        "created_at": "2041-02-01T10:01:00+00:00",
                        "decision": {
                            "customer_name_status": "name_followup_required",
                            "customer_name_source": "voice_start",
                            "metadata": {
                                "customer_name_status": "name_followup_required",
                                "customer_name_source": "voice_start",
                            },
                        },
                    },
                    {
                        "turn_id": f"{marker}_turn_down_2",
                        "created_at": "2041-02-01T10:03:00+00:00",
                        "decision": {
                            "customer_name_status": "name_obtained",
                            "customer_name_value": "Нурлан",
                            "customer_name_source": "voice_followup",
                            "metadata": {
                                "customer_name_status": "name_obtained",
                                "customer_name_value": "Нурлан",
                                "customer_name_source": "voice_followup",
                            },
                        },
                    },
                ],
            },
            {
                "voice_session_id": f"{marker}_voice_followup",
                "ai_session_id": f"{marker}_ai_followup",
                "call_id": f"{marker}_call_followup",
                "interaction_id": f"{marker}_int_followup",
                "started_at": "2041-02-02T11:00:00+00:00",
                "queue_id": None,
                "interaction_queue_id": queue_support,
                "language": None,
                "voice_start_language": None,
                "status": "handoff_required",
                "handoff_reason": "requested_human",
                "call_queue_id": None,
                "call_voice_start_language": None,
                "claimed_by_user": "operator_support",
                "operator_extension": None,
                "customer_name_source": "voice_start",
                "with_call_row": False,
                "turns": [
                    {
                        "turn_id": f"{marker}_turn_followup_1",
                        "created_at": "2041-02-02T11:02:00+00:00",
                        "decision": {
                            "customer_name_status": "name_followup_required",
                            "customer_name_source": "voice_start",
                            "metadata": {
                                "customer_name_status": "name_followup_required",
                                "customer_name_source": "voice_start",
                            },
                        },
                    },
                ],
            },
            {
                "voice_session_id": f"{marker}_voice_missing",
                "ai_session_id": None,
                "call_id": f"{marker}_call_missing",
                "interaction_id": f"{marker}_int_missing",
                "started_at": "2041-02-02T12:00:00+00:00",
                "queue_id": queue_ru,
                "interaction_queue_id": queue_ru,
                "language": "ru",
                "voice_start_language": "ru",
                "status": "closed",
                "handoff_reason": None,
                "call_queue_id": queue_ru,
                "call_voice_start_language": "ru",
                "claimed_by_user": None,
                "operator_extension": None,
                "customer_name_source": "voice_start",
                "turns": [],
                "timeline_event": {
                    "timestamp": "2041-02-02T12:01:00+00:00",
                    "metadata": {
                        "customer_name_status": "name_not_obtained",
                        "customer_name_source": "voice_start",
                        "language": "ru",
                    },
                },
            },
            {
                "voice_session_id": f"{marker}_voice_manual",
                "ai_session_id": f"{marker}_ai_manual",
                "call_id": f"{marker}_call_manual",
                "interaction_id": f"{marker}_int_manual",
                "started_at": "2041-02-02T13:00:00+00:00",
                "queue_id": queue_ru,
                "interaction_queue_id": queue_ru,
                "language": "ru",
                "voice_start_language": "ru",
                "status": "human_owned",
                "handoff_reason": "requested_human",
                "call_queue_id": queue_ru,
                "call_voice_start_language": "ru",
                "claimed_by_user": "operator_manual",
                "operator_extension": "2201",
                "customer_name_source": "manual",
                "turns": [
                    {
                        "turn_id": f"{marker}_turn_manual_1",
                        "created_at": "2041-02-02T13:02:00+00:00",
                        "decision": {
                            "customer_name_status": "name_not_obtained",
                            "customer_name_source": "voice_followup",
                            "metadata": {
                                "customer_name_status": "name_not_obtained",
                                "customer_name_source": "voice_followup",
                            },
                        },
                    },
                ],
            },
        ]

        entities = []
        for item in sessions:
            interaction_id = item["interaction_id"]
            call_id = item["call_id"]
            voice_session_id = item["voice_session_id"]
            ai_session_id = item["ai_session_id"]

            entities.append(
                Interaction(
                    interaction_id=interaction_id,
                    channel="voice",
                    subject=f"{marker} {interaction_id}",
                    customer_id=f"{marker}_cust_{interaction_id}",
                    queue_id=item["interaction_queue_id"],
                    priority=3,
                    status="closed" if item["status"] == "closed" else "in_progress",
                    assigned_to=item["claimed_by_user"],
                    created_at=item["started_at"],
                    updated_at=item["started_at"],
                )
            )
            if item.get("with_call_row", True):
                entities.append(
                    AsteriskCallLinkRow(
                        call_id=call_id,
                        linked_id=f"{call_id}_linked",
                        queue_code="voice_support",
                        queue_id=item["call_queue_id"],
                        interaction_id=interaction_id,
                        caller_number=f"+7700{abs(hash(call_id)) % 1000000:06d}",
                        caller_name="Voice Caller",
                        status="active",
                        telephony_status="connected",
                        claimed_by_user=item["claimed_by_user"],
                        claimed_at=item["started_at"] if item["claimed_by_user"] else None,
                        operator_extension=item["operator_extension"],
                        channel_name="PJSIP/1001-000001",
                        started_at=item["started_at"],
                        connected_at=item["started_at"],
                        ended_at=None,
                        updated_at=item["started_at"],
                        voice_start_language=item["call_voice_start_language"],
                        customer_name_status="name_obtained" if item["customer_name_source"] == "manual" else None,
                        customer_name_value="Manual Name" if item["customer_name_source"] == "manual" else None,
                        customer_name_source=item["customer_name_source"],
                        customer_name_resolved_at=item["started_at"] if item["customer_name_source"] == "manual" else None,
                        voice_session_id=voice_session_id,
                        ai_state="human_owned" if item["claimed_by_user"] else "closed",
                        ai_handoff_reason=item["handoff_reason"],
                    )
                )
            entities.append(
                VoiceAISessionRow(
                    session_id=voice_session_id,
                    call_id=call_id,
                    linked_id=f"{call_id}_linked",
                    interaction_id=interaction_id,
                    customer_id=f"{marker}_cust_{interaction_id}",
                    queue_id=item["queue_id"],
                    ai_session_id=ai_session_id,
                    agent_profile="voice_support",
                    language=item["language"],
                    asr_provider="openai",
                    tts_provider="yandex",
                    status=item["status"],
                    handoff_reason=item["handoff_reason"],
                    handoff_target_queue_id=item["interaction_queue_id"] or item["call_queue_id"],
                    disclosure_played_at=item["started_at"],
                    last_user_utterance_at=None,
                    last_ai_reply_at=None,
                    started_at=item["started_at"],
                    updated_at=item["started_at"],
                    ended_at=None,
                    voice_start_language=item["voice_start_language"],
                    customer_name_status="name_obtained" if item["customer_name_source"] == "manual" else None,
                    customer_name_value="Manual Name" if item["customer_name_source"] == "manual" else None,
                    customer_name_source=item["customer_name_source"],
                    customer_name_resolved_at=item["started_at"] if item["customer_name_source"] == "manual" else None,
                )
            )
            if ai_session_id:
                entities.append(
                    AISessionRow(
                        session_id=ai_session_id,
                        channel="voice",
                        thread_id=None,
                        interaction_id=interaction_id,
                        customer_id=f"{marker}_cust_{interaction_id}",
                        agent_profile="voice_support",
                        language=item["language"] or item["voice_start_language"] or item["call_voice_start_language"] or "unknown",
                        status=item["status"],
                        summary_text="",
                        last_user_message_id=None,
                        last_ai_message_id=None,
                        handoff_reason=item["handoff_reason"],
                        created_at=item["started_at"],
                        updated_at=item["started_at"],
                        closed_at=item["started_at"] if item["status"] == "closed" else None,
                    )
                )
            for turn in item["turns"]:
                entities.append(
                    AITurnRow(
                        turn_id=turn["turn_id"],
                        session_id=ai_session_id,
                        thread_id=None,
                        interaction_id=interaction_id,
                        role="assistant",
                        source_type="voice_policy",
                        text="voice policy",
                        payload_json=json.dumps({"decision": turn["decision"]}, ensure_ascii=False),
                        model="stub",
                        finish_reason="stop",
                        latency_ms=120,
                        created_at=turn["created_at"],
                    )
                )
            if item.get("timeline_event"):
                entities.append(
                    InteractionTimeline(
                        interaction_id=interaction_id,
                        timestamp=item["timeline_event"]["timestamp"],
                        action="voice.start.completed",
                        metadata_json=json.dumps(item["timeline_event"]["metadata"], ensure_ascii=False),
                    )
                )

        session.add_all(entities)
        session.commit()
        return {
            "from_ts": window_from,
            "to_ts": window_to,
            "queue_ru": queue_ru,
            "queue_kz": queue_kz,
            "queue_support": queue_support,
        }
    finally:
        session.close()


def cleanup_voice_name_flow_analytics_dataset(marker: str) -> None:
    session = get_session()
    try:
        session.query(AITurnRow).filter(AITurnRow.turn_id.like(f"{marker}_turn_%")).delete(synchronize_session=False)
        session.query(AISessionRow).filter(AISessionRow.session_id.like(f"{marker}_ai_%")).delete(synchronize_session=False)
        session.query(VoiceAISessionRow).filter(VoiceAISessionRow.session_id.like(f"{marker}_voice_%")).delete(synchronize_session=False)
        session.query(AsteriskCallLinkRow).filter(AsteriskCallLinkRow.call_id.like(f"{marker}_call_%")).delete(synchronize_session=False)
        session.query(InteractionTimeline).filter(InteractionTimeline.interaction_id.like(f"{marker}_int_%")).delete(synchronize_session=False)
        session.query(Interaction).filter(Interaction.interaction_id.like(f"{marker}_int_%")).delete(synchronize_session=False)
        session.commit()
    finally:
        session.close()


def test_ai_language_detection_prefers_kz_letters():
    assert ai_module._infer_language("Сәлем, көмек керек") == "kz"
    assert ai_module._infer_language("Здравствуйте, нужна помощь") == "ru"


def test_voice_module_import_helper_returns_app_module():
    loaded = voice_module._app()
    assert getattr(loaded, "__name__", "") == "services.ai_orchestrator_service.app"
    assert hasattr(loaded, "_customer_id_is_real")
    assert hasattr(loaded, "_interaction_request")


def test_text_openai_prompt_uses_operator_persona_without_ai_disclosure():
    messages = ai_module._openai_prompt(
        customer=None,
        interaction=SimpleNamespace(
            interaction_id="int_text_prompt",
            status="new",
            queue_id="que_text",
            subject="Need help",
            customer_id="cust_text",
        ),
        thread=SimpleNamespace(thread_id="thr_text", chat_id="chat_text", display_name="Customer"),
        messages=[
            SimpleNamespace(
                author_type="customer",
                author_id="cust_text",
                direction="inbound",
                text="Хочу узнать тариф",
                created_at=utc_now_iso(),
            )
        ],
        kb_results=[],
        language="ru",
        channel_label="Telegram",
        channel_key="telegram",
    )

    system_prompt = messages[0]["content"]
    assert "AI assistant" not in system_prompt
    assert "Always disclose" not in system_prompt
    assert "human operator" in system_prompt
    assert "Do not mention a knowledge base" in system_prompt


def test_voice_decision_reuses_recent_topic_instead_of_repeating_same_prompt():
    transcript_window = [
        SimpleNamespace(speaker="assistant", text=voice_module._voice_greeting("ru"), sequence_no=1),
        SimpleNamespace(speaker="caller", text="График работы узнать", sequence_no=2),
        SimpleNamespace(
            speaker="assistant",
            text="Подскажите, график работы какого филиала, адреса или города вас интересует?",
            sequence_no=3,
        ),
        SimpleNamespace(speaker="caller", text="О каком запросе?", sequence_no=4),
    ]

    decision = voice_module._voice_decision(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(interaction_id="int_voice_test"),
        transcript_text="О каком запросе?",
        transcript_window=transcript_window,
        kb_results=[],
        disclosure_required=False,
    )

    assert decision["intent"] == "clarification"
    assert decision["needs_handoff"] is False
    assert decision["reply_text"] != transcript_window[2].text
    assert "подскажите точнее" in decision["reply_text"].lower()
    assert "график работы" in decision["reply_text"].lower()


def test_voice_decision_handoffs_after_repeated_clarification_loop():
    transcript_window = [
        SimpleNamespace(speaker="assistant", text=voice_module._voice_greeting("ru"), sequence_no=1),
        SimpleNamespace(speaker="caller", text="Привет", sequence_no=2),
        SimpleNamespace(
            speaker="assistant",
            text="Чтобы помочь быстрее, скажите в двух словах, что вам нужно: график работы, статус заявки, тариф или оператор.",
            sequence_no=3,
        ),
        SimpleNamespace(speaker="caller", text="Угу", sequence_no=4),
        SimpleNamespace(
            speaker="assistant",
            text="Сейчас уточняю цель звонка. Скажите коротко, что именно нужно: график работы, статус заявки, тариф или оператор.",
            sequence_no=5,
        ),
        SimpleNamespace(speaker="caller", text="Не понял", sequence_no=6),
        SimpleNamespace(
            speaker="assistant",
            text="Сейчас уточняю цель звонка. Скажите коротко, что именно нужно: график работы, статус заявки, тариф или оператор.",
            sequence_no=7,
        ),
        SimpleNamespace(speaker="caller", text="О каком запросе?", sequence_no=8),
        SimpleNamespace(
            speaker="assistant",
            text="Сейчас уточняю цель звонка. Скажите коротко, что именно нужно: график работы, статус заявки, тариф или оператор.",
            sequence_no=9,
        ),
    ]

    decision = voice_module._voice_decision(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(interaction_id="int_voice_test"),
        transcript_text="О каком запросе?",
        transcript_window=transcript_window,
        kb_results=[],
        disclosure_required=False,
    )

    assert decision["intent"] == "handoff_request"
    assert decision["needs_handoff"] is True
    assert "перевожу на оператора" in decision["reply_text"].lower()
    assert "нескольких попыток" in decision["handoff_reason"].lower()


def test_voice_kb_search_matches_relaxed_phrase_and_returns_kb_answer():
    query = _u(r"\u0425\u043e\u0447\u0443 \u0443\u0437\u043d\u0430\u0442\u044c \u0442\u0430\u0440\u0438\u0444 relaxbasicx")
    seed_kb_article(
        "Tariff relaxbasicx",
        "Tariff relaxbasicx activates after the request is confirmed.",
        "relaxbasicx",
    )

    session = get_session()
    try:
        kb_results = ai_module._kb_search(session, query)
    finally:
        session.close()

    assert kb_results
    assert kb_results[0].title == "Tariff relaxbasicx"

    decision = voice_module._voice_decision(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(interaction_id="int_voice_kb_relaxed"),
        transcript_text=query,
        transcript_window=[SimpleNamespace(speaker="caller", text=query, sequence_no=1)],
        kb_results=kb_results,
        disclosure_required=False,
    )

    assert decision["intent"] == "kb_answer"
    assert decision["needs_handoff"] is False
    assert "relaxbasicx" in decision["reply_text"].lower()


def test_voice_llm_guarded_decision_uses_operator_style_without_ai_or_kb(monkeypatch):
    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "llm_guarded")
    monkeypatch.setattr(ai_module, "_ai_provider", lambda: "openai_compatible")
    captured: dict[str, object] = {}

    def _fake_structured(messages, **kwargs):
        captured["messages"] = messages
        return {
            "language": "ru",
            "intent": "kb_answer",
            "reply_text": "Сейчас подскажу: филиал в Алматы работает с 9:00 до 18:00 по будням.",
            "confidence": 0.88,
            "needs_handoff": False,
            "handoff_reason": None,
            "case_action": "keep_open",
            "kb_refs": ["kba_voice_1"],
            "_model": "gpt-test",
            "_latency_ms": 42,
            "_finish_reason": "stop",
        }

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _fake_structured)

    kb_article = SimpleNamespace(article_id="kba_voice_1", title="График работы", body="Будни 9:00-18:00")
    decision = voice_module._voice_decision(
        language="ru",
        customer=SimpleNamespace(customer_id="cus_voice_1", display_name="Айдос"),
        interaction=SimpleNamespace(interaction_id="int_voice_llm", status="new", queue_id="que_voice", subject="hours"),
        transcript_text="Как работает филиал в Алматы?",
        transcript_window=[SimpleNamespace(speaker="caller", text="Как работает филиал в Алматы?", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso())],
        kb_results=[kb_article],
        disclosure_required=False,
        customer_name_value="Айдос",
        customer_name_status="name_obtained",
    )

    assert decision["intent"] == "kb_answer"
    assert decision["model"] == "gpt-test"
    assert "ai" not in decision["reply_text"].lower()
    assert "база знаний" not in decision["reply_text"].lower()
    system_prompt = captured["messages"][0]["content"]
    assert "human operator" in system_prompt
    assert "Do not say or imply that you are an AI" in system_prompt


def test_voice_v2_fast_conversational_adds_ack_metadata_and_compacts_reply(monkeypatch):
    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_fast_conversational")
    monkeypatch.setattr(ai_module, "_ai_provider", lambda: "openai_compatible")

    def _fake_structured(messages, **kwargs):
        del messages
        return {
            "language": "ru",
            "intent": "kb_answer",
            "reply_text": (
                "Сейчас сориентирую по графику работы филиала в Алматы. "
                "Он работает с понедельника по пятницу с 9:00 до 18:00. "
                "Если нужно, подскажу и по субботе."
            ),
            "confidence": 0.91,
            "needs_handoff": False,
            "handoff_reason": None,
            "case_action": "keep_open",
            "kb_refs": ["kba_voice_v2"],
            "_model": "gpt-test",
            "_latency_ms": 35,
            "_finish_reason": "stop",
        }

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _fake_structured)

    decision = voice_module._voice_decision(
        language="ru",
        customer=SimpleNamespace(customer_id="cus_voice_v2", display_name="Ернор"),
        interaction=SimpleNamespace(interaction_id="int_voice_v2", status="new", queue_id="que_voice", subject="hours"),
        transcript_text="Хочу узнать график работы филиала в Алматы",
        transcript_window=[
            SimpleNamespace(
                speaker="caller",
                text="Хочу узнать график работы филиала в Алматы",
                sequence_no=1,
                source_type="voice_asr",
                barge_in_interrupted=False,
                created_at=utc_now_iso(),
            )
        ],
        kb_results=[SimpleNamespace(article_id="kba_voice_v2", title="График", body="Будни 9:00-18:00")],
        disclosure_required=False,
        customer_name_value="Ернор",
        customer_name_status="name_obtained",
        request_metadata={"voice_v2_enabled": True, "response_plan_id": "rsp_test"},
    )

    assert decision["intent"] == "kb_answer"
    assert decision["metadata"]["voice_v2_enabled"] is True
    assert decision["metadata"]["early_intent"] == "schedule"
    assert decision["metadata"]["ack_kind"] == "understanding"
    assert decision["metadata"]["response_plan_id"] == "rsp_test"
    assert len(decision["reply_text"]) <= 180


def test_voice_v2_off_domain_request_returns_fast_operator_fallback_without_llm(monkeypatch):
    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_fast_conversational")

    def _unexpected_llm(messages, **kwargs):
        raise AssertionError(f"LLM should not be called for off-domain fallback: {messages!r}")

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)

    decision = voice_module._voice_decision(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(interaction_id="int_voice_off_domain", status="new", queue_id="que_voice", subject="unknown"),
        transcript_text="Мне надо узнать, как работает ядерный реактор.",
        transcript_window=[
            SimpleNamespace(
                speaker="caller",
                text="Мне надо узнать, как работает ядерный реактор.",
                sequence_no=1,
                source_type="voice_asr",
                barge_in_interrupted=False,
                created_at=utc_now_iso(),
            )
        ],
        kb_results=[],
        disclosure_required=False,
        request_metadata={"voice_v2_enabled": True, "response_plan_id": "rsp_off_domain"},
    )

    assert decision["intent"] == "clarification"
    assert decision["needs_handoff"] is False
    assert decision["model"] == "voice_policy_off_domain"
    assert "наших услуг" in decision["reply_text"].lower()
    assert "оператор" in decision["reply_text"].lower()
    assert "ai" not in decision["reply_text"].lower()
    assert "база знаний" not in decision["reply_text"].lower()
    assert decision["metadata"]["voice_v2_enabled"] is True
    assert decision["metadata"]["early_intent"] == "unknown"
    assert decision["metadata"]["response_plan_id"] == "rsp_off_domain"


def test_voice_v2_streaming_duplex_early_plan_returns_fast_safe_reply_without_llm(monkeypatch):
    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex")

    def _unexpected_llm(messages, **kwargs):
        raise AssertionError(f"LLM should not be called for early plan: {messages!r}")

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)

    decision = voice_module._voice_decision(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(interaction_id="int_voice_early_plan", status="new", queue_id="que_voice", subject="unknown"),
        transcript_text="Расскажи, как устроен кондиционер",
        transcript_window=[],
        kb_results=[],
        disclosure_required=False,
        request_metadata={
            "voice_v2_enabled": True,
            "reply_phase": "early_plan",
            "response_plan_id": "rsp_early",
        },
    )

    assert decision["model"] == "voice_early_plan_off_domain"
    assert decision["metadata"]["reply_phase"] == "early_plan"
    assert decision["metadata"]["voice_v2_enabled"] is True
    assert decision["metadata"]["response_plan_id"] == "rsp_early"
    assert "кондиционер" not in decision["reply_text"].lower()
    assert "оператор" in decision["reply_text"].lower()


def test_voice_v2_streaming_duplex_early_plan_returns_domain_followup_without_llm(monkeypatch):
    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex")

    def _unexpected_llm(messages, **kwargs):
        raise AssertionError(f"LLM should not be called for early plan: {messages!r}")

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)

    decision = voice_module._voice_decision(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(interaction_id="int_voice_early_schedule", status="new", queue_id="que_voice", subject="unknown"),
        transcript_text="РњРЅРµ РЅР°РґРѕ СѓР·РЅР°С‚СЊ РіСЂР°С„РёРє СЂР°Р±РѕС‚С‹",
        transcript_window=[],
        kb_results=[],
        disclosure_required=False,
        request_metadata={
            "voice_v2_enabled": True,
            "reply_phase": "early_plan",
            "response_plan_id": "rsp_early_schedule",
            "early_intent": "schedule",
        },
    )

    assert decision["model"] == "voice_early_plan_domain"
    assert decision["reply_text"]
    assert decision["needs_handoff"] is False
    assert decision["metadata"]["reply_phase"] == "early_plan"
    assert decision["metadata"]["early_intent"] == "schedule"


def test_voice_decision_hearing_check_keeps_active_topic_without_llm(monkeypatch):
    def _unexpected_llm(messages, **kwargs):
        raise AssertionError(f"LLM should not be called for hearing check: {messages!r}")

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)

    decision = voice_module._voice_decision(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(interaction_id="int_voice_hearing", status="new", queue_id="que_voice", subject="schedule"),
        transcript_text="Алло, ты меня слышишь?",
        transcript_window=[
            SimpleNamespace(speaker="caller", text="Мне надо узнать график работы.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="assistant", text="Подскажите, какой именно график работы вас интересует?", sequence_no=2, source_type="voice_policy", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="caller", text="В городе Алмата.", sequence_no=3, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="caller", text="Алло, ты меня слышишь?", sequence_no=4, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
        ],
        kb_results=[],
        disclosure_required=False,
        request_metadata={"voice_v2_enabled": True, "response_plan_id": "rsp_hearing"},
    )

    assert decision["model"] == "voice_policy_hearing_check"
    assert decision["reply_text"].startswith("Да, вас слышу.")
    assert "Как я могу помочь" not in decision["reply_text"]
    assert "филиал" in decision["reply_text"] or "адрес" in decision["reply_text"]


def test_voice_postprocess_reply_rewrites_false_lookup_promise():
    reply_text = voice_module._voice_postprocess_reply_text(
        language="ru",
        transcript_text="В городе Алмата.",
        transcript_window=[
            SimpleNamespace(speaker="caller", text="Мне надо узнать график работы.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="caller", text="В городе Алмата.", sequence_no=2, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
        ],
        reply_text="Спасибо. Я уточню график работы в Алмате. Минуточку, пожалуйста.",
        kb_results=[],
        needs_handoff=False,
    )

    assert "уточню" not in reply_text.lower()
    assert "минуточ" not in reply_text.lower()
    assert "филиал" in reply_text.lower() or "адрес" in reply_text.lower()


def test_voice_postprocess_reply_rewrites_midcall_greeting_with_followup_question():
    reply_text = voice_module._voice_postprocess_reply_text(
        language="ru",
        transcript_text="Здравствуйте, мне надо узнать...",
        transcript_window=[
            SimpleNamespace(speaker="assistant", text=voice_module._voice_greeting("ru"), sequence_no=1, source_type="voice_policy", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="caller", text="Здравствуйте, мне надо узнать...", sequence_no=2, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
        ],
        reply_text="Здравствуйте, Ернур! О чем именно вы хотите узнать?",
        kb_results=[],
        needs_handoff=False,
    )

    normalized = reply_text.lower()
    assert "здравствуйте" not in normalized
    assert "о чем именно" not in normalized


def test_voice_postprocess_reply_reuses_active_topic_after_frustration_turn():
    reply_text = voice_module._voice_postprocess_reply_text(
        language="ru",
        transcript_text="Сколько раз повторять тебе?",
        transcript_window=[
            SimpleNamespace(speaker="caller", text="Мне надо узнать график работы.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="caller", text="В городе Алма-Ата.", sequence_no=2, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="caller", text="Меня интересует филиал в Ауэзовском районе.", sequence_no=3, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
            SimpleNamespace(speaker="caller", text="Сколько раз повторять тебе?", sequence_no=4, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
        ],
        reply_text="Пожалуйста, уточните, какая услуга вас интересует, чтобы я мог подсказать тариф.",
        kb_results=[],
        needs_handoff=False,
    )

    normalized = reply_text.lower()
    assert "тариф" not in normalized
    assert "услуг" not in normalized
    assert "филиал" in normalized or "адрес" in normalized


def test_turn_voice_session_early_plan_does_not_persist_partial_turns(monkeypatch):
    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex")

    def _unexpected_llm(messages, **kwargs):
        raise AssertionError(f"LLM should not be called for early plan: {messages!r}")

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)

    seeded = seed_voice_downstream_session(
        marker=f"voice_early_plan_{new_id('seed')}",
        name_status="name_not_obtained",
        customer_display_name="+77010009999",
    )

    decision = voice_module.turn_voice_session(
        seeded["session_id"],
        VoiceAITurnIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            transcript_text="Расскажи, как устроен кондиционер",
            language="ru",
            sequence_no=1,
            metadata={
                "voice_v2_enabled": True,
                "reply_phase": "early_plan",
                "response_plan_id": "rsp_early_turn",
            },
        ),
    )

    assert decision.metadata["reply_phase"] == "early_plan"
    assert decision.metadata["response_plan_id"] == "rsp_early_turn"
    assert decision.status == "active"

    session = get_session()
    try:
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == seeded["session_id"])
        ).scalar_one()
        ai_turns = session.execute(
            select(AITurnRow).where(AITurnRow.interaction_id == seeded["interaction_id"])
        ).scalars().all()
        transcript_segments = session.execute(
            select(VoiceTranscriptSegmentRow).where(VoiceTranscriptSegmentRow.session_id == seeded["session_id"])
        ).scalars().all()
        assert voice_session.ai_session_id is None
        assert voice_session.status == "active"
        assert ai_turns == []
        assert transcript_segments == []
    finally:
        session.close()


def test_voice_llm_prompt_includes_context_summary_and_uses_12_segments():
    transcript_window = [
        SimpleNamespace(
            speaker="caller" if index % 2 == 0 else "assistant",
            text=f"segment {index}",
            sequence_no=index,
            source_type="voice_policy",
            barge_in_interrupted=False,
            created_at=utc_now_iso(),
        )
        for index in range(1, 16)
    ]

    messages = voice_module._voice_llm_prompt_messages(
        language="ru",
        customer=None,
        interaction=SimpleNamespace(
            interaction_id="int_voice_prompt",
            status="open",
            queue_id="que_voice",
            subject="schedule",
            customer_id="cus_voice_prompt",
        ),
        transcript_text="Мне нужен график работы",
        transcript_window=transcript_window,
        conversation_summary_text="Customer name: Ернур | Active intent: schedule | Confirmed facts: city=Алмата",
        kb_results=[],
        name_value="Ернур",
        name_status="name_obtained",
    )

    payload = json.loads(messages[1]["content"])
    assert payload["conversation_summary"].startswith("Customer name: Ернур")
    assert len(payload["history"]) == 12
    assert payload["history"][0]["sequence_no"] == 4


def test_voice_postprocess_reply_uses_summary_context_when_raw_window_lost_topic():
    reply_text = voice_module._voice_postprocess_reply_text(
        language="ru",
        transcript_text="Сколько раз повторять тебе?",
        transcript_window=[
            SimpleNamespace(speaker="caller", text="Сколько раз повторять тебе?", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
        ],
        context_summary=json.dumps(
            {
                "customer_name": "Ернур",
                "active_intent": "schedule",
                "active_request_text": "Мне нужен график работы в городе Алмата",
                "confirmed_facts": {"city": "Алмата", "branch_hint": None, "service_hint": "график работы", "request_number": None},
                "open_slots": ["branch_or_address"],
            },
            ensure_ascii=False,
        ),
        reply_text="Пожалуйста, уточните, какая услуга вас интересует, чтобы я мог подсказать тариф.",
        kb_results=[],
        needs_handoff=False,
    )

    normalized = reply_text.lower()
    assert "тариф" not in normalized
    assert "услуг" not in normalized
    assert "филиал" in normalized or "адрес" in normalized


def test_voice_decision_uses_summary_city_slot_instead_of_asking_city_again(monkeypatch):
    def _unexpected_llm(messages, **kwargs):
        raise AssertionError(f"LLM should not be called when summary slot prompt is enough: {messages!r}")

    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)

    decision = voice_module._voice_decision(
        language="ru",
        customer=SimpleNamespace(display_name="Ернур"),
        interaction=SimpleNamespace(interaction_id="int_voice_summary_slot", customer_id=None, status="open", queue_id=None, subject=None),
        transcript_text="Матта.",
        transcript_window=[
            SimpleNamespace(speaker="caller", text="Матта.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
        ],
        context_summary=json.dumps(
            {
                "customer_name": "Ернур",
                "active_intent": "schedule",
                "active_request_text": "Хочу узнать график работы",
                "confirmed_facts": {"city": "Алмата", "branch_hint": None, "service_hint": "график работы", "request_number": None},
                "open_slots": ["branch_or_address"],
            },
            ensure_ascii=False,
        ),
        kb_results=[],
        disclosure_required=False,
        customer_name_value="Ернур",
        customer_name_status="name_obtained",
        request_metadata={"voice_v2_enabled": True},
    )

    assert decision["intent"] == "clarification"
    assert decision["needs_handoff"] is False
    assert "Алмата" in decision["reply_text"]
    assert "город" in decision["reply_text"].lower()
    assert "филиал" in decision["reply_text"].lower() or "адрес" in decision["reply_text"].lower()


def test_ai_enqueue_creates_outbound_ai_reply_and_delivery_flow(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)
    seed_kb_article("Тариф Basic", "Тариф basic активируется за 5 минут после подтверждения заявки.", "basic")

    created = create_inbound_thread(telegram_client, "chat_ai_reply", "basic")
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "done"

    summary = fetch_ai_summary(telegram_client, created["thread_id"])
    assert summary.status_code == 200
    assert summary.json() is None

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{created['thread_id']}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    assert messages.json()[-1]["author_type"] == "ai"
    assert messages.json()[-1]["delivery_status"] == "pending"

    monkeypatch.setattr(
        telegram_module,
        "_send_telegram_message",
        lambda chat_id, text: {"ok": True, "result": {"message_id": 5551, "chat": {"id": chat_id}, "text": text}},
    )
    telegram_module._deliver_pending_telegram_reply(messages.json()[-1]["message_id"])

    delivered = telegram_client.get(
        f"/integrations/telegram/threads/{created['thread_id']}/messages",
        headers=admin_headers(),
    )
    assert delivered.status_code == 200
    assert delivered.json()[-1]["delivery_status"] == "sent"
    assert delivered.json()[-1]["telegram_message_id_external"] == "5551"


def test_ai_enqueue_persists_internal_context_summary(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)

    created = create_inbound_thread(
        telegram_client,
        f"chat_ai_context_summary_{new_id('chat')}",
        "Хочу узнать график работы в городе Алмата",
    )
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )

    assert enqueue.status_code == 200
    assert enqueue.json()["status"] in {"done", "handoff_required", "running"}

    deadline = time.time() + 2.0
    last_summary = None
    while time.time() < deadline:
        session = get_session()
        try:
            thread = session.execute(
                select(TelegramThreadRow).where(TelegramThreadRow.thread_id == created["thread_id"])
            ).scalar_one()
            if not thread.ai_session_id:
                time.sleep(0.05)
                continue
            ai_session = session.execute(
                select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id)
            ).scalar_one()
            last_summary = json.loads(ai_session.context_summary_json or "{}")
            if (
                last_summary.get("active_intent") == "schedule"
                and last_summary.get("confirmed_facts", {}).get("city") == "Алмата"
            ):
                break
        finally:
            session.close()
        time.sleep(0.05)

    assert last_summary is not None
    assert last_summary["active_intent"] == "schedule"
    assert last_summary["confirmed_facts"]["city"] == "Алмата"
    assert last_summary["open_slots"] == ["branch_or_address"]


def test_ai_enqueue_relaxed_kb_search_answers_phrase_query(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)
    seed_kb_article(
        "Tariff tgrelaxx",
        "Tariff tgrelaxx activates after the request is confirmed.",
        "tgrelaxx",
    )

    created = create_inbound_thread(
        telegram_client,
        "chat_ai_relaxed_kb",
        _u(r"\u0425\u043e\u0447\u0443 \u0443\u0437\u043d\u0430\u0442\u044c \u0442\u0430\u0440\u0438\u0444 tgrelaxx"),
    )
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "done"

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{created['thread_id']}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    assert messages.json()[-1]["author_type"] == "ai"
    assert "tgrelaxx" in messages.json()[-1]["text"].lower()


def test_ai_enqueue_uses_language_filtered_kb_localization(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)
    monkeypatch.setattr(ai_module, "_infer_language", lambda text: "kz")

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)

    seeded = seed_kb_article(
        "Localized sharedkbx RU",
        "rulocalizedanswerx",
        "sharedkbx",
        language="ru",
    )
    seed_kb_article(
        "Localized sharedkbx KZ",
        "kzlocalizedanswerx",
        "sharedkbx",
        language="kz",
        article_group_id=seeded["article_group_id"],
    )

    created = create_inbound_thread(telegram_client, "chat_ai_kz_localized_kb", "sharedkbx")
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "done"

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{created['thread_id']}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    reply_text = messages.json()[-1]["text"].lower()
    assert "kzlocalizedanswerx" in reply_text
    assert "rulocalizedanswerx" not in reply_text


def test_ai_enqueue_deduplicates_same_trigger_message(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)
    seed_kb_article("FAQ handoff", "Ответ по FAQ для повторного запроса.", "faq-dedupe")

    created = create_inbound_thread(telegram_client, "chat_ai_dedupe", "faq-dedupe")
    first = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert first.status_code == 200
    second = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert second.status_code == 200
    assert second.json()["deduplicated"] is True
    deliver_latest_pending_message(monkeypatch, telegram_client, created["thread_id"], external_id=7002)

    session = get_session()
    try:
        jobs = session.execute(
            select(AIJobRow).where(AIJobRow.thread_id == created["thread_id"])
        ).scalars().all()
        assert len(jobs) == 1
        assert jobs[0].trigger_message_id == created["message_id"]
    finally:
        session.close()


def test_ai_handoff_required_leaves_thread_available_for_claim(monkeypatch):
    patch_interaction_request(monkeypatch)
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)

    created = create_inbound_thread(telegram_client, "chat_ai_handoff", "Хочу человека, соедините с оператором")
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "handoff_required"

    thread = telegram_client.get(
        f"/integrations/telegram/threads/{created['thread_id']}",
        headers=admin_headers(),
    )
    assert thread.status_code == 200
    assert thread.json()["ai_state"] == "handoff_required"
    assert thread.json()["claimed_by_user"] is None

    summary = fetch_ai_summary(telegram_client, created["thread_id"])
    assert summary.status_code == 200
    payload = summary.json()
    assert payload["thread_id"] == created["thread_id"]
    assert payload["status_label"] == "AI передал без ответа"
    assert payload["status_tone"] == "handoff"
    assert payload["customer_request_text"] == created["text"]
    assert payload["ai_outcome_text"] == "AI передал диалог оператору без ответа клиенту."
    assert payload["handoff_reason"]
    assert payload["recommended_next_step"] == "Заберите чат и ответьте клиенту вручную."

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{created['thread_id']}/claim",
        headers=operator_headers("operator_summary"),
    )
    assert claimed.status_code == 200
    assert claimed.json()["ai_state"] == "human_owned"

    summary_after_claim = fetch_ai_summary(
        telegram_client,
        created["thread_id"],
        headers=operator_headers("operator_summary"),
    )
    assert summary_after_claim.status_code == 200
    claim_payload = summary_after_claim.json()
    assert claim_payload["session_id"] == payload["session_id"]
    assert claim_payload["status_label"] == payload["status_label"]
    assert claim_payload["status_tone"] == payload["status_tone"]
    assert claim_payload["handoff_reason"] == payload["handoff_reason"]
    assert claim_payload["recommended_next_step"] == "Продолжайте диалог вручную; AI больше не отвечает в этот thread."

    returned = telegram_client.post(
        f"/integrations/telegram/threads/{created['thread_id']}/return-to-ai",
        headers=operator_headers("operator_summary"),
    )
    assert returned.status_code == 200

    summary_after_return = fetch_ai_summary(telegram_client, created["thread_id"])
    assert summary_after_return.status_code == 200
    assert summary_after_return.json() is None

    session = get_session()
    try:
        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == created["thread_id"])
        ).scalar_one()
        ai_session = session.execute(
            select(AISessionRow).where(AISessionRow.session_id == payload["session_id"])
        ).scalar_one()
        assert thread.ai_state == "queued"
        assert thread.ai_handoff_reason is None
        assert ai_session.status == "active"
        assert ai_session.handoff_reason is None
    finally:
        session.close()


def test_select_trigger_message_ignores_non_customer_trigger():
    session = get_session()
    try:
        now = utc_now_iso()
        thread_id = new_id("tgt")
        interaction_id = new_id("int")
        customer_message_id = new_id("tgm")
        system_message_id = new_id("tgm")
        session.add(
            TelegramThreadRow(
                thread_id=thread_id,
                chat_id="chat_select_trigger",
                interaction_id=interaction_id,
                telegram_user_id="tg_select_trigger",
                username="select_trigger_user",
                display_name="Select Trigger",
                queue_id="q_telegram",
                status="new",
                claimed_by_user=None,
                claimed_at=None,
                ai_session_id=None,
                ai_state="queued",
                ai_handoff_reason=None,
                ai_last_model_at=None,
                last_message_at=now,
                last_message_preview="latest",
                created_at=now,
                updated_at=now,
            )
        )
        session.add(
            TelegramMessageRow(
                message_id=customer_message_id,
                thread_id=thread_id,
                interaction_id=interaction_id,
                chat_id="chat_select_trigger",
                text="Customer message",
                customer_external_id="cus_select_trigger",
                direction="inbound",
                telegram_message_id_external="1001",
                operator_user=None,
                author_type="customer",
                author_id="tg_select_trigger",
                delivery_status="received",
                payload_json="{}",
                created_at=now,
            )
        )
        session.add(
            TelegramMessageRow(
                message_id=system_message_id,
                thread_id=thread_id,
                interaction_id=interaction_id,
                chat_id="chat_select_trigger",
                text="[Unsupported Telegram content: photo]",
                customer_external_id="cus_select_trigger",
                direction="system",
                telegram_message_id_external="1002",
                operator_user=None,
                author_type="system",
                author_id="tg_select_trigger",
                delivery_status="received",
                payload_json="{}",
                created_at=now,
            )
        )
        session.commit()

        selected = ai_module._select_trigger_message(
            session,
            thread_id=thread_id,
            trigger_message_id=system_message_id,
        )

        assert selected is not None
        assert selected.message_id == customer_message_id
    finally:
        session.close()


def test_ai_reply_conflict_marks_job_human_owned_instead_of_error(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)
    patch_interaction_request(monkeypatch)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)

    def fake_interaction_request(method: str, path: str, *, payload: dict | None = None) -> dict:
        response = interaction_client.request(method, path, json=payload, headers=admin_headers())
        response.raise_for_status()
        return response.json()

    created = create_inbound_thread(telegram_client, "chat_ai_conflict", "basic")

    def fake_telegram_request(method: str, path: str, *, payload: dict | None = None) -> dict:
        if path.endswith("/ai/reply"):
            claimed = telegram_client.post(
                f"/integrations/telegram/threads/{created['thread_id']}/claim",
                headers=operator_headers("operator_race"),
            )
            assert claimed.status_code == 200
            request = httpx.Request(method, f"http://testserver{path}", json=payload)
            response = httpx.Response(
                409,
                request=request,
                json={"detail": "Telegram thread is owned by a human operator"},
            )
            raise httpx.HTTPStatusError("409 Conflict", request=request, response=response)

        response = telegram_client.request(method, path, json=payload, headers=admin_headers())
        response.raise_for_status()
        return response.json()

    monkeypatch.setattr(ai_module, "_telegram_request", fake_telegram_request)
    monkeypatch.setattr(ai_module, "_interaction_request", fake_interaction_request)
    seed_kb_article("Basic plan", "Basic answer.", "basic")

    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "human_owned"

    session = get_session()
    try:
        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == created["thread_id"])
        ).scalar_one()
        ai_session = session.execute(
            select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id)
        ).scalar_one()
        job = session.execute(
            select(AIJobRow).where(AIJobRow.thread_id == created["thread_id"]).order_by(AIJobRow.id.desc())
        ).scalar_one()
        assert thread.ai_state == "human_owned"
        assert thread.claimed_by_user == "operator_race"
        assert ai_session.status == "human_owned"
        assert job.status == "done"
        assert thread.ai_handoff_reason is None
    finally:
        session.close()


def test_ai_always_reply_mode_answers_greeting_without_handoff(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_TELEGRAM_ALWAYS_REPLY", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)

    created = create_inbound_thread(telegram_client, "chat_ai_always_reply", "Привет")
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "done"

    thread = telegram_client.get(
        f"/integrations/telegram/threads/{created['thread_id']}",
        headers=admin_headers(),
    )
    assert thread.status_code == 200
    assert thread.json()["ai_state"] == "active"
    assert thread.json()["claimed_by_user"] is None

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{created['thread_id']}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    assert messages.json()[-1]["author_type"] == "ai"
    assert messages.json()[-1]["text"]

    summary = fetch_ai_summary(telegram_client, created["thread_id"])
    assert summary.status_code == 200
    assert summary.json() is None


def test_claiming_ai_thread_transfers_to_human_and_blocks_auto_replies(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    enqueue_calls: list[tuple[str, str | None]] = []
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)))
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)
    patch_interaction_request(monkeypatch)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)
    seed_kb_article("Takeover FAQ", "FAQ answer before human takeover.", "takeover")

    created = create_inbound_thread(telegram_client, "chat_ai_takeover", "takeover")
    enqueue_calls.clear()
    ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    deliver_latest_pending_message(monkeypatch, telegram_client, created["thread_id"], external_id=7003)
    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{created['thread_id']}/claim",
        headers=operator_headers("operator_ai"),
    )
    assert claimed.status_code == 200
    assert claimed.json()["ai_state"] == "human_owned"
    enqueue_calls.clear()

    follow_up = telegram_client.post(
        "/integrations/telegram/webhook",
        json={
            "chat_id": "chat_ai_takeover",
            "text": "ещё вопрос после takeover",
            "payload": {"telegram_user_id": "user-chat_ai_takeover", "username": "takeover_user"},
        },
    )
    assert follow_up.status_code == 200
    time.sleep(0.05)
    assert enqueue_calls == []

    session = get_session()
    try:
        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == created["thread_id"])
        ).scalar_one()
        ai_session = session.execute(
            select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id)
        ).scalar_one()
        assert thread.ai_state == "human_owned"
        assert ai_session.status == "human_owned"
    finally:
        session.close()


def test_ai_summary_is_hidden_after_close(monkeypatch):
    patch_interaction_request(monkeypatch)
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)

    created = create_inbound_thread(telegram_client, "chat_ai_summary_close", "Хочу поговорить с человеком")
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "handoff_required"

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{created['thread_id']}/claim",
        headers=operator_headers("operator_close_summary"),
    )
    assert claimed.status_code == 200

    closed = telegram_client.post(
        f"/integrations/telegram/threads/{created['thread_id']}/close",
        headers=operator_headers("operator_close_summary"),
    )
    assert closed.status_code == 200

    summary = fetch_ai_summary(telegram_client, created["thread_id"])
    assert summary.status_code == 200
    assert summary.json() is None


def test_ai_summary_marks_answered_when_ai_replied_before_handoff(monkeypatch):
    patch_interaction_request(monkeypatch)
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    monkeypatch.setenv("AI_TELEGRAM_ALWAYS_REPLY", "1")
    monkeypatch.setenv("AI_PROVIDER", "stub")
    monkeypatch.setattr(telegram_module, "_ai_enqueue_request", lambda thread_id, trigger_message_id: None)
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)
    ai_client = TestClient(ai_app)
    patch_ai_internal_calls(monkeypatch, telegram_client, interaction_client)

    created = create_inbound_thread(telegram_client, "chat_ai_summary_answered", "Привет")
    enqueue = ai_client.post(
        f"/ai/telegram/threads/{created['thread_id']}/enqueue",
        headers=admin_headers(),
        json={"trigger_message_id": created["message_id"]},
    )
    assert enqueue.status_code == 200
    assert enqueue.json()["status"] == "done"

    handoff = telegram_client.post(
        f"/integrations/telegram/threads/{created['thread_id']}/ai/handoff",
        headers=admin_headers(),
        json={
            "reason": "Нужен оператор после автоответа.",
            "agent_profile": "telegram_support",
            "trigger_message_id": created["message_id"],
            "payload": {},
        },
    )
    assert handoff.status_code == 200

    summary = fetch_ai_summary(telegram_client, created["thread_id"])
    assert summary.status_code == 200
    payload = summary.json()
    assert payload["status_label"] == "AI ответил клиенту"
    assert payload["status_tone"] == "answered"
    assert payload["ai_outcome_text"]


def test_ai_analytics_overview_aggregates_mixed_telegram_and_whatsapp_sessions():
    marker = f"aiov_{new_id('seed')}"
    seeded = seed_ai_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    try:
        response = ai_client.get(
            "/ai/analytics/overview",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "channel": "all",
            },
            headers=admin_headers(),
        )

        assert response.status_code == 200
        payload = response.json()
        assert payload["totals"]["sessions_started"] == 4
        assert payload["totals"]["sessions_contained"] == 2
        assert payload["totals"]["sessions_handoff"] == 1
        assert payload["totals"]["sessions_closed"] == 3
        assert payload["totals"]["sessions_closed_without_operator"] == 2
        assert payload["totals"]["assistant_turns"] == 4
        assert payload["metrics"]["containment_rate"] == 50.0
        assert payload["metrics"]["handoff_rate"] == 25.0
        assert payload["metrics"]["ai_latency_avg_ms"] == 400.0
        assert payload["metrics"]["ai_latency_p95_ms"] == 900.0
        assert payload["metrics"]["closed_without_operator_rate"] == 66.67
        assert payload["metrics"]["human_touched_rate"] == 50.0
        assert payload["coverage"]["sessions_with_interaction_id"] == 4
        assert payload["coverage"]["sessions_with_queue_id"] == 4
        assert payload["coverage"]["sessions_with_latency_turns"] == 3
        assert payload["coverage"]["sessions_with_terminal_state"] == 4
        assert payload["coverage"]["sessions_with_handoff_reason"] == 1

        by_channel = {item["channel"]: item for item in payload["breakdowns"]["by_channel"]}
        assert by_channel["telegram"]["sessions_started"] == 2
        assert by_channel["telegram"]["containment_rate"] == 50.0
        assert by_channel["telegram"]["human_touched_sessions"] == 1
        assert by_channel["telegram"]["ai_latency_avg_ms"] == 150.0
        assert by_channel["whatsapp"]["sessions_started"] == 2
        assert by_channel["whatsapp"]["handoff_rate"] == 50.0
        assert by_channel["whatsapp"]["closed_without_operator_rate"] == 100.0
        assert by_channel["whatsapp"]["ai_latency_avg_ms"] == 650.0

        by_outcome = {item["outcome"]: item for item in payload["breakdowns"]["by_outcome"]}
        assert by_outcome["contained"]["sessions"] == 2
        assert by_outcome["handoff"]["sessions"] == 1
        assert by_outcome["human_touched"]["sessions"] == 1
        assert by_outcome["closed_without_operator"]["sessions"] == 2

        by_reason = {item["reason_key"]: item for item in payload["breakdowns"]["by_handoff_reason"]}
        assert by_reason["requested_human"]["sessions"] == 1
        assert by_reason["requested_human"]["label"]
    finally:
        cleanup_ai_analytics_dataset(marker)


def test_ai_analytics_overview_filters_by_queue_and_channel_and_handles_unsupported_channels():
    marker = f"aiflt_{new_id('seed')}"
    seeded = seed_ai_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    try:
        queue_filtered = ai_client.get(
            "/ai/analytics/overview",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "queue_id": seeded["queue_thread"],
                "channel": "whatsapp",
            },
            headers=admin_headers(),
        )
        assert queue_filtered.status_code == 200
        queue_payload = queue_filtered.json()
        assert queue_payload["totals"]["sessions_started"] == 1
        assert queue_payload["totals"]["sessions_contained"] == 1
        assert queue_payload["coverage"]["sessions_with_queue_id"] == 1
        assert queue_payload["breakdowns"]["by_channel"][0]["channel"] == "whatsapp"

        telegram_only = ai_client.get(
            "/ai/analytics/overview",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "channel": "telegram",
                "queue_id": seeded["queue_tg"],
            },
            headers=admin_headers(),
        )
        assert telegram_only.status_code == 200
        assert telegram_only.json()["totals"]["sessions_started"] == 2

        unsupported = ai_client.get(
            "/ai/analytics/overview",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "channel": "voice",
            },
            headers=admin_headers(),
        )
        assert unsupported.status_code == 200
        unsupported_payload = unsupported.json()
        assert unsupported_payload["totals"]["sessions_started"] == 0
        assert unsupported_payload["breakdowns"]["by_channel"] == []
    finally:
        cleanup_ai_analytics_dataset(marker)


def test_ai_analytics_timeseries_returns_stable_points_and_ignores_non_model_turns():
    marker = f"aits_{new_id('seed')}"
    seeded = seed_ai_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    try:
        response = ai_client.get(
            "/ai/analytics/timeseries",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "metric": "ai_latency_avg_ms",
                "interval": "day",
                "channel": "whatsapp",
                "queue_id": seeded["queue_thread"],
            },
            headers=admin_headers(),
        )
        assert response.status_code == 200
        payload = response.json()
        assert payload["metric"] == "ai_latency_avg_ms"
        assert payload["interval"] == "day"
        assert len(payload["points"]) == 2
        assert payload["points"][0]["value"] is None
        assert payload["points"][0]["sessions"] == 0
        assert payload["points"][0]["assistant_turns"] == 0
        assert payload["points"][1]["value"] == 400.0
        assert payload["points"][1]["sessions"] == 1
        assert payload["points"][1]["assistant_turns"] == 1

        latency_response = ai_client.get(
            "/ai/analytics/timeseries",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "metric": "ai_latency_avg_ms",
                "interval": "day",
                "channel": "voice",
            },
            headers=admin_headers(),
        )
        assert latency_response.status_code == 200
        latency_payload = latency_response.json()
        assert latency_payload["points"] == []

        human_touched = ai_client.get(
            "/ai/analytics/timeseries",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "metric": "human_touched_rate",
                "interval": "day",
                "channel": "all",
            },
            headers=admin_headers(),
        )
        assert human_touched.status_code == 200
        human_touched_payload = human_touched.json()
        assert human_touched_payload["metric"] == "human_touched_rate"
        assert [point["value"] for point in human_touched_payload["points"]] == [50.0, 50.0]
    finally:
        cleanup_ai_analytics_dataset(marker)


def test_ai_analytics_drilldown_filters_by_slice_reason_status_queue_channel_and_query():
    marker = f"aidd_{new_id('seed')}"
    seeded = seed_ai_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    try:
        handoff = ai_client.get(
            "/ai/analytics/drilldown",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "slice": "handoff",
                "reason_key": "requested_human",
                "status": "human_owned",
                "channel": "whatsapp",
                "queue_id": seeded["queue_wa"],
                "q": f"{marker}_sess_wa",
                "sort_by": "updated_at",
                "sort_dir": "desc",
            },
            headers=admin_headers(),
        )
        assert handoff.status_code == 200
        payload = handoff.json()
        assert payload["total"] == 1
        assert payload["filters"]["slice"] == "handoff"
        assert payload["filters"]["reason_key"] == "requested_human"
        assert payload["filters"]["status"] == "human_owned"
        assert payload["filters"]["q"] == f"{marker}_sess_wa"
        assert payload["filters"]["sort_by"] == "updated_at"
        assert payload["filters"]["sort_dir"] == "desc"
        assert payload["items"][0]["session_id"] == f"{marker}_sess_wa"
        assert payload["items"][0]["reason_key"] == "requested_human"
        assert payload["items"][0]["status"] == "human_owned"

        all_sorted = ai_client.get(
            "/ai/analytics/drilldown",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "slice": "all",
                "sort_by": "ai_latency_avg_ms",
                "sort_dir": "desc",
            },
            headers=admin_headers(),
        )
        assert all_sorted.status_code == 200
        sorted_payload = all_sorted.json()
        assert sorted_payload["items"][0]["session_id"] == f"{marker}_sess_wa"
        assert sorted_payload["items"][0]["ai_latency_avg_ms"] == 900.0
    finally:
        cleanup_ai_analytics_dataset(marker)


def test_ai_analytics_session_detail_is_metadata_only_and_reason_taxonomy_supports_manual_claim():
    marker = f"aidetail_{new_id('seed')}"
    seeded = seed_ai_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    session = get_session()
    try:
        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == f"{marker}_thread_tg_human")
        ).scalar_one()
        thread.claimed_by_user = "manual_takeover"
        thread.ai_handoff_reason = None
        session.commit()
    finally:
        session.close()

    try:
        manual_claim = ai_client.get(
            "/ai/analytics/drilldown",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "slice": "human_touched",
                "q": f"{marker}_sess_tg_human",
            },
            headers=admin_headers(),
        )
        assert manual_claim.status_code == 200
        manual_payload = manual_claim.json()
        assert manual_payload["total"] == 1
        assert manual_payload["items"][0]["reason_key"] == "manual_claim"
        assert manual_payload["items"][0]["reason_label"]

        detail = ai_client.get(
            f"/ai/analytics/sessions/{marker}_sess_wa",
            headers=admin_headers(),
        )
        assert detail.status_code == 200
        detail_payload = detail.json()
        assert detail_payload["session"]["session_id"] == f"{marker}_sess_wa"
        assert detail_payload["session"]["reason_key"] == "requested_human"
        assert detail_payload["interaction"]["interaction_id"] == f"{marker}_int_wa"
        assert "summary_text" not in detail_payload["session"]
        assert detail_payload["timeline"]
        assert all("text" not in event for event in detail_payload["timeline"])
        assert all("summary_text" not in event for event in detail_payload["timeline"])
    finally:
        cleanup_ai_analytics_dataset(marker)


def test_voice_name_flow_overview_aggregates_start_downstream_handoff_and_manual_overlay():
    marker = f"vname_{new_id('seed')}"
    seeded = seed_voice_name_flow_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    try:
        response = ai_client.get(
            "/ai/analytics/voice-name-flow/overview",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
            },
            headers=admin_headers(),
        )
        assert response.status_code == 200
        payload = response.json()

        assert payload["totals"]["scenario_calls"] == 5
        assert payload["totals"]["start_obtained"] == 1
        assert payload["totals"]["downstream_ai_obtained"] == 1
        assert payload["totals"]["followup_required"] == 1
        assert payload["totals"]["name_not_obtained"] == 2
        assert payload["totals"]["manual_corrected"] == 1
        assert payload["totals"]["handoff_confirmed_name"] == 1
        assert payload["totals"]["handoff_unconfirmed_name"] == 2
        assert payload["totals"]["needed_downstream"] == 4

        assert payload["metrics"]["start_capture_rate"] == 20.0
        assert payload["metrics"]["downstream_rescue_rate"] == 25.0
        assert payload["metrics"]["handoff_unconfirmed_rate"] == 66.67
        assert payload["metrics"]["manual_correction_rate"] == 33.33

        funnel = {item["stage"]: item for item in payload["breakdowns"]["funnel"]}
        assert funnel["scenario_calls"]["sessions"] == 5
        assert funnel["start_obtained"]["sessions"] == 1
        assert funnel["needed_downstream"]["sessions"] == 4
        assert funnel["downstream_ai_obtained"]["sessions"] == 1
        assert funnel["handoff_confirmed_name"]["sessions"] == 1
        assert funnel["handoff_unconfirmed_name"]["sessions"] == 2

        by_language = {item["language"]: item for item in payload["breakdowns"]["by_language"]}
        assert by_language["ru"]["scenario_calls"] == 3
        assert by_language["kz"]["downstream_ai_obtained"] == 1
        assert by_language["unknown"]["followup_required"] == 1

        by_queue = {item["queue_id"]: item for item in payload["breakdowns"]["by_queue"]}
        assert by_queue[seeded["queue_ru"]]["scenario_calls"] == 3
        assert by_queue[seeded["queue_kz"]]["downstream_ai_obtained"] == 1
        assert by_queue[seeded["queue_support"]]["handoff_unconfirmed_name"] == 1

        handoff = {item["outcome"]: item for item in payload["breakdowns"]["handoff"]}
        assert handoff["confirmed_name"]["sessions"] == 1
        assert handoff["unconfirmed_name"]["sessions"] == 2

        assert payload["coverage"]["sessions_with_start_decision"] == 5
        assert payload["coverage"]["sessions_with_final_ai_state"] == 5
        assert payload["coverage"]["sessions_with_manual_overlay"] == 1
        assert payload["coverage"]["note"]
    finally:
        cleanup_voice_name_flow_analytics_dataset(marker)


def test_voice_name_flow_overview_filters_by_queue_and_language_with_fallbacks():
    marker = f"vnameflt_{new_id('seed')}"
    seeded = seed_voice_name_flow_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    try:
        queue_filtered = ai_client.get(
            "/ai/analytics/voice-name-flow/overview",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "queue_id": seeded["queue_kz"],
            },
            headers=admin_headers(),
        )
        assert queue_filtered.status_code == 200
        queue_payload = queue_filtered.json()
        assert queue_payload["totals"]["scenario_calls"] == 1
        assert queue_payload["totals"]["downstream_ai_obtained"] == 1
        assert queue_payload["breakdowns"]["by_queue"][0]["queue_id"] == seeded["queue_kz"]

        language_filtered = ai_client.get(
            "/ai/analytics/voice-name-flow/overview",
            params={
                "from_ts": seeded["from_ts"],
                "to_ts": seeded["to_ts"],
                "language": "unknown",
            },
            headers=admin_headers(),
        )
        assert language_filtered.status_code == 200
        language_payload = language_filtered.json()
        assert language_payload["totals"]["scenario_calls"] == 1
        assert language_payload["totals"]["followup_required"] == 1
        assert language_payload["breakdowns"]["by_language"][0]["language"] == "unknown"
    finally:
        cleanup_voice_name_flow_analytics_dataset(marker)


def test_voice_name_flow_timeseries_returns_stable_points_for_all_metrics():
    marker = f"vnamets_{new_id('seed')}"
    seeded = seed_voice_name_flow_analytics_dataset(marker)
    ai_client = TestClient(ai_app)

    expected_values = {
        "scenario_calls": [2.0, 3.0],
        "start_capture_rate": [50.0, 0.0],
        "downstream_rescue_rate": [100.0, 0.0],
        "handoff_unconfirmed_rate": [0.0, 100.0],
        "manual_correction_rate": [0.0, 50.0],
    }

    try:
        for metric, expected in expected_values.items():
            response = ai_client.get(
                "/ai/analytics/voice-name-flow/timeseries",
                params={
                    "from_ts": seeded["from_ts"],
                    "to_ts": seeded["to_ts"],
                    "metric": metric,
                },
                headers=admin_headers(),
            )
            assert response.status_code == 200
            payload = response.json()
            assert payload["metric"] == metric
            assert payload["interval"] == "day"
            assert [point["value"] for point in payload["points"]] == expected
    finally:
        cleanup_voice_name_flow_analytics_dataset(marker)


def test_voice_name_collection_config_get_returns_defaults_when_not_persisted():
    ai_client = TestClient(ai_app)

    response = ai_client.get("/ai/voice/config/name-collection", headers=admin_headers())

    assert response.status_code == 200
    payload = response.json()
    assert payload["source"] == "defaults"
    assert payload["updated_at"] is None
    assert payload["config"]["enabled"] is True
    assert payload["config"]["start"]["known_customer_behavior"] == "trust_and_handoff"
    assert payload["config"]["downstream"]["missing_name_behavior"] == "ask_inline_once"
    assert "{name}" in payload["config"]["texts"]["ru"]["personalized_greeting_template"]
    assert "{name}" in payload["config"]["texts"]["kz"]["confirmation_greeting_template"]


def test_voice_name_collection_config_put_persists_custom_payload():
    ai_client = TestClient(ai_app)
    payload = voice_config_module.voice_name_collection_default_config().model_dump()
    payload["enabled"] = False
    payload["start"]["known_customer_behavior"] = "confirm_in_downstream"
    payload["texts"]["ru"]["start_prompt"] = "Представьтесь, пожалуйста."

    response = ai_client.put(
        "/ai/voice/config/name-collection",
        headers=admin_headers(),
        json=payload,
    )

    assert response.status_code == 200
    saved = response.json()
    assert saved["source"] == "database"
    assert saved["updated_at"]
    assert saved["config"]["enabled"] is False
    assert saved["config"]["start"]["known_customer_behavior"] == "confirm_in_downstream"
    assert saved["config"]["texts"]["ru"]["start_prompt"] == "Представьтесь, пожалуйста."

    session = get_session()
    try:
        row = session.execute(select(VoiceNameCollectionSettingsRow)).scalar_one()
        assert "Представьтесь, пожалуйста." in row.config_json
    finally:
        session.close()


def test_voice_name_collection_config_put_rejects_invalid_name_template():
    ai_client = TestClient(ai_app)
    payload = voice_config_module.voice_name_collection_default_config().model_dump()
    payload["texts"]["ru"]["confirmation_greeting_template"] = "Подтвердите имя клиента."

    response = ai_client.put(
        "/ai/voice/config/name-collection",
        headers=admin_headers(),
        json=payload,
    )

    assert response.status_code == 422


def test_voice_tts_config_get_returns_defaults_when_not_persisted():
    ai_client = TestClient(ai_app)

    response = ai_client.get("/ai/voice/config/tts", headers=admin_headers())

    assert response.status_code == 200
    payload = response.json()
    assert payload["source"] == "defaults"
    assert payload["updated_at"] is None
    assert payload["config"]["provider"] == "yandex"
    assert "elevenlabs" in payload["provider_options"]
    assert payload["voice_options"]["elevenlabs"]["ru"][0]["label"] == "Brian"


def test_voice_tts_config_put_persists_custom_payload():
    ai_client = TestClient(ai_app)
    payload = voice_tts_config_module.voice_tts_default_config().model_dump()
    payload["provider"] = "elevenlabs"
    payload["elevenlabs"]["ru"]["voice"] = "nPczCjzI2devNBz1zQrb"
    payload["elevenlabs"]["ru"]["model_id"] = "eleven_v3"
    payload["elevenlabs"]["kz"]["language_code"] = "kk"

    response = ai_client.put(
        "/ai/voice/config/tts",
        headers=admin_headers(),
        json=payload,
    )

    assert response.status_code == 200
    saved = response.json()
    assert saved["source"] == "database"
    assert saved["config"]["provider"] == "elevenlabs"
    assert saved["config"]["elevenlabs"]["ru"]["voice"] == "nPczCjzI2devNBz1zQrb"
    assert saved["config"]["elevenlabs"]["ru"]["model_id"] == "eleven_v3"

    session = get_session()
    try:
        row = session.execute(select(VoiceTTSSettingsRow)).scalar_one()
        assert "elevenlabs" in row.config_json
        assert "nPczCjzI2devNBz1zQrb" in row.config_json
    finally:
        session.close()


def test_ai_operator_config_get_returns_ainur_defaults():
    ai_client = TestClient(ai_app)

    response = ai_client.get("/ai/operator/config", headers=admin_headers())

    assert response.status_code == 200
    payload = response.json()
    assert payload["source"] == "defaults"
    assert payload["updated_at"] is None
    assert payload["config"]["agent_name"] == "Айнур"
    assert "Айнур" in payload["config"]["base_system_prompt"]
    assert "Айнур" in payload["config"]["identity_reply_ru"]


def test_ai_operator_config_put_persists_custom_prompt():
    ai_client = TestClient(ai_app)
    payload = ai_operator_config_module.ai_operator_default_config().model_dump()
    payload["company_name"] = "Kazakhtelecom"
    payload["base_system_prompt"] = "Ты {agent_name}, единый оператор {company_name}."
    payload["identity_reply_ru"] = "Я {agent_name}, оператор {company_name}."

    response = ai_client.put("/ai/operator/config", headers=admin_headers(), json=payload)

    assert response.status_code == 200
    saved = response.json()
    assert saved["source"] == "database"
    assert saved["config"]["company_name"] == "Kazakhtelecom"
    assert saved["config"]["base_system_prompt"] == "Ты {agent_name}, единый оператор {company_name}."

    session = get_session()
    try:
        row = session.execute(select(AIOperatorSettingsRow)).scalar_one()
        assert "Kazakhtelecom" in row.config_json
    finally:
        session.close()


def test_identity_question_uses_same_ainur_reply_without_handoff():
    operator_config = ai_operator_config_module.ai_operator_default_config()

    decision = ai_module._decide_reply(
        customer=None,
        interaction=SimpleNamespace(),
        thread=SimpleNamespace(),
        messages=[SimpleNamespace(author_type="customer", text="Скажи мне, кто ты такой")],
        kb_results=[],
        language="ru",
        operator_config=operator_config,
    )

    assert decision["intent"] == "identity_question"
    assert "Айнур" in decision["reply_text"]
    assert decision["needs_handoff"] is False
    assert decision["_model"] == "operator_identity_policy"


def test_voice_start_disabled_hands_off_without_prompt():
    session = get_session()
    try:
        config = voice_config_module.voice_name_collection_default_config().model_dump()
        config["enabled"] = False
        voice_config_module.save_voice_name_collection_config(session, config)
    finally:
        session.close()

    seeded = seed_voice_start_session(marker=f"voice_start_disabled_{new_id('seed')}")

    started = voice_module.start_voice_session(
        seeded["session_id"],
        VoiceAIStartIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            customer_id=None,
            language_hint="ru",
            agent_profile="voice_start",
            metadata={
                "stage": "voice_start",
                "next_queue_code": seeded["next_queue_code"],
                "next_queue_id": seeded["next_queue_id"],
            },
        ),
    )

    assert started.needs_handoff is True
    assert started.greeting_text == ""
    assert started.metadata["customer_name_status"] == "name_not_obtained"
    assert started.metadata["customer_name_value"] is None


def test_voice_start_known_customer_can_require_downstream_confirmation():
    session = get_session()
    try:
        config = voice_config_module.voice_name_collection_default_config().model_dump()
        config["start"]["known_customer_behavior"] = "confirm_in_downstream"
        voice_config_module.save_voice_name_collection_config(session, config)
    finally:
        session.close()

    seeded = seed_voice_start_session(
        marker=f"voice_start_known_{new_id('seed')}",
        customer_display_name="Айдос",
        caller_name="Текущий caller",
    )

    started = voice_module.start_voice_session(
        seeded["session_id"],
        VoiceAIStartIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            customer_id=seeded["customer_id"],
            language_hint="ru",
            agent_profile="voice_start",
            metadata={
                "stage": "voice_start",
                "next_queue_code": seeded["next_queue_code"],
                "next_queue_id": seeded["next_queue_id"],
            },
        ),
    )

    assert started.needs_handoff is True
    assert started.start_result.customer_name_status == "name_followup_required"
    assert started.start_result.customer_name_value == "Айдос"
    assert started.start_result.customer_name_source == "known_customer"


def test_voice_start_unknown_customer_can_skip_start_prompt():
    session = get_session()
    try:
        config = voice_config_module.voice_name_collection_default_config().model_dump()
        config["start"]["unknown_customer_behavior"] = "skip_to_downstream"
        voice_config_module.save_voice_name_collection_config(session, config)
    finally:
        session.close()

    seeded = seed_voice_start_session(marker=f"voice_start_skip_{new_id('seed')}")

    started = voice_module.start_voice_session(
        seeded["session_id"],
        VoiceAIStartIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            customer_id=None,
            language_hint="ru",
            agent_profile="voice_start",
            metadata={
                "stage": "voice_start",
                "next_queue_code": seeded["next_queue_code"],
                "next_queue_id": seeded["next_queue_id"],
            },
        ),
    )

    assert started.needs_handoff is True
    assert started.greeting_text == ""
    assert started.start_result.customer_name_status == "name_not_obtained"


def test_downstream_voice_turn_can_disable_inline_name_followup():
    session = get_session()
    try:
        config = voice_config_module.voice_name_collection_default_config().model_dump()
        config["downstream"]["missing_name_behavior"] = "do_not_ask"
        voice_config_module.save_voice_name_collection_config(session, config)
    finally:
        session.close()

    seeded = seed_voice_downstream_session(
        marker=f"voice_no_inline_{new_id('seed')}",
        name_status="name_not_obtained",
        customer_display_name="+77010009999",
    )

    decision = voice_module.turn_voice_session(
        seeded["session_id"],
        VoiceAITurnIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            transcript_text="Хочу узнать график работы",
            language="ru",
            sequence_no=1,
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert decision.metadata["customer_name_status"] == "name_not_obtained"
    assert "как мне к вам обращаться" not in decision.reply_text.lower()


def test_downstream_voice_start_can_finalize_uncertain_name_immediately():
    session = get_session()
    try:
        config = voice_config_module.voice_name_collection_default_config().model_dump()
        config["downstream"]["uncertain_name_behavior"] = "finalize_immediately"
        voice_config_module.save_voice_name_collection_config(session, config)
    finally:
        session.close()

    seeded = seed_voice_downstream_session(
        marker=f"voice_finalize_now_{new_id('seed')}",
        name_status="name_followup_required",
        name_value="Айдос",
        name_source="voice_start",
        customer_display_name="+77010009999",
    )

    started = voice_module.start_voice_session(
        seeded["session_id"],
        VoiceAIStartIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            customer_id=seeded["customer_id"],
            language_hint="ru",
            agent_profile="voice_support",
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert started.metadata["customer_name_status"] == "name_obtained"
    assert started.metadata["customer_name_value"] == "Айдос"
    assert "Айдос" in started.greeting_text


def test_downstream_voice_start_can_discard_uncertain_name_candidate():
    session = get_session()
    try:
        config = voice_config_module.voice_name_collection_default_config().model_dump()
        config["downstream"]["uncertain_name_behavior"] = "discard_and_collect"
        voice_config_module.save_voice_name_collection_config(session, config)
    finally:
        session.close()

    seeded = seed_voice_downstream_session(
        marker=f"voice_discard_name_{new_id('seed')}",
        name_status="name_followup_required",
        name_value="Айдос",
        name_source="voice_start",
        customer_display_name="+77010009999",
    )

    started = voice_module.start_voice_session(
        seeded["session_id"],
        VoiceAIStartIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            customer_id=seeded["customer_id"],
            language_hint="ru",
            agent_profile="voice_support",
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert started.metadata["customer_name_status"] == "name_not_obtained"
    assert started.metadata["customer_name_value"] is None
    assert "Айдос" not in started.greeting_text


def test_custom_voice_name_texts_are_used_in_start_and_followup():
    session = get_session()
    try:
        config = voice_config_module.voice_name_collection_default_config().model_dump()
        config["texts"]["ru"]["start_prompt"] = "Представьтесь, пожалуйста."
        config["texts"]["ru"]["inline_followup_prompt"] = "Как к вам обращаться сейчас?"
        config["texts"]["ru"]["personalized_greeting_template"] = "Здравствуйте, {name}. Чем помочь дальше?"
        config["texts"]["ru"]["confirmation_greeting_template"] = "Правильно понял, вас зовут {name}? Чем помочь?"
        voice_config_module.save_voice_name_collection_config(session, config)
    finally:
        session.close()

    start_seed = seed_voice_start_session(marker=f"voice_custom_start_{new_id('seed')}")
    started = voice_module.start_voice_session(
        start_seed["session_id"],
        VoiceAIStartIn(
            voice_session_id=start_seed["session_id"],
            call_id=start_seed["call_id"],
            interaction_id=start_seed["interaction_id"],
            customer_id=None,
            language_hint="ru",
            agent_profile="voice_start",
            metadata={
                "stage": "voice_start",
                "next_queue_code": start_seed["next_queue_code"],
                "next_queue_id": start_seed["next_queue_id"],
            },
        ),
    )
    assert started.greeting_text == "Представьтесь, пожалуйста."

    downstream_seed = seed_voice_downstream_session(
        marker=f"voice_custom_followup_{new_id('seed')}",
        name_status="name_not_obtained",
        customer_display_name="+77010009999",
    )
    decision = voice_module.turn_voice_session(
        downstream_seed["session_id"],
        VoiceAITurnIn(
            voice_session_id=downstream_seed["session_id"],
            call_id=downstream_seed["call_id"],
            interaction_id=downstream_seed["interaction_id"],
            transcript_text="Хочу узнать график работы",
            language="ru",
            sequence_no=1,
            metadata={"voice_start_language": "ru"},
        ),
    )
    assert "как к вам обращаться сейчас" not in decision.reply_text.lower()
    assert "город" in decision.reply_text.lower()


def test_downstream_voice_start_personalizes_greeting_and_finalizes_confirmed_name():
    seeded = seed_voice_downstream_session(
        marker=f"voice_start_{new_id('seed')}",
        name_status="name_obtained",
        name_value="айдос",
        name_source="voice_start",
        customer_display_name="+77010009999",
    )

    started = voice_module.start_voice_session(
        seeded["session_id"],
        VoiceAIStartIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            customer_id=seeded["customer_id"],
            language_hint="ru",
            agent_profile="voice_support",
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert "Айдос" in started.greeting_text
    assert "как мне к вам обращаться" not in started.greeting_text.lower()
    assert started.metadata["customer_name_status"] == "name_obtained"
    assert started.metadata["customer_name_value"] == "Айдос"

    session = get_session()
    try:
        customer = session.execute(
            select(Customer).where(Customer.customer_id == seeded["customer_id"])
        ).scalar_one()
        identity = session.execute(
            select(CustomerExternalIdentity).where(
                CustomerExternalIdentity.customer_id == seeded["customer_id"],
                CustomerExternalIdentity.channel == "voice",
            )
        ).scalar_one()
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == seeded["session_id"])
        ).scalar_one()
        assert customer.display_name == "Айдос"
        assert identity.display_name_snapshot == "Айдос"
        assert voice_session.customer_name_value == "Айдос"
    finally:
        session.close()


def test_downstream_voice_turn_adds_inline_name_followup_then_finalizes_provided_name():
    seeded = seed_voice_downstream_session(
        marker=f"voice_inline_{new_id('seed')}",
        name_status="name_not_obtained",
        customer_display_name="+77010009999",
    )

    first_turn = voice_module.turn_voice_session(
        seeded["session_id"],
        VoiceAITurnIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            transcript_text="Хочу узнать график работы",
            language="ru",
            sequence_no=1,
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert first_turn.metadata["customer_name_status"] == "name_not_obtained"
    assert "как мне к вам обращаться" not in first_turn.reply_text.lower()
    assert "город" in first_turn.reply_text.lower()

    second_turn = voice_module.turn_voice_session(
        seeded["session_id"],
        VoiceAITurnIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            transcript_text="Меня зовут Айдос, хочу узнать график работы",
            language="ru",
            sequence_no=2,
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert second_turn.metadata["customer_name_status"] == "name_obtained"
    assert second_turn.metadata["customer_name_value"] == "Айдос"
    assert second_turn.metadata["customer_name_source"] == "voice_followup"
    assert "Айдос" in second_turn.reply_text

    session = get_session()
    try:
        customer = session.execute(
            select(Customer).where(Customer.customer_id == seeded["customer_id"])
        ).scalar_one()
        voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == seeded["session_id"])
        ).scalar_one()
        assert customer.display_name == "Айдос"
        assert voice_session.customer_name_status == "name_obtained"
        assert voice_session.customer_name_source == "voice_followup"
        assert voice_session.customer_name_value == "Айдос"
    finally:
        session.close()


def test_downstream_voice_turn_extracts_explicit_name_without_restarting_name_flow():
    seeded = seed_voice_downstream_session(
        marker=f"voice_explicit_name_intent_{new_id('seed')}",
        name_status="name_not_obtained",
        customer_display_name="+77010009999",
    )

    turn = voice_module.turn_voice_session(
        seeded["session_id"],
        VoiceAITurnIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            transcript_text="Меня зовут Ания, мне нужно узнать график работы",
            language="ru",
            sequence_no=1,
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert turn.metadata["customer_name_status"] == "name_obtained"
    assert turn.metadata["customer_name_value"] == "Ания"
    assert "город" in turn.reply_text.lower()
    assert "филиал" in turn.reply_text.lower()
    assert "какой вопрос по работе" not in turn.reply_text.lower()
    assert "как мне к вам обращаться" not in turn.reply_text.lower()


@pytest.mark.parametrize(
    ("transcript_text", "expected_name"),
    [
        ("Да, Айдос", "Айдос"),
        ("Нет, меня зовут Марат", "Марат"),
    ],
)
def test_downstream_voice_turn_resolves_followup_name(transcript_text: str, expected_name: str):
    seeded = seed_voice_downstream_session(
        marker=f"voice_followup_{new_id('seed')}",
        name_status="name_followup_required",
        name_value="Айдос",
        name_source="voice_start",
        customer_display_name="+77010009999",
    )

    started = voice_module.start_voice_session(
        seeded["session_id"],
        VoiceAIStartIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            customer_id=seeded["customer_id"],
            language_hint="ru",
            agent_profile="voice_support",
            metadata={"voice_start_language": "ru"},
        ),
    )
    assert "Айдос" in started.greeting_text

    decision = voice_module.turn_voice_session(
        seeded["session_id"],
        VoiceAITurnIn(
            voice_session_id=seeded["session_id"],
            call_id=seeded["call_id"],
            interaction_id=seeded["interaction_id"],
            transcript_text=transcript_text,
            language="ru",
            sequence_no=1,
            metadata={"voice_start_language": "ru"},
        ),
    )

    assert decision.metadata["customer_name_status"] == "name_obtained"
    assert decision.metadata["customer_name_value"] == expected_name
    assert decision.metadata["customer_name_source"] == "voice_followup"
    assert expected_name in decision.reply_text

    session = get_session()
    try:
        customer = session.execute(
            select(Customer).where(Customer.customer_id == seeded["customer_id"])
        ).scalar_one()
        assert customer.display_name == expected_name
    finally:
        session.close()
