From 5af47991dcc32d894ac32b11e752325278b913fc Mon Sep 17 00:00:00 2001 From: Yera All Date: Thu, 16 Apr 2026 02:30:39 +0500 Subject: [PATCH] fix(voice): stabilize turn finalization after asr timeouts --- .../ai_voice_runtime_service/media_runtime.py | 58 +++++- tests/test_ai_voice_media_runtime.py | 170 ++++++++++++++++++ 2 files changed, 226 insertions(+), 2 deletions(-) diff --git a/services/ai_voice_runtime_service/media_runtime.py b/services/ai_voice_runtime_service/media_runtime.py index 395e1e8..9bf79b5 100644 --- a/services/ai_voice_runtime_service/media_runtime.py +++ b/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, diff --git a/tests/test_ai_voice_media_runtime.py b/tests/test_ai_voice_media_runtime.py index bae2715..0c8029b 100644 --- a/tests/test_ai_voice_media_runtime.py +++ b/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]] = []