diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 1ceac84..090b880 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -537,6 +537,8 @@ class _ElevenLabsRealtimeStreamState: send_lock: threading.Lock = field(default_factory=threading.Lock) resample_state: object | None = None resample_lock: threading.Lock = field(default_factory=threading.Lock) + committed_segments: list[str] = field(default_factory=list) + finalize_requested: bool = False class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): @@ -674,23 +676,44 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): 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: + if update.is_final: + # A committed segment from the server. Under commit_strategy=vad + # the server can autonomously commit more than one segment across + # a single long utterance while the caller keeps talking, so + # accumulate every segment instead of treating the first one as + # the whole turn's answer — only our own finalize() call (which + # sets finalize_requested) marks the turn as actually complete. + segment_text = str(update.text or "").strip() + if segment_text: + state.committed_segments.append(segment_text) + accumulated_text = " ".join(state.committed_segments).strip() + accumulated = StreamingASRPartial( + text=accumulated_text, + language=update.language or state.language, + confidence=update.confidence, + is_final=state.finalize_requested, + is_stable=True, + ) + state.stable_partial = accumulated + state.latest_partial = accumulated + if state.finalize_requested: + state.final_transcription = ASRTranscription( + text=accumulated_text, + language=update.language or state.language, + confidence=update.confidence, + ) + state.final_event.set() + continue + if 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_final=False, is_stable=True, ) update = state.stable_partial state.latest_partial = update - if update.is_final: - state.final_transcription = ASRTranscription( - text=update.text, - language=update.language or state.language, - confidence=update.confidence, - ) - state.final_event.set() @staticmethod def _transcription_from_partial( @@ -755,18 +778,19 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): language, text[:160], ) - state.updates_queue.put( - StreamingASRPartial( - text=text, - language=language, - confidence=None, - is_final=True, - is_stable=True, - ) + # Route empty commits through the same queue/accumulation path as + # non-empty ones (rather than setting final_event directly here) so + # an autonomous empty vad-mode commit mid-utterance doesn't end the + # turn early — only our own finalize_requested gate does that. + state.updates_queue.put( + StreamingASRPartial( + text=text, + language=language, + confidence=None, + is_final=True, + is_stable=True, ) - else: - with state.lock: - state.final_event.set() + ) continue if message_type.startswith("scribe") and "error" in message_type.lower(): raise StreamingASRUnavailable(self._payload_error_message(payload)) @@ -868,6 +892,14 @@ class ElevenLabsRealtimeStreamingASRProvider(StreamingASRProvider): ) return transcription + # Drain anything already queued (e.g. an autonomous vad-mode commit that + # arrived mid-utterance) *before* marking finalize_requested, so those + # segments get accumulated as ongoing speech rather than mistaken for the + # response to the commit we're about to send. + self._drain_updates(state) + with state.lock: + state.finalize_requested = True + silence = b"\x00\x00" * int(self._sample_rate_hz * 0.1) try: self._send_json( diff --git a/tests/test_ai_voice_asr_provider.py b/tests/test_ai_voice_asr_provider.py index 60ba1dd..67fd208 100644 --- a/tests/test_ai_voice_asr_provider.py +++ b/tests/test_ai_voice_asr_provider.py @@ -458,6 +458,77 @@ def test_elevenlabs_realtime_streaming_provider_finalizes_from_stable_partial_on assert websocket.sent_payloads[-1]["commit"] is True +def test_elevenlabs_realtime_streaming_provider_accumulates_autonomous_vad_commits(): + # Regression test: under commit_strategy=vad the server can autonomously commit + # an early segment mid-utterance, before the client ever asks it to finalize. + # That must be accumulated, not mistaken for the whole turn's answer — otherwise + # everything the caller says after that early commit gets silently dropped. + class _FakeRealtimeWebSocket: + def __init__(self) -> None: + self.sent_payloads: list[dict] = [] + self.incoming: queue.Queue[str | None] = queue.Queue() + self.incoming.put( + json.dumps( + { + "message_type": "committed_transcript", + "text": "Здравствуйте", + "language_code": "ru", + } + ) + ) + + def send(self, raw_payload: str) -> None: + payload = json.loads(raw_payload) + self.sent_payloads.append(payload) + if payload.get("commit"): + self.incoming.put( + json.dumps( + { + "message_type": "committed_transcript", + "text": "нужна помощь с газом", + "language_code": "ru", + } + ) + ) + + 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=3, + finalize_timeout_seconds=1, + websocket_factory=lambda url, *, header, timeout: websocket, + ) + stream_id = provider.open_stream("session-1", language_hint="ru") + + early = None + for _ in range(20): + early = provider.poll_partial(stream_id) + if early is not None and early.text: + break + time.sleep(0.02) + assert early is not None + assert early.text == "Здравствуйте" + assert early.is_final is False + + provider.push_pcm(stream_id, b"\x01\x00" * 160) + final = provider.finalize(stream_id) + provider.close_stream(stream_id) + + assert final.text == "Здравствуйте нужна помощь с газом" + assert final.language == "ru" + + def test_yandex_grpc_streaming_provider_returns_partial_and_final(): calls: list[dict] = []