feat: enhance ElevenLabs ASR provider to accumulate autonomous VAD commits

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(

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] = []