# Delegation & Handoff

Patterns for transferring work between agents: task decomposition, context transfer, and seamless handoffs.

---

## Delegation vs Handoff

| Aspect | Delegation | Handoff |
|---|---|---|
| **Initiator** | Manager/Orchestrator | Current agent |
| **Relationship** | Hierarchical | Peer or hierarchical |
| **Control** | Delegator retains oversight | Control transfers |
| **Return** | Expected (result back) | May not return |
| **Context** | Partial (task-specific) | Full (conversation + state) |

---

## Delegation Patterns

### 1. Explicit Delegation (Manager-Initiated)
```python
class DelegationManager:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
    
    async def delegate(self, task: Task, assignee: str, 
                      context: DelegationContext) -> Result:
        agent = self.agents[assignee]
        
        # Prepare delegation package
        delegation = DelegationPackage(
            task=task,
            context=context,
            delegator=self.manager_id,
            assignee=assignee,
            deadline=context.deadline,
            priority=context.priority,
            expected_output=context.expected_output,
            constraints=context.constraints
        )
        
        # Execute
        result = await agent.execute_delegation(delegation)
        
        # Validate
        if not self.validate_result(delegation, result):
            raise DelegationFailed("Result validation failed")
        
        return result
```

### 2. Capability-Based Delegation
```python
class CapabilityBasedDelegation:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
        self.capability_index = self._build_capability_index()
    
    def _build_capability_index(self) -> dict[str, list[str]]:
        """Map capability → list of agent IDs"""
        index = defaultdict(list)
        for agent_id, agent in self.agents.items():
            for cap in agent.capabilities:
                index[cap].append(agent_id)
        return index
    
    def find_best_agent(self, required_capabilities: list[str]) -> str | None:
        candidates = set()
        for cap in required_capabilities:
            agents = set(self.capability_index.get(cap, []))
            if candidates:
                candidates &= agents
            else:
                candidates = agents
        
        if not candidates:
            return None
        
        # Select best (least loaded, highest skill)
        return min(candidates, key=lambda a: self.agents[a].current_load)
```

### 3. Recursive Delegation
```python
class RecursiveDelegation:
    def __init__(self, agents: dict[str, Agent], max_depth: int = 5):
        self.agents = agents
        self.max_depth = max_depth
    
    async def delegate_recursive(self, task: Task, agent_id: str, 
                                 depth: int = 0) -> Result:
        if depth >= self.max_depth:
            raise MaxDelegationDepthExceeded()
        
        agent = self.agents[agent_id]
        
        # Agent decides: execute or sub-delegate
        decision = await agent.decide_delegation(task)
        
        if decision.action == "execute":
            return await agent.execute(task)
        
        elif decision.action == "delegate":
            # Sub-delegate to specialist
            subtask = decision.subtask
            specialist = decision.specialist
            return await self.delegate_recursive(subtask, specialist, depth + 1)
        
        elif decision.action == "decompose":
            # Break into parallel subtasks
            subtasks = decision.subtasks
            results = await asyncio.gather(*[
                self.delegate_recursive(st.task, st.assignee, depth + 1)
                for st in subtasks
            ])
            return self.merge_results(results)
```

---

## Handoff Patterns

### 1. Context-Preserving Handoff
```python
class HandoffPackage:
    def __init__(self, 
                 conversation_history: list[Message],
                 current_state: AgentState,
                 active_tools: list[Tool],
                 active_goal: Goal,
                 pending_actions: list[Action],
                 metadata: dict):
        self.conversation_history = conversation_history
        self.current_state = current_state
        self.active_tools = active_tools
        self.active_goal = active_goal
        self.pending_actions = pending_actions
        self.metadata = metadata
        self.handoff_id = str(uuid4())
        self.timestamp = datetime.utcnow()
        self.from_agent = None
        self.to_agent = None

class HandoffManager:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
    
    async def handoff(self, from_agent: str, to_agent: str, 
                      reason: HandoffReason) -> HandoffResult:
        from_agent_obj = self.agents[from_agent]
        to_agent_obj = self.agents[to_agent]
        
        # 1. Prepare handoff package
        package = await from_agent_obj.prepare_handoff(reason)
        package.from_agent = from_agent
        package.to_agent = to_agent
        
        # 2. Validate package completeness
        if not self.validate_package(package):
            raise IncompleteHandoffPackage()
        
        # 3. Execute handoff
        await to_agent_obj.receive_handoff(package)
        
        # 3. Confirm
        confirmation = await to_agent_obj.confirm_handoff(package.handoff_id)
        
        return HandoffResult(
            handoff_id=package.handoff_id,
            success=True,
            new_agent=to_agent,
            context_preserved=True
        )
```

