Merge pull request #4145 from pipecat-ai/filipi/tts_improvements_remove_reset

TTS improvements.

changelog/4145.fixed.2.md

@@ -0,0 +1 @@
+- Fixed websocket TTS word timestamps so interrupted contexts cannot leak stale words or backward PTS values into later turns.

changelog/4145.fixed.md

@@ -0,0 +1 @@
+- Fixed a race condition in `InterruptibleTTSService` where, if `run_tts` had been invoked but `BotStartedSpeakingFrame` had not yet been received, a user interruption could allow stale audio to leak through.

changelog/4145.removed.md

@@ -0,0 +1 @@
+- ⚠️ `TTSService.add_word_timestamps()` no longer supports the `"Reset"` and `"TTSStoppedFrame"` sentinel strings. If you have a custom TTS service that called `await self.add_word_timestamps([("Reset", 0)])` or `await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id)`, replace them with `await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))` and let `_handle_audio_context` manage the word-timestamp reset automatically.

src/pipecat/services/asyncai/tts.py

@@ -435,6 +435,7 @@ class AsyncAITTSService(WebsocketTTSService):
     async def on_audio_context_interrupted(self, context_id: str):
         """Close the Async AI context when the bot is interrupted."""
         await self._close_context(context_id)
+        await super().on_audio_context_interrupted(context_id)
 
     async def on_audio_context_completed(self, context_id: str):
         """Close the Async AI context after all audio has been played.
@@ -444,6 +445,7 @@ class AsyncAITTSService(WebsocketTTSService):
         ``close_context: True`` to free server-side resources.
         """
         await self._close_context(context_id)
+        await super().on_audio_context_completed(context_id)
 
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:

src/pipecat/services/azure/tts.py

@@ -611,8 +611,6 @@ class AzureTTSService(TTSService, AzureBaseTTSService):
         await super().push_frame(frame, direction)
         if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
             self._reset_state()
-            if isinstance(frame, TTSStoppedFrame) and self._current_context_id:
-                await self.add_word_timestamps([("Reset", 0)], self._current_context_id)
 
     def _reset_state(self):
         """Reset TTS state between turns."""

src/pipecat/services/cartesia/tts.py

@@ -574,6 +574,7 @@ class CartesiaTTSService(WebsocketTTSService):
         if context_id:
             cancel_msg = json.dumps({"context_id": context_id, "cancel": True})
             await self._get_websocket().send(cancel_msg)
+        await super().on_audio_context_interrupted(context_id)
 
     async def on_audio_context_completed(self, context_id: str):
         """Close the Cartesia context after all audio has been played.
@@ -582,7 +583,7 @@ class CartesiaTTSService(WebsocketTTSService):
         done once it has sent its ``done`` message, which is handled in
         ``_process_messages``.
         """
-        pass
+        await super().on_audio_context_completed(context_id)
 
     async def flush_audio(self, context_id: Optional[str] = None):
         """Flush any pending audio and finalize the current context.
@@ -606,7 +607,7 @@ class CartesiaTTSService(WebsocketTTSService):
             ctx_id = msg["context_id"]
             if msg["type"] == "done":
                 await self.stop_ttfb_metrics()
-                await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id)
+                await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))
                 await self.remove_audio_context(ctx_id)
             elif msg["type"] == "timestamps":
                 # Process the timestamps based on language before adding them

src/pipecat/services/deepgram/sagemaker/tts.py

@@ -309,6 +309,7 @@ class DeepgramSageMakerTTSService(TTSService):
                 await self._client.send_json({"type": "Clear"})
             except Exception as e:
                 logger.error(f"{self} error sending Clear message: {e}")
+        await super().on_audio_context_interrupted(context_id)
 
     async def flush_audio(self, context_id: Optional[str] = None):
         """Flush any pending audio synthesis by sending Flush command.

src/pipecat/services/deepgram/tts.py

@@ -277,6 +277,7 @@ class DeepgramTTSService(WebsocketTTSService):
                 await self._websocket.send(json.dumps({"type": "Clear"}))
             except Exception as e:
                 logger.error(f"{self} error sending Clear message: {e}")
+        await super().on_audio_context_interrupted(context_id)
 
     async def _receive_messages(self):
         """Receive and process messages from Deepgram WebSocket."""

src/pipecat/services/elevenlabs/tts.py

@@ -620,18 +620,6 @@ class ElevenLabsTTSService(WebsocketTTSService):
         msg = {"context_id": flush_id, "flush": True}
         await self._websocket.send(json.dumps(msg))
 
