feat(voice): start replies from stable streaming partials

services/ai_orchestrator_service/voice.py

@@ -267,6 +267,10 @@ def _normalize_name_candidate(text: str | None) -> str | None:
     filtered = [word for word, lowered_word in zip(words, lowered) if lowered_word not in stop_words]
     if not filtered:
         return None
+    if len(filtered) == 2 and len(filtered[1]) == 1:
+        filtered = [filtered[0]]
+    elif len(filtered) == 2 and len(filtered[0]) == 1:
+        filtered = [filtered[1]]
     invalid_tokens = {
         "\u0434\u0430",
         "\u0445\u0430\u0447\u0443",
@@ -1430,9 +1434,12 @@ def _voice_early_plan(
     *,
     language: str,
     transcript_text: str,
+    context_summary: str | dict[str, Any] | None = None,
     request_metadata: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     v2_metadata = _voice_v2_metadata(transcript_text, request_metadata)
+    payload = request_metadata if isinstance(request_metadata, dict) else {}
+    early_intent = str(payload.get("early_intent") or v2_metadata.get("early_intent") or "").strip()
     lower_text = str(transcript_text or "").strip().lower()
     if _app()._looks_like_human_request(lower_text) or _app()._is_sensitive_request(lower_text):
         return {
@@ -1471,6 +1478,55 @@ def _voice_early_plan(
                 "reply_phase": "early_plan",
             },
         }
+    if early_intent in {"schedule", "address", "price", "status", "problem", "operator_request"}:
+        reply_text = _voice_summary_slot_prompt(language, context_summary) or _voice_topic_prompt(language, [transcript_text])
+        if early_intent == "operator_request":
+            reply_text = _voice_handoff_reply(language)
+        if not reply_text and language == "kz":
+            if early_intent == "schedule":
+                reply_text = "Qai filialdyn, mekenjaidyn nemese qalanyng jumys uaqyty qyzyqtyratynyn aitnyz."
+            elif early_intent == "address":
+                reply_text = "Qai filial, mekenjai nemese qala qyzyqtyratynyn aitnyz."
+            elif early_intent == "price":
+                reply_text = "Qai qyzmet, tarif nemese bagasy qyzyqtyratynyn naqtylanyz."
+            elif early_intent == "status":
+                reply_text = "Otinish nomirin nemese resimdegen telefon nomirin aitnyz."
+            elif early_intent == "problem":
+                reply_text = "Bir soilemmen naqty aitynyz: ne istep turmagan?"
+        if not reply_text:
+            if early_intent == "schedule":
+                reply_text = "\u041f\u043e\u0434\u0441\u043a\u0430\u0436\u0438\u0442\u0435, \u0433\u0440\u0430\u0444\u0438\u043a \u0440\u0430\u0431\u043e\u0442\u044b \u043a\u0430\u043a\u043e\u0433\u043e \u0444\u0438\u043b\u0438\u0430\u043b\u0430, \u0430\u0434\u0440\u0435\u0441\u0430 \u0438\u043b\u0438 \u0433\u043e\u0440\u043e\u0434\u0430 \u0432\u0430\u0441 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u0443\u0435\u0442?"
+            elif early_intent == "address":
+                reply_text = "\u041f\u043e\u0434\u0441\u043a\u0430\u0436\u0438\u0442\u0435, \u043a\u0430\u043a\u043e\u0439 \u0444\u0438\u043b\u0438\u0430\u043b, \u0430\u0434\u0440\u0435\u0441 \u0438\u043b\u0438 \u0433\u043e\u0440\u043e\u0434 \u0432\u0430\u0441 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u0443\u0435\u0442."
+            elif early_intent == "price":
+                reply_text = "\u0423\u0442\u043e\u0447\u043d\u0438\u0442\u0435, \u043a\u0430\u043a\u0430\u044f \u0443\u0441\u043b\u0443\u0433\u0430, \u0442\u0430\u0440\u0438\u0444 \u0438\u043b\u0438 \u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u044c \u0432\u0430\u0441 \u0438\u043d\u0442\u0435\u0440\u0435\u0441\u0443\u0435\u0442."
+            elif early_intent == "status":
+                reply_text = "\u041d\u0430\u0437\u043e\u0432\u0438\u0442\u0435, \u043f\u043e\u0436\u0430\u043b\u0443\u0439\u0441\u0442\u0430, \u043d\u043e\u043c\u0435\u0440 \u0437\u0430\u044f\u0432\u043a\u0438 \u0438\u043b\u0438 \u0442\u0435\u043b\u0435\u0444\u043e\u043d, \u043f\u043e \u043a\u043e\u0442\u043e\u0440\u043e\u043c\u0443 \u043e\u043d\u0430 \u043e\u0444\u043e\u0440\u043c\u043b\u044f\u043b\u0430\u0441\u044c."
+            elif early_intent == "problem":
+                reply_text = "\u041e\u043f\u0438\u0448\u0438\u0442\u0435, \u043f\u043e\u0436\u0430\u043b\u0443\u0439\u0441\u0442\u0430, \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u0443 \u043e\u0434\u043d\u0438\u043c \u043f\u0440\u0435\u0434\u043b\u043e\u0436\u0435\u043d\u0438\u0435\u043c: \u0447\u0442\u043e \u0438\u043c\u0435\u043d\u043d\u043e \u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442?"
+        if reply_text:
+            return {
+                "language": language,
+                "intent": "handoff_request" if early_intent == "operator_request" else "clarification",
+                "reply_text": _voice_compact_reply_text(reply_text, language=language),
+                "confidence": 0.62,
+                "needs_handoff": early_intent == "operator_request",
+                "handoff_reason": (
+                    "Запрос требует участия живого оператора."
+                    if early_intent == "operator_request"
+                    else None
+                ),
+                "case_action": "keep_open",
+                "kb_refs": [],
+                "summary_text": "Early domain plan is prepared.",
+                "model": "voice_early_plan_domain",
+                "latency_ms": 1,
+                "metadata": {
+                    **v2_metadata,
+                    "reply_phase": "early_plan",
+                    "early_intent": early_intent,
+                },
+            }
     return {
         "language": language,
         "intent": "clarification",
@@ -1835,6 +1891,7 @@ def _voice_decision(
         return _voice_early_plan(
             language=language,
             transcript_text=transcript_text,
+            context_summary=context_summary,
             request_metadata=request_metadata,
         )
 
@@ -2579,7 +2636,8 @@ def turn_voice_session(session_id: str, payload: VoiceAITurnIn) -> VoiceAITurnDe
             resolved_at=effective_name_resolved_at,
         )
         decision_metadata.update(decision.get("metadata") or {})
-        if effective_name_status == "name_obtained" and effective_name_value:
+        suppress_name_prefix = bool(request_metadata.get("suppress_name_prefix")) if isinstance(request_metadata, dict) else False
+        if effective_name_status == "name_obtained" and effective_name_value and not early_plan_only and not suppress_name_prefix:
             decision["reply_text"] = _voice_reply_with_name(
                 decision["language"],
                 decision["reply_text"],

services/ai_voice_runtime_service/media_runtime.py

@@ -87,9 +87,16 @@ class MediaActor:
     playback_active: bool = False
     early_ack_started: bool = False
     partial_transcript: str | None = None
+    stable_partial_transcript: str | None = None
+    last_partial_transcript: str | None = None
+    partial_transcript_streak: int = 0
     partial_intent: str | None = None
     partial_intent_streak: int = 0
     stable_partial_intent: str | None = None
+    early_plan_task: asyncio.Task | None = None
+    early_plan_transcript: str | None = None
+    early_plan_intent: str | None = None
+    early_plan_generation: int = 0
     response_plan_id: str | None = None
     playback_generation: int = 0
     tts_generation: int = 0
@@ -199,6 +206,25 @@ class AudioSocketMediaRuntime:
         self._streaming_asr_push_drain_timeout_seconds = 1.5
         self._thinking_continuation_grace_seconds = 0.45
         self._thinking_continuation_max_bytes = int(1800 * 16)
+        self._partial_first_final_enabled = (
+            str(os.getenv("AI_VOICE_V2_PARTIAL_FIRST_FINAL", "1")).strip().lower()
+            in {"1", "true", "yes", "on"}
+        )
+        self._early_plan_enabled = (
+            str(os.getenv("AI_VOICE_V2_EARLY_PLAN_ENABLED", "1")).strip().lower()
+            in {"1", "true", "yes", "on"}
+        )
+        raw_early_plan_intents = str(
+            os.getenv(
+                "AI_VOICE_V2_EARLY_PLAN_INTENTS",
+                "schedule,address,price,status,problem,operator_request",
+            )
+        )
+        self._early_plan_intents = {
+            item.strip().lower()
+            for item in raw_early_plan_intents.split(",")
+            if item.strip()
+        }
 
     @staticmethod
     def _normalize_intent_text(text: str) -> str:
@@ -409,10 +435,20 @@ class AudioSocketMediaRuntime:
         actor.input_active = True
         actor.early_ack_started = False
         actor.partial_transcript = None
+        actor.stable_partial_transcript = None
+        actor.last_partial_transcript = None
+        actor.partial_transcript_streak = 0
         actor.partial_intent = None
         actor.partial_intent_streak = 0
         actor.stable_partial_intent = None
-        actor.response_plan_id = None
+        actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}"
+        actor.early_plan_transcript = None
+        actor.early_plan_intent = None
+        actor.early_plan_generation = 0
+        early_plan_task = actor.early_plan_task
+        actor.early_plan_task = None
+        if early_plan_task is not None and not early_plan_task.done():
+            early_plan_task.cancel()
         actor.partial_asr_attempted = False
         actor.asr_streaming_failed = False
         actor.asr_poll_due_monotonic = 0.0
@@ -636,6 +672,134 @@ class AudioSocketMediaRuntime:
         if actor.partial_intent_streak >= 2:
             actor.stable_partial_intent = normalized
 
+    @staticmethod
+    def _update_stable_partial_transcript(actor: MediaActor, transcript_text: str, *, provider_stable: bool) -> None:
+        normalized = " ".join(str(transcript_text or "").strip().lower().split())
+        if not normalized:
+            return
+        if actor.last_partial_transcript == normalized:
+            actor.partial_transcript_streak += 1
+        else:
+            actor.last_partial_transcript = normalized
+            actor.partial_transcript_streak = 1
+        if provider_stable or actor.partial_transcript_streak >= 2:
+            actor.stable_partial_transcript = transcript_text
+
+    def _should_schedule_early_plan(self, actor: MediaActor, transcript_text: str, intent: str) -> bool:
+        if not self._early_plan_enabled or not self._should_use_voice_v2(actor.registration):
+            return False
+        if actor.closed or self._is_low_signal_transcript(transcript_text):
+            return False
+        normalized_intent = str(intent or "").strip().lower() or "unknown"
+        if normalized_intent not in self._early_plan_intents:
+            return False
+        task = actor.early_plan_task
+        if task is not None and task.done() and actor.early_plan_transcript == transcript_text and actor.early_plan_intent == normalized_intent:
+            return False
+        return task is None or task.done()
+
+    def _schedule_early_plan(
+        self,
+        actor: MediaActor,
+        *,
+        transcript_text: str,
+        language: str | None,
+        intent: str,
+        source: str,
+    ) -> None:
+        if not self._should_schedule_early_plan(actor, transcript_text, intent):
+            return
+        if not actor.response_plan_id:
+            actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}"
+        actor.early_plan_generation = actor.utterance_generation
+        actor.early_plan_transcript = transcript_text
+        actor.early_plan_intent = str(intent or "").strip() or "unknown"
+        metadata = {
+            "voice_v2_enabled": actor.registration.voice_v2_enabled,
+            "reply_phase": "early_plan",
+            "response_plan_id": actor.response_plan_id,
+            "playback_generation": actor.playback_generation + 1,
+            "tts_generation": actor.tts_generation + 1,
+            "partial_transcript": transcript_text,
+            "early_intent": actor.early_plan_intent,
+            "transcript_source": source,
+            "suppress_name_prefix": True,
+        }
+        actor.early_plan_task = asyncio.create_task(
+            asyncio.to_thread(
+                self._process_turn,
+                actor.registration.voice_session_id,
+                transcript_text,
+                language or actor.registration.language,
+                False,
+                metadata,
+            ),
+            name=f"voice-early-plan-{actor.registration.voice_session_id}",
+        )
+        logger.info(
+            "audiosocket.early_plan_started session_id=%s generation=%s intent=%s source=%s text_len=%s",
+            actor.registration.voice_session_id,
+            actor.early_plan_generation,
+            actor.early_plan_intent,
+            source,
+            len(transcript_text),
+        )
+
+    async def _take_early_plan_decision(
+        self,
+        actor: MediaActor,
+        *,
+        timeout_seconds: float,
+    ) -> VoiceAITurnDecisionOut | None:
+        task = actor.early_plan_task
+        if task is None:
+            return None
+        if actor.early_plan_generation != actor.utterance_generation:
+            return None
+        if not task.done() and timeout_seconds > 0:
+            with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError):
+                await asyncio.wait_for(asyncio.shield(task), timeout=timeout_seconds)
+        if not task.done():
+            return None
+        try:
+            decision = task.result()
+        except Exception as exc:
+            logger.warning(
+                "audiosocket.early_plan_failed session_id=%s generation=%s error=%s",
+                actor.registration.voice_session_id,
+                actor.early_plan_generation,
+                str(exc)[:500],
+            )
+            return None
+        if not getattr(decision, "reply_text", None) and not getattr(decision, "needs_handoff", False):
+            return None
+        return decision
+
+    async def _reconcile_final_decision_after_early_plan(
+        self,
+        actor: MediaActor,
+        *,
+        decision_task: asyncio.Task,
+        early_decision: VoiceAITurnDecisionOut,
+        transcript_text: str,
+    ) -> None:
+        try:
+            final_decision = await decision_task
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            logger.warning(
+                "audiosocket.final_plan_reconcile_failed session_id=%s error=%s",
+                actor.registration.voice_session_id,
+                str(exc)[:500],
+            )
+            return
+        if actor.closed:
+            return
+        if final_decision.needs_handoff and not early_decision.needs_handoff:
+            await self._set_actor_state(actor, "handoff_requested", final_decision.handoff_reason)
+            self._start_handoff_request(actor, transcript_text, final_decision)
+
     async def _poll_streaming_partial(self, actor: MediaActor) -> None:
         if actor.closed or not actor.asr_streaming_enabled or not actor.asr_stream_id:
             return
@@ -650,8 +814,19 @@ class AudioSocketMediaRuntime:
         if not transcript_text:
             return
         actor.partial_transcript = transcript_text
+        self._update_stable_partial_transcript(actor, transcript_text, provider_stable=bool(partial.is_stable or partial.is_final))
         intent = self._detect_early_intent(transcript_text)
         self._update_stable_partial_intent(actor, intent)
+        stable_text = str(actor.stable_partial_transcript or "").strip()
+        stable_intent = str(actor.stable_partial_intent or "").strip()
+        if stable_text and stable_intent:
+            self._schedule_early_plan(
+                actor,
+                transcript_text=stable_text,
+                language=partial.language or actor.registration.language,
+                intent=stable_intent,
+                source="streaming_stable_partial",
+            )
 
     async def _run_streaming_partial_poll(self, actor: MediaActor, utterance_generation: int) -> None:
         try:
@@ -780,7 +955,16 @@ class AudioSocketMediaRuntime:
         if not transcript_text:
             return
         actor.partial_transcript = transcript_text
+        self._update_stable_partial_transcript(actor, transcript_text, provider_stable=True)
         actor.partial_intent = self._detect_early_intent(transcript_text)
+        actor.stable_partial_intent = actor.partial_intent
+        self._schedule_early_plan(
+            actor,
+            transcript_text=transcript_text,
+            language=transcription.language or actor.registration.language,
+            intent=actor.partial_intent,
+            source="partial_asr_probe",
+        )
         logger.info(
             "audiosocket.partial_asr_ready session_id=%s generation=%s text_len=%s intent=%s",
             actor.registration.voice_session_id,
@@ -880,6 +1064,9 @@ class AudioSocketMediaRuntime:
     ) -> None:
         if actor.closed or not text or not pcm_8k:
             return
+        actor.playback_interrupt.clear()
+        actor.barge_in_detected_monotonic = 0.0
+        actor.barge_in_speech_ms = 0
         await self._set_actor_state(actor, "speaking")
         actor.current_reply_phase = reply_phase
         interrupted = False
@@ -1275,9 +1462,9 @@ class AudioSocketMediaRuntime:
         pcm_bytes = await self._extend_with_thinking_continuation(actor, pcm_bytes)
         actor.playback_generation += 1
         actor.tts_generation += 1
-        actor.response_plan_id = f"rsp_{uuid.uuid4().hex[:10]}"
+        actor.response_plan_id = actor.response_plan_id or f"rsp_{uuid.uuid4().hex[:10]}"
         utterance_generation = actor.utterance_generation
-        partial_transcript = str(actor.partial_transcript or "").strip()
+        partial_transcript = str(actor.stable_partial_transcript or 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 = {
             "turn_duration_ms": int(len(pcm_bytes) / 16),
@@ -1291,15 +1478,22 @@ class AudioSocketMediaRuntime:
             "early_intent": partial_intent,
         }
         if self._should_use_voice_v2(actor.registration) and not actor.early_ack_started:
-            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):
-                        await asyncio.wait_for(asyncio.shield(partial_task), timeout=0.08)
-                        partial_transcript = str(actor.partial_transcript or "").strip()
-                        partial_intent = str(actor.partial_intent or "").strip() or self._detect_early_intent(partial_transcript)
-                        base_metadata["partial_transcript"] = partial_transcript
-                        base_metadata["early_intent"] = partial_intent
+            partial_task = actor.partial_asr_task
+            if partial_task is not None and not partial_task.done():
+                with contextlib.suppress(asyncio.TimeoutError, asyncio.CancelledError):
+                    await asyncio.wait_for(asyncio.shield(partial_task), timeout=0.08)
+            partial_transcript = str(actor.stable_partial_transcript or 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
+            if partial_transcript:
+                self._schedule_early_plan(
+                    actor,
+                    transcript_text=partial_transcript,
+                    language=actor.registration.language,
+                    intent=partial_intent,
+                    source="turn_close_partial",
+                )
             if self._should_emit_partial_ack(actor, partial_transcript, partial_intent):
                 await self._emit_early_ack(
                     actor,
@@ -1316,9 +1510,11 @@ class AudioSocketMediaRuntime:
                     ack_kind="unknown",
                 )
 
+        transcript_source = "batch"
         if actor.asr_streaming_enabled and not actor.asr_streaming_failed:
             try:
                 transcription = await self._finalize_streaming_transcription(actor)
+                transcript_source = "streaming_final"
             except StreamingASRUnavailable as exc:
                 logger.warning(
                     "audiosocket.streaming_asr_finalize_failed session_id=%s error=%s",
@@ -1327,12 +1523,22 @@ class AudioSocketMediaRuntime:
                 )
                 self._mark_streaming_asr_backoff(actor)
                 await self._close_streaming_asr(actor, drain=False)
-                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,
-                )
+                partial_first_text = str(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):
+                    transcription = ASRTranscription(
+                        text=partial_first_text,
+                        language=actor.registration.language,
+                        confidence=None,
+                    )
+                    transcript_source = "streaming_partial_after_finalize_failure"
+                else:
+                    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,
+                    )
+                    transcript_source = "batch_fallback_after_streaming_failure"
         else:
             wav_bytes = pcm16le_to_wav_bytes(pcm_bytes, sample_rate_hz=8000)
             transcription = await asyncio.to_thread(
@@ -1340,6 +1546,7 @@ class AudioSocketMediaRuntime:
                 wav_bytes,
                 language_hint=actor.registration.language,
             )
