Claude's Corner: Traversal, The AI That Fires Your On-Call Rotation

# Build Your Own AI SRE Agent (Traversal Clone)
## 7-Step Developer Guide

---

### Step 1: Design the Production World Model (Graph Database)

**Goal:** Build a causal graph of your entire infrastructure as a living, queryable data structure.

**DB Schema (PostgreSQL + pgvector + recursive CTEs):**

```sql
-- Core node table: every infrastructure entity
CREATE TABLE infra_nodes (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  external_id TEXT NOT NULL,          -- k8s pod name, service name, host FQDN, etc.
  node_type TEXT NOT NULL,            -- 'pod', 'service', 'host', 'database', 'queue', 'container', 'ingress', etc.
  namespace TEXT,
  cluster TEXT,
  metadata JSONB DEFAULT '{}',        -- labels, annotations, resource limits
  health_status TEXT DEFAULT 'unknown',
  last_seen_at TIMESTAMPTZ DEFAULT now(),
  created_at TIMESTAMPTZ DEFAULT now(),
  UNIQUE(external_id, node_type, cluster)
);

-- Directed causal edges between nodes
CREATE TABLE infra_edges (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  source_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
  target_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
  edge_type TEXT NOT NULL,            -- 'calls', 'depends_on', 'runs_on', 'stores_in', 'routes_to'
  weight NUMERIC DEFAULT 1.0,         -- traffic volume or dependency strength
  latency_p99_ms NUMERIC,
  error_rate NUMERIC,
  metadata JSONB DEFAULT '{}',
  updated_at TIMESTAMPTZ DEFAULT now()
);

-- Time-series metric snapshots per node
CREATE TABLE metric_snapshots (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  node_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
  ts TIMESTAMPTZ NOT NULL,
  cpu_percent NUMERIC,
  memory_percent NUMERIC,
  latency_p50_ms NUMERIC,
  latency_p99_ms NUMERIC,
  error_rate NUMERIC,
  request_rate NUMERIC,
  custom_metrics JSONB DEFAULT '{}'
);
CREATE INDEX ON metric_snapshots (node_id, ts DESC);

-- Deployments / change events
CREATE TABLE change_events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  node_id UUID REFERENCES infra_nodes(id) ON DELETE CASCADE,
  event_type TEXT NOT NULL,           -- 'deployment', 'config_change', 'scale_event', 'restart'
  description TEXT,
  diff JSONB,
  triggered_by TEXT,
  occurred_at TIMESTAMPTZ NOT NULL,
  created_at TIMESTAMPTZ DEFAULT now()
);

-- Incidents (root cause + resolution)
CREATE TABLE incidents (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  title TEXT NOT NULL,
  severity TEXT,                      -- 'critical', 'high', 'medium', 'low'
  status TEXT DEFAULT 'open',         -- 'open', 'investigating', 'resolved'
  root_cause_node_id UUID REFERENCES infra_nodes(id),
  root_cause_summary TEXT,
  rca_confidence NUMERIC,             -- 0.0 - 1.0
  remediation_applied TEXT,
  ttd_seconds INTEGER,                -- time to detect
  ttr_seconds INTEGER,                -- time to resolve
  started_at TIMESTAMPTZ NOT NULL,
  resolved_at TIMESTAMPTZ,
  created_at TIMESTAMPTZ DEFAULT now()
);

-- Knowledge bank: runbooks + historical context
CREATE TABLE knowledge_entries (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  entry_type TEXT NOT NULL,           -- 'runbook', 'past_incident', 'post_mortem', 'procedure'
  title TEXT,
  content TEXT,
  tags TEXT[],
  embedding VECTOR(1536),             -- pgvector embedding for semantic search
  metadata JSONB DEFAULT '{}',
  created_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX ON knowledge_entries USING ivfflat (embedding vector_cosine_ops);
```

**Implementation notes:**
- Use TimescaleDB or partition `metric_snapshots` by time if ingesting high-volume metrics
- Neo4j is a viable alternative for the graph layer if you prefer native graph traversal
- Keep the graph synced in real-time using Kubernetes watch APIs + WebSocket listeners

---

### Step 2: Build Agentless Data Collectors

**Goal:** Pull metrics and topology from existing infrastructure without installing agents.

**Kubernetes Collector (Python):**

```python
from kubernetes import client, watch, config
import asyncio
import psycopg2

class K8sCollector:
    def __init__(self, kubeconfig_path=None):
        if kubeconfig_path:
            config.load_kube_config(kubeconfig_path)
        else:
            config.load_incluster_config()  # in-cluster service account
        self.v1 = client.CoreV1Api()
        self.apps_v1 = client.AppsV1Api()

    async def sync_topology(self, db_conn):
        """Sync pods, services, deployments into infra_nodes + infra_edges."""
        # Collect all pods
        pods = self.v1.list_pod_for_all_namespaces()
        for pod in pods.items:
            upsert_node(db_conn, {
                'external_id': pod.metadata.name,
                'node_type': 'pod',
                'namespace': pod.metadata.namespace,
                'cluster': pod.metadata.cluster_name or 'default',
                'metadata': {
                    'labels': pod.metadata.labels,
                    'node_name': pod.spec.node_name,
                    'containers': [c.name for c in pod.spec.containers],
                    'status': pod.status.phase,
                    'restart_count': sum(
                        cs.restart_count for cs in (pod.status.container_statuses or [])
                    )
                }
            })

        # Collect services and wire edges
        services = self.v1.list_service_for_all_namespaces()
        for svc in services.items:
            svc_id = upsert_node(db_conn, {
                'external_id': svc.metadata.name,
                'node_type': 'service',
                'namespace': svc.metadata.namespace,
                'metadata': {'selector': svc.spec.selector}
            })
            # Wire service -> pods via selector matching
            if svc.spec.selector:
                for pod in pods.items:
                    if matches_selector(pod.metadata.labels, svc.spec.selector):
                        upsert_edge(db_conn, svc_id, pod_id_map[pod.metadata.name], 'routes_to')

    async def watch_events(self, db_conn):
        """Stream real-time change events."""
        w = watch.Watch()
        for event in w.stream(self.v1.list_event_for_all_namespaces):
            obj = event['object']
            if obj.reason in ['BackOff', 'Failed', 'OOMKilling', 'Unhealthy']:
                emit_change_event(db_conn, obj)
```

**Additional collectors to build:**
- **Prometheus scraper**: Pull metrics from `/metrics` endpoints every 15s
- **Cloud provider APIs**: AWS CloudWatch, GCP Monitoring, Azure Monitor
- **Database metrics**: pg_stat_activity, Redis INFO, MySQL performance_schema
- **ArgoCD webhook**: Capture deployment events in real-time

---

### Step 3: Build the Causal Inference Engine

**Goal:** Distinguish root causes from correlated symptoms using causal graph analysis.

The key insight: most monitoring tools use correlation (Pearson/Spearman). You need causation (Granger, PC algorithm, or structural causal models).

**Causal Discovery (Python, using causallearn):**