-    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        """Push a frame and handle state changes.
-
-        Args:
-            frame: The frame to push.
-            direction: The direction to push the frame.
-        """
-        await super().push_frame(frame, direction)
-        if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
-            if isinstance(frame, TTSStoppedFrame):
-                await self.add_word_timestamps([("Reset", 0)], self.get_active_audio_context_id())
-
     async def _connect(self):
         await super()._connect()
 
@@ -743,6 +731,7 @@ class ElevenLabsTTSService(WebsocketTTSService):
     async def on_audio_context_interrupted(self, context_id: str):
         """Close the ElevenLabs context when the bot is interrupted."""
         await self._close_context(context_id)
+        await super().on_audio_context_interrupted(context_id)
 
     async def on_audio_context_completed(self, context_id: str):
         """Close the ElevenLabs context after all audio has been played.
@@ -752,6 +741,7 @@ class ElevenLabsTTSService(WebsocketTTSService):
         ``close_context: True`` to free server-side resources.
         """
         await self._close_context(context_id)
+        await super().on_audio_context_completed(context_id)
 
     async def _receive_messages(self):
         """Handle incoming WebSocket messages from ElevenLabs."""
@@ -1130,10 +1120,6 @@ class ElevenLabsHttpTTSService(TTSService):
         if isinstance(frame, (InterruptionFrame, TTSStoppedFrame)):
             # Reset timing on interruption or stop
             self._reset_state()
-
-            if isinstance(frame, TTSStoppedFrame):
-                await self.add_word_timestamps([("Reset", 0)])
-
         elif isinstance(frame, LLMFullResponseEndFrame):
             # End of turn - reset previous text
             self._previous_text = ""

src/pipecat/services/fish/tts.py

@@ -363,6 +363,7 @@ class FishAudioTTSService(InterruptibleTTSService):
     async def on_audio_context_interrupted(self, context_id: str):
         """Stop all metrics when audio context is interrupted."""
         await self.stop_all_metrics()
+        await super().on_audio_context_interrupted(context_id)
 
     async def _receive_messages(self):
         async for message in self._get_websocket():

src/pipecat/services/gradium/tts.py

@@ -301,6 +301,7 @@ class GradiumTTSService(WebsocketTTSService):
         audio context no longer exists.
         """
         await self.stop_all_metrics()
+        await super().on_audio_context_interrupted(context_id)
 
     async def on_audio_context_completed(self, context_id: str):
         """Called after an audio context has finished playing all of its audio.
@@ -309,7 +310,7 @@ class GradiumTTSService(WebsocketTTSService):
         ``end_of_stream`` message (handled in ``_receive_messages``), after
         which the server-side context is already closed.
         """
-        pass
+        await super().on_audio_context_completed(context_id)
 
     async def _receive_messages(self):
         """Process incoming websocket messages, demultiplexing by client_req_id."""
@@ -342,7 +343,7 @@ class GradiumTTSService(WebsocketTTSService):
 
             elif msg["type"] == "end_of_stream":
                 if ctx_id and self.audio_context_available(ctx_id):
-                    await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id)
+                    await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))
                     await self.remove_audio_context(ctx_id)
                 if ctx_id:
                     self._setup_context_ids.discard(ctx_id)

src/pipecat/services/hume/tts.py

@@ -235,9 +235,6 @@ class HumeTTSService(TTSService):
             # Reset timing on interruption or stop
             self._reset_state()
 
-            if isinstance(frame, TTSStoppedFrame):
-                await self.add_word_timestamps([("Reset", 0)])
-
     async def update_setting(self, key: str, value: Any) -> None:
         """Runtime updates via key/value pair.
 

src/pipecat/services/inworld/tts.py

@@ -252,8 +252,6 @@ class InworldHttpTTSService(TTSService):
         await super().push_frame(frame, direction)
         if isinstance(frame, (InterruptionFrame, TTSStoppedFrame)):
             self._cumulative_time = 0.0
-            if isinstance(frame, TTSStoppedFrame):
-                await self.add_word_timestamps([("Reset", 0)])
 
     def _calculate_word_times(
         self,
@@ -759,8 +757,6 @@ class InworldTTSService(WebsocketTTSService):
             )
             self._cumulative_time = 0.0
             self._generation_end_time = 0.0
