Merge branch 'main' into filipi/deepgram

# Conflicts:
#	src/pipecat/services/deepgram/stt_sagemaker.py

tests/test_context_aggregators.py

@@ -21,7 +21,6 @@ from pipecat.frames.frames import (
     FunctionCallResultProperties,
     InterimTranscriptionFrame,
     InterruptionFrame,
-    InterruptionTaskFrame,
     LLMContextAssistantTimestampFrame,
     LLMContextFrame,
     LLMFullResponseEndFrame,
@@ -567,7 +566,7 @@ class BaseTestUserContextAggregator:
             SleepFrame(),
             UserStoppedSpeakingFrame(),
         ]
-        expected_up_frames = [InterruptionTaskFrame]
+        expected_up_frames = [InterruptionFrame]
         expected_down_frames = [
             BotStartedSpeakingFrame,
             UserStartedSpeakingFrame,

tests/test_frame_processor.py

@@ -9,8 +9,6 @@ import unittest
 from dataclasses import dataclass, field
 from typing import List
 
-from loguru import logger
-
 from pipecat.frames.frames import (
     DataFrame,
     EndFrame,
@@ -85,50 +83,38 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
         assert before_push_called
         assert after_push_called
 
-    async def test_interruption_and_wait(self):
-        class DelayFrameProcessor(FrameProcessor):
-            """This processors just gives time to the event loop to change
-            between tasks. Otherwise things happen to fast."""
-
-            async def process_frame(self, frame: Frame, direction: FrameDirection):
-                await super().process_frame(frame, direction)
-                await asyncio.sleep(0.1)
-                await self.push_frame(frame, direction)
+    async def test_broadcast_interruption(self):
+        """Test that broadcast_interruption() pushes InterruptionFrame both
+        directions and allows subsequent code to run."""
 
         class InterruptFrameProcessor(FrameProcessor):
             async def process_frame(self, frame: Frame, direction: FrameDirection):
                 await super().process_frame(frame, direction)
 
                 if isinstance(frame, TextFrame):
-                    await self.push_interruption_task_frame_and_wait()
+                    await self.broadcast_interruption()
                     await self.push_frame(OutputTransportMessageUrgentFrame(message=frame.text))
                 else:
                     await self.push_frame(frame, direction)
 
-        pipeline = Pipeline([DelayFrameProcessor(), InterruptFrameProcessor()])
+        pipeline = Pipeline([InterruptFrameProcessor()])
 
         frames_to_send = [
-            # Just a random interruption to make sure we don't clear anything
-            # before the actual `InterruptionTaskFrame` interruption.
-            InterruptionFrame(),
-            # This will generate an `InterruptionTaskFrame` and will wait for an
-            # `InterruptionFrame`.
             TextFrame(text="Hello from Pipecat!"),
-            # Just give time for everything to complete.
             SleepFrame(sleep=0.5),
-            EndFrame(),
         ]
         expected_down_frames = [
-            InterruptionFrame,
             InterruptionFrame,
             OutputTransportMessageUrgentFrame,
-            EndFrame,
+        ]
+        expected_up_frames = [
+            InterruptionFrame,
         ]
         await run_test(
             pipeline,
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
-            send_end_frame=False,
+            expected_up_frames=expected_up_frames,
         )
 
     async def test_interruptible_frames(self):
@@ -454,33 +440,20 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
         stop_frames = [f for f in received_frames if isinstance(f, StopFrame)]
         self.assertEqual(len(stop_frames), 1, "StopFrame should survive interruption")
 
-    async def test_interruption_frame_complete_sets_event(self):
-        """Test that InterruptionFrame.complete() sets the event."""
-        event = asyncio.Event()
-        frame = InterruptionFrame(event=event)
-        self.assertFalse(event.is_set())
-        frame.complete()
-        self.assertTrue(event.is_set())
-
-    async def test_interruption_frame_complete_without_event(self):
-        """Test that InterruptionFrame.complete() is safe without an event."""
-        frame = InterruptionFrame()
-        frame.complete()  # Should not raise
-
-    async def test_interruption_event_set_at_pipeline_sink(self):
-        """Test that the event from push_interruption_task_frame_and_wait()
-        is set when the InterruptionFrame reaches the pipeline sink."""
-        event_was_set = False
+    async def test_broadcast_interruption_allows_subsequent_code(self):
+        """Test that broadcast_interruption() returns immediately, allowing the
+        caller to run code afterwards (e.g. push an urgent frame)."""
+        code_after_ran = False
 
         class InterruptOnTextProcessor(FrameProcessor):
             async def process_frame(self, frame: Frame, direction: FrameDirection):
-                nonlocal event_was_set
+                nonlocal code_after_ran
 
                 await super().process_frame(frame, direction)
                 if isinstance(frame, TextFrame):
-                    await self.push_interruption_task_frame_and_wait()
+                    await self.broadcast_interruption()
 
-                    event_was_set = True
+                    code_after_ran = True
                     await self.push_frame(OutputTransportMessageUrgentFrame(message="done"))
                 else:
                     await self.push_frame(frame, direction)
@@ -499,63 +472,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase):
             frames_to_send=frames_to_send,
             expected_down_frames=expected_down_frames,
         )
