AI Agents Keep Dying in Production. The Fix Was Invented in 1986.

from dataclasses import dataclass, field
from enum import Enum
from typing import Any
import asyncio
import time
import uuid
 
 
class RestartStrategy(Enum):
    ONE_FOR_ONE = "one_for_one"      # restart only the failed child
    ONE_FOR_ALL = "one_for_all"      # restart all children
    REST_FOR_ONE = "rest_for_one"    # restart failed + all children started after it
 
 
class ChildState(Enum):
    RUNNING = "running"
    STOPPED = "stopped"
    FAILED = "failed"
    RESTARTING = "restarting"
 
 
@dataclass
class ChildSpec:
    """Specification for a supervised child process."""
    id: str
    start_fn: Any          # async callable that returns a result
    max_restarts: int = 3  # max restarts within restart_window
    restart_window: int = 60  # seconds
    shutdown_timeout: int = 5  # seconds to wait for graceful shutdown
    state: ChildState = ChildState.STOPPED
    restart_count: int = 0
    restart_timestamps: list = field(default_factory=list)
    last_error: Exception | None = None

class Supervisor:
    def __init__(
        self,
        name: str,
        strategy: RestartStrategy = RestartStrategy.ONE_FOR_ONE,
        max_intensity: int = 5,    # max total restarts
        intensity_window: int = 60  # within this window (seconds)
    ):
        self.name = name
        self.strategy = strategy
        self.max_intensity = max_intensity
        self.intensity_window = intensity_window
        self.children: list[ChildSpec] = []
        self.restart_history: list[float] = []
 
    def add_child(self, spec: ChildSpec) -> None:
        self.children.append(spec)
 
    async def start(self) -> dict[str, Any]:
        """Start all children and supervise them."""
        results = {}
        for child in self.children:
            results[child.id] = await self._start_child(child)
        return results
 
    async def _start_child(self, child: ChildSpec) -> Any:
        """Start a single child with supervision."""
        child.state = ChildState.RUNNING
        try:
            result = await asyncio.wait_for(
                child.start_fn(),
                timeout=child.shutdown_timeout * 10  # generous timeout
            )
            child.state = ChildState.STOPPED
            return result
        except Exception as e:
            child.state = ChildState.FAILED
            child.last_error = e
            return await self._handle_failure(child, e)
 
    async def _handle_failure(self, failed_child: ChildSpec, error: Exception) -> Any:
        """Apply the restart strategy."""
        print(f"[{self.name}] Child '{failed_child.id}' failed: {error}")
 
        if not self._can_restart(failed_child):
            raise SupervisorShutdown(
                f"Child '{failed_child.id}' exceeded max restarts "
                f"({failed_child.max_restarts} in {failed_child.restart_window}s)"
            )
 
        if not self._check_intensity():
            raise SupervisorShutdown(
                f"Supervisor '{self.name}' exceeded max restart intensity "
                f"({self.max_intensity} in {self.intensity_window}s)"
            )
 
        match self.strategy:
            case RestartStrategy.ONE_FOR_ONE:
                return await self._restart_one(failed_child)
            case RestartStrategy.ONE_FOR_ALL:
                return await self._restart_all(failed_child)
            case RestartStrategy.REST_FOR_ONE:
                return await self._restart_rest(failed_child)
 
    def _can_restart(self, child: ChildSpec) -> bool:
        """Check if child hasn't exceeded its restart budget."""
        now = time.time()
        child.restart_timestamps = [
            t for t in child.restart_timestamps
            if now - t < child.restart_window
        ]
        if len(child.restart_timestamps) >= child.max_restarts:
            return False
        child.restart_timestamps.append(now)
        child.restart_count += 1
        return True
 
    def _check_intensity(self) -> bool:
        """Check if supervisor hasn't exceeded global restart intensity."""
        now = time.time()
        self.restart_history = [
            t for t in self.restart_history
            if now - t < self.intensity_window
        ]
        if len(self.restart_history) >= self.max_intensity:
            return False
        self.restart_history.append(now)
        return True
 
    async def _restart_one(self, child: ChildSpec) -> Any:
        """Restart only the failed child."""
        print(f"[{self.name}] Restarting '{child.id}' (attempt {child.restart_count})")
        child.state = ChildState.RESTARTING
        return await self._start_child(child)
 
    async def _restart_all(self, failed_child: ChildSpec) -> dict[str, Any]:
        """Restart all children (order matters)."""
        print(f"[{self.name}] Restarting ALL children due to '{failed_child.id}' failure")
        results = {}
        for child in self.children:
            child.state = ChildState.RESTARTING
            results[child.id] = await self._start_child(child)
        return results
 
    async def _restart_rest(self, failed_child: ChildSpec) -> dict[str, Any]:
        """Restart the failed child and all children started after it."""
        idx = self.children.index(failed_child)
        to_restart = self.children[idx:]
        print(f"[{self.name}] Restarting '{failed_child.id}' + {len(to_restart)-1} subsequent children")
        results = {}
        for child in to_restart:
            child.state = ChildState.RESTARTING
            results[child.id] = await self._start_child(child)
        return results
 
 
