From 5cc9b7e0d164776ebbb0c73e9af75b68f627fef1 Mon Sep 17 00:00:00 2001
From: Mark Backman <mark@daily.co>
Date: Fri, 20 Jun 2025 15:46:18 -0400
Subject: [PATCH] Fix: Correctly close the context for ElevenLabsTTSService

---
 CHANGELOG.md                           |  3 ++
 src/pipecat/services/elevenlabs/tts.py | 39 ++++++++++++--------------
 2 files changed, 21 insertions(+), 21 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 466b3f28b..aecab04f4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -41,6 +41,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue with `ElevenLabsTTSService` where the context was not being
+  closed.
+
 - Fixed function calling in `AWSNovaSonicLLMService`.
 
 - Fixed an issue that would cause multiple `PipelineTask.on_idle_timeout`
diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py
index e0301360e..f48ba5552 100644
--- a/src/pipecat/services/elevenlabs/tts.py
+++ b/src/pipecat/services/elevenlabs/tts.py
@@ -284,7 +284,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         logger.trace(f"{self}: flushing audio")
         msg = {"context_id": self._context_id, "flush": True}
         await self._websocket.send(json.dumps(msg))
-        self._context_id = None
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         await super().push_frame(frame, direction)
@@ -380,6 +379,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         if self._context_id and self._websocket:
             logger.trace(f"Closing context {self._context_id} due to interruption")
             try:
+                # ElevenLabs requires that Pipecat manages the contexts and closes them
+                # when they're not longer in use. Since a StartInterruptionFrame is pushed
+                # every time the user speaks, we'll use this as a trigger to close the context
+                # and reset the state.
+                # Note: We do not need to call remove_audio_context here, as the context is
+                # automatically reset when super ()._handle_interruption is called.
                 await self._websocket.send(
                     json.dumps({"context_id": self._context_id, "close_context": True})
                 )
@@ -391,10 +396,20 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
     async def _receive_messages(self):
         async for message in self._get_websocket():
             msg = json.loads(message)
-            # Check if this message belongs to the current context
+
             received_ctx_id = msg.get("contextId")
+
+            # Handle final messages first, regardless of context availability
+            # At the moment, this message is received AFTER the close_context message is
+            # sent, so it doesn't serve any functional purpose. For now, we'll just log it.
+            if msg.get("isFinal") is True:
+                logger.trace(f"Received final message for context {received_ctx_id}")
+                continue
+
+            # Check if this message belongs to the current context.
+            # This should never happen, so warn about it.
             if not self.audio_context_available(received_ctx_id):
-                logger.trace(f"Ignoring message from unavailable context: {received_ctx_id}")
+                logger.warning(f"Ignoring message from unavailable context: {received_ctx_id}")
                 continue
 
             if msg.get("audio"):
@@ -408,13 +423,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
                 await self.add_word_timestamps(word_times)
                 self._cumulative_time = word_times[-1][1]
-            if msg.get("isFinal"):
-                logger.trace(f"Received final message for context {received_ctx_id}")
-                await self.remove_audio_context(received_ctx_id)
-                # Reset context tracking if this was our active context
-                if self._context_id == received_ctx_id:
-                    self._context_id = None
-                    self._started = False
 
     async def _keepalive_task_handler(self):
         while True:
@@ -441,14 +449,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 await self._connect()
 
             try:
-                # Close previous context if there was one
-                if self._context_id and not self._started:
-                    await self._websocket.send(
-                        json.dumps({"context_id": self._context_id, "close_context": True})
-                    )
-                    await self.remove_audio_context(self._context_id)
-                    self._context_id = None
-
                 if not self._started:
                     await self.start_ttfb_metrics()
                     yield TTSStartedFrame()
@@ -473,9 +473,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 logger.error(f"{self} error sending message: {e}")
                 yield TTSStoppedFrame()
                 self._started = False
-                if self._context_id:
-                    await self.remove_audio_context(self._context_id)
-                    self._context_id = None
                 return
             yield None
         except Exception as e: