diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 3e85ddfb8..ad0b448e9 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -111,6 +111,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): for timing in report.processor_timings: logger.info(f" {timing.processor_name}: {timing.duration_secs:.3f}s") + @startup_observer.event_handler("on_transport_readiness_measured") + async def on_transport_readiness_measured(observer, report): + logger.info(f"Transport readiness: {report.readiness_secs:.3f}s") + turn_observer = task.turn_tracking_observer if turn_observer: diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 126f3c001..b5e368c53 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1910,6 +1910,17 @@ class StopFrame(ControlFrame, UninterruptibleFrame): pass +@dataclass +class ClientConnectedFrame(SystemFrame): + """Frame indicating that a client has connected to the transport. + + Pushed downstream by the input transport when a client (participant) + connects. Used by observers to measure transport readiness timing. + """ + + pass + + @dataclass class OutputTransportReadyFrame(ControlFrame): """Frame indicating that the output transport is ready. diff --git a/src/pipecat/observers/startup_timing_observer.py b/src/pipecat/observers/startup_timing_observer.py index 0f3ad0b7a..d6b1c8fa9 100644 --- a/src/pipecat/observers/startup_timing_observer.py +++ b/src/pipecat/observers/startup_timing_observer.py @@ -12,6 +12,10 @@ when a ``StartFrame`` arrives at a processor (``on_process_frame``) versus when it leaves (``on_push_frame``), giving the exact ``start()`` duration for each processor in the pipeline. +It also measures transport readiness — the time from ``StartFrame`` to the +first ``ClientConnectedFrame`` — via a separate ``on_transport_readiness_measured`` +event. + Example:: observer = StartupTimingObserver() @@ -21,6 +25,10 @@ Example:: for t in report.processor_timings: print(f"{t.processor_name}: {t.duration_secs:.3f}s") + @observer.event_handler("on_transport_readiness_measured") + async def on_readiness(observer, report): + print(f"Transport ready in {report.readiness_secs:.3f}s") + task = PipelineTask(pipeline, observers=[observer]) """ @@ -29,11 +37,11 @@ from typing import Dict, List, Optional, Tuple, Type from loguru import logger -from pipecat.frames.frames import StartFrame +from pipecat.frames.frames import ClientConnectedFrame, StartFrame from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.pipeline.pipeline import PipelineSink, PipelineSource -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.frame_processor import FrameProcessor # Internal pipeline types excluded from tracking by default. _INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline) @@ -65,6 +73,17 @@ class StartupTimingReport: processor_timings: List[ProcessorStartupTiming] = field(default_factory=list) +@dataclass +class TransportReadinessReport: + """Time from pipeline start to first client connection. + + Parameters: + readiness_secs: Seconds from StartFrame to first ClientConnectedFrame. + """ + + readiness_secs: float + + class StartupTimingObserver(BaseObserver): """Observer that measures processor startup times during pipeline initialization. @@ -73,6 +92,10 @@ class StartupTimingObserver(BaseObserver): pushed downstream. This captures WebSocket connections, API authentication, model loading, and other initialization work. + Also measures transport readiness — the time from ``StartFrame`` to the + first ``ClientConnectedFrame`` — indicating how long it takes for a client + to connect after the pipeline starts. + By default, internal pipeline processors (``PipelineSource``, ``PipelineSink``, ``Pipeline``) are excluded from the report. Pass ``processor_types`` to measure only specific types. @@ -81,6 +104,8 @@ class StartupTimingObserver(BaseObserver): - on_startup_timing_report: Called once after startup completes with the full timing report. + - on_transport_readiness_measured: Called once when the first client connects with the + transport readiness timing. Example:: @@ -93,6 +118,10 @@ class StartupTimingObserver(BaseObserver): for t in report.processor_timings: logger.info(f"{t.processor_name}: {t.duration_secs:.3f}s") + @observer.event_handler("on_transport_readiness_measured") + async def on_readiness(observer, report): + logger.info(f"Transport ready in {report.readiness_secs:.3f}s") + task = PipelineTask(pipeline, observers=[observer]) Args: @@ -125,10 +154,17 @@ class StartupTimingObserver(BaseObserver): # Lock onto the first StartFrame we see (by frame ID). self._start_frame_id: Optional[str] = None - # Whether we've already emitted the report. - self._reported = False + # Whether we've already emitted the startup timing report. + self._startup_timing_reported = False + + # Whether we've already measured transport readiness. + self._transport_readiness_measured = False + + # Timestamp (ns) when we first see a StartFrame arrive at a processor. + self._start_frame_arrival_ns: Optional[int] = None self._register_event_handler("on_startup_timing_report") + self._register_event_handler("on_transport_readiness_measured") def _should_track(self, processor: FrameProcessor) -> bool: """Check if a processor should be tracked for timing. @@ -153,18 +189,16 @@ class StartupTimingObserver(BaseObserver): Args: data: The frame processing event data. """ - if self._reported: + if self._startup_timing_reported: return if not isinstance(data.frame, StartFrame): return - if data.direction != FrameDirection.DOWNSTREAM: - return - # Lock onto the first StartFrame. if self._start_frame_id is None: self._start_frame_id = data.frame.id + self._start_frame_arrival_ns = data.timestamp elif data.frame.id != self._start_frame_id: return @@ -182,18 +216,21 @@ class StartupTimingObserver(BaseObserver): async def on_push_frame(self, data: FramePushed): """Record when a StartFrame leaves a processor and compute the delta. + Also handles ``ClientConnectedFrame`` to measure transport readiness. + Args: data: The frame push event data. """ - if self._reported: + if isinstance(data.frame, ClientConnectedFrame): + await self._handle_client_connected(data) + return + + if self._startup_timing_reported: return if not isinstance(data.frame, StartFrame): return - if data.direction != FrameDirection.DOWNSTREAM: - return - if self._start_frame_id is not None and data.frame.id != self._start_frame_id: return @@ -212,11 +249,22 @@ class StartupTimingObserver(BaseObserver): ) ) + async def _handle_client_connected(self, data: FramePushed): + """Measure transport readiness on first client connection.""" + if self._transport_readiness_measured or self._start_frame_arrival_ns is None: + return + + self._transport_readiness_measured = True + delta_ns = data.timestamp - self._start_frame_arrival_ns + readiness_secs = delta_ns / 1e9 + report = TransportReadinessReport(readiness_secs=readiness_secs) + await self._call_event_handler("on_transport_readiness_measured", report) + async def _emit_report(self): """Build and emit the startup timing report.""" - if self._reported: + if self._startup_timing_reported: return - self._reported = True + self._startup_timing_reported = True total = sum(t.duration_secs for t in self._timings) @@ -225,8 +273,4 @@ class StartupTimingObserver(BaseObserver): processor_timings=self._timings, ) - logger.debug(f"Pipeline startup completed in {total:.3f}s") - for t in self._timings: - logger.debug(f" {t.processor_name}: {t.duration_secs:.3f}s") - await self._call_event_handler("on_startup_timing_report", report) diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 9575fd51b..cb24b23fa 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -25,6 +25,7 @@ from pydantic import BaseModel from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams from pipecat.frames.frames import ( CancelFrame, + ClientConnectedFrame, DataFrame, EndFrame, Frame, @@ -2716,6 +2717,8 @@ class DailyTransport(BaseTransport): await self._call_event_handler("on_participant_joined", participant) # Also call on_client_connected for compatibility with other transports await self._call_event_handler("on_client_connected", participant) + if self._input: + await self._input.push_frame(ClientConnectedFrame()) async def _on_participant_left(self, participant, reason): """Handle participant left events.""" diff --git a/src/pipecat/transports/heygen/transport.py b/src/pipecat/transports/heygen/transport.py index dbeded3e5..77ccda09f 100644 --- a/src/pipecat/transports/heygen/transport.py +++ b/src/pipecat/transports/heygen/transport.py @@ -26,6 +26,7 @@ from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, + ClientConnectedFrame, EndFrame, Frame, InputAudioRawFrame, @@ -387,6 +388,8 @@ class HeyGenTransport(BaseTransport): async def _on_client_connected(self, participant: Any): """Handle client connected events.""" await self._call_event_handler("on_client_connected", participant) + if self._input: + await self._input.push_frame(ClientConnectedFrame()) async def _on_client_disconnected(self, participant: Any): """Handle client disconnected events.""" diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index 1902e7cd3..e4435016c 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -24,6 +24,7 @@ from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.frames.frames import ( AudioRawFrame, CancelFrame, + ClientConnectedFrame, EndFrame, ImageRawFrame, OutputAudioRawFrame, @@ -1143,6 +1144,8 @@ class LiveKitTransport(BaseTransport): async def _on_participant_connected(self, participant_id: str): """Handle participant connected events.""" await self._call_event_handler("on_participant_connected", participant_id) + if self._input: + await self._input.push_frame(ClientConnectedFrame()) async def _on_participant_disconnected(self, participant_id: str): """Handle participant disconnected events.""" diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index dc91588a3..36f883278 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -23,6 +23,7 @@ from pydantic import BaseModel from pipecat.frames.frames import ( CancelFrame, + ClientConnectedFrame, EndFrame, Frame, InputAudioRawFrame, @@ -964,6 +965,8 @@ class SmallWebRTCTransport(BaseTransport): async def _on_client_connected(self, webrtc_connection): """Handle client connection events.""" await self._call_event_handler("on_client_connected", webrtc_connection) + if self._input: + await self._input.push_frame(ClientConnectedFrame()) async def _on_client_disconnected(self, webrtc_connection): """Handle client disconnection events.""" diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index dd63cb790..114f33ca0 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -22,6 +22,7 @@ from pydantic import BaseModel from pipecat.frames.frames import ( CancelFrame, + ClientConnectedFrame, EndFrame, Frame, InputAudioRawFrame, @@ -786,6 +787,8 @@ class TavusTransport(BaseTransport): async def _on_client_connected(self, participant: Any): """Handle client connected events.""" await self._call_event_handler("on_client_connected", participant) + if self._input: + await self._input.push_frame(ClientConnectedFrame()) async def _on_client_disconnected(self, participant: Any): """Handle client disconnected events.""" diff --git a/src/pipecat/transports/websocket/fastapi.py b/src/pipecat/transports/websocket/fastapi.py index f52123e52..0fde2b9ae 100644 --- a/src/pipecat/transports/websocket/fastapi.py +++ b/src/pipecat/transports/websocket/fastapi.py @@ -23,6 +23,7 @@ from pydantic import BaseModel from pipecat.frames.frames import ( CancelFrame, + ClientConnectedFrame, EndFrame, Frame, InputAudioRawFrame, @@ -260,6 +261,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): if not self._monitor_websocket_task and self._params.session_timeout: self._monitor_websocket_task = self.create_task(self._monitor_websocket()) await self._client.trigger_client_connected() + await self.push_frame(ClientConnectedFrame()) if not self._receive_task: self._receive_task = self.create_task(self._receive_messages()) await self.set_transport_ready(frame) diff --git a/src/pipecat/transports/websocket/server.py b/src/pipecat/transports/websocket/server.py index e5f628fa4..fa3645d37 100644 --- a/src/pipecat/transports/websocket/server.py +++ b/src/pipecat/transports/websocket/server.py @@ -22,11 +22,11 @@ from pydantic import BaseModel from pipecat.frames.frames import ( CancelFrame, + ClientConnectedFrame, EndFrame, Frame, InputAudioRawFrame, InputTransportMessageFrame, - InputTransportMessageUrgentFrame, InterruptionFrame, OutputAudioRawFrame, OutputTransportMessageFrame, @@ -504,6 +504,8 @@ class WebsocketServerTransport(BaseTransport): if self._output: await self._output.set_client_connection(websocket) await self._call_event_handler("on_client_connected", websocket) + if self._input: + await self._input.push_frame(ClientConnectedFrame()) else: logger.error("A WebsocketServerTransport output is missing in the pipeline") diff --git a/tests/test_startup_timing_observer.py b/tests/test_startup_timing_observer.py index e3cd7c2b7..efabf5bc7 100644 --- a/tests/test_startup_timing_observer.py +++ b/tests/test_startup_timing_observer.py @@ -1,10 +1,11 @@ import asyncio import unittest -from pipecat.frames.frames import Frame, StartFrame, TextFrame +from pipecat.frames.frames import ClientConnectedFrame, Frame, StartFrame, TextFrame from pipecat.observers.startup_timing_observer import ( StartupTimingObserver, StartupTimingReport, + TransportReadinessReport, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.tests.utils import run_test @@ -181,6 +182,79 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase): f"Internal processor {t.processor_name} should be excluded by default", ) + async def test_transport_readiness_measured(self): + """Test that ClientConnectedFrame after startup emits on_transport_readiness_measured.""" + observer = StartupTimingObserver() + processor = FastProcessor() + + readiness_reports = [] + + @observer.event_handler("on_transport_readiness_measured") + async def on_readiness(obs, report): + readiness_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(readiness_reports), 1) + report = readiness_reports[0] + self.assertIsInstance(report, TransportReadinessReport) + self.assertGreater(report.readiness_secs, 0) + + async def test_transport_readiness_only_first(self): + """Test that only the first ClientConnectedFrame triggers the event.""" + observer = StartupTimingObserver() + processor = FastProcessor() + + readiness_reports = [] + + @observer.event_handler("on_transport_readiness_measured") + async def on_readiness(obs, report): + readiness_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(readiness_reports), 1) + + async def test_transport_readiness_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_readiness_measured) + if __name__ == "__main__": unittest.main()