From 0ccdd808e6b5454ee2099b238c6513e0c8d838c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 May 2026 09:37:49 -0700 Subject: [PATCH 1/7] Fix traced_tts so metrics.ttfb reflects the real TTFB Previously @traced_tts scoped the span to the lifetime of run_tts(). For streaming TTS services run_tts() returns as soon as the synthesis request is sent, long before audio chunks arrive, so: - The span duration measured the WebSocket-send time, not synthesis time. - The first synthesis recorded the WS-send duration as metrics.ttfb (via the in-progress fallback in FrameProcessorMetrics.ttfb). - Subsequent syntheses recorded the previous call's TTFB on the current span (off-by-one). The decorator now uses a __set_name__ descriptor to wrap the owning class's setup() at class definition time. setup() installs per-instance patches on create_audio_context, append_to_audio_context, remove_audio_context, on_audio_context_completed, and reset_active_audio_context. These patches own the span lifetime: - create_audio_context: open span, set baseline attributes. - append_to_audio_context: record metrics.ttfb on the first TTSAudioRawFrame (when stop_ttfb_metrics has produced a real value), end span on appended TTSStoppedFrame. - on_audio_context_completed: end span on natural completion (handles services that auto-push TTSStoppedFrame via push_frame, bypassing append_to_audio_context). - remove_audio_context: safety net for explicit removal paths. - reset_active_audio_context: interruption hook (always reached from _handle_interruption); marks the span tts.interrupted=true only when nothing else has closed it. The run_tts wrapper now only attaches per-call attributes (text, metrics.character_count) to the already-open span. No changes required in tts_service.py or in any of the per-service files. --- .../utils/tracing/service_decorators.py | 280 +++++++++++++----- 1 file changed, 203 insertions(+), 77 deletions(-) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index f4de1dd85..94770db52 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -11,7 +11,6 @@ rich information about service execution including configuration, parameters, and performance metrics. """ -import contextlib import functools import inspect import json @@ -24,6 +23,8 @@ if TYPE_CHECKING: from opentelemetry import context as context_api from opentelemetry import trace +from pipecat.frames.frames import MetricsFrame, TTSStoppedFrame +from pipecat.metrics.metrics import TTFBMetricsData from pipecat.processors.aggregators.llm_context import NOT_GIVEN from pipecat.utils.tracing.service_attributes import ( add_gemini_live_span_attributes, @@ -175,6 +176,13 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call - Character count and text content - Performance metrics like TTFB + The span is scoped to the full synthesis operation, from + ``create_audio_context`` until ``TTSStoppedFrame`` (or + ``remove_audio_context`` as a safety net), so TTFB and any other + runtime-computed metrics land on the correct span even when audio + chunks are delivered after ``run_tts`` returns (e.g. WebSocket + streaming TTS services). + Works with both async functions and generators. Args: @@ -190,99 +198,217 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call def decorator(f): is_async_generator = inspect.isasyncgenfunction(f) - @contextlib.asynccontextmanager - async def tracing_context(self, text): - """Async context manager for TTS tracing. + def end_tts_span(service, context_id, *, interrupted=False): + """End the TTS span for ``context_id`` if still open. Idempotent.""" + entry = service._tts_spans.pop(context_id, None) + if not entry: + return + try: + span = entry["span"] + if interrupted: + span.set_attribute("tts.interrupted", True) + span.end() + except Exception as e: + logging.warning(f"Error closing TTS span: {e}") - Args: - self: The TTS service instance. - text: The text being synthesized. + def install_audio_context_patches(service): + """Install per-instance wrappers on the audio-context methods. - Yields: - The active span for the TTS operation. + The wrappers own the lifetime of the TTS span: + + - ``create_audio_context``: opens the span and records + baseline attributes. + - ``append_to_audio_context``: ends the span on + ``TTSStoppedFrame``. + - ``push_frame``: records ``metrics.ttfb`` from the + canonical ``TTFBMetricsData`` payload of any + ``MetricsFrame`` pushed by ``stop_ttfb_metrics``. Reading + the value from the metrics event (instead of polling + ``_metrics.ttfb`` when the first audio is queued) avoids + the ``ttfb`` property's in-progress fallback, which would + otherwise report an under-estimate whenever a context's + audio waits behind earlier queued audio before + ``_handle_audio_context`` actually stops the TTFB + measurement. + - ``remove_audio_context``: ends any still-open span as a + safety net for error and cancellation paths. + - ``on_audio_context_completed``: ends the span on natural + completion. Needed because services that rely on the + base class to auto-push ``TTSStoppedFrame`` (via + ``push_frame`` in ``_handle_audio_context``) bypass the + ``append_to_audio_context`` hook entirely. + - ``reset_active_audio_context``: ends the currently + playing context's span if still open. Always called from + ``_handle_interruption``, so this is the interruption + hook. + + The patches check ``_tracing_enabled`` at invocation time, + so they are safe to install regardless of whether tracing + is enabled. """ - # Check if tracing is enabled for this service instance - if not getattr(self, "_tracing_enabled", False): - yield None + if getattr(service, "__tts_tracing_patches_installed__", False): + return + service.__tts_tracing_patches_installed__ = True + service._tts_spans = {} + + orig_create = service.create_audio_context + orig_append = service.append_to_audio_context + orig_remove = service.remove_audio_context + orig_completed = service.on_audio_context_completed + orig_reset_active = service.reset_active_audio_context + orig_push_frame = service.push_frame + + async def traced_create_audio_context(context_id): + if getattr(service, "_tracing_enabled", False): + try: + parent = _get_turn_context(service) or _get_parent_service_context(service) + tracer = trace.get_tracer("pipecat") + span = tracer.start_span("tts", context=parent) + service._tts_spans[context_id] = {"span": span, "ttfb_recorded": False} + + settings = getattr(service, "_settings", None) + add_tts_span_attributes( + span=span, + service_name=service.__class__.__name__, + model=_get_model_name(service), + voice_id=getattr(settings, "voice", "unknown"), + settings=settings, + operation_name="tts", + ) + except Exception as e: + logging.warning(f"Error opening TTS span: {e}") + return await orig_create(context_id) + + async def traced_append_to_audio_context(context_id, frame): + entry = service._tts_spans.get(context_id) + if entry and frame is not None: + try: + if isinstance(frame, TTSStoppedFrame): + entry["span"].end() + service._tts_spans.pop(context_id, None) + except Exception as e: + logging.warning(f"Error updating TTS span: {e}") + return await orig_append(context_id, frame) + + async def traced_push_frame(frame, direction=FrameDirection.DOWNSTREAM): + await orig_push_frame(frame, direction) + if not getattr(service, "_tracing_enabled", False): + return + if not isinstance(frame, MetricsFrame): + return + try: + playing_id = getattr(service, "_playing_context_id", None) + if playing_id is None: + return + entry = service._tts_spans.get(playing_id) + if not entry or entry["ttfb_recorded"]: + return + for data in frame.data: + if isinstance(data, TTFBMetricsData): + entry["span"].set_attribute("metrics.ttfb", data.value) + entry["ttfb_recorded"] = True + break + except Exception as e: + logging.warning(f"Error recording TTS ttfb from MetricsFrame: {e}") + + async def traced_remove_audio_context(context_id): + entry = service._tts_spans.pop(context_id, None) + if entry: + try: + entry["span"].end() + except Exception as e: + logging.warning(f"Error closing TTS span: {e}") + return await orig_remove(context_id) + + async def traced_on_audio_context_completed(context_id): + end_tts_span(service, context_id) + return await orig_completed(context_id) + + def traced_reset_active_audio_context(): + playing_id = getattr(service, "_playing_context_id", None) + if playing_id is not None: + end_tts_span(service, playing_id, interrupted=True) + return orig_reset_active() + + service.create_audio_context = traced_create_audio_context + service.append_to_audio_context = traced_append_to_audio_context + service.push_frame = traced_push_frame + service.remove_audio_context = traced_remove_audio_context + service.on_audio_context_completed = traced_on_audio_context_completed + service.reset_active_audio_context = traced_reset_active_audio_context + + def patch_setup(owner): + """Wrap ``owner.setup`` so audio-context patches install per-instance. + + Idempotent: if a parent class has already been wrapped, + skip. The patches check ``_tracing_enabled`` at invocation + time, so wrapping is always safe. + """ + original_setup = owner.setup + if getattr(original_setup, "__tts_tracing_setup_wrapped__", False): return - service_class_name = self.__class__.__name__ - span_name = "tts" + @functools.wraps(original_setup) + async def patched_setup(self, setup): + await original_setup(self, setup) + install_audio_context_patches(self) - # Get parent context - parent_context = _get_turn_context(self) or _get_parent_service_context(self) + patched_setup.__tts_tracing_setup_wrapped__ = True + owner.setup = patched_setup - # Create span - tracer = trace.get_tracer("pipecat") - with tracer.start_as_current_span(span_name, context=parent_context) as span: - try: - settings = getattr(self, "_settings", None) - add_tts_span_attributes( - span=span, - service_name=service_class_name, - model=_get_model_name(self), - voice_id=getattr(settings, "voice", "unknown"), - text=text, - settings=settings, - character_count=len(text), - operation_name="tts", - cartesia_version=getattr(self, "_cartesia_version", None), - context_id=getattr(self, "_context_id", None), - ) + def attach_run_tts_attributes(service, text, args, kwargs): + """Attach text-specific attributes to the in-flight TTS span.""" + if not getattr(service, "_tracing_enabled", False): + return + try: + context_id = args[0] if args else kwargs.get("context_id") + entry = getattr(service, "_tts_spans", {}).get(context_id) + if entry and text: + span = entry["span"] + span.set_attribute("text", text) + span.set_attribute("metrics.character_count", len(text)) + except Exception as e: + logging.warning(f"Error attaching TTS text to span: {e}") - yield span + def make_run_tts_wrapper(): + """Build the wrapper around ``run_tts`` that adds per-call attributes. - except Exception as e: - logging.warning(f"Error in TTS tracing: {e}") - raise - finally: - # Update TTFB metric at the end - ttfb: float | None = getattr(getattr(self, "_metrics", None), "ttfb", None) - if ttfb is not None: - span.set_attribute("metrics.ttfb", ttfb) + Span lifetime is owned by the audio-context patches. This + wrapper only attaches the text and character count to the + span that was opened by ``create_audio_context`` just + before ``run_tts`` was invoked. + """ + if is_async_generator: - if is_async_generator: - - @functools.wraps(f) - async def gen_wrapper(self, text, *args, **kwargs): - if not getattr(self, "_tracing_enabled", False): - async for item in f(self, text, *args, **kwargs): - yield item - return - - fn_called = False - try: - async with tracing_context(self, text): - fn_called = True - async for item in f(self, text, *args, **kwargs): - yield item - except Exception as e: - if fn_called: - raise - logging.error(f"Error in TTS tracing (continuing without tracing): {e}") + @functools.wraps(f) + async def gen_wrapper(self, text, *args, **kwargs): + attach_run_tts_attributes(self, text, args, kwargs) async for item in f(self, text, *args, **kwargs): yield item - return gen_wrapper - else: + return gen_wrapper @functools.wraps(f) - async def wrapper(self, text, *args, **kwargs): - if not getattr(self, "_tracing_enabled", False): - return await f(self, text, *args, **kwargs) + async def coro_wrapper(self, text, *args, **kwargs): + attach_run_tts_attributes(self, text, args, kwargs) + return await f(self, text, *args, **kwargs) - fn_called = False - try: - async with tracing_context(self, text): - fn_called = True - return await f(self, text, *args, **kwargs) - except Exception as e: - if fn_called: - raise - logging.error(f"Error in TTS tracing (continuing without tracing): {e}") - return await f(self, text, *args, **kwargs) + return coro_wrapper - return wrapper + class _TracedTTSDescriptor: + """Class-level descriptor that wires up TTS tracing at class definition time. + + ``__set_name__`` fires when the class body finishes evaluating, + giving us a chance to wrap the owner's ``setup()`` so that the + audio-context patches install on every instance before any + ``create_audio_context`` call (including the very first one). + """ + + def __set_name__(self, owner, attr_name): + patch_setup(owner) + setattr(owner, attr_name, make_run_tts_wrapper()) + + return _TracedTTSDescriptor() if func is not None: return decorator(func) From 079282d140e419611944020c52417764d97888f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 May 2026 09:38:28 -0700 Subject: [PATCH 2/7] Add changelog for #4467 --- changelog/4467.fixed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4467.fixed.md diff --git a/changelog/4467.fixed.md b/changelog/4467.fixed.md new file mode 100644 index 000000000..a365da620 --- /dev/null +++ b/changelog/4467.fixed.md @@ -0,0 +1 @@ +- Fixed incorrect `metrics.ttfb` on TTS OpenTelemetry spans for streaming services. From 111e59a7b11e5b7b90dc8d2d749283da080423eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 May 2026 12:00:46 -0700 Subject: [PATCH 3/7] Apply the same span-scope fix to traced_stt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit @traced_stt had the same root issue as @traced_tts: the span lifetime was tied to a per-transcript handler call, which doesn't match the operation we want to trace. Now uses the __set_name__ pattern to install: - A push_frame wrapper that drives one STT span per finalized TranscriptionFrame. The span is anchored at speech start (VADUserStartedSpeakingFrame.timestamp - start_secs) but lazy-opened on the first TranscriptionFrame. Opening earlier (on VAD or UserStartedSpeakingFrame) races with TurnTraceObserver._handle_turn_started, which runs as a background task via _call_event_handler (sync=False), so the span would end up parented to the previous turn. Deferring the open to the first TranscriptionFrame avoids that race because STT only emits transcripts well after the turn observer has set the current turn's context. - A stop_ttfb_metrics wrapper that closes the span on the TTFB-timeout path (called with end_time != None from stt_service.py:566). The span is marked stt.timed_out=True and its end_time is pinned to the timeout's end_time (= _last_transcript_time) so the duration reflects when STT actually stopped responding, not when the timeout fired. Span lifecycle: - Open: lazy on first TranscriptionFrame of a segment. - Close (success): finalized=True attaches metrics.ttfb and closes the span. Multiple finalized transcripts in a single turn produce multiple spans. - Close (timeout): stop_ttfb_metrics(end_time=...) closes with stt.timed_out=True. - Close (orphan): UserStoppedSpeakingFrame closes any still-open span with stt.incomplete=True (covers turns where no finalized transcript and no timeout fired). No changes required outside service_decorators.py — stt_service.py and every per-service file are untouched. --- .../utils/tracing/service_decorators.py | 244 ++++++++++++++---- 1 file changed, 196 insertions(+), 48 deletions(-) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 94770db52..fe4bbe96e 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -23,9 +23,16 @@ if TYPE_CHECKING: from opentelemetry import context as context_api from opentelemetry import trace -from pipecat.frames.frames import MetricsFrame, TTSStoppedFrame +from pipecat.frames.frames import ( + MetricsFrame, + TranscriptionFrame, + TTSStoppedFrame, + UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, +) from pipecat.metrics.metrics import TTFBMetricsData from pipecat.processors.aggregators.llm_context import NOT_GIVEN +from pipecat.processors.frame_processor import FrameDirection from pipecat.utils.tracing.service_attributes import ( add_gemini_live_span_attributes, add_llm_span_attributes, @@ -425,69 +432,210 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call - Language information - Performance metrics like TTFB + The span is scoped to one STT segment, from + ``VADUserStartedSpeakingFrame`` (or the first ``TranscriptionFrame`` + when VAD did not fire, e.g. whispered speech) until a finalized + ``TranscriptionFrame``. Multiple finalized transcripts in a single + user turn produce multiple sequential spans, each anchored at the + point speech for that segment began. ``metrics.ttfb`` is read after + the base ``push_frame`` runs ``stop_ttfb_metrics`` for the + finalized frame, so the value is correct for the closing span. + Args: func: The STT method to trace. name: Custom span name. Defaults to function name. Returns: - Wrapped method with STT-specific tracing. + The original method unchanged. The decorator's class-definition- + time work is to install a ``push_frame`` wrapper on the owning + class that owns the span lifetime. """ if not is_tracing_available(): return _noop_decorator if func is None else _noop_decorator(func) def decorator(f): - @functools.wraps(f) - async def wrapper(self, transcript, is_final, language=None): - if not getattr(self, "_tracing_enabled", False): - return await f(self, transcript, is_final, language) + def patch_push_frame(owner): + """Wrap ``owner.push_frame`` to drive the STT span lifecycle. - fn_called = False - try: - service_class_name = self.__class__.__name__ - span_name = "stt" + Idempotent: if a parent class has already been wrapped, skip. + The wrapper checks ``_tracing_enabled`` at invocation time, + so it is safe to install regardless of whether tracing is + enabled. + """ + original_push_frame = owner.push_frame + if getattr(original_push_frame, "__stt_tracing_push_frame_wrapped__", False): + return - # Get the turn context first, then fall back to service context - parent_context = _get_turn_context(self) or _get_parent_service_context(self) - - # Create a new span as child of the turn span or service span + def open_span(service, state): + """Open the STT span, anchored at ``segment_start_time`` if set.""" + parent = _get_turn_context(service) or _get_parent_service_context(service) tracer = trace.get_tracer("pipecat") - with tracer.start_as_current_span( - span_name, context=parent_context - ) as current_span: + kwargs = {"context": parent} + if state["segment_start_time"] is not None: + kwargs["start_time"] = int(state["segment_start_time"] * 1e9) + span = tracer.start_span("stt", **kwargs) + try: + settings = getattr(service, "_settings", None) + add_stt_span_attributes( + span=span, + service_name=service.__class__.__name__, + model=_get_model_name(service), + settings=settings, + vad_enabled=getattr(service, "vad_enabled", False), + ) + except Exception as e: + logging.warning(f"Error setting STT span baseline attributes: {e}") + state["span"] = span + + def handle_pre_push(service, frame, state): + """Record speech-start anchor. + + The span itself is lazy-opened in ``handle_post_push`` + when the first ``TranscriptionFrame`` arrives. Opening + on ``VADUserStartedSpeakingFrame`` or + ``UserStartedSpeakingFrame`` would race with + ``TurnTraceObserver._handle_turn_started``, which runs + in a background task fired by ``_call_event_handler`` + (``base_object.py:232``) and may not have set the new + turn's context yet — that produces STT spans parented + to the previous turn. By the time STT actually emits a + transcript, the turn observer has run. + """ + if isinstance(frame, VADUserStartedSpeakingFrame): + # Anchor the next span at the moment speech began. + # Skip if we already have an anchor (intra-turn VAD + # re-trigger) or a span open. + if state["span"] is None and state["segment_start_time"] is None: + state["segment_start_time"] = frame.timestamp - frame.start_secs + + def handle_post_push(service, frame, state): + """Lazy-open span on first transcript; attach attrs; close on finalized. + + One STT span per finalized transcript: the span opens + on the first ``TranscriptionFrame`` (anchored at + speech start via ``segment_start_time``) and closes on + ``finalized=True``. Multiple finalized transcripts in + a single turn produce multiple spans; subsequent + spans lazy-open with no explicit anchor (start_time + defaults to "now"). Spans that never see a finalized + transcript are closed by ``UserStoppedSpeakingFrame`` + (marked ``stt.incomplete``) or by the TTFB timeout + (marked ``stt.timed_out``). + """ + if isinstance(frame, UserStoppedSpeakingFrame): + prev_span = state["span"] + if prev_span is not None: + prev_span.set_attribute("stt.incomplete", True) + prev_span.end() + state["span"] = None + state["segment_start_time"] = None + elif isinstance(frame, TranscriptionFrame): + if state["span"] is None: + open_span(service, state) + span = state["span"] + if frame.text: + span.set_attribute("transcript", frame.text) + span.set_attribute("is_final", bool(frame.finalized)) + if frame.language: + span.set_attribute("language", str(frame.language)) + if frame.user_id: + span.set_attribute("user_id", frame.user_id) + if frame.finalized: + ttfb = getattr(getattr(service, "_metrics", None), "ttfb", None) + if ttfb is not None: + span.set_attribute("metrics.ttfb", ttfb) + span.end() + state["span"] = None + state["segment_start_time"] = None + + @functools.wraps(original_push_frame) + async def patched_push_frame(self, frame, direction=FrameDirection.DOWNSTREAM): + state = getattr(self, "_stt_span_state", None) + if state is None: + state = {"span": None, "segment_start_time": None} + self._stt_span_state = state + + if getattr(self, "_tracing_enabled", False): try: - # Get TTFB metric if available - ttfb: float | None = getattr(getattr(self, "_metrics", None), "ttfb", None) - - # Use settings from the service if available - settings = getattr(self, "_settings", None) - - add_stt_span_attributes( - span=current_span, - service_name=service_class_name, - model=_get_model_name(self), - transcript=transcript, - is_final=is_final, - language=str(language) if language else None, - user_id=getattr(self, "_user_id", None), - vad_enabled=getattr(self, "vad_enabled", False), - settings=settings, - ttfb=ttfb, - ) - - # Call the original function - fn_called = True - return await f(self, transcript, is_final, language) + handle_pre_push(self, frame, state) except Exception as e: - # Log any exception but don't disrupt the main flow - logging.warning(f"Error in STT transcription tracing: {e}") - raise - except Exception as e: - if fn_called: - raise - logging.error(f"Error in STT tracing (continuing without tracing): {e}") - return await f(self, transcript, is_final, language) + logging.warning(f"Error in STT pre-push tracing: {e}") - return wrapper + await original_push_frame(self, frame, direction) + + if getattr(self, "_tracing_enabled", False): + try: + handle_post_push(self, frame, state) + except Exception as e: + logging.warning(f"Error in STT post-push tracing: {e}") + + patched_push_frame.__stt_tracing_push_frame_wrapped__ = True + owner.push_frame = patched_push_frame + + def patch_stop_ttfb_metrics(owner): + """Wrap ``owner.stop_ttfb_metrics`` to handle the TTFB-timeout path. + + When ``stop_ttfb_metrics`` is invoked with ``end_time`` set, + that signals the STT service's TTFB-timeout handler firing + without a finalized ``TranscriptionFrame`` ever arriving + (`stt_service.py:566`). In that case we attach + ``metrics.ttfb`` and close the span with + ``stt.timed_out=true``. The default ``end_time=None`` path + (called from ``push_frame`` on finalized frames) only + attaches ``metrics.ttfb``; closing happens in the + ``push_frame`` post-hook so the final transcript attributes + land first. + """ + original_stop = owner.stop_ttfb_metrics + if getattr(original_stop, "__stt_tracing_stop_ttfb_wrapped__", False): + return + + @functools.wraps(original_stop) + async def patched_stop(self, *, end_time=None): + await original_stop(self, end_time=end_time) + if not getattr(self, "_tracing_enabled", False): + return + state = getattr(self, "_stt_span_state", None) + if not state or state["span"] is None: + return + span = state["span"] + try: + ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None) + if ttfb is not None: + span.set_attribute("metrics.ttfb", ttfb) + if end_time is not None: + span.set_attribute("stt.timed_out", True) + # Use end_time (= _last_transcript_time) so the + # span ends where the timeout handler actually + # finalized TTFB, rather than after the timeout + # sleep. + span.end(end_time=int(end_time * 1e9)) + state["span"] = None + state["segment_start_time"] = None + except Exception as e: + logging.warning(f"Error in STT stop_ttfb_metrics tracing: {e}") + + patched_stop.__stt_tracing_stop_ttfb_wrapped__ = True + owner.stop_ttfb_metrics = patched_stop + + class _TracedSTTDescriptor: + """Class-level descriptor that wires up STT tracing at class definition time. + + ``__set_name__`` fires when the class body finishes evaluating, + giving us a chance to wrap the owner's ``push_frame`` so that + VAD, transcription, and finalization events drive the span + lifecycle, and to wrap ``stop_ttfb_metrics`` so the + TTFB-timeout path can attach metrics and close the span when + no finalized transcript ever arrives. The decorated method + itself runs unchanged. + """ + + def __set_name__(self, owner, attr_name): + patch_push_frame(owner) + patch_stop_ttfb_metrics(owner) + setattr(owner, attr_name, f) + + return _TracedSTTDescriptor() if func is not None: return decorator(func) From e70ee603b21621c6d80538a3730d1dc5e2bf6e42 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 May 2026 12:01:00 -0700 Subject: [PATCH 4/7] Add STT changelog entry for #4467 --- changelog/4467.fixed.2.md | 1 + .../utils/tracing/service_decorators.py | 172 ++++++++++++------ 2 files changed, 120 insertions(+), 53 deletions(-) create mode 100644 changelog/4467.fixed.2.md diff --git a/changelog/4467.fixed.2.md b/changelog/4467.fixed.2.md new file mode 100644 index 000000000..80de8a289 --- /dev/null +++ b/changelog/4467.fixed.2.md @@ -0,0 +1 @@ +- Fixed incorrect `metrics.ttfb` on STT OpenTelemetry spans, and parented them to the current turn span. diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index fe4bbe96e..5828e7fb7 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -466,6 +466,29 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call if getattr(original_push_frame, "__stt_tracing_push_frame_wrapped__", False): return + def update_transcript(state, new_text): + """Append or extend the current segment in ``state['segments']``. + + If ``new_text`` starts with the last recorded segment, + treat it as a continuation (interim accumulation) and + replace the last segment. Otherwise treat it as a new + segment and append. Some STT services (Deepgram with + utterance_end_ms enabled, for example) emit several + ``TranscriptionFrame``s per turn where each carries a + different segment rather than a cumulative update — + without this logic the span's transcript would only + show the last segment and the beginning would be lost. + """ + if not new_text: + return + segments = state["segments"] + if not segments: + segments.append(new_text) + elif new_text.startswith(segments[-1]): + segments[-1] = new_text + else: + segments.append(new_text) + def open_span(service, state): """Open the STT span, anchored at ``segment_start_time`` if set.""" parent = _get_turn_context(service) or _get_parent_service_context(service) @@ -488,18 +511,24 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call state["span"] = span def handle_pre_push(service, frame, state): - """Record speech-start anchor. + """Record speech-start anchor; lazy-open span on first transcript. - The span itself is lazy-opened in ``handle_post_push`` - when the first ``TranscriptionFrame`` arrives. Opening - on ``VADUserStartedSpeakingFrame`` or - ``UserStartedSpeakingFrame`` would race with + Lazy-opening on ``TranscriptionFrame`` (rather than on + ``VADUserStartedSpeakingFrame`` or + ``UserStartedSpeakingFrame``) avoids racing with ``TurnTraceObserver._handle_turn_started``, which runs in a background task fired by ``_call_event_handler`` (``base_object.py:232``) and may not have set the new turn's context yet — that produces STT spans parented - to the previous turn. By the time STT actually emits a - transcript, the turn observer has run. + to the previous turn. By the time STT actually emits + a transcript, the turn observer has run. + + Opening happens in pre-push (rather than post-push) so + that the recursive ``push_frame`` that + ``STTService.push_frame`` triggers for the + ``MetricsFrame`` (via ``stop_ttfb_metrics`` at + ``stt_service.py:465``) sees the span already open and + can attribute ``metrics.ttfb`` to it. """ if isinstance(frame, VADUserStartedSpeakingFrame): # Anchor the next span at the moment speech began. @@ -507,52 +536,91 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call # re-trigger) or a span open. if state["span"] is None and state["segment_start_time"] is None: state["segment_start_time"] = frame.timestamp - frame.start_secs + elif isinstance(frame, TranscriptionFrame) and state["span"] is None: + open_span(service, state) - def handle_post_push(service, frame, state): - """Lazy-open span on first transcript; attach attrs; close on finalized. + async def handle_post_push(service, frame, state): + """Attach per-frame attrs; close on finalized; record TTFB from MetricsFrame. + + ``metrics.ttfb`` is read off the ``TTFBMetricsData`` + payload of any ``MetricsFrame`` pushed by + ``stop_ttfb_metrics`` — the canonical value the rest + of the system uses — rather than from + ``_metrics.ttfb``, which has an in-progress fallback + branch (``frame_processor_metrics.py:48-62``) that + would return an under-estimate if read at the wrong + time. One STT span per finalized transcript: the span opens - on the first ``TranscriptionFrame`` (anchored at - speech start via ``segment_start_time``) and closes on - ``finalized=True``. Multiple finalized transcripts in - a single turn produce multiple spans; subsequent - spans lazy-open with no explicit anchor (start_time - defaults to "now"). Spans that never see a finalized - transcript are closed by ``UserStoppedSpeakingFrame`` - (marked ``stt.incomplete``) or by the TTFB timeout - (marked ``stt.timed_out``). + lazily on the first ``TranscriptionFrame`` (pre-push, + anchored at speech start via ``segment_start_time``) + and closes on ``finalized=True``. Multiple finalized + transcripts in a single turn produce multiple spans. + + For services that never set ``frame.finalized=True`` + (e.g. Deepgram, which only marks it via + ``confirm_finalize()``), the span closes on + ``UserStoppedSpeakingFrame``. To capture + ``metrics.ttfb`` for those spans we force-stop any + pending TTFB measurement before closing — that pushes + a ``MetricsFrame``, our post-push attributes the + value, and ``patched_stop_ttfb_metrics`` closes the + span. The ``stt.incomplete=true`` flag is only set if + neither a finalized transcript nor a TTFB measurement + ever finalized for the span. """ if isinstance(frame, UserStoppedSpeakingFrame): prev_span = state["span"] - if prev_span is not None: - prev_span.set_attribute("stt.incomplete", True) - prev_span.end() + if prev_span is None: + return + metrics = getattr(service, "_metrics", None) + if metrics is not None and getattr(metrics, "_start_ttfb_time", 0) > 0: + last_transcript_time = getattr(service, "_last_transcript_time", 0) or None + try: + await service.stop_ttfb_metrics(end_time=last_transcript_time) + except Exception as e: + logging.warning(f"Error force-stopping STT TTFB on user turn end: {e}") + # patched_stop_ttfb_metrics may have closed the span + # via the timeout path; re-check. + if state["span"] is None: + state["segments"] = [] + return + state["span"].set_attribute("stt.incomplete", True) + state["span"].end() state["span"] = None state["segment_start_time"] = None - elif isinstance(frame, TranscriptionFrame): - if state["span"] is None: - open_span(service, state) + state["segments"] = [] + elif isinstance(frame, MetricsFrame): span = state["span"] + if span is None: + return + for data in frame.data: + if isinstance(data, TTFBMetricsData): + span.set_attribute("metrics.ttfb", data.value) + break + elif isinstance(frame, TranscriptionFrame): + span = state["span"] + if span is None: + return if frame.text: - span.set_attribute("transcript", frame.text) + update_transcript(state, frame.text) + span.set_attribute("transcript", " ".join(state["segments"]).strip()) span.set_attribute("is_final", bool(frame.finalized)) if frame.language: span.set_attribute("language", str(frame.language)) if frame.user_id: span.set_attribute("user_id", frame.user_id) if frame.finalized: - ttfb = getattr(getattr(service, "_metrics", None), "ttfb", None) - if ttfb is not None: - span.set_attribute("metrics.ttfb", ttfb) span.end() state["span"] = None state["segment_start_time"] = None + state["segments"] = [] @functools.wraps(original_push_frame) async def patched_push_frame(self, frame, direction=FrameDirection.DOWNSTREAM): state = getattr(self, "_stt_span_state", None) if state is None: - state = {"span": None, "segment_start_time": None} + state = {"span": None, "segment_start_time": None, "segments": []} self._stt_span_state = state if getattr(self, "_tracing_enabled", False): @@ -565,7 +633,7 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call if getattr(self, "_tracing_enabled", False): try: - handle_post_push(self, frame, state) + await handle_post_push(self, frame, state) except Exception as e: logging.warning(f"Error in STT post-push tracing: {e}") @@ -573,18 +641,22 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call owner.push_frame = patched_push_frame def patch_stop_ttfb_metrics(owner): - """Wrap ``owner.stop_ttfb_metrics`` to handle the TTFB-timeout path. + """Wrap ``owner.stop_ttfb_metrics`` to close the span on the timeout path. When ``stop_ttfb_metrics`` is invoked with ``end_time`` set, - that signals the STT service's TTFB-timeout handler firing - without a finalized ``TranscriptionFrame`` ever arriving - (`stt_service.py:566`). In that case we attach - ``metrics.ttfb`` and close the span with - ``stt.timed_out=true``. The default ``end_time=None`` path - (called from ``push_frame`` on finalized frames) only - attaches ``metrics.ttfb``; closing happens in the - ``push_frame`` post-hook so the final transcript attributes - land first. + that signals the TTFB-timeout handler firing + (`stt_service.py:566`), or our own force-stop from the + ``UserStoppedSpeakingFrame`` handler. In either case we + anchor the span's end at ``end_time`` + (= ``_last_transcript_time``) rather than at whenever the + coroutine resumed. + + ``metrics.ttfb`` attribution is not done here — the + ``MetricsFrame`` that ``stop_ttfb_metrics`` pushes flows + through ``push_frame`` and gets recorded by + ``handle_post_push``, which reads the canonical + ``TTFBMetricsData.value`` rather than the in-progress + ``_metrics.ttfb`` property. """ original_stop = owner.stop_ttfb_metrics if getattr(original_stop, "__stt_tracing_stop_ttfb_wrapped__", False): @@ -593,25 +665,19 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call @functools.wraps(original_stop) async def patched_stop(self, *, end_time=None): await original_stop(self, end_time=end_time) + if end_time is None: + return if not getattr(self, "_tracing_enabled", False): return state = getattr(self, "_stt_span_state", None) if not state or state["span"] is None: return - span = state["span"] try: - ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None) - if ttfb is not None: - span.set_attribute("metrics.ttfb", ttfb) - if end_time is not None: - span.set_attribute("stt.timed_out", True) - # Use end_time (= _last_transcript_time) so the - # span ends where the timeout handler actually - # finalized TTFB, rather than after the timeout - # sleep. - span.end(end_time=int(end_time * 1e9)) - state["span"] = None - state["segment_start_time"] = None + span = state["span"] + span.end(end_time=int(end_time * 1e9)) + state["span"] = None + state["segment_start_time"] = None + state["segments"] = [] except Exception as e: logging.warning(f"Error in STT stop_ttfb_metrics tracing: {e}") From a3ce963b5488cc39822179d5393088a34940e59c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 May 2026 12:04:08 -0700 Subject: [PATCH 5/7] Capture partial LLM output on interruption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit traced_llm only attached the aggregated ``output`` attribute to the span after the wrapped function returned successfully. When the LLM call was cancelled mid-stream (e.g. interruption during generation), the accumulated text was discarded — the span had no ``output``. Moved the attribute assignment into the ``finally`` block alongside the existing TTFB write so the partial text we already captured via the patched ``push_frame`` lands on the span regardless of whether ``f`` returned normally, raised, or was cancelled. --- src/pipecat/utils/tracing/service_decorators.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 5828e7fb7..8117e06fd 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -889,10 +889,6 @@ def traced_llm(func: Callable | None = None, *, name: str | None = None) -> Call fn_called = True result = await f(self, context, *args, **kwargs) - # Add aggregated output after function completes, if available - if output_text: - current_span.set_attribute("output", output_text) - return result finally: @@ -905,6 +901,15 @@ def traced_llm(func: Callable | None = None, *, name: str | None = None) -> Call ): self.start_llm_usage_metrics = original_start_llm_usage_metrics + # Attach whatever output text we accumulated so + # far. Doing this in finally captures partial + # output when ``f`` is cancelled or raises mid- + # stream (e.g. interruption during LLM + # generation), rather than only on clean + # completion. + if output_text: + current_span.set_attribute("output", output_text) + # Update TTFB metric ttfb: float | None = getattr(getattr(self, "_metrics", None), "ttfb", None) if ttfb is not None: From 52d8008783a576b83ad9a29119bc21eee1b5bb94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 May 2026 12:04:18 -0700 Subject: [PATCH 6/7] Add LLM interruption changelog entry for #4467 --- changelog/4467.fixed.3.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4467.fixed.3.md diff --git a/changelog/4467.fixed.3.md b/changelog/4467.fixed.3.md new file mode 100644 index 000000000..628623c16 --- /dev/null +++ b/changelog/4467.fixed.3.md @@ -0,0 +1 @@ +- Fixed missing `output` attribute on LLM OpenTelemetry spans when the LLM call is interrupted mid-stream. From 7cc7968abb86bb4fb50bf3e25c02fdd58397c8da Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 12 May 2026 12:15:04 -0700 Subject: [PATCH 7/7] Fix pyright errors in service_decorators.py --- .../utils/tracing/service_decorators.py | 28 +++++++++++++------ 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 8117e06fd..c7e7583d0 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -360,7 +360,7 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call await original_setup(self, setup) install_audio_context_patches(self) - patched_setup.__tts_tracing_setup_wrapped__ = True + setattr(patched_setup, "__tts_tracing_setup_wrapped__", True) owner.setup = patched_setup def attach_run_tts_attributes(service, text, args, kwargs): @@ -418,7 +418,11 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call return _TracedTTSDescriptor() if func is not None: - return decorator(func) + # ``decorator(func)`` returns a descriptor placeholder that + # Python replaces with the real wrapped function once + # ``__set_name__`` runs at class definition time. Pyright sees + # only the descriptor instance, hence the ignore. + return decorator(func) # type: ignore[return-value] return decorator @@ -493,10 +497,12 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call """Open the STT span, anchored at ``segment_start_time`` if set.""" parent = _get_turn_context(service) or _get_parent_service_context(service) tracer = trace.get_tracer("pipecat") - kwargs = {"context": parent} - if state["segment_start_time"] is not None: - kwargs["start_time"] = int(state["segment_start_time"] * 1e9) - span = tracer.start_span("stt", **kwargs) + start_time_ns = ( + int(state["segment_start_time"] * 1e9) + if state["segment_start_time"] is not None + else None + ) + span = tracer.start_span("stt", context=parent, start_time=start_time_ns) try: settings = getattr(service, "_settings", None) add_stt_span_attributes( @@ -637,7 +643,7 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call except Exception as e: logging.warning(f"Error in STT post-push tracing: {e}") - patched_push_frame.__stt_tracing_push_frame_wrapped__ = True + setattr(patched_push_frame, "__stt_tracing_push_frame_wrapped__", True) owner.push_frame = patched_push_frame def patch_stop_ttfb_metrics(owner): @@ -681,7 +687,7 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call except Exception as e: logging.warning(f"Error in STT stop_ttfb_metrics tracing: {e}") - patched_stop.__stt_tracing_stop_ttfb_wrapped__ = True + setattr(patched_stop, "__stt_tracing_stop_ttfb_wrapped__", True) owner.stop_ttfb_metrics = patched_stop class _TracedSTTDescriptor: @@ -704,7 +710,11 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call return _TracedSTTDescriptor() if func is not None: - return decorator(func) + # ``decorator(func)`` returns a descriptor placeholder that + # Python replaces with the real wrapped function once + # ``__set_name__`` runs at class definition time. Pyright sees + # only the descriptor instance, hence the ignore. + return decorator(func) # type: ignore[return-value] return decorator