AI Agent Backup and Recovery: What It Is
AI agent backup and recovery is the systematic process of preserving an agent's internal state — its memory, conversation history, vector embeddings, tool outputs, and decision-making context — so that it can be fully restored after a crash, migration, or system failure. Think of it as a save point for your autonomous AI system. Without it, every unexpected shutdown means your agent wakes up with amnesia, losing hours or days of accumulated context and progress.
An AI agent's "memory" isn't just a chat log. It typically includes:
- Short-term memory — the current conversation window, pending actions, and in-flight reasoning chains
- Long-term memory — vectorized knowledge stored in databases like Pinecone, Chroma, or Weaviate
- Episodic memory — records of past interactions, decisions, and their outcomes
- Tool execution state — partially completed API calls, file operations, or multi-step workflows
- Agent configuration — dynamic parameters, feature flags, and runtime-injected prompts
Backup and recovery gives you resilience. It transforms your agent from a fragile prototype into a production-grade system that can survive pod restarts, cloud outages, or even deliberate shutdowns for maintenance.
Why Agent Memory Backup Matters
Consider a customer support agent that has been working on a complex ticket for 45 minutes. It has gathered 17 pieces of information, made 3 tool calls, and is about to escalate to a human with a perfect summary. If the container restarts right then, all of that evaporates. The human gets nothing. The customer has to start over. The business loses trust.
Here's what's actually at stake:
- User experience continuity — users expect agents to remember context across sessions
- Cost efficiency — re-computing lost context via LLM calls costs money and time
- Compliance and auditability — regulated industries require traceable decision trails
- Multi-turn workflow integrity — complex agent workflows with tool chains can't survive mid-process crashes
- Developer sanity — debugging an agent that randomly loses memory is a nightmare
Backup and recovery isn't optional once your agent serves real users or performs consequential tasks. It's foundational infrastructure, just like database backups for a web application.
Architecture of Agent State Backup
The Serializable State Pattern
The core idea is to design your agent so that its complete internal state can be serialized to a portable format (JSON, MessagePack, or Protocol Buffers) at any checkpoint, and then deserialized back into a live agent later. This requires you to think about state boundaries carefully.
Here's a minimal agent state interface in Python using Pydantic for strong typing:
from pydantic import BaseModel, Field
from datetime import datetime
from typing import List, Dict, Any, Optional
import uuid
class Message(BaseModel):
role: str # 'user', 'assistant', 'system', 'tool'
content: str
timestamp: datetime = Field(default_factory=datetime.utcnow)
message_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
class ToolCall(BaseModel):
tool_name: str
arguments: Dict[str, Any]
result: Optional[Any] = None
status: str = 'pending' # 'pending', 'running', 'completed', 'failed'
call_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
class AgentState(BaseModel):
session_id: str
conversation_history: List[Message] = []
pending_tool_calls: List[ToolCall] = []
long_term_memory_keys: List[str] = [] # references to vector DB entries
scratchpad: str = "" # reasoning chain in progress
checkpoint_number: int = 0
created_at: datetime = Field(default_factory=datetime.utcnow)
last_updated: datetime = Field(default_factory=datetime.utcnow)This state object becomes the single source of truth that you'll persist and restore.
How to Implement Backup and Recovery
Step 1: Make Your Agent Checkpoint-Aware
Your agent loop needs to periodically pause and save state. Here's a simplified agent run loop with checkpoint integration:
import json
import os
from typing import Callable
class CheckpointManager:
def __init__(self, base_path: str = "./agent_checkpoints"):
self.base_path = base_path
os.makedirs(base_path, exist_ok=True)
def save(self, state: AgentState) -> str:
"""Save agent state to disk, return the checkpoint path."""
state.checkpoint_number += 1
state.last_updated = datetime.utcnow()
checkpoint_path = os.path.join(
self.base_path,
f"{state.session_id}_checkpoint_{state.checkpoint_number}.json"
)
with open(checkpoint_path, 'w') as f:
f.write(state.model_dump_json(indent=2))
return checkpoint_path
def load_latest(self, session_id: str) -> Optional[AgentState]:
"""Load the most recent checkpoint for a session."""
pattern = f"{session_id}_checkpoint_"
checkpoints = [
f for f in os.listdir(self.base_path)
if f.startswith(pattern)
]
if not checkpoints:
return None
# Sort by checkpoint number
checkpoints.sort(
key=lambda x: int(x.split('_checkpoint_')[1].split('.')[0]),
reverse=True
)
with open(os.path.join(self.base_path, checkpoints[0]), 'r') as f:
data = json.load(f)
return AgentState.model_validate(data)
# Usage inside the agent loop
async def agent_loop(
state: AgentState,
process_turn: Callable,
checkpoint_manager: CheckpointManager,
checkpoint_every_n_turns: int = 5
):
turn_count = 0
while True:
user_input = await get_next_input()
state.conversation_history.append(
Message(role="user", content=user_input)
)
# Process the turn
response = await process_turn(state)
state.conversation_history.append(
Message(role="assistant", content=response)
)
turn_count += 1
if turn_count % checkpoint_every_n_turns == 0:
checkpoint_path = checkpoint_manager.save(state)
print(f"Checkpoint saved: {checkpoint_path}")
Step 2: Backup Vector Memory and External Stores
Long-term memory in vector databases requires separate backup strategies. Most vector DBs provide snapshot or export capabilities. Here's how to back up ChromaDB and integrate it with your agent state backup:
import shutil
import chromadb
from pathlib import Path
class VectorMemoryBackup:
def __init__(self, chroma_persist_path: str, backup_dir: str = "./vector_backups"):
self.client = chromadb.PersistentClient(path=chroma_persist_path)
self.backup_dir = Path(backup_dir)
self.backup_dir.mkdir(exist_ok=True, parents=True)
def backup_collection(self, collection_name: str) -> str:
"""Export a collection to a JSON backup file."""
collection = self.client.get_collection(collection_name)
# Get all documents with embeddings and metadata
result = collection.get(
include=["embeddings", "metadatas", "documents"]
)
backup_data = {
"collection_name": collection_name,
"ids": result["ids"],
"embeddings": result["embeddings"],
"metadatas": result["metadatas"],
"documents": result["documents"],
"timestamp": datetime.utcnow().isoformat()
}
backup_path = self.backup_dir / f"{collection_name}_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json"
with open(backup_path, 'w') as f:
json.dump(backup_data, f, default=str)
return str(backup_path)
def restore_collection(self, backup_path: str) -> str:
"""Restore a collection from a backup file."""
with open(backup_path, 'r') as f:
backup_data = json.load(f)
# Delete existing collection if present
try:
self.client.delete_collection(backup_data["collection_name"])
except Exception:
pass
new_collection = self.client.create_collection(
name=backup_data["collection_name"]
)
new_collection.add(
ids=backup_data["ids"],
embeddings=backup_data["embeddings"],
metadatas=backup_data["metadatas"],
documents=backup_data["documents"]
)
return backup_data["collection_name"]
# Integrated backup function
async def full_agent_backup(
state: AgentState,
checkpoint_manager: CheckpointManager,
vector_backup: VectorMemoryBackup,
collections_to_backup: List[str]
):
# Save agent state
checkpoint_path = checkpoint_manager.save(state)
# Save vector memory
vector_backup_paths = []
for collection in collections_to_backup:
path = vector_backup.backup_collection(collection)
vector_backup_paths.append(path)
return {
"checkpoint": checkpoint_path,
"vector_backups": vector_backup_paths,
"timestamp": datetime.utcnow().isoformat()
}Step 3: Automated Recovery on Startup
When your agent starts, it should automatically detect and restore the most recent checkpoint. This transforms a crash from a catastrophic event into a minor hiccup:
class ResilientAgent:
def __init__(
self,
session_id: str,
checkpoint_manager: CheckpointManager,
vector_backup: VectorMemoryBackup
):
self.session_id = session_id
self.checkpoint_manager = checkpoint_manager
self.vector_backup = vector_backup
self.state = self._recover_or_initialize()
def _recover_or_initialize(self) -> AgentState:
"""Attempt recovery; fall back to fresh state if nothing found."""
latest_state = self.checkpoint_manager.load_latest(self.session_id)
if latest_state is not None:
print(f"✅ Recovered session {self.session_id} from checkpoint {latest_state.checkpoint_number}")
print(f" Conversation turns restored: {len(latest_state.conversation_history)}")
print(f" Pending tool calls: {len(latest_state.pending_tool_calls)}")
# Restore vector memory references
if latest_state.long_term_memory_keys:
print(f" Reconnecting vector memory keys: {latest_state.long_term_memory_keys}")
# Logic to re-establish vector DB connections goes here
return latest_state
print(f"🆕 No checkpoint found for session {self.session_id}. Starting fresh.")
return AgentState(session_id=self.session_id)
async def run(self):
"""Main agent loop with recovery-aware startup."""
state = self.state
# The agent continues from wherever it left off
await agent_loop(state, self.process_turn, self.checkpoint_manager)
Step 4: Cloud-Native Backup with S3 / Object Storage
For production deployments, local disk checkpoints won't survive pod restarts. You need durable, off-instance storage. Here's an S3-backed checkpoint manager:
import boto3
import pickle
from io import BytesIO
class S3CheckpointManager:
def __init__(self, bucket_name: str, prefix: str = "agent_checkpoints/"):
self.s3 = boto3.client('s3')
self.bucket = bucket_name
self.prefix = prefix
def save(self, state: AgentState) -> str:
state.checkpoint_number += 1
state.last_updated = datetime.utcnow()
key = f"{self.prefix}{state.session_id}/checkpoint_{state.checkpoint_number}.json"
body = state.model_dump_json(indent=2)
self.s3.put_object(
Bucket=self.bucket,
Key=key,
Body=body,
ContentType='application/json'
)
return f"s3://{self.bucket}/{key}"
def load_latest(self, session_id: str) -> Optional[AgentState]:
prefix = f"{self.prefix}{session_id}/"
response = self.s3.list_objects_v2(
Bucket=self.bucket,
Prefix=prefix
)
if 'Contents' not in response:
return None
# Sort by last modified, get newest
objects = sorted(
response['Contents'],
key=lambda x: x['LastModified'],
reverse=True
)
latest_key = objects[0]['Key']
obj = self.s3.get_object(Bucket=self.bucket, Key=latest_key)
data = json.loads(obj['Body'].read().decode('utf-8'))
return AgentState.model_validate(data)
def list_checkpoints(self, session_id: str) -> List[Dict[str, Any]]:
"""List all checkpoints with timestamps for a session."""
prefix = f"{self.prefix}{session_id}/"
response = self.s3.list_objects_v2(
Bucket=self.bucket,
Prefix=prefix
)
if 'Contents' not in response:
return []
return [
{
"key": obj['Key'],
"last_modified": obj['LastModified'].isoformat(),
"size_bytes": obj['Size']
}
for obj in response['Contents']
]Incremental Checkpointing for Efficiency
Full state snapshots every few turns can become expensive with large conversation histories. Incremental checkpointing stores only what changed since the last snapshot:
class IncrementalCheckpointManager:
def __init__(self, base_path: str = "./incremental_checkpoints"):
self.base_path = Path(base_path)
self.base_path.mkdir(exist_ok=True, parents=True)
def save_incremental(self, state: AgentState, previous_checkpoint_number: int) -> str:
"""Save only the delta since the last full checkpoint."""
# Load the previous full checkpoint as baseline
previous = self._load_checkpoint(state.session_id, previous_checkpoint_number)
delta = {
"session_id": state.session_id,
"base_checkpoint": previous_checkpoint_number,
"new_checkpoint_number": previous_checkpoint_number + 1,
"conversation_additions": state.conversation_history[len(previous.conversation_history):],
"tool_call_changes": [
tc for tc in state.pending_tool_calls
if tc.status != 'completed'
],
"scratchpad": state.scratchpad,
"timestamp": datetime.utcnow().isoformat()
}
delta_path = self.base_path / f"{state.session_id}_delta_{previous_checkpoint_number + 1}.json"
with open(delta_path, 'w') as f:
json.dump(delta, f, default=str, indent=2)
return str(delta_path)
def reconstruct_state(self, session_id: str, target_checkpoint: int) -> AgentState:
"""Rebuild state from base checkpoint plus incremental deltas."""
# Find the most recent full checkpoint before target
full_checkpoints = sorted([
int(f.stem.split('_checkpoint_')[1])
for f in self.base_path.glob(f"{session_id}_checkpoint_*.json")
])
base_num = max([c for c in full_checkpoints if c <= target_checkpoint])
state = self._load_checkpoint(session_id, base_num)
# Apply deltas from base_num+1 to target_checkpoint
for delta_num in range(base_num + 1, target_checkpoint + 1):
delta_path = self.base_path / f"{session_id}_delta_{delta_num}.json"
if delta_path.exists():
with open(delta_path, 'r') as f:
delta = json.load(f)
state.conversation_history.extend(delta["conversation_additions"])
for tc_change in delta["tool_call_changes"]:
# Merge tool call updates
existing = next(
(tc for tc in state.pending_tool_calls if tc.call_id == tc_change["call_id"]),
None
)
if existing:
existing.status = tc_change.get("status", existing.status)
existing.result = tc_change.get("result", existing.result)
else:
state.pending_tool_calls.append(ToolCall(**tc_change))
state.scratchpad = delta.get("scratchpad", state.scratchpad)
state.checkpoint_number = target_checkpoint
return state
Best Practices for Agent Backup and Recovery
1. Design for Serializability from Day One
Retrofitting backup onto an agent that wasn't designed for it is painful. From the start, keep all mutable state in a single, serializable state object. Avoid state scattered across global variables, module-level caches, or in-memory data structures that can't be easily captured.
2. Use Structured Formats, Not Raw Text
Don't just dump conversation as a text blob. Use typed message objects with roles, timestamps, and IDs. This allows you to replay conversations exactly, filter by time range, or even partially restore. JSON with Pydantic models gives you validation on restore — corrupted checkpoints will fail loudly rather than silently injecting bad state.
3. Establish a Checkpoint Cadence Based on Risk
- Every turn — for high-stakes agents (financial transactions, medical advice)
- Every N turns (N=5-10) — for general conversational agents
- Before and after tool calls — always checkpoint before an irreversible action (sending email, placing order)
- On graceful shutdown signal — catch SIGTERM and do a final checkpoint
4. Implement Write-Ahead Logging for Critical Operations
Before the agent executes a destructive or irreversible tool call, write the intent to a log that survives crashes. On recovery, the agent can check whether the action actually completed:
class WriteAheadLogger:
def __init__(self, log_path: str = "./walogs"):
self.log_path = Path(log_path)
self.log_path.mkdir(exist_ok=True)
def log_intent(self, session_id: str, tool_call: ToolCall):
"""Log an intended action before execution."""
intent_record = {
"session_id": session_id,
"call_id": tool_call.call_id,
"tool_name": tool_call.tool_name,
"arguments": tool_call.arguments,
"status": "intended",
"timestamp": datetime.utcnow().isoformat()
}
log_file = self.log_path / f"{session_id}_walog.jsonl"
with open(log_file, 'a') as f:
f.write(json.dumps(intent_record) + '\n')
def mark_completed(self, session_id: str, call_id: str, result: Any):
"""Mark an intended action as completed."""
completion_record = {
"session_id": session_id,
"call_id": call_id,
"status": "completed",
"result": result,
"timestamp": datetime.utcnow().isoformat()
}
log_file = self.log_path / f"{session_id}_walog.jsonl"
with open(log_file, 'a') as f:
f.write(json.dumps(completion_record) + '\n')
def get_in_flight_intents(self, session_id: str) -> List[Dict]:
"""On recovery, find intentions that never completed."""
log_file = self.log_path / f"{session_id}_walog.jsonl"
if not log_file.exists():
return []
intents = {}
with open(log_file, 'r') as f:
for line in f:
record = json.loads(line)
call_id = record["call_id"]
if record["status"] == "intended":
intents[call_id] = record
elif record["status"] == "completed":
intents.pop(call_id, None)
return list(intents.values())
5. Test Recovery, Not Just Backup
A backup that can't be restored is worse than no backup — it gives false confidence. Build automated integration tests that:
- Run the agent through a complex multi-turn scenario
- Simulate a crash mid-execution
- Restore from the latest checkpoint
- Verify that the agent continues correctly without repeating or skipping actions
- Assert that pending tool calls are properly resolved
async def test_recovery_after_tool_call_crash():
"""Integration test: agent recovers correctly after crash mid-tool-execution."""
checkpoint_mgr = CheckpointManager("./test_checkpoints")
agent = ResilientAgent("test_session_123", checkpoint_mgr, None)
# Run through initial turns
state = agent.state
state.conversation_history.append(Message(role="user", content="Book a flight to Paris"))
# Simulate the agent starting a tool call
flight_tool = ToolCall(
tool_name="search_flights",
arguments={"destination": "Paris", "date": "2025-06-15"},
status="running",
call_id="call_flight_001"
)
state.pending_tool_calls.append(flight_tool)
# Checkpoint
checkpoint_mgr.save(state)
# --- Simulate crash ---
# Re-initialize agent (this triggers recovery)
recovered_agent = ResilientAgent("test_session_123", checkpoint_mgr, None)
recovered_state = recovered_agent.state
# Assertions
assert len(recovered_state.conversation_history) == 1
assert recovered_state.pending_tool_calls[0].tool_name == "search_flights"
assert recovered_state.pending_tool_calls[0].status == "running"
assert recovered_state.pending_tool_calls[0].call_id == "call_flight_001"
print("✅ Recovery test passed: agent resumed mid-flight booking correctly")
6. Implement Retention Policies
Checkpoints accumulate. Without cleanup, you'll pay for storage of agent sessions from months ago. Implement automatic expiration:
class CheckpointRetentionPolicy:
def __init__(self, checkpoint_manager, max_age_days: int = 30, max_checkpoints_per_session: int = 50):
self.manager = checkpoint_manager
self.max_age_days = max_age_days
self.max_checkpoints = max_checkpoints
def enforce(self):
"""Remove expired and excess checkpoints."""
cutoff = datetime.utcnow() - timedelta(days=self.max_age_days)
for checkpoint_file in Path(self.manager.base_path).glob("*_checkpoint_*.json"):
if checkpoint_file.stat().st_mtime < cutoff.timestamp():
checkpoint_file.unlink()
print(f"🧹 Removed expired: {checkpoint_file.name}")
# Enforce per-session max
sessions = set()
for f in Path(self.manager.base_path).glob("*_checkpoint_*.json"):
session_id = f.stem.split('_checkpoint_')[0]
sessions.add(session_id)
for session_id in sessions:
checkpoints = sorted(
Path(self.manager.base_path).glob(f"{session_id}_checkpoint_*.json"),
key=lambda x: x.stat().st_mtime,
reverse=True
)
for old_checkpoint in checkpoints[self.max_checkpoints:]:
old_checkpoint.unlink()
print(f"🧹 Removed excess: {old_checkpoint.name}")
7. Encrypt Sensitive State
Agent checkpoints often contain user data, PII, or proprietary reasoning. Encrypt at rest:
from cryptography.fernet import Fernet
class EncryptedCheckpointManager:
def __init__(self, base_path: str, encryption_key: bytes):
self.base_path = Path(base_path)
self.base_path.mkdir(exist_ok=True)
self.cipher = Fernet(encryption_key)
def save(self, state: AgentState) -> str:
state.checkpoint_number += 1
state.last_updated = datetime.utcnow()
plaintext = state.model_dump_json(indent=2).encode('utf-8')
encrypted = self.cipher.encrypt(plaintext)
checkpoint_path = self.base_path / f"{state.session_id}_checkpoint_{state.checkpoint_number}.enc"
checkpoint_path.write_bytes(encrypted)
return str(checkpoint_path)
def load_latest(self, session_id: str) -> Optional[AgentState]:
checkpoints = sorted(
self.base_path.glob(f"{session_id}_checkpoint_*.enc"),
key=lambda x: int(x.stem.split('_checkpoint_')[1]),
reverse=True
)
if not checkpoints:
return None
encrypted = checkpoints[0].read_bytes()
plaintext = self.cipher.decrypt(encrypted).decode('utf-8')
return AgentState.model_validate_json(plaintext)
Handling Partial Recovery and Conflict Resolution
Sometimes recovery isn't clean. The agent may have been in the middle of a tool call that partially executed. You need a strategy for each tool's idempotency:
class IdempotentToolWrapper:
def __init__(self, tool_func: Callable, idempotency_key: str):
self.tool_func = tool_func
self.idempotency_key = idempotency_key
async def execute(self, *args, **kwargs) -> Any:
"""Execute tool with idempotency check."""
# Check if this exact call already completed
if self._has_completed(self.idempotency_key):
return self._get_cached_result(self.idempotency_key)
result = await self.tool_func(*args, **kwargs)
self._cache_result(self.idempotency_key, result)
return result
def _has_completed(self, key: str) -> bool:
# Check persistent store for completion marker
completion_path = Path("./idempotency_log") / f"{key}.done"
return completion_path.exists()
def _cache_result(self, key: str, result: Any):
result_path = Path("./idempotency_log") / f"{key}.result.json"
result_path.parent.mkdir(exist_ok=True)
result_path.write_text(json.dumps(result))
# Mark as done
(Path("./idempotency_log") / f"{key}.done").touch()
def _get_cached_result(self, key: str) -> Any:
result_path = Path("./idempotency_log") / f"{key}.result.json"
return json.loads(result_path.read_text())
On recovery, the agent replays pending tool calls through idempotent wrappers. If a call already completed before the crash, the wrapper returns the cached result and the agent moves on without double-executing.
Monitoring and Alerting for Backup Health
Your backup system itself needs monitoring. Silent backup failures are dangerous:
class BackupHealthMonitor:
def __init__(self, checkpoint_manager):
self.checkpoint_manager = checkpoint_manager
self.last_successful_backup: Optional[datetime] = None
self.consecutive_failures = 0
self.alert_threshold = 3
def record_success(self, session_id: str, checkpoint_path: str):
self.last_successful_backup = datetime.utcnow()
self.consecutive_failures = 0
print(f"✅ Backup healthy: {checkpoint_path}")
def record_failure(self, session_id: str, error: Exception):
self.consecutive_failures += 1
print(f"❌ Backup failure #{self.consecutive_failures}: {error}")
if self.consecutive_failures >= self.alert_threshold:
self._trigger_alert(
f"Agent backup system has failed {self.consecutive_failures} times consecutively. "
f"Last success: {self.last_successful_backup or 'never'}"
)
def _trigger_alert(self, message: str):
# Send to PagerDuty, Slack, email, etc.
print(f"🚨 ALERT: {message}")
# In production: requests.post(webhook_url, json={"text": message})
def check_backup_freshness(self, max_minutes_since_backup: int = 15):
"""Alert if no backup has succeeded recently."""
if self.last_successful_backup is None:
self._trigger_alert("No successful backup has ever occurred")
return
age_minutes = (datetime.utcnow() - self.last_successful_backup).total_seconds() / 60
if age_minutes > max_minutes_since_backup:
self._trigger_alert(
f"Last successful backup was {age_minutes:.1f} minutes ago, "
f"exceeding threshold of {max_minutes_since_backup} minutes"
)
Putting It All Together: A Production-Ready Agent with Backup
Here's a complete agent class that integrates checkpointing, vector backup, write-ahead logging, encryption, and health monitoring into a single cohesive system:
class ProductionAgent:
def __init__(
self,
session_id: str,
s3_bucket: str,
encryption_key: bytes,
chroma_path: str = "./chroma_data",
max_checkpoint_age_days: int = 30
):
self.session_id = session_id
# Storage layers
self.checkpoint_manager = EncryptedCheckpointManager(
base_path=f"./checkpoints/{session_id}",
encryption_key=encryption_key
)
self.s3_backup = S3CheckpointManager(
bucket_name=s3_bucket,
prefix=f"agents/{session_id}/"
)
self.vector_backup = VectorMemoryBackup(chroma_persist_path=chroma_path)
self.wal_logger = WriteAheadLogger(f"./walogs/{session_id}")
# Health and retention
self.health_monitor = BackupHealthMonitor(self.checkpoint_manager)
self.retention = CheckpointRetentionPolicy(
self.checkpoint_manager,
max_age_days=max_checkpoint_age_days
)
# Recover or initialize
self.state = self._recover_or_initialize()
def _recover_or_initialize(self) -> AgentState:
"""Multi-source recovery: try S3 first, then local, then fresh."""
# Try S3 (cross-instance durability)
try:
s3_state = self.s3_backup.load_latest(self.session_id)
if s3_state:
print("🌐 Recovered from S3")
return s3_state
except Exception as e:
print(f"S3 recovery failed (will try local): {e}")
# Try local encrypted checkpoints
local_state = self.checkpoint_manager.load_latest(self.session_id)
if local_state:
print("💾 Recovered from local encrypted checkpoint")
return local_state
print("🆕 Starting fresh session")
return AgentState(session_id=self.session_id)
async def checkpoint(self):
"""Perform a comprehensive checkpoint."""
try:
# Save locally with encryption
local_path = self.checkpoint_manager.save(self.state)
# Replicate to S3 for durability
s3_path = self.s3_backup.save(self.state)
# Backup vector collections
vector_paths = []
for collection in ["agent_knowledge", "user_context"]:
vp = self.vector_backup.backup_collection(collection)
vector_paths.append(vp)
# Enforce retention policies
self.retention.enforce()
# Record health
self.health_monitor.record_success(self.session_id, local_path)
print(f"📦 Full checkpoint complete: local={local_path}, s3={s3_path}")
return {"local": local_path, "s3": s3_path, "vectors": vector_paths}
except Exception as e:
self.health_monitor.record_failure(self.session_id, e)
raise
async def execute_with_safety(self, tool_call: ToolCall):
"""Execute a tool call with write-ahead logging and idempotency."""
# Log intent before execution
self.wal_logger.log_intent(self.session_id, tool_call)
# Wrap in idempotency
wrapper = IdempotentToolWrapper(
tool_func=lambda: self._actual_tool_execution(tool_call),
idempotency_key=tool_call.call_id
)
try:
result = await wrapper.execute()
self.wal_logger.mark_completed(
self.session_id, tool_call.call_id, result
)
tool_call.status = 'completed'
tool_call.result = result
except Exception as e:
tool_call.status = 'failed'
tool_call.result = str(e)
raise
# Checkpoint after tool execution
await self.checkpoint()
return result
def graceful_shutdown(self):
"""Handle SIGTERM gracefully."""
print("🛑 Graceful shutdown initiated — saving final checkpoint...")
try:
self.checkpoint_manager.save(self.state)
print("✅ Final checkpoint saved. Safe to terminate.")
except Exception as e:
print(f"⚠️ Could not save final checkpoint: {e}")
Conclusion
AI agent backup and recovery transforms your agent from a fragile, ephemeral process into a resilient, production-grade system. The core principles are straightforward: design for serializability, checkpoint at meaningful boundaries, use write-ahead logging for irreversible actions, store checkpoints in durable off-instance storage like S3, encrypt sensitive state, and monitor backup health continuously. The investment