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 1/4] 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.""" From 9b7e15c9bc6719db2404d7edb8f6350ada8c25c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 9 Apr 2026 16:50:46 -0700 Subject: [PATCH 2/4] Add changelog for #4267 --- changelog/4267.fixed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4267.fixed.md diff --git a/changelog/4267.fixed.md b/changelog/4267.fixed.md new file mode 100644 index 000000000..5f495bc73 --- /dev/null +++ b/changelog/4267.fixed.md @@ -0,0 +1 @@ +- Fixed `ValueError: write to closed file` during pipeline shutdown when observers were active. Observer proxy tasks are now cancelled before observer resources are cleaned up. From 360e4480e07840f99021d3d9a8fd1e2ea3642a38 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 9 Apr 2026 17:38:46 -0700 Subject: [PATCH 3/4] Remove deprecated _load_observer_files in favor of setup files --- src/pipecat/pipeline/task.py | 32 -------------------------------- 1 file changed, 32 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 8ad2cfd63..1f9de78a6 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -727,7 +727,6 @@ class PipelineTask(BasePipelineTask): # Do any additional pipeline task setup externally. await self._load_setup_files() - await self._load_observer_files() # Start task observer. await self._observer.start() @@ -967,37 +966,6 @@ class PipelineTask(BasePipelineTask): except Exception as e: logger.error(f"{self} error running external setup from {f}: {e}") - async def _load_observer_files(self): - """Dynamically load observers from files listed in PIPECAT_OBSERVER_FILES.""" - observer_files = [f for f in os.environ.get("PIPECAT_OBSERVER_FILES", "").split(":") if f] - for f in observer_files: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Observer files (and environment variable `PIPECAT_OBSERVER_FILES`) is deprecated, use setup files instead (and `PIPECAT_SETUP_FILES`) instead.", - DeprecationWarning, - ) - - try: - path = Path(f).resolve() - module_name = path.stem - spec = importlib.util.spec_from_file_location(module_name, str(path)) - if spec: - logger.debug(f"{self} loading observers from {path}") - - # Load module. - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - - # Create observers. - observers = await module.create_observers(self) - for observer in observers: - self.add_observer(observer) - except Exception as e: - logger.error(f"{self} error loading external observers from {f}: {e}") - def _print_dangling_tasks(self): """Log any dangling tasks that haven't been properly cleaned up.""" tasks = [t.get_name() for t in self._task_manager.current_tasks()] From f9117e6d4ab617ddd1ee74684acc6c1df9477814 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 9 Apr 2026 17:39:54 -0700 Subject: [PATCH 4/4] Add changelog for PIPECAT_OBSERVER_FILES removal --- changelog/4267.removed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4267.removed.md diff --git a/changelog/4267.removed.md b/changelog/4267.removed.md new file mode 100644 index 000000000..bf1e1cb82 --- /dev/null +++ b/changelog/4267.removed.md @@ -0,0 +1 @@ +- ⚠️ Removed deprecated `PIPECAT_OBSERVER_FILES` environment variable support. Use `PIPECAT_SETUP_FILES` instead.