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 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 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. 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/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..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. @@ -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 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 866d0405f..44126a7a7 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() @@ -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 = "" 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 a0b32df31..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.""" @@ -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) 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 beefc8d1e..cc0350abf 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/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): diff --git a/src/pipecat/services/resembleai/tts.py b/src/pipecat/services/resembleai/tts.py index fc70d814b..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.""" @@ -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": diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index 4dca789e5..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. @@ -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. diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index da3164f20..a65a5e244 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 @@ -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 - 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: - 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) @@ -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): - """Append audio or control frame to an existing context. + async def append_to_audio_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. - 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") @@ -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 - # 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 ) @@ -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.