Stop Blaming the LLM: Why Your AI Agent's Bugs Are Actually Distributed Systems Failures

┌─────────────────────────────────────────────────────────────┐
│                    ORCHESTRATOR (Your Code)                  │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐        │
│  │ Planner │→ │ Executor │→ │ Critic  │→ │ Memory  │        │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘        │
└───────┼────────────┼────────────┼────────────┼──────────────┘
        │            │            │            │
        ▼            ▼            ▼            ▼
┌───────────────┐ ┌─────────┐ ┌──────────┐ ┌────────────┐
│  LLM Provider │ │  Tools  │ │  Vector  │ │  State     │
│  (API Gateway)│ │ (Code,  │ │  Store   │ │  Store     │
│               │ │  Browse,│ │ (Pinecone│ │ (Redis,    │
│  - Rate limits│ │  API)   │ │  , etc.) │ │  Postgres) │
│  - Latency    │ │         │ │          │ │            │
│  - Errors     │ │ - Sand- │ │ - Index  │ │ - TTL      │
│  - Quotas     │ │   boxes │ │   lag    │ │ - Locks    │
└───────────────┘ └─────────┘ └──────────┘ └────────────┘

# ponytail: naive sequential execution, parallelize independent calls if latency matters
async def agent_turn(user_input: str, ctx: AgentContext) -> AgentResponse:
    plan = await llm.plan(user_input, ctx.history)           # 1.2s
    tool_results = await execute_tools(plan.tools)           # 2.8s (parallel)
    critique = await llm.critique(plan, tool_results)        # 0.9s
    memory.update(ctx.session_id, critique)                  # 0.1s
    return await llm.synthesize(plan, tool_results, critique) # 1.1s
# Total: ~6s sequential, ~3s with parallelization

# What the model actually produced (streaming):
"I need to check the database... [tool: query_db] ... 
 The result shows 42 users. Now I'll calculate... [tool: python] ...
 The answer is 42 * 1.5 = 63. So the final answer is—"

# What your orchestrator receives (gateway timeout at 60s):
"I need to check the database... [tool: query_db] ... 
 The result shows 42 users. Now I'll calculate... [tool: python] ...
 The answer is 42 * 1.5 = 63. So the final answer is"

# Your code sees: incomplete response → treats as model error → retries

# Detect this by logging model fingerprint headers
# OpenAI returns: x-request-id, x-ratelimit-*, and sometimes model fingerprint
curl -v https://api.openai.com/v1/chat/completions ... 2>&1 | grep -i fingerprint
# x-model-fingerprint: fp_abc123  ← log this every request

# DANGEROUS: Non-idempotent operations in a retry loop
async def run_agent(task: Task) -> Result:
    for attempt in range(3):
        try:
            plan = await llm.plan(task)           # Idempotent (read-only)
            result = await execute_plan(plan)     # NOT idempotent!
            return result
        except TransientError:
            continue
    raise MaxRetriesExceeded()

# execute_plan might:
# - Charge a credit card (twice)
# - Send an email (twice)
# - Mutate database state (twice)
# - Create a Jira ticket (twice)

import hashlib
import json
from dataclasses import dataclass
from typing import Any

@dataclass
class ToolInvocation:
    tool_name: str
    args: dict[str, Any]
    idempotency_key: str
    attempt: int = 0

    @classmethod
def create(cls, tool_name: str, args: dict, execution_id: str, step: int) -> "ToolInvocation":
        # Deterministic key: execution_id + step + tool + args hash
        content = f"{execution_id}:{step}:{tool_name}:{json.dumps(args, sort_keys=True)}"
        key = hashlib.sha256(content.encode()).hexdigest()[:32]
        return cls(tool_name=tool_name, args=args, idempotency_key=key)

# Tool executor checks idempotency store before executing
async def execute_tool(invocation: ToolInvocation) -> ToolResult:
    existing = await idempotency_store.get(invocation.idempotency_key)
    if existing:
        return ToolResult.from_cached(existing)  # Return previous result

    result = await actual_tool_call(invocation.tool_name, invocation.args)
    await idempotency_store.set(invocation.idempotency_key, result, ttl=86400)
    return result

# BETTER: Granular retry at tool level, not agent level
async def execute_plan(plan: Plan, ctx: ExecutionContext) -> PlanResult:
    results = {}
    for step in plan.steps:
        if step.depends_on and not all(results[d].success for d in step.depends_on):
            results[step.id] = StepResult(skipped=True, reason="dependency_failed")
            continue

        # Retry ONLY this step, with its own budget
        results[step.id] = await execute_step_with_retry(
            step, 
            max_attempts=3,
            timeout=step.timeout or 30,
            idempotency_key=ctx.make_key(step.id)
        )
    return PlanResult(results)

