Merge pull request #2476 from pipecat-ai/aleix/add-asyncio-timeout

implement custom asyncio.wait_for()

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `pipecat.utils.timeout.wait_for()` which uses `wait_for2` package to
+  avoid `asyncio.wait_for()` issues in Python < 3.12.
+
 - Allow passing custom pipeline sink and source processors to a
   `Pipeline`. Pipeline source and sink processors are used to know and control
   what's coming in and out of a `Pipeline` processor.
@@ -46,6 +49,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Replaced `asyncio.wait_for()` for Pipecat's `wait_for()`. In Python 3.10,
+  `asyncio.wait_for()` has issues regarding task cancellation (i.e. cancellation
+  is never propagated).
+  See https://bugs.python.org/issue42130
+
 - Fixed an `AudioBufferProcessor` issues that would cause audio overlap when
   setting a max buffer size.
 

pyproject.toml

@@ -36,6 +36,7 @@ dependencies = [
     "openai>=1.74.0,<=1.99.1",
     # Pinning numba to resolve package dependencies
     "numba==0.61.2",
+    "wait_for2>=0.4.1; python_version<'3.12'",
 ]
 
 [project.urls]

scripts/evals/eval.py

@@ -43,6 +43,7 @@ from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.utils.asyncio.timeout import wait_for
 
 SCRIPT_DIR = Path(__file__).resolve().parent
 
@@ -122,7 +123,7 @@ class EvalRunner:
             logger.error(f"ERROR: Unable to run {example_file}: {e}")
 
         try:
-            result = await asyncio.wait_for(self._queue.get(), timeout=1.0)
+            result = await wait_for(self._queue.get(), timeout=1.0)
         except asyncio.TimeoutError:
             result = False
 

src/pipecat/pipeline/task.py

@@ -51,6 +51,7 @@ from pipecat.utils.asyncio.task_manager import (
     TaskManager,
     TaskManagerParams,
 )
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 from pipecat.utils.tracing.setup import is_tracing_available
 from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
@@ -654,7 +655,7 @@ class PipelineTask(BasePipelineTask):
         wait_time = HEARTBEAT_MONITOR_SECONDS
         while True:
             try:
-                frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time)
+                frame = await wait_for(self._heartbeat_queue.get(), timeout=wait_time)
                 process_time = (self._clock.get_time() - frame.timestamp) / 1_000_000_000
                 logger.trace(f"{self}: heartbeat frame processed in {process_time} seconds")
                 self._heartbeat_queue.task_done()
@@ -677,9 +678,7 @@ class PipelineTask(BasePipelineTask):
 
         while running:
             try:
-                frame = await asyncio.wait_for(
+                frame = await wait_for(self._idle_queue.get(), timeout=self._idle_timeout_secs)
-                    self._idle_queue.get(), timeout=self._idle_timeout_secs
-                )
 
                 if not isinstance(frame, InputAudioRawFrame):
                     frame_buffer.append(frame)

src/pipecat/processors/aggregators/dtmf_aggregator.py

@@ -25,6 +25,7 @@ from pipecat.frames.frames import (
     TranscriptionFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.time import time_now_iso8601
 
 
@@ -136,7 +137,7 @@ class DTMFAggregator(FrameProcessor):
         """Background task that handles timeout-based flushing."""
         while True:
             try:
-                await asyncio.wait_for(self._digit_event.wait(), timeout=self._idle_timeout)
+                await wait_for(self._digit_event.wait(), timeout=self._idle_timeout)
                 self._digit_event.clear()
             except asyncio.TimeoutError:
                 self.reset_watchdog()

src/pipecat/processors/aggregators/llm_response.py

@@ -60,6 +60,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.time import time_now_iso8601
 
 
@@ -670,7 +671,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
                         if self._vad_params
                         else self._params.turn_emulated_vad_timeout
                     )
-                await asyncio.wait_for(self._aggregation_event.wait(), timeout)
+                await wait_for(self._aggregation_event.wait(), timeout=timeout)
                 await self._maybe_emulate_user_speaking()
             except asyncio.TimeoutError:
                 if not self._user_speaking:

src/pipecat/processors/idle_frame_processor.py

