From 33db71ec32db8c892197652f7c167075c83540c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 8 May 2026 13:51:46 -0700 Subject: [PATCH] Call super().setup() in PipelineTask to honor BaseObject contract PipelineTask owns its TaskManager (still constructed in __init__ since TaskObserver needs it eagerly). Adding the explicit `await super().setup(self._task_manager)` in `_setup()` formalizes the BaseObject lifecycle so any future wiring added to BaseObject.setup is picked up automatically. --- src/pipecat/pipeline/task.py | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 479aaada9..38b8dd03d 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -654,27 +654,19 @@ class PipelineTask(BasePipelineTask): async def _create_tasks(self): """Create and start all pipeline processing tasks.""" - self._process_push_task = self.create_task( - self._process_push_queue(), "_process_push_queue" - ) + self._process_push_task = self.create_task(self._process_push_queue()) return self._process_push_task def _maybe_start_heartbeat_tasks(self): """Start heartbeat tasks if heartbeats are enabled and not already running.""" if self._params.enable_heartbeats and self._heartbeat_push_task is None: - self._heartbeat_push_task = self.create_task( - self._heartbeat_push_handler(), "_heartbeat_push_handler" - ) - self._heartbeat_monitor_task = self.create_task( - self._heartbeat_monitor_handler(), "_heartbeat_monitor_handler" - ) + self._heartbeat_push_task = self.create_task(self._heartbeat_push_handler()) + self._heartbeat_monitor_task = self.create_task(self._heartbeat_monitor_handler()) def _maybe_start_idle_task(self): """Start idle monitoring task if idle timeout is configured.""" if self._idle_timeout_secs: - self._idle_monitor_task = self.create_task( - self._idle_monitor_handler(), "_idle_monitor_handler" - ) + self._idle_monitor_task = self.create_task(self._idle_monitor_handler()) async def _cancel_tasks(self): """Cancel all running pipeline tasks.""" @@ -759,12 +751,14 @@ class PipelineTask(BasePipelineTask): async def _setup(self, params: PipelineTaskParams): """Set up the pipeline task and all processors.""" + await super().setup(self._task_manager) + mgr_params = TaskManagerParams(loop=params.loop) - self._task_manager.setup(mgr_params) + self.task_manager.setup(mgr_params) setup = FrameProcessorSetup( clock=self._clock, - task_manager=self._task_manager, + task_manager=self.task_manager, observer=self._observer, pipeline_task=self, # Populate the deprecated `tool_resources` field for backwards