+            transcript_source = "batch"
         transcript_text = str(transcription.text or "").strip() or partial_transcript
         logger.info(
             "audiosocket.asr_turn_ready session_id=%s provider=%s utterance_ms=%s text_len=%s empty=%s",
@@ -1372,6 +1579,7 @@ class AudioSocketMediaRuntime:
             "partial_transcript": actor.partial_transcript,
             "early_intent": actor.partial_intent,
             "reply_phase": "final",
+            "transcript_source": transcript_source,
         }
         decision_task = asyncio.create_task(
             asyncio.to_thread(
@@ -1386,18 +1594,30 @@ class AudioSocketMediaRuntime:
                 },
             )
         )
+        early_plan_decision: VoiceAITurnDecisionOut | None = None
         if self._should_use_voice_v2(actor.registration):
-            try:
-                decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v2_ack_wait_seconds)
-            except asyncio.TimeoutError:
-                if not actor.early_ack_started:
-                    await self._emit_early_ack(
-                        actor,
-                        language=transcription.language or actor.registration.language,
-                        metadata=metadata,
-                        ack_source="decision_timeout",
-                    )
-                decision = await decision_task
+            early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.04)
+            if early_plan_decision is not None:
+                decision = early_plan_decision
+                logger.info(
+                    "audiosocket.early_plan_used session_id=%s generation=%s intent=%s",
+                    actor.registration.voice_session_id,
+                    actor.early_plan_generation,
+                    actor.early_plan_intent,
+                )
+            else:
+                try:
+                    decision = await asyncio.wait_for(asyncio.shield(decision_task), timeout=self._v2_ack_wait_seconds)
+                except asyncio.TimeoutError:
+                    if not actor.early_ack_started:
+                        await self._emit_early_ack(
+                            actor,
+                            language=transcription.language or actor.registration.language,
+                            metadata=metadata,
+                            ack_source="decision_timeout",
+                        )
+                    early_plan_decision = await self._take_early_plan_decision(actor, timeout_seconds=0.12)
+                    decision = early_plan_decision if early_plan_decision is not None else await decision_task
         else:
             decision = await decision_task
         handoff_task: asyncio.Task | None = None
