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

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

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``,
+    By default, internal pipeline processors (``PipelineSource``, ``Pipeline``)
-    ``Pipeline``) are excluded from the report. Pass ``processor_types`` to
+    are excluded from the report. Pass ``processor_types`` to measure only
-    measure only specific types.
+    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

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

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