```python
import numpy as np
from causallearn.search.ConstraintBased.PC import pc
from causallearn.utils.cit import fisherz

class CausalSearchEngine:
    def __init__(self, db_conn, lookback_minutes=60):
        self.db = db_conn
        self.lookback = lookback_minutes

    def get_metric_matrix(self, incident_ts, node_ids):
        """Pull aligned time-series for all nodes in the blast radius."""
        # Returns shape: (time_steps, n_nodes)
        rows = self.db.execute("""
            SELECT node_id, ts, error_rate, latency_p99_ms, cpu_percent
            FROM metric_snapshots
            WHERE node_id = ANY(%s)
              AND ts BETWEEN %s - interval '%s minutes' AND %s
            ORDER BY ts ASC
        """, [node_ids, incident_ts, self.lookback, incident_ts])
        return pivot_to_matrix(rows)

    def find_root_cause(self, incident, blast_radius_nodes):
        """Run PC algorithm to discover causal structure, then identify root node."""
        X = self.get_metric_matrix(incident.started_at, blast_radius_nodes)
        
        # PC algorithm for causal graph discovery
        cg = pc(X, alpha=0.05, indep_test=fisherz)
        
        # Find nodes with no incoming causal edges (potential root causes)
        G = cg.G.graph
        root_candidates = [
            i for i in range(G.shape[0])
            if G[:, i].sum() == 0  # no parents
        ]
        
        # Score candidates by: (anomaly severity) × (causal distance to symptoms)
        scored = self.score_candidates(root_candidates, blast_radius_nodes, incident)
        return scored[0]  # top candidate

    def score_candidates(self, candidates, all_nodes, incident):
        """Combine anomaly score with graph centrality for final ranking."""
        scores = []
        for c in candidates:
            node = all_nodes[c]
            anomaly_score = self.compute_anomaly_score(node, incident.started_at)
            change_recency = self.recent_change_score(node, incident.started_at)
            scores.append((node, anomaly_score * 0.6 + change_recency * 0.4))
        return sorted(scores, key=lambda x: x[1], reverse=True)
```

**Blast radius discovery** (recursive CTE to find all downstream nodes):

```sql
WITH RECURSIVE blast_radius AS (
  SELECT id, 0 AS depth FROM infra_nodes WHERE id = :alert_node_id
  UNION ALL
  SELECT e.target_id, br.depth + 1
  FROM infra_edges e
  JOIN blast_radius br ON e.source_id = br.id
  WHERE br.depth < 5
)
SELECT DISTINCT n.*, br.depth
FROM blast_radius br
JOIN infra_nodes n ON n.id = br.id
ORDER BY br.depth;
```

---

### Step 4: Build the Alert Intelligence Layer

**Goal:** Reduce alert noise and prioritize by business impact, not just technical severity.

```python
from anthropic import Anthropic
import numpy as np

client = Anthropic()

class AlertIntelligence:
    def __init__(self, db_conn, knowledge_bank):
        self.db = db_conn
        self.kb = knowledge_bank

    def triage_alert(self, alert):
        """Classify alert: noise / duplicate / real incident / cascade."""
        # 1. Check if this alert is a known cascade symptom
        if self.is_downstream_symptom(alert):
            return {'action': 'suppress', 'reason': 'downstream_of_known_incident'}

        # 2. Semantic similarity to recent alerts (dedup)
        similar = self.kb.find_similar_alerts(alert['message'], top_k=5, hours=2)
        if similar and similar[0]['score'] > 0.92:
            return {'action': 'group', 'parent_incident_id': similar[0]['incident_id']}

        # 3. Score business impact
        impact = self.score_business_impact(alert)
        
        return {
            'action': 'escalate' if impact > 0.7 else 'monitor',
            'impact_score': impact,
            'suggested_priority': 'P1' if impact > 0.85 else 'P2'
        }

    def score_business_impact(self, alert):
        """Use LLM to assess business impact given service criticality context."""
        service_context = self.get_service_criticality(alert['node_id'])
        
        # Prompt caching for the system context (static across many alerts)
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=256,
            system=[{
                "type": "text",
                "text": "You are an SRE impact assessor. Score alerts 0.0-1.0 by business impact.",
                "cache_control": {"type": "ephemeral"}
            }],
            messages=[{
                "role": "user",
                "content": f"Alert: {alert['message']}\nService tier: {service_context['tier']}\nDependent services: {service_context['dependents']}\nReturn only a JSON: {{\"score\": 0.0-1.0, \"reason\": \"...\"}}"
            }]
        )
        return parse_json(response.content[0].text)['score']
```

---

### Step 5: Build the LLM Reasoning & RCA Generation Layer

**Goal:** Generate human-readable root cause summaries, remediation suggestions, and post-mortem drafts.

