Add VADUserTurnStopStrategy for VAD-only realtime pipelines

Provides a dedicated stop strategy that triggers on VADUserStoppedSpeakingFrame
without requiring a turn analyzer or transcriptions. Complements the existing
VADTurnAnalyzerUserTurnStopStrategy and matches the one-per-modality convention
already used in user_start/.

.github/workflows/format.yaml

@@ -41,3 +41,7 @@ jobs:
       - name: Ruff linter (all rules)
         id: ruff-check
         run: uv run ruff check
+
+      - name: Type check (pyright)
+        id: pyright
+        run: uv run pyright

README.md

@@ -28,6 +28,10 @@
 
 ## 🌐 Pipecat Ecosystem
 
+### 🧩 Multi-agent systems
+
+Need multiple AI agents working together? [Pipecat Subagents](https://github.com/pipecat-ai/pipecat-subagents) lets you build distributed multi-agent systems where each agent runs its own pipeline and communicates through a shared message bus. Hand off conversations between specialists, dispatch background tasks, and scale agents across processes or machines.
+
 ### 📱 Client SDKs
 
 Building client applications? You can connect to Pipecat from any platform using our official SDKs:

changelog/4199.added.2.md

@@ -0,0 +1 @@
+- Added `VADUserTurnStopStrategy`, a VAD-only turn stop strategy that triggers the end of a user turn as soon as VAD reports the user stopped speaking. Intended for realtime speech-to-speech pipelines that rely solely on VAD and don't use a turn analyzer or STT transcriptions to decide end of turn.

changelog/4199.added.md

@@ -0,0 +1 @@
+- Added `VADTurnAnalyzerUserTurnStopStrategy`, a turn stop strategy that triggers immediately when the turn analyzer reports COMPLETE without waiting for STT transcriptions. This reduces end-of-turn latency for speech-to-speech pipelines (e.g. Gemini Live) where audio goes directly to the LLM.

changelog/4314.changed.md

@@ -0,0 +1 @@
+- Reduced debug log noise for LLM services. The system instruction is now logged once when composed (e.g. when turn completion is enabled) instead of on every LLM call. Per-call logs now show only the conversation messages, consistent across Google, Anthropic, AWS, and OpenAI services.

changelog/4324.added.md

@@ -0,0 +1 @@
+- Added incremental `pyright` type checking. A `pyrightconfig.json` at the repo root uses `typeCheckingMode: "basic"` with an explicit `include` list of modules that pass cleanly (`clocks`, `metrics`, `transcriptions`, `frames`, `observers`, `extensions`, `turns`, `pipeline`, `runner`). Remaining modules will be added in subsequent PRs. CI enforces the checked set via `uv run pyright` in the format workflow.

changelog/4324.changed.md

@@ -0,0 +1 @@
+- `LiveKitRunnerArguments.token` is now a required `str` (previously `str | None` with a default of `None`). LiveKit requires a token to join a room, so the type now reflects reality. This only affects custom runners that construct `LiveKitRunnerArguments` directly; code consuming the argument from the standard runner is unaffected.

changelog/4326.added.md

@@ -0,0 +1 @@
+- Added multilingual support to `DeepgramFluxSTTService` via a new `language_hints: list[Language]` setting. Works with Deepgram's new `flux-general-multi` model to bias transcription across English, Spanish, French, German, Hindi, Russian, Portuguese, Japanese, Italian, and Dutch. Omit the hints to use auto-detection, or pass a subset to bias toward expected languages. Hints can be updated mid-stream via `STTUpdateSettingsFrame` (sent as a Deepgram `Configure` control message, no reconnect) to support detect-then-lock flows.

changelog/4326.changed.md

@@ -0,0 +1 @@
+- `TranscriptionFrame.language` and `InterimTranscriptionFrame.language` emitted by `DeepgramFluxSTTService` now reflect the language Deepgram detected for each turn (read from the `languages` field on Flux's `TurnInfo` event). On `flux-general-multi` this gives per-turn accuracy for downstream consumers (e.g. TTS voice selection). `flux-general-en` continues to emit `Language.EN`.

examples/realtime/realtime-gemini-live-local-vad.py

@@ -10,6 +10,7 @@ import os
 from dotenv import load_dotenv
 from loguru import logger
 
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
@@ -28,6 +29,9 @@ from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService, Gemini
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.turns.user_start import VADUserTurnStartStrategy
+from pipecat.turns.user_stop import VADTurnAnalyzerUserTurnStopStrategy
+from pipecat.turns.user_turn_strategies import UserTurnStrategies
 
 load_dotenv(override=True)
 
@@ -73,6 +77,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
         context,
         user_params=LLMUserAggregatorParams(
+            user_turn_strategies=UserTurnStrategies(
+                start=[VADUserTurnStartStrategy()],
+                stop=[
+                    VADTurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())
+                ],
+            ),
             vad_analyzer=SileroVADAnalyzer(),
         ),
     )

examples/update-settings/stt/stt-deepgram-flux.py

