scripts/judge_crosswalk.py

#!/usr/bin/env python3
"""Build a judge-facing crosswalk from official criteria to ProofFrame evidence."""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
SCHEMA = "proofframe.judge_crosswalk.v1"
DEFAULT_JSON = ROOT / "docs" / "assets" / "judge-crosswalk.json"
DEFAULT_MD = ROOT / "docs" / "assets" / "judge-crosswalk.md"

SOURCE_SCHEMAS = {
    "docs/assets/devpost-event-snapshot.json": "proofframe.devpost_event_snapshot.v1",
    "docs/assets/judge-brief.json": "proofframe.judge_brief.v1",
    "docs/assets/sponsor-fit-audit.json": "proofframe.sponsor_fit_audit.v1",
    "docs/assets/award-readiness-report.json": "proofframe.award_readiness.v1",
    "docs/assets/final-submission-control.json": "proofframe.final_submission_control.v1",
}

REQUIRED_EVIDENCE_FILES = [
    "README.md",
    "docs/prd.md",
    "docs/spec.md",
    "docs/sponsor_fit_matrix.md",
    "docs/assets/devpost-form-kit.md",
    "docs/assets/devpost-submission-checklist.md",
    "docs/assets/public-video-check.md",
    "apps/web/index.html",
    "scripts/run_b2_live_proof.py",
    "scripts/run_final_live_proof.py",
    "scripts/submission_audit.py",
]

DEFAULT_CRITERIA = [
    "Real-world Utility",
    "Production Readiness",
    "B2 Storage + Data Orchestration",
    "Use of Genblaze",
]

SUBMISSION_REQUIREMENT_LABELS = {
    "working_app_url": "Working application URL",
    "github_repo_url": "GitHub repository URL",
    "demo_video": "Demonstration video",
    "video_under_three_minutes": "Video under three minutes",
    "public_video_host": "Public video host",
    "b2_usage": "Backblaze B2 usage",
    "genblaze_usage": "Genblaze usage",
}


def utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def load_json(path: Path) -> dict[str, Any] | None:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return None


def task_statuses(root: Path) -> dict[str, str]:
    data = load_json(root / "tasks.json") or {}
    return {
        str(task.get("id", "")).upper(): str(task.get("status", "missing"))
        for task in data.get("tasks", [])
    }


def source_record(root: Path, relative_path: str, expected_schema: str) -> dict[str, Any]:
    report = load_json(root / relative_path)
    actual_schema = report.get("schema") if report else None
    return {
        "path": relative_path,
        "present": report is not None,
        "expected_schema": expected_schema,
        "actual_schema": actual_schema,
        "ok": report is not None and actual_schema == expected_schema,
    }


def source_reports(root: Path) -> list[dict[str, Any]]:
    return [
        source_record(root, relative_path, schema)
        for relative_path, schema in SOURCE_SCHEMAS.items()
    ]


def evidence_file_records(root: Path) -> list[dict[str, Any]]:
    records = []
    for relative_path in REQUIRED_EVIDENCE_FILES:
        path = root / relative_path
        records.append({"path": relative_path, "present": path.exists()})
    return records


def official_criteria(event_snapshot: dict[str, Any] | None) -> list[dict[str, Any]]:
    criteria = (event_snapshot or {}).get("rules", {}).get("judging_criteria") or []
    if not criteria:
        return [{"name": name, "present": False} for name in DEFAULT_CRITERIA]
    return [
        {"name": str(item.get("name", "")), "present": bool(item.get("present"))}
        for item in criteria
    ]


def criterion_present(criteria: list[dict[str, Any]], name: str) -> bool:
    return any(item["name"] == name and item["present"] for item in criteria)


def blocking_items(final_control: dict[str, Any] | None) -> list[dict[str, str]]:
    return [
        {
            "id": str(item.get("id", "")),
            "label": str(item.get("label", "")),
            "detail": str(item.get("detail", "")),
            "evidence": str(item.get("evidence", "")),
        }
        for item in (final_control or {}).get("blocking_items", [])
    ]


