eval/tools/run_eval.py

"""Tool-output eval runner (PR-2 Layer 2).

Runs every case in `cases.py` — each builds a fixture with a property planted and
asserts the tool reports it — then prints a per-tool table + aggregate summary and
writes a timestamped JSON report under `results/` (never overwritten).

**No LLM, no DB, no network.** The only compute is the tools themselves (which may
fit sklearn/statsmodels — fast, local, deterministic). This is the shared, committed
version of the fixtures that otherwise live gitignored under `tests/`.

Invoke as a module so `src` imports resolve:

    PYTHONPATH=. python -m eval.tools.run_eval
    PYTHONPATH=. python -m eval.tools.run_eval --tool analyze_cluster
"""

from __future__ import annotations

import argparse
import asyncio
import json
import time
import traceback
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path

from eval.tools.cases import CASES, EvalCase
from src.tools.registry import ACTIVE_ANALYTICS_TOOLS

_HERE = Path(__file__).resolve().parent
RESULTS_DIR = _HERE / "results"


@dataclass
class CaseResult:
    id: str
    tool: str
    prop: str
    passed: bool
    latency_ms: float
    error: str | None


async def run_case(case: EvalCase) -> CaseResult:
    start = time.perf_counter()
    error: str | None = None
    try:
        await case.fn()
        passed = True
    except Exception as exc:  # noqa: BLE001 — a failed assertion is a failed case, recorded
        passed = False
        error = f"{type(exc).__name__}: {exc}".strip()
    latency_ms = round((time.perf_counter() - start) * 1000, 1)
    return CaseResult(case.id, case.tool, case.prop, passed, latency_ms, error)


def _coverage() -> dict[str, list[str]]:
    """Which ACTIVE analytics tools have >= 1 case, and which have none (§4 acceptance)."""
    covered = {c.tool for c in CASES}
    active = {t.name for t in ACTIVE_ANALYTICS_TOOLS}
    return {
        "covered": sorted(active & covered),
        "uncovered": sorted(active - covered),
    }


def format_table(results: list[CaseResult]) -> str:
    header = f"{'ID':<38} {'TOOL':<22} {'ms':>6}  OK"
    rule = "-" * len(header)
    lines = [rule, header, rule]
    for r in results:
        ok = "ok" if r.passed else "X"
        lines.append(f"{r.id:<38} {r.tool:<22} {r.latency_ms:>6.1f}  {ok}")
        if not r.passed:
            lines.append(f"    -> {r.error}")
    lines.append(rule)
    return "\n".join(lines)


def format_summary(results: list[CaseResult], coverage: dict[str, list[str]]) -> str:
    passed = sum(r.passed for r in results)
    total = len(results)
    lines = ["SUMMARY", f"  {passed}/{total} cases passed"]

    by_tool: dict[str, list[CaseResult]] = {}
    for r in results:
        by_tool.setdefault(r.tool, []).append(r)
    lines.append("")
    lines.append("  By tool")
    for tool in sorted(by_tool):
        sub = by_tool[tool]
        lines.append(f"    {tool:<24} {sum(s.passed for s in sub)}/{len(sub)}")

    lines.append("")
    lines.append(
        f"  Coverage: {len(coverage['covered'])}/"
        f"{len(coverage['covered']) + len(coverage['uncovered'])} ACTIVE analytics tools"
    )
    if coverage["uncovered"]:
        lines.append(f"    UNCOVERED: {', '.join(coverage['uncovered'])}")

    failures = [r for r in results if not r.passed]
    if failures:
        lines.append("")
        lines.append(f"  FAILURES ({len(failures)})")
        for r in failures:
            lines.append(f"    {r.id:<38} {r.error}")
    return "\n".join(lines)


async def main() -> int:
    parser = argparse.ArgumentParser(description="Tool-output eval (PR-2 Layer 2)")
    parser.add_argument("--tool", default="", help="run only cases for this tool")
    parser.add_argument("--no-table", action="store_true", help="skip the detail table")
    args = parser.parse_args()

    cases = [c for c in CASES if not args.tool or c.tool == args.tool]

    started = datetime.now()
    print(f"Tool-Output Eval -- {started:%Y-%m-%d %H:%M:%S}")
    print(f"cases: {len(cases)}  (no LLM, no DB)")

    results = [await run_case(c) for c in cases]
    coverage = _coverage()

    if not args.no_table:
        print(format_table(results))
    print(format_summary(results, coverage))

    report = {
        "run": {
            "timestamp": started.isoformat(timespec="seconds"),
            "total": len(results),
            "passed": sum(r.passed for r in results),
            "runtime_ms": round(sum(r.latency_ms for r in results), 1),
            "llm_calls": 0,
        },
        "coverage": coverage,
        "cases": [asdict(r) for r in results],
    }
    RESULTS_DIR.mkdir(parents=True, exist_ok=True)
    out_path = RESULTS_DIR / f"tools_result_{started:%Y-%m-%d_%H%M%S}.json"
    out_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"\n-> saved: {out_path.relative_to(_HERE.parent.parent)}")

    # Non-zero exit on any failure so CI can gate on it.
    return 0 if all(r.passed for r in results) else 1


if __name__ == "__main__":
    try:
        raise SystemExit(asyncio.run(main()))
    except KeyboardInterrupt:
        traceback.print_exc()
        raise SystemExit(130) from None