The AI Code Reviewer: Clones, Tests, and Reports on Every PR

import asyncio
import json
import os
import hmac
import hashlib
import httpx
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

SANDBOX_URL = os.environ["SANDBOX_URL"]
SANDBOX_API_KEY = os.environ["SANDBOX_API_KEY"]
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
GITHUB_WEBHOOK_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]
HEADERS = {
    "Authorization": f"Bearer {SANDBOX_API_KEY}",
    "Content-Type": "application/json",
}

# Files that indicate frontend changes
FRONTEND_PATTERNS = [
    "src/components/", "src/pages/", "src/app/",
    "public/", ".css", ".tsx", ".jsx",
    "templates/", "static/",
]


@app.post("/webhook/github")
async def handle_github_webhook(request: Request):
    # Verify webhook signature
    body = await request.body()
    sig = request.headers.get("X-Hub-Signature-256", "")
    expected = "sha256=" + hmac.new(
        GITHUB_WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        raise HTTPException(401, "Invalid signature")

    event = request.headers.get("X-GitHub-Event")
    payload = json.loads(body)

    if event == "pull_request" and payload["action"] in ("opened", "synchronize"):
        pr = payload["pull_request"]
        asyncio.create_task(run_review(
            repo=payload["repository"]["full_name"],
            pr_number=pr["number"],
            branch=pr["head"]["ref"],
            base_branch=pr["base"]["ref"],
            clone_url=payload["repository"]["clone_url"],
            changed_files=[f["filename"] for f in pr.get("files", [])],
        ))

    return {"status": "accepted"}

async def run_review(
    repo: str,
    pr_number: int,
    branch: str,
    base_branch: str,
    clone_url: str,
    changed_files: list,
):
    has_frontend = any(
        any(pattern in f for pattern in FRONTEND_PATTERNS)
        for f in changed_files
    )

    async with httpx.AsyncClient(timeout=600) as client:
        # 1. Create sandbox
        sandbox = (await client.post(
            f"{SANDBOX_URL}/api/sandboxes", headers=HEADERS,
            json={
                "name": f"pr-review-{repo.split('/')[-1]}-{pr_number}",
                "instance_type": "ab0t.small",
                "auto_stop_minutes": 15,
            },
        )).json()
        sandbox_id = sandbox["sandbox_id"]

        # 2. Wait for boot
        for _ in range(24):
            status = (await client.get(
                f"{SANDBOX_URL}/api/sandboxes/{sandbox_id}",
                headers=HEADERS)).json()
            if status["status"] == "running": break
            await asyncio.sleep(5)

        # 3. Clone, install, test
        clone_cmd = f"""
cd /workspace &&
git clone https://x-access-token:{GITHUB_TOKEN}@github.com/{repo}.git project &&
cd project &&
git checkout {branch} &&
npm install 2>&1 | tail -5 &&
echo '--- TESTS ---' &&
npm test 2>&1;
echo "EXIT_CODE=$?"
"""
        test_result = (await client.post(
            f"{SANDBOX_URL}/api/sandboxes/{sandbox_id}/execute",
            headers=HEADERS,
            json={"command": clone_cmd},
        )).json()

        test_output = test_result["stdout"]
        test_passed = "EXIT_CODE=0" in test_output

        # 4. Get the diff for AI review
        diff_result = (await client.post(
            f"{SANDBOX_URL}/api/sandboxes/{sandbox_id}/execute",
            headers=HEADERS,
            json={"command": f"cd /workspace/project && git diff {base_branch}...{branch} | head -2000"},
        )).json()

        # 5. Visual review (if frontend)
        visual_report = ""
        if has_frontend:
            visual_report = await visual_review(
                client, sandbox_id, repo, branch, base_branch)

        # 6. AI-powered code review
        review_notes = await ai_code_review(
            diff_result["stdout"], test_output, changed_files)

        # 7. Post PR comment
        comment = format_pr_comment(
            test_passed, test_output, review_notes, visual_report, changed_files)
        await post_pr_comment(repo, pr_number, comment)

        # 8. Cleanup
        await client.delete(
            f"{SANDBOX_URL}/api/sandboxes/{sandbox_id}", headers=HEADERS)

import anthropic

async def ai_code_review(
    diff: str,
    test_output: str,
    changed_files: list,
) -> str:
    """Use Claude to review the code diff and write review notes."""
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-4-6-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""You are a senior software engineer reviewing a pull request.

## Changed files
{json.dumps(changed_files, indent=2)}

## Diff
```
{diff[:8000]}
```

## Test output
```
{test_output[-3000:]}
```

Write a concise code review. Focus on:
1. **Bugs** — logic errors, edge cases, null/undefined risks
2. **Security** — injection, XSS, secrets in code, insecure defaults
3. **Performance** — N+1 queries, unnecessary re-renders, missing indexes
4. **Test coverage** — are the new code paths tested?
5. **Style** — only if inconsistent with the surrounding code

For each finding, cite the specific file and line. Use this format:

### [category emoji] Finding title
**File:** `path/to/file.ts:42`
[1-2 sentence description of the issue and suggested fix]

If the code looks good, say so. Don't fabricate issues.
End with a one-line overall assessment: APPROVE, REQUEST_CHANGES, or COMMENT.""",
        }],
    )

    return response.content[0].text

async def visual_review(
    client,
    sandbox_id: str,
    repo: str,
    branch: str,
    base_branch: str,
) -> str:
    """
    Boot the app, screenshot routes on base and PR branch,
    compare visually.
    """
    routes_to_check = ["/", "/dashboard", "/settings", "/pricing"]

    # Screenshot base branch
    await client.post(
        f"{SANDBOX_URL}/api/sandboxes/{sandbox_id}/execute",
        headers=HEADERS,
        json={"command": f"cd /workspace/project && git checkout {base_branch} && npm run build && npm start &"},
    )
    await asyncio.sleep(10)  # Wait for dev server

    base_screenshots = await capture_screenshots(
        client, routes_to_check, sandbox_id, prefix="base")

    # Kill dev server, switch to PR branch, rebuild
    await client.post(
        f"{SANDBOX_URL}/api/sandboxes/{sandbox_id}/execute",
        headers=HEADERS,
        json={"command": f"pkill -f 'npm start'; cd /workspace/project && git checkout {branch} && npm run build && npm start &"},
    )
    await asyncio.sleep(10)

    pr_screenshots = await capture_screenshots(
        client, routes_to_check, sandbox_id, prefix="pr")

    # Compare screenshots with Claude vision
    diffs = await compare_screenshots(base_screenshots, pr_screenshots)

    return format_visual_report(diffs)


async def capture_screenshots(
    client,
    routes: list,
    sandbox_id: str,
    prefix: str,
) -> dict:
    """Create a browser container, navigate to each route, screenshot."""
    browser_data = (await client.post(
        f"{SANDBOX_URL}/api/browsers", headers=HEADERS,
        json={"browser_type": "chrome", "enable_debug_port": True},
    )).json()
    container_id = browser_data["container_id"]

    # Wait for browser
    for _ in range(40):
        status = (await client.get(
            f"{SANDBOX_URL}/api/containers/{container_id}/status",
            headers=HEADERS)).json()
        if status["state"] == "assigned": break
        await asyncio.sleep(3)

    cdp_url = status["cdp_url"]
    screenshots = {}

    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(cdp_url)
        page = await browser.contexts[0].new_page()
        await page.set_viewport_size({"width": 1280, "height": 720})

        # The dev server runs on the sandbox — use its internal IP
        sandbox_ip = (await client.get(
            f"{SANDBOX_URL}/api/sandboxes/{sandbox_id}",
            headers=HEADERS)).json()["instance_ip"]

        for route in routes:
            url = f"http://{sandbox_ip}:3000{route}"
            try:
                await page.goto(url, wait_until="networkidle", timeout=15000)
                await page.wait_for_timeout(1000)
                screenshots[route] = await page.screenshot()
            except Exception as e:
                screenshots[route] = None

    # Release browser back to pool
    await client.post(
        f"{SANDBOX_URL}/api/containers/{container_id}/release",
        headers=HEADERS)

    return screenshots

async def compare_screenshots(base: dict, pr: dict) -> list:
    """Compare base and PR screenshots using Claude vision."""
    client = anthropic.Anthropic()
    diffs = []

    for route in base:
        if not base[route] or not pr.get(route):
            diffs.append({"route": route, "status": "error", "note": "Screenshot missing"})
            continue

        b64_base = base64.b64encode(base[route]).decode()
        b64_pr = base64.b64encode(pr[route]).decode()

        response = client.messages.create(
            model="claude-sonnet-4-6-20250514",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": f"Compare these two screenshots of the '{route}' route. The first is the base branch, the second is the PR branch. Describe any visual differences. If they look identical, say 'No visual changes.' Be concise."},
                    {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": b64_base}},
                    {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": b64_pr}},
                ],
            }],
        )

        diff_text = response.content[0].text
        diffs.append({
            "route": route,
            "status": "changed" if "no visual changes" not in diff_text.lower() else "unchanged",
            "description": diff_text,
        })

    return diffs

def format_pr_comment(
    test_passed: bool,
    test_output: str,
    review_notes: str,
    visual_report: str,
    changed_files: list,
) -> str:
    status_icon = "✅" if test_passed else "❌"
    status_text = "All tests passed" if test_passed else "Tests failed"

    comment = f"""## AI Code Review

{status_icon} **Tests: {status_text}**


Test output (click to expand)

```
{test_output[-5000:]}
```



### Files changed ({len(changed_files)})
{chr(10).join(f'- `{f}`' for f in changed_files[:20])}

---

### Code review

{review_notes}
"""

    if visual_report:
        comment += f"""
---

### Visual review

{visual_report}
"""

    comment += """
---
Reviewed by AI Code Reviewer on Sandbox Platform. Compute cost: ~$0.05."""

    return comment


async def post_pr_comment(repo: str, pr_number: int, comment: str):
    """Post a comment on the GitHub PR."""
    async with httpx.AsyncClient() as client:
        await client.post(
            f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments",
            headers={
                "Authorization": f"Bearer {GITHUB_TOKEN}",
                "Accept": "application/vnd.github+json",
            },
            json={"body": comment},
        )

# Clone, checkout, and ask Claude Code to review
cd /workspace
git clone https://github.com/your-org/your-repo.git project
cd project
git checkout feature/new-auth-flow

# Run Claude Code with the review task
claude "Review this branch against main.

1. Run the full test suite and report results.
2. Read the diff (git diff main...HEAD) and write a code review.
3. Focus on bugs, security issues, and missing test coverage.
4. Save your review to /workspace/review.md.
5. If there are frontend changes, start the dev server and
   use the Sandbox Platform API to create a Chrome browser container,
   screenshot the main routes, and note any visual changes."

# Determine affected packages from changed files
def affected_packages(changed_files: list) -> set:
    packages = set()
    for f in changed_files:
        parts = f.split("/")
        if len(parts) >= 2 and parts[0] == "packages":
            packages.add(parts[1])
    return packages

# Run tests only for affected packages
packages = affected_packages(changed_files)
for pkg in packages:
    test_cmd = f"cd /workspace/project/packages/{pkg} && npm test"
    result = await execute(sandbox_id, test_cmd)
    # Collect results per package