GEWU/scripts/build_benchmark_manifest.py

#!/usr/bin/env python3
"""Create a reproducibility manifest for a clean, pinned GEWU benchmark run."""

from __future__ import annotations

import argparse
import hashlib
import json
import platform
import subprocess
from datetime import UTC, datetime
from pathlib import Path
from typing import Any


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        while chunk := stream.read(1024 * 1024):
            digest.update(chunk)
    return digest.hexdigest()


def _git(repo: Path, *args: str) -> str:
    result = subprocess.run(
        ["git", "-C", str(repo), *args],
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()


def _pack_digest(pack: Path) -> str:
    digest = hashlib.sha256()
    for path in sorted(item for item in pack.rglob("*") if item.is_file()):
        relative = path.relative_to(pack).as_posix()
        digest.update(relative.encode("utf-8"))
        digest.update(b"\0")
        digest.update(bytes.fromhex(sha256(path)))
    return digest.hexdigest()


def _extractions(directory: Path) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    for path in sorted(directory.rglob("*.json")):
        value = json.loads(path.read_text(encoding="utf-8"))
        record = value.get("record", value)
        required = ("extraction_id", "parser_name", "parser_version", "parser_config_sha256")
        if not all(isinstance(record.get(key), str) for key in required):
            raise ValueError(f"Extraction manifest is missing parser identity fields: {path}")
        rows.append({key: record[key] for key in required})
    if not rows:
        raise ValueError("No extraction records found")
    return rows


def create_manifest(
    *,
    repo: Path,
    pack: Path,
    sources: list[Path],
    reference: Path,
    extractions: Path,
    model_id: str,
    search_mode: str,
    budget_tokens: int,
) -> dict[str, Any]:
    dirty = _git(repo, "status", "--porcelain", "--untracked-files=all")
    if dirty:
        raise ValueError("Backend checkout is dirty; use a clean isolated checkout")
    if not model_id.strip() or model_id.strip().lower() in {"default", "unknown", "latest"}:
        raise ValueError("A concrete model ID is required; aliases are not reproducible pins")
    if not 256 <= budget_tokens <= 50_000:
        raise ValueError("budget_tokens must be between 256 and 50000")
    for path in [repo, pack, reference, extractions, *sources]:
        if not path.exists():
            raise ValueError(f"Required benchmark input does not exist: {path}")
    manifest_path = pack / "manifest.json"
    pack_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    return {
        "schema_version": 1,
        "created_utc": datetime.now(UTC).isoformat(),
        "backend": {
            "git_commit": _git(repo, "rev-parse", "HEAD"),
            "clean_worktree": True,
        },
        "agent_pack": {
            "pack_id": pack_manifest["pack_id"],
            "pack_version": pack_manifest["pack_version"],
            "manifest_sha256": sha256(manifest_path),
            "tree_sha256": _pack_digest(pack),
        },
        "model_id": model_id.strip(),
        "search_mode": search_mode,
        "budget_tokens": budget_tokens,
        "reference_sha256": sha256(reference),
        "sources": [
            {"filename": path.name, "sha256": sha256(path), "bytes": path.stat().st_size}
            for path in sorted(sources)
        ],
        "extractions": _extractions(extractions),
        "runtime": {"python": platform.python_version(), "platform": platform.platform()},
    }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--pack", type=Path, required=True)
    parser.add_argument("--source", type=Path, action="append", required=True)
    parser.add_argument("--reference", type=Path, required=True)
    parser.add_argument("--extractions", type=Path, required=True)
    parser.add_argument("--model-id", required=True)
    parser.add_argument("--search-mode", required=True)
    parser.add_argument("--budget-tokens", type=int, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    manifest = create_manifest(
        repo=args.repo.resolve(),
        pack=args.pack.resolve(),
        sources=[path.resolve() for path in args.source],
        reference=args.reference.resolve(),
        extractions=args.extractions.resolve(),
        model_id=args.model_id,
        search_mode=args.search_mode,
        budget_tokens=args.budget_tokens,
    )
    with args.output.open("x", encoding="utf-8") as output:
        output.write(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n")
    print(json.dumps({"status": "created", "output": str(args.output)}))


if __name__ == "__main__":
    main()