@@ -51,7 +51,16 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    stt = DeepgramFluxSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+    # Start with the multilingual model and broad hints so Flux can auto-detect
+    # across English and Spanish. TranscriptionFrame.language will reflect
+    # whichever language Flux detected for each turn.
+    stt = DeepgramFluxSTTService(
+        api_key=os.getenv("DEEPGRAM_API_KEY"),
+        settings=DeepgramFluxSTTService.Settings(
+            model="flux-general-multi",
+            language_hints=[Language.EN, Language.ES],
+        ),
+    )
 
     tts = CartesiaTTSService(
         api_key=os.getenv("CARTESIA_API_KEY"),
@@ -108,17 +117,21 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         logger.info("Updating Deepgram Flux STT settings: eot_threshold, keyterm")
         await task.queue_frame(
             STTUpdateSettingsFrame(
-                delta=DeepgramFluxSTTSettings(
+                delta=DeepgramFluxSTTService.Settings(
                     eot_threshold=0.8,
                     keyterm=["Pipecat", "Deepgram"],
                 )
             )
         )
 
+        # Detect-then-lock: narrow the hints to a single language mid-stream.
+        # Sent as a Configure message — no reconnect needed.
         await asyncio.sleep(10)
-        logger.info("Updating Deepgram Flux STT settings: language=es")
+        logger.info("Updating Deepgram Flux STT settings: language_hints=[es]")
         await task.queue_frame(
-            STTUpdateSettingsFrame(delta=DeepgramFluxSTTService.Settings(language=Language.ES))
+            STTUpdateSettingsFrame(
+                delta=DeepgramFluxSTTService.Settings(language_hints=[Language.ES])
+            )
         )
 
     @transport.event_handler("on_client_disconnected")

examples/voice/voice-krisp-viva.py

@@ -4,20 +4,24 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Interruptible bot with Krisp VIVA noise filtering and turn detection.
+"""Interruptible bot with Krisp VIVA noise filtering, turn detection, and IP.
 
 This example demonstrates a conversational bot with:
 - Krisp VIVA noise reduction on incoming audio
-- Krisp VIVA Turn detection for natural interruptions
+- Krisp VIVA Turn detection for end-of-turn
+- Krisp Interruption Prediction (IP) to filter backchannels from real interruptions
 - Voice activity detection (VAD)
 
 Required environment variables:
 - KRISP_VIVA_FILTER_MODEL_PATH: Path to the Krisp noise filter model file (.kef)
 - KRISP_VIVA_TURN_MODEL_PATH: Path to the Krisp turn detection model file (.kef)
-- DEEPGRAM_API_KEY: Deepgram API key for STT/TTS
+- KRISP_VIVA_IP_MODEL_PATH: Path to the Krisp IP model file (.kef)
+- DEEPGRAM_API_KEY: Deepgram API key for STT
+- CARTESIA_API_KEY: Cartesia API key for TTS
 - OPENAI_API_KEY: OpenAI API key for LLM
 
 Optional environment variables:
+- KRISP_VIVA_API_KEY: Krisp SDK API key (or set in code)
 - KRISP_NOISE_SUPPRESSION_LEVEL: Noise suppression level 0-100 (default: 100)
   Higher values = more aggressive noise reduction
 """
@@ -49,31 +53,30 @@ from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.turns.user_start import (
+    KrispVivaIPUserTurnStartStrategy,
+    TranscriptionUserTurnStartStrategy,
+)
 from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy
 from pipecat.turns.user_turn_strategies import UserTurnStrategies
 
 load_dotenv(override=True)
 
-# We use lambdas to defer transport parameter creation until the transport
-# type is selected at runtime.
-
-krisp_viva_filter = KrispVivaFilter()
-
 transport_params = {
     "daily": lambda: DailyParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=krisp_viva_filter,
+        audio_in_filter=KrispVivaFilter(),
     ),
     "twilio": lambda: FastAPIWebsocketParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=krisp_viva_filter,
+        audio_in_filter=KrispVivaFilter(),
     ),
     "webrtc": lambda: TransportParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=krisp_viva_filter,
+        audio_in_filter=KrispVivaFilter(),
     ),
 }
 
@@ -102,7 +105,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         context,
         user_params=LLMUserAggregatorParams(
             user_turn_strategies=UserTurnStrategies(
-                stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=KrispVivaTurn())]
+                start=[
+                    KrispVivaIPUserTurnStartStrategy(threshold=0.5),
+                    TranscriptionUserTurnStartStrategy(),
+                ],
+                stop=[TurnAnalyzerUserTurnStopStrategy(turn_analyzer=KrispVivaTurn())],
             ),
             vad_analyzer=SileroVADAnalyzer(),  # or KrispVivaVadAnalyzer
         ),

pyrightconfig.json

@@ -0,0 +1,36 @@
+{
+  "typeCheckingMode": "basic",
+  "pythonVersion": "3.11",
+  "pythonPlatform": "All",
+  "include": [
+    "src/pipecat/clocks",
+    "src/pipecat/metrics",
+    "src/pipecat/transcriptions",
+    "src/pipecat/frames",
+    "src/pipecat/observers",
+    "src/pipecat/extensions",
+    "src/pipecat/turns",
+    "src/pipecat/pipeline",
+    "src/pipecat/runner"
+  ],
+  "exclude": [
+    "**/*_pb2.py",
+    "**/__pycache__"
+  ],
+  "ignore": [
+    "src/pipecat/adapters",
+    "src/pipecat/audio",
+    "src/pipecat/processors",
+    "src/pipecat/serializers",
+    "src/pipecat/services",
+    "src/pipecat/sync",
+    "src/pipecat/tests",
+    "src/pipecat/transports",
+    "src/pipecat/utils",
+    "tests",
+    "examples",
+    "scripts",
+    "docs"
+  ],
+  "reportMissingImports": false
+}

scripts/fix-ruff-and-typecheck.sh

@@ -6,5 +6,9 @@ PROJECT_ROOT="$(dirname "$SCRIPT_DIR")"
 
 echo "Running ruff format..."
 uv run ruff format "$PROJECT_ROOT"
+
 echo "Running ruff check..."
 uv run ruff check --fix "$PROJECT_ROOT"
+
+echo "Running pyright check..."
+uv run pyright

src/pipecat/adapters/schemas/direct_function.py

@@ -211,9 +211,9 @@ class BaseDirectFunctionWrapper:
             return {"type": "number"}
         elif type_hint is bool:
             return {"type": "boolean"}
-        elif type_hint is dict or type_hint is dict:
+        elif type_hint is dict:
             return {"type": "object"}
-        elif type_hint is list or type_hint is list:
+        elif type_hint is list:
             return {"type": "array"}
 
         # Get origin and arguments for complex types
@@ -225,11 +225,11 @@ class BaseDirectFunctionWrapper:
             return {"anyOf": [self._typehint_to_jsonschema(arg) for arg in args]}
 
         # Handle List, Tuple, Set with specific item types
-        if origin in (list, list, tuple, tuple, set, set) and args:
+        if origin in (list, tuple, set) and args:
             return {"type": "array", "items": self._typehint_to_jsonschema(args[0])}
 
         # Handle Dict with specific key/value types
-        if origin in (dict, dict) and len(args) == 2:
+        if origin is dict and len(args) == 2:
             # For JSON Schema, keys must be strings
             return {"type": "object", "additionalProperties": self._typehint_to_jsonschema(args[1])}
 

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/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/services/anthropic/llm.py

@@ -345,9 +345,7 @@ class AnthropicLLMService(LLMService):
 
             adapter = self.get_llm_adapter()
             messages_for_logging = adapter.get_messages_for_logging(context)
-            logger.debug(
-                f"{self}: Generating chat from context [{params_from_context['system']}] | {messages_for_logging}"
-            )
+            logger.debug(f"{self}: Generating chat from context {messages_for_logging}")
 
             await self.start_ttfb_metrics()
 

src/pipecat/services/aws/llm.py

@@ -475,9 +475,7 @@ class AWSBedrockLLMService(LLMService):
             # Log request params with messages redacted for logging
             adapter = self.get_llm_adapter()
             messages_for_logging = adapter.get_messages_for_logging(context)
-            logger.debug(
-                f"{self}: Generating chat from context [{system}] | {messages_for_logging}"
-            )
+            logger.debug(f"{self}: Generating chat from context {messages_for_logging}")
 
             async with self._aws_session.client(
                 service_name="bedrock-runtime", **self._aws_params

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/google/llm.py

@@ -392,7 +392,7 @@ class GoogleLLMService(LLMService):
         )
 
         logger.debug(
-            f"{self}: Generating chat from context [{params['system_instruction']}] | {adapter.get_messages_for_logging(context)}"
+            f"{self}: Generating chat from context {adapter.get_messages_for_logging(context)}"
         )
 
         messages = params["messages"]

