whetstone_DSL/tools/mcp/summarize_ast_once_vs_direct.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any, Dict, List


def parse_args() -> argparse.Namespace:
    ap = argparse.ArgumentParser(description="Summarize AST-once vs direct benchmark results.")
    ap.add_argument("--results", required=True)
    ap.add_argument("--languages", required=True, help="Comma-separated language matrix")
    ap.add_argument("--out", required=True)
    return ap.parse_args()


def pct(n: int, d: int) -> float:
    if d <= 0:
        return 0.0
    return round((n / d) * 100.0, 2)


def load_rows(path: Path) -> List[Dict[str, Any]]:
    rows: List[Dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            rows.append(json.loads(line))
    return rows


def summarize(rows: List[Dict[str, Any]], n_langs: int) -> Dict[str, Any]:
    by_lang = defaultdict(lambda: {
        "total": 0,
        "runnable": 0,
        "infra_unavailable": 0,
        "direct_ready": 0,
        "ast_ready": 0,
        "direct_fix_applied": 0,
        "ast_fix_applied": 0,
        "direct_tokens_sum": 0,
        "ast_tokens_sum": 0,
        "ast_tokens_with_base_sum": 0,
        "ast_tokens_with_base_amortized_sum": 0.0,
    })
    by_category = defaultdict(lambda: {"total": 0, "direct_ready": 0, "ast_ready": 0})
    ast_failures = Counter()
    direct_failures = Counter()

    total = len(rows)
    direct_ready = 0
    ast_ready = 0
    direct_tokens_sum = 0
    ast_tokens_sum = 0
    ast_tokens_with_base_sum = 0
    ast_tokens_with_base_amortized_sum = 0.0

    for r in rows:
        lang = str(r.get("language_exec", "unknown"))
        cat = str(r.get("category", "unknown"))
        token = r.get("token_accounting", {}) or {}
        base_once = token.get("ast_base_once", {}) or {}
        direct = token.get("direct", {}) or {}
        ast = token.get("ast_path", {}) or {}

        base_total = int(base_once.get("total_tokens", 0) or 0)
        direct_total = int(direct.get("total_tokens", 0) or 0)
        ast_total = int(ast.get("total_tokens", 0) or 0)
        ast_with_base = ast_total + base_total
        ast_with_base_amortized = ast_total + (base_total / max(n_langs, 1))

        by_lang[lang]["total"] += 1
        by_lang[lang]["direct_tokens_sum"] += direct_total
        by_lang[lang]["ast_tokens_sum"] += ast_total
        by_lang[lang]["ast_tokens_with_base_sum"] += ast_with_base
        by_lang[lang]["ast_tokens_with_base_amortized_sum"] += ast_with_base_amortized

        by_category[cat]["total"] += 1
        direct_obj = (r.get("direct") or {})
        ast_obj = (r.get("ast_path") or {})
        direct_infra = bool(direct_obj.get("infra_unavailable"))
        ast_infra = bool(ast_obj.get("infra_unavailable"))
        infra_unavailable = direct_infra or ast_infra
        if infra_unavailable:
            by_lang[lang]["infra_unavailable"] += 1
        else:
            by_lang[lang]["runnable"] += 1

        if bool(direct_obj.get("fix_applied")):
            by_lang[lang]["direct_fix_applied"] += 1
        if bool(ast_obj.get("fix_applied")):
            by_lang[lang]["ast_fix_applied"] += 1

        direct_ok = bool(direct_obj.get("overall_ready"))
        ast_ok = bool(ast_obj.get("overall_ready"))
        if direct_ok:
            direct_ready += 1
            by_lang[lang]["direct_ready"] += 1
            by_category[cat]["direct_ready"] += 1
        if ast_ok:
            ast_ready += 1
            by_lang[lang]["ast_ready"] += 1
            by_category[cat]["ast_ready"] += 1

        direct_tokens_sum += direct_total
        ast_tokens_sum += ast_total
        ast_tokens_with_base_sum += ast_with_base
        ast_tokens_with_base_amortized_sum += ast_with_base_amortized

        for reason in (r.get("direct") or {}).get("failure_reasons") or []:
            direct_failures[str(reason)] += 1
        for reason in (r.get("ast_path") or {}).get("failure_reasons") or []:
            ast_failures[str(reason)] += 1

    by_lang_out: Dict[str, Any] = {}
    for lang, s in sorted(by_lang.items()):
        t = s["total"]
        by_lang_out[lang] = {
            "total": t,
            "runnable": s["runnable"],
            "infra_unavailable": s["infra_unavailable"],
            "direct_ready": s["direct_ready"],
            "direct_ready_rate_pct": pct(s["direct_ready"], t),
            "direct_ready_rate_runnable_pct": pct(s["direct_ready"], s["runnable"]),
            "direct_fix_applied": s["direct_fix_applied"],
            "direct_fix_applied_rate_pct": pct(s["direct_fix_applied"], t),
            "ast_ready": s["ast_ready"],
            "ast_ready_rate_pct": pct(s["ast_ready"], t),
            "ast_ready_rate_runnable_pct": pct(s["ast_ready"], s["runnable"]),
            "ast_fix_applied": s["ast_fix_applied"],
            "ast_fix_applied_rate_pct": pct(s["ast_fix_applied"], t),
            "avg_tokens_direct": round(s["direct_tokens_sum"] / t, 2) if t else 0.0,
            "avg_tokens_ast_path_only": round(s["ast_tokens_sum"] / t, 2) if t else 0.0,
            "avg_tokens_ast_with_full_base": round(s["ast_tokens_with_base_sum"] / t, 2) if t else 0.0,
            "avg_tokens_ast_with_amortized_base": round(s["ast_tokens_with_base_amortized_sum"] / t, 2) if t else 0.0,
        }

    by_category_out: Dict[str, Any] = {}
    for cat, s in sorted(by_category.items()):
        t = s["total"]
        by_category_out[cat] = {
            "total": t,
            "direct_ready": s["direct_ready"],
            "direct_ready_rate_pct": pct(s["direct_ready"], t),
            "ast_ready": s["ast_ready"],
            "ast_ready_rate_pct": pct(s["ast_ready"], t),
        }

    return {
        "total_runs": total,
        "direct": {
            "ready": direct_ready,
            "ready_rate_pct": pct(direct_ready, total),
            "avg_tokens": round(direct_tokens_sum / total, 2) if total else 0.0,
            "top_failure_reasons": direct_failures.most_common(15),
        },
        "ast_path": {
            "ready": ast_ready,
            "ready_rate_pct": pct(ast_ready, total),
            "avg_tokens_path_only": round(ast_tokens_sum / total, 2) if total else 0.0,
            "avg_tokens_with_full_base": round(ast_tokens_with_base_sum / total, 2) if total else 0.0,
            "avg_tokens_with_amortized_base": round(ast_tokens_with_base_amortized_sum / total, 2) if total else 0.0,
            "top_failure_reasons": ast_failures.most_common(15),
        },
        "by_language": by_lang_out,
        "by_category": by_category_out,
    }


def main() -> None:
    args = parse_args()
    rows = load_rows(Path(args.results))
    langs = [x for x in args.languages.split(",") if x.strip()]
    summary = summarize(rows, max(len(langs), 1))
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()