processors(rtvi): fix task cleanup

src/pipecat/processors/async_frame_processor.py

@@ -59,5 +59,6 @@ class AsyncFrameProcessor(FrameProcessor):
                 (frame, direction) = await self._push_queue.get()
                 await self.push_frame(frame, direction)
                 running = not isinstance(frame, EndFrame)
+                self._push_queue.task_done()
             except asyncio.CancelledError:
                 break

src/pipecat/processors/frameworks/rtvi.py

@@ -316,6 +316,7 @@ class RTVIProcessor(FrameProcessor):
             try:
                 (frame, direction) = await self._frame_queue.get()
                 await self._handle_frame(frame, direction)
+                self._frame_queue.task_done()
             except asyncio.CancelledError:
                 break
 

src/pipecat/services/cartesia.py

@@ -114,9 +114,11 @@ class CartesiaTTSService(TTSService):
         try:
             if self._context_appending_task:
                 self._context_appending_task.cancel()
+                await self._context_appending_task
                 self._context_appending_task = None
             if self._receive_task:
                 self._receive_task.cancel()
+                await self._receive_task
                 self._receive_task = None
             if self._websocket:
                 ws = self._websocket

src/pipecat/transports/base_input.py

@@ -104,6 +104,7 @@ class BaseInputTransport(FrameProcessor):
             try:
                 (frame, direction) = await self._push_queue.get()
                 await self.push_frame(frame, direction)
+                self._push_queue.task_done()
             except asyncio.CancelledError:
                 break
 
@@ -185,6 +186,8 @@ class BaseInputTransport(FrameProcessor):
                 # Push audio downstream if passthrough.
                 if audio_passthrough:
                     await self._internal_push_frame(frame)
+
+                self._audio_in_queue.task_done()
             except asyncio.CancelledError:
                 break
             except Exception as e:

src/pipecat/transports/base_output.py

@@ -204,6 +204,7 @@ class BaseOutputTransport(FrameProcessor):
             try:
                 (frame, direction) = await self._push_queue.get()
                 await self.push_frame(frame, direction)
+                self._push_queue.task_done()
             except asyncio.CancelledError:
                 break