wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3721 from pipecat-ai/fix/pipeline-scoped-tracing-context

Browse Source 
Replace singleton context providers with pipeline-scoped TracingContext
This commit is contained in:
Mark Backman
2026-02-13 11:24:37 -05:00
committed by GitHub
parent 9e6f27c9f1 25ca296477
commit 347eaf582d
15 changed files with 792 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.github/workflows/coverage.yaml vendored
View File

@@ -41,6 +41,7 @@ jobs:
--extra livekit \
--extra local-smart-turn-v3 \
--extra piper \
--extra tracing \
--extra websocket
- name: Run tests with coverage

1
.github/workflows/tests.yaml vendored
View File

@@ -45,6 +45,7 @@ jobs:
--extra livekit \
--extra local-smart-turn-v3 \
--extra piper \
--extra tracing \
--extra websocket
- name: Test with pytest

3
src/pipecat/frames/frames.py
View File

@@ -42,6 +42,7 @@ from pipecat.utils.utils import obj_count, obj_id
if TYPE_CHECKING:
from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.utils.tracing.tracing_context import TracingContext
class DeprecatedKeypadEntry:
@@ -1036,6 +1037,7 @@ class StartFrame(SystemFrame):
Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead.
report_only_initial_ttfb: Whether to report only initial time-to-first-byte.
tracing_context: Pipeline-scoped tracing context for span hierarchy.
"""
audio_in_sample_rate: int = 16000
@@ -1046,6 +1048,7 @@ class StartFrame(SystemFrame):
enable_usage_metrics: bool = False
interruption_strategies: List[BaseInterruptionStrategy] = field(default_factory=list)
report_only_initial_ttfb: bool = False
tracing_context: Optional["TracingContext"] = None
@dataclass

6
src/pipecat/pipeline/task.py
View File

@@ -53,6 +53,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, F
from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIObserverParams, RTVIProcessor
from pipecat.utils.asyncio.task_manager import BaseTaskManager, TaskManager, TaskManagerParams
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.tracing_context import TracingContext
from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
HEARTBEAT_SECS = 1.0
@@ -290,10 +291,13 @@ class PipelineTask(BasePipelineTask):
self._turn_tracking_observer: Optional[TurnTrackingObserver] = None
self._user_bot_latency_observer: Optional[UserBotLatencyObserver] = None
self._turn_trace_observer: Optional[TurnTraceObserver] = None
self._tracing_context: Optional[TracingContext] = None
if self._enable_turn_tracking:
self._turn_tracking_observer = TurnTrackingObserver()
observers.append(self._turn_tracking_observer)
if self._enable_tracing and self._turn_tracking_observer:
# Create pipeline-scoped tracing context
self._tracing_context = TracingContext()
# Create latency observer for tracing
self._user_bot_latency_observer = UserBotLatencyObserver()
observers.append(self._user_bot_latency_observer)
@@ -303,6 +307,7 @@ class PipelineTask(BasePipelineTask):
latency_tracker=self._user_bot_latency_observer,
conversation_id=self._conversation_id,
additional_span_attributes=self._additional_span_attributes,
tracing_context=self._tracing_context,
)
observers.append(self._turn_trace_observer)
@@ -813,6 +818,7 @@ class PipelineTask(BasePipelineTask):
enable_usage_metrics=self._params.enable_usage_metrics,
report_only_initial_ttfb=self._params.report_only_initial_ttfb,
interruption_strategies=self._params.interruption_strategies,
tracing_context=self._tracing_context,
)
start_frame.metadata = self._create_start_metadata()
await self._pipeline.queue_frame(start_frame)

5
src/pipecat/services/ai_service.py
View File

@@ -44,6 +44,8 @@ class AIService(FrameProcessor):
self._model_name: str = ""
self._settings: Dict[str, Any] = {}
self._session_properties: Dict[str, Any] = {}
self._tracing_enabled: bool = False
self._tracing_context = None
@property
def model_name(self) -> str:
@@ -72,7 +74,8 @@ class AIService(FrameProcessor):
Args:
frame: The start frame containing initialization parameters.
"""
pass
self._tracing_enabled = frame.enable_tracing
self._tracing_context = frame.tracing_context
async def stop(self, frame: EndFrame):
"""Stop the AI service.

2
src/pipecat/services/llm_service.py
View File

@@ -198,7 +198,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {}
self._function_call_tasks: Dict[Optional[asyncio.Task], FunctionCallRunnerItem] = {}
self._sequential_runner_task: Optional[asyncio.Task] = None
self._tracing_enabled: bool = False
self._skip_tts: Optional[bool] = None
self._summary_task: Optional[asyncio.Task] = None
@@ -285,7 +284,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
await super().start(frame)
if not self._run_in_parallel:
await self._create_sequential_runner_task()
self._tracing_enabled = frame.enable_tracing
async def stop(self, frame: EndFrame):
"""Stop the LLM service.

