Merge pull request #2537 from pipecat-ai/aleix/pipeline-task-cleanup-observers

PipelineTask: cleanup observers

src/pipecat/pipeline/task.py

@@ -571,6 +571,10 @@ class PipelineTask(BasePipelineTask):
         # Cleanup base object.
         await self.cleanup()
 
+        # Cleanup observers.
+        if self._observer:
+            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"):
             self._turn_trace_observer.end_conversation_tracing()

src/pipecat/pipeline/task_observer.py

@@ -119,6 +119,16 @@ class TaskObserver(BaseObserver):
         for proxy in self._proxies.values():
             await self._task_manager.cancel_task(proxy.task)
 
+    async def cleanup(self):
+        """Cleanup all proxy observers."""
+        await super().cleanup()
+
+        if not self._proxies:
+            return
+
+        for proxy in self._proxies:
+            await proxy.cleanup()
+
     async def on_process_frame(self, data: FramePushed):
         """Queue frame data for all managed observers.