Merge branch 'main' into filipi/includes_inter_frame_spaces

# Conflicts:
#	uv.lock

src/pipecat/audio/krisp_instance.py

@@ -15,9 +15,11 @@ from loguru import logger
 try:
     import krisp_audio
 except ModuleNotFoundError as e:
-    logger.error(f"Exception: {e}")
-    logger.error("In order to use the Krisp instance, you need to install krisp_audio.")
-    raise Exception(f"Missing module: {e}")
+    raise ImportError(
+        "krisp_audio is required for Krisp audio features. "
+        "Install it to use KrispVivaFilter, KrispVivaVadAnalyzer, "
+        "KrispVivaTurn, or KrispVivaIPUserTurnStartStrategy."
+    ) from e
 
 
 # Mapping of sample rates (Hz) to Krisp SDK SamplingRate enums

src/pipecat/audio/turn/krisp_viva_turn.py

@@ -7,7 +7,9 @@
 """Krisp turn analyzer for end-of-turn detection using Krisp VIVA SDK.
 
 This module provides a turn analyzer implementation using Krisp's turn detection
-(Tt) API to determine when a user has finished speaking in a conversation.
+v3 (Tt) API to determine when a user has finished speaking in a conversation.
+The Tt API accepts an external VAD flag alongside audio frames, allowing the
+model to leverage voice activity information for more accurate turn detection.
 
 Note: This analyzer uses a different model than KrispVivaFilter. The model path
 can be specified via the KRISP_VIVA_TURN_MODEL_PATH environment variable or
@@ -33,7 +35,7 @@ try:
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use KrispVivaTurn, you need to install krisp_audio.")
-    raise Exception(f"Missing module: {e}")
+    raise ImportError(f"Missing module: {e}") from e
 
 
 class KrispTurnParams(BaseTurnParams):
@@ -53,8 +55,10 @@ class KrispTurnParams(BaseTurnParams):
 class KrispVivaTurn(BaseTurnAnalyzer):
     """Turn analyzer using Krisp VIVA SDK for end-of-turn detection.
 
-    Uses Krisp's turn detection (Tt) API to determine when a user has finished
-    speaking. This analyzer requires a valid Krisp model file to operate.
+    Uses Krisp's turn detection v3 (Tt) API to determine when a user has
+    finished speaking. The Tt API receives an external VAD flag with each
+    audio frame, which the ``is_speech`` parameter of ``append_audio``
+    provides. This analyzer requires a valid Krisp model file to operate.
     """
 
     def __init__(
@@ -158,14 +162,14 @@ class KrispVivaTurn(BaseTurnAnalyzer):
         """Create a turn detection session with the specified sample rate.
 
         Args:
-            sample_rate: Sample rate for the session
+            sample_rate: Sample rate for the session.
 
         Returns:
-            krisp_audio.TtFloat instance
+            krisp_audio.TtFloat instance.
 
         Raises:
-            ValueError: If sample rate or frame duration is not supported
-            RuntimeError: If session creation fails
+            ValueError: If sample rate or frame duration is not supported.
+            RuntimeError: If session creation fails.
         """
         try:
             model_info = krisp_audio.ModelInfo()
@@ -306,12 +310,7 @@ class KrispVivaTurn(BaseTurnAnalyzer):
                 # Instead, we wait for the model's probability check below to confirm
                 # end-of-turn based on the threshold.
 
-                prob = self._tt_session.process(frame.tolist())
-
-                # Negative values indicate the model is not ready yet (working with 100ms data)
-                # Skip processing until we get positive probabilities
-                if prob < 0:
-                    continue
+                prob = self._tt_session.process(frame.tolist(), is_speech, False)
 
                 # Store the probability for external access
                 self._last_probability = prob

src/pipecat/extensions/ivr/ivr_navigator.py

@@ -30,6 +30,7 @@ from pipecat.frames.frames import (
     VADParamsUpdateFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
+from pipecat.processors.aggregators.llm_context import LLMContextMessage
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.llm_service import LLMService
 from pipecat.utils.text.pattern_pair_aggregator import (
@@ -87,7 +88,7 @@ class IVRProcessor(FrameProcessor):
         self._classifier_prompt = classifier_prompt
 
         # Store saved context messages
-        self._saved_messages: list[dict] = []
+        self._saved_messages: list[LLMContextMessage] = []
 
         # XML pattern aggregation
         self._aggregator = PatternPairAggregator()
@@ -97,18 +98,18 @@ class IVRProcessor(FrameProcessor):
         self._register_event_handler("on_conversation_detected")
         self._register_event_handler("on_ivr_status_changed")
 
-    def update_saved_messages(self, messages: list[dict]) -> None:
+    def update_saved_messages(self, messages: list[LLMContextMessage]) -> None:
         """Update the saved context messages.
 
         Sets the messages that are saved when switching between
         conversation and IVR navigation modes.
 
         Args:
-            messages: List of message dictionaries to save.
+            messages: List of context messages to save.
         """
         self._saved_messages = messages
 
-    def _get_conversation_history(self) -> list[dict]:
+    def _get_conversation_history(self) -> list[LLMContextMessage]:
         """Get saved context messages without the system message.
 
         Returns:
@@ -144,7 +145,9 @@ class IVRProcessor(FrameProcessor):
             await self.push_frame(frame, direction)
 
             # Set the classifier prompt and push it upstream
-            messages = [{"role": "system", "content": self._classifier_prompt}]
+            messages: list[LLMContextMessage] = [
+                {"role": "developer", "content": self._classifier_prompt}
+            ]
             llm_update_frame = LLMMessagesUpdateFrame(messages=messages)
             await self.push_frame(llm_update_frame, FrameDirection.UPSTREAM)
 
@@ -261,7 +264,7 @@ class IVRProcessor(FrameProcessor):
         logger.debug("IVR detected - switching to IVR navigation mode")
 
         # Create new context with IVR system prompt and saved messages
-        messages = [{"role": "system", "content": self._ivr_prompt}]
+        messages: list[LLMContextMessage] = [{"role": "developer", "content": self._ivr_prompt}]
 
         # Add saved conversation history if available
         conversation_history = self._get_conversation_history()

src/pipecat/extensions/voicemail/voicemail_detector.py

