diff --git a/netra/instrumentation/livekit/audio_capture.py b/netra/instrumentation/livekit/audio_capture.py index fe79f75..67634c7 100644 --- a/netra/instrumentation/livekit/audio_capture.py +++ b/netra/instrumentation/livekit/audio_capture.py @@ -259,6 +259,39 @@ def on_speaking_end(self, role: SpeakerRole, *, span_id: str = "") -> None: trace_id=ended_trace_id, ) + def on_tts_node_started(self, *, parent_span_id: str) -> None: + """Clear stale agent speech when a ``tts_node`` from a different turn starts. + + Both ``agent_speaking`` and ``tts_node`` are children of the same + ``agent_turn`` span. When a ``tts_node`` starts under a *different* + ``agent_turn`` than the stale ``agent_speaking``, the audio it + generates belongs to a new turn and must not be attributed to the + old span. Clearing ``_active_speech`` here makes subsequent frames + hit the existing ``is None`` guard in :meth:`on_frame`. + + A ``tts_node`` from the *same* turn (same parent) is a no-op, so + trailing frames from the current utterance are never disturbed. + + Args: + parent_span_id: Hex id of the ``tts_node``'s parent + (the ``agent_turn`` span), or ``""`` if unknown. + """ + if not self._agent_speech_ended: + return + active = self._active_speech[SpeakerRole.AGENT] + if active is None: + return + if not active.parent_span_id or not parent_span_id: + return + if parent_span_id != active.parent_span_id: + self._active_speech[SpeakerRole.AGENT] = None + logger.debug( + "netra.audio: tts_node parent %s differs from agent_speaking parent %s " + "— clearing stale agent speech", + parent_span_id, + active.parent_span_id, + ) + # -- frame callbacks ---------------------------------------------------- def on_frame(self, role: SpeakerRole, frame: "AudioFrame") -> None: @@ -338,6 +371,8 @@ def on_playback_finished(self, event: "PlaybackFinishedEvent") -> None: if not getattr(event, "interrupted", False): self._agent_playback_started_at = None self._agent_capture_started_at = None + if self._agent_speech_ended: + self._active_speech[SpeakerRole.AGENT] = None return span_id = self._interrupted_agent_span_id or self._last_agent_span_id if not span_id or self._sender is None: diff --git a/netra/instrumentation/livekit/audio_processor.py b/netra/instrumentation/livekit/audio_processor.py index 9e7b3cc..52bb590 100644 --- a/netra/instrumentation/livekit/audio_processor.py +++ b/netra/instrumentation/livekit/audio_processor.py @@ -5,6 +5,13 @@ it is what lets a frame captured milliseconds later be filed under the turn it belongs to. +It also watches ``tts_node`` spans so the coordinator can detect when a *new* +agent turn starts synthesizing speech. When the ``tts_node``'s parent +(``agent_turn``) differs from the stale ``agent_speaking``'s parent, the +coordinator clears the stale reference and subsequent frames are dropped rather +than misattributed — see +:meth:`~netra.instrumentation.livekit.audio_capture.SessionAudioCoordinator.on_tts_node_started`. + Registered once for the process, while coordinators are per call — hence the lookup by the span's trace id in :data:`~netra.instrumentation.livekit.audio_capture.audio_coordinators`. @@ -20,7 +27,7 @@ from opentelemetry.trace import SpanContext from netra.instrumentation.livekit.audio_capture import SessionAudioCoordinator, audio_coordinators -from netra.instrumentation.livekit.audio_types import SPEAKING_SPAN_ROLES, SpeakerRole +from netra.instrumentation.livekit.audio_types import SPEAKING_SPAN_ROLES, TTS_NODE_SPAN_NAME, SpeakerRole logger = logging.getLogger(__name__) @@ -44,26 +51,44 @@ class _SpeakingSpan(NamedTuple): parent_span_id: str +class _TtsNodeSpan(NamedTuple): + """A ``tts_node`` span resolved to its call's coordinator. + + Attributes: + coordinator: The coordinator capturing that call's audio. + parent_span_id: Hex id of the ``tts_node``'s parent (``agent_turn``), or ``""``. + """ + + coordinator: SessionAudioCoordinator + parent_span_id: str + + class AudioSpanProcessor(SpanProcessor): # type: ignore[misc] """Opens and closes an audio recording alongside each speaking span.""" def on_start(self, span: Span, parent_context: Optional[otel_context.Context] = None) -> None: """Start attributing this speaker's audio to the span that just opened. + Also detects ``tts_node`` spans so the coordinator can clear stale + ``agent_speaking`` attribution when a new turn's TTS begins. + Args: span: The span that was started. parent_context: The parent context (unused). """ speaking = _resolve_speaking_span(span) - if speaking is None: + if speaking is not None: + speaking.coordinator.on_speaking_start( + speaking.role, + trace_id=format(speaking.span_context.trace_id, _TRACE_ID_HEX_DIGITS), + span_id=format(speaking.span_context.span_id, _SPAN_ID_HEX_DIGITS), + parent_span_id=speaking.parent_span_id, + ) return - speaking.coordinator.on_speaking_start( - speaking.role, - trace_id=format(speaking.span_context.trace_id, _TRACE_ID_HEX_DIGITS), - span_id=format(speaking.span_context.span_id, _SPAN_ID_HEX_DIGITS), - parent_span_id=speaking.parent_span_id, - ) + tts = _resolve_tts_node_span(span) + if tts is not None: + tts.coordinator.on_tts_node_started(parent_span_id=tts.parent_span_id) def on_end(self, span: ReadableSpan) -> None: """Close the recording for the speaking span that just ended. @@ -131,11 +156,43 @@ def _resolve_speaking_span(span: Union[Span, ReadableSpan]) -> Optional[_Speakin return None +def _resolve_tts_node_span(span: Union[Span, ReadableSpan]) -> Optional[_TtsNodeSpan]: + """Identify a ``tts_node`` span and the call whose audio it may affect. + + Never raises, for the same reason as ``_resolve_speaking_span``. + + Args: + span: The span that started. + + Returns: + The resolved TTS node span, or ``None`` when *span* is not a + ``tts_node`` or its call is not capturing audio. + """ + try: + if (span.name or "") != TTS_NODE_SPAN_NAME: + return None + + span_context = span.get_span_context() + if span_context is None or not span_context.is_valid: + return None + + coordinator = audio_coordinators.get(span_context.trace_id) + if coordinator is None: + return None + return _TtsNodeSpan( + coordinator=coordinator, + parent_span_id=_parent_span_id_hex(span), + ) + except Exception: + logger.debug("netra.audio: could not resolve a tts_node span", exc_info=True) + return None + + def _parent_span_id_hex(span: Union[Span, ReadableSpan]) -> str: """Return the hex id of *span*'s parent, or ``""`` when there is none. Args: - span: The speaking span whose parent to read. + span: The span whose parent to read. Returns: A 16-digit lowercase hex span id, or an empty string for a root span diff --git a/netra/instrumentation/livekit/audio_types.py b/netra/instrumentation/livekit/audio_types.py index 7d85e59..fa7510d 100644 --- a/netra/instrumentation/livekit/audio_types.py +++ b/netra/instrumentation/livekit/audio_types.py @@ -43,6 +43,12 @@ class SpeakerRole(str, Enum): "agent_speaking": SpeakerRole.AGENT, } +# The TTS wrapper span whose parent is the ``agent_turn`` that owns the +# synthesis. Used by ``AudioSpanProcessor`` to detect when a new turn's TTS +# starts and clear stale ``agent_speaking`` attribution — see +# ``SessionAudioCoordinator.on_tts_node_started``. +TTS_NODE_SPAN_NAME = "tts_node" + # --------------------------------------------------------------------------- # PCM format diff --git a/tests/test_audio_integration.py b/tests/test_audio_integration.py index 9181520..64e793f 100644 --- a/tests/test_audio_integration.py +++ b/tests/test_audio_integration.py @@ -754,6 +754,7 @@ def test_agent_trailing_frames_are_attributed_after_span_end(self) -> None: span_id=AGENT_SPAN_ID, parent_span_id=PARENT_SPAN_ID, ) + coordinator.on_playback_started(created_at=time.time()) coordinator.on_speaking_end(SpeakerRole.AGENT, span_id=AGENT_SPAN_ID) coordinator.on_frame(SpeakerRole.AGENT, make_frame()) @@ -787,6 +788,93 @@ def test_close_is_idempotent(self) -> None: assert sender.mark_audio_end.call_count == 1 +class TestSessionAudioCoordinatorStaleSpeechClearing: + """Tests for tts_node-driven and playback_finished clearing of stale agent speech.""" + + TURN_A_PARENT = "aaaa000000000001" + TURN_B_PARENT = "bbbb000000000002" + + def test_tts_node_from_different_turn_clears_stale_agent_speech(self) -> None: + sender = MagicMock() + coordinator = SessionAudioCoordinator(sender=sender) + coordinator.on_speaking_start( + SpeakerRole.AGENT, + trace_id=TRACE_ID, + span_id=AGENT_SPAN_ID, + parent_span_id=self.TURN_A_PARENT, + ) + coordinator.on_speaking_end(SpeakerRole.AGENT, span_id=AGENT_SPAN_ID) + + coordinator.on_tts_node_started(parent_span_id=self.TURN_B_PARENT) + coordinator.on_frame(SpeakerRole.AGENT, make_frame()) + + sender.enqueue.assert_not_called() + + def test_tts_node_from_same_turn_does_not_clear(self) -> None: + sender = MagicMock() + coordinator = SessionAudioCoordinator(sender=sender) + coordinator.on_speaking_start( + SpeakerRole.AGENT, + trace_id=TRACE_ID, + span_id=AGENT_SPAN_ID, + parent_span_id=self.TURN_A_PARENT, + ) + coordinator.on_speaking_end(SpeakerRole.AGENT, span_id=AGENT_SPAN_ID) + + coordinator.on_tts_node_started(parent_span_id=self.TURN_A_PARENT) + coordinator.on_frame(SpeakerRole.AGENT, make_frame()) + + assert sender.enqueue.call_args.kwargs["span_id"] == AGENT_SPAN_ID + + def test_tts_node_while_agent_is_still_speaking_is_a_noop(self) -> None: + sender = MagicMock() + coordinator = SessionAudioCoordinator(sender=sender) + coordinator.on_speaking_start( + SpeakerRole.AGENT, + trace_id=TRACE_ID, + span_id=AGENT_SPAN_ID, + parent_span_id=self.TURN_A_PARENT, + ) + + coordinator.on_tts_node_started(parent_span_id=self.TURN_B_PARENT) + coordinator.on_frame(SpeakerRole.AGENT, make_frame()) + + assert sender.enqueue.call_args.kwargs["span_id"] == AGENT_SPAN_ID + + def test_playback_finished_after_speech_ended_clears_stale_agent_speech(self) -> None: + sender = MagicMock() + coordinator = SessionAudioCoordinator(sender=sender) + coordinator.on_speaking_start( + SpeakerRole.AGENT, + trace_id=TRACE_ID, + span_id=AGENT_SPAN_ID, + parent_span_id=self.TURN_A_PARENT, + ) + coordinator.on_playback_started(created_at=time.time()) + coordinator.on_speaking_end(SpeakerRole.AGENT, span_id=AGENT_SPAN_ID) + + coordinator.on_playback_finished(MagicMock(interrupted=False, playback_position=2.0)) + coordinator.on_frame(SpeakerRole.AGENT, make_frame()) + + sender.enqueue.assert_not_called() + + def test_playback_finished_while_still_speaking_does_not_clear(self) -> None: + sender = MagicMock() + coordinator = SessionAudioCoordinator(sender=sender) + coordinator.on_speaking_start( + SpeakerRole.AGENT, + trace_id=TRACE_ID, + span_id=AGENT_SPAN_ID, + parent_span_id=self.TURN_A_PARENT, + ) + coordinator.on_playback_started(created_at=time.time()) + + coordinator.on_playback_finished(MagicMock(interrupted=False, playback_position=2.0)) + coordinator.on_frame(SpeakerRole.AGENT, make_frame()) + + assert sender.enqueue.call_args.kwargs["span_id"] == AGENT_SPAN_ID + + class TestSessionAudioCoordinatorInterrupts: @staticmethod def _interrupted_coordinator(sender: MagicMock) -> SessionAudioCoordinator: @@ -1008,6 +1096,37 @@ def test_a_span_that_is_not_speech_is_ignored(self) -> None: span.get_span_context.assert_not_called() + def test_a_tts_node_span_triggers_on_tts_node_started(self) -> None: + trace_id = 0xAAAABBBBCCCCDDDD + sender = MagicMock() + coordinator = SessionAudioCoordinator(sender=sender) + coordinator.on_speaking_start( + SpeakerRole.AGENT, + trace_id=format(trace_id, "032x"), + span_id=AGENT_SPAN_ID, + parent_span_id=format(0x1111000000000001, "016x"), + ) + coordinator.on_speaking_end(SpeakerRole.AGENT, span_id=AGENT_SPAN_ID) + audio_coordinators.register(trace_id, coordinator) + processor = AudioSpanProcessor() + + tts_span = make_span( + "tts_node", + trace_id=trace_id, + span_id=0xBBBB, + parent_span_id=0x2222000000000002, + ) + processor.on_start(tts_span) + + coordinator.on_frame(SpeakerRole.AGENT, make_frame()) + sender.enqueue.assert_not_called() + + def test_a_tts_node_span_without_a_coordinator_is_ignored(self) -> None: + processor = AudioSpanProcessor() + tts_span = make_span("tts_node", trace_id=0x9999, span_id=0xCCCC) + + processor.on_start(tts_span) + # --------------------------------------------------------------------------- # Session wiring