reproduction-bundle/src/audit_upstream_release.py

#!/usr/bin/env python3
"""Fail-closed audit of materials released for the anchored empirical claims."""

from __future__ import annotations

import argparse
import json
import re
import subprocess
from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]
UPSTREAM = ROOT / "vendor" / "LogDiff"


def git_output(*args: str) -> str:
    return subprocess.check_output(["git", "-C", str(UPSTREAM), *args], text=True).strip()


def extract_scalar(path: Path, key: str) -> str:
    match = re.search(rf"^{re.escape(key)}:\s*(.+?)\s*$", path.read_text(), re.MULTILINE)
    if not match:
        raise ValueError(f"{key!r} not found in {path}")
    return match.group(1).strip("'\"")


def checkpoint_record(config: Path, key: str) -> dict[str, object]:
    value = extract_scalar(config, key)
    return {
        "config": str(config.relative_to(UPSTREAM)),
        "key": key,
        "declared_path": value,
        "present_in_release": (UPSTREAM / value).is_file(),
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--output",
        type=Path,
        default=ROOT / "outputs" / "upstream-audit" / "release_audit.json",
    )
    args = parser.parse_args()

    expected_commit = "94ef35bafd4b4239e9832d8295128c09e8fc1472"
    actual_commit = git_output("rev-parse", "HEAD")
    tracked_files = git_output("ls-files").splitlines()
    configs = [
        UPSTREAM / "configs" / "cmnist_inference.yaml",
        UPSTREAM / "configs" / "shapes3d_inference.yaml",
        UPSTREAM / "configs" / "celeba_inference.yaml",
    ]
    checkpoint_records = [
        checkpoint_record(config, key)
        for config in configs
        for key in (
            "checkpoint_path",
            "composition_classifier_checkpoint",
            "judge_classifier_checkpoint",
        )
    ]

    composition_config = UPSTREAM / "configs" / "cs_celeba_composition.yaml"
    default_match = re.search(r"- /dataset:\s*(\S+)", composition_config.read_text())
    if not default_match:
        raise ValueError("CelebA composition dataset default not found")
    requested_default = default_match.group(1)
    requested_name = requested_default if requested_default.endswith(".yaml") else requested_default + ".yaml"
    requested_path = UPSTREAM / "configs" / "dataset" / requested_name

    evaluator = (UPSTREAM / "logdiff" / "evaluate" / "evaluate_celeba.py").read_text()
    paper_protocol_path = ROOT / "repro" / "data" / "paper_protocol.json"
    paper_protocol = json.loads(paper_protocol_path.read_text())
    paper_mentions_clean_fid = paper_protocol["specified_fid_implementation"] == "clean-fid"
    eval_total_match = re.search(r"EVAL_TOTAL_SAMPLES\s*=\s*(\d+)", evaluator)
    eval_total = int(eval_total_match.group(1)) if eval_total_match else None
    recorded_expression_present = "EVAL_TOTAL_SAMPLES * BATCH_SIZE" in evaluator
    default_batch_size = 100

    molecular_markers = [
        path
        for path in tracked_files
        if re.search(r"molecul|protein|ligand|targetdiff|dualdiff|vina", path, re.IGNORECASE)
    ]
    report = {
        "paper": "OAM1jJsMGp",
        "release": {
            "repository": "TanjaBien/LogDiff",
            "expected_commit": expected_commit,
            "actual_commit": actual_commit,
            "commit_matches": actual_commit == expected_commit,
            "tracked_file_count": len(tracked_files),
        },
        "declared_checkpoints": checkpoint_records,
        "all_declared_checkpoints_missing": all(
            not item["present_in_release"] for item in checkpoint_records
        ),
        "celeba_composition_config": {
            "requested_dataset_default": requested_default,
            "resolved_expected_path": str(requested_path.relative_to(UPSTREAM)),
            "resolved_expected_path_exists": requested_path.is_file(),
            "nearby_copy_suffixed_file_exists": (
                UPSTREAM / "configs" / "dataset" / "celeba_cs_latents_male_haircolors copy.yaml"
            ).is_file(),
        },
        "celeba_fid_protocol": {
            "paper_protocol_record": str(paper_protocol_path.relative_to(ROOT)),
            "paper_specifies_clean_fid": paper_mentions_clean_fid,
            "released_code_imports_torchmetrics_fid": (
                "from torchmetrics.image.fid import FrechetInceptionDistance" in evaluator
            ),
            "released_code_imports_clean_fid": "clean_fid" in evaluator,
            "protocol_match": bool(
                paper_mentions_clean_fid
                and "clean_fid" in evaluator
                and "FrechetInceptionDistance" not in evaluator
            ),
        },
        "sample_accounting": {
            "actual_generation_target_per_task": eval_total,
            "default_batch_size": default_batch_size,
            "results_writer_uses_total_times_batch_size": recorded_expression_present,
            "recorded_samples_under_default": (
                eval_total * default_batch_size
                if eval_total and recorded_expression_present
                else None
            ),
            "accounting_matches": not recorded_expression_present,
        },
        "molecular_release": {
            "tracked_files_matching_molecular_domain": molecular_markers,
            "implementation_present_in_pinned_release": bool(molecular_markers),
        },
        "anchored_claim_3_reproduction_ready": False,
        "anchored_claim_4_reproduction_ready": False,
        "interpretation": {
            "claim_3": (
                "The table transcription can be audited, but the released package cannot reproduce the "
                "CelebA FID/CS values without absent checkpoints and a repaired config; the FID "
                "implementation also differs from the paper specification."
            ),
            "claim_4": (
                "The table transcription can be audited, but the pinned release contains no molecular "
                "implementation, targets, generated ligands, docking outputs, or checkpoints."
            ),
        },
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(report, indent=2) + "\n")
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()