@@ -35,7 +35,7 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline
-from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
 from pipecat.processors.aggregators.llm_response_universal import (
     LLMContextAggregatorPair,
     LLMUserAggregatorParams,
@@ -617,9 +617,9 @@ VOICEMAIL SYSTEM (respond "VOICEMAIL"):
             self._validate_prompt(custom_system_prompt)
 
         # Set up the LLM context with the classification prompt
-        self._messages = [
+        self._messages: list[LLMContextMessage] = [
             {
-                "role": "system",
+                "role": "developer",
                 "content": self._prompt,
             },
         ]

src/pipecat/frames/frames.py

@@ -20,7 +20,6 @@ from typing import (
     TYPE_CHECKING,
     Any,
     Literal,
-    Optional,
 )
 
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
@@ -559,11 +558,11 @@ class LLMMessagesAppendFrame(DataFrame):
     current context.
 
     Parameters:
-        messages: List of message dictionaries to append.
+        messages: List of context messages to append.
         run_llm: Whether the context update should be sent to the LLM.
     """
 
-    messages: list[dict]
+    messages: list[LLMContextMessage]
     run_llm: bool | None = None
 
 
@@ -575,11 +574,11 @@ class LLMMessagesUpdateFrame(DataFrame):
     context LLM messages.
 
     Parameters:
-        messages: List of message dictionaries to replace current context.
+        messages: List of context messages to replace current context.
         run_llm: Whether the context update should be sent to the LLM.
     """
 
-    messages: list[dict]
+    messages: list[LLMContextMessage]
     run_llm: bool | None = None
 
 

src/pipecat/observers/startup_timing_observer.py

@@ -49,6 +49,15 @@ from pipecat.processors.frame_processor import FrameProcessor
 _INTERNAL_TYPES = (PipelineSource, BasePipeline)
 
 
+@dataclass
+class _StartFrameInfo:
+    """Captured once when the first StartFrame arrives at a processor."""
+
+    frame_id: int
+    arrival_ns: int
+    wall_clock: float
+
+
 @dataclass
 class _ArrivalInfo:
     """Internal record of when a StartFrame arrived at a processor."""
@@ -175,8 +184,8 @@ class StartupTimingObserver(BaseObserver):
         # Collected timings in pipeline order.
         self._timings: list[ProcessorStartupTiming] = []
 
-        # Lock onto the first StartFrame we see (by frame ID).
-        self._start_frame_id: str | None = None
+        # Captured once when the first StartFrame arrives.
+        self._start_frame: _StartFrameInfo | None = None
 
         # Whether we've already emitted the startup timing report.
         self._startup_timing_reported = False
@@ -184,15 +193,9 @@ class StartupTimingObserver(BaseObserver):
         # Whether we've already measured transport timing.
         self._transport_timing_reported = False
 
-        # Timestamp (ns) when we first see a StartFrame arrive at a processor.
-        self._start_frame_arrival_ns: int | None = None
-
         # Bot connected timing (stored for inclusion in the transport report).
         self._bot_connected_secs: float | None = None
 
-        # Wall clock time when the StartFrame was first seen.
-        self._start_wall_clock: float | None = None
-
         self._register_event_handler("on_startup_timing_report")
         self._register_event_handler("on_transport_timing_report")
 
@@ -233,11 +236,13 @@ class StartupTimingObserver(BaseObserver):
             return
 
         # Lock onto the first StartFrame.
-        if self._start_frame_id is None:
-            self._start_frame_id = data.frame.id
-            self._start_frame_arrival_ns = data.timestamp
-            self._start_wall_clock = time.time()
-        elif data.frame.id != self._start_frame_id:
+        if self._start_frame is None:
+            self._start_frame = _StartFrameInfo(
+                frame_id=data.frame.id,
+                arrival_ns=data.timestamp,
+                wall_clock=time.time(),
+            )
+        elif data.frame.id != self._start_frame.frame_id:
             return
 
         if self._should_track(data.processor):
@@ -268,16 +273,16 @@ class StartupTimingObserver(BaseObserver):
         if not isinstance(data.frame, StartFrame):
             return
 
-        if self._start_frame_id is not None and data.frame.id != self._start_frame_id:
+        if self._start_frame is not None and data.frame.id != self._start_frame.frame_id:
             return
 
         arrival = self._arrivals.pop(data.source.id, None)
-        if arrival is None:
+        if arrival is None or self._start_frame is None:
             return
 
         duration_ns = data.timestamp - arrival.arrival_ts_ns
         duration_secs = duration_ns / 1e9
-        start_offset_secs = (arrival.arrival_ts_ns - self._start_frame_arrival_ns) / 1e9
+        start_offset_secs = (arrival.arrival_ts_ns - self._start_frame.arrival_ns) / 1e9
 
         self._timings.append(
             ProcessorStartupTiming(
@@ -289,22 +294,22 @@ class StartupTimingObserver(BaseObserver):
 
     def _handle_bot_connected(self, data: FramePushed):
         """Record bot connected timing on first BotConnectedFrame."""
-        if self._bot_connected_secs is not None or self._start_frame_arrival_ns is None:
+        if self._bot_connected_secs is not None or self._start_frame is None:
             return
 
-        delta_ns = data.timestamp - self._start_frame_arrival_ns
+        delta_ns = data.timestamp - self._start_frame.arrival_ns
         self._bot_connected_secs = delta_ns / 1e9
 
     async def _handle_client_connected(self, data: FramePushed):
         """Emit transport timing report on first ClientConnectedFrame."""
-        if self._transport_timing_reported or self._start_frame_arrival_ns is None:
+        if self._transport_timing_reported or self._start_frame is None:
             return
 
         self._transport_timing_reported = True
-        delta_ns = data.timestamp - self._start_frame_arrival_ns
+        delta_ns = data.timestamp - self._start_frame.arrival_ns
         client_connected_secs = delta_ns / 1e9
         report = TransportTimingReport(
-            start_time=self._start_wall_clock or 0.0,
+            start_time=self._start_frame.wall_clock,
             bot_connected_secs=self._bot_connected_secs,
             client_connected_secs=client_connected_secs,
         )
@@ -319,7 +324,7 @@ class StartupTimingObserver(BaseObserver):
         total = sum(t.duration_secs for t in self._timings)
 
         report = StartupTimingReport(
-            start_time=self._start_wall_clock or 0.0,
+            start_time=self._start_frame.wall_clock if self._start_frame else 0.0,
             total_duration_secs=total,
             processor_timings=self._timings,
         )

src/pipecat/pipeline/llm_switcher.py

@@ -6,7 +6,7 @@
 
 """LLM switcher for switching between different LLMs at runtime, with different switching strategies."""
 
-from typing import Any
+from typing import Any, cast
 
 from pipecat.adapters.schemas.direct_function import DirectFunction
 from pipecat.pipeline.service_switcher import (
@@ -15,6 +15,7 @@ from pipecat.pipeline.service_switcher import (
     StrategyType,
 )
 from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.services.llm_service import LLMService
 
 
@@ -38,7 +39,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
             strategy_type: The strategy class to use for switching between LLMs.
                 Defaults to ``ServiceSwitcherStrategyManual``.
         """
-        super().__init__(llms, strategy_type)
+        super().__init__(cast(list[FrameProcessor], llms), strategy_type)
 
     @property
     def llms(self) -> list[LLMService]:
@@ -47,7 +48,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
         Returns:
             List of LLM services managed by this switcher.
         """
-        return self.services
+        return cast(list[LLMService], self.services)
 
     @property
     def active_llm(self) -> LLMService:
@@ -56,7 +57,7 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
         Returns:
             The currently active LLM service, or None if no LLM is active.
         """
-        return self.strategy.active_service
+        return cast(LLMService, self.strategy.active_service)
 
     async def run_inference(self, context: LLMContext, **kwargs) -> str | None:
         """Run a one-shot, out-of-band (i.e. out-of-pipeline) inference with the given LLM context, using the currently active LLM.

src/pipecat/pipeline/pipeline.py

@@ -11,7 +11,7 @@ in sequence and manages frame flow between them, along with helper classes
 for pipeline source and sink operations.
 """
 
-from collections.abc import Callable, Coroutine
+from collections.abc import Callable, Coroutine, Sequence
 
 from pipecat.frames.frames import Frame
 from pipecat.pipeline.base_pipeline import BasePipeline
@@ -98,7 +98,7 @@ class Pipeline(BasePipeline):
 
     def __init__(
         self,
-        processors: list[FrameProcessor],
+        processors: Sequence[FrameProcessor],
         *,
         source: FrameProcessor | None = None,
         sink: FrameProcessor | None = None,
@@ -106,7 +106,7 @@ class Pipeline(BasePipeline):
         """Initialize the pipeline with a list of processors.
 
         Args:
-            processors: List of frame processors to connect in sequence.
+            processors: Sequence of frame processors to connect in sequence.
             source: An optional pipeline source processor.
             sink: An optional pipeline sink processor.
         """
@@ -116,7 +116,7 @@ class Pipeline(BasePipeline):
         # downstream outside of the pipeline.
         self._source = source or PipelineSource(self.push_frame, name=f"{self}::Source")
         self._sink = sink or PipelineSink(self.push_frame, name=f"{self}::Sink")
-        self._processors: list[FrameProcessor] = [self._source] + processors + [self._sink]
+        self._processors: list[FrameProcessor] = [self._source, *processors, self._sink]
 
         self._link_processors()
 

src/pipecat/pipeline/task.py

@@ -742,7 +742,7 @@ class PipelineTask(BasePipelineTask):
         await self._observer.cleanup()
 
         # End conversation tracing if it's active - this will also close any active turn span
-        if self._enable_tracing and hasattr(self, "_turn_trace_observer"):
+        if self._enable_tracing and self._turn_trace_observer:
             self._turn_trace_observer.end_conversation_tracing()
 
         # Cleanup pipeline processors.

src/pipecat/pipeline/task_observer.py

@@ -173,6 +173,8 @@ class TaskObserver(BaseObserver):
         return proxies
 
     async def _send_to_proxy(self, data: Any):
+        if not self._proxies:
+            return
         for proxy in self._proxies.values():
             await proxy.queue.put(data)
 

src/pipecat/processors/metrics/sentry.py

@@ -97,13 +97,20 @@ class SentryMetrics(FrameProcessorMetrics):
 
         Args:
             end_time: Optional end timestamp override.
+
+        Returns:
+            MetricsFrame produced by the base class, or None if not measuring.
+            Returning the frame is required so ``FrameProcessor.stop_ttfb_metrics``
+            can push it downstream to observers.
         """
-        await super().stop_ttfb_metrics(end_time=end_time)
+        frame = await super().stop_ttfb_metrics(end_time=end_time)
 
         if self._sentry_available and self._ttfb_metrics_tx:
             await self._sentry_queue.put(self._ttfb_metrics_tx)
             self._ttfb_metrics_tx = None
 
+        return frame
+
     async def start_processing_metrics(self, *, start_time: float | None = None):
         """Start tracking frame processing metrics.
 
@@ -126,13 +133,20 @@ class SentryMetrics(FrameProcessorMetrics):
 
         Args:
             end_time: Optional end timestamp override.
+
+        Returns:
+            MetricsFrame produced by the base class, or None if not measuring.
+            Returning the frame is required so ``FrameProcessor.stop_processing_metrics``
+            can push it downstream to observers.
         """
-        await super().stop_processing_metrics(end_time=end_time)
+        frame = await super().stop_processing_metrics(end_time=end_time)
 
         if self._sentry_available and self._processing_metrics_tx:
             await self._sentry_queue.put(self._processing_metrics_tx)
             self._processing_metrics_tx = None
 
+        return frame
+
     async def _sentry_task_handler(self):
         """Background task handler for completing Sentry transactions."""
         running = True

src/pipecat/runner/daily.py

@@ -228,7 +228,7 @@ async def configure(
             room_properties.enable_dialout = True
 
         # Add SIP configuration if enabled
-        if sip_enabled:
+        if sip_enabled and sip_caller_phone:
             sip_params = DailyRoomSipParams(
                 display_name=sip_caller_phone,
                 video=sip_enable_video,

src/pipecat/runner/run.py

@@ -156,6 +156,8 @@ def _get_bot_module():
                 spec = importlib.util.spec_from_file_location(
                     module_name, os.path.join(cwd, filename)
                 )
+                if spec is None or spec.loader is None:
+                    continue
                 module = importlib.util.module_from_spec(spec)
                 spec.loader.exec_module(module)
 
@@ -386,29 +388,27 @@ def _add_lifespan_to_app(app: FastAPI, new_lifespan):
 
 def _setup_whatsapp_routes(app: FastAPI, args: argparse.Namespace):
     """Set up WhatsApp-specific routes."""
-    WHATSAPP_APP_SECRET = os.getenv("WHATSAPP_APP_SECRET")
-    WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID")
-    WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN")
-    WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN")
-
-    if not all(
-        [
-            WHATSAPP_APP_SECRET,
-            WHATSAPP_PHONE_NUMBER_ID,
-            WHATSAPP_TOKEN,
-            WHATSAPP_WEBHOOK_VERIFICATION_TOKEN,
-        ]
-    ):
+    required_vars = [
+        "WHATSAPP_APP_SECRET",
+        "WHATSAPP_PHONE_NUMBER_ID",
+        "WHATSAPP_TOKEN",
+        "WHATSAPP_WEBHOOK_VERIFICATION_TOKEN",
+    ]
+    missing = [v for v in required_vars if not os.getenv(v)]
+    if missing:
+        missing_list = "\n    ".join(missing)
         logger.error(
-            """Missing required environment variables for WhatsApp transport:
-    WHATSAPP_APP_SECRET
-    WHATSAPP_PHONE_NUMBER_ID
-    WHATSAPP_TOKEN
-    WHATSAPP_WEBHOOK_VERIFICATION_TOKEN
+            f"""Missing required environment variables for WhatsApp transport:
+    {missing_list}
             """
         )
         return
 
+    WHATSAPP_APP_SECRET = os.environ["WHATSAPP_APP_SECRET"]
+    WHATSAPP_PHONE_NUMBER_ID = os.environ["WHATSAPP_PHONE_NUMBER_ID"]
+    WHATSAPP_TOKEN = os.environ["WHATSAPP_TOKEN"]
+    WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.environ["WHATSAPP_WEBHOOK_VERIFICATION_TOKEN"]
+
     try:
         from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
         from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest

src/pipecat/runner/types.py

@@ -122,4 +122,4 @@ class LiveKitRunnerArguments(RunnerArguments):
 
     room_name: str
     url: str
-    token: str | None = None
+    token: str

src/pipecat/runner/utils.py

@@ -416,9 +416,9 @@ def _get_transport_params(transport_key: str, transport_params: dict[str, Callab
 
 async def _create_telephony_transport(
     websocket: WebSocket,
-    params: Any | None = None,
-    transport_type: str = None,
-    call_data: dict = None,
+    params: Any,
+    transport_type: str,
+    call_data: dict,
 ) -> BaseTransport:
     """Create a telephony transport with pre-parsed WebSocket data.
 
@@ -433,12 +433,6 @@ async def _create_telephony_transport(
     """
     from pipecat.transports.websocket.fastapi import FastAPIWebsocketTransport
 
-    if params is None:
-        raise ValueError(
-            "FastAPIWebsocketParams must be provided. "
-            "The serializer and add_wav_header will be set automatically."
-        )
-
     # Always set add_wav_header to False for telephony
     params.add_wav_header = False
 

src/pipecat/serializers/exotel.py

@@ -59,7 +59,9 @@ class ExotelFrameSerializer(FrameSerializer):
             call_sid: The associated Exotel Call SID (optional, not used in this implementation).
             params: Configuration parameters.
         """
-        super().__init__(params or ExotelFrameSerializer.InputParams())
+        params = params or ExotelFrameSerializer.InputParams()
+        super().__init__(params)
+        self._params: ExotelFrameSerializer.InputParams = params
 
         self._stream_sid = stream_sid
         self._call_sid = call_sid

src/pipecat/serializers/genesys.py

@@ -166,7 +166,9 @@ class GenesysAudioHookSerializer(FrameSerializer):
             params: Configuration parameters.
             **kwargs: Additional arguments passed to BaseObject (e.g., name).
         """
-        super().__init__(params or GenesysAudioHookSerializer.InputParams(), **kwargs)
+        params = params or GenesysAudioHookSerializer.InputParams()
+        super().__init__(params, **kwargs)
+        self._params: GenesysAudioHookSerializer.InputParams = params
 
         self._genesys_sample_rate = self._params.genesys_sample_rate
         self._sample_rate = 0  # Pipeline input rate, set in setup()

src/pipecat/serializers/plivo.py

@@ -8,6 +8,7 @@
 
 import base64
 import json
+from typing import cast
 
 from loguru import logger
 
@@ -71,7 +72,24 @@ class PlivoFrameSerializer(FrameSerializer):
             auth_token: Plivo auth token (required for auto hang-up).
             params: Configuration parameters.
         """
-        super().__init__(params or PlivoFrameSerializer.InputParams())
+        params = params or PlivoFrameSerializer.InputParams()
+        super().__init__(params)
+        self._params: PlivoFrameSerializer.InputParams = params
+
+        # Validate hangup-related parameters if auto_hang_up is enabled
+        if self._params.auto_hang_up:
+            missing_credentials = []
+            if not call_id:
+                missing_credentials.append("call_id")
+            if not auth_id:
+                missing_credentials.append("auth_id")
+            if not auth_token:
+                missing_credentials.append("auth_token")
+
+            if missing_credentials:
+                raise ValueError(
+                    f"auto_hang_up is enabled but missing required parameters: {', '.join(missing_credentials)}"
+                )
 
         self._stream_id = stream_id
         self._call_id = call_id
@@ -152,23 +170,11 @@ class PlivoFrameSerializer(FrameSerializer):
         try:
             import aiohttp
 
-            auth_id = self._auth_id
-            auth_token = self._auth_token
-            call_id = self._call_id
-
-            if not call_id or not auth_id or not auth_token:
-                missing = []
-                if not call_id:
-                    missing.append("call_id")
-                if not auth_id:
-                    missing.append("auth_id")
-                if not auth_token:
-                    missing.append("auth_token")
-
-                logger.warning(
-                    f"Cannot hang up Plivo call: missing required parameters: {', '.join(missing)}"
-                )
-                return
+            # __init__ guarantees these are non-None whenever auto_hang_up is True,
+            # which is the only path that reaches this method.
+            auth_id = cast(str, self._auth_id)
+            auth_token = cast(str, self._auth_token)
+            call_id = cast(str, self._call_id)
 
             # Plivo API endpoint for hanging up calls
             endpoint = f"https://api.plivo.com/v1/Account/{auth_id}/Call/{call_id}/"

src/pipecat/serializers/protobuf.py

@@ -83,23 +83,24 @@ class ProtobufFrameSerializer(FrameSerializer):
             Serialized frame as bytes, or None if frame type is not serializable.
         """
         # Wrapping this messages as a JSONFrame to send
+        serializable: Frame | MessageFrame = frame
         if isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
             if self.should_ignore_frame(frame):
                 return None
-            frame = MessageFrame(
+            serializable = MessageFrame(
                 data=json.dumps(frame.message),
             )
 
-        proto_frame = frame_protos.Frame()
-        if type(frame) not in self.SERIALIZABLE_TYPES:
-            logger.warning(f"Frame type {type(frame)} is not serializable")
+        proto_frame = frame_protos.Frame()  # type: ignore[attr-defined]
+        if type(serializable) not in self.SERIALIZABLE_TYPES:
+            logger.warning(f"Frame type {type(serializable)} is not serializable")
             return None
 
         # ignoring linter errors; we check that type(frame) is in this dict above
-        proto_optional_name = self.SERIALIZABLE_TYPES[type(frame)]  # type: ignore
+        proto_optional_name = self.SERIALIZABLE_TYPES[type(serializable)]  # type: ignore
         proto_attr = getattr(proto_frame, proto_optional_name)
-        for field in dataclasses.fields(frame):  # type: ignore
-            value = getattr(frame, field.name)
+        for field in dataclasses.fields(serializable):  # type: ignore
+            value = getattr(serializable, field.name)
             if value and hasattr(proto_attr, field.name):
                 setattr(proto_attr, field.name, value)
 
@@ -114,7 +115,7 @@ class ProtobufFrameSerializer(FrameSerializer):
         Returns:
             Deserialized frame instance, or None if deserialization fails.
         """
-        proto = frame_protos.Frame.FromString(data)
+        proto = frame_protos.Frame.FromString(data)  # type: ignore[attr-defined]
         which = proto.WhichOneof("frame")
         if which not in self.DESERIALIZABLE_FIELDS:
             logger.error("Unable to deserialize a valid frame")

src/pipecat/serializers/telnyx.py

@@ -8,10 +8,10 @@
 
 import base64
 import json
+from typing import cast
 
 import aiohttp
 from loguru import logger
-from pydantic import BaseModel
 
 from pipecat.audio.dtmf.types import KeypadEntry
 from pipecat.audio.utils import (
@@ -46,7 +46,7 @@ class TelnyxFrameSerializer(FrameSerializer):
     credentials to be provided.
     """
 
-    class InputParams(BaseModel):
+    class InputParams(FrameSerializer.InputParams):
         """Configuration parameters for TelnyxFrameSerializer.
 
         Parameters:
@@ -82,10 +82,26 @@ class TelnyxFrameSerializer(FrameSerializer):
             api_key: Your Telnyx API key (required for auto hang-up).
             params: Configuration parameters.
         """
+        params = params or TelnyxFrameSerializer.InputParams()
+        super().__init__(params)
+        self._params: TelnyxFrameSerializer.InputParams = params
+
+        # Validate hangup-related parameters if auto_hang_up is enabled
+        if self._params.auto_hang_up:
+            missing_credentials = []
+            if not call_control_id:
+                missing_credentials.append("call_control_id")
+            if not api_key:
+                missing_credentials.append("api_key")
+
+            if missing_credentials:
+                raise ValueError(
+                    f"auto_hang_up is enabled but missing required parameters: {', '.join(missing_credentials)}"
+                )
+
         self._stream_id = stream_id
         self._call_control_id = call_control_id
         self._api_key = api_key
-        self._params = params or TelnyxFrameSerializer.InputParams()
         self._params.outbound_encoding = outbound_encoding
         self._params.inbound_encoding = inbound_encoding
 
@@ -163,14 +179,10 @@ class TelnyxFrameSerializer(FrameSerializer):
     async def _hang_up_call(self):
         """Hang up the Telnyx call using Telnyx's REST API."""
         try:
-            call_control_id = self._call_control_id
-            api_key = self._api_key
-
-            if not call_control_id or not api_key:
-                logger.warning(
-                    "Cannot hang up Telnyx call: call_control_id and api_key must be provided"
-                )
-                return
+            # __init__ guarantees these are non-None whenever auto_hang_up is True,
+            # which is the only path that reaches this method.
+            call_control_id = cast(str, self._call_control_id)
+            api_key = cast(str, self._api_key)
 
             # Telnyx API endpoint for hanging up a call
             endpoint = f"https://api.telnyx.com/v2/calls/{call_control_id}/actions/hangup"

src/pipecat/serializers/twilio.py

@@ -8,6 +8,7 @@
 
 import base64
 import json
+from typing import cast
 
 from loguru import logger
 
@@ -75,7 +76,9 @@ class TwilioFrameSerializer(FrameSerializer):
             edge: Twilio edge location (e.g., "sydney", "dublin"). Must be specified with region.
             params: Configuration parameters.
         """
-        super().__init__(params or TwilioFrameSerializer.InputParams())
+        params = params or TwilioFrameSerializer.InputParams()
+        super().__init__(params)
+        self._params: TwilioFrameSerializer.InputParams = params
 
         # Validate hangup-related parameters if auto_hang_up is enabled
         if self._params.auto_hang_up:
@@ -178,9 +181,11 @@ class TwilioFrameSerializer(FrameSerializer):
         try:
             import aiohttp
 
-            account_sid = self._account_sid
-            auth_token = self._auth_token
-            call_sid = self._call_sid
+            # __init__ guarantees these are non-None whenever auto_hang_up is True,
+            # which is the only path that reaches this method.
+            account_sid = cast(str, self._account_sid)
+            auth_token = cast(str, self._auth_token)
+            call_sid = cast(str, self._call_sid)
             region = self._region
             edge = self._edge
 

src/pipecat/serializers/vonage.py

@@ -54,7 +54,9 @@ class VonageFrameSerializer(FrameSerializer):
         Args:
             params: Configuration parameters.
         """
-        super().__init__(params or VonageFrameSerializer.InputParams())
+        params = params or VonageFrameSerializer.InputParams()
+        super().__init__(params)
+        self._params: VonageFrameSerializer.InputParams = params
 
         self._vonage_sample_rate = self._params.vonage_sample_rate
         self._sample_rate = 0  # Pipeline input rate

src/pipecat/services/deepgram/flux/base.py

@@ -27,11 +27,59 @@ from pipecat.frames.frames import (
 )
 from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven
 from pipecat.services.stt_service import STTService
-from pipecat.transcriptions.language import Language
+from pipecat.transcriptions.language import Language, resolve_language
 from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.service_decorators import traced_stt
 
 
+def language_to_deepgram_flux_language(language: Language) -> str | None:
+    """Convert a Pipecat Language to a Deepgram Flux language code.
+
+    Only honored by the ``flux-general-multi`` model. Locale variants
+    (e.g. ``Language.EN_GB``) fall back to the base code.
+    """
+    LANGUAGE_MAP = {
+        Language.DE: "de",
+        Language.EN: "en",
+        Language.ES: "es",
+        Language.FR: "fr",
+        Language.HI: "hi",
+        Language.IT: "it",
+        Language.JA: "ja",
+        Language.NL: "nl",
+        Language.PT: "pt",
+        Language.RU: "ru",
+    }
+    return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
+
+
+def _prepare_language_hints(hints: list[Language] | None) -> list[str]:
+    """Convert a list of Pipecat Languages to Deepgram Flux codes.
+
+    Drops entries that can't be mapped and deduplicates while preserving order.
+    """
+    if not hints:
+        return []
+    seen: set[str] = set()
+    prepared: list[str] = []
+    for hint in hints:
+        code = language_to_deepgram_flux_language(hint)
+        if code is None or code in seen:
+            continue
+        seen.add(code)
+        prepared.append(code)
+    return prepared
+
+
+def _code_to_pipecat_language(code: str) -> Language | None:
+    """Convert a Deepgram-returned language code to a Pipecat Language."""
+    try:
+        return Language(code)
+    except ValueError:
+        logger.debug(f"Unmapped Deepgram Flux detected language code: {code}")
+        return None
+
+
 class FluxMessageType(StrEnum):
     """Deepgram Flux WebSocket message types.
 
@@ -73,6 +121,10 @@ class DeepgramFluxSTTSettings(STTSettings):
             confidence (default 5000).
         keyterm: Keyterms to boost recognition accuracy for specialized terminology.
         min_confidence: Minimum confidence required to create a TranscriptionFrame.
+        language_hints: Languages to bias transcription toward. Only honored by the
+            ``flux-general-multi`` model. An empty list clears any active hints;
+            ``None``/``NOT_GIVEN`` means no hints (auto-detect). Can be updated
+            mid-stream via ``STTUpdateSettingsFrame``.
     """
 
     eager_eot_threshold: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
@@ -80,6 +132,7 @@ class DeepgramFluxSTTSettings(STTSettings):
     eot_timeout_ms: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
     keyterm: list | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
     min_confidence: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    language_hints: list[Language] | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
 
 
 class DeepgramFluxSTTBase(STTService):
@@ -93,7 +146,14 @@ class DeepgramFluxSTTBase(STTService):
 
     Settings = DeepgramFluxSTTSettings
     _settings: Settings
-    _CONFIGURE_FIELDS = {"keyterm", "eot_threshold", "eager_eot_threshold", "eot_timeout_ms"}
+    _CONFIGURE_FIELDS = {
+        "keyterm",
+        "eot_threshold",
+        "eager_eot_threshold",
+        "eot_timeout_ms",
+        "language_hints",
+    }
+    _MULTILINGUAL_MODEL = "flux-general-multi"
 
     def __init__(
         self,
@@ -200,6 +260,18 @@ class DeepgramFluxSTTBase(STTService):
         for tag_value in self._tag:
             params.append(urlencode({"tag": tag_value}))
 
+        # Add language_hint parameters (only valid on flux-general-multi)
+        hints = self._settings.language_hints
+        if hints and not isinstance(hints, _NotGiven):
+            if self._settings.model == self._MULTILINGUAL_MODEL:
+                for code in _prepare_language_hints(hints):
+                    params.append(urlencode({"language_hint": code}))
+            else:
+                logger.warning(
+                    f"language_hints only supported on {self._MULTILINGUAL_MODEL}; "
+                    f"ignoring hints for model {self._settings.model!r}"
+                )
+
         return "&".join(params)
 
     async def _send_silence(self, duration_secs: float = 0.5):
@@ -266,6 +338,21 @@ class DeepgramFluxSTTBase(STTService):
         if thresholds:
             message["thresholds"] = thresholds
 
+        if "language_hints" in fields:
+            if self._settings.model != self._MULTILINGUAL_MODEL:
+                logger.warning(
+                    f"language_hints only supported on {self._MULTILINGUAL_MODEL}; "
+                    f"skipping Configure update for model {self._settings.model!r}"
+                )
+            else:
+                hints = self._settings.language_hints
+                # Empty list clears hints; NOT_GIVEN/None also treated as clear
+                # since we only reach this branch when the user set the field.
+                if hints is None or isinstance(hints, _NotGiven):
+                    message["language_hints"] = []
+                else:
+                    message["language_hints"] = _prepare_language_hints(hints)
+
         logger.debug(f"{self}: sending Configure message: {message}")
         await self._transport_send_json(message)
 
@@ -281,8 +368,9 @@ class DeepgramFluxSTTBase(STTService):
         """Apply a settings delta.
 
         Configure-able fields (keyterm, eot_threshold, eager_eot_threshold,
-        eot_timeout_ms) are sent to Deepgram via a Configure message.
-        Other fields are stored but cannot be applied to the active connection.
+        eot_timeout_ms, language_hints) are sent to Deepgram via a Configure
+        message. Other fields are stored but cannot be applied to the active
+        connection.
         """
         changed = await super()._update_settings(delta)
 
@@ -520,6 +608,20 @@ class DeepgramFluxSTTBase(STTService):
             return None
         return sum(confidences) / len(confidences)
 
+    def _primary_detected_language(self, data: dict[str, Any]) -> Language | None:
+        """Extract the primary detected language from a TurnInfo payload.
+
+        On ``flux-general-multi`` the language is read from TurnInfo's
+        ``languages`` field. On ``flux-general-en`` the field is absent, so we
+        fall back to ``Language.EN`` to match the model's fixed language.
+        """
+        codes = data.get("languages") or []
+        if codes:
+            return _code_to_pipecat_language(codes[0])
+        if self._settings.model == "flux-general-en":
+            return Language.EN
+        return None
+
     async def _handle_end_of_turn(self, transcript: str, data: dict[str, Any]):
         """Handle EndOfTurn events from Deepgram Flux.
 
@@ -543,6 +645,7 @@ class DeepgramFluxSTTBase(STTService):
 
         # Compute the average confidence
         average_confidence = self._calculate_average_confidence(data)
+        detected_language = self._primary_detected_language(data)
 
         if not self._settings.min_confidence or average_confidence > self._settings.min_confidence:
             # EndOfTurn means Flux has determined the turn is complete,
@@ -552,7 +655,7 @@ class DeepgramFluxSTTBase(STTService):
                     transcript,
                     self._user_id,
                     time_now_iso8601(),
-                    self._settings.language,
+                    detected_language,
                     result=data,
                     finalized=True,
                 )
@@ -562,7 +665,7 @@ class DeepgramFluxSTTBase(STTService):
                 f"Transcription confidence below min_confidence threshold: {average_confidence}"
             )
 
-        await self._handle_transcription(transcript, True, self._settings.language)
+        await self._handle_transcription(transcript, True, detected_language)
         await self.stop_processing_metrics()
         await self.broadcast_frame(UserStoppedSpeakingFrame)
         await self._call_event_handler("on_end_of_turn", transcript)
@@ -606,7 +709,7 @@ class DeepgramFluxSTTBase(STTService):
                 transcript,
                 self._user_id,
                 time_now_iso8601(),
-                self._settings.language,
+                self._primary_detected_language(data),
                 result=data,
             )
         )

src/pipecat/services/deepgram/flux/sagemaker/stt.py

@@ -23,7 +23,6 @@ from pipecat.services.deepgram.flux.base import (
     DeepgramFluxSTTBase,
     DeepgramFluxSTTSettings,
 )
-from pipecat.transcriptions.language import Language
 
 
 @dataclass
@@ -112,12 +111,13 @@ class DeepgramFluxSageMakerSTTService(DeepgramFluxSTTBase):
         # Initialize default settings
         default_settings = self.Settings(
             model="flux-general-en",
-            language=Language.EN,
+            language=None,
             eager_eot_threshold=None,
             eot_threshold=None,
             eot_timeout_ms=None,
             keyterm=[],
             min_confidence=None,
+            language_hints=None,
         )
 
         # Apply settings delta

src/pipecat/services/deepgram/flux/stt.py

@@ -24,7 +24,6 @@ from pipecat.services.deepgram.flux.base import (
     FluxMessageType,
 )
 from pipecat.services.websocket_service import WebsocketService
-from pipecat.transcriptions.language import Language
 
 try:
     from websockets.asyncio.client import connect as websocket_connect
@@ -50,6 +49,12 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService):
     Supports configurable models, VAD events, and various audio processing options
     including advanced turn detection and EagerEndOfTurn events for improved conversational AI performance.
 
+    For multilingual use, set ``model="flux-general-multi"`` and pass
+    ``language_hints`` to bias detection toward specific languages. Hints can
+    be updated mid-stream via ``STTUpdateSettingsFrame`` (e.g. to implement a
+    detect-then-lock flow). ``TranscriptionFrame.language`` reflects whichever
+    language Flux detected for each turn.
+
     Event handlers available (in addition to base events):
 
     - on_start_of_turn(service, transcript): Deepgram detected start of speech
@@ -156,6 +161,16 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService):
                         tag=["production", "voice-agent"],
                     ),
                 )
+
+            Multilingual usage with language hints::
+
+                stt = DeepgramFluxSTTService(
+                    api_key="your-api-key",
+                    settings=DeepgramFluxSTTService.Settings(
+                        model="flux-general-multi",
+                        language_hints=[Language.EN, Language.ES],
+                    ),
+                )
         """
         # Note: For DeepgramFluxSTTService, differently from other processes, we need to create
         # the _receive_task inside _connect_websocket, because the websocket should only be
@@ -171,12 +186,13 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService):
         # 1. Initialize default_settings with hardcoded defaults
         default_settings = self.Settings(
             model="flux-general-en",
-            language=Language.EN,
+            language=None,
             eager_eot_threshold=None,
             eot_threshold=None,
             eot_timeout_ms=None,
             keyterm=[],
             min_confidence=None,
+            language_hints=None,
         )
 
         # 2. Apply direct init arg overrides (deprecated)

src/pipecat/services/deepgram/stt.py

@@ -621,6 +621,7 @@ class DeepgramSTTService(STTService):
         """
         while True:
             connect_kwargs = self._build_connect_kwargs()
+            keepalive_task = None
             try:
                 async with self._client.listen.v1.connect(**connect_kwargs) as connection:
                     self._connection = connection
@@ -639,7 +640,8 @@ class DeepgramSTTService(STTService):
             finally:
                 self._connection_ready.clear()
                 self._connection = None
-                await self.cancel_task(keepalive_task)
+                if keepalive_task:
+                    await self.cancel_task(keepalive_task)
 
     async def _keepalive_handler(self):
         """Periodically send KeepAlive frames to prevent server-side timeout.

src/pipecat/services/elevenlabs/tts.py

@@ -245,6 +245,35 @@ class ElevenLabsHttpTTSSettings(TTSSettings):
     )
 
 
+def _strip_leading_space(
+    alignment: Mapping[str, Any], keys: tuple[str, str, str]
+) -> Mapping[str, Any]:
+    """Return alignment with a prepended space char removed, if present.
+
+    Normalized alignment chunks from ElevenLabs begin with a leading space that
+    marks the prosody/chunk boundary. Left in place, it would prematurely
+    terminate a partial word carried over from the previous chunk. Stripping it
+    is lossless for timing: the dropped space's duration is still reflected in
+    the next char's `charStartTimesMs`, and the chunk's last-element values
+    (used to advance cumulative time) are untouched.
+
+    Args:
+        alignment: Alignment dict from the API.
+        keys: Tuple of (chars_key, start_times_key, durations_or_end_times_key)
+            naming the three parallel arrays — these differ between the
+            WebSocket and HTTP response schemas.
+    """
+    chars_key, starts_key, tail_key = keys
+    chars = alignment.get(chars_key) or []
+    if chars and chars[0] == " ":
+        return {
+            chars_key: chars[1:],
+            starts_key: alignment.get(starts_key, [])[1:],
+            tail_key: alignment.get(tail_key, [])[1:],
+        }
+    return alignment
+
+
 def calculate_word_times(
     alignment_info: Mapping[str, Any],
     cumulative_time: float,
@@ -790,8 +819,15 @@ class ElevenLabsTTSService(WebsocketTTSService):
                 frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=received_ctx_id)
                 await self.append_to_audio_context(received_ctx_id, frame)
 
-            if msg.get("alignment"):
-                alignment = msg["alignment"]
+            if msg.get("normalizedAlignment"):
+                # Use normalizedAlignment (what was actually spoken) rather than
+                # alignment (the input text), so word timestamps stay accurate
+                # when a pronunciation dictionary or text normalization rewrites
+                # the input.
+                alignment = _strip_leading_space(
+                    msg["normalizedAlignment"],
+                    ("chars", "charStartTimesMs", "charDurationsMs"),
+                )
                 word_times, self._partial_word, self._partial_word_start_time = (
                     calculate_word_times(
                         alignment,
@@ -1296,21 +1332,30 @@ class ElevenLabsHttpTTSService(TTSService):
                                 audio, self.sample_rate, 1, context_id=context_id
                             )
 
-                        # Process alignment if present
-                        if data and "alignment" in data:
-                            alignment = data["alignment"]
-                            if alignment:  # Ensure alignment is not None
-                                # Get end time of the last character in this chunk
-                                char_end_times = alignment.get("character_end_times_seconds", [])
-                                if char_end_times:
-                                    chunk_end_time = char_end_times[-1]
-                                    # Update to the longest end time seen so far
-                                    utterance_duration = max(utterance_duration, chunk_end_time)
+                        # Process alignment if present. Use normalized_alignment
+                        # (what was actually spoken) so word timestamps stay
+                        # accurate when a pronunciation dictionary or text
+                        # normalization rewrites the input.
+                        if data and data.get("normalized_alignment"):
+                            alignment = _strip_leading_space(
+                                data["normalized_alignment"],
+                                (
+                                    "characters",
+                                    "character_start_times_seconds",
+                                    "character_end_times_seconds",
+                                ),
+                            )
+                            # Get end time of the last character in this chunk
+                            char_end_times = alignment.get("character_end_times_seconds", [])
+                            if char_end_times:
+                                chunk_end_time = char_end_times[-1]
+                                # Update to the longest end time seen so far
+                                utterance_duration = max(utterance_duration, chunk_end_time)
 
-                                # Calculate word timestamps
-                                word_times = self.calculate_word_times(alignment)
-                                if word_times:
-                                    await self.add_word_timestamps(word_times, context_id)
+                            # Calculate word timestamps
+                            word_times = self.calculate_word_times(alignment)
+                            if word_times:
+                                await self.add_word_timestamps(word_times, context_id)
                     except json.JSONDecodeError as e:
                         logger.warning(f"Failed to parse JSON from stream: {e}")
                         continue

src/pipecat/services/nvidia/llm.py

@@ -2,21 +2,44 @@
 # Copyright (c) 2024-2026, Daily
 #
 # SPDX-License-Identifier: BSD 2-Clause License
+# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES.
 #
 
 """NVIDIA NIM API service implementation.
 
 This module provides a service for interacting with NVIDIA's NIM (NVIDIA Inference
 Microservice) API while maintaining compatibility with the OpenAI-style interface.
+
+Refer to the NVIDIA NIM LLM API documentation for available models and usage:
+https://docs.api.nvidia.com/nim/reference/llm-apis
 """
 
+from collections.abc import AsyncIterator
 from dataclasses import dataclass
+from enum import StrEnum
 
+from loguru import logger
+from openai.types.chat import ChatCompletionChunk
+
+from pipecat.frames.frames import (
+    LLMThoughtEndFrame,
+    LLMThoughtStartFrame,
+    LLMThoughtTextFrame,
+)
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.openai.base_llm import BaseOpenAILLMService
 from pipecat.services.openai.llm import OpenAILLMService
 
+_THINK_OPEN = "<think>"
+_THINK_CLOSE = "</think>"
+
+
+class _ThinkTagState(StrEnum):
+    DETECTING = "detecting"
+    IN_THOUGHT = "in_thought"
+    CONTENT = "content"
+
 
 @dataclass
 class NvidiaLLMSettings(BaseOpenAILLMService.Settings):
@@ -28,9 +51,19 @@ class NvidiaLLMSettings(BaseOpenAILLMService.Settings):
 class NvidiaLLMService(OpenAILLMService):
     """A service for interacting with NVIDIA's NIM (NVIDIA Inference Microservice) API.
 
-    This service extends OpenAILLMService to work with NVIDIA's NIM API while maintaining
-    compatibility with the OpenAI-style interface. It specifically handles the difference
-    in token usage reporting between NIM (incremental) and OpenAI (final summary).
+    This service extends OpenAILLMService to work with NVIDIA's NIM API while
+    maintaining compatibility with the OpenAI-style interface. It handles:
+
+    - Incremental token usage reporting (NIM sends per-chunk counts instead
+      of a final summary)
+    - Detection and filtering of leading ``<think>``/``</think>`` content for
+      models that emit reasoning inline before visible output (e.g.
+      DeepSeek-R1, some nemotron models)
+    - Extraction of ``reasoning_content`` from the streaming delta for models
+      with API-level reasoning separation (e.g. Nemotron Nano models)
+
+    Reasoning content is emitted as ``LLMThought*Frame`` objects, keeping it
+    accessible to observers and logging without sending it to TTS.
     """
 
     Settings = NvidiaLLMSettings
@@ -39,7 +72,7 @@ class NvidiaLLMService(OpenAILLMService):
     def __init__(
         self,
         *,
-        api_key: str,
+        api_key: str | None = None,
         base_url: str = "https://integrate.api.nvidia.com/v1",
         model: str | None = None,
         settings: Settings | None = None,
@@ -48,10 +81,12 @@ class NvidiaLLMService(OpenAILLMService):
         """Initialize the NvidiaLLMService.
 
         Args:
-            api_key: The API key for accessing NVIDIA's NIM API.
-            base_url: The base URL for NIM API. Defaults to "https://integrate.api.nvidia.com/v1".
+            api_key: NVIDIA API key for authentication. Required when using the
+                cloud endpoint. Not needed for local NIM deployments.
+            base_url: The base URL for NIM API. Defaults to NVIDIA's cloud endpoint.
+                For local deployments, pass the local address (e.g. ``http://localhost:8000/v1``).
             model: The model identifier to use. Defaults to
-                "nvidia/llama-3.1-nemotron-70b-instruct".
+                "nvidia/nemotron-3-nano-30b-a3b".
 
                 .. deprecated:: 0.0.105
                     Use ``settings=NvidiaLLMService.Settings(model=...)`` instead.
@@ -61,7 +96,7 @@ class NvidiaLLMService(OpenAILLMService):
             **kwargs: Additional keyword arguments passed to OpenAILLMService.
         """
         # 1. Initialize default_settings with hardcoded defaults
-        default_settings = self.Settings(model="nvidia/llama-3.1-nemotron-70b-instruct")
+        default_settings = self.Settings(model="nvidia/nemotron-3-nano-30b-a3b")
 
         # 2. Apply direct init arg overrides (deprecated)
         if model is not None:
@@ -75,6 +110,14 @@ class NvidiaLLMService(OpenAILLMService):
             default_settings.apply_update(settings)
 
         super().__init__(api_key=api_key, base_url=base_url, settings=default_settings, **kwargs)
+
+        if "api.nvidia.com" in base_url and not api_key:
+            logger.warning(
+                "NvidiaLLMService: Using the cloud endpoint but no API key was provided. "
+                "An API key is required for the cloud endpoint. "
+                "Set base_url to your local NIM endpoint for local deployments."
+            )
+
         # Counters for accumulating token usage metrics
         self._prompt_tokens = 0
         self._completion_tokens = 0
@@ -82,24 +125,202 @@ class NvidiaLLMService(OpenAILLMService):
         self._has_reported_prompt_tokens = False
         self._is_processing = False
 
-    async def _process_context(self, context: LLMContext):
-        """Process a context through the LLM and accumulate token usage metrics.
+    def _reset_response_state(self):
+        """Reset per-response state at the start of each LLM call.
 
-        This method overrides the parent class implementation to handle NVIDIA's
-        incremental token reporting style, accumulating the counts and reporting
-        them once at the end of processing.
-
-        Args:
-            context: The context to process, containing messages and other information
-                needed for the LLM interaction.
+        Resets token accumulation counters, leading-think-tag detection state,
+        and reasoning-content field tracking.
         """
-        # Reset all counters and flags at the start of processing
         self._prompt_tokens = 0
         self._completion_tokens = 0
         self._total_tokens = 0
         self._has_reported_prompt_tokens = False
         self._is_processing = True
 
+        self._think_tag_state = _ThinkTagState.DETECTING
+        self._think_tag_buffer = ""
+
+        # reasoning_content field tracking
+        self._has_reasoning_field = False
+
+    async def _filter_thinking_content(self, text: str) -> str | None:
+        """Filter leading ``<think>`` tags from content and emit thought frames.
+
+        Uses a three-state machine optimized for the common provider pattern
+        where a response either begins with a ``<think>`` block or contains no
+        think tags at all. It returns only visible content to the base OpenAI
+        processing loop while emitting hidden reasoning as ``LLMThought*Frame``
+        side effects.
+
+        - ``detecting``: Buffers the start of the stream to check for
+          ``<think>``.
+        - ``in_thought``: Inside a leading think block; emits
+          ``LLMThoughtTextFrame`` until ``</think>`` is found.
+        - ``content``: Normal content; passthrough.
+
+        Non-reasoning models transition from ``detecting`` to ``content``
+        on the first chunk with zero buffering overhead after that.
+
+        Args:
+            text: The text content from the LLM to filter.
+
+        Returns:
+            The non-reasoning content that should continue through the base
+            OpenAI content path, or ``None`` if this chunk should not emit
+            normal content.
+
+        """
+        if self._think_tag_state == _ThinkTagState.CONTENT:
+            return text
+
+        self._think_tag_buffer += text
+
+        if self._think_tag_state == _ThinkTagState.DETECTING:
+            if len(self._think_tag_buffer) < len(_THINK_OPEN):
+                if _THINK_OPEN.startswith(self._think_tag_buffer):
+                    return None
+                self._think_tag_state = _ThinkTagState.CONTENT
+                passthrough = self._think_tag_buffer
+                self._think_tag_buffer = ""
+                return passthrough
+
+            if self._think_tag_buffer.startswith(_THINK_OPEN):
+                self._think_tag_state = _ThinkTagState.IN_THOUGHT
+                await self.push_frame(LLMThoughtStartFrame())
+                self._think_tag_buffer = self._think_tag_buffer[len(_THINK_OPEN) :]
+            else:
+                self._think_tag_state = _ThinkTagState.CONTENT
+                passthrough = self._think_tag_buffer
+                self._think_tag_buffer = ""
+                return passthrough
+
+        if self._think_tag_state == _ThinkTagState.IN_THOUGHT:
+            idx = self._think_tag_buffer.find(_THINK_CLOSE)
+            if idx != -1:
+                thought = self._think_tag_buffer[:idx]
+                if thought:
+                    await self.push_frame(LLMThoughtTextFrame(text=thought))
+                await self.push_frame(LLMThoughtEndFrame())
+                remainder = self._think_tag_buffer[idx + len(_THINK_CLOSE) :]
+                self._think_tag_buffer = ""
+                self._think_tag_state = _ThinkTagState.CONTENT
+                return remainder or None
+            else:
+                safe_end = len(self._think_tag_buffer) - len(_THINK_CLOSE) + 1
+                if safe_end > 0:
+                    await self.push_frame(
+                        LLMThoughtTextFrame(text=self._think_tag_buffer[:safe_end])
+                    )
+                    self._think_tag_buffer = self._think_tag_buffer[safe_end:]
+        return None
+
+    async def _flush_reasoning_state(self):
+        """Flush buffered reasoning state at normal stream completion.
+
+        Emits any buffered trailing thought text, closes open thought frames,
+        and forwards any buffered pre-content text that was held while deciding
+        whether the stream began with ``<think>``.
+        """
+        if self._think_tag_state == _ThinkTagState.IN_THOUGHT:
+            if self._think_tag_buffer:
+                await self.push_frame(LLMThoughtTextFrame(text=self._think_tag_buffer))
+            await self.push_frame(LLMThoughtEndFrame())
+        elif self._think_tag_state == _ThinkTagState.DETECTING and self._think_tag_buffer:
+            await super()._push_llm_text(self._think_tag_buffer)
+
+        self._think_tag_buffer = ""
+        self._think_tag_state = _ThinkTagState.CONTENT
+
+        if self._has_reasoning_field:
+            await self.push_frame(LLMThoughtEndFrame())
+            self._has_reasoning_field = False
+
+    async def get_chat_completions(self, context: LLMContext) -> AsyncIterator[ChatCompletionChunk]:
+        """Wrap the chat completion stream to handle ``reasoning_content``.
+
+        Models with API-level reasoning separation (e.g. Nemotron Nano)
+        include a ``reasoning_content`` field on the streaming delta. This
+        wrapper extracts those chunks and emits them as ``LLMThought*Frame``
+        objects. It also rewrites streamed ``delta.content`` so leading
+        ``<think>`` sections are removed before the base OpenAI loop processes
+        visible content.
+
+        Args:
+            context: The LLM context for the completion request.
+
+        Returns:
+            An async iterator of chat completion chunks where
+            ``reasoning_content`` has been emitted as ``LLMThought*Frame``
+            side effects.
+        """
+        stream = await super().get_chat_completions(context)
+        return self._handle_reasoning_content(stream)
+
+    async def _handle_reasoning_content(
+        self, stream: AsyncIterator[ChatCompletionChunk]
+    ) -> AsyncIterator[ChatCompletionChunk]:
+        """Handle ``reasoning_content`` and leading ``<think>`` tags in a chunk stream.
+
+        Inspects each chunk for a ``reasoning_content`` field on the delta and
+        emits ``LLMThoughtStartFrame`` / ``LLMThoughtTextFrame`` /
+        ``LLMThoughtEndFrame`` as side effects. It also strips ``<think>``
+        blocks from ``delta.content`` before yielding the chunk so the base
+        OpenAI loop only sees user-facing content. Every chunk is still yielded
+        so the base streaming loop can process metadata such as token usage,
+        model name, tool calls, and audio transcripts.
+
+        Notes:
+            Stream cleanup is owned by the base OpenAI processing loop
+            (``BaseOpenAILLMService._process_context``), which wraps the stream
+            in its own closing context manager.
+
+        Args:
+            stream: The original chat completion stream.
+
+        Yields:
+            Chat completion chunks with any leading ``<think>`` content removed
+            from ``delta.content`` before they reach the base OpenAI loop.
+        """
+        async for chunk in stream:
+            if chunk.choices and len(chunk.choices) > 0 and chunk.choices[0].delta:
+                delta = chunk.choices[0].delta
+                rc = getattr(delta, "reasoning_content", None)
+                if rc:
+                    if not self._has_reasoning_field:
+                        self._has_reasoning_field = True
+                        await self.push_frame(LLMThoughtStartFrame())
+                    await self.push_frame(LLMThoughtTextFrame(text=rc))
+                elif self._has_reasoning_field and delta.content:
+                    await self.push_frame(LLMThoughtEndFrame())
+                    self._has_reasoning_field = False
+
+                if delta.content:
+                    delta.content = await self._filter_thinking_content(delta.content)
+            yield chunk
+
+        await self._flush_reasoning_state()
+
+    async def _process_context(self, context: LLMContext):
+        """Process a context through the LLM and accumulate token usage metrics.
+
+        Delegates to the base OpenAI streaming loop while adding
+        NVIDIA-specific behavior:
+
+        - ``reasoning_content`` and leading ``<think>`` content are
+          intercepted via the ``get_chat_completions`` stream wrapper and
+          emitted as
+          ``LLMThought*Frame`` objects.
+        - Incremental token counts are accumulated and reported as final
+          totals.
+
+        Args:
+            context: The context to process, containing messages and other
+                information needed for the LLM interaction.
+        """
+        self._reset_response_state()
+
+        # Wrap in try/finally to guarantee accumulated token metrics are
+        # reported and _is_processing is cleared even on cancellation.
         try:
             await super()._process_context(context)
         finally:

src/pipecat/services/sarvam/stt.py

@@ -91,6 +91,7 @@ class ModelConfig:
         supports_prompt: Whether the model accepts prompt parameter.
         supports_mode: Whether the model accepts mode parameter.
         supports_language: Whether the model accepts language parameter.
+        supports_vad_params: Whether the model accepts fine-grained VAD parameters.
         default_language: Default language code (None = auto-detect).
         default_mode: Default mode (None = not applicable).
         use_translate_endpoint: Whether to use speech_to_text_translate_streaming endpoint.
@@ -100,6 +101,7 @@ class ModelConfig:
     supports_prompt: bool
     supports_mode: bool
     supports_language: bool
+    supports_vad_params: bool
     default_language: str | None
     default_mode: str | None
     use_translate_endpoint: bool
@@ -111,6 +113,7 @@ MODEL_CONFIGS: dict[str, ModelConfig] = {
         supports_prompt=False,
         supports_mode=False,
         supports_language=True,
+        supports_vad_params=False,
         default_language="unknown",
         default_mode=None,
         use_translate_endpoint=False,
@@ -120,6 +123,7 @@ MODEL_CONFIGS: dict[str, ModelConfig] = {
         supports_prompt=True,
         supports_mode=False,
         supports_language=False,
+        supports_vad_params=False,
         default_language=None,  # Auto-detects language
         default_mode=None,
         use_translate_endpoint=True,
@@ -129,6 +133,7 @@ MODEL_CONFIGS: dict[str, ModelConfig] = {
         supports_prompt=False,
         supports_mode=True,
         supports_language=True,
+        supports_vad_params=True,
         default_language="unknown",
         default_mode="transcribe",
         use_translate_endpoint=False,
@@ -146,11 +151,43 @@ class SarvamSTTSettings(STTSettings):
             Only applicable to models that support prompts (e.g., saaras:v2.5).
         vad_signals: Enable VAD signals in response.
         high_vad_sensitivity: Enable high VAD sensitivity.
+        positive_speech_threshold: VAD probability threshold (0.0-1.0) above which
+            a frame is considered speech. Only for saaras:v3.
+        negative_speech_threshold: VAD probability threshold (0.0-1.0) below which
+            a frame is considered silence. Only for saaras:v3.
+        min_speech_frames: Minimum consecutive speech frames to start a speech
+            segment. Only for saaras:v3.
+        first_turn_min_speech_frames: Minimum speech frames for the first user
+            turn. Only for saaras:v3.
+        negative_frames_count: Number of silence frames within the window to end
+            a speech segment. Only for saaras:v3.
+        negative_frames_window: Sliding window size (in frames) for counting
+            negative frames. Only for saaras:v3.
+        start_speech_volume_threshold: Volume level (dB) below which audio is
+            too quiet to be speech. Only for saaras:v3.
+        interrupt_min_speech_frames: Minimum speech frames to register a
+            barge-in/interruption. Only for saaras:v3.
+        pre_speech_pad_frames: Number of audio frames to prepend before detected
+            speech onset. Only for saaras:v3.
+        num_initial_ignored_frames: Number of leading audio frames to skip at
+            connection start. Only for saaras:v3.
     """
 
     prompt: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
     vad_signals: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
     high_vad_sensitivity: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    positive_speech_threshold: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    negative_speech_threshold: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    min_speech_frames: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    first_turn_min_speech_frames: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    negative_frames_count: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    negative_frames_window: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    start_speech_volume_threshold: float | None | _NotGiven = field(
+        default_factory=lambda: NOT_GIVEN
+    )
+    interrupt_min_speech_frames: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    pre_speech_pad_frames: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    num_initial_ignored_frames: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
 
 
 class SarvamSTTService(STTService):
@@ -190,7 +227,7 @@ class SarvamSTTService(STTService):
             mode: Mode of operation for saaras:v3 models only. Options: transcribe, translate,
                 verbatim, translit, codemix. Defaults to "transcribe" for saaras:v3.
             vad_signals: Enable VAD signals in response. Defaults to None.
-            high_vad_sensitivity: Enable high VAD (Voice Activity Detection) sensitivity. Defaults to None.
+            high_vad_sensitivity: Enable high VAD sensitivity. Defaults to None.
         """
 
         language: Language | None = None
@@ -244,11 +281,21 @@ class SarvamSTTService(STTService):
         """
         # --- 1. Hardcoded defaults ---
         default_settings = self.Settings(
-            model="saarika:v2.5",
+            model="saaras:v3",
             language=None,
             prompt=None,
             vad_signals=None,
             high_vad_sensitivity=None,
+            positive_speech_threshold=None,
+            negative_speech_threshold=None,
+            min_speech_frames=None,
+            first_turn_min_speech_frames=None,
+            negative_frames_count=None,
+            negative_frames_window=None,
+            start_speech_volume_threshold=None,
+            interrupt_min_speech_frames=None,
+            pre_speech_pad_frames=None,
+            num_initial_ignored_frames=None,
         )
 
         # --- 2. Deprecated direct-arg overrides ---
@@ -289,6 +336,26 @@ class SarvamSTTService(STTService):
                 f"Model '{resolved_model}' does not support language parameter (auto-detects language)."
             )
 
+        if not self._config.supports_vad_params:
+            vad_param_names = [
+                "positive_speech_threshold",
+                "negative_speech_threshold",
+                "min_speech_frames",
+                "first_turn_min_speech_frames",
+                "negative_frames_count",
+                "negative_frames_window",
+                "start_speech_volume_threshold",
+                "interrupt_min_speech_frames",
+                "pre_speech_pad_frames",
+                "num_initial_ignored_frames",
+            ]
+            for param_name in vad_param_names:
+                if getattr(default_settings, param_name) is not None:
+                    raise ValueError(
+                        f"Model '{resolved_model}' does not support {param_name} parameter. "
+                        f"Fine-grained VAD parameters are only supported by saaras:v3."
+                    )
+
         # Resolve mode default from model config
         if mode is None:
             mode = self._config.default_mode
@@ -393,10 +460,44 @@ class SarvamSTTService(STTService):
                     f"Model '{self._settings.model}' does not support prompt parameter."
                 )
 
+        if isinstance(delta, self.Settings) and not self._config.supports_vad_params:
+            vad_param_names = [
+                "positive_speech_threshold",
+                "negative_speech_threshold",
+                "min_speech_frames",
+                "first_turn_min_speech_frames",
+                "negative_frames_count",
+                "negative_frames_window",
+                "start_speech_volume_threshold",
+                "interrupt_min_speech_frames",
+                "pre_speech_pad_frames",
+                "num_initial_ignored_frames",
+            ]
+            for param_name in vad_param_names:
+                val = getattr(delta, param_name, NOT_GIVEN)
+                if is_given(val) and val is not None:
+                    raise ValueError(
+                        f"Model '{self._settings.model}' does not support {param_name} "
+                        f"parameter. Fine-grained VAD parameters are only supported by saaras:v3."
+                    )
+
         changed = await super()._update_settings(delta)
 
-        # Language and prompt are WebSocket connect-time parameters; reconnect to apply.
-        reconnect_fields = {"language", "prompt"}
+        # These are all WebSocket connect-time parameters; reconnect to apply.
+        reconnect_fields = {
+            "language",
+            "prompt",
+            "positive_speech_threshold",
+            "negative_speech_threshold",
+            "min_speech_frames",
+            "first_turn_min_speech_frames",
+            "negative_frames_count",
+            "negative_frames_window",
+            "start_speech_volume_threshold",
+            "interrupt_min_speech_frames",
+            "pre_speech_pad_frames",
+            "num_initial_ignored_frames",
+        }
         if changed.keys() & reconnect_fields:
             await self._disconnect()
             await self._connect()
@@ -535,6 +636,24 @@ class SarvamSTTService(STTService):
                     "true" if self._settings.high_vad_sensitivity else "false"
                 )
 
+            # Fine-grained VAD parameters (saaras:v3 only, sent as strings per SDK spec)
+            if self._config.supports_vad_params:
+                _vad_params = {
+                    "positive_speech_threshold": self._settings.positive_speech_threshold,
+                    "negative_speech_threshold": self._settings.negative_speech_threshold,
+                    "min_speech_frames": self._settings.min_speech_frames,
+                    "first_turn_min_speech_frames": self._settings.first_turn_min_speech_frames,
+                    "negative_frames_count": self._settings.negative_frames_count,
+                    "negative_frames_window": self._settings.negative_frames_window,
+                    "start_speech_volume_threshold": self._settings.start_speech_volume_threshold,
+                    "interrupt_min_speech_frames": self._settings.interrupt_min_speech_frames,
+                    "pre_speech_pad_frames": self._settings.pre_speech_pad_frames,
+                    "num_initial_ignored_frames": self._settings.num_initial_ignored_frames,
+                }
+                for k, v in _vad_params.items():
+                    if v is not None:
+                        connect_kwargs[k] = str(v)
+
             # Add language_code for models that support it
             language_string = self._get_language_string()
             if language_string is not None:

src/pipecat/services/smallest/tts.py

@@ -125,7 +125,7 @@ class SmallestTTSService(InterruptibleTTSService):
         self,
         *,
         api_key: str,
-        base_url: str = "wss://waves-api.smallest.ai",
+        base_url: str = "wss://api.smallest.ai",
         sample_rate: int | None = None,
         settings: Settings | None = None,
         **kwargs,
@@ -174,6 +174,10 @@ class SmallestTTSService(InterruptibleTTSService):
         """
         return True
 
+    async def flush_audio(self, context_id: str | None = None):
+        """Flush any pending audio data."""
+        logger.trace(f"{self}: flushing audio")
+
     def language_to_service_language(self, language: Language) -> str | None:
         """Convert a Language enum to Smallest service language format.
 
@@ -217,7 +221,7 @@ class SmallestTTSService(InterruptibleTTSService):
 
     def _build_websocket_url(self) -> str:
         """Build the WebSocket URL from base URL and model."""
-        return f"{self._base_url}/api/v1/{self._settings.model}/get_speech/stream"
+        return f"{self._base_url}/waves/v1/{self._settings.model}/get_speech/stream"
 
     async def start(self, frame: StartFrame):
         """Start the Smallest TTS service.
@@ -350,17 +354,15 @@ class SmallestTTSService(InterruptibleTTSService):
             await self._send_keepalive()
 
     async def _send_keepalive(self):
-        """Send a flush message to keep the connection alive."""
+        """Send a silent message to keep the WebSocket connection alive."""
         if self._websocket and self._websocket.state is State.OPEN:
-            msg = {"flush": True}
+            msg = {
+                "text": " ",
+                "voice_id": self._settings.voice,
+                "language": self._settings.language,
+            }
             await self._websocket.send(json.dumps(msg))
 
-    async def flush_audio(self, context_id: str | None = None):
-        """Flush any pending audio synthesis."""
-        if not self._websocket or self._websocket.state is State.CLOSED:
-            return
-        await self._get_websocket().send(json.dumps({"flush": True}))
-
     async def _receive_messages(self):
         """Receive and process messages from the Smallest WebSocket API."""
         async for message in self._get_websocket():

src/pipecat/services/stt_latency.py

@@ -57,3 +57,4 @@ WHISPER_TTFS_P99: float = DEFAULT_TTFS_P99
 # No benchmark available yet; using conservative default
 MISTRAL_TTFS_P99: float = DEFAULT_TTFS_P99
 SMALLEST_TTFS_P99: float = DEFAULT_TTFS_P99
+XAI_TTFS_P99: float = DEFAULT_TTFS_P99

src/pipecat/services/tts_service.py

@@ -293,6 +293,7 @@ class TTSService(AIService):
         self._processing_text: bool = False
         self._tts_contexts: dict[str, TTSContext] = {}
         self._streamed_text: str = ""
+        self._sent_non_whitespace_in_context: bool = False
         self._text_aggregation_metrics_started: bool = False
 
         # Word timestamp state
@@ -684,6 +685,7 @@ class TTSService(AIService):
 
             # Reset aggregator state
             self._processing_text = False
+            self._sent_non_whitespace_in_context = False
             if isinstance(frame, LLMFullResponseEndFrame):
                 if self._push_text_frames:
                     # Route through the serialization queue so the frame is
@@ -698,6 +700,8 @@ class TTSService(AIService):
         elif isinstance(frame, TTSSpeakFrame):
             # Store if we were processing text or not so we can set it back.
             processing_text = self._processing_text
+            saved_sent_non_whitespace = self._sent_non_whitespace_in_context
+            self._sent_non_whitespace_in_context = False
             # TTSSpeakFrame is independent — temporarily clear the turn context
             # so create_context_id() generates a fresh UUID for this utterance.
             saved_turn_context_id = self._turn_context_id
@@ -718,6 +722,7 @@ class TTSService(AIService):
             # the TTS. We pause to avoid audio overlapping.
             await self._maybe_pause_frame_processing()
             self._turn_context_id = saved_turn_context_id
+            self._sent_non_whitespace_in_context = saved_sent_non_whitespace
             self._processing_text = processing_text
         elif isinstance(frame, TTSUpdateSettingsFrame):
             if frame.service is not None and frame.service is not self:
@@ -844,6 +849,7 @@ class TTSService(AIService):
 
     async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
         self._processing_text = False
+        self._sent_non_whitespace_in_context = False
         await self._text_aggregator.handle_interruption()
         for filter in self._text_filters:
             await filter.handle_interruption()
@@ -901,13 +907,22 @@ class TTSService(AIService):
             await self.push_frame(src_frame)
             return
 
-        # Remove leading newlines only
-        text = text.lstrip("\n")
-
-        # Don't send only whitespace. This causes problems for some TTS models. But also don't
-        # strip all whitespace, as whitespace can influence prosody.
-        if not text.strip():
-            return
+        # Whitespace gating depends on aggregation mode:
+        # - Token streaming: drop all leading whitespace at the start of a context, as
+        #   nothing substantive has been sent yet for it to attach to. Once a non-whitespace
+        #   token has been sent, send whitespace as-is since it can influence prosody between
+        #   non-whitespace tokens.
+        #
+        # - Sentence aggregation: strip leading newlines only and drop pure-whitespace frames.
+        if self._is_streaming_tokens:
+            if not self._sent_non_whitespace_in_context:
+                text = text.lstrip()
+            if not text:
+                return
+        else:
+            text = text.lstrip("\n")
+            if not text.strip():
+                return
 
         # This is just a flag that indicates if we sent something to the TTS
         # service. It will be cleared if we sent text because of a TTSSpeakFrame
@@ -929,9 +944,15 @@ class TTSService(AIService):
             await filter.reset_interruption()
             text = await filter.filter(text)
 
-        if not text.strip():
-            if not self._is_streaming_tokens:
-                await self.stop_processing_metrics()
+        # Post-filter whitespace gate. Mirrors the pre-filter logic so filter
+        # output that collapses to whitespace-only is handled consistently.
+        if self._is_streaming_tokens:
+            # If empty, or only-whitespace and we haven't sent any non-whitespace, skip.
+            if not text or (not text.strip() and not self._sent_non_whitespace_in_context):
+                return
+            self._sent_non_whitespace_in_context = True
+        elif not text.strip():
+            await self.stop_processing_metrics()
             return
 
         # Create context ID and store metadata

src/pipecat/services/xai/stt.py

@@ -0,0 +1,411 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""xAI speech-to-text service implementation.
+
+This module provides integration with xAI's real-time speech-to-text WebSocket
+API documented at https://docs.x.ai/developers/rest-api-reference/inference/voice.
+"""
+
+import asyncio
+import json
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass, field
+from typing import Any
+from urllib.parse import urlencode
+
+from loguru import logger
+
+from pipecat import version as pipecat_version
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    StartFrame,
+    TranscriptionFrame,
+)
+from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven
+from pipecat.services.stt_latency import XAI_TTFS_P99
+from pipecat.services.stt_service import WebsocketSTTService
+from pipecat.transcriptions.language import Language, resolve_language
+from pipecat.utils.time import time_now_iso8601
+from pipecat.utils.tracing.service_decorators import traced_stt
+
+try:
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error('In order to use xAI STT, you need to `pip install "pipecat-ai[xai]"`.')
+    raise Exception(f"Missing module: {e}")
+
+
+def language_to_xai_stt_language(language: Language) -> str | None:
+    """Convert a Language enum to the xAI STT language code.
+
+    xAI STT accepts two-letter language codes (e.g. ``en``, ``fr``, ``de``,
+    ``ja``). When set, the server applies Inverse Text Normalization.
+
+    Args:
+        language: The Language enum value to convert.
+
+    Returns:
+        The corresponding xAI STT language code, or None if not supported.
+    """
+    LANGUAGE_MAP = {
+        Language.AR: "ar",
+        Language.BN: "bn",
+        Language.DE: "de",
+        Language.EN: "en",
+        Language.ES: "es",
+        Language.FR: "fr",
+        Language.HI: "hi",
+        Language.ID: "id",
+        Language.IT: "it",
+        Language.JA: "ja",
+        Language.KO: "ko",
+        Language.PT: "pt",
+        Language.RU: "ru",
+        Language.TR: "tr",
+        Language.VI: "vi",
+        Language.ZH: "zh",
+    }
+
+    return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
+
+
+@dataclass
+class XAISTTSettings(STTSettings):
+    """Settings for XAISTTService.
+
+    Parameters:
+        interim_results: When True, partial transcripts are emitted
+            approximately every 500ms.
+        endpointing: Silence duration in milliseconds that triggers a
+            speech-final event. Range 0-5000. Server default is 10ms.
+        multichannel: When True, transcribes each interleaved channel
+            independently. Requires ``channels`` >= 2.
+        channels: Number of interleaved channels (2-8). Required when
+            ``multichannel`` is True.
+        diarize: When True, the server attaches a ``speaker`` field to each
+            word identifying the detected speaker.
+    """
+
+    interim_results: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    endpointing: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    multichannel: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    channels: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    diarize: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+
+
+class XAISTTService(WebsocketSTTService):
+    """xAI real-time speech-to-text service.
+
+    Streams audio to xAI's WebSocket STT endpoint and emits interim and final
+    transcription frames. The ``XAI_API_KEY`` is passed directly as a Bearer
+    token on the WebSocket handshake.
+
+    The connection is persistent: audio is streamed continuously and the
+    server emits ``transcript.partial`` events with ``is_final`` and
+    ``speech_final`` flags to mark utterance boundaries. If the connection
+    drops mid-session, the base class reconnects automatically.
+    """
+
+    Settings = XAISTTSettings
+    _settings: Settings
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        ws_url: str = "wss://api.x.ai/v1/stt",
+        sample_rate: int = 16000,
+        encoding: str = "pcm",
+        settings: Settings | None = None,
+        ttfs_p99_latency: float | None = XAI_TTFS_P99,
+        **kwargs,
+    ):
+        """Initialize the xAI STT service.
+
+        Args:
+            api_key: xAI API key (used as Bearer for the WebSocket handshake).
+            ws_url: WebSocket endpoint URL. Defaults to ``wss://api.x.ai/v1/stt``.
+            sample_rate: Audio sample rate in Hz. Supported values: 8000,
+                16000, 22050, 24000, 44100, 48000. Defaults to 16000.
+            encoding: Audio encoding. One of ``"pcm"`` (signed 16-bit LE),
+                ``"mulaw"``, or ``"alaw"``. Defaults to ``"pcm"``.
+            settings: Runtime-updatable settings overriding defaults.
+            ttfs_p99_latency: P99 latency from speech end to final transcript
+                in seconds. See https://github.com/pipecat-ai/stt-benchmark.
+            **kwargs: Additional arguments passed to WebsocketSTTService.
+        """
+        default_settings = self.Settings(
+            model=None,
+            language=Language.EN,
+            interim_results=True,
+            endpointing=None,
+            multichannel=None,
+            channels=None,
+            diarize=None,
+        )
+        if settings is not None:
+            default_settings.apply_update(settings)
+
+        super().__init__(
+            sample_rate=sample_rate,
+            settings=default_settings,
+            ttfs_p99_latency=ttfs_p99_latency,
+            **kwargs,
+        )
+
+        self._api_key = api_key
+        self._ws_url = ws_url
+        self._encoding = encoding
+
+        self._receive_task: asyncio.Task | None = None
+        self._session_ready = asyncio.Event()
+
+    def can_generate_metrics(self) -> bool:
+        """Check if the service can generate metrics.
+
+        Returns:
+            True if metrics generation is supported.
+        """
+        return True
+
+    def language_to_service_language(self, language: Language) -> str | None:
+        """Convert a Language enum to the xAI STT language code."""
+        return language_to_xai_stt_language(language)
+
+    async def _update_settings(self, delta: Settings) -> dict[str, Any]:
+        """Apply a settings delta and reconnect to apply changes.
+
+        xAI STT configures the session via WebSocket query parameters, so any
+        change requires a fresh connection.
+        """
+        changed = await super()._update_settings(delta)
+        if not changed:
+            return changed
+        await self._disconnect()
+        await self._connect()
+        return changed
+
+    async def start(self, frame: StartFrame):
+        """Start the speech-to-text service."""
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the speech-to-text service."""
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the speech-to-text service."""
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        """Forward raw audio bytes to the xAI STT WebSocket.
+
+        Transcription frames are pushed from the receive task, not yielded
+        from this coroutine.
+        """
+        if self._websocket and self._websocket.state is State.OPEN and self._session_ready.is_set():
+            try:
+                await self._websocket.send(audio)
+            except Exception as e:
+                await self.push_error(error_msg=f"xAI STT send failed: {e}", exception=e)
+        yield None
+
+    def _build_ws_url(self) -> str:
+        """Build the WebSocket URL with session query parameters."""
+        s = self._settings
+        params: dict[str, Any] = {
+            "sample_rate": self.sample_rate,
+            "encoding": self._encoding,
+        }
+
+        if s.language is not None:
+            params["language"] = s.language
+
+        optional_fields = {
+            "interim_results": s.interim_results,
+            "endpointing": s.endpointing,
+            "multichannel": s.multichannel,
+            "channels": s.channels,
+            "diarize": s.diarize,
+        }
+        for key, val in optional_fields.items():
+            if val is None:
+                continue
+            if isinstance(val, bool):
+                params[key] = str(val).lower()
+            else:
+                params[key] = val
+
+        return f"{self._ws_url}?{urlencode(params)}"
+
+    async def _connect(self):
+        """Establish the WebSocket connection and start the receive task."""
+        await super()._connect()
+        await self._connect_websocket()
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+    async def _disconnect(self):
+        """Tear down the WebSocket connection and cancel the receive task."""
+        await super()._disconnect()
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                await self._websocket.send(json.dumps({"type": "audio.done"}))
+        except Exception as e:
+            logger.debug(f"{self} error sending audio.done during disconnect: {e}")
+
+        if self._receive_task:
+            await self.cancel_task(self._receive_task)
+            self._receive_task = None
+
+        await self._disconnect_websocket()
+
+    async def _connect_websocket(self):
+        """Open a WebSocket connection to the xAI STT endpoint."""
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                return
+
+            logger.debug("Connecting to xAI STT WebSocket")
+            self._session_ready.clear()
+
+            ws_url = self._build_ws_url()
+            headers = {
+                "Authorization": f"Bearer {self._api_key}",
+                "User-Agent": f"xAI/1.0 (integration=Pipecat/{pipecat_version()})",
+            }
+            self._websocket = await websocket_connect(ws_url, additional_headers=headers)
+            await self._call_event_handler("on_connected")
+            logger.debug(f"{self} connected to xAI STT WebSocket")
+        except Exception as e:
+            await self.push_error(error_msg=f"Unable to connect to xAI STT: {e}", exception=e)
+            raise
+
+    async def _disconnect_websocket(self):
+        """Close the WebSocket connection."""
+        try:
+            if self._websocket:
+                logger.debug("Disconnecting from xAI STT WebSocket")
+                await self._websocket.close()
+        except Exception as e:
+            await self.push_error(error_msg=f"Error closing xAI STT websocket: {e}", exception=e)
+        finally:
+            self._websocket = None
+            self._session_ready.clear()
+            await self._call_event_handler("on_disconnected")
+
+    async def _receive_messages(self):
+        """Receive and dispatch xAI STT WebSocket messages."""
+        if not self._websocket:
+            raise Exception("Websocket not connected")
+        async for message in self._websocket:
+            try:
+                data = json.loads(message)
+            except json.JSONDecodeError:
+                logger.warning(f"{self} received non-JSON message: {message}")
+                continue
+            await self._handle_message(data)
+
+    async def _handle_message(self, message: dict[str, Any]):
+        """Branch on xAI STT event type."""
+        msg_type = message.get("type")
+
+        if msg_type == "transcript.created":
+            self._session_ready.set()
+            logger.debug(f"{self} xAI STT session ready")
+        elif msg_type == "transcript.partial":
+            await self._handle_transcript(message)
+        elif msg_type == "transcript.done":
+            if message.get("text"):
+                await self._push_final_transcript(message, speech_final=True)
+        elif msg_type == "error":
+            await self.push_error(
+                error_msg=f"xAI STT error: {message.get('message', message)}",
+                exception=Exception(message),
+            )
+        else:
+            logger.debug(f"{self} unhandled xAI STT message: {message}")
+
+    async def _handle_transcript(self, message: dict[str, Any]):
+        text = message.get("text", "")
+        if not text:
+            return
+
+        is_final = bool(message.get("is_final"))
+        speech_final = bool(message.get("speech_final"))
+        language = self._language_for_frame()
+
+        if is_final:
+            await self._push_final_transcript(
+                message, speech_final=speech_final, language=language, text=text
+            )
+        else:
+            await self.push_frame(
+                InterimTranscriptionFrame(
+                    text,
+                    self._user_id,
+                    time_now_iso8601(),
+                    language,
+                    result=message,
+                )
+            )
+
+    async def _push_final_transcript(
+        self,
+        message: dict[str, Any],
+        *,
+        speech_final: bool,
+        language: Language | None = None,
+        text: str | None = None,
+    ):
+        text = text if text is not None else message.get("text", "")
+        if not text:
+            return
+        language = language if language is not None else self._language_for_frame()
+
+        await self.push_frame(
+            TranscriptionFrame(
+                text,
+                self._user_id,
+                time_now_iso8601(),
+                language,
+                result=message,
+                finalized=speech_final,
+            )
+        )
+        await self._trace_transcription(text, True, language)
+        if speech_final:
+            await self.stop_processing_metrics()
+
+    def _language_for_frame(self) -> Language:
+        """Return a Language enum suitable for transcription frames.
+
+        Settings stores the service-specific string (e.g. ``"en"``); frames
+        carry the enum value.
+        """
+        lang = self._settings.language
+        if isinstance(lang, Language):
+            return lang
+        if isinstance(lang, str):
+            try:
+                return Language(lang)
+            except ValueError:
+                return Language.EN
+        return Language.EN
+
+    @traced_stt
+    async def _trace_transcription(self, transcript: str, is_final: bool, language: Language):
+        """Record transcription event for tracing."""
+        pass

src/pipecat/services/xai/tts.py

@@ -6,22 +6,48 @@
 
 """xAI text-to-speech service implementation.
 
-Uses xAI's HTTP TTS endpoint documented at:
-https://docs.x.ai/developers/model-capabilities/audio/text-to-speech
+Provides two TTS services against xAI's voice API:
+
+- :class:`XAIHttpTTSService` uses the batch HTTP endpoint at
+  ``https://api.x.ai/v1/tts``.
+- :class:`XAITTSService` uses the streaming WebSocket endpoint at
+  ``wss://api.x.ai/v1/tts``.
+
+See https://docs.x.ai/developers/rest-api-reference/inference/voice.
 """
 
+import base64
+import json
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass
+from typing import Any
+from urllib.parse import urlencode
 
 import aiohttp
 from loguru import logger
 
-from pipecat.frames.frames import ErrorFrame, Frame, TTSAudioRawFrame
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    StartFrame,
+    TTSAudioRawFrame,
+    TTSStoppedFrame,
+)
 from pipecat.services.settings import TTSSettings
-from pipecat.services.tts_service import TTSService
+from pipecat.services.tts_service import InterruptibleTTSService, TTSService
 from pipecat.transcriptions.language import Language, resolve_language
 from pipecat.utils.tracing.service_decorators import traced_tts
 
+try:
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use XAITTSService, you need to `pip install pipecat-ai[xai]`.")
+    raise Exception(f"Missing module: {e}")
+
 
 def language_to_xai_language(language: Language) -> str | None:
     """Convert a Language enum to xAI language code.
@@ -214,3 +240,249 @@ class XAIHttpTTSService(TTSService):
                     )
         except Exception as e:
             yield ErrorFrame(error=f"Unknown error occurred: {e}")
+
+
+@dataclass
+class XAIWebsocketTTSSettings(TTSSettings):
+    """Settings for XAITTSService (WebSocket streaming)."""
+
+    pass
+
+
+class XAITTSService(InterruptibleTTSService):
+    """xAI streaming text-to-speech service.
+
+    Connects to xAI's WebSocket TTS endpoint and streams audio chunks back as
+    they are synthesized. Text can be sent incrementally via ``text.delta``
+    messages and each utterance is terminated with ``text.done``. The server
+    responds with ``audio.delta`` chunks followed by an ``audio.done`` message.
+
+    Audio parameters (voice, language, codec, sample rate, bit rate) are passed
+    as query string parameters on the WebSocket URL; changing any of them at
+    runtime reconnects the WebSocket.
+    """
+
+    Settings = XAIWebsocketTTSSettings
+    _settings: Settings
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str = "wss://api.x.ai/v1/tts",
+        sample_rate: int | None = None,
+        codec: str = "pcm",
+        settings: Settings | None = None,
+        **kwargs,
+    ):
+        """Initialize the xAI WebSocket TTS service.
+
+        Args:
+            api_key: xAI API key for authentication.
+            base_url: xAI TTS WebSocket endpoint. Defaults to
+                ``wss://api.x.ai/v1/tts``.
+            sample_rate: Output audio sample rate in Hz. If None, uses the
+                pipeline default.
+            codec: Output audio codec. One of ``pcm``, ``wav``, ``mulaw``,
+                ``alaw``. Defaults to ``pcm`` so emitted ``TTSAudioRawFrame``
+                objects need no decoding downstream.
+            settings: Runtime-updatable settings.
+            **kwargs: Additional arguments passed to parent
+                ``InterruptibleTTSService``.
+        """
+        default_settings = self.Settings(
+            model=None,
+            voice="eve",
+            language=Language.EN,
+        )
+
+        if settings is not None:
+            default_settings.apply_update(settings)
+
+        super().__init__(
+            push_start_frame=True,
+            push_stop_frames=True,
+            sample_rate=sample_rate,
+            settings=default_settings,
+            **kwargs,
+        )
+
+        self._api_key = api_key
+        self._base_url = base_url
+        self._codec = codec
+        self._receive_task = None
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics."""
+        return True
+
+    def language_to_service_language(self, language: Language) -> str | None:
+        """Convert a Language enum to xAI language format."""
+        return language_to_xai_language(language)
+
+    async def start(self, frame: StartFrame):
+        """Start the xAI WebSocket TTS service."""
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the xAI WebSocket TTS service."""
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the xAI WebSocket TTS service."""
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def _connect(self):
+        await super()._connect()
+
+        await self._connect_websocket()
+
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+    async def _disconnect(self):
+        await super()._disconnect()
+
+        if self._receive_task:
+            await self.cancel_task(self._receive_task)
+            self._receive_task = None
+
+        await self._disconnect_websocket()
+
+    async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
+        """Apply a settings delta. Reconnects if any URL-baked field changes."""
+        changed = await super()._update_settings(delta)
+
+        if changed:
+            await self._disconnect()
+            await self._connect()
+
+        return changed
+
+    def _build_url(self) -> str:
+        language = self._settings.language
+        if isinstance(language, Language):
+            language_value = language_to_xai_language(language) or language.value
+        else:
+            language_value = str(language) if language is not None else "auto"
+
+        params: dict[str, Any] = {
+            "voice": self._settings.voice,
+            "language": language_value,
+            "codec": self._codec,
+            "sample_rate": self.sample_rate,
+        }
+        return f"{self._base_url}?{urlencode(params)}"
+
+    async def _connect_websocket(self):
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                return
+
+            logger.debug("Connecting to xAI TTS")
+
+            url = self._build_url()
+            headers = {"Authorization": f"Bearer {self._api_key}"}
+            self._websocket = await websocket_connect(url, additional_headers=headers)
+
+            await self._call_event_handler("on_connected")
+        except Exception as e:
+            await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
+            self._websocket = None
+            await self._call_event_handler("on_connection_error", f"{e}")
+
+    async def _disconnect_websocket(self):
+        try:
+            await self.stop_all_metrics()
+
+            if self._websocket:
+                logger.debug("Disconnecting from xAI TTS")
+                await self._websocket.close()
+        except Exception as e:
+            await self.push_error(error_msg=f"Error disconnecting from xAI TTS: {e}", exception=e)
+        finally:
+            self._websocket = None
+            await self._call_event_handler("on_disconnected")
+
+    def _get_websocket(self):
+        if self._websocket:
+            return self._websocket
+        raise Exception("Websocket not connected")
+
+    async def flush_audio(self, context_id: str | None = None):
+        """Signal end-of-utterance so xAI begins synthesizing what it has buffered."""
+        if not self._websocket or self._websocket.state is State.CLOSED:
+            return
+        await self._get_websocket().send(json.dumps({"type": "text.done"}))
+
+    async def _receive_messages(self):
+        async for message in self._get_websocket():
+            if isinstance(message, bytes):
+                logger.warning(f"{self}: unexpected binary frame from xAI TTS")
+                continue
+            try:
+                msg = json.loads(message)
+            except json.JSONDecodeError:
+                logger.error(f"{self}: invalid JSON message: {message}")
+                continue
+
+            msg_type = msg.get("type")
+            context_id = self.get_active_audio_context_id()
+
+            if msg_type == "audio.delta":
+                audio_b64 = msg.get("delta")
+                if not audio_b64:
+                    continue
+                audio = base64.b64decode(audio_b64)
+                await self.stop_ttfb_metrics()
+                if context_id:
+                    frame = TTSAudioRawFrame(
+                        audio=audio,
+                        sample_rate=self.sample_rate,
+                        num_channels=1,
+                        context_id=context_id,
+                    )
+                    await self.append_to_audio_context(context_id, frame)
+            elif msg_type == "audio.done":
+                await self.stop_all_metrics()
+                if 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":
+                await self.stop_all_metrics()
+                error_detail = msg.get("message") or msg.get("error") or str(msg)
+                if context_id:
+                    await self.append_to_audio_context(
+                        context_id, TTSStoppedFrame(context_id=context_id)
+                    )
+                    await self.remove_audio_context(context_id)
+                await self.push_error(error_msg=f"xAI TTS error: {error_detail}")
+            else:
+                logger.debug(f"{self}: unhandled xAI message type: {msg_type}")
+
+    @traced_tts
+    async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
+        """Generate TTS audio from text using xAI's streaming WebSocket API."""
+        logger.debug(f"{self}: Generating TTS [{text}]")
+
+        try:
+            if not self._websocket or self._websocket.state is State.CLOSED:
+                await self._connect()
+
+            try:
+                await self._get_websocket().send(json.dumps({"type": "text.delta", "delta": text}))
+                await self.start_tts_usage_metrics(text)
+            except Exception as e:
+                yield ErrorFrame(error=f"Unknown error occurred: {e}")
+                yield TTSStoppedFrame(context_id=context_id)
+                await self._disconnect()
+                await self._connect()
+                return
+            yield None
+        except Exception as e:
+            yield ErrorFrame(error=f"Unknown error occurred: {e}")

src/pipecat/turns/user_start/__init__.py

@@ -11,9 +11,15 @@ from .transcription_user_turn_start_strategy import TranscriptionUserTurnStartSt
 from .vad_user_turn_start_strategy import VADUserTurnStartStrategy
 from .wake_phrase_user_turn_start_strategy import WakePhraseUserTurnStartStrategy
 
+try:
+    from .krisp_viva_ip_user_turn_start_strategy import KrispVivaIPUserTurnStartStrategy
+except ImportError:
+    KrispVivaIPUserTurnStartStrategy = None
+
 __all__ = [
     "BaseUserTurnStartStrategy",
     "ExternalUserTurnStartStrategy",
+    "KrispVivaIPUserTurnStartStrategy",
     "MinWordsUserTurnStartStrategy",
     "TranscriptionUserTurnStartStrategy",
     "UserTurnStartedParams",

src/pipecat/turns/user_start/base_user_turn_start_strategy.py

@@ -101,7 +101,7 @@ class BaseUserTurnStartStrategy(BaseObject):
         """Reset the strategy to its initial state."""
         pass
 
-    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult | None:
         """Process an incoming frame.
 
         Subclasses should override this to implement logic that decides whether
@@ -111,8 +111,8 @@ class BaseUserTurnStartStrategy(BaseObject):
             frame: The frame to be processed.
 
         Returns:
-            A ProcessFrameResult indicating the outcome. Subclasses that return
-            None are treated as CONTINUE for backward compatibility.
+            A ProcessFrameResult indicating the outcome, or None (treated as
+            CONTINUE for backward compatibility).
         """
         pass
 

src/pipecat/turns/user_start/krisp_viva_ip_user_turn_start_strategy.py

@@ -0,0 +1,282 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User turn start strategy using Krisp Interruption Prediction (IP).
+
+This strategy uses Krisp's IP model to distinguish genuine user interruptions
+from backchannels (e.g. "uh-huh", "yeah"). Instead of triggering a user turn
+on every VAD speech event, it collects audio after VAD detects speech and runs
+the IP model to predict whether the speech is a real interruption.
+
+Only when the IP model's probability exceeds the configured threshold is
+``trigger_user_turn_started()`` called. This prevents the bot from being
+interrupted by brief acknowledgements or filler words.
+"""
+
+import os
+
+import numpy as np
+from loguru import logger
+
+from pipecat.audio.krisp_instance import (
+    KrispVivaSDKManager,
+    int_to_krisp_frame_duration,
+    int_to_krisp_sample_rate,
+)
+from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
+    Frame,
+    InputAudioRawFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
+)
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
+
+try:
+    import krisp_audio
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use KrispVivaIPUserTurnStartStrategy, you need to install krisp_audio."
+    )
+    raise Exception(f"Missing module: {e}")
+
+
+class KrispVivaIPUserTurnStartStrategy(BaseUserTurnStartStrategy):
+    """User turn start strategy using Krisp VIVA Interruption Prediction.
+
+    When VAD detects user speech, this strategy feeds audio frames into
+    the Krisp VIVA IP model. The model outputs a probability indicating
+    whether the speech is a genuine interruption (as opposed to a
+    backchannel). A user turn is triggered only when this probability
+    exceeds the configured threshold.
+
+    This strategy is designed to work alongside other start strategies
+    (e.g. ``TranscriptionUserTurnStartStrategy`` as a fallback) via the
+    strategy list in ``UserTurnStrategies``.
+
+    Example::
+
+        from pipecat.turns.user_start import KrispVivaIPUserTurnStartStrategy
+
+        strategies = UserTurnStrategies(
+            start=[
+                KrispVivaIPUserTurnStartStrategy(
+                    model_path="/path/to/ip_model.kef",
+                    threshold=0.5,
+                ),
+                TranscriptionUserTurnStartStrategy(),
+            ],
+        )
+    """
+
+    def __init__(
+        self,
+        *,
+        model_path: str | None = None,
+        threshold: float = 0.5,
+        frame_duration_ms: int = 20,
+        api_key: str = "",
+        **kwargs,
+    ):
+        """Initialize the Krisp VIVA IP user turn start strategy.
+
+        Args:
+            model_path: Path to the Krisp VIVA IP model file (.kef). If None,
+                uses the KRISP_VIVA_IP_MODEL_PATH environment variable.
+            threshold: IP probability threshold (0.0 to 1.0). When the model's
+                output exceeds this value, the speech is classified as a genuine
+                interruption.
+            frame_duration_ms: Frame duration in milliseconds for IP processing.
+                Supported values: 10, 15, 20, 30, 32.
+            api_key: Krisp SDK API key. If empty, falls back to the
+                KRISP_VIVA_API_KEY environment variable.
+            **kwargs: Additional arguments passed to BaseUserTurnStartStrategy.
+        """
+        super().__init__(**kwargs)
+
+        self._threshold = threshold
+        self._frame_duration_ms = frame_duration_ms
+        self._api_key = api_key
+
+        self._model_path = model_path or os.getenv("KRISP_VIVA_IP_MODEL_PATH")
+        if not self._model_path:
+            raise ValueError(
+                "IP model path must be provided via model_path or "
+                "KRISP_VIVA_IP_MODEL_PATH environment variable."
+            )
+        if not self._model_path.endswith(".kef"):
+            raise ValueError("Model is expected with .kef extension")
+        if not os.path.isfile(self._model_path):
+            raise FileNotFoundError(f"IP model file not found: {self._model_path}")
+
+        self._sdk_acquired = False
+        self._ip_session = None
+        self._samples_per_frame: int | None = None
+        self._sample_rate: int | None = None
+
+        # State tracking
+        self._speech_active = False
+        self._audio_buffer = bytearray()
+        self._decision_made = False
+
+        # Acquire SDK
+        try:
+            KrispVivaSDKManager.acquire(api_key=api_key)
+            self._sdk_acquired = True
+        except Exception as e:
+            raise RuntimeError(f"Failed to initialize Krisp SDK: {e}")
+
+    async def cleanup(self):
+        """Release Krisp SDK resources."""
+        if self._sdk_acquired:
+            try:
+                self._ip_session = None
+                KrispVivaSDKManager.release()
+                self._sdk_acquired = False
+            except Exception as e:
+                logger.error(f"Error cleaning up Krisp VIVA IP strategy: {e}", exc_info=True)
+
+    def _ensure_session(self, sample_rate: int):
+        """Create or re-create the IP session when sample rate changes.
+
+        Args:
+            sample_rate: Audio sample rate in Hz.
+        """
+        if self._sample_rate == sample_rate and self._ip_session is not None:
+            return
+
+        self._sample_rate = sample_rate
+        self._samples_per_frame = int((sample_rate * self._frame_duration_ms) / 1000)
+
+        model_info = krisp_audio.ModelInfo()
+        model_info.path = self._model_path
+
+        ip_cfg = krisp_audio.IpSessionConfig()
+        ip_cfg.inputSampleRate = int_to_krisp_sample_rate(sample_rate)
+        ip_cfg.inputFrameDuration = int_to_krisp_frame_duration(self._frame_duration_ms)
+        ip_cfg.modelInfo = model_info
+
+        self._ip_session = krisp_audio.IpFloat.create(ip_cfg)
+        logger.debug(f"Krisp VIVA IP session created (sample_rate={sample_rate})")
+
+    def _reset_state(self):
+        """Reset speech tracking state for the next candidate interruption."""
+        self._speech_active = False
+        self._audio_buffer.clear()
+        self._decision_made = False
+
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        await super().reset()
+        self._reset_state()
+
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
+        """Process a frame to detect genuine user interruptions.
+
+        On ``VADUserStartedSpeakingFrame``, begins collecting audio.
+        On ``InputAudioRawFrame``, feeds audio through the IP model and
+        triggers a user turn if the interruption probability exceeds the
+        threshold.
+        On ``VADUserStoppedSpeakingFrame`` or ``BotStoppedSpeakingFrame``,
+        resets the candidate state.
+
+        Args:
+            frame: The incoming frame.
+
+        Returns:
+            STOP if a genuine interruption was detected, CONTINUE otherwise.
+        """
+        if isinstance(frame, VADUserStartedSpeakingFrame):
+            return await self._handle_vad_started(frame)
+        elif isinstance(frame, InputAudioRawFrame):
+            return await self._handle_audio(frame)
+        elif isinstance(frame, (VADUserStoppedSpeakingFrame, BotStoppedSpeakingFrame)):
+            return await self._handle_reset(frame)
+
+        return ProcessFrameResult.CONTINUE
+
+    async def _handle_vad_started(self, frame: VADUserStartedSpeakingFrame) -> ProcessFrameResult:
+        """Begin collecting audio for interruption classification.
+
+        Args:
+            frame: The VAD speech-start frame.
+
+        Returns:
+            Always CONTINUE; the decision is deferred until enough audio is processed.
+        """
+        logger.trace("Krisp VIVA IP: VAD speech started, collecting audio for classification")
+        self._speech_active = True
+        self._audio_buffer.clear()
+        self._decision_made = False
+        return ProcessFrameResult.CONTINUE
+
+    async def _handle_audio(self, frame: InputAudioRawFrame) -> ProcessFrameResult:
+        """Feed audio to the IP model and check for genuine interruption.
+
+        Args:
+            frame: Raw audio input frame.
+
+        Returns:
+            STOP if the model detects a genuine interruption, CONTINUE otherwise.
+        """
+        if not self._speech_active or self._decision_made:
+            return ProcessFrameResult.CONTINUE
+
+        self._ensure_session(frame.sample_rate)
+
+        if self._ip_session is None or self._samples_per_frame is None:
+            logger.warning("IP session not ready, skipping frame")
+            return ProcessFrameResult.CONTINUE
+
+        self._audio_buffer.extend(frame.audio)
+
+        total_samples = len(self._audio_buffer) // 2  # 2 bytes per int16 sample
+        num_complete_frames = total_samples // self._samples_per_frame
+
+        if num_complete_frames == 0:
+            return ProcessFrameResult.CONTINUE
+
+        complete_samples_count = num_complete_frames * self._samples_per_frame
+        bytes_to_process = complete_samples_count * 2
+
+        audio_to_process = bytes(self._audio_buffer[:bytes_to_process])
+        self._audio_buffer = self._audio_buffer[bytes_to_process:]
+
+        audio_int16 = np.frombuffer(audio_to_process, dtype=np.int16)
+        audio_float32 = audio_int16.astype(np.float32) / 32768.0
+        frames = audio_float32.reshape(-1, self._samples_per_frame)
+
+        for ip_frame in frames:
+            ip_prob = self._ip_session.process(ip_frame.tolist(), self._speech_active)
+
+            if ip_prob >= self._threshold:
+                logger.debug(
+                    f"Krisp VIVA IP: genuine interruption detected (prob={ip_prob:.3f}, "
+                    f"threshold={self._threshold})"
+                )
+                self._decision_made = True
+                await self.trigger_user_turn_started()
+                return ProcessFrameResult.STOP
+
+        return ProcessFrameResult.CONTINUE
+
+    async def _handle_reset(
+        self, frame: VADUserStoppedSpeakingFrame | BotStoppedSpeakingFrame
+    ) -> ProcessFrameResult:
+        """Reset state when the candidate interruption window ends.
+
+        Args:
+            frame: The frame signaling end of speech or bot output.
+
+        Returns:
+            Always CONTINUE.
+        """
+        if self._speech_active:
+            logger.trace("Krisp VIVA IP: speech segment ended, resetting state")
+            self._reset_state()
+        return ProcessFrameResult.CONTINUE

src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py

@@ -89,7 +89,7 @@ class BaseUserTurnStopStrategy(BaseObject):
         """Reset the strategy to its initial state."""
         pass
 
-    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult | None:
         """Process an incoming frame to decide whether the user stopped speaking.
 
         Subclasses should override this to implement logic that decides whether
@@ -99,8 +99,8 @@ class BaseUserTurnStopStrategy(BaseObject):
             frame: The frame to be analyzed.
 
         Returns:
-            A ProcessFrameResult indicating the outcome. Subclasses that return
-            None are treated as CONTINUE for backward compatibility.
+            A ProcessFrameResult indicating the outcome, or None (treated as
+            CONTINUE for backward compatibility).
         """
         pass
 

src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py

@@ -7,7 +7,6 @@
 """Speech timeout-based user turn stop strategy."""
 
 import asyncio
-import time
 
 from loguru import logger
 
@@ -25,20 +24,25 @@ from pipecat.utils.asyncio.task_manager import BaseTaskManager
 
 
 class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
-    """User turn stop strategy that uses a configurable timeout to determine if the user is done speaking.
+    """User turn stop strategy using two independent timers after VAD stop.
 
-    After the user stops speaking (detected by VAD), this strategy waits for a
-    configurable timeout before triggering the end of the user's turn. The
-    timeout accounts for two factors:
+    After the user stops speaking (detected by VAD), this strategy runs two
+    independent timers. The user turn stop is triggered only when both have
+    finished and at least one transcript has been received:
 
-    - user_speech_timeout: Time to wait for the user to potentially say more
-      after they pause.
-    - stt_timeout: The P99 time for the STT service to return a transcription
-      after the user stops speaking, adjusted by the VAD stop_secs.
+    - user_speech_timeout: Policy floor — the window in which the user may
+      resume speaking after a pause. Always runs to completion.
+    - stt_timeout: Safety net for STT latency — the P99 time for the STT
+      service to return a final transcript after VAD stop, adjusted by the
+      VAD stop_secs. Short-circuited when the STT service emits a finalized
+      transcript (TranscriptionFrame.finalized=True), since finalization
+      means STT has nothing more to send.
 
-    For services that support finalization (TranscriptionFrame.finalized=True),
-    the turn can be triggered immediately once the finalized transcript is
-    received and the user resume speaking timeout has elapsed.
+    Fallback: when a transcript arrives without a VAD stop event, the
+    user_speech_timeout timer measures inactivity since the last transcript
+    (rearmed on each transcript). stt_timeout has no meaning here since it
+    is defined relative to VAD stop, and STT has already emitted a
+    transcript — so the stt wait is marked done immediately.
     """
 
     def __init__(self, *, user_speech_timeout: float = 0.6, **kwargs):
@@ -59,8 +63,11 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = False
         self._transcript_finalized = False
         self._vad_stopped_time: float | None = None
-        self._timeout_task: asyncio.Task | None = None
-        self._timeout_expired: bool = False
+
+        self._user_speech_timeout_task: asyncio.Task | None = None
+        self._stt_timeout_task: asyncio.Task | None = None
+        self._user_speech_wait_done: bool = False
+        self._stt_wait_done: bool = False
 
     async def reset(self):
         """Reset the strategy to its initial state."""
@@ -69,10 +76,9 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = False
         self._transcript_finalized = False
         self._vad_stopped_time = None
-        self._timeout_expired = False
-        if self._timeout_task:
-            await self.task_manager.cancel_task(self._timeout_task)
-            self._timeout_task = None
+        self._user_speech_wait_done = False
+        self._stt_wait_done = False
+        await self._cancel_all_tasks()
 
     async def setup(self, task_manager: BaseTaskManager):
         """Initialize the strategy with the given task manager.
@@ -85,9 +91,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
     async def cleanup(self):
         """Cleanup the strategy."""
         await super().cleanup()
-        if self._timeout_task:
-            await self.task_manager.cancel_task(self._timeout_task)
-            self._timeout_task = None
+        await self._cancel_all_tasks()
 
     async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to update strategy state.
@@ -105,8 +109,10 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
             self._stt_timeout = frame.ttfs_p99_latency
             self._stop_secs_warned = False
         elif isinstance(frame, VADUserStartedSpeakingFrame):
+            logger.debug(f"{self} VADUserStartedSpeakingFrame received")
             await self._handle_vad_user_started_speaking(frame)
         elif isinstance(frame, VADUserStoppedSpeakingFrame):
+            logger.debug(f"{self} VADUserStoppedSpeakingFrame received")
             await self._handle_vad_user_stopped_speaking(frame)
         elif isinstance(frame, TranscriptionFrame):
             await self._handle_transcription(frame)
@@ -118,11 +124,9 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = True
         self._transcript_finalized = False
         self._vad_stopped_time = None
-        self._timeout_expired = False
-        # Cancel any pending timeout
-        if self._timeout_task:
-            await self.task_manager.cancel_task(self._timeout_task)
-            self._timeout_task = None
+        self._user_speech_wait_done = False
+        self._stt_wait_done = False
+        await self._cancel_all_tasks()
 
     async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame):
         """Handle when the VAD indicates the user has stopped speaking."""
@@ -150,59 +154,69 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
                     f"user_turn_stop_timeout parameter in the LLMUserAggregatorParams."
                 )
 
-        # Start the timeout task
-        timeout = self._calculate_timeout()
-        self._timeout_task = self.task_manager.create_task(
-            self._timeout_handler(timeout), f"{self}::_timeout_handler"
-        )
-        # Make sure the task is scheduled.
+        # user_speech_timeout is the policy floor and always runs. A prior
+        # fallback-mode run of the same timer is superseded here.
+        await self._restart_user_speech_timer()
+
+        # stt_timeout is a safety net. Short-circuit it if the transcript is
+        # already finalized, or if the VAD stop_secs already covered it.
+        self._stt_wait_done = False
+        effective_stt_wait = max(0.0, self._stt_timeout - self._stop_secs)
+        if self._transcript_finalized or effective_stt_wait <= 0:
+            self._stt_wait_done = True
+        else:
+            self._stt_timeout_task = self.task_manager.create_task(
+                self._stt_timeout_handler(effective_stt_wait),
+                f"{self}::_stt_timeout_handler",
+            )
+
+        # Make sure the tasks are scheduled.
         await asyncio.sleep(0)
 
     async def _handle_transcription(self, frame: TranscriptionFrame):
         """Handle user transcription."""
         self._text += frame.text
+
         if frame.finalized:
             self._transcript_finalized = True
-            # For finalized transcripts, check if we can trigger early
+            # Short-circuit the stt_timeout safety net: STT has told us
+            # there's nothing more coming.
+            if not self._stt_wait_done:
+                self._stt_wait_done = True
+                if self._stt_timeout_task:
+                    await self.task_manager.cancel_task(self._stt_timeout_task)
+                    self._stt_timeout_task = None
+
+        # If both waits are already done, the turn was waiting on text —
+        # trigger now.
+        if self._user_speech_wait_done and self._stt_wait_done:
             await self._maybe_trigger_user_turn_stopped()
-        elif self._timeout_expired:
-            # The p99 timeout already elapsed without a transcript. Now that
-            # we have one, trigger the turn stop immediately.
-            await self.trigger_user_turn_stopped()
             return
 
-        # Fallback: handle transcripts when no VAD stop was received.
-        # This handles edge cases where transcripts arrive without VAD firing.
-        # _vad_stopped_time is None means VAD stopped hasn't been received yet.
-        # In fallback mode, reset timeout on each transcript to wait for inactivity.
+        # Fallback: transcript arrived without a VAD stop. Measure inactivity
+        # since the last transcript with the user_speech_timer. stt_timeout
+        # has no meaning here (it's defined relative to VAD stop), so mark
+        # the stt wait done immediately.
         if not self._vad_user_speaking and self._vad_stopped_time is None:
-            # Cancel existing fallback timeout if any
-            if self._timeout_task:
-                await self.task_manager.cancel_task(self._timeout_task)
-            timeout = self._calculate_timeout()
-            self._timeout_task = self.task_manager.create_task(
-                self._timeout_handler(timeout), f"{self}::_timeout_handler"
-            )
-            # Make sure the task is scheduled.
-            await asyncio.sleep(0)
+            self._stt_wait_done = True
+            await self._restart_user_speech_timer()
 
-    def _calculate_timeout(self) -> float:
-        """Calculate the timeout value based on current state.
+    async def _restart_user_speech_timer(self):
+        """Cancel any running user_speech timer and start a fresh one."""
+        if self._user_speech_timeout_task:
+            await self.task_manager.cancel_task(self._user_speech_timeout_task)
+            self._user_speech_timeout_task = None
+        self._user_speech_wait_done = False
+        self._user_speech_timeout_task = self.task_manager.create_task(
+            self._user_speech_timeout_handler(self._user_speech_timeout),
+            f"{self}::_user_speech_timeout_handler",
+        )
+        # Make sure the task is scheduled so it can't be cancelled before
+        # starting (which would leave its coroutine un-awaited).
+        await asyncio.sleep(0)
 
-        Returns:
-            The timeout in seconds to wait after VAD stopped speaking.
-        """
-        # Adjust STT timeout by VAD stop_secs since that time has already elapsed
-        effective_stt_wait = max(0, self._stt_timeout - self._stop_secs)
-
-        # If transcript is already finalized, we don't need to wait for STT
-        if self._transcript_finalized:
-            return self._user_speech_timeout
-
-        return max(effective_stt_wait, self._user_speech_timeout)
-
-    async def _timeout_handler(self, timeout: float):
-        """Wait for the timeout then trigger user turn stopped if conditions met.
+    async def _user_speech_timeout_handler(self, timeout: float):
+        """Wait user_speech_timeout then attempt to trigger user turn stopped.
 
         Args:
             timeout: The timeout in seconds to wait.
@@ -212,36 +226,46 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         except asyncio.CancelledError:
             return
         finally:
-            self._timeout_task = None
+            self._user_speech_timeout_task = None
 
-        self._timeout_expired = True
+        self._user_speech_wait_done = True
+        await self._maybe_trigger_user_turn_stopped()
+
+    async def _stt_timeout_handler(self, timeout: float):
+        """Wait stt_timeout then attempt to trigger user turn stopped.
+
+        Args:
+            timeout: The timeout in seconds to wait.
+        """
+        try:
+            await asyncio.sleep(timeout)
+        except asyncio.CancelledError:
+            return
+        finally:
+            self._stt_timeout_task = None
+
+        self._stt_wait_done = True
         await self._maybe_trigger_user_turn_stopped()
 
     async def _maybe_trigger_user_turn_stopped(self):
-        """Trigger user turn stopped if conditions are met.
+        """Trigger user turn stopped if all required conditions are met.
 
-        Conditions:
-        - User is not currently speaking
-        - We have transcription text
-        - Either the timeout has elapsed OR we have a finalized transcript
-          and user_speech_timeout has elapsed
+        Both timers must be done (stt is marked done immediately on the
+        fallback path and when finalization short-circuits the safety net),
+        the user must not be currently speaking, and at least one transcript
+        must have been received.
         """
         if self._vad_user_speaking or not self._text:
             return
 
-        # For finalized transcripts, check if user_speech_timeout has elapsed.
-        # If elapsed, trigger user turn stopped immediately. Else, wait for user resume
-        # speaking timeout.
-        if self._transcript_finalized and self._vad_stopped_time is not None:
-            elapsed = time.time() - self._vad_stopped_time
-            if elapsed >= self._user_speech_timeout:
-                # Cancel any remaining timeout since we're triggering now
-                if self._timeout_task:
-                    await self.task_manager.cancel_task(self._timeout_task)
-                    self._timeout_task = None
-                await self.trigger_user_turn_stopped()
-                return
-
-        # For non-finalized, only trigger if timeout task has completed
-        if self._timeout_task is None:
+        if self._user_speech_wait_done and self._stt_wait_done:
             await self.trigger_user_turn_stopped()
+
+    async def _cancel_all_tasks(self):
+        """Cancel any running timer tasks and clear the handles."""
+        if self._user_speech_timeout_task:
+            await self.task_manager.cancel_task(self._user_speech_timeout_task)
+            self._user_speech_timeout_task = None
+        if self._stt_timeout_task:
+            await self.task_manager.cancel_task(self._stt_timeout_task)
+            self._stt_timeout_task = None

src/pipecat/turns/user_turn_completion_mixin.py

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
     LLMRunFrame,
     LLMTextFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 # Turn completion markers
 USER_TURN_COMPLETE_MARKER = "✓"
@@ -178,7 +178,7 @@ class UserTurnCompletionConfig:
         return self.incomplete_long_prompt or DEFAULT_INCOMPLETE_LONG_PROMPT
 
 
-class UserTurnCompletionLLMServiceMixin:
+class UserTurnCompletionLLMServiceMixin(FrameProcessor):
     """Mixin that adds turn completion detection to LLM services.
 
     This mixin provides methods to push LLM text with turn completion detection.
@@ -292,7 +292,7 @@ class UserTurnCompletionLLMServiceMixin:
 
             # Push through pipeline to trigger LLM response
             await self.push_frame(
-                LLMMessagesAppendFrame(messages=[{"role": "system", "content": prompt}])
+                LLMMessagesAppendFrame(messages=[{"role": "developer", "content": prompt}])
             )
             await self.push_frame(LLMRunFrame())
 

src/pipecat/utils/context/llm_context_summarization.py

@@ -20,7 +20,11 @@ if TYPE_CHECKING:
 
 from loguru import logger
 
-from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
+from pipecat.processors.aggregators.llm_context import (
+    LLMContext,
+    LLMContextMessage,
+    LLMSpecificMessage,
+)
 
 # Fallback timeout (seconds) used when summarization_timeout is None.
 DEFAULT_SUMMARIZATION_TIMEOUT = 120.0
@@ -269,7 +273,7 @@ class LLMMessagesToSummarize:
         last_summarized_index: Index of the last message being summarized
     """
 
-    messages: list[dict]
+    messages: list[LLMContextMessage]
     last_summarized_index: int
 
 
@@ -415,7 +419,7 @@ class LLMContextSummarizationUtil:
 
     @staticmethod
     def _get_earliest_function_call_not_resolved_in_range(
-        messages: list[dict], start_idx: int, summary_end: int
+        messages: list[LLMContextMessage], start_idx: int, summary_end: int
     ) -> int:
         """Find the earliest message index with incomplete function calls.
 
@@ -470,9 +474,10 @@ class LLMContextSummarizationUtil:
             if role == "tool":
                 tool_call_id = msg.get("tool_call_id")
                 if tool_call_id and tool_call_id in pending_tool_calls:
-                    if not LLMContextSummarizationUtil._is_tool_message_pending(
-                        msg.get("content", "")
-                    ):
+                    content = msg.get("content", "")
+                    if not isinstance(content, str):
+                        content = ""
+                    if not LLMContextSummarizationUtil._is_tool_message_pending(content):
                         pending_tool_calls.pop(tool_call_id)
 
             # Check for async tool completion — a developer message with
@@ -480,7 +485,10 @@ class LLMContextSummarizationUtil:
             # async result has arrived and the call is now resolved.
             if role == "developer":
                 try:
-                    parsed = json.loads(msg.get("content", ""))
+                    content = msg.get("content", "")
+                    if not isinstance(content, str):
+                        continue
+                    parsed = json.loads(content)
                     if (
                         isinstance(parsed, dict)
                         and parsed.get("type") == "async_tool"

src/pipecat/utils/frame_queue.py

@@ -58,7 +58,7 @@ class FrameQueue(asyncio.Queue):
         Returns:
             True if at least one enqueued frame is an instance of ``frame_type``.
         """
-        for item in self._queue:
+        for item in self._queue:  # pyright: ignore[reportAttributeAccessIssue]
             if isinstance(self._frame_getter(item), frame_type):
                 return True
         return False

src/pipecat/utils/string.py

@@ -234,7 +234,7 @@ class TextPartForConcatenation:
     includes_inter_part_spaces: bool
 
     def __str__(self):
-        return f"{self.name}(text: [{self.text}], includes_inter_part_spaces: {self.includes_inter_part_spaces})"
+        return f"{type(self).__name__}(text: [{self.text}], includes_inter_part_spaces: {self.includes_inter_part_spaces})"
 
 
 def concatenate_aggregated_text(text_parts: list[TextPartForConcatenation]) -> str:

src/pipecat/utils/text/base_text_aggregator.py

@@ -125,7 +125,7 @@ class BaseTextAggregator(ABC):
         """
         pass
         # Make this a generator to satisfy type checker
-        yield  # pragma: no cover
+        yield  # pyright: ignore[reportReturnType]  # pragma: no cover
 
     @abstractmethod
     async def flush(self) -> Aggregation | None:

src/pipecat/utils/text/pattern_pair_aggregator.py

@@ -273,7 +273,7 @@ class PatternPairAggregator(SimpleTextAggregator):
             # Which is why we base the return on the first found.
             if start_count > end_count:
                 start_index = text.find(start)
-                return [start_index, pattern_info]
+                return (start_index, pattern_info)
 
         return None
 

src/pipecat/utils/tracing/service_attributes.py

@@ -440,7 +440,7 @@ def add_openai_realtime_span_attributes(
             if isinstance(tool, dict) and "name" in tool:
                 tool_names.append(tool["name"])
             elif hasattr(tool, "name"):
-                tool_names.append(tool.name)
+                tool_names.append(getattr(tool, "name"))
             elif isinstance(tool, dict) and "function" in tool and "name" in tool["function"]:
                 tool_names.append(tool["function"]["name"])
 
@@ -455,7 +455,7 @@ def add_openai_realtime_span_attributes(
         if function_calls:
             call = function_calls[0]
             if hasattr(call, "name"):
-                span.set_attribute("function_calls.first_name", call.name)
+                span.set_attribute("function_calls.first_name", getattr(call, "name"))
             elif isinstance(call, dict) and "name" in call:
                 span.set_attribute("function_calls.first_name", call["name"])