-        self.assertTrue(event_was_set, "Event should be set after InterruptionFrame completes")
-
-    async def test_interruption_completion_timeout_warning(self):
-        """Test that a warning is logged when an InterruptionFrame is blocked
-        and never reaches the pipeline sink."""
-        warnings = []
-        handler_id = logger.add(
-            lambda msg: warnings.append(str(msg)), level="WARNING", format="{message}"
-        )
-
-        try:
-
-            class BlockInterruptionProcessor(FrameProcessor):
-                """Blocks InterruptionFrames, completing them after a delay."""
-
-                async def process_frame(self, frame: Frame, direction: FrameDirection):
-                    await super().process_frame(frame, direction)
-                    if isinstance(frame, InterruptionFrame):
-                        # Complete after the timeout so the warning fires
-                        # but the test doesn't hang.
-                        async def delayed_complete():
-                            await asyncio.sleep(1.0)
-                            frame.complete()
-
-                        asyncio.create_task(delayed_complete())
-                        return
-                    await self.push_frame(frame, direction)
-
-            class InterruptOnTextProcessor(FrameProcessor):
-                async def process_frame(self, frame: Frame, direction: FrameDirection):
-                    await super().process_frame(frame, direction)
-                    if isinstance(frame, TextFrame):
-                        await self.push_interruption_task_frame_and_wait(timeout=0.5)
-                        await self.push_frame(OutputTransportMessageUrgentFrame(message="done"))
-                    else:
-                        await self.push_frame(frame, direction)
-
-            pipeline = Pipeline([BlockInterruptionProcessor(), InterruptOnTextProcessor()])
-
-            frames_to_send = [
-                TextFrame(text="trigger"),
-            ]
-            expected_down_frames = [
-                OutputTransportMessageUrgentFrame,
-            ]
-            await run_test(
-                pipeline,
-                frames_to_send=frames_to_send,
-                expected_down_frames=expected_down_frames,
-            )
-        finally:
-            logger.remove(handler_id)
-
-        self.assertTrue(
-            any("InterruptionFrame has not completed" in w for w in warnings),
-            "Expected a timeout warning about InterruptionFrame not completing",
-        )
+        self.assertTrue(code_after_ran, "Code after broadcast_interruption() should execute")
 
 
 if __name__ == "__main__":

tests/test_startup_timing_observer.py

