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

Add ClientConnectedFrame and transport readiness timing

Browse Source 
Introduce ClientConnectedFrame (SystemFrame) pushed by all transports
when a client connects. StartupTimingObserver uses this to measure
transport readiness — the time from StartFrame to first client
connection — via a new on_transport_readiness_measured event.
This commit is contained in:
Mark Backman
2026-03-01 08:45:59 -05:00
parent 58aa8e1ba5
commit 0836066898
11 changed files with 172 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
examples/foundational/29-turn-tracking-observer.py
View File

@@ -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:

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

@@ -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.

80
src/pipecat/observers/startup_timing_observer.py
View File

@@ -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)

3
src/pipecat/transports/daily/transport.py
View File

@@ -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."""

3
src/pipecat/transports/heygen/transport.py
View File

@@ -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."""

3
src/pipecat/transports/livekit/transport.py
View File

@@ -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."""

3
src/pipecat/transports/smallwebrtc/transport.py
View File

@@ -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."""

3
src/pipecat/transports/tavus/transport.py
View File

@@ -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."""

2
src/pipecat/transports/websocket/fastapi.py
View File

@@ -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)

4
src/pipecat/transports/websocket/server.py
View File

@@ -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")

76
tests/test_startup_timing_observer.py
View File

@@ -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()