def requirement_crosswalk(event_snapshot: dict[str, Any] | None) -> list[dict[str, Any]]:
    requirements = (event_snapshot or {}).get("rules", {}).get("requirements") or {}
    evidence = {
        "working_app_url": "Public mock demo URL in README and Devpost form kit",
        "github_repo_url": "Public GitHub repository and submission bundle manifest",
        "demo_video": "Public video check and storyboard reports; final URL still gated",
        "video_under_three_minutes": "Demo storyboard duration gate",
        "public_video_host": "Public video check URL rules",
        "b2_usage": "B2 adapter, B2 live setup record, and T020 live proof runner",
        "genblaze_usage": "Genblaze provider adapter and T021 final live proof runner",
    }
    return [
        {
            "id": requirement_id,
            "label": SUBMISSION_REQUIREMENT_LABELS.get(requirement_id, requirement_id),
            "official_marker_present": bool(present),
            "evidence": evidence.get(requirement_id, "Tracked in submission docs."),
        }
        for requirement_id, present in requirements.items()
    ]


def row(
    *,
    criterion_id: str,
    criterion: str,
    official_present: bool,
    evidence: list[str],
    safe_claim: str,
    final_gate: str,
    demo_shot: str,
    readiness: str,
) -> dict[str, Any]:
    return {
        "id": criterion_id,
        "criterion": criterion,
        "official_present": official_present,
        "readiness": readiness,
        "evidence": evidence,
        "safe_claim": safe_claim,
        "final_gate": final_gate,
        "demo_shot": demo_shot,
    }


def build_rows(criteria: list[dict[str, Any]], statuses: dict[str, str]) -> list[dict[str, Any]]:
    b2_done = statuses.get("T020") == "done"
    genblaze_done = statuses.get("T021") == "done"
    return [
        row(
            criterion_id="real_world_utility",
            criterion="Real-world Utility",
            official_present=criterion_present(criteria, "Real-world Utility"),
            readiness="public_mock_ready",
            evidence=[
                "docs/prd.md",
                "docs/spec.md",
                "apps/web/index.html",
                "docs/assets/judge-brief.md",
                "Downloadable evidence ZIP from /api/campaigns/{id}/packet.zip",
            ],
            safe_claim=(
                "ProofFrame is a working provenance and review desk for generated media, "
                "with the public demo running in credential-free local/mock mode."
            ),
            final_gate="Final video must show one creator workflow end-to-end with live proof context.",
            demo_shot="Judge Demo packet, asset review, approval state, manifest preview, evidence ZIP.",
        ),
        row(
            criterion_id="production_readiness",
            criterion="Production Readiness",
            official_present=criterion_present(criteria, "Production Readiness"),
            readiness="pre_live_fail_closed",
            evidence=[
                ".github/workflows/ci.yml",
                "scripts/secret_scan.py",
                "scripts/claim_lint.py",
                "scripts/final_submission_control.py",
                "scripts/submission_audit.py",
            ],
            safe_claim=(
                "The repo has fail-closed readiness gates, CI checks, secret scanning, "
                "claim linting, and generated submission reports."
            ),
            final_gate="T041A secret scan, T041 strict audit, and T042 receipt must close before submit.",
            demo_shot="Final gate dashboard and judge recording slate showing current safe-to-submit state.",
        ),
        row(
            criterion_id="b2_storage_data_orchestration",
            criterion="B2 Storage + Data Orchestration",
            official_present=criterion_present(criteria, "B2 Storage + Data Orchestration"),
            readiness="live_verified" if b2_done else "code_ready_live_proof_pending",
            evidence=[
                "src/proofframe/storage.py",
                "docs/assets/b2-live-setup.md",
                "scripts/run_b2_live_proof.py",
                "docs/assets/live-credential-handoff.md",
                "docs/assets/final-submission-control.md",
            ],
            safe_claim=(
                "ProofFrame has captured live B2 media and manifest evidence with sanitized object keys and checksums."
                if b2_done
                else "ProofFrame includes B2-compatible storage paths and B2-ready manifests; completed live B2 proof is not claimed until T020 is done."
            ),
            final_gate=(
                "B2 proof is done; final submission still needs public video, audit, scan, and Devpost receipt."
                if b2_done
                else "T020 must capture sanitized B2 asset and manifest object keys plus checksums."
            ),
            demo_shot="B2 Object Route and manifest checksum fields in the first-screen evidence model.",
        ),
        row(
            criterion_id="use_of_genblaze",
            criterion="Use of Genblaze",
            official_present=criterion_present(criteria, "Use of Genblaze"),
            readiness="live_verified" if genblaze_done else "adapter_ready_live_proof_pending",
            evidence=[
                "src/proofframe/providers.py",
                "scripts/run_final_live_proof.py",
                ".env.final.example",
                "docs/assets/live-credential-handoff.md",
                "docs/assets/devpost-submission-packet.md",
            ],
            safe_claim=(
                "ProofFrame has captured Genblaze Pipeline proof with provider/model metadata and B2-backed manifest output."
                if genblaze_done
                else "ProofFrame includes a Genblaze-compatible provider path and manifest fields; completed live Genblaze proof is not claimed until T021 is done."
            ),
            final_gate=(
                "Genblaze proof is done; final submission still needs public video, audit, scan, and Devpost receipt."
                if genblaze_done
                else "T021 must capture live provider/model metadata from the final Genblaze path."
            ),
            demo_shot="Genblaze Step, provider/model metadata, and prompt lineage in the manifest preview.",
        ),
    ]


