From 389d0c3fb6adbbf7d926a3e3a6fea49a8aaae3d0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 2 Mar 2026 14:33:55 -0500 Subject: [PATCH] Use on_pipeline_started from PipelineTask for startup report Replace the PipelineSink detection in StartupTimingObserver with an on_pipeline_started() callback from PipelineTask via TaskObserver. This fixes premature report emission when using ParallelPipeline, which has its own inner PipelineSinks per branch. --- src/pipecat/observers/base_observer.py | 8 +++++ .../observers/startup_timing_observer.py | 31 +++++++++---------- src/pipecat/pipeline/task.py | 1 + src/pipecat/pipeline/task_observer.py | 14 ++++++++- 4 files changed, 37 insertions(+), 17 deletions(-) diff --git a/src/pipecat/observers/base_observer.py b/src/pipecat/observers/base_observer.py index 78e36fec8..70c79224a 100644 --- a/src/pipecat/observers/base_observer.py +++ b/src/pipecat/observers/base_observer.py @@ -100,3 +100,11 @@ class BaseObserver(BaseObject): data: The event data containing details about the frame transfer. """ pass + + async def on_pipeline_started(self): + """Called when the pipeline has fully started. + + Fired after the ``StartFrame`` has been processed by all processors + in the pipeline, including nested ``ParallelPipeline`` branches. + """ + pass diff --git a/src/pipecat/observers/startup_timing_observer.py b/src/pipecat/observers/startup_timing_observer.py index d4a010d33..a1ea04d47 100644 --- a/src/pipecat/observers/startup_timing_observer.py +++ b/src/pipecat/observers/startup_timing_observer.py @@ -43,11 +43,11 @@ from pydantic import BaseModel, Field 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 +from pipecat.pipeline.pipeline import PipelineSource from pipecat.processors.frame_processor import FrameProcessor # Internal pipeline types excluded from tracking by default. -_INTERNAL_TYPES = (PipelineSink, PipelineSource, BasePipeline) +_INTERNAL_TYPES = (PipelineSource, BasePipeline) @dataclass @@ -118,9 +118,9 @@ class StartupTimingObserver(BaseObserver): - ``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 - measure only specific types. + By default, internal pipeline processors (``PipelineSource``, ``Pipeline``) + are excluded from the report. Pass ``processor_types`` to measure only + specific types. Event handlers available: @@ -211,12 +211,19 @@ class StartupTimingObserver(BaseObserver): # Default: exclude internal pipeline plumbing. return not isinstance(processor, _INTERNAL_TYPES) + async def on_pipeline_started(self): + """Emit the startup timing report when the pipeline has fully started. + + Called by the ``PipelineTask`` after the ``StartFrame`` has been + processed by all processors, including nested ``ParallelPipeline`` + branches. + """ + if self._timings: + await self._emit_report() + async def on_process_frame(self, data: FrameProcessed): """Record when a StartFrame arrives at a processor. - When a ``StartFrame`` reaches a ``PipelineSink``, startup is complete - (the frame has traversed the entire pipeline) and the report is emitted. - Args: data: The frame processing event data. """ @@ -234,14 +241,6 @@ class StartupTimingObserver(BaseObserver): elif data.frame.id != self._start_frame_id: return - # When the StartFrame reaches a PipelineSink, all processors have - # completed start(). PipelineSinks use direct mode so the outermost - # sink fires last within the same synchronous call chain. - if isinstance(data.processor, PipelineSink): - if self._timings: - await self._emit_report() - return - if self._should_track(data.processor): self._arrivals[data.processor.id] = _ArrivalInfo( processor=data.processor, arrival_ts_ns=data.timestamp diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index deae6290c..906d55eb6 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -915,6 +915,7 @@ class PipelineTask(BasePipelineTask): if isinstance(frame, StartFrame): await self._call_event_handler("on_pipeline_started", frame) + await self._observer.on_pipeline_started() # Start heartbeat tasks now that StartFrame has been processed # by all processors in the pipeline diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index 4d33fd60e..dc2040e07 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -39,6 +39,12 @@ class Proxy: observer: BaseObserver +class _PipelineStartedSignal: + """Internal sentinel queued to observers when the pipeline has started.""" + + pass + + class TaskObserver(BaseObserver): """Proxy observer that manages multiple observers without blocking the pipeline. @@ -129,6 +135,10 @@ class TaskObserver(BaseObserver): for proxy in self._proxies: await proxy.cleanup() + async def on_pipeline_started(self): + """Forward pipeline started signal to all managed observers.""" + await self._send_to_proxy(_PipelineStartedSignal()) + async def on_process_frame(self, data: FrameProcessed): """Queue frame data for all managed observers. @@ -186,7 +196,9 @@ class TaskObserver(BaseObserver): while True: data = await queue.get() - if isinstance(data, FramePushed): + if isinstance(data, _PipelineStartedSignal): + await observer.on_pipeline_started() + elif isinstance(data, FramePushed): if on_push_frame_deprecated: await observer.on_push_frame( data.source, data.destination, data.frame, data.direction, data.timestamp