[NET-1409] fix: Forcefully end the user turn span and also buffer the user audio transcripts

netra/instrumentation/libraries/livekit/trace_processor.py

@@ -259,6 +259,17 @@ def _numeric(value: Any) -> Optional[float]:
 _user_turn_spans: "weakref.WeakValueDictionary[int, Span]" = weakref.WeakValueDictionary()
 _user_turn_lock = threading.Lock()
 
+# Streaming transcript for the open ``user_turn`` in each call, built from LiveKit's
+# ``user_input_transcribed`` events. LiveKit only writes ``lk.user_transcript`` onto
+# the span when end-of-utterance *commits* the turn — so a mid-speech hangup leaves
+# the span with no transcript even though the LiveKit UI already showed one. This
+# buffer is what lets ``end_orphaned_user_turn_span`` stamp what was heard.
+#
+# Mirrors LiveKit's own accumulation in ``AudioRecognition``: finals append, interims
+# overlay. Cleared when a turn starts or ends so one turn cannot inherit another's
+# words.
+_user_transcripts: Dict[int, Tuple[str, str]] = {}
+
 
 def _register_user_turn_span(span: Span, parent_context: Optional[otel_context.Context]) -> None:
     """Make *span* the turn STT usage is attributed to in its call.
@@ -277,6 +288,8 @@ def _register_user_turn_span(span: Span, parent_context: Optional[otel_context.C
     setattr(span, _CALL_ID_FIELD, call_id)
     with _user_turn_lock:
         _user_turn_spans[call_id] = span
+        # Fresh turn — drop any leftover from a previous one in this call.
+        _user_transcripts[call_id] = ("", "")
 
 
 def _deregister_user_turn_span(span: ReadableSpan) -> None:
@@ -300,6 +313,99 @@ def _deregister_user_turn_span(span: ReadableSpan) -> None:
         registered = _user_turn_spans.get(call_id)
         if registered is not None and registered.get_span_context().span_id == context.span_id:
             del _user_turn_spans[call_id]
+            _user_transcripts.pop(call_id, None)
+
+
+def record_user_transcript(call_id: int, transcript: str, *, is_final: bool) -> None:
+    """Accumulate one LiveKit ``user_input_transcribed`` sample for the open turn.
+
+    The LiveKit UI shows these as they stream; the OTel ``user_turn`` span only
+    receives ``lk.user_transcript`` when the turn is committed by EOU. Keeping the
+    running text here is what fills that gap when the session closes mid-speech.
+
+    Args:
+        call_id: The call the transcript belongs to.
+        transcript: The text LiveKit reported on this event — a final fragment or
+            the current interim overlay.
+        is_final: Whether this is a final (append) or interim (replace overlay).
+    """
+    if not transcript:
+        return
+    with _user_turn_lock:
+        # Only accumulate while a turn is open for this call — otherwise a late
+        # event after deregister would seed the *next* turn's buffer.
+        if call_id not in _user_turn_spans:
+            return
+        finals, _interim = _user_transcripts.get(call_id, ("", ""))
+        if is_final:
+            finals = f"{finals} {transcript}".strip()
+            _user_transcripts[call_id] = (finals, "")
+        else:
+            _user_transcripts[call_id] = (finals, transcript)
+
+
+def _transcript_for(call_id: int) -> Optional[str]:
+    """Return the accumulated transcript for *call_id*, or ``None`` if empty.
+
+    Args:
+        call_id: The call to read.
+
+    Returns:
+        Finals plus any current interim overlay, or ``None``.
+    """
+    finals, interim = _user_transcripts.get(call_id, ("", ""))
+    if interim:
+        combined = f"{finals} {interim}".strip()
+        return combined or None
+    return finals or None
+
+
+def end_orphaned_user_turn_span(call_id: int) -> None:
+    """End a still-recording ``user_turn`` span that LiveKit left open.
+
+    When the user hangs up mid-speech, LiveKit's ``_aclose_impl`` may end
+    ``user_speaking`` without ending its parent ``user_turn``, because the turn
+    never completed naturally — no end-of-utterance, no end-of-speech.  An unended
+    span is never queued by ``BatchSpanProcessor`` and therefore never exported,
+    leaving ``user_speaking`` orphaned in the backend: it carries a ``parent_id``
+    that resolves to nothing.
+
+    LiveKit also never writes ``lk.user_transcript`` in that path — that attribute
+    is set only when EOU commits the turn — even though streaming transcripts were
+    already emitted on ``user_input_transcribed``. Any text buffered by
+    :func:`record_user_transcript` is stamped here before the span ends.
+
+    Called from ``wrap_aclose`` after LiveKit's ``_aclose_impl`` has returned, so
+    LiveKit has had its full chance to end the span itself.  Only acts on spans
+    that are still recording — a span LiveKit did end is left untouched.
+
+    Args:
+        call_id: The call the turn belongs to — the ``livekit-call`` span's own
+            span id, as ``call_id_of_session`` reports it.
+    """
+    with _user_turn_lock:
+        span = _user_turn_spans.get(call_id)
+        transcript = _transcript_for(call_id)
+    if span is None:
+        return
+    if not span.is_recording():
+        return
+    logger.debug("netra.livekit: ending orphaned user_turn span for call %x", call_id)
+    try:
+        span.set_attribute("netra.turn.interrupted_by_session_close", True)
+    except Exception:
+        logger.debug("netra.livekit: could not mark orphaned user_turn span", exc_info=True)
+    if transcript:
+        try:
+            # Goes through the set_attribute wrapper so CONVERSATION_MAP fills
+            # gen_ai.completion the same way a normal committed turn does.
+            span.set_attribute("lk.user_transcript", transcript)
+        except Exception:
+            logger.debug("netra.livekit: could not stamp orphaned user_turn transcript", exc_info=True)
+    try:
+        span.end()
+    except Exception:
+        logger.debug("netra.livekit: could not end orphaned user_turn span", exc_info=True)
 
 
 def record_stt_usage(call_id: int, metrics_payload: Any) -> None:

netra/instrumentation/libraries/livekit/utils.py

@@ -386,8 +386,10 @@ class AudioPricingAttributes(NamedTuple):
     # tts_request: the text handed to the TTS provider. The words are the agent's.
     "lk.input_text": ConversationTarget(ConversationSide.PROMPT, "assistant"),
     # user_turn: the STT transcript — the output of the transcription. The words are
-    # the caller's.
+    # the caller's. ``lk.pii.user_transcript`` is the same key after LiveKit's PII
+    # rename; both are accepted so older and newer livekit-agents agree.
     "lk.user_transcript": ConversationTarget(ConversationSide.COMPLETION, "user"),
+    "lk.pii.user_transcript": ConversationTarget(ConversationSide.COMPLETION, "user"),
 }
 
 # span name -> ``netra.span.type``.

netra/instrumentation/libraries/livekit/wrappers.py

@@ -42,7 +42,11 @@
     end_call_span_of_session,
     start_call_span,
 )
-from netra.instrumentation.libraries.livekit.trace_processor import record_stt_usage
+from netra.instrumentation.libraries.livekit.trace_processor import (
+    end_orphaned_user_turn_span,
+    record_stt_usage,
+    record_user_transcript,
+)
 from netra.instrumentation.libraries.livekit.utils import NETRA_CLOSE_REASON, STT_METRICS_TYPE
 from netra.session_manager import SessionManager
 
@@ -60,12 +64,20 @@
 # LiveKit's ``AgentSession`` event carrying every plugin's metrics.
 _METRICS_EVENT = "metrics_collected"
 
+# LiveKit's ``AgentSession`` event carrying streaming user transcripts (interim and
+# final). Fired as the user speaks — the same feed the LiveKit UI renders — and
+# the only place the text reaches us before EOU commits it onto the span.
+_TRANSCRIPT_EVENT = "user_input_transcribed"
+
 # Instance attribute marking a session whose metrics this package already listens
 # to. One subscription per session, however many times ``start()`` is called on it:
 # a second listener would record every STT sample twice and double the audio
 # duration and token counts the call is billed on.
 _METRICS_SUBSCRIBED_FIELD = "_netra_livekit_metrics_subscribed"
 
+# Same idea for the transcript listener.
+_TRANSCRIPT_SUBSCRIBED_FIELD = "_netra_livekit_transcript_subscribed"
+
 
 # ---------------------------------------------------------------------------
 # Session-id resolution
@@ -315,14 +327,75 @@ def _listen_for_metrics(instance: "AgentSession", handler: Callable[[Any], None]
         instance: The ``AgentSession`` to subscribe to.
         handler: The callback to register.
     """
