The agent that delegated

pip install anthropic
export ANTHROPIC_API_KEY="sk-ant-..."

import anthropic
client = anthropic.AsyncAnthropic()
print("SDK version:", anthropic.__version__)

FINDING_SCHEMA = {
    "type": "object",
    "properties": {
        "findings": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "file": {"type": "string"},
                    "line": {"type": "integer"},
                    "severity": {
                        "type": "string",
                        "enum": ["high", "medium", "low"]
                    },
                "title": {"type": "string"},
                "description": {"type": "string"}
            },
        "required": [
            "file", "line", "severity",
            "title", "description"
        ]
}
}
},
"required": ["findings"]
}

import anthropic
import asyncio
client = anthropic.AsyncAnthropic()
DIMENSIONS = [
    {
        "key": "bugs",
        "system": (
            "You are a code reviewer focused exclusively on "
            "correctness bugs. Ignore style, performance, and "
            "security. Return only genuine logic errors, off-by-one "
            "mistakes, null dereferences, and race conditions."
        ),
    },
    {
        "key": "perf",
        "system": (
            "You are a code reviewer focused exclusively on "
            "performance. Ignore correctness and security. Return "
            "only measurable inefficiencies: unnecessary allocations, "
            "O(n^2) where O(n) exists, missing indexes, redundant I/O."
        ),
    },
    {
        "key": "security",
        "system": (
            "You are a code reviewer focused exclusively on security "
            "vulnerabilities. Ignore style and performance. Return "
            "only exploitable issues: injection, auth bypass, secrets "
            "in code, unsafe deserialization."
        ),
    },
]

async def review_dimension(dimension, diff):
    """Run a single specialist reviewer on the diff."""
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=dimension["system"],
        messages=[
            {"role": "user", "content": f"Review this diff:\n\n{diff}"}
        ],
        tools=[
            {
                "name": "submit_findings",
                "description": "Submit structured review findings",
                "input_schema": FINDING_SCHEMA,
            }
        ],
        tool_choice={"type": "tool", "name": "submit_findings"},
    )
    for block in response.content:
        if block.type == "tool_use":
            return {"dimension": dimension["key"], **block.input}
            return {"dimension": dimension["key"], "findings": []}

            async def run_parallel_review(diff):
                """Fan out all specialist reviewers concurrently."""
                tasks = [review_dimension(d, diff) for d in DIMENSIONS]
                return await asyncio.gather(*tasks)

VERDICT_SCHEMA = {
    "type": "object",
    "properties": {
        "is_real": {"type": "boolean"},
        "reasoning": {"type": "string"},
    },
    "required": ["is_real", "reasoning"],
}

async def verify_finding(finding, diff):
    """Spawn a skeptic to try to refute a single finding."""
    prompt = (
        f"Finding: {finding['title']}\n"
        f"Description: {finding['description']}\n"
        f"Location: {finding['file']}:{finding['line']}\n\n"
        f"Diff:\n{diff}"
    )
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=(
            "You are a skeptical code reviewer. Your only job is "
            "to REFUTE the following finding. If you cannot refute "
            "it with evidence from the code, mark it as real. "
            "Default to refuted if uncertain."
        ),
        messages=[{"role": "user", "content": prompt}],
        tools=[
            {
                "name": "submit_verdict",
                "description": "Submit verification verdict",
                "input_schema": VERDICT_SCHEMA,
            }
        ],
        tool_choice={"type": "tool", "name": "submit_verdict"},
    )
    for block in response.content:
        if block.type == "tool_use":
            return {**finding, **block.input}
            return {**finding, "is_real": False, "reasoning": "Verification failed"}

            async def verify_all(results, diff):
                """Run adversarial verification on all findings."""
                all_findings = [
                    {**f, "dimension": r["dimension"]}
                    for r in results
                    for f in r["findings"]
                ]
                tasks = [verify_finding(f, diff) for f in all_findings]
                verified = await asyncio.gather(*tasks)
                return [f for f in verified if f["is_real"]]

async def main():
    diff = open("changes.diff").read()
    # Phase 1: parallel specialist review
    results = await run_parallel_review(diff)
    total = sum(len(r["findings"]) for r in results)
    print(f"Phase 1: {total} findings across {len(DIMENSIONS)} dimensions")
    for r in results:
        print(f"\n [{r['dimension'].upper()}] {len(r['findings'])} findings")
        for f in r["findings"]:
            print(f" [{f['severity']}] {f['file']}:{f['line']} {f['title']}")
            # Phase 2: adversarial verification
            confirmed = await verify_all(results, diff)
            print(f"\nPhase 2: {len(confirmed)} confirmed out of {total}")
            for f in confirmed:
                print(f" [{f['severity']}] {f['file']}:{f['line']} {f['title']}")
                print(f" {f['reasoning']}")

                asyncio.run(main())