Capstone A, built end to end

Chapter 53 argued the repository-audit fleet as a design; this chapter is the runbook: every command and every file, in the order you run them, from an empty directory to a validated overnight run. The stack is Bedrock (model plane), Strands (the open-source worker framework), a sandbox for running checks, and Step Functions distributed map for fan-out. This box has no AWS account, so the AWS steps are follow-along and their outputs are illustrative; the local simulation is fully runnable.

Goal, gates, patterns. Audit a large repo overnight; return deduplicated, verified, provenance-linked findings under a token budget, resumable after a crash. Dominant gates: 3 (budgets), 6 (resumability), 8 (verification). Patterns: orchestrator-workers, debate/judge panel, checkpoint-resume, budget-bounded admission, ReAct.

Step 0a: rehearse locally first

Before any AWS command, climb rung 1 and 2. For this capstone that is three checks, all free:

# rung 1: the whole pipeline's logic (shard -> budget -> panel -> merge)
python3 capstone_hive.py

# rung 1: the finder's prompt shape against a replaying fake model
python3 -m pytest tests/test_finder.py -q

# rung 2: does the state machine definition actually parse and fan out?
docker run --rm -d -p 8083:8083 amazon/aws-stepfunctions-local
aws stepfunctions --endpoint-url http://localhost:8083 create-state-machine \
  --name hive-local --definition file://statemachine.json \
  --role-arn arn:aws:iam::123456789012:role/DummyRole

The Step Functions Local check is the highest-value one here: a malformed ItemReader or a MaxConcurrency typo costs seconds to find locally and an hour to find after a container build and deploy.

Architecture

  repo -> planner -> shards in S3
                        |
            Step Functions distributed map (MaxConcurrency = N*)
             /          |            \
       worker Lambda (finder: Strands + Bedrock + sandbox tool)
             \          |            /
                 raw findings -> DynamoDB
                        |
              dedup -> verifier panel (3 blind refuters, frontier)
                        |
                 verified findings -> merge.py -> report in S3

Step 0: prerequisites

Install the tooling and confirm Bedrock access. Run these once.

# AWS CLI v2 and Docker with buildx (ARM64), Python 3.11+
aws --version && docker buildx version && python3 --version

# Configure credentials and a region where Bedrock is enabled
aws configure          # set key, secret, region (e.g. us-east-1)
export AWS_REGION=us-east-1 ACCOUNT=$(aws sts get-caller-identity --query Account --output text)

# Enable model access in the console once (Bedrock -> Model access),
# then confirm the two models resolve:
aws bedrock list-foundation-models --region $AWS_REGION \
  --query "modelSummaries[?contains(modelId,'claude')].modelId" --output text

Create the project:

mkdir hive-audit && cd hive-audit
python3 -m venv .venv && source .venv/bin/activate
printf 'strands-agents\nboto3\n' > requirements.txt
pip install -r requirements.txt

Step 1: create the state store and buckets

aws s3 mb s3://hive-audit-$ACCOUNT --region $AWS_REGION

aws dynamodb create-table --table-name hive-audit-state \
  --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=S \
  --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST --region $AWS_REGION

Step 2: the planner (planner.py)

Create planner.py verbatim, then run it to shard a repo snapshot you have already synced to s3://hive-audit-$ACCOUNT/repo/.

# planner.py
import boto3, json, os, sys

BUCKET = f"hive-audit-{os.environ['ACCOUNT']}"
s3 = boto3.client("s3")

def plan(repo_prefix="repo/", shard_size=25):
    keys, token = [], None
    while True:
        kw = {"Bucket": BUCKET, "Prefix": repo_prefix}
        if token: kw["ContinuationToken"] = token
        resp = s3.list_objects_v2(**kw)
        keys += [o["Key"] for o in resp.get("Contents", [])]
        token = resp.get("NextContinuationToken")
        if not token: break
    keys.sort()                                  # related files adjacent
    shards = [keys[i:i+shard_size] for i in range(0, len(keys), shard_size)]
    for i, files in enumerate(shards):
        s3.put_object(Bucket=BUCKET, Key=f"shards/{i:04d}.json",
                      Body=json.dumps({"shard": i, "files": files}))
    print(f"wrote {len(shards)} shards")

