AI Agent Disaster Recovery: Backups, Failover & RTO/RPO

import boto3
import requests
from datetime import datetime, timezone

QDRANT_URL = "http://qdrant:6333"
S3_BUCKET = "agent-backups"
COLLECTION = "agent_memory"

def snapshot_vector_store():
    """Create a Qdrant snapshot and ship it to S3."""
    # 1. Trigger an on-demand snapshot
    resp = requests.post(f"{QDRANT_URL}/collections/{COLLECTION}/snapshots")
    resp.raise_for_status()
    snapshot_name = resp.json()["result"]["name"]

    # 2. Download the snapshot
    snap = requests.get(
        f"{QDRANT_URL}/collections/{COLLECTION}/snapshots/{snapshot_name}",
        stream=True,
    )

    # 3. Ship to S3 with a timestamped key
    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    key = f"vector/{COLLECTION}/{ts}_{snapshot_name}"

    s3 = boto3.client("s3")
    s3.upload_fileobj(snap.raw, S3_BUCKET, key, ExtraArgs={
        "StorageClass": "STANDARD_IA",
        "Metadata": {"collection": COLLECTION, "snapshot": snapshot_name},
    })

    # 4. Delete server-side snapshot after successful upload
    requests.delete(f"{QDRANT_URL}/collections/{COLLECTION}/snapshots/{snapshot_name}")

    return {"s3_key": key, "collection": COLLECTION, "timestamp": ts}

# Run every 5 minutes via cron or a sidecar container

import json
import uuid
import psycopg2
from contextlib import contextmanager

DB = psycopg2.connect("postgres://agent:***@db/agent_state")

def start_task(payload: dict) -> str:
    task_id = str(uuid.uuid4())
    with DB, DB.cursor() as cur:
        cur.execute(
            """INSERT INTO tasks (id, status, payload, created_at)
               VALUES (%s, 'pending', %s, now())""",
            (task_id, json.dumps(payload)),
        )
    return task_id

@contextmanager
def step(task_id: str, step_name: str, idempotency_key: str):
    """Durable step that survives crashes. Skips if already committed."""
    with DB, DB.cursor() as cur:
        cur.execute(
            "SELECT status FROM task_steps WHERE task_id=%s AND name=%s",
            (task_id, step_name),
        )
        row = cur.fetchone()

        if row and row[0] == "completed":
            # Already done before the crash — skip
            yield None
            return

        cur.execute(
            """INSERT INTO task_steps (task_id, name, idempotency_key, status)
               VALUES (%s, %s, %s, 'running')
               ON CONFLICT (task_id, name) DO UPDATE SET status='running'""",
            (task_id, step_name, idempotency_key),
        )

    try:
        result = {}
        yield result
        with DB, DB.cursor() as cur:
            cur.execute(
                """UPDATE task_steps SET status='completed', result=%s, finished_at=now()
                   WHERE task_id=%s AND name=%s""",
                (json.dumps(result), task_id, step_name),
            )
    except Exception as e:
        with DB, DB.cursor() as cur:
            cur.execute(
                """UPDATE task_steps SET status='failed', error=%s, finished_at=now()
                   WHERE task_id=%s AND name=%s""",
                (str(e), task_id, step_name),
            )
        raise

# Usage
def run_research_task(task_id: str, query: str):
    with step(task_id, "search_web", idempotency_key=f"search:{query}") as r:
        if r is not None:
            r["results"] = web_search(query)

    with step(task_id, "summarize", idempotency_key=f"sum:{task_id}") as r:
        if r is not None:
            r["summary"] = llm_summarize(query)

    with step(task_id, "email_report", idempotency_key=f"email:{task_id}") as r:
        if r is not None:
            # External side effect — idempotency key prevents double-send
            send_email(subject=query, body=r.get("summary", ""))

import asyncio
import json
import nats
from nats.js.api import StreamConfig, RetentionPolicy, StorageType

STREAM = "AGENT_TASKS"
SUBJECT = "agent.tasks.*"

async def setup_stream():
    nc = await nats.connect("nats://nats:4222")
    js = nc.jetstream()

    # File-backed, replicated 3x, retain until ACK
    await js.add_stream(StreamConfig(
        name=STREAM,
        subjects=[SUBJECT],
        storage=StorageType.FILE,
        num_replicas=3,
        retention=RetentionPolicy.WORK_QUEUE,
        max_age=86400 * 7,  # 7-day safety net
    ))
    return js

async def enqueue(js, task_id: str, payload: dict):
    """Publish task with msg-id for exactly-once semantics."""
    await js.publish(
        f"agent.tasks.{task_id}",
        json.dumps(payload).encode(),
        headers={"Nats-Msg-Id": task_id},  # dedupes within window
    )