2
src/pipecat/services/stt_service.py
View File

@@ -114,7 +114,6 @@ class STTService(AIService):
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._settings: Dict[str, Any] = {}
self._tracing_enabled: bool = False
self._muted: bool = False
self._user_id: str = ""
self._ttfs_p99_latency = ttfs_p99_latency
@@ -220,7 +219,6 @@ class STTService(AIService):
"""
await super().start(frame)
self._sample_rate = self._init_sample_rate or frame.audio_in_sample_rate
self._tracing_enabled = frame.enable_tracing
async def cleanup(self):
"""Clean up STT service resources."""

3
src/pipecat/services/tts_service.py
View File

@@ -208,8 +208,6 @@ class TTSService(AIService):
# TODO: Deprecate _text_filters when added to LLMTextProcessor
self._text_filters: Sequence[BaseTextFilter] = text_filters or []
self._transport_destination: Optional[str] = transport_destination
self._tracing_enabled: bool = False
if text_filter:
import warnings
@@ -349,7 +347,6 @@ class TTSService(AIService):
self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
if self._push_stop_frames and not self._stop_frame_task:
self._stop_frame_task = self.create_task(self._stop_frame_handler())
self._tracing_enabled = frame.enable_tracing
async def stop(self, frame: EndFrame):
"""Stop the TTS service.

114
src/pipecat/utils/tracing/conversation_context_provider.py
View File

@@ -1,114 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Conversation context provider for OpenTelemetry tracing in Pipecat.
This module provides a singleton context provider that manages the current
conversation's tracing context, allowing services to create child spans
that are properly associated with the conversation.
"""
import uuid
from typing import TYPE_CHECKING, Optional
# Import types for type checking only
if TYPE_CHECKING:
from opentelemetry.context import Context
from opentelemetry.trace import SpanContext
from pipecat.utils.tracing.setup import is_tracing_available
if is_tracing_available():
from opentelemetry.context import Context
from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context
class ConversationContextProvider:
"""Provides access to the current conversation's tracing context.
This is a singleton that can be used to get the current conversation's
span context to create child spans (like turns).
"""
_instance = None
_current_conversation_context: Optional["Context"] = None
_conversation_id: Optional[str] = None
@classmethod
def get_instance(cls):
"""Get the singleton instance.
Returns:
The singleton ConversationContextProvider instance.
"""
if cls._instance is None:
cls._instance = ConversationContextProvider()
return cls._instance
def set_current_conversation_context(
self, span_context: Optional["SpanContext"], conversation_id: Optional[str] = None
):
"""Set the current conversation context.
Args:
span_context: The span context for the current conversation or None to clear it.
conversation_id: Optional ID for the conversation.
"""
if not is_tracing_available():
return
self._conversation_id = conversation_id
if span_context:
# Create a non-recording span from the span context
non_recording_span = NonRecordingSpan(span_context)
self._current_conversation_context = set_span_in_context(non_recording_span)
else:
self._current_conversation_context = None
def get_current_conversation_context(self) -> Optional["Context"]:
"""Get the OpenTelemetry context for the current conversation.
Returns:
The current conversation context or None if not available.
"""
return self._current_conversation_context
def get_conversation_id(self) -> Optional[str]:
"""Get the ID for the current conversation.
Returns:
The current conversation ID or None if not available.
"""
return self._conversation_id
def generate_conversation_id(self) -> str:
"""Generate a new conversation ID.
Returns:
A new randomly generated UUID string.
"""
return str(uuid.uuid4())
def get_current_conversation_context() -> Optional["Context"]:
"""Get the OpenTelemetry context for the current conversation.
Returns:
The current conversation context or None if not available.
"""
provider = ConversationContextProvider.get_instance()
return provider.get_current_conversation_context()
def get_conversation_id() -> Optional[str]:
"""Get the ID for the current conversation.
Returns:
The current conversation ID or None if not available.
"""
provider = ConversationContextProvider.get_instance()
return provider.get_conversation_id()

33
src/pipecat/utils/tracing/service_decorators.py
View File

@@ -25,7 +25,6 @@ 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,
@@ -34,7 +33,6 @@ from pipecat.utils.tracing.service_attributes import (
add_tts_span_attributes,
)
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.turn_context_provider import get_current_turn_context
if is_tracing_available():
from opentelemetry import context as context_api
@@ -56,6 +54,19 @@ def _noop_decorator(func):
return func
def _get_turn_context(self):
"""Get the current turn's tracing context if available.
Args:
self: The service instance.
Returns:
The turn context, or None if unavailable.
"""
tracing_ctx = getattr(self, "_tracing_context", None)
return tracing_ctx.get_turn_context() if tracing_ctx else None
def _get_parent_service_context(self):
"""Get the parent service span context (internal use only).
@@ -76,7 +87,8 @@ def _get_parent_service_context(self):
return trace.set_span_in_context(self._span)
# Fall back to conversation context if available
conversation_context = get_current_conversation_context()
tracing_ctx = getattr(self, "_tracing_context", None)
conversation_context = tracing_ctx.get_conversation_context() if tracing_ctx else None
if conversation_context:
return conversation_context
@@ -183,8 +195,7 @@ def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) -
span_name = "tts"
# Get parent context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
parent_context = _get_turn_context(self) or _get_parent_service_context(self)
# Create span
tracer = trace.get_tracer("pipecat")
@@ -290,8 +301,7 @@ def traced_stt(func: Optional[Callable] = None, *, name: Optional[str] = None) -
span_name = "stt"
# Get the turn context first, then fall back to service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
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
tracer = trace.get_tracer("pipecat")
@@ -372,8 +382,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
span_name = "llm"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
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
tracer = trace.get_tracer("pipecat")
@@ -582,8 +591,7 @@ def traced_gemini_live(operation: str) -> Callable:
span_name = f"{operation}"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
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
tracer = trace.get_tracer("pipecat")
@@ -889,8 +897,7 @@ def traced_openai_realtime(operation: str) -> Callable:
span_name = f"{operation}"
# Get the parent context - turn context if available, otherwise service context
turn_context = get_current_turn_context()
parent_context = turn_context or _get_parent_service_context(self)
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
tracer = trace.get_tracer("pipecat")