src/pipecat/services/heygen/api_interactive_avatar.py

@@ -225,7 +225,7 @@ class HeyGenApi(BaseAvatarApi):
             "activity_idle_timeout": request_data.activity_idle_timeout,
         }
         session_info = await self._request("/streaming.new", params)
-        print("heygen session info", session_info)
+        logger.debug(f"HeyGen session info: {session_info}")
 
         heygen_session = HeyGenSession.model_validate(session_info)
 

src/pipecat/services/heygen/api_liveavatar.py

@@ -276,6 +276,13 @@ class LiveAvatarApi(BaseAvatarApi):
             # Fall back to VP8 encoding if video_settings is not provided
             params["video_settings"] = {"encoding": VideoEncoding.VP8.value}
 
+        if request_data.livekit_config is not None:
+            params["livekit_config"] = {
+                "livekit_url": request_data.livekit_config.livekit_url,
+                "livekit_room": request_data.livekit_config.livekit_room,
+                "livekit_client_token": request_data.livekit_config.livekit_client_token,
+            }
+
         logger.debug(f"Creating LiveAvatar session token with params: {params}")
         response = await self._request("POST", "/sessions/token", params)
         logger.debug(f"LiveAvatar session token created")

src/pipecat/services/heygen/client.py

@@ -156,6 +156,7 @@ class HeyGenClient:
         self._in_sample_rate = 0
         self._out_sample_rate = 0
         self._connected = False
+        self._keep_alive_task = None
         self._session_request = session_request
         self._callbacks = callbacks
         self._event_queue: asyncio.Queue | None = None
@@ -170,7 +171,6 @@ class HeyGenClient:
         # would be sending it to quickly. Instead, we want to block to emulate
         # an audio device, this is what the send interval is. It will be
         # computed on StartFrame.
-        self._send_interval = 0
         self._next_send_time = 0
         self._audio_seconds_sent = 0.0
         self._transport_ready = False
@@ -232,7 +232,7 @@ class HeyGenClient:
         except Exception as e:
             logger.error(f"Exception during cleanup: {e}")
 
-    async def start(self, frame: StartFrame, audio_chunk_size: int) -> None:
+    async def start(self, frame: StartFrame) -> None:
         """Start the client and establish all necessary connections.
 
         Initializes WebSocket and LiveKit connections using the provided configuration.
@@ -240,7 +240,6 @@ class HeyGenClient:
 
         Args:
             frame: Initial configuration frame containing audio parameters
-            audio_chunk_size: Audio chunk size for output processing
         """
         if self._websocket:
             logger.debug("heygen client already started")
@@ -249,10 +248,11 @@ class HeyGenClient:
         logger.debug(f"HeyGenClient starting")
         self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
         self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
-        self._send_interval = (audio_chunk_size / self._out_sample_rate) / 2
-        logger.debug(f"HeyGenClient send_interval: {self._send_interval}")
         await self._ws_connect()
         await self._livekit_connect()
+        self._keep_alive_task = self._task_manager.create_task(
+            self._keep_alive_handler(), name="HeyGenClient_KeepAlive"
+        )
         self._call_event_callback(self._callbacks.on_connected)
 
     async def stop(self) -> None:
@@ -261,6 +261,9 @@ class HeyGenClient:
         Disconnects from WebSocket and LiveKit endpoints, and performs cleanup.
         """
         logger.debug(f"HeyGenVideoService stopping")
+        if self._keep_alive_task:
+            await self._task_manager.cancel_task(self._keep_alive_task)
+            self._keep_alive_task = None
         await self._ws_disconnect()
         await self._livekit_disconnect()
         await self.cleanup()
@@ -286,21 +289,29 @@ class HeyGenClient:
 
     async def _ws_receive_task_handler(self):
         """Handle incoming WebSocket messages."""
-        while self._connected:
-            try:
-                message = await self._websocket.recv()
-                parsed_message = json.loads(message)
-                await self._handle_ws_server_event(parsed_message)
-            except ConnectionClosedOK:
-                break
-            except Exception as e:
-                logger.error(f"Error processing WebSocket message: {e}")
-                break
+        try:
+            while self._connected:
+                try:
+                    message = await self._websocket.recv()
+                    parsed_message = json.loads(message)
+                    await self._handle_ws_server_event(parsed_message)
+                except ConnectionClosedOK:
+                    logger.debug("HeyGenClient: WebSocket closed normally")
+                    break
+                except Exception as e:
+                    logger.error(f"Error processing WebSocket message: {e}")
+                    break
+        finally:
+            self._connected = False
+            logger.debug("HeyGenClient: WS receive handler exited, state cleaned up")
 
     async def _handle_ws_server_event(self, event: dict) -> None:
         """Handle an event from HeyGen websocket."""
         event_type = event.get("type")
-        if event_type == "agent.state":
+        if event_type == "session.state_updated":
+            state = event.get("state")
+            logger.debug(f"HeyGenClient ws session state updated: {state}")
+        elif event_type == "agent.state":
             logger.debug(f"HeyGenClient ws received agent status: {event}")
         else:
             logger.trace(f"HeyGenClient ws received unknown event: {event_type}")
@@ -328,6 +339,22 @@ class HeyGenClient:
             logger.error(f"Error sending message to HeyGen websocket: {e}")
             raise e
 
+    async def _keep_alive_handler(self):
+        """Periodically send keep-alive to prevent session timeout (5 min inactivity limit)."""
+        while self._connected:
+            await asyncio.sleep(150)  # 2.5 minutes
+            if self._connected:
+                try:
+                    await self._ws_send(
+                        {
+                            "type": "session.keep_alive",
+                            "event_id": str(uuid.uuid4()),
+                        }
+                    )
+                    logger.debug("HeyGenClient: Sent keep-alive")
+                except Exception as e:
+                    logger.warning(f"HeyGenClient: Keep-alive failed: {e}")
+
     async def interrupt(self, event_id: str) -> None:
         """Interrupt the avatar's current action.
 
@@ -394,7 +421,7 @@ class HeyGenClient:
         """Send audio data to the agent speak.
 
         Args:
-            audio: Audio data in base64 encoded format
+            audio: Audio data as raw bytes (will be base64 encoded)
             event_id: Unique identifier for the event
         """
         audio_base64 = base64.b64encode(audio).decode("utf-8")
@@ -406,30 +433,36 @@ class HeyGenClient:
             }
         )
         # Simulate audio playback with a sleep.
-        await self._write_audio_sleep()
+        await self._write_audio_sleep(len(audio))
 
     def _reset_audio_timing(self):
         """Reset audio timing control variables."""
         self._audio_seconds_sent = 0.0
         self._next_send_time = 0
 
