import asyncio import time import uuid from services.ai_voice_runtime_service.audiosocket import EnergyVAD from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaActor, MediaRegistration from services.ai_voice_runtime_service.providers.asr import ( ASRProvider, ASRTranscription, StreamingASRPartial, StreamingASRProvider, StreamingASRUnavailable, ) from services.ai_voice_runtime_service.providers.tts import TTSSynthesis, TTSProvider from services.shared.models import VoiceAITurnDecisionOut class _StubTTSProvider(TTSProvider): name = "stub-tts" def synthesize( self, text: str, *, language: str | None = None, style_hints: dict[str, object] | None = None, ) -> TTSSynthesis: del language, style_hints assert text return TTSSynthesis( text=text, audio_bytes=(b"\x10\x00" * 960), sample_rate_hz=24000, ) 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[tuple[str, str]] = [] class _BatchASRProvider(ASRProvider): name = "batch-asr" def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: raise AssertionError("batch ASR should not be used when streaming sidecar is active") class _StreamingProvider(StreamingASRProvider): name = "streaming-sidecar" supports_streaming = True def __init__(self) -> None: self.events: list[tuple[str, float]] = [] self.chunk_count = 0 def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: assert session_id assert language_hint == "ru" self.events.append(("open", time.monotonic())) return "stream-1" def push_pcm(self, stream_id: str, pcm_bytes: bytes) -> None: assert stream_id == "stream-1" assert pcm_bytes self.chunk_count += 1 self.events.append(("push", time.monotonic())) def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: assert stream_id == "stream-1" if self.chunk_count < 2: return None self.events.append(("partial", time.monotonic())) return StreamingASRPartial( text="need work schedule", language="ru", confidence=0.84, is_final=False, is_stable=True, ) def finalize(self, stream_id: str) -> ASRTranscription: assert stream_id == "stream-1" self.events.append(("finalize", time.monotonic())) return ASRTranscription(text="need work schedule", language="ru", confidence=0.91) def close_stream(self, stream_id: str) -> None: assert stream_id == "stream-1" self.events.append(("close", time.monotonic())) streaming_provider = _StreamingProvider() registration = MediaRegistration( voice_session_id="avs_voice_v2_streaming", call_id="call_voice_v2_streaming", interaction_id="int_voice_v2_streaming", ai_session_id="ais_voice_v2_streaming", 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", voice_v2_prebaked_ack=False, ) 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=_BatchASRProvider(), 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_started=lambda session_id, text, is_greeting, reply_phase=None: None, mark_reply_delivered=lambda session_id, text, is_greeting, reply_phase=None: None, mark_reply_discarded=lambda session_id, text, reply_phase, reason: None, 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((str((metadata or {}).get("reply_phase") or "final"), transcript_text)) or VoiceAITurnDecisionOut( language=language or "ru", intent="schedule", reply_text="Please name the city.", confidence=0.92, 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, is_greeting, style_hints reply_starts.append((text, reply_phase, time.monotonic())) runtime._speak_reply = _fake_speak_reply # type: ignore[method-assign] 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, ) runtime._reset_live_turn_state(actor) await runtime._ensure_streaming_asr(actor) assert actor.asr_stream_id == "stream-1" pcm_chunk = (1000).to_bytes(2, "little", signed=True) * 160 await asyncio.to_thread(streaming_provider.push_pcm, actor.asr_stream_id, pcm_chunk) await asyncio.to_thread(streaming_provider.push_pcm, actor.asr_stream_id, pcm_chunk) actor.asr_poll_due_monotonic = 0.0 await runtime._poll_streaming_partial(actor) actor.speech_started_monotonic = time.monotonic() actor.speech_ended_monotonic = actor.speech_started_monotonic await runtime._process_utterance(actor, pcm_chunk * 14, False) asyncio.run(_scenario()) phases = [phase for _, phase, _ in reply_starts] finalize_started_at = next(ts for name, ts in streaming_provider.events if name == "finalize") 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"] assert reply_starts[0][2] <= finalize_started_at assert any(text == "Please name the city." and phase == "main" for text, phase, _ in reply_starts) def test_media_runtime_voice_v2_falls_back_when_streaming_sidecar_is_unavailable_stable(): turns: list[str] = [] class _BatchASRProvider(ASRProvider): name = "batch-asr" def transcribe(self, audio_bytes: bytes, *, language_hint: str | None = None) -> ASRTranscription: assert audio_bytes return ASRTranscription(text="need operator", language=language_hint or "ru", confidence=0.88) class _UnavailableStreamingProvider(StreamingASRProvider): name = "missing-sidecar" supports_streaming = True def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: del session_id, language_hint raise StreamingASRUnavailable("sidecar down") registration = MediaRegistration( voice_session_id="avs_voice_v2_fallback", call_id="call_voice_v2_fallback", interaction_id="int_voice_v2_fallback", ai_session_id="ais_voice_v2_fallback", 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", voice_v2_prebaked_ack=False, ) 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=_BatchASRProvider(), streaming_asr_provider=_UnavailableStreamingProvider(), 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_started=lambda session_id, text, is_greeting, reply_phase=None: None, mark_reply_delivered=lambda session_id, text, is_greeting, reply_phase=None: None, mark_reply_discarded=lambda session_id, text, reply_phase, reason: None, 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) or VoiceAITurnDecisionOut( language=language or "ru", intent="handoff", reply_text="Connecting you to an operator.", confidence=0.9, 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=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, ) runtime._reset_live_turn_state(actor) await runtime._ensure_streaming_asr(actor) assert actor.asr_streaming_enabled is False pcm_chunk = (1000).to_bytes(2, "little", signed=True) * 160 actor.speech_started_monotonic = time.monotonic() actor.speech_ended_monotonic = actor.speech_started_monotonic await runtime._process_utterance(actor, pcm_chunk * 4, False) asyncio.run(_scenario()) assert turns == ["need operator"] assert registration.voice_v2_duplex is True assert registration.voice_v2_partial_asr is True