109
src/pipecat/utils/tracing/tracing_context.py Normal file
View File

@@ -0,0 +1,109 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Pipeline-scoped tracing context for OpenTelemetry tracing in Pipecat.
This module provides a per-pipeline tracing context that holds the current
conversation and turn span contexts. Each PipelineTask creates its own
TracingContext, ensuring concurrent pipelines do not interfere with each other.
"""
import uuid
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from opentelemetry.context import Context
from opentelemetry.trace import SpanContext
from pipecat.utils.tracing.setup import is_tracing_available
if is_tracing_available():
from opentelemetry.context import Context
from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context
class TracingContext:
"""Pipeline-scoped tracing context.
Holds the current conversation and turn span contexts for a single pipeline.
Created by PipelineTask, passed to TurnTraceObserver (writer) and services
(readers) via StartFrame.
"""
def __init__(self):
"""Initialize the tracing context with empty state."""
self._conversation_context: Optional["Context"] = None
self._turn_context: Optional["Context"] = None
self._conversation_id: Optional[str] = None
def set_conversation_context(
self, span_context: Optional["SpanContext"], conversation_id: Optional[str] = None
):
"""Set the current conversation context.
Args:
span_context: The span context for the current conversation or None to clear it.
conversation_id: Optional ID for the conversation.
"""
if not is_tracing_available():
return
self._conversation_id = conversation_id
if span_context:
non_recording_span = NonRecordingSpan(span_context)
self._conversation_context = set_span_in_context(non_recording_span)
else:
self._conversation_context = None
def get_conversation_context(self) -> Optional["Context"]:
"""Get the OpenTelemetry context for the current conversation.
Returns:
The current conversation context or None if not available.
"""
return self._conversation_context
def set_turn_context(self, span_context: Optional["SpanContext"]):
"""Set the current turn context.
Args:
span_context: The span context for the current turn or None to clear it.
"""
if not is_tracing_available():
return
if span_context:
non_recording_span = NonRecordingSpan(span_context)
self._turn_context = set_span_in_context(non_recording_span)
else:
self._turn_context = None
def get_turn_context(self) -> Optional["Context"]:
"""Get the OpenTelemetry context for the current turn.
Returns:
The current turn context or None if not available.
"""
return self._turn_context
@property
def conversation_id(self) -> Optional[str]:
"""Get the ID for the current conversation.
Returns:
The current conversation ID or None if not available.
"""
return self._conversation_id
@staticmethod
def generate_conversation_id() -> str:
"""Generate a new conversation ID.
Returns:
A new randomly generated UUID string.
"""
return str(uuid.uuid4())

81
src/pipecat/utils/tracing/turn_context_provider.py
View File

@@ -1,81 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Turn context provider for OpenTelemetry tracing in Pipecat.
This module provides a singleton context provider that manages the current
turn's tracing context, allowing services to create child spans that are
properly associated with the conversation turn.
"""
from typing import TYPE_CHECKING, Optional
# Import types for type checking only
if TYPE_CHECKING:
from opentelemetry.context import Context
from opentelemetry.trace import SpanContext
from pipecat.utils.tracing.setup import is_tracing_available
if is_tracing_available():
from opentelemetry.context import Context
from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context
class TurnContextProvider:
"""Provides access to the current turn's tracing context.
This is a singleton that services can use to get the current turn's
span context to create child spans.
"""
_instance = None
_current_turn_context: Optional["Context"] = None
@classmethod
def get_instance(cls):
"""Get the singleton instance.
Returns:
The singleton TurnContextProvider instance.
"""
if cls._instance is None:
cls._instance = TurnContextProvider()
return cls._instance
def set_current_turn_context(self, span_context: Optional["SpanContext"]):
"""Set the current turn context.
Args:
span_context: The span context for the current turn or None to clear it.
"""
if not is_tracing_available():
return
if span_context:
# Create a non-recording span from the span context
non_recording_span = NonRecordingSpan(span_context)
self._current_turn_context = set_span_in_context(non_recording_span)
else:
self._current_turn_context = None
def get_current_turn_context(self) -> Optional["Context"]:
"""Get the OpenTelemetry context for the current turn.
Returns:
The current turn context or None if not available.
"""
return self._current_turn_context
def get_current_turn_context() -> Optional["Context"]:
"""Get the OpenTelemetry context for the current turn.
Returns:
The current turn context or None if not available.
"""
provider = TurnContextProvider.get_instance()
return provider.get_current_turn_context()

