# # Copyright (c) 2024-2026, Daily # # SPDX-License-Identifier: BSD 2-Clause License # import asyncio import threading import unittest from unittest.mock import patch 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, CancelFrame, EndFrame, StartFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) from pipecat.observers.base_observer import FramePushed 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.processors.frame_processor import FrameDirection 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] def _get_turn_spans(self): """Return finished conversation turns, including their numbered names.""" return [s for s in self._exporter.get_finished_spans() if "turn.number" in s.attributes] def test_trace_url_can_be_retrieved_for_persistence(self): _, _, trace_observer, _ = self._create_observers(conversation_id="persisted-call") trace_observer.start_conversation_tracing("persisted-call") trace_id = trace_observer.get_trace_id() self.assertIsNotNone(trace_id) expected_url = f"https://langfuse.example/project/test/traces/{trace_id}" with patch("langfuse.get_client") as get_client: get_client.return_value.get_trace_url.return_value = expected_url self.assertEqual(trace_observer.get_trace_url(), expected_url) get_client.return_value.get_trace_url.assert_called_once_with(trace_id=trace_id) trace_observer.end_conversation_tracing() 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 PipelineWorker._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_turn_spans() self.assertEqual(len(turn_spans), 2) self.assertEqual({s.name for s in turn_spans}, {"turn-1", "turn-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_turn_spans() 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_turn_spans() 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 PipelineWorker._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_turn_spans() 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_terminal_frames_preserve_parent_for_downstream_service_spans(self): """Early terminal notifications must preserve parentage until cleanup.""" for terminal_frame_type in (EndFrame, CancelFrame): with self.subTest(frame=terminal_frame_type.__name__): self._exporter.clear() turn_tracker, _, trace_observer, tracing_context = self._create_observers() processor = IdentityFilter() async def notify_tracker(frame, tracker=turn_tracker, source=processor): await tracker.on_push_frame( FramePushed( source=source, destination=source, frame=frame, direction=FrameDirection.DOWNSTREAM, timestamp=0, ) ) await notify_tracker(StartFrame()) await notify_tracker(BotStartedSpeakingFrame()) await notify_tracker(BotStoppedSpeakingFrame()) turn_span = trace_observer._current_span turn_context = tracing_context.get_turn_context() end_timer = turn_tracker._end_turn_timer self.assertIsNotNone(turn_span) self.assertIsNotNone(turn_context) self.assertIsNotNone(end_timer) try: await notify_tracker(terminal_frame_type()) self.assertTrue(end_timer.cancelled()) self.assertIsNone(turn_tracker._end_turn_timer) self.assertIs(trace_observer._current_span, turn_span) self.assertTrue(turn_span.is_recording()) self.assertIs(tracing_context.get_turn_context(), turn_context) with self._tracer.start_as_current_span( "downstream-service", context=tracing_context.get_turn_context() ): pass service_span = self._get_spans_by_name("downstream-service")[0] self.assertEqual( service_span.parent.span_id, turn_span.get_span_context().span_id ) finally: trace_observer.end_conversation_tracing() self.assertFalse(turn_span.is_recording()) self.assertIsNone(tracing_context.get_turn_context()) self.assertEqual(len(self._get_turn_spans()), 1) 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_turn_spans() 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()