# 08 · GenAI Incident Co-Pilot (Amazon Bedrock + Claude)

**Format:** AWS CDK (TypeScript) + Python Lambda · **Est. cost:** ~$0.01–0.05 per incident brief (Bedrock tokens); everything else is near-free at idle

When any CloudWatch alarm fires, the co-pilot acts as the **first responder**. It collects evidence, reasons about it with
Claude, and posts a brief to Slack/email **before a human has opened a laptop**.

```
CloudWatch alarm ─ALARM─▶ EventBridge ─▶ Lambda co-pilot
                                           ├─ metric history (last 60 min)
                                           ├─ Logs Insights: errors/timeouts (last 30 min)
                                           ├─ CloudTrail: write events (last 2 h)  ← "what changed?"
                                           ├─ Bedrock Knowledge Base: matching runbook (optional)
                                           └─ Bedrock Converse (Claude) ─▶ brief ─▶ Slack + SNS + DynamoDB
```

## Prerequisites
- **Bedrock model access** for an Anthropic Claude model in your region (Bedrock console → Model access)
- Node.js 18+, AWS CDK v2

## Deploy
```bash
npm install
npx cdk deploy -c alertEmail=you@example.com
# choose a different model / inference profile:
npx cdk deploy -c modelId=<your-claude-model-or-inference-profile-id>
```
Optional Slack: paste your incoming-webhook URL into the `SlackSecretArn` secret.

## Demo
```bash
aws lambda update-function-configuration --function-name demo-flaky-checkout \
  --environment "Variables={FAIL_RATE=0.7}"
for i in $(seq 1 40); do aws lambda invoke --function-name demo-flaky-checkout /dev/null >/dev/null; done
```
Within ~1–2 minutes the alarm fires and you get a brief like:
> **Summary:** Checkout is failing because payments-api calls time out. **Severity:** SEV2 …
> **Most likely cause (high):** connection pool exhausted (log evidence: "50/50 connections in use") …
> **Recent change:** `UpdateFunctionConfiguration` on demo-flaky-checkout at 14:02 by alice …

Reset: set `FAIL_RATE=0`.

## Add your runbooks (RAG)
1. Put Markdown runbooks in `runbooks/` (deployed to the `RunbookBucketName` bucket).
2. Bedrock console → **Knowledge bases → Create** → data source = that bucket → quick-create vector store → Sync.
3. Redeploy with `-c knowledgeBaseId=<KB_ID>`.

## Guardrails
- The co-pilot has **read-only** access to evidence. It never changes infrastructure.
- Prompts say "use only the evidence" and to state uncertainty explicitly.
- Every brief plus its evidence is stored in DynamoDB for post-incident review.

## Clean up
```bash
npx cdk destroy
```

#!/usr/bin/env node
import * as cdk from 'aws-cdk-lib';
import { IncidentCopilotStack } from '../lib/incident-copilot-stack';

const app = new cdk.App();
new IncidentCopilotStack(app, 'IncidentCopilot', {
  env: { account: process.env.CDK_DEFAULT_ACCOUNT, region: process.env.CDK_DEFAULT_REGION },
  alertEmail: app.node.tryGetContext('alertEmail'),
  // Use the newest Claude model / inference profile enabled in your Bedrock console.
  modelId: app.node.tryGetContext('modelId') ?? 'us.anthropic.claude-sonnet-4-5-20250929-v1:0',
  knowledgeBaseId: app.node.tryGetContext('knowledgeBaseId'),
  logGroupPrefixes: (app.node.tryGetContext('logGroupPrefixes') ?? '/aws/lambda/,/ecs/').split(','),
  withDemo: app.node.tryGetContext('withDemo') !== 'false',
});

import * as path from 'path';
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
import * as events from 'aws-cdk-lib/aws-events';
import * as targets from 'aws-cdk-lib/aws-events-targets';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as s3deploy from 'aws-cdk-lib/aws-s3-deployment';
import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager';
import * as sns from 'aws-cdk-lib/aws-sns';
import * as subs from 'aws-cdk-lib/aws-sns-subscriptions';

