evaluate.py

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any


# Matching helpers

def normalize_method_name(sig: str) -> str:
    """Reduce a method signature/symbol to its short, lower-cased method name."""
    if not sig:
        return ""
    base = sig.split("(", 1)[0]
    return base.split(".")[-1].strip().lower()


def _dedupe_preserve_order(items: list[str]) -> list[str]:
    seen: set[str] = set()
    out: list[str] = []
    for x in items:
        key = normalize_method_name(x)
        if not key or key in seen:
            continue
        seen.add(key)
        out.append(x)
    return out


# Per-approach prediction extractors

def extract_autofl_standard(data: dict) -> tuple[list[str], float | None]:
    """Ordered predictions + elapsed seconds for AutoFL standard mode."""
    preds: list[str] = []
    for loc in data.get("predicted_locations") or []:
        sig = loc.get("signature")
        if sig:
            preds.append(sig)
    if not preds:
        preds = list((data.get("buggy_methods") or {}).keys())
    # AutoFL writes an absolute epoch timestamp into ``time``; it is not a
    # duration, so a wall-clock cannot be derived from a single file.
    return _dedupe_preserve_order(preds), None


def extract_autofl_agent(data: dict) -> tuple[list[str], float | None]:
    """Ordered predictions + elapsed seconds for AutoFL agent mode."""
    preds: list[str] = []
    for key in ("final_prediction", "agent_suspects"):
        for sig in data.get(key) or []:
            if sig:
                preds.append(sig)
        if preds:
            break
    if not preds:
        for loc in data.get("predicted_locations") or []:
            sig = loc.get("signature")
            if sig:
                preds.append(sig)
    if not preds:
        preds = list((data.get("buggy_methods") or {}).keys())
    return _dedupe_preserve_order(preds), None


def extract_langgraph(data: dict) -> tuple[list[str], float | None]:
    """Ordered predictions + elapsed seconds for the LangGraph agent."""
    preds = [v.get("method") for v in (data.get("vulnerabilities") or []) if v.get("method")]
    if not preds:
        preds = list((data.get("buggy_methods") or {}).keys())
    return _dedupe_preserve_order(preds), data.get("elapsed_seconds")


_BP_AGENT_METHOD_RE = re.compile(r"method=['\"]([^'\"]+)['\"]")


def _bp_method_for(vuln: dict) -> str | None:
    """Prefer the agent's original method emission over pipeline auto-corrections."""
    note = vuln.get("method_normalization_note") or ""
    m = _BP_AGENT_METHOD_RE.search(note)
    if m:
        return m.group(1)
    return vuln.get("method")


def extract_beyond_patching(data: dict) -> tuple[list[str], float | None]:
    """Ordered predictions for Beyond Patching."""
    vulns = (data.get("detection_results") or {}).get("vulnerabilities")
    if vulns is None:
        vulns = data.get("vulnerabilities") or []
    preds = [m for m in (_bp_method_for(v) for v in vulns) if m]
    elapsed = data.get("elapsed_seconds")
    if elapsed is None:
        elapsed = data.get("detect_duration_seconds")
    return _dedupe_preserve_order(preds), elapsed


# Metric computation

def compute_metrics(predictions: list[str], ground_truth: list[str]) -> dict[str, Any]:
    gt_keys = {normalize_method_name(g) for g in ground_truth if g}
    pred_keys = [normalize_method_name(p) for p in predictions]

    hits = sum(1 for k in pred_keys if k in gt_keys)
    matched_gt = {k for k in pred_keys if k in gt_keys}
    matched_gt_names = [normalize_method_name(g) for g in ground_truth if normalize_method_name(g) in matched_gt]

    # Recall: fraction of GT methods that appear anywhere in the prediction list.
    recall = len(matched_gt) / len(gt_keys) if gt_keys else 0.0
    # Precision: fraction of predictions that hit a GT method.
    precision = hits / len(pred_keys) if pred_keys else 0.0
    # F1: harmonic mean; defined as 0 when both are 0.
    f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0
    # Acc@1: 1 iff the top-ranked prediction matches any GT method.
    acc1 = 1 if pred_keys and pred_keys[0] in gt_keys else 0

    return {
        "recall": round(recall, 4),
        "precision": round(precision, 4),
        "f1": round(f1, 4),
        "acc_at_1": acc1,
        "n_predictions": len(pred_keys),
        "predicted_methods": [normalize_method_name(p) for p in predictions],
        "gt_found": matched_gt_names,
        "n_gt_found": len(matched_gt),
        "n_gt_total": len(gt_keys),
    }


# Ground-truth loader

def load_ground_truth(path: Path) -> list[str]:
    data = json.loads(path.read_text(encoding="utf-8"))
    locs = data.get("vulnerability_locations") or []
    syms = [loc.get("symbol") for loc in locs if loc.get("symbol")]
    if not syms:
        raise ValueError(f"No 'vulnerability_locations[].symbol' entries in {path}")
    return syms


