Blame for core/flow.py

"""
Pipecat Flow Handlers for RTVI Integration

This module provides handlers for integrating Pipecat flows with RTVI punt system,
allowing client-side code execution while maintaining server-side flow management.
"""

from datetime import datetime
from typing import Callable, Optional
from pipecat.processors.frameworks.rtvi import RTVIProcessor, RTVIServerMessageFrame
from pipecat_flows import FlowConfig, FlowManager, FlowsFunctionSchema
from loguru import logger

# Define the set of action types that should get no-op handlers
NO_OP_ACTION_TYPES = {
    "save_summary_conversation",
    "node_summary_conversation",
    "save_html_conversation",
}


def create_stub_handler(
    node_name: str,
    function_name: str,
    rtvi_client: RTVIProcessor,
    transition_to: Optional[str] = None,
) -> Callable:
    """
    Create a stub handler that forwards function calls to client via RTVI.

    This enables "RTVI stubs" - when server-side flow encounters a client function,
    it sends the request to the client for execution while maintaining flow state.

    Args:
        node_name: Name of the node that contains the function
        function_name: Name of the function to punt to client
        rtvi_client: RTVI client for sending messages
        transition_to: Optional node to transition to after function execution

    Returns:
        Async handler function for use with FlowManager
    """

    async def stub_handler(params, flow_manager: FlowManager):
        """Fire-and-forget stub handler"""
        try:
            # Create function request message
            message = {
                "type": "function_request",
                "function_name": function_name,
                "params": [params],
                "timestamp": datetime.now().isoformat(),
            }

            # Send to client via RTVI server message frame
            frame = RTVIServerMessageFrame(data=message)
            await flow_manager.task.queue_frame(frame)
            logger.info(f"🚀 Punted function '{function_name}' to client")

            if transition_to:
                state_message = {
                    "type": "flow_state_sync",
                    "current_node": transition_to,
                    "previous_node": node_name,
                    "timestamp": datetime.now().isoformat(),
                }
                frame = RTVIServerMessageFrame(data=state_message)
                await flow_manager.task.queue_frame(frame)
                logger.info(f"🔄 State sync: {node_name} → {transition_to}")

            return {"status": "success"}

        except Exception as e:
            logger.error(f"❌ Failed to punt function '{function_name}': {e}")
            return {"status": "error", "error": str(e)}

    return stub_handler


async def send_state_sync(flow_manager: FlowManager, node_name: str):
    state_message = {
        "type": "flow_state_sync",
        "current_node": node_name,
        "timestamp": datetime.now().isoformat(),
    }
    frame = RTVIServerMessageFrame(data=state_message)
    await flow_manager.task.queue_frame(frame)
    logger.info(f"🔄 State sync: {node_name}")


def create_no_op_action_handler() -> Callable:
    """
    Create a no-op handler for actions that should be ignored on server-side.

    Returns:
        Async handler function that does nothing
    """

    async def no_op_handler(action: dict, flow_manager: FlowManager):
        """No-op handler for client-side only actions"""
        logger.debug(
            f"No-op handler called for action type: {action.get('type', 'unknown')}"
        )
        pass

    return no_op_handler


async def end_conversation_handler(action: dict, flow_manager: FlowManager):
    """Handler for end_conversation action - works for both Daily and Twilio"""
    try:
        # Import here to avoid circular imports
        from utils.twilio import terminate_call, get_call_context

        # Determine call source
        if not action or action == {}:
            call_source = "function_call"
        else:
            call_source = "post_action"

        logger.info(f"🏁 End conversation called with source {call_source}")

        # Create end_conversation message with source information
        message = {
            "type": "end_conversation",
            "timestamp": datetime.now().isoformat(),
            "call_source": call_source,
            "immediate_disconnect": call_source == "function_call",
        }

        # Try to send RTVI message (works for Daily/WebRTC clients)
        try:
            frame = RTVIServerMessageFrame(data=message)
            await flow_manager.task.queue_frame(frame)
            logger.info("🏁 Sent end_conversation message to client")
        except Exception as rtvi_error:
            logger.debug(f"RTVI message not sent (expected for Twilio): {rtvi_error}")

        # For Twilio calls, terminate via REST API
        call_context = get_call_context()
        if call_context and call_context.get("call_sid"):
            call_sid = call_context["call_sid"]
            logger.info(f"🏁 Terminating Twilio call via REST API: {call_sid}")
            await terminate_call(call_sid)

        # Cancel the task to end the call (works for both Daily and Twilio)
        # This ensures the call is terminated regardless of transport type
        if call_source == "function_call":
            logger.info("🏁 Cancelling task to end call")
            await flow_manager.task.cancel()

    except Exception as e:
        logger.error(f"❌ Failed to end conversation: {e}")