export interface IncidentCopilotStackProps extends cdk.StackProps {
  alertEmail?: string;
  modelId: string;
  knowledgeBaseId?: string;
  logGroupPrefixes: string[];
  withDemo: boolean;
}

/**
 * Alarm fires -> co-pilot gathers evidence (metric, logs, recent changes, runbooks)
 * -> Claude on Amazon Bedrock writes a first-response brief -> Slack / email + incident log.
 */
export class IncidentCopilotStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: IncidentCopilotStackProps) {
    super(scope, id, props);

    const topic = new sns.Topic(this, 'IncidentTopic', { displayName: 'Incident co-pilot' });
    if (props.alertEmail) topic.addSubscription(new subs.EmailSubscription(props.alertEmail));

    // Paste your Slack incoming-webhook URL into this secret after deploy (optional).
    const slackSecret = new secretsmanager.Secret(this, 'SlackWebhook', {
      description: 'Slack incoming webhook URL for incident briefs (leave as "none" to disable)',
      secretStringValue: cdk.SecretValue.unsafePlainText('none'),
    });

    const incidents = new dynamodb.TableV2(this, 'Incidents', {
      partitionKey: { name: 'incident_id', type: dynamodb.AttributeType.STRING },
      sortKey: { name: 'created_at', type: dynamodb.AttributeType.STRING },
      billing: dynamodb.Billing.onDemand(),
      timeToLiveAttribute: 'expires_at',
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Runbooks: point a Bedrock Knowledge Base at this bucket (see README).
    const runbooks = new s3.Bucket(this, 'RunbookBucket', {
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      enforceSSL: true,
      autoDeleteObjects: true,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });
    new s3deploy.BucketDeployment(this, 'SampleRunbooks', {
      sources: [s3deploy.Source.asset(path.join(__dirname, '..', 'runbooks'))],
      destinationBucket: runbooks,
    });

    const copilot = new lambda.Function(this, 'Copilot', {
      runtime: lambda.Runtime.PYTHON_3_12,
      architecture: lambda.Architecture.ARM_64,
      handler: 'copilot.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, '..', 'lambda', 'copilot')),
      timeout: cdk.Duration.minutes(3),
      memorySize: 512,
      environment: {
        MODEL_ID: props.modelId,
        KNOWLEDGE_BASE_ID: props.knowledgeBaseId ?? '',
        LOG_GROUP_PREFIXES: props.logGroupPrefixes.join(','),
        TOPIC_ARN: topic.topicArn,
        SLACK_SECRET_ARN: slackSecret.secretArn,
        INCIDENT_TABLE: incidents.tableName,
      },
      logGroup: new logs.LogGroup(this, 'CopilotLogs', {
        retention: logs.RetentionDays.ONE_MONTH,
        removalPolicy: cdk.RemovalPolicy.DESTROY,
      }),
    });
    topic.grantPublish(copilot);
    slackSecret.grantRead(copilot);
    incidents.grantWriteData(copilot);
    copilot.addToRolePolicy(
      new iam.PolicyStatement({
        sid: 'ReadOnlyEvidence',
        actions: [
          'cloudwatch:DescribeAlarms',
          'cloudwatch:GetMetricData',
          'logs:DescribeLogGroups',
          'logs:StartQuery',
          'logs:GetQueryResults',
          'cloudtrail:LookupEvents',
        ],
        resources: ['*'],
      }),
    );
    copilot.addToRolePolicy(
      new iam.PolicyStatement({
        sid: 'InvokeClaude',
        actions: ['bedrock:InvokeModel'], // Converse API is authorized by bedrock:InvokeModel
        resources: [
          `arn:${this.partition}:bedrock:*::foundation-model/anthropic.*`,
          `arn:${this.partition}:bedrock:*:${this.account}:inference-profile/*`,
        ],
      }),
    );
    if (props.knowledgeBaseId) {
      copilot.addToRolePolicy(
        new iam.PolicyStatement({
          actions: ['bedrock:Retrieve'],
          resources: [`arn:${this.partition}:bedrock:${this.region}:${this.account}:knowledge-base/${props.knowledgeBaseId}`],
        }),
      );
    }

    new events.Rule(this, 'AlarmToCopilot', {
      description: 'Every alarm that enters ALARM state gets an AI first-responder brief',
      eventPattern: {
        source: ['aws.cloudwatch'],
        detailType: ['CloudWatch Alarm State Change'],
        detail: { state: { value: ['ALARM'] } },
      },
      targets: [new targets.LambdaFunction(copilot, { retryAttempts: 1 })],
    });

    // ---------------------------------------------------------------- demo: a service that breaks on demand
    if (props.withDemo) {
      const flaky = new lambda.Function(this, 'FlakyCheckout', {
        functionName: 'demo-flaky-checkout',
        runtime: lambda.Runtime.PYTHON_3_12,
        handler: 'index.handler',
        environment: { FAIL_RATE: '0' },
        logGroup: new logs.LogGroup(this, 'FlakyLogs', {
          logGroupName: '/aws/lambda/demo-flaky-checkout',
          retention: logs.RetentionDays.ONE_WEEK,
          removalPolicy: cdk.RemovalPolicy.DESTROY,
        }),
        code: lambda.Code.fromInline(`
import os, random, time
def handler(event, context):
    rate = float(os.environ.get("FAIL_RATE", "0"))
    if random.random() < rate:
        print("ERROR PaymentGatewayTimeout: upstream payments-api did not respond within 3000ms (pool exhausted: 50/50 connections in use)")
        raise TimeoutError("payments-api timeout")
    time.sleep(0.05)
    return {"ok": True}
`),
      });
      new cloudwatch.Alarm(this, 'FlakyErrors', {
        alarmName: 'demo-checkout-errors',
        alarmDescription: 'Checkout errors above 3 per minute (demo)',
        metric: flaky.metricErrors({ period: cdk.Duration.minutes(1), statistic: 'Sum' }),
        threshold: 3,
        evaluationPeriods: 1,
        treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING,
      });
      new cdk.CfnOutput(this, 'DemoFunction', { value: flaky.functionName });
    }

    new cdk.CfnOutput(this, 'RunbookBucketName', { value: runbooks.bucketName });
    new cdk.CfnOutput(this, 'SlackSecretArn', { value: slackSecret.secretArn });
    new cdk.CfnOutput(this, 'IncidentTableName', { value: incidents.tableName });
  }
}

"""AI incident co-pilot.

Triggered by "CloudWatch Alarm State Change" (ALARM). Gathers evidence with
read-only APIs, asks Claude on Amazon Bedrock for a first-responder brief, then
posts it to SNS (+ Slack if configured) and records it in DynamoDB.
"""
import datetime as dt
import json
import os
import time
import urllib.request
import uuid

import boto3

MODEL_ID = os.environ["MODEL_ID"]
KB_ID = os.environ.get("KNOWLEDGE_BASE_ID", "")
LOG_PREFIXES = [p for p in os.environ.get("LOG_GROUP_PREFIXES", "").split(",") if p]
TOPIC_ARN = os.environ["TOPIC_ARN"]
SLACK_SECRET_ARN = os.environ["SLACK_SECRET_ARN"]
TABLE = boto3.resource("dynamodb").Table(os.environ["INCIDENT_TABLE"])

cw = boto3.client("cloudwatch")
logs = boto3.client("logs")
trail = boto3.client("cloudtrail")
bedrock = boto3.client("bedrock-runtime")
kb = boto3.client("bedrock-agent-runtime")
sns = boto3.client("sns")
secrets = boto3.client("secretsmanager")

SYSTEM_PROMPT = """You are a senior SRE acting as the first responder for an AWS production incident.
Using ONLY the evidence provided, write a concise brief for the on-call engineer:
1. *Summary* - one sentence, plain English.
2. *Severity* - SEV1/SEV2/SEV3 with a one-line justification.
3. *Most likely cause* - rank up to 3 hypotheses with confidence (high/medium/low) and the evidence for each.
4. *Next steps* - numbered, concrete commands or console actions, safest first.
5. *What NOT to do* - one or two risky actions to avoid.
If evidence is missing or contradictory, say so explicitly instead of guessing."""


def handler(event, _context):
    detail = event["detail"]
    alarm_name = detail["alarmName"]
    now = dt.datetime.now(dt.timezone.utc)

