add on_turn_context_created hook instead

src/pipecat/services/inworld/tts.py

@@ -56,7 +56,6 @@ from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
     InterruptionFrame,
-    LLMFullResponseStartFrame,
     StartFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
@@ -654,10 +653,10 @@ class InworldTTSService(WebsocketTTSService):
         # Track the end time of the last word in the current generation
         self._generation_end_time = 0.0
 
-        # Context ID that was pre-opened on the server during process_frame
-        # (LLMFullResponseStartFrame) to avoid context creation latency when
-        # enough context for TTS is available.
-        self._prewarmed_context_id: Optional[str] = None
+        # Context IDs already sent to the server via _send_context, used to
+        # make _send_context idempotent so on_turn_context_created can eagerly
+        # open contexts without causing duplicate creates in run_tts.
+        self._sent_context_ids: set[str] = set()
 
         # Init-only config (not runtime-updatable).
         self._audio_encoding = encoding
@@ -732,28 +731,16 @@ class InworldTTSService(WebsocketTTSService):
             if isinstance(frame, TTSStoppedFrame):
                 await self.add_word_timestamps([("Reset", 0)])
 
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process incoming frames and pre-open context on LLM response start.
+    async def on_turn_context_created(self, context_id: str):
+        """Eagerly open the context on the server when a new turn starts.
 
-        Eagerly sends the context configuration to the server when
-        LLMFullResponseStartFrame arrives, so the context is ready by the time
-        enough context for TTS is available. The base class assigns ``_turn_context_id`` before
-        this runs, which is reused for all ``run_tts`` calls within the turn.
+        This overlaps server-side context creation with sentence aggregation
+        time, so the context is ready by the time text arrives in run_tts.
         """
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, LLMFullResponseStartFrame):
-            if self._prewarmed_context_id:
-                try:
-                    await self._send_close_context(self._prewarmed_context_id)
-                except Exception as e:
-                    logger.warning(f"{self}: Failed to close previous prewarmed context: {e}")
-                self._prewarmed_context_id = None
-            try:
-                await self._send_context(self._turn_context_id)
-                self._prewarmed_context_id = self._turn_context_id
-            except Exception as e:
-                logger.warning(f"{self}: Failed to pre-open context: {e}")
+        try:
+            await self._send_context(context_id)
+        except Exception as e:
+            logger.warning(f"{self}: Failed to pre-open context: {e}")
 
     def _calculate_word_times(self, timestamp_info: Dict[str, Any]) -> List[Tuple[str, float]]:
         """Calculate word timestamps from Inworld WebSocket API response.
@@ -800,6 +787,7 @@ class InworldTTSService(WebsocketTTSService):
                 await self._send_close_context(context_id)
             except Exception as e:
                 await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
+        self._sent_context_ids.discard(context_id)
         self._cumulative_time = 0.0
         self._generation_end_time = 0.0
 
@@ -916,7 +904,7 @@ class InworldTTSService(WebsocketTTSService):
         finally:
             await self.remove_active_audio_context()
             self._websocket = None
-            self._prewarmed_context_id = None
+            self._sent_context_ids.clear()
             self._cumulative_time = 0.0
             self._generation_end_time = 0.0
             await self._call_event_handler("on_disconnected")
@@ -961,8 +949,11 @@ class InworldTTSService(WebsocketTTSService):
                 await self.push_error(error_msg=str(msg["error"]))
                 continue
 
+            # Handle context created confirmation
+            if "contextCreated" in result:
+                logger.trace(f"{self}: Context created on server: {ctx_id}")
             # If the context isn't available recreate it (handles race conditions during interruption recovery).
-            if ctx_id and not self.audio_context_available(ctx_id):
+            elif ctx_id and not self.audio_context_available(ctx_id):
                 logger.trace(f"{self}: Recreating audio context for current context: {ctx_id}")
                 await self.create_audio_context(ctx_id)
 
@@ -987,10 +978,6 @@ class InworldTTSService(WebsocketTTSService):
                 if word_times:
                     await self.add_word_timestamps(word_times, ctx_id)
 
-            # Handle context created confirmation
-            if "contextCreated" in result:
-                logger.trace(f"{self}: Context created on server: {ctx_id}")
-
             # Handle flush completion, which indicates the end of a generation
             if "flushCompleted" in result:
                 logger.trace(
@@ -1031,15 +1018,15 @@ class InworldTTSService(WebsocketTTSService):
     async def _send_context(self, context_id: str):
         """Send a context to the Inworld WebSocket TTS service.
 
-        Skips the send if this context was already pre-opened on the server
-        (prewarmed during process_frame).
+        Idempotent: skips the send if this context was already opened on the
+        server (e.g., eagerly via on_turn_context_created).
 
         Args:
             context_id: The context ID.
         """
-        if context_id == self._prewarmed_context_id:
-            self._prewarmed_context_id = None
+        if context_id in self._sent_context_ids:
             return
+        self._sent_context_ids.add(context_id)
 
         audio_config = {
             "audioEncoding": self._audio_encoding,

src/pipecat/services/tts_service.py

@@ -633,6 +633,18 @@ class TTSService(AIService):
 
         await self.queue_frame(TTSSpeakFrame(text))
 
+    async def on_turn_context_created(self, context_id: str):
+        """Called when a new turn context ID has been created.
+
+        Override to perform provider-specific setup (e.g., eagerly opening a
+        server-side context) before text starts flowing. This is called from
+        ``process_frame`` when an ``LLMFullResponseStartFrame`` arrives.
+
+        Args:
+            context_id: The newly created turn context ID.
+        """
+        pass
+
     async def on_turn_context_completed(self):
         """Handle the completion of a turn."""
         # For HTTP services they emit the frames synchronously, so close the audio context here
@@ -685,6 +697,7 @@ class TTSService(AIService):
             self._llm_response_started = True
             # New LLM turn → assign a fresh context ID shared by all sentences
             self._turn_context_id = self.create_context_id()
+            await self.on_turn_context_created(self._turn_context_id)
             await self.push_frame(frame, direction)
         elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
             # We pause processing incoming frames if the LLM response included