```python
from anthropic import Anthropic

client = Anthropic()

class RCAEngine:
    def generate_rca(self, incident, causal_chain, knowledge_context):
        """Generate a full RCA report with remediation steps."""
        
        # Build causal chain description
        chain_description = self.format_causal_chain(causal_chain)
        
        # Retrieve relevant runbooks via semantic search
        runbooks = knowledge_bank.search(
            f"{incident.title} {causal_chain.root_node.node_type}",
            top_k=3
        )
        
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system=[
                {
                    "type": "text",
                    "text": """You are an expert SRE generating incident root cause analyses.
Be precise, technical, and actionable. Reference specific services and metrics.""",
                    "cache_control": {"type": "ephemeral"}  # cache across incidents
                },
                {
                    "type": "text", 
                    "text": f"Relevant runbooks:\n{format_runbooks(runbooks)}",
                    "cache_control": {"type": "ephemeral"}  # cache per incident session
                }
            ],
            messages=[{
                "role": "user",
                "content": f"""Incident: {incident.title}
Severity: {incident.severity}
Started: {incident.started_at}

Causal chain discovered:
{chain_description}

Metrics at time of incident:
{format_metrics(incident.metrics_snapshot)}

Recent changes (30min window):
{format_changes(incident.recent_changes)}

Generate:
1. Root cause summary (2-3 sentences, specific)
2. Impact assessment 
3. Immediate remediation steps (ordered)
4. Prevention recommendations
5. Post-mortem action items"""
            }]
        )
        
        return parse_rca_response(response.content[0].text)

    def format_causal_chain(self, chain):
        """Convert graph path to readable chain: A → B → C → symptom"""
        nodes = chain.path
        return " → ".join([
            f"{n.external_id} ({n.node_type}, {n.anomaly_description})"
            for n in nodes
        ])
```

---

### Step 6: Build the Self-Healing Automation Engine

**Goal:** Execute remediation actions safely, with configurable autonomy levels and circuit breakers.

```python
import asyncio
from enum import Enum

class AutonomyLevel(Enum):
    SUGGEST = "suggest"       # AI suggests, human approves
    AUTO_LOW = "auto_low"     # Auto-execute safe actions (restarts, scaling)
    AUTO_HIGH = "auto_high"   # Auto-execute all actions including rollbacks

class SelfHealingEngine:
    def __init__(self, k8s_client, db_conn, autonomy_level=AutonomyLevel.SUGGEST):
        self.k8s = k8s_client
        self.db = db_conn
        self.autonomy = autonomy_level
        self.circuit_breaker = CircuitBreaker(max_actions_per_hour=20)

    async def remediate(self, incident, rca_result):
        """Execute or queue remediation based on autonomy level."""
        actions = self.plan_remediation(rca_result)
        
        for action in actions:
            if not self.circuit_breaker.allow(action):
                await self.escalate_to_human(incident, action, "circuit_breaker_tripped")
                continue
                
            if self.autonomy == AutonomyLevel.SUGGEST:
                await self.notify_oncall(incident, action)
            elif self.autonomy == AutonomyLevel.AUTO_LOW and action.is_safe:
                await self.execute(action, incident)
            elif self.autonomy == AutonomyLevel.AUTO_HIGH:
                await self.execute(action, incident)
            else:
                await self.notify_oncall(incident, action)

    def plan_remediation(self, rca):
        """Map root cause type to remediation actions."""
        playbook = {
            'pod_oom': [RestartPodAction, ScaleDeploymentAction],
            'connection_pool_exhausted': [RestartServiceAction, ScalePoolAction],
            'bad_deployment': [RollbackDeploymentAction],
            'database_slow': [KillLongRunningQueriesAction, ScaleReplicasAction],
            'disk_full': [CompressLogsAction, EscalateToHumanAction],
        }
        return [ActionClass(rca) for ActionClass in playbook.get(rca.root_cause_type, [EscalateToHumanAction])]

    async def execute(self, action, incident):
        """Execute a remediation action and record the result."""
        try:
            result = await action.run(self.k8s)
            self.db.execute("""
                INSERT INTO remediation_log (incident_id, action_type, result, executed_at)
                VALUES (%s, %s, %s, now())
            """, [incident.id, action.type, result])
            return result
        except Exception as e:
            await self.escalate_to_human(incident, action, str(e))
            raise

class RestartPodAction:
    is_safe = True
    type = 'restart_pod'
    
    async def run(self, k8s):
        k8s.v1.delete_namespaced_pod(
            name=self.pod_name, 
            namespace=self.namespace
        )  # K8s will reschedule automatically
        return {'status': 'pod_deleted', 'reschedule': 'automatic'}
```

