fix(voice): stabilize turn finalization after asr timeouts

services/ai_voice_runtime_service/media_runtime.py

@@ -99,10 +99,13 @@ class MediaActor:
     asr_stream_id: str | None = None
     asr_streaming_enabled: bool = False
     asr_poll_due_monotonic: float = 0.0
+    streaming_asr_backoff_until_monotonic: float = 0.0
     barge_in_speech_ms: int = 0
     barge_in_detected_monotonic: float = 0.0
     speech_started_monotonic: float = 0.0
     speech_ended_monotonic: float = 0.0
+    thinking_continuation_pcm: bytearray = field(default_factory=bytearray)
+    thinking_continuation_deadline_monotonic: float = 0.0
     last_ack_text: str | None = None
     last_ack_completed_monotonic: float = 0.0
     last_ack_variant: str | None = None
@@ -180,6 +183,9 @@ class AudioSocketMediaRuntime:
         self._partial_poll_interval_seconds = 0.20
         self._stable_partial_hold_seconds = 0.40
         self._barge_in_trigger_ms = 220
+        self._streaming_asr_reopen_backoff_seconds = 2.0
+        self._thinking_continuation_grace_seconds = 0.45
+        self._thinking_continuation_max_bytes = int(1800 * 16)
 
     @staticmethod
     def _normalize_intent_text(text: str) -> str:
@@ -377,6 +383,8 @@ class AudioSocketMediaRuntime:
         actor.barge_in_speech_ms = 0
         actor.barge_in_detected_monotonic = 0.0
         actor.current_reply_phase = None
+        actor.thinking_continuation_pcm.clear()
+        actor.thinking_continuation_deadline_monotonic = 0.0
         partial_task = actor.partial_asr_task
         actor.partial_asr_task = None
         if partial_task is not None and not partial_task.done():
@@ -396,6 +404,8 @@ class AudioSocketMediaRuntime:
             return
         if not self._is_streaming_v2_session(actor.registration):
             return
+        if time.monotonic() < actor.streaming_asr_backoff_until_monotonic:
+            return
         try:
             stream_id = await asyncio.to_thread(
                 self._streaming_asr_provider.open_stream,
@@ -415,6 +425,11 @@ class AudioSocketMediaRuntime:
         actor.asr_streaming_enabled = True
         actor.asr_poll_due_monotonic = 0.0
 
+    def _mark_streaming_asr_backoff(self, actor: MediaActor) -> None:
+        actor.streaming_asr_backoff_until_monotonic = (
+            time.monotonic() + self._streaming_asr_reopen_backoff_seconds
+        )
+
     async def _close_streaming_asr(self, actor: MediaActor) -> None:
         stream_id = actor.asr_stream_id
         actor.asr_stream_id = None
@@ -423,6 +438,37 @@ class AudioSocketMediaRuntime:
             return
         await asyncio.to_thread(self._streaming_asr_provider.close_stream, stream_id)
 
+    def _buffer_thinking_continuation(self, actor: MediaActor, pcm_frame: bytes, *, is_speech: bool, now: float) -> None:
+        if not is_speech and not actor.thinking_continuation_pcm:
+            return
+        remaining = self._thinking_continuation_max_bytes - len(actor.thinking_continuation_pcm)
+        if remaining <= 0:
+            actor.thinking_continuation_deadline_monotonic = now + self._thinking_continuation_grace_seconds
+            return
+        actor.thinking_continuation_pcm.extend(pcm_frame[:remaining])
+        actor.thinking_continuation_deadline_monotonic = now + self._thinking_continuation_grace_seconds
+
+    async def _extend_with_thinking_continuation(self, actor: MediaActor, pcm_bytes: bytes) -> bytes:
+        deadline = time.monotonic() + self._thinking_continuation_grace_seconds
+        observed_size = len(actor.thinking_continuation_pcm)
+        while time.monotonic() < deadline:
+            await asyncio.sleep(0.05)
+            if actor.closed:
+                break
+            if actor.thinking_continuation_pcm:
+                observed_size = len(actor.thinking_continuation_pcm)
+                deadline = max(deadline, actor.thinking_continuation_deadline_monotonic)
+                continue
+            if observed_size > 0:
+                break
+        if not actor.thinking_continuation_pcm:
+            actor.thinking_continuation_deadline_monotonic = 0.0
+            return pcm_bytes
+        merged = pcm_bytes + bytes(actor.thinking_continuation_pcm)
+        actor.thinking_continuation_pcm.clear()
+        actor.thinking_continuation_deadline_monotonic = 0.0
+        return merged
+
     def _update_stable_partial_intent(self, actor: MediaActor, intent: str) -> None:
         normalized = str(intent or "").strip() or "unknown"
         if actor.partial_intent == normalized:
@@ -920,6 +966,9 @@ class AudioSocketMediaRuntime:
             return
 
         is_speech = audioop.rms(pcm_frame, 2) >= 250
+        if actor.state == "thinking":
+            self._buffer_thinking_continuation(actor, pcm_frame, is_speech=is_speech, now=now)
+            return
         if actor.state == "speaking":
             if is_speech:
                 actor.barge_in_speech_ms += actor.frame_ms
@@ -940,7 +989,8 @@ class AudioSocketMediaRuntime:
         if vad_result.speech_started:
             self._reset_live_turn_state(actor)
             await self._ensure_streaming_asr(actor)
-        if actor.asr_streaming_enabled and actor.asr_stream_id and actor.input_active:
+        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:
             try:
                 await asyncio.to_thread(
                     self._streaming_asr_provider.push_pcm,
@@ -953,6 +1003,7 @@ class AudioSocketMediaRuntime:
                     actor.registration.voice_session_id,
                     str(exc)[:500],
                 )
+                self._mark_streaming_asr_backoff(actor)
                 await self._close_streaming_asr(actor)
             else:
                 try:
@@ -967,6 +1018,8 @@ class AudioSocketMediaRuntime:
                         actor.registration.voice_session_id,
                         str(exc)[:500],
                     )