if __name__ == "__main__":
    plan(sys.argv[1] if len(sys.argv) > 1 else "repo/")

python planner.py repo/
aws s3 ls s3://hive-audit-$ACCOUNT/shards/ | head       # confirm shards exist

Step 3: the finder agent (finder.py)

# finder.py
import json
from strands import Agent, tool
from strands.models import BedrockModel

@tool
def read_file(path: str) -> str:
    """Read a repo file already mounted at /workspace."""
    with open(f"/workspace/{path}") as f:
        return f.read()

@tool
def run_check(code: str) -> str:
    """Run a short Python check in the no-egress sandbox; return stdout."""
    from sandbox import exec_no_egress          # step 5
    return exec_no_egress(code)

FINDER_SYSTEM = (
    "You audit a shard of a repository for bugs, security smells, and dead "
    "code. Output a JSON array; each item {\"file\",\"line\",\"kind\","
    "\"severity\",\"evidence\"}. Cite the exact line as evidence. Use "
    "run_check only to confirm a suspected bug. Report nothing you cannot "
    "evidence.")

finder = Agent(model=BedrockModel(model_id="anthropic.claude-sonnet-5"),
               tools=[read_file, run_check], system_prompt=FINDER_SYSTEM)

def audit_shard(files: list[str]) -> list[dict]:
    result = finder(f"Audit these files: {files}")
    try:
        return json.loads(str(result))
    except json.JSONDecodeError:
        return []                                # a malformed pass -> no findings

Step 4: the verifier panel (verifier.py)

# verifier.py
import json
from strands import Agent, tool
from finder import read_file

refuter = Agent(
    model=BedrockModel(model_id="anthropic.claude-opus-4-8"),   # frontier
    tools=[read_file],
    system_prompt=("Given a claimed bug and its evidence, try to REFUTE it "
                   "against the cited file. Default to refuted when uncertain. "
                   "Answer only JSON: {\"real\": true|false}."))
from strands.models import BedrockModel          # noqa: E402 (import order)

def verify(finding: dict, votes: int = 3) -> bool:
    kept = 0
    for _ in range(votes):
        out = json.loads(str(refuter(f"Claim: {json.dumps(finding)}")))
        kept += 1 if out.get("real") else 0
    return kept > votes // 2                      # majority keeps

Step 5: the sandbox (sandbox.py)

A no-egress executor. For the runbook, use a local subprocess with the network disabled via a restricted environment; in production swap in E2B or a Firecracker sandbox (the interface is the same one function).

# sandbox.py
import subprocess, sys

def exec_no_egress(code: str, timeout: int = 10) -> str:
    """Run untrusted code with no network and a hard timeout.
    Production: replace with an E2B/Firecracker session (same signature)."""
    try:
        p = subprocess.run([sys.executable, "-c", code], capture_output=True,
                           text=True, timeout=timeout, env={"PATH": "/usr/bin"})
        return (p.stdout or p.stderr)[:2000]
    except subprocess.TimeoutExpired:
        return "error: check timed out"

Step 6: the worker entrypoint (worker.py)

This is what the distributed map invokes per shard: read the shard, admit against the budget, run the finder, write findings and a checkpoint event to DynamoDB.

# worker.py
import boto3, json, os
from finder import audit_shard

BUCKET = f"hive-audit-{os.environ['ACCOUNT']}"
s3, ddb = boto3.client("s3"), boto3.client("dynamodb")
FIND_COST = 22_000

def admit(run_id: str, cost: int, cap: int) -> bool:
    try:
        ddb.update_item(
            TableName="hive-audit-state",
            Key={"pk": {"S": f"run#{run_id}"}, "sk": {"S": "budget"}},
            UpdateExpression="SET spent = spent + :c",
            ConditionExpression="spent + :c <= :cap",
            ExpressionAttributeValues={":c": {"N": str(cost)}, ":cap": {"N": str(cap)}})
        return True
    except ddb.exceptions.ConditionalCheckFailedException:
        return False                              # budget exhausted

