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 streaming_asr_push_task: asyncio.Task | None = None streaming_asr_push_queue: asyncio.Queue[bytes | None] | None = None partial_asr_attempted: bool = False asr_stream_id: str | None = None asr_streaming_enabled: bool = False asr_streaming_failed: bool = False asr_poll_due_monotonic: float = 0.0 streaming_asr_backoff_until_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 thinking_continuation_pcm: bytearray = field(default_factory=bytearray) thinking_continuation_deadline_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, vad_rms_threshold: int = 250, 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._vad_rms_threshold = max(int(vad_rms_threshold), 1) 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 self._streaming_asr_reopen_backoff_seconds = 2.0 self._streaming_asr_push_queue_max_frames = 250 self._streaming_asr_push_batch_max_bytes = self._frame_bytes * 8 self._streaming_asr_push_drain_timeout_seconds = 1.5 self._thinking_continuation_grace_seconds = 0.45 self._thinking_continuation_max_bytes = int(1800 * 16) @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_streaming_failed = 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 actor.thinking_continuation_pcm.clear() actor.thinking_continuation_deadline_monotonic = 0.0 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 if time.monotonic() < actor.streaming_asr_backoff_until_monotonic: 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_streaming_failed = False actor.asr_poll_due_monotonic = 0.0 actor.streaming_asr_push_queue = asyncio.Queue(maxsize=self._streaming_asr_push_queue_max_frames) actor.streaming_asr_push_task = asyncio.create_task( self._streaming_asr_push_loop(actor, stream_id), name=f"streaming-asr-push-{actor.registration.voice_session_id}", ) def _mark_streaming_asr_backoff(self, actor: MediaActor) -> None: actor.streaming_asr_backoff_until_monotonic = ( time.monotonic() + self._streaming_asr_reopen_backoff_seconds ) async def _close_streaming_asr(self, actor: MediaActor, *, drain: bool = True) -> None: stream_id = actor.asr_stream_id actor.asr_stream_id = None actor.asr_streaming_enabled = False await self._stop_streaming_asr_push_loop(actor, drain=drain) if not stream_id: return await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id) async def _streaming_asr_push_loop(self, actor: MediaActor, stream_id: str) -> None: queue = actor.streaming_asr_push_queue if queue is None: return while True: item = await queue.get() if item is None: queue.task_done() return batch = bytearray(item) task_done_count = 1 stop_after_batch = False while len(batch) < self._streaming_asr_push_batch_max_bytes: try: extra = queue.get_nowait() except asyncio.QueueEmpty: break if extra is None: stop_after_batch = True queue.task_done() break batch.extend(extra) task_done_count += 1 try: await asyncio.to_thread( self._streaming_asr_provider.push_pcm, stream_id, bytes(batch), ) with contextlib.suppress(StreamingASRUnavailable): await self._poll_streaming_partial(actor) except StreamingASRUnavailable as exc: actor.asr_streaming_failed = True actor.asr_streaming_enabled = False self._mark_streaming_asr_backoff(actor) logger.warning( "audiosocket.streaming_asr_push_failed session_id=%s error=%s", actor.registration.voice_session_id, str(exc)[:500], ) return finally: for _ in range(task_done_count): queue.task_done() if stop_after_batch: return async def _stop_streaming_asr_push_loop(self, actor: MediaActor, *, drain: bool) -> None: task = actor.streaming_asr_push_task queue = actor.streaming_asr_push_queue actor.streaming_asr_push_task = None actor.streaming_asr_push_queue = None if task is None: return if task.done(): with contextlib.suppress(asyncio.CancelledError, Exception): await task return if queue is not None: if drain: try: await asyncio.wait_for( queue.join(), timeout=self._streaming_asr_push_drain_timeout_seconds, ) except asyncio.TimeoutError: actor.asr_streaming_failed = True actor.asr_streaming_enabled = False self._mark_streaming_asr_backoff(actor) logger.warning( "audiosocket.streaming_asr_push_drain_timeout session_id=%s queued_frames=%s", actor.registration.voice_session_id, queue.qsize(), ) sentinel_enqueued = False with contextlib.suppress(asyncio.QueueFull): queue.put_nowait(None) sentinel_enqueued = True if not sentinel_enqueued: task.cancel() try: await asyncio.wait_for(task, timeout=0.5) except asyncio.TimeoutError: task.cancel() with contextlib.suppress(asyncio.CancelledError, Exception): await task def _queue_streaming_asr_pcm(self, actor: MediaActor, pcm_frame: bytes) -> None: queue = actor.streaming_asr_push_queue if queue is None or actor.asr_streaming_failed: return try: queue.put_nowait(pcm_frame) except asyncio.QueueFull: actor.asr_streaming_failed = True actor.asr_streaming_enabled = False self._mark_streaming_asr_backoff(actor) logger.warning( "audiosocket.streaming_asr_push_queue_full session_id=%s max_frames=%s", actor.registration.voice_session_id, self._streaming_asr_push_queue_max_frames, ) asyncio.create_task(self._close_streaming_asr(actor, drain=False)) def _buffer_thinking_continuation(self, actor: MediaActor, pcm_frame: bytes, *, is_speech: bool, now: float) -> None: if not is_speech and not actor.thinking_continuation_pcm: return remaining = self._thinking_continuation_max_bytes - len(actor.thinking_continuation_pcm) if remaining <= 0: actor.thinking_continuation_deadline_monotonic = now + self._thinking_continuation_grace_seconds return actor.thinking_continuation_pcm.extend(pcm_frame[:remaining]) actor.thinking_continuation_deadline_monotonic = now + self._thinking_continuation_grace_seconds async def _extend_with_thinking_continuation(self, actor: MediaActor, pcm_bytes: bytes) -> bytes: deadline = time.monotonic() + self._thinking_continuation_grace_seconds observed_size = len(actor.thinking_continuation_pcm) while time.monotonic() < deadline: await asyncio.sleep(0.05) if actor.closed: break if actor.thinking_continuation_pcm: observed_size = len(actor.thinking_continuation_pcm) deadline = max(deadline, actor.thinking_continuation_deadline_monotonic) continue if observed_size > 0: break if not actor.thinking_continuation_pcm: actor.thinking_continuation_deadline_monotonic = 0.0 return pcm_bytes merged = pcm_bytes + bytes(actor.thinking_continuation_pcm) actor.thinking_continuation_pcm.clear() actor.thinking_continuation_deadline_monotonic = 0.0 return merged 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 await self._stop_streaming_asr_push_loop(actor, drain=True) if actor.asr_streaming_failed: raise StreamingASRUnavailable("Streaming ASR push failed before finalize") 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, rms_threshold=self._vad_rms_threshold, ), 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) >= self._vad_rms_threshold if actor.state == "thinking": self._buffer_thinking_continuation(actor, pcm_frame, is_speech=is_speech, now=now) return 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() return else: actor.barge_in_speech_ms = 0 if not actor.barge_in_pending: return 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) logger.info( "audiosocket.speech_started session_id=%s", actor.registration.voice_session_id, ) await self._ensure_streaming_asr(actor) stream_input_active = actor.input_active or actor.barge_in_pending if actor.asr_streaming_enabled and actor.asr_stream_id and stream_input_active: self._queue_streaming_asr_pcm(actor, pcm_frame) 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() logger.info( "audiosocket.utterance_finalized session_id=%s duration_ms=%s barge_in=%s", actor.registration.voice_session_id, int(len(vad_result.utterance_pcm) / 16), actor.barge_in_pending, ) 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") pcm_bytes = await self._extend_with_thinking_continuation(actor, pcm_bytes) 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 and not actor.asr_streaming_failed: try: transcription = await self._finalize_streaming_transcription(actor) except StreamingASRUnavailable as exc: logger.warning( "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s", actor.registration.voice_session_id, str(exc)[:500], ) self._mark_streaming_asr_backoff(actor) await self._close_streaming_asr(actor, drain=False) 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, ) 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 == "listening" 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, drain=False) 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)