@@ -11,6 +11,7 @@ from typing import Awaitable, Callable, List, Optional
 
 from pipecat.frames.frames import Frame, StartFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_event import WatchdogEvent
 
 
@@ -85,7 +86,7 @@ class IdleFrameProcessor(FrameProcessor):
         """Handle idle timeout monitoring and callback execution."""
         while True:
             try:
-                await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout)
+                await wait_for(self._idle_event.wait(), timeout=self._timeout)
             except asyncio.TimeoutError:
                 await self._callback(self)
             finally:

src/pipecat/processors/user_idle_processor.py

@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_event import WatchdogEvent
 
 
@@ -191,7 +192,7 @@ class UserIdleProcessor(FrameProcessor):
         """
         while True:
             try:
-                await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout)
+                await wait_for(self._idle_event.wait(), timeout=self._timeout)
             except asyncio.TimeoutError:
                 if not self._interrupted:
                     self._retry_count += 1

src/pipecat/services/anthropic/llm.py

@@ -53,6 +53,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
 from pipecat.utils.tracing.service_decorators import traced_llm
 
@@ -185,9 +186,7 @@ class AnthropicLLMService(LLMService):
         """
         if self._retry_on_timeout:
             try:
-                response = await asyncio.wait_for(
+                response = await wait_for(api_call(**params), timeout=self._retry_timeout_secs)
-                    api_call(**params), timeout=self._retry_timeout_secs
-                )
                 return response
             except (APITimeoutError, asyncio.TimeoutError):
                 # Retry, this time without a timeout so we get a response

src/pipecat/services/assemblyai/stt.py

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.service_decorators import traced_stt
 
@@ -219,10 +220,7 @@ class AssemblyAISTTService(STTService):
                 await self._websocket.send(json.dumps({"type": "Terminate"}))
 
                 try:
-                    await asyncio.wait_for(
+                    await wait_for(self._termination_event.wait(), timeout=5.0)
-                        self._termination_event.wait(),
-                        timeout=5.0,
-                    )
                 except asyncio.TimeoutError:
                     logger.warning("Timed out waiting for termination message from server")
 
@@ -247,7 +245,7 @@ class AssemblyAISTTService(STTService):
         try:
             while self._connected:
                 try:
-                    message = await asyncio.wait_for(self._websocket.recv(), timeout=1.0)
+                    message = await wait_for(self._websocket.recv(), timeout=1.0)
                     data = json.loads(message)
                     await self._handle_message(data)
                 except asyncio.TimeoutError:

src/pipecat/services/aws/llm.py

@@ -52,6 +52,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import LLMService
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.tracing.service_decorators import traced_llm
 
 try:
@@ -801,8 +802,8 @@ class AWSBedrockLLMService(LLMService):
         """
         if self._retry_on_timeout:
             try:
-                response = await asyncio.wait_for(
+                response = await wait_for(
-                    client.converse_stream(**request_params), timeout=self._retry_timeout_secs
+                    await client.converse_stream(**request_params), timeout=self._retry_timeout_secs
                 )
                 return response
             except (ReadTimeoutError, asyncio.TimeoutError) as e:

src/pipecat/services/aws/stt.py

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
 from pipecat.services.aws.utils import build_event_message, decode_event, get_presigned_url
 from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.service_decorators import traced_stt
 
@@ -480,7 +481,7 @@ class AWSTranscribeSTTService(STTService):
                 break
 
             try:
-                response = await asyncio.wait_for(self._ws_client.recv(), timeout=1.0)
+                response = await wait_for(self._ws_client.recv(), timeout=1.0)
 
                 headers, payload = decode_event(response)
 

src/pipecat/services/heygen/client.py

@@ -31,6 +31,7 @@ from pipecat.processors.frame_processor import FrameProcessorSetup
 from pipecat.services.heygen.api import HeyGenApi, HeyGenSession, NewSessionRequest
 from pipecat.transports.base_transport import TransportParams
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 
 try:
@@ -231,7 +232,7 @@ class HeyGenClient:
         """Handle incoming WebSocket messages."""
         while self._connected:
             try:
-                message = await asyncio.wait_for(self._websocket.recv(), timeout=1.0)
+                message = await wait_for(self._websocket.recv(), timeout=1.0)
                 parsed_message = json.loads(message)
                 await self._handle_ws_server_event(parsed_message)
             except asyncio.TimeoutError:

src/pipecat/services/heygen/video.py

@@ -40,6 +40,7 @@ from pipecat.services.ai_service import AIService
 from pipecat.services.heygen.api import NewSessionRequest
 from pipecat.services.heygen.client import HEY_GEN_SAMPLE_RATE, HeyGenCallbacks, HeyGenClient
 from pipecat.transports.base_transport import TransportParams
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 
 # Using the same values that we do in the BaseOutputTransport
@@ -320,7 +321,7 @@ class HeyGenVideoService(AIService):
 
         while True:
             try:
-                frame = await asyncio.wait_for(self._queue.get(), timeout=AVATAR_VAD_STOP_SECS)
+                frame = await wait_for(self._queue.get(), timeout=AVATAR_VAD_STOP_SECS)
                 if self._is_interrupting:
                     break
                 if isinstance(frame, TTSAudioRawFrame):

src/pipecat/services/openai/base_llm.py

@@ -39,6 +39,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
 from pipecat.utils.tracing.service_decorators import traced_llm
 
@@ -196,7 +197,7 @@ class BaseOpenAILLMService(LLMService):
 
         if self._retry_on_timeout:
             try:
-                chunks = await asyncio.wait_for(
+                chunks = await wait_for(
                     self._client.chat.completions.create(**params), timeout=self._retry_timeout_secs
                 )
                 return chunks

src/pipecat/services/riva/tts.py

@@ -14,6 +14,7 @@ import asyncio
 import os
 from typing import AsyncGenerator, Mapping, Optional
 
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.tracing.service_decorators import traced_tts
 
 # Suppress gRPC fork warnings
@@ -168,7 +169,7 @@ class RivaTTSService(TTSService):
             await asyncio.to_thread(read_audio_responses, queue)
 
             # Wait for the thread to start.
-            resp = await asyncio.wait_for(queue.get(), RIVA_TTS_TIMEOUT_SECS)
+            resp = await wait_for(queue.get(), timeout=RIVA_TTS_TIMEOUT_SECS)
             while resp:
                 await self.stop_ttfb_metrics()
                 frame = TTSAudioRawFrame(
@@ -177,7 +178,7 @@ class RivaTTSService(TTSService):
                     num_channels=1,
                 )
                 yield frame
-                resp = await asyncio.wait_for(queue.get(), RIVA_TTS_TIMEOUT_SECS)
+                resp = await wait_for(queue.get(), timeout=RIVA_TTS_TIMEOUT_SECS)
         except asyncio.TimeoutError:
             logger.error(f"{self} timeout waiting for audio response")
 

src/pipecat/services/speechmatics/stt.py

@@ -33,6 +33,7 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.tracing.service_decorators import traced_stt
 
 try:
@@ -576,7 +577,7 @@ class SpeechmaticsSTTService(STTService):
         # Disconnect the client
         try:
             if self._client:
-                await asyncio.wait_for(self._client.close(), timeout=1.0)
+                await wait_for(self._client.close(), timeout=1.0)
         except asyncio.TimeoutError:
             logger.warning("Timeout while closing Speechmatics client connection")
         except Exception as e:

src/pipecat/services/tts_service.py

@@ -37,6 +37,7 @@ from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
 from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
 from pipecat.utils.text.base_text_filter import BaseTextFilter
@@ -427,8 +428,8 @@ class TTSService(AIService):
         has_started = False
         while True:
             try:
-                frame = await asyncio.wait_for(
+                frame = await wait_for(
-                    self._stop_frame_queue.get(), self._stop_frame_timeout_s
+                    self._stop_frame_queue.get(), timeout=self._stop_frame_timeout_s
                 )
                 if isinstance(frame, TTSStartedFrame):
                     has_started = True
@@ -858,7 +859,7 @@ class AudioContextWordTTSService(WebsocketWordTTSService):
         running = True
         while running:
             try:
-                frame = await asyncio.wait_for(queue.get(), timeout=AUDIO_CONTEXT_TIMEOUT)
+                frame = await wait_for(queue.get(), timeout=AUDIO_CONTEXT_TIMEOUT)
                 self.reset_watchdog()
                 if frame:
                     await self.push_frame(frame)

src/pipecat/transports/base_input.py

@@ -49,6 +49,8 @@ from pipecat.frames.frames import (
 from pipecat.metrics.metrics import MetricsData
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.transports.base_transport import TransportParams
+from pipecat.utils.asyncio.timeout import wait_for
+from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 
 AUDIO_INPUT_TIMEOUT_SECS = 0.5
 
@@ -395,7 +397,7 @@ class BaseInputTransport(FrameProcessor):
     def _create_audio_task(self):
         """Create the audio processing task if audio input is enabled."""
         if not self._audio_task and self._params.audio_in_enabled:
-            self._audio_in_queue = asyncio.Queue()
+            self._audio_in_queue = WatchdogQueue(self.task_manager)
             self._audio_task = self.create_task(self._audio_task_handler())
 
     async def _cancel_audio_task(self):
@@ -474,7 +476,7 @@ class BaseInputTransport(FrameProcessor):
         vad_state: VADState = VADState.QUIET
         while True:
             try:
-                frame: InputAudioRawFrame = await asyncio.wait_for(
+                frame: InputAudioRawFrame = await wait_for(
                     self._audio_in_queue.get(), timeout=AUDIO_INPUT_TIMEOUT_SECS
                 )
 

src/pipecat/transports/base_output.py

@@ -46,6 +46,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.transports.base_transport import TransportParams
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_priority_queue import WatchdogPriorityQueue
 from pipecat.utils.time import nanoseconds_to_seconds
 
@@ -623,9 +624,7 @@ class BaseOutputTransport(FrameProcessor):
             async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[Frame, None]:
                 while True:
                     try:
-                        frame = await asyncio.wait_for(
+                        frame = await wait_for(self._audio_queue.get(), timeout=vad_stop_secs)
-                            self._audio_queue.get(), timeout=vad_stop_secs
-                        )
                         self._transport.reset_watchdog()
                         yield frame
                     except asyncio.TimeoutError:

src/pipecat/transports/network/small_webrtc.py

@@ -40,6 +40,7 @@ from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
 
 try:
@@ -289,7 +290,7 @@ class SmallWebRTCClient:
                 continue
 
             try:
-                frame = await asyncio.wait_for(self._video_input_track.recv(), timeout=2.0)
+                frame = await wait_for(self._video_input_track.recv(), timeout=2.0)
             except asyncio.TimeoutError:
                 if self._webrtc_connection.is_connected():
                     logger.warning("Timeout: No video frame received within the specified time.")
@@ -332,7 +333,7 @@ class SmallWebRTCClient:
                 continue
 
             try:
-                frame = await asyncio.wait_for(self._audio_input_track.recv(), timeout=2.0)
+                frame = await wait_for(self._audio_input_track.recv(), timeout=2.0)
             except asyncio.TimeoutError:
                 if self._webrtc_connection.is_connected():
                     logger.warning("Timeout: No audio frame received within the specified time.")

src/pipecat/transports/services/daily.py

@@ -49,6 +49,7 @@ from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.asyncio.timeout import wait_for
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 
 try:
@@ -694,7 +695,7 @@ class DailyTransportClient(EventHandler):
             },
         )
 
-        return await asyncio.wait_for(future, timeout=10)
+        return await wait_for(future, timeout=10)
 
     async def leave(self):
         """Leave the Daily room and cleanup resources."""
@@ -735,7 +736,7 @@ class DailyTransportClient(EventHandler):
         """Execute the actual room leave operation."""
         future = self._get_event_loop().create_future()
         self._client.leave(completion=completion_callback(future))
-        return await asyncio.wait_for(future, timeout=10)
+        return await wait_for(future, timeout=10)
 
     def _cleanup(self):
         """Cleanup the Daily client instance."""

src/pipecat/utils/asyncio/task_manager.py

@@ -20,6 +20,8 @@ from typing import Coroutine, Dict, Optional, Sequence
 
 from loguru import logger
 
+from pipecat.utils.asyncio.timeout import wait_for
+
 WATCHDOG_TIMEOUT = 5.0
 
 
@@ -285,7 +287,7 @@ class TaskManager(BaseTaskManager):
         name = task.get_name()
         try:
             if timeout:
-                await asyncio.wait_for(task, timeout=timeout)
+                await wait_for(task, timeout=timeout)
             else:
                 await task
         except asyncio.TimeoutError:
@@ -315,7 +317,7 @@ class TaskManager(BaseTaskManager):
             # Make sure to reset watchdog if a task is cancelled.
             self.reset_watchdog(task)
             if timeout:
-                await asyncio.wait_for(task, timeout=timeout)
+                await wait_for(task, timeout=timeout)
             else:
                 await task
         except asyncio.TimeoutError:
@@ -402,7 +404,7 @@ class TaskManager(BaseTaskManager):
                     break
 
                 start_time = time.time()
-                await asyncio.wait_for(timer.wait(), timeout=watchdog_timeout)
+                await wait_for(timer.wait(), timeout=watchdog_timeout)
                 total_time = time.time() - start_time
                 if enable_watchdog_logging:
                     logger.debug(f"{name} time between watchdog timer resets: {total_time:.20f}")

src/pipecat/utils/asyncio/timeout.py

@@ -0,0 +1,28 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Implementation of Python's `asyncio.wait_for()`.
+
+This module uses `wait_for2` package to implement `asyncio.wait_for()` for
+Python < 3.12.
+
+In Python 3.12, `asyncio.wait_for()` is implemented in terms of
+`asyncio.timeout()` which fixed a bunch of issues. However, this was never
+backported (because of the lack of `async.timeout()`) and there are still many
+remainig issues, specially in Python 3.10, in `async.wait_for()`.
+
+See https://github.com/python/cpython/pull/98518
+"""
+
+import asyncio
+import sys
+
+import wait_for2
+
+if sys.version_info >= (3, 12):
+    wait_for = asyncio.wait_for
+else:
+    wait_for = wait_for2.wait_for