class SupervisorShutdown(Exception):
    """Raised when a supervisor exhausts its restart budget."""
    pass

import anthropic
import httpx
 
client = anthropic.Anthropic()
 
async def web_search_agent(query: str) -> list[str]:
    """Search the web and return relevant URLs."""
    response = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=1024,
        tools=[{
            "name": "web_search",
            "description": "Search the web",
            "input_schema": {"type": "object", "properties": {"q": {"type": "string"}}}
        }],
        messages=[{"role": "user", "content": f"Find sources about: {query}"}]
    )
    # Extract URLs from tool use results
    urls = parse_search_results(response)
    if not urls:
        raise ValueError("Search returned no results")
    return urls
 
 
async def extract_agent(urls: list[str]) -> list[dict]:
    """Extract key information from each URL."""
    results = []
    async with httpx.AsyncClient(timeout=10) as http:
        for url in urls[:5]:  # limit to 5 sources
            try:
                resp = await http.get(url)
                content = resp.text[:8000]
                extraction = client.messages.create(
                    model="claude-sonnet-4-6-20250514",
                    max_tokens=2048,
                    messages=[{
                        "role": "user",
                        "content": f"Extract key facts from:\n\n{content}"
                    }]
                )
                results.append({
                    "url": url,
                    "facts": extraction.content[0].text
                })
            except httpx.HTTPError:
                continue  # individual URL failures are fine
    if not results:
        raise ValueError("All extractions failed")
    return results
 
 
async def synthesize_agent(facts: list[dict], question: str) -> str:
    """Synthesize extracted facts into a coherent answer."""
    context = "\n\n".join(
        f"Source: {f['url']}\nFacts: {f['facts']}" for f in facts
    )
    response = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": (
                f"Based on these sources, answer: {question}\n\n"
                f"Sources:\n{context}"
            )
        }]
    )
    return response.content[0].text

async def run_supervised_research(question: str) -> str:
    """Run a research pipeline with Erlang-style supervision."""
 
    # State flows through the pipeline
    state = {"question": question, "urls": [], "facts": [], "answer": ""}
 
    # Define each step as a supervised child
    search_spec = ChildSpec(
        id="web_search",
        start_fn=lambda: web_search_agent(question),
        max_restarts=3,
        restart_window=30
    )
 
    extract_spec = ChildSpec(
        id="extraction",
        start_fn=lambda: extract_agent(state["urls"]),
        max_restarts=2,
        restart_window=60
    )
 
    synth_spec = ChildSpec(
        id="synthesis",
        start_fn=lambda: synthesize_agent(state["facts"], question),
        max_restarts=2,
        restart_window=60
    )
 
    # REST_FOR_ONE: if search fails, re-run search + everything after it
    # but if synthesis fails, only re-run synthesis
    supervisor = Supervisor(
        name="research_pipeline",
        strategy=RestartStrategy.REST_FOR_ONE,
        max_intensity=8,
        intensity_window=120
    )
 
    supervisor.add_child(search_spec)
    supervisor.add_child(extract_spec)
    supervisor.add_child(synth_spec)
 
    # Run pipeline: each step feeds the next
    state["urls"] = await supervisor._start_child(search_spec)
    state["facts"] = await supervisor._start_child(extract_spec)
    state["answer"] = await supervisor._start_child(synth_spec)
 
    return state["answer"]

@dataclass
class CircuitBreaker:
    max_cost_usd: float = 1.0
    max_tokens: int = 100_000
    max_wall_time: int = 120  # seconds
    current_cost: float = 0.0
    current_tokens: int = 0
    start_time: float = field(default_factory=time.time)
 
    def track(self, input_tokens: int, output_tokens: int, model: str) -> None:
        """Track usage and trip if limits exceeded."""
        cost = self._calculate_cost(input_tokens, output_tokens, model)
        self.current_cost += cost
        self.current_tokens += input_tokens + output_tokens
 
        if self.current_cost > self.max_cost_usd:
            raise CircuitBreakerTripped(
                f"Cost limit exceeded: ${self.current_cost:.2f} > ${self.max_cost_usd:.2f}"
            )
        if self.current_tokens > self.max_tokens:
            raise CircuitBreakerTripped(
                f"Token limit exceeded: {self.current_tokens} > {self.max_tokens}"
            )
        if time.time() - self.start_time > self.max_wall_time:
            raise CircuitBreakerTripped(
                f"Wall time exceeded: {time.time() - self.start_time:.0f}s > {self.max_wall_time}s"
            )
 
    def _calculate_cost(self, inp: int, out: int, model: str) -> float:
        rates = {
            "claude-sonnet-4-6-20250514": (0.003, 0.015),
            "claude-haiku-4-5-20251001": (0.0008, 0.004),
        }
        input_rate, output_rate = rates.get(model, (0.003, 0.015))
        return (inp / 1000 * input_rate) + (out / 1000 * output_rate)
 
 
class CircuitBreakerTripped(Exception):
    pass

MODEL_CHAIN = [
    "claude-sonnet-4-6-20250514",
    "claude-haiku-4-5-20251001",
]
 
async def resilient_llm_call(messages, tools=None, model_idx=0):
    """Try models in order until one succeeds."""
    if model_idx >= len(MODEL_CHAIN):
        raise AllModelsExhausted("No models available")
    try:
        return client.messages.create(
            model=MODEL_CHAIN[model_idx],
            max_tokens=4096,
            messages=messages,
            tools=tools or []
        )
    except (anthropic.RateLimitError, anthropic.APIStatusError):
        return await resilient_llm_call(messages, tools, model_idx + 1)

import json
from pathlib import Path
 
class CheckpointStore:
    def __init__(self, run_id: str):
        self.path = Path(f"/tmp/agent_checkpoints/{run_id}")
        self.path.mkdir(parents=True, exist_ok=True)
 
    def save(self, step: str, data: Any) -> None:
        (self.path / f"{step}.json").write_text(json.dumps(data))
 
    def load(self, step: str) -> Any | None:
        f = self.path / f"{step}.json"
        return json.loads(f.read_text()) if f.exists() else None
 
    def has(self, step: str) -> bool:
        return (self.path / f"{step}.json").exists()

async def with_heartbeat(fn, timeout_per_beat=30):
    """Wraps an async function with a heartbeat timeout.
    The function must yield periodically to prove it's making progress."""
    last_beat = time.time()
 
    async def monitor():
        nonlocal last_beat
        while True:
            await asyncio.sleep(5)
            if time.time() - last_beat > timeout_per_beat:
                raise HeartbeatTimeout("Agent stopped making progress")
 
    async def tracked():
        nonlocal last_beat
        async for partial in fn():
            last_beat = time.time()
            yield partial
 
    monitor_task = asyncio.create_task(monitor())
    try:
        async for result in tracked():
            yield result
    finally:
        monitor_task.cancel()

class PoisonPill:
    def __init__(self, max_failures: int = 2):
        self.failures: dict[str, int] = {}
        self.max_failures = max_failures
 
    def record_failure(self, input_hash: str) -> None:
        self.failures[input_hash] = self.failures.get(input_hash, 0) + 1
 
    def is_poisoned(self, input_hash: str) -> bool:
        return self.failures.get(input_hash, 0) >= self.max_failures