36
src/pipecat/utils/tracing/turn_trace_observer.py
View File

@@ -19,9 +19,8 @@ from pipecat.frames.frames import StartFrame
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.utils.tracing.conversation_context_provider import ConversationContextProvider
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.turn_context_provider import TurnContextProvider
from pipecat.utils.tracing.tracing_context import TracingContext
# Import types for type checking only
if TYPE_CHECKING:
@@ -49,6 +48,7 @@ class TurnTraceObserver(BaseObserver):
latency_tracker: UserBotLatencyObserver,
conversation_id: Optional[str] = None,
additional_span_attributes: Optional[dict] = None,
tracing_context: Optional[TracingContext] = None,
**kwargs,
):
"""Initialize the turn trace observer.
@@ -58,11 +58,13 @@ class TurnTraceObserver(BaseObserver):
latency_tracker: The latency tracking observer for user-bot latency.
conversation_id: Optional conversation ID for grouping turns.
additional_span_attributes: Additional attributes to add to spans.
tracing_context: Pipeline-scoped tracing context for span hierarchy.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(**kwargs)
self._turn_tracker = turn_tracker
self._latency_tracker = latency_tracker
self._tracing_context = tracing_context or TracingContext()
self._current_span: Optional["Span"] = None
self._current_turn_number: int = 0
self._trace_context_map: Dict[int, "SpanContext"] = {}
@@ -123,9 +125,8 @@ class TurnTraceObserver(BaseObserver):
return
# Generate a conversation ID if not provided
context_provider = ConversationContextProvider.get_instance()
if conversation_id is None:
conversation_id = context_provider.generate_conversation_id()
conversation_id = TracingContext.generate_conversation_id()
logger.debug(f"Generated new conversation ID: {conversation_id}")
self._conversation_id = conversation_id
@@ -140,8 +141,8 @@ class TurnTraceObserver(BaseObserver):
for k, v in (self._additional_span_attributes or {}).items():
self._conversation_span.set_attribute(k, v)
# Update the conversation context provider
context_provider.set_current_conversation_context(
# Update the tracing context
self._tracing_context.set_conversation_context(
self._conversation_span.get_span_context(), conversation_id
)
@@ -161,9 +162,8 @@ class TurnTraceObserver(BaseObserver):
self._current_span.end()
self._current_span = None
# Clear the turn context provider
context_provider = TurnContextProvider.get_instance()
context_provider.set_current_turn_context(None)
# Clear the turn context
self._tracing_context.set_turn_context(None)
# Now end the conversation span if it exists
if self._conversation_span:
@@ -171,9 +171,8 @@ class TurnTraceObserver(BaseObserver):
self._conversation_span.end()
self._conversation_span = None
# Clear the context provider
context_provider = ConversationContextProvider.get_instance()
context_provider.set_current_conversation_context(None)
# Clear the conversation context
self._tracing_context.set_conversation_context(None)
logger.debug(f"Ended tracing for Conversation {self._conversation_id}")
self._conversation_id = None
@@ -189,8 +188,7 @@ class TurnTraceObserver(BaseObserver):
# Get the parent context - conversation if available, otherwise use root context
parent_context = None
if self._conversation_span:
context_provider = ConversationContextProvider.get_instance()
parent_context = context_provider.get_current_conversation_context()
parent_context = self._tracing_context.get_conversation_context()
# Create a new span for this turn
self._current_span = self._tracer.start_span("turn", context=parent_context)
@@ -207,9 +205,8 @@ class TurnTraceObserver(BaseObserver):
# Store the span context so services can become children of this span
self._trace_context_map[turn_number] = self._current_span.get_span_context()
# Update the context provider so services can access this span
context_provider = TurnContextProvider.get_instance()
context_provider.set_current_turn_context(self._current_span.get_span_context())
# Update the tracing context so services can access this span
self._tracing_context.set_turn_context(self._current_span.get_span_context())
logger.debug(f"Started tracing for Turn {turn_number}")
@@ -228,9 +225,8 @@ class TurnTraceObserver(BaseObserver):
self._current_span.end()
self._current_span = None
# Clear the context provider
context_provider = TurnContextProvider.get_instance()
context_provider.set_current_turn_context(None)
# Clear the turn context
self._tracing_context.set_turn_context(None)
logger.debug(f"Ended tracing for Turn {turn_number}")

127
tests/test_tracing_context.py Normal file
View File

@@ -0,0 +1,127 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import unittest
try:
from opentelemetry.sdk.trace import TracerProvider
HAS_OPENTELEMETRY = True
except ImportError:
HAS_OPENTELEMETRY = False
from pipecat.utils.tracing.tracing_context import TracingContext
@unittest.skipUnless(HAS_OPENTELEMETRY, "opentelemetry not installed")
class TestTracingContext(unittest.TestCase):
"""Tests for TracingContext."""
@classmethod
def setUpClass(cls):
"""Set up a tracer provider for generating span contexts."""
cls._provider = TracerProvider()
cls._tracer = cls._provider.get_tracer("test")
def test_initial_state_is_empty(self):
"""Test that a new TracingContext starts with no context set."""
ctx = TracingContext()
self.assertIsNone(ctx.get_conversation_context())
self.assertIsNone(ctx.get_turn_context())
self.assertIsNone(ctx.conversation_id)
def test_set_and_get_conversation_context(self):
"""Test setting and retrieving conversation context."""
ctx = TracingContext()
span = self._tracer.start_span("conv")
span_context = span.get_span_context()
ctx.set_conversation_context(span_context, "conv-123")
self.assertIsNotNone(ctx.get_conversation_context())
self.assertEqual(ctx.conversation_id, "conv-123")
span.end()
def test_clear_conversation_context(self):
"""Test clearing conversation context by passing None."""
ctx = TracingContext()
span = self._tracer.start_span("conv")
ctx.set_conversation_context(span.get_span_context(), "conv-123")
self.assertIsNotNone(ctx.get_conversation_context())
ctx.set_conversation_context(None)
self.assertIsNone(ctx.get_conversation_context())
self.assertIsNone(ctx.conversation_id)
span.end()
def test_set_and_get_turn_context(self):
"""Test setting and retrieving turn context."""
ctx = TracingContext()
span = self._tracer.start_span("turn")
span_context = span.get_span_context()
ctx.set_turn_context(span_context)
self.assertIsNotNone(ctx.get_turn_context())
span.end()
def test_clear_turn_context(self):
"""Test clearing turn context by passing None."""
ctx = TracingContext()
span = self._tracer.start_span("turn")
ctx.set_turn_context(span.get_span_context())
self.assertIsNotNone(ctx.get_turn_context())
ctx.set_turn_context(None)
self.assertIsNone(ctx.get_turn_context())
span.end()
def test_generate_conversation_id(self):
"""Test that generated conversation IDs are unique UUIDs."""
id1 = TracingContext.generate_conversation_id()
id2 = TracingContext.generate_conversation_id()
self.assertIsInstance(id1, str)
self.assertNotEqual(id1, id2)
def test_instances_are_isolated(self):
"""Test that two TracingContext instances do not share state."""
ctx_a = TracingContext()
ctx_b = TracingContext()
span = self._tracer.start_span("turn")
ctx_a.set_turn_context(span.get_span_context())
ctx_a.set_conversation_context(span.get_span_context(), "conv-a")
# ctx_b should still be empty
self.assertIsNone(ctx_b.get_turn_context())
self.assertIsNone(ctx_b.get_conversation_context())
self.assertIsNone(ctx_b.conversation_id)
span.end()
def test_conversation_and_turn_are_independent(self):
"""Test that clearing turn context does not affect conversation context."""
ctx = TracingContext()
conv_span = self._tracer.start_span("conv")
turn_span = self._tracer.start_span("turn")
ctx.set_conversation_context(conv_span.get_span_context(), "conv-1")
ctx.set_turn_context(turn_span.get_span_context())
# Clear turn but conversation should remain
ctx.set_turn_context(None)
self.assertIsNone(ctx.get_turn_context())
self.assertIsNotNone(ctx.get_conversation_context())
self.assertEqual(ctx.conversation_id, "conv-1")
conv_span.end()
turn_span.end()
if __name__ == "__main__":
unittest.main()

505
tests/test_turn_trace_observer.py Normal file
View File

@@ -0,0 +1,505 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import threading
import unittest
try:
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult
HAS_OPENTELEMETRY = True
except ImportError:
HAS_OPENTELEMETRY = False
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
from pipecat.processors.filters.identity_filter import IdentityFilter
from pipecat.tests.utils import SleepFrame, run_test
from pipecat.utils.tracing.tracing_context import TracingContext
from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
if HAS_OPENTELEMETRY:
class _InMemorySpanExporter(SpanExporter):
"""Simple in-memory span exporter for testing."""
def __init__(self):
"""Initialize the exporter."""
self._spans = []
self._lock = threading.Lock()
def export(self, spans):
"""Export spans to memory."""
with self._lock:
self._spans.extend(spans)
return SpanExportResult.SUCCESS
def get_finished_spans(self):
"""Return collected spans."""
with self._lock:
return list(self._spans)
def clear(self):
"""Clear collected spans."""
with self._lock:
self._spans.clear()
@unittest.skipUnless(HAS_OPENTELEMETRY, "opentelemetry not installed")
class TestTurnTraceObserver(unittest.IsolatedAsyncioTestCase):
"""Tests for TurnTraceObserver."""
def setUp(self):
"""Set up a fresh provider and exporter for each test.
We create a dedicated TracerProvider per test and inject its tracer
directly into the observer, avoiding the global provider singleton.
"""
self._exporter = _InMemorySpanExporter()
self._provider = TracerProvider()
self._provider.add_span_processor(SimpleSpanProcessor(self._exporter))
self._tracer = self._provider.get_tracer("pipecat.turn")
def tearDown(self):
"""Shut down the provider to flush spans."""
self._provider.shutdown()
def _create_observers(self, conversation_id=None, tracing_context=None):
"""Create a standard set of turn/trace observers.
Args:
conversation_id: Optional conversation ID.
tracing_context: Optional TracingContext instance.
Returns:
Tuple of (turn_tracker, latency_tracker, trace_observer, tracing_context).
"""
tracing_context = tracing_context or TracingContext()
turn_tracker = TurnTrackingObserver(turn_end_timeout_secs=0.2)
latency_tracker = UserBotLatencyObserver()
trace_observer = TurnTraceObserver(
turn_tracker,
latency_tracker=latency_tracker,
conversation_id=conversation_id,
tracing_context=tracing_context,
)
# Inject the test tracer so spans go to our in-memory exporter
trace_observer._tracer = self._tracer
return turn_tracker, latency_tracker, trace_observer, tracing_context
def _all_observers(self, trace_observer):
"""Return the list of observers needed for run_test."""
return [trace_observer._turn_tracker, trace_observer._latency_tracker, trace_observer]
def _get_spans_by_name(self, name):
"""Return finished spans with the given name."""
return [s for s in self._exporter.get_finished_spans() if s.name == name]
async def test_conversation_span_created_on_start_frame(self):
"""Test that a conversation span is created when StartFrame is observed."""
_, _, trace_observer, _ = self._create_observers(conversation_id="test-conv")
processor = IdentityFilter()
frames_to_send = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=self._all_observers(trace_observer),
)
# End conversation to flush the conversation span (normally done by PipelineTask._cleanup)
trace_observer.end_conversation_tracing()
conv_spans = self._get_spans_by_name("conversation")
self.assertEqual(len(conv_spans), 1)
self.assertEqual(conv_spans[0].attributes["conversation.id"], "test-conv")
self.assertEqual(conv_spans[0].attributes["conversation.type"], "voice")
async def test_turn_spans_created_for_each_turn(self):
"""Test that a turn span is created for each conversation turn."""
_, _, trace_observer, _ = self._create_observers()
processor = IdentityFilter()
frames_to_send = [
# Turn 1
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.05),
# Turn 2
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=self._all_observers(trace_observer),
)
turn_spans = self._get_spans_by_name("turn")
self.assertEqual(len(turn_spans), 2)
turn_numbers = {s.attributes["turn.number"] for s in turn_spans}
self.assertEqual(turn_numbers, {1, 2})
async def test_turn_spans_are_children_of_conversation(self):
"""Test that turn spans are parented under the conversation span."""
_, _, trace_observer, _ = self._create_observers()
processor = IdentityFilter()
frames_to_send = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=self._all_observers(trace_observer),
)
# End conversation to flush the conversation span
trace_observer.end_conversation_tracing()
conv_spans = self._get_spans_by_name("conversation")
turn_spans = self._get_spans_by_name("turn")
self.assertEqual(len(conv_spans), 1)
self.assertEqual(len(turn_spans), 1)
# Turn span's parent should be the conversation span
conv_span_id = conv_spans[0].context.span_id
turn_parent_id = turn_spans[0].parent.span_id
self.assertEqual(turn_parent_id, conv_span_id)
async def test_interrupted_turn_marked(self):
"""Test that an interrupted turn span has was_interrupted=True."""
_, _, trace_observer, _ = self._create_observers()
processor = IdentityFilter()
frames_to_send = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
# User interrupts
UserStartedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
UserStartedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=self._all_observers(trace_observer),
)
# End conversation to flush remaining spans
trace_observer.end_conversation_tracing()
turn_spans = self._get_spans_by_name("turn")
self.assertGreaterEqual(len(turn_spans), 1)
# First turn should be interrupted
interrupted_turns = [s for s in turn_spans if s.attributes.get("turn.was_interrupted")]
self.assertGreaterEqual(len(interrupted_turns), 1)
async def test_tracing_context_updated_during_turn(self):
"""Test that TracingContext is populated during a turn and cleared after."""
tracing_ctx = TracingContext()
_, _, trace_observer, _ = self._create_observers(tracing_context=tracing_ctx)
processor = IdentityFilter()
frames_to_send = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=self._all_observers(trace_observer),
)
# After the turn ends, turn context should be cleared
self.assertIsNone(tracing_ctx.get_turn_context())
async def test_tracing_context_cleared_after_conversation_end(self):
"""Test that TracingContext is cleared when conversation tracing ends."""
tracing_ctx = TracingContext()
_, _, trace_observer, _ = self._create_observers(tracing_context=tracing_ctx)
processor = IdentityFilter()
frames_to_send = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=self._all_observers(trace_observer),
)
# Manually end conversation tracing (as PipelineTask._cleanup does)
trace_observer.end_conversation_tracing()
self.assertIsNone(tracing_ctx.get_conversation_context())
self.assertIsNone(tracing_ctx.get_turn_context())
self.assertIsNone(tracing_ctx.conversation_id)
async def test_additional_span_attributes(self):
"""Test that additional span attributes are added to the conversation span."""
extra_attrs = {"deployment.id": "abc-123", "customer.tier": "premium"}
tracing_ctx = TracingContext()
turn_tracker = TurnTrackingObserver(turn_end_timeout_secs=0.2)
latency_tracker = UserBotLatencyObserver()
trace_observer = TurnTraceObserver(
turn_tracker,
latency_tracker=latency_tracker,
additional_span_attributes=extra_attrs,
tracing_context=tracing_ctx,
)
trace_observer._tracer = self._tracer
processor = IdentityFilter()
frames_to_send = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=[turn_tracker, latency_tracker, trace_observer],
)
# End conversation to flush the conversation span
trace_observer.end_conversation_tracing()
conv_spans = self._get_spans_by_name("conversation")
self.assertEqual(len(conv_spans), 1)
self.assertEqual(conv_spans[0].attributes["deployment.id"], "abc-123")
self.assertEqual(conv_spans[0].attributes["customer.tier"], "premium")
async def test_concurrent_pipelines_are_isolated(self):
"""Test that two pipelines with separate TracingContexts don't interfere."""
tracing_ctx_a = TracingContext()
tracing_ctx_b = TracingContext()
_, _, trace_observer_a, _ = self._create_observers(
conversation_id="conv-a", tracing_context=tracing_ctx_a
)
_, _, trace_observer_b, _ = self._create_observers(
conversation_id="conv-b", tracing_context=tracing_ctx_b
)
processor_a = IdentityFilter()
processor_b = IdentityFilter()
frames = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
# Run both pipelines concurrently
await asyncio.gather(
run_test(
processor_a,
frames_to_send=frames,
expected_down_frames=expected,
observers=self._all_observers(trace_observer_a),
),
run_test(
processor_b,
frames_to_send=frames,
expected_down_frames=expected,
observers=self._all_observers(trace_observer_b),
),
)
# End both conversations to flush spans
trace_observer_a.end_conversation_tracing()
trace_observer_b.end_conversation_tracing()
# Each TracingContext should have its own conversation ID
conv_spans = self._get_spans_by_name("conversation")
conv_ids = {s.attributes["conversation.id"] for s in conv_spans}
self.assertEqual(conv_ids, {"conv-a", "conv-b"})
# Turn spans should be children of their own conversation span, not cross-linked
turn_spans = self._get_spans_by_name("turn")
conv_span_map = {s.context.span_id: s.attributes["conversation.id"] for s in conv_spans}
for turn_span in turn_spans:
parent_id = turn_span.parent.span_id
turn_conv_id = turn_span.attributes["conversation.id"]
parent_conv_id = conv_span_map[parent_id]
self.assertEqual(
turn_conv_id,
parent_conv_id,
f"Turn span for {turn_conv_id} parented under {parent_conv_id}",
)
async def test_end_conversation_closes_active_turn(self):
"""Test that end_conversation_tracing closes any active turn span."""
_, _, trace_observer, _ = self._create_observers()
# Manually start conversation and a turn
trace_observer.start_conversation_tracing("conv-end-test")
await trace_observer._handle_turn_started(1)
self.assertIsNotNone(trace_observer._current_span)
self.assertIsNotNone(trace_observer._conversation_span)
# End conversation — should close both turn and conversation
trace_observer.end_conversation_tracing()
self.assertIsNone(trace_observer._current_span)
self.assertIsNone(trace_observer._conversation_span)
# Check span attributes
turn_spans = self._get_spans_by_name("turn")
self.assertEqual(len(turn_spans), 1)
self.assertTrue(turn_spans[0].attributes["turn.was_interrupted"])
self.assertTrue(turn_spans[0].attributes["turn.ended_by_conversation_end"])
async def test_conversation_id_auto_generated(self):
"""Test that a conversation ID is auto-generated when none is provided."""
_, _, trace_observer, _ = self._create_observers(conversation_id=None)
processor = IdentityFilter()
frames_to_send = [
UserStartedSpeakingFrame(),
UserStoppedSpeakingFrame(),
BotStartedSpeakingFrame(),
BotStoppedSpeakingFrame(),
SleepFrame(sleep=0.4),
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
]
await run_test(
processor,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
observers=self._all_observers(trace_observer),
)
# End conversation to flush the conversation span
trace_observer.end_conversation_tracing()
conv_spans = self._get_spans_by_name("conversation")
self.assertEqual(len(conv_spans), 1)
# Should have an auto-generated UUID as conversation.id
conv_id = conv_spans[0].attributes["conversation.id"]
self.assertIsNotNone(conv_id)
self.assertGreater(len(conv_id), 0)
if __name__ == "__main__":
unittest.main()