error-recovery-retry

name	error-recovery-retry
description	Design robust error recovery, retry logic, and fallback strategies for production AI agents. Covers transient failure handling, circuit breakers, exponential backoff, state recovery, graceful degradation, and dead-letter queues for agent systems.
metadata	{"author":"cosmicstack-labs","version":"1.0.0","category":"ai-ml","tags":["error-recovery","retry-logic","circuit-breaker","fallback-strategies","fault-tolerance","graceful-degradation"]}

Error Recovery & Retry Logic for Agents

Overview

Agents fail. APIs time out. Models return garbage. Tools throw exceptions. The difference between a production-grade system and a prototype is how gracefully it fails. This skill covers comprehensive error recovery patterns — from simple retries to circuit breakers, stateful recovery, and human escalation paths.

---

Core Concepts

Failure Taxonomy

Failure Type	Example	Frequency	Recoverable?
Transient	API timeout, network glitch	Common	✅ Yes — retry
Rate Limited	429 Too Many Requests	Common	✅ Yes — backoff
Validation	Invalid tool parameters	Occasional	✅ Yes — fix and retry
Model Error	LLM returns nonsense	Occasional	⚠️ Maybe — retry with different prompt
Context Overflow	Token limit exceeded	Rare	✅ Yes — compress and retry
Permission	Agent lacks access	Rare	❌ No — escalate
Security	Injection attempt detected	Rare	❌ No — alert and block
Permanent	Tool deleted, endpoint gone	Rare	❌ No — escalate to human

Recovery Strategy Decision Tree

                    ┌──────────────┐
                    │  Agent Error  │
                    └──────┬───────┘
                           │
               ┌───────────┴───────────┐
               │                       │
          Transient?              Permanent?
               │                       │
          ┌────┴────┐            ┌──────┴──────┐
          │         │            │             │
       Retry    Circuit      Fallback      Escalate
       +backoff  Breaker     Agent         to Human

---

Step-by-Step Implementation

Step 1: Retry with Exponential Backoff

import asyncio
import random
from functools import wraps
from typing import Callable, Any

async def retry_with_backoff(
    fn: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    backoff_factor: float = 2.0,
    jitter: bool = True,
    retryable_exceptions: tuple = (TimeoutError, ConnectionError, 
                                    RateLimitError)
) -> Any:
    """Execute a function with exponential backoff retry logic."""
    last_exception = None
    
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except retryable_exceptions as e:
            last_exception = e
            
            if attempt == max_retries:
                raise  # Exhausted retries
            
            # Calculate delay with exponential backoff
            delay = min(base_delay * (backoff_factor ** attempt), max_delay)
            
            # Add jitter to prevent thundering herd
            if jitter:
                delay = delay * (0.5 + random.random() * )
            
            logger.warning(
                
                
            )
            
             asyncio.sleep(delay)
    
     last_exception  


 :
    
    
     ():
        .name = name
        .max_retries = max_retries
        .base_delay = base_delay
        .max_delay = max_delay
        .backoff_factor = backoff_factor
        .consecutive_failures = 
    
      () -> :
        
        :
            result =  retry_with_backoff(
                fn,
                max_retries=.max_retries,
                base_delay=.base_delay,
                max_delay=.max_delay,
                backoff_factor=.backoff_factor
            )
            .consecutive_failures = 
             result
         Exception  e:
            .consecutive_failures += 
            
    
     () -> :
        
         .consecutive_failures >= threshold


RETRY_POLICIES = {
    : RetryPolicy(, max_retries=, base_delay=),
    : RetryPolicy(, max_retries=, base_delay=),
    : RetryPolicy(, max_retries=, base_delay=),
    : RetryPolicy(, max_retries=, base_delay=),
}

Step 2: Circuit Breaker Pattern

