Merge pull request #2552 from pipecat-ai/filipi/freeze_issues

Fixed an issue where the pipeline could freeze.

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added a timeout around cancel input tasks to prevent indefinite 
+  hangs when cancellation is swallowed by third-party code.
+
 - Added `pipecat.extensions.ivr` for automated IVR system navigation with
   configurable goals and conversation handling. Supports DTMF input, verbal
   responses, and intelligent menu traversal.
@@ -72,6 +75,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue where Deepgram swallowed `asyncio.CancelledError` during 
+  disconnect, preventing tasks from being cancelled.
+
 - Fixed an issue where `PipelineTask` was not cleaning up the observers.
 
 ### Performance

src/pipecat/processors/frame_processor.py

@@ -125,6 +125,11 @@ class FrameProcessorQueue(asyncio.PriorityQueue):
         return item
 
 
+# Timeout in seconds for cancelling the input frame processing task.
+# This prevents hanging if a library swallows asyncio.CancelledError.
+INPUT_TASK_CANCEL_TIMEOUT_SECS = 3
+
+
 class FrameProcessor(BaseObject):
     """Base class for all frame processors in the pipeline.
 
@@ -742,7 +747,10 @@ class FrameProcessor(BaseObject):
     async def __cancel_input_task(self):
         """Cancel the frame input processing task."""
         if self.__input_frame_task:
-            await self.cancel_task(self.__input_frame_task)
+            # Apply a timeout as a safeguard: if a library swallows asyncio.CancelledError,
+            # the task would otherwise never be cancelled. With a timeout, we can detect this
+            # situation and surface it in the logs instead of hanging indefinitely.
+            await self.cancel_task(self.__input_frame_task, INPUT_TASK_CANCEL_TIMEOUT_SECS)
             self.__input_frame_task = None
 
     def __create_process_task(self):

src/pipecat/services/deepgram/stt.py

@@ -6,6 +6,7 @@
 
 """Deepgram speech-to-text service implementation."""
 
+import asyncio
 from typing import AsyncGenerator, Dict, Optional
 
 from loguru import logger
@@ -237,7 +238,16 @@ class DeepgramSTTService(STTService):
     async def _disconnect(self):
         if self._connection.is_connected:
             logger.debug("Disconnecting from Deepgram")
-            await self._connection.finish()
+            result = await self._connection.finish()
+            # Deepgram swallows asyncio.CancelledError internally and returns False instead.
+            # This prevents proper cancellation propagation: tasks awaiting on queue.get()
+            # would remain stuck if we didn’t normalize this back into a CancelledError.
+            # GH issue: https://github.com/deepgram/deepgram-python-sdk/issues/570
+            if not result:
+                logger.warning(f"{self}: Deepgram connection failed to disconnect, forcing cancel.")
+                raise asyncio.CancelledError(
+                    f"{self}: Deepgram connection cancelled during disconnect"
+                )
 
     async def start_metrics(self):
         """Start TTFB and processing metrics collection."""