my-homelab-configs/scripts/ai-evals

#!/usr/bin/env python3
"""Deterministic eval harness for Jeannie and homelab RAG behavior."""

from __future__ import annotations

import argparse
import csv
import json
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
CASES_FILE = REPO_ROOT / "infra" / "ai-evals" / "cases.tsv"
FIXTURES_DIR = REPO_ROOT / "infra" / "ai-evals" / "fixtures"


@dataclass
class Case:
    case_id: str
    category: str
    mode: str
    fixture: str
    required_terms: list[str]
    forbidden_terms: list[str]
    notes: str


def split_terms(value: str) -> list[str]:
    return [item.strip() for item in value.split(";") if item.strip()]


def load_cases() -> list[Case]:
    with CASES_FILE.open(encoding="utf-8", newline="") as handle:
        rows = csv.DictReader(handle, delimiter="\t")
        return [
            Case(
                case_id=row["id"],
                category=row["category"],
                mode=row["mode"],
                fixture=row["fixture"],
                required_terms=split_terms(row["required_terms"]),
                forbidden_terms=split_terms(row["forbidden_terms"]),
                notes=row["notes"],
            )
            for row in rows
        ]


def fixture_text(case: Case) -> str:
    path = FIXTURES_DIR / case.fixture
    if not path.is_file():
        raise SystemExit(f"Missing fixture for {case.case_id}: {path}")
    return path.read_text(encoding="utf-8")


def run_explain_output(case: Case) -> str:
    process = subprocess.run(
        [str(REPO_ROOT / "scripts" / "explain"), "output"],
        input=fixture_text(case),
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        cwd=REPO_ROOT,
        check=False,
    )
    return process.stdout


def run_repo_scan(case: Case) -> str:
    haystack_parts: list[str] = [fixture_text(case), "\nRETRIEVED CONTEXT\n"]
    for path in [
        REPO_ROOT / "README.md",
        REPO_ROOT / "docs" / "jeannie.1.md",
        REPO_ROOT / "docs" / "runbooks" / "cluster-stop-start-failures.md",
        REPO_ROOT / "docs" / "runbooks" / "edge-failures.md",
        REPO_ROOT / "docs" / "runbooks" / "gitea-failures.md",
    ]:
        if path.is_file():
            rel = path.relative_to(REPO_ROOT)
            lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
            matches: list[str] = []
            for index, line in enumerate(lines):
                lowered = line.lower()
                if any(term.lower() in lowered for term in case.required_terms):
                    start = max(index - 1, 0)
                    end = min(index + 2, len(lines))
                    matches.extend(lines[start:end])
            if matches:
                haystack_parts.append(f"\nSOURCE {rel}\n")
                haystack_parts.extend(matches[:24])
    return "\n".join(haystack_parts)


def run_case(case: Case) -> dict[str, object]:
    if case.mode == "explain-output":
        output = run_explain_output(case)
    elif case.mode == "repo-scan":
        output = run_repo_scan(case)
    else:
        return {
            "id": case.case_id,
            "status": "fail",
            "reason": f"unknown mode {case.mode}",
            "missing": [],
            "forbidden_hits": [],
            "output": "",
        }

    lowered = output.lower()
    missing = [term for term in case.required_terms if term.lower() not in lowered]
    forbidden_hits = [term for term in case.forbidden_terms if term.lower() in lowered]
    status = "pass" if not missing and not forbidden_hits else "fail"
    reason = "ok" if status == "pass" else "term checks failed"
    return {
        "id": case.case_id,
        "category": case.category,
        "status": status,
        "reason": reason,
        "missing": missing,
        "forbidden_hits": forbidden_hits,
        "notes": case.notes,
        "output": output,
    }


def print_list(cases: list[Case]) -> int:
    for case in cases:
        print(f"{case.case_id}\t{case.category}\t{case.mode}\t{case.notes}")
    return 0


def print_show(cases: list[Case], case_id: str) -> int:
    for case in cases:
        if case.case_id == case_id:
            print(f"id: {case.case_id}")
            print(f"category: {case.category}")
            print(f"mode: {case.mode}")
            print(f"fixture: {case.fixture}")
            print(f"required: {', '.join(case.required_terms)}")
            print(f"forbidden: {', '.join(case.forbidden_terms)}")
            print(f"notes: {case.notes}")
            return 0
    print(f"Unknown eval case: {case_id}", file=sys.stderr)
    return 2


def print_run(cases: list[Case], case_id: str | None, as_json: bool, details: bool) -> int:
    selected = [case for case in cases if case_id in (None, case.case_id)]
    if not selected:
        print(f"Unknown eval case: {case_id}", file=sys.stderr)
        return 2

    results = [run_case(case) for case in selected]
    if as_json:
        print(json.dumps(results, indent=2, sort_keys=True))
    else:
        failures = sum(1 for result in results if result["status"] != "pass")
        print("Homelab AI Evals")
        print("================")
        print(f"cases={len(results)} failures={failures}")
        print()
        for result in results:
            print(f"{result['status']:5} {result['id']} - {result['reason']}")
            if result["missing"]:
                print(f"      missing: {', '.join(result['missing'])}")
            if result["forbidden_hits"]:
                print(f"      forbidden: {', '.join(result['forbidden_hits'])}")
            if details:
                print("      output:")
                for line in str(result["output"]).splitlines()[:40]:
                    print(f"        {line}")
    return 1 if any(result["status"] != "pass" for result in results) else 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("list")
    show_parser = subparsers.add_parser("show")
    show_parser.add_argument("case_id")
    run_parser = subparsers.add_parser("run")
    run_parser.add_argument("--case", dest="case_id")
    run_parser.add_argument("--json", action="store_true")
    run_parser.add_argument("--details", action="store_true")
    args = parser.parse_args()

    cases = load_cases()
    if args.command == "list":
        return print_list(cases)
    if args.command == "show":
        return print_show(cases, args.case_id)
    if args.command == "run":
        return print_run(cases, args.case_id, args.json, args.details)
    return 2


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