-            if isinstance(frame, TTSStoppedFrame):
-                await self.add_word_timestamps([("Reset", 0)])
 
     async def on_turn_context_created(self, context_id: str):
         """Eagerly open the context on the server when a new turn starts.
@@ -836,6 +832,7 @@ class InworldTTSService(WebsocketTTSService):
         """Callback invoked when an audio context has been interrupted."""
         await self._maybe_push_fallback_text(context_id)
         await self._close_context(context_id)
+        await super().on_audio_context_interrupted(context_id)
 
     async def on_audio_context_completed(self, context_id: str):
         """Callback invoked when an audio context has been completed."""
@@ -1059,7 +1056,7 @@ class InworldTTSService(WebsocketTTSService):
                 logger.trace(f"{self}: Context closed on server: {ctx_id}")
                 await self._maybe_push_fallback_text(ctx_id)
                 await self.stop_ttfb_metrics()
-                await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id)
+                await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))
                 await self.remove_audio_context(ctx_id)
 
     async def _keepalive_task_handler(self):

src/pipecat/services/resembleai/tts.py

@@ -258,6 +258,7 @@ class ResembleAITTSService(WebsocketTTSService):
     async def on_audio_context_interrupted(self, context_id: str):
         """Stop metrics when the bot is interrupted."""
         await self.stop_all_metrics()
+        await super().on_audio_context_interrupted(context_id)
 
     async def on_audio_context_completed(self, context_id: str):
         """Stop metrics after the Resemble AI context finishes playing.
@@ -266,7 +267,7 @@ class ResembleAITTSService(WebsocketTTSService):
         ``audio_end`` message (handled in ``_process_messages``), after which
         the server-side context is already closed.
         """
-        pass
+        await super().on_audio_context_completed(context_id)
 
     async def flush_audio(self, context_id: Optional[str] = None):
         """Flush any pending audio and finalize the current context."""
@@ -387,7 +388,9 @@ class ResembleAITTSService(WebsocketTTSService):
                     if request_id in self._request_id_to_context:
                         del self._request_id_to_context[request_id]
 
-                await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], context_id)
+                await self.append_to_audio_context(
+                    context_id, TTSStoppedFrame(context_id=context_id)
+                )
                 await self.remove_audio_context(context_id)
 
             elif msg_type == "error":

src/pipecat/services/rime/tts.py

@@ -516,6 +516,7 @@ class RimeTTSService(WebsocketTTSService):
     async def on_audio_context_interrupted(self, context_id: str):
         """Clear the Rime speech queue and stop metrics when the bot is interrupted."""
         await self._close_context(context_id)
+        await super().on_audio_context_interrupted(context_id)
 
     async def on_audio_context_completed(self, context_id: str):
         """Clear server-side state and stop metrics after the Rime context finishes playing.
@@ -525,6 +526,7 @@ class RimeTTSService(WebsocketTTSService):
         any residual server-side state once all audio has been delivered.
         """
         await self._close_context(context_id)
+        await super().on_audio_context_completed(context_id)
 
     def _calculate_word_times(self, words: list, starts: list, ends: list) -> list:
         """Calculate word timing pairs with proper spacing and punctuation.
@@ -604,18 +606,6 @@ class RimeTTSService(WebsocketTTSService):
                 await self.push_error(error_msg=f"Error: {msg['message']}")
                 self.reset_active_audio_context()
 
-    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        """Push frame and handle end-of-turn conditions.
-
-        Args:
-            frame: The frame to push.
-            direction: The direction to push the frame.
-        """
-        await super().push_frame(frame, direction)
-        if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
-            if isinstance(frame, TTSStoppedFrame):
-                await self.add_word_timestamps([("Reset", 0)])
-
     @traced_tts
     async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
         """Generate speech from text using Rime's streaming API.

src/pipecat/services/tts_service.py

@@ -342,7 +342,7 @@ class TTSService(AIService):
         self._initial_word_timestamp: int = -1
         self._initial_word_times: List[Tuple[str, float, Optional[str]]] = []
         # PTS of the last word frame pushed via _add_word_timestamps, used to assign
-        # correct PTS to sentinel frames ("TTSStoppedFrame", "Reset") that follow.
+        # correct PTS to TTSStoppedFrame and LLMFullResponseEndFrame.
         self._word_last_pts: int = 0
         self._llm_response_started: bool = False
         self._reuse_context_id_within_turn: bool = reuse_context_id_within_turn
@@ -1120,6 +1120,9 @@ class TTSService(AIService):
     async def reset_word_timestamps(self):
         """Reset word timestamp tracking."""
         self._initial_word_timestamp = -1
+        # Discard any pre-audio word timestamps from the interrupted turn so they
+        # cannot be flushed into the next context after the audio baseline resets.
+        self._initial_word_times = []
 
     async def add_word_timestamps(
         self, word_times: List[Tuple[str, float]], context_id: Optional[str] = None
@@ -1137,12 +1140,13 @@ class TTSService(AIService):
         """
         if context_id and self.audio_context_available(context_id):
             for word, timestamp in word_times:
-                await self._audio_contexts[context_id].put(
+                await self.append_to_audio_context(
+                    context_id,
                     _WordTimestampEntry(
                         word=word,
                         timestamp=timestamp,
                         context_id=context_id,
-                    )
+                    ),
                 )
         else:
             await self._add_word_timestamps(word_times=word_times, context_id=context_id)
@@ -1150,48 +1154,31 @@ class TTSService(AIService):
     async def _add_word_timestamps(
         self, word_times: List[Tuple[str, float]], context_id: Optional[str] = None
     ):
-        """Process word timestamps directly, building and pushing frames inline.
+        """Process word timestamps directly, building and pushing TTSTextFrames inline.
 
-        This is the single processing path for all word timestamp events, used both
+        Used both from _handle_audio_context (via _WordTimestampEntry) and from services
-        from _handle_audio_context (via _WordTimestampEntry) and from services that
+        that do not use audio contexts. Each entry emits a TTSTextFrame with a PTS
-        do not use audio contexts. Sentinel entries drive control-frame emission:
+        relative to the baseline established by start_word_timestamps().
 
-        - ("Reset", 0): reset timestamp baseline; emit LLMFullResponseEndFrame if needed.
+        When the baseline (_initial_word_timestamp) is not yet set, entries are cached
-        - ("TTSStoppedFrame", 0): emit TTSStoppedFrame.
+        in _initial_word_times and flushed once start_word_timestamps() is called
-        - Any other entry: emit TTSTextFrame with a PTS relative to the baseline.
+        (i.e. when the first audio chunk is received).
-
-        When the baseline (_initial_word_timestamp) is not yet set, regular word entries
-        are cached in _initial_word_times and flushed once start_word_timestamps() is
-        called (i.e. when the first audio chunk is received).
         """
         for word, timestamp in word_times:
-            if word == "Reset" and timestamp == 0:
+            ts_ns = seconds_to_nanoseconds(timestamp)
-                await self.reset_word_timestamps()
+            if self._initial_word_timestamp == -1:
-                if self._llm_response_started:
+                # Cache until we have audio and can compute PTS.
-                    self._llm_response_started = False
+                self._initial_word_times.append((word, timestamp, context_id))
-                    frame = LLMFullResponseEndFrame()
-                    frame.pts = self._word_last_pts
-                    await self.push_frame(frame)
-            elif word == "TTSStoppedFrame" and timestamp == 0:
-                frame = TTSStoppedFrame(context_id=context_id)
-                frame.pts = self._word_last_pts
-                frame.context_id = context_id
-                await self.push_frame(frame)
             else:
-                ts_ns = seconds_to_nanoseconds(timestamp)
+                # Assumption: word-by-word text frames don't include spaces, so
-                if self._initial_word_timestamp == -1:
+                # we can rely on the default includes_inter_frame_spaces=False
-                    # Cache until we have audio and can compute PTS.
+                frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD)
-                    self._initial_word_times.append((word, timestamp, context_id))
+                frame.pts = self._initial_word_timestamp + ts_ns
-                else:
+                frame.context_id = context_id
-                    # Assumption: word-by-word text frames don't include spaces, so
+                if context_id in self._tts_contexts:
-                    # we can rely on the default includes_inter_frame_spaces=False
+                    frame.append_to_context = self._tts_contexts[context_id].append_to_context
-                    frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD)
+                self._word_last_pts = frame.pts
-                    frame.pts = self._initial_word_timestamp + ts_ns
+                await self.push_frame(frame)
-                    frame.context_id = context_id
-                    if context_id in self._tts_contexts:
-                        frame.append_to_context = self._tts_contexts[context_id].append_to_context
-                    self._word_last_pts = frame.pts
-                    await self.push_frame(frame)
 
     #
     # Audio context methods (active when using websocket-based TTS with context management)
@@ -1207,12 +1194,19 @@ class TTSService(AIService):
         self._audio_contexts[context_id] = asyncio.Queue()
         logger.trace(f"{self} created audio context {context_id}")
 
-    async def append_to_audio_context(self, context_id: str, frame: Frame):
+    async def append_to_audio_context(
-        """Append audio or control frame to an existing context.
+        self, context_id: str, frame: Frame | _WordTimestampEntry | None
+    ):
+        """Append a frame or word-timestamp entry to an existing audio context queue.
+
+        Passing ``None`` signals end-of-context (used by remove_audio_context to mark
+        the queue for deletion). If the context no longer exists but the context_id
+        matches the active turn, the context is transparently recreated before appending.
 
         Args:
-            context_id: The context to append audio to.
+            context_id: The context to append to.
-            frame: The audio or control frame to append.
+            frame: The frame, word-timestamp entry, or ``None`` (end-of-context sentinel)
+                to append.
         """
         if not context_id:
             logger.debug(f"{self} unable to append audio to context: no context ID provided")
