From c4253a7d98fd6f013d119608d4b7cc5bb3979deb Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 25 Mar 2026 17:21:55 -0300 Subject: [PATCH 1/9] Refactoring to invoke append_to_audio_context instead of direct queue put. --- src/pipecat/services/tts_service.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index da3164f20..ba07eb483 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -1137,12 +1137,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) @@ -1207,7 +1208,9 @@ 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( + self, context_id: str, frame: Frame | _WordTimestampEntry | None + ): """Append audio or control frame to an existing context. Args: @@ -1241,7 +1244,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}") From b4096f9a111fe8c398684dd314e01ca9cb5e1025 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 25 Mar 2026 17:47:24 -0300 Subject: [PATCH 2/9] Refactoring to remove "Reset" and "TTSStoppedFrame" from word. --- src/pipecat/services/azure/tts.py | 2 - src/pipecat/services/cartesia/tts.py | 2 +- src/pipecat/services/elevenlabs/tts.py | 16 -------- src/pipecat/services/gradium/tts.py | 2 +- src/pipecat/services/hume/tts.py | 3 -- src/pipecat/services/inworld/tts.py | 6 +-- src/pipecat/services/resembleai/tts.py | 4 +- src/pipecat/services/rime/tts.py | 12 ------ src/pipecat/services/tts_service.py | 57 +++++++++++++------------- 9 files changed, 35 insertions(+), 69 deletions(-) diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index a9491e9aa..79dc8a2e1 100644 --- a/src/pipecat/services/azure/tts.py +++ b/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.""" diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index b713a0d9a..9e6073b57 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -606,7 +606,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 diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 866d0405f..32da01c81 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/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() @@ -1130,10 +1118,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 = "" diff --git a/src/pipecat/services/gradium/tts.py b/src/pipecat/services/gradium/tts.py index a0b32df31..1c80a2ece 100644 --- a/src/pipecat/services/gradium/tts.py +++ b/src/pipecat/services/gradium/tts.py @@ -342,7 +342,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) diff --git a/src/pipecat/services/hume/tts.py b/src/pipecat/services/hume/tts.py index e591ebec2..ea43a530d 100644 --- a/src/pipecat/services/hume/tts.py +++ b/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. diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index e8db2ca33..44361ebd7 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -248,8 +248,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, @@ -728,8 +726,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. @@ -996,7 +992,7 @@ class InworldTTSService(WebsocketTTSService): if "contextClosed" in result: logger.trace(f"{self}: Context closed on server: {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): diff --git a/src/pipecat/services/resembleai/tts.py b/src/pipecat/services/resembleai/tts.py index fc70d814b..f9fd6549e 100644 --- a/src/pipecat/services/resembleai/tts.py +++ b/src/pipecat/services/resembleai/tts.py @@ -387,7 +387,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": diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index 4dca789e5..a359986b6 100644 --- a/src/pipecat/services/rime/tts.py +++ b/src/pipecat/services/rime/tts.py @@ -604,18 +604,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. diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index ba07eb483..c63607c4a 100644 --- a/src/pipecat/services/tts_service.py +++ b/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 @@ -1166,33 +1166,20 @@ class TTSService(AIService): called (i.e. when the first audio chunk is received). """ for word, timestamp in word_times: - if word == "Reset" and timestamp == 0: - await self.reset_word_timestamps() - if self._llm_response_started: - self._llm_response_started = False - 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) + ts_ns = seconds_to_nanoseconds(timestamp) + if self._initial_word_timestamp == -1: + # Cache until we have audio and can compute PTS. + self._initial_word_times.append((word, timestamp, context_id)) else: - ts_ns = seconds_to_nanoseconds(timestamp) - if self._initial_word_timestamp == -1: - # Cache until we have audio and can compute PTS. - self._initial_word_times.append((word, timestamp, context_id)) - else: - # Assumption: word-by-word text frames don't include spaces, so - # we can rely on the default includes_inter_frame_spaces=False - frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD) - frame.pts = self._initial_word_timestamp + ts_ns - 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) + # Assumption: word-by-word text frames don't include spaces, so + # we can rely on the default includes_inter_frame_spaces=False + frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD) + frame.pts = self._initial_word_timestamp + ts_ns + 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) @@ -1223,7 +1210,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}") @@ -1348,6 +1336,15 @@ class TTSService(AIService): self._serialization_queue.task_done() + async def _maybe_reset_word_timestamps(self): + 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] @@ -1382,6 +1379,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) @@ -1392,6 +1392,7 @@ class TTSService(AIService): logger.trace(f"{self} time out on audio context {context_id}") 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() break if should_push_stop_frame and self._push_stop_frames: From 2ff4b3f4a341039d90ba8996639055a070734414 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 25 Mar 2026 17:52:05 -0300 Subject: [PATCH 3/9] Improving docstring based on the recent changes. --- src/pipecat/services/tts_service.py | 44 +++++++++++++++++------------ 1 file changed, 26 insertions(+), 18 deletions(-) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index c63607c4a..8195b4f86 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -1151,19 +1151,15 @@ 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 - from _handle_audio_context (via _WordTimestampEntry) and from services that - do not use audio contexts. Sentinel entries drive control-frame emission: + Used both from _handle_audio_context (via _WordTimestampEntry) and from services + that do not use audio contexts. Each entry emits a TTSTextFrame with a PTS + relative to the baseline established by start_word_timestamps(). - - ("Reset", 0): reset timestamp baseline; emit LLMFullResponseEndFrame if needed. - - ("TTSStoppedFrame", 0): emit TTSStoppedFrame. - - Any other entry: emit TTSTextFrame with a PTS relative to the baseline. - - 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). + When the baseline (_initial_word_timestamp) is not yet set, 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: ts_ns = seconds_to_nanoseconds(timestamp) @@ -1198,11 +1194,16 @@ class TTSService(AIService): async def append_to_audio_context( self, context_id: str, frame: Frame | _WordTimestampEntry | None ): - """Append audio or control frame to an existing context. + """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. - frame: The audio or control frame to append. + context_id: The context to append to. + 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") @@ -1337,6 +1338,14 @@ 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: @@ -1360,9 +1369,8 @@ class TTSService(AIService): elif frame is None: running = False elif isinstance(frame, _WordTimestampEntry): - # _add_word_timestamps is the single processing path: it handles - # sentinel entries ("Reset", "TTSStoppedFrame") and regular words - # inline, keeping all word-frame logic in one place. + # Route word timestamps through _add_word_timestamps so they are + # processed in playback order alongside audio frames. await self._add_word_timestamps( [(frame.word, frame.timestamp)], frame.context_id ) @@ -1392,11 +1400,11 @@ class TTSService(AIService): logger.trace(f"{self} time out on audio context {context_id}") 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() break 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. From f7ec6befe1ac138b4143294df8b08984ec115a59 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 26 Mar 2026 10:14:21 -0300 Subject: [PATCH 4/9] Invoking superclass method when audio context is interrupted or completed. --- src/pipecat/services/asyncai/tts.py | 2 ++ src/pipecat/services/cartesia/tts.py | 3 ++- src/pipecat/services/deepgram/sagemaker/tts.py | 1 + src/pipecat/services/deepgram/tts.py | 1 + src/pipecat/services/elevenlabs/tts.py | 2 ++ src/pipecat/services/fish/tts.py | 1 + src/pipecat/services/gradium/tts.py | 3 ++- src/pipecat/services/inworld/tts.py | 1 + src/pipecat/services/resembleai/tts.py | 3 ++- src/pipecat/services/rime/tts.py | 2 ++ 10 files changed, 16 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index d2ac74445..0f140281d 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/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]: diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 9e6073b57..d879ff694 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/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. diff --git a/src/pipecat/services/deepgram/sagemaker/tts.py b/src/pipecat/services/deepgram/sagemaker/tts.py index 36541b4be..6f43475dc 100644 --- a/src/pipecat/services/deepgram/sagemaker/tts.py +++ b/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. diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index 9f2dc3976..3b5d04202 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/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.""" diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 32da01c81..44126a7a7 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -731,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. @@ -740,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.""" diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index ab57522d4..6333ac1c4 100644 --- a/src/pipecat/services/fish/tts.py +++ b/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(): diff --git a/src/pipecat/services/gradium/tts.py b/src/pipecat/services/gradium/tts.py index 1c80a2ece..e8d69a7da 100644 --- a/src/pipecat/services/gradium/tts.py +++ b/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.""" diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index 44361ebd7..a47308d39 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -800,6 +800,7 @@ class InworldTTSService(WebsocketTTSService): async def on_audio_context_interrupted(self, context_id: str): """Callback invoked when an audio context has been interrupted.""" await self._close_context(context_id) + await super().on_audio_context_interrupted(context_id) def _get_websocket(self): """Get the websocket for the Inworld WebSocket TTS service. diff --git a/src/pipecat/services/resembleai/tts.py b/src/pipecat/services/resembleai/tts.py index f9fd6549e..d4bb77fb0 100644 --- a/src/pipecat/services/resembleai/tts.py +++ b/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.""" diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index a359986b6..e02933e26 100644 --- a/src/pipecat/services/rime/tts.py +++ b/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. From df602b900d9ceac7520ad6019f665d2bce98de2f Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 26 Mar 2026 10:39:44 -0300 Subject: [PATCH 5/9] Preventing a race condition in the InterruptibleTTSServices in cases where run_tts has been invoked but the BotStartedSpeakingFrame has not yet been received. --- src/pipecat/services/tts_service.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 8195b4f86..1e34ab79a 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -1509,6 +1509,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. From c4466ba67876904a9901479a13d2331e30300257 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 26 Mar 2026 10:58:57 -0300 Subject: [PATCH 6/9] Adding changelog for the InterruptibleTTSService race condition fix --- changelog/4145.fixed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4145.fixed.md diff --git a/changelog/4145.fixed.md b/changelog/4145.fixed.md new file mode 100644 index 000000000..831e1fcd7 --- /dev/null +++ b/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. \ No newline at end of file From a06bf47ed2b6a77afaaa9bf2bc055d9332474169 Mon Sep 17 00:00:00 2001 From: namanbansal013 Date: Thu, 26 Mar 2026 11:42:24 -0300 Subject: [PATCH 7/9] Discard any pre-audio word timestamps from the interrupted turn. --- src/pipecat/services/tts_service.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 1e34ab79a..a65a5e244 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -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 From d2869912578df1df4d538b43e24f3cf3069301f6 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Thu, 26 Mar 2026 11:51:31 -0300 Subject: [PATCH 8/9] Changelog entry for the changes involving add_word_timestamp. --- changelog/4145.removed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4145.removed.md diff --git a/changelog/4145.removed.md b/changelog/4145.removed.md new file mode 100644 index 000000000..ac780c1dc --- /dev/null +++ b/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. From 800fd6a916be4602bb9cbff622dd09429e928123 Mon Sep 17 00:00:00 2001 From: namanbansal013 Date: Thu, 26 Mar 2026 11:52:34 -0300 Subject: [PATCH 9/9] Changelog entry for the websocket word context leak. --- changelog/4145.fixed.2.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/4145.fixed.2.md diff --git a/changelog/4145.fixed.2.md b/changelog/4145.fixed.2.md new file mode 100644 index 000000000..4fceb9bb1 --- /dev/null +++ b/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. \ No newline at end of file