def handler(event, _ctx=None):                    # event = one shard item
    run_id, cap = event["run_id"], int(event["cap"])
    shard = json.loads(s3.get_object(Bucket=BUCKET, Key=event["shard_key"])
                       ["Body"].read())
    if not admit(run_id, FIND_COST, cap):
        return {"shard": shard["shard"], "status": "exhausted"}
    findings = audit_shard(shard["files"])
    for i, f in enumerate(findings):
        ddb.put_item(TableName="hive-audit-state", Item={
            "pk": {"S": f"run#{run_id}"},
            "sk": {"S": f"finding#{shard['shard']:04d}#{i:03d}"},
            "data": {"S": json.dumps(f)}})
    ddb.put_item(TableName="hive-audit-state", Item={   # checkpoint
        "pk": {"S": f"run#{run_id}"},
        "sk": {"S": f"unit#{shard['shard']:04d}"},
        "status": {"S": "done"}})
    return {"shard": shard["shard"], "findings": len(findings)}

Step 7: the merger (merge.py)

# merge.py
import boto3, json, os, sys
from verifier import verify

ddb = boto3.client("dynamodb")

def merge(run_id: str) -> dict:
    items = ddb.query(TableName="hive-audit-state",
        KeyConditionExpression="pk = :p AND begins_with(sk, :s)",
        ExpressionAttributeValues={":p": {"S": f"run#{run_id}"},
                                   ":s": {"S": "finding#"}})["Items"]
    raw = [json.loads(i["data"]["S"]) for i in items]
    seen, deduped = set(), []
    for f in raw:                                 # dedup by file+line+kind
        key = (f["file"], f["line"], f["kind"])
        if key not in seen:
            seen.add(key); deduped.append(f)
    verified = [f for f in deduped if verify(f)]  # the panel
    return {"raw": len(raw), "deduped": len(deduped), "verified": verified}

if __name__ == "__main__":
    out = merge(sys.argv[1])
    print(f"raw {out['raw']} -> deduped {out['deduped']} -> "
          f"verified {len(out['verified'])}")
    print(json.dumps(out["verified"][:5], indent=2))

Step 8: containerize and push (Dockerfile)

# Dockerfile
FROM public.ecr.aws/lambda/python:3.12-arm64
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY finder.py verifier.py sandbox.py worker.py ./
CMD ["worker.handler"]

aws ecr create-repository --repository-name hive-audit --region $AWS_REGION
aws ecr get-login-password --region $AWS_REGION | \
  docker login --username AWS --password-stdin $ACCOUNT.dkr.ecr.$AWS_REGION.amazonaws.com
docker buildx build --platform linux/arm64 \
  -t $ACCOUNT.dkr.ecr.$AWS_REGION.amazonaws.com/hive-audit:latest --push .

Step 9: the IAM execution role

