diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index d7cbaf8..6f54478 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/services/ai_voice_runtime_service/media_runtime.py @@ -283,6 +283,13 @@ class AudioSocketMediaRuntime: 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: @@ -1184,25 +1191,69 @@ class AudioSocketMediaRuntime: 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", + try: + greeting_text = await asyncio.to_thread( + self._get_pending_greeting, 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") + 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) + 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__, + } + logger.warning( + "audiosocket.turn_failed session_id=%s error=%s", + actor.registration.voice_session_id, + 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") diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index e554e80..ec9e85f 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/tests/test_ai_voice_media_runtime.py @@ -187,6 +187,109 @@ def test_media_runtime_streams_greeting_and_turn(): assert errors == [] +def test_media_runtime_speaks_technical_fallback_when_asr_transcribe_fails(): + registrations: dict[str, MediaRegistration] = {} + delivered: list[tuple[str, str, bool]] = [] + errors: list[tuple[str, str]] = [] + states: list[tuple[str, str]] = [] + + class _FailingASRProvider(ASRProvider): + name = "failing-asr" + + def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: + del audio_bytes, language_hint + raise RuntimeError("AI_VOICE_ASR_YANDEX_API_KEY is required for Yandex ASR") + + media_uuid = str(uuid.uuid4()) + registrations[media_uuid] = MediaRegistration( + voice_session_id="avs_media_runtime_asr_error", + call_id="call_media_runtime_asr_error", + interaction_id="int_media_runtime_asr_error", + ai_session_id="ais_media_runtime_asr_error", + language="ru", + media_uuid=media_uuid, + ) + + runtime = AudioSocketMediaRuntime( + enabled=True, + host="127.0.0.1", + port=0, + frame_ms=20, + idle_timeout_seconds=2.0, + registration_wait_timeout_seconds=0.5, + min_speech_ms=40, + trailing_silence_ms=40, + max_turn_ms=400, + asr_provider=_FailingASRProvider(), + tts_provider=_StubTTSProvider(), + load_registration_by_media_uuid=lambda value: registrations.get(value), + mark_media_connected=lambda session_id, value: None, + mark_media_ended=lambda session_id, reason: None, + touch_media_frame=lambda session_id: None, + set_state=lambda session_id, state, handoff_reason, metadata: states.append((session_id, state)), + get_pending_greeting=lambda session_id: "greeting" if session_id == "avs_media_runtime_asr_error" else None, + mark_reply_delivered=lambda session_id, text, is_greeting: delivered.append((session_id, text, is_greeting)), + plan_reply=lambda session_id, text, metadata, kind: None, + process_turn=lambda session_id, transcript_text, language, barge_in, metadata: VoiceAITurnDecisionOut( + language=language or "ru", + intent="answer", + reply_text="reply", + confidence=0.9, + needs_handoff=False, + handoff_reason=None, + case_action="keep_open", + kb_refs=[], + summary_text="reply ready", + model="stub-voice", + latency_ms=1, + status="active", + ), + request_handoff=lambda session_id, customer_request_text, decision: None, + handle_media_error=lambda session_id, message, metadata: errors.append((session_id, message)), + ) + + async def _scenario() -> None: + await runtime.start() + port = runtime._server.sockets[0].getsockname()[1] + + reader, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(encode_packet(AUDIO_SOCKET_PACKET_UUID, uuid.UUID(media_uuid).bytes)) + await writer.drain() + + packet_type, _ = await read_packet(reader, timeout=2.0) + assert packet_type == AUDIO_SOCKET_PACKET_PCM16 + await asyncio.sleep(0.15) + + speech_frame = (1000).to_bytes(2, "little", signed=True) * 160 + silence_frame = b"\x00\x00" * 160 + for _ in range(2): + writer.write(encode_audio_packet(speech_frame)) + for _ in range(2): + writer.write(encode_audio_packet(silence_frame)) + await writer.drain() + + for _ in range(40): + if len(delivered) >= 2 and errors: + break + await asyncio.sleep(0.05) + + writer.close() + await writer.wait_closed() + await asyncio.sleep(0.2) + await runtime.stop() + + asyncio.run(_scenario()) + + assert delivered[0] == ("avs_media_runtime_asr_error", "greeting", True) + assert delivered[-1] == ( + "avs_media_runtime_asr_error", + "Возникла техническая проблема со связью. Соединяю с оператором.", + False, + ) + assert errors == [("avs_media_runtime_asr_error", "AI_VOICE_ASR_YANDEX_API_KEY is required for Yandex ASR")] + assert ("avs_media_runtime_asr_error", "handoff_requested") in states + + def test_media_runtime_waits_for_late_registration(): registrations: dict[str, MediaRegistration] = {} delivered: list[tuple[str, str, bool]] = []