def process_flow_config_handlers(
    llm_flow: FlowConfig,
    rtvi_client: RTVIProcessor,
) -> FlowConfig:
    """
    Process flow configuration and replace string handlers with RTVI punt handlers.

    This is the core of the RTVI punt system - it finds string handlers like
    "__function__:handleTransition" and replaces them with lambda functions
    that send function requests to the client.

    Args:
        llm_flow: LLM flow configuration
        rtvi_client: RTVI client for sending messages

    Returns:
        Processed flow config with punt handlers
    """
    processed_config = llm_flow.copy()

    # Get the handlers once
    no_op_handler = create_no_op_action_handler()

    # Process each node
    for node_name, node_config in processed_config.get("nodes", {}).items():
        # Process functions
        functions = node_config.get("functions", [])

        # Inject end_conversation function at every node
        end_conversation_function = FlowsFunctionSchema(
            name="end_conversation",
            description="Call this function when the user explicitly asks to stop, end, or terminate the current interview or conversation. This should be called when user clearly expresses they want to finish the interview session.",
            properties={},
            required=[],
            handler=end_conversation_handler,
        )
        functions.append(end_conversation_function)
        logger.info(f"💬 Injected end_conversation function in node: {node_name}")

        for i, func_outer in enumerate(functions):
            # Handle both dict format and FlowsFunctionSchema
            if isinstance(func_outer, FlowsFunctionSchema):
                func = func_outer
                func_name = func.name
            else:
                func = func_outer["function"]
                func_name = func["name"]

            # Skip processing if this is the end_conversation function we just added
            if func_name == "end_conversation":
                continue

            # Convert to FlowsFunctionSchema if it's not already
            if isinstance(func_outer, dict):
                func_data = func_outer["function"]
                name = func_data["name"]
                description = func_data.get("description", "")
                parameters = func_data.get("parameters", {}) or {}
                properties = parameters.get("properties", {})
                required = parameters.get("required", [])

                # Extract Flows-specific fields
                transition_to = func_data.get("transition_to")

                # Create RTVI stub handler
                handler = create_stub_handler(
                    node_name, func_name, rtvi_client, transition_to=transition_to
                )
                logger.info(f"📤 Converted {func_name} to RTVI stub")

                new_func = FlowsFunctionSchema(
                    name=name,
                    description=description,
                    properties=properties,
                    required=required,
                    handler=handler,
                    transition_to=transition_to,
                )

                functions[i] = new_func

        # Process pre_actions
        pre_actions = node_config.get("pre_actions", [])
        for action in pre_actions:
            action_type = action.get("type")
            if action_type in NO_OP_ACTION_TYPES:
                action["handler"] = no_op_handler
                logger.info(
                    f"Added no-op handler to pre_action: {action_type} in node: {node_name}"
                )

        # Process post_actions
        post_actions = node_config.get("post_actions", [])
        for action in post_actions:
            action_type = action.get("type")
            if action_type in NO_OP_ACTION_TYPES:
                action["handler"] = no_op_handler
                logger.info(
                    f"Added no-op handler to post_action: {action_type} in node: {node_name}"
                )

    logger.info("✅ Flow config handlers processed for RTVI stubs and actions")
    return processed_config