Avoiding Circular Delegation: Deadlocks in Multi-Agent Handoffs

When agents delegate tasks to each other, they create a web of dependencies. Without proper safeguards, this web can become a trap—agents endlessly passing tasks back and forth in circular patterns that never converge on a solution.

The Circular Delegation Problem

Consider a simple scenario:

1. Agent A (Data Analyst) receives a request to analyze customer sentiment
2. Agent A delegates data validation to Agent B
3. Agent B (Validator) finds issues and delegates data cleaning to Agent C
4. Agent C (Cleaner) needs business rules and delegates to Agent A
5. Agent A receives the task and delegates to Agent B again…

The cycle continues indefinitely, consuming tokens, time, and resources without producing results.

Why Circles Form

Circular delegation emerges from several common patterns:

1. Ambiguous Responsibilities: When agent boundaries overlap, tasks bounce between them
2. Missing Termination Conditions: Agents don’t know when to stop or escalate
3. Recursive “Helper” Behavior: Agents designed to be helpful keep delegating instead of completing
4. Lack of Global Context: Each agent sees only its local view, not the full delegation chain

Prevention Strategies

1. Hard Hop Limits (Time-to-Live)

The simplest defense is a maximum delegation depth:

from dataclasses import dataclass
from typing import List, Optional

@dataclass
class DelegationEnvelope:
    task_id: str
    originator: str  # Human or root agent
    delegation_chain: List[str]  # Agent IDs in order
    hop_count: int
    max_hops: int = 5
    
    def can_delegate(self) -> bool:
        return self.hop_count < self.max_hops
    
    def add_hop(self, agent_id: str) -> 'DelegationEnvelope':
        if not self.can_delegate():
            raise DelegationLimitExceeded(
                f"Maximum hop count {self.max_hops} exceeded"
            )
        return DelegationEnvelope(
            task_id=self.task_id,
            originator=self.originator,
            delegation_chain=self.delegation_chain + [agent_id],
            hop_count=self.hop_count + 1,
            max_hops=self.max_hops
        )

In the A2A protocol, this is implemented via the hop_count field in the message envelope. Each agent increments this count, and requests are rejected when the limit is exceeded.

2. Cycle Detection

Track which agents have already processed a task:

class CycleDetector:
    def __init__(self):
        self.task_history: Dict[str, Set[str]] = {}
        
    def check_and_record(self, task_id: str, agent_id: str) -> bool:
        """Returns True if cycle detected."""
        if task_id not in self.task_history:
            self.task_history[task_id] = set()
            
        if agent_id in self.task_history[task_id]:
            return True  # Cycle detected
            
        self.task_history[task_id].add(agent_id)
        return False
    
    def get_chain(self, task_id: str) -> List[str]:
        """Return the delegation chain for a task."""
        return list(self.task_history.get(task_id, set()))

The delegation chain specification from the Agentic Control Plane project defines this formally:

Cycle prevention: An agent MUST NOT delegate to a profile already in the chain. No mutual recursion, no infinite loops.

3. Mandatory Final States

Every task must have explicit termination conditions:

from enum import Enum

