fix(voice): harden noisy streaming turns

services/ai_orchestrator_service/voice.py

@@ -930,6 +930,19 @@ def _voice_topic_prompt(language: str, caller_texts: list[str]) -> str | None:
     return None
 
 
+def _voice_summary_slot_prompt(language: str, context_summary: str | dict[str, Any] | None) -> str | None:
+    summary = load_context_summary(context_summary)
+    intent = str(summary.get("active_intent") or "").strip()
+    facts = summary.get("confirmed_facts") if isinstance(summary.get("confirmed_facts"), dict) else {}
+    city = str(facts.get("city") or "").strip()
+    branch_hint = str(facts.get("branch_hint") or "").strip()
+    if intent == "schedule" and city and not branch_hint:
+        if language == "kz":
+            return f"{city} qalasy boiynsha qaysy filial nemese mekenjai qyzyqtyratynyn aitnyz."
+        return f"По городу {city} уточните, пожалуйста, филиал или адрес."
+    return None
+
+
 def _voice_generic_prompt(language: str) -> str:
     if language == "kz":
         return "Jyldam komektesu ushin eki ush sozben ne kerek ekenin aitnyz: jumys uaqyty, otinish statusy, tarif nemese operator."
@@ -1850,6 +1863,26 @@ def _voice_decision(
             decision["metadata"] = v2_metadata
         return decision
 
+    summary_slot_prompt = _voice_summary_slot_prompt(language, context_summary)
+    if not kb_results and summary_slot_prompt:
+        decision = {
+            "language": language,
+            "intent": "clarification",
+            "reply_text": summary_slot_prompt,
+            "confidence": 0.72,
+            "needs_handoff": False,
+            "handoff_reason": None,
+            "case_action": "keep_open",
+            "kb_refs": [],
+            "summary_text": "AI продолжил активный сценарий из conversation summary и уточнил недостающий слот.",
+            "model": "voice_policy_context_summary",
+            "latency_ms": 1,
+        }
+        if v2_metadata:
+            decision["reply_text"] = _voice_compact_reply_text(decision["reply_text"], language=language)
+            decision["metadata"] = {**v2_metadata, "reply_phase": "final"}
+        return decision
+
     llm_decision = _voice_llm_decision(
         language=language,
         customer=customer,

services/ai_voice_runtime_service/media_runtime.py

@@ -5,6 +5,7 @@ import audioop
 import contextlib
 import hashlib
 import logging
+import os
 import threading
 import time
 import uuid
@@ -186,6 +187,10 @@ class AudioSocketMediaRuntime:
         self._immediate_ack_min_ms = 700
         self._v2_ack_post_gap_seconds = 0.10
         self._partial_poll_interval_seconds = 0.20
+        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
@@ -197,7 +202,10 @@ class AudioSocketMediaRuntime:
 
     @staticmethod
     def _normalize_intent_text(text: str) -> str:
-        return " ".join(str(text or "").strip().lower().split())
+        compact = str(text or "").strip().lower()
+        for char in ",.!?;:…\"'()[]{}":
+            compact = compact.replace(char, " ")
+        return " ".join(compact.split())
 
     @classmethod
     def _is_low_signal_transcript(cls, text: str | None) -> bool:
@@ -220,9 +228,12 @@ class AudioSocketMediaRuntime:
             "привет",
             "слышу",
             "слышно",
+            "твой",
             "угу",
             "хорошо",
             "ясно",
+            "давай",
+            "поргай что это",
         }
 
     def _detect_early_intent(self, text: str) -> str:
@@ -270,7 +281,7 @@ class AudioSocketMediaRuntime:
             return "Сейчас сориентирую."
         if ack_kind == "clarify":
             return "Сейчас уточню."
-        return "Сейчас подскажу."
+        return "Секунду."
 
     @staticmethod
     def _should_use_emotive_ack(registration: MediaRegistration, language: str | None) -> bool:
@@ -323,11 +334,10 @@ class AudioSocketMediaRuntime:
                 "Ага, сейчас сориентирую.",
             )
         return (
-            "Угу, сейчас подскажу.",
             "Мхм, секунду.",
-            "Ага, сейчас подскажу.",
+            "Секунду.",
             "Хм, секунду.",
-            "Хорошо, сейчас подскажу.",
+            "Хорошо, секунду.",
         )
 
     def _select_ack_payload(
@@ -623,6 +633,36 @@ class AudioSocketMediaRuntime:
         intent = self._detect_early_intent(transcript_text)
         self._update_stable_partial_intent(actor, intent)
 
+    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")
@@ -756,10 +796,11 @@ class AudioSocketMediaRuntime:
         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 = self._ack_kind_for_intent(actor.partial_intent or "unknown")
+        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,
@@ -774,7 +815,7 @@ class AudioSocketMediaRuntime:
             metadata={
                 **metadata,
                 "partial_transcript": actor.partial_transcript,
-                "early_intent": actor.partial_intent,
+                "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",
@@ -1127,6 +1168,7 @@ class AudioSocketMediaRuntime:
         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:
@@ -1183,14 +1225,7 @@ class AudioSocketMediaRuntime:
             "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:
+            if not actor.asr_streaming_enabled:
                 partial_task = actor.partial_asr_task
                 if partial_task is not None and not partial_task.done():
                     with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError):
@@ -1212,6 +1247,7 @@ class AudioSocketMediaRuntime:
                     language=actor.registration.language,
                     metadata=base_metadata,
                     ack_source="immediate_turn_close",
+                    ack_kind="unknown",
                 )
 
         if actor.asr_streaming_enabled and not actor.asr_streaming_failed:

