What Is Disaster Recovery for AI Agent Businesses?
Disaster Recovery (DR) for AI agent businesses refers to the comprehensive set of policies, procedures, and technical implementations that ensure your AI-powered services can survive and quickly recover from catastrophic failures. Unlike traditional application DR, AI agent systems introduce unique challenges: they depend on large language model APIs, maintain complex conversation state, rely on vector databases for retrieval-augmented generation (RAG), and often orchestrate multiple tool-calling sequences that must be preserved or safely replayed.
A well-designed DR plan for an AI agent business answers one fundamental question: When critical infrastructure fails, how do we restore agent functionality with minimal data loss and downtime? The answer involves backing up not just code and databases, but also model configurations, embedding indices, prompt templates, agent memory stores, and the intricate web of third-party API dependencies that agents rely on to perform tasks.
Why Disaster Recovery Matters for AI Agents
AI agent businesses face several failure modes that traditional SaaS applications rarely encounter:
- LLM Provider Outages: If OpenAI, Anthropic, or your chosen provider goes down, your agents stop thinking entirely. Without a multi-provider fallback strategy, your entire product becomes inert.
- Vector Database Corruption: RAG-powered agents depend on embedding indices. A corrupted vector store silently returns garbage context, causing agents to hallucinate or provide incorrect answers at scale.
- Agent State Loss: Long-running agent workflows (multi-step reasoning, tool-calling loops) hold in-flight state. Losing this state mid-execution can leave customer tasks incomplete, charge customers incorrectly, or corrupt business data.
- API Key Compromise or Expiry: Agents use multiple API keys for tools, models, and data sources. A single expired key can cascade into widespread agent failures.
- Prompt Registry Destruction: If your curated prompt templates and few-shot examples are lost, agents revert to unpredictable behavior, potentially damaging customer trust.
For AI agent businesses, downtime doesn't just mean lost revenue—it means active harm. An agent that continues running with degraded reasoning or missing context can make erroneous decisions, send incorrect communications to customers, or corrupt downstream systems. DR is therefore not optional; it's a fundamental requirement of responsible agent deployment.
Core Components of an AI Agent DR Plan
An effective DR plan for AI agent businesses must address these critical layers:
1. Model Endpoint Redundancy
Your agents must not be hard-coupled to a single LLM provider. Implement a routing layer that can switch providers based on availability, latency, or cost. This layer should also handle API key rotation and quota management.
2. Prompt and Configuration Backup
Prompt templates, system messages, few-shot examples, and agent tool definitions should be treated as version-controlled assets stored in a durable, replicated object store (e.g., S3 with cross-region replication).
3. Vector Store Snapshots and Replication
Embedding indices must support point-in-time snapshots and incremental backup. For Pinecone, Weaviate, or pgvector-based stores, implement regular snapshot routines and test restoration procedures.
4. Agent State and Memory Persistence
Conversation histories, agent scratchpads, and tool-calling intermediate results must be persisted to a durable, replicated data store. This enables warm recovery of in-progress agent sessions.
5. Tool and API Dependency Management
Catalog all external APIs your agents call (search APIs, CRM integrations, payment processors) and implement circuit breakers with graceful degradation so agents can continue operating—albeit with reduced capability—when a dependency fails.
How to Implement Disaster Recovery: A Step-by-Step Guide
Step 1: Inventory Your Agent Dependencies
Begin by creating a comprehensive dependency map. Document every external service, model endpoint, database, and configuration source your agents touch. The following script automates dependency discovery by scanning your agent codebase for API calls and configuration references:
#!/usr/bin/env python3
"""Agent Dependency Scanner - discovers all external dependencies in your agent codebase."""
import ast
import os
import re
from pathlib import Path
from collections import defaultdict
class DependencyScanner:
def __init__(self, root_path: str):
self.root_path = Path(root_path)
self.dependencies = defaultdict(list)
def scan_python_files(self):
for py_file in self.root_path.rglob("*.py"):
with open(py_file, 'r') as f:
try:
tree = ast.parse(f.read(), filename=str(py_file))
self._extract_calls(tree, py_file)
except SyntaxError:
continue
return self.dependencies
def _extract_calls(self, tree, source_file):
for node in ast.walk(tree):
# Detect HTTP requests
if isinstance(node, ast.Call):
if isinstance(node.func, ast.Attribute):
func_name = node.func.attr
if func_name in ('get', 'post', 'put', 'patch', 'delete'):
self._record_dependency(node, source_file, 'HTTP')
elif isinstance(node.func, ast.Name):
if node.func.id in ('openai', 'anthropic', 'pinecone', 'weaviate'):
self._record_dependency(node, source_file, 'SDK_CALL')
def _record_dependency(self, node, source_file, call_type):
# Extract string arguments (URLs, endpoints)
for arg in node.args:
if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
self.dependencies[call_type].append({
'file': str(source_file),
'value': arg.value,
'line': node.lineno
})
def scan_env_files(self):
env_pattern = re.compile(r'^[A-Z_]+=.*$')
for env_file in self.root_path.rglob(".env*"):
with open(env_file, 'r') as f:
for line in f:
if env_pattern.match(line.strip()):
key = line.split('=')[0].strip()
self.dependencies['ENV_VAR'].append({
'file': str(env_file),
'key': key
})
return self.dependencies
def generate_report(self):
report = "# Agent Dependency Report\n\n"
for category, items in self.dependencies.items():
report += f"## {category} ({len(items)} found)\n\n"
for item in items:
report += f"- **{item.get('value', item.get('key'))}** in `{item['file']}` (line {item.get('line', 'N/A')})\n"
report += "\n"
return report
if __name__ == "__main__":
scanner = DependencyScanner("./agent_codebase")
scanner.scan_python_files()
scanner.scan_env_files()
print(scanner.generate_report())
# Save report
with open("dependency_report.md", "w") as f:
f.write(scanner.generate_report())Run this scanner regularly and store the dependency report alongside your DR documentation. Every dependency in this report needs a recovery strategy.
Step 2: Build a Multi-Provider LLM Router
The heart of AI agent DR is the ability to seamlessly switch between LLM providers. Below is a production-grade routing layer that implements health checking, automatic failover, and graceful degradation:
import asyncio
import time
import logging
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
logger = logging.getLogger("llm_router")
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded" # Slow but responding
FAILING = "failing" # Intermittent errors
DOWN = "down" # Complete outage
@dataclass
class ProviderConfig:
name: str
api_key_env: str
base_url: str
models: List[str]
max_retries: int = 3
timeout_seconds: float = 30.0
weight: float = 1.0 # Preference weight when multiple providers are healthy
@dataclass
class ProviderState:
config: ProviderConfig
status: ProviderStatus = ProviderStatus.HEALTHY
consecutive_failures: int = 0
last_check_time: float = 0.0
average_latency_ms: float = 0.0
circuit_open: bool = False
circuit_open_until: float = 0.0
@dataclass
class RoutingDecision:
provider_name: str
model: str
fallback_used: bool
providers_attempted: int
class LLMRouter:
"""Multi-provider router with health checks, circuit breakers, and automatic failover."""
def __init__(self, providers: List[ProviderConfig], health_check_interval: float = 10.0):
self.providers: Dict[str, ProviderState] = {}
self.health_check_interval = health_check_interval
self.failure_threshold = 5 # Consecutive failures before marking DOWN
self.recovery_time_window = 60.0 # Seconds before attempting recovery
self._health_check_task = None
for config in providers:
self.providers[config.name] = ProviderState(config=config)
async def start(self):
"""Begin background health checking."""
self._health_check_task = asyncio.create_task(self._health_check_loop())
logger.info("LLM Router started with %d providers", len(self.providers))
async def stop(self):
if self._health_check_task:
self._health_check_task.cancel()
async def _health_check_loop(self):
while True:
await asyncio.sleep(self.health_check_interval)
for name, state in self.providers.items():
if state.circuit_open and time.time() < state.circuit_open_until:
continue
await self._check_provider(name, state)
async def _check_provider(self, name: str, state: ProviderState):
"""Ping provider with a minimal completion request."""
import os
api_key = os.getenv(state.config.api_key_env)
if not api_key:
state.status = ProviderStatus.DOWN
logger.error("Provider %s: missing API key env var %s", name, state.config.api_key_env)
return
try:
start = time.monotonic()
# Lightweight health check - simple token count or models list
# This avoids consuming expensive tokens for health checks
import httpx
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(
f"{state.config.base_url}/models",
headers={"Authorization": f"Bearer {api_key}"}
)
latency_ms = (time.monotonic() - start) * 1000
state.last_check_time = time.time()
state.average_latency_ms = (
0.7 * state.average_latency_ms + 0.3 * latency_ms
if state.average_latency_ms > 0 else latency_ms
)
if response.status_code == 200:
state.status = ProviderStatus.HEALTHY
state.consecutive_failures = 0
state.circuit_open = False
logger.debug("Provider %s: healthy (latency %.1fms)", name, latency_ms)
else:
state.consecutive_failures += 1
self._evaluate_failure(name, state)
except Exception as e:
state.consecutive_failures += 1
logger.warning("Provider %s health check failed: %s", name, str(e))
self._evaluate_failure(name, state)
def _evaluate_failure(self, name: str, state: ProviderState):
if state.consecutive_failures >= self.failure_threshold:
state.status = ProviderStatus.DOWN
state.circuit_open = True
state.circuit_open_until = time.time() + self.recovery_time_window
logger.error("Provider %s: circuit OPEN - marked DOWN for %.0fs",
name, self.recovery_time_window)
elif state.consecutive_failures >= 3:
state.status = ProviderStatus.FAILING
async def route_completion(
self,
messages: List[Dict[str, str]],
preferred_model: Optional[str] = None,
max_fallbacks: int = 3
) -> RoutingDecision:
"""Route a completion request to the best available provider."""
import os, httpx
attempted = 0
last_error = None
# Build priority queue: healthy providers sorted by weight and latency
candidates = sorted(
[
s for s in self.providers.values()
if s.status in (ProviderStatus.HEALTHY, ProviderStatus.DEGRADED)
and not s.circuit_open
],
key=lambda s: (-s.config.weight, s.average_latency_ms)
)
if not candidates:
raise Exception("No healthy LLM providers available")
for state in candidates:
if attempted >= max_fallbacks:
break
attempted += 1
model = preferred_model or state.config.models[0]
api_key = os.getenv(state.config.api_key_env)
try:
async with httpx.AsyncClient(timeout=state.config.timeout_seconds) as client:
response = await client.post(
f"{state.config.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": messages,
"max_tokens": 1024
}
)
if response.status_code == 200:
return RoutingDecision(
provider_name=state.config.name,
model=model,
fallback_used=(attempted > 1),
providers_attempted=attempted
)
else:
last_error = f"HTTP {response.status_code}: {response.text[:200]}"
logger.warning("Provider %s returned %s", state.config.name, last_error)
except Exception as e:
last_error = str(e)
logger.warning("Provider %s failed: %s", state.config.name, str(e))
state.consecutive_failures += 1
self._evaluate_failure(state.config.name, state)
raise Exception(f"All providers exhausted after {attempted} attempts. Last error: {last_error}")
# ==== USAGE EXAMPLE ====
async def main():
providers = [
ProviderConfig(
name="openai",
api_key_env="OPENAI_API_KEY",
base_url="https://api.openai.com/v1",
models=["gpt-4-turbo", "gpt-3.5-turbo"],
weight=1.0
),
ProviderConfig(
name="anthropic",
api_key_env="ANTHROPIC_API_KEY",
base_url="https://api.anthropic.com/v1",
models=["claude-3-opus", "claude-3-sonnet"],
weight=0.9
),
ProviderConfig(
name="azure_openai",
api_key_env="AZURE_OPENAI_API_KEY",
base_url="https://your-resource.openai.azure.com",
models=["gpt-4-deployment"],
weight=0.8
),
ProviderConfig(
name="local_fallback",
api_key_env="LOCAL_LLM_API_KEY",
base_url="http://localhost:8080/v1",
models=["llama-3-70b"],
weight=0.5,
timeout_seconds=10.0
),
]
router = LLMRouter(providers, health_check_interval=15.0)
await router.start()
try:
decision = await router.route_completion(
messages=[{"role": "user", "content": "Summarize the quarterly earnings report."}],
preferred_model="gpt-4-turbo"
)
print(f"Routed to {decision.provider_name} using {decision.model}")
print(f"Fallback used: {decision.fallback_used}, Attempts: {decision.providers_attempted}")
except Exception as e:
print(f"Critical routing failure: {e}")
# Trigger full DR mode - queue request for later or notify ops team
await router.stop()
if __name__ == "__main__":
asyncio.run(main())This router provides automatic health checking, circuit breaking, and weighted provider selection. When your primary provider fails, requests seamlessly flow to the next available option without any code changes in your agent logic.
Step 3: Vector Store Backup and Restoration
Your vector database is the memory backbone of RAG-powered agents. Losing it means your agents lose access to your entire knowledge base. Implement regular snapshot backups with verified restoration:
#!/usr/bin/env python3
"""Vector Store Snapshot Manager - backs up and restores Pinecone indexes with verification."""
import os
import json
import time
import hashlib
import logging
from datetime import datetime, timezone
from typing import List, Dict, Optional
from dataclasses import dataclass
import boto3
from pinecone import Pinecone
logger = logging.getLogger("vector_snapshot")
@dataclass
class SnapshotMetadata:
snapshot_id: str
index_name: str
namespace: str
vector_count: int
created_at: str
checksum: str
dimensions: int
class VectorStoreSnapshotManager:
"""Manages automated snapshots of Pinecone vector indexes with S3 backup."""
def __init__(
self,
pinecone_api_key: str,
pinecone_environment: str,
s3_bucket: str,
s3_prefix: str = "vector-snapshots/",
aws_region: str = "us-east-1"
):
self.pc = Pinecone(api_key=pinecone_api_key)
self.s3 = boto3.client('s3', region_name=aws_region)
self.bucket = s3_bucket
self.prefix = s3_prefix
self.index_cache = {}
def list_indexes(self) -> List[str]:
"""Return all available Pinecone indexes."""
return [idx['name'] for idx in self.pc.list_indexes()]
def create_snapshot(self, index_name: str, namespace: str = "") -> SnapshotMetadata:
"""Create a full snapshot of a Pinecone index namespace and upload to S3."""
logger.info(f"Starting snapshot for index={index_name}, namespace={namespace or 'default'}")
index = self.pc.Index(index_name)
# Get index stats for vector count
stats = index.describe_index_stats()
ns_stats = stats.get('namespaces', {}).get(namespace, {})
vector_count = ns_stats.get('vector_count', 0)
if vector_count == 0:
logger.warning(f"Namespace {namespace or 'default'} is empty, skipping snapshot")
return None
snapshot_id = f"snap-{index_name}-{namespace or 'default'}-{int(time.time() * 1000)}"
snapshot_data = []
total_vectors = 0
# Fetch all vectors in batches using pagination
# Pinecone fetch limit: fetch IDs in chunks of 1000
# First, list all vector IDs via a dummy query or by maintaining an ID registry
# For production, maintain a separate vector ID registry table
# Here we demonstrate the fetch approach for known IDs
vector_ids = self._get_all_vector_ids(index, namespace)
batch_size = 100 # Pinecone fetch limit
for i in range(0, len(vector_ids), batch_size):
batch_ids = vector_ids[i:i + batch_size]
fetch_result = index.fetch(ids=batch_ids, namespace=namespace)
for vector_id, vector_data in fetch_result.vectors.items():
snapshot_data.append({
'id': vector_id,
'values': vector_data.values,
'metadata': vector_data.metadata if vector_data.metadata else {},
'sparse_values': vector_data.sparse_values if hasattr(vector_data, 'sparse_values') else None
})
total_vectors += 1
logger.debug(f"Fetched batch {i//batch_size + 1}: {total_vectors}/{vector_count} vectors")
# Serialize and compute checksum
snapshot_json = json.dumps(snapshot_data, ensure_ascii=False)
checksum = hashlib.sha256(snapshot_json.encode()).hexdigest()
metadata = SnapshotMetadata(
snapshot_id=snapshot_id,
index_name=index_name,
namespace=namespace or 'default',
vector_count=total_vectors,
created_at=datetime.now(timezone.utc).isoformat(),
checksum=checksum,
dimensions=stats.get('dimension', 0)
)
# Upload to S3 with metadata
snapshot_key = f"{self.prefix}{snapshot_id}/vectors.json"
metadata_key = f"{self.prefix}{snapshot_id}/metadata.json"
self.s3.put_object(
Bucket=self.bucket,
Key=snapshot_key,
Body=snapshot_json,
ContentType='application/json',
Metadata={'checksum': checksum, 'vector_count': str(total_vectors)}
)
self.s3.put_object(
Bucket=self.bucket,
Key=metadata_key,
Body=json.dumps(metadata.__dict__, indent=2),
ContentType='application/json'
)
logger.info(f"Snapshot {snapshot_id} complete: {total_vectors} vectors uploaded to s3://{self.bucket}/{snapshot_key}")
return metadata
def _get_all_vector_ids(self, index, namespace: str) -> List[str]:
"""Retrieve all vector IDs from a namespace.
In production, use a dedicated ID registry table (DynamoDB/Postgres)
rather than relying on list operations which may be limited."""
ids = []
# For Pinecone serverless, use a paginated list approach
# This is a simplified example - production systems should maintain
# a separate durable ID registry
try:
# Attempt to list IDs if supported by the index type
stats = index.describe_index_stats()
total = stats.get('namespaces', {}).get(namespace, {}).get('vector_count', 0)
if total > 0:
# In practice, maintain a PostgreSQL/DynamoDB table of vector IDs
logger.info(f"Namespace has {total} vectors - use ID registry for production")
# Placeholder: return empty if no registry
return []
except Exception as e:
logger.warning(f"Could not list vector IDs: {e}")
return ids
def restore_from_snapshot(
self,
snapshot_id: str,
target_index: str,
target_namespace: str = "",
batch_size: int = 100
) -> bool:
"""Restore vectors from an S3 snapshot to a Pinecone index."""
logger.info(f"Restoring snapshot {snapshot_id} to index={target_index}, namespace={target_namespace}")
# Load metadata and verify
metadata_key = f"{self.prefix}{snapshot_id}/metadata.json"
try:
metadata_obj = self.s3.get_object(Bucket=self.bucket, Key=metadata_key)
metadata = SnapshotMetadata(**json.loads(metadata_obj['Body'].read()))
except Exception as e:
logger.error(f"Failed to load snapshot metadata: {e}")
return False
# Load vectors
snapshot_key = f"{self.prefix}{snapshot_id}/vectors.json"
try:
vector_obj = self.s3.get_object(Bucket=self.bucket, Key=snapshot_key)
vector_data = json.loads(vector_obj['Body'].read())
except Exception as e:
logger.error(f"Failed to load vector data: {e}")
return False
# Verify checksum
computed_checksum = hashlib.sha256(
json.dumps(vector_data, ensure_ascii=False).encode()
).hexdigest()
if computed_checksum != metadata.checksum:
logger.error(f"Checksum mismatch! Expected {metadata.checksum}, got {computed_checksum}")
return False
# Upsert vectors in batches
target_idx = self.pc.Index(target_index)
total_upserted = 0
for i in range(0, len(vector_data), batch_size):
batch = vector_data[i:i + batch_size]
vectors_to_upsert = []
for vec in batch:
upsert_vec = {
'id': vec['id'],
'values': vec['values'],
'metadata': vec.get('metadata', {})
}
vectors_to_upsert.append(upsert_vec)
target_idx.upsert(vectors=vectors_to_upsert, namespace=target_namespace)
total_upserted += len(vectors_to_upsert)
logger.info(f"Restored batch: {total_upserted}/{len(vector_data)} vectors")
logger.info(f"Restoration complete: {total_upserted} vectors restored to {target_index}")
return True
def list_snapshots(self) -> List[SnapshotMetadata]:
"""List all available snapshots in S3."""
snapshots = []
paginator = self.s3.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket=self.bucket, Prefix=self.prefix, Delimiter='/'):
for prefix in page.get('CommonPrefixes', []):
snap_id = prefix['Prefix'].rstrip('/').split('/')[-1]
try:
meta = self.s3.get_object(Bucket=self.bucket, Key=f"{prefix['Prefix']}metadata.json")
snapshots.append(SnapshotMetadata(**json.loads(meta['Body'].read())))
except:
continue
return sorted(snapshots, key=lambda s: s.created_at, reverse=True)
# ==== USAGE EXAMPLE ====
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
manager = VectorStoreSnapshotManager(
pinecone_api_key=os.environ["PINECONE_API_KEY"],
pinecone_environment="us-east-1",
s3_bucket="my-agent-dr-bucket",
s3_prefix="vector-snapshots/production/",
aws_region="us-east-1"
)
# Create daily snapshot
snapshot_meta = manager.create_snapshot("my-knowledge-index", namespace="legal-docs")
if snapshot_meta:
print(f"Snapshot created: {snapshot_meta.snapshot_id} with {snapshot_meta.vector_count} vectors")
# List available snapshots
all_snapshots = manager.list_snapshots()
for snap in all_snapshots[:5]:
print(f" {snap.snapshot_id} | {snap.vector_count} vectors | {snap.created_at}")
# Restore from latest snapshot to a recovery index
if all_snapshots:
latest = all_snapshots[0]
restored = manager.restore_from_snapshot(
snapshot_id=latest.snapshot_id,
target_index="my-knowledge-index-recovery",
target_namespace="legal-docs"
)
print(f"Restore {'succeeded' if restored else 'failed'}")Step 4: Agent State Persistence and Warm Recovery
Long-running agent workflows need durable state persistence so that in-progress operations can resume after a crash. This example shows a checkpoint system using Redis with PostgreSQL as the durable backing store:
import json
import hashlib
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from datetime import datetime, timezone
import asyncpg
import redis.asyncio as redis
@dataclass
class AgentCheckpoint:
session_id: str
step_index: int
messages: List[Dict[str, Any]]
tool_calls_pending: List[Dict[str, Any]]
tool_results: Dict[str, Any]
scratchpad: str
created_at: str
checksum: str
class AgentStateManager:
"""Durable agent state persistence with Redis hot cache and PostgreSQL cold storage."""
def __init__(self, redis_url: str, postgres_dsn: str):
self.redis = None
self.pg_pool = None
self.redis_url = redis_url
self.postgres_dsn = postgres_dsn
self.checkpoint_retention_days = 30
async def start(self):
self.redis = await redis.from_url(self.redis_url)
self.pg_pool = await asyncpg.create_pool(self.postgres_dsn, min_size=2, max_size=10)
# Ensure PostgreSQL schema exists
async with self.pg_pool.acquire() as conn:
await conn.execute("""
CREATE TABLE IF NOT EXISTS agent_checkpoints (
session_id TEXT NOT NULL,
step_index INTEGER NOT NULL,
checkpoint_data JSONB NOT NULL,
checksum TEXT NOT NULL,
created_at TIMESTAMPTZ DEFAULT NOW(),
PRIMARY KEY (session_id, step_index)
);
CREATE TABLE IF NOT EXISTS agent_sessions (
session_id TEXT PRIMARY KEY,
status TEXT NOT NULL DEFAULT 'active',
current_step INTEGER DEFAULT 0,
total_steps INTEGER,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_checkpoints_session
ON agent_checkpoints(session_id, created_at DESC);
""")
print("AgentStateManager started")
async def save_checkpoint(self, checkpoint: AgentCheckpoint) -> bool:
"""Save agent checkpoint to Redis (hot) and PostgreSQL (durable)."""
checkpoint_json = json.dumps({
'session_id': checkpoint.session_id,
'step_index': checkpoint.step_index,
'messages': checkpoint.messages,
'tool_calls_pending': checkpoint.tool_calls_pending,
'tool_results': checkpoint.tool_results,
'scratchpad': checkpoint.scratchpad,
'created_at': checkpoint.created_at
}, ensure_ascii=False)
computed_checksum = hashlib.sha256(checkpoint_json.encode()).hexdigest()
checkpoint.checksum = computed_checksum
# Write to Redis first (fast, hot path)
redis_key = f"agent:checkpoint:{checkpoint.session_id}:{checkpoint.step_index}"
await self.redis.set(redis_key, checkpoint_json, ex=3600) # 1 hour TTL in Redis
# Persist to PostgreSQL (durable, cold storage)
async with self.pg_pool.acquire() as conn:
await conn.execute("""
INSERT INTO agent_checkpoints (session_id, step_index, checkpoint_data, checksum, created_at)
VALUES ($1, $2, $3::jsonb, $4, $5)
ON CONFLICT (session_id, step_index) DO UPDATE SET
checkpoint_data = EXCLUDED.checkpoint_data,
checksum = EXCLUDED.checksum,
created_at = EXCLUDED.created_at
""", checkpoint.session_id, checkpoint.step_index, checkpoint_json,
computed_checksum, datetime.now(timezone.utc))
# Update session tracking
await conn.execute("""
INSERT INTO agent_sessions (session_id, status, current_step, updated_at)
VALUES ($1, 'active', $2, NOW())
ON CONFLICT (session_id) DO UPDATE SET
current_step = GREATEST(agent_sessions.current_step, $2),
updated_at = NOW()
""", checkpoint.session_id, checkpoint.step_index)
return True
async def load_latest_checkpoint(self, session_id: str) -> Optional[AgentCheckpoint]:
"""Load the most recent checkpoint for a session. Checks Redis first, then PostgreSQL."""
# Try Redis first for hot data
session_pattern = f"agent:checkpoint:{session_id}:*"
redis_keys = await self.redis.keys(session_pattern)
if redis_keys:
# Find the highest step index
max_step = -1
latest_data = None
for key in redis_keys:
step = int(key.decode().split(':')[-1])
if step > max_step:
max_step = step
latest_data = await self.redis.get(key)
if latest_data:
data = json.loads(latest_data.decode())
return AgentCheckpoint(**data)
# Fall back to PostgreSQL
async with self.pg_pool.acquire() as conn:
row = await conn.fetchrow("""
SELECT session_id, step_index, checkpoint_data, checksum,
created_at::text as created_at
FROM agent_checkpoints
WHERE session_id = $1
ORDER BY step_index DESC
LIMIT 1
""", session_id)
if row:
checkpoint_data = json.loads(row['checkpoint_data'])
return AgentCheckpoint(
session_id=row['session_id'],
step_index=row['step_index'],
messages=checkpoint_data.get('messages', []),
tool_calls_pending=checkpoint_data.get('tool_calls_pending', []),
tool_results=checkpoint_data.get('tool_results', {}),
scratchpad=checkpoint_data.get('scratchpad', ''),
created_at=row['created_at'],
checksum=row['checksum']
)
return None
async def resume_session(self, session_id: str) -> Optional[Dict[str, Any]]:
"""Resume an agent session from the latest checkpoint."""
checkpoint = await self.load_latest_checkpoint(session_id)
if not checkpoint:
return None
# Verify checksum integrity
raw_data = json.dumps({
'session_id': checkpoint.session_id,
'step_index': checkpoint.step_index,
'messages': checkpoint.messages,
'tool_calls_pending': checkpoint.tool_calls_pending,
'tool_results': checkpoint.tool_results,
'scratchpad': checkpoint.scratchpad,
'created_at': checkpoint.created_at
}, ensure_ascii=False)
if hashlib.sha256(raw_data.encode()).hexdigest() != checkpoint.checksum:
raise ValueError(f"Checksum mismatch for session {session_id} - possible data corruption")
return {
'session_id': checkpoint.session_id,
'res