-    async def _write_audio_sleep(self):
-        """Simulate audio playback timing with appropriate delays."""
-        # Only sleep after we've sent the first second of audio
-        # This appears to reduce the latency to receive the answer from HeyGen
+    async def _write_audio_sleep(self, audio_bytes: int):
+        """Simulate audio playback timing with appropriate delays.
+
+        Args:
+            audio_bytes: Number of raw audio bytes sent (24kHz, 16-bit mono).
+        """
+        # Compute actual audio duration from bytes (24kHz, 16-bit mono = 2 bytes/sample)
+        chunk_duration = audio_bytes / (HEY_GEN_SAMPLE_RATE * 2)
+
+        # Skip sleeping for the first 3 seconds of audio to reduce initial latency
         if self._audio_seconds_sent < 3.0:
-            self._audio_seconds_sent += self._send_interval
-            self._next_send_time = time.monotonic() + self._send_interval
+            self._audio_seconds_sent += chunk_duration
+            self._next_send_time = time.monotonic() + chunk_duration
             return
 
-        # After first second, use normal timing
+        # After first 3 seconds, pace sends to match real-time playback
         current_time = time.monotonic()
         sleep_duration = max(0, self._next_send_time - current_time)
         if sleep_duration > 0:
             await asyncio.sleep(sleep_duration)
-            self._next_send_time += self._send_interval
+            self._next_send_time += chunk_duration
         else:
-            self._next_send_time = time.monotonic() + self._send_interval
+            self._next_send_time = time.monotonic() + chunk_duration
 
     async def agent_speak_end(self, event_id: str) -> None:
         """Send signaling that the agent has finished speaking.

src/pipecat/services/heygen/video.py

@@ -211,9 +211,11 @@ class HeyGenVideoService(AIService):
             frame: The start frame containing initialization parameters.
         """
         await super().start(frame)
-        # 40 ms of audio, match the default behavior from the output transport
-        self._audio_chunk_size = int((HEY_GEN_SAMPLE_RATE * 2) / 25)
-        await self._client.start(frame, self._audio_chunk_size)
+        # First chunk: 400ms for faster initial response
+        self._first_chunk_size = int(HEY_GEN_SAMPLE_RATE * 2 * 0.4)  # 19200 bytes
+        # Subsequent chunks: 1000ms for efficient streaming
+        self._chunk_size = int(HEY_GEN_SAMPLE_RATE * 2 * 1.0)  # 48000 bytes
+        await self._client.start(frame)
         await self._create_send_task()
 
     async def stop(self, frame: EndFrame):
@@ -339,12 +341,14 @@ class HeyGenVideoService(AIService):
         """Handle sending audio frames to the HeyGen client.
 
         Continuously processes audio frames from the queue and sends them to the
-        HeyGen client. Handles timeouts and silence detection for proper audio
-        streaming management.
+        HeyGen client. Uses 600ms for the first chunk of each utterance for faster
+        initial response, then 1000ms chunks for efficient streaming. Handles
+        timeouts and silence detection for proper audio streaming management.
         """
         sample_rate = self._client.out_sample_rate
         audio_buffer = bytearray()
         self._event_id = None
+        is_first_chunk = True
 
         while True:
             try:
@@ -355,20 +359,30 @@ class HeyGenVideoService(AIService):
                     # starting the new inference
                     if self._event_id is None:
                         self._event_id = str(frame.id)
+                        is_first_chunk = True
 
                     audio = await self._resampler.resample(
                         frame.audio, frame.sample_rate, sample_rate
                     )
                     audio_buffer.extend(audio)
-                    while len(audio_buffer) >= self._audio_chunk_size:
-                        chunk = audio_buffer[: self._audio_chunk_size]
-                        audio_buffer = audio_buffer[self._audio_chunk_size :]
 
+                    current_chunk_size = (
+                        self._first_chunk_size if is_first_chunk else self._chunk_size
+                    )
+                    while len(audio_buffer) >= current_chunk_size:
+                        chunk = audio_buffer[:current_chunk_size]
+                        audio_buffer = audio_buffer[current_chunk_size:]
                         await self._client.agent_speak(bytes(chunk), self._event_id)
+                        if is_first_chunk:
+                            is_first_chunk = False
+                            current_chunk_size = self._chunk_size
                 self._queue.task_done()
             except TimeoutError:
                 # Bot has stopped speaking
                 if self._event_id is not None:
+                    # Flush any remaining buffered audio
+                    if len(audio_buffer) > 0:
+                        await self._client.agent_speak(bytes(audio_buffer), self._event_id)
+                        audio_buffer.clear()
                     await self._client.agent_speak_end(self._event_id)
                     self._event_id = None
-                    audio_buffer.clear()

src/pipecat/services/llm_service.py

@@ -347,6 +347,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
             parts.append(ASYNC_TOOL_CANCELLATION_INSTRUCTIONS)
         composed = "\n\n".join(p for p in parts if p)
         self._settings.system_instruction = composed or None
+        logger.debug(f"{self}: System instruction composed: {self._settings.system_instruction}")
 
     async def _update_settings(self, delta: LLMSettings) -> dict[str, Any]:
         """Apply a settings delta, handling turn-completion fields.

src/pipecat/transports/heygen/transport.py

@@ -21,6 +21,7 @@ from typing import Any
 import aiohttp
 from loguru import logger
 
+from pipecat.audio.utils import create_stream_resampler
 from pipecat.frames.frames import (
     AudioRawFrame,
     BotConnectedFrame,
@@ -40,7 +41,12 @@ from pipecat.frames.frames import (
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 from pipecat.services.heygen.api_interactive_avatar import NewSessionRequest
 from pipecat.services.heygen.api_liveavatar import LiveAvatarNewSessionRequest
-from pipecat.services.heygen.client import HeyGenCallbacks, HeyGenClient, ServiceType
+from pipecat.services.heygen.client import (
+    HEY_GEN_SAMPLE_RATE,
+    HeyGenCallbacks,
+    HeyGenClient,
+    ServiceType,
+)
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport, TransportParams
@@ -164,6 +170,7 @@ class HeyGenOutputTransport(BaseOutputTransport):
         super().__init__(params, **kwargs)
         self._client = client
         self._params = params
+        self._resampler = create_stream_resampler()
 
         # Whether we have seen a StartFrame already.
         self._initialized = False
@@ -195,7 +202,7 @@ class HeyGenOutputTransport(BaseOutputTransport):
             return
 
         self._initialized = True
-        await self._client.start(frame, self.audio_chunk_size)
+        await self._client.start(frame)
         await self.set_transport_ready(frame)
         self._client.transport_ready()
 
@@ -266,10 +273,15 @@ class HeyGenOutputTransport(BaseOutputTransport):
     async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the HeyGen transport.
 
+        Resamples audio to 24kHz if needed before sending.
+
         Args:
             frame: The audio frame to write.
         """