@@ -1425,6 +1645,24 @@ class AudioSocketMediaRuntime:
                     },
                 )
             await self._speak_reply(actor, decision.reply_text, is_greeting=False, reply_phase="main")
+        if early_plan_decision is not None:
+            if decision_task.done():
+                await self._reconcile_final_decision_after_early_plan(
+                    actor,
+                    decision_task=decision_task,
+                    early_decision=early_plan_decision,
+                    transcript_text=transcript_text,
+                )
+            else:
+                asyncio.create_task(
+                    self._reconcile_final_decision_after_early_plan(
+                        actor,
+                        decision_task=decision_task,
+                        early_decision=early_plan_decision,
+                        transcript_text=transcript_text,
+                    ),
+                    name=f"voice-final-plan-reconcile-{actor.registration.voice_session_id}",
+                )
         if actor.closed:
             return
         if decision.needs_handoff:
@@ -1523,6 +1761,9 @@ class AudioSocketMediaRuntime:
     ) -> None:
         if actor.closed or not text:
             return
+        actor.playback_interrupt.clear()
+        actor.barge_in_detected_monotonic = 0.0
+        actor.barge_in_speech_ms = 0
         await self._set_actor_state(actor, "speaking")
         actor.current_reply_phase = reply_phase
         synth_started_at = time.monotonic()

