gistfile1.txt

#!/usr/bin/env python3
"""
Multi-agent code review tool.

Runs 4 review agents in parallel (2 Claude, 2 Codex):
  - Claude high-level design reviewer
  - Codex high-level design reviewer
  - Claude low-level bug finder (real bugs only)
  - Codex low-level bug finder (real bugs only)

Then feeds all results through a final Claude synthesis pass that filters
and surfaces the most important findings.

Usage:
    python3 ~/code_review.py                     # reviews staged/unstaged diff
    python3 ~/code_review.py --diff-cmd "git diff main...HEAD"
    python3 ~/code_review.py --file path/to/file.py
    echo "<diff>" | python3 ~/code_review.py --stdin
"""

import argparse
import asyncio
import subprocess
import sys
import textwrap

HIGH_LEVEL_DESIGN_PROMPT = textwrap.dedent("""\
    You are a senior software architect performing a HIGH-LEVEL DESIGN review.
    Focus on:
    - Overall architecture and design patterns
    - API design and contracts
    - Separation of concerns and modularity
    - Naming, abstractions, and layering
    - Whether the approach is the right one for the problem
    - Missing error handling strategies (not individual checks, but systemic gaps)
    - Scalability and maintainability concerns

    Do NOT comment on formatting, style, typos, or low-level implementation details.
    Be concise. Only surface findings you are confident matter.

    Here is the code to review:

    {code}
""")

LOW_LEVEL_BUGS_PROMPT = textwrap.dedent("""\
    You are an expert bug hunter performing a LOW-LEVEL DETAIL review.
    Your job is to find ACTUAL BUGS — not hypothetical ones, not style issues,
    not "could be a problem if..." scenarios.

    Only report issues where you can point to a specific code path that will
    produce wrong behavior, crash, data corruption, security vulnerability,
    or resource leak under concrete conditions.

    For each bug, provide:
    1. The exact location (file + line/region)
    2. What goes wrong and under what conditions
    3. A concrete scenario or input that triggers it
    4. Suggested fix

    If you find no real bugs, say "No bugs found." Do not pad the review
    with speculative issues.

    Here is the code to review:

    {code}
""")

SYNTHESIS_PROMPT = textwrap.dedent("""\
    You are a lead engineer synthesizing feedback from four independent code
    reviewers. Two focused on high-level design, two on low-level bugs.

    Your job:
    1. Deduplicate — merge overlapping findings into single items.
    2. Filter — remove anything speculative, hypothetical, or low-signal.
       Keep only findings that would actually block or meaningfully improve a PR.
    3. Rank — order by severity (blocking issues first, then important, then nice-to-have).
    4. Attribute — note which reviewers agreed (consensus = higher confidence).
    5. Format — produce a clean, actionable review the author can work from.

    Output format:

    ## Critical Issues (must fix)
    - ...

    ## Important Suggestions (should fix)
    - ...

    ## Minor Notes (consider)
    - ...

    If a section has no items, omit it. If all reviewers found nothing
    significant, say so plainly.

    ---

    ### Claude Design Review:
    {claude_design}

    ### Codex Design Review:
    {codex_design}

    ### Claude Bug Review:
    {claude_bugs}

    ### Codex Bug Review:
    {codex_bugs}
""")


def get_diff(diff_cmd: str) -> str:
    result = subprocess.run(
        diff_cmd, shell=True, capture_output=True, text=True
    )
    if result.returncode != 0 and not result.stdout:
        print(f"Error running '{diff_cmd}': {result.stderr}", file=sys.stderr)
        sys.exit(1)
    return result.stdout


def read_files(paths: list[str]) -> str:
    chunks = []
    for path in paths:
        with open(path) as f:
            content = f.read()
        chunks.append(f"=== {path} ===\n{content}")
    return "\n\n".join(chunks)


async def run_claude(prompt: str, label: str) -> str:
    print(f"  Starting {label}...", file=sys.stderr)
    proc = await asyncio.create_subprocess_exec(
        "claude", "-p", "--bare",
        prompt,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    output = stdout.decode()
    if proc.returncode != 0:
        err = stderr.decode()
        print(f"  Warning: {label} exited {proc.returncode}: {err[:200]}", file=sys.stderr)
        return f"[{label} failed: {err[:500]}]"
    print(f"  Finished {label}.", file=sys.stderr)
    return output


async def run_codex(prompt: str, label: str) -> str:
    print(f"  Starting {label}...", file=sys.stderr)
    proc = await asyncio.create_subprocess_exec(
        "codex", "exec",
        "--full-auto",
        prompt,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    output = stdout.decode()
    if proc.returncode != 0:
        err = stderr.decode()
        print(f"  Warning: {label} exited {proc.returncode}: {err[:200]}", file=sys.stderr)
        return f"[{label} failed: {err[:500]}]"
    print(f"  Finished {label}.", file=sys.stderr)
    return output


async def run_all_agents(code: str) -> dict[str, str]:
    claude_design_prompt = HIGH_LEVEL_DESIGN_PROMPT.format(code=code)
    codex_design_prompt = HIGH_LEVEL_DESIGN_PROMPT.format(code=code)
    claude_bugs_prompt = LOW_LEVEL_BUGS_PROMPT.format(code=code)
    codex_bugs_prompt = LOW_LEVEL_BUGS_PROMPT.format(code=code)

    print("Running 4 review agents in parallel...", file=sys.stderr)

    claude_design, codex_design, claude_bugs, codex_bugs = await asyncio.gather(
        run_claude(claude_design_prompt, "Claude Design Review"),
        run_codex(codex_design_prompt, "Codex Design Review"),
        run_claude(claude_bugs_prompt, "Claude Bug Review"),
        run_codex(codex_bugs_prompt, "Codex Bug Review"),
    )

    return {
        "claude_design": claude_design,
        "codex_design": codex_design,
        "claude_bugs": claude_bugs,
        "codex_bugs": codex_bugs,
    }


async def synthesize(results: dict[str, str]) -> str:
    prompt = SYNTHESIS_PROMPT.format(**results)
    print("Running synthesis pass...", file=sys.stderr)
    output = await run_claude(prompt, "Synthesis")
    return output


async def main():
    parser = argparse.ArgumentParser(
        description="Multi-agent code review (2 Claude + 2 Codex + synthesis)"
    )
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "--diff-cmd",
        default=None,
        help='Shell command to produce a diff (default: "git diff HEAD")',
    )
    group.add_argument(
        "--file", "-f",
        nargs="+",
        dest="files",
        help="File(s) to review directly",
    )
    group.add_argument(
        "--stdin",
        action="store_true",
        help="Read code/diff from stdin",
    )
    parser.add_argument(
        "--raw",
        action="store_true",
        help="Print raw agent outputs before synthesis",
    )

    args = parser.parse_args()

    if args.stdin:
        code = sys.stdin.read()
    elif args.files:
        code = read_files(args.files)
    else:
        diff_cmd = args.diff_cmd or "git diff HEAD"
        code = get_diff(diff_cmd)

    if not code.strip():
        print("No code or diff to review.", file=sys.stderr)
        sys.exit(1)

    results = await run_all_agents(code)

    if args.raw:
        for label, text in results.items():
            print(f"\n{'='*60}")
            print(f"  {label}")
            print(f"{'='*60}\n")
            print(text)

    synthesis = await synthesize(results)

    print(synthesis)


if __name__ == "__main__":
    asyncio.run(main())