advancedadvancedtimeouterror-handlingproductioncircuit-breaker
Chapter 4 of 8
Solving Loop Timeout Issues: Patterns for Production AI Agents
Configurable timeouts, checkpoint/resume, circuit breakers, and kill switches for long-running workflows.
Timeouts are one of the most common failure modes in production AI agent loops. When an autonomous agent runs unbounded -- making LLM calls, invoking external tools, and validating outputs -- any single step can hang, fail silently, or consume resources indefinitely. Unlike a simple HTTP request you can retry after 30 seconds, an agent turn can span 20 tool calls and 10 minutes of wall-clock time. If the process dies at minute 9, naive designs throw away all progress.
This tutorial covers five production-tested patterns for handling timeouts in agent loops, drawing on real failures documented in the developer community and solutions implemented in open-source tools like Claude Code (github.com/anthropics/claude-code), Aider (github.com/paul-gauthier/aider, 30K+ stars), OpenHands (github.com/All-Hands-AI/OpenHands), and SWE-Agent (github.com/princeton-nlp/SWE-Agent, 15K+ stars).
For background on how agent loops work, see the Agent Loop guide.
Why Timeouts Happen: Real Failure Scenarios
| Cause | Frequency | Severity | Real-World Example |
|---|---|---|---|
| LLM API rate limiting | High | Medium | Claude 3.5 Sonnet at $3.00/1M input tokens; rate limits hit during bulk operations |
| Large context causing slow inference | High | Medium | Traditional prompt stuffing sends 20,000+ tokens but much of it is irrelevant (enterprise report) |
| External tool/API hangs | Medium | Critical | OpenHands agent calling a build server that becomes unresponsive |
| Infinite loop in agent logic | Low | Critical | Agent retries a failing tool call without a circuit breaker |
| Context overflow from blind retries | Medium | Critical | Agent retries append to context until the window is exhausted |
| Resource exhaustion (memory, CPU) | Low | Critical | MetaGPT (github.com/geekan/MetaGPT, 45K+ stars) multi-agent pipeline exhausting memory |
Pattern 1: Configurable Timeouts
The simplest and most immediate defense is applying per-operation timeouts tuned to each operation's expected latency. One-size-fits-all timeouts do not work: an LLM call might legitimately need 30 seconds, but a validation check should finish in 5 seconds.
Tools like Claude Code implement this pattern internally. When you run a task with claude --timeout 300, the CLI agent applies different timeout budgets to model inference, tool execution, and file I/O operations. Similarly, Codex CLI (github.com/openai/codex) uses configurable sandbox timeouts to prevent individual tool calls from blocking the entire agent loop.
import asyncio
from dataclasses import dataclass
from typing import Any, Callable
@dataclass
class TimeoutConfig:
"""Granular timeout settings for different loop operations."""
llm_call: float = 30.0 # Per LLM API call
tool_execution: float = 15.0 # Per external tool call
validation: float = 5.0 # Per output validation
total_iteration: float = 60.0 # Max time per full iteration
total_loop: float = 600.0 # Max time for entire loop run
async def call_with_timeout(
func: Callable,
args: tuple = (),
kwargs: dict = None,
timeout_seconds: float = 30.0,
operation_name: str = "operation"
) -> Any:
"""
Execute a function with a configurable timeout.
Raises TimeoutError if the function does not complete in time.
"""
kwargs = kwargs or {}
try:
return await asyncio.wait_for(
func(*args, **kwargs),
timeout=timeout_seconds
)
except asyncio.TimeoutError:
raise TimeoutError(
f"{operation_name} timed out after {timeout_seconds}s"
)
async def run_iteration_with_timeouts(
prompt: str,
state: dict,
config: TimeoutConfig = None
):
"""
Run a single iteration with operation-level timeouts.
Each phase gets its own timeout based on its latency profile.
"""
if config is None:
config = TimeoutConfig()
loop = asyncio.get_event_loop()
start = loop.time()
# Phase 1: LLM call with timeout
response = await call_with_timeout(
call_agent,
args=(prompt, state),
timeout_seconds=config.llm_call,
operation_name="LLM API call"
)
# Phase 2: Tool execution with timeout
tool_result = await call_with_timeout(
execute_tools,
args=(response,),
timeout_seconds=config.tool_execution,
operation_name="Tool execution"
)
# Phase 3: Validation with timeout
is_valid = await call_with_timeout(
validate_output,
args=(tool_result,),
timeout_seconds=config.validation,
operation_name="Output validation"
)
# Check total iteration budget
elapsed = loop.time() - start
if elapsed > config.total_iteration:
raise TimeoutError(
f"Iteration exceeded {config.total_iteration}s limit "
f"(actual: {elapsed:.1f}s)"
)
return response, tool_result, is_valid
# Example usage
async def run_agent_loop(task: str, max_iterations: int = 50):
"""Run an agent loop with configurable timeouts."""
config = TimeoutConfig(
llm_call=45.0,
tool_execution=20.0,
validation=5.0,
total_iteration=90.0,
total_loop=900.0 # 15-minute hard cap
)
state = {"history": [], "task": task}
loop_start = asyncio.get_event_loop().time()
for i in range(max_iterations):
elapsed_total = asyncio.get_event_loop().time() - loop_start
if elapsed_total > config.total_loop:
raise TimeoutError(
f"Total loop time exceeded {config.total_loop}s"
)
try:
response, tool_result, is_valid = await run_iteration_with_timeouts(
task, state, config
)
state["history"].append(response)
if is_valid:
return tool_result
except TimeoutError as e:
print(f"[Iteration {i}] {e}")
raise
Why this works: By giving each operation its own timeout budget, you fail fast on the right operation rather than waiting for a global timeout that is either too short (killing legitimate slow operations) or too long (letting a hung tool block everything). This is the same principle behind Cursor's Agent mode in Cursor 2.0, which runs parallel agents (up to 8) with per-agent timeouts to prevent one slow operation from stalling the entire workspace.
Pattern 2: Checkpoint and Resume
Configurable timeouts handle individual operation failures but do not protect against losing progress. If your agent loop runs 50 iterations and the process dies at iteration 47, you lose everything. The checkpoint/resume pattern saves state at meaningful points so you can resume instead of restarting.
import json
import hashlib
from pathlib import Path
from datetime import datetime
from typing import Optional
from dataclasses import dataclass, asdict
@dataclass
class Checkpoint:
"""A saved point in loop execution."""
checkpoint_id: str
iteration: int
timestamp: str
state_hash: str
loop_state: dict
partial_result: dict
class CheckpointManager:
"""Manages checkpoint creation and resumption."""
def __init__(
self,
checkpoint_dir: str = "./checkpoints",
checkpoint_interval: int = 3,
max_checkpoints: int = 20
):
self.checkpoint_dir = Path(checkpoint_dir)
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
self.checkpoint_interval = checkpoint_interval
self.max_checkpoints = max_checkpoints
def should_checkpoint(self, iteration: int) -> bool:
"""
Determine whether to save a checkpoint at this iteration.
Too-frequent checkpoints thrash storage; too-infrequent
checkpoints lose too much progress on failure.
"""
if iteration == 0:
return True
return (iteration + 1) % self.checkpoint_interval == 0
def create_checkpoint(
self,
iteration: int,
loop_state: dict,
partial_result: dict
) -> str:
"""Save current loop state as a checkpoint."""
state_json = json.dumps(loop_state, sort_keys=True)
checkpoint_id = hashlib.sha256(
f"{iteration}:{state_json}".encode()
).hexdigest()[:12]
checkpoint = Checkpoint(
checkpoint_id=checkpoint_id,
iteration=iteration,
timestamp=datetime.utcnow().isoformat(),
state_hash=state_json[:64],
loop_state=loop_state,
partial_result=partial_result
)
path = self.checkpoint_dir / f"checkpoint_{checkpoint_id}.json"
path.write_text(json.dumps(asdict(checkpoint), indent=2))
self._cleanup_old_checkpoints()
return checkpoint_id
def resume_from_latest(self) -> Optional[Checkpoint]:
"""Resume from the most recent checkpoint."""
checkpoints = sorted(
self.checkpoint_dir.glob("checkpoint_*.json")
)
if not checkpoints:
return None
latest = checkpoints[-1]
data = json.loads(latest.read_text())
print(
f"Resuming from checkpoint: {data['checkpoint_id']} "
f"(iteration {data['iteration']}, "
f"saved {data['timestamp']})"
)
return Checkpoint(**data)
def _cleanup_old_checkpoints(self):
"""Remove oldest checkpoints beyond the limit."""
checkpoints = sorted(
self.checkpoint_dir.glob("checkpoint_*.json")
)
while len(checkpoints) > self.max_checkpoints:
checkpoints[0].unlink()
checkpoints.pop(0)
# Integration with agent loop
async def run_loop_with_checkpoints(
task: str,
max_iterations: int = 50
):
"""Run a loop that periodically saves checkpoints."""
checkpoint_mgr = CheckpointManager(
checkpoint_interval=3,
max_checkpoints=20
)
# Try to resume from checkpoint
checkpoint = checkpoint_mgr.resume_from_latest()
start_iteration = checkpoint.iteration + 1 if checkpoint else 0
state = checkpoint.loop_state if checkpoint else {"history": [], "task": task}
print(
f"Starting at iteration {start_iteration}"
f" (resumed={checkpoint is not None})"
)
for i in range(start_iteration, max_iterations):
try:
result = await run_iteration(task, state)
if checkpoint_mgr.should_checkpoint(i):
checkpoint_id = checkpoint_mgr.create_checkpoint(
iteration=i,
loop_state=state,
partial_result=result
)
print(f"[Iteration {i}] Checkpoint saved: {checkpoint_id}")
state["history"].append(result)
except TimeoutError as e:
# Save emergency checkpoint before failing
checkpoint_id = checkpoint_mgr.create_checkpoint(
iteration=i,
loop_state=state,
partial_result={"error": str(e), "last_result": result}
)
print(
f"[Iteration {i}] Timeout: {e}\n"
f"Emergency checkpoint saved: {checkpoint_id}\n"
f"Resume to continue from iteration {i + 1}."
)
raise
For deeper coverage of state management, see State Persistence.
Pattern 3: Circuit Breakers
Circuit breakers prevent cascading failures by temporarily halting calls to a failing service. The pattern has three states: CLOSED (normal operation), OPEN (blocking calls), and HALF_OPEN (testing whether recovery is possible). This is critical for AI agent loops where LLM providers can experience regional outages or rate-limiting events.
Tools like Claude Code handle this at the CLI level by backing off on rate-limited API calls and retrying with exponential backoff. OpenHands implements circuit-breaker patterns in its runtime sandbox to prevent hung tool executions from blocking the agent's decision loop.
import time
from enum import Enum
from typing import Callable, Any, Optional
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitOpenError(Exception):
"""Raised when the circuit breaker is open and blocking calls."""
pass
class SemanticFailureError(Exception):
"""
Raised when an LLM call returns a semantically invalid result.
AI circuit breakers must detect not just connection errors
but also semantic failures like hallucinations and malformed outputs.
"""
pass
class CircuitBreaker:
"""
Circuit breaker for LLM API calls with semantic failure detection.
Unlike HTTP circuit breakers that only track transport errors, this
implementation counts semantic failures toward the failure threshold.
"""
def __init__(
self,
name: str,
failure_threshold: int = 3,
recovery_timeout: float = 60.0,
half_open_max_calls: int = 1,
semantic_failure_detector: Optional[Callable] = None
):
self.name = name
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.semantic_failure_detector = semantic_failure_detector
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: float = 0
def call(self, func: Callable, *args, **kwargs) -> Any:
"""Execute a function through the circuit breaker."""
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
print(f"[{self.name}] Circuit -> HALF_OPEN (testing recovery)")
else:
remaining = self.recovery_timeout - (
time.time() - self.last_failure_time
)
raise CircuitOpenError(
f"[{self.name}] Circuit is OPEN. "
f"{remaining:.0f}s remaining."
)
try:
result = func(*args, **kwargs)
if self.semantic_failure_detector and not self.semantic_failure_detector(result):
self._record_failure()
raise SemanticFailureError(
f"[{self.name}] Semantic validation failed"
)
self._record_success()
return result
except (CircuitOpenError, SemanticFailureError):
raise
except Exception as e:
self._record_failure()
raise
def _record_success(self):
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.half_open_max_calls:
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
print(f"[{self.name}] Circuit -> CLOSED (recovered)")
elif self.state == CircuitState.CLOSED:
self.failure_count = 0
def _record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
print(
f"[{self.name}] Failure recorded "
f"({self.failure_count}/{self.failure_threshold})"
)
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
elif (
self.state == CircuitState.CLOSED
and self.failure_count >= self.failure_threshold
):
self.state = CircuitState.OPEN
print(
f"[{self.name}] Circuit -> OPEN "
f"(threshold {self.failure_threshold} reached)"
)
def validate_llm_response(result: dict) -> bool:
"""Check for semantic validity of an LLM response."""
if not isinstance(result, dict):
return False
if "content" not in result or not result["content"].strip():
return False
tool_calls = result.get("tool_calls", [])
for tc in tool_calls:
if not tc.get("name") or not tc.get("arguments"):
return False
return True
class ResilientAgentCaller:
"""
Agent caller with circuit breakers and multi-provider fallback.
Falls back across providers when one is unavailable.
"""
def __init__(self):
self.circuits = {
"anthropic": CircuitBreaker(
name="anthropic",
failure_threshold=3,
recovery_timeout=60.0,
semantic_failure_detector=validate_llm_response
),
"openai": CircuitBreaker(
name="openai",
failure_threshold=3,
recovery_timeout=60.0,
semantic_failure_detector=validate_llm_response
),
}
self.fallback_order = ["anthropic", "openai"]
def call_with_fallback(self, prompt: str, **kwargs) -> dict:
"""Try providers in order, using circuit breakers."""
errors = []
for provider in self.fallback_order:
circuit = self.circuits[provider]
try:
return circuit.call(
self._call_provider,
provider,
prompt,
**kwargs
)
except (CircuitOpenError, SemanticFailureError) as e:
errors.append(f"{provider}: {e}")
print(f"[Fallback] {provider} failed, trying next...")
continue
except Exception as e:
errors.append(f"{provider}: {e}")
continue
raise RuntimeError(
f"All providers failed: {'; '.join(errors)}"
)
def _call_provider(self, provider: str, prompt: str, **kwargs) -> dict:
"""Dispatch to the appropriate provider API."""
if provider == "anthropic":
return self._call_anthropic(prompt, **kwargs)
elif provider == "openai":
return self._call_openai(prompt, **kwargs)
raise ValueError(f"Unknown provider: {provider}")
def _call_anthropic(self, prompt: str, **kwargs) -> dict:
# Actual Anthropic API call implementation
pass
def _call_openai(self, prompt: str, **kwargs) -> dict:
# Actual OpenAI API call implementation
pass
async def run_loop_with_circuit_breaker(task: str):
"""Run an agent loop protected by circuit breakers."""
caller = ResilientAgentCaller()
state = {"history": [], "task": task}
for i in range(50):
try:
response = caller.call_with_fallback(task)
state["history"].append(response)
if response.get("done", False):
return response
except RuntimeError as e:
print(f"[Iteration {i}] All providers exhausted: {e}")
raise
Why AI circuit breakers differ from HTTP circuit breakers: A traditional circuit breaker trips on connection errors and HTTP 5xx responses. An AI circuit breaker must also detect semantic failures -- responses that are technically successful HTTP calls but contain hallucinations, invalid tool call structures, or nonsensical content. Without semantic failure detection, your circuit breaker will pass through malformed LLM responses that cause downstream failures in tool execution or validation.
Pattern 4: Kill Switches
Kill switches are hard limits that act as absolute safety nets. They prevent runaway loops regardless of the timeout or circuit breaker configuration. Every production agent loop should have at least one kill switch -- ideally all three: iteration count, cost ceiling, and wall-clock time.
import time
import asyncio
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class KillSwitchConfig:
"""
Hard limits that terminate the loop regardless of other state.
These are absolute safety nets. If any kill switch triggers,
the loop stops permanently.
"""
max_iterations: int = 50
max_cost_usd: float = 5.00
max_wall_clock_seconds: float = 600.0
@dataclass
class BudgetTracker:
"""Track cumulative cost across iterations."""
total_cost_usd: float = 0.0
iteration_count: int = 0
start_time: float = field(default_factory=time.time)
def record_iteration(self, cost_usd: float):
self.total_cost_usd += cost_usd
self.iteration_count += 1
def elapsed_seconds(self) -> float:
return time.time() - self.start_time
def check_kill_switches(
self,
config: KillSwitchConfig
) -> Optional[str]:
"""
Check all kill switches. Returns a reason string if any
switch has been triggered, or None if all are safe.
"""
if self.iteration_count >= config.max_iterations:
return (
f"Kill switch: max iterations reached "
f"({self.iteration_count}/{config.max_iterations})"
)
if self.total_cost_usd >= config.max_cost_usd:
return (
f"Kill switch: budget exhausted "
f"(${self.total_cost_usd:.2f}/${config.max_cost_usd:.2f})"
)
elapsed = self.elapsed_seconds()
if elapsed >= config.max_wall_clock_seconds:
return (
f"Kill switch: wall-clock time exceeded "
f"({elapsed:.0f}s/{config.max_wall_clock_seconds:.0f}s)"
)
return None
class KillSwitchError(Exception):
"""Raised when a kill switch is triggered."""
def __init__(self, reason: str, budget: BudgetTracker):
self.reason = reason
self.budget = budget
super().__init__(reason)
async def run_loop_with_kill_switches(
task: str,
kill_switch_config: KillSwitchConfig = None,
timeout_config: TimeoutConfig = None
):
"""
Run an agent loop with all three kill switches active.
Kill switches are checked BEFORE each iteration.
"""
if kill_switch_config is None:
kill_switch_config = KillSwitchConfig()
if timeout_config is None:
timeout_config = TimeoutConfig()
budget = BudgetTracker()
state = {"history": [], "task": task}
print(
f"Loop starting with kill switches: "
f"max_iterations={kill_switch_config.max_iterations}, "
f"max_cost=${kill_switch_config.max_cost_usd:.2f}, "
f"max_time={kill_switch_config.max_wall_clock_seconds:.0f}s"
)
while True:
reason = budget.check_kill_switches(kill_switch_config)
if reason:
raise KillSwitchError(reason, budget)
try:
result = await run_iteration_with_timeouts(
task, state, timeout_config
)
iteration_cost = estimate_cost(result)
budget.record_iteration(iteration_cost)
state["history"].append(result)
print(
f"[Iteration {budget.iteration_count}] "
f"cost=${iteration_cost:.4f} "
f"total=${budget.total_cost_usd:.2f} "
f"elapsed={budget.elapsed_seconds():.0f}s"
)
if result.get("done", False):
print(f"Loop completed in {budget.iteration_count} iterations")
return result
except TimeoutError as e:
budget.record_iteration(0.0)
reason = budget.check_kill_switches(kill_switch_config)
if reason:
raise KillSwitchError(reason, budget)
print(f"[Iteration {budget.iteration_count}] Timeout: {e}")
def estimate_cost(result: dict) -> float:
"""Estimate cost based on token usage."""
usage = result.get("usage", {})
input_tokens = usage.get("input_tokens", 0)
output_tokens = usage.get("output_tokens", 0)
# Claude 3.5 Sonnet: $3/MTok input, $15/MTok output
return (input_tokens * 3e-6) + (output_tokens * 15e-6)
# Example usage
async def main():
try:
result = await run_loop_with_kill_switches(
task="Analyze the quarterly report and extract key metrics",
kill_switch_config=KillSwitchConfig(
max_iterations=100,
max_cost_usd=2.00,
max_wall_clock_seconds=300.0 # 5 minutes
)
)
print(f"Result: {result}")
except KillSwitchError as e:
print(
f"Loop terminated safely: {e.reason}\n"
f"Iterations: {e.budget.iteration_count}\n"
f"Total cost: ${e.budget.total_cost_usd:.2f}\n"
f"Wall clock: {e.budget.elapsed_seconds():.0f}s"
)
Kill switches vs. timeouts: Timeouts are per-operation limits that allow recovery and continuation. Kill switches are absolute hard limits that stop the loop permanently. You need both: timeouts for graceful degradation, kill switches for absolute safety. Claude Code enforces a similar pattern at the CLI level -- claude --max-turns 20 acts as an iteration kill switch, while per-operation timeouts handle individual call failures.
Pattern 5: Durable Execution
For production systems that must survive process crashes, container restarts, and deployment interruptions, durable execution decouples the agent loop's state from the process's in-memory state. The agent loop can be interrupted at any point and resumed from where it left off.
OpenHands (github.com/All-Hands-AI/OpenHands) implements this pattern with its runtime sandbox and event-stream architecture. Agent actions are recorded as an event stream that persists across restarts. SWE-Agent (github.com/princeton-nlp/SWE-Agent, 15K+ stars) uses a similar approach for its research-oriented software engineering tasks, where a single agent session solving a GitHub issue can run for hours and must survive interruptions.
import json
from pathlib import Path
from typing import Any, Optional
from dataclasses import dataclass, field
from enum import Enum
class StepStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class StepRecord:
"""A single recorded step in the durable execution log."""
step_id: int
step_type: str
input_data: dict
output_data: Optional[dict] = None
status: StepStatus = StepStatus.PENDING
error: Optional[str] = None
cost_usd: float = 0.0
@dataclass
class DurableLoopState:
"""Full durable state for an agent loop execution."""
execution_id: str
current_step: int = 0
total_steps: int = 0
steps: list = field(default_factory=list)
loop_state: dict = field(default_factory=dict)
is_complete: bool = False
class DurableExecutionStore:
"""
Persistent storage for durable execution state.
In production, replace file storage with a database
(PostgreSQL, SQLite, Redis, or a workflow engine's built-in store).
"""
def __init__(self, store_dir: str = "./durable_store"):
self.store_dir = Path(store_dir)
self.store_dir.mkdir(parents=True, exist_ok=True)
def save(self, state: DurableLoopState):
path = self.store_dir / f"{state.execution_id}.json"
tmp_path = path.with_suffix(".tmp")
tmp_path.write_text(json.dumps({
"execution_id": state.execution_id,
"current_step": state.current_step,
"total_steps": state.total_steps,
"steps": [
{
"step_id": s.step_id,
"step_type": s.step_type,
"input_data": s.input_data,
"output_data": s.output_data,
"status": s.status.value,
"error": s.error,
"cost_usd": s.cost_usd,
}
for s in state.steps
],
"loop_state": state.loop_state,
"is_complete": state.is_complete,
}, indent=2))
tmp_path.rename(path) # Atomic write
def load(self, execution_id: str) -> Optional[DurableLoopState]:
path = self.store_dir / f"{execution_id}.json"
if not path.exists():
return None
data = json.loads(path.read_text())
steps = []
for s in data["steps"]:
s["status"] = StepStatus(s["status"])
steps.append(StepRecord(**s))
return DurableLoopState(
execution_id=data["execution_id"],
current_step=data["current_step"],
total_steps=data["total_steps"],
steps=steps,
loop_state=data["loop_state"],
is_complete=data["is_complete"],
)
class DurableAgentLoop:
"""
An agent loop with durable execution semantics.
Every step is recorded before execution. On restart, the loop
resumes from the last incomplete step. Completed steps are
never re-executed (idempotency guarantee).
"""
def __init__(
self,
execution_id: str,
store: Optional[DurableExecutionStore] = None,
kill_switch_config: Optional[KillSwitchConfig] = None,
):
self.execution_id = execution_id
self.store = store or DurableExecutionStore()
self.kill_switch_config = kill_switch_config or KillSwitchConfig()
self.budget = BudgetTracker()
def execute_step(self, step: StepRecord) -> StepRecord:
"""Execute a single step and record the result."""
step.status = StepStatus.IN_PROGRESS
self._persist()
try:
if step.step_type == "llm_call":
step.output_data = self._execute_llm_call(step.input_data)
elif step.step_type == "tool_execution":
step.output_data = self._execute_tool(step.input_data)
elif step.step_type == "validation":
step.output_data = self._execute_validation(step.input_data)
else:
raise ValueError(f"Unknown step type: {step.step_type}")
step.status = StepStatus.COMPLETED
step.cost_usd = estimate_cost(step.output_data or {})
self.budget.record_iteration(step.cost_usd)
except Exception as e:
step.status = StepStatus.FAILED
step.error = str(e)
self._persist()
return step
def run(self, task: str) -> Any:
"""Run the full agent loop with durable execution."""
state = self.store.load(self.execution_id)
if state is None:
state = DurableLoopState(
execution_id=self.execution_id,
current_step=0,
loop_state={"task": task, "history": []},
)
state.steps = self._plan_steps(task)
state.total_steps = len(state.steps)
self._persist_state(state)
print(
f"Durable execution: resuming at step "
f"{state.current_step}/{state.total_steps}"
)
for step in state.steps[state.current_step:]:
reason = self.budget.check_kill_switches(self.kill_switch_config)
if reason:
raise KillSwitchError(reason, self.budget)
if step.status == StepStatus.COMPLETED:
continue # Skip already-completed steps (idempotency)
self.execute_step(step)
state.current_step += 1
state.loop_state["history"].append(
step.output_data or {"error": step.error}
)
self._persist_state(state)
state.is_complete = True
self._persist_state(state)
return state.loop_state
def _plan_steps(self, task: str) -> list:
"""Plan the steps for this task (simplified)."""
steps = []
step_id = 0
for i in range(self.kill_switch_config.max_iterations):
steps.append(StepRecord(
step_id=step_id,
step_type="llm_call",
input_data={"prompt": task, "iteration": i},
))
step_id += 1
steps.append(StepRecord(
step_id=step_id,
step_type="tool_execution",
input_data={"iteration": i},
))
step_id += 1
steps.append(StepRecord(
step_id=step_id,
step_type="validation",
input_data={"iteration": i},
))
step_id += 1
return steps
def _execute_llm_call(self, input_data: dict) -> dict:
return {"content": "...", "usage": {"input_tokens": 100, "output_tokens": 50}}
def _execute_tool(self, input_data: dict) -> dict:
return {"tool_result": "..."}
def _execute_validation(self, input_data: dict) -> dict:
return {"is_valid": True}
def _persist(self):
state = self.store.load(self.execution_id)
if state:
self._persist_state(state)
def _persist_state(self, state: DurableLoopState):
self.store.save(state)
# Usage: survives process restarts
def run_durable_agent(task: str, execution_id: str):
"""
Run a durable agent loop. Call this function again with the same
execution_id after a crash to resume exactly where you left off.
"""
loop = DurableAgentLoop(
execution_id=execution_id,
kill_switch_config=KillSwitchConfig(
max_iterations=50,
max_cost_usd=5.00,
max_wall_clock_seconds=600.0,
),
)
try:
result = loop.run(task)
print(f"Durable execution complete: {execution_id}")
return result
except KillSwitchError as e:
print(f"Kill switch triggered: {e.reason}")
print(f"Resume with execution_id={execution_id}")
raise
Checkpoints vs. durable execution: Checkpoints save the loop state at intervals and allow resuming from those saved points. Durable execution records every step, guarantees idempotency (steps are never re-executed), provides locking (preventing concurrent corruption), and survives arbitrary process interruptions. True durable execution requires all four guarantees: persistence, locking, idempotency, and consistency. Simple file-based checkpoints only provide persistence.
Combining Patterns in Production
For production systems, layer all five patterns together. Each layer handles a different class of failure:
Layer 5: Durable Execution (survive process death)
| Agent state lives outside process memory.
| Any step can be replayed without side effects.
|
Layer 4: Kill Switch (absolute safety net)
| Hard limits on iterations, cost, and wall-clock time.
| Prevents runaway loops regardless of other mechanisms.
|
Layer 3: Checkpoint/Resume (recover lost progress)
| Save state at meaningful intervals.
| Resume from last good state after interruption.
|
Layer 2: Configurable Timeouts (per-operation limits)
| Different timeouts for LLM calls, tools, validation.
| Fail fast on the right operation.
|
Layer 1: Circuit Breaker (fast fail on provider outage)
Detect semantic failures, not just HTTP errors.
Multi-provider fallback for resilience.
Each layer independently provides protection, and together they cover the full spectrum of timeout-related failures. This layered approach mirrors how production tools like Windsurf (windsurf.ai) implement its Cascade agent with multi-step execution, and how CrewAI (github.com/crewAIInc/crewAI) orchestrates team-based multi-agent workflows with built-in error handling and retry policies.
Configuration Best Practices
Recommended Timeout Values
| Operation Type | Recommended Timeout | Rationale |
|---|---|---|
| LLM API call | 30-45s | Models vary in latency; allow for large contexts |
| Tool execution | 10-20s | Most tools respond within seconds; some are slower |
| Output validation | 5s | Should be fast; if slow, the validation itself is a problem |
| Per iteration (total) | 60-120s | Sum of operation timeouts plus overhead |
| Per loop (total) | 300-900s | Depends on task complexity and budget |
| Circuit breaker recovery | 30-120s | Time before testing if provider recovered |
| Kill switch wall-clock | 300-3600s | Hard maximum; should match your SLA |
Production Readiness Checklist
- Per-operation timeouts configured for all LLM, tool, and validation calls
- Circuit breaker with semantic failure detection (not just HTTP errors)
- Multi-provider fallback configured for all critical LLM calls
- Checkpoint/resume with appropriate frequency (not every tick, not every 100 iterations)
- Emergency checkpoint saved on any unhandled exception
- Kill switches for max iterations, max cost, and max wall-clock time
- Cost tracking with real token counting (not estimates)
- Durable execution for state persistence across process restarts
- Idempotency guarantees so replayed steps do not duplicate side effects
- Logging for all timeout events, circuit breaker transitions, and kill switch triggers
- Alerts configured for circuit breaker OPEN state and kill switch triggers
- Runbook for manual intervention when automated recovery is insufficient
Next Steps
- Token Overload Prevention -- Prevent context window exhaustion that causes slow inference and timeouts
- Production Loop Standards -- Full production readiness checklist and architectural standards
- Auto-Correction Loop -- Automatic recovery mechanisms that complement timeout handling
- Custom Loop Script in Python -- Build a complete agent loop with all resilience patterns