FrameProcessor: wait_for_task is now deprecated

CHANGELOG.md

@@ -48,6 +48,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Updated `daily-python` to 0.19.7.
 
+### Deprecated
+
+- `FrameProcessor.wait_for_task()` is deprecated. Use `await task` or `await
+  asyncio.wait_for(task, timeout)` instead.
+
 ### Removed
 
 - Watchdog timers have been removed. They were introduced in 0.0.72 to help

src/pipecat/audio/utils.py

@@ -28,7 +28,7 @@ SPEAKING_THRESHOLD = 20
 def create_default_resampler(**kwargs) -> BaseAudioResampler:
     """Create a default audio resampler instance.
 
-    . deprecated:: 0.0.74
+    .. deprecated:: 0.0.74
         This function is deprecated and will be removed in a future version.
         Use `create_stream_resampler` for real-time processing scenarios or
         `create_file_resampler` for batch processing of complete audio files.

src/pipecat/pipeline/task.py

@@ -363,9 +363,9 @@ class PipelineTask(BasePipelineTask):
 
             # Create all main tasks and wait of the main push task. This is the
             # task that pushes frames to the very beginning of our pipeline (our
-            # controlled PipelineTaskSource processor).
+            # controlled source processor).
             push_task = await self._create_tasks()
-            await self._task_manager.wait_for_task(push_task)
+            await push_task
 
             # We have already cleaned up the pipeline inside the task.
             cleanup_pipeline = False

src/pipecat/processors/aggregators/llm_response.py

@@ -985,10 +985,6 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
 
     def _context_updated_task_finished(self, task: asyncio.Task):
         self._context_updated_tasks.discard(task)
-        # The task is finished so this should exit immediately. We need to do
-        # this because otherwise the task manager would report a dangling task
-        # if we don't remove it.
-        asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop())
 
 
 class LLMUserResponseAggregator(LLMUserContextAggregator):

src/pipecat/processors/frame_processor.py

@@ -442,11 +442,27 @@ class FrameProcessor(BaseObject):
     async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None):
         """Wait for a task to complete.
 
