realtime_voice_service/core/session.py

from __future__ import annotations

import asyncio
import contextlib
import datetime
import difflib
import io
import logging
import os
import re
import time
import wave
from collections.abc import AsyncGenerator
from collections.abc import Iterable
from enum import Enum

from realtime_voice_service.core.filler_audio import FillerAudioLibrary
from realtime_voice_service.core.kazakh_names import TOP_KAZAKH_NAMES
from realtime_voice_service.core.vad import BaseVAD, SileroVADDetector
from realtime_voice_service.providers.base import BaseLLM, BaseSTT, BaseSTTStream, BaseTTS, MockLLM, MockSTT, MockTTS
from realtime_voice_service.transports.base import BaseMediaTransport


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


def _read_bounded_int_env(name: str, default: int, *, minimum: int, maximum: int) -> int:
    raw = str(os.getenv(name, str(default))).strip()
    try:
        value = int(float(raw))
    except Exception:
        value = default
    return max(minimum, min(value, maximum))


def _semantic_hold_ms() -> int:
    return _read_bounded_int_env("SEMANTIC_ENDPOINTING_HOLD_MS", 700, minimum=200, maximum=2000)


def _filler_delay_ms() -> int:
    return _read_bounded_int_env("REALTIME_VOICE_FILLER_DELAY_MS", 250, minimum=0, maximum=3000)


def _tts_chunk_soft_min_chars() -> int:
    return _read_bounded_int_env("REALTIME_VOICE_TTS_CHUNK_SOFT_MIN_CHARS", 10, minimum=1, maximum=80)


def _tts_chunk_soft_min_words() -> int:
    return _read_bounded_int_env("REALTIME_VOICE_TTS_CHUNK_SOFT_MIN_WORDS", 2, minimum=1, maximum=10)


SEMANTIC_ENDPOINTING_HOLD_MS = _semantic_hold_ms()
FILLER_AUDIO_DELAY_MS = _filler_delay_ms()
TTS_CHUNK_SOFT_MIN_CHARS = _tts_chunk_soft_min_chars()
TTS_CHUNK_SOFT_MIN_WORDS = _tts_chunk_soft_min_words()

_COURTESY_OR_FAREWELL_MARKERS: frozenset[str] = frozenset({
    "спасибо",
    "благодарю",
    "до свидания",
    "всего доброго",
    "пока",
    "до встречи",
    "хорошего дня",
    "ничего не нужно",
    "ничего не надо",
    "мне ничего не нужно",
    "мне ничего не надо",
    "больше ничего",
    "это все",
    "это всё",
    "все понятно",
    "всё понятно",
    "вопросов нет",
    "не надо",
    "не нужно",
    "goodbye",
    "bye",
})

_TERMINAL_DIRECT_FAREWELL_MARKERS: frozenset[str] = frozenset({
    "до свидания",
    "всего доброго",
    "пока",
    "до встречи",
    "хорошего дня",
    "goodbye",
    "bye",
})

_TERMINAL_CLOSE_REQUEST_MARKERS: frozenset[str] = frozenset({
    "можно завершить",
    "можем завершить",
    "можете завершать",
    "давайте завершим",
    "давайте закончим",
    "завершить разговор",
    "закончить разговор",
    "заканчиваем разговор",
    "завершайте разговор",
    "кладу трубку",
    "положу трубку",
})

_TERMINAL_NO_MORE_HELP_MARKERS: frozenset[str] = frozenset({
    "ничего не нужно",
    "ничего не надо",
    "мне ничего не нужно",
    "мне ничего не надо",
    "больше ничего",
    "больше ничего не нужно",
    "больше ничего не надо",
    "ничего больше не нужно",
    "ничего больше не надо",
    "это все",
    "это всё",
    "на этом все",
    "на этом всё",
    "вопросов нет",
    "больше вопросов нет",
})

_FAREWELL_QUERY_MARKERS: frozenset[str] = frozenset({
    "вопрос",
    "подскаж",
    "скажи",
    "почему",
    "как",
    "что",
    "где",
    "когда",
    "можно",
    "нужно ли",
    "нужно еще",
    "нужно ещё",
    "еще",
    "ещё",
    "теперь",
    "тогда",
    "адрес",
    "тариф",
    "оператор",
    "статус",
    "проблем",
    "помощ",
    "помочь",
    "узнать",
    "нужен",
    "нужна",
    "нужны",
})

_FAREWELL_HARD_QUERY_MARKERS: frozenset[str] = frozenset(
    marker for marker in _FAREWELL_QUERY_MARKERS if marker not in {"вопрос", "можно"}
)
FAREWELL_RESPONSE_TEXT = "Спасибо за обращение в DigiOps. Всего доброго, до свидания!"


def _contains_voice_phrase(normalized_text: str, phrase: str) -> bool:
    phrase_parts = [re.escape(part) for part in phrase.split()]
    if not phrase_parts:
        return False
    phrase_pattern = r"\s+".join(phrase_parts)
    return re.search(
        rf"(?<![^\W\d_]){phrase_pattern}(?![^\W\d_])",
        normalized_text,
        flags=re.UNICODE,
    ) is not None


def _ends_with_voice_phrase(normalized_text: str, phrase: str) -> bool:
    phrase_parts = [re.escape(part) for part in phrase.split()]
    if not phrase_parts:
        return False
    phrase_pattern = r"\s+".join(phrase_parts)
    return re.search(
        rf"(?<![^\W\d_]){phrase_pattern}(?:\s+(?:спасибо|благодарю))?$",
        normalized_text,
        flags=re.UNICODE,
    ) is not None


def _is_farewell_transcript(text: str) -> bool:
    normalized = _voice_text_key(text)
    return any(_contains_voice_phrase(normalized, marker) for marker in _COURTESY_OR_FAREWELL_MARKERS)