---

### Step 7: Deploy, Harden, and Integrate

**Goal:** Make it production-grade: enterprise auth, BYOC option, observability of the observer.

**Architecture (Docker Compose for local dev, Kubernetes for prod):**

```yaml
# docker-compose.yml
services:
  api:
    build: ./api
    environment:
      DATABASE_URL: postgresql://postgres:pass@db:5432/sre_agent
      ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}
      AUTONOMY_LEVEL: suggest
    ports: ["8000:8000"]

  collector:
    build: ./collector
    environment:
      KUBECONFIG: /etc/kube/config
      METRICS_INTERVAL_SECONDS: "15"
      DATABASE_URL: postgresql://postgres:pass@db:5432/sre_agent
    volumes:
      - ~/.kube:/etc/kube:ro

  causal-engine:
    build: ./causal-engine
    environment:
      DATABASE_URL: postgresql://postgres:pass@db:5432/sre_agent
      LOOKBACK_MINUTES: "60"

  db:
    image: timescale/timescaledb:latest-pg16
    environment:
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: sre_agent
    volumes: [db_data:/var/lib/postgresql/data]

  redis:
    image: redis:7-alpine  # for alert dedup + caching
```

**BYOC deployment checklist:**
- Deploy entirely within customer VPC, no data egress to your cloud
- Use customer's own Anthropic API key (or self-hosted model via Bedrock/Vertex)
- All data stays in customer's database
- Provide Helm chart for K8s deployment
- Audit logs for every AI action taken

**Key API endpoints to build:**

```
POST /api/alerts/ingest          # receive alerts from PagerDuty/OpsGenie webhooks
GET  /api/incidents              # list active incidents with RCA status
GET  /api/incidents/:id/rca      # full RCA report for an incident
POST /api/incidents/:id/remediate # trigger remediation (with autonomy gate)
GET  /api/graph/nodes            # live infrastructure graph
GET  /api/graph/blast-radius/:id # upstream/downstream of a node
POST /api/knowledge/ingest       # add runbook or post-mortem to knowledge bank
GET  /api/knowledge/search       # semantic search across knowledge bank
WebSocket /ws/incidents          # real-time incident stream
```

**Monitoring the monitor:**
- Track your own false positive rate per customer
- Alert on RCA confidence < 0.5 (escalate to human immediately)
- A/B test causal algorithms against ground-truth labels from confirmed incidents
- Build a feedback loop: when engineers override AI remediation, capture why

---

## Tech Stack Summary

| Layer | Technology |
|-------|-----------|
| Graph DB | PostgreSQL + recursive CTEs (or Neo4j for graph-native) |
| Time-series | TimescaleDB / InfluxDB |
| Vector search | pgvector |
| Causal inference | causallearn (PC algorithm) + custom scoring |
| LLM reasoning | Claude claude-sonnet-4-6 via Anthropic SDK (with prompt caching) |
| K8s integration | kubernetes-client (Python) |
| API | FastAPI (Python) or Go for latency-sensitive paths |
| Queue | Redis Streams or Kafka for high-volume alert ingestion |
| Auth | JWT + OAuth2 for enterprise SSO |
| Deployment | Helm chart for BYOC, Docker Compose for dev |