### 2. Streaming Handoff (Live Transfer)
```python
class StreamingHandoff:
    def __init__(self, from_agent: Agent, to_agent: Agent):
        self.from_agent = from_agent
        self.to_agent = to_agent
        self.stream_queue = asyncio.Queue()
    
    async def start_streaming_handoff(self):
        # 1. Start streaming state
        asyncio.create_task(self._stream_state())
        
        # 2. Transfer control
        await self.to_agent.prepare_for_handoff()
        
        # 3. Stream events
        await self.stream_queue.put(HandoffEvent.START)
        
        while not self.handoff_complete:
            event = await self.from_agent.get_next_event()
            await self.stream_queue.put(event)
            
            if event.type == "tool_result" and event.is_final:
                self.handoff_complete = True
        
        await self.stream_queue.put(HandoffEvent.COMPLETE)
    
    async def _stream_state(self):
        # Stream memory, context, tools incrementally
        for chunk in self.from_agent.stream_state(chunk_size=1000):
            await self.to_agent.receive_state_chunk(chunk)
```

### 3. Human-Initiated Handoff
```python
class HumanHandoffManager:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
    
    async def request_human_handoff(self, agent_id: str, 
                                    reason: str, 
                                    urgency: Urgency) -> HandoffResult:
        agent = self.agents[agent_id]
        
        # 1. Prepare summary for human
        summary = await agent.prepare_human_summary()
        
        # 2. Notify human (Slack, email, dashboard)
        notification = HandoffNotification(
            agent_id=agent_id,
            reason=reason,
            urgency=urgency,
            summary=summary,
            conversation_link=f"/conversations/{agent.conversation_id}"
        )
        await self.notify_human(notification)
        
        # 3. Wait for human response
        response = await self.wait_for_human_response(
            timeout=300 if urgency == Urgency.NORMAL else 60
        )
        
        if response.action == "take_over":
            # Human takes over directly
            return await self.human_takes_over(agent, response)
        elif response.action == "reassign":
            # Reassign to another agent
            return await self.reassign_agent(agent, response.target_agent)
        else:
            # Continue with agent
            return HandoffResult(continued=True)
```

---

## OpenAI Agents SDK Handoffs

### Built-in Handoffs
```python
from agents import Agent, Runner, handoff

# Define specialist agents
billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing inquiries, refunds, subscriptions.",
    tools=[refund_tool, subscription_tool]
)

technical_agent = Agent(
    name="Technical Support",
    instructions="Handle technical issues, bugs, integrations.",
    tools=[diagnose_tool, logs_tool]
)

# Triage agent with handoffs
triage_agent = Agent(
    name="Triage",
    instructions="Route to appropriate specialist.",
    handoffs=[billing_agent, technical_agent]
)

# Automatic handoff based on LLM decision
result = await Runner.run(triage_agent, "I was charged twice!")
# LLM decides: "This is a billing issue" → handoffs to billing_agent
```

### Custom Handoff Configuration
```python
from agents import handoff

# Custom handoff with input transformation
billing_handoff = handoff(
    agent=billing_agent,
    tool_name_override="transfer_to_billing",
    tool_description_override="Transfer to billing for refunds, charges, subscriptions",
    on_handoff=lambda ctx: log_handoff("billing", ctx)
)

# Input filter for handoff
def filter_billing_context(ctx):
    return {
        "issue": ctx.context.get("issue"),
        "account_id": ctx.context.get("account_id"),
        "amount": ctx.context.get("amount")
    }

billing_handoff = handoff(
    agent=billing_agent,
    input_filter=filter_billing_context
)
```

### Handoff Patterns in SDK
```python
# 1. Explicit handoff tool
result = await Runner.run(triage_agent, "Help with my bill")
# Agent calls: transfer_to_billing(issue="double charge", account_id="123")

# 2. Automatic handoff via instructions
# "If billing question, handoff to billing_agent"

# 3. Chained handoffs
# triage → billing → refund_specialist

# 4. Conditional handoff
# "If amount > $1000, handoff to senior_billing"
```

---

## LangGraph Handoffs

### Subgraph Handoffs
```python
from langgraph.graph import StateGraph, START, END
from langgraph.graph import send

# Subgraph as specialist
research_subgraph = StateGraph(ResearchState)
research_subgraph.add_node("researcher", researcher_node)
research_subgraph.add_node("analyst", analyst_node)
research_subgraph.add_edge(START, "researcher")
research_subgraph.add_edge("researcher", "analyst")
research_subgraph.add_edge("analyst", END)
compiled_research = research_subgraph.compile()

# Main graph with handoff
main_graph = StateGraph(State)
main_graph.add_node("orchestrator", orchestrator_node)
main_graph.add_node("research_team", compiled_research)  # Handoff to subgraph
main_graph.add_node("writer", writer_node)

main_graph.add_edge(START, "orchestrator")
main_graph.add_conditional_edges("orchestrator", route_to_team)
main_graph.add_edge("research_team", "writer")
main_graph.add_edge("writer", END)
```

