LLM Incident Runbooks: Ground On-Call Answers in CloudWatch Signals

// incident-bot/context.ts
export type IncidentContext = {
  alarmArn: string;
  service: string;          // e.g. checkout-api
  env: "prod" | "staging";
  region: string;
  startedAt: string;        // ISO
  severity: "SEV1" | "SEV2" | "SEV3";
};

export function systemPrompt(ctx: IncidentContext): string {
  return [
    "You are an on-call assistant. Cite CloudWatch/X-Ray evidence for every claim.",
    `Service=${ctx.service} env=${ctx.env} region=${ctx.region} sev=${ctx.severity}`,
    `Alarm=${ctx.alarmArn} started=${ctx.startedAt}`,
    "Never invent metric names. Prefer tool results over memory.",
    "Remediations must map to allowlisted runbook IDs.",
  ].join("\n");
}

# tools/logs_insights.py
import time, boto3

logs = boto3.client("logs")

APPROVED = {
  "error_rate_by_path": """
fields @timestamp, path, status
| filter status >= 500 and service = '{service}'
| stats count() as errors by path
| sort errors desc
| limit 20
""",
  "p99_latency": """
fields @timestamp, latencyMs, path
| filter service = '{service}'
| stats pct(latencyMs, 99) as p99 by path
| sort p99 desc
| limit 20
""",
}

def run_approved(query_id: str, service: str, log_group: str, start: int, end: int) -> dict:
    if query_id not in APPROVED:
        raise ValueError("query_not_allowlisted")
    q = APPROVED[query_id].format(service=service)
    # ❌ Never: interpolate raw model text into the query string
    start_q = logs.start_query(
        logGroupName=log_group,
        startTime=start,
        endTime=end,
        queryString=q,
    )
    qid = start_q["queryId"]
    while True:
        r = logs.get_query_results(queryId=qid)
        if r["status"] in ("Complete", "Failed", "Cancelled"):
            return {"queryId": qid, "status": r["status"], "results": r.get("results", [])}
        time.sleep(0.4)

// tools/xray.ts
import { XRayClient, GetTraceSummariesCommand, BatchGetTracesCommand } from "@aws-sdk/client-xray";

const xray = new XRayClient({});

export async function slowTraces(service: string, start: Date, end: Date) {
  const summaries = await xray.send(new GetTraceSummariesCommand({
    StartTime: start,
    EndTime: end,
    FilterExpression: `service("${service}") AND responseTime > 2`,
  }));
  const ids = (summaries.TraceSummaries ?? []).slice(0, 5).map(t => t.Id!).filter(Boolean);
  if (!ids.length) return { traces: [], note: "no_slow_traces" };
  const batch = await xray.send(new BatchGetTracesCommand({ TraceIds: ids }));
  // ✅ Return truncated segment names + durations, not full payloads with PII
  return {
    traces: (batch.Traces ?? []).map(t => ({
      id: t.Id,
      duration: t.Duration,
      segments: (t.Segments ?? []).slice(0, 12).map(s => s.Document?.slice(0, 400)),
    })),
  };
}

# runbooks/allowlist.yaml
actions:
  - id: rollback_ecs_service
    requires: ["evidence.deploy_marker", "sev<=SEV2"]
    params: ["service", "previousTaskDef"]
  - id: scale_out_asg
    requires: ["evidence.cpu_or_latency"]
    params: ["asgName", "desiredCapacity"]
    maxDesired: 20
  - id: flip_feature_flag
    requires: ["evidence.error_burst"]
    params: ["flagKey", "value"]
# ❌ no: arbitrary aws cli / kubectl from the model

function assertAction(action: { id: string; params: Record<string, unknown> }, allow: Set<string>) {
  if (!allow.has(action.id)) throw new Error(`deny:${action.id}`);
  // ✅ Human approval gate for SEV1 destructive actions
}