From 3064326834737e4384c2f14538bab9c5754db956 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 27 Jun 2025 10:23:16 -0700
Subject: [PATCH] utils.asyncio: added watchdog_coroutine()

---
 CHANGELOG.md                                  |  9 +++
 .../utils/asyncio/watchdog_async_iterator.py  |  3 +-
 .../utils/asyncio/watchdog_coroutine.py       | 61 +++++++++++++++++++
 3 files changed, 72 insertions(+), 1 deletion(-)
 create mode 100644 src/pipecat/utils/asyncio/watchdog_coroutine.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 019cafd58..2d1d5a72e 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/src/pipecat/utils/asyncio/watchdog_async_iterator.py b/src/pipecat/utils/asyncio/watchdog_async_iterator.py
index e71b37ae3..d9d3e2f79 100644
--- a/src/pipecat/utils/asyncio/watchdog_async_iterator.py
+++ b/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
diff --git a/src/pipecat/utils/asyncio/watchdog_coroutine.py b/src/pipecat/utils/asyncio/watchdog_coroutine.py
new file mode 100644
index 000000000..84855c3e6
--- /dev/null
+++ b/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()