class CircuitBreaker:
    """Prevent repeated calls to failing services."""
    
    STATES = ["CLOSED", "OPEN", "HALF_OPEN"]
    
    def __init__(self, name: str, failure_threshold: int = 5,
                 recovery_timeout: float = 30.0,
                 half_open_max_calls: int = 3):
        self.name = name
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        
        self.state = "CLOSED"
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0
    
    async def call(self, fn: Callable, fallback: Callable = None) -> Any:
        """Execute with circuit breaking."""
        
        if self.state == "OPEN":
            if self._should_attempt_recovery():
                self.state = "HALF_OPEN"
                self.half_open_calls = 
            :
                  ._use_fallback(fn, fallback)
        
         .state == :
             .half_open_calls >= .half_open_max_calls:
                  ._use_fallback(fn, fallback)
            .half_open_calls += 
        
        :
            result =  fn()
            ._on_success()
             result
         Exception  e:
            ._on_failure(e)
            
             .state == :
                .state = 
                .last_failure_time = time.time()
            
              ._use_fallback(fn, fallback)
    
     ():
        .failure_count = 
         .state == :
            .state = 
    
     ():
        .failure_count += 
        .last_failure_time = time.time()
        
         .failure_count >= .failure_threshold:
            .state = 
            logger.warning(
                
                
            )
    
     () -> :
          .last_failure_time:
             
        elapsed = time.time() - .last_failure_time
         elapsed >= .recovery_timeout
    
      () -> :
         fallback:
              fallback()
         CircuitBreakerOpenError()


 :
    
    
     ():
        .breakers: [, CircuitBreaker] = {}
    
     () -> CircuitBreaker:
         name   .breakers:
            .breakers[name] = CircuitBreaker(name, **kwargs)
         .breakers[name]
    
     () -> :
         {
            name: {
                : cb.state,
                : cb.failure_count,
                : cb.last_failure_time,
            }
             name, cb  .breakers.items()
        }

Step 3: Stateful Agent Recovery

class AgentStateRecovery:
    """Recover agent state after failures to resume work."""

    def __init__(self, storage):
        self.storage = storage
    
    async def checkpoint(self, agent_id: str, state: dict):
        """Save agent state at a checkpoint."""
        checkpoint = {
            "agent_id": agent_id,
            "state": state,
            "timestamp": time.time(),
            "version": state.get("_version", 0) + 1
        }
        await self.storage.set(
            f"checkpoint:{agent_id}",
            checkpoint
        )
    
    async def recover(self, agent_id: str) -> dict:
        """Restore agent state from last checkpoint."""
        checkpoint = await self.storage.get(f"checkpoint:{agent_id}")
        if not checkpoint:
            return {}  # No checkpoint, start fresh
        
        return checkpoint["state"]
    
    async def replay_from_checkpoint() -> :
        
        
        
        agent.context = checkpoint.get(, {})
        
        
        agent.memory.working_memory = checkpoint.get(, [])
        
        
        completed_steps = checkpoint.get(, [])
        
        
        plan = checkpoint.get(, [])
        remaining = [
            step  step  plan 
             step[]   completed_steps
        ]
        
          remaining:
             checkpoint.get(, )
        
        
        agent.current_plan = remaining
          agent.execute_plan()

Step 4: Graceful Degradation

class GracefulDegradation:
    """Define fallback behaviors when capabilities degrade."""
    
    def __init__(self, agent):
        self.agent = agent
        self.capability_levels = {
            "full": ["search", "analyze", "write", "execute"],
            "reduced": ["search", "analyze"],
            "minimal": ["search"],
            "fallback": []
        }
        self.current_level = "full"
    
    def degrade(self, reason: str):
        """Reduce capabilities when something fails."""
        levels = ["full", "reduced", "minimal", "fallback"]
        current_idx = levels.index(self.current_level)
        
        if current_idx < len(levels) - 1:
            self.current_level = levels[current_idx + 1]
            logger.warning(
                f"Agent {self.agent.name} degraded to {self.current_level}: {reason}"
            )
        
        # Update available tools
        self.agent.tools = [
            t for t in self.agent.tools 
             t.name  .capability_levels[.current_level]
        ]
    
      () -> :
        
        :
              fn()
         Exception  e:
            .degrade()
            
             fallback_fn:
                logger.info()
                  fallback_fn()
            
            
             {
                : ,
                : operation,
                : 
            }
    
     () -> :
         {
            : .agent.name,
            : .current_level,
            : [t.name  t  .agent.tools],
            : .current_level != 
        }

Step 5: Dead-Letter Queue for Unrecoverable Tasks