services/ai_voice_runtime_service/providers/asr.py

@@ -131,14 +131,20 @@ def _elevenlabs_realtime_commit_strategy() -> str:
 
 
 def _elevenlabs_realtime_finalize_timeout_seconds() -> float:
+    raw_ms = os.getenv("AI_VOICE_V2_STREAMING_FINAL_HARD_TIMEOUT_MS", "").strip()
+    if raw_ms:
+        try:
+            return max(float(raw_ms) / 1000.0, 0.25)
+        except ValueError:
+            pass
     raw = os.getenv("AI_VOICE_ASR_ELEVENLABS_REALTIME_FINALIZE_TIMEOUT_SECONDS", "").strip()
     if not raw:
-        return max(_streaming_asr_timeout_seconds(), 1.0)
+        return 0.7
     try:
         value = float(raw)
     except ValueError:
-        value = _streaming_asr_timeout_seconds()
-    return max(value, 1.0)
+        value = 0.7
+    return max(value, 0.25)
 
 
 def _normalize_elevenlabs_realtime_language(language: str | None) -> str:
@@ -513,6 +519,9 @@ class _ElevenLabsRealtimeStreamState:
     updates_queue: queue.Queue[StreamingASRPartial]
     thread: threading.Thread | None = None
     latest_partial: StreamingASRPartial | None = None