async def worker(js, handler):
    sub = await js.pull_subscribe(SUBJECT, durable="agent_worker")
    while True:
        msgs = await sub.fetch(batch=1, timeout=30)
        for msg in msgs:
            task = json.loads(msg.data)
            try:
                await handler(task)
                await msg.ack()
            except Exception:
                await msg.nak(delay=5)  # retry with backoff

# Primary region: full deployment
module "agent_primary" {
  source    = "./modules/agent-stack"
  region    = "us-east-1"
  replicas  = 3
  role      = "primary"
}

# DR region: pilot light — 1 replica, 0 traffic
module "agent_dr" {
  source    = "./modules/agent-stack"
  region    = "us-west-2"
  replicas  = 1            # scales to 3 on failover
  role      = "standby"
}

# Cross-region S3 replication for vector backups
resource "aws_s3_bucket_replication_configuration" "vector_backups" {
  bucket = aws_s3_bucket.primary_vector_backups.id
  role   = aws_iam_role.replication.arn

  rule {
    id       = "cross-region-vector-backups"
    status   = "Enabled"
    priority = 1
    destination {
      bucket        = aws_s3_bucket.dr_vector_backups.arn
      storage_class = "STANDARD_IA"
      replication_time {
        status = "Enabled"
        time { minutes = 15 }
      }
    }
  }
}

# Route53 health check + failover record
resource "aws_route53_health_check" "primary" {
  fqdn              = module.agent_primary.health_endpoint
  type              = "HTTPS"
  failure_threshold = 3
  request_interval  = 30
}

resource "aws_route53_record" "agent_failover_primary" {
  zone_id = var.zone_id
  name    = "agents.rapidclaw.dev"
  type    = "A"
  set_identifier = "primary"
  failover_routing_policy { type = "PRIMARY" }
  health_check_id = aws_route53_health_check.primary.id
  alias {
    name                   = module.agent_primary.lb_dns
    zone_id                = module.agent_primary.lb_zone_id
    evaluate_target_health = true
  }
}

import time
from dataclasses import dataclass, field

@dataclass
class Provider:
    name: str
    call: callable
    failures: list[float] = field(default_factory=list)
    opened_at: float | None = None

    def is_open(self, now: float) -> bool:
        if self.opened_at is None:
            return False
        # Half-open after 60 seconds
        if now - self.opened_at > 60:
            self.opened_at = None
            self.failures = []
            return False
        return True

    def record_failure(self, now: float):
        # Keep last 60 seconds of failures
        self.failures = [t for t in self.failures if t > now - 60] + [now]
        if len(self.failures) >= 3:
            self.opened_at = now

class ProviderRouter:
    def __init__(self, providers: list[Provider]):
        self.providers = providers

    def call(self, prompt: str, **kwargs):
        now = time.time()
        last_error = None
        for p in self.providers:
            if p.is_open(now):
                continue
            try:
                return p.call(prompt, **kwargs)
            except Exception as e:
                p.record_failure(now)
                last_error = e
                continue
        raise RuntimeError(f"All providers unavailable: {last_error}")

# Configure priority order: primary, secondary, self-hosted fallback
router = ProviderRouter([
    Provider("anthropic", call=anthropic_call),
    Provider("openai",    call=openai_call),
    Provider("local_llama", call=local_llama_call),  # degraded quality, always up
])

response = router.call("Summarize this document: ...")

# Agent DR Runbook

## Scenario 1 — Primary region unhealthy (Route53 health check failing)
1. Confirm the outage: check /health on each primary pod + CloudWatch.
2. If > 3 min of failed health checks, Route53 will fail over automatically.
3. Scale the DR region from 1 -> 3 replicas:
   $ kubectl scale -n agents deploy/agent --replicas=3 --context=dr-cluster
4. Verify vector store is hydrated:
   $ curl https://qdrant-dr/collections/agent_memory | jq .result.points_count
5. Smoke test: $ ./scripts/dr-smoke-test.sh
6. Post to #incidents with timeline.

## Scenario 2 — Vector memory corruption detected
1. Put agent in read-only mode:
   $ kubectl set env deploy/agent READ_ONLY=true
2. Restore latest healthy snapshot from S3:
   $ ./scripts/restore-vector-snapshot.sh --snapshot=20260419T020000Z
3. Verify restored collection count matches expected baseline.
4. Remove read-only flag, monitor error rate for 30 min.

## Scenario 3 — Primary LLM provider down
1. Router handles this automatically (see provider_router.py).
2. If persistent (> 30 min), page on-call to review quality metrics.
3. Consider pausing non-critical agent workloads until primary recovers.

## Scenario 4 — In-flight task queue lost
1. NATS JetStream is 3x replicated — this should not happen.
2. If it does: tasks are replayable from agent_tasks table.
3. $ ./scripts/replay-tasks.sh --since=<last-known-good-timestamp>