def _is_terminal_farewell(text: str) -> bool:
    normalized = _voice_text_key(text)
    has_direct_farewell = any(
        _contains_voice_phrase(normalized, marker)
        for marker in _TERMINAL_DIRECT_FAREWELL_MARKERS
    )
    has_close_request = any(
        _contains_voice_phrase(normalized, marker)
        for marker in _TERMINAL_CLOSE_REQUEST_MARKERS
    )
    has_no_more_help = any(
        _ends_with_voice_phrase(normalized, marker)
        for marker in _TERMINAL_NO_MORE_HELP_MARKERS
    )
    if not (has_direct_farewell or has_close_request or has_no_more_help):
        return False
    query_markers = _FAREWELL_QUERY_MARKERS if has_direct_farewell else _FAREWELL_HARD_QUERY_MARKERS
    if any(marker in normalized for marker in query_markers):
        return False
    word_count = len(re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", normalized, flags=re.UNICODE))
    return word_count <= 16


SEMANTIC_CONTINUATION_TOKENS = {
    "а",
    "в",
    "во",
    "да",
    "до",
    "и",
    "или",
    "к",
    "ко",
    "на",
    "но",
    "ну",
    "о",
    "об",
    "от",
    "по",
    "при",
    "с",
    "со",
    "так",
    "то",
    "у",
    "что",
    "чтобы",
    "эээ",
    "ммм",
}


def _preview_text(text: str, *, limit: int = 160) -> str:
    normalized = " ".join(str(text or "").split())
    if len(normalized) <= limit:
        return normalized
    return f"{normalized[:limit]}..."


_KAZAKH_SPECIFIC_LETTERS = set("әғқңөұүһі")
_RUSSIAN_SPECIFIC_LETTERS = set("ыэъё")
_KAZAKH_LANGUAGE_MARKERS = frozenset({
    "казахский",
    "казахском",
    "казахски",
    "казакша",
    "казак",
    "қазақ",
    "қазақша",
    "сәлем",
    "салем",
    "сәлеметсіз",
    "салеметсиз",
    "рахмет",
    "жақсы",
    "жаксы",
    "иә",
    "ия",
    "жоқ",
    "жок",
    "менің",
    "менин",
    "атым",
    "маған",
    "маган",
    "керек",
    "көмек",
    "комек",
    "сұрақ",
    "сурак",
})
_RUSSIAN_LANGUAGE_MARKERS = frozenset({
    "русский",
    "русском",
    "русски",
    "орыс",
    "орысша",
})
_ARABIC_SCRIPT_RE = re.compile(r"[\u0600-\u06ff]")
_CYRILLIC_SCRIPT_RE = re.compile(r"[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]")


def _detect_preferred_session_language(text: str) -> str | None:
    normalized = _voice_text_key(text)
    if not normalized:
        return None
    words = set(normalized.split())
    if words & _KAZAKH_LANGUAGE_MARKERS:
        return "kk"
    if words & _RUSSIAN_LANGUAGE_MARKERS:
        return "ru"
    if re.search(r"\bпо\s+русски\b", normalized):
        return "ru"
    if re.search(r"\bна\s+казахском\b", normalized):
        return "kk"
    if re.search(r"\bна\s+русском\b", normalized):
        return "ru"
    return None




def _detect_name_collection_language(text: str) -> str | None:
    normalized = str(text or "").strip().lower()
    if not normalized:
        return None
    preferred_language = _detect_preferred_session_language(normalized)
    if preferred_language:
        return preferred_language
    if any(ch in _KAZAKH_SPECIFIC_LETTERS for ch in normalized):
        return "kk"
    return None

def _detect_session_language(text: str) -> str | None:
    normalized = str(text or "").strip().lower()
    if not normalized:
        return None
    preferred_language = _detect_preferred_session_language(normalized)
    if preferred_language:
        return preferred_language
    has_kazakh = any(ch in _KAZAKH_SPECIFIC_LETTERS for ch in normalized)
    if has_kazakh:
        return "kk"
    has_russian = any(ch in _RUSSIAN_SPECIFIC_LETTERS for ch in normalized)
    if has_russian:
        return "ru"
    return None


def _stt_language_candidate_score(text: str, expected_language: str) -> int:
    normalized_language = expected_language.lower()
    normalized_text = str(text or "").strip().lower()
    if not normalized_text:
        return -100

    score = 0
    detected_preference = _detect_preferred_session_language(normalized_text)
    if detected_preference == normalized_language:
        score += 40
    elif detected_preference:
        score -= 40

    if any(ch in _KAZAKH_SPECIFIC_LETTERS for ch in normalized_text):
        score += 12 if normalized_language == "kk" else -8
    if any(ch in _RUSSIAN_SPECIFIC_LETTERS for ch in normalized_text):
        score += 8 if normalized_language == "ru" else -4
    if _CYRILLIC_SCRIPT_RE.search(normalized_text):
        score += 4
    if _ARABIC_SCRIPT_RE.search(normalized_text):
        score -= 30

    latin_chars = sum(1 for ch in normalized_text if "a" <= ch <= "z")
    cyrillic_chars = len(_CYRILLIC_SCRIPT_RE.findall(normalized_text))
    if latin_chars > cyrillic_chars and latin_chars >= 6:
        score -= 12

    return score


def _choose_language_specific_transcript(ru_transcript: str, kk_transcript: str) -> tuple[str, str | None, int, int]:
    ru_text = str(ru_transcript or "").strip()
    kk_text = str(kk_transcript or "").strip()
    ru_score = _stt_language_candidate_score(ru_text, "ru")
    kk_score = _stt_language_candidate_score(kk_text, "kk")

    if ru_score >= kk_score and ru_text:
        detected_language = _detect_name_collection_language(ru_text) or _detect_session_language(ru_text)
        return ru_text, detected_language, ru_score, kk_score
    if kk_text:
        detected_language = _detect_name_collection_language(kk_text) or _detect_session_language(kk_text)
        return kk_text, detected_language, ru_score, kk_score
    return ru_text or kk_text, None, ru_score, kk_score


_TTS_LANGUAGE_CODE_MAP = {
    "ru": "ru",
    "kk": "kk",
}

_STT_YANDEX_LANGUAGE_MAP = {
    "ru": "ru-RU",
    "kk": "kk-KZ",
}

_STT_ELEVENLABS_LANGUAGE_MAP = {
    "ru": "rus",
    "kk": "kaz",
}


def _tts_language_for(session_language: str | None) -> str | None:
    if not session_language:
        return None
    return _TTS_LANGUAGE_CODE_MAP.get(session_language.lower())


def _stt_language_for(session_language: str | None) -> str | None:
    if not session_language:
        return None
    return session_language.lower() if session_language.lower() in {"ru", "kk"} else None


def _audio_duration_ms(audio_bytes: bytes, *, sample_rate_hz: int) -> int:
    if not audio_bytes or sample_rate_hz <= 0:
        return 0
    return int(((len(audio_bytes) // 2) / float(sample_rate_hz)) * 1000.0)


def _audio_byte_count_duration_ms(byte_count: int, *, sample_rate_hz: int) -> int:
    if byte_count <= 0 or sample_rate_hz <= 0:
        return 0
    return int(((byte_count // 2) / float(sample_rate_hz)) * 1000.0)


RU_DIGIT_WORDS = ("ноль", "один", "два", "три", "четыре", "пять", "шесть", "семь", "восемь", "девять")
RU_UNITS_MASCULINE = ("", "один", "два", "три", "четыре", "пять", "шесть", "семь", "восемь", "девять")
RU_UNITS_FEMININE = ("", "одна", "две", "три", "четыре", "пять", "шесть", "семь", "восемь", "девять")
RU_TEENS = (
    "десять",
    "одиннадцать",
    "двенадцать",
    "тринадцать",
    "четырнадцать",
    "пятнадцать",
    "шестнадцать",
    "семнадцать",
    "восемнадцать",
    "девятнадцать",
)
RU_TENS = ("", "", "двадцать", "тридцать", "сорок", "пятьдесят", "шестьдесят", "семьдесят", "восемьдесят", "девяносто")
RU_HUNDREDS = (
    "",
    "сто",
    "двести",
    "триста",
    "четыреста",
    "пятьсот",
    "шестьсот",
    "семьсот",
    "восемьсот",
    "девятьсот",
)


RU_DAY_ORDINALS = {
    1: "первое",
    2: "второе",
    3: "третье",
    4: "четвертое",
    5: "пятое",
    6: "шестое",
    7: "седьмое",
    8: "восьмое",
    9: "девятое",
    10: "десятое",
    11: "одиннадцатое",
    12: "двенадцатое",
    13: "тринадцатое",
    14: "четырнадцатое",
    15: "пятнадцатое",
    16: "шестнадцатое",
    17: "семнадцатое",
    18: "восемнадцатое",
    19: "девятнадцатое",
    20: "двадцатое",
    21: "двадцать первое",
    22: "двадцать второе",
    23: "двадцать третье",
    24: "двадцать четвертое",
    25: "двадцать пятое",
    26: "двадцать шестое",
    27: "двадцать седьмое",
    28: "двадцать восьмое",
    29: "двадцать девятое",
    30: "тридцатое",
    31: "тридцать первое",
}
RU_MONTHS_GENITIVE = {
    1: "января",
    2: "февраля",
    3: "марта",
    4: "апреля",
    5: "мая",
    6: "июня",
    7: "июля",
    8: "августа",
    9: "сентября",
    10: "октября",
    11: "ноября",
    12: "декабря",
}


def _ru_plural(value: int, one: str, few: str, many: str) -> str:
    value = abs(value) % 100
    if 11 <= value <= 19:
        return many
    last_digit = value % 10
    if last_digit == 1:
        return one
    if 2 <= last_digit <= 4:
        return few
    return many


def _ru_under_1000(value: int, *, feminine: bool = False) -> list[str]:
    value = max(min(int(value), 999), 0)
    words: list[str] = []
    hundreds = value // 100
    if hundreds:
        words.append(RU_HUNDREDS[hundreds])
    remainder = value % 100
    if 10 <= remainder <= 19:
        words.append(RU_TEENS[remainder - 10])
        return words
    tens = remainder // 10
    units = remainder % 10
    if tens:
        words.append(RU_TENS[tens])
    if units:
        words.append((RU_UNITS_FEMININE if feminine else RU_UNITS_MASCULINE)[units])
    return words


def _ru_int_words(value: int) -> str:
    value = int(value)
    if value == 0:
        return RU_DIGIT_WORDS[0]
    if value < 0:
        return f"минус {_ru_int_words(abs(value))}"

    words: list[str] = []
    billions = value // 1_000_000_000
    if billions:
        words.extend(_ru_under_1000(billions))
        words.append(_ru_plural(billions, "миллиард", "миллиарда", "миллиардов"))
    value %= 1_000_000_000

    millions = value // 1_000_000
    if millions:
        words.extend(_ru_under_1000(millions))
        words.append(_ru_plural(millions, "миллион", "миллиона", "миллионов"))
    value %= 1_000_000

    thousands = value // 1000
    if thousands:
        words.extend(_ru_under_1000(thousands, feminine=True))
        words.append(_ru_plural(thousands, "тысяча", "тысячи", "тысяч"))
    value %= 1000

    if value:
        words.extend(_ru_under_1000(value))
    return " ".join(word for word in words if word)


def _ru_digit_sequence(raw: str) -> str:
    return " ".join(RU_DIGIT_WORDS[int(char)] for char in raw if char.isdigit())


def _ru_number_words(raw: str) -> str:
    compact = str(raw or "").strip().replace(" ", "")
    decimal_match = re.fullmatch(r"(\d+)[,.](\d+)", compact)
    if decimal_match:
        integer_part = _ru_number_words(decimal_match.group(1))
        fractional_part = _ru_digit_sequence(decimal_match.group(2))
        return f"{integer_part} целых {fractional_part}"

    digits = re.sub(r"\D+", "", compact)
    if not digits:
        return str(raw or "")
    if digits.startswith("0") or len(digits) > 6:
        return _ru_digit_sequence(digits)
    return _ru_int_words(int(digits))


def _ru_genitive_number_words(raw: str) -> str:
    digits = re.sub(r"\D+", "", str(raw or ""))
    if not digits:
        return _ru_number_words(raw)
    value = int(digits)
    small_genitive = {
        1: "одного",
        2: "двух",
        3: "трех",
        4: "четырех",
        5: "пяти",
        6: "шести",
        7: "семи",
        8: "восьми",
        9: "девяти",
        10: "десяти",
    }
    return small_genitive.get(value, _ru_number_words(raw))


def _ru_percent_phrase(raw: str) -> str:
    words = _ru_number_words(raw)
    compact = str(raw or "").strip().replace(" ", "")
    if re.fullmatch(r"\d+[,.]\d+", compact):
        return f"{words} процента"
    digits = re.sub(r"\D+", "", compact)
    if not digits or digits.startswith("0") or len(digits) > 6:
        return f"{words} процентов"
    value = int(digits)
    return f"{words} {_ru_plural(value, 'процент', 'процента', 'процентов')}"


def _ru_date_words(day_raw: str, month_raw: str, year_raw: str) -> str:
    day = int(day_raw)
    month = int(month_raw)
    year = int(year_raw)
    day_words = RU_DAY_ORDINALS.get(day, _ru_number_words(str(day)))
    month_words = RU_MONTHS_GENITIVE.get(month, _ru_number_words(str(month)))
    return f"{day_words} {month_words} {_ru_number_words(str(year))}"


VOICE_PRONUNCIATION_REPLACEMENTS: tuple[tuple[str, str], ...] = (
    (r"\bDigiOps\b", "ДиджиОпс"),
    (r"\bA\.?I\.?\b", "эй-ай"),
    (r"\bAI\b", "эй-ай"),
    (r"\bCRM\b", "си-ар-эм"),
    (r"\bB2B\b", "би-ту-би"),
    (r"\bIT\b", "ай-ти"),
)


def _normalize_voice_pronunciation(text: str) -> str:
    normalized = str(text or "")
    for pattern, replacement in VOICE_PRONUNCIATION_REPLACEMENTS:
        normalized = re.sub(pattern, replacement, normalized, flags=re.IGNORECASE)
    return normalized


def _ru_phone_words(raw: str) -> str:
    payload = str(raw or "")
    digits = re.sub(r"\D+", "", payload)
    if len(digits) < 7:
        return _ru_number_words(payload)
    prefix = "плюс " if payload.strip().startswith("+") else ""
    return f"{prefix}{_ru_digit_sequence(digits)}".strip()


def _normalize_voice_numbers(text: str) -> str:
    normalized = str(text or "")
    normalized = re.sub(
        r"(?<!\w)(\+?\d(?:[\s()\-]*\d){6,})(?!\w)",
        lambda match: _ru_phone_words(match.group(1)),
        normalized,
    )
    normalized = re.sub(
        r"(?<!\w)(\d{1,2})[.](\d{1,2})[.](\d{2,4})(?!\w)",
        lambda match: _ru_date_words(match.group(1), match.group(2), match.group(3)),
        normalized,
    )
    normalized = re.sub(
        r"(?<!\w)(\d+(?:[,.]\d+)?)\s*%",
        lambda match: _ru_percent_phrase(match.group(1)),
        normalized,
    )
    normalized = re.sub(
        r"(?<!\w)(\d+)\s*:\s*(\d+)(?!\w)",
        lambda match: f"{_ru_number_words(match.group(1))} {_ru_number_words(match.group(2))}",
        normalized,
    )
    normalized = re.sub(
        r"(?<!\w)(\d+[,.]\d+)\s*/\s*(\d+)(?!\w)",
        lambda match: f"{_ru_number_words(match.group(1))} из {_ru_genitive_number_words(match.group(2))}",
        normalized,
    )
    normalized = re.sub(
        r"(?<!\w)(\d+)\s*/\s*(\d+)(?!\w)",
        lambda match: f"{_ru_number_words(match.group(1))} из {_ru_genitive_number_words(match.group(2))}",
        normalized,
    )
    normalized = re.sub(
        r"(?<!\w)(\d+[,.]\d+)(?!\w)",
        lambda match: _ru_number_words(match.group(1)),
        normalized,
    )
    normalized = re.sub(
        r"(?<!\w)(\d+)(?!\w)",
        lambda match: _ru_number_words(match.group(1)),
        normalized,
    )
    return normalized


def _combine_user_transcripts(parts: Iterable[str]) -> str:
    cleaned = [re.sub(r"\s+", " ", str(part or "")).strip() for part in parts]
    cleaned = [part for part in cleaned if part]
    if len(cleaned) <= 1:
        return cleaned[0] if cleaned else ""
    normalized_parts: list[str] = []
    for part in cleaned:
        if normalized_parts and not normalized_parts[-1].endswith((".", "!", "?", "…")):
            normalized_parts[-1] = f"{normalized_parts[-1]}."
        normalized_parts.append(part)
    return " ".join(normalized_parts)


def _sanitize_voice_text(text: str) -> str:
    sanitized = str(text or "")
    sanitized = re.sub(r"\[([^\]]+)\]\((?:https?://|www\.)[^)\s]+[^)]*\)", r"\1", sanitized)
    sanitized = re.sub(r"(?:https?://|www\.)\S+", "", sanitized)
    sanitized = sanitized.replace("[", "").replace("]", "").replace("(", "").replace(")", "")
    sanitized = _normalize_voice_pronunciation(sanitized)
    sanitized = re.sub(r"\s+", " ", sanitized).strip()
    sanitized = _normalize_voice_numbers(sanitized)
    sanitized = re.sub(r"\s+", " ", sanitized).strip()
    return sanitized


DEFAULT_NAME_RETRY_TEXT = "Подскажите, пожалуйста, как я могу к вам обращаться?"
DEFAULT_PERSONALIZED_GREETING_TEMPLATE = '{name}, я работаю на основе искусственного интеллекта и постараюсь максимально внимательно разобраться с вашим вопросом. Чем я могу помочь?'
KAZAKH_NAME_RETRY_TEXT = "Өтінемін, сізге қалай жүгінсем болады?"
KAZAKH_PERSONALIZED_GREETING_TEMPLATE = "{name}, мен жасанды интеллект негізінде жұмыс істеймін және сұрағыңызды мұқият түсінуге тырысамын. Қалай көмектесе аламын?"
_NAME_CAPTURE_CONFUSION_ALIASES = {
    "парень": "Арнур",
    "арнер": "Арнур",
    "ар нур": "Арнур",
    "шомарт": "Жомарт",
    "шумарт": "Жомарт",
}


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


_KAZAKH_NAME_BY_KEY = {_voice_text_key(name): name for name in TOP_KAZAKH_NAMES}


def _voice_letters_key(text: str | None) -> str:
    return "".join(re.findall(r"[^\W\d_]+", str(text or "").lower(), flags=re.UNICODE))


def _canonical_name(text: str) -> str:
    words = [part for part in re.split(r"\s+", text.strip()) if part]
    return " ".join(word[:1].upper() + word[1:].lower() if len(word) > 1 else word.upper() for word in words)


def _looks_like_supported_name_text(text: str) -> bool:
    return bool(
        re.fullmatch(
            r"[A-Za-zА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]+(?:[ -][A-Za-zА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]+){0,2}",
            str(text or "").strip(),
        )
    )


def _correct_name_capture_candidate(name: str | None) -> str | None:
    candidate = _canonical_name(str(name or "").strip())
    if not candidate:
        return None
    key = _voice_text_key(candidate)
    exact_match = _KAZAKH_NAME_BY_KEY.get(key)
    if exact_match:
        return exact_match
    alias_match = _NAME_CAPTURE_CONFUSION_ALIASES.get(key)
    if alias_match:
        return alias_match

    best_name = None
    best_ratio = 0.0
    for gazetteer_key, gazetteer_name in _KAZAKH_NAME_BY_KEY.items():
        ratio = difflib.SequenceMatcher(None, key, gazetteer_key).ratio()
        if ratio > best_ratio:
            best_ratio = ratio
            best_name = gazetteer_name
    if best_name is None:
        return candidate
    threshold = 0.82 if len(key) >= 5 else 0.9
    if best_ratio >= threshold:
        LOGGER.info(
            "name capture candidate corrected via Kazakh names: candidate=%r corrected=%r ratio=%.3f",
            candidate,
            best_name,
            best_ratio,
        )
        return best_name
    return candidate


def _normalize_name_candidate(text: str | None) -> str | None:
    raw = str(text or "").strip(" \t\r\n,.;:!?\"'()[]{}")
    if not raw or any(ch.isdigit() for ch in raw):
        return None
    words = re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", raw, flags=re.UNICODE)
    if not words or len(words) > 4:
        return None
    lowered = [_voice_text_key(word) for word in words]
    stop_words = {
        "меня",
        "зовут",
        "это",
        "я",
        "мое",
        "моё",
        "имя",
        "менің",
        "атым",
        "аты",
        "болады",
        "язык",
        "тіл",
    }
    stop_words.update(_KAZAKH_LANGUAGE_MARKERS)
    stop_words.update(_RUSSIAN_LANGUAGE_MARKERS)
    filtered = [word for word, lowered_word in zip(words, lowered) if lowered_word not in stop_words]
    if not filtered:
        return None
    if len(filtered) == 2 and len(filtered[1]) == 1:
        filtered = [filtered[0]]
    elif len(filtered) == 2 and len(filtered[0]) == 1:
        filtered = [filtered[1]]
    invalid_tokens = {
        "да",
        "нет",
        "алло",
        "привет",
        "здравствуйте",
        "добрый",
        "день",
        "хочу",
        "хотел",
        "узнать",
        "помощь",
        "вопрос",
        "тариф",
        "статус",
        "оператор",
        "проблема",
        "интернет",
        "click",
        "noise",
        "beep",
        "tap",
        "tick",
        "hum",
        "um",
        "uh",
        "umm",
        "угу",
        "ага",
        "эм",
        "эмм",
        "эммм",
        "мм",
        "ммм",
        "мммм",
        "м",
        "ум",
        "ээ",
        "эээ",
        "ээээ",
        "аа",
        "ааа",
        "аааа",
        "хм",
        "хмм",
        "хммм",
        "звони",
        "слушаю",
        "слушаюсь",
        "хорошо",
        "подожди",
        "керек",
        "сұрақ",
        "көмек",
    }
    invalid_tokens.update(_KAZAKH_LANGUAGE_MARKERS)
    invalid_tokens.update(_RUSSIAN_LANGUAGE_MARKERS)
    invalid_letter_tokens = {_voice_letters_key(token) for token in invalid_tokens}
    filler_letters = {"а", "э", "е", "ё", "у", "о", "ы", "м", "m", "h"}
    for word in filtered:
        token_key = _voice_text_key(word)
        letters_key = _voice_letters_key(word)
        if token_key in invalid_tokens or letters_key in invalid_letter_tokens:
            return None
        if len(letters_key) >= 2 and len(set(letters_key)) == 1 and letters_key[0] in filler_letters:
            return None
        if letters_key in {"хм", "хмм", "хммм"}:
            return None
    if len(filtered) > 2:
        return None
    if len(filtered) == 2:
        similarity = difflib.SequenceMatcher(
            None,
            _voice_text_key(filtered[0]),
            _voice_text_key(filtered[1]),
        ).ratio()
        if similarity >= 0.72:
            return None
    candidate = _canonical_name(" ".join(filtered))
    if not _looks_like_supported_name_text(candidate):
        return None
    return candidate


def _name_followup_needed(text: str) -> bool:
    normalized = _voice_text_key(text)
    request_markers = (
        "хотел",
        "хочу",
        "нужн",
        "помог",
        "вопрос",
        "проблем",
        "тариф",
        "статус",
        "оператор",
        "адрес",
        "график",
        "интернет",
        "керек",
        "сұра",
        "көмек",
    )
    return any(marker in normalized for marker in request_markers)


def _extract_name_candidate(text: str | None) -> tuple[str | None, bool]:
    raw = str(text or "").strip()
    if not raw:
        return None, False
    if any(marker in raw for marker in ("[", "]", "<", ">")):
        return None, False
    normalized = _voice_text_key(raw)
    explicit_patterns = (
        r"(?:меня\s+зовут|мо[её]\s+имя|my name is|i am|this is)\s+(.+)",
        r"(?:менің\s+атым|аты[мң]?|mening atym)\s+(.+)",
    )
    cutoff_tokens = {
        "мне",
        "надо",
        "нужно",
        "хочу",
        "хотел",
        "узнать",
        "график",
        "работы",
        "адрес",
        "филиал",
        "город",
        "тариф",
        "статус",
        "оператор",
        "вопрос",
        "проблема",
        "интернет",
        "керек",
        "сұра",
        "көмек",
    }
    cutoff_tokens.update(_KAZAKH_LANGUAGE_MARKERS)
    cutoff_tokens.update(_RUSSIAN_LANGUAGE_MARKERS)
    for pattern in explicit_patterns:
        match = re.search(pattern, normalized, flags=re.IGNORECASE)
        if not match:
            continue
        tail = match.group(1).strip()
        tail_words = re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", tail, flags=re.UNICODE)
        candidate_words: list[str] = []
        for word in tail_words:
            if _voice_text_key(word) in cutoff_tokens:
                break
            candidate_words.append(word)
            if len(candidate_words) >= 3:
                break
        candidate = _normalize_name_candidate(" ".join(candidate_words)) or _normalize_name_candidate(tail)
        if candidate:
            return candidate, False

    candidate = _normalize_name_candidate(raw)
    if candidate:
        lower_candidate = candidate.lower()
        stopwords = {
            "здравствуйте",
            "привет",
            "алло",
            "да",
            "нет",
            "добрый",
            "день",
            "вопрос",
            "интернет",
            "у",
            "меня",
        }
        candidate_words = set(lower_candidate.split())
        if candidate_words.issubset(stopwords) or len(lower_candidate) < 2:
            return None, False
        word_count = len(re.findall(r"[^\W\d_]+(?:[-'][^\W\d_]+)*", raw, flags=re.UNICODE))
        if len(candidate.split()) == 1 and len(candidate) < 3:
            return None, False
        if word_count <= 2 and not _name_followup_needed(raw):
            return candidate, False
        return candidate, True

    return None, False


def _voice_start_name_outcome(text: str | None) -> tuple[str, str | None]:
    raw = str(text or "").strip()
    if not raw:
        return "name_not_obtained", None
    candidate, needs_followup = _extract_name_candidate(raw)
    if candidate:
        corrected_candidate = _correct_name_capture_candidate(candidate) or candidate
        if corrected_candidate != candidate:
            LOGGER.info(
                "name capture candidate normalized: candidate=%r corrected=%r transcript=%r",
                candidate,
                corrected_candidate,
                _preview_text(raw),
            )
        candidate = corrected_candidate
    if candidate and not needs_followup:
        return "name_obtained", candidate
    if candidate:
        return "name_followup_required", candidate
    if _name_followup_needed(raw):
        return "name_followup_required", None
    return "name_not_obtained", None


def _voice_short_name(name: str | None) -> str | None:
    canonical = _normalize_name_candidate(name)
    if not canonical:
        raw = str(name or "").strip()
        if not raw:
            return None
        canonical = _canonical_name(raw)
    short_name = canonical.split(" ", 1)[0].strip()
    return short_name or None


class SessionState(str, Enum):
    LISTENING = "LISTENING"
    USER_SPEAKING = "USER_SPEAKING"
    ASSISTANT_THINKING = "ASSISTANT_THINKING"
    ASSISTANT_SPEAKING = "ASSISTANT_SPEAKING"


class GenerationInterrupted(RuntimeError):
    pass


class TextChunker:
    _HARD_BOUNDARY_CHARS = ".!?\n"
    _SOFT_BOUNDARY_CHARS = ",:"
    _SOFT_BOUNDARY_TOKENS = ("\u2014",)

    def __init__(
        self,
        *,
        soft_min_chars: int = TTS_CHUNK_SOFT_MIN_CHARS,
        soft_min_words: int = TTS_CHUNK_SOFT_MIN_WORDS,
    ) -> None:
        self._buffer = ""
        self._soft_min_chars = max(soft_min_chars, 1)
        self._soft_min_words = max(soft_min_words, 1)

    def feed(self, fragment: str) -> list[str]:
        if not fragment:
            return []
        self._buffer += fragment
        return self._extract_ready_chunks()

    def flush(self) -> list[str]:
        chunks = self._extract_ready_chunks()
        tail = self._buffer.strip()
        self._buffer = ""
        if tail:
            chunks.append(tail)
        return chunks

    def reset(self) -> None:
        self._buffer = ""

    def _extract_ready_chunks(self) -> list[str]:
        chunks: list[str] = []
        while True:
            next_chunk = self._pop_next_chunk()
            if next_chunk is None:
                break
            chunks.append(next_chunk)
        return chunks

    def _pop_next_chunk(self) -> str | None:
        for index, char in enumerate(self._buffer):
            if char in self._HARD_BOUNDARY_CHARS:
                return self._consume_chunk(index + 1)
            if char in self._SOFT_BOUNDARY_CHARS or char in self._SOFT_BOUNDARY_TOKENS:
                candidate = self._buffer[: index + 1].strip()
                if self._is_soft_chunk_ready(candidate):
                    return self._consume_chunk(index + 1)
        return None

    def _consume_chunk(self, end_index: int) -> str | None:
        while end_index < len(self._buffer) and self._buffer[end_index] in self._HARD_BOUNDARY_CHARS:
            end_index += 1
        while end_index < len(self._buffer) and self._buffer[end_index].isspace():
            end_index += 1
        chunk = self._buffer[:end_index].strip()
        self._buffer = self._buffer[end_index:]
        return chunk or None

    def _is_soft_chunk_ready(self, candidate: str) -> bool:
        if len(candidate) < self._soft_min_chars:
            return False
        return len(candidate.split()) >= self._soft_min_words


class CallSession:
    def __init__(
        self,
        *,
        session_id: str,
        transport: BaseMediaTransport,
        vad: BaseVAD | None = None,
        stt: BaseSTT | None = None,
        llm: BaseLLM | None = None,
        tts: BaseTTS | None = None,
        filler_audio: FillerAudioLibrary | None = None,
        initial_greeting_text: str | None = None,
        initial_greeting_segments: Iterable[tuple[str, str | None]] | None = None,
    ) -> None:
        self.session_id = session_id
        self.transport = transport
        self.state = SessionState.LISTENING
        self.generation_epoch = 0
        self.interruptions: list[str] = []
        self.last_latency_ms: dict[str, int] = {}
        self._vad = vad or SileroVADDetector(sample_rate_hz=transport.sample_rate_hz)
        self._stt = stt or MockSTT(sample_rate_hz=transport.sample_rate_hz)
        self._llm = llm or MockLLM()
        self._tts = tts or MockTTS(sample_rate_hz=transport.sample_rate_hz)
        self._conversation: list[tuple[str, str]] = []
        self._pending_unanswered_texts: list[str] = []
        self._pending_unanswered_audio: list[bytes] = []
        self._active_turn_epoch: int | None = None
        self._active_user_audio: bytes | None = None
        self._active_user_transcript: str | None = None
        self._active_answer_audio_started = False
        self._active_unanswered_captured = False
        self._assistant_task: asyncio.Task[None] | None = None
        self._filler_task: asyncio.Task[None] | None = None
        self._greeting_task: asyncio.Task[None] | None = None
        self._sentence_queue: asyncio.Queue[str | None] | None = None
        self._closed = False
        self._filler_audio = filler_audio
        self._initial_greeting_segments = self._normalize_initial_greeting_segments(
            initial_greeting_segments,
            fallback_text=initial_greeting_text,
        )
        self._initial_greeting_text = _combine_user_transcripts(text for text, _ in self._initial_greeting_segments)
        self._initial_greeting_discarded_bytes = 0
        self._awaiting_customer_name = bool(self._initial_greeting_segments)
        self._customer_name: str | None = None
        self._name_retry_text = DEFAULT_NAME_RETRY_TEXT
        self._personalized_greeting_template = DEFAULT_PERSONALIZED_GREETING_TEMPLATE
        self._live_stt_stream: BaseSTTStream | None = None
        self._latest_partial_transcript = ""
        self._session_language: str | None = None
        self._default_vad_silence_timeout_ms = getattr(self._vad, "default_speech_end_silence_ms", 550)
        self._semantic_hold_silence_timeout_ms = max(self._default_vad_silence_timeout_ms, SEMANTIC_ENDPOINTING_HOLD_MS)
        LOGGER.info(
            "realtime session %s initialized: protocol=%s sample_rate=%s frame_ms=%s frame_bytes=%s "
            "initial_greeting=%s default_vad_silence_ms=%s semantic_hold_ms=%s",
            self.session_id,
            self.transport.protocol,
            self.transport.sample_rate_hz,
            self.transport.frame_duration_ms,
            self.transport.frame_bytes,
            bool(self._initial_greeting_text),
            self._default_vad_silence_timeout_ms,
            self._semantic_hold_silence_timeout_ms,
        )

    @staticmethod
    def _normalize_initial_greeting_segments(
        segments: Iterable[tuple[str, str | None]] | None,
        *,
        fallback_text: str | None,
    ) -> tuple[tuple[str, str | None], ...]:
        normalized_segments: list[tuple[str, str | None]] = []
        for text, language_code in segments or ():
            normalized_text = str(text or "").strip()
            if not normalized_text:
                continue
            normalized_language = str(language_code or "").strip() or None
            normalized_segments.append((normalized_text, normalized_language))
        if normalized_segments:
            return tuple(normalized_segments)
        normalized_fallback = str(fallback_text or "").strip()
        if normalized_fallback:
            return ((normalized_fallback, None),)
        return ()

    @property
    def conversation(self) -> tuple[tuple[str, str], ...]:
        return tuple(self._conversation)

    async def run(self) -> None:
        if self._assistant_task is not None:
            raise RuntimeError("CallSession.run() can only be called once per session")
        LOGGER.info("realtime session %s run started", self.session_id)
        try:
            self._start_initial_greeting()
            await self.media_loop()
        finally:
            LOGGER.info("realtime session %s run stopping", self.session_id)
            await self.stop()

    async def media_loop(self) -> None:
        while not self._closed:
            audio_chunk = await self.transport.receive_audio()
            if audio_chunk is None:
                LOGGER.info("realtime session %s transport closed", self.session_id)
                break
            if not audio_chunk:
                continue

            if self._initial_greeting_in_progress():
                self._initial_greeting_discarded_bytes += len(audio_chunk)
                continue

            vad_result = self._vad.feed(audio_chunk)

            if vad_result.is_speech and self.state != SessionState.USER_SPEAKING:
                if self.state in {SessionState.ASSISTANT_THINKING, SessionState.ASSISTANT_SPEAKING}:
                    self.interrupt("barge-in")
                self._set_state(
                    SessionState.USER_SPEAKING,
                    reason=f"speech detected prob={vad_result.speech_probability:.3f}",
                )

            if vad_result.speech_started:
                LOGGER.info(
                    "realtime session %s VAD speech_started: prob=%.3f chunk_bytes=%s current_timeout_ms=%s",
                    self.session_id,
                    vad_result.speech_probability,
                    len(audio_chunk),
                    getattr(self._vad, "current_speech_end_silence_ms", self._default_vad_silence_timeout_ms),
                )
                await self._start_live_stt_stream()
            elif self._live_stt_stream is not None:
                await self._push_live_stt_audio(audio_chunk)

            if vad_result.speech_ended and vad_result.utterance_audio:
                speech_end_monotonic = time.perf_counter()
                LOGGER.info(
                    "realtime session %s VAD speech_ended: utterance_bytes=%s utterance_ms=%s "
                    "chunk_bytes=%s current_timeout_ms=%s",
                    self.session_id,
                    len(vad_result.utterance_audio),
                    _audio_duration_ms(vad_result.utterance_audio, sample_rate_hz=self.transport.sample_rate_hz),
                    len(audio_chunk),
                    getattr(self._vad, "current_speech_end_silence_ms", self._default_vad_silence_timeout_ms),
                )
                self._set_state(SessionState.ASSISTANT_THINKING, reason="speech end detected")
                live_stt_stream = self._detach_live_stt_stream()
                self._start_assistant_turn(
                    epoch=self.generation_epoch,
                    stt_stream=live_stt_stream,
                    utterance_audio=vad_result.utterance_audio,
                    speech_end_monotonic=speech_end_monotonic,
                )

    def _initial_greeting_in_progress(self) -> bool:
        return self._greeting_task is not None and not self._greeting_task.done()

    def interrupt(self, reason: str = "interrupt") -> int:
        self.generation_epoch += 1
        self.interruptions.append(reason)
        if self.state in {SessionState.ASSISTANT_THINKING, SessionState.ASSISTANT_SPEAKING}:
            self._capture_unanswered_user_turn(reason=reason)
        self._clear_sentence_queue()
        self.transport.clear_buffer()
        if self._greeting_task is not None and not self._greeting_task.done():
            self._greeting_task.cancel()
        if self._filler_task is not None and not self._filler_task.done():
            self._filler_task.cancel()
        if self._assistant_task is not None and not self._assistant_task.done():
            self._capture_unanswered_user_turn(reason="superseded by new assistant turn")
            self._assistant_task.cancel()
        LOGGER.info(
            "realtime session %s interrupted: epoch=%s reason=%s state=%s conversation_entries=%s",
            self.session_id,
            self.generation_epoch,
            reason,
            self.state.value,
            len(self._conversation),
        )
        return self.generation_epoch

    async def stop(self) -> None:
        if self._closed:
            return
        self._closed = True
        LOGGER.info(
            "realtime session %s stop requested: state=%s epoch=%s interruptions=%s conversation_entries=%s",
            self.session_id,
            self.state.value,
            self.generation_epoch,
            self.interruptions,
            len(self._conversation),
        )
        self._vad.reset()
        self._reset_semantic_endpointing()
        self._clear_sentence_queue()
        self.transport.clear_buffer()
        await self._cancel_live_stt_stream()
        if self._greeting_task is not None and not self._greeting_task.done():
            self._greeting_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._greeting_task
        if self._filler_task is not None and not self._filler_task.done():
            self._filler_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._filler_task
        if self._assistant_task is not None and not self._assistant_task.done():
            self._assistant_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._assistant_task
        await self.transport.close()

    async def _close_after_farewell(self) -> None:
        if self._closed:
            return
        self._closed = True
        LOGGER.info(
            "realtime session %s closing transport after farewell: state=%s epoch=%s conversation_entries=%s",
            self.session_id,
            self.state.value,
            self.generation_epoch,
            len(self._conversation),
        )
        await self._cancel_live_stt_stream()
        await self.transport.close()

    def _start_assistant_turn(
        self,
        *,
        epoch: int,
        stt_stream: BaseSTTStream | None,
        utterance_audio: bytes,
        speech_end_monotonic: float,
    ) -> None:
        if self._assistant_task is not None and not self._assistant_task.done():
            self._capture_unanswered_user_turn(reason="superseded by new assistant turn")
            self._assistant_task.cancel()
        LOGGER.info(
            "realtime session %s assistant turn scheduled: epoch=%s utterance_bytes=%s utterance_ms=%s live_stt=%s",
            self.session_id,
            epoch,
            len(utterance_audio),
            _audio_duration_ms(utterance_audio, sample_rate_hz=self.transport.sample_rate_hz),
            stt_stream is not None,
        )
        self._assistant_task = asyncio.create_task(
            self._run_assistant_turn(
                epoch=epoch,
                stt_stream=stt_stream,
                utterance_audio=utterance_audio,
                speech_end_monotonic=speech_end_monotonic,
            ),
            name=f"{self.session_id}-assistant-{epoch}",
        )

    async def _run_assistant_turn(
        self,
        *,
        epoch: int,
        stt_stream: BaseSTTStream | None,
        utterance_audio: bytes,
        speech_end_monotonic: float,
    ) -> None:
        sentence_queue: asyncio.Queue[str | None] | None = None
        playback_task: asyncio.Task[None] | None = None
        assistant_fragments: list[str] = []
        actually_spoken_chunks: list[str] = []
        assistant_audio_started = [False]
        transcript = ""
        filler_started = asyncio.Event()
        self._set_active_unanswered_turn(epoch=epoch, utterance_audio=utterance_audio)
        try:
            LOGGER.info(
                "realtime session %s assistant turn started: epoch=%s utterance_bytes=%s live_stt=%s",
                self.session_id,
                epoch,
                len(utterance_audio),
                stt_stream is not None,
            )
            transcript = await self._resolve_transcript(
                stt_stream=stt_stream,
                utterance_audio=utterance_audio,
            )
            self._ensure_generation(epoch)
            self._log_latency(
                "stt_latency",
                speech_end_monotonic,
                epoch=epoch,
                message="speech end -> transcript ready",
            )
            transcript = transcript.strip()
            if not transcript:
                LOGGER.info("realtime session %s produced empty transcript", self.session_id)
                self._set_state(SessionState.LISTENING, reason="empty transcript")
                return

            pending_unanswered = await self._consume_pending_unanswered_transcripts()
            if pending_unanswered:
                original_transcript = transcript
                transcript = _combine_user_transcripts([*pending_unanswered, transcript])
                LOGGER.info(
                    "realtime session %s merged unanswered user turns: epoch=%s pending=%s "
                    "current=%r merged=%r",
                    self.session_id,
                    epoch,
                    len(pending_unanswered),
                    _preview_text(original_transcript),
                    _preview_text(transcript),
                )
            self._active_user_transcript = transcript

            if self._session_language is None:
                detected_language = (
                    _detect_name_collection_language(transcript)
                    if self._awaiting_customer_name
                    else _detect_session_language(transcript)
                )
                if detected_language:
                    self._session_language = detected_language
                    LOGGER.info(
                        "realtime session %s language locked: epoch=%s language=%s transcript=%r",
                        self.session_id,
                        epoch,
                        detected_language,
                        _preview_text(transcript),
                    )

            LOGGER.info(
                "realtime session %s transcript accepted: epoch=%s chars=%s text=%r",
                self.session_id,
                epoch,
                len(transcript),
                _preview_text(transcript),
            )
            self._conversation.append(("user", transcript))
            LOGGER.info(
                "realtime session %s conversation append user: entries=%s",
                self.session_id,
                len(self._conversation),
            )
            if _is_terminal_farewell(transcript):
                LOGGER.info(
                    "realtime session %s farewell detected before assistant response: epoch=%s transcript=%r",
                    self.session_id,
                    epoch,
                    _preview_text(transcript),
                )
                await self._play_prepared_response(
                    epoch=epoch,
                    response_text=FAREWELL_RESPONSE_TEXT,
                    actually_spoken_chunks=actually_spoken_chunks,
                )
                self._ensure_generation(epoch)
                self._commit_assistant_context(
                    llm_generated_text=FAREWELL_RESPONSE_TEXT,
                    actually_spoken_chunks=actually_spoken_chunks,
                    interrupted=False,
                )
                await self._close_after_farewell()
                return
            name_collection_response = self._build_name_collection_response(transcript)
            if name_collection_response is not None:
                LOGGER.info(
                    "realtime session %s name collection response prepared: epoch=%s name=%r text=%r",
                    self.session_id,
                    epoch,
                    self._customer_name,
                    _preview_text(name_collection_response),
                )
                await self._play_prepared_response(
                    epoch=epoch,
                    response_text=name_collection_response,
                    actually_spoken_chunks=actually_spoken_chunks,
                )
                self._ensure_generation(epoch)
                self._commit_assistant_context(
                    llm_generated_text=name_collection_response,
                    actually_spoken_chunks=actually_spoken_chunks,
                    interrupted=False,
                )
                self._set_state(SessionState.LISTENING, reason="name collection completed")
                return
            if _is_terminal_farewell(transcript):
                LOGGER.info(
                    "realtime session %s farewell response prepared: epoch=%s text=%r",
                    self.session_id,
                    epoch,
                    _preview_text(FAREWELL_RESPONSE_TEXT),
                )
                await self._play_prepared_response(
                    epoch=epoch,
                    response_text=FAREWELL_RESPONSE_TEXT,
                    actually_spoken_chunks=actually_spoken_chunks,
                )
                self._ensure_generation(epoch)
                self._commit_assistant_context(
                    llm_generated_text=FAREWELL_RESPONSE_TEXT,
                    actually_spoken_chunks=actually_spoken_chunks,
                    interrupted=False,
                )
                await self._close_after_farewell()
                return
            sentence_queue = asyncio.Queue()
            self._sentence_queue = sentence_queue
            playback_task = asyncio.create_task(
                self._stream_tts_pipeline(
                    epoch=epoch,
                    sentence_queue=sentence_queue,
                    actually_spoken_chunks=actually_spoken_chunks,
                    answer_audio_started=assistant_audio_started,
                ),
                name=f"{self.session_id}-playback-{epoch}",
            )
            llm_started_monotonic = time.perf_counter()
            chunker = TextChunker()
            first_token_seen = False
            filler_trigger_task = asyncio.create_task(
                self._trigger_filler_audio_after_delay(
                    epoch=epoch,
                    tool_name=None,
                    started_monotonic=llm_started_monotonic,
                    filler_started=filler_started,
                    suppress=_is_terminal_farewell(transcript) or _is_farewell_transcript(transcript),
                ),
                name=f"{self.session_id}-filler-delay-{epoch}",
            )
            async for event in self._llm.generate_stream(
                transcript,
                self._build_llm_context(),
                language_code=self._session_language,
            ):
                self._ensure_generation(epoch)
                if event.type == "tool_call_start":
                    LOGGER.info(
                        "realtime session %s llm tool_call_start: epoch=%s name=%s tool_call_id=%s filler_started=%s",
                        self.session_id,
                        epoch,
                        event.name,
                        event.tool_call_id,
                        filler_started.is_set(),
                    )
                    if not filler_trigger_task.done():
                        filler_trigger_task.cancel()
                    if not filler_started.is_set():
                        self._start_filler_audio(
                            epoch=epoch,
                            tool_name=event.name,
                            started_monotonic=llm_started_monotonic,
                            filler_started=filler_started,
                        )
                    continue

                token = str(event.content or "")
                if not token:
                    continue
                if not first_token_seen:
                    first_token_seen = True
                    if not filler_trigger_task.done():
                        filler_trigger_task.cancel()
                    self._log_latency(
                        "ttft",
                        llm_started_monotonic,
                        epoch=epoch,
                        message="llm request -> first token",
                    )
                assistant_fragments.append(token)
                await self._enqueue_chunks(
                    epoch=epoch,
                    sentence_queue=sentence_queue,
                    chunks=chunker.feed(token),
                )

            assistant_text = "".join(assistant_fragments).strip()
            LOGGER.info(
                "realtime session %s llm stream completed: epoch=%s generated_chars=%s spoken_chunks_so_far=%s",
                self.session_id,
                epoch,
                len(assistant_text),
                len(actually_spoken_chunks),
            )
            if not assistant_text:
                await self._finish_sentence_queue(sentence_queue)
                LOGGER.warning("realtime session %s llm produced no text", self.session_id)
                self._set_state(SessionState.LISTENING, reason="empty llm response")
                return

            await self._enqueue_chunks(
                epoch=epoch,
                sentence_queue=sentence_queue,
                chunks=chunker.flush(),
            )
            await self._finish_sentence_queue(sentence_queue)
            if playback_task is not None:
                await playback_task
            self._ensure_generation(epoch)
            self._commit_assistant_context(
                llm_generated_text=assistant_text,
                actually_spoken_chunks=actually_spoken_chunks,
                interrupted=False,
            )
            LOGGER.info(
                "realtime session %s assistant turn completed: epoch=%s generated_chars=%s spoken_chunks=%s",
                self.session_id,
                epoch,
                len(assistant_text),
                len(actually_spoken_chunks),
            )
            self._set_state(SessionState.LISTENING, reason="assistant turn completed")
        except GenerationInterrupted:
            if assistant_audio_started[0] or self._active_answer_audio_started:
                self._commit_assistant_context(
                    llm_generated_text="".join(assistant_fragments).strip(),
                    actually_spoken_chunks=actually_spoken_chunks,
                    interrupted=True,
                )
            else:
                self._capture_unanswered_user_turn(reason="stale generation before answer audio")
            LOGGER.info(
                "realtime session %s ignored stale generation %s",
                self.session_id,
                epoch,
            )
        except asyncio.CancelledError:
            if assistant_audio_started[0] or self._active_answer_audio_started:
                self._commit_assistant_context(
                    llm_generated_text="".join(assistant_fragments).strip(),
                    actually_spoken_chunks=actually_spoken_chunks,
                    interrupted=not self._closed,
                )
            elif not self._closed:
                self._capture_unanswered_user_turn(reason="cancelled before answer audio")
            LOGGER.info(
                "realtime session %s cancelled generation %s",
                self.session_id,
                epoch,
            )
            raise
        except Exception:
            LOGGER.exception(
                "realtime session %s generation %s failed",
                self.session_id,
                epoch,
            )
            if epoch == self.generation_epoch and not self._closed:
                self._set_state(SessionState.LISTENING, reason="assistant generation failed")
        finally:
            filler_trigger_task = locals().get("filler_trigger_task")
            if filler_trigger_task is not None and not filler_trigger_task.done():
                filler_trigger_task.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await filler_trigger_task
            if stt_stream is not None:
                with contextlib.suppress(Exception):
                    await stt_stream.cancel()
            self._clear_sentence_queue(sentence_queue)
            if self._filler_task is not None and not self._filler_task.done():
                self._filler_task.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await self._filler_task
            self._filler_task = None
            if playback_task is not None and not playback_task.done():
                playback_task.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await playback_task
            current_task = asyncio.current_task()
            if self._assistant_task is current_task:
                self._clear_active_unanswered_turn()
                self._assistant_task = None

    def _set_active_unanswered_turn(self, *, epoch: int, utterance_audio: bytes) -> None:
        self._active_turn_epoch = epoch
        self._active_user_audio = utterance_audio
        self._active_user_transcript = None
        self._active_answer_audio_started = False
        self._active_unanswered_captured = False

    def _clear_active_unanswered_turn(self) -> None:
        self._active_turn_epoch = None
        self._active_user_audio = None
        self._active_user_transcript = None
        self._active_answer_audio_started = False
        self._active_unanswered_captured = False

    def _capture_unanswered_user_turn(self, *, reason: str) -> None:
        if self._closed or self._active_unanswered_captured or self._active_answer_audio_started:
            return

        captured_kind = ""
        text = str(self._active_user_transcript or "").strip()
        if text:
            if self._conversation and self._conversation[-1] == ("user", text):
                self._conversation.pop()
                LOGGER.info(
                    "realtime session %s removed unanswered user turn from context: entries=%s text=%r",
                    self.session_id,
                    len(self._conversation),
                    _preview_text(text),
                )
            if not self._pending_unanswered_texts or self._pending_unanswered_texts[-1] != text:
                self._pending_unanswered_texts.append(text)
            captured_kind = "text"
        elif self._active_user_audio:
            self._pending_unanswered_audio.append(self._active_user_audio)
            captured_kind = "audio"
        else:
            return

        self._active_unanswered_captured = True
        self._active_user_audio = None
        self._active_user_transcript = None
        LOGGER.info(
            "realtime session %s captured unanswered user turn: epoch=%s kind=%s reason=%s "
            "pending_texts=%s pending_audio=%s",
            self.session_id,
            self._active_turn_epoch,
            captured_kind,
            reason,
            len(self._pending_unanswered_texts),
            len(self._pending_unanswered_audio),
        )

    async def _consume_pending_unanswered_transcripts(self) -> list[str]:
        pending_texts = [text for text in self._pending_unanswered_texts if text.strip()]
        pending_audio = list(self._pending_unanswered_audio)
        self._pending_unanswered_texts = []
        self._pending_unanswered_audio = []

        for audio_bytes in pending_audio:
            if not audio_bytes:
                continue
            self._maybe_dump_audio(audio_bytes)
            try:
                LOGGER.info(
                    "realtime session %s resolving pending unanswered audio: bytes=%s ms=%s",
                    self.session_id,
                    len(audio_bytes),
                    _audio_duration_ms(audio_bytes, sample_rate_hz=self.transport.sample_rate_hz),
                )
                pending_transcript = (await self._transcribe_batch_audio(audio_bytes)).strip()
            except Exception:
                LOGGER.exception("realtime session %s failed to transcribe pending unanswered audio", self.session_id)
                continue
            if pending_transcript:
                pending_texts.append(pending_transcript)

        if pending_texts:
            LOGGER.info(
                "realtime session %s pending unanswered transcripts ready: count=%s preview=%r",
                self.session_id,
                len(pending_texts),
                _preview_text(_combine_user_transcripts(pending_texts)),
            )
        return pending_texts

    def _name_capture_keyterms(self) -> tuple[str, ...]:
        return TOP_KAZAKH_NAMES

    def _name_capture_language_code(self) -> str | None:
        raw = (
            os.getenv("STT_NAME_CAPTURE_LANGUAGE_CODE", "").strip()
            or os.getenv("STT_NAME_CAPTURE_LANGUAGE", "").strip()
        )
        return raw or None

    async def _transcribe_batch_audio(self, audio_bytes: bytes) -> str:
        if self._awaiting_customer_name and self._session_language is None and self._name_capture_language_code() is None:
            return await self._transcribe_initial_language_choice(audio_bytes)
        return await self._stt.transcribe(
            audio_bytes,
            keyterms=self._name_capture_keyterms() if self._awaiting_customer_name else None,
            language_code=(
                self._name_capture_language_code()
                if self._awaiting_customer_name
                else _stt_language_for(self._session_language)
            ),
            force_batch=self._awaiting_customer_name,
        )

    async def _transcribe_initial_language_choice(self, audio_bytes: bytes) -> str:
        LOGGER.info(
            "realtime session %s resolving initial language choice with ru/kk STT requests",
            self.session_id,
        )
        ru_task = self._stt.transcribe(
            audio_bytes,
            keyterms=self._name_capture_keyterms(),
            language_code="ru",
            force_batch=True,
        )
        kk_task = self._stt.transcribe(
            audio_bytes,
            keyterms=self._name_capture_keyterms(),
            language_code="kk",
            force_batch=True,
        )
        ru_raw, kk_raw = await asyncio.gather(ru_task, kk_task, return_exceptions=True)
        if isinstance(ru_raw, Exception):
            LOGGER.warning(
                "realtime session %s initial Russian STT request failed: %s",
                self.session_id,
                ru_raw,
            )
            ru_result = ""
        else:
            ru_result = str(ru_raw or "")
        if isinstance(kk_raw, Exception):
            LOGGER.warning(
                "realtime session %s initial Kazakh STT request failed: %s",
                self.session_id,
                kk_raw,
            )
            kk_result = ""
        else:
            kk_result = str(kk_raw or "")
        transcript, detected_language, ru_score, kk_score = _choose_language_specific_transcript(ru_result, kk_result)
        LOGGER.info(
            "realtime session %s initial language STT selected: detected=%s ru_score=%s kk_score=%s "
            "ru=%r kk=%r selected=%r",
            self.session_id,
            detected_language,
            ru_score,
            kk_score,
            _preview_text(ru_result),
            _preview_text(kk_result),
            _preview_text(transcript),
        )
        if detected_language and self._session_language is None:
            self._session_language = detected_language
            LOGGER.info(
                "realtime session %s language locked during initial language choice: language=%s transcript=%r",
                self.session_id,
                detected_language,
                _preview_text(transcript),
            )
        return transcript

    def _build_name_collection_response(self, transcript: str) -> str | None:
        if not self._awaiting_customer_name:
            return None
        if self._session_language is None:
            detected_language = _detect_name_collection_language(transcript)
            if detected_language:
                self._session_language = detected_language
                LOGGER.info(
                    "realtime session %s language locked during name collection: language=%s transcript=%r",
                    self.session_id,
                    detected_language,
                    _preview_text(transcript),
                )
        status, name_value = _voice_start_name_outcome(transcript)
        LOGGER.info(
            "realtime session %s name collection outcome: status=%s name=%r transcript=%r",
            self.session_id,
            status,
            name_value,
            _preview_text(transcript),
        )
        if status == "name_obtained" and name_value:
            short_name = _voice_short_name(name_value) or str(name_value).strip()
            if short_name:
                self._customer_name = short_name
                self._awaiting_customer_name = False
                return self._personalized_greeting_text(short_name)
        return self._name_retry_text_for_current_language()

    def _name_retry_text_for_current_language(self) -> str:
        if self._session_language == "kk":
            return KAZAKH_NAME_RETRY_TEXT
        return self._name_retry_text

    def _personalized_greeting_text(self, name: str) -> str:
        if self._session_language == "kk":
            return KAZAKH_PERSONALIZED_GREETING_TEMPLATE.format(name=name)
        return self._personalized_greeting_template.format(name=name)

    def _build_llm_context(self) -> list[object]:
        context: list[object] = list(self._conversation)
        if self._customer_name:
            context.append(
                {
                    "role": "system",
                    "content": (
                        f"Имя пользователя: {self._customer_name}. "
                        "Не повторяй имя в каждом ответе. Используй имя только когда это естественно и полезно: "
                        "после первого знакомства, при уточнении, важном подтверждении, извинении или прощании. "
                        "В обычных ответах не обращайся по имени."
                    ),
                }
            )
        return context

    async def _play_prepared_response(
        self,
        *,
        epoch: int,
        response_text: str,
        actually_spoken_chunks: list[str],
    ) -> None:
        sentence_queue: asyncio.Queue[str | None] = asyncio.Queue()
        self._sentence_queue = sentence_queue
        playback_task = asyncio.create_task(
            self._stream_tts_pipeline(
                epoch=epoch,
                sentence_queue=sentence_queue,
                actually_spoken_chunks=actually_spoken_chunks,
            ),
            name=f"{self.session_id}-prepared-playback-{epoch}",
        )
        try:
            chunker = TextChunker()
            await self._enqueue_chunks(
                epoch=epoch,
                sentence_queue=sentence_queue,
                chunks=chunker.feed(response_text),
            )
            await self._enqueue_chunks(
                epoch=epoch,
                sentence_queue=sentence_queue,
                chunks=chunker.flush(),
            )
            await self._finish_sentence_queue(sentence_queue)
            await playback_task
        finally:
            self._clear_sentence_queue(sentence_queue)
            if not playback_task.done():
                playback_task.cancel()
                with contextlib.suppress(asyncio.CancelledError):
                    await playback_task

    def _start_initial_greeting(self) -> None:
        if not self._initial_greeting_segments or self._closed:
            return
        if self._greeting_task is not None and not self._greeting_task.done():
            return
        self._greeting_task = asyncio.create_task(
            self._play_initial_greeting(
                epoch=self.generation_epoch,
                greeting_text=self._initial_greeting_text,
                greeting_segments=self._initial_greeting_segments,
            ),
            name=f"{self.session_id}-greeting-{self.generation_epoch}",
        )
        LOGGER.info(
            "realtime session %s initial greeting scheduled: epoch=%s segments=%s chars=%s text=%r",
            self.session_id,
            self.generation_epoch,
            [(language, len(text)) for text, language in self._initial_greeting_segments],
            len(self._initial_greeting_text),
            _preview_text(self._initial_greeting_text),
        )

    async def _play_initial_greeting(
        self,
        *,
        epoch: int,
        greeting_text: str,
        greeting_segments: Iterable[tuple[str, str | None]],
    ) -> None:
        first_audio_seen = False
        started_monotonic = time.perf_counter()
        try:
            audio_chunk_count = 0
            audio_byte_count = 0
            for segment_index, (segment_text, language_code) in enumerate(greeting_segments, start=1):
                cached_segment = (
                    self._filler_audio.pick(
                        f"initial_greeting_{segment_index}_{language_code or 'default'}",
                        allow_fallback=False,
                    )
                    if self._filler_audio is not None
                    else None
                )
                if cached_segment:
                    self._ensure_generation(epoch)
                    audio_chunk_count += 1
                    audio_byte_count += len(cached_segment)
                    if not first_audio_seen:
                        first_audio_seen = True
                        self._log_latency(
                            "initial_greeting_cached_ttfa",
                            started_monotonic,
                            epoch=epoch,
                            message="session start -> cached initial greeting first audio",
                        )
                        self._set_state(SessionState.ASSISTANT_SPEAKING, reason="cached initial greeting started")
                    LOGGER.info(
                        "realtime session %s initial greeting cached segment: epoch=%s index=%s language=%s bytes=%s",
                        self.session_id,
                        epoch,
                        segment_index,
                        language_code,
                        len(cached_segment),
                    )
                    await self.transport.send_audio(cached_segment)
                    continue

                async def one_shot_text_stream() -> AsyncGenerator[str, None]:
                    yield segment_text

                LOGGER.info(
                    "realtime session %s initial greeting segment start: epoch=%s index=%s language=%s chars=%s text=%r",
                    self.session_id,
                    epoch,
                    segment_index,
                    language_code,
                    len(segment_text),
                    _preview_text(segment_text),
                )
                audio_segment_chunk_count = 0
                async for audio_chunk in self._tts.synthesize_stream(
                    one_shot_text_stream(),
                    language_code=language_code,
                ):
                    self._ensure_generation(epoch)
                    if not audio_chunk:
                        continue
                    audio_segment_chunk_count += 1
                    audio_chunk_count += 1
                    audio_byte_count += len(audio_chunk)
                    if not first_audio_seen:
                        first_audio_seen = True
                        self._log_latency(
                            "initial_greeting_ttfa",
                            started_monotonic,
                            epoch=epoch,
                            message="session start -> initial greeting first audio",
                        )
                        self._set_state(SessionState.ASSISTANT_SPEAKING, reason="initial greeting started")
                    await self.transport.send_audio(audio_chunk)
                
                if audio_segment_chunk_count == 0:
                    LOGGER.warning(
                        "realtime session %s initial greeting segment produced NO audio: language=%s",
                        self.session_id,
                        language_code,
                    )
                else:
                    LOGGER.info(
                        "realtime session %s initial greeting segment completed: language=%s chunks=%s",
                        self.session_id,
                        language_code,
                        audio_segment_chunk_count,
                    )
            await self.transport.flush_audio()
            self._ensure_generation(epoch)
            LOGGER.info(
                "realtime session %s initial greeting audio completed: epoch=%s chunks=%s bytes=%s",
                self.session_id,
                epoch,
                audio_chunk_count,
                audio_byte_count,
            )
            if self._initial_greeting_discarded_bytes:
                LOGGER.info(
                    "realtime session %s ignored inbound audio during initial greeting: bytes=%s ms=%s",
                    self.session_id,
                    self._initial_greeting_discarded_bytes,
                    _audio_byte_count_duration_ms(
                        self._initial_greeting_discarded_bytes,
                        sample_rate_hz=self.transport.sample_rate_hz,
                    ),
                )
                self._initial_greeting_discarded_bytes = 0
            if not self._conversation or self._conversation[-1] != ("assistant", greeting_text):
                self._conversation.append(("assistant", greeting_text))
                LOGGER.info(
                    "realtime session %s conversation append initial greeting: entries=%s",
                    self.session_id,
                    len(self._conversation),
                )
            if self.state == SessionState.ASSISTANT_SPEAKING and self._assistant_task is None:
                self._set_state(SessionState.LISTENING, reason="initial greeting completed")
        except GenerationInterrupted:
            LOGGER.info("realtime session %s initial greeting interrupted", self.session_id)
        except asyncio.CancelledError:
            raise
        except Exception:
            LOGGER.exception("realtime session %s initial greeting failed", self.session_id)
            if self.state == SessionState.ASSISTANT_SPEAKING and self._assistant_task is None:
                self._set_state(SessionState.LISTENING, reason="initial greeting failed")
        finally:
            current_task = asyncio.current_task()
            if self._greeting_task is current_task:
                self._greeting_task = None

    def _ensure_generation(self, epoch: int) -> None:
        if self._closed or epoch != self.generation_epoch:
            raise GenerationInterrupted(f"stale generation {epoch}")

    def _log_latency(
        self,
        metric_name: str,
        started_monotonic: float,
        *,
        epoch: int,
        message: str,
    ) -> None:
        latency_ms = max(int((time.perf_counter() - started_monotonic) * 1000.0), 0)
        self.last_latency_ms[metric_name] = latency_ms
        LOGGER.info(
            "realtime session %s %s=%sms epoch=%s (%s)",
            self.session_id,
            metric_name,
            latency_ms,
            epoch,
            message,
        )

    async def _enqueue_chunks(
        self,
        *,
        epoch: int,
        sentence_queue: asyncio.Queue[str | None],
        chunks: Iterable[str],
    ) -> None:
        for chunk in chunks:
            self._ensure_generation(epoch)
            normalized = chunk.strip()
            if not normalized:
                continue
            normalized = _sanitize_voice_text(normalized)
            if not normalized:
                LOGGER.info(
                    "realtime session %s skipped non-voice tts chunk after sanitization: epoch=%s raw=%r",
                    self.session_id,
                    epoch,
                    _preview_text(chunk),
                )
                continue
            LOGGER.info(
                "realtime session %s enqueue_tts_text: epoch=%s chars=%s preview=%r",
                self.session_id,
                epoch,
                len(normalized),
                _preview_text(normalized),
            )
            await sentence_queue.put(normalized)

    async def _finish_sentence_queue(self, sentence_queue: asyncio.Queue[str | None]) -> None:
        await sentence_queue.put(None)

    async def _stream_tts_pipeline(
        self,
        *,
        epoch: int,
        sentence_queue: asyncio.Queue[str | None],
        actually_spoken_chunks: list[str],
        answer_audio_started: list[bool] | None = None,
    ) -> None:
        first_audio_seen = False
        tts_started_monotonic: list[float | None] = [None]
        try:
            audio_chunk_count = 0
            audio_byte_count = 0
            async for audio_chunk in self._tts.synthesize_stream(
                self._iter_tts_text_chunks(
                    epoch=epoch,
                    sentence_queue=sentence_queue,
                    started_holder=tts_started_monotonic,
                    actually_spoken_chunks=actually_spoken_chunks,
                ),
                language_code=_tts_language_for(self._session_language),
            ):
                self._ensure_generation(epoch)
                if not audio_chunk:
                    continue
                audio_chunk_count += 1
                audio_byte_count += len(audio_chunk)
                if not first_audio_seen:
                    first_audio_seen = True
                    if answer_audio_started is not None:
                        answer_audio_started[0] = True
                    self._active_answer_audio_started = True
                    self._log_latency(
                        "ttfa",
                        tts_started_monotonic[0] or time.perf_counter(),
                        epoch=epoch,
                        message="tts request -> first audio",
                    )
                    self._set_state(SessionState.ASSISTANT_SPEAKING, reason="assistant playback started")
                await self.transport.send_audio(audio_chunk)
            await self.transport.flush_audio()
            LOGGER.info(
                "realtime session %s tts pipeline completed: epoch=%s audio_chunks=%s audio_bytes=%s spoken_chunks=%s",
                self.session_id,
                epoch,
                audio_chunk_count,
                audio_byte_count,
                len(actually_spoken_chunks),
            )
        finally:
            if not self._closed:
                self.transport.clear_buffer()

    async def _iter_tts_text_chunks(
        self,
        *,
        epoch: int,
        sentence_queue: asyncio.Queue[str | None],
        started_holder: list[float | None],
        actually_spoken_chunks: list[str],
    ) -> AsyncGenerator[str, None]:
        while True:
            self._ensure_generation(epoch)
            sentence = await sentence_queue.get()
            if sentence is None:
                return
            normalized = sentence.strip()
            if not normalized:
                continue
            await self._wait_for_filler_audio()
            if started_holder[0] is None:
                started_holder[0] = time.perf_counter()
            actually_spoken_chunks.append(normalized)
            LOGGER.info(
                "realtime session %s tts_text_yield: epoch=%s spoken_index=%s chars=%s preview=%r",
                self.session_id,
                epoch,
                len(actually_spoken_chunks),
                len(normalized),
                _preview_text(normalized),
            )
            yield normalized

    async def _wait_for_filler_audio(self) -> None:
        filler_task = self._filler_task
        if filler_task is None or filler_task.done():
            return
        with contextlib.suppress(asyncio.CancelledError):
            await filler_task

    async def _trigger_filler_audio_after_delay(
        self,
        *,
        epoch: int,
        tool_name: str | None,
        started_monotonic: float,
        filler_started: asyncio.Event,
        suppress: bool = False,
    ) -> None:
        if suppress:
            return
        if FILLER_AUDIO_DELAY_MS <= 0:
            self._start_filler_audio(
                epoch=epoch,
                tool_name=tool_name,
                started_monotonic=started_monotonic,
                filler_started=filler_started,
            )
            return
        await asyncio.sleep(FILLER_AUDIO_DELAY_MS / 1000.0)
        self._ensure_generation(epoch)
        if self._active_answer_audio_started or filler_started.is_set():
            return
        self._start_filler_audio(
            epoch=epoch,
            tool_name=tool_name,
            started_monotonic=started_monotonic,
            filler_started=filler_started,
        )

    def _start_filler_audio(
        self,
        *,
        epoch: int,
        tool_name: str | None,
        started_monotonic: float,
        filler_started: asyncio.Event,
    ) -> None:
        if filler_started.is_set():
            LOGGER.info(
                "realtime session %s filler audio skipped: already started for current turn epoch=%s tool=%s",
                self.session_id,
                epoch,
                tool_name or "generic",
            )
            return
        if self._filler_task is not None and not self._filler_task.done():
            return
        filler_started.set()
        LOGGER.info(
            "realtime session %s filler audio scheduled: epoch=%s tool=%s",
            self.session_id,
            epoch,
            tool_name or "generic",
        )
        self._filler_task = asyncio.create_task(
            self._play_filler_audio(
                epoch=epoch,
                tool_name=tool_name,
                started_monotonic=started_monotonic,
                filler_started=filler_started,
            ),
            name=f"{self.session_id}-filler-{epoch}",
        )

    async def _play_filler_audio(
        self,
        *,
        epoch: int,
        tool_name: str | None,
        started_monotonic: float,
        filler_started: asyncio.Event,
    ) -> None:
        if self._filler_audio is None:
            LOGGER.info("realtime session %s filler audio skipped: no library", self.session_id)
            filler_started.clear()
            return
        filler_pcm = self._filler_audio.pick(tool_name)
        if not filler_pcm:
            LOGGER.info(
                "realtime session %s filler audio skipped: no clip tool=%s",
                self.session_id,
                tool_name or "generic",
            )
            filler_started.clear()
            return
        first_audio_seen = False
        sent_chunks = 0
        sent_bytes = 0
        for offset in range(0, len(filler_pcm), max(self.transport.frame_bytes * 4, self.transport.frame_bytes)):
            self._ensure_generation(epoch)
            audio_chunk = filler_pcm[offset : offset + max(self.transport.frame_bytes * 4, self.transport.frame_bytes)]
            if not audio_chunk:
                continue
            sent_chunks += 1
            sent_bytes += len(audio_chunk)
            if not first_audio_seen:
                first_audio_seen = True
                self._log_latency(
                    "filler_ttfa",
                    started_monotonic,
                    epoch=epoch,
                    message=f"tool call -> filler audio started ({tool_name or 'generic'})",
                )
                self._set_state(SessionState.ASSISTANT_SPEAKING, reason="filler audio playback started")
            await self.transport.send_audio(audio_chunk)
        await self.transport.flush_audio()
        LOGGER.info(
            "realtime session %s filler audio completed: epoch=%s tool=%s chunks=%s bytes=%s",
            self.session_id,
            epoch,
            tool_name or "generic",
            sent_chunks,
            sent_bytes,
        )

    def _clear_sentence_queue(self, sentence_queue: asyncio.Queue[str | None] | None = None) -> None:
        queue = sentence_queue if sentence_queue is not None else self._sentence_queue
        if queue is None:
            return
        removed = 0
        while True:
            try:
                queue.get_nowait()
                removed += 1
            except asyncio.QueueEmpty:
                break
        with contextlib.suppress(asyncio.QueueFull):
            queue.put_nowait(None)
        LOGGER.info(
            "realtime session %s sentence queue cleared: removed=%s owns_queue=%s",
            self.session_id,
            removed,
            queue is self._sentence_queue,
        )
        if queue is self._sentence_queue:
            self._sentence_queue = None

    def _set_state(self, state: SessionState, *, reason: str | None = None) -> None:
        if self.state == state:
            return
        previous_state = self.state
        self.state = state
        suffix = f" ({reason})" if reason else ""
        LOGGER.info(
            "realtime session %s state %s -> %s%s",
            self.session_id,
            previous_state.value,
            state.value,
            suffix,
        )

    async def _start_live_stt_stream(self) -> None:
        await self._cancel_live_stt_stream()
        self._reset_semantic_endpointing()
        if self._awaiting_customer_name:
            LOGGER.info(
                "realtime session %s live STT stream skipped for name capture; will use batch STT with keyterms",
                self.session_id,
            )
            return
        try:
            LOGGER.info("realtime session %s live STT stream start requested", self.session_id)
            live_stt_stream = await self._stt.start_stream(
                partial_callback=self._handle_partial_transcript,
                keyterms=None,
                language_code=_stt_language_for(self._session_language),
            )
        except Exception:
            LOGGER.exception("realtime session %s failed to start live STT stream", self.session_id)
            return
        if live_stt_stream is None:
            LOGGER.info("realtime session %s live STT stream unavailable; will use batch STT", self.session_id)
            return
        self._live_stt_stream = live_stt_stream
        seed_audio = self._vad.current_utterance_audio()
        if seed_audio:
            try:
                LOGGER.info(
                    "realtime session %s live STT seed audio: bytes=%s ms=%s",
                    self.session_id,
                    len(seed_audio),
                    _audio_duration_ms(seed_audio, sample_rate_hz=self.transport.sample_rate_hz),
                )
                await live_stt_stream.push_audio(seed_audio)
            except Exception:
                LOGGER.exception("realtime session %s failed to seed live STT stream", self.session_id)
                await self._cancel_live_stt_stream()

    async def _push_live_stt_audio(self, audio_chunk: bytes) -> None:
        live_stt_stream = self._live_stt_stream
        if live_stt_stream is None or not audio_chunk:
            return
        try:
            await live_stt_stream.push_audio(audio_chunk)
        except Exception:
            LOGGER.exception("realtime session %s live STT push failed", self.session_id)
            await self._cancel_live_stt_stream()

    def _detach_live_stt_stream(self) -> BaseSTTStream | None:
        live_stt_stream = self._live_stt_stream
        self._live_stt_stream = None
        self._latest_partial_transcript = ""
        self._reset_semantic_endpointing()
        LOGGER.info(
            "realtime session %s live STT stream detached: present=%s",
            self.session_id,
            live_stt_stream is not None,
        )
        return live_stt_stream

    async def _cancel_live_stt_stream(self) -> None:
        live_stt_stream = self._detach_live_stt_stream()
        if live_stt_stream is None:
            return
        LOGGER.info("realtime session %s live STT stream canceling", self.session_id)
        with contextlib.suppress(Exception):
            await live_stt_stream.cancel()

    async def _resolve_transcript(
        self,
        *,
        stt_stream: BaseSTTStream | None,
        utterance_audio: bytes,
    ) -> str:
        self._maybe_dump_audio(utterance_audio)
        if stt_stream is not None:
            try:
                LOGGER.info("realtime session %s resolving transcript via live STT stream", self.session_id)
                transcript = (await stt_stream.finish()).strip()
                if transcript:
                    return transcript
            except Exception:
                LOGGER.exception("realtime session %s live STT stream failed; falling back to batch STT", self.session_id)
        LOGGER.info(
            "realtime session %s resolving transcript via batch STT: utterance_bytes=%s utterance_ms=%s",
            self.session_id,
            len(utterance_audio),
            _audio_duration_ms(utterance_audio, sample_rate_hz=self.transport.sample_rate_hz),
        )
        return await self._transcribe_batch_audio(utterance_audio)

    def _maybe_dump_audio(self, audio_bytes: bytes) -> None:
        if str(os.getenv("ENABLE_AUDIO_DUMP", "")).strip().lower() not in {"1", "true", "yes", "on"}:
            return
        if not audio_bytes:
            return

        try:
            dump_dir = os.getenv("AUDIO_DUMP_DIR", "debug_audio").strip() or "debug_audio"
            os.makedirs(dump_dir, exist_ok=True)
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
            safe_session_id = re.sub(r"[^A-Za-z0-9_.-]+", "_", self.session_id)
            filename = os.path.join(dump_dir, f"utterance_{safe_session_id}_{timestamp}.wav")
            dump_sample_rate_hz = self.transport.sample_rate_hz

            with wave.open(filename, "wb") as wav_file:
                wav_file.setnchannels(1)
                wav_file.setsampwidth(2)
                wav_file.setframerate(dump_sample_rate_hz)
                wav_file.writeframes(audio_bytes)

            LOGGER.info(
                "realtime session %s audio dumped: path=%s bytes=%s sample_rate=%s channels=1 sample_width=16bit",
                self.session_id,
                filename,
                len(audio_bytes),
                dump_sample_rate_hz,
            )
        except Exception:
            LOGGER.exception("realtime session %s failed to dump audio", self.session_id)

    async def _play_cached_initial_greeting(
        self,
        *,
        epoch: int,
        greeting_text: str,
        pcm_audio: bytes,
        started_monotonic: float,
    ) -> None:
        audio_chunk_count = 0
        audio_byte_count = 0
        step = max(self.transport.frame_bytes * 4, self.transport.frame_bytes)
        first_audio_seen = False
        try:
            for offset in range(0, len(pcm_audio), step):
                self._ensure_generation(epoch)
                audio_chunk = pcm_audio[offset : offset + step]
                if not audio_chunk:
                    continue
                audio_chunk_count += 1
                audio_byte_count += len(audio_chunk)
                if not first_audio_seen:
                    first_audio_seen = True
                    self._log_latency(
                        "initial_greeting_cached_ttfa",
                        started_monotonic,
                        epoch=epoch,
                        message="session start -> cached initial greeting first audio",
                    )
                    self._set_state(SessionState.ASSISTANT_SPEAKING, reason="cached initial greeting started")
                await self.transport.send_audio(audio_chunk)
            await self.transport.flush_audio()
            self._ensure_generation(epoch)
            LOGGER.info(
                "realtime session %s cached initial greeting completed: epoch=%s chunks=%s bytes=%s",
                self.session_id,
                epoch,
                audio_chunk_count,
                audio_byte_count,
            )
            if self._initial_greeting_discarded_bytes:
                LOGGER.info(
                    "realtime session %s ignored inbound audio during cached initial greeting: bytes=%s ms=%s",
                    self.session_id,
                    self._initial_greeting_discarded_bytes,
                    _audio_byte_count_duration_ms(
                        self._initial_greeting_discarded_bytes,
                        sample_rate_hz=self.transport.sample_rate_hz,
                    ),
                )
                self._initial_greeting_discarded_bytes = 0
            if not self._conversation or self._conversation[-1] != ("assistant", greeting_text):
                self._conversation.append(("assistant", greeting_text))
                LOGGER.info(
                    "realtime session %s conversation append cached initial greeting: entries=%s",
                    self.session_id,
                    len(self._conversation),
                )
            if self.state == SessionState.ASSISTANT_SPEAKING and self._assistant_task is None:
                self._set_state(SessionState.LISTENING, reason="cached initial greeting completed")
        except asyncio.CancelledError:
            raise
        except Exception:
            LOGGER.exception("realtime session %s cached initial greeting failed", self.session_id)

    async def _handle_partial_transcript(self, partial_text: str) -> None:
        normalized = " ".join(str(partial_text or "").strip().lower().split())
        if not normalized or self._closed or self._live_stt_stream is None:
            return
        self._latest_partial_transcript = normalized
        LOGGER.info(
            "realtime session %s partial transcript received: chars=%s text=%r",
            self.session_id,
            len(normalized),
            _preview_text(normalized),
        )
        target_timeout_ms = (
            self._semantic_hold_silence_timeout_ms
            if self._should_hold_endpointing(normalized)
            else self._default_vad_silence_timeout_ms
        )
        current_timeout_ms = getattr(self._vad, "current_speech_end_silence_ms", self._default_vad_silence_timeout_ms)
        if target_timeout_ms == current_timeout_ms:
            return
        self._vad.set_speech_end_silence_ms(target_timeout_ms)
        LOGGER.info(
            "realtime session %s semantic endpointing timeout=%sms partial=%r",
            self.session_id,
            target_timeout_ms,
            normalized[-80:],
        )

    def _reset_semantic_endpointing(self) -> None:
        self._latest_partial_transcript = ""
        self._vad.restore_default_speech_end_silence_ms()

    def _should_hold_endpointing(self, partial_text: str) -> bool:
        normalized = partial_text.strip().lower()
        if not normalized:
            return False
        if normalized.endswith(("...", "…", ",", ":", ";", "-", "—")):
            return True
        if len(normalized) < 12:
            return True
        words = normalized.split()
        if len(words) < 3:
            return True
        last_word = words[-1].strip(".,!?;:()[]{}\"'")
        if last_word in SEMANTIC_CONTINUATION_TOKENS:
            return True
        if len(last_word) <= 2 and len(words) < 5:
            return True
        if normalized.endswith((".", "!", "?")):
            return False
        return len(words) < 6

    def _commit_assistant_context(
        self,
        *,
        llm_generated_text: str,
        actually_spoken_chunks: list[str],
        interrupted: bool,
    ) -> None:
        if self._closed:
            return
        spoken_text = self._compose_assistant_history_text(
            actually_spoken_chunks=actually_spoken_chunks,
            fallback_text=llm_generated_text,
            interrupted=interrupted,
        )
        if not spoken_text:
            return
        self._conversation.append(("assistant", spoken_text))
        LOGGER.info(
            "realtime session %s conversation append assistant: entries=%s interrupted=%s "
            "generated_chars=%s spoken_chars=%s spoken_chunks=%s text=%r",
            self.session_id,
            len(self._conversation),
            interrupted,
            len(llm_generated_text),
            len(spoken_text),
            len(actually_spoken_chunks),
            _preview_text(spoken_text),
        )

    @staticmethod
    def _compose_assistant_history_text(
        *,
        actually_spoken_chunks: list[str],
        fallback_text: str,
        interrupted: bool,
    ) -> str:
        spoken_text = " ".join(chunk.strip() for chunk in actually_spoken_chunks if chunk and chunk.strip()).strip()
        if not spoken_text and not interrupted:
            spoken_text = fallback_text.strip()
        if not spoken_text:
            return ""
        if interrupted:
            return f"{spoken_text} [interrupted]"
        return spoken_text