+    stable_partial: StreamingASRPartial | None = None
+    last_partial_text: str = ""
+    partial_streak: int = 0
     final_transcription: ASRTranscription | None = None
     error: BaseException | None = None
     close_requested: bool = False
@@ -549,7 +558,7 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
                 if finalize_timeout_seconds is not None
                 else _elevenlabs_realtime_finalize_timeout_seconds()
             ),
-            1.0,
+            0.25,
         )
         self._model_id = str(model_id or _elevenlabs_realtime_model_id()).strip() or "scribe_v2_realtime"
         self._audio_format = str(audio_format or _elevenlabs_realtime_audio_format()).strip().lower() or "pcm_16000"
@@ -650,6 +659,21 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
             except queue.Empty:
                 return
             with state.lock:
+                normalized_text = " ".join(str(update.text or "").strip().lower().split())
+                if normalized_text and normalized_text == state.last_partial_text:
+                    state.partial_streak += 1
+                else:
+                    state.last_partial_text = normalized_text
+                    state.partial_streak = 1 if normalized_text else 0
+                if update.is_final or update.is_stable or state.partial_streak >= 2:
+                    state.stable_partial = StreamingASRPartial(
+                        text=update.text,
+                        language=update.language or state.language,
+                        confidence=update.confidence,
+                        is_final=update.is_final,
+                        is_stable=True,
+                    )
+                    update = state.stable_partial
                 state.latest_partial = update
                 if update.is_final:
                     state.final_transcription = ASRTranscription(
@@ -659,6 +683,25 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
                     )
                     state.final_event.set()
 
