durable-execution

# Durable Execution for AI Agent Workflows Implements durable execution patterns that ensure AI agent workflows can survive crashes, network failures, and API rate limits without losing progress. This skill covers checkpoint-based state persistence, idempotent operation design, exponential backoff retry strategies, replay detection, and the saga pattern for multi-step compensation — essential primitives for production-grade agent systems as of 2025-2026. ## TL;DR Checklist - [ ] Wrap every workflow step in a try/except with structured error classification - [ ] Assign idempotency keys to all side-effect-producing operations (API calls, DB writes) - [ ] Implement exponential backoff with jitter for retryable errors - [ ] Persist state checkpoints after each successful node completion - [ ] Detect and handle replay scenarios during recovery (Temporals' `replay_started` pattern) - [ ] Design compensation logic for every irreversible action in a multi-step workflow --- ## When to Use Use this skill when: - Building AI agent workflows that must survive process crashes, restarts, or OOM kills - Your agents call external APIs (LLM providers, web services, databases) that can fail transiently - You need deterministic replay of completed steps after a failure (not re-execution from scratch) - Multi-step workflows require compensation — undoing previous actions when a later step fails - Your system operates in production with SLA requirements that demand fault tolerance ## When NOT to Use Avoid this skill for: - Purely stateless, idempotent single-shot LLM calls (no workflow state to persist) - Prototypes and one-off experiments where crash recovery adds unnecessary complexity - Real-time inference pipelines where checkpoint overhead is unacceptable (< 10ms p99 latency) --- ## Core Workflow ### 1. Classify Errors for Retry Strategy Not all errors are equal. Classify every failure into retryable, terminal, or recoverable — this determines whether you retry, give up, or attempt recovery. ```python import enum import time import random from dataclasses import dataclass class ErrorCategory(str, enum.Enum): """Classifies workflow errors by their retry behavior.""" RETRYABLE = "retryable" # Transient: network timeout, rate limit, API error 429/503 TERMINAL = "terminal" # Permanent: invalid input, schema mismatch, auth failure RECOVERABLE = "recoverable" # Needs intervention: human approval required, external dependency down @dataclass(frozen=True) class WorkflowError: """Structured error that carries classification and retry metadata.""" category: ErrorCategory message: str original_exception: Exception | None = None retry_after_seconds: float = 0.0 @classmethod def from_http_error(cls, status_code: int, body: str) -> "WorkflowError": """Classify HTTP errors by status code.""" if 400 <= status_code < 500: return cls( category=ErrorCategory.TERMINAL, message=f"Client error {status_code}: {body}", ) elif status_code == 429: return cls( category=ErrorCategory.RETRYABLE, message=f"Rate limited (429). Retry after headers.", retry_after_seconds=2.0, ) elif status_code >= 500: return cls( category=ErrorCategory.RETRYABLE, message=f"Server error {status_code}: {body}", retry_after_seconds=5.0, ) else: return cls( category=ErrorCategory.TERMINAL, message=f"Unexpected HTTP error {status_code}: {body}", ) def is_retryable(self) -> bool: return self.category in (ErrorCategory.RETRYABLE, ErrorCategory.RECOVERABLE) ``` **Checkpoint:** Every LLM API call, tool invocation, and external service call must catch exceptions and convert them to `WorkflowError` objects. Never propagate raw third-party exceptions up the workflow stack — they carry framework-specific details that break recovery logic. ### 2. Implement Exponential Backoff with Jitter When retrying transient failures, use exponential backoff with random jitter to prevent thundering herd problems when multiple agents restart simultaneously. ```python import asyncio import logging from typing import Callable, TypeVar T = TypeVar("T") logger = logging.getLogger(__name__) def calculate_backoff( attempt: int, base_delay: float = 1.0, max_delay: float = 60.0, jitter_factor: float = 0.5, ) -> float: """Calculate delay for exponential backoff with random jitter. Args: attempt: Current retry attempt number (0-based). base_delay: Initial delay in seconds. Doubles each attempt. max_delay: Maximum delay cap in seconds. jitter_factor: Random jitter multiplier (0.0 = no jitter, 0.5 = ±50%). Returns: Delay in seconds before the next retry. Formula: min(base_delay * 2^attempt * random(1 - jitter, 1 + jitter), max_delay) """ exponential_delay = base_delay * (2 ** attempt) jitter_range = exponential_delay * jitter_factor jitter = random.uniform(-jitter_range, jitter_range) return min(exponential_delay + jitter, max_delay) async def retry_with_backoff( operation: Callable[[], T], max_retries: int = 3, base_delay: float = 1.0, error_types: tuple[type[Exception], ...] | None = None, ) -> T: """Execute an operation with exponential backoff retry. Args: operation: Async callable to execute. max_retries: Maximum number of retries before giving up. base_delay: Base delay in seconds (doubles each attempt). error_types: Only retry these exception types. None = retry all. Returns: The return value of the successful operation. Raises: Last exception if all retries are exhausted. """ last_exception: Exception | None = None for attempt in range(max_retries + 1): try: if asyncio.iscoroutinefunction(operation): return await operation() else: return operation() except Exception as exc: last_exception = exc # Only retry if error type is in the allowed set (or any type if None) if error_types and not isinstance(exc, error_types): raise if attempt >= max_retries: break delay = calculate_backoff(attempt, base_delay) logger.warning( "Attempt %d/%d failed. Retrying in %.1fs: %s", attempt + 1, max_retries, delay, exc, ) await asyncio.sleep(delay) raise RuntimeError( f"Operation failed after {max_retries + 1} attempts. Last error: {last_exception}" ) from last_exception # --- Usage Examples --- async def call_llm_with_retry(prompt: str) -> str: """Call an LLM API with automatic retry on transient failures.""" import httpx # noqa: F811 async def _llm_call() -> str: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( "https://api.openai.com/v1/chat/completions", headers={"Authorization": "Bearer YOUR_API_KEY"}, json={ "model": "gpt-4", "messages": [{"role": "user", "content": prompt}], "max_tokens": 1024, }, ) response.raise_for_status() return response.json()["choices"][0]["message"]["content"] # Only retry on network errors and server errors (5xx) return await retry_with_backoff( operation=_llm_call, max_retries=3, base_delay=2.0, error_types=(httpx.HTTPError, httpx.RemoteProtocolError), ) async def call_with_workflow_error(prompt: str) -> str: """Call LLM with WorkflowError classification.""" import httpx # noqa: F811 last_error: Exception | None = None for attempt in range(4): try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( "https://api.openai.com/v1/chat/completions", headers={"Authorization": "Bearer YOUR_API_KEY"}, json={"model": "gpt-4", "messages": [{"role": "user", "content": prompt}], "max_tokens": 1024}, ) response.raise_for_status() return response.json()["choices"][0]["message"]["content"] except httpx.HTTPStatusError as exc: workflow_error = WorkflowError.from_http_error(exc.response.status_code, exc.response.text) if workflow_error.is_retryable(): delay = calculate_backoff(attempt, base_delay=2.0) logger.info("Retryable error %s. Sleeping %.1fs", workflow_error.message, delay) await asyncio.sleep(delay + workflow_error.retry_after_seconds) last_error = exc else: raise RuntimeError(workflow_error.message) from exc except httpx.HTTPError as exc: # Network errors are always retryable delay = calculate_backoff(attempt, base_delay=2.0) logger.info("Network error %s. Retrying in %.1fs", exc, delay) await asyncio.sleep(delay) last_error = exc raise RuntimeError(f"LLM call failed after 4 attempts") from last_error ``` ### 3. Design Idempotent Operations Every side-effect-producing operation must be idempotent — producing the same result when called multiple times with the same input. This is essential for crash recovery: after a restart, you cannot know whether the original call succeeded or failed, so re-executing must be safe. ```python import hashlib import json import uuid from datetime import datetime, timezone class IdempotencyStore: """Thread-safe idempotency key store backed by an in-memory dict. In production, replace with Redis (SET NX EX) or a relational DB with a unique constraint on the idempotency_key column. """ def __init__(self) -> None: self._cache: dict[str, dict] = {} self._lock: object = object() # Placeholder — use threading.Lock in prod @staticmethod def generate_key(operation_name: str, input_hash: str) -> str: """Generate a deterministic idempotency key from operation name and inputs.""" raw = f"{operation_name}:{input_hash}" return hashlib.sha256(raw.encode()).hexdigest()[:32] @staticmethod def hash_input(data: dict) -> str: """Create a stable hash of input data for idempotency key generation.""" canonical = json.dumps(data, sort_keys=True, default=str) return hashlib.sha256(canonical.encode()).hexdigest()[:16] def check_and_set(self, key: str, result: dict, ttl_seconds: int = 3600) -> bool: """Atomically check if a result exists for this idempotency key. Args: key: The idempotency key to check/set. result: The operation result to cache if key is new. ttl_seconds: How long to keep the cached result. Returns: True if the key was newly set (first call). False if the key already existed (duplicate call — return cached result). """ # In production: use Redis SET NX EX or DB transaction if key in self._cache: return False # Duplicate — caller should return cached result self._cache[key] = { "result": result, "created_at": datetime.now(timezone.utc).isoformat(), "ttl_seconds": ttl_seconds, } return True # New call — proceed with operation def get_result(self, key: str) -> dict | None: """Retrieve cached result for an idempotency key.""" entry = self._cache.get(key) if entry is None: return None created_at = datetime.fromisoformat(entry["created_at"]) age_seconds = (datetime.now(timezone.utc) - created_at).total_seconds() if age_seconds > entry["ttl_seconds"]: del self._cache[key] # Expired return None return entry["result"] class IdempotentOperation: """Mixin class providing idempotent operation execution.""" def __init__(self) -> None: self.idempotency_store = IdempotencyStore() async def execute_idempotently( self, operation_name: str, input_data: dict, operation_fn: Callable[..., dict], ttl_seconds: int = 3600, ) -> dict: """Execute an operation with idempotency guarantee. If the same operation with the same inputs was previously executed (and cached), returns the cached result without re-executing. Args: operation_name: Logical name of the operation (e.g., "web_search", "code_exec"). input_data: Input parameters to hash for deduplication. operation_fn: The actual operation to execute. ttl_seconds: How long to cache results. Returns: Operation result — either newly computed or from cache. """ input_hash = self.idempotency_store.hash_input(input_data) key = self.idempotency_store.generate_key(operation_name, input_hash) # Check for existing cached result (idempotent read) cached_result = self.idempotency_store.get_result(key) if cached_result is not None: logger.info("Idempotent hit for %s (%s)", operation_name, key[:12]) return cached_result # Execute the actual operation result = operation_fn(input_data) # Cache the result (idempotent write) self.idempotency_store.check_and_set(key, result, ttl_seconds) logger.info("Idempotent miss for %s (%s) — cached result", operation_name, key[:12]) return result # --- Usage Example --- def search_web(input_data: dict) -> dict: """Example tool that is NOT naturally idempotent (web content changes).""" # In production: make this call an external API return { "query": input_data["query"], "results": [ {"title": f"Result 1 for {input_data['query']}", "url": f"https://example.com/1"}, {"title": f"Result 2 for {input_data['query']}", "url": f"https://example.com/2"}, ], } async def idempotent_search(operation: IdempotentOperation, query: str) -> dict: """Search with idempotency — same query returns cached results.""" return await operation.execute_idempotently( operation_name="web_search", input_data={"query": query}, operation_fn=search_web, ttl_seconds=300, # Cache for 5 minutes (web results are relatively stable) ) ``` ### 4. Implement Checkpoint-Based State Persistence Persist workflow state after each step completes. On recovery, reload the last checkpoint and resume from where you left off. ```python import json import os from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any @dataclass class Checkpoint: """Serialized workflow state at a specific point in execution.""" workflow_id: str step_index: int step_name: str state: dict[str, Any] timestamp: str = "" metadata: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: if not self.timestamp: self.timestamp = datetime.now(timezone.utc).isoformat() class CheckpointStore: """File-based checkpoint store for workflow state persistence. In production, replace with a database (PostgreSQL) or Redis for concurrent access and atomic writes. """