fix(voice): export uncommitted user_turn spans on close

livekit-agents/livekit/agents/voice/audio_recognition.py

@@ -773,36 +773,42 @@ def _push_audio(
 
     async def _aclose(self) -> None:
         self._closing.set()
-        if self._commit_user_turn_atask is not None:
-            await aio.cancel_and_wait(self._commit_user_turn_atask)
+        try:
+            if self._commit_user_turn_atask is not None:
+                await aio.cancel_and_wait(self._commit_user_turn_atask)
 
-        if self._stt_pipeline is not None:
-            await self._stt_pipeline.aclose()
-            self._stt_pipeline = None
+            if self._stt_pipeline is not None:
+                await self._stt_pipeline.aclose()
+                self._stt_pipeline = None
 
-        await aio.cancel_and_wait(*self._tasks)
+            await aio.cancel_and_wait(*self._tasks)
 
-        if self._stt_consumer_atask is not None:
-            await aio.cancel_and_wait(self._stt_consumer_atask)
+            if self._stt_consumer_atask is not None:
+                await aio.cancel_and_wait(self._stt_consumer_atask)
 
-        if self._vad_atask is not None:
-            await aio.cancel_and_wait(self._vad_atask)
+            if self._vad_atask is not None:
+                await aio.cancel_and_wait(self._vad_atask)
 
-        if self._interruption_atask is not None:
-            await aio.cancel_and_wait(self._interruption_atask)
+            if self._interruption_atask is not None:
+                await aio.cancel_and_wait(self._interruption_atask)
 
-        if self._end_of_turn_task is not None:
-            await aio.cancel_and_wait(self._end_of_turn_task)
+            if self._end_of_turn_task is not None:
+                await aio.cancel_and_wait(self._end_of_turn_task)
 
-        if self._turn_detector_stream is not None:
-            await self._turn_detector_stream.aclose()
-            self._turn_detector_stream = None
-        self._turn_detector_prediction_fut = None
+            if self._turn_detector_stream is not None:
+                await self._turn_detector_stream.aclose()
+                self._turn_detector_stream = None
+            self._turn_detector_prediction_fut = None
 
-        if self._backchannel_boundary_timer is not None:
-            self._backchannel_boundary_timer.cancel()
-            self._backchannel_boundary_timer = None
-            self._backchannel_boundary_callback = None
+            if self._backchannel_boundary_timer is not None:
+                self._backchannel_boundary_timer.cancel()
+                self._backchannel_boundary_timer = None
+                self._backchannel_boundary_callback = None
+        finally:
+            # a speech segment may never produce a transcript or a committed turn.
+            # the eou detection ends the span on the normal path, it is cancelled above,
+            # so end it here once recognition stopped to keep the span exported
+            self._end_user_turn_span()
 
     def _update_stt(
         self,
@@ -999,9 +1005,7 @@ def _clear_user_turn(self) -> None:
         self._turn_tracker = _UserTurnTracker()
 
         # end any in-progress user_turn span so the next speech starts a fresh one
-        if self._user_turn_span is not None and self._user_turn_span.is_recording():
-            self._user_turn_span.end()
-        self._user_turn_span = None
+        self._end_user_turn_span()
         self._stt_request_ids = []
 
         # reset stt to clear the buffer from previous user turn
@@ -1866,3 +1870,9 @@ def _ensure_user_turn_span(self, start_time: float | None = None) -> trace.Span:
             )
 
         return self._user_turn_span
+
+    def _end_user_turn_span(self) -> None:
+        if self._user_turn_span is not None and self._user_turn_span.is_recording():
+            self._user_turn_span.end()
+        self._user_turn_span = None
+        self._user_turn_start = None

tests/test_audio_recognition_aclose.py

@@ -9,7 +9,7 @@
 """
 
 import asyncio
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -43,6 +43,7 @@ def _create_audio_recognition(self) -> AudioRecognition:
         audio_recognition._audio_input_atask = None
         audio_recognition._backchannel_boundary_timer = None
         audio_recognition._AudioRecognition__stt_context = None
+        audio_recognition._user_turn_span = None
 
         return audio_recognition
 
@@ -150,3 +151,51 @@ async def long_running_task():
         # Both tasks are now done (not orphaned)
         assert commit_task.done()
         assert end_of_turn_task.done()
+
+    @pytest.mark.asyncio
+    async def test_aclose_ends_in_progress_user_turn_span(self):
+        """A turn that never completed still gets its span ended (and exported) on close."""
+        audio_recognition = self._create_audio_recognition()
+
+        span = MagicMock()
+        span.is_recording.return_value = True
+        audio_recognition._user_turn_span = span
+
+        await audio_recognition._aclose()
+
+        span.end.assert_called_once()
+        assert audio_recognition._user_turn_span is None
+
+    @pytest.mark.asyncio
+    async def test_aclose_ends_user_turn_span_when_teardown_raises(self):
+        """Teardown blowing up must not take the span with it — closing on error is the
+        case that leaks spans in the first place."""
+        audio_recognition = self._create_audio_recognition()
+
+        span = MagicMock()
+        span.is_recording.return_value = True
+        audio_recognition._user_turn_span = span
+
+        stt_pipeline = MagicMock()
+        stt_pipeline.aclose = AsyncMock(side_effect=RuntimeError("vendor stream teardown failed"))
+        audio_recognition._stt_pipeline = stt_pipeline
+
+        with pytest.raises(RuntimeError, match="vendor stream teardown failed"):
+            await audio_recognition._aclose()
+
+        span.end.assert_called_once()
+        assert audio_recognition._user_turn_span is None
+
+    @pytest.mark.asyncio
+    async def test_aclose_does_not_reend_completed_user_turn_span(self):
+        """The eou detection already ended the span, aclose must not end it twice."""
+        audio_recognition = self._create_audio_recognition()
+
+        span = MagicMock()
+        span.is_recording.return_value = False
+        audio_recognition._user_turn_span = span
+
+        await audio_recognition._aclose()
+
+        span.end.assert_not_called()
+        assert audio_recognition._user_turn_span is None

tests/test_audio_recognition_turn_detection.py

@@ -345,11 +345,13 @@ async def test_clear_user_turn_allows_next_turn_to_emit(self) -> None:
         ar._audio_preflight_transcript = ""
         ar._stt_request_ids = []
         ar._turn_detector_stream.flush = MagicMock()
-        ar.update_stt = MagicMock()  # type: ignore[method-assign]
+        ar._update_stt = MagicMock()  # type: ignore[method-assign]
+        ar._user_turn_start = 123.0
 
-        ar.clear_user_turn()
+        ar._clear_user_turn()
 
         assert ar._last_emitted_prediction is None
+        assert ar._user_turn_start is None
 
 
 class TestBackchannelOpportunityEmit: