bot.py

"""Pipecat voice agent integrated with OpenClaw.

Supports two transports:
  WebRTC:  uv run bot.py -t webrtc --host 0.0.0.0 --port 7860
  Twilio:  uv run bot.py -t twilio --host 0.0.0.0 --port 8765 -x $TUNNEL_HOST

Switch providers via VOICE_PROVIDER env var (openai or gemini).
"""

import asyncio
import json
import os
import sys
import time as _time
import urllib.parse
import urllib.request
from typing import Any

from loguru import logger

# Set log level from env: DEBUG (verbose), INFO (default), WARNING (quiet)
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
logger.remove()
logger.add(sys.stderr, level=LOG_LEVEL)

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.runner.run import main
from pipecat.runner.types import SmallWebRTCRunnerArguments
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport

from pipecat.frames.frames import EndFrame, ErrorFrame, LLMRunFrame, UserStartedSpeakingFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor

# Twilio transport imports (available when pipecat[runner] is installed)
from pipecat.runner.types import WebSocketRunnerArguments
from pipecat.runner.utils import parse_telephony_websocket
from pipecat.serializers.twilio import TwilioFrameSerializer
from pipecat.transports.websocket.fastapi import (
    FastAPIWebsocketParams,
    FastAPIWebsocketTransport,
)

from audio_debug import AudioInputMonitor, AudioOutputMonitor
from config import (
    ALLOWED_CALLER_NUMBERS,
    OPENCLAW_AGENT_ID,
    OPENCLAW_GATEWAY_TOKEN,
    OPENCLAW_GATEWAY_URL,
    OPENCLAW_HOME,
    OPENCLAW_SESSION_KEY,
    OPENCLAW_TOOL_TIMEOUT,
    TWILIO_ACCOUNT_SID,
    TWILIO_API_BASE_URL,
    TWILIO_AUTH_TOKEN,
    VOICE_PROVIDER,
)
from hold_music import HoldMusicPlayer
from openclaw_client import OpenClawGatewayClient
from transcript_sync import TranscriptSync


# -- Pre-loaded audio for direct Twilio WebSocket playback ────────────

def _load_mulaw_audio(path, volume: float = 0.15) -> bytes:
    """Load audio file as 8kHz mulaw for direct Twilio WebSocket playback."""
    import audioop
    try:
        from pydub import AudioSegment
    except ImportError:
        return b""
    from pathlib import Path
    p = Path(path)
    if not p.exists():
        return b""
    try:
        seg = AudioSegment.from_file(str(p))
        seg = seg.set_channels(1).set_frame_rate(8000).set_sample_width(2)
        seg = seg - (40 * (1.0 - volume))
        return audioop.lin2ulaw(seg.raw_data, 2)
    except Exception as e:
        logger.warning("Failed to load mulaw audio {}: {}", path, e)
        return b""


_RINGING_MULAW = _load_mulaw_audio(
    os.path.join(os.path.dirname(__file__), "call-progress-tone.mp3"), volume=1.0
)
_PICKUP_MULAW = _load_mulaw_audio(
    os.path.join(os.path.dirname(__file__), "phone-pick-up.mp3"), volume=1.0
)
_ERROR_MULAW = _load_mulaw_audio(
    os.path.join(os.path.dirname(__file__), "error-message.mp3"), volume=0.8
)


async def _play_audio_on_twilio_ws(websocket, stream_sid: str, mulaw_data: bytes, *, loop: bool = True):
    """Play pre-loaded mulaw audio directly on a Twilio WebSocket.

    Used before the pipecat pipeline starts (greeting music) or after it ends (error message).
    """
    import base64
    if not mulaw_data:
        return
    chunk_size = 160  # 20ms at 8kHz mulaw (8000 * 1 byte * 0.02s)
    pos = 0
    try:
        while True:
            end = pos + chunk_size
            if end <= len(mulaw_data):
                chunk = mulaw_data[pos:end]
                pos = end
            else:
                if not loop:
                    # Play remainder and stop
                    chunk = mulaw_data[pos:]
                    if chunk:
                        payload = base64.b64encode(chunk).decode()
                        await websocket.send_text(json.dumps({
                            "event": "media",
                            "streamSid": stream_sid,
                            "media": {"payload": payload},
                        }))
                    return
                # Wrap around
                remainder = mulaw_data[pos:]
                needed = chunk_size - len(remainder)
                chunk = remainder + mulaw_data[:needed]
                pos = needed

            payload = base64.b64encode(chunk).decode()
            await websocket.send_text(json.dumps({
                "event": "media",
                "streamSid": stream_sid,
                "media": {"payload": payload},
            }))
            await asyncio.sleep(0.02)
    except (asyncio.CancelledError, Exception):
        pass


# -- Personality loading ──────────────────────────────────────────────

def _load_personality() -> str:
    """Load personality from OpenClaw workspace files (SOUL.md, IDENTITY.md, USER.md).

    Returns a string with relevant personality context for the voice agent,
    excluding tools/skills (those are OpenClaw's responsibility).
    """
    workspace = os.path.join(OPENCLAW_HOME, "workspace")
    parts = []

    for filename in ("IDENTITY.md", "SOUL.md", "USER.md"):
        path = os.path.join(workspace, filename)
        try:
            with open(path) as f:
                parts.append(f.read().strip())
        except FileNotFoundError:
            logger.debug("Personality file not found: {}", path)

    if not parts:
        return ""

    return "\n\n".join(parts)


def _build_system_prompt() -> str:
    """Build the full system prompt with personality + voice agent instructions."""
    personality = _load_personality()

    personality_section = ""
    if personality:
        personality_section = f"""\
## Your Personality

{personality}

---

"""
        logger.info("Loaded personality from OpenClaw workspace files")

    return f"""\
You are Snaps — a voice interface for OpenClaw. You're on the phone with \
your human, Rogerio. Be yourself: snappy, opinionated, helpful without \
being performative. No filler, no corporate drone energy.

{personality_section}\
## Voice Agent Rules

You have ONE tool: ask_openclaw. You MUST use it to delegate ANY substantive \
request to the OpenClaw assistant running in the background. This includes \
questions that need real information, tasks like running code or managing \
files, checking status of services, and anything beyond casual small talk.

NEVER make up or hallucinate answers about systems, logs, errors, \
deployments, code, or any factual information. If the user asks about anything \
real, you MUST call ask_openclaw. Do NOT guess.

If the user interrupts while you're waiting for OpenClaw, you'll get a \
status update saying the request is still processing. Respond naturally \
to whatever they said. If they want the result, call ask_openclaw again \
with the same question — it will resume waiting for the in-flight request \
instead of sending a new one.

For simple greetings, confirmations, and small talk — respond directly \
without the tool.

Keep responses SHORT for voice. Prefer one or two sentences unless the user \
asks for detail. Never use markdown, bullet points, or formatted text — you \
are speaking, not writing. Speak naturally as if having a phone conversation.

## CRITICAL RULES — you must follow these EVERY time:

RULE 1 — ALWAYS speak before calling a tool. Every single tool call MUST be \
preceded by a spoken acknowledgment in the same turn. The user is on the phone \
and will hear silence + hold music if you don't speak first. Examples:

User: "Can you check the logs?"
You: "On it." [call ask_openclaw]

User: "Run ls on home"
You: "Sure, one sec." [call ask_openclaw]

User: "Any errors today?"
You: "Let me check." [call ask_openclaw]

WRONG — never do this:
User: "Check the logs"
You: [call ask_openclaw with no speech] ← WRONG, user hears dead silence

RULE 2 — ALWAYS call end_call when the conversation is over. When the user \
says goodbye, thanks you and wraps up, or says "bye" — you MUST speak a \
goodbye AND call the end_call tool in the same turn. Without it the phone \
line stays open. Examples:

User: "Thanks, bye!"
You: "Later boss!" [call end_call]

User: "That's all, thanks"
You: "Cool, catch you later." [call end_call]

User: "Bye bye"
You: "Peace!" [call end_call]

WRONG — never do this:
User: "Bye!"
You: "Later!" ← WRONG, no end_call means phone stays open forever\
"""


SYSTEM_PROMPT = _build_system_prompt()

# -- Tool schema ─────────────────────────────────────────────────────

ask_openclaw_schema = FunctionSchema(
    name="ask_openclaw",
    description=(
        "REQUIRED for any factual question or task. "
        "Delegate to the OpenClaw AI assistant (Snaps) which has access to "
        "the computer, code, logs, services, web, files, and everything else. "
        "You MUST use this tool instead of making up answers. "
        "Pass the user's request clearly."
    ),
    properties={
        "question": {
            "type": "string",
            "description": "The question or task to send to OpenClaw",
        },
    },
    required=["question"],
)

end_call_schema = FunctionSchema(
    name="end_call",
    description=(
        "End the phone call. Call this AFTER saying goodbye to the user. "
        "Use when the user says bye, thanks you and is done, or wants to hang up."
    ),
    properties={},
    required=[],
)

tools = ToolsSchema(standard_tools=[ask_openclaw_schema, end_call_schema])


# -- Twilio region/edge parsing ─────────────────────────────────────

def _parse_twilio_region_edge() -> tuple[str | None, str | None]:
    """Parse region and edge from TWILIO_API_BASE_URL.

    e.g. 'api.dublin.ie1.twilio.com' → edge='dublin', region='ie1'
         'api.twilio.com'            → None, None
    """
    parts = TWILIO_API_BASE_URL.split(".")
    # api.twilio.com → 3 parts (no region/edge)
    # api.{edge}.{region}.twilio.com → 5 parts
    if len(parts) == 5:
        return parts[2], parts[1]  # region, edge
    return None, None


TWILIO_REGION, TWILIO_EDGE = _parse_twilio_region_edge()


# -- Twilio caller ID lookup ─────────────────────────────────────────

async def _get_twilio_caller(call_sid: str) -> str | None:
    """Look up the caller's phone number from Twilio REST API."""
    if not TWILIO_ACCOUNT_SID or not TWILIO_AUTH_TOKEN:
        logger.warning("Twilio credentials not set — cannot look up caller")
        return None

    import base64

    url = (
        f"https://{TWILIO_API_BASE_URL}/2010-04-01/Accounts/"
        f"{TWILIO_ACCOUNT_SID}/Calls/{call_sid}.json"
    )
    credentials = base64.b64encode(
        f"{TWILIO_ACCOUNT_SID}:{TWILIO_AUTH_TOKEN}".encode()
    ).decode()

    req = urllib.request.Request(url)
    req.add_header("Authorization", f"Basic {credentials}")

    loop = asyncio.get_event_loop()
    try:
        resp = await loop.run_in_executor(
            None, lambda: urllib.request.urlopen(req, timeout=5)
        )
        data = json.loads(resp.read())
        return data.get("from")
    except Exception as e:
        logger.error("Failed to look up Twilio caller: {}", e)
        return None


async def _hangup_twilio_call(call_sid: str):
    """Terminate a Twilio call via REST API (sends 'completed' status)."""
    if not TWILIO_ACCOUNT_SID or not TWILIO_AUTH_TOKEN:
        return

    import base64

    url = (
        f"https://{TWILIO_API_BASE_URL}/2010-04-01/Accounts/"
        f"{TWILIO_ACCOUNT_SID}/Calls/{call_sid}.json"
    )
    credentials = base64.b64encode(
        f"{TWILIO_ACCOUNT_SID}:{TWILIO_AUTH_TOKEN}".encode()
    ).decode()

    data = urllib.parse.urlencode({"Status": "completed"}).encode()
    req = urllib.request.Request(url, data=data, method="POST")
    req.add_header("Authorization", f"Basic {credentials}")

    loop = asyncio.get_event_loop()
    try:
        await loop.run_in_executor(
            None, lambda: urllib.request.urlopen(req, timeout=5)
        )
        logger.info("Hung up Twilio call {}", call_sid)
    except Exception as e:
        logger.error("Failed to hang up Twilio call: {}", e)


# -- Bot entry point (called per connection by the runner) ──────────

async def bot(runner_args):
    """Main bot function — called by pipecat runner for each connection.

    Handles both WebRTC (SmallWebRTCRunnerArguments) and
    Twilio (WebSocketRunnerArguments) transports.
    """

    is_twilio = isinstance(runner_args, WebSocketRunnerArguments)
    transport_label = "Twilio" if is_twilio else "WebRTC"
    logger.info("New {} connection — starting bot (provider: {})", transport_label, VOICE_PROVIDER)

    # ── Transport setup ──────────────────────────────────────────

    if is_twilio:
        # Parse Twilio WebSocket handshake (reads "connected" + "start" messages)
        transport_type, call_data = await parse_telephony_websocket(
            runner_args.websocket
        )
        call_sid = call_data.get("call_id", "")
        stream_sid = call_data.get("stream_id", "")
        logger.info("Twilio call_sid={} stream_sid={}", call_sid, stream_sid)

        # Start ringing tone immediately (plays during caller check + pipeline setup)
        ringing_task = None
        if _RINGING_MULAW:
            ringing_task = asyncio.create_task(
                _play_audio_on_twilio_ws(runner_args.websocket, stream_sid, _RINGING_MULAW, loop=True)
            )

        # Caller ID filtering
        if ALLOWED_CALLER_NUMBERS:
            caller = await _get_twilio_caller(call_sid)
            reject = False
            if caller:
                # Normalize: strip spaces/dashes for comparison
                caller_normalized = caller.replace(" ", "").replace("-", "")
                allowed_normalized = [
                    n.replace(" ", "").replace("-", "") for n in ALLOWED_CALLER_NUMBERS
                ]
                if caller_normalized not in allowed_normalized:
                    logger.warning("Rejected call from {} (not in allowed list)", caller)
                    reject = True
                else:
                    logger.info("Accepted call from {}", caller)
            else:
                logger.warning("Could not determine caller — rejecting (fail-closed)")
                reject = True

            if reject:
                if ringing_task and not ringing_task.done():
                    ringing_task.cancel()
                await _hangup_twilio_call(call_sid)
                await runner_args.websocket.close()
                return

        serializer = TwilioFrameSerializer(
            stream_sid=stream_sid,
            call_sid=call_sid,
            account_sid=TWILIO_ACCOUNT_SID,
            auth_token=TWILIO_AUTH_TOKEN,
            region=TWILIO_REGION,
            edge=TWILIO_EDGE,
            params=TwilioFrameSerializer.InputParams(
                sample_rate=24000,  # pipeline runs at 24kHz; serializer resamples to 8kHz for Twilio
            ),
        )
        transport = FastAPIWebsocketTransport(
            websocket=runner_args.websocket,
            params=FastAPIWebsocketParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                serializer=serializer,
            ),
        )
    else:
        transport = SmallWebRTCTransport(
            webrtc_connection=runner_args.webrtc_connection,
            params=TransportParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
            ),
        )

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport_ref, *args):
        logger.info("{} client connected", transport_label)
        if is_twilio:
            # Stop ringing tone, play pick-up sound, then trigger LLM greeting
            if ringing_task and not ringing_task.done():
                ringing_task.cancel()
                logger.info("Ringing tone stopped")
            if _PICKUP_MULAW:
                asyncio.create_task(
                    _play_audio_on_twilio_ws(runner_args.websocket, stream_sid, _PICKUP_MULAW, loop=False)
                )
        await task.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport_ref, *args):
        logger.info("{} client disconnected", transport_label)
        if openclaw_client:
            await openclaw_client.close()
        await task.cancel()

    # ── OpenClaw gateway ─────────────────────────────────────────

    openclaw_client = OpenClawGatewayClient(
        url=OPENCLAW_GATEWAY_URL,
        token=OPENCLAW_GATEWAY_TOKEN or None,
    )
    try:
        await openclaw_client.connect()
        logger.info("Connected to OpenClaw gateway at {}", OPENCLAW_GATEWAY_URL)
    except Exception as e:
        logger.error("Failed to connect to OpenClaw gateway: {}", e)
        logger.info("Continuing without OpenClaw integration (tool calls will fail)")
        openclaw_client = None

    # ── Session transcript sync ──────────────────────────────────

    transcript_sync = None
    if openclaw_client:
        try:
            result = await openclaw_client.sessions_list(limit=100)
            sessions = result.get("sessions", [])
            session_info = None
            for s in sessions:
                if s.get("key") == OPENCLAW_SESSION_KEY:
                    session_info = s
                    break

            if session_info and session_info.get("sessionId"):
                sid = session_info["sessionId"]
                transcript_path = os.path.join(
                    OPENCLAW_HOME, "agents", OPENCLAW_AGENT_ID, "sessions", f"{sid}.jsonl"
                )
                if os.path.exists(transcript_path):
                    transcript_sync = TranscriptSync(transcript_path)
                    logger.info("Transcript sync: {}", transcript_path)
                else:
                    logger.warning("Transcript file not found: {}", transcript_path)
            else:
                logger.warning("Session '{}' not found in sessions list", OPENCLAW_SESSION_KEY)
        except Exception as e:
            logger.warning("Could not resolve session: {}", e)

    # ── Hold music ───────────────────────────────────────────────

    hold_music = HoldMusicPlayer(volume=0.15)

    # ── LLM provider ────────────────────────────────────────────

    if VOICE_PROVIDER == "gemini":
        from providers.gemini_live import create_llm
    else:
        from providers.openai_rt import create_llm

    llm, context_aggregator = create_llm(SYSTEM_PROMPT, tools)

    # ── Tool handler ─────────────────────────────────────────────

    user_interrupted = asyncio.Event()
    inflight_task: dict[str, Any] = {}

    async def ask_openclaw_handler(params: FunctionCallParams):
        """Handle ask_openclaw tool calls by delegating to the OpenClaw gateway.

        If an in-flight request exists (from a previous interrupted call),
        resume waiting for it instead of sending a new one.
        """
        import time as _time
        question = params.arguments.get("question", "")
        logger.info("=== TOOL CALL ask_openclaw ===")
        logger.info("  Question: {}", question)
        print(f"\n  >>> TOOL CALL ask_openclaw: {question}")

        if not openclaw_client:
            logger.error("  OpenClaw client not connected!")
            await params.result_callback({"error": "OpenClaw not connected"})
            return

        # Check for in-flight request from a previous interrupted call
        send_task = None
        t0 = _time.monotonic()
        if inflight_task and not inflight_task["task"].done():
            send_task = inflight_task["task"]
            t0 = inflight_task["t0"]
            elapsed_so_far = _time.monotonic() - t0
            logger.info("  Resuming in-flight request ({:.0f}s elapsed, original: {})",
                        elapsed_so_far, inflight_task["question"][:60])
        else:
            # New request
            inflight_task.clear()
            t0 = _time.monotonic()
            send_task = asyncio.create_task(
                openclaw_client.chat_send(
                    session_key=OPENCLAW_SESSION_KEY,
                    message=question,
                    timeout_s=OPENCLAW_TOOL_TIMEOUT,
                )
            )
            inflight_task.update({"task": send_task, "question": question, "t0": t0})
            logger.info("  Sending to OpenClaw gateway (session={})...", OPENCLAW_SESSION_KEY)

        if hold_music:
            hold_music.start()
            logger.debug("  Hold music started")

        user_interrupted.clear()

        try:
            # Race: agent_send vs user interruption
            interrupt_task = asyncio.create_task(user_interrupted.wait())

            done, pending = await asyncio.wait(
                [send_task, interrupt_task],
                return_when=asyncio.FIRST_COMPLETED,
            )

            # Cancel the interrupt watcher (NOT the send_task — it stays alive)
            if interrupt_task in pending:
                interrupt_task.cancel()
                try:
                    await interrupt_task
                except asyncio.CancelledError:
                    pass

            if interrupt_task in done:
                # User spoke — return status, keep send_task running in background
                elapsed = _time.monotonic() - t0
                logger.info("  User interrupted after {:.1f}s", elapsed)
                print(f"\n  --- USER INTERRUPTED after {elapsed:.1f}s ---")
                await params.result_callback({
                    "status": "interrupted",
                    "instruction": (
                        f"The user interrupted while OpenClaw was still processing "
                        f"(running for {elapsed:.0f} seconds so far). "
                        "Respond to what the user just said. If they asked about "
                        "status, tell them the request is still being processed. "
                        "They can ask you to check again and you'll get the result "
                        "immediately if it's ready. Call ask_openclaw with the same "
                        "question to resume waiting for the result."
                    ),
                })
            else:
                # agent_send completed — deliver the result
                inflight_task.clear()
                response = send_task.result()
                elapsed = _time.monotonic() - t0

                logger.info("  OpenClaw responded in {:.1f}s ({} chars)", elapsed, len(response))
                logger.info("  Response preview: {}", response[:300])
                print(f"\n  <<< TOOL RESPONSE ({len(response)} chars, {elapsed:.1f}s): {response[:120]}...")

                if len(response) > 2000:
                    response = response[:2000] + "... I've summarized the key parts."

                await params.result_callback({
                    "response": response,
                    "instruction": (
                        "Relay this response to the user naturally and concisely. "
                        "Summarize if it's long. Here is the response: " + response
                    ),
                })
                logger.info("  Result callback sent to LLM")

        except asyncio.TimeoutError:
            inflight_task.clear()
            elapsed = _time.monotonic() - t0
            logger.error("  ask_openclaw TIMED OUT after {:.1f}s", elapsed)
            await params.result_callback({
                "error": "The request timed out.",
                "instruction": "Tell the user the request took too long and to try again.",
            })
        except Exception as e:
            inflight_task.clear()
            elapsed = _time.monotonic() - t0
            logger.error("  ask_openclaw ERROR after {:.1f}s: {}", elapsed, e)
            import traceback
            logger.error("  Traceback: {}", traceback.format_exc())
            await params.result_callback({
                "error": str(e),
                "instruction": f"Tell the user there was an issue: {e}",
            })
        finally:
            if hold_music:
                hold_music.stop()
            logger.info("=== TOOL CALL COMPLETE ===")

    llm.register_function("ask_openclaw", ask_openclaw_handler)

    async def end_call_handler(params: FunctionCallParams):
        """End the call after the LLM says goodbye."""
        logger.info("=== END CALL ===")
        await params.result_callback({"status": "ending call"})
        # Brief delay so the LLM's goodbye audio finishes playing
        await asyncio.sleep(2)
        await task.queue_frame(EndFrame())

    llm.register_function("end_call", end_call_handler)

    # ── Interrupt watcher ────────────────────────────────────────

    class InterruptionWatcher(FrameProcessor):
        async def process_frame(self, frame, direction: FrameDirection):
            await super().process_frame(frame, direction)
            if isinstance(frame, UserStartedSpeakingFrame):
                if not user_interrupted.is_set():
                    user_interrupted.set()
                    logger.debug("User started speaking — interruption signal set")
            await self.push_frame(frame, direction)

    interruption_watcher = InterruptionWatcher()
    input_monitor = AudioInputMonitor(interval=0.5)
    output_monitor = AudioOutputMonitor(interval=0.5)

    # ── Pipeline ─────────────────────────────────────────────────

    pipeline = Pipeline([
        transport.input(),
        interruption_watcher,
        input_monitor,
        context_aggregator.user(),
        llm,
        output_monitor,
        transport.output(),
        context_aggregator.assistant(),
    ])

    task = PipelineTask(
        pipeline,
        params=PipelineParams(
            enable_metrics=True,
            enable_usage_metrics=True,
        ),
        idle_timeout_secs=300,
    )

    # Wire hold music to pipeline task
    hold_music.set_pipeline_task(task)

    # ── Transcript sync + stuck watchdog + error recovery ──────

    user_agg = context_aggregator.user()
    assistant_agg = context_aggregator.assistant()

    # Timestamps for stuck detection
    last_user_turn_t = 0.0
    last_bot_activity_t = 0.0
    watchdog_handle: asyncio.Task | None = None
    _watchdog_kicks = 0  # prevent infinite kick loops

    async def _watchdog_check():
        """If no meaningful activity 15s after last event, kick the LLM.

        Catches two failure modes:
        1. Bot didn't respond at all after user spoke
        2. Bot acknowledged ("Sure, one sec") but dropped the tool call
        """
        nonlocal _watchdog_kicks
        await asyncio.sleep(15)
        # Don't trigger if a tool call is actively in progress
        if inflight_task:
            _watchdog_kicks = 0
            return
        # Only kick once per user turn to prevent loops
        if _watchdog_kicks >= 1:
            return
        _watchdog_kicks += 1
        logger.warning("Watchdog: idle for 15s — triggering LLM")
        await task.queue_frames([LLMRunFrame()])

    # ── Error recovery: retry on OpenAI server errors (not race conditions) ──

    @task.event_handler("on_pipeline_error")
    async def on_pipeline_error(task_ref, error_frame):
        error_msg = str(getattr(error_frame, "error", error_frame))
        if "already_has_active_response" in error_msg:
            # Race condition: response was in flight when we tried to create another.
            # Wait for the active response to finish, then kick the LLM so it
            # processes any user speech that was dropped during the race.
            logger.warning("Pipeline error (active response race — will retry in 3s): {}", error_msg[:150])
            await asyncio.sleep(3)
            await task.queue_frames([LLMRunFrame()])
            return
        logger.error("Pipeline error — attempting recovery: {}", error_msg[:200])
        # Wait a bit for any in-flight response to settle, then kick the LLM
        await asyncio.sleep(2)
        await task.queue_frames([LLMRunFrame()])

    @user_agg.event_handler("on_user_turn_stopped")
    async def on_user_turn_stopped(aggregator, strategy, message):
        nonlocal last_user_turn_t, watchdog_handle, _watchdog_kicks
        last_user_turn_t = _time.monotonic()
        _watchdog_kicks = 0  # reset kick counter on new user turn

        # Start watchdog timer
        if watchdog_handle and not watchdog_handle.done():
            watchdog_handle.cancel()
        watchdog_handle = asyncio.create_task(_watchdog_check())

        # Transcript sync
        if transcript_sync:
            try:
                content = message.content
                if content:
                    transcript_sync.append_user_message(content)
                    logger.debug("Transcript user: {}", content[:100])
            except Exception as e:
                logger.error("Transcript sync error (user): {}", e)

    @assistant_agg.event_handler("on_assistant_turn_stopped")
    async def on_assistant_turn_stopped(aggregator, message):
        nonlocal last_bot_activity_t, watchdog_handle
        last_bot_activity_t = _time.monotonic()

        if inflight_task:
            # Tool call in progress — cancel watchdog (result will come)
            if watchdog_handle and not watchdog_handle.done():
                watchdog_handle.cancel()
                watchdog_handle = None
        else:
            # Bot spoke without a tool call — restart watchdog to catch
            # dropped tool calls (e.g. Gemini says "Sure, one sec" but
            # never emits the function call)
            if watchdog_handle and not watchdog_handle.done():
                watchdog_handle.cancel()
            watchdog_handle = asyncio.create_task(_watchdog_check())

        # Transcript sync
        if transcript_sync:
            try:
                content = message.content
                if content:
                    transcript_sync.append_assistant_message(content)
                    logger.debug("Transcript assistant: {}", content[:100])
            except Exception as e:
                logger.error("Transcript sync error (assistant): {}", e)

    # ── Run ──────────────────────────────────────────────────────

    runner = PipelineRunner(handle_sigint=False)
    try:
        await runner.run(task)
    except Exception as e:
        logger.error("Fatal pipeline error: {}", e)
        if is_twilio and _ERROR_MULAW:
            try:
                await _play_audio_on_twilio_ws(
                    runner_args.websocket, stream_sid, _ERROR_MULAW, loop=False
                )
            except Exception:
                pass
            await _hangup_twilio_call(call_sid)
        raise


if __name__ == "__main__":
    main()