services/shared/ai_context_summary.py

@@ -33,6 +33,7 @@ _LOW_SIGNAL_TEXTS = {
 _CITY_ALIASES = {
     "алма ата": "Алма-Ата",
     "алмат": "Алмата",
+    "матта": "Алмата",
     "астан": "Астана",
     "актау": "Актау",
     "актоб": "Актобе",

tests/test_ai_context_summary.py

@@ -24,6 +24,30 @@ def test_context_summary_tracks_city_and_branch_slot_for_schedule():
     assert summary["open_slots"] == ["branch_or_address"]
 
 
+def test_context_summary_maps_common_asr_city_mishearing_to_almaty():
+    summary = update_context_summary_from_user_turn(
+        None,
+        channel="voice",
+        language="ru",
+        customer_name="Ернур",
+        text="Хочу узнать график работы.",
+        now="2026-04-12T00:00:00+00:00",
+    )
+
+    updated = update_context_summary_from_user_turn(
+        json.dumps(summary, ensure_ascii=False),
+        channel="voice",
+        language="ru",
+        customer_name="Ернур",
+        text="Матта.",
+        now="2026-04-12T00:00:02+00:00",
+    )
+
+    assert updated["active_intent"] == "schedule"
+    assert updated["confirmed_facts"]["city"] == "Алмата"
+    assert updated["open_slots"] == ["branch_or_address"]
+
+
 def test_context_summary_does_not_overwrite_meaningful_request_with_low_signal():
     summary = update_context_summary_from_user_turn(
         None,

tests/test_ai_orchestrator_service.py

@@ -1655,6 +1655,44 @@ def test_voice_postprocess_reply_uses_summary_context_when_raw_window_lost_topic
     assert "филиал" in normalized or "адрес" in normalized
 
 
+def test_voice_decision_uses_summary_city_slot_instead_of_asking_city_again(monkeypatch):
+    def _unexpected_llm(messages):
+        raise AssertionError(f"LLM should not be called when summary slot prompt is enough: {messages!r}")
+
+    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)
+
+    decision = voice_module._voice_decision(
+        language="ru",
+        customer=SimpleNamespace(display_name="Ернур"),
+        interaction=SimpleNamespace(interaction_id="int_voice_summary_slot", customer_id=None, status="open", queue_id=None, subject=None),
+        transcript_text="Матта.",
+        transcript_window=[
+            SimpleNamespace(speaker="caller", text="Матта.", sequence_no=1, source_type="voice_asr", barge_in_interrupted=False, created_at=utc_now_iso()),
+        ],
+        context_summary=json.dumps(
+            {
+                "customer_name": "Ернур",
+                "active_intent": "schedule",
+                "active_request_text": "Хочу узнать график работы",
+                "confirmed_facts": {"city": "Алмата", "branch_hint": None, "service_hint": "график работы", "request_number": None},
+                "open_slots": ["branch_or_address"],
+            },
+            ensure_ascii=False,
+        ),
+        kb_results=[],
+        disclosure_required=False,
+        customer_name_value="Ернур",
+        customer_name_status="name_obtained",
+        request_metadata={"voice_v2_enabled": True},
+    )
+
+    assert decision["intent"] == "clarification"
+    assert decision["needs_handoff"] is False
+    assert "Алмата" in decision["reply_text"]
+    assert "город" in decision["reply_text"].lower()
+    assert "филиал" in decision["reply_text"].lower() or "адрес" in decision["reply_text"].lower()
+
+
 def test_ai_enqueue_creates_outbound_ai_reply_and_delivery_flow(monkeypatch):
     monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
     monkeypatch.setenv("AI_PROVIDER", "stub")

tests/test_ai_voice_media_runtime.py

@@ -1047,6 +1047,12 @@ def test_media_runtime_voice_v2_emits_blind_ack_on_first_turn_without_partial_si
     assert speak_events[1][0] == "Подскажите подробнее, пожалуйста."
 
 
+def test_media_runtime_low_signal_filter_catches_short_asr_noise():
+    assert AudioSocketMediaRuntime._is_low_signal_transcript("Давай")
+    assert AudioSocketMediaRuntime._is_low_signal_transcript("твой")
+    assert AudioSocketMediaRuntime._is_low_signal_transcript("Поргай, что это")
+
+
 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]] = []