+    @staticmethod
+    def _transcription_from_partial(
+        partial: StreamingASRPartial,
+        *,
+        fallback_language: str,
+    ) -> ASRTranscription:
+        return ASRTranscription(
+            text=partial.text,
+            language=partial.language or fallback_language,
+            confidence=partial.confidence,
+        )
+
+    def _best_partial_transcription(self, state: _ElevenLabsRealtimeStreamState) -> ASRTranscription | None:
+        with state.lock:
+            partial = state.stable_partial or state.latest_partial
+            if partial is None:
+                return None
+            return self._transcription_from_partial(partial, fallback_language=state.language)
+
     def _run_reader(self, state: _ElevenLabsRealtimeStreamState) -> None:
         try:
             while True:
@@ -770,36 +813,38 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider):
         with state.lock:
             if state.error is not None and state.latest_partial is None:
                 raise StreamingASRUnavailable(str(state.error)[:500])
-            return state.latest_partial
+            return state.stable_partial or state.latest_partial
 
     def finalize(self, stream_id: str) -> ASRTranscription:
         state = self._state(stream_id)
         silence = b"\x00\x00" * int(self._sample_rate_hz * 0.1)
-        self._send_json(
-            state,
-            {
-                "message_type": "input_audio_chunk",
-                "audio_base_64": base64.b64encode(silence).decode("ascii"),
-                "commit": True,
-                "sample_rate": self._sample_rate_hz,
-            },
-        )
+        try:
+            self._send_json(
+                state,
+                {
+                    "message_type": "input_audio_chunk",
+                    "audio_base_64": base64.b64encode(silence).decode("ascii"),
+                    "commit": True,
+                    "sample_rate": self._sample_rate_hz,
+                },
+            )
+        except StreamingASRUnavailable:
+            partial_transcription = self._best_partial_transcription(state)
+            if partial_transcription is not None and partial_transcription.text.strip():
+                return partial_transcription
+            raise
         deadline = time.monotonic() + self._finalize_timeout_seconds
         while True:
             self._drain_updates(state)
             with state.lock:
                 if state.final_transcription is not None:
                     return state.final_transcription
-                latest_partial = state.latest_partial
+                latest_partial = state.stable_partial or state.latest_partial
                 error = state.error
                 event_is_set = state.final_event.is_set()
             if event_is_set or time.monotonic() >= deadline:
                 if latest_partial is not None:
-                    return ASRTranscription(
-                        text=latest_partial.text,
-                        language=latest_partial.language or state.language,
-                        confidence=latest_partial.confidence,
-                    )
+                    return self._transcription_from_partial(latest_partial, fallback_language=state.language)
                 if error is not None:
                     raise StreamingASRUnavailable(str(error)[:500])
                 if time.monotonic() >= deadline:

tests/test_ai_orchestrator_service.py

@@ -1451,6 +1451,37 @@ def test_voice_v2_streaming_duplex_early_plan_returns_fast_safe_reply_without_ll
     assert "оператор" in decision["reply_text"].lower()
 
 
