fix(voice): avoid silence on ASR runtime failures

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]] = []