call-center/services/ai_voice_runtime_service/media_runtime.py

from __future__ import annotations

import asyncio
import audioop
import contextlib
import hashlib
import logging
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable

from services.ai_voice_runtime_service.ack_bank import PrebakedAckBank
from services.ai_voice_runtime_service.audiosocket import (
    AUDIO_SOCKET_PACKET_DTMF,
    AUDIO_SOCKET_PACKET_HANGUP,
    AUDIO_SOCKET_PACKET_PCM16,
    AUDIO_SOCKET_PACKET_UUID,
    EnergyVAD,
    chunk_audio,
    encode_audio_packet,
    normalize_media_uuid,
    pcm16le_to_wav_bytes,
    read_packet,
    resample_pcm16le,
)
from services.ai_voice_runtime_service.providers.asr import (
    ASRTranscription,
    ASRProvider,
    StreamingASRPartial,
    StreamingASRProvider,
    StreamingASRUnavailable,
)
from services.ai_voice_runtime_service.providers.tts import TTSProvider
from services.shared.models import VoiceAITurnDecisionOut


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


@dataclass(slots=True)
class MediaRegistration:
    voice_session_id: str
    call_id: str
    interaction_id: str
    ai_session_id: str | None = None
    language: str | None = None
    media_uuid: str | None = None
    queue_code: str | None = None
    queue_id: str | None = None
    agent_profile: str | None = None
    voice_v2_enabled: bool = False
    voice_v2_ack_mode: str = "disabled"
    voice_v2_streaming_tts: bool = False
    voice_v2_partial_asr: bool = False
    voice_v2_duplex: bool = False
    voice_v2_streaming_asr_backend: str | None = None
    voice_v2_prebaked_ack: bool = False
    voice_v2_emotive_ack: bool = False
    voice_v2_emotive_ack_ru_only: bool = True


@dataclass(slots=True)
class MediaActor:
    registration: MediaRegistration
    reader: asyncio.StreamReader
    writer: asyncio.StreamWriter
    vad: EnergyVAD
    frame_ms: int
    frame_bytes: int
    turn_queue: asyncio.Queue[tuple[bytes, bool]] = field(default_factory=asyncio.Queue)
    worker_task: asyncio.Task | None = None
    keepalive_task: asyncio.Task | None = None
    handoff_task: asyncio.Task | None = None
    playback_interrupt: asyncio.Event = field(default_factory=asyncio.Event)
    write_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    closed: bool = False
    state: str = "greeting"
    barge_in_pending: bool = False
    last_media_touch_monotonic: float = 0.0
    last_outbound_audio_monotonic: float = 0.0
    first_pcm_logged: bool = False
    keepalive_loop_logged: bool = False
    input_active: bool = False
    playback_active: bool = False
    early_ack_started: bool = False
    partial_transcript: str | None = None
    partial_intent: str | None = None
    partial_intent_streak: int = 0
    stable_partial_intent: str | None = None
    response_plan_id: str | None = None
    playback_generation: int = 0
    tts_generation: int = 0
    utterance_generation: int = 0
    finalized_utterance_generation: int = 0
    partial_asr_task: asyncio.Task | None = None
    partial_asr_attempted: bool = False
    asr_stream_id: str | None = None
    asr_streaming_enabled: bool = False
    asr_poll_due_monotonic: float = 0.0
    barge_in_speech_ms: int = 0
    barge_in_detected_monotonic: float = 0.0
    speech_started_monotonic: float = 0.0
    speech_ended_monotonic: float = 0.0
    last_ack_text: str | None = None
    last_ack_completed_monotonic: float = 0.0
    last_ack_variant: str | None = None
    current_reply_phase: str | None = None
    finalized_caller_turn_count: int = 0