@dataclass
class SagaStep:
    name: str
    execute: Callable[[], Awaitable[StepResult]]
    compensate: Callable[[StepResult], Awaitable[None]]  # Undo action
    idempotency_key: str

class AgentSaga:
    def __init__(self, execution_id: str):
        self.execution_id = execution_id
        self.completed_steps: list[SagaStep] = []
        self.step_results: dict[str, StepResult] = {}

    async def run(self, steps: list[SagaStep]) -> SagaResult:
        for step in steps:
            try:
                result = await step.execute()
                self.step_results[step.name] = result
                self.completed_steps.append(step)
            except Exception as e:
                # Compensate in reverse order
                await self._compensate()
                return SagaResult(success=False, failed_step=step.name, error=e)
        return SagaResult(success=True, results=self.step_results)

    async def _compensate(self):
        for step in reversed(self.completed_steps):
            try:
                await step.compensate(self.step_results[step.name])
            except Exception as comp_err:
                # Log but continue compensating other steps
                logger.error(f"Compensation failed for {step.name}: {comp_err}")
                await alert_oncall(f"Saga {self.execution_id} compensation failed at {step.name}")

from contextvars import ContextVar
import uuid

current_trace: ContextVar[str] = ContextVar("current_trace", default="")
current_span: ContextVar[str] = ContextVar("current_span", default="")

class TracedAgent:
    async def run(self, input: str) -> AgentOutput:
        trace_id = str(uuid.uuid4())
        current_trace.set(trace_id)

        with span(trace_id, "agent.run", {"input": input}) as root_span:
            plan = await self._plan_with_trace(input, root_span)
            tool_results = await self._execute_with_trace(plan, root_span)
            output = await self._synthesize_with_trace(plan, tool_results, root_span)
            root_span.set_attribute("output", output)
            return output

    async def _plan_with_trace(self, input: str, parent_span) -> Plan:
        with span(parent_span.trace_id, "llm.plan", {"model": self.model}, parent_span.span_id) as span:
            span.set_attribute("prompt_tokens", estimate_tokens(input))
            result = await self.llm.plan(input)
            span.set_attribute("completion_tokens", estimate_tokens(result))
            span.set_attribute("latency_ms", span.duration_ms)
            return result

{
  "timestamp": "2024-01-15T10:23:45.123Z",
  "level": "INFO",
  "trace_id": "abc-123",
  "span_id": "span-456",
  "component": "tool_executor",
  "tool": "web_search",
  "args": {"query": "quarterly revenue", "max_results": 5},
  "idempotency_key": "abc123...",
  "attempt": 2,
  "latency_ms": 2340,
  "status": "success",
  "result_summary": {"results_count": 5, "tokens": 1200}
}

# test_replay.py
import pytest
from agent.tracing import TraceRecord
from agent.orchestrator import AgentOrchestrator

@pytest.mark.parametrize("trace", load_production_traces("2024-01", sample=1000))
async def test_agent_replay(trace: TraceRecord):
    """Replay a real production trace through current code."""
    orchestrator = AgentOrchestrator(
        llm=FakeLLM(responses=trace.llm_responses),  # Deterministic replay
        tools=FakeTools(responses=trace.tool_responses),
        memory=FakeMemory(initial_state=trace.initial_memory)
    )

    result = await orchestrator.run(trace.input)

    # Assert structural equivalence, not exact string match
    assert result.tool_calls == trace.expected_tool_calls
    assert result.final_answer.semantic_equiv(trace.expected_answer)
    assert result.latency_ms < trace.original_latency_ms * 1.5  # No regression

class ChaosAgentWrapper:
    """Wraps an agent and injects failures per configuration."""
    def __init__(self, agent, chaos_config: ChaosConfig):
        self.agent = agent
        self.chaos = chaos_config

    async def run(self, input: str) -> AgentOutput:
        if self.chaos.should_fail("llm_timeout"):
            raise TimeoutError("Injected LLM timeout")
        if self.chaos.should_fail("tool_error"):
            raise ToolError("Injected tool failure")
        if self.chaos.should_delay("network_latency"):
            await asyncio.sleep(self.chaos.latency_ms / 1000)

        return await self.agent.run(input)

# Run in CI: 1000 iterations with 5% random failure injection
# Assert: no uncompensated mutations, no lost user data, graceful degradation

# evals/regression_suite.yaml
cases:
  - id: "finance_calc_001"
    input: "What's 15% of $2,340?"
    expected_tools: ["calculator"]
    expected_answer_pattern: "351"
    max_latency_ms: 5000
    tags: ["math", "single_tool"]

  - id: "multi_hop_research_001"
    input: "Compare Q3 revenue of Apple vs Microsoft"
    expected_tools: ["web_search", "calculator", "web_search"]
    expected_answer_contains: ["Apple", "Microsoft", "revenue", "billion"]
    max_latency_ms: 30000
    tags: ["multi_hop", "research"]