src/pipecat/utils/asyncio/watchdog_async_iterator.py

@@ -15,6 +15,7 @@ import asyncio
 from typing import AsyncIterator, Optional
 
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.asyncio.timeout import wait_for
 
 
 class WatchdogAsyncIterator:
@@ -77,9 +78,8 @@ class WatchdogAsyncIterator:
                 if not self._current_anext_task:
                     self._current_anext_task = asyncio.create_task(self._iter.__anext__())
 
-                item = await asyncio.wait_for(
+                item = await wait_for(
-                    asyncio.shield(self._current_anext_task),
+                    asyncio.shield(self._current_anext_task), timeout=self._timeout
-                    timeout=self._timeout,
                 )
 
                 self._manager.task_reset_watchdog()

src/pipecat/utils/asyncio/watchdog_coroutine.py

@@ -16,6 +16,7 @@ from typing import Optional
 
 from pipecat.pipeline import task
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.asyncio.timeout import wait_for
 
 
 class WatchdogCoroutine:
@@ -59,9 +60,8 @@ class WatchdogCoroutine:
                 if not self._current_coro_task:
                     self._current_coro_task = asyncio.create_task(self._coroutine)
 
-                result = await asyncio.wait_for(
+                result = await wait_for(
-                    asyncio.shield(self._current_coro_task),
+                    asyncio.shield(self._current_coro_task), timeout=self._timeout
-                    timeout=self._timeout,
                 )
 
                 self._manager.task_reset_watchdog()

src/pipecat/utils/asyncio/watchdog_event.py

@@ -14,6 +14,7 @@ watchdog timeouts during legitimate waiting periods.
 import asyncio
 
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.asyncio.timeout import wait_for
 
 
 class WatchdogEvent(asyncio.Event):
@@ -55,7 +56,7 @@ class WatchdogEvent(asyncio.Event):
         """Wait for event while periodically resetting watchdog timer."""
         while True:
             try:
-                await asyncio.wait_for(super().wait(), timeout=self._timeout)
+                await wait_for(super().wait(), timeout=self._timeout)
                 self._manager.task_reset_watchdog()
                 return True
             except asyncio.TimeoutError:

src/pipecat/utils/asyncio/watchdog_priority_queue.py

@@ -17,6 +17,7 @@ from dataclasses import dataclass
 from loguru import logger
 
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.asyncio.timeout import wait_for
 
 
 @dataclass
@@ -124,7 +125,7 @@ class WatchdogPriorityQueue(asyncio.PriorityQueue):
         """Get item from queue while periodically resetting watchdog timer."""
         while True:
             try:
-                item = await asyncio.wait_for(super().get(), timeout=self._timeout)
+                item = await wait_for(super().get(), timeout=self._timeout)
                 self._manager.task_reset_watchdog()
                 return item
             except asyncio.TimeoutError:

src/pipecat/utils/asyncio/watchdog_queue.py

@@ -17,6 +17,7 @@ from dataclasses import dataclass
 from loguru import logger
 
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.asyncio.timeout import wait_for
 
 
 @dataclass
@@ -103,7 +104,7 @@ class WatchdogQueue(asyncio.Queue):
         """Get item from queue while periodically resetting watchdog timer."""
         while True:
             try:
-                item = await asyncio.wait_for(super().get(), timeout=self._timeout)
+                item = await wait_for(super().get(), timeout=self._timeout)
                 self._manager.task_reset_watchdog()
                 return item
             except asyncio.TimeoutError:

tests/test_pipeline.py

@@ -24,6 +24,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.filters.identity_filter import IdentityFilter
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.tests.utils import HeartbeatsObserver, run_test
+from pipecat.utils.asyncio.timeout import wait_for
 
 
 class TestPipeline(unittest.IsolatedAsyncioTestCase):
@@ -250,7 +251,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         await task.queue_frame(TextFrame(text="Hello Downstream!"))
 
         try:
-            await asyncio.wait_for(
+            await wait_for(
                 asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))),
                 timeout=1.0,
             )