def _load_json(path: Path | None) -> dict | None:
    if path is None:
        return None
    if not path.exists():
        print(f"WARN: {path} not found - skipping", file=sys.stderr)
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        print(f"WARN: {path} is not valid JSON ({exc}) - skipping", file=sys.stderr)
        return None


# Reporting

APPROACHES: list[tuple[str, str, Any]] = [
    ("AutoFL Standard",   "autofl_standard",   extract_autofl_standard),
    ("AutoFL Agent Mode", "autofl_agent",      extract_autofl_agent),
    ("LangGraph",         "langgraph",         extract_langgraph),
    ("Beyond Patching",   "beyond_patching",   extract_beyond_patching),
]


def _fmt_time(t: float | None) -> str:
    return "N/A" if t is None else f"{t:.1f}"


def render_table(sample_name: str, gt: list[str], rows: list[dict]) -> str:
    lines: list[str] = []
    lines.append("=== Fault Localization Evaluation Results ===")
    lines.append(f"Sample: {sample_name}")
    lines.append(f"Ground truth methods: {len(gt)}")
    lines.append("")

    headers = ["Approach", "Ground Truth Methods", "Predicted Methods", "GT Found", "Found/Total",
               "Recall", "Precision", "F1", "Acc@1", "Predictions", "Time(s)"]
    gt_names = [normalize_method_name(g) for g in gt]
    gt_str = ", ".join(gt_names) if gt_names else "-"
    table: list[list[str]] = [headers]
    for r in rows:
        m = r["metrics"]
        preds = m.get("predicted_methods", []) or ["-"]
        found = m.get("gt_found", []) or ["-"]
        table.append([
            r["approach"],
            gt_str,
            ", ".join(preds),
            ", ".join(found),
            f"{m.get('n_gt_found', 0)}/{m.get('n_gt_total', len(gt))}",
            f"{m['recall']:.2f}",
            f"{m['precision']:.2f}",
            f"{m['f1']:.2f}",
            str(m["acc_at_1"]),
            str(m["n_predictions"]),
            _fmt_time(r["time_seconds"]),
        ])

    widths = [max(len(row[i]) for row in table) for i in range(len(headers))]
    for idx, row in enumerate(table):
        lines.append("  ".join(cell.ljust(widths[i]) for i, cell in enumerate(row)))
        if idx == 0:
            lines.append("  ".join("-" * w for w in widths))
    return "\n".join(lines)


# Entry point

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Evaluate fault-localization approaches against a ground-truth file.")
    parser.add_argument("--autofl-standard", type=Path, default=None)
    parser.add_argument("--autofl-agent",    type=Path, default=None)
    parser.add_argument("--langgraph",       type=Path, default=None)
    parser.add_argument("--beyond-patching", type=Path, default=None)
    parser.add_argument("--ground-truth",    type=Path, required=True)
    parser.add_argument("--out", type=Path, default=Path("evaluation_results.json"))
    parser.add_argument("--sample-name", default=None,
                        help="Label for the report (default: ground-truth parent dir)")
    args = parser.parse_args(argv)

    ground_truth = load_ground_truth(args.ground_truth)
    sample_name = args.sample_name or args.ground_truth.resolve().parent.parent.name

    inputs = {
        "autofl_standard":   args.autofl_standard,
        "autofl_agent":      args.autofl_agent,
        "langgraph":         args.langgraph,
        "beyond_patching":   args.beyond_patching,
    }

    rows: list[dict] = []
    for label, key, extractor in APPROACHES:
        path = inputs[key]
        data = _load_json(path)
        if data is None:
            rows.append({
                "approach": label,
                "key": key,
                "source_file": str(path) if path else None,
                "metrics": {
                    "recall": 0.0, "precision": 0.0, "f1": 0.0,
                    "acc_at_1": 0, "n_predictions": 0,
                    "predicted_methods": [], "gt_found": [],
                    "n_gt_found": 0, "n_gt_total": len(ground_truth),
                },
                "predictions": [],
                "time_seconds": None,
                "skipped": True,
            })
            continue

        preds, elapsed = extractor(data)
        rows.append({
            "approach": label,
            "key": key,
            "source_file": str(path),
            "metrics": compute_metrics(preds, ground_truth),
            "predictions": preds,
            "time_seconds": elapsed,
            "skipped": False,
        })

    table_text = render_table(sample_name, ground_truth, rows)
    print(table_text)

    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(
        json.dumps({
            "sample": sample_name,
            "ground_truth": ground_truth,
            "ground_truth_file": str(args.ground_truth),
            "approaches": rows,
        }, indent=2),
        encoding="utf-8",
    )
    table_path = args.out.with_suffix(".txt")
    table_path.write_text(table_text + "\n", encoding="utf-8")
    print(f"\nWrote {args.out}")
    print(f"Wrote {table_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())