async def run_regression_suite(agent, suite_path: str) -> EvaluationReport:
    results = []
    for case in load_suite(suite_path):
        trace = await agent.run_with_trace(case.input)
        results.append(EvalResult(
            case_id=case.id,
            passed=check_case(case, trace),
            latency_ms=trace.total_latency_ms,
            tokens_used=trace.total_tokens,
            trace_id=trace.trace_id
        ))
    return EvaluationReport(results)

from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing fast
    HALF_OPEN = "half_open" # Testing recovery

class ProviderCircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_max_calls: int = 3
    ):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.half_open_calls = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls

    async def call(self, func: Callable, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitOpenError("Circuit breaker open")

        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls >= self.half_open_max_calls:
                raise CircuitOpenError("Half-open limit reached")
            self.half_open_calls += 1

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

llm_clients = {
    "openai": CircuitBreakerClient(OpenAIClient(), ProviderCircuitBreaker()),
    "anthropic": CircuitBreakerClient(AnthropicClient(), ProviderCircuitBreaker()),
    "local": CircuitBreakerClient(LocalLLMClient(), ProviderCircuitBreaker(failure_threshold=10)),
}

async def call_llm_with_fallback(messages, preferred="openai"):
    for provider in [preferred, "anthropic", "local"]:
        try:
            return await llm_clients[provider].complete(messages)
        except (CircuitOpenError, RateLimitError, TimeoutError):
            continue
    raise AllProvidersUnavailable()

class AdaptiveTimeout:
    def __init__(self, percentile: float = 0.99, buffer_multiplier: float = 1.5):
        self.percentile = percentile
        self.buffer = buffer_multiplier
        self.latencies: list[float] = []  # Rolling window
        self.max_samples = 1000

    def record(self, latency_ms: float):
        self.latencies.append(latency_ms)
        if len(self.latencies) > self.max_samples:
            self.latencies.pop(0)

    def get_timeout(self) -> float:
        if len(self.latencies) < 10:
            return 30000  # Default 30s
        sorted_lat = sorted(self.latencies)
        idx = int(len(sorted_lat) * self.percentile)
        return sorted_lat[idx] * self.buffer

# Per-operation adaptive timeouts
timeouts = {
    "llm.plan": AdaptiveTimeout(0.99, 1.5),
    "llm.critique": AdaptiveTimeout(0.99, 1.5),
    "tool.web_search": AdaptiveTimeout(0.95, 2.0),  # More variable
    "tool.code_exec": AdaptiveTimeout(0.99, 1.2),
    "memory.read": AdaptiveTimeout(0.999, 1.1),
}

async def execute_with_timeout(op_name: str, coro):
    timeout = timeouts[op_name].get_timeout()
    try:
        start = time.time()
        result = await asyncio.wait_for(coro, timeout=timeout / 1000)
        timeouts[op_name].record((time.time() - start) * 1000)
        return result
    except asyncio.TimeoutError:
        timeouts[op_name].record(timeout)  # Record timeout as latency
        raise

class DegradedAgent:
    """Agent that progressively sheds capabilities under stress."""

    async def run(self, input: str, context: AgentContext) -> AgentOutput:
        # Level 0: Full capability
        if not self.degraded:
            return await self._full_run(input, context)

        # Level 1: No web search (slow, unreliable)
        if self.degraded_level >= 1:
            context.available_tools.discard("web_search")

        # Level 2: No code execution (sandbox overhead)
        if self.degraded_level >= 2:
            context.available_tools.discard("python")

        # Level 3: Single-turn only, no planning loop
        if self.degraded_level >= 3:
            return await self._single_turn_fallback(input, context)

        # Level 4: Canned responses only
        return await self._canned_fallback(input)

    def update_health(self, metrics: HealthMetrics):
        """Called periodically by health monitor."""
        if metrics.error_rate > 0.2 or metrics.p99_latency > 60000:
            self.degraded_level = min(self.degraded_level + 1, 4)
        elif metrics.error_rate < 0.01 and metrics.p99_latency < 15000:
            self.degraded_level = max(self.degraded_level - 1, 0)

Trace: agent_7x9k2
├─ llm.plan (1.2s) ✓
├─ tool.web_search "quarterly revenue" (28.3s) → TIMEOUT (30s gateway)
├─ llm.plan RETRY (1.1s) → "Let me search for that..."
├─ tool.web_search "quarterly revenue" (29.1s) → TIMEOUT
├─ llm.plan RETRY (1.3s) → "Let me search for that..."
├─ tool.web_search "quarterly revenue" (31.2s) → TIMEOUT
├─ llm.synthesize (0.8s) → Generic answer (no tool results)

# 1. Increase gateway timeout to 60s (match search p99 + buffer)
# 2. Make web_search idempotent (cache results for 5min)
# 3. Retry ONLY the failed tool, not the whole plan

# Before (broken):
async def run_agent():
    for attempt in range(3):
        try:
            return await full_agent_loop()
        except ToolTimeout:
            continue

# After (fixed):
async def run_agent():
    plan = await llm.plan()  # No retry needed, idempotent
    results = {}
    for step in plan.steps:
        results[step.id] = await execute_with_retry(
            step, max_attempts=2, timeout=60  # Per-step config
        )
    return await llm.synthesize(plan, results)

# Agent side: generate and attach a key
payment_id = "pay_12345"
idempotency_key = f"checkout-{payment_id}-{int(time.time())}"
result = await agent.call_tool(
    "payments_process",
    {"amount": 999, "currency": "USD", "idempotency_key": idempotency_key},
)

# Service side: check cache before executing
async def process_payment(data: dict):
    key = data.pop("idempotency_key", None)
    cached = idempotency_store.get(key)
    if cached:
        return cached  # return original result unchanged
    # ... actual payment processing ...
    idempotency_store.set(key, result, ttl=86400)
    return result

async with asyncio.timeout(5):  # 5-second hard limit
    result = await agent.call_tool("some_remote_service", payload)

# Circuit breaker state (persisted across restarts)
class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure = None
        self.state = "closed"  # closed/open/half-open

    async def call(self, fn, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = "half-open"
            else:
                raise CircuitOpenError("service temporarily unavailable")

        try:
            result = await fn(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failures = 0
        self.state = "closed"

    def _on_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

import asyncio
import time
import random
from typing import Optional, Dict, Any

# --- Mock infrastructure ---

class InMemoryIdempotencyStore:
    def __init__(self):
        self._store: Dict[str, Any] = {}

    def get(self, key: str) -> Optional[Any]:
        return self._store.get(key)

    def set(self, key: str, value: Any, ttl: int = 3600):
        self._store[key] = value


class SimpleCircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure: Optional[float] = None
        self.state = "closed"  # closed / open / half-open

    async def call(self, fn, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = "half-open"
            else:
                raise RuntimeError("CircuitOpen: service unavailable")

        try:
            result = await fn(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failures = 0
        self.state = "closed"

    def _on_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"


# --- Agent core ---

class DistributedAgent:
    def __init__(self):
        self.idempotency = InMemoryIdempotencyStore()
        self.circuit = SimpleCircuitBreaker()
        self.metrics = {"calls": 0, "errors": 0, "retries": 0}

    async def call_external(self, payload: dict) -> dict:
        """Wrap external calls with timeout, idempotency, and circuit breaking."""
        # Attach idempotency key if present
        key = payload.pop("idempotency_key", None)
        if key:
            cached = self.idempotency.get(key)
            if cached:
                self.metrics["calls"] += 1
                return cached

        # Enforce timeout
        try:
            async with asyncio.timeout(5):
                result = await self.circuit.call(self._do_call, payload)
        except asyncio.TimeoutError:
            self.metrics["errors"] += 1
            raise RuntimeError("Call timed out after 5s") from None

        # Cache result for idempotency
        if key:
            self.idempotency.set(key, result, ttl=3600)

        self.metrics["calls"] += 1
        return result

    async def _do_call(self, payload: dict) -> dict:
        """Actual external call—here simulated with random latency."""
        # Simulate flaky external service
        await asyncio.sleep(random.uniform(0.1, 2.0))
        if random.random() < 0.3:
            raise RuntimeError("Simulated external failure")
        return {"status": "ok", "data": payload.get("value")}

    async def run_workflow(self, input_value: str) -> dict:
        """Example workflow that uses all four pillars."""
        # Step 1: idempotent lookup
        lookup_key = f"workflow-{input_value}"
        cached = self.idempotency.get(lookup_key)
        if cached:
            self.metrics["retries"] += 1
            return cached

        # Step 2: perform external action with timeout + circuit breaker
        payload = {"value": input_value, "idempotency_key": lookup_key}
        result = await self.call_external(payload)

        # Step 3: store result for future idempotent repeats
        self.idempotency.set(lookup_key, result, ttl=3600)

        # Emit metrics (in real code, push to Prometheus/Datadog)
        self._emit_metrics()

        return result

    def _emit_metrics(self):
        # Placeholder for metric emission
        pass


# --- Example usage ---

async def main():
    agent = DistributedAgent()

    # First call—may succeed or fail
    print("Attempt 1:", await agent.run_workflow("hello-world"))

    # Second call with same input—idempotent cache hit
    print("Attempt 2 (cached):", await agent.run_workflow("hello-world"))


if __name__ == "__main__":
    asyncio.run(main())