-        await self._client.agent_speak(bytes(frame.audio), self._event_id)
+        audio = frame.audio
+        if frame.sample_rate != HEY_GEN_SAMPLE_RATE:
+            audio = await self._resampler.resample(audio, frame.sample_rate, HEY_GEN_SAMPLE_RATE)
+        await self._client.agent_speak(bytes(audio), self._event_id)
         return True
 
 

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/__init__.py

@@ -8,11 +8,13 @@ from .base_user_turn_stop_strategy import BaseUserTurnStopStrategy, UserTurnStop
 from .external_user_turn_stop_strategy import ExternalUserTurnStopStrategy
 from .speech_timeout_user_turn_stop_strategy import SpeechTimeoutUserTurnStopStrategy
 from .turn_analyzer_user_turn_stop_strategy import TurnAnalyzerUserTurnStopStrategy
+from .vad_turn_analyzer_user_turn_stop_strategy import VADTurnAnalyzerUserTurnStopStrategy
 
 __all__ = [
     "BaseUserTurnStopStrategy",
     "ExternalUserTurnStopStrategy",
     "SpeechTimeoutUserTurnStopStrategy",
-    "UserTurnStoppedParams",
     "TurnAnalyzerUserTurnStopStrategy",
+    "UserTurnStoppedParams",
+    "VADTurnAnalyzerUserTurnStopStrategy",
 ]

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/vad_turn_analyzer_user_turn_stop_strategy.py

@@ -0,0 +1,130 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User turn stop strategy based on VAD and turn detection analyzers.
+
+This strategy uses a turn analyzer to detect end-of-turn but does not
+require STT transcriptions. It triggers immediately when the turn analyzer
+indicates COMPLETE, making it suitable for speech-to-speech pipelines
+where transcriptions arrive too late to be useful for turn decisions.
+"""
+
+from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, EndOfTurnState
+from pipecat.frames.frames import (
+    Frame,
+    InputAudioRawFrame,
+    MetricsFrame,
+    SpeechControlParamsFrame,
+    StartFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
+)
+from pipecat.metrics.metrics import MetricsData
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
+from pipecat.utils.asyncio.task_manager import BaseTaskManager
+
+
+class VADTurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
+    """User turn stop strategy that uses a turn analyzer without waiting for transcriptions.
+
+    This strategy feeds audio and VAD frames to a turn detection model
+    (``BaseTurnAnalyzer``) and triggers immediately when the model indicates
+    the turn is complete. Unlike ``TurnAnalyzerUserTurnStopStrategy``, it does
+    not wait for STT transcriptions, making it ideal for speech-to-speech
+    pipelines (e.g. Gemini Live) where audio goes directly to the LLM.
+
+    The ``UserTurnController`` provides a safety-net timeout
+    (``user_turn_stop_timeout``, default 5s) if the turn analyzer never
+    returns COMPLETE.
+    """
+
+    def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, **kwargs):
+        """Initialize the user turn stop strategy.
+
+        Args:
+            turn_analyzer: The turn detection analyzer instance to detect end of user turn.
+            **kwargs: Additional keyword arguments.
+        """
+        super().__init__(**kwargs)
+        self._turn_analyzer = turn_analyzer
+        self._vad_user_speaking = False
+
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        await super().reset()
+        self._vad_user_speaking = False
+
+    async def setup(self, task_manager: BaseTaskManager):
+        """Initialize the strategy with the given task manager.
+
+        Args:
+            task_manager: The task manager to be associated with this instance.
+        """
+        await super().setup(task_manager)
+
+    async def cleanup(self):
+        """Cleanup the strategy."""
+        await super().cleanup()
+        await self._turn_analyzer.cleanup()
+
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
+        """Process an incoming frame to update the turn analyzer and strategy state.
+
+        Args:
+            frame: The frame to be analyzed.
+
+        Returns:
+            Always returns CONTINUE so subsequent stop strategies are evaluated.
+        """
+        await super().process_frame(frame)
+
+        if isinstance(frame, StartFrame):
+            await self._start(frame)
+        elif isinstance(frame, VADUserStartedSpeakingFrame):
+            await self._handle_vad_user_started_speaking(frame)
+        elif isinstance(frame, VADUserStoppedSpeakingFrame):
+            await self._handle_vad_user_stopped_speaking(frame)
+        elif isinstance(frame, InputAudioRawFrame):
+            await self._handle_input_audio(frame)
+
+        return ProcessFrameResult.CONTINUE
+
+    async def _start(self, frame: StartFrame):
+        """Process the start frame to configure the turn analyzer."""
+        self._turn_analyzer.set_sample_rate(frame.audio_in_sample_rate)
+        await self.broadcast_frame(SpeechControlParamsFrame, turn_params=self._turn_analyzer.params)
+
+    async def _handle_input_audio(self, frame: InputAudioRawFrame):
+        """Handle input audio to check if the turn is completed."""
+        state = self._turn_analyzer.append_audio(frame.audio, self._vad_user_speaking)
+
+        # Streaming analyzers (e.g. KrispVivaTurn) detect turn completion
+        # frame-by-frame inside append_audio, so COMPLETE is returned here.
+        if state == EndOfTurnState.COMPLETE:
+            _, prediction = await self._turn_analyzer.analyze_end_of_turn()
+            await self._handle_prediction_result(prediction)
+            await self.trigger_user_turn_stopped()
+
+    async def _handle_vad_user_started_speaking(self, frame: VADUserStartedSpeakingFrame):
+        """Handle when the VAD indicates the user is speaking."""
+        self._turn_analyzer.update_vad_start_secs(frame.start_secs)
+        self._vad_user_speaking = True
+
+    async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame):
+        """Handle when the VAD indicates the user has stopped speaking."""
+        self._vad_user_speaking = False
+
+        state, prediction = await self._turn_analyzer.analyze_end_of_turn()
+        await self._handle_prediction_result(prediction)
+
+        if state == EndOfTurnState.COMPLETE:
+            await self.trigger_user_turn_stopped()
+
+    async def _handle_prediction_result(self, result: MetricsData | None):
+        """Handle a prediction result event from the turn analyzer."""
+        if result:
+            await self.push_frame(MetricsFrame(data=[result]))

src/pipecat/turns/user_stop/vad_user_turn_stop_strategy.py

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User turn stop strategy based on VAD events only.
+
+This strategy triggers the end of a user turn immediately when VAD indicates
+the user has stopped speaking. It does not use a turn analyzer or
+transcriptions, making it suitable for realtime speech-to-speech pipelines
+that rely solely on VAD for turn detection.
+"""
+
+from pipecat.frames.frames import Frame, VADUserStoppedSpeakingFrame
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
+
+
+class VADUserTurnStopStrategy(BaseUserTurnStopStrategy):
+    """User turn stop strategy based on VAD (Voice Activity Detection).
+
+    This strategy triggers the end of a user turn as soon as a VAD frame
+    indicates the user has stopped speaking. It is intended for realtime
+    speech-to-speech pipelines where neither a turn analyzer nor STT
+    transcriptions are used to decide end of turn.
+    """
+
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
+        """Process an incoming frame to detect user turn stop.
+
+        Args:
+            frame: The frame to be analyzed.
+
+        Returns:
+            Always returns CONTINUE so subsequent stop strategies are evaluated.
+        """
+        await super().process_frame(frame)
+
+        if isinstance(frame, VADUserStoppedSpeakingFrame):
+            await self.trigger_user_turn_stopped()
+
+        return ProcessFrameResult.CONTINUE

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())
 

