From dea7c22020dad6e87902ea34a126ff113ab395e4 Mon Sep 17 00:00:00 2001
From: Filipi Fuchter <filipi87@gmail.com>
Date: Tue, 2 Sep 2025 13:58:41 -0300
Subject: [PATCH] Fixed an issue where the pipeline could freeze.

---
 CHANGELOG.md                              |  6 ++++++
 src/pipecat/processors/frame_processor.py | 10 +++++++++-
 src/pipecat/services/deepgram/stt.py      | 12 +++++++++++-
 3 files changed, 26 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 60e4151af..dbc383391 100644
--- a/CHANGELOG.md
+++ b/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.
@@ -70,6 +73,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.
 
 ## [0.0.82] - 2025-08-28
diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index 641883cf5..cac1a0806 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/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):
diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py
index 8381135d9..e2fea34fb 100644
--- a/src/pipecat/services/deepgram/stt.py
+++ b/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."""