@@ -0,0 +1,337 @@
+import asyncio
+import unittest
+
+from pipecat.frames.frames import (
+    BotConnectedFrame,
+    ClientConnectedFrame,
+    Frame,
+    StartFrame,
+    TextFrame,
+)
+from pipecat.observers.startup_timing_observer import (
+    StartupTimingObserver,
+    StartupTimingReport,
+    TransportTimingReport,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.tests.utils import run_test
+
+
+class SlowStartProcessor(FrameProcessor):
+    """A processor that sleeps during start to simulate slow initialization."""
+
+    def __init__(self, delay: float = 0.1, **kwargs):
+        super().__init__(**kwargs)
+        self._delay = delay
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if isinstance(frame, StartFrame):
+            await asyncio.sleep(self._delay)
+        await self.push_frame(frame, direction)
+
+
+class FastProcessor(FrameProcessor):
+    """A processor with no start delay."""
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        await self.push_frame(frame, direction)
+
+
+class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase):
+    """Tests for StartupTimingObserver."""
+
+    async def test_timing_reported(self):
+        """Test that startup timing is measured and reported."""
+        observer = StartupTimingObserver()
+        processor = SlowStartProcessor(delay=0.1)
+
+        reports = []
+
+        @observer.event_handler("on_startup_timing_report")
+        async def on_report(obs, report):
+            reports.append(report)
+
+        frames_to_send = [TextFrame(text="hello")]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(reports), 1)
+        report = reports[0]
+        self.assertGreater(report.total_duration_secs, 0)
+        self.assertGreater(len(report.processor_timings), 0)
+
+        # Find our slow processor in the timings.
+        slow_timings = [
+            t for t in report.processor_timings if "SlowStartProcessor" in t.processor_name
+        ]
+        self.assertEqual(len(slow_timings), 1)
+        self.assertGreaterEqual(slow_timings[0].duration_secs, 0.05)
+
+    async def test_processor_types_filter(self):
+        """Test that processor_types filter limits which processors appear."""
+        observer = StartupTimingObserver(processor_types=(SlowStartProcessor,))
+        processor = SlowStartProcessor(delay=0.05)
+
+        reports = []
+
+        @observer.event_handler("on_startup_timing_report")
+        async def on_report(obs, report):
+            reports.append(report)
+
+        frames_to_send = [TextFrame(text="hello")]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(reports), 1)
+        report = reports[0]
+
+        # Only SlowStartProcessor should be in the timings.
+        for t in report.processor_timings:
+            self.assertIn("SlowStartProcessor", t.processor_name)
+
+    async def test_report_emits_once(self):
+        """Test that the report is emitted only once even with multiple frames."""
+        observer = StartupTimingObserver()
+        processor = FastProcessor()
+
+        reports = []
+
+        @observer.event_handler("on_startup_timing_report")
+        async def on_report(obs, report):
+            reports.append(report)
+
+        frames_to_send = [
+            TextFrame(text="first"),
+            TextFrame(text="second"),
+            TextFrame(text="third"),
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[TextFrame, TextFrame, TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(reports), 1)
+
+    async def test_event_handler_receives_report(self):
+        """Test that the event handler receives a proper StartupTimingReport."""
+        observer = StartupTimingObserver()
+        processor = SlowStartProcessor(delay=0.05)
+
+        reports = []
+
+        @observer.event_handler("on_startup_timing_report")
+        async def on_report(obs, report):
+            reports.append(report)
+
+        frames_to_send = [TextFrame(text="hello")]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(reports), 1)
+        report = reports[0]
+        self.assertIsInstance(report, StartupTimingReport)
+        self.assertIsInstance(report.total_duration_secs, float)
+        self.assertGreater(report.start_time, 0)
+        for timing in report.processor_timings:
+            self.assertIsInstance(timing.processor_name, str)
+            self.assertIsInstance(timing.duration_secs, float)
+            self.assertGreaterEqual(timing.start_offset_secs, 0)
+
+    async def test_excludes_internal_processors(self):
+        """Test that internal pipeline processors are excluded by default."""
+        observer = StartupTimingObserver()
+        processor = FastProcessor()
+
+        reports = []
+
+        @observer.event_handler("on_startup_timing_report")
+        async def on_report(obs, report):
+            reports.append(report)
+
+        frames_to_send = [TextFrame(text="hello")]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(reports), 1)
+        report = reports[0]
+
+        # No internal processors (PipelineSource, PipelineSink, Pipeline) in the report.
+        internal_names = ("Pipeline#", "PipelineTask#")
+        for t in report.processor_timings:
+            for prefix in internal_names:
+                self.assertNotIn(
+                    prefix,
+                    t.processor_name,
+                    f"Internal processor {t.processor_name} should be excluded by default",
+                )
+
+    async def test_transport_timing_client_only(self):
+        """Test that ClientConnectedFrame emits on_transport_timing_report."""
+        observer = StartupTimingObserver()
+        processor = FastProcessor()
+
+        transport_reports = []
+
+        @observer.event_handler("on_transport_timing_report")
+        async def on_transport(obs, report):
+            transport_reports.append(report)
+
+        frames_to_send = [ClientConnectedFrame(), TextFrame(text="hello")]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[ClientConnectedFrame, TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(transport_reports), 1)
+        report = transport_reports[0]
+        self.assertIsInstance(report, TransportTimingReport)
+        self.assertGreater(report.start_time, 0)
+        self.assertGreater(report.client_connected_secs, 0)
+        self.assertIsNone(report.bot_connected_secs)
+
+    async def test_transport_timing_only_first_client(self):
+        """Test that only the first ClientConnectedFrame triggers the event."""
+        observer = StartupTimingObserver()
+        processor = FastProcessor()
+
+        transport_reports = []
+
+        @observer.event_handler("on_transport_timing_report")
+        async def on_transport(obs, report):
+            transport_reports.append(report)
+
+        frames_to_send = [
+            ClientConnectedFrame(),
+            ClientConnectedFrame(),
+            TextFrame(text="hello"),
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[ClientConnectedFrame, ClientConnectedFrame, TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(transport_reports), 1)
+
+    async def test_transport_timing_without_start_frame(self):
+        """Test that ClientConnectedFrame before StartFrame does not crash."""
+        observer = StartupTimingObserver()
+
+        # Directly call on_push_frame with a ClientConnectedFrame before any
+        # StartFrame has been seen. This should be a no-op (no crash).
+        from pipecat.observers.base_observer import FramePushed
+
+        processor = FastProcessor()
+        destination = FastProcessor()
+        data = FramePushed(
+            source=processor,
+            destination=destination,
+            frame=ClientConnectedFrame(),
+            direction=FrameDirection.DOWNSTREAM,
+            timestamp=1000,
+        )
+        await observer.on_push_frame(data)
+
+        # No event should have been emitted.
+        self.assertFalse(observer._transport_timing_reported)
+
+    async def test_bot_and_client_connected(self):
+        """Test that BotConnectedFrame timing is included in the transport report."""
+        observer = StartupTimingObserver()
+        processor = FastProcessor()
+
+        transport_reports = []
+
+        @observer.event_handler("on_transport_timing_report")
+        async def on_transport(obs, report):
+            transport_reports.append(report)
+
+        frames_to_send = [
+            BotConnectedFrame(),
+            ClientConnectedFrame(),
+            TextFrame(text="hello"),
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[BotConnectedFrame, ClientConnectedFrame, TextFrame],
+            observers=[observer],
+        )
+
+        self.assertEqual(len(transport_reports), 1)
+        report = transport_reports[0]
+        self.assertGreater(report.client_connected_secs, 0)
+        self.assertIsNotNone(report.bot_connected_secs)
+        self.assertGreater(report.bot_connected_secs, 0)
+
+        # Client connected should be >= bot connected.
+        self.assertGreaterEqual(report.client_connected_secs, report.bot_connected_secs)
+
+    async def test_bot_connected_only_first(self):
+        """Test that only the first BotConnectedFrame is recorded."""
+        observer = StartupTimingObserver()
+        processor = FastProcessor()
+
+        transport_reports = []
+
+        @observer.event_handler("on_transport_timing_report")
+        async def on_transport(obs, report):
+            transport_reports.append(report)
+
+        frames_to_send = [
+            BotConnectedFrame(),
+            BotConnectedFrame(),
+            ClientConnectedFrame(),
+            TextFrame(text="hello"),
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=[
+                BotConnectedFrame,
+                BotConnectedFrame,
+                ClientConnectedFrame,
+                TextFrame,
+            ],
+            observers=[observer],
+        )
+
+        # Only one transport report, with bot timing from first frame.
+        self.assertEqual(len(transport_reports), 1)
+        self.assertIsNotNone(transport_reports[0].bot_connected_secs)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_stt_mute_filter.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import asyncio
 import unittest
 
 from pipecat.frames.frames import (
@@ -329,17 +328,13 @@ class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase):
             expected_down_frames=expected_returned_frames,
         )
 
