feat: Let the AI hold the line (New Wait Tool with Music & Interrupts)

api/alembic/versions/ceeaf3c37b2f_add_wait_to_tool_category.py

@@ -0,0 +1,64 @@
+"""add wait to tool category
+
+Revision ID: ceeaf3c37b2f
+Revises: gg11dd223344
+Create Date: 2026-07-22 12:42:18.521312
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+from alembic_postgresql_enum import TableReference
+
+# revision identifiers, used by Alembic.
+revision: str = 'ceeaf3c37b2f'
+down_revision: Union[str, None] = '0a1b2c3d4e5f'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.sync_enum_values(
+        enum_schema="public",
+        enum_name="tool_category",
+        new_values=[
+            "http_api",
+            "end_call",
+            "transfer_call",
+            "calculator",
+            "native",
+            "integration",
+            "mcp",
+            "wait",
+        ],
+        affected_columns=[
+            TableReference(
+                table_schema="public", table_name="tools", column_name="category"
+            )
+        ],
+        enum_values_to_rename=[],
+    )
+
+def downgrade() -> None:
+    op.sync_enum_values(
+        enum_schema="public",
+        enum_name="tool_category",
+        new_values=[
+            "http_api",
+            "end_call",
+            "transfer_call",
+            "calculator",
+            "native",
+            "integration",
+            "mcp",
+        ],
+        affected_columns=[
+            TableReference(
+                table_schema="public", table_name="tools", column_name="category"
+            )
+        ],
+        enum_values_to_rename=[],
+    )

api/enums.py

@@ -173,6 +173,7 @@ class ToolCategory(Enum):
     NATIVE = "native"  # Built-in integrations (future: dtmf_input)
     INTEGRATION = "integration"  # Third-party integrations (future: Google Calendar, Salesforce, etc.)
     MCP = "mcp"  # Customer-provided MCP server exposing a tool catalog
+    WAIT = "wait"  # Built-in wait_for_user tool
 
 
 class ToolStatus(Enum):

api/schemas/tool.py

@@ -25,6 +25,7 @@
     "end_call",
     "transfer_call",
     "calculator",
+    "wait",
     "native",
     "integration",
     "mcp",
@@ -435,6 +436,13 @@ class CalculatorToolDefinition(BaseModel):
     type: Literal["calculator"] = Field(description="Tool type.")
 
 
+class WaitToolDefinition(BaseModel):
+    """Tool definition for Wait tools."""
+
+    schema_version: int = Field(default=1, description="Schema version.")
+    type: Literal["wait"] = Field(description="Tool type.")
+
+
 class McpToolDefinition(BaseModel):
     """Persisted MCP tool definition."""
 
@@ -449,6 +457,7 @@ class McpToolDefinition(BaseModel):
         EndCallToolDefinition,
         TransferCallToolDefinition,
         CalculatorToolDefinition,
+        WaitToolDefinition,
         McpToolDefinition,
     ],
     Field(discriminator="type"),

api/services/pipecat/audio_playback.py

@@ -174,19 +174,32 @@ async def play_audio_loop(
     duration = num_samples / sample_rate
 
     logger.debug(f"Audio loop: playing at {sample_rate}Hz")
+
+    # 500ms chunks to allow near-instant cancellation without music bleeding
+    chunk_duration_sec = 0.5
+    chunk_size = int(sample_rate * chunk_duration_sec * 2)  # 2 bytes per sample
+
     try:
         while not stop_event.is_set():
-            frame = OutputAudioRawFrame(
-                audio=audio_data,
-                sample_rate=sample_rate,
-                num_channels=1,
-            )
-            await queue_frame(frame)
-            try:
-                await asyncio.wait_for(stop_event.wait(), timeout=duration + 1.5)
-                break
-            except asyncio.TimeoutError:
-                pass
+            for i in range(0, len(audio_data), chunk_size):
+                if stop_event.is_set():
+                    break
+
+                chunk = audio_data[i:i + chunk_size]
+                frame = OutputAudioRawFrame(
+                    audio=chunk,
+                    sample_rate=sample_rate,
+                    num_channels=1,
+                )
+                await queue_frame(frame)
+
+                chunk_play_time = (len(chunk) // 2) / sample_rate
+
+                try:
+                    await asyncio.wait_for(stop_event.wait(), timeout=chunk_play_time)
+                    break
+                except asyncio.TimeoutError:
+                    pass
     except Exception as e:
         logger.error(f"Audio loop error: {e}")
     logger.debug("Audio loop: stopped")

api/services/workflow/pipecat_engine_custom_tools.py

@@ -28,6 +28,7 @@
 from api.services.telephony.factory import get_telephony_provider_for_run
 from api.services.telephony.transfer_event_protocol import TransferContext
 from api.services.workflow.tools.calculator import get_calculator_tools, safe_calculator
+from api.services.workflow.tools.wait import get_wait_tools
 from api.services.workflow.tools.custom_tool import (
     execute_http_tool,
     tool_to_function_schema,
@@ -172,6 +173,21 @@ async def get_tool_schemas(
 
             schemas: list[FunctionSchema] = []
             for tool in tools:
+                if tool.category == "wait":
+                    self._register_wait_handler()
+                    logger.debug(f"Registered wait tool handler (tool_uuid: {tool.tool_uuid})")
+                    for tool_def in get_wait_tools():
+                        func = tool_def["function"]
+                        schemas.append(
+                            get_function_schema(
+                                func["name"],
+                                func["description"],
+                                properties=func["parameters"]["properties"],
+                                required=func["parameters"]["required"],
+                            )
+                        )
+                    continue
+
                 if tool.category == ToolCategory.CALCULATOR.value:
                     # Built-in calculator: return pre-defined schemas
                     for tool_def in get_calculator_tools():
@@ -250,6 +266,11 @@ async def register_handlers(
             tools = await db_client.get_tools_by_uuids(tool_uuids, organization_id)
 
             for tool in tools:
+                if tool.category == "wait":
+                    self._register_wait_handler()
+                    logger.debug(f"Registered wait tool handler (tool_uuid: {tool.tool_uuid})")
+                    continue
+
                 if tool.category == ToolCategory.CALCULATOR.value:
                     self._register_calculator_handler()
                     logger.debug(
@@ -359,6 +380,186 @@ def _transfer_handler_timeout_secs(self, tool: Any) -> float:
 
         return float(transfer_timeout) + resolver_timeout + 15.0
 
+    def _register_wait_handler(self) -> None:
+        """Register the built-in wait function with the LLM."""
+
+        async def wait_func(function_call_params: FunctionCallParams) -> None:
+            logger.info("LLM Function Call EXECUTED: wait_for_user")
+            logger.info(f"Arguments: {function_call_params.arguments}")
+
+            try:
+                seconds_arg = function_call_params.arguments.get("seconds", 60)
+                try:
+                    seconds = int(seconds_arg)
+                    if seconds < 0:
+                        raise ValueError("Seconds cannot be negative.")
+                except (ValueError, TypeError):
+                    await function_call_params.result_callback({"error": f"Invalid 'seconds' parameter: {seconds_arg}. Must be a positive integer."})
+                    return
+
+                # Use a hard cap of 300 seconds for the built-in wait tool
+                max_wait = 300
+                seconds = min(max(seconds, 1), max_wait)
+
+                message = function_call_params.arguments.get("message")
+                if message:
+                    from pipecat.frames.frames import TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame
+                    from pipecat.observers.base_observer import BaseObserver as _BaseObserver, FramePushed as _FramePushed
+
+                    logger.info(f"Playing wait acknowledgment message: {message}")
+
+                    # Register a temporary observer BEFORE queuing the TTS so we
+                    # don't miss TTSStartedFrame if the pipeline is very fast.
+                    # We require TTSStartedFrame before accepting TTSStoppedFrame to
+                    # guard against stale TTSStoppedFrames from the previous response.
+                    tts_done_event = asyncio.Event()
+
+                    class _TTSDoneObserver(_BaseObserver):
+                        def __init__(self):
+                            super().__init__()
+                            self._seen_started = False
+
+                        async def on_push_frame(self, data: _FramePushed):
+                            if isinstance(data.frame, TTSStartedFrame):
+                                self._seen_started = True
+                            elif isinstance(data.frame, TTSStoppedFrame) and self._seen_started:
+                                logger.info("Wait tool: acknowledgment TTS finished (TTSStoppedFrame received).")
+                                tts_done_event.set()
+
+                    tts_done_observer = _TTSDoneObserver()
+                    _task = self._engine.task
+                    if _task:
+                        _task.add_observer(tts_done_observer)
+
+                    await self._engine.task.queue_frame(
+                        TTSSpeakFrame(
+                            message,
+                            append_to_context=False,
+                            persist_to_logs=True,
+                        )
+                    )
+
+                    # Wait for TTS to actually finish, with a generous fallback timeout.
+                    try:
+                        await asyncio.wait_for(tts_done_event.wait(), timeout=15.0)
+                    except asyncio.TimeoutError:
+                        logger.warning("Wait tool: TTS done event timed out after 15s ÔÇö proceeding anyway.")
+
+                    if _task:
+                        try:
+                            await _task.remove_observer(tts_done_observer)
+                        except Exception as e:
+                            logger.warning(f"Could not remove TTS done observer: {e}")
+
+                logger.info(f"Pausing for {seconds} seconds...")
+
+                stop_event = asyncio.Event()
+                hold_music_task = None
+
+                if seconds >= 5 and self._engine._audio_config and self._engine._transport_output:
+                    from api.constants import APP_ROOT_DIR
+                    sample_rate = self._engine._audio_config.pipeline_sample_rate
+                    audio_file = str(APP_ROOT_DIR / "assets" / f"wait_music_{sample_rate}.wav")
+
+                    hold_music_task = asyncio.create_task(
+                        play_audio_loop(
+                            stop_event=stop_event,
+                            sample_rate=sample_rate,
+                            queue_frame=self._engine._transport_output.queue_frame,
+                            audio_file=audio_file,
+                        )
+                    )
+
+                user_speech_event = asyncio.Event()
+                # Issue 2 fix: capture the actual transcript text so we can include it
+                # in the tool result. Without this, the TranscriptionFrame is muted by
+                # FunctionCallUserMuteStrategy and the LLM never sees what the user said,
+                # forcing the user to repeat themselves.
+                import time
+                user_speech_text: list[str] = []  # mutable container for closure capture
+
+                # We use a pipeline observer (not the mute-filtered aggregator)
+                # because FunctionCallUserMuteStrategy suppresses all
+                # TranscriptionFrames and UserStartedSpeakingFrames while a
+                # tool is executing. Pipeline observers receive frames *before*
+                # the mute strategy suppresses them.
+                from pipecat.observers.base_observer import BaseObserver, FramePushed
+                from pipecat.frames.frames import TranscriptionFrame
+
+                class _WaitInterruptObserver(BaseObserver):
+                    async def on_push_frame(self, data: FramePushed):
+                        if isinstance(data.frame, TranscriptionFrame) and data.frame.text.strip():
+                            text = data.frame.text.strip()
+                            logger.info(f"Wait tool observer: transcription received: '{text}', signalling interrupt.")
+                            # Capture the text (Issue 2 fix: include in tool result)
+                            if not user_speech_text:
+                                user_speech_text.append(text)
+                            user_speech_event.set()
+
+                wait_observer = _WaitInterruptObserver()
+                task = self._engine.task
+                if task:
+                    task.add_observer(wait_observer)
+
+                elapsed = 0.0
+                try:
+                    # Fix #1: Never wait less than 15 seconds to avoid overlapping TTS
+                    # acknowledgment and immediate "are you still there" follow-up.
+                    effective_seconds = min(max(float(seconds), 15.0), float(max_wait))
+
+                    while round(elapsed, 1) < effective_seconds:
+                        if self._engine.is_call_disposed():
+                            break
+
+                        if user_speech_event.is_set():
+                            logger.info("Wait interrupted by user speech.")
+                            break
+
+                        await asyncio.sleep(0.1)
+                        elapsed += 0.1
+                finally:
+                    stop_event.set()
+                    if hold_music_task:
+                        hold_music_task.cancel()
+                        try:
+                            await hold_music_task
+                        except asyncio.CancelledError:
+                            pass
+
+                    if task:
+                        try:
+                            await task.remove_observer(wait_observer)
+                        except Exception as e:
+                            logger.warning(f"Could not remove wait observer: {e}")
+
+                logger.info(f"Wait completed after {elapsed} seconds (requested {seconds}).")
+
+                if function_call_params and function_call_params.result_callback:
+                    clean_elapsed = round(elapsed, 1)
+                    msg = f"Wait finished after {clean_elapsed} seconds. "
+                    if user_speech_event.is_set():
+                        # Issue 2 fix: The TranscriptionFrame was muted by
+                        # FunctionCallUserMuteStrategy so the LLM context never received
+                        # it. We inject the captured text directly into the tool result
+                        # so the LLM can respond to the user's exact words.
+                        captured = user_speech_text[0] if user_speech_text else None
+                        if captured:
+                            msg += f'The user has returned and said: "{captured}". Respond directly to what they said. Do NOT thank them for waiting, because YOU were the one waiting for THEM.'
+                        else:
+                            msg += "The user has returned and spoken. Please respond naturally to what they just said. Do NOT thank them for waiting, because YOU were the one waiting for THEM."
+                    else:
+                        msg += "The time is up. Please ask the user if they are still there and if they need further assistance. Do NOT thank them for waiting, because YOU were the one waiting for THEM."
+
+                    await function_call_params.result_callback(
+                        {"status": "success", "message": msg}
+                    )
+            except Exception as e:
+                logger.error(f"Wait tool error: {e}")
+                await function_call_params.result_callback({"error": str(e)})
+
+        # Register with a large timeout to prevent the LLM caller from timing out the wait.
+        self._engine.llm.register_function("wait_for_user", wait_func, timeout_secs=330.0)
+
     def _register_calculator_handler(self) -> None:
         """Register the built-in calculator function with the LLM."""