+def test_voice_v2_streaming_duplex_early_plan_returns_domain_followup_without_llm(monkeypatch):
+    monkeypatch.setenv("AI_VOICE_POLICY_MODE", "v2_streaming_duplex")
+
+    def _unexpected_llm(messages):
+        raise AssertionError(f"LLM should not be called for early plan: {messages!r}")
+
+    monkeypatch.setattr(ai_module, "_request_structured_model_decision", _unexpected_llm)
+
+    decision = voice_module._voice_decision(
+        language="ru",
+        customer=None,
+        interaction=SimpleNamespace(interaction_id="int_voice_early_schedule", status="new", queue_id="que_voice", subject="unknown"),
+        transcript_text="Мне надо узнать график работы",
+        transcript_window=[],
+        kb_results=[],
+        disclosure_required=False,
+        request_metadata={
+            "voice_v2_enabled": True,
+            "reply_phase": "early_plan",
+            "response_plan_id": "rsp_early_schedule",
+            "early_intent": "schedule",
+        },
+    )
+
+    assert decision["model"] == "voice_early_plan_domain"
+    assert decision["reply_text"]
+    assert decision["needs_handoff"] is False
+    assert decision["metadata"]["reply_phase"] == "early_plan"
+    assert decision["metadata"]["early_intent"] == "schedule"
+
+
 def test_voice_decision_hearing_check_keeps_active_topic_without_llm(monkeypatch):
     def _unexpected_llm(messages):
         raise AssertionError(f"LLM should not be called for hearing check: {messages!r}")

tests/test_ai_voice_asr_provider.py

@@ -399,6 +399,65 @@ def test_elevenlabs_realtime_streaming_provider_returns_partial_and_final():
     assert websocket.sent_payloads[1]["commit"] is True
 
 
+def test_elevenlabs_realtime_streaming_provider_finalizes_from_stable_partial_on_timeout():
+    class _FakeRealtimeWebSocket:
+        def __init__(self) -> None:
+            self.sent_payloads: list[dict] = []
+            self.incoming: queue.Queue[str | None] = queue.Queue()
+
+        def send(self, raw_payload: str) -> None:
+            payload = json.loads(raw_payload)
+            self.sent_payloads.append(payload)
+            if not payload.get("commit"):
+                partial = json.dumps(
+                    {
+                        "message_type": "partial_transcript",
+                        "text": "need schedule",
+                        "language_code": "ru",
+                    }
+                )
+                self.incoming.put(partial)
+                self.incoming.put(partial)
+
+        def recv(self) -> str:
+            item = self.incoming.get(timeout=1)
+            if item is None:
+                raise RuntimeError("closed")
+            return item
+
+        def close(self) -> None:
+            self.incoming.put(None)
+
+    websocket = _FakeRealtimeWebSocket()
+
+    provider = asr_module.ElevenLabsRealtimeStreamingASRProvider(
+        api_base="https://api.elevenlabs.example",
+        api_key="asr-key",
+        timeout_seconds=1,
+        finalize_timeout_seconds=0.25,
+        websocket_factory=lambda url, *, header, timeout: websocket,
+    )
+    stream_id = provider.open_stream("session-1", language_hint="ru")
+    provider.push_pcm(stream_id, b"\x01\x00" * 160)
+
+    partial = None
+    for _ in range(20):
+        partial = provider.poll_partial(stream_id)
+        if partial is not None and partial.is_stable:
+            break
+        time.sleep(0.02)
+
+    assert partial is not None
+    assert partial.is_stable is True
+
+    final = provider.finalize(stream_id)
+    provider.close_stream(stream_id)
+
+    assert final.text == "need schedule"
+    assert final.language == "ru"
+    assert websocket.sent_payloads[-1]["commit"] is True
+
+
 def test_yandex_grpc_streaming_provider_returns_partial_and_final():
     calls: list[dict] = []
 

tests/test_ai_voice_media_runtime.py

@@ -1847,6 +1847,263 @@ def test_media_runtime_streaming_partial_poll_does_not_block_turn_close_ack():
     assert "main" in events
 
 