-    async def test_interruption_frame_completed_when_muted(self):
-        """Test that InterruptionFrame.complete() is called when the frame is
-        suppressed due to muting, so push_interruption_task_frame_and_wait()
-        doesn't hang."""
+    async def test_interruption_frame_suppressed_when_muted(self):
+        """Test that InterruptionFrame is suppressed when the filter is muted."""
         filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.ALWAYS}))
 
-        event = asyncio.Event()
-
         frames_to_send = [
             BotStartedSpeakingFrame(),
-            InterruptionFrame(event=event),
+            InterruptionFrame(),
             BotStoppedSpeakingFrame(),
         ]
 
@@ -354,8 +349,6 @@ class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase):
             expected_down_frames=expected_returned_frames,
         )
 
-        self.assertTrue(event.is_set(), "InterruptionFrame.complete() should be called when muted")
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_user_bot_latency_observer.py

@@ -2,12 +2,28 @@ import unittest
 
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
+    ClientConnectedFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
+    InterruptionFrame,
+    MetricsFrame,
+    UserStoppedSpeakingFrame,
     VADUserStartedSpeakingFrame,
     VADUserStoppedSpeakingFrame,
 )
-from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver
+from pipecat.metrics.metrics import (
+    TextAggregationMetricsData,
+    TTFBMetricsData,
+)
+from pipecat.observers.user_bot_latency_observer import (
+    FunctionCallMetrics,
+    LatencyBreakdown,
+    TextAggregationBreakdownMetrics,
+    TTFBBreakdownMetrics,
+    UserBotLatencyObserver,
+)
 from pipecat.processors.filters.identity_filter import IdentityFilter
