[NET-1409] fix: Use parent span id to ignore unwanted tts audio

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:

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

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

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