class DeadLetterQueue:
    """Handle tasks that cannot be processed after all retries."""
    
    def __init__(self, storage):
        self.storage = storage
    
    async def send(self, task: dict, error: str, 
                    retries_exhausted: bool = True):
        """Send a failed task to the dead-letter queue."""
        dlq_entry = {
            "task_id": task.get("id"),
            "original_task": task,
            "error": error,
            "retries_exhausted": retries_exhausted,
            "failed_at": time.time(),
            "status": "pending_review"
        }
        
        await self.storage.append(
            f"dlq:{datetime.now().strftime('%Y-%m-%d')}",
            dlq_entry
        )
        
        logger.error(f"Task {task.get('id')} sent to DLQ: {error}")
    
    async def replay(self, dlq_id: str, agent_executor) -> bool:
        """Replay a task from the dead-letter queue."""
        entry = await self.storage.get(f"dlq_entry:{dlq_id}")
        if  entry:
             
        
        :
            result =  agent_executor(entry[])
             .storage.(, {
                **entry,
                : ,
                : time.time(),
                : result
            })
             
         Exception  e:
             .storage.(, {
                **entry,
                : ,
                : (e),
                : entry.get(, ) + 
            })
             
    
      () -> :
        
        today = datetime.now().strftime()
        entries =  .storage.query()
        
         {
            : (entries),
            : Counter(e[]  e  entries).most_common(),
            : (  e  entries  e[] == ),
            : (  e  entries  e[] == ),
        }

Step 6: Comprehensive Error Handler

class AgentErrorHandler:
    """Central error handler for all agent failures."""
    
    def __init__(self, retry_policies: dict, circuit_breakers: CircuitBreakerRegistry,
                 dlq: DeadLetterQueue, degradation: GracefulDegradation):
        self.retry_policies = retry_policies
        self.circuit_breakers = circuit_breakers
        self.dlq = dlq
        self.degradation = degradation
    
    async def handle(self, operation: str, fn: Callable, 
                      context: dict = None) -> Any:
        """Handle an operation with full error recovery stack."""
        
        try:
            # 1. Get retry policy
            policy = self.retry_policies.get(operation, RETRY_POLICIES["tool_call"])
            
            # 2. Check circuit breaker
            cb = self.circuit_breakers.get_or_create(operation)
            
            # 3. Execute with retries and circuit breaking
            return await policy.execute(
                lambda: cb.call(fn)
            )
            
        except CircuitBreakerOpenError:
            # 4. Circuit is open — use degraded fallback
            return await self.degradation.attempt_operation(
                operation, fn
            )
            
        except Exception as e:
            
             .dlq.send(
                context  {},
                error=(e)
            )
            
            
             {
                : ,
                : operation,
                : (e),
                : 
            }

---

Recovery Configuration

YAML Configuration

# recovery-config.yaml
retry_policies:
  llm_call:
    max_retries: 3
    base_delay: 2.0
    max_delay: 30.0
    retryable_errors: [timeout, rate_limit, server_error]
  
  tool_execution:
    max_retries: 2
    base_delay: 0.5
    max_delay: 10.0
    retryable_errors: [timeout, connection_error]

circuit_breakers:
  tool_api:
    failure_threshold: 5
    recovery_timeout: 30
    half_open_max_calls: 2
  
  model_api:
    failure_threshold: 3
    recovery_timeout: 60
    half_open_max_calls: 1

fallbacks:
  search_tool:
    primary: vector_search
    fallback: keyword_search
    last_resort: return_cached_results
  
  llm_generation:
    primary: gpt-4o
    fallback: gpt-4o-mini
    last_resort: template_response

---

Trigger Phrases

Phrase	Action
"Retry that"	Retry the last failed operation
"What went wrong?"	Show error details and trace
"Check circuit breakers"	Show circuit breaker status
"Clear circuit breaker"	Manually reset a circuit breaker
"Show dead letter queue"	List unrecoverable failed tasks
"Replay from DLQ"	Retry a task from dead-letter queue
"Degrade gracefully"	Switch to reduced capability mode
"Run recovery"	Attempt state recovery from checkpoint

---

Anti-Patterns

Anti-Pattern	Why It Fails	Fix
Infinite retries	Never gives up, burns tokens	Always set max retries
No backoff	Retry instantly, overload service	Exponential backoff + jitter
Retrying permanent errors	Wastes time and tokens	Classify errors: retryable vs not
No circuit breaker	Cascade failures across system	Circuit breaker per dependency
Ignoring partial success	All-or-nothing mindset	Checkpoint partial progress
No human escalation	Tasks stuck in retry loops forever	Dead-letter queue + alert
Retry without idempotency	Duplicate side effects	Ensure tool idempotency