-from pipecat.tests.utils import run_test
+from pipecat.tests.utils import SleepFrame, run_test
 
 
 class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase):
@@ -97,22 +113,226 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase):
         self.assertGreater(latencies[0], 0)
         self.assertGreater(latencies[1], 0)
 
-    async def test_no_measurement_without_user_stop(self):
-        """Test that latency is not measured if bot starts without user stopping first."""
-        # Create observer
+    async def test_breakdown_with_metrics(self):
+        """Test that metrics collected between VADUserStopped and BotStarted appear in breakdown."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        breakdowns = []
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        stt_ttfb = TTFBMetricsData(processor="DeepgramSTTService#0", value=0.080)
+        llm_ttfb = TTFBMetricsData(processor="OpenAILLMService#0", model="gpt-4o", value=0.250)
+        tts_ttfb = TTFBMetricsData(processor="CartesiaTTSService#0", value=0.070)
+        text_agg = TextAggregationMetricsData(processor="CartesiaTTSService#0", value=0.030)
+
+        frames_to_send = [
+            VADUserStoppedSpeakingFrame(),
+            MetricsFrame(data=[stt_ttfb]),
+            MetricsFrame(data=[llm_ttfb, text_agg]),
+            MetricsFrame(data=[tts_ttfb]),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            VADUserStoppedSpeakingFrame,
+            MetricsFrame,
+            MetricsFrame,
+            MetricsFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(breakdowns), 1)
+        bd = breakdowns[0]
+        self.assertEqual(len(bd.ttfb), 3)
+        self.assertEqual(bd.ttfb[0].processor, "DeepgramSTTService#0")
+        self.assertEqual(bd.ttfb[1].processor, "OpenAILLMService#0")
+        self.assertEqual(bd.ttfb[2].processor, "CartesiaTTSService#0")
+        self.assertIsNotNone(bd.text_aggregation)
+        self.assertEqual(bd.text_aggregation.duration_secs, 0.030)
+
+    async def test_interruption_resets_accumulators(self):
+        """Test that InterruptionFrame clears stale metrics from earlier cycles."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        breakdowns = []
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        # First cycle metrics (will be interrupted)
+        stale_llm = TTFBMetricsData(processor="OpenAILLMService#0", value=0.245)
+        # Second cycle metrics (the ones that matter)
+        final_llm = TTFBMetricsData(processor="OpenAILLMService#0", value=0.224)
+        final_tts = TTFBMetricsData(processor="CartesiaTTSService#0", value=0.142)
+
+        frames_to_send = [
+            VADUserStoppedSpeakingFrame(),
+            MetricsFrame(data=[stale_llm]),
+            InterruptionFrame(),
+            MetricsFrame(data=[final_llm]),
+            MetricsFrame(data=[final_tts]),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            VADUserStoppedSpeakingFrame,
+            MetricsFrame,
+            InterruptionFrame,
+            MetricsFrame,
+            MetricsFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(breakdowns), 1)
+        bd = breakdowns[0]
+        # Only the post-interruption metrics should be present
+        self.assertEqual(len(bd.ttfb), 2)
+        self.assertEqual(bd.ttfb[0].processor, "OpenAILLMService#0")
+        self.assertEqual(bd.ttfb[0].duration_secs, 0.224)
+        self.assertEqual(bd.ttfb[1].processor, "CartesiaTTSService#0")
+        self.assertEqual(bd.ttfb[1].duration_secs, 0.142)
+
+    async def test_only_first_text_aggregation_kept(self):
+        """Test that only the first text aggregation metric is kept per cycle."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        breakdowns = []
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        text_agg_1 = TextAggregationMetricsData(processor="CartesiaTTSService#0", value=0.030)
+        text_agg_2 = TextAggregationMetricsData(processor="CartesiaTTSService#0", value=0.080)
+
+        frames_to_send = [
+            VADUserStoppedSpeakingFrame(),
+            MetricsFrame(data=[text_agg_1]),
+            MetricsFrame(data=[text_agg_2]),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            VADUserStoppedSpeakingFrame,
+            MetricsFrame,
+            MetricsFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(breakdowns), 1)
+        self.assertIsNotNone(breakdowns[0].text_aggregation)
+        self.assertEqual(breakdowns[0].text_aggregation.duration_secs, 0.030)
+
+    async def test_user_turn_measured(self):
+        """Test that pre-LLM wait from user silence to UserStopped is captured."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        breakdowns = []
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        frames_to_send = [
+            VADUserStoppedSpeakingFrame(),
+            SleepFrame(sleep=0.1),  # Simulate turn analyzer wait
+            UserStoppedSpeakingFrame(),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            VADUserStoppedSpeakingFrame,
+            UserStoppedSpeakingFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(breakdowns), 1)
+        self.assertIsNotNone(breakdowns[0].user_turn_secs)
+        self.assertGreaterEqual(breakdowns[0].user_turn_secs, 0.1)
+
+    async def test_user_turn_none_without_user_stopped(self):
+        """Test that user_turn is None when no UserStoppedSpeakingFrame arrives."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        breakdowns = []
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        frames_to_send = [
+            VADUserStoppedSpeakingFrame(),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            VADUserStoppedSpeakingFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(breakdowns), 1)
+        self.assertIsNone(breakdowns[0].user_turn_secs)
+
+    async def test_no_measurement_without_user_stop(self):
+        """Test that BotStartedSpeaking without prior user stop emits nothing."""
         observer = UserBotLatencyObserver()
-
-        # Create identity filter
         processor = IdentityFilter()
 
-        # Capture latency events
         latencies = []
+        breakdowns = []
 
         @observer.event_handler("on_latency_measured")
         async def on_latency(obs, latency_seconds):
             latencies.append(latency_seconds)
 
-        # Define frame sequence - bot starts without user stop
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
         frames_to_send = [
             BotStartedSpeakingFrame(),
         ]
@@ -121,7 +341,6 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase):
             BotStartedSpeakingFrame,
         ]
 
-        # Run test
         await run_test(
             processor,
             frames_to_send=frames_to_send,
@@ -129,8 +348,283 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase):
             observers=[observer],
         )
 
-        # Verify no latency was measured
         self.assertEqual(len(latencies), 0)
+        self.assertEqual(len(breakdowns), 0)
+
+    async def test_first_bot_speech_latency(self):
+        """Test first bot speech latency and breakdown from ClientConnected to BotStartedSpeaking."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        first_speech_latencies = []
+        breakdowns = []
+
+        @observer.event_handler("on_first_bot_speech_latency")
+        async def on_first_bot_speech(obs, latency_seconds):
+            first_speech_latencies.append(latency_seconds)
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        llm_ttfb = TTFBMetricsData(processor="OpenAILLMService#0", value=0.250)
+        tts_ttfb = TTFBMetricsData(processor="CartesiaTTSService#0", value=0.070)
+
+        frames_to_send = [
+            ClientConnectedFrame(),
+            MetricsFrame(data=[llm_ttfb]),
+            MetricsFrame(data=[tts_ttfb]),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            ClientConnectedFrame,
+            MetricsFrame,
+            MetricsFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(first_speech_latencies), 1)
+        self.assertGreater(first_speech_latencies[0], 0)
+        self.assertLess(first_speech_latencies[0], 1.0)
+
+        # Breakdown should also be emitted with the accumulated metrics
+        self.assertEqual(len(breakdowns), 1)
+        self.assertEqual(len(breakdowns[0].ttfb), 2)
+        self.assertEqual(breakdowns[0].ttfb[0].processor, "OpenAILLMService#0")
+        self.assertEqual(breakdowns[0].ttfb[1].processor, "CartesiaTTSService#0")
+
+    async def test_first_bot_speech_only_once(self):
+        """Test that first bot speech latency is only emitted once."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        first_speech_latencies = []
+
+        @observer.event_handler("on_first_bot_speech_latency")
+        async def on_first_bot_speech(obs, latency_seconds):
+            first_speech_latencies.append(latency_seconds)
+
+        frames_to_send = [
+            ClientConnectedFrame(),
+            BotStartedSpeakingFrame(),
+            # Second bot speech should not trigger the event again
+            VADUserStoppedSpeakingFrame(),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            ClientConnectedFrame,
+            BotStartedSpeakingFrame,
+            VADUserStoppedSpeakingFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(first_speech_latencies), 1)
+
+    async def test_first_bot_speech_skipped_when_user_speaks_first(self):
+        """Test that first bot speech event is not emitted when user speaks before the bot."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        first_speech_latencies = []
+
+        @observer.event_handler("on_first_bot_speech_latency")
+        async def on_first_bot_speech(obs, latency_seconds):
+            first_speech_latencies.append(latency_seconds)
+
+        frames_to_send = [
+            ClientConnectedFrame(),
+            # User speaks before bot has a chance to greet
+            VADUserStartedSpeakingFrame(),
+            VADUserStoppedSpeakingFrame(),
+            BotStartedSpeakingFrame(),
+        ]
+
+        expected_down_frames = [
+            ClientConnectedFrame,
+            VADUserStartedSpeakingFrame,
+            VADUserStoppedSpeakingFrame,
+            BotStartedSpeakingFrame,
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(first_speech_latencies), 0)
+
+    async def test_function_call_latency_in_breakdown(self):
+        """Test that function call duration appears in the latency breakdown."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        breakdowns = []
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        tool_call_id = "call_abc123"
+
+        frames_to_send = [
+            VADUserStoppedSpeakingFrame(),
+            FunctionCallInProgressFrame(
+                function_name="get_weather",
+                tool_call_id=tool_call_id,
+                arguments={"location": "Atlanta"},
+            ),
+            SleepFrame(sleep=0.1),
+            FunctionCallResultFrame(
+                function_name="get_weather",
+                tool_call_id=tool_call_id,
+                arguments={"location": "Atlanta"},
+                result={"temperature": "75"},
+            ),
+            BotStartedSpeakingFrame(),
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(breakdowns), 1)
+        self.assertEqual(len(breakdowns[0].function_calls), 1)
+        fc = breakdowns[0].function_calls[0]
+        self.assertEqual(fc.function_name, "get_weather")
+        self.assertGreaterEqual(fc.duration_secs, 0.1)
+
+    async def test_function_call_reset_on_interruption(self):
+        """Test that function call metrics are cleared on interruption."""
+        observer = UserBotLatencyObserver()
+        processor = IdentityFilter()
+
+        breakdowns = []
+
+        @observer.event_handler("on_latency_breakdown")
+        async def on_breakdown(obs, breakdown):
+            breakdowns.append(breakdown)
+
+        frames_to_send = [
+            VADUserStoppedSpeakingFrame(),
+            FunctionCallInProgressFrame(
+                function_name="get_weather",
+                tool_call_id="call_1",
+                arguments={},
+            ),
+            FunctionCallResultFrame(
+                function_name="get_weather",
+                tool_call_id="call_1",
+                arguments={},
+                result={},
+            ),
+            InterruptionFrame(),
+            BotStartedSpeakingFrame(),
+        ]
+
+        await run_test(
+            processor,
+            frames_to_send=frames_to_send,
+            observers=[observer],
+        )
+
+        self.assertEqual(len(breakdowns), 1)
+        self.assertEqual(len(breakdowns[0].function_calls), 0)
+
+
+class TestLatencyBreakdownChronologicalEvents(unittest.TestCase):
+    """Tests for LatencyBreakdown.chronological_events()."""
+
+    def test_events_sorted_by_start_time(self):
+        """Test that events are returned in chronological order."""
+        breakdown = LatencyBreakdown(
+            user_turn_start_time=100.0,
+            user_turn_secs=0.150,
+            ttfb=[
+                TTFBBreakdownMetrics(
+                    processor="OpenAILLMService#0",
+                    model="gpt-4o",
+                    start_time=100.200,
+                    duration_secs=0.250,
+                ),
+                TTFBBreakdownMetrics(
+                    processor="DeepgramSTTService#0",
+                    start_time=100.050,
+                    duration_secs=0.080,
+                ),
+                TTFBBreakdownMetrics(
+                    processor="CartesiaTTSService#0",
+                    start_time=100.500,
+                    duration_secs=0.070,
+                ),
+            ],
+            function_calls=[
+                FunctionCallMetrics(
+                    function_name="get_weather",
+                    start_time=100.450,
+                    duration_secs=0.120,
+                ),
+            ],
+            text_aggregation=TextAggregationBreakdownMetrics(
+                processor="CartesiaTTSService#0",
+                start_time=100.480,
+                duration_secs=0.030,
+            ),
+        )
+
+        events = breakdown.chronological_events()
+
+        self.assertEqual(len(events), 6)
+        self.assertEqual(events[0], "User turn: 0.150s")
+        self.assertEqual(events[1], "DeepgramSTTService#0: TTFB 0.080s")
+        self.assertEqual(events[2], "OpenAILLMService#0: TTFB 0.250s")
+        self.assertEqual(events[3], "get_weather: 0.120s")
+        self.assertEqual(events[4], "CartesiaTTSService#0: text aggregation 0.030s")
+        self.assertEqual(events[5], "CartesiaTTSService#0: TTFB 0.070s")
+
+    def test_empty_breakdown(self):
+        """Test that an empty breakdown returns no events."""
+        breakdown = LatencyBreakdown()
+        self.assertEqual(breakdown.chronological_events(), [])
+
+    def test_user_turn_requires_both_fields(self):
+        """Test that user turn is only included when both start_time and secs are set."""
+        # Only start_time, no duration
+        breakdown = LatencyBreakdown(user_turn_start_time=100.0)
+        self.assertEqual(breakdown.chronological_events(), [])
+
+        # Only duration, no start_time
+        breakdown = LatencyBreakdown(user_turn_secs=0.150)
+        self.assertEqual(breakdown.chronological_events(), [])
+
+    def test_ttfb_only(self):
+        """Test breakdown with only TTFB metrics."""
+        breakdown = LatencyBreakdown(
+            ttfb=[
+                TTFBBreakdownMetrics(processor="LLM#0", start_time=100.0, duration_secs=0.200),
+            ],
+        )
+        events = breakdown.chronological_events()
+        self.assertEqual(events, ["LLM#0: TTFB 0.200s"])
 
 
 if __name__ == "__main__":