+def test_media_runtime_voice_v2_uses_early_plan_before_slow_final_decision():
+    events: list[str | tuple[str, str]] = []
+
+    class _StreamingProvider(StreamingASRProvider):
+        name = "streaming-sidecar"
+        supports_streaming = True
+
+        def finalize(self, stream_id: str) -> ASRTranscription:
+            assert stream_id == "stream-1"
+            return ASRTranscription(text="work schedule", language="ru", confidence=0.9)
+
+        def close_stream(self, stream_id: str) -> None:
+            assert stream_id == "stream-1"
+
+    def _process_turn(session_id, transcript_text, language, barge_in, metadata):
+        del session_id, transcript_text, barge_in
+        if metadata and metadata.get("reply_phase") == "early_plan":
+            events.append("early_plan")
+            return VoiceAITurnDecisionOut(
+                language=language or "ru",
+                intent="schedule",
+                reply_text="early reply",
+                confidence=0.8,
+                needs_handoff=False,
+                handoff_reason=None,
+                case_action="keep_open",
+                kb_refs=[],
+                summary_text="early ready",
+                model="early",
+                latency_ms=1,
+                status="active",
+            )
+        events.append("final_start")
+        time.sleep(0.25)
+        events.append("final_done")
+        return VoiceAITurnDecisionOut(
+            language=language or "ru",
+            intent="schedule",
+            reply_text="final reply",
+            confidence=0.9,
+            needs_handoff=False,
+            handoff_reason=None,
+            case_action="keep_open",
+            kb_refs=[],
+            summary_text="final ready",
+            model="final",
+            latency_ms=1,
+            status="active",
+        )
+
+    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=_StreamingProvider(),
+        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=_process_turn,
+        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, is_greeting, style_hints
+        events.append((str(reply_phase), text))
+
+    runtime._speak_reply = _fake_speak_reply  # type: ignore[method-assign]
+
+    async def _scenario() -> None:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_early_plan",
+                call_id="call_media_runtime_early_plan",
+                interaction_id="int_media_runtime_early_plan",
+                ai_session_id="ais_media_runtime_early_plan",
+                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"
+        actor.partial_transcript = "work schedule"
+        actor.stable_partial_transcript = "work schedule"
+        actor.partial_intent = "schedule"
+        actor.stable_partial_intent = "schedule"
+        pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        await runtime._process_utterance(actor, pcm_frame * 40, False)
+        for _ in range(50):
+            if "final_done" in events:
+                break
+            await asyncio.sleep(0.01)
+
+    asyncio.run(_scenario())
+
+    assert ("main", "early reply") in events
+    assert ("main", "final reply") not in events
+    assert events.index(("main", "early reply")) < events.index("final_done")
+
+
+def test_media_runtime_voice_v2_uses_partial_as_final_when_streaming_finalize_fails():
+    captured: list[tuple[str, dict | None]] = []
+
+    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:
+            del audio_bytes
+            self.transcribe_count += 1
+            return ASRTranscription(text="batch text", language=language_hint or "ru", confidence=0.9)
+
+    class _FailingStreamingProvider(StreamingASRProvider):
+        name = "failing-streaming"
+        supports_streaming = True
+
+        def finalize(self, stream_id: str) -> ASRTranscription:
+            assert stream_id == "stream-1"
+            raise StreamingASRUnavailable("finalize timeout")
+
+        def close_stream(self, stream_id: str) -> None:
+            assert stream_id == "stream-1"
+
+    asr_provider = _CountingASRProvider()
+    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=_FailingStreamingProvider(),
+        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: (
+            captured.append((transcript_text, metadata))
+            or 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, 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_partial_final",
+                call_id="call_media_runtime_partial_final",
+                interaction_id="int_media_runtime_partial_final",
+                ai_session_id="ais_media_runtime_partial_final",
+                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"
+        actor.partial_transcript = "work schedule"
+        actor.stable_partial_transcript = "work schedule"
+        actor.partial_intent = "schedule"
+        actor.stable_partial_intent = "schedule"
+        pcm_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        await runtime._process_utterance(actor, pcm_frame * 40, False)
+
+    asyncio.run(_scenario())
+
+    assert asr_provider.transcribe_count == 0
+    final_turns = [item for item in captured if item[1] and item[1].get("reply_phase") == "final"]
+    assert final_turns
+    assert final_turns[0][0] == "work schedule"
+    assert final_turns[0][1]["transcript_source"] == "streaming_partial_after_finalize_failure"
+
+
 def test_media_runtime_merges_thinking_continuation_into_current_utterance():
     runtime = AudioSocketMediaRuntime(
         enabled=True,

tests/test_ai_voice_media_runtime_streaming_stable.py

@@ -36,7 +36,7 @@ class _StubTTSProvider(TTSProvider):
 
 def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr_stable():
     reply_starts: list[tuple[str, str | None, float]] = []
-    turns: list[str] = []
+    turns: list[tuple[str, str]] = []
 
     class _BatchASRProvider(ASRProvider):
         name = "batch-asr"
@@ -131,7 +131,7 @@ def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr
         plan_reply=lambda session_id, text, metadata, kind: None,
         record_latency=lambda session_id, metric, latency_ms: None,
         process_turn=lambda session_id, transcript_text, language, barge_in, metadata: (
-            turns.append(transcript_text)
+            turns.append((str((metadata or {}).get("reply_phase") or "final"), transcript_text))
             or VoiceAITurnDecisionOut(
                 language=language or "ru",
                 intent="schedule",
@@ -191,7 +191,7 @@ def test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr
 
     phases = [phase for _, phase, _ in reply_starts]
     finalize_started_at = next(ts for name, ts in streaming_provider.events if name == "finalize")
-    assert turns == ["need work schedule"]
+    assert [text for phase, text in turns if phase == "final"] == ["need work schedule"]
     assert "open" in [name for name, _ in streaming_provider.events]
     assert "close" in [name for name, _ in streaming_provider.events]
     assert phases[:2] == ["ack", "main"]

tests/test_voice_start_policy.py

@@ -48,6 +48,10 @@ def test_voice_start_name_outcome_rejects_garbage_name_inside_mixed_request():
     assert source == "none"
 
 
+def test_normalize_name_candidate_drops_trailing_asr_initial():
+    assert voice_policy._normalize_name_candidate("\u0415\u0440\u043d\u0443\u0440 \u0418") == "\u0415\u0440\u043d\u0443\u0440"
+
+
 def test_display_name_looks_trusted_rejects_phone_and_accepts_real_name():
     customer = SimpleNamespace(display_name="+77010000001")
     assert voice_policy._display_name_looks_trusted(customer, "+77010000001", None) is False