diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 68dda6562..f151a16ae 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -32,6 +32,7 @@ from pipecat.utils.tracing.service_attributes import ( add_stt_span_attributes, add_tts_span_attributes, ) +from pipecat.utils.tracing.conversation_context_provider import get_current_conversation_context from pipecat.utils.tracing.setup import is_tracing_available from pipecat.utils.tracing.turn_context_provider import get_current_turn_context @@ -58,7 +59,8 @@ def _noop_decorator(func): def _get_parent_service_context(self): """Get the parent service span context (internal use only). - This looks for the service span that was created when the service was initialized. + This looks for the service span that was created when the service was initialized, + or falls back to the conversation context if available. Args: self: The service instance. @@ -73,7 +75,12 @@ def _get_parent_service_context(self): if hasattr(self, "_span") and self._span: return trace.set_span_in_context(self._span) - # If we can't find a stored span, default to current context + # Fall back to conversation context if available + conversation_context = get_current_conversation_context() + if conversation_context: + return conversation_context + + # Last resort: use current context (may create orphan spans) return context_api.get_current() diff --git a/src/pipecat/utils/tracing/turn_trace_observer.py b/src/pipecat/utils/tracing/turn_trace_observer.py index a76d9bb8f..59b0bd2ae 100644 --- a/src/pipecat/utils/tracing/turn_trace_observer.py +++ b/src/pipecat/utils/tracing/turn_trace_observer.py @@ -15,6 +15,7 @@ from typing import TYPE_CHECKING, Dict, Optional from loguru import logger +from pipecat.frames.frames import StartFrame from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.observers.turn_tracking_observer import TurnTrackingObserver from pipecat.utils.tracing.conversation_context_provider import ConversationContextProvider @@ -81,13 +82,15 @@ class TurnTraceObserver(BaseObserver): async def on_push_frame(self, data: FramePushed): """Process a frame without modifying it. - This observer doesn't need to process individual frames as it - relies on turn start/end events from the turn tracker. + Handles StartFrame to begin conversation tracing early, ensuring + that any spans created before Turn 1 (e.g., from flow initialization) + are properly attached to the conversation trace. Args: data: The frame push event data. """ - pass + if isinstance(data.frame, StartFrame) and not self._conversation_span: + self.start_conversation_tracing(self._conversation_id) def start_conversation_tracing(self, conversation_id: Optional[str] = None): """Start a new conversation span. @@ -159,8 +162,6 @@ class TurnTraceObserver(BaseObserver): if not is_tracing_available() or not self._tracer: return - # If this is the first turn and no conversation span exists yet, - # start the conversation tracing (will generate ID if needed) if turn_number == 1 and not self._conversation_span: self.start_conversation_tracing(self._conversation_id)