Handoff Strategies

from registries.scenariostore.loader import (
    build_handoff_map_from_scenario,
    get_handoff_config,
)

# Build handoff routing from scenario
handoff_map = build_handoff_map_from_scenario(scenario_name)
# → {"handoff_fraud_agent": "FraudAgent", "handoff_concierge": "Concierge", ...}

async def _execute_tool_call(self, name: str, args: Dict[str, Any]) -> None:
    # Check if this is a handoff tool
    target_agent = self._handoff_map.get(name)
    if target_agent:
        # Get handoff configuration from scenario
        handoff_cfg = get_handoff_config(
            scenario_name=self._scenario_name,
            from_agent=self._active_agent_name,
            tool_name=name,
        )

        # Determine greeting behavior
        should_greet = handoff_cfg.greet_on_switch  # True if "announced"

        # Execute the switch with appropriate behavior
        await self._switch_to(
            target_agent,
            system_vars,
            greet=should_greet,
        )
        return

    # Otherwise execute as business tool
    result = await execute_tool(name, args)

# registries/toolstore/handoffs.py

@register_tool(
    name="handoff_new_specialist",
    description="Transfer to the new specialist for domain expertise",
    is_handoff=True,
)
async def handoff_new_specialist(reason: str, details: str = "") -> Dict[str, Any]:
    return {
        "success": True,
        "handoff_context": {"reason": reason, "details": details},
        "handoff_summary": f"Transferring to specialist: {reason}",
    }

# tests/test_scenario_handoffs.py
import pytest
from registries.scenariostore.loader import (
    load_scenario,
    build_handoff_map_from_scenario,
    get_handoff_config,
)

def test_scenario_handoff_map():
    handoff_map = build_handoff_map_from_scenario("banking")

    assert handoff_map["handoff_fraud_agent"] == "FraudAgent"
    assert handoff_map["handoff_concierge"] == "Concierge"

def test_handoff_config_lookup():
    cfg = get_handoff_config(
        scenario_name="banking",
        from_agent="Concierge",
        tool_name="handoff_investment_advisor",
    )

    assert cfg.to_agent == "InvestmentAdvisor"
    assert cfg.type == "discrete"
    assert cfg.greet_on_switch == False

def test_announced_vs_discrete():
    # Auth handoff should be announced
    auth_cfg = get_handoff_config("banking", "Concierge", "handoff_to_auth")
    assert auth_cfg.type == "announced"
    assert auth_cfg.greet_on_switch == True

    # Investment handoff should be discrete
    invest_cfg = get_handoff_config("banking", "Concierge", "handoff_investment_advisor")
    assert invest_cfg.type == "discrete"
    assert invest_cfg.greet_on_switch == False

@dataclass
class ScenarioConfig:
    name: str                                    # Scenario identifier
    description: str = ""                        # Human-readable description
    agents: list[str] = field(...)               # Agents included (empty = all)
    start_agent: str | None = None               # Initial agent
    handoff_type: str = "announced"              # Default: "announced" or "discrete"
    handoffs: list[HandoffConfig] = field(...)   # Handoff route definitions
    agent_defaults: AgentOverride | None = None  # Global template vars
    global_template_vars: dict[str, Any] = field(...)  # Template variables

@dataclass
class HandoffConfig:
    from_agent: str = ""           # Source agent initiating handoff
    to_agent: str = ""             # Target agent receiving handoff
    tool: str = ""                 # Tool name that triggers this route
    type: str = "announced"        # "discrete" or "announced"
    share_context: bool = True     # Pass conversation context?

    @property
    def greet_on_switch(self) -> bool:
        """Returns True if type is 'announced'."""
        return self.type == "announced"

@dataclass
class HandoffContext:
    """
    Context passed during agent handoffs.

    Captures all relevant information for smooth agent transitions.
    """
    source_agent: str                           # Agent initiating the handoff
    target_agent: str                           # Agent receiving the handoff
    reason: str = ""                            # Why the handoff is occurring
    user_last_utterance: str = ""               # User's most recent speech
    context_data: Dict[str, Any] = field(...)   # Additional context (caller info)
    session_overrides: Dict[str, Any] = field(...)  # Config for new agent
    greeting: Optional[str] = None              # Explicit greeting override

    def to_system_vars(self) -> Dict[str, Any]:
        """Convert to system_vars dict for agent session application."""
        ...

@dataclass
class HandoffResult:
    """
    Result from a handoff operation.

    This is a **signal** returned by execute_handoff() that tells the
    orchestrator what to do next. The actual agent switch happens in
    the orchestrator based on this result.
    """
    success: bool                           # Whether handoff completed
    target_agent: Optional[str] = None      # Agent to switch to
    message: Optional[str] = None           # Message to speak after handoff
    error: Optional[str] = None             # Error if handoff failed
    should_interrupt: bool = True           # Cancel current TTS playback?

def sanitize_handoff_context(raw: Any) -> Dict[str, Any]:
    """
    Remove control flags from raw handoff context.

    Control flags like 'success', 'target_agent', 'handoff_summary' are
    internal signaling mechanisms and should not appear in agent prompts.
    """

def build_handoff_system_vars(
    *,
    source_agent: str,
    target_agent: str,
    tool_result: Dict[str, Any],
    tool_args: Dict[str, Any],
    current_system_vars: Dict[str, Any],
    user_last_utterance: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Build system_vars dict for agent handoff from tool result and session state.

    This shared logic ensures consistent handoff context:
    1. Extracts and sanitizes handoff_context from tool result
    2. Builds handoff_reason from multiple fallback sources
    3. Carries forward session variables (profile, client_id, etc.)
    4. Applies session_overrides if present
    """

# Control flags that are automatically removed:
_HANDOFF_CONTROL_FLAGS = frozenset({
    "success",
    "handoff",
    "target_agent",
    "message",
    "handoff_summary",
    "should_interrupt_playback",
    "session_overrides",
})