fix(voice): race slow streaming asr finalize

services/ai_orchestrator_service/voice.py

@@ -607,8 +607,12 @@ def _voice_reply_with_name(language: str, reply_text: str, name: str | None) ->
     short_name = _voice_short_name(name)
     if not short_name:
         return reply_text
+    reply_text = _voice_strip_reply_greeting_prefix(reply_text, name=short_name)
     prefix = _voice_disclosure_prefix(language)
     normalized_short = _voice_text_key(short_name)
+    first_words = _voice_text_key(" ".join(str(reply_text or "").split()[:8])).split()
+    if normalized_short in first_words:
+        return reply_text
     if reply_text.startswith(prefix):
         rest = reply_text[len(prefix) :].lstrip()
         if _voice_text_key(rest).startswith(normalized_short):
@@ -1344,6 +1348,28 @@ def _voice_is_midcall_greeting_reply(text: str | None) -> bool:
     )
 
 
+def _voice_strip_reply_greeting_prefix(reply_text: str | None, *, name: str | None = None) -> str:
+    text = str(reply_text or "").strip()
+    if not text:
+        return text
+    greeting_pattern = (
+        r"^\s*(?:здравствуйте|здравствуй|привет(?:ствую)?|"
+        r"добрый\s+(?:день|вечер|утро)|сәлеметсіз\s+бе|сәлем)\b[\s,!.:-]*"
+    )
+    stripped = re.sub(greeting_pattern, "", text, count=1, flags=re.IGNORECASE).strip()
+    if name and stripped != text:
+        short_name = _voice_short_name(name)
+        if short_name:
+            stripped = re.sub(
+                rf"^\s*{re.escape(short_name)}\b[\s,!.:-]*",
+                "",
+                stripped,
+                count=1,
+                flags=re.IGNORECASE,
+            ).strip()
+    return stripped or text
+
+
 def _voice_contains_false_lookup_promise(text: str | None) -> bool:
     normalized = _voice_text_key(text)
     if not normalized:
@@ -1388,15 +1414,16 @@ def _voice_postprocess_reply_text(
     prior_caller_texts = _voice_caller_context_before_current(caller_texts, transcript_text)
     active_topic_texts = _voice_service_context_texts(prior_caller_texts or caller_texts)
     active_topic_prompt = _voice_topic_prompt(language, active_topic_texts) if active_topic_texts else None
+    has_assistant_context = any(segment.speaker == "assistant" for segment in transcript_window)
     if _voice_is_hearing_check_caller_text(transcript_text):
         return _voice_hearing_check_reply(language, prior_caller_texts or caller_texts)
     if _voice_contains_false_lookup_promise(normalized_reply) and not kb_results:
         return active_topic_prompt or _voice_topic_prompt(language, caller_texts) or _voice_generic_prompt(language)
-    if _voice_is_midcall_greeting_reply(normalized_reply) and any(
-        segment.speaker == "assistant" for segment in transcript_window
-    ):
+    if _voice_is_midcall_greeting_reply(normalized_reply) and has_assistant_context:
         context_texts = prior_caller_texts if _voice_is_low_signal_caller_text(transcript_text) else caller_texts
         return _voice_topic_prompt(language, context_texts) or _voice_generic_prompt(language)
+    if has_assistant_context or _voice_has_service_topic(transcript_text):
+        normalized_reply = _voice_strip_reply_greeting_prefix(normalized_reply)
     if (
         active_topic_prompt
         and not _voice_has_service_topic(transcript_text)

services/ai_voice_runtime_service/media_runtime.py

@@ -40,6 +40,16 @@ from services.shared.models import VoiceAITurnDecisionOut
 logger = logging.getLogger("uvicorn.error")
 
 
+def _env_float(name: str, default: float) -> float:
+    raw = str(os.getenv(name, "")).strip()
+    if not raw:
+        return default
+    try:
+        return float(raw)
+    except ValueError:
+        return default
+
+
 @dataclass(slots=True)
 class MediaRegistration:
     voice_session_id: str
@@ -204,6 +214,10 @@ class AudioSocketMediaRuntime:
         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._streaming_finalize_race_grace_seconds = max(
+            0.0,
+            _env_float("AI_VOICE_V2_STREAMING_FINALIZE_RACE_GRACE_SECONDS", 0.35),
+        )
         self._thinking_continuation_grace_seconds = 0.45
         self._thinking_continuation_max_bytes = int(1800 * 16)
         self._partial_first_final_enabled = (
@@ -1056,6 +1070,68 @@ class AudioSocketMediaRuntime:
         finally:
             await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id)
 
+    def _observe_late_streaming_finalize(self, actor: MediaActor, task: asyncio.Task) -> None:
+        def _done(done_task: asyncio.Task) -> None:
+            try:
+                done_task.result()
+            except asyncio.CancelledError:
+                return
+            except StreamingASRUnavailable as exc:
+                logger.warning(
+                    "audiosocket.streaming_asr_late_finalize_failed session_id=%s error=%s",
+                    actor.registration.voice_session_id,
+                    str(exc)[:500],
+                )
+                self._mark_streaming_asr_backoff(actor)
+            except Exception as exc:
+                logger.warning(
+                    "audiosocket.streaming_asr_late_finalize_error session_id=%s error=%s",
+                    actor.registration.voice_session_id,
+                    str(exc)[:500],
+                )
+
+        task.add_done_callback(_done)
+
+    async def _batch_transcribe_turn(
+        self,
+        actor: MediaActor,
+        *,
+        pcm_bytes: bytes,
+    ) -> ASRTranscription:
+        wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
+        return await asyncio.to_thread(
+            self._asr_provider.transcribe,
+            wav_bytes,
+            language_hint=actor.registration.language,
+        )
+
+    async def _finalize_transcription_after_streaming_failure(
+        self,
+        actor: MediaActor,
+        *,
+        pcm_bytes: bytes,
+        partial_transcript: str,
+        error: Exception,
+    ) -> tuple[ASRTranscription, str]:
+        logger.warning(
+            "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s",
+            actor.registration.voice_session_id,
+            str(error)[:500],
+        )
+        self._mark_streaming_asr_backoff(actor)
+        partial_first_text = str(partial_transcript or 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_transcript(partial_first_text):
+            return (
+                ASRTranscription(
+                    text=partial_first_text,
+                    language=actor.registration.language,
+                    confidence=None,
+                ),
+                "streaming_partial_after_finalize_failure",
+            )
+        transcription = await self._batch_transcribe_turn(actor, pcm_bytes=pcm_bytes)
+        return transcription, "batch_fallback_after_streaming_failure"
+
     async def _finalize_turn_transcription(
         self,
         actor: MediaActor,
@@ -1066,45 +1142,91 @@ class AudioSocketMediaRuntime:
     ) -> tuple[ASRTranscription, str]:
         if detached_stream is not None and detached_stream[0]:
             stream_id, push_task, push_queue, streaming_failed = detached_stream
-            try:
-                transcription = await self._finalize_detached_streaming_transcription(
+            streaming_task = asyncio.create_task(
+                self._finalize_detached_streaming_transcription(
                     actor,
                     stream_id=stream_id,
                     push_task=push_task,
                     push_queue=push_queue,
                     streaming_failed=streaming_failed,
                 )
-                return transcription, "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],
+            )
+            if self._streaming_finalize_race_grace_seconds > 0:
+                done, _pending = await asyncio.wait(
+                    {streaming_task},
+                    timeout=self._streaming_finalize_race_grace_seconds,
                 )
-                self._mark_streaming_asr_backoff(actor)
-                partial_first_text = str(partial_transcript or 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_transcript(partial_first_text):
-                    return (
-                        ASRTranscription(
-                            text=partial_first_text,
-                            language=actor.registration.language,
-                            confidence=None,
-                        ),
-                        "streaming_partial_after_finalize_failure",
+                if streaming_task in done:
+                    try:
+                        return streaming_task.result(), "streaming_final"
+                    except StreamingASRUnavailable as exc:
+                        return await self._finalize_transcription_after_streaming_failure(
+                            actor,
+                            pcm_bytes=pcm_bytes,
+                            partial_transcript=partial_transcript,
+                            error=exc,
+                        )
+            partial_first_text = str(partial_transcript or 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_transcript(partial_first_text):
+                self._observe_late_streaming_finalize(actor, streaming_task)
+                return (
+                    ASRTranscription(
+                        text=partial_first_text,
+                        language=actor.registration.language,
+                        confidence=None,
+                    ),
+                    "streaming_partial_before_slow_finalize",
+                )
+            batch_task = asyncio.create_task(
+                self._batch_transcribe_turn(actor, pcm_bytes=pcm_bytes),
+                name=f"voice-batch-asr-race-{actor.registration.voice_session_id}",
+            )
+            done, _pending = await asyncio.wait(
+                {streaming_task, batch_task},
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+            if streaming_task in done:
+                try:
+                    return streaming_task.result(), "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],
                     )
-                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,
-                )
-                return transcription, "batch_fallback_after_streaming_failure"
-        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,
-        )
+                    self._mark_streaming_asr_backoff(actor)
+                    partial_first_text = str(
+                        partial_transcript or 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_transcript(partial_first_text)
+                    ):
+                        return (
+                            ASRTranscription(
+                                text=partial_first_text,
+                                language=actor.registration.language,
+                                confidence=None,
+                            ),
+                            "streaming_partial_after_finalize_failure",
+                        )
+                    return await batch_task, "batch_fallback_after_streaming_failure"
+            try:
+                transcription = batch_task.result()
+            except Exception:
+                try:
+                    return await streaming_task, "streaming_final"
+                except StreamingASRUnavailable as exc:
+                    return await self._finalize_transcription_after_streaming_failure(
+                        actor,
+                        pcm_bytes=pcm_bytes,
+                        partial_transcript=partial_transcript,
+                        error=exc,
+                    )
+            self._observe_late_streaming_finalize(actor, streaming_task)
+            return transcription, "batch_race_before_streaming_finalize"
+        transcription = await self._batch_transcribe_turn(actor, pcm_bytes=pcm_bytes)
         return transcription, "batch"
 
     async def _record_reply_status(

tests/test_ai_orchestrator_service.py

@@ -1546,6 +1546,19 @@ def test_voice_postprocess_reply_rewrites_midcall_greeting_with_followup_questio
     assert "о чем именно" not in normalized
 
 
+def test_voice_reply_with_name_strips_midcall_greeting_and_does_not_double_prefix():
+    reply_text = voice_module._voice_reply_with_name(
+        "ru",
+        "Здравствуйте, Ернур! Чтобы уточнить график работы, назовите город или филиал.",
+        "Ернур",
+    )
+
+    normalized = reply_text.lower()
+    assert reply_text.startswith("Ернур, Чтобы уточнить")
+    assert normalized.count("ернур") == 1
+    assert "здравствуйте" not in normalized
+
+
 def test_voice_postprocess_reply_reuses_active_topic_after_frustration_turn():
     reply_text = voice_module._voice_postprocess_reply_text(
         language="ru",

tests/test_ai_voice_media_runtime.py

@@ -2240,6 +2240,127 @@ def test_media_runtime_voice_v2_uses_partial_as_final_when_streaming_finalize_fa
     assert final_turns[0][1]["transcript_source"] == "streaming_partial_after_finalize_failure"
 
 
+def test_media_runtime_races_batch_asr_when_streaming_finalize_is_slow():
+    class _CountingASRProvider(ASRProvider):
+        name = "counting-asr"
+
+        def __init__(self) -> None:
+            self.transcribe_count = 0
+
+        def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription:
+            assert audio_bytes
+            self.transcribe_count += 1
+            return ASRTranscription(text="batch schedule", language=language_hint or "ru", confidence=0.9)
+
+    class _SlowStreamingProvider(StreamingASRProvider):
+        name = "slow-streaming"
+        supports_streaming = True
+
+        def __init__(self) -> None:
+            self.finalize_started = False
+            self.finalize_done = False
+
+        def finalize(self, stream_id: str) -> ASRTranscription:
+            assert stream_id == "stream-1"
+            self.finalize_started = True
+            time.sleep(0.35)
+            self.finalize_done = True
+            return ASRTranscription(text="streaming schedule", language="ru", confidence=0.9)
+
+        def close_stream(self, stream_id: str) -> None:
+            assert stream_id == "stream-1"
+
+    asr_provider = _CountingASRProvider()
+    streaming_provider = _SlowStreamingProvider()
+    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=asr_provider,
+        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,
+    )
+    runtime._streaming_finalize_race_grace_seconds = 0.01
+
+    async def _scenario() -> tuple[ASRTranscription, str, float]:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_asr_race",
+                call_id="call_media_runtime_asr_race",
+                interaction_id="int_media_runtime_asr_race",
+                ai_session_id="ais_media_runtime_asr_race",
+                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,
+        )
+        pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        started = time.monotonic()
+        result = await runtime._finalize_turn_transcription(
+            actor,
+            pcm_bytes=pcm_frame * 40,
+            partial_transcript="",
+            detached_stream=("stream-1", None, None, False),
+        )
+        elapsed = time.monotonic() - started
+        for _ in range(50):
+            if streaming_provider.finalize_done:
+                break
+            await asyncio.sleep(0.01)
+        return result[0], result[1], elapsed
+
+    transcription, source, elapsed = asyncio.run(_scenario())
+
+    assert streaming_provider.finalize_started is True
+    assert asr_provider.transcribe_count == 1
+    assert transcription.text == "batch schedule"
+    assert source == "batch_race_before_streaming_finalize"
+    assert elapsed < 0.2
+
+
 def test_media_runtime_merges_thinking_continuation_into_current_utterance():
     runtime = AudioSocketMediaRuntime(
         enabled=True,