Merge branch 'main' into richtermb/push-more-error-frames

src/pipecat/observers/loggers/user_bot_latency_log_observer.py

@@ -7,11 +7,14 @@
 """Observer for measuring user-to-bot response latency."""
 
 import time
+from statistics import mean
 
 from loguru import logger
 
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
+    CancelFrame,
+    EndFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -35,6 +38,7 @@ class UserBotLatencyLogObserver(BaseObserver):
         super().__init__()
         self._processed_frames = set()
         self._user_stopped_time = 0
+        self._latencies = []
 
     async def on_push_frame(self, data: FramePushed):
         """Process frames to track speech timing and calculate latency.
@@ -56,6 +60,18 @@ class UserBotLatencyLogObserver(BaseObserver):
             self._user_stopped_time = 0
         elif isinstance(data.frame, UserStoppedSpeakingFrame):
             self._user_stopped_time = time.time()
+        elif isinstance(data.frame, (EndFrame, CancelFrame)):
+            if self._latencies:
+                avg_latency = mean(self._latencies)
+                min_latency = min(self._latencies)
+                max_latency = max(self._latencies)
+                logger.info(
+                    f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING - Avg: {avg_latency:.3f}s, Min: {min_latency:.3f}s, Max: {max_latency:.3f}s"
+                )
         elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time:
             latency = time.time() - self._user_stopped_time
-            logger.debug(f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency}")
+            self._user_stopped_time = 0
+            self._latencies.append(latency)
+            logger.debug(
+                f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency:.3f}s"
+            )

src/pipecat/pipeline/task_observer.py

@@ -153,22 +153,23 @@ class TaskObserver(BaseObserver):
 
     async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver):
         """Handle frame processing for a single observer."""
-        warning_reported = False
+        on_push_frame_deprecated = False
+        signature = inspect.signature(observer.on_push_frame)
+        if len(signature.parameters) > 1:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Observer `on_push_frame(source, destination, frame, direction, timestamp)` is deprecated, us `on_push_frame(data: FramePushed)` instead.",
+                    DeprecationWarning,
+                )
+
+            on_push_frame_deprecated = True
+
         while True:
             data = await queue.get()
-
-            signature = inspect.signature(observer.on_push_frame)
-            if len(signature.parameters) > 1:
-                if not warning_reported:
-                    import warnings
-
-                    with warnings.catch_warnings():
-                        warnings.simplefilter("always")
-                        warnings.warn(
-                            "Observer `on_push_frame(source, destination, frame, direction, timestamp)` is deprecated, us `on_push_frame(data: FramePushed)` instead.",
-                            DeprecationWarning,
-                        )
-                    warning_reported = True
+            if on_push_frame_deprecated:
                 await observer.on_push_frame(
                     data.src, data.dst, data.frame, data.direction, data.timestamp
                 )

src/pipecat/runner/daily.py

@@ -7,17 +7,14 @@
 """Daily room and token configuration utilities.
 
 This module provides helper functions for creating and configuring Daily rooms
-and authentication tokens. It handles both command-line argument parsing and
-environment variable configuration.
+and authentication tokens. It automatically creates temporary rooms for
+development or uses existing rooms specified via environment variables.
 
-The module supports creating temporary rooms for development or using existing
-rooms specified via arguments or environment variables.
+Environment variables:
 
-Required environment variables:
-
-- DAILY_API_KEY - Daily API key for room/token creation
-- DAILY_SAMPLE_ROOM_URL (optional) - Existing room URL to use
-- DAILY_SAMPLE_ROOM_TOKEN (optional) - Existing token to use
+- DAILY_API_KEY - Daily API key for room/token creation (required)
+- DAILY_SAMPLE_ROOM_URL (optional) - Existing room URL to use. If not provided,
+  a temporary room will be created automatically.
 
 Example::
 
@@ -29,17 +26,26 @@ Example::
         # Use room_url and token with DailyTransport
 """
 
-import argparse
 import os
-from typing import Optional
+import time
+import uuid
+from typing import Tuple
 
 import aiohttp
 
-from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+from pipecat.transports.services.helpers.daily_rest import (
+    DailyRESTHelper,
+    DailyRoomParams,
+    DailyRoomProperties,
+)
 
 
-async def configure(aiohttp_session: aiohttp.ClientSession):
-    """Configure Daily room URL and token from arguments or environment.
+async def configure(aiohttp_session: aiohttp.ClientSession) -> Tuple[str, str]:
+    """Configure Daily room URL and token from environment variables.
+
+    This function will either:
+    1. Use an existing room URL from DAILY_SAMPLE_ROOM_URL environment variable
+    2. Create a new temporary room automatically if no URL is provided
 
     Args:
         aiohttp_session: HTTP session for making API requests.
@@ -48,65 +54,79 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
         Tuple containing the room URL and authentication token.
 
     Raises:
-        Exception: If room URL or API key are not provided.
+        Exception: If DAILY_API_KEY is not provided in environment variables.
     """
-    (url, token, _) = await configure_with_args(aiohttp_session)
-    return (url, token)
-
-
-async def configure_with_args(
-    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
-):
-    """Configure Daily room with command-line argument parsing.
-
-    Args:
-        aiohttp_session: HTTP session for making API requests.
-        parser: Optional argument parser. If None, creates a default one.
-
-    Returns:
-        Tuple containing room URL, authentication token, and parsed arguments.
-
-    Raises:
-        Exception: If room URL or API key are not provided via arguments or environment.
-    """
-    if not parser:
-        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
-    parser.add_argument(
-        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
-    )
-    parser.add_argument(
-        "-k",
-        "--apikey",
-        type=str,
-        required=False,
-        help="Daily API Key (needed to create an owner token for the room)",
-    )
-
-    args, unknown = parser.parse_known_args()
-
-    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
-    key = args.apikey or os.getenv("DAILY_API_KEY")
-
-    if not url:
+    # Check for required API key
+    api_key = os.getenv("DAILY_API_KEY")
+    if not api_key:
         raise Exception(
-            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+            "DAILY_API_KEY environment variable is required. "
+            "Get your API key from https://dashboard.daily.co/developers"
         )
 
-    if not key:
-        raise Exception(
-            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
-        )
+    # Check for existing room URL
+    existing_room_url = os.getenv("DAILY_SAMPLE_ROOM_URL")
 
     daily_rest_helper = DailyRESTHelper(
-        daily_api_key=key,
+        daily_api_key=api_key,
         daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
         aiohttp_session=aiohttp_session,
     )
 
-    # Create a meeting token for the given room with an expiration 2 hours in
-    # the future.
+    if existing_room_url:
+        # Use existing room
+        print(f"Using existing Daily room: {existing_room_url}")
+        room_url = existing_room_url
+    else:
+        # Create a new temporary room
+        room_name = f"pipecat-{uuid.uuid4().hex[:8]}"
+        print(f"Creating new Daily room: {room_name}")
+
+        # Calculate expiration time: current time + 2 hours
+        expiration_time = time.time() + (2 * 60 * 60)  # 2 hours from now
+
+        # Create room properties with absolute timestamp
+        room_properties = DailyRoomProperties(
+            exp=expiration_time,  # Absolute Unix timestamp
+            eject_at_room_exp=True,
+        )
+
+        # Create room parameters
+        room_params = DailyRoomParams(name=room_name, properties=room_properties)
+
+        room_response = await daily_rest_helper.create_room(room_params)
+        room_url = room_response.url
+        print(f"Created Daily room: {room_url}")
+
+    # Create a meeting token for the room with an expiration 2 hours in the future
     expiry_time: float = 2 * 60 * 60
+    token = await daily_rest_helper.get_token(room_url, expiry_time)
 
-    token = await daily_rest_helper.get_token(url, expiry_time)
+    return (room_url, token)
 
-    return (url, token, args)
+
+# Keep this for backwards compatibility, but mark as deprecated
+async def configure_with_args(aiohttp_session: aiohttp.ClientSession, parser=None):
+    """Configure Daily room with command-line argument parsing.
+
+    .. deprecated:: 0.0.78
+        This function is deprecated. Use configure() instead which uses
+        environment variables only.
+
+    Args:
+        aiohttp_session: HTTP session for making API requests.
+        parser: Ignored. Kept for backwards compatibility.
+
+    Returns:
+        Tuple containing room URL, authentication token, and None (for args).
+    """
+    import warnings
+
+    warnings.warn(
+        "configure_with_args is deprecated. Use configure() instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+
+    room_url, token = await configure(aiohttp_session)
+    return (room_url, token, None)

src/pipecat/runner/run.py

@@ -82,7 +82,7 @@ from pipecat.runner.types import (
 try:
     import uvicorn
     from dotenv import load_dotenv
-    from fastapi import BackgroundTasks, FastAPI, WebSocket
+    from fastapi import BackgroundTasks, FastAPI, Request, WebSocket
     from fastapi.middleware.cors import CORSMiddleware
     from fastapi.responses import HTMLResponse, RedirectResponse
 except ImportError as e:
@@ -261,17 +261,43 @@ def _setup_daily_routes(app: FastAPI):
         async with aiohttp.ClientSession() as session:
             room_url, token = await configure(session)
 
-            # Start the bot in the background
+            # Start the bot in the background with empty body for GET requests
             bot_module = _get_bot_module()
             runner_args = DailyRunnerArguments(room_url=room_url, token=token, body={})
             asyncio.create_task(bot_module.bot(runner_args))
             return RedirectResponse(room_url)
 
-    @app.post("/connect")
-    async def rtvi_connect():
-        """Launch a Daily bot and return connection info for RTVI clients."""
+    async def _handle_rtvi_request(request: Request):
+        """Common handler for both /start and /connect endpoints.
+
+        Expects POST body like::
+
+            {
+                "createDailyRoom": true,
+                "dailyRoomProperties": { "start_video_off": true },
+                "body": { "custom_data": "value" }
+            }
+        """
         print("Starting bot with Daily transport")
 
+        # Parse the request body
+        try:
+            request_data = await request.json()
+            logger.debug(f"Received request: {request_data}")
+        except Exception as e:
+            logger.error(f"Failed to parse request body: {e}")
+            request_data = {}
+
+        # Extract the body data that should be passed to the bot
+        # This mimics Pipecat Cloud's behavior
+        bot_body = request_data.get("body", {})
+
+        # Log the extracted body data for debugging
+        if bot_body:
+            logger.info(f"Extracted body data for bot: {bot_body}")
+        else:
+            logger.debug("No body data provided in request")
+
         import aiohttp
 
         from pipecat.runner.daily import configure
@@ -279,11 +305,30 @@ def _setup_daily_routes(app: FastAPI):
         async with aiohttp.ClientSession() as session:
             room_url, token = await configure(session)
 
-            # Start the bot in the background
+            # Start the bot in the background with extracted body data
             bot_module = _get_bot_module()
-            runner_args = DailyRunnerArguments(room_url=room_url, token=token, body={})
+            runner_args = DailyRunnerArguments(room_url=room_url, token=token, body=bot_body)
             asyncio.create_task(bot_module.bot(runner_args))
-            return {"room_url": room_url, "token": token}
+            # Match PCC /start endpoint response format:
+            return {"dailyRoom": room_url, "dailyToken": token}
+
+    @app.post("/start")
+    async def rtvi_start(request: Request):
+        """Launch a Daily bot and return connection info for RTVI clients."""
+        return await _handle_rtvi_request(request)
+
+    @app.post("/connect")
+    async def rtvi_connect(request: Request):
+        """Launch a Daily bot and return connection info for RTVI clients.
+
+        .. deprecated:: 0.0.78
+            Use /start instead. This endpoint will be removed in a future version.
+        """
+        logger.warning(
+            "DEPRECATED: /connect endpoint is deprecated. Please use /start instead. "
+            "This endpoint will be removed in a future version."
+        )
+        return await _handle_rtvi_request(request)
 
 
 def _setup_telephony_routes(app: FastAPI, transport_type: str, proxy: str):
@@ -345,6 +390,7 @@ async def _run_daily_direct():
     async with aiohttp.ClientSession() as session:
         room_url, token = await configure(session)
 
+        # Direct connections have no request body, so use empty dict
         runner_args = DailyRunnerArguments(room_url=room_url, token=token, body={})
 
         # Get the bot module and run it directly
@@ -357,6 +403,27 @@ async def _run_daily_direct():
         await bot_module.bot(runner_args)
 
 
+def _validate_and_clean_proxy(proxy: str) -> str:
+    """Validate and clean proxy hostname, removing protocol if present."""
+    if not proxy:
+        return proxy
+
+    original_proxy = proxy
+
+    # Strip common protocols
+    if proxy.startswith(("http://", "https://")):
+        proxy = proxy.split("://", 1)[1]
+        logger.warning(
+            f"Removed protocol from proxy URL. Using '{proxy}' instead of '{original_proxy}'. "
+            f"The --proxy argument expects only the hostname (e.g., 'mybot.ngrok.io')."
+        )
+
+    # Remove trailing slashes
+    proxy = proxy.rstrip("/")
+
+    return proxy
+
+
 def main():
     """Start the Pipecat development runner.
 
@@ -408,6 +475,10 @@ def main():
 
     args = parser.parse_args()
 
+    # Validate and clean proxy hostname
+    if args.proxy:
+        args.proxy = _validate_and_clean_proxy(args.proxy)
+
     # Auto-set transport to daily if --direct is used without explicit transport
     if args.direct and args.transport == "webrtc":  # webrtc is the default
         args.transport = "daily"
@@ -438,17 +509,16 @@ def main():
     if args.transport == "webrtc":
         print()
         if args.esp32:
-            print(
-                f"🚀 WebRTC server starting at http://{args.host}:{args.port}/client (ESP32 mode)"
-            )
+            print(f"🚀 Bot ready! (ESP32 mode)")
+            print(f"   → Open http://{args.host}:{args.port}/client in your browser")
         else:
-            print(f"🚀 WebRTC server starting at http://{args.host}:{args.port}/client")
-        print(f"   Open this URL in your browser to connect!")
+            print(f"🚀 Bot ready!")
+            print(f"   → Open http://{args.host}:{args.port}/client in your browser")
         print()
     elif args.transport == "daily":
         print()
-        print(f"🚀 Daily server starting at http://{args.host}:{args.port}")
-        print(f"   Open this URL in your browser to start a session!")
+        print(f"🚀 Bot ready!")
+        print(f"   → Open http://{args.host}:{args.port} in your browser to start a session")
         print()
 
     # Create the app with transport-specific setup

src/pipecat/services/cartesia/tts.py

@@ -34,6 +34,10 @@ from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
 from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
 from pipecat.utils.tracing.service_decorators import traced_tts
 
+# Suppress regex warnings from pydub (used by cartesia)
+warnings.filterwarnings("ignore", message="invalid escape sequence", category=SyntaxWarning)
+
+
 # See .env.example for Cartesia configuration needed
 try:
     from cartesia import AsyncCartesia

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -983,17 +983,14 @@ class GeminiMultimodalLiveLLMService(LLMService):
         with audio and video inputs, preventing temporal misalignment that can occur
         when different modalities are processed through separate API pathways.
 
-        After sending the text, we signal turn completion to trigger a model response
-        for text-only interactions.
+        For realtimeInput, turn completion is automatically inferred by the API based
+        on user activity, so no explicit turnComplete signal is needed.
 
         Args:
             text: The text to send as user input.
         """
         evt = events.TextInputMessage.from_text(text)
         await self.send_client_event(evt)
-        # After sending text, we need to signal that the turn is complete.
-        evt = events.ClientContentMessage.model_validate({"clientContent": {"turnComplete": True}})
-        await self.send_client_event(evt)
 
     async def _send_user_video(self, frame):
         """Send user video frame to Gemini Live API."""

src/pipecat/services/llm_service.py

@@ -25,7 +25,6 @@ from loguru import logger
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.direct_function import DirectFunction, DirectFunctionWrapper
-from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
 from pipecat.frames.frames import (
     CancelFrame,
@@ -108,6 +107,7 @@ class FunctionCallRegistryItem:
     function_name: Optional[str]
     handler: FunctionCallHandler | "DirectFunctionWrapper"
     cancel_on_interruption: bool
+    handler_deprecated: bool
 
 
 @dataclass
@@ -282,12 +282,25 @@ class LLMService(AIService):
             cancel_on_interruption: Whether to cancel this function call when an
                 interruption occurs. Defaults to True.
         """
+        signature = inspect.signature(handler)
+        handler_deprecated = len(signature.parameters) > 1
+        if handler_deprecated:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Function calls with parameters `(function_name, tool_call_id, arguments, llm, context, result_callback)` are deprecated, use a single `FunctionCallParams` parameter instead.",
+                    DeprecationWarning,
+                )
+
         # Registering a function with the function_name set to None will run
         # that handler for all functions
         self._functions[function_name] = FunctionCallRegistryItem(
             function_name=function_name,
             handler=handler,
             cancel_on_interruption=cancel_on_interruption,
+            handler_deprecated=handler_deprecated,
         )
 
         # Start callbacks are now deprecated.
@@ -325,6 +338,7 @@ class LLMService(AIService):
             function_name=wrapper.name,
             handler=wrapper,
             cancel_on_interruption=cancel_on_interruption,
+            handler_deprecated=False,
         )
 
     def unregister_function(self, function_name: Optional[str]):
@@ -552,17 +566,7 @@ class LLMService(AIService):
             )
         else:
             # Handler is a FunctionCallHandler
-            signature = inspect.signature(item.handler)
-            if len(signature.parameters) > 1:
-                import warnings
-
-                with warnings.catch_warnings():
-                    warnings.simplefilter("always")
-                    warnings.warn(
-                        "Function calls with parameters `(function_name, tool_call_id, arguments, llm, context, result_callback)` are deprecated, use a single `FunctionCallParams` parameter instead.",
-                        DeprecationWarning,
-                    )
-
+            if item.handler_deprecated:
                 await item.handler(
                     runner_item.function_name,
                     runner_item.tool_call_id,

src/pipecat/services/piper/tts.py

@@ -84,7 +84,9 @@ class PiperTTSService(TTSService):
         try:
             await self.start_ttfb_metrics()
 
-            async with self._session.post(self._base_url, json=text, headers=headers) as response:
+            async with self._session.post(
+                self._base_url, json={"text": text}, headers=headers
+            ) as response:
                 if response.status != 200:
                     error = await response.text()
                     logger.error(

src/pipecat/services/simli/video.py

@@ -18,6 +18,8 @@ from pipecat.frames.frames import (
     OutputImageRawFrame,
     StartInterruptionFrame,
     TTSAudioRawFrame,
+    TTSStoppedFrame,
+    UserStartedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, StartFrame
 from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
@@ -45,24 +47,39 @@ class SimliVideoService(FrameProcessor):
         simli_config: SimliConfig,
         use_turn_server: bool = False,
         latency_interval: int = 0,
+        simli_url: str = "https://api.simli.ai",
+        is_trinity_avatar: bool = False,
     ):
         """Initialize the Simli video service.
 
         Args:
             simli_config: Configuration object for Simli client settings.
             use_turn_server: Whether to use TURN server for connection. Defaults to False.
-            latency_interval: Latency interval setting for video processing. Defaults to 0.
+            latency_interval: Latency interval setting for sending health checks to check the latency to Simli Servers. Defaults to 0.
+            simli_url: URL of the simli servers. Can be changed for custom deployments of enterprise users.
+            is_trinity_avatar: boolean to tell simli client that this is a Trinity avatar which reduces latency when using Trinity.
+
         """
         super().__init__()
-        self._simli_client = SimliClient(simli_config, use_turn_server, latency_interval)
+        self._initialized = False
+        simli_config.maxIdleTime += 5
+        simli_config.maxSessionLength += 5
+        self._simli_client = SimliClient(
+            simli_config,
+            use_turn_server,
+            latency_interval,
+            simliURL=simli_url,
+        )
 
-        self._pipecat_resampler_event = asyncio.Event()
         self._pipecat_resampler: AudioResampler = None
+        self._pipecat_resampler_event = asyncio.Event()
         self._simli_resampler = AudioResampler("s16", "mono", 16000)
 
-        self._initialized = False
         self._audio_task: asyncio.Task = None
         self._video_task: asyncio.Task = None
+        self._is_trinity_avatar = is_trinity_avatar
+        self._previously_interrupted = is_trinity_avatar
+        self._audio_buffer = bytearray()
 
     async def _start_connection(self):
         """Start the connection to Simli service and begin processing tasks."""
@@ -71,11 +88,9 @@ class SimliVideoService(FrameProcessor):
             self._initialized = True
 
         # Create task to consume and process audio and video
-        if not self._audio_task:
-            self._audio_task = self.create_task(self._consume_and_process_audio())
-
-        if not self._video_task:
-            self._video_task = self.create_task(self._consume_and_process_video())
+        await self._simli_client.sendSilence()
+        self._audio_task = self.create_task(self._consume_and_process_audio())
+        self._video_task = self.create_task(self._consume_and_process_video())
 
     async def _consume_and_process_audio(self):
         """Consume audio frames from Simli and push them downstream."""
@@ -118,7 +133,6 @@ class SimliVideoService(FrameProcessor):
         """
         await super().process_frame(frame, direction)
         if isinstance(frame, StartFrame):
-            await self.push_frame(frame, direction)
             await self._start_connection()
         elif isinstance(frame, TTSAudioRawFrame):
             # Send audio frame to Simli
@@ -137,19 +151,41 @@ class SimliVideoService(FrameProcessor):
 
                 resampled_frames = self._simli_resampler.resample(old_frame)
                 for resampled_frame in resampled_frames:
-                    await self._simli_client.send(
-                        resampled_frame.to_ndarray().astype(np.int16).tobytes()
-                    )
+                    audioBytes = resampled_frame.to_ndarray().astype(np.int16).tobytes()
+                    if self._previously_interrupted:
+                        self._audio_buffer.extend(audioBytes)
+                        if len(self._audio_buffer) >= 128000:
+                            try:
+                                for flushFrame in self._simli_resampler.resample(None):
+                                    self._audio_buffer.extend(
+                                        flushFrame.to_ndarray().astype(np.int16).tobytes()
+                                    )
+                            finally:
+                                await self._simli_client.playImmediate(self._audio_buffer)
+                                self._previously_interrupted = False
+                                self._audio_buffer = bytearray()
+                    else:
+                        await self._simli_client.send(audioBytes)
+                return
             except Exception as e:
                 logger.exception(f"{self} exception: {e}")
+        elif isinstance(frame, TTSStoppedFrame):
+            try:
+                if self._previously_interrupted and len(self._audio_buffer) > 0:
+                    await self._simli_client.playImmediate(self._audio_buffer)
+                    self._previously_interrupted = False
+                    self._audio_buffer = bytearray()
+            except Exception as e:
+                logger.exception(f"{self} exception: {e}")
+            return
         elif isinstance(frame, (EndFrame, CancelFrame)):
             await self._stop()
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, StartInterruptionFrame):
-            await self._simli_client.clearBuffer()
-            await self.push_frame(frame, direction)
-        else:
-            await self.push_frame(frame, direction)
+        elif isinstance(frame, (StartInterruptionFrame, UserStartedSpeakingFrame)):
+            if not self._previously_interrupted:
+                await self._simli_client.clearBuffer()
+            self._previously_interrupted = self._is_trinity_avatar
+
+        await self.push_frame(frame, direction)
 
     async def _stop(self):
         """Stop the Simli client and cancel processing tasks."""

src/pipecat/services/speechmatics/stt.py

@@ -191,7 +191,7 @@ class SpeakerFragments:
             passive_format = active_format
         return {
             "text": self._format_text(active_format if self.is_active else passive_format),
-            "user_id": self.speaker_id,
+            "user_id": self.speaker_id or "",
             "timestamp": self.timestamp,
             "language": self.language,
             "result": [frag.result for frag in self.fragments],

src/pipecat/services/tavus/video.py

@@ -24,6 +24,7 @@ from pipecat.frames.frames import (
     Frame,
     OutputAudioRawFrame,
     OutputImageRawFrame,
+    OutputTransportReadyFrame,
     StartFrame,
     StartInterruptionFrame,
     TTSAudioRawFrame,
@@ -81,6 +82,7 @@ class TavusVideoService(AIService):
         self._send_task: Optional[asyncio.Task] = None
         # This is the custom track destination expected by Tavus
         self._transport_destination: Optional[str] = "stream"
+        self._transport_ready = False
 
     async def setup(self, setup: FrameProcessorSetup):
         """Set up the Tavus video service.
@@ -145,7 +147,8 @@ class TavusVideoService(AIService):
             format=video_frame.color_format,
         )
         frame.transport_source = video_source
-        await self.push_frame(frame)
+        if self._transport_ready:
+            await self.push_frame(frame)
 
     async def _on_participant_audio_data(
         self, participant_id: str, audio: AudioData, audio_source: str
@@ -157,7 +160,8 @@ class TavusVideoService(AIService):
             num_channels=audio.num_channels,
         )
         frame.transport_source = audio_source
-        await self.push_frame(frame)
+        if self._transport_ready:
+            await self.push_frame(frame)
 
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate processing metrics.
@@ -221,6 +225,9 @@ class TavusVideoService(AIService):
             await self.push_frame(frame, direction)
         elif isinstance(frame, TTSAudioRawFrame):
             await self._handle_audio_frame(frame)
+        elif isinstance(frame, OutputTransportReadyFrame):
+            self._transport_ready = True
+            await self.push_frame(frame, direction)
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/transports/services/tavus.py

@@ -245,6 +245,10 @@ class TavusTransportClient:
                 on_recording_started=partial(self._on_handle_callback, "on_recording_started"),
                 on_recording_stopped=partial(self._on_handle_callback, "on_recording_stopped"),
                 on_recording_error=partial(self._on_handle_callback, "on_recording_error"),
+                on_transcription_stopped=partial(
+                    self._on_handle_callback, "on_transcription_stopped"
+                ),
+                on_transcription_error=partial(self._on_handle_callback, "on_transcription_error"),
             )
             self._client = DailyTransportClient(
                 room_url, None, "Pipecat", self._params, daily_callbacks, self._bot_name