+                    self._mark_streaming_asr_backoff(actor)
+                    await self._close_streaming_asr(actor)
         elif actor.registration.voice_v2_partial_asr:
             self._maybe_schedule_partial_asr(actor)
         if vad_result.utterance_pcm:
@@ -998,6 +1051,7 @@ class AudioSocketMediaRuntime:
 
     async def _process_utterance(self, actor: MediaActor, pcm_bytes: bytes, barge_in: bool) -> None:
         await self._set_actor_state(actor, "thinking")
+        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]}"
@@ -1360,7 +1414,7 @@ class AudioSocketMediaRuntime:
             handoff_reason,
         )
         actor.state = state
-        actor.input_active = state in {"listening", "thinking"}
+        actor.input_active = state == "listening"
         actor.playback_active = state == "speaking"
         await asyncio.to_thread(
             self._set_state,

tests/test_ai_voice_media_runtime.py

@@ -1413,6 +1413,176 @@ def test_media_runtime_voice_v2_emotive_ack_avoids_same_variant_back_to_back():
     assert first_text != second_text
 
 
+def test_media_runtime_streaming_timeout_enters_backoff_before_reopen():
+    class _FlakyStreamingProvider(StreamingASRProvider):
+        name = "flaky-streaming"
+        supports_streaming = True
+
+        def __init__(self) -> None:
+            self.open_calls = 0
+
+        def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
+            del session_id, language_hint
+            self.open_calls += 1
+            return "stream-1"
+
+        def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None:
+            del stream_id, pcm_8k_chunk
+            raise StreamingASRUnavailable("timed out")
+
+    media_uuid = str(uuid.uuid4())
+    registration = MediaRegistration(
+        voice_session_id="avs_media_runtime_stream_backoff",
+        call_id="call_media_runtime_stream_backoff",
+        interaction_id="int_media_runtime_stream_backoff",
+        ai_session_id="ais_media_runtime_stream_backoff",
+        language="ru",
+        media_uuid=media_uuid,
+        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",
+    )
+    streaming_provider = _FlakyStreamingProvider()
+    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="clarification",
+            reply_text="Подскажите подробнее, пожалуйста.",
+            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 _scenario() -> None:
+        actor = MediaActor(
+            registration=registration,
+            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,
+            state="listening",
+        )
+        runtime._reset_live_turn_state(actor)
+        await runtime._ensure_streaming_asr(actor)
+        assert actor.asr_streaming_enabled is True
+        speech_frame = (1000).to_bytes(2, "little", signed=True) * 160
+        await runtime._handle_pcm(actor, speech_frame)
+        await runtime._handle_pcm(actor, speech_frame)
+        assert actor.asr_streaming_enabled is False
+        assert actor.streaming_asr_backoff_until_monotonic > time.monotonic()
+        await runtime._ensure_streaming_asr(actor)
+        assert streaming_provider.open_calls == 1
+
+    asyncio.run(_scenario())
+
+
+def test_media_runtime_merges_thinking_continuation_into_current_utterance():
+    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=StreamingASRProvider(),
+        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="clarification",
+            reply_text="Подскажите подробнее, пожалуйста.",
+            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,
+    )
+    base_pcm = (1000).to_bytes(2, "little", signed=True) * 160
+    continuation_pcm = (900).to_bytes(2, "little", signed=True) * 160
+
+    async def _scenario() -> bytes:
+        actor = MediaActor(
+            registration=MediaRegistration(
+                voice_session_id="avs_media_runtime_thinking_merge",
+                call_id="call_media_runtime_thinking_merge",
+                interaction_id="int_media_runtime_thinking_merge",
+                ai_session_id="ais_media_runtime_thinking_merge",
+                language="ru",
+                media_uuid=str(uuid.uuid4()),
+            ),
+            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,
+            state="thinking",
+        )
+        await runtime._handle_pcm(actor, continuation_pcm)
+        await runtime._handle_pcm(actor, continuation_pcm)
+        merged = await runtime._extend_with_thinking_continuation(actor, base_pcm)
+        assert actor.thinking_continuation_pcm == bytearray()
+        return merged
+
+    merged = asyncio.run(_scenario())
+
+    assert len(merged) > len(base_pcm)
+    assert merged.endswith(continuation_pcm + continuation_pcm)
+
+
 def _legacy_test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr():
     registrations: dict[str, MediaRegistration] = {}
     reply_starts: list[tuple[str, str | None, float]] = []