eval_compare.py

#!/usr/bin/env python3
"""
Evaluate agentic reviewer results against ground truth.

File-level matching:
  - TP: We commented on a file that ground truth also commented on
  - FP: We commented on a file that ground truth did NOT comment on
  - FN: Ground truth commented on a file that we did NOT comment on

Usage:
    python eval_compare.py \
        --results-dir bench_new \
        --ground-truth-dir ground_truth \
        [--baseline-dir bench_latest] \
        [--verbose]
"""

import argparse
import json
import os
import sys
from collections import defaultdict
from pathlib import Path


def load_ground_truth(gt_dir: str) -> dict[tuple[str, int], set[str]]:
    """Load ground truth: {(repo, pr_number): set of file paths with comments}."""
    gt = {}
    for tool_dir in Path(gt_dir).iterdir():
        if not tool_dir.is_dir():
            continue
        for gt_file in sorted(tool_dir.glob("*.json")):
            with open(gt_file) as f:
                data = json.load(f)
            repo = data["repo"]
            for pr in data["prs"]:
                pr_num = pr["pr_number"]
                files = {c["path"] for c in pr.get("review_comments", []) if c.get("path")}
                if files:
                    gt[(repo, pr_num)] = files
    return gt


def load_results(results_dir: str) -> dict[tuple[str, int], dict]:
    """Load reviewer results: {(repo, pr_number): result_dict}."""
    results = {}
    for f in sorted(Path(results_dir).glob("*.json")):
        with open(f) as fh:
            data = json.load(fh)
        key = (data["repo_name"], data["pr_number"])
        results[key] = data
    return results


def get_result_files(result: dict) -> set[str]:
    """Extract set of file paths from reviewer result comments."""
    files = set()
    for c in result.get("comments", []):
        path = c.get("file_path") or c.get("path") or c.get("file")
        if path:
            files.add(path)
    return files


def compute_metrics(tp: int, fp: int, fn: int) -> dict:
    precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0
    return {"precision": precision, "recall": recall, "f1": f1, "tp": tp, "fp": fp, "fn": fn}


def evaluate(gt: dict, results: dict, verbose: bool = False) -> dict:
    total_tp, total_fp, total_fn = 0, 0, 0
    per_pr = []
    repo_stats = defaultdict(lambda: {"tp": 0, "fp": 0, "fn": 0, "prs": 0})

    for key in sorted(gt.keys()):
        repo, pr_num = key
        gt_files = gt[key]

        if key not in results:
            # Missing result = all FN
            fn = len(gt_files)
            total_fn += fn
            per_pr.append({"repo": repo, "pr": pr_num, "tp": 0, "fp": 0, "fn": fn,
                           "missed": gt_files, "extra": set()})
            repo_stats[repo]["fn"] += fn
            repo_stats[repo]["prs"] += 1
            continue

        result = results[key]
        if result.get("status") != "success":
            fn = len(gt_files)
            total_fn += fn
            per_pr.append({"repo": repo, "pr": pr_num, "tp": 0, "fp": 0, "fn": fn,
                           "missed": gt_files, "extra": set(), "error": result.get("error")})
            repo_stats[repo]["fn"] += fn
            repo_stats[repo]["prs"] += 1
            continue

        our_files = get_result_files(result)
        tp = len(gt_files & our_files)
        fp = len(our_files - gt_files)
        fn = len(gt_files - our_files)

        total_tp += tp
        total_fp += fp
        total_fn += fn

        repo_stats[repo]["tp"] += tp
        repo_stats[repo]["fp"] += fp
        repo_stats[repo]["fn"] += fn
        repo_stats[repo]["prs"] += 1

        per_pr.append({
            "repo": repo, "pr": pr_num,
            "tp": tp, "fp": fp, "fn": fn,
            "missed": gt_files - our_files,
            "extra": our_files - gt_files,
            "comments": result.get("comment_count", 0),
            "time": result.get("elapsed_seconds", 0),
        })

    aggregate = compute_metrics(total_tp, total_fp, total_fn)

    # Per-repo metrics
    per_repo = {}
    for repo, s in sorted(repo_stats.items()):
        per_repo[repo] = compute_metrics(s["tp"], s["fp"], s["fn"])
        per_repo[repo]["prs"] = s["prs"]

    return {
        "aggregate": aggregate,
        "per_repo": per_repo,
        "per_pr": per_pr,
        "total_prs": len(gt),
        "matched_prs": sum(1 for p in per_pr if p["tp"] + p["fp"] > 0),
    }