+        .. deprecated:: 0.0.81
+            This function is deprecated, use `await task` or
+            `await asyncio.wait_for(task, timeout) instead.
+
         Args:
             task: The task to wait for.
             timeout: Optional timeout for waiting.
         """
-        await self.task_manager.wait_for_task(task, timeout)
+        import warnings
+
+        warnings.warn(
+            "`FrameProcessor.wait_for_task()` is deprecated. "
+            "Use `await task` or `await asyncio.wait_for(task, timeout)` instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+
+        if timeout:
+            await asyncio.wait_for(task, timeout)
+        else:
+            await task
 
     async def setup(self, setup: FrameProcessorSetup):
         """Set up the processor with required components.

src/pipecat/processors/metrics/sentry.py

@@ -63,7 +63,7 @@ class SentryMetrics(FrameProcessorMetrics):
         await super().cleanup()
         if self._sentry_task:
             await self._sentry_queue.put(None)
-            await self.task_manager.wait_for_task(self._sentry_task)
+            await self._sentry_task
             self._sentry_task = None
             logger.trace(f"{self} Flushing Sentry metrics")
             sentry_sdk.flush(timeout=5.0)

src/pipecat/services/llm_service.py

@@ -487,7 +487,7 @@ class LLMService(AIService):
             self._function_call_tasks[task] = runner_item
             # Since we run tasks sequentially we don't need to call
             # task.add_done_callback(self._function_call_task_finished).
-            await self.wait_for_task(task)
+            await task
             del self._function_call_tasks[task]
 
     async def _run_function_call(self, runner_item: FunctionCallRunnerItem):
@@ -616,7 +616,3 @@ class LLMService(AIService):
     def _function_call_task_finished(self, task: asyncio.Task):
         if task in self._function_call_tasks:
             del self._function_call_tasks[task]
-            # The task is finished so this should exit immediately. We need to
-            # do this because otherwise the task manager would report a dangling
-            # task if we don't remove it.
-            asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop())

src/pipecat/services/soniox/stt.py

@@ -208,7 +208,7 @@ class SonioxSTTService(STTService):
         if self._receive_task:
             # Task cannot cancel itself. If task called _cleanup() we expect it to cancel itself.
             if self._receive_task != asyncio.current_task():
-                await self.wait_for_task(self._receive_task)
+                await self._receive_task
             self._receive_task = None
 
     async def stop(self, frame: EndFrame):

src/pipecat/services/tts_service.py

@@ -794,7 +794,7 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
             # Indicate no more audio contexts are available. this will end the
             # task cleanly after all contexts have been processed.
             await self._contexts_queue.put(None)
-            await self.wait_for_task(self._audio_context_task)
+            await self._audio_context_task
             self._audio_context_task = None
 
     async def cancel(self, frame: CancelFrame):

src/pipecat/transports/base_output.py

@@ -435,9 +435,9 @@ class BaseOutputTransport(FrameProcessor):
             # also need to wait for these tasks before cancelling the video task
             # because it might be still rendering.
             if self._audio_task:
-                await self._transport.wait_for_task(self._audio_task)
+                await self._audio_task
             if self._clock_task:
-                await self._transport.wait_for_task(self._clock_task)
+                await self._clock_task
 
             # Stop audio mixer.
             if self._mixer:

src/pipecat/transports/network/websocket_server.py

@@ -154,7 +154,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
             await self.cancel_task(self._monitor_task)
             self._monitor_task = None
         if self._server_task:
-            await self.wait_for_task(self._server_task)
+            await self._server_task
             self._server_task = None
 
     async def cancel(self, frame: CancelFrame):

src/pipecat/utils/asyncio/task_manager.py

@@ -69,21 +69,6 @@ class BaseTaskManager(ABC):
         """
         pass
 
-    @abstractmethod
-    async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None):
-        """Wait for an asyncio.Task to complete with optional timeout handling.
-
-        This function awaits the specified asyncio.Task and handles scenarios for
-        timeouts, cancellations, and other exceptions. It also ensures that the task
-        is removed from the set of registered tasks upon completion or failure.
-
-        Args:
-            task: The asyncio Task to wait for.
-            timeout: The maximum number of seconds to wait for the task to complete.
-                If None, waits indefinitely.
-        """
-        pass
-
     @abstractmethod
     async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None):
         """Cancels the given asyncio Task and awaits its completion with an optional timeout.
@@ -189,35 +174,6 @@ class TaskManager(BaseTaskManager):
         logger.trace(f"{name}: task created")
         return task
 
-    async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None):
-        """Wait for an asyncio.Task to complete with optional timeout handling.
-
-        This function awaits the specified asyncio.Task and handles scenarios for
-        timeouts, cancellations, and other exceptions. It also ensures that the task
-        is removed from the set of registered tasks upon completion or failure.
-
-        Args:
-            task: The asyncio Task to wait for.
-            timeout: The maximum number of seconds to wait for the task to complete.
-                If None, waits indefinitely.
-        """
-        name = task.get_name()
-        try:
-            if timeout:
-                await asyncio.wait_for(task, timeout=timeout)
-            else:
-                await task
-        except asyncio.TimeoutError:
-            logger.warning(f"{name}: timed out waiting for task to finish")
-        except asyncio.CancelledError:
-            logger.trace(f"{name}: unexpected task cancellation (maybe Ctrl-C?)")
-            raise
-        except Exception as e:
-            logger.exception(f"{name}: unexpected exception while stopping task: {e}")
-        except BaseException as e:
-            logger.critical(f"{name}: fatal base exception while stopping task: {e}")
-            raise
-
     async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None):
         """Cancels the given asyncio Task and awaits its completion with an optional timeout.