    evidence = {
        "alarm": {
            "name": alarm_name,
            "reason": detail["state"].get("reason"),
            "description": detail.get("configuration", {}).get("description"),
            "since": detail["state"].get("timestamp"),
            "region": event.get("region"),
            "account": event.get("account"),
        },
        "metric_last_hour": metric_history(alarm_name, now),
        "error_logs": error_logs(alarm_name, detail, now),
        "recent_changes": recent_changes(now),
        "runbook_excerpts": runbooks(alarm_name, detail),
    }

    brief = ask_claude(evidence)
    incident_id = f"{now:%Y%m%d}-{uuid.uuid4().hex[:6]}"
    text = f":rotating_light: *{alarm_name}* - incident `{incident_id}`\n\n{brief}"

    sns.publish(TopicArn=TOPIC_ARN, Subject=f"[co-pilot] {alarm_name}"[:100], Message=text)
    post_to_slack(text)
    TABLE.put_item(Item={
        "incident_id": incident_id,
        "created_at": now.isoformat(),
        "alarm": alarm_name,
        "brief": brief,
        "evidence": json.dumps(evidence, default=str)[:350_000],
        "expires_at": int(time.time()) + 90 * 86400,
    })
    return {"incident_id": incident_id}


# ------------------------------------------------------------------ evidence
def metric_history(alarm_name, now):
    alarms = cw.describe_alarms(AlarmNames=[alarm_name])["MetricAlarms"]
    if not alarms or "MetricName" not in alarms[0]:
        return "composite or math alarm - metric history skipped"
    a = alarms[0]
    resp = cw.get_metric_data(
        MetricDataQueries=[{
            "Id": "m",
            "MetricStat": {
                "Metric": {"Namespace": a["Namespace"], "MetricName": a["MetricName"], "Dimensions": a.get("Dimensions", [])},
                "Period": 60,
                "Stat": a.get("Statistic", "Average"),
            },
        }],
        StartTime=now - dt.timedelta(hours=1),
        EndTime=now,
    )
    r = resp["MetricDataResults"][0]
    points = sorted(zip(r["Timestamps"], r["Values"]))[-30:]
    return {
        "metric": f"{a['Namespace']}/{a['MetricName']}",
        "dimensions": a.get("Dimensions", []),
        "threshold": f"{a.get('ComparisonOperator')} {a.get('Threshold')}",
        "points": [(t.strftime("%H:%M"), round(v, 2)) for t, v in points],
    }


def candidate_log_groups(alarm_name, detail):
    # Prefer log groups whose names match the alarm's dimension values (e.g. FunctionName).
    hints = [alarm_name]
    for m in detail.get("configuration", {}).get("metrics", []):
        for v in m.get("metricStat", {}).get("metric", {}).get("dimensions", {}).values():
            hints.append(str(v))
    groups = []
    for prefix in LOG_PREFIXES:
        for page in logs.get_paginator("describe_log_groups").paginate(logGroupNamePrefix=prefix):
            groups += [g["logGroupName"] for g in page["logGroups"]]
    matched = [g for g in groups if any(h and h in g for h in hints)]
    return (matched or groups)[:10]


def error_logs(alarm_name, detail, now):
    groups = candidate_log_groups(alarm_name, detail)
    if not groups:
        return "no log groups found"
    q = logs.start_query(
        logGroupNames=groups,
        startTime=int((now - dt.timedelta(minutes=30)).timestamp()),
        endTime=int(now.timestamp()),
        queryString=(
            "fields @timestamp, @log, @message "
            "| filter @message like /(?i)(error|exception|timeout|fatal|denied|throttl)/ "
            "| sort @timestamp desc | limit 25"
        ),
        limit=25,
    )
    for _ in range(30):
        res = logs.get_query_results(queryId=q["queryId"])
        if res["status"] in ("Complete", "Failed", "Cancelled", "Timeout"):
            break
        time.sleep(1)
    rows = [{f["field"]: f["value"] for f in row} for row in res.get("results", [])]
    return {"log_groups": groups, "matches": [f"{r.get('@timestamp')} {r.get('@log')} {r.get('@message', '')[:400]}" for r in rows]}


def recent_changes(now):
    events = trail.lookup_events(
        LookupAttributes=[{"AttributeKey": "ReadOnly", "AttributeValue": "false"}],
        StartTime=now - dt.timedelta(hours=2),
        EndTime=now,
        MaxResults=25,
    )["Events"]
    return [
        f"{e['EventTime']:%H:%M} {e['EventSource']}:{e['EventName']} by {e.get('Username', '?')} "
        f"on {[r.get('ResourceName') for r in e.get('Resources', [])][:3]}"
        for e in events
    ]


def runbooks(alarm_name, detail):
    if not KB_ID:
        return "no knowledge base configured"
    query = f"Runbook for alarm {alarm_name}: {detail.get('configuration', {}).get('description', '')}"
    res = kb.retrieve(knowledgeBaseId=KB_ID, retrievalQuery={"text": query},
                      retrievalConfiguration={"vectorSearchConfiguration": {"numberOfResults": 3}})
    return [r["content"]["text"][:1500] for r in res.get("retrievalResults", [])]


# ------------------------------------------------------------------ reasoning + delivery
def ask_claude(evidence):
    resp = bedrock.converse(
        modelId=MODEL_ID,
        system=[{"text": SYSTEM_PROMPT}],
        messages=[{
            "role": "user",
            "content": [{"text": "Evidence (JSON):\n" + json.dumps(evidence, default=str)[:60_000]}],
        }],
        inferenceConfig={"maxTokens": 1200, "temperature": 0.2},
    )
    return "".join(block.get("text", "") for block in resp["output"]["message"]["content"])


def post_to_slack(text):
    url = secrets.get_secret_value(SecretId=SLACK_SECRET_ARN)["SecretString"].strip()
    if not url.startswith("https://hooks.slack.com/"):
        return
    req = urllib.request.Request(url, data=json.dumps({"text": text}).encode(),
                                 headers={"Content-Type": "application/json"})
    urllib.request.urlopen(req, timeout=10)

# Runbook: checkout errors / payments-api timeouts

**Alarm:** `demo-checkout-errors` · **Service owner:** payments-team · **Escalation:** #payments-oncall

## Symptoms
- Spike in `PaymentGatewayTimeout` errors in `/aws/lambda/demo-flaky-checkout`
- Messages mention connection pool exhaustion (`50/50 connections in use`)

## Known causes (most common first)
1. **Connection pool exhaustion** after a traffic burst or slow downstream. Fix: raise the pool to 100 via the
   `POOL_SIZE` env var, then restart.
2. **Recent deploy** changed timeouts. Check CloudTrail for `UpdateFunctionConfiguration` /
   `UpdateFunctionCode` in the last 2 hours; roll back to the previous version alias.
3. **Payments provider outage**. Check the provider status page; enable the "queue orders, charge later" feature flag.

## Safe first actions
1. Confirm the blast radius: error % vs total invocations.
2. Roll back the latest change if one happened in the last 2 hours.
3. Enable the degraded-mode flag before touching infrastructure.

## Do NOT
- Do not raise the Lambda timeout above 10s (it hides the problem and multiplies cost).
- Do not purge the orders queue.

{
  "app": "npx ts-node --prefer-ts-exts bin/app.ts",
  "watch": { "include": ["**"], "exclude": ["README.md", "cdk*.json", "**/*.d.ts", "**/*.js", "node_modules", "cdk.out"] },
  "context": {}
}

{
  "name": "08-genai-incident-copilot",
  "version": "1.0.0",
  "private": true,
  "bin": { "app": "bin/app.js" },
  "scripts": {
    "build": "tsc",
    "synth": "cdk synth",
    "deploy": "cdk deploy --all",
    "destroy": "cdk destroy --all"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "aws-cdk": "^2.1000.0",
    "ts-node": "^10.9.2",
    "typescript": "~5.6.3"
  },
  "dependencies": {
    "aws-cdk-lib": "^2.200.0",
    "constructs": "^10.4.2"
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "commonjs",
    "lib": ["es2022"],
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "noEmit": true,
    "types": ["node"]
  },
  "exclude": ["node_modules", "cdk.out"]
}