def build_crosswalk(root: Path = ROOT) -> dict[str, Any]:
    root = root.resolve()
    reports = {
        relative_path: load_json(root / relative_path)
        for relative_path in SOURCE_SCHEMAS
    }
    event_snapshot = reports["docs/assets/devpost-event-snapshot.json"]
    final_control = reports["docs/assets/final-submission-control.json"]
    judge_brief = reports["docs/assets/judge-brief.json"]
    sponsor_audit = reports["docs/assets/sponsor-fit-audit.json"]
    award = reports["docs/assets/award-readiness-report.json"]
    statuses = task_statuses(root)
    sources = source_reports(root)
    evidence_files = evidence_file_records(root)
    criteria = official_criteria(event_snapshot)
    source_ok = all(item["ok"] for item in sources)
    evidence_ok = all(item["present"] for item in evidence_files)
    official_ok = bool((event_snapshot or {}).get("validation", {}).get("ok")) and all(
        item["present"] for item in criteria
    )
    safe_to_submit = bool((final_control or {}).get("safe_to_submit"))
    ok = source_ok and evidence_ok and official_ok
    return {
        "schema": SCHEMA,
        "created_at": utc_now(),
        "mode": (
            "final_crosswalk_ready"
            if ok and safe_to_submit
            else "pre_live_crosswalk_ready"
            if ok
            else "needs_crosswalk_sources"
        ),
        "ok": ok,
        "safe_to_submit": safe_to_submit,
        "event": {
            "name": (event_snapshot or {}).get("event", {}).get("name", "Backblaze Generative Media Hackathon"),
            "deadline_utc": (event_snapshot or {}).get("event", {}).get("deadline_utc"),
            "deadline_beijing": (event_snapshot or {}).get("event", {}).get("deadline_beijing"),
            "devpost_url": (event_snapshot or {}).get("event", {}).get("devpost_url"),
            "rules_url": (event_snapshot or {}).get("event", {}).get("rules_url"),
        },
        "source_reports": sources,
        "missing_evidence_files": [item["path"] for item in evidence_files if not item["present"]],
        "official_criteria": criteria,
        "submission_requirements": requirement_crosswalk(event_snapshot),
        "rows": build_rows(criteria, statuses),
        "summary": {
            "judge_brief_safe_to_submit": (judge_brief or {}).get("status", {}).get("safe_to_submit"),
            "sponsor_fit_mode": (sponsor_audit or {}).get("mode"),
            "sponsor_fit_ok": (sponsor_audit or {}).get("ok"),
            "award_score": (award or {}).get("score"),
            "award_max_score": (award or {}).get("max_score"),
            "final_control_mode": (final_control or {}).get("mode"),
        },
        "task_statuses": {
            task_id: statuses.get(task_id, "missing")
            for task_id in ["T020", "T021", "T040", "T041", "T041A", "T042"]
        },
        "blocking_items": blocking_items(final_control),
        "claim_boundaries": [
            (
                "B2 live storage proof may be claimed with sanitized evidence."
                if statuses.get("T020") == "done"
                else "Do not claim completed B2 live storage until T020 is done and evidence is sanitized."
            ),
            (
                "Genblaze Pipeline proof may be claimed with sanitized provider/model evidence."
                if statuses.get("T021") == "done"
                else "Do not claim completed Genblaze live generation until T021 is done and provider/model metadata is captured."
            ),
            "Do not mark safe_to_submit true until public video, strict audit, final secret scan, and Devpost receipt are complete.",
        ],
        "next_actions": [
            "Use this crosswalk as the judge-facing map while recording the final video and filling Devpost.",
            (
                "After the public video URL is ready, rerun final reports and regenerate this crosswalk."
                if statuses.get("T020") == "done" and statuses.get("T021") == "done"
                else "After T020/T021 and the public video URL are ready, rerun final reports and regenerate this crosswalk."
            ),
            "Submit only after final_submission_control.py --strict-final and submission_audit.py --strict-final pass.",
        ],
    }


