What is AI Agent Uptime?
AI agent uptime refers to the continuous operational availability of an autonomous AI agent — ensuring it remains responsive, functional, and capable of handling tasks around the clock. Unlike traditional web services backed by dedicated DevOps teams, AI agents often run as standalone processes, background workers, or long-lived scripts that are surprisingly fragile. A single unhandled exception, memory leak, API rate limit, or third-party service outage can silently kill your agent without anyone noticing until hours or days later.
The core challenge is this: AI agents are stateful, asynchronous programs that interact with external APIs, databases, and language models. They can crash for dozens of reasons that have nothing to do with your code quality. Uptime engineering for AI agents means building resilience into the agent's runtime environment so that failures are detected quickly, recovered automatically, and — when necessary — escalated to a human through the right channels.
Why AI Agent Uptime Matters
When an AI agent goes down, the cost is not just downtime. It's lost context, abandoned conversations, missed opportunities, and degraded user trust. Consider these scenarios:
- Customer-facing agents: A support chatbot that stops responding mid-conversation creates frustration and churn
- Background automation agents: A data ingestion agent that silently crashes means stale dashboards and missed insights for days
- Multi-agent systems: One agent in a pipeline failing can cascade into a complete workflow stall with no clear root cause
- Developer productivity agents: A code review agent that dies means PRs pile up and the team loses velocity
For solo developers and small teams without a dedicated DevOps function, uptime becomes a critical skill. The good news is that modern tooling and architectural patterns can give you production-grade reliability with minimal overhead. You don't need Kubernetes, a dedicated SRE team, or a six-figure observability budget. You need the right patterns baked into your agent's DNA.
Common Failure Modes for AI Agents
Before we dive into solutions, let's catalog what actually kills agents in production. Understanding failure modes shapes your resilience strategy:
- Unhandled exceptions: An unexpected API response shape, a null value where one wasn't expected, or a malformed LLM output crashes the entire process
- API rate limiting and throttling: Hitting OpenAI, Anthropic, or other provider limits without backoff handling causes cascading failures
- Memory exhaustion: Accumulating conversation history or embedding caches in-process without bounds leads to OOM kills
- Websocket / connection drops: Long-lived connections to streaming APIs silently disconnect and the agent hangs waiting for data that never arrives
- Token limit overflows: Context windows exceeding model limits produce hard errors that naive retry loops can't fix
- External dependency outages: Vector databases, search APIs, or tool-calling endpoints going down blocks the agent's reasoning loop
- State corruption: Concurrent access to shared state (file-based or in-memory) without proper locking leaves the agent in an unrecoverable position
- Timeout cascades: One slow tool call causes the orchestrator to time out, which causes upstream timeouts in a domino effect
Architecture Patterns for Self-Healing Agents
There are four fundamental patterns that, when combined, create a robust uptime foundation. Each can be implemented independently and layered for defense in depth.
Pattern 1: The Supervisor Process
A lightweight external process whose sole job is to launch, monitor, and restart your agent. The supervisor does not contain any agent logic — it's a thin watchdog. If the agent exits for any reason (crash, graceful shutdown after an error, or silent hang), the supervisor restarts it. This pattern is inspired by Erlang/OTP supervisors and Unix process supervisors like supervisord, but simplified for single-process agents.
Pattern 2: Internal Health Checks with Heartbeat Timers
The agent periodically writes a heartbeat — a timestamp to a file, a database row, or an in-memory flag — and a separate health-check routine verifies that the heartbeat is recent. If the agent is stuck in an infinite loop, deadlocked, or blocked on a hanging API call, the heartbeat stops and the health check triggers a restart.
Pattern 3: Graceful Degradation with Circuit Breakers
Instead of crashing when a dependency fails, the agent catches the failure, logs it, and either skips the non-critical step or uses a cached fallback. Circuit breakers track failure counts and temporarily disable calls to a flaky dependency, preventing cascading resource exhaustion.
Pattern 4: External Watchdog with Dead Man's Switch
A separate service (can be a free cron job, a serverless function, or a monitoring SaaS) expects a regular "I'm alive" ping from your agent. If the ping stops arriving, the external watchdog alerts you via email, Slack, SMS, or PagerDuty. This is your last line of defense — it catches failures that the supervisor and health checks miss.
Practical Implementation: Building a Self-Healing Agent Runtime
Let's build a complete, production-ready agent wrapper in Python that implements all four patterns. We'll create a reusable AgentRuntime class that you can drop around any AI agent logic.
1. The Supervisor Shell Script
Start with the simplest possible supervisor — a bash script that runs your agent in an infinite restart loop with a cooldown delay. This lives completely outside your Python code.
#!/bin/bash
# supervisor.sh - Restart the agent indefinitely
# Usage: ./supervisor.sh
AGENT_CMD="python agent_main.py"
RESTART_DELAY=5 # seconds between restart attempts
MAX_RESTARTS_PER_HOUR=20
RESTART_COUNT=0
HOUR_START=$(date +%s)
log() {
echo "[$(date -u '+%Y-%m-%d %H:%M:%S UTC')] SUPERVISOR: $1"
}
while true; do
CURRENT_TIME=$(date +%s)
HOUR_ELAPSED=$(( CURRENT_TIME - HOUR_START ))
# Reset counter each hour
if [ $HOUR_ELAPSED -ge 3600 ]; then
RESTART_COUNT=0
HOUR_START=$CURRENT_TIME
fi
if [ $RESTART_COUNT -ge $MAX_RESTARTS_PER_HOUR ]; then
log "ERROR: Exceeded max restarts ($MAX_RESTARTS_PER_HOUR per hour). Sleeping 10 min."
sleep 600
RESTART_COUNT=0
HOUR_START=$(date +%s)
fi
log "Starting agent process..."
$AGENT_CMD
EXIT_CODE=$?
RESTART_COUNT=$(( RESTART_COUNT + 1 ))
log "Agent exited with code $EXIT_CODE. Restart count: $RESTART_COUNT. Restarting in $RESTART_DELAY seconds..."
sleep $RESTART_DELAY
doneThis script provides a solid first line of defense. It tracks restart frequency to prevent tight crash loops from burning CPU, logs every event with timestamps, and never gives up. Run it in a tmux session, a systemd service, or as a background process with nohup.
2. The AgentRuntime Base Class
Now we build the in-process resilience layer. This is a reusable wrapper that adds heartbeat health checks, structured logging, exception isolation, and graceful degradation to any agent loop.
import os
import time
import json
import signal
import threading
import traceback
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Callable, Optional, Dict, Any
from enum import Enum
class AgentStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
STARTING = "starting"
SHUTTING_DOWN = "shutting_down"
@dataclass
class CircuitBreakerState:
failure_count: int = 0
last_failure_time: float = 0.0
half_open: bool = False
open_until: float = 0.0
class AgentRuntime:
"""
Production-grade wrapper for AI agents that provides:
- Heartbeat-based health checks
- Automatic crash isolation and recovery
- Circuit breakers for external dependencies
- Structured logging with context
- Graceful shutdown handling
"""
def __init__(
self,
agent_name: str,
heartbeat_path: str = "/tmp/agent_heartbeat.json",
heartbeat_interval_seconds: int = 10,
circuit_breaker_threshold: int = 5,
circuit_breaker_recovery_seconds: int = 60,
max_consecutive_failures: int = 3,
log_dir: str = "./agent_logs"
):
self.agent_name = agent_name
self.heartbeat_path = heartbeat_path
self.heartbeat_interval = heartbeat_interval_seconds
self.circuit_breaker_threshold = circuit_breaker_threshold
self.circuit_breaker_recovery = circuit_breaker_recovery_seconds
self.max_consecutive_failures = max_consecutive_failures
self.status = AgentStatus.STARTING
self.consecutive_failures = 0
self.circuit_breakers: Dict[str, CircuitBreakerState] = {}
self._shutdown_requested = False
self._heartbeat_thread: Optional[threading.Thread] = None
os.makedirs(log_dir, exist_ok=True)
self.log_dir = log_dir
# Register signal handlers for graceful shutdown
signal.signal(signal.SIGTERM, self._handle_signal)
signal.signal(signal.SIGINT, self._handle_signal)
def log(self, level: str, message: str, extra: Optional[Dict[str, Any]] = None):
"""Structured logging with timestamps and agent context."""
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"agent": self.agent_name,
"status": self.status.value,
"level": level,
"message": message,
"extra": extra or {}
}
# Write to structured log file
log_file = os.path.join(self.log_dir, f"{self.agent_name}.jsonl")
with open(log_file, "a") as f:
f.write(json.dumps(entry) + "\n")
# Also emit to stdout for supervisor visibility
print(f"[{entry['timestamp']}] {level.upper()}: {message}", flush=True)
def _write_heartbeat(self):
"""Atomically write heartbeat state to disk."""
heartbeat = {
"agent": self.agent_name,
"status": self.status.value,
"timestamp": datetime.now(timezone.utc).isoformat(),
"pid": os.getpid(),
"consecutive_failures": self.consecutive_failures
}
tmp_path = self.heartbeat_path + ".tmp"
with open(tmp_path, "w") as f:
json.dump(heartbeat, f)
os.rename(tmp_path, self.heartbeat_path) # Atomic on POSIX
def _heartbeat_loop(self):
"""Background thread that writes periodic heartbeats."""
self.log("info", "Heartbeat thread started")
while not self._shutdown_requested:
try:
self._write_heartbeat()
except Exception as e:
self.log("error", f"Heartbeat write failed: {e}")
time.sleep(self.heartbeat_interval)
self.log("info", "Heartbeat thread stopped")
def start_heartbeat(self):
"""Launch the heartbeat background thread."""
self._heartbeat_thread = threading.Thread(
target=self._heartbeat_loop, daemon=True, name="heartbeat-thread"
)
self._heartbeat_thread.start()
self._write_heartbeat()
def _handle_signal(self, signum, frame):
"""Handle SIGTERM/SIGINT for graceful shutdown."""
self.log("info", f"Received signal {signum}, initiating graceful shutdown")
self._shutdown_requested = True
self.status = AgentStatus.SHUTTING_DOWN
self._write_heartbeat()
def check_circuit_breaker(self, dependency_name: str) -> bool:
"""
Check if a circuit breaker allows the call.
Returns True if the call is allowed, False if the circuit is open.
"""
cb = self.circuit_breakers.get(dependency_name)
if cb is None:
return True # No circuit breaker configured, allow
now = time.time()
if cb.open_until > 0 and now < cb.open_until:
return False # Circuit is open
if cb.open_until > 0 and now >= cb.open_until:
# Transition to half-open: allow a single trial call
cb.half_open = True
cb.open_until = 0
self.log("info", f"Circuit breaker '{dependency_name}' entering half-open state")
return True
return True # Circuit is closed or half-open
def record_circuit_success(self, dependency_name: str):
"""Reset circuit breaker after a successful call."""
cb = self.circuit_breakers.get(dependency_name)
if cb:
cb.failure_count = 0
cb.half_open = False
cb.open_until = 0
def record_circuit_failure(self, dependency_name: str):
"""Record a failure and potentially open the circuit."""
if dependency_name not in self.circuit_breakers:
self.circuit_breakers[dependency_name] = CircuitBreakerState()
cb = self.circuit_breakers[dependency_name]
cb.failure_count += 1
cb.last_failure_time = time.time()
if cb.failure_count >= self.circuit_breaker_threshold:
cb.open_until = time.time() + self.circuit_breaker_recovery
cb.half_open = False
self.log(
"error",
f"Circuit breaker '{dependency_name}' OPENED after {cb.failure_count} failures. "
f"Recovery in {self.circuit_breaker_recovery}s"
)
def run_with_resilience(self, agent_loop: Callable[[], None]):
"""
Execute the main agent loop with crash isolation.
If the agent loop raises an unhandled exception, we log it,
update metrics, and re-raise so the supervisor can restart.
"""
self.status = AgentStatus.STARTING
self.start_heartbeat()
self.log("info", f"Agent '{self.agent_name}' starting up")
self.status = AgentStatus.HEALTHY
self._write_heartbeat()
try:
while not self._shutdown_requested:
try:
agent_loop()
self.consecutive_failures = 0
self.status = AgentStatus.HEALTHY
self._write_heartbeat()
except Exception as e:
self.consecutive_failures += 1
self.log(
"error",
f"Agent loop iteration failed: {e}",
{"traceback": traceback.format_exc()}
)
if self.consecutive_failures >= self.max_consecutive_failures:
self.status = AgentStatus.UNHEALTHY
self._write_heartbeat()
self.log(
"critical",
f"Agent reached {self.consecutive_failures} consecutive failures. "
"Exiting so supervisor can restart with fresh state."
)
raise # Re-raise to trigger supervisor restart
self.status = AgentStatus.DEGRADED
self._write_heartbeat()
time.sleep(2) # Brief cooldown before retry
except Exception as fatal_error:
self.status = AgentStatus.UNHEALTHY
self._write_heartbeat()
self.log("critical", f"Fatal error: {fatal_error}")
raise
finally:
self.log("info", "Agent runtime shutting down")
self._shutdown_requested = True
self._write_heartbeat()3. A Complete Agent with the Runtime Wrapper
Here's how a real AI agent uses the AgentRuntime class. This example agent processes support tickets using an LLM, with circuit breakers protecting the external API calls and proper error isolation.
import time
import random
from agent_runtime import AgentRuntime # The class we built above
def fetch_pending_tickets():
"""
Simulate fetching tickets from a support queue.
In production, this would query Zendesk, Intercom, a database, etc.
"""
# Simulate occasional transient failures
if random.random() < 0.05:
raise ConnectionError("Database connection timeout")
# Return some mock tickets
return [
{"id": f"T-{i}", "subject": "Billing question", "priority": "high"}
for i in range(random.randint(1, 3))
]
def analyze_ticket_with_llm(ticket, runtime):
"""
Call an LLM API with circuit breaker protection.
"""
dependency = "llm_api"
if not runtime.check_circuit_breaker(dependency):
runtime.log("warn", f"Skipping LLM call for ticket {ticket['id']} — circuit breaker open")
return {"classification": "unknown", "confidence": 0.0}
try:
# Simulate LLM API call
if random.random() < 0.1:
raise RuntimeError("LLM API rate limited")
# Simulate successful classification
result = {
"classification": random.choice(["billing", "technical", "account"]),
"confidence": random.uniform(0.7, 0.99)
}
runtime.record_circuit_success(dependency)
return result
except Exception as e:
runtime.record_circuit_failure(dependency)
runtime.log("error", f"LLM call failed: {e}")
raise
def process_ticket(ticket, runtime):
"""Full ticket processing pipeline with error isolation per ticket."""
try:
analysis = analyze_ticket_with_llm(ticket, runtime)
runtime.log(
"info",
f"Ticket {ticket['id']} classified as {analysis['classification']} "
f"with confidence {analysis['confidence']:.2f}"
)
# In production: route ticket, send response, update database, etc.
except Exception as e:
runtime.log("error", f"Failed to process ticket {ticket['id']}: {e}")
# Per-ticket failures don't crash the whole agent
def main_agent_loop(runtime):
"""Single iteration of the agent's work loop."""
tickets = fetch_pending_tickets()
runtime.log("info", f"Fetched {len(tickets)} pending tickets")
for ticket in tickets:
process_ticket(ticket, runtime)
time.sleep(0.5) # Rate limit between tickets
# If no tickets, sleep before next poll
if not tickets:
time.sleep(10)
if __name__ == "__main__":
runtime = AgentRuntime(
agent_name="support-ticket-classifier",
heartbeat_path="/tmp/support_agent_heartbeat.json",
heartbeat_interval_seconds=5,
circuit_breaker_threshold=3,
circuit_breaker_recovery_seconds=90
)
# This call blocks. If it raises, the supervisor script restarts us.
runtime.run_with_resilience(lambda: main_agent_loop(runtime))4. External Health Check Script
The supervisor restarts the agent if it crashes, but what if the agent hangs without crashing? We need an external health checker that reads the heartbeat file and alerts if the agent is stale. Run this as a separate cron job or a parallel process.
#!/usr/bin/env python3
"""
external_health_check.py - Verify agent is alive via heartbeat file.
Run this every minute via cron or as a daemon.
"""
import json
import os
import sys
from datetime import datetime, timezone, timedelta
import requests # For Slack/Discord webhook alerts
HEARTBEAT_PATH = "/tmp/support_agent_heartbeat.json"
MAX_STALE_SECONDS = 120 # 2 minutes without heartbeat = alert
ALERT_WEBHOOK_URL = os.environ.get("ALERT_WEBHOOK_URL", "")
def check_heartbeat():
if not os.path.exists(HEARTBEAT_PATH):
print(f"CRITICAL: Heartbeat file {HEARTBEAT_PATH} not found")
return False
try:
with open(HEARTBEAT_PATH, "r") as f:
data = json.load(f)
except json.JSONDecodeError:
print(f"CRITICAL: Heartbeat file corrupted")
return False
timestamp_str = data.get("timestamp", "")
status = data.get("status", "unknown")
agent_name = data.get("agent", "unknown")
try:
last_beat = datetime.fromisoformat(timestamp_str)
except ValueError:
print(f"CRITICAL: Invalid timestamp format in heartbeat")
return False
now = datetime.now(timezone.utc)
age_seconds = (now - last_beat).total_seconds()
if age_seconds > MAX_STALE_SECONDS:
print(
f"CRITICAL: Agent '{agent_name}' heartbeat is {int(age_seconds)}s stale. "
f"Status: {status}. Agent may be hung."
)
return False
if status == "unhealthy":
print(f"WARNING: Agent '{agent_name}' reports UNHEALTHY status")
return False
print(f"OK: Agent '{agent_name}' healthy. Last heartbeat {int(age_seconds)}s ago.")
return True
def send_alert(message):
"""Send alert via webhook (Slack, Discord, etc.)"""
if not ALERT_WEBHOOK_URL:
return
try:
payload = {"text": f"🚨 AGENT ALERT: {message}"}
requests.post(ALERT_WEBHOOK_URL, json=payload, timeout=5)
except Exception as e:
print(f"Failed to send alert: {e}")
if __name__ == "__main__":
healthy = check_heartbeat()
if not healthy:
send_alert(f"Agent health check failed. Check supervisor and restart if needed.")
sys.exit(1)
sys.exit(0)5. Systemd Service Unit (Linux)
On Linux servers, wrapping the supervisor in a systemd service gives you automatic startup on boot, log capture via journald, and integration with the OS process tree. This replaces the bare bash supervisor loop with OS-level supervision.
[Unit]
Description=AI Support Ticket Classifier Agent
After=network.target
[Service]
Type=simple
User=agentuser
WorkingDirectory=/opt/agents/support-classifier
ExecStart=/opt/agents/support-classifier/supervisor.sh
Restart=no
# Restart is handled by the supervisor script internally
StandardOutput=journal
StandardError=journal
SyslogIdentifier=support-agent
# Prevent the agent from being killed by OOM killer if possible
OOMScoreAdjust=-500
# Give the agent time for graceful shutdown
TimeoutStopSec=30
KillSignal=SIGTERM
[Install]
WantedBy=multi-user.targetEnable and start with:
sudo systemctl enable support-agent.service
sudo systemctl start support-agent.service
# Check status and logs
sudo systemctl status support-agent.service
journalctl -u support-agent.service -f6. Docker Compose Deployment with Health Checks
For containerized deployments, Docker's built-in health check mechanism can serve as your external watchdog without any additional infrastructure.
# docker-compose.yml
version: '3.8'
services:
support-agent:
build: .
container_name: support-agent
restart: unless-stopped
volumes:
- ./agent_logs:/app/agent_logs
- agent_heartbeat:/tmp
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- ALERT_WEBHOOK_URL=${ALERT_WEBHOOK_URL}
healthcheck:
test: ["CMD", "python", "/app/external_health_check.py"]
interval: 60s
timeout: 10s
retries: 3
start_period: 30s
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
volumes:
agent_heartbeat:The Docker health check runs the same external health check script we wrote earlier. If it fails 3 times consecutively, Docker marks the container as unhealthy and — combined with restart: unless-stopped — can trigger an automatic restart.
Best Practices for Long-Running AI Agents
1. Never Trust a Single Process
Always assume your agent process can and will die. Build your architecture around that assumption. The supervisor pattern costs almost nothing to implement and catches 80% of production failures. A bash script with 20 lines is production-grade if it reliably restarts your agent.
2. Heartbeats Must Be External to the Main Loop
The heartbeat must run in a separate thread or process. If your agent's main loop blocks on a hung API call for 5 minutes, a heartbeat written inside the same loop won't fire either. The background thread approach we used above solves this — it keeps ticking even when the main loop is stuck.
3. Circuit Breakers Prevent Cascade Failures
A single flaky dependency shouldn't take down your entire agent. Circuit breakers are simple state machines that cost ~30 lines of code and prevent the agent from repeatedly hammering a failing service. The key insight: after N consecutive failures, stop trying for a cooldown period. This gives downstream services time to recover and prevents your agent from burning through retries that will also fail.
4. Log Like Someone Will Read It at 3 AM
Structured logging (JSON lines) with timestamps, agent names, and status fields makes debugging trivial. When your agent crashes at 2:47 AM, you want to open one file and immediately see: what was the agent doing, which dependency failed, was the circuit breaker open, and what was the exact exception traceback. Plain text logs scattered across stdout don't cut it.
5. Separate Crash Recovery from Bug Fixes
Your supervisor restarts the agent after a crash, but if the crash is caused by a deterministic bug (e.g., a malformed API response that always triggers the same exception), the agent will crash again immediately. Track restart frequency. The supervisor script we built caps restarts per hour and sleeps for 10 minutes if the threshold is exceeded — this prevents CPU-burning crash loops and gives you time to deploy a fix.
6. Test Failure Scenarios Explicitly
Don't just test that your agent works. Test that it fails well. Simulate:
- Killing the process with
kill -9— does the supervisor restart it? - Blocking the LLM API with a firewall rule — does the circuit breaker open?
- Filling the disk where heartbeat is written — does the external health check detect the stale heartbeat?
- Running the agent for 48 hours straight — does memory usage grow unbounded?
7. Monitor Memory and Set Resource Limits
AI agents accumulate state: conversation histories, embedding caches, tool call results. Use memory profiling and set explicit limits. In Python, consider resource.setrlimit or container memory limits. If an agent approaches its limit, log a warning and trigger a graceful restart (exit with a specific code that the supervisor recognizes as "intentional restart due to memory pressure" rather than a crash).
8. Use Idempotent Operations and Durable State
When your agent restarts, it must not re-process work it already completed. Use external durable state (database, queue with checkpoints) rather than in-memory state. Each unit of work should have a status field: pending, processing, completed, failed. On restart, the agent scans for processing items (work that was interrupted mid-flight) and handles them appropriately — either retry with caution or mark as failed for human review.
Conclusion
Keeping an AI agent alive 24/7 without a DevOps team is not about complex infrastructure — it's about disciplined patterns baked into your agent's runtime. A 20-line bash supervisor, a background heartbeat thread, a circuit breaker for each external dependency, and an external health check that alerts you when things go wrong. That's it. These four patterns, implemented with the code in this tutorial, will catch the vast majority of production failures and recover from them automatically.
The key mindset shift: stop treating your AI agent as a script and start treating it as a service. Services crash, hang, and degrade. They need supervision, health checks, and graceful failure handling. The patterns here give you that without requiring Kubernetes, PagerDuty, or an SRE team. Build them in from day one, and your agent will survive the chaos of production — while you sleep through the night.