+    _listen_for_event(instance, _METRICS_EVENT, handler)
+
+
+def _subscribe_user_transcript(instance: "AgentSession") -> None:
+    """Buffer streaming user transcripts for the open ``user_turn`` span.
+
+    LiveKit's UI shows these as they arrive; the OTel span only gets
+    ``lk.user_transcript`` when EOU commits the turn. Buffering them here is what
+    lets a mid-speech hangup still export the words that were already transcribed.
+
+    Subscribed at most once per session, for the same retry reasons as
+    :func:`_subscribe_stt_usage`.
+
+    Args:
+        instance: The ``AgentSession`` that is starting.
+    """
+    if getattr(instance, _TRANSCRIPT_SUBSCRIBED_FIELD, False):
+        return
+
+    def on_transcript(event: Any) -> None:
+        """Record one ``user_input_transcribed`` event.
+
+        Args:
+            event: LiveKit's ``UserInputTranscribedEvent``.
+        """
+        try:
+            call_id = call_id_of_session(instance)
+            if call_id is None:
+                return
+            transcript = getattr(event, "transcript", None)
+            if not isinstance(transcript, str) or not transcript:
+                return
+            is_final = bool(getattr(event, "is_final", False))
+            record_user_transcript(call_id, transcript, is_final=is_final)
+        except Exception:
+            logger.debug("netra.livekit: user transcript could not be recorded", exc_info=True)
+
+    _listen_for_event(instance, _TRANSCRIPT_EVENT, on_transcript)
+
+    try:
+        setattr(instance, _TRANSCRIPT_SUBSCRIBED_FIELD, True)
+    except Exception:
+        logger.debug("netra.livekit: could not mark the session as transcript-subscribed", exc_info=True)
+
+
+def _listen_for_event(instance: "AgentSession", event_name: str, handler: Callable[[Any], None]) -> None:
+    """Subscribe *handler* to a session event via ``rtc.EventEmitter``.
+
+    Prefers the base ``EventEmitter.on`` so session-level deprecation warnings on
+    specific events (e.g. ``metrics_collected``) do not appear in the user's logs
+    for listeners the user did not add.
+
+    Args:
+        instance: The ``AgentSession`` to subscribe to.
+        event_name: The event name.
+        handler: The callback to register.
+    """
     try:
         from livekit import rtc
     except ImportError:
