What is AI Agent Backup and Recovery?
AI Agent Backup and Recovery is a systematic approach to preserving and restoring an autonomous agent's complete operational state — its memory, context, learned patterns, and ongoing workflows. Just as database administrators maintain disaster recovery plans for transactional data, AI agent developers must implement robust backup strategies to safeguard the agent's accumulated intelligence. When an agent crashes, gets redeployed, or encounters a catastrophic failure, a properly designed backup and recovery system ensures the agent can resume exactly where it left off, without losing weeks of learning, conversation context, or critical decision-making history.
Agent memory encompasses far more than simple chat logs. It includes vector embeddings stored in retrieval systems, fine-tuned model weights, active task queues, persistent state machines, and the intricate web of relationships the agent has built between concepts, users, and actions. Losing this memory means forcing the agent to start from zero — a costly regression that destroys user trust and wastes computational resources invested in the agent's learning trajectory.
Why Agent Memory Backup Matters
The stakes for AI agent memory integrity grow proportionally with the agent's autonomy and deployment longevity. Consider these critical scenarios:
- Production Failures: A memory corruption bug in a customer-facing agent erases six months of personalized interaction history, instantly degrading the experience from tailored conversations to generic, impersonal responses.
- Infrastructure Migration: Moving an agent from an on-premise cluster to cloud infrastructure without proper state transfer causes complete loss of ongoing multi-step workflows — abandoned orders, half-completed analyses, and dropped conversations.
- Model Evolution: When upgrading the underlying LLM or embedding model, the agent's vector store requires careful migration. Without backup, a failed migration could corrupt the entire knowledge base irreversibly.
- Compliance and Audit: Regulated industries require immutable records of agent decisions. Backup systems serve as the audit trail for why an agent made specific choices, protecting organizations during compliance reviews.
Beyond disaster prevention, robust backup enables time-travel debugging — the ability to restore the agent's exact state at any historical point to understand emergent behaviors, debug subtle bugs, or reproduce user-reported issues with perfect fidelity.
Anatomy of Agent Memory: What to Back Up
A production AI agent typically maintains several distinct memory layers. Each requires a tailored backup strategy:
1. Conversation and Interaction History
The chronological record of exchanges between the agent and users, including metadata like timestamps, user IDs, session contexts, and message-level annotations. This is typically stored in a relational database or time-series storage and forms the foundation for personalization and context-aware responses.
2. Vector Stores and Semantic Memory
Embedding-based retrieval systems (Pinecone, Weaviate, pgvector, ChromaDB) hold the agent's long-term knowledge. These contain dense vector representations of documents, memories, and facts — along with their metadata payloads. Vector stores are often the largest component and the most challenging to back up efficiently due to their size and indexing structures.
3. State Machine and Active Workflows
Multi-step operations in progress — like a travel booking agent mid-reservation or a coding agent in the middle of a complex refactor. The state includes the current step, accumulated partial results, pending confirmations, and timeout windows. This state is often ephemeral but critical to preserve for seamless recovery.
4. Agent Configuration and Policy Memory
The rules, constraints, and behavioral guidelines that define the agent's operational boundaries. This includes system prompts, safety filters, tool access permissions, rate-limiting configurations, and user-specific preferences. Configuration drift is a common failure mode — backing up configuration snapshots prevents silent behavioral changes.
5. Learned Patterns and Fine-tuned Artifacts
If the agent employs reinforcement learning from feedback, stores user preference models, or maintains locally fine-tuned adapter weights, these artifacts represent expensive learned intelligence that must be preserved across deployments.
Implementing Backup and Recovery: A Complete Blueprint
Let's build a production-grade backup system for an AI agent with conversation memory, a vector store, and active workflow state. We'll implement snapshot-based backups with point-in-time recovery capabilities.
Step 1: Define the Backup Manifest Schema
First, create a unified manifest that catalogs all backup components and their versions. This serves as the recovery roadmap.
# backup_manifest.py
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional, Dict
import json
import hashlib
@dataclass
class BackupComponent:
component_type: str # 'conversation', 'vector_store', 'workflow_state', 'config'
storage_backend: str # 'postgres', 'pinecone', 'redis', 's3'
snapshot_id: str # unique identifier for this snapshot
checksum: str # SHA-256 of serialized data
timestamp: datetime
record_count: Optional[int] = None
size_bytes: Optional[int] = None
metadata: Dict = field(default_factory=dict)
@dataclass
class BackupManifest:
agent_id: str
backup_id: str # UUID-based unique backup identifier
created_at: datetime
components: List[BackupComponent]
agent_version: str
model_versions: Dict[str, str] # e.g., {'llm': 'gpt-4-turbo-20240410', 'embedding': 'text-embedding-3-small'}
total_size_bytes: int
def to_json(self) -> str:
return json.dumps(self.__dict__, default=str, indent=2)
@classmethod
def from_json(cls, json_str: str) -> 'BackupManifest':
data = json.loads(json_str)
data['created_at'] = datetime.fromisoformat(data['created_at'])
data['components'] = [BackupComponent(**c) for c in data['components']]
return cls(**data)Step 2: Implement the Conversation History Backup
For conversation history stored in PostgreSQL, we'll use transaction-consistent snapshots with incremental change tracking. This allows both full backups and lightweight delta backups for efficiency.
# conversation_backup.py
import psycopg2
import hashlib
import json
from datetime import datetime, timedelta
from typing import Optional, List, Dict
class ConversationBackup:
def __init__(self, db_connection_string: str):
self.conn_string = db_connection_string
self.last_backup_pointer = None # tracks last backed up message ID
def create_full_snapshot(self, agent_id: str, before_timestamp: datetime) -> Dict:
"""Create a complete snapshot of all conversations up to a point in time."""
conn = psycopg2.connect(self.conn_string)
try:
with conn.cursor() as cursor:
# Use a transaction with REPEATABLE READ for consistency
cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
cursor.execute("""
SELECT conversation_id, user_id, messages, metadata, created_at, updated_at
FROM agent_conversations
WHERE agent_id = %s AND updated_at <= %s
ORDER BY conversation_id, created_at
""", (agent_id, before_timestamp))
rows = cursor.fetchall()
snapshot_data = {
'agent_id': agent_id,
'snapshot_type': 'full',
'snapshot_timestamp': before_timestamp.isoformat(),
'conversations': []
}
total_bytes = 0
for row in rows:
conv = {
'conversation_id': row[0],
'user_id': row[1],
'messages': json.loads(row[2]) if isinstance(row[2], str) else row[2],
'metadata': json.loads(row[3]) if isinstance(row[3], str) else row[3],
'created_at': row[4].isoformat(),
'updated_at': row[5].isoformat()
}
snapshot_data['conversations'].append(conv)
serialized = json.dumps(snapshot_data, default=str).encode('utf-8')
checksum = hashlib.sha256(serialized).hexdigest()
# Record the latest message timestamp as pointer for future incrementals
cursor.execute("""
SELECT MAX(updated_at) FROM agent_conversations
WHERE agent_id = %s AND updated_at <= %s
""", (agent_id, before_timestamp))
max_ts = cursor.fetchone()[0]
return {
'data': snapshot_data,
'checksum': checksum,
'size_bytes': len(serialized),
'last_timestamp': max_ts.isoformat() if max_ts else None,
'record_count': len(rows)
}
finally:
conn.close()
def create_incremental_snapshot(self, agent_id: str,
from_timestamp: datetime,
to_timestamp: Optional[datetime] = None) -> Dict:
"""Capture only changes since the last backup for efficiency."""
if to_timestamp is None:
to_timestamp = datetime.utcnow()
conn = psycopg2.connect(self.conn_string)
try:
with conn.cursor() as cursor:
cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
cursor.execute("""
SELECT conversation_id, user_id, messages, metadata, created_at, updated_at
FROM agent_conversations
WHERE agent_id = %s
AND updated_at > %s
AND updated_at <= %s
ORDER BY conversation_id, updated_at
""", (agent_id, from_timestamp, to_timestamp))
rows = cursor.fetchall()
snapshot_data = {
'agent_id': agent_id,
'snapshot_type': 'incremental',
'base_timestamp': from_timestamp.isoformat(),
'snapshot_timestamp': to_timestamp.isoformat(),
'changes': []
}
for row in rows:
change = {
'conversation_id': row[0],
'user_id': row[1],
'messages': json.loads(row[2]) if isinstance(row[2], str) else row[2],
'metadata': json.loads(row[3]) if isinstance(row[3], str) else row[3],
'created_at': row[4].isoformat(),
'updated_at': row[5].isoformat()
}
snapshot_data['changes'].append(change)
serialized = json.dumps(snapshot_data, default=str).encode('utf-8')
checksum = hashlib.sha256(serialized).hexdigest()
return {
'data': snapshot_data,
'checksum': checksum,
'size_bytes': len(serialized),
'record_count': len(rows),
'from_timestamp': from_timestamp.isoformat(),
'to_timestamp': to_timestamp.isoformat()
}
finally:
conn.close()Step 3: Vector Store Backup with Efficient Serialization
Vector stores present unique challenges: they can be enormous (millions of embeddings) and tightly coupled to specific embedding model versions. We'll implement a streaming backup approach that exports vectors in batches while preserving index structure metadata.
# vector_store_backup.py
import numpy as np
import hashlib
import json
import pickle
from typing import List, Dict, Generator
import struct
class VectorStoreBackup:
def __init__(self, vector_db_client, index_name: str):
self.client = vector_db_client # Abstracted client (Pinecone, Weaviate, etc.)
self.index_name = index_name
self.batch_size = 1000
def export_vectors_streaming(self, namespace: str = "") -> Generator:
"""Stream vectors in batches to avoid memory exhaustion on large stores."""
# Fetch vector IDs in batches using cursor-based pagination
pagination_token = None
while True:
batch_result = self.client.fetch_batch(
index=self.index_name,
namespace=namespace,
limit=self.batch_size,
pagination_token=pagination_token,
include_vectors=True,
include_metadata=True
)
if not batch_result.vectors:
break
for vector_record in batch_result.vectors:
yield {
'id': vector_record.id,
'values': vector_record.values.tolist() if hasattr(vector_record.values, 'tolist') else vector_record.values,
'metadata': vector_record.metadata,
'sparse_values': vector_record.sparse_values if hasattr(vector_record, 'sparse_values') else None
}
pagination_token = batch_result.pagination_token
if not pagination_token:
break
def create_backup_file(self, namespace: str = "", output_path: str = "vector_backup.jsonl"):
"""Write vectors to a JSONL file for portable backup."""
import os
total_vectors = 0
file_checksums = []
with open(output_path, 'w') as f:
for vector_batch in self._batch_generator(self.export_vectors_streaming(namespace)):
for vector in vector_batch:
serialized = json.dumps(vector, default=str)
f.write(serialized + '\n')
total_vectors += 1
# Compute rolling checksum per 1000 vectors
if total_vectors % 1000 == 0:
current_checksum = hashlib.sha256(
serialized.encode('utf-8')
).hexdigest()
file_checksums.append(current_checksum)
# Final checksum of the entire file
with open(output_path, 'rb') as f:
full_checksum = hashlib.sha256(f.read()).hexdigest()
return {
'file_path': output_path,
'total_vectors': total_vectors,
'checksum': full_checksum,
'batch_checksums': file_checksums,
'index_name': self.index_name,
'namespace': namespace,
'size_bytes': os.path.getsize(output_path)
}
def _batch_generator(self, stream: Generator, batch_size: int = 100):
"""Group streamed items into batches for efficient I/O."""
batch = []
for item in stream:
batch.append(item)
if len(batch) >= batch_size:
yield batch
batch = []
if batch:
yield batch
def restore_vectors(self, backup_file_path: str, namespace: str = "",
batch_size: int = 100, verify_checksum: bool = True):
"""Restore vectors from backup file with integrity verification."""
if verify_checksum:
with open(backup_file_path, 'rb') as f:
actual_checksum = hashlib.sha256(f.read()).hexdigest()
# Compare against stored manifest checksum
# (Implementation would fetch manifest checksum from metadata store)
vectors_restored = 0
batch = []
with open(backup_file_path, 'r') as f:
for line in f:
vector_data = json.loads(line.strip())
# Reconstruct vector with proper types
vector_id = vector_data['id']
values = vector_data['values']
metadata = vector_data.get('metadata', {})
batch.append({
'id': vector_id,
'values': values,
'metadata': metadata
})
if len(batch) >= batch_size:
self.client.upsert_batch(
index=self.index_name,
namespace=namespace,
vectors=batch
)
vectors_restored += len(batch)
batch = []
# Upsert remaining batch
if batch:
self.client.upsert_batch(
index=self.index_name,
namespace=namespace,
vectors=batch
)
vectors_restored += len(batch)
return {'vectors_restored': vectors_restored, 'status': 'completed'}Step 4: Workflow State and Active Task Backup
Active workflows represent in-progress operations that span multiple steps. Backing these up requires capturing the exact state machine position, accumulated partial results, and expiration timers.
# workflow_state_backup.py
import redis
import json
import hashlib
from datetime import datetime, timedelta
from typing import Dict, Optional, List
class WorkflowStateBackup:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
self.workflow_prefix = "agent:workflow:"
def snapshot_active_workflows(self, agent_id: str) -> Dict:
"""Capture all active workflow states with TTL preservation."""
pattern = f"{self.workflow_prefix}{agent_id}:*"
workflow_keys = self.redis.keys(pattern)
workflows = {}
total_count = 0
for key in workflow_keys:
decoded_key = key.decode('utf-8') if isinstance(key, bytes) else key
# Extract workflow state hash
workflow_data = self.redis.hgetall(decoded_key)
if workflow_data:
decoded_data = {}
for field, value in workflow_data.items():
f = field.decode('utf-8') if isinstance(field, bytes) else field
v = value.decode('utf-8') if isinstance(value, bytes) else value
decoded_data[f] = v
# Capture TTL for timeout-aware recovery
ttl_seconds = self.redis.ttl(decoded_key)
workflows[decoded_key] = {
'state': decoded_data,
'ttl_remaining_seconds': ttl_seconds if ttl_seconds > 0 else None,
'snapshot_time': datetime.utcnow().isoformat()
}
total_count += 1
serialized = json.dumps(workflows, default=str).encode('utf-8')
return {
'agent_id': agent_id,
'workflow_count': total_count,
'workflows': workflows,
'checksum': hashlib.sha256(serialized).hexdigest(),
'size_bytes': len(serialized),
'snapshot_timestamp': datetime.utcnow().isoformat()
}
def restore_workflows(self, backup_data: Dict, adjust_timers: bool = True):
"""Restore workflow states with optional TTL adjustment."""
restored_count = 0
now = datetime.utcnow()
for workflow_key, workflow_info in backup_data['workflows'].items():
state_dict = workflow_info['state']
ttl_remaining = workflow_info.get('ttl_remaining_seconds')
# Restore the hash fields
self.redis.hset(workflow_key, mapping=state_dict)
# Adjust TTL: subtract elapsed time since snapshot
if adjust_timers and ttl_remaining is not None:
snapshot_time = datetime.fromisoformat(workflow_info['snapshot_time'])
elapsed = (now - snapshot_time).total_seconds()
new_ttl = max(1, int(ttl_remaining - elapsed))
self.redis.expire(workflow_key, new_ttl)
elif ttl_remaining is not None:
self.redis.expire(workflow_key, int(ttl_remaining))
restored_count += 1
return {'restored_workflows': restored_count}Step 5: Orchestrating the Complete Backup Pipeline
The orchestrator coordinates all backup components, generates the manifest, and handles atomicity — ensuring all snapshots are taken at consistent logical points.
# backup_orchestrator.py
import uuid
import json
import os
import shutil
from datetime import datetime
from typing import Dict, Optional
import boto3 # For S3/cloud storage persistence
class AgentBackupOrchestrator:
def __init__(self,
conversation_backup: 'ConversationBackup',
vector_backup: 'VectorStoreBackup',
workflow_backup: 'WorkflowStateBackup',
config_store, # Configuration management system
storage_backend=None): # S3, GCS, local filesystem
self.conversation_backup = conversation_backup
self.vector_backup = vector_backup
self.workflow_backup = workflow_backup
self.config_store = config_store
self.storage_backend = storage_backend or LocalStorageBackend("./agent_backups")
def create_full_backup(self, agent_id: str, backup_label: str = "") -> BackupManifest:
"""Execute a complete point-in-time backup of all agent memory components."""
backup_id = str(uuid.uuid4())
snapshot_time = datetime.utcnow()
components = []
total_size = 0
print(f"[Backup] Starting full backup {backup_id} for agent {agent_id}")
# 1. Backup conversation history
print("[Backup] Snapshotting conversation history...")
conv_result = self.conversation_backup.create_full_snapshot(
agent_id=agent_id,
before_timestamp=snapshot_time
)
conv_component = BackupComponent(
component_type='conversation',
storage_backend='postgres',
snapshot_id=f"conv_{backup_id}",
checksum=conv_result['checksum'],
timestamp=snapshot_time,
record_count=conv_result['record_count'],
size_bytes=conv_result['size_bytes']
)
components.append(conv_component)
total_size += conv_result['size_bytes']
# Store conversation backup data
self.storage_backend.store(
key=f"{agent_id}/{backup_id}/conversations.json",
data=json.dumps(conv_result['data'], default=str).encode('utf-8')
)
# 2. Backup vector store
print("[Backup] Exporting vector store...")
vector_backup_path = f"/tmp/vector_backup_{backup_id}.jsonl"
vector_result = self.vector_backup.create_backup_file(
namespace=agent_id,
output_path=vector_backup_path
)
# Upload vector backup to persistent storage
with open(vector_backup_path, 'rb') as f:
self.storage_backend.store(
key=f"{agent_id}/{backup_id}/vectors.jsonl",
data=f.read()
)
vector_component = BackupComponent(
component_type='vector_store',
storage_backend=self.vector_backup.index_name,
snapshot_id=f"vec_{backup_id}",
checksum=vector_result['checksum'],
timestamp=snapshot_time,
record_count=vector_result['total_vectors'],
size_bytes=vector_result['size_bytes']
)
components.append(vector_component)
total_size += vector_result['size_bytes']
# Clean up temp file
os.unlink(vector_backup_path)
# 3. Backup active workflow states
print("[Backup] Capturing workflow states...")
workflow_result = self.workflow_backup.snapshot_active_workflows(agent_id)
workflow_component = BackupComponent(
component_type='workflow_state',
storage_backend='redis',
snapshot_id=f"wf_{backup_id}",
checksum=workflow_result['checksum'],
timestamp=snapshot_time,
record_count=workflow_result['workflow_count'],
size_bytes=workflow_result['size_bytes']
)
components.append(workflow_component)
total_size += workflow_result['size_bytes']
self.storage_backend.store(
key=f"{agent_id}/{backup_id}/workflows.json",
data=json.dumps(workflow_result, default=str).encode('utf-8')
)
# 4. Backup agent configuration
print("[Backup] Capturing configuration snapshot...")
config_snapshot = self.config_store.export_configuration(agent_id)
config_json = json.dumps(config_snapshot, default=str).encode('utf-8')
config_component = BackupComponent(
component_type='config',
storage_backend='config_store',
snapshot_id=f"cfg_{backup_id}",
checksum=hashlib.sha256(config_json).hexdigest(),
timestamp=snapshot_time,
size_bytes=len(config_json)
)
components.append(config_component)
total_size += len(config_json)
self.storage_backend.store(
key=f"{agent_id}/{backup_id}/config.json",
data=config_json
)
# Build final manifest
manifest = BackupManifest(
agent_id=agent_id,
backup_id=backup_id,
created_at=snapshot_time,
components=components,
agent_version=self.config_store.get_version(agent_id),
model_versions={
'llm': self.config_store.get_model_version('llm'),
'embedding': self.config_store.get_model_version('embedding')
},
total_size_bytes=total_size
)
# Store manifest as the authoritative recovery index
manifest_json = manifest.to_json().encode('utf-8')
self.storage_backend.store(
key=f"{agent_id}/{backup_id}/manifest.json",
data=manifest_json
)
# Update latest pointer for quick recovery reference
self.storage_backend.store(
key=f"{agent_id}/latest_backup_id",
data=backup_id.encode('utf-8')
)
print(f"[Backup] Complete. Backup ID: {backup_id}, Total size: {total_size / 1024 / 1024:.2f} MB")
return manifestStep 6: Recovery — Restoring Agent Memory from Backup
Recovery reconstructs the agent's complete memory state from the backup manifest. The process respects component dependencies and validates integrity at each step.
# recovery_engine.py
import json
import hashlib
from datetime import datetime
from typing import Optional
class AgentRecoveryEngine:
def __init__(self,
conversation_backup,
vector_backup,
workflow_backup,
config_store,
storage_backend):
self.conversation_backup = conversation_backup
self.vector_backup = vector_backup
self.workflow_backup = workflow_backup
self.config_store = config_store
self.storage_backend = storage_backend
def restore_from_backup(self, agent_id: str,
backup_id: Optional[str] = None,
point_in_time: Optional[datetime] = None,
dry_run: bool = False) -> Dict:
"""
Restore agent memory from a specific backup or latest snapshot.
Args:
agent_id: The agent to restore
backup_id: Specific backup ID to restore. If None, uses latest.
point_in_time: For time-travel recovery, finds nearest backup before this time.
dry_run: Validate backup integrity without actually restoring.
Returns:
Detailed restoration report with component statuses.
"""
# Determine which backup to use
if backup_id is None:
if point_in_time is not None:
backup_id = self._find_nearest_backup_before(agent_id, point_in_time)
else:
backup_id = self.storage_backend.retrieve(
f"{agent_id}/latest_backup_id"
).decode('utf-8')
print(f"[Recovery] Restoring agent {agent_id} from backup {backup_id}")
# Fetch and parse manifest
manifest_data = self.storage_backend.retrieve(
f"{agent_id}/{backup_id}/manifest.json"
)
manifest = BackupManifest.from_json(manifest_data.decode('utf-8'))
recovery_report = {
'backup_id': backup_id,
'restore_timestamp': datetime.utcnow().isoformat(),
'dry_run': dry_run,
'components': {},
'overall_status': 'pending'
}
# Restore components in dependency order
# 1. Configuration first (other components may depend on config)
config_component = next(
c for c in manifest.components if c.component_type == 'config'
)
config_data = self.storage_backend.retrieve(
f"{agent_id}/{backup_id}/config.json"
)
# Verify checksum
config_checksum = hashlib.sha256(config_data).hexdigest()
if config_checksum != config_component.checksum:
raise IntegrityError(
f"Config checksum mismatch! Expected {config_component.checksum}, "
f"got {config_checksum}"
)
if not dry_run:
self.config_store.import_configuration(
agent_id,
json.loads(config_data.decode('utf-8'))
)
recovery_report['components']['config'] = {'status': 'restored', 'checksum_verified': True}
# 2. Restore vector store
vector_component = next(
c for c in manifest.components if c.component_type == 'vector_store'
)
vector_backup_data = self.storage_backend.retrieve(
f"{agent_id}/{backup_id}/vectors.jsonl"
)
# Verify vector backup checksum
vector_checksum = hashlib.sha256(vector_backup_data).hexdigest()
if vector_checksum != vector_component.checksum:
raise IntegrityError(
f"Vector store checksum mismatch! Expected {vector_component.checksum}, "
f"got {vector_checksum}"
)
if not dry_run:
# Write to temporary file for streaming restore
temp_path = f"/tmp/vector_restore_{backup_id}.jsonl"
with open(temp_path, 'wb') as f:
f.write(vector_backup_data)
restore_result = self.vector_backup.restore_vectors(
backup_file_path=temp_path,
namespace=agent_id,
verify_checksum=True
)
import os
os.unlink(temp_path)
recovery_report['components']['vector_store'] = {
'status': 'restored',
'vectors_restored': restore_result['vectors_restored'],
'checksum_verified': True
}
# 3. Restore conversation history
conv_component = next(
c for c in manifest.components if c.component_type == 'conversation'
)
conv_data = self.storage_backend.retrieve(
f"{agent_id}/{backup_id}/conversations.json"
)
conv_checksum = hashlib.sha256(conv_data).hexdigest()
if conv_checksum != conv_component.checksum:
raise IntegrityError(
f"Conversation checksum mismatch! Expected {conv_component.checksum}, "
f"got {conv_checksum}"
)
if not dry_run:
conversations = json.loads(conv_data.decode('utf-8'))
# Bulk insert conversations with conflict handling
self.conversation_backup.restore_conversations(
agent_id, conversations['conversations']
)
recovery_report['components']['conversation'] = {
'status': 'restored',
'record_count': conv_component.record_count,
'checksum_verified': True
}
# 4. Restore workflow states
wf_component = next(
c for c in manifest.components if c.component_type == 'workflow_state'
)
wf_data = self.storage_backend.retrieve(
f"{agent_id}/{backup_id}/workflows.json"
)
wf_checksum = hashlib.sha256(wf_data).hexdigest()
if wf_checksum != wf_component.checksum:
raise IntegrityError(
f"Workflow checksum mismatch! Expected {wf_component.checksum}, "
f"got {wf_checksum}"
)
if not dry_run:
workflow_backup_data = json.loads(wf_data.decode('utf-8'))
restore_result = self.workflow_backup.restore_workflows(
workflow_backup_data,
adjust_timers=True # Compensate for elapsed time since backup
)
recovery_report['components']['workflow_state'] = {
'status': 'restored',
'workflow_count': wf_component.record_count,
'checksum_verified': True
}
recovery_report['overall_status'] = 'success'
print(f"[Recovery] Restoration complete. Agent {agent_id} is ready.")
return recovery_report
def _find_nearest_backup_before(self, agent_id: str,
point_in_time: datetime) -> str:
"""Find the backup ID closest to (but not after) a point in time."""
# List all backups for this agent
backup_listing = self.storage_backend.list_prefix(
f"{agent_id}/"
)
# Parse manifests to find timestamps
backups_with_times = []
for key in backup_listing:
if key.endswith('/manifest.json'):
manifest_data = self.storage_backend.retrieve(key)
manifest = BackupManifest.from_json(manifest_data.decode('utf-8'))
if manifest.created_at <= point_in_time:
backups_with_times.append((manifest.created_at, manifest.backup_id))
if not backups_with_times:
raise NoSuitableBackupError(
f"No backup found before {point_in_time.isoformat()}"
)
# Return the most recent backup before the point in time
backups_with_times.sort(key=lambda x: x[0], reverse=True)
return backups_with_times[0][1]Step 7: Scheduling Automated Backups
Automated backup scheduling ensures regular snapshots without manual intervention. Use cron-style scheduling with configurable retention policies.
# backup_scheduler.py
import schedule
import time
import threading
from datetime import datetime
from typing import Callable, Dict
class BackupScheduler:
def __init__(self, orchestrator: AgentBackupOrchestrator,
retention_policy: Dict = None):
self.orchestrator = orchestrator
self.retention_policy = retention_policy or {
'hourly': 24, # Keep 24 hourly backups
'daily': 30, # Keep 30 daily backups
'weekly': 12, # Keep 12 weekly backups
'monthly': 12 # Keep 12 monthly backups
}
self.running = False
self._scheduler_thread = None
def setup_schedule(self, agent_id: str):
"""Configure backup frequency for an agent."""
# Hourly backups during business hours
schedule.every().hour.at(":00").do(
self._scheduled_backup, agent_id=agent_id, frequency='hourly'
)
# Daily backup at midnight
schedule.every().day.at("00:00").do(
self._scheduled_backup, agent_id=agent_id