tests/test_krisp_ip_user_turn_start_strategy.py

@@ -0,0 +1,236 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+
+# Mock package version check before importing pipecat (development mode)
+_version_patcher = patch("importlib.metadata.version", return_value="0.0.0-dev")
+_version_patcher.start()
+
+# Mock krisp_audio before any pipecat import that loads krisp_instance / VIVA IP strategy
+mock_krisp_audio = MagicMock()
+mock_krisp_audio.SamplingRate.Sr8000Hz = 8000
+mock_krisp_audio.SamplingRate.Sr16000Hz = 16000
+mock_krisp_audio.SamplingRate.Sr24000Hz = 24000
+mock_krisp_audio.SamplingRate.Sr32000Hz = 32000
+mock_krisp_audio.SamplingRate.Sr44100Hz = 44100
+mock_krisp_audio.SamplingRate.Sr48000Hz = 48000
+mock_krisp_audio.FrameDuration.Fd10ms = "10ms"
+mock_krisp_audio.FrameDuration.Fd15ms = "15ms"
+mock_krisp_audio.FrameDuration.Fd20ms = "20ms"
+mock_krisp_audio.FrameDuration.Fd30ms = "30ms"
+mock_krisp_audio.FrameDuration.Fd32ms = "32ms"
+
+sys.modules["krisp_audio"] = mock_krisp_audio
+
+mock_pipecat_krisp = MagicMock()
+sys.modules["pipecat_ai_krisp"] = mock_pipecat_krisp
+sys.modules["pipecat_ai_krisp.audio"] = MagicMock()
+sys.modules["pipecat_ai_krisp.audio.krisp_processor"] = MagicMock()
+
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    InputAudioRawFrame,
+    TranscriptionFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
+)
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_start.krisp_viva_ip_user_turn_start_strategy import (
+    KrispVivaIPUserTurnStartStrategy,
+)
+
+STRATEGY_MODULE = "pipecat.turns.user_start.krisp_viva_ip_user_turn_start_strategy"
+
+
+def _int16_silence(num_samples: int) -> bytes:
+    return np.zeros(num_samples, dtype=np.int16).tobytes()
+
+
+class TestKrispVivaIPUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
+    """Tests for KrispVivaIPUserTurnStartStrategy with mocked krisp_audio."""
+
+    def setUp(self):
+        self.temp_model_file = tempfile.NamedTemporaryFile(suffix=".kef", delete=False)
+        self.temp_model_file.write(b"dummy")
+        self.temp_model_file.close()
+        self.model_path = self.temp_model_file.name
+
+        self.mock_krisp_audio = mock_krisp_audio
+        self.mock_krisp_audio.reset_mock()
+        self.mock_krisp_audio.ModelInfo.reset_mock()
+        self.mock_krisp_audio.IpSessionConfig.reset_mock()
+        self.mock_krisp_audio.IpFloat.reset_mock()
+
+        self.mock_model_info = MagicMock()
+        self.mock_krisp_audio.ModelInfo.return_value = self.mock_model_info
+
+        self.mock_ip_cfg = MagicMock()
+        self.mock_krisp_audio.IpSessionConfig.return_value = self.mock_ip_cfg
+
+        self.mock_ip_session = MagicMock()
+        self.mock_krisp_audio.IpFloat.create.return_value = self.mock_ip_session
+
+        self.krisp_patch = patch(f"{STRATEGY_MODULE}.krisp_audio", self.mock_krisp_audio)
+        self.krisp_patch.start()
+
+        self.sdk_patcher = patch(f"{STRATEGY_MODULE}.KrispVivaSDKManager")
+        self.mock_sdk_manager = self.sdk_patcher.start()
+        self.mock_sdk_manager.acquire = MagicMock()
+        self.mock_sdk_manager.release = MagicMock()
+
+    def tearDown(self):
+        self.krisp_patch.stop()
+        self.sdk_patcher.stop()
+        if os.path.exists(self.model_path):
+            os.unlink(self.model_path)
+
+    def _make_strategy(self, *, threshold: float = 0.5, frame_duration_ms: int = 20):
+        return KrispVivaIPUserTurnStartStrategy(
+            model_path=self.model_path,
+            threshold=threshold,
+            frame_duration_ms=frame_duration_ms,
+            api_key="test-key",
+        )
+
+    def _audio_frame(self, sample_rate: int = 16000, frame_duration_ms: int = 20):
+        samples = int(sample_rate * frame_duration_ms / 1000)
+        return InputAudioRawFrame(
+            audio=_int16_silence(samples),
+            sample_rate=sample_rate,
+            num_channels=1,
+        )
+
+    async def test_interruption_detected_emits_turn_and_stop(self):
+        self.mock_ip_session.process = MagicMock(return_value=0.87)
+
+        strategy = self._make_strategy(threshold=0.5)
+        try:
+            fired = False
+
+            @strategy.event_handler("on_user_turn_started")
+            async def on_user_turn_started(strategy, params):
+                nonlocal fired
+                fired = True
+
+            await strategy.process_frame(VADUserStartedSpeakingFrame())
+            result = await strategy.process_frame(self._audio_frame())
+
+            self.assertTrue(fired)
+            self.assertEqual(result, ProcessFrameResult.STOP)
+            self.mock_ip_session.process.assert_called()
+        finally:
+            await strategy.cleanup()
+
+    async def test_backchannel_suppressed_no_event_continue(self):
+        self.mock_ip_session.process = MagicMock(return_value=0.23)
+
+        strategy = self._make_strategy(threshold=0.5)
+        try:
+            fired = False
+
+            @strategy.event_handler("on_user_turn_started")
+            async def on_user_turn_started(strategy, params):
+                nonlocal fired
+                fired = True
+
+            await strategy.process_frame(VADUserStartedSpeakingFrame())
+            result = await strategy.process_frame(self._audio_frame())
+
+            self.assertFalse(fired)
+            self.assertEqual(result, ProcessFrameResult.CONTINUE)
+        finally:
+            await strategy.cleanup()
+
+    async def test_reset_on_vad_stopped_clears_state(self):
+        self.mock_ip_session.process = MagicMock(return_value=0.1)
+
+        strategy = self._make_strategy(threshold=0.5)
+        try:
+            await strategy.process_frame(VADUserStartedSpeakingFrame())
+            await strategy.process_frame(self._audio_frame())
+            self.mock_ip_session.process.reset_mock()
+
+            await strategy.process_frame(VADUserStoppedSpeakingFrame())
+            result = await strategy.process_frame(self._audio_frame())
+
+            self.assertEqual(result, ProcessFrameResult.CONTINUE)
+            self.mock_ip_session.process.assert_not_called()
+        finally:
+            await strategy.cleanup()
+
+    async def test_reset_on_bot_stopped_clears_state(self):
+        self.mock_ip_session.process = MagicMock(return_value=0.1)
+
+        strategy = self._make_strategy(threshold=0.5)
+        try:
+            await strategy.process_frame(VADUserStartedSpeakingFrame())
+            await strategy.process_frame(self._audio_frame())
+            self.mock_ip_session.process.reset_mock()
+
+            await strategy.process_frame(BotStoppedSpeakingFrame())
+            result = await strategy.process_frame(self._audio_frame())
+
+            self.assertEqual(result, ProcessFrameResult.CONTINUE)
+            self.mock_ip_session.process.assert_not_called()
+        finally:
+            await strategy.cleanup()
+
+    async def test_no_op_before_vad_start(self):
+        self.mock_ip_session.process = MagicMock(return_value=0.99)
+
+        strategy = self._make_strategy()
+        try:
+            result = await strategy.process_frame(self._audio_frame())
+            self.assertEqual(result, ProcessFrameResult.CONTINUE)
+            self.mock_ip_session.process.assert_not_called()
+        finally:
+            await strategy.cleanup()
+
+    async def test_decision_sticks_no_double_trigger(self):
+        self.mock_ip_session.process = MagicMock(return_value=0.9)
+
+        strategy = self._make_strategy(threshold=0.5)
+        try:
+            count = 0
+
+            @strategy.event_handler("on_user_turn_started")
+            async def on_user_turn_started(strategy, params):
+                nonlocal count
+                count += 1
+
+            await strategy.process_frame(VADUserStartedSpeakingFrame())
+            r1 = await strategy.process_frame(self._audio_frame())
+            r2 = await strategy.process_frame(self._audio_frame())
+
+            self.assertEqual(r1, ProcessFrameResult.STOP)
+            self.assertEqual(r2, ProcessFrameResult.CONTINUE)
+            self.assertEqual(count, 1)
+        finally:
+            await strategy.cleanup()
+
+    async def test_unrelated_frames_continue(self):
+        strategy = self._make_strategy()
+        try:
+            r1 = await strategy.process_frame(BotStartedSpeakingFrame())
+            r2 = await strategy.process_frame(
+                TranscriptionFrame(text="hi", user_id="", timestamp="")
+            )
+            self.assertEqual(r1, ProcessFrameResult.CONTINUE)
+            self.assertEqual(r2, ProcessFrameResult.CONTINUE)
+        finally:
+            await strategy.cleanup()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_krisp_sdk_manager.py