@@ -1624,6 +1630,120 @@ def test_media_runtime_streaming_sidecar_push_does_not_block_vad_finalization():
     assert streaming_provider.push_count >= 1
 
 
+def test_media_runtime_streaming_partial_poll_does_not_block_turn_close_ack():
+    events: list[str] = []
+
+    class _StreamingProvider(StreamingASRProvider):
+        name = "streaming-sidecar"
+        supports_streaming = True
+
+        def __init__(self) -> None:
+            self.poll_count = 0
+
+        def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
+            del stream_id
+            self.poll_count += 1
+            time.sleep(0.25)
+            return StreamingASRPartial(text="need schedule", language="ru", confidence=0.8, is_stable=True)
+
+        def finalize(self, stream_id: str) -> ASRTranscription:
+            assert stream_id == "stream-1"
+            events.append("finalize")
+            return ASRTranscription(text="need schedule", language="ru", confidence=0.9)
+
+        def close_stream(self, stream_id: str) -> None:
+            assert stream_id == "stream-1"
+
+    streaming_provider = _StreamingProvider()
+    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=_StubASRProvider(),
+        streaming_asr_provider=streaming_provider,
+        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: None,
+        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="schedule",
+            reply_text="reply",
+            confidence=0.8,
+            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 _fake_speak_reply(
+        actor: MediaActor,
+        text: str,
+        *,
+        is_greeting: bool,
+        style_hints: dict[str, object] | None = None,
+        reply_phase: str | None = "main",
+    ) -> None:
+        del actor, text, is_greeting, style_hints
+        events.append(str(reply_phase))
+
+    runtime._speak_reply = _fake_speak_reply  # type: ignore[method-assign]
+
+    async def _scenario() -> None:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_turn_close_no_poll",
+                call_id="call_media_runtime_turn_close_no_poll",
+                interaction_id="int_media_runtime_turn_close_no_poll",
+                ai_session_id="ais_media_runtime_turn_close_no_poll",
+                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=True,
+                voice_v2_duplex=True,
+                voice_v2_streaming_asr_backend="local_sidecar",
+            ),
+            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,
+        )
+        actor.asr_streaming_enabled = True
+        actor.asr_stream_id = "stream-1"
+        pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        await runtime._process_utterance(actor, pcm_frame * 40, False)
+
+    asyncio.run(_scenario())
+
+    assert streaming_provider.poll_count == 0
+    assert events[:2] == ["ack", "finalize"]
+    assert "main" in events
+
+
 def test_media_runtime_merges_thinking_continuation_into_current_utterance():
     runtime = AudioSocketMediaRuntime(
         enabled=True,