def compact_list(items: list[str]) -> str:
    return "<br>".join(f"`{item}`" if item.startswith(("docs/", "scripts/", "src/", "apps/", ".github/")) else item for item in items)


def render_markdown(report: dict[str, Any]) -> str:
    lines = [
        "# ProofFrame Judge Crosswalk",
        "",
        f"Created: `{report['created_at']}`",
        f"Mode: `{report['mode']}`",
        f"OK: `{str(report['ok']).lower()}`",
        f"Safe to submit: `{str(report['safe_to_submit']).lower()}`",
        f"Event: {report['event']['name']}",
        f"Deadline: `{report['event']['deadline_utc']}` / `{report['event']['deadline_beijing']}`",
        "",
        "## Source Health",
        "",
    ]
    for source in report["source_reports"]:
        marker = "OK" if source["ok"] else "TODO"
        lines.append(
            f"- {marker} `{source['path']}` schema `{source['actual_schema']}` "
            f"(expected `{source['expected_schema']}`)"
        )
    if report["missing_evidence_files"]:
        lines.extend(["", "Missing evidence files:"])
        lines.extend(f"- `{path}`" for path in report["missing_evidence_files"])

    lines.extend(
        [
            "",
            "## Official Requirements",
            "",
            "| Requirement | Official Marker | Evidence |",
            "| --- | --- | --- |",
        ]
    )
    for requirement in report["submission_requirements"]:
        marker = "yes" if requirement["official_marker_present"] else "no"
        lines.append(f"| {requirement['label']} | `{marker}` | {requirement['evidence']} |")

    lines.extend(
        [
            "",
            "## Judging Crosswalk",
            "",
            "| Official angle | Current evidence | Safe claim | Final gate | Demo shot |",
            "| --- | --- | --- | --- | --- |",
        ]
    )
    for item in report["rows"]:
        criterion = f"{item['criterion']}<br>`{item['readiness']}`"
        evidence = compact_list(item["evidence"])
        lines.append(
            f"| {criterion} | {evidence} | {item['safe_claim']} | "
            f"{item['final_gate']} | {item['demo_shot']} |"
        )

    lines.extend(["", "## Blocking Items", ""])
    if report["blocking_items"]:
        for item in report["blocking_items"]:
            lines.append(f"- `{item['id']}`: {item['detail']} Evidence: `{item['evidence']}`")
    else:
        lines.append("- None.")

    lines.extend(["", "## Claim Boundaries", ""])
    lines.extend(f"- {item}" for item in report["claim_boundaries"])
    lines.extend(["", "## Next Actions", ""])
    lines.extend(f"- {item}" for item in report["next_actions"])
    return "\n".join(lines) + "\n"


def write_outputs(report: dict[str, Any], json_path: Path, markdown_path: Path) -> None:
    json_path.parent.mkdir(parents=True, exist_ok=True)
    markdown_path.parent.mkdir(parents=True, exist_ok=True)
    json_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    markdown_path.write_text(render_markdown(report), encoding="utf-8")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Build the ProofFrame judge evidence crosswalk.")
    parser.add_argument("--root", type=Path, default=ROOT)
    parser.add_argument("--json-out", type=Path, default=DEFAULT_JSON)
    parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MD)
    return parser


def main() -> None:
    args = build_parser().parse_args()
    report = build_crosswalk(args.root)
    write_outputs(report, args.json_out, args.markdown_out)
    print(
        json.dumps(
            {
                "ok": report["ok"],
                "mode": report["mode"],
                "safe_to_submit": report["safe_to_submit"],
                "json": str(args.json_out),
                "markdown": str(args.markdown_out),
                "blocking_items": len(report["blocking_items"]),
            },
            indent=2,
        )
    )
    raise SystemExit(0 if report["ok"] else 2)


if __name__ == "__main__":
    main()