From 68e8732e72ba5f20ec3b03129f97d4fa2271b190 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 2 Mar 2026 10:41:05 -0500 Subject: [PATCH] Add BotConnectedFrame and on_transport_timing_report event Add BotConnectedFrame (SystemFrame) pushed by SFU transports (Daily, LiveKit, HeyGen, Tavus) when the bot joins the room. Replace the on_transport_readiness_measured event with on_transport_timing_report which includes both bot_connected_secs and client_connected_secs. --- changelog/3881.added.3.md | 1 + .../foundational/29-turn-tracking-observer.py | 8 +- src/pipecat/frames/frames.py | 12 ++ .../observers/startup_timing_observer.py | 93 +++++++++----- src/pipecat/services/heygen/client.py | 7 +- src/pipecat/services/heygen/video.py | 5 + src/pipecat/services/tavus/video.py | 5 + src/pipecat/transports/daily/transport.py | 3 + src/pipecat/transports/heygen/transport.py | 9 ++ src/pipecat/transports/livekit/transport.py | 3 + src/pipecat/transports/tavus/transport.py | 12 ++ tests/test_startup_timing_observer.py | 114 +++++++++++++++--- 12 files changed, 215 insertions(+), 57 deletions(-) create mode 100644 changelog/3881.added.3.md diff --git a/changelog/3881.added.3.md b/changelog/3881.added.3.md new file mode 100644 index 000000000..cad26e876 --- /dev/null +++ b/changelog/3881.added.3.md @@ -0,0 +1 @@ +Added `BotConnectedFrame` for SFU transports and `on_transport_timing_report` event to `StartupTimingObserver` with bot and client connection timing. diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index ad0b448e9..4af28f1ed 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -111,9 +111,11 @@ 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") + @startup_observer.event_handler("on_transport_timing_report") + async def on_transport_timing_report(observer, report): + if report.bot_connected_secs is not None: + logger.info(f"Bot connected: {report.bot_connected_secs:.3f}s") + logger.info(f"Client connected: {report.client_connected_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 b5e368c53..86778e564 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1910,6 +1910,18 @@ class StopFrame(ControlFrame, UninterruptibleFrame): pass +@dataclass +class BotConnectedFrame(SystemFrame): + """Frame indicating the bot has connected to the transport service. + + Pushed downstream by SFU transports (Daily, LiveKit, HeyGen, Tavus) + when the bot successfully joins the room. Non-SFU transports do not + emit this frame. + """ + + pass + + @dataclass class ClientConnectedFrame(SystemFrame): """Frame indicating that a client has connected to the transport. diff --git a/src/pipecat/observers/startup_timing_observer.py b/src/pipecat/observers/startup_timing_observer.py index d6b1c8fa9..555a10cb0 100644 --- a/src/pipecat/observers/startup_timing_observer.py +++ b/src/pipecat/observers/startup_timing_observer.py @@ -12,9 +12,9 @@ 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. +It also measures transport timing — the time from ``StartFrame`` to the +first ``BotConnectedFrame`` (SFU transports only) and ``ClientConnectedFrame`` +— via a separate ``on_transport_timing_report`` event. Example:: @@ -25,9 +25,11 @@ 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") + @observer.event_handler("on_transport_timing_report") + async def on_transport(observer, report): + if report.bot_connected_secs is not None: + print(f"Bot connected in {report.bot_connected_secs:.3f}s") + print(f"Client connected in {report.client_connected_secs:.3f}s") task = PipelineTask(pipeline, observers=[observer]) """ @@ -35,9 +37,7 @@ Example:: from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple, Type -from loguru import logger - -from pipecat.frames.frames import ClientConnectedFrame, StartFrame +from pipecat.frames.frames import BotConnectedFrame, 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 @@ -74,14 +74,17 @@ class StartupTimingReport: @dataclass -class TransportReadinessReport: - """Time from pipeline start to first client connection. +class TransportTimingReport: + """Time from pipeline start to transport connection milestones. Parameters: - readiness_secs: Seconds from StartFrame to first ClientConnectedFrame. + bot_connected_secs: Seconds from StartFrame to first BotConnectedFrame + (only set for SFU transports). + client_connected_secs: Seconds from StartFrame to first ClientConnectedFrame. """ - readiness_secs: float + bot_connected_secs: Optional[float] = None + client_connected_secs: Optional[float] = None class StartupTimingObserver(BaseObserver): @@ -92,9 +95,13 @@ 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. + Also measures transport timing, the time from ``StartFrame`` to connection + milestones: + + - ``bot_connected_secs``: When the bot joins the transport room + (SFU transports only, triggered by ``BotConnectedFrame``). + - ``client_connected_secs``: When a remote participant connects + (triggered by ``ClientConnectedFrame``). By default, internal pipeline processors (``PipelineSource``, ``PipelineSink``, ``Pipeline``) are excluded from the report. Pass ``processor_types`` to @@ -104,8 +111,9 @@ 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. + - on_transport_timing_report: Called once when the first client connects with a + TransportTimingReport containing client_connected_secs and bot_connected_secs + (if available). Example:: @@ -118,9 +126,11 @@ 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") + @observer.event_handler("on_transport_timing_report") + async def on_transport(observer, report): + if report.bot_connected_secs is not None: + logger.info(f"Bot connected in {report.bot_connected_secs:.3f}s") + logger.info(f"Client connected in {report.client_connected_secs:.3f}s") task = PipelineTask(pipeline, observers=[observer]) @@ -157,14 +167,17 @@ class StartupTimingObserver(BaseObserver): # 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 + # Whether we've already measured transport timing. + self._transport_timing_reported = False # Timestamp (ns) when we first see a StartFrame arrive at a processor. self._start_frame_arrival_ns: Optional[int] = None + # Bot connected timing (stored for inclusion in the transport report). + self._bot_connected_secs: Optional[float] = None + self._register_event_handler("on_startup_timing_report") - self._register_event_handler("on_transport_readiness_measured") + self._register_event_handler("on_transport_timing_report") def _should_track(self, processor: FrameProcessor) -> bool: """Check if a processor should be tracked for timing. @@ -216,11 +229,16 @@ 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. + Also handles ``BotConnectedFrame`` and ``ClientConnectedFrame`` to + measure transport timing. Args: data: The frame push event data. """ + if isinstance(data.frame, BotConnectedFrame): + self._handle_bot_connected(data) + return + if isinstance(data.frame, ClientConnectedFrame): await self._handle_client_connected(data) return @@ -249,16 +267,27 @@ 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: + def _handle_bot_connected(self, data: FramePushed): + """Record bot connected timing on first BotConnectedFrame.""" + if self._bot_connected_secs is not None 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) + self._bot_connected_secs = delta_ns / 1e9 + + async def _handle_client_connected(self, data: FramePushed): + """Emit transport timing report on first ClientConnectedFrame.""" + if self._transport_timing_reported or self._start_frame_arrival_ns is None: + return + + self._transport_timing_reported = True + delta_ns = data.timestamp - self._start_frame_arrival_ns + client_connected_secs = delta_ns / 1e9 + report = TransportTimingReport( + bot_connected_secs=self._bot_connected_secs, + client_connected_secs=client_connected_secs, + ) + await self._call_event_handler("on_transport_timing_report", report) async def _emit_report(self): """Build and emit the startup timing report.""" diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py index 4018d3858..6d45d6114 100644 --- a/src/pipecat/services/heygen/client.py +++ b/src/pipecat/services/heygen/client.py @@ -62,10 +62,12 @@ class HeyGenCallbacks(BaseModel): """Callback handlers for HeyGen events. Parameters: - on_participant_connected: Called when a participant connects - on_participant_disconnected: Called when a participant disconnects + on_connected: Called when the bot connects to the LiveKit room. + on_participant_connected: Called when a participant connects. + on_participant_disconnected: Called when a participant disconnects. """ + on_connected: Callable[[], Awaitable[None]] on_participant_connected: Callable[[str], Awaitable[None]] on_participant_disconnected: Callable[[str], Awaitable[None]] @@ -251,6 +253,7 @@ class HeyGenClient: logger.debug(f"HeyGenClient send_interval: {self._send_interval}") await self._ws_connect() await self._livekit_connect() + self._call_event_callback(self._callbacks.on_connected) async def stop(self) -> None: """Stop the client and terminate all connections. diff --git a/src/pipecat/services/heygen/video.py b/src/pipecat/services/heygen/video.py index b97f4a5ed..7f3624f35 100644 --- a/src/pipecat/services/heygen/video.py +++ b/src/pipecat/services/heygen/video.py @@ -128,6 +128,7 @@ class HeyGenVideoService(AIService): session_request=self._session_request, service_type=self._service_type, callbacks=HeyGenCallbacks( + on_connected=self._on_connected, on_participant_connected=self._on_participant_connected, on_participant_disconnected=self._on_participant_disconnected, ), @@ -144,6 +145,10 @@ class HeyGenVideoService(AIService): await self._client.cleanup() self._client = None + async def _on_connected(self): + """Handle bot connected to LiveKit room.""" + logger.info("HeyGen bot connected to LiveKit room") + async def _on_participant_connected(self, participant_id: str): """Handle participant connected events.""" logger.info(f"Participant connected {participant_id}") diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index d9f259797..8c63ff354 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -94,6 +94,7 @@ class TavusVideoService(AIService): """ await super().setup(setup) callbacks = TavusCallbacks( + on_joined=self._on_joined, on_participant_joined=self._on_participant_joined, on_participant_left=self._on_participant_left, ) @@ -119,6 +120,10 @@ class TavusVideoService(AIService): await self._client.cleanup() self._client = None + async def _on_joined(self, data): + """Handle bot joined the Daily room.""" + logger.info("Tavus bot joined Daily room") + async def _on_participant_left(self, participant, reason): """Handle participant leaving the session.""" participant_id = participant["id"] diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index cb24b23fa..97aebe915 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -24,6 +24,7 @@ from pydantic import BaseModel from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams from pipecat.frames.frames import ( + BotConnectedFrame, CancelFrame, ClientConnectedFrame, DataFrame, @@ -2579,6 +2580,8 @@ class DailyTransport(BaseTransport): if error: await self._on_error(f"Unable to start transcription: {error}") await self._call_event_handler("on_joined", data) + if self._input: + await self._input.push_frame(BotConnectedFrame()) async def _on_left(self): """Handle room left events.""" diff --git a/src/pipecat/transports/heygen/transport.py b/src/pipecat/transports/heygen/transport.py index 77ccda09f..d79d0080e 100644 --- a/src/pipecat/transports/heygen/transport.py +++ b/src/pipecat/transports/heygen/transport.py @@ -23,6 +23,7 @@ from loguru import logger from pipecat.frames.frames import ( AudioRawFrame, + BotConnectedFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, @@ -340,6 +341,7 @@ class HeyGenTransport(BaseTransport): session_request=session_request, service_type=service_type, callbacks=HeyGenCallbacks( + on_connected=self._on_connected, on_participant_connected=self._on_participant_connected, on_participant_disconnected=self._on_participant_disconnected, ), @@ -350,9 +352,16 @@ class HeyGenTransport(BaseTransport): # Register supported handlers. The user will only be able to register # these handlers. + self._register_event_handler("on_connected") self._register_event_handler("on_client_connected") self._register_event_handler("on_client_disconnected") + async def _on_connected(self): + """Handle bot connected to LiveKit room.""" + await self._call_event_handler("on_connected") + if self._input: + await self._input.push_frame(BotConnectedFrame()) + async def _on_participant_disconnected(self, participant_id: str): logger.debug(f"HeyGen participant {participant_id} disconnected") if participant_id != "heygen": diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index e4435016c..7e9c1de35 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -23,6 +23,7 @@ from pipecat.audio.utils import create_stream_resampler from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.frames.frames import ( AudioRawFrame, + BotConnectedFrame, CancelFrame, ClientConnectedFrame, EndFrame, @@ -1132,6 +1133,8 @@ class LiveKitTransport(BaseTransport): async def _on_connected(self): """Handle room connected events.""" await self._call_event_handler("on_connected") + if self._input: + await self._input.push_frame(BotConnectedFrame()) async def _on_disconnected(self): """Handle room disconnected events.""" diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index 114f33ca0..6db44d431 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -21,6 +21,7 @@ from loguru import logger from pydantic import BaseModel from pipecat.frames.frames import ( + BotConnectedFrame, CancelFrame, ClientConnectedFrame, EndFrame, @@ -133,10 +134,12 @@ class TavusCallbacks(BaseModel): """Callback handlers for Tavus events. Parameters: + on_joined: Called when the bot joins the Daily room. on_participant_joined: Called when a participant joins the conversation. on_participant_left: Called when a participant leaves the conversation. """ + on_joined: Callable[[Mapping[str, Any]], Awaitable[None]] on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]] on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]] @@ -271,6 +274,7 @@ class TavusTransportClient: async def _on_joined(self, data): """Handle joined event.""" logger.debug("TavusTransportClient joined!") + await self._callbacks.on_joined(data) async def _on_left(self): """Handle left event.""" @@ -703,6 +707,7 @@ class TavusTransport(BaseTransport): self._params = params callbacks = TavusCallbacks( + on_joined=self._on_joined, on_participant_joined=self._on_participant_joined, on_participant_left=self._on_participant_left, ) @@ -721,9 +726,16 @@ class TavusTransport(BaseTransport): # Register supported handlers. The user will only be able to register # these handlers. + self._register_event_handler("on_joined") self._register_event_handler("on_client_connected") self._register_event_handler("on_client_disconnected") + async def _on_joined(self, data): + """Handle bot joined room event.""" + await self._call_event_handler("on_joined", data) + if self._input: + await self._input.push_frame(BotConnectedFrame()) + async def _on_participant_left(self, participant, reason): """Handle participant left events.""" persona_name = await self._client.get_persona_name() diff --git a/tests/test_startup_timing_observer.py b/tests/test_startup_timing_observer.py index efabf5bc7..3c89b9ca3 100644 --- a/tests/test_startup_timing_observer.py +++ b/tests/test_startup_timing_observer.py @@ -1,11 +1,17 @@ import asyncio import unittest -from pipecat.frames.frames import ClientConnectedFrame, Frame, StartFrame, TextFrame +from pipecat.frames.frames import ( + BotConnectedFrame, + ClientConnectedFrame, + Frame, + StartFrame, + TextFrame, +) from pipecat.observers.startup_timing_observer import ( StartupTimingObserver, StartupTimingReport, - TransportReadinessReport, + TransportTimingReport, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.tests.utils import run_test @@ -182,16 +188,16 @@ 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.""" + async def test_transport_timing_client_only(self): + """Test that ClientConnectedFrame emits on_transport_timing_report.""" observer = StartupTimingObserver() processor = FastProcessor() - readiness_reports = [] + transport_reports = [] - @observer.event_handler("on_transport_readiness_measured") - async def on_readiness(obs, report): - readiness_reports.append(report) + @observer.event_handler("on_transport_timing_report") + async def on_transport(obs, report): + transport_reports.append(report) frames_to_send = [ClientConnectedFrame(), TextFrame(text="hello")] @@ -202,21 +208,22 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase): observers=[observer], ) - self.assertEqual(len(readiness_reports), 1) - report = readiness_reports[0] - self.assertIsInstance(report, TransportReadinessReport) - self.assertGreater(report.readiness_secs, 0) + self.assertEqual(len(transport_reports), 1) + report = transport_reports[0] + self.assertIsInstance(report, TransportTimingReport) + self.assertGreater(report.client_connected_secs, 0) + self.assertIsNone(report.bot_connected_secs) - async def test_transport_readiness_only_first(self): + async def test_transport_timing_only_first_client(self): """Test that only the first ClientConnectedFrame triggers the event.""" observer = StartupTimingObserver() processor = FastProcessor() - readiness_reports = [] + transport_reports = [] - @observer.event_handler("on_transport_readiness_measured") - async def on_readiness(obs, report): - readiness_reports.append(report) + @observer.event_handler("on_transport_timing_report") + async def on_transport(obs, report): + transport_reports.append(report) frames_to_send = [ ClientConnectedFrame(), @@ -231,9 +238,9 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase): observers=[observer], ) - self.assertEqual(len(readiness_reports), 1) + self.assertEqual(len(transport_reports), 1) - async def test_transport_readiness_without_start_frame(self): + async def test_transport_timing_without_start_frame(self): """Test that ClientConnectedFrame before StartFrame does not crash.""" observer = StartupTimingObserver() @@ -253,7 +260,74 @@ class TestStartupTimingObserver(unittest.IsolatedAsyncioTestCase): await observer.on_push_frame(data) # No event should have been emitted. - self.assertFalse(observer._transport_readiness_measured) + 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__":