-        logger.debug("netra.livekit: livekit.rtc is unavailable; subscribing through the session", exc_info=True)
-        instance.on(_METRICS_EVENT, handler)
+        logger.debug(
+            "netra.livekit: livekit.rtc is unavailable; subscribing to %s through the session",
+            event_name,
+            exc_info=True,
+        )
+        instance.on(event_name, handler)
         return
 
-    rtc.EventEmitter.on(instance, _METRICS_EVENT, handler)
+    rtc.EventEmitter.on(instance, event_name, handler)
 
 
 # ---------------------------------------------------------------------------
@@ -598,6 +671,15 @@ async def wrap_start(
                     exc_info=True,
                 )
 
+            try:
+                _subscribe_user_transcript(instance)
+            except Exception:
+                logger.warning(
+                    "netra.livekit: could not subscribe to user transcripts; interrupted user_turn "
+                    "spans may export without the words already shown in the LiveKit UI",
+                    exc_info=True,
+                )
+
             result = await wrapped(*args, **kwargs)
     except BaseException:
         # A session that never started will never be closed, so neither end path
@@ -672,6 +754,18 @@ async def wrap_aclose(
     try:
         return await wrapped(*args, **kwargs)
     finally:
+        # LiveKit's ``_aclose_impl`` may end ``user_speaking`` without ending its
+        # parent ``user_turn`` when the user hangs up mid-speech (the turn never
+        # completed naturally).  An unended span is never exported, so
+        # ``user_speaking`` arrives at the backend with a ``parent_id`` that
+        # resolves to nothing.  Ending the orphan here — after ``_aclose_impl``
+        # has had its chance — ensures both spans reach the exporter.
+        try:
+            call_id = call_id_of_session(instance)
+            if call_id is not None:
+                end_orphaned_user_turn_span(call_id)
+        except Exception:
+            logger.debug("netra.livekit: could not end orphaned user_turn span", exc_info=True)
         try:
             await _after_close(instance)
         except Exception: