fix(voice): suppress low-signal turns and false lookup replies

tests/test_ai_voice_media_runtime.py

@@ -842,7 +842,7 @@ def test_media_runtime_voice_v2_emits_generic_ack_before_full_asr_without_partia
             frame_ms=20,
             frame_bytes=320,
         )
-        await runtime._process_utterance(actor, pcm_frame * 30, False)
+        await runtime._process_utterance(actor, pcm_frame * 45, False)
 
     asyncio.run(_scenario())
 
@@ -940,7 +940,7 @@ def test_media_runtime_voice_v2_emits_ack_for_short_utterance_after_reduced_thre
             frame_ms=20,
             frame_bytes=320,
         )
-        await runtime._process_utterance(actor, pcm_frame * 20, False)
+        await runtime._process_utterance(actor, pcm_frame * 40, False)
 
     asyncio.run(_scenario())
 
@@ -949,6 +949,92 @@ def test_media_runtime_voice_v2_emits_ack_for_short_utterance_after_reduced_thre
     assert speak_events[0][1] < timings["full_finished"]
 
 
+def test_media_runtime_ignores_low_signal_utterance_without_ack_or_turn():
+    planned: list[tuple[str, str, str, dict | None]] = []
+    delivered: list[tuple[str, str, bool]] = []
+    turns: list[str] = []
+
+    class _LowSignalASRProvider(ASRProvider):
+        name = "low-signal-asr"
+
+        def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+            assert audio_bytes
+            return ASRTranscription(text="Ой", language=language_hint or "ru", confidence=0.82)
+
+    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=2000,
+        asr_provider=_LowSignalASRProvider(),
+        tts_provider=_StubTTSProvider(),
+        load_registration_by_media_uuid=lambda value: None,
+        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: None,
+        get_pending_greeting=lambda session_id: 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: planned.append((session_id, text, kind, metadata)),
+        process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
+            turns.append(transcript_text)
+            or VoiceAITurnDecisionOut(
+                language=language or "ru",
+                intent="clarification",
+                reply_text="Подскажите подробнее.",
+                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: None,
+    )
+
+    async def _scenario() -> None:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_low_signal",
+                call_id="call_media_runtime_low_signal",
+                interaction_id="int_media_runtime_low_signal",
+                ai_session_id="ais_media_runtime_low_signal",
+                language="ru",
+                media_uuid=str(uuid.uuid4()),
+                queue_code="voice_lab_ai",
+                queue_id="que_voice_lab_ai",
+                agent_profile="voice_support",
+                voice_v2_enabled=True,
+                voice_v2_ack_mode="immediate_short",
+                voice_v2_streaming_tts=True,
+                voice_v2_partial_asr=False,
+            ),
+            reader=asyncio.StreamReader(),
+            writer=None,  # type: ignore[arg-type]
+            vad=EnergyVAD(frame_ms=20, min_speech_ms=40, trailing_silence_ms=40, max_turn_ms=2000),
+            frame_ms=20,
+            frame_bytes=320,
+        )
+        pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        await runtime._process_utterance(actor, pcm_frame * 12, False)
+
+    asyncio.run(_scenario())
+
+    assert turns == []
+    assert planned == []
+    assert delivered == []
+
+
 def test_media_runtime_voice_v2_inserts_small_gap_between_ack_and_main_reply():
     speak_events: list[tuple[str, float]] = []