diff --git a/netra/instrumentation/libraries/livekit/trace_processor.py b/netra/instrumentation/libraries/livekit/trace_processor.py index 1227ed3..9c9d26a 100644 --- a/netra/instrumentation/libraries/livekit/trace_processor.py +++ b/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: diff --git a/netra/instrumentation/libraries/livekit/utils.py b/netra/instrumentation/libraries/livekit/utils.py index e8ce974..a27a540 100644 --- a/netra/instrumentation/libraries/livekit/utils.py +++ b/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``. diff --git a/netra/instrumentation/libraries/livekit/wrappers.py b/netra/instrumentation/libraries/livekit/wrappers.py index 1cae6bb..468b3bf 100644 --- a/netra/instrumentation/libraries/livekit/wrappers.py +++ b/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: