utils.asyncio: added watchdog_coroutine()

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Added `watchdog_coroutine()`. This is a watchdog helper for couroutines. So,
+  if you have a coroutine that is waiting for a result and that takes a long
+  time, you will need to wrap it with `watchdog_coroutine()` so the watchdog
+  timers are reset regularly.
+
 ## [0.0.73] - 2025-06-26
 
 ### Fixed

src/pipecat/utils/asyncio/watchdog_async_iterator.py

@@ -55,7 +55,8 @@ class WatchdogAsyncIterator:
 
                 self._manager.task_reset_watchdog()
 
-                # The task has finish, so we will create a new one for th next item.
+                # The task has finished, so we will create a new one for the
+                # next item.
                 self._current_anext_task = None
 
                 return item

src/pipecat/utils/asyncio/watchdog_coroutine.py

@@ -0,0 +1,61 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from typing import Optional
+
+from pipecat.utils.asyncio.task_manager import BaseTaskManager
+
+
+class WatchdogCoroutine:
+    """An asynchronous iterator that monitors activity and resets the current
+    task watchdog timer. This is necessary to avoid task watchdog timers to
+    expire while we are waiting to get an item from the iterator.
+
+    """
+
+    def __init__(
+        self,
+        coroutine,
+        *,
+        manager: BaseTaskManager,
+        timeout: float = 2.0,
+    ):
+        self._coroutine = coroutine
+        self._manager = manager
+        self._timeout = timeout
+        self._current_coro_task: Optional[asyncio.Task] = None
+
+    async def __call__(self):
+        if self._manager.task_watchdog_enabled:
+            return await self._watchdog_call()
+        else:
+            return await self._coroutine
+
+    async def _watchdog_call(self):
+        while True:
+            try:
+                if not self._current_coro_task:
+                    self._current_coro_task = asyncio.create_task(self._coroutine)
+
+                result = await asyncio.wait_for(
+                    asyncio.shield(self._current_coro_task),
+                    timeout=self._timeout,
+                )
+
+                self._manager.task_reset_watchdog()
+
+                # The task has finished.
+                self._current_coro_task = None
+
+                return result
+            except asyncio.TimeoutError:
+                self._manager.task_reset_watchdog()
+
+
+async def watchdog_coroutine(coroutine, *, manager: BaseTaskManager, timeout: float = 2.0):
+    watchdog_coro = WatchdogCoroutine(coroutine, manager=manager, timeout=timeout)
+    return await watchdog_coro()