def print_results(eval_result: dict, verbose: bool = False, baseline: dict = None):
    agg = eval_result["aggregate"]

    print("\n" + "=" * 70)
    print("AGGREGATE RESULTS")
    print("=" * 70)
    print(f"  PRs evaluated:  {eval_result['total_prs']}")
    print(f"  F1:             {agg['f1']:.1%}")
    print(f"  Precision:      {agg['precision']:.1%}")
    print(f"  Recall:         {agg['recall']:.1%}")
    print(f"  TP: {agg['tp']}  FP: {agg['fp']}  FN: {agg['fn']}")

    if baseline:
        base_agg = baseline["aggregate"]
        print(f"\n  vs Baseline:")
        print(f"    F1:        {base_agg['f1']:.1%} -> {agg['f1']:.1%}  ({agg['f1'] - base_agg['f1']:+.1%})")
        print(f"    Precision: {base_agg['precision']:.1%} -> {agg['precision']:.1%}  ({agg['precision'] - base_agg['precision']:+.1%})")
        print(f"    Recall:    {base_agg['recall']:.1%} -> {agg['recall']:.1%}  ({agg['recall'] - base_agg['recall']:+.1%})")

    # Per-repo table
    print("\n" + "-" * 70)
    print(f"{'Repo':<35} {'PRs':>4} {'F1':>7} {'Prec':>7} {'Rec':>7} {'TP':>4} {'FP':>4} {'FN':>4}")
    print("-" * 70)
    for repo, m in sorted(eval_result["per_repo"].items()):
        print(f"{repo:<35} {m['prs']:>4} {m['f1']:>6.1%} {m['precision']:>6.1%} {m['recall']:>6.1%} {m['tp']:>4} {m['fp']:>4} {m['fn']:>4}")
    print("-" * 70)

    if verbose:
        print("\n" + "=" * 70)
        print("PER-PR DETAILS")
        print("=" * 70)
        for p in eval_result["per_pr"]:
            m = compute_metrics(p["tp"], p["fp"], p["fn"])
            status = "OK" if p["tp"] > 0 else "MISS"
            print(f"\n  {p['repo']}#{p['pr']}  [{status}]  F1={m['f1']:.0%}  TP={p['tp']} FP={p['fp']} FN={p['fn']}")
            if p.get("error"):
                print(f"    ERROR: {p['error']}")
            if p.get("missed"):
                for f in sorted(p["missed"]):
                    print(f"    - MISSED: {f}")
            if p.get("extra"):
                for f in sorted(p["extra"]):
                    print(f"    + EXTRA:  {f}")


def main():
    parser = argparse.ArgumentParser(description="Evaluate reviewer results against ground truth")
    parser.add_argument("--results-dir", required=True)
    parser.add_argument("--ground-truth-dir", required=True)
    parser.add_argument("--baseline-dir", help="Previous results dir for comparison")
    parser.add_argument("--verbose", action="store_true")
    args = parser.parse_args()

    gt = load_ground_truth(args.ground_truth_dir)
    print(f"Loaded ground truth: {len(gt)} PRs with comments")

    results = load_results(args.results_dir)
    print(f"Loaded results: {len(results)} PRs")

    eval_result = evaluate(gt, results, verbose=args.verbose)

    baseline = None
    if args.baseline_dir:
        baseline_results = load_results(args.baseline_dir)
        baseline = evaluate(gt, baseline_results)

    print_results(eval_result, verbose=args.verbose, baseline=baseline)


if __name__ == "__main__":
    main()