Production Loop Standards

Overview

Moving an agent loop from prototype to production requires systematic attention to reliability, observability, and operational safety. A production-ready loop is one that can run continuously without human intervention, degrade gracefully under failure, and provide sufficient visibility for operators to diagnose and resolve issues.

What Makes a Loop "Production-Ready"

A production-ready loop satisfies these core requirements:

• Reliability: Handles errors predictably and recovers automatically -- not through blind retries, but through classified error handling
• Observability: Exposes metrics, logs, and traces for real-time monitoring of loop health
• Efficiency: Uses resources within defined budgets; at Claude 3.5 Sonnet pricing ($3.00/1M input tokens, per Anthropic), unbounded loops can become expensive rapidly
• Testability: Can be validated through automated tests before deployment
• Operability: Can be configured, deployed, and maintained without code changes

Real-World Production Loops

Several production tools demonstrate these standards in practice:

Monitoring and Observability

Monitoring is the foundation of production operations. Every loop should expose the following signals:

Essential Metrics

Metric	Type	Description	Alert Threshold
loop_iterations_total	Counter	Total iterations executed	N/A
loop_duration_seconds	Histogram	Wall-clock time per iteration	p99 > 60s
loop_errors_total	Counter	Total errors by type	Any spike
loop_convergence_rate	Gauge	% of tasks reaching quality threshold	< 90%
loop_active_tasks	Gauge	Currently executing tasks	> max_capacity
loop_tokens_used	Gauge	Token consumption per iteration	> budget per task

Structured Logging

import structlog

logger = structlog.get_logger()

def log_iteration(task_id: str, iteration: int, result: IterationResult):
    logger.info("loop.iteration.completed",
        task_id=task_id,
        iteration=iteration,
        score=result.score,
        passed=result.passed,
        action=result.action,
        duration_ms=result.duration_ms,
        tokens_used=result.tokens_used,
        error=result.error_type if result.error else None
    )

Distributed Tracing

For nested and multi-agent loops, use distributed tracing to follow request paths across loop boundaries. This is essential when using frameworks like LangGraph (github.com/langchain-ai/langgraph) for graph-based orchestration, where a single task may traverse multiple agent nodes:

from opentelemetry import trace

tracer = trace.get_tracer("loop-engineering")

class TracedLoop:
    def run(self, task: Task) -> Result:
        with tracer.start_as_current_span("loop.run") as span:
            span.set_attribute("task.id", task.id)
            span.set_attribute("task.type", task.type)
            result = self._execute(task)
            span.set_attribute("result.score", result.score)
            span.set_attribute("result.iterations", result.iterations)
            return result

Error Handling and Graceful Degradation

Production loops must handle failures at every layer. emphasizes that classified error handling is critical: not all errors warrant the same response, and blind retries amplify failures rather than resolving them.

Degradation Hierarchy

Level 0: Full Operation     -- All components healthy
Level 1: Reduced Quality    -- Skip optional enhancement loops
Level 2: Fallback Model     -- Use lighter/cheaper model if primary fails
Level 3: Cached Response    -- Return cached/precomputed results
Level 4: Fail Open          -- Return best-effort result with warning
Level 5: Fail Closed        -- Reject request with clear error message

Implementation with Classified Errors

The key insight from production failure analysis is that errors must be classified before retry. A RateLimitError warrants a backoff-and-retry; a ContextOverflowError warrants truncation and restart, not a retry of the same oversized context; a SchemaValidationError warrants immediate feedback to the agent, not a retry:

class ProductionLoop:
    def run_with_degradation(self, task: Task) -> Result:
        try:
            return self.primary_loop.run(task)
        except RateLimitError:
            logger.warning("Rate limited, backing off")
            time.sleep(exponential_backoff(self.retry_count))
            return self.primary_loop.run(task)
        except ContextOverflowError:
            logger.warning("Context overflow, truncating history")
            task.context = self._truncate_context(task.context)
            return self.primary_loop.run(task)
        except PrimaryModelError:
            logger.warning("Primary model failed, falling back to secondary")
            try:
                return self.secondary_loop.run(task)
            except SecondaryModelError:
                if self.cache.has(task.id):
                    logger.warning("Using cached result")
                    return self.cache.get(task.id)
                return Result(status="failed", error="all_models_unavailable")
        except TimeoutError:
            return Result(status="degraded", output=self.best_partial_result)

Circuit Breaker Pattern

Implement circuit breakers around all external dependencies. This prevents cascading failures -- a critical pattern when using multi-agent frameworks like CrewAI (github.com/crewAIInc/crewAI) or AutoGen (github.com/microsoft/autogen), where one agent's failure can stall an entire pipeline:

from pybreaker import CircuitBreaker

model_breaker = CircuitBreaker(fail_max=3, reset_timeout=60)

@model_breaker
def call_model(prompt: str, model: str) -> str:
    return client.messages.create(model=model, messages=[{"role": "user", "content": prompt}])

Resource Limits and Timeout Management

Without resource limits, a misbehaving loop can consume unbounded resources. Reports indicate that traditional prompt-stuffing approaches routinely send 20,000+ tokens of which much is irrelevant -- a direct consequence of uncontrolled context growth in iterative loops.

Timeout Configuration

Define timeouts at every level. Claude Code itself enforces timeout guards through its session configuration; similarly, Cursor in Agent mode bounds task execution to prevent runaway parallel agent spawns:

@dataclass
class LoopTimeouts:
    iteration_timeout: float = 30.0      # Max time per iteration
    total_timeout: float = 300.0          # Max time for entire loop
    subtask_timeout: float = 10.0         # Max time per subtask (nested)
    api_call_timeout: float = 5.0          # Max time per external API call
    queue_wait_timeout: float = 60.0       # Max time waiting in queue

class TimeoutGuard:
    def __init__(self, timeouts: LoopTimeouts):
        self.timeouts = timeouts

    def run_with_timeout(self, func, timeout_key: str):
        timeout = getattr(self.timeouts, timeout_key)
        return asyncio.wait_for(func(), timeout=timeout)

Token Budget Enforcement

@dataclass
class ResourceBudget:
    max_tokens_per_task: int = 50_000
    max_tokens_per_iteration: int = 10_000
    max_memory_mb: int = 512
    max_concurrent_tasks: int = 10

class BudgetEnforcer:
    def check(self, usage: ResourceUsage, budget: ResourceBudget) -> bool:
        if usage.tokens_total > budget.max_tokens_per_task:
            raise TokenBudgetExceeded(usage.tokens_total, budget.max_tokens_per_task)
        if usage.concurrent_tasks > budget.max_concurrent_tasks:
            raise ConcurrencyLimitExceeded(usage.concurrent_tasks, budget.max_concurrent_tasks)
        return True

Anti-Infinite-Loop Guards

The infinite loop scenario is one of the four documented production failure modes. Prevent it with hard iteration caps and convergence detection:

class ConvergenceGuard:
    def __init__(self, max_iterations: int = 20, plateau_threshold: int = 3):
        self.max_iterations = max_iterations
        self.plateau_threshold = plateau_threshold
        self._score_history: list[float] = []

    def check(self, iteration: int, score: float) -> bool:
        if iteration >= self.max_iterations:
            raise MaxIterationsReached(iteration, self.max_iterations)

        # Detect oscillation: score hasn't improved in N iterations
        if len(self._score_history) >= self.plateau_threshold:
            recent = self._score_history[-self.plateau_threshold:]
            if max(recent) == min(recent):
                raise ConvergencePlateau(recent, self.plateau_threshold)

        self._score_history.append(score)
        return True

Testing Strategies for Loops

Test Type	Scope	Tools	Coverage
Unit Tests	Individual components (executor, evaluator)	pytest, mocks	Core logic correctness
Integration Tests	Full loop with mocked dependencies	pytest, fixtures	Loop flow and state transitions
Property Tests	Invariants across random inputs	hypothesis	Edge cases and failure modes
Load Tests	Loop behavior under concurrent load	locust, k6	Performance and resource limits
Chaos Tests	Loop resilience under failures	toxiproxy, fault injection	Error handling and degradation
End-to-End Tests	Full system with real dependencies	pytest + real services	Production-like behavior

Property-Based Testing Example

Property-based tests are especially valuable for loops because they can verify invariants that hold across a wide range of inputs, catching edge cases that example-based tests miss:

from hypothesis import given, strategies as st

@given(st.builds(Task, max_iterations=st.integers(min_value=1, max_value=5)))
def test_loop_always_terminates(task):
    """A production loop must ALWAYS terminate within max_iterations."""
    loop = SingleLoop(queue, executor, evaluator, feedback_channel)
    result = loop.run(task)
    assert result is not None
    assert result.iterations <= task.max_iterations

@given(st.builds(Task))
def test_loop_preserves_best_result(task):
    """A loop must never return a score lower than its initial evaluation."""
    loop = SingleLoop(queue, executor, evaluator, feedback_channel)
    result = loop.run(task)
    assert result.score >= task.min_acceptable_score or result.status == "max_iterations"

Testing Against the Four Failure Modes

Design tests that specifically target each of the four documented production failure scenarios:

def test_context_overflow_does_not_retry_unchanged():
    """Context overflow must truncate, not blindly retry the same oversized prompt."""
    task = Task(context="x" * 1_000_000)  # Intentionally oversized
    loop = ProductionLoop(budget=ResourceBudget(max_tokens_per_task=100_000))
    result = loop.run(task)
    # Should NOT raise MaxRetriesExceeded from blind retries
    assert result.status in ("degraded", "failed")

def test_exception_does_not_enter_infinite_retry():
    """Classified errors must not trigger infinite retry loops."""
    loop = ProductionLoop(max_retries=3, retry_backoff_base=2.0)
    with patch.object(loop, '_execute', side_effect=RateLimitError("429")):
        result = loop.run(Task(prompt="test"))
    assert loop._retry_count == 3  # Hard cap, not infinite

Configuration Management

Production loops must be configurable without code changes. This follows the context engineering principle of dynamic context -- adapting behavior based on runtime state rather than hardcoding it.

Claude Code CLAUDE.md Pattern

Claude Code (github.com/anthropics/claude-code, docs at code.claude.com/docs) uses CLAUDE.md files to configure agent behavior per project. This is a practical pattern for production loop configuration:

# CLAUDE.md - Production Loop Configuration

## Loop Parameters
- max_iterations: 15
- iteration_timeout: 30s
- token_budget_per_task: 50000
- convergence_plateau_threshold: 3

## Degradation Policy
- fallback_model: claude-3-haiku-20240307
- cache_ttl: 3600s
- fail_open_after_retries: 3

Aider Configuration

Similarly, Aider uses a layered config approach with .aider.conf.yml that demonstrates externalized configuration for agent behavior:

# .aider.conf.yml
model: claude-3-5-sonnet-20241022
edit-format: diff
max-tokens: 8192
auto-commits: true

Windsurf Cascade Configuration

Windsurf (windsurf.ai) extends this through its Cascade agent, which supports multi-step execution plans with configurable retry policies and tool permissions -- analogous to the degradation hierarchy defined above.

Production Readiness Checklist

Use this checklist before deploying any loop to production:

Reliability

• All exceptions are caught and classified (recoverable vs fatal vs retryable)
• Retry logic uses exponential backoff with jitter, not fixed intervals
• Maximum retry limits are enforced as hard caps
• Graceful degradation chain is defined and tested
• State persistence ensures recovery after restarts
• Infinite loop guards are in place (hard iteration caps + convergence detection)

Observability

• Structured logging on every iteration with consistent schema
• Key metrics are exported (Prometheus, StatsD, or equivalent)
• Distributed tracing is configured for nested/multi-agent loops
• Alert rules are defined for all critical metrics
• Dashboards display loop health, convergence rate, and token spend

Resource Management

• Per-iteration and per-task timeouts are set and enforced
• Token/compute budgets are enforced with hard rejection on overflow
• Concurrent task limits are configured
• Memory usage is monitored and bounded
• Queue sizes are limited to prevent backlog buildup

Testing

• Unit tests cover all core components
• Integration tests validate the full loop flow
• Property tests verify termination and convergence invariants
• Tests specifically target the four production failure modes
• Load tests confirm resource limits hold under concurrent execution
• Chaos tests validate error handling with injected failures
• End-to-end tests run against staging environment with real services

Operations

• Configuration is externalized (CLAUDE.md, YAML, or environment variables)
• Feature flags allow runtime behavior changes without redeployment
• Health check endpoint verifies loop responsiveness, not just process existence
• Deployment can be rolled back without data loss (Git-first like Aider)
• Runbook documents common failure scenarios, including the four fault categories

Best Practices

1. Deploy with feature flags so you can disable or degrade individual loops without redeploying. Cursor follows this pattern in Agent mode, where each parallel agent task can be individually verified or rolled back.

2. Use circuit breakers around all external dependencies. This is critical in multi-agent systems built with CrewAI, LangGraph, or AutoGen, where a single agent failure can cascade.

3. Classify errors before retrying. The production failure analysis shows that blind retries amplify problems. A RateLimitError needs backoff; a SchemaValidationError needs feedback to the agent; a ContextOverflowError needs truncation.

4. Implement health checks that verify loop responsiveness and convergence, not just process existence.

5. Set SLOs for loop convergence rate and latency. Define what "healthy" looks like numerically before you need to diagnose a problem.

6. Use layered model strategies to control costs. The documented 80% token reduction technique assigns heavier models (Claude 3.5 Sonnet) to evaluation stages and lighter models (Haiku) to generation stages, keeping quality while reducing spend.

7. Review loop metrics weekly to detect gradual degradation patterns. An oscillating convergence rate that slowly declines is a leading indicator of context drift or model quality changes.