cat > trust.json <<'JSON'
{"Version":"2012-10-17","Statement":[{"Effect":"Allow",
 "Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}
JSON
aws iam create-role --role-name hive-audit-worker \
  --assume-role-policy-document file://trust.json

cat > policy.json <<JSON
{"Version":"2012-10-17","Statement":[
 {"Effect":"Allow","Action":["bedrock:InvokeModel","bedrock:InvokeModelWithResponseStream"],
  "Resource":["arn:aws:bedrock:*::foundation-model/anthropic.claude-sonnet-5",
              "arn:aws:bedrock:*::foundation-model/anthropic.claude-opus-4-8"]},
 {"Effect":"Allow","Action":["s3:GetObject","s3:PutObject"],
  "Resource":"arn:aws:s3:::hive-audit-$ACCOUNT/*"},
 {"Effect":"Allow","Action":["dynamodb:PutItem","dynamodb:UpdateItem","dynamodb:Query"],
  "Resource":"arn:aws:dynamodb:$AWS_REGION:$ACCOUNT:table/hive-audit-state"},
 {"Effect":"Allow","Action":["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents"],
  "Resource":"arn:aws:logs:*:$ACCOUNT:*"}]}
JSON
aws iam put-role-policy --role-name hive-audit-worker \
  --policy-name hive-audit --policy-document file://policy.json

Step 10: the worker Lambda and the fan-out state machine

aws lambda create-function --function-name hive-audit-worker \
  --package-type Image \
  --code ImageUri=$ACCOUNT.dkr.ecr.$AWS_REGION.amazonaws.com/hive-audit:latest \
  --role arn:aws:iam::$ACCOUNT:role/hive-audit-worker \
  --timeout 600 --memory-size 2048 \
  --environment "Variables={ACCOUNT=$ACCOUNT}" --region $AWS_REGION

statemachine.json, a distributed map over the shards (set MaxConcurrency to your N* from the sizing arithmetic, not 10,000):

{
  "Comment": "Hive repo-audit fan-out",
  "StartAt": "FanOut",
  "States": {
    "FanOut": {
      "Type": "Map", "ItemProcessor": {
        "ProcessorConfig": {"Mode": "DISTRIBUTED", "ExecutionType": "STANDARD"},
        "StartAt": "AuditShard",
        "States": {"AuditShard": {
          "Type": "Task", "Resource": "arn:aws:states:::lambda:invoke",
          "Parameters": {"FunctionName": "hive-audit-worker",
            "Payload": {"shard_key.$": "$.Key",
                        "run_id.$": "$$.Execution.Input.run_id",
                        "cap.$": "$$.Execution.Input.cap"}},
          "End": true}}},
      "ItemReader": {"Resource": "arn:aws:states:::s3:listObjectsV2",
        "Parameters": {"Bucket.$": "$.bucket", "Prefix": "shards/"}},
      "MaxConcurrency": 120,
      "ToleratedFailurePercentage": 5,
      "End": true
    }
  }
}

Seed the budget row, then create and start the machine:

aws dynamodb put-item --table-name hive-audit-state --region $AWS_REGION \
  --item '{"pk":{"S":"run#night1"},"sk":{"S":"budget"},"spent":{"N":"0"}}'

aws stepfunctions create-state-machine --name hive-audit \
  --definition file://statemachine.json \
  --role-arn arn:aws:iam::$ACCOUNT:role/hive-audit-sfn --region $AWS_REGION

Step 11: run it

aws stepfunctions start-execution \
  --state-machine-arn arn:aws:states:$AWS_REGION:$ACCOUNT:stateMachine:hive-audit \
  --input "{\"run_id\":\"night1\",\"cap\":2000000,\"bucket\":\"hive-audit-$ACCOUNT\"}"
# watch it:
aws stepfunctions describe-execution --execution-arn <arn-from-above> \
  --query '{status:status}'                       # RUNNING -> SUCCEEDED

Step 12: validate the results

Validate in four steps, in order:

1. Pre-flight the whole pipeline locally (fully runnable, no AWS):

python3 capstone_hive.py       # the book's Capstone A simulation

--- verification mesh (blind panel, dedup first) ---
  raw findings     : 28 (15 real, precision 53%)
  after panel      : 16 (15 real, precision 93%, recall 83%)

If the local sim's precision lift or budget math is wrong, the design is wrong; fix it before the cloud run.

2. Merge and read the verified report:

python merge.py night1
# raw 41 -> deduped 33 -> verified 22      (illustrative)

3. Confirm the gates from DynamoDB:

# budget held under cap (gate 3):
aws dynamodb get-item --table-name hive-audit-state \
  --key '{"pk":{"S":"run#night1"},"sk":{"S":"budget"}}' --query 'Item.spent.N'
# checkpoints exist (gate 6):
aws dynamodb query --table-name hive-audit-state \
  --key-condition-expression "pk = :p AND begins_with(sk, :s)" \
  --expression-attribute-values '{":p":{"S":"run#night1"},":s":{"S":"unit#"}}' \
  --select COUNT

4. Spot-audit provenance: pick five verified findings from step 2 and follow each to its cited file and line. If a finding cannot be traced, merge.py dropped provenance, fix the merge, not the finders.

Troubleshoot

Don't be confused: the finder's confidence vs the panel's verdict. A finder reporting a bug confidently is a candidate, not a verified finding. The panel's job is to disagree with confident finders, and its verdict, not the finder's self-assessment, is what reaches the report. Skipping the panel because the finders "seem sure" is exactly how you ship the 53%-precision report.

👉 Next: Capstone B, built end to end, the incident responder on Bedrock and LangGraph, as a runbook with the read-only IAM and the human-approval gate wired step by step.