@@ -62,8 +62,15 @@ class TestKrispVivaSDKManager:
     def setup_method(self):
         """Reset mocks and SDK state before each test."""
         mock_krisp_audio.reset_mock()
+        mock_krisp_audio.globalInit.side_effect = None
         mock_krisp_audio.getVersion.return_value = mock_version
 
+        # Ensure krisp_instance module uses THIS test's mock, not a stale
+        # reference cached from a different test file's sys.modules entry.
+        import pipecat.audio.krisp_instance as _ki
+
+        _ki.krisp_audio = mock_krisp_audio
+
         # Reset the SDK manager state for clean tests
         # We access internal state to ensure tests are isolated
         with KrispVivaSDKManager._lock:

tests/test_vad_turn_analyzer_user_turn_stop_strategy.py

@@ -0,0 +1,213 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for VADTurnAnalyzerUserTurnStopStrategy."""
+
+import unittest
+from unittest.mock import AsyncMock
+
+import pytest
+
+from pipecat.audio.turn.base_turn_analyzer import (
+    BaseTurnAnalyzer,
+    BaseTurnParams,
+    EndOfTurnState,
+)
+from pipecat.frames.frames import (
+    InputAudioRawFrame,
+    MetricsFrame,
+    SpeechControlParamsFrame,
+    StartFrame,
+    TranscriptionFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
+)
+from pipecat.metrics.metrics import MetricsData
+from pipecat.turns.user_stop.vad_turn_analyzer_user_turn_stop_strategy import (
+    VADTurnAnalyzerUserTurnStopStrategy,
+)
+from pipecat.utils.asyncio.task_manager import TaskManager
+
+
+class MockTurnAnalyzer(BaseTurnAnalyzer):
+    """Mock turn analyzer for testing."""
+
+    def __init__(self):
+        super().__init__()
+        self._append_audio_state = EndOfTurnState.INCOMPLETE
+        self._analyze_state = EndOfTurnState.INCOMPLETE
+        self._prediction: MetricsData | None = None
+        self._speech_triggered = False
+
+    @property
+    def speech_triggered(self) -> bool:
+        return self._speech_triggered
+
+    @property
+    def params(self) -> BaseTurnParams:
+        return BaseTurnParams()
+
+    def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
+        return self._append_audio_state
+
+    async def analyze_end_of_turn(self) -> tuple[EndOfTurnState, MetricsData | None]:
+        return self._analyze_state, self._prediction
+
+    def update_vad_start_secs(self, vad_start_secs: float):
+        pass
+
+    def clear(self):
+        pass
+
+
+class TestVADTurnAnalyzerUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.analyzer = MockTurnAnalyzer()
+        self.strategy = VADTurnAnalyzerUserTurnStopStrategy(turn_analyzer=self.analyzer)
+        self.task_manager = TaskManager()
+        await self.strategy.setup(self.task_manager)
+
+        self.turn_stopped_called = False
+
+        @self.strategy.event_handler("on_user_turn_stopped")
+        async def on_user_turn_stopped(strategy, params):
+            self.turn_stopped_called = True
+
+        self.pushed_frames = []
+
+        @self.strategy.event_handler("on_push_frame")
+        async def on_push_frame(strategy, frame, direction):
+            self.pushed_frames.append(frame)
+
+        self.broadcast_frames = []
+
+        @self.strategy.event_handler("on_broadcast_frame")
+        async def on_broadcast_frame(strategy, frame_cls, **kwargs):
+            self.broadcast_frames.append((frame_cls, kwargs))
+
+    async def asyncTearDown(self):
+        await self.strategy.cleanup()
+
+    async def test_vad_stop_complete_triggers_immediately(self):
+        """VAD stop with COMPLETE should trigger user turn stopped immediately."""
+        self.analyzer._analyze_state = EndOfTurnState.COMPLETE
+
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3))
+
+        assert self.turn_stopped_called
+
+    async def test_vad_stop_incomplete_does_not_trigger(self):
+        """VAD stop with INCOMPLETE should not trigger user turn stopped."""
+        self.analyzer._analyze_state = EndOfTurnState.INCOMPLETE
+
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3))
+
+        assert not self.turn_stopped_called
+
+    async def test_streaming_complete_via_append_audio_triggers(self):
+        """Streaming COMPLETE from append_audio should trigger immediately."""
+        self.analyzer._append_audio_state = EndOfTurnState.COMPLETE
+        # analyze_end_of_turn is called after append_audio returns COMPLETE
+        self.analyzer._analyze_state = EndOfTurnState.COMPLETE
+
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        await self.strategy.process_frame(
+            InputAudioRawFrame(audio=b"\x00" * 320, sample_rate=16000, num_channels=1)
+        )
+
+        assert self.turn_stopped_called
+
+    async def test_streaming_incomplete_does_not_trigger(self):
+        """Streaming INCOMPLETE from append_audio should not trigger."""
+        self.analyzer._append_audio_state = EndOfTurnState.INCOMPLETE
+
+        await self.strategy.process_frame(
+            InputAudioRawFrame(audio=b"\x00" * 320, sample_rate=16000, num_channels=1)
+        )
+
+        assert not self.turn_stopped_called
+
+    async def test_vad_start_resets_speaking_state(self):
+        """VAD start should set speaking state to True."""
+        assert not self.strategy._vad_user_speaking
+
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+
+        assert self.strategy._vad_user_speaking
+
+    async def test_vad_stop_resets_speaking_state(self):
+        """VAD stop should set speaking state to False."""
+        self.analyzer._analyze_state = EndOfTurnState.INCOMPLETE
+
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        assert self.strategy._vad_user_speaking
+
+        await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3))
+        assert not self.strategy._vad_user_speaking
+
+    async def test_reset_clears_state(self):
+        """Reset should clear speaking state."""
+        self.strategy._vad_user_speaking = True
+
+        await self.strategy.reset()
+
+        assert not self.strategy._vad_user_speaking
+
+    async def test_metrics_frame_pushed_on_prediction(self):
+        """MetricsFrame should be pushed when prediction result is available."""
+        prediction = MetricsData(processor="test_analyzer")
+        self.analyzer._analyze_state = EndOfTurnState.COMPLETE
+        self.analyzer._prediction = prediction
+
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3))
+
+        assert len(self.pushed_frames) == 1
+        assert isinstance(self.pushed_frames[0], MetricsFrame)
+        assert self.pushed_frames[0].data == [prediction]
+
+    async def test_no_metrics_frame_when_no_prediction(self):
+        """No MetricsFrame should be pushed when prediction is None."""
+        self.analyzer._analyze_state = EndOfTurnState.COMPLETE
+        self.analyzer._prediction = None
+
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3))
+
+        assert len(self.pushed_frames) == 0
+
+    async def test_speech_control_params_broadcast_on_start(self):
+        """SpeechControlParamsFrame should be broadcast on StartFrame."""
+        await self.strategy.process_frame(
+            StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000)
+        )
+
+        assert len(self.broadcast_frames) == 1
+        frame_cls, kwargs = self.broadcast_frames[0]
+        assert frame_cls is SpeechControlParamsFrame
+
+    async def test_transcription_frames_ignored(self):
+        """TranscriptionFrame should not affect state or trigger turn stop."""
+        self.analyzer._analyze_state = EndOfTurnState.INCOMPLETE
+
+        await self.strategy.process_frame(
+            TranscriptionFrame(text="hello", user_id="user1", timestamp="now")
+        )
+
+        assert not self.turn_stopped_called
+
+    async def test_process_frame_returns_continue(self):
+        """process_frame should always return CONTINUE."""
+        from pipecat.turns.types import ProcessFrameResult
+
+        result = await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        assert result == ProcessFrameResult.CONTINUE
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_vad_user_turn_stop_strategy.py

@@ -0,0 +1,79 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for VADUserTurnStopStrategy."""
+
+import unittest
+
+from pipecat.frames.frames import (
+    InputAudioRawFrame,
+    TranscriptionFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
+)
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_stop.vad_user_turn_stop_strategy import VADUserTurnStopStrategy
+from pipecat.utils.asyncio.task_manager import TaskManager
+
+
+class TestVADUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.strategy = VADUserTurnStopStrategy()
+        self.task_manager = TaskManager()
+        await self.strategy.setup(self.task_manager)
+
+        self.turn_stopped_called = False
+
+        @self.strategy.event_handler("on_user_turn_stopped")
+        async def on_user_turn_stopped(strategy, params):
+            self.turn_stopped_called = True
+
+    async def asyncTearDown(self):
+        await self.strategy.cleanup()
+
+    async def test_vad_stop_triggers_user_turn_stopped(self):
+        """VADUserStoppedSpeakingFrame should trigger user turn stopped."""
+        await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3))
+
+        assert self.turn_stopped_called
+
+    async def test_vad_start_does_not_trigger(self):
+        """VADUserStartedSpeakingFrame should not trigger user turn stopped."""
+        await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+
+        assert not self.turn_stopped_called
+
+    async def test_transcription_frames_ignored(self):
+        """TranscriptionFrame should not trigger user turn stopped."""
+        await self.strategy.process_frame(
+            TranscriptionFrame(text="hello", user_id="user1", timestamp="now")
+        )
+
+        assert not self.turn_stopped_called
+
+    async def test_input_audio_frames_ignored(self):
+        """InputAudioRawFrame should not trigger user turn stopped."""
+        await self.strategy.process_frame(
+            InputAudioRawFrame(audio=b"\x00" * 320, sample_rate=16000, num_channels=1)
+        )
+
+        assert not self.turn_stopped_called
+
+    async def test_process_frame_returns_continue(self):
+        """process_frame should always return CONTINUE."""
+        result = await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3))
+        assert result == ProcessFrameResult.CONTINUE
+
+        result = await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2))
+        assert result == ProcessFrameResult.CONTINUE
+
+    async def test_reset_does_not_crash(self):
+        """reset() should complete without error."""
+        await self.strategy.reset()
+
+
+if __name__ == "__main__":
+    unittest.main()

uv.lock

@@ -1,5 +1,5 @@
 version = 1
-revision = 3
+revision = 2
 requires-python = ">=3.11"
 resolution-markers = [
     "python_full_version >= '3.14'",