class TaskState(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    NEEDS_HUMAN = "needs_human"
    CYCLE_DETECTED = "cycle_detected"

class AgentTask:
    def __init__(self, task_id: str, description: str):
        self.task_id = task_id
        self.description = description
        self.state = TaskState.PENDING
        self.attempts = 0
        self.max_attempts = 3
        
    def should_terminate(self) -> bool:
        """Check if task should terminate regardless of completion."""
        return (
            self.state in [TaskState.COMPLETED, TaskState.FAILED, 
                          TaskState.NEEDS_HUMAN, TaskState.CYCLE_DETECTED] or
            self.attempts >= self.max_attempts
        )

Include termination signals in agent prompts:

TERMINATION_PROMPT = """
Upon successful completion, respond with 'TASK_COMPLETED' followed by the result.
If unable to complete after {max_attempts} attempts, respond with 'TASK_FAILED' with the reason.
If you detect a delegation cycle, respond with 'CYCLE_DETECTED'.
"""

4. Circuit Breakers on Handoffs

Monitor delegation patterns and trip circuit breakers when cycles form:

class DelegationCircuitBreaker:
    def __init__(self, failure_threshold: int = 3, 
                 recovery_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.consecutive_failures = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
        
    def record_handoff(self, from_agent: str, to_agent: str, 
                      success: bool):
        if not success:
            self.consecutive_failures += 1
            self.last_failure_time = time.time()
            
            if self.consecutive_failures >= self.failure_threshold:
                self.state = "open"
        else:
            self.consecutive_failures = 0
            self.state = "closed"
    
    def allow_handoff(self, from_agent: str, to_agent: str) -> bool:
        if self.state == "closed":
            return True
            
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
                return True
            return False
            
        # half-open: allow one test handoff
        return True

5. Delegation Chain Propagation

The A2A protocol implements this through delegation chains in the message envelope:

@dataclass
class A2AMessage:
    task_id: str
    context_id: str
    delegation_chain: List[DelegationHop]
    max_hops: int = 5
    
    def add_delegation(self, agent_id: str, 
                      permissions: Set[str]) -> 'A2AMessage':
        # Check for cycles
        existing_agents = {hop.agent_id for hop in self.delegation_chain}
        if agent_id in existing_agents:
            raise CycleDetectedError(
                f"Agent {agent_id} already in chain: "
                f"{[hop.agent_id for hop in self.delegation_chain]}"
            )
        
        # Check hop limit
        if len(self.delegation_chain) >= self.max_hops:
            raise HopLimitExceededError(
                f"Max hops {self.max_hops} reached"
            )
        
        # Calculate permissions (intersection = most restrictive wins)
        parent_permissions = self.delegation_chain[-1].permissions if self.delegation_chain else permissions
        child_permissions = parent_permissions.intersection(permissions)
        
        new_hop = DelegationHop(
            agent_id=agent_id,
            permissions=child_permissions,
            timestamp=datetime.utcnow()
        )
        
        return A2AMessage(
            task_id=self.task_id,
            context_id=self.context_id,
            delegation_chain=self.delegation_chain + [new_hop],
            max_hops=self.max_hops
        )

Detection and Monitoring

Tracing Delegation Patterns

Log and visualize delegation chains to detect cycles early:

import logging
from typing import Dict, List

class DelegationTracer:
    def __init__(self):
        self.logger = logging.getLogger("delegation_tracer")
        self.chains: Dict[str, List[str]] = {}
        
    def record_delegation(self, task_id: str, from_agent: str, 
                         to_agent: str):
        if task_id not in self.chains:
            self.chains[task_id] = []
        
        chain = self.chains[task_id]
        chain.append(f"{from_agent} -> {to_agent}")
        
        # Check for repeated patterns
        if self._detect_repetition(chain):
            self.logger.warning(
                f"Potential cycle detected in task {task_id}: "
                f"Chain: {' -> '.join(chain)}"
            )
    
    def _detect_repetition(self, chain: List[str]) -> bool:
        """Detect if the same delegation pattern repeats."""
        if len(chain) < 4:
            return False
            
        # Check for A -> B -> A pattern
        last_three = chain[-3:]
        if len(set(last_three)) < 3:
            return True
            
        return False

Metrics to Monitor

Track these metrics to identify problematic patterns:

class DelegationMetrics:
    def __init__(self):
        self.metrics = {
            "total_delegations": 0,
            "cycles_detected": 0,
            "hop_limit_exceeded": 0,
            "avg_chain_length": 0,
            "max_chain_length": 0
        }
        
    def record_delegation(self, chain_length: int, cycle_detected: bool):
        self.metrics["total_delegations"] += 1
        
        if cycle_detected:
            self.metrics["cycles_detected"] += 1
            
        if chain_length > self.metrics["max_chain_length"]:
            self.metrics["max_chain_length"] = chain_length
            
        # Update average
        n = self.metrics["total_delegations"]
        self.metrics["avg_chain_length"] = (
            (self.metrics["avg_chain_length"] * (n - 1) + chain_length) / n
        )

Best Practices for Production

1. Design Clear Agent Boundaries

• Each agent should have a single, well-defined responsibility
• Avoid agents that “help with everything”
• Document what each agent does and does not handle

2. Implement Graceful Degradation

When a cycle is detected, don’t fail silently:

async def handle_cycle_detected(task_id: str, chain: List[str]):
    # Log the cycle for debugging
    logger.error(f"Cycle detected in task {task_id}: {chain}")
    
    # Escalate to human operator
    await escalation_service.notify(
        task_id=task_id,
        issue="delegation_cycle",
        chain=chain,
        message="Multi-agent delegation cycle detected. Human intervention required."
    )
    
    # Return partial results if available
    return DegradedResult(
        status="cycle_detected",
        partial_results=get_partial_results(task_id),
        escalation_required=True
    )

3. Set Appropriate Limits

Parameter	Recommended Value	Rationale
Max hops	5-7	Balances flexibility with safety
Cycle detection window	Last 3-4 delegations	Catches immediate loops
Circuit breaker threshold	3 failures	Prevents rapid-fire retries
Recovery timeout	30-60 seconds	Allows temporary issues to resolve

4. Test for Cycles

Include cycle scenarios in your test suite:

async def test_cycle_detection():
    # Create a circular delegation scenario
    agent_a = MockAgent("A", delegates_to=["B"])
    agent_b = MockAgent("B", delegates_to=["A"])
    
    task = AgentTask("test-1", "Test task")
    
    with pytest.raises(CycleDetectedError):
        await run_delegation_chain(
            task=task,
            agents={"A": agent_a, "B": agent_b},
            max_hops=10
        )

async def test_hop_limit_enforcement():
    # Create a linear chain that exceeds limit
    agents = {f"agent_{i}": MockAgent(f"agent_{i}", 
              delegates_to=[f"agent_{i+1}"]) 
              for i in range(10)}
    
    task = AgentTask("test-2", "Long chain task")
    
    with pytest.raises(HopLimitExceededError):
        await run_delegation_chain(
            task=task,
            agents=agents,
            max_hops=5
        )

Conclusion

Circular delegation is a real risk in multi-agent systems, but it’s preventable with proper safeguards:

1. Hard limits on delegation depth (hop counts)
2. Cycle detection by tracking agent participation
3. Clear termination conditions in agent prompts
4. Circuit breakers to halt problematic patterns
5. Transparent escalation when cycles are detected

The key insight is that prevention is better than detection. Design your agents with clear boundaries, implement delegation limits from the start, and monitor for patterns that indicate emerging cycles.

In production systems, the cost of an undetected cycle can be enormous—wasted tokens, degraded performance, and frustrated users. By implementing these strategies, you build resilient multi-agent systems that deliver results instead of spinning in circles.