class AudioSocketMediaRuntime:
    def __init__(
        self,
        *,
        enabled: bool,
        host: str,
        port: int,
        frame_ms: int,
        idle_timeout_seconds: float,
        registration_wait_timeout_seconds: float,
        min_speech_ms: int,
        trailing_silence_ms: int,
        max_turn_ms: int,
        asr_provider: ASRProvider,
        streaming_asr_provider: StreamingASRProvider | None = None,
        tts_provider: TTSProvider,
        load_registration_by_media_uuid: Callable[[str], MediaRegistration | None],
        mark_media_connected: Callable[[str, str], None],
        mark_media_ended: Callable[[str, str], None],
        touch_media_frame: Callable[[str], None],
        set_state: Callable[[str, str, str | None, dict[str, Any] | None], None],
        get_pending_greeting: Callable[[str], str | None],
        mark_reply_started: Callable[[str, str, bool, str | None], None] | None = None,
        mark_reply_delivered: Callable[[str, str, bool, str | None], None],
        mark_reply_discarded: Callable[[str, str, str | None, str], None] | None = None,
        plan_reply: Callable[[str, str, dict[str, Any] | None, str], None],
        record_latency: Callable[[str, str, int], None] | None = None,
        process_turn: Callable[[str, str, str | None, bool, dict[str, Any] | None], VoiceAITurnDecisionOut],
        request_handoff: Callable[[str, str, VoiceAITurnDecisionOut], None],
        handle_media_error: Callable[[str, str, dict[str, Any] | None], None],
    ) -> None:
        self._enabled = enabled
        self._host = host
        self._port = port
        self._frame_ms = max(frame_ms, 20)
        self._frame_bytes = int((8000 * self._frame_ms / 1000.0) * 2)
        self._idle_timeout_seconds = max(idle_timeout_seconds, 5.0)
        self._registration_wait_timeout_seconds = max(registration_wait_timeout_seconds, 0.0)
        self._outbound_keepalive_interval_seconds = 0.75
        self._min_speech_ms = max(min_speech_ms, self._frame_ms)
        self._trailing_silence_ms = max(trailing_silence_ms, self._frame_ms)
        self._max_turn_ms = max(max_turn_ms, self._frame_ms)
        self._asr_provider = asr_provider
        self._streaming_asr_provider = streaming_asr_provider or StreamingASRProvider()
        self._tts_provider = tts_provider
        self._ack_bank = PrebakedAckBank(tts_provider=tts_provider)
        self._load_registration_by_media_uuid = load_registration_by_media_uuid
        self._mark_media_connected = mark_media_connected
        self._mark_media_ended = mark_media_ended
        self._touch_media_frame = touch_media_frame
        self._set_state = set_state
        self._get_pending_greeting = get_pending_greeting
        self._mark_reply_started = mark_reply_started or (lambda session_id, text, is_greeting, phase: None)
        self._mark_reply_delivered = mark_reply_delivered
        self._mark_reply_discarded = mark_reply_discarded or (lambda session_id, text, phase, status: None)
        self._plan_reply = plan_reply
        self._record_latency = record_latency or (lambda session_id, metric, latency_ms: None)
        self._process_turn = process_turn
        self._request_handoff = request_handoff
        self._handle_media_error = handle_media_error
        self._server: asyncio.base_events.Server | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        self._actors: dict[str, MediaActor] = {}
        self._v2_ack_wait_seconds = 0.18
        self._partial_asr_min_ms = 320
        self._immediate_ack_min_ms = 700
        self._v2_ack_post_gap_seconds = 0.10
        self._partial_poll_interval_seconds = 0.20
        self._stable_partial_hold_seconds = 0.40
        self._barge_in_trigger_ms = 220

    @staticmethod
    def _normalize_intent_text(text: str) -> str:
        return " ".join(str(text or "").strip().lower().split())

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

    def _detect_early_intent(self, text: str) -> str:
        normalized = self._normalize_intent_text(text)
        if not normalized:
            return "unknown"
        if any(token in normalized for token in ("оператор", "оператором", "человеком", "менеджер", "сотрудник")):
            return "operator_request"
        if any(token in normalized for token in ("график", "распис", "время работы", "work schedule", "жұмыс")):
            return "schedule"
        if any(token in normalized for token in ("адрес", "филиал", "офис", "где вы", "мекен", "қайда")):
            return "address"
        if any(token in normalized for token in ("тариф", "цена", "стоимость", "сколько стоит", "баға")):
            return "price"
        if any(token in normalized for token in ("статус", "заявк", "заказ", "обращени", "өтінім")):
            return "status"
        if any(token in normalized for token in ("не работает", "ошибка", "проблем", "сломал", "істемей")):
            return "problem"
        return "unknown"

    @staticmethod
    def _ack_kind_for_intent(intent: str) -> str:
        if intent == "operator_request":
            return "handoff"
        if intent in {"schedule", "address", "price", "status", "problem"}:
            return "understanding"
        if intent == "unknown":
            return "unknown"
        return "clarify"

    @staticmethod
    def _ack_text(language: str | None, ack_kind: str) -> str:
        normalized = str(language or "").strip().lower()
        if normalized == "kz":
            if ack_kind == "handoff":
                return "Бір сәт."
            if ack_kind == "understanding":
                return "Қазір айтып шығамын."
            if ack_kind == "clarify":
                return "Қазір нақтылайын."
            return "Қазір айтайын."
        if ack_kind == "handoff":
            return "Секунду."
        if ack_kind == "understanding":
            return "Сейчас сориентирую."
        if ack_kind == "clarify":
            return "Сейчас уточню."
        return "Сейчас подскажу."

    @staticmethod
    def _should_use_emotive_ack(registration: MediaRegistration, language: str | None) -> bool:
        if not registration.voice_v2_emotive_ack:
            return False
        if not registration.voice_v2_emotive_ack_ru_only:
            return True
        normalized = str(language or registration.language or "").strip().lower()
        return normalized.startswith("ru")

    def _should_emit_partial_ack(self, actor: MediaActor, partial_transcript: str, partial_intent: str) -> bool:
        transcript_text = str(partial_transcript or "").strip()
        if not transcript_text or self._is_low_signal_transcript(transcript_text):
            return False
        normalized_intent = str(partial_intent or "").strip() or "unknown"
        if actor.finalized_caller_turn_count > 0:
            return True
        return normalized_intent != "unknown"

    def _should_emit_blind_ack(self, actor: MediaActor, pcm_bytes: bytes) -> bool:
        if len(pcm_bytes) < self._immediate_ack_min_bytes:
            return False
        return actor.finalized_caller_turn_count > 0

    @staticmethod
    def _base_ack_variants(language: str | None, ack_kind: str) -> tuple[str, ...]:
        normalized = str(language or "").strip().lower()
        if normalized == "kz":
            return (AudioSocketMediaRuntime._ack_text(language, ack_kind),)
        if ack_kind == "handoff":
            return (
                "Угу, секунду.",
                "Понял вас, секунду.",
                "Мхм, соединяю.",
                "Хорошо, сейчас соединю.",
            )
        if ack_kind == "understanding":
            return (
                "Угу, сейчас подскажу.",
                "Мхм, сориентирую.",
                "Ага, понял вас.",
                "Хм, сейчас уточню.",
                "Хорошо, сейчас подскажу.",
                "Понял вас, секунду.",
            )
        if ack_kind == "clarify":
            return (
                "Угу, сейчас уточню.",
                "Мхм, одну секунду.",
                "Хорошо, сейчас уточню.",
                "Ага, сейчас сориентирую.",
            )
        return (
            "Угу, сейчас подскажу.",
            "Мхм, я в контексте.",
            "Ага, понял.",
            "Хм, секунду.",
            "Хорошо, сейчас подскажу.",
        )

    def _select_ack_payload(
        self,
        actor: MediaActor,
        *,
        language: str | None,
        ack_kind: str,
    ) -> tuple[str, dict[str, object] | None, str]:
        effective_language = language or actor.registration.language
        emotive_ack = self._should_use_emotive_ack(actor.registration, effective_language)
        if emotive_ack:
            variants = self._base_ack_variants(effective_language, ack_kind)
        else:
            variants = (self._ack_text(effective_language, ack_kind),)
        seed = (
            f"{actor.registration.voice_session_id}:"
            f"{actor.response_plan_id or actor.utterance_generation}:"
            f"{effective_language or ''}:"
            f"{ack_kind}"
        ).encode("utf-8")
        index = int.from_bytes(hashlib.sha256(seed).digest()[:4], "big") % len(variants)
        ack_text = variants[index]
        if actor.last_ack_text == ack_text and len(variants) > 1:
            index = (index + 1) % len(variants)
            ack_text = variants[index]
        actor.last_ack_text = ack_text
        actor.last_ack_variant = f"{ack_kind}:{index}"
        style_hints: dict[str, object] | None = None
        if emotive_ack:
            style_hints = {"role": "good"}
        return ack_text, style_hints, actor.last_ack_variant

    def _should_use_voice_v2(self, registration: MediaRegistration) -> bool:
        return bool(registration.voice_v2_enabled and str(registration.voice_v2_ack_mode or "").strip() == "immediate_short")

    @property
    def _partial_asr_min_bytes(self) -> int:
        return self._partial_asr_min_ms * 16

    @property
    def _immediate_ack_min_bytes(self) -> int:
        return self._immediate_ack_min_ms * 16

    @staticmethod
    def _reset_live_turn_state(actor: MediaActor) -> None:
        actor.utterance_generation += 1
        actor.finalized_utterance_generation = 0
        actor.input_active = True
        actor.early_ack_started = False
        actor.partial_transcript = None
        actor.partial_intent = None
        actor.partial_intent_streak = 0
        actor.stable_partial_intent = None
        actor.response_plan_id = None
        actor.partial_asr_attempted = False
        actor.asr_poll_due_monotonic = 0.0
        actor.speech_started_monotonic = time.monotonic()
        actor.speech_ended_monotonic = 0.0
        actor.barge_in_speech_ms = 0
        actor.barge_in_detected_monotonic = 0.0
        actor.current_reply_phase = None
        partial_task = actor.partial_asr_task
        actor.partial_asr_task = None
        if partial_task is not None and not partial_task.done():
            partial_task.cancel()

    @staticmethod
    def _is_streaming_v2_session(registration: MediaRegistration) -> bool:
        return bool(
            registration.voice_v2_enabled
            and registration.voice_v2_duplex
            and registration.voice_v2_partial_asr
            and registration.voice_v2_streaming_asr_backend
        )

    async def _ensure_streaming_asr(self, actor: MediaActor) -> None:
        if actor.closed or actor.asr_streaming_enabled:
            return
        if not self._is_streaming_v2_session(actor.registration):
            return
        try:
            stream_id = await asyncio.to_thread(
                self._streaming_asr_provider.open_stream,
                actor.registration.voice_session_id,
                language_hint=actor.registration.language,
            )
        except StreamingASRUnavailable as exc:
            actor.asr_streaming_enabled = False
            logger.warning(
                "audiosocket.streaming_asr_unavailable session_id=%s backend=%s error=%s",
                actor.registration.voice_session_id,
                actor.registration.voice_v2_streaming_asr_backend,
                str(exc)[:500],
            )
            return
        actor.asr_stream_id = stream_id
        actor.asr_streaming_enabled = True
        actor.asr_poll_due_monotonic = 0.0

    async def _close_streaming_asr(self, actor: MediaActor) -> None:
        stream_id = actor.asr_stream_id
        actor.asr_stream_id = None
        actor.asr_streaming_enabled = False
        if not stream_id:
            return
        await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id)

    def _update_stable_partial_intent(self, actor: MediaActor, intent: str) -> None:
        normalized = str(intent or "").strip() or "unknown"
        if actor.partial_intent == normalized:
            actor.partial_intent_streak += 1
        else:
            actor.partial_intent = normalized
            actor.partial_intent_streak = 1
        if actor.partial_intent_streak >= 2:
            actor.stable_partial_intent = normalized

    async def _poll_streaming_partial(self, actor: MediaActor) -> None:
        if actor.closed or not actor.asr_streaming_enabled or not actor.asr_stream_id:
            return
        now = time.monotonic()
        if now < actor.asr_poll_due_monotonic:
            return
        actor.asr_poll_due_monotonic = now + self._partial_poll_interval_seconds
        partial = await asyncio.to_thread(self._streaming_asr_provider.poll_partial, actor.asr_stream_id)
        if partial is None:
            return
        transcript_text = str(partial.text or "").strip()
        if not transcript_text:
            return
        actor.partial_transcript = transcript_text
        intent = self._detect_early_intent(transcript_text)
        self._update_stable_partial_intent(actor, intent)

    async def _finalize_streaming_transcription(self, actor: MediaActor) -> ASRTranscription:
        if not actor.asr_streaming_enabled or not actor.asr_stream_id:
            raise StreamingASRUnavailable("Streaming ASR stream is not active")
        stream_id = actor.asr_stream_id
        try:
            return await asyncio.to_thread(self._streaming_asr_provider.finalize, stream_id)
        finally:
            await self._close_streaming_asr(actor)

    async def _record_reply_status(
        self,
        actor: MediaActor,
        *,
        text: str,
        is_greeting: bool,
        phase: str | None,
        status: str,
    ) -> None:
        if status == "started":
            await asyncio.to_thread(self._invoke_reply_status_callback, self._mark_reply_started, actor.registration.voice_session_id, text, is_greeting, phase)
            return
        if status == "delivered":
            await asyncio.to_thread(self._invoke_reply_status_callback, self._mark_reply_delivered, actor.registration.voice_session_id, text, is_greeting, phase)
            return
        if status in {"interrupted", "discarded"}:
            await asyncio.to_thread(
                self._invoke_discard_callback,
                actor.registration.voice_session_id,
                text,
                phase,
                status,
            )

    async def _record_latency_metric(self, actor: MediaActor, metric: str, start_monotonic: float) -> None:
        if start_monotonic <= 0:
            return
        latency_ms = int(max((time.monotonic() - start_monotonic) * 1000.0, 0.0))
        await asyncio.to_thread(
            self._record_latency,
            actor.registration.voice_session_id,
            metric,
            latency_ms,
        )

    @staticmethod
    def _invoke_reply_status_callback(callback, session_id: str, text: str, is_greeting: bool, phase: str | None) -> None:
        try:
            callback(session_id, text, is_greeting, phase)
        except TypeError:
            callback(session_id, text, is_greeting)

    def _invoke_discard_callback(self, session_id: str, text: str, phase: str | None, status: str) -> None:
        try:
            self._mark_reply_discarded(session_id, text, phase, status)
        except TypeError:
            return

    async def _run_partial_asr_probe(
        self,
        actor: MediaActor,
        *,
        utterance_generation: int,
        pcm_bytes: bytes,
    ) -> None:
        try:
            wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
            transcription = await asyncio.to_thread(
                lambda: self._asr_provider.transcribe_partial(
                    wav_bytes,
                    language_hint=actor.registration.language,
                )
            )
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            logger.warning(
                "audiosocket.partial_asr_failed session_id=%s generation=%s error=%s",
                actor.registration.voice_session_id,
                utterance_generation,
                str(exc)[:500],
            )
            return

        if actor.closed:
            return
        if utterance_generation != actor.utterance_generation:
            return
        if actor.finalized_utterance_generation >= utterance_generation:
            return

        transcript_text = str(transcription.text or "").strip()
        if not transcript_text:
            return
        actor.partial_transcript = transcript_text
        actor.partial_intent = self._detect_early_intent(transcript_text)
        logger.info(
            "audiosocket.partial_asr_ready session_id=%s generation=%s text_len=%s intent=%s",
            actor.registration.voice_session_id,
            utterance_generation,
            len(transcript_text),
            actor.partial_intent,
        )

    def _maybe_schedule_partial_asr(self, actor: MediaActor) -> None:
        if actor.closed or not actor.registration.voice_v2_partial_asr:
            return
        if actor.partial_asr_attempted:
            return
        if not actor.vad.is_active:
            return
        snapshot = actor.vad.snapshot_utterance_pcm()
        if len(snapshot) < self._partial_asr_min_bytes:
            return
        actor.partial_asr_attempted = True
        utterance_generation = actor.utterance_generation
        actor.partial_asr_task = asyncio.create_task(
            self._run_partial_asr_probe(
                actor,
                utterance_generation=utterance_generation,
                pcm_bytes=snapshot,
            )
        )

    async def _emit_early_ack(
        self,
        actor: MediaActor,
        *,
        language: str | None,
        metadata: dict[str, Any],
        ack_source: str,
    ) -> None:
        if actor.closed or actor.early_ack_started:
            return
        ack_kind = self._ack_kind_for_intent(actor.partial_intent or "unknown")
        ack_text, style_hints, ack_variant = self._select_ack_payload(
            actor,
            language=language,
            ack_kind=ack_kind,
        )
        actor.early_ack_started = True
        actor.current_reply_phase = "ack"
        await self._plan_reply_segment(
            actor,
            ack_text,
            kind="ack",
            metadata={
                **metadata,
                "partial_transcript": actor.partial_transcript,
                "early_intent": actor.partial_intent,
                "ack_kind": ack_kind,
                "ack_variant": ack_variant,
                "voice_style": "emotive_ack" if style_hints else "neutral_ack",
                "phase": "ack",
                "partial_ack_source": ack_source,
            },
        )
        if actor.registration.voice_v2_prebaked_ack and str(language or actor.registration.language or "").strip().lower().startswith("ru"):
            await self._play_prebaked_ack(actor, ack_text, style_hints=style_hints)
        else:
            await self._speak_reply(actor, ack_text, is_greeting=False, style_hints=style_hints, reply_phase="ack")
        actor.last_ack_completed_monotonic = time.monotonic()
        if not actor.closed:
            await self._set_actor_state(actor, "thinking")

    async def _plan_reply_segment(
        self,
        actor: MediaActor,
        text: str,
        *,
        kind: str,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        if actor.closed or not str(text or "").strip():
            return
        await asyncio.to_thread(
            self._plan_reply,
            actor.registration.voice_session_id,
            text,
            metadata or {},
            kind,
        )

    async def _play_pcm_payload(
        self,
        actor: MediaActor,
        *,
        text: str,
        pcm_8k: bytes,
        is_greeting: bool,
        reply_phase: str | None,
    ) -> None:
        if actor.closed or not text or not pcm_8k:
            return
        await self._set_actor_state(actor, "speaking")
        actor.current_reply_phase = reply_phase
        interrupted = False
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="started",
        )
        if reply_phase == "ack":
            await self._record_latency_metric(actor, "speech_end_to_ack_start", actor.speech_ended_monotonic)
            await self._record_latency_metric(actor, "speech_start_to_ack_start", actor.speech_started_monotonic)
        elif reply_phase == "main":
            await self._record_latency_metric(actor, "speech_end_to_main_reply_start", actor.speech_ended_monotonic)
        for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes):
            if actor.closed or actor.playback_interrupt.is_set():
                interrupted = True
                break
            await self._write_audio_packet(actor, frame)
            await asyncio.sleep(actor.frame_ms / 1000.0)
        if actor.playback_interrupt.is_set():
            interrupted = True
        actor.playback_interrupt.clear()
        actor.current_reply_phase = None
        if interrupted or actor.closed:
            await self._record_latency_metric(actor, "barge_in_detected_to_playback_stopped", actor.barge_in_detected_monotonic)
            await self._record_reply_status(
                actor,
                text=text,
                is_greeting=is_greeting,
                phase=reply_phase,
                status="interrupted" if interrupted else "discarded",
            )
            return
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="delivered",
        )
        logger.info(
            "audiosocket.reply_delivered session_id=%s greeting=%s phase=%s",
            actor.registration.voice_session_id,
            is_greeting,
            reply_phase,
        )

    async def _play_prebaked_ack(
        self,
        actor: MediaActor,
        text: str,
        *,
        style_hints: dict[str, object] | None = None,
    ) -> None:
        clip = await asyncio.to_thread(
            self._ack_bank.get_clip,
            text=text,
            language=actor.registration.language,
            style_hints=style_hints,
        )
        await self._play_pcm_payload(
            actor,
            text=text,
            pcm_8k=clip.pcm_8k_bytes,
            is_greeting=False,
            reply_phase="ack",
        )

    async def _speak_reply(
        self,
        actor: MediaActor,
        text: str,
        *,
        is_greeting: bool,
        style_hints: dict[str, object] | None = None,
        reply_phase: str | None = "main",
    ) -> None:
        try:
            await self._speak_text(
                actor,
                text,
                is_greeting=is_greeting,
                style_hints=style_hints,
                reply_phase=reply_phase,
            )
        except TypeError:
            await self._speak_text(
                actor,
                text,
                is_greeting=is_greeting,
                style_hints=style_hints,
            )

    @property
    def address(self) -> str:
        return f"{self._host}:{self._port}"

    async def start(self) -> None:
        if not self._enabled or self._server is not None:
            return
        self._loop = asyncio.get_running_loop()
        self._server = await asyncio.start_server(self._handle_connection, self._host, self._port)

    async def _await_registration(
        self,
        media_uuid: str,
        *,
        writer: asyncio.StreamWriter | None = None,
        frame_bytes: int | None = None,
    ) -> MediaRegistration | None:
        deadline = time.monotonic() + self._registration_wait_timeout_seconds
        last_keepalive_at = 0.0
        keepalive_payload = b"\x00" * max(frame_bytes or 0, 0)
        keepalive_logged = False
        while True:
            registration = await asyncio.to_thread(self._load_registration_by_media_uuid, media_uuid)
            if registration is not None:
                return registration
            if time.monotonic() >= deadline:
                logger.warning("audiosocket.registration_timeout media_uuid=%s", media_uuid)
                return None
            if writer is not None and keepalive_payload:
                now = time.monotonic()
                if now - last_keepalive_at >= self._outbound_keepalive_interval_seconds:
                    writer.write(encode_audio_packet(keepalive_payload))
                    await writer.drain()
                    if not keepalive_logged:
                        keepalive_logged = True
                        logger.info(
                            "audiosocket.prereg_keepalive media_uuid=%s frame_bytes=%s",
                            media_uuid,
                            len(keepalive_payload),
                        )
                    last_keepalive_at = now
            await asyncio.sleep(0.05)

    async def stop(self) -> None:
        actor_ids = list(self._actors.keys())
        for session_id in actor_ids:
            await self._close_session(session_id, reason="runtime_shutdown")
        if self._server is not None:
            self._server.close()
            await self._server.wait_closed()
            self._server = None

    def close_session_sync(self, session_id: str, *, reason: str) -> None:
        if not self._loop:
            return
        future = asyncio.run_coroutine_threadsafe(
            self._close_session(session_id, reason=reason),
            self._loop,
        )
        with contextlib.suppress(Exception):
            future.result(timeout=5.0)

    async def _handle_connection(
        self,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
    ) -> None:
        actor: MediaActor | None = None
        close_reason = "connection_closed"
        close_as_error = False
        peer = writer.get_extra_info("peername")
        try:
            logger.info("audiosocket.accept peer=%s", peer)
            packet_type, payload = await read_packet(reader, timeout=self._idle_timeout_seconds)
            logger.info(
                "audiosocket.initial_packet peer=%s packet_type=%s payload_len=%s",
                peer,
                packet_type,
                len(payload),
            )
            if packet_type != AUDIO_SOCKET_PACKET_UUID:
                raise RuntimeError("AudioSocket UUID handshake is required")
            media_uuid = normalize_media_uuid(payload)
            logger.info("audiosocket.handshake peer=%s media_uuid=%s", peer, media_uuid)
            registration = await self._await_registration(
                media_uuid,
                writer=writer,
                frame_bytes=self._frame_bytes,
            )
            if registration is None:
                raise RuntimeError(f"Unknown AudioSocket media_uuid: {media_uuid}")
            logger.info(
                "audiosocket.registered peer=%s session_id=%s media_uuid=%s",
                peer,
                registration.voice_session_id,
                media_uuid,
            )

            actor = MediaActor(
                registration=registration,
                reader=reader,
                writer=writer,
                vad=EnergyVAD(
                    frame_ms=self._frame_ms,
                    min_speech_ms=self._min_speech_ms,
                    trailing_silence_ms=self._trailing_silence_ms,
                    max_turn_ms=self._max_turn_ms,
                ),
                frame_ms=self._frame_ms,
                frame_bytes=self._frame_bytes,
            )
            previous_actor = self._actors.get(registration.voice_session_id)
            if previous_actor is not None:
                await self._cleanup_actor(previous_actor, reason="media_replaced", error=False)
            self._actors[registration.voice_session_id] = actor
            actor.last_outbound_audio_monotonic = time.monotonic()
            await asyncio.to_thread(self._mark_media_connected, registration.voice_session_id, media_uuid)
            actor.keepalive_task = asyncio.create_task(self._keepalive_loop(actor))
            actor.worker_task = asyncio.create_task(self._worker(actor))

            while not actor.closed:
                packet_type, payload = await read_packet(reader, timeout=self._idle_timeout_seconds)
                if packet_type == AUDIO_SOCKET_PACKET_PCM16:
                    await self._handle_pcm(actor, payload)
                    continue
                if packet_type == AUDIO_SOCKET_PACKET_DTMF:
                    logger.info(
                        "audiosocket.dtmf session_id=%s payload_len=%s",
                        actor.registration.voice_session_id,
                        len(payload),
                    )
                    continue
                if packet_type == AUDIO_SOCKET_PACKET_HANGUP:
                    close_reason = "audiosocket_hangup"
                    break
                if packet_type == AUDIO_SOCKET_PACKET_UUID:
                    logger.info(
                        "audiosocket.extra_uuid session_id=%s payload_len=%s",
                        actor.registration.voice_session_id,
                        len(payload),
                    )
                    continue
                raise RuntimeError(f"Unsupported AudioSocket packet type: {packet_type}")
        except asyncio.TimeoutError:
            close_reason = "media_idle_timeout"
            close_as_error = True
            logger.warning("audiosocket.timeout peer=%s", peer)
        except asyncio.IncompleteReadError:
            close_reason = "connection_closed"
            logger.info("audiosocket.peer_closed peer=%s", peer)
        except Exception as exc:
            close_reason = str(exc)[:1000] or "media_runtime_error"
            close_as_error = True
            logger.warning("audiosocket.error peer=%s reason=%s", peer, close_reason)
        finally:
            if actor is not None:
                await self._cleanup_actor(actor, reason=close_reason, error=close_as_error)
            else:
                writer.close()
                with contextlib.suppress(Exception):
                    await writer.wait_closed()

    async def _handle_pcm(self, actor: MediaActor, pcm_frame: bytes) -> None:
        if actor.closed or not pcm_frame:
            return
        if not actor.first_pcm_logged:
            actor.first_pcm_logged = True
            logger.info(
                "audiosocket.first_pcm session_id=%s state=%s frame_bytes=%s",
                actor.registration.voice_session_id,
                actor.state,
                len(pcm_frame),
            )
        now = time.monotonic()
        if now - actor.last_media_touch_monotonic >= 1.0:
            actor.last_media_touch_monotonic = now
            await asyncio.to_thread(self._touch_media_frame, actor.registration.voice_session_id)

        if actor.state not in {"listening", "speaking", "thinking"}:
            return

        is_speech = audioop.rms(pcm_frame, 2) >= 250
        if actor.state == "speaking":
            if is_speech:
                actor.barge_in_speech_ms += actor.frame_ms
                if actor.barge_in_speech_ms >= self._barge_in_trigger_ms and not actor.playback_interrupt.is_set():
                    actor.barge_in_detected_monotonic = time.monotonic()
                    actor.playback_interrupt.set()
                    actor.barge_in_pending = True
                    actor.speech_started_monotonic = actor.speech_started_monotonic or time.monotonic()
            else:
                actor.barge_in_speech_ms = 0
        else:
            actor.barge_in_speech_ms = 0

        vad_result = actor.vad.feed(pcm_frame)
        if vad_result.speech_started:
            self._reset_live_turn_state(actor)
            await self._ensure_streaming_asr(actor)
        if actor.asr_streaming_enabled and actor.asr_stream_id and actor.input_active:
            try:
                await asyncio.to_thread(
                    self._streaming_asr_provider.push_pcm,
                    actor.asr_stream_id,
                    pcm_frame,
                )
            except StreamingASRUnavailable as exc:
                logger.warning(
                    "audiosocket.streaming_asr_push_failed session_id=%s error=%s",
                    actor.registration.voice_session_id,
                    str(exc)[:500],
                )
                await self._close_streaming_asr(actor)
            else:
                try:
                    await self._poll_streaming_partial(actor)
                except StreamingASRUnavailable as exc:
                    actor.asr_poll_due_monotonic = time.monotonic() + max(
                        self._partial_poll_interval_seconds,
                        0.60,
                    )
                    logger.warning(
                        "audiosocket.streaming_asr_partial_poll_failed session_id=%s error=%s",
                        actor.registration.voice_session_id,
                        str(exc)[:500],
                    )
        elif actor.registration.voice_v2_partial_asr:
            self._maybe_schedule_partial_asr(actor)
        if vad_result.utterance_pcm:
            actor.input_active = False
            actor.speech_ended_monotonic = time.monotonic()
            await actor.turn_queue.put((vad_result.utterance_pcm, actor.barge_in_pending))
            actor.barge_in_pending = False

    async def _worker(self, actor: MediaActor) -> None:
        greeting_text = await asyncio.to_thread(
            self._get_pending_greeting,
            actor.registration.voice_session_id,
        )
        if greeting_text:
            logger.info(
                "audiosocket.greeting session_id=%s text_len=%s",
                actor.registration.voice_session_id,
                len(greeting_text),
            )
            await self._speak_reply(actor, greeting_text, is_greeting=True, reply_phase="greeting")
        if not actor.closed:
            await self._set_actor_state(actor, "listening")

        while not actor.closed:
            pcm_bytes, barge_in = await actor.turn_queue.get()
            if actor.closed:
                break
            await self._process_utterance(actor, pcm_bytes, barge_in)

    async def _process_utterance(self, actor: MediaActor, pcm_bytes: bytes, barge_in: bool) -> None:
        await self._set_actor_state(actor, "thinking")
        actor.playback_generation += 1
        actor.tts_generation += 1
        actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}"
        utterance_generation = actor.utterance_generation
        partial_transcript = str(actor.partial_transcript or "").strip()
        partial_intent = str(actor.stable_partial_intent or actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
        base_metadata = {
            "turn_duration_ms": int(len(pcm_bytes) / 16),
            "media_uuid": actor.registration.media_uuid,
            "queue_code": actor.registration.queue_code,
            "voice_v2_enabled": actor.registration.voice_v2_enabled,
            "response_plan_id": actor.response_plan_id,
            "playback_generation": actor.playback_generation,
            "tts_generation": actor.tts_generation,
            "partial_transcript": partial_transcript,
            "early_intent": partial_intent,
        }
        if self._should_use_voice_v2(actor.registration) and not actor.early_ack_started:
            if actor.asr_streaming_enabled:
                with contextlib.suppress(StreamingASRUnavailable):
                    await self._poll_streaming_partial(actor)
                    partial_transcript = str(actor.partial_transcript or "").strip()
                    partial_intent = str(actor.stable_partial_intent or actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
                    base_metadata["partial_transcript"] = partial_transcript
                    base_metadata["early_intent"] = partial_intent
            else:
                partial_task = actor.partial_asr_task
                if partial_task is not None and not partial_task.done():
                    with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError):
                        await asyncio.wait_for(asyncio.shield(partial_task), timeout=0.08)
                        partial_transcript = str(actor.partial_transcript or "").strip()
                        partial_intent = str(actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
                        base_metadata["partial_transcript"] = partial_transcript
                        base_metadata["early_intent"] = partial_intent
            if self._should_emit_partial_ack(actor, partial_transcript, partial_intent):
                await self._emit_early_ack(
                    actor,
                    language=actor.registration.language,
                    metadata=base_metadata,
                    ack_source="streaming_partial" if actor.asr_streaming_enabled else "precomputed_partial_asr",
                )
            elif self._should_emit_blind_ack(actor, pcm_bytes):
                await self._emit_early_ack(
                    actor,
                    language=actor.registration.language,
                    metadata=base_metadata,
                    ack_source="immediate_turn_close",
                )

        if actor.asr_streaming_enabled:
            transcription = await self._finalize_streaming_transcription(actor)
        else:
            wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
            transcription = await asyncio.to_thread(
                self._asr_provider.transcribe,
                wav_bytes,
                language_hint=actor.registration.language,
            )
        transcript_text = str(transcription.text or "").strip() or partial_transcript
        if not transcript_text:
            await self._set_actor_state(actor, "listening")
            return
        if self._is_low_signal_transcript(transcript_text):
            actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation)
            logger.info(
                "audiosocket.low_signal_ignored session_id=%s transcript=%s",
                actor.registration.voice_session_id,
                transcript_text[:120],
            )
            await self._set_actor_state(actor, "listening")
            return

        actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation)
        actor.finalized_caller_turn_count += 1
        actor.partial_transcript = transcript_text if actor.registration.voice_v2_partial_asr else None
        actor.partial_intent = self._detect_early_intent(transcript_text)
        actor.stable_partial_intent = actor.partial_intent
        metadata = {
            **base_metadata,
            "partial_transcript": actor.partial_transcript,
            "early_intent": actor.partial_intent,
            "reply_phase": "final",
        }
        decision_task = asyncio.create_task(
            asyncio.to_thread(
                self._process_turn,
                actor.registration.voice_session_id,
                transcript_text,
                transcription.language or actor.registration.language,
                barge_in,
                {
                    **metadata,
                    "runtime_defer_reply_planned": self._should_use_voice_v2(actor.registration),
                },
            )
        )
        if self._should_use_voice_v2(actor.registration):
            try:
                decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v2_ack_wait_seconds)
            except asyncio.TimeoutError:
                if not actor.early_ack_started:
                    await self._emit_early_ack(
                        actor,
                        language=transcription.language or actor.registration.language,
                        metadata=metadata,
                        ack_source="decision_timeout",
                    )
                decision = await decision_task
        else:
            decision = await decision_task
        handoff_task: asyncio.Task | None = None
        if decision.needs_handoff:
            await self._set_actor_state(actor, "handoff_requested", decision.handoff_reason)
            handoff_task = self._start_handoff_request(actor, transcript_text, decision)
        if decision.reply_text:
            if self._should_use_voice_v2(actor.registration) and actor.early_ack_started:
                remaining_gap = self._v2_ack_post_gap_seconds - max(
                    0.0,
                    time.monotonic() - actor.last_ack_completed_monotonic,
                )
                if remaining_gap > 0:
                    await asyncio.sleep(remaining_gap)
            if self._should_use_voice_v2(actor.registration):
                await self._plan_reply_segment(
                    actor,
                    decision.reply_text,
                    kind="reply",
                    metadata={
                        **metadata,
                        **(decision.metadata if isinstance(decision.metadata, dict) else {}),
                        "phase": "main",
                        "early_ack_started": actor.early_ack_started,
                    },
                )
            await self._speak_reply(actor, decision.reply_text, is_greeting=False, reply_phase="main")
        if actor.closed:
            return
        if decision.needs_handoff:
            if handoff_task is not None and handoff_task.done():
                with contextlib.suppress(asyncio.CancelledError, Exception):
                    await handoff_task
            return
        actor.input_active = False
        await self._set_actor_state(actor, "listening")

    def _start_handoff_request(
        self,
        actor: MediaActor,
        customer_request_text: str,
        decision: VoiceAITurnDecisionOut,
    ) -> asyncio.Task:
        existing = actor.handoff_task
        if existing is not None and not existing.done():
            return existing

        async def _run() -> None:
            try:
                await asyncio.to_thread(
                    self._request_handoff,
                    actor.registration.voice_session_id,
                    customer_request_text,
                    decision,
                )
            except Exception as exc:
                logger.warning(
                    "audiosocket.handoff_request_failed session_id=%s error=%s",
                    actor.registration.voice_session_id,
                    str(exc)[:500],
                )

        actor.handoff_task = asyncio.create_task(_run())
        return actor.handoff_task

    async def _stream_tts_chunks(
        self,
        actor: MediaActor,
        text: str,
        *,
        style_hints: dict[str, object] | None = None,
    ):
        if not actor.registration.voice_v2_streaming_tts:
            synthesis = await asyncio.to_thread(
                lambda: self._tts_provider.synthesize(
                    text,
                    language=actor.registration.language,
                    style_hints=style_hints,
                )
            )
            if synthesis.audio_bytes:
                yield synthesis
            return

        loop = asyncio.get_running_loop()
        queue: asyncio.Queue[Any] = asyncio.Queue()
        done = object()

        def _producer() -> None:
            try:
                for synthesis in self._tts_provider.synthesize_chunks(
                    text,
                    language=actor.registration.language,
                    style_hints=style_hints,
                ):
                    asyncio.run_coroutine_threadsafe(queue.put(synthesis), loop).result()
            except Exception as exc:  # pragma: no cover - defensive bridge from thread to loop
                asyncio.run_coroutine_threadsafe(queue.put(exc), loop).result()
            finally:
                asyncio.run_coroutine_threadsafe(queue.put(done), loop).result()

        threading.Thread(
            target=_producer,
            name=f"tts-stream-{actor.registration.voice_session_id}",
            daemon=True,
        ).start()
        while True:
            item = await queue.get()
            if item is done:
                break
            if isinstance(item, Exception):
                raise item
            yield item

    async def _speak_text(
        self,
        actor: MediaActor,
        text: str,
        *,
        is_greeting: bool,
        style_hints: dict[str, object] | None = None,
        reply_phase: str | None = "main",
    ) -> None:
        if actor.closed or not text:
            return
        await self._set_actor_state(actor, "speaking")
        actor.current_reply_phase = reply_phase
        synth_started_at = time.monotonic()
        first_frame_sent = False
        total_audio_bytes = 0
        interrupted = False
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="started",
        )
        async for synthesis in self._stream_tts_chunks(actor, text, style_hints=style_hints):
            if not synthesis.audio_bytes:
                continue
            total_audio_bytes += len(synthesis.audio_bytes)
            if not first_frame_sent:
                logger.info(
                    "audiosocket.tts_ready session_id=%s greeting=%s synth_ms=%s audio_bytes=%s",
                    actor.registration.voice_session_id,
                    is_greeting,
                    int((time.monotonic() - synth_started_at) * 1000),
                    total_audio_bytes,
                )
                if reply_phase == "ack":
                    await self._record_latency_metric(actor, "speech_end_to_ack_start", actor.speech_ended_monotonic)
                    await self._record_latency_metric(actor, "speech_start_to_ack_start", actor.speech_started_monotonic)
                elif reply_phase == "main":
                    await self._record_latency_metric(actor, "speech_end_to_main_reply_start", actor.speech_ended_monotonic)
            pcm_8k = resample_pcm16le(
                synthesis.audio_bytes,
                input_rate_hz=synthesis.sample_rate_hz,
                output_rate_hz=8000,
            )
            for frame in chunk_audio(pcm_8k, frame_bytes=actor.frame_bytes):
                if actor.closed or actor.playback_interrupt.is_set():
                    interrupted = True
                    break
                await self._write_audio_packet(actor, frame)
                if not first_frame_sent:
                    first_frame_sent = True
                    logger.info(
                        "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s",
                        actor.registration.voice_session_id,
                        is_greeting,
                        len(frame),
                    )
                await asyncio.sleep(actor.frame_ms / 1000.0)
            if actor.closed or actor.playback_interrupt.is_set():
                interrupted = True
                break

        if total_audio_bytes <= 0:
            raise RuntimeError("TTS provider returned empty audio")

        interrupted = interrupted or actor.playback_interrupt.is_set()
        actor.playback_interrupt.clear()
        actor.current_reply_phase = None
        if interrupted or actor.closed:
            await self._record_latency_metric(actor, "barge_in_detected_to_playback_stopped", actor.barge_in_detected_monotonic)
            await self._record_reply_status(
                actor,
                text=text,
                is_greeting=is_greeting,
                phase=reply_phase,
                status="interrupted" if interrupted else "discarded",
            )
            return
        await self._record_reply_status(
            actor,
            text=text,
            is_greeting=is_greeting,
            phase=reply_phase,
            status="delivered",
        )
        logger.info(
            "audiosocket.reply_delivered session_id=%s greeting=%s phase=%s",
            actor.registration.voice_session_id,
            is_greeting,
            reply_phase,
        )

    async def _write_audio_packet(self, actor: MediaActor, pcm_frame: bytes) -> None:
        if actor.closed:
            return
        async with actor.write_lock:
            if actor.closed:
                return
            actor.writer.write(encode_audio_packet(pcm_frame))
            await actor.writer.drain()
            actor.last_outbound_audio_monotonic = time.monotonic()

    async def _keepalive_loop(self, actor: MediaActor) -> None:
        silence_frame = b"\x00" * actor.frame_bytes
        while not actor.closed:
            await asyncio.sleep(0.25)
            if actor.closed:
                break
            if (time.monotonic() - actor.last_outbound_audio_monotonic) < self._outbound_keepalive_interval_seconds:
                continue
            if not actor.keepalive_loop_logged:
                actor.keepalive_loop_logged = True
                logger.info(
                    "audiosocket.keepalive_loop session_id=%s state=%s frame_bytes=%s",
                    actor.registration.voice_session_id,
                    actor.state,
                    actor.frame_bytes,
                )
            await self._write_audio_packet(actor, silence_frame)

    async def _set_actor_state(
        self,
        actor: MediaActor,
        state: str,
        handoff_reason: str | None = None,
    ) -> None:
        if actor.closed and state not in {"error", "closed"}:
            return
        if actor.state == state and (handoff_reason or None) is None:
            return
        logger.info(
            "audiosocket.state session_id=%s from_state=%s to_state=%s handoff_reason=%s",
            actor.registration.voice_session_id,
            actor.state,
            state,
            handoff_reason,
        )
        actor.state = state
        actor.input_active = state in {"listening", "speaking", "thinking"}
        actor.playback_active = state == "speaking"
        await asyncio.to_thread(
            self._set_state,
            actor.registration.voice_session_id,
            state,
            handoff_reason,
            {"media_uuid": actor.registration.media_uuid},
        )

    async def _close_session(self, session_id: str, *, reason: str) -> None:
        actor = self._actors.get(session_id)
        if actor is None:
            return
        await self._cleanup_actor(actor, reason=reason, error=False)

    async def _cleanup_actor(self, actor: MediaActor, *, reason: str, error: bool) -> None:
        if actor.closed:
            return
        actor.closed = True
        logger.info(
            "audiosocket.cleanup session_id=%s reason=%s error=%s first_pcm=%s",
            actor.registration.voice_session_id,
            reason,
            error,
            actor.first_pcm_logged,
        )
        self._actors.pop(actor.registration.voice_session_id, None)
        if actor.keepalive_task is not None:
            actor.keepalive_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.keepalive_task
        if actor.handoff_task is not None:
            actor.handoff_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.handoff_task
        if actor.partial_asr_task is not None:
            actor.partial_asr_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.partial_asr_task
        with contextlib.suppress(Exception):
            await self._close_streaming_asr(actor)
        if actor.worker_task is not None:
            actor.worker_task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await actor.worker_task
        actor.writer.close()
        with contextlib.suppress(Exception):
            await actor.writer.wait_closed()

        if error:
            await asyncio.to_thread(
                self._handle_media_error,
                actor.registration.voice_session_id,
                reason,
                {"media_uuid": actor.registration.media_uuid},
            )
            return
        await asyncio.to_thread(self._mark_media_ended, actor.registration.voice_session_id, reason)