From 00ea86fda8df6febd6890eecb20a5a6088284063 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 9 Apr 2026 16:50:17 -0700 Subject: [PATCH] Fix observer cleanup ordering to stop proxy tasks before closing resources During pipeline shutdown, proxy tasks must be cancelled before observer resources are cleaned up. Previously, stop() was called inside _cancel_tasks() and start() was called in _start_tasks(), which could lead to proxy tasks still consuming frames after observer resources were closed. Now the lifecycle is explicit in _handle_start_frame: start() after all observers are loaded, and stop() before cleanup() on shutdown. Also fixes misleading variable name in TaskObserver.cleanup() where iterating self._proxies yields observer keys, not Proxy values. Fixes #4195 --- src/pipecat/pipeline/task.py | 22 +++++++++------------- src/pipecat/pipeline/task_observer.py | 4 ++-- 2 files changed, 11 insertions(+), 15 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 507389341..8ad2cfd63 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -613,9 +613,6 @@ class PipelineTask(BasePipelineTask): self._process_push_task = self._task_manager.create_task( self._process_push_queue(), f"{self}::_process_push_queue" ) - - await self._observer.start() - return self._process_push_task def _maybe_start_heartbeat_tasks(self): @@ -637,8 +634,6 @@ class PipelineTask(BasePipelineTask): async def _cancel_tasks(self): """Cancel all running pipeline tasks.""" - await self._observer.stop() - if self._process_push_task: await self._task_manager.cancel_task(self._process_push_task) self._process_push_task = None @@ -720,12 +715,6 @@ class PipelineTask(BasePipelineTask): async def _setup(self, params: PipelineTaskParams): """Set up the pipeline task and all processors.""" - # Do any additional pipeline task setup externally. - await self._load_setup_files() - - # Load additional observers. - await self._load_observer_files() - mgr_params = TaskManagerParams(loop=params.loop) self._task_manager.setup(mgr_params) @@ -736,14 +725,21 @@ class PipelineTask(BasePipelineTask): ) await self._pipeline.setup(setup) + # Do any additional pipeline task setup externally. + await self._load_setup_files() + await self._load_observer_files() + + # Start task observer. + await self._observer.start() + async def _cleanup(self, cleanup_pipeline: bool): """Clean up the pipeline task and processors.""" # Cleanup base object. await self.cleanup() # Cleanup observers. - if self._observer: - await self._observer.cleanup() + await self._observer.stop() + await self._observer.cleanup() # End conversation tracing if it's active - this will also close any active turn span if self._enable_tracing and hasattr(self, "_turn_trace_observer"): diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index 84f463443..44326d7c3 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -131,8 +131,8 @@ class TaskObserver(BaseObserver): if not self._proxies: return - for proxy in self._proxies: - await proxy.cleanup() + for observer in self._proxies: + await observer.cleanup() async def on_pipeline_started(self): """Forward pipeline started signal to all managed observers."""