@@ -1220,7 +1214,8 @@ class TTSService(AIService):
         if self.audio_context_available(context_id):
             logger.trace(f"{self} appending audio {frame} to audio context {context_id}")
             await self._audio_contexts[context_id].put(frame)
-        elif context_id == self._turn_context_id:
+        # In case the frame is None, we should not recreate the context.
+        elif context_id == self._turn_context_id and frame:
             # Sometimes the HTTP service can take more than 3 seconds without sending any audio
             # So we are now recreating the context id while we are in the same turn
             logger.debug(f"{self} recreating audio context {context_id}")
@@ -1241,7 +1236,7 @@ class TTSService(AIService):
             # None. Once we reach None while handling audio we know we can
             # safely remove the context.
             logger.trace(f"{self} marking audio context {context_id} for deletion")
-            await self._audio_contexts[context_id].put(None)
+            await self.append_to_audio_context(context_id, None)
         else:
             logger.warning(f"{self} unable to remove context {context_id}")
 
@@ -1345,6 +1340,23 @@ class TTSService(AIService):
 
             self._serialization_queue.task_done()
 
+    async def _maybe_reset_word_timestamps(self):
+        """Reset word-timestamp state and emit LLMFullResponseEndFrame if needed.
+
+        Called at the end of an audio context (either on clean completion timeout or
+        when the context queue is drained). Resets the PTS baseline so the next turn
+        starts fresh. If an LLM response is still marked as in-progress and text frames
+        are not being pushed (which would have already emitted the frame), an
+        LLMFullResponseEndFrame is pushed with the PTS of the last word frame.
+        """
+        await self.reset_word_timestamps()
+        # If self._push_text_frames is True, we have already pushed the original LLMFullResponseEndFrame
+        if self._llm_response_started and not self._push_text_frames:
+            self._llm_response_started = False
+            frame = LLMFullResponseEndFrame()
+            frame.pts = self._word_last_pts
+            await self.push_frame(frame)
+
     async def _handle_audio_context(self, context_id: str):
         """Process items from an audio context queue until it is exhausted."""
         queue = self._audio_contexts[context_id]
@@ -1360,9 +1372,8 @@ class TTSService(AIService):
                 elif frame is None:
                     running = False
                 elif isinstance(frame, _WordTimestampEntry):
-                    # _add_word_timestamps is the single processing path: it handles
+                    # Route word timestamps through _add_word_timestamps so they are
-                    # sentinel entries ("Reset", "TTSStoppedFrame") and regular words
+                    # processed in playback order alongside audio frames.
-                    # inline, keeping all word-frame logic in one place.
                     await self._add_word_timestamps(
                         [(frame.word, frame.timestamp)], frame.context_id
                     )
@@ -1379,6 +1390,9 @@ class TTSService(AIService):
                         should_push_stop_frame = self._push_stop_frames
                     elif isinstance(frame, TTSStoppedFrame):
                         should_push_stop_frame = False
+                        # Setting the last word timestamp as the TTSStoppedFrame PTS
+                        if not frame.pts:
+                            frame.pts = self._word_last_pts
 
                     if isinstance(frame, ErrorFrame):
                         await self.push_error_frame(frame)
@@ -1393,6 +1407,7 @@ class TTSService(AIService):
 
         if should_push_stop_frame and self._push_stop_frames:
             await self.push_frame(TTSStoppedFrame(context_id=context_id))
+        await self._maybe_reset_word_timestamps()
 
     async def on_audio_context_interrupted(self, context_id: str):
         """Called when an audio context is cancelled due to an interruption.
@@ -1497,6 +1512,20 @@ class InterruptibleTTSService(WebsocketTTSService):
             await self._disconnect()
             await self._connect()
 
+    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        """Push a frame downstream with TTS-specific handling.
+
+        Args:
+            frame: The frame to push.
+            direction: The direction to push the frame.
+        """
+        # This prevents a race condition in cases where run_tts has been invoked but the
+        # BotStartedSpeakingFrame has not yet been received, which could allow stale audio to leak through.
+        if isinstance(frame, TTSStartedFrame):
+            self._bot_speaking = True
+
+        await super().push_frame(frame, direction)
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames with bot speaking state tracking.