Merge pull request #3649 from itsderek23/fix/tracing-orphan-spans

Fix orphan otel spans during flow initialization and transitions

changelog/3649.fixed.md

@@ -0,0 +1 @@
+- Fixed orphan OpenTelemetry spans during flow initialization and transitions in tracing.

src/pipecat/utils/tracing/service_decorators.py

@@ -25,6 +25,7 @@ if TYPE_CHECKING:
 
 from pipecat.processors.aggregators.llm_context import NOT_GIVEN, LLMContext
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.utils.tracing.conversation_context_provider import get_current_conversation_context
 from pipecat.utils.tracing.service_attributes import (
     add_gemini_live_span_attributes,
     add_llm_span_attributes,
@@ -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()
 
 

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)