from __future__ import annotations import asyncio import contextlib import hashlib import logging import os 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_stateful, ) from services.ai_voice_runtime_service.providers.asr import ( ASRTranscription, ASRProvider, StreamingASRPartial, StreamingASRProvider, StreamingASRUnavailable, ) from services.shared.audioop_compat import audioop 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 stable_partial_transcript: str | None = None last_partial_transcript: str | None = None partial_transcript_streak: int = 0 partial_intent: str | None = None partial_intent_streak: int = 0 stable_partial_intent: str | None = None early_plan_task: asyncio.Task | None = None early_plan_transcript: str | None = None early_plan_intent: str | None = None early_plan_generation: int = 0 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 listening_since_monotonic: float = 0.0 reprompt_attempts: int = 0 consecutive_low_signal_turns: int = 0 @dataclass(slots=True) class _FramePacer: """Wall-clock-anchored pacing for one playback: paces frames to the first frame's send time instead of naive per-frame relative sleeps, whose drift accumulates under event-loop scheduling pressure (audible as stutter).""" frame_seconds: float start_monotonic: float | None = None frames_written: int = 0 async def wait_for_next_frame(self) -> None: if self.start_monotonic is None: self.start_monotonic = time.monotonic() self.frames_written += 1 target_monotonic = self.start_monotonic + self.frames_written * self.frame_seconds sleep_seconds = target_monotonic - time.monotonic() if sleep_seconds > 0: await asyncio.sleep(sleep_seconds) 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._v1_ack_wait_seconds = 0.6 self._partial_poll_interval_seconds = 0.20 # Small startup cushion for streamed TTS playback: absorb ElevenLabs # network delivery jitter before we start pacing frames out to the # caller, so a brief mid-download stall doesn't read as a dead-air gap. self._tts_stream_prebuffer_ms = 200 self._streaming_asr_partial_poll_enabled = ( str(os.getenv("AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED", "0")).strip().lower() in {"1", "true", "yes", "on"} ) 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) self._partial_first_final_enabled = ( str(os.getenv("AI_VOICE_V2_PARTIAL_FIRST_FINAL", "1")).strip().lower() in {"1", "true", "yes", "on"} ) self._early_plan_enabled = ( str(os.getenv("AI_VOICE_V2_EARLY_PLAN_ENABLED", "1")).strip().lower() in {"1", "true", "yes", "on"} ) self._no_speech_reprompt_enabled = ( str(os.getenv("AI_VOICE_NO_SPEECH_REPROMPT_ENABLED", "1")).strip().lower() in {"1", "true", "yes", "on"} ) self._no_speech_reprompt_seconds = 7.0 self._no_speech_reprompt_low_signal_turns = 2 self._no_speech_reprompt_max_attempts = 2 raw_early_plan_intents = str( os.getenv( "AI_VOICE_V2_EARLY_PLAN_INTENTS", "schedule,address,price,status,problem,operator_request", ) ) self._early_plan_intents = { item.strip().lower() for item in raw_early_plan_intents.split(",") if item.strip() } @staticmethod def _normalize_intent_text(text: str) -> str: compact = str(text or "").strip().lower() for char in ",.!?;:…\"'()[]{}": compact = compact.replace(char, " ") return " ".join(compact.split()) _FINAL_LOW_SIGNAL_PHRASES = { "алло", "ой", "слышу", "слышно", "твой", "поргай что это", } @classmethod def _is_low_signal_partial_transcript(cls, text: str | None) -> bool: """Broad filler check used only for mid-utterance ack/early-plan gating.""" normalized = cls._normalize_intent_text(text) if not normalized: return True return normalized in cls._FINAL_LOW_SIGNAL_PHRASES | { "ага", "да", "добрый день", "здравствуйте", "ладно", "неа", "нет", "ок", "понял", "поняла", "привет", "угу", "хорошо", "ясно", "давай", } @classmethod def _is_low_signal_final_transcript(cls, text: str | None) -> bool: """Narrow filler check for a *finalized* caller turn. Unlike the partial-transcript check, this must not swallow real answers such as "да"/"нет"/"хорошо" — those are legitimate replies and a caller who says only that deserves a response, not silence. """ normalized = cls._normalize_intent_text(text) if not normalized: return True return normalized in cls._FINAL_LOW_SIGNAL_PHRASES 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 _asr_failure_category(exc: Exception) -> str: text = str(exc or "").lower() if "unauthorized" in text or "unknown api key" in text or "invalid api key" in text: return "auth" if "timeout" in text: return "timeout" if "not found" in text: return "provider_not_found" if "service unavailable" in text or "temporarily unavailable" in text: return "provider_unavailable" return "provider_error" @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 _no_speech_reprompt_text(language: str | None) -> str: normalized = str(language or "").strip().lower() if normalized == "kz": return "Алло, сізді тыңдап тұрмын. Немен көмектесе аламын?" return "Алло, я вас слушаю. Подскажите, чем могу помочь?" @staticmethod def _technical_issue_text(language: str | None) -> str: normalized = str(language or "").strip().lower() if normalized == "kz": 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_partial_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: del actor return len(pcm_bytes) >= self._immediate_ack_min_bytes @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 @property def _tts_stream_prebuffer_bytes(self) -> int: return self._tts_stream_prebuffer_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.stable_partial_transcript = None actor.last_partial_transcript = None actor.partial_transcript_streak = 0 actor.partial_intent = None actor.partial_intent_streak = 0 actor.stable_partial_intent = None actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}" actor.early_plan_transcript = None actor.early_plan_intent = None actor.early_plan_generation = 0 early_plan_task = actor.early_plan_task actor.early_plan_task = None if early_plan_task is not None and not early_plan_task.done(): early_plan_task.cancel() 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), ) 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 if ( not is_speech and actor.thinking_continuation_deadline_monotonic > 0 and now > actor.thinking_continuation_deadline_monotonic ): return remaining = self._thinking_continuation_max_bytes - len(actor.thinking_continuation_pcm) if remaining <= 0: if is_speech: actor.thinking_continuation_deadline_monotonic = now + self._thinking_continuation_grace_seconds return actor.thinking_continuation_pcm.extend(pcm_frame[:remaining]) if is_speech: 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 @staticmethod def _update_stable_partial_transcript(actor: MediaActor, transcript_text: str, *, provider_stable: bool) -> None: normalized = " ".join(str(transcript_text or "").strip().lower().split()) if not normalized: return if actor.last_partial_transcript == normalized: actor.partial_transcript_streak += 1 else: actor.last_partial_transcript = normalized actor.partial_transcript_streak = 1 if provider_stable or actor.partial_transcript_streak >= 2: actor.stable_partial_transcript = transcript_text def _should_schedule_early_plan(self, actor: MediaActor, transcript_text: str, intent: str) -> bool: if not self._early_plan_enabled or not self._should_use_voice_v2(actor.registration): return False if actor.closed or self._is_low_signal_partial_transcript(transcript_text): return False normalized_intent = str(intent or "").strip().lower() or "unknown" if normalized_intent not in self._early_plan_intents: return False task = actor.early_plan_task if task is not None and task.done() and actor.early_plan_transcript == transcript_text and actor.early_plan_intent == normalized_intent: return False return task is None or task.done() def _schedule_early_plan( self, actor: MediaActor, *, transcript_text: str, language: str | None, intent: str, source: str, ) -> None: if not self._should_schedule_early_plan(actor, transcript_text, intent): return if not actor.response_plan_id: actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}" actor.early_plan_generation = actor.utterance_generation actor.early_plan_transcript = transcript_text actor.early_plan_intent = str(intent or "").strip() or "unknown" metadata = { "voice_v2_enabled": actor.registration.voice_v2_enabled, "reply_phase": "early_plan", "response_plan_id": actor.response_plan_id, "playback_generation": actor.playback_generation + 1, "tts_generation": actor.tts_generation + 1, "partial_transcript": transcript_text, "early_intent": actor.early_plan_intent, "transcript_source": source, "suppress_name_prefix": True, } actor.early_plan_task = asyncio.create_task( asyncio.to_thread( self._process_turn, actor.registration.voice_session_id, transcript_text, language or actor.registration.language, False, metadata, ), name=f"voice-early-plan-{actor.registration.voice_session_id}", ) logger.info( "audiosocket.early_plan_started session_id=%s generation=%s intent=%s source=%s text_len=%s", actor.registration.voice_session_id, actor.early_plan_generation, actor.early_plan_intent, source, len(transcript_text), ) async def _take_early_plan_decision( self, actor: MediaActor, *, timeout_seconds: float, ) -> VoiceAITurnDecisionOut | None: task = actor.early_plan_task if task is None: return None if actor.early_plan_generation != actor.utterance_generation: return None if not task.done() and timeout_seconds > 0: with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError): await asyncio.wait_for(asyncio.shield(task), timeout=timeout_seconds) if not task.done(): return None try: decision = task.result() except Exception as exc: logger.warning( "audiosocket.early_plan_failed session_id=%s generation=%s error=%s", actor.registration.voice_session_id, actor.early_plan_generation, str(exc)[:500], ) return None if not getattr(decision, "reply_text", None) and not getattr(decision, "needs_handoff", False): return None return decision async def _reconcile_final_decision_after_early_plan( self, actor: MediaActor, *, decision_task: asyncio.Task, early_decision: VoiceAITurnDecisionOut, transcript_text: str, ) -> None: try: final_decision = await decision_task except asyncio.CancelledError: raise except Exception as exc: logger.warning( "audiosocket.final_plan_reconcile_failed session_id=%s error=%s", actor.registration.voice_session_id, str(exc)[:500], ) return if actor.closed: return if final_decision.needs_handoff and not early_decision.needs_handoff: await self._set_actor_state(actor, "handoff_requested", final_decision.handoff_reason) self._start_handoff_request(actor, transcript_text, final_decision) 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 if transcript_text != actor.partial_transcript: logger.info( "audiosocket.streaming_asr_partial session_id=%s stable=%s final=%s text=%s", actor.registration.voice_session_id, bool(partial.is_stable), bool(partial.is_final), transcript_text[:160], ) actor.partial_transcript = transcript_text self._update_stable_partial_transcript(actor, transcript_text, provider_stable=bool(partial.is_stable or partial.is_final)) intent = self._detect_early_intent(transcript_text) self._update_stable_partial_intent(actor, intent) stable_text = str(actor.stable_partial_transcript or "").strip() stable_intent = str(actor.stable_partial_intent or "").strip() if stable_text and stable_intent: self._schedule_early_plan( actor, transcript_text=stable_text, language=partial.language or actor.registration.language, intent=stable_intent, source="streaming_stable_partial", ) async def _run_streaming_partial_poll(self, actor: MediaActor, utterance_generation: int) -> None: try: if utterance_generation != actor.utterance_generation: return await self._poll_streaming_partial(actor) except asyncio.CancelledError: raise except Exception as exc: logger.warning( "audiosocket.streaming_asr_partial_failed session_id=%s generation=%s error=%s", actor.registration.voice_session_id, utterance_generation, str(exc)[:500], ) def _maybe_schedule_streaming_partial_poll(self, actor: MediaActor) -> None: if not self._streaming_asr_partial_poll_enabled: return if actor.closed or not actor.asr_streaming_enabled or not actor.asr_stream_id: return if time.monotonic() < actor.asr_poll_due_monotonic: return task = actor.partial_asr_task if task is not None and not task.done(): return actor.partial_asr_task = asyncio.create_task( self._run_streaming_partial_poll(actor, actor.utterance_generation), name=f"streaming-asr-partial-{actor.registration.voice_session_id}", ) 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 start_monotonic = time.monotonic() await self._stop_streaming_asr_push_loop(actor, drain=True) if actor.asr_streaming_failed: raise StreamingASRUnavailable("Streaming ASR push failed before finalize") try: transcription = await asyncio.to_thread(self._streaming_asr_provider.finalize, stream_id) finally: await self._close_streaming_asr(actor) logger.info( "audiosocket.streaming_asr_finalized session_id=%s elapsed_ms=%s text=%s", actor.registration.voice_session_id, int((time.monotonic() - start_monotonic) * 1000), str(transcription.text or "")[:160], ) return transcription 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 self._update_stable_partial_transcript(actor, transcript_text, provider_stable=True) actor.partial_intent = self._detect_early_intent(transcript_text) actor.stable_partial_intent = actor.partial_intent self._schedule_early_plan( actor, transcript_text=transcript_text, language=transcription.language or actor.registration.language, intent=actor.partial_intent, source="partial_asr_probe", ) 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, ack_kind: str | None = None, ) -> None: if actor.closed or actor.early_ack_started: return ack_kind = ack_kind or 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": metadata.get("early_intent") or 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 actor.playback_interrupt.clear() actor.barge_in_detected_monotonic = 0.0 actor.barge_in_speech_ms = 0 await self._set_actor_state(actor, "speaking") actor.current_reply_phase = reply_phase 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) pacer = _FramePacer(frame_seconds=actor.frame_ms / 1000.0) interrupted = await self._write_paced_pcm_frames(actor, pcm_8k, pacer) 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) if actor.asr_streaming_enabled and actor.asr_stream_id: # The local VAD only flags speech_started after min_speech_ms of # buffered pre-roll audio (see EnergyVAD.feed), and that pre-roll # is folded into snapshot_utterance_pcm() together with the current # frame. Without forwarding it here, every utterance's opening # ~min_speech_ms would never reach the realtime ASR stream, clipping # the first word(s) of each turn. leadin_pcm = actor.vad.snapshot_utterance_pcm() if len(leadin_pcm) > len(pcm_frame): self._queue_streaming_asr_pcm(actor, leadin_pcm[: -len(pcm_frame)]) 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) self._maybe_schedule_streaming_partial_poll(actor) 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: try: 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) except asyncio.CancelledError: raise except Exception as exc: await self._handle_turn_exception(actor, exc) async def _handle_turn_exception(self, actor: MediaActor, exc: Exception) -> None: error_text = str(exc)[:500] or "voice_turn_failed" metadata = { "media_uuid": actor.registration.media_uuid, "queue_code": actor.registration.queue_code, "reply_phase": actor.current_reply_phase or "error_handoff", "error_class": exc.__class__.__name__, "error_category": self._asr_failure_category(exc), } logger.warning( "audiosocket.turn_failed session_id=%s category=%s error=%s", actor.registration.voice_session_id, metadata["error_category"], error_text, ) await self._set_actor_state(actor, "handoff_requested", error_text) fallback_text = self._technical_issue_text(actor.registration.language) with contextlib.suppress(Exception): await self._plan_reply_segment( actor, fallback_text, kind="reply", metadata={ **metadata, "phase": "error_handoff", "technical_fallback": True, }, ) await self._speak_reply( actor, fallback_text, is_greeting=False, reply_phase="error_handoff", ) await asyncio.to_thread( self._handle_media_error, actor.registration.voice_session_id, error_text, metadata, ) 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 = actor.response_plan_id or f"rsp_{uuid.uuid4().hex[:10]}" utterance_generation = actor.utterance_generation partial_transcript = str(actor.stable_partial_transcript or 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: 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.stable_partial_transcript or 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 if partial_transcript: self._schedule_early_plan( actor, transcript_text=partial_transcript, language=actor.registration.language, intent=partial_intent, source="turn_close_partial", ) 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", ack_kind="unknown", ) transcript_source = "batch" asr_start_monotonic = time.monotonic() if actor.asr_streaming_enabled and not actor.asr_streaming_failed: try: transcription = await self._finalize_streaming_transcription(actor) transcript_source = "streaming_final" 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) partial_first_text = str(actor.stable_partial_transcript or actor.partial_transcript or "").strip() if self._partial_first_final_enabled and partial_first_text and not self._is_low_signal_final_transcript(partial_first_text): transcription = ASRTranscription( text=partial_first_text, language=actor.registration.language, confidence=None, ) transcript_source = "streaming_partial_after_finalize_failure" 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_source = "batch_fallback_after_streaming_failure" 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_source = "batch" transcript_text = str(transcription.text or "").strip() or partial_transcript logger.info( "audiosocket.asr_turn_ready session_id=%s provider=%s source=%s utterance_ms=%s elapsed_ms=%s " "text_len=%s empty=%s text=%s", actor.registration.voice_session_id, getattr(self._asr_provider, "name", "unknown"), transcript_source, int(len(pcm_bytes) / 16), int((time.monotonic() - asr_start_monotonic) * 1000), len(str(transcript_text or "").strip()), not bool(str(transcript_text or "").strip()), transcript_text[:160], ) if not transcript_text: actor.consecutive_low_signal_turns += 1 await self._set_actor_state(actor, "listening") if actor.consecutive_low_signal_turns >= self._no_speech_reprompt_low_signal_turns: await self._maybe_play_no_speech_reprompt(actor, trigger="empty_transcript") return if self._is_low_signal_final_transcript(transcript_text): actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation) actor.consecutive_low_signal_turns += 1 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") if actor.consecutive_low_signal_turns >= self._no_speech_reprompt_low_signal_turns: await self._maybe_play_no_speech_reprompt(actor, trigger="low_signal_transcript") return actor.finalized_utterance_generation = max(actor.finalized_utterance_generation, utterance_generation) actor.finalized_caller_turn_count += 1 actor.consecutive_low_signal_turns = 0 actor.reprompt_attempts = 0 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", "transcript_source": transcript_source, } 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), }, ) ) early_plan_decision: VoiceAITurnDecisionOut | None = None if self._should_use_voice_v2(actor.registration): early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.04) if early_plan_decision is not None: decision = early_plan_decision logger.info( "audiosocket.early_plan_used session_id=%s generation=%s intent=%s", actor.registration.voice_session_id, actor.early_plan_generation, actor.early_plan_intent, ) else: 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", ) early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.12) decision = early_plan_decision if early_plan_decision is not None else await decision_task else: try: decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v1_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="v1_decision_timeout", ack_kind="unknown", ) 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 early_plan_decision is not None: if decision_task.done(): await self._reconcile_final_decision_after_early_plan( actor, decision_task=decision_task, early_decision=early_plan_decision, transcript_text=transcript_text, ) else: asyncio.create_task( self._reconcile_final_decision_after_early_plan( actor, decision_task=decision_task, early_decision=early_plan_decision, transcript_text=transcript_text, ), name=f"voice-final-plan-reconcile-{actor.registration.voice_session_id}", ) 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 actor.playback_interrupt.clear() actor.barge_in_detected_monotonic = 0.0 actor.barge_in_speech_ms = 0 await self._set_actor_state(actor, "speaking") actor.current_reply_phase = reply_phase synth_started_at = time.monotonic() first_chunk_logged = False total_audio_bytes = 0 interrupted = False pending_pcm = bytearray() resample_state: object | None = None prebuffered = False await self._record_reply_status( actor, text=text, is_greeting=is_greeting, phase=reply_phase, status="started", ) pacer = _FramePacer(frame_seconds=actor.frame_ms / 1000.0) def _log_first_frame(frame: bytes) -> None: logger.info( "audiosocket.first_frame session_id=%s greeting=%s frame_bytes=%s", actor.registration.voice_session_id, is_greeting, len(frame), ) async def _write_pcm_frames(pcm_bytes: bytes) -> bool: return await self._write_paced_pcm_frames(actor, pcm_bytes, pacer, on_first_frame=_log_first_frame) 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_chunk_logged: first_chunk_logged = True 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) # Carry ratecv state across chunks: resampling each network chunk # independently introduces an audible click/discontinuity at every # chunk boundary, which gets much more noticeable with a fast model # that streams many small chunks (e.g. eleven_flash_v2_5). pcm_8k, resample_state = resample_pcm16le_stateful( synthesis.audio_bytes, input_rate_hz=synthesis.sample_rate_hz, output_rate_hz=8000, state=resample_state, ) if prebuffered: # Past the initial cushion: feed each resampled chunk straight # through instead of round-tripping it through pending_pcm. interrupted = await _write_pcm_frames(pcm_8k) if interrupted: break continue pending_pcm.extend(pcm_8k) if len(pending_pcm) < self._tts_stream_prebuffer_bytes: continue prebuffered = True interrupted = await _write_pcm_frames(bytes(pending_pcm)) pending_pcm.clear() if interrupted: break if not interrupted and pending_pcm: interrupted = await _write_pcm_frames(bytes(pending_pcm)) pending_pcm.clear() 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 _write_paced_pcm_frames( self, actor: MediaActor, pcm_bytes: bytes, pacer: _FramePacer, *, on_first_frame: Callable[[bytes], None] | None = None, ) -> bool: """Write pcm_bytes out as frame_bytes-sized packets, paced by `pacer`. Returns True if playback was interrupted (barge-in or actor closed). """ for frame in chunk_audio(pcm_bytes, frame_bytes=actor.frame_bytes): if actor.closed or actor.playback_interrupt.is_set(): return True await self._write_audio_packet(actor, frame) if pacer.frames_written == 0 and on_first_frame is not None: on_first_frame(frame) await pacer.wait_for_next_frame() return False 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 await self._maybe_reprompt_on_silence(actor) 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 _maybe_play_no_speech_reprompt(self, actor: MediaActor, *, trigger: str) -> None: if actor.closed or actor.state != "listening": return if actor.reprompt_attempts >= self._no_speech_reprompt_max_attempts: return actor.reprompt_attempts += 1 actor.consecutive_low_signal_turns = 0 logger.info( "audiosocket.no_speech_reprompt session_id=%s attempt=%s trigger=%s", actor.registration.voice_session_id, actor.reprompt_attempts, trigger, ) reprompt_text = self._no_speech_reprompt_text(actor.registration.language) await self._speak_reply(actor, reprompt_text, is_greeting=False, reply_phase="no_speech_reprompt") if not actor.closed: await self._set_actor_state(actor, "listening") async def _maybe_reprompt_on_silence(self, actor: MediaActor) -> None: if not self._no_speech_reprompt_enabled: return if actor.state != "listening" or actor.listening_since_monotonic <= 0: return if time.monotonic() - actor.listening_since_monotonic < self._no_speech_reprompt_seconds: return await self._maybe_play_no_speech_reprompt(actor, trigger="silence_timeout") 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" if state == "listening": actor.listening_since_monotonic = time.monotonic() 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)