### Explicit Handoff with `send()`
```python
from langgraph.graph import send

def orchestrator_node(state):
    # Decide which team handles this
    if state["task_type"] == "research":
        # Handoff to research subgraph
        return send("research_team", {"topic": state["topic"], "depth": "deep"})
    elif state["task_type"] == "coding":
        return send("coding_team", {"spec": state["spec"]})
    else:
        return {"next": "writer"}

# Inter-agent communication via send()
def researcher_node(state):
    # Can send to another agent within graph
    if need_fact_check:
        return send("fact_checker", {"claim": state["claim"]})
    return {"findings": findings}
```

---

## Handoff Best Practices

### 1. Context Preservation Checklist
- [ ] Full conversation history (or compressed summary)
- [ ] Current agent state (variables, memory, tools)
- [ ] Active goal and sub-goals
- [ ] Pending actions and tool calls
- [ ] Tool configurations and auth state
- [ ] User preferences and context
- [ ] Error history and recovery state

### 2. Handoff Validation
```python
def validate_handoff(package: HandoffPackage) -> ValidationResult:
    errors = []
    
    # Required fields
    if not package.conversation_history:
        errors.append("Missing conversation history")
    if not package.active_goal:
        errors.append("Missing active goal")
    if not package.current_state:
        errors.append("Missing agent state")
    
    # Context continuity
    if package.conversation_history:
        last_msg = package.conversation_history[-1]
        if last_msg.role == "assistant" and not last_msg.tool_calls:
            errors.append("Last message appears incomplete")
    
    # Tool continuity
    if package.active_tools:
        for tool in package.active_tools:
            if tool.requires_auth and not tool.auth_valid:
                errors.append(f"Tool {tool.name} auth expired")
    
    return ValidationResult(valid=len(errors) == 0, errors=errors)
```

### 3. Graceful Degradation
```python
async def robust_handoff(from_agent: Agent, to_agent: Agent, 
                         max_retries: int = 3) -> HandoffResult:
    for attempt in range(max_retries):
        try:
            return await handoff_manager.handoff(from_agent, to_agent)
        except HandoffError as e:
            if attempt == max_retries - 1:
                raise
            
            # Degraded handoff: minimal context
            minimal_package = from_agent.prepare_minimal_handoff()
            await to_agent.receive_handoff(minimal_package)
            logger.warning(f"Degraded handoff on attempt {attempt + 1}")
    
    raise HandoffFailed("All handoff attempts failed")
```

---

## Handoff Patterns in Frameworks

| Framework | Mechanism | Features |
|---|---|---|
| **OpenAI Agents SDK** | `handoff()` function | Input filters, `on_handoff` callbacks, chaining |
| **LangGraph** | `send()`, subgraphs | Streaming, interrupts, human-in-loop |
| **AutoGen** | `initiate_chat()`, GroupChat | Nested chats, speaker selection |
| **CrewAI** | Process delegation | Sequential/hierarchical, manager agent |
| **OpenAI Agents SDK** | Handoff tools | Input filters, `on_handoff` hooks |
| **PydanticAI** | Agent as tool | Structured handoff with validation |

---

## Handoff Anti-Patterns

| Anti-Pattern | Problem | Fix |
|---|---|---|
| **Context loss** | New agent lacks history | Preserve full context or summary |
| **Tool mismatch** | New agent lacks tools | Ensure tool parity or document gaps |
| **Auth loss** | Tokens not transferred | Include auth in handoff package |
| **Goal drift** | New agent pursues different goal | Explicit goal transfer |
| **Infinite handoffs** | Ping-pong between agents | Max handoff depth, circuit breaker |
| **Silent failure** | Handoff fails silently | Explicit confirmation, timeouts |

---

## Related Notes

- [[MultiAgent/Communication-Protocols.md]] — Messages for handoff
- [[MultiAgent/Coordination-Patterns.md]] — Delegation in coordination
- [[MultiAgent/Negotiation-Consensus.md]] — Handoff as negotiation
- [[MultiAgent/Safety-Coordination.md]] — Safe handoffs
- [[Frameworks/OpenAI-Agents-SDK.md]] — SDK handoffs
- [[Frameworks/LangGraph.md]] — LangGraph `send()` and subgraphs
- [[Frameworks/AutoGen.md]] — AutoGen nested chats
- [[Frameworks/CrewAI.md]] — CrewAI delegation
- [[Frameworks/OpenAI-Agents-SDK.md]] — SDK handoffs
- [[Architectures/Control-Flow.md]] — Handoff as control flow
- [[Architectures/Human-in-the-Loop.md]] — Human handoffs
- [[Production/Reliability.md]] — Handoff reliability
- [[Frontiers/Agent-Computer-Interface.md]] — Desktop handoff

---

*Sources: OpenAI Agents SDK docs, LangGraph `send()` docs, AutoGen GroupChat, CrewAI processes, FIPA ACL handoff, delegation patterns, handoff patterns in microservices, human-agent interaction research.*