@@ -286,7 +287,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
 
         await task.queue_frame(TextFrame(text="Hello!"))
         try:
-            await asyncio.wait_for(
+            await wait_for(
                 asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))),
                 timeout=1.0,
             )
@@ -306,7 +307,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase):
         pipeline = Pipeline([identity])
         task = PipelineTask(pipeline, idle_timeout_secs=0.2, cancel_on_idle_timeout=False)
         try:
-            await asyncio.wait_for(
+            await wait_for(
                 asyncio.shield(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))),
                 timeout=0.3,
             )

uv.lock

@@ -4195,6 +4195,7 @@ dependencies = [
     { name = "pyloudnorm" },
     { name = "resampy" },
     { name = "soxr" },
+    { name = "wait-for2", marker = "python_full_version < '3.12'" },
 ]
 
 [package.optional-dependencies]
@@ -4466,6 +4467,7 @@ requires-dist = [
     { name = "transformers", marker = "extra == 'ultravox'", specifier = ">=4.48.0" },
     { name = "uvicorn", marker = "extra == 'runner'", specifier = ">=0.32.0,<1.0.0" },
     { name = "vllm", marker = "extra == 'ultravox'", specifier = ">=0.9.0" },
+    { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" },
     { name = "websockets", marker = "extra == 'assemblyai'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'asyncai'", specifier = ">=13.1,<15.0" },
     { name = "websockets", marker = "extra == 'aws'", specifier = ">=13.1,<15.0" },
@@ -7316,6 +7318,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f4/72/c14ff1acac64294f45782769b9c8144a1c3e8d4f2228d4648197511b015a/vllm-0.9.2-cp38-abi3-manylinux1_x86_64.whl", hash = "sha256:f3c5da29a286f4933b480a5b4749fab226564f35c96928eeef547f88d385cd34", size = 383350132, upload-time = "2025-07-08T04:48:54.133Z" },
 ]
 
+[[package]]
+name = "wait-for2"
+version = "0.4.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/8f/7c/ea09d6a11990a8aa3ceac206fb7ea82366ea2c200caa87966611e0e18597/wait_for2-0.4.1.tar.gz", hash = "sha256:7f415415d21845c441391d6b4abe68f5959d2c0fbe927c2f61be28a297bc2acb", size = 17519, upload-time = "2025-06-13T19:45:00.081Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/90/56/0f88040567af7ff376ec9eaabe18fd980a4f5089d3bf8c7a32598ef06b8d/wait_for2-0.4.1-py3-none-any.whl", hash = "sha256:c694503e8c7420929e8a86bcffd9b00d55acaec2c14223a2b1e92bdc2ebf2154", size = 10985, upload-time = "2025-06-13T19:44:58.82Z" },
+]
+
 [[package]]
 name = "watchfiles"
 version = "1.1.0"