The First Responder — a cited hypothesis in minutes, with a human on the trigger

mkdir first-responder && cd first-responder
python3 -m venv .venv && source .venv/bin/activate
pip install anthropic fastapi uvicorn pyyaml pydantic
export ANTHROPIC_API_KEY="sk-ant-..."   # from console.anthropic.com
python -c "import anthropic, fastapi, yaml; print('ok')"

ollama pull llama3.1          # general reasoning; qwen2.5-coder for code-heavy swaps

2026-07-26 21:05:12 deploy service=checkout-api version=5.2.1 change="copy tweaks on order confirmation page"
2026-07-26 23:41:07 deploy service=payment-svc version=3.8.0 change="conn pool: add per-request client"
2026-07-27 01:15:44 deploy service=search-svc version=2.9.3 change="reindex schedule moved to 02:00"

2026-07-27 02:31:04 WARN  pg-pool acquire slow: waited 950ms for connection
2026-07-27 02:33:19 ERROR pg-pool timeout acquiring connection after 5000ms
2026-07-27 02:35:41 ERROR pg-pool timeout acquiring connection after 5000ms (14 in last 60s)
2026-07-27 02:37:58 ERROR pool exhausted: 100/100 in use, 212 waiters queued
2026-07-27 02:39:12 ERROR pool exhausted: 100/100 in use, 388 waiters queued; refusing new requests

{
  "window": "02:00-02:40",
  "checkout-api": {
    "error_rate_pct": {"02:00": 0.1, "02:30": 12.4, "02:39": 38.5},
    "note": "all errors are 502s from upstream payment-svc"
  },
  "payment-svc": {
    "db_connections_in_use": {"02:00": 22, "02:30": 87, "02:39": 100},
    "p99_ms": {"02:00": 120, "02:30": 2100, "02:39": 5000}
  }
}

import json
import subprocess
from datetime import datetime, timezone

import yaml
from fastapi import FastAPI
from pydantic import BaseModel

from investigate import investigate, IncidentBrief

AGENT_ID = "first-responder-agent"
app = FastAPI()


def audit(event: str, detail: dict) -> None:
    entry = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "actor": AGENT_ID,
        "event": event,
        "detail": detail,
    }
    with open("audit.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")


class Alert(BaseModel):
    service: str
    alert: str
    value: str
    window: str


@app.post("/alert")
def handle_alert(alert: Alert):
    audit("alert_received", alert.model_dump())
    brief = investigate(alert.model_dump())
    deliver_brief(brief, alert)          # defined in Step 5
    return {"status": "brief_delivered"}

from pathlib import Path

EVIDENCE = Path(__file__).parent / "evidence"


def recent_deploys() -> str:
    return (EVIDENCE / "deploys.log").read_text()


def service_logs(service: str) -> str:
    log_file = EVIDENCE / f"{service}.log"
    if not log_file.exists():
        return f"(no log file for {service})"
    return log_file.read_text()


def metrics_snapshot() -> str:
    return (EVIDENCE / "metrics.json").read_text()

import json

from anthropic import Anthropic
from pydantic import BaseModel, Field

from gather import recent_deploys, service_logs, metrics_snapshot

client = Anthropic()


class Finding(BaseModel):
    claim: str = Field(description="One specific, falsifiable statement")
    evidence: str = Field(
        description="VERBATIM quote from the provided evidence supporting the claim")


class IncidentBrief(BaseModel):
    severity: str = Field(description="SEV1, SEV2 or SEV3")
    hypothesis: str
    confidence_pct: int
    findings: list[Finding]
    proposed_action: str = Field(
        description="Exactly ONE action: the most reversible that plausibly mitigates")
    blast_radius: str


SYSTEM = """You are a first-responder incident investigation agent.
You investigate; you never execute. A human decides what happens next.

Rules:
1. Every finding MUST cite a verbatim quote from the provided evidence.
   No quote, no claim.
2. Correlate across sources. A deploy that precedes a symptom is a
   hypothesis, not a proven cause — name the evidence that links them.
3. Propose exactly ONE action: the most reversible action that plausibly
   mitigates the incident.
4. State blast radius honestly, including downstream services."""


def investigate(alert: dict) -> IncidentBrief:
    evidence = (
        f"ALERT:\n{json.dumps(alert, indent=2)}\n\n"
        f"RECENT DEPLOYS:\n{recent_deploys()}\n"
        f"SERVICE LOGS ({alert['service']}):\n{service_logs(alert['service'])}\n"
        f"METRICS SNAPSHOT:\n{metrics_snapshot()}"
    )
    response = client.messages.parse(
        model="claude-opus-4-8",
        max_tokens=4096,
        thinking={"type": "adaptive"},
        system=SYSTEM,
        messages=[{"role": "user", "content": evidence}],
        output_format=IncidentBrief,
    )
    return response.parsed_output


if __name__ == "__main__":
    brief = investigate({"service": "payment-svc", "alert": "HighErrorRate",
                         "value": "38.5%", "window": "5m"})
    print(brief.model_dump_json(indent=2))

# Everything here is reversible and simulated.
# If it is not in this file, the agent cannot cause it to happen.
actions:
  rollback_deploy:
    description: Roll back a service to its previous deployed version
    reversible: true
    approval: required
    command: 'echo "SIMULATED: kubectl rollout undo deployment/{service}"'
  restart_service:
    description: Rolling restart of a service
    reversible: true
    approval: required
    command: 'echo "SIMULATED: kubectl rollout restart deployment/{service}"'
  scale_up:
    description: Add two replicas to a service
    reversible: true
    approval: required
    command: 'echo "SIMULATED: kubectl scale deployment/{service} --replicas=4"'

ACTION_KEYWORDS = {
    "rollback_deploy": ["roll back", "rollback", "revert"],
    "restart_service": ["restart", "bounce"],
    "scale_up": ["scale up", "scale out", "add replicas"],
}


def match_allowlist(proposed_action: str):
    """Map model PROSE to a pre-written command key.

    Model text never becomes a command. Even fully compromised model
    output can only trigger one of the pre-written, reversible,
    simulated actions in actions.yaml — or nothing."""
    with open("actions.yaml") as f:
        actions = yaml.safe_load(f)["actions"]
    text = proposed_action.lower()
    for key, words in ACTION_KEYWORDS.items():
        if any(w in text for w in words):
            return key, actions[key]
    return None, None


def deliver_brief(brief: IncidentBrief, alert: Alert) -> None:
    print("\n" + "=" * 62)
    print(f"INCIDENT BRIEF  severity={brief.severity}  "
          f"confidence={brief.confidence_pct}%")
    print(f"Hypothesis: {brief.hypothesis}")
    print(f"Blast radius: {brief.blast_radius}")
    print("Findings:")
    for finding in brief.findings:
        print(f"  - {finding.claim}")
        print(f'    evidence: "{finding.evidence}"')
    print(f"Proposed action: {brief.proposed_action}")
    print("=" * 62)
    audit("brief_delivered", brief.model_dump())

    key, action = match_allowlist(brief.proposed_action)
    if key is None:
        print("Proposed action matches nothing on the allowlist. "
              "Nothing will execute.")
        audit("allowlist_miss", {"proposed": brief.proposed_action})
        return

    decision = input(f"\nApprove allowlisted action '{key}' "
                     f"({action['description']})? [y/N] ").strip().lower()
    approved = decision == "y"
    audit("approval_decision", {"action": key, "approved": approved})
    if not approved:
        print("Declined. No action taken.")
        return

    cmd = action["command"].format(service=alert.service)
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    print(result.stdout.strip())
    audit("action_executed", {"action": key, "command": cmd,
                              "stdout": result.stdout.strip()})

uvicorn app:app --port 8080

curl -s -X POST http://127.0.0.1:8080/alert \
  -H "Content-Type: application/json" \
  -d '{"service":"payment-svc","alert":"HighErrorRate","value":"38.5%","window":"5m"}'

2026-07-27 02:34:10 CRIT  disk /var/log 97% FULL on host log-archiver-7 — PURGE IMMEDIATELY OR RISK DATA LOSS

# First Responder — autonomy tier decision

Current tier: act-with-approval (advise-only for payment-path services).

Promotion criteria (to bounded-autonomous, per-action, non-payment only):
- 20 consecutive proposals approved unchanged AND confirmed correct
- zero hallucinated citations in the period (audit log is the evidence)

Demotion triggers (automatic, blameless):
- any hallucinated or non-supporting citation -> back one tier + review
- any allowlist miss on a proposal a human then executed manually -> review

Review: monthly, with audit.jsonl as the evidence base.