Add CTT paper artifact audit script

workspace/scripts/audit_ctt_paper_artifacts.py

@@ -0,0 +1,420 @@
+#!/usr/bin/env python
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+
+FORBIDDEN_PATTERNS: tuple[tuple[str, str, str], ...] = (
+    (
+        "submission_status_language",
+        r"\bA\*|\bQ1\b|Best Paper",
+        "Submission draft must not contain internal venue/readiness language.",
+    ),
+    (
+        "future_work_should",
+        r"future work should",
+        "Main method text must not defer the method to future work.",
+    ),
+    (
+        "generic_q_phi_generator",
+        r"q_phi|q_\s*\\phi|q_\s*\{\\phi\}|q\^\s*\\phi",
+        "The main generator must be CTT transport, not a generic q_phi noise model.",
+    ),
+    (
+        "distance_proxy_called_ptr",
+        r"(PTR[^.\n]{0,80}(distance|proxy))|((distance|proxy)[^.\n]{0,80}PTR)",
+        "Distance-only support diagnostics must be called PPTC, not PTR.",
+    ),
+)
+
+
+REQUIRED_PHRASES: tuple[tuple[str, str, str], ...] = (
+    (
+        "title_ctt",
+        "Causal Tangent Transport",
+        "Paper title/introduction names CTT as the central method.",
+    ),
+    (
+        "transport_operator",
+        "T_{\\phi}(z_s,z_t,\\xi_s^+)",
+        "Paper defines train-positive source-to-target transport.",
+    ),
+    (
+        "outcome_ptr",
+        "OutcomePTR",
+        "Paper names measured rollout positive-tangent recall separately.",
+    ),
+    (
+        "pptc",
+        "PPTC",
+        "Paper names distance-only Proxy Positive Tangent Coverage.",
+    ),
+    (
+        "support_gap",
+        "SupportGap",
+        "Paper exposes the support part of CAR.",
+    ),
+    (
+        "selector_gap",
+        "SelectorGap",
+        "Paper exposes the selector part of CAR.",
+    ),
+)
+
+
+REQUIRED_PATHS: tuple[tuple[str, str, str], ...] = (
+    ("metrics_module", "cil/metrics.py", "Canonical measured/proxy metrics."),
+    ("metrics_eval", "scripts/eval_metrics.py", "JSON/TeX metric export and proxy guards."),
+    ("metrics_tests", "tests/test_metrics.py", "Regression tests for metric separation."),
+    ("chart_export", "scripts/export_cil_charts.py", "Chart database export."),
+    ("chart_audit", "scripts/audit_cil_charts.py", "Chart leakage audit."),
+    ("data_accounting", "runs/data_accounting/table.json", "Scripted data accounting."),
+    ("data_accounting_table", "runs/data_accounting/table.tex", "Data accounting table."),
+    ("leakage_audit", "runs/leakage_audit/report.json", "Leakage audit artifact."),
+    ("ctt_model", "cil/models/ctt.py", "Causal Tangent Transport module."),
+    ("tangent_encoder", "cil/models/tangent_encoder.py", "Tangent-code encoder/decoder helpers."),
+    ("chart_encoder", "cil/models/chart_encoder.py", "Chart encoder."),
+    ("utility_energy", "cil/models/utility_energy.py", "Utility energy scorer."),
+    ("ctt_train", "scripts/train_ctt.py", "CTT training script."),
+    ("ctt_proxy_eval", "scripts/eval_ctt_proxy.py", "Proxy support evaluation."),
+    ("ctt_rollout_eval", "scripts/eval_ctt_rollout.py", "Measured rollout evaluation."),
+    ("utility_train", "scripts/train_utility_energy.py", "Utility energy training."),
+    ("dominance_calibration", "scripts/calibrate_dominance.py", "Calibrated dominance rule."),
+    ("theory_tex", "paper/sections/theory.tex", "Theory section included by paper."),
+    ("paper_pdf", "latex/main.pdf", "Compiled paper PDF."),
+)
+
+
+REQUIRED_RUN_FILES: tuple[str, ...] = (
+    "table.tex",
+    "metrics.json",
+    "command.txt",
+    "git_hash.txt",
+    "data_hash.txt",
+    "split_hash.txt",
+)
+
+
+ADVISOR_RUN_FILES: tuple[str, ...] = (
+    "config.yaml",
+    "train.log",
+    "eval.log",
+    "metrics_by_task.json",
+    "metrics_by_seed.json",
+    "report.md",
+)
+
+
+INPUT_RE = re.compile(r"\\input\{([^}]+)\}")
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Audit the CTT paper against the claim-to-artifact contract. "
+            "Outputs JSON and TeX so the repo can keep README.md as the only Markdown file."
+        )
+    )
+    parser.add_argument("--repo-root", type=Path, default=Path.cwd())
+    parser.add_argument("--paper", type=Path, default=Path("latex/main.tex"))
+    parser.add_argument("--out-dir", type=Path, default=Path("runs/paper_ctt_audit"))
+    parser.add_argument(
+        "--skip-implementation-checks",
+        action="store_true",
+        help="Only scan paper text and input artifacts; useful for isolated unit tests.",
+    )
+    args = parser.parse_args(argv)
+
+    repo_root = args.repo_root.resolve()
+    paper_path = _resolve(repo_root, args.paper)
+    if not paper_path.exists():
+        raise SystemExit(f"paper not found: {paper_path}")
+    paper_text = paper_path.read_text()
+
+    forbidden = _forbidden_findings(paper_text)
+    phrase_checks = _phrase_checks(paper_text)
+    paper_inputs = _paper_input_checks(repo_root, paper_path, paper_text)
+    required_paths = [] if args.skip_implementation_checks else _required_path_checks(repo_root)
+    run_artifacts = _run_artifact_checks(repo_root, paper_inputs["run_dirs"])
+    summary = _summary(forbidden, phrase_checks, paper_inputs, required_paths, run_artifacts)
+
+    payload = {
+        "schema_version": 1,
+        "audit_policy": {
+            "markdown_policy": "consolidated_readme_only",
+            "markdown_note": (
+                "Advisor checklist asks for report.md files, but the current workspace "
+                "policy keeps README.md as the only Markdown document. Missing report.md "
+                "is recorded as a warning, not regenerated here."
+            ),
+        },
+        "paper": str(paper_path.relative_to(repo_root)),
+        "summary": summary,
+        "forbidden_patterns": forbidden,
+        "required_phrases": phrase_checks,
+        "paper_inputs": paper_inputs,
+        "required_paths": required_paths,
+        "run_artifacts": run_artifacts,
+    }
+
+    out_dir = _resolve(repo_root, args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    (out_dir / "audit.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    (out_dir / "metrics.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    (out_dir / "table.tex").write_text(_latex_table(payload) + "\n")
+    (out_dir / "config.yaml").write_text(
+        "\n".join(
+            [
+                "paper: " + str(paper_path.relative_to(repo_root)),
+                "markdown_policy: consolidated_readme_only",
+                "strict_report_md: false",
+            ]
+        )
+        + "\n"
+    )
+    (out_dir / "command.txt").write_text(
+        "python scripts/audit_ctt_paper_artifacts.py " + " ".join(sys.argv[1:]) + "\n"
+    )
+    (out_dir / "git_hash.txt").write_text(_git_hash(repo_root) + "\n")
+    (out_dir / "data_hash.txt").write_text(_first_existing_hash(repo_root, "runs/data_accounting/table.json") + "\n")
+    (out_dir / "split_hash.txt").write_text(_split_hash(repo_root) + "\n")
+
+    print(json.dumps({"out_dir": str(out_dir), **summary}, indent=2, sort_keys=True))
+    return 1 if summary["status"] == "fail" else 0
+
+
+def _forbidden_findings(text: str) -> list[dict[str, Any]]:
+    findings: list[dict[str, Any]] = []
+    for name, pattern, detail in FORBIDDEN_PATTERNS:
+        for match in re.finditer(pattern, text, flags=re.IGNORECASE):
+            findings.append(
+                {
+                    "name": name,
+                    "status": "fail",
+                    "detail": detail,
+                    "match": match.group(0),
+                    "line": _line_for_offset(text, match.start()),
+                }
+            )
+    return findings
+
+
+def _phrase_checks(text: str) -> list[dict[str, Any]]:
+    checks = []
+    for name, phrase, detail in REQUIRED_PHRASES:
+        checks.append(
+            {
+                "name": name,
+                "status": "pass" if phrase in text else "fail",
+                "detail": detail,
+                "needle": phrase,
+            }
+        )
+    return checks
+
+
+def _paper_input_checks(repo_root: Path, paper_path: Path, text: str) -> dict[str, Any]:
+    rows = []
+    run_dirs: list[str] = []
+    for match in INPUT_RE.finditer(text):
+        raw = match.group(1)
+        resolved = _resolve_tex_input(paper_path.parent, raw)
+        status = "pass" if resolved.exists() else "fail"
+        row = {
+            "raw": raw,
+            "line": _line_for_offset(text, match.start()),
+            "resolved": _display_path(repo_root, resolved),
+            "status": status,
+        }
+        run_dir = _run_dir_for_input(repo_root, resolved)
+        if run_dir is not None:
+            row["run_dir"] = _display_path(repo_root, run_dir)
+            if row["run_dir"] not in run_dirs:
+                run_dirs.append(row["run_dir"])
+        rows.append(row)
+    return {
+        "num_inputs": len(rows),
+        "num_missing": sum(1 for row in rows if row["status"] != "pass"),
+        "rows": rows,
+        "run_dirs": sorted(run_dirs),
+    }
+
+
+def _required_path_checks(repo_root: Path) -> list[dict[str, Any]]:
+    checks = []
+    for name, rel_path, detail in REQUIRED_PATHS:
+        path = repo_root / rel_path
+        checks.append(
+            {
+                "name": name,
+                "path": rel_path,
+                "status": "pass" if path.exists() else "fail",
+                "detail": detail,
+            }
+        )
+    ctt_configs = sorted((repo_root / "configs/ctt").glob("*.yaml"))
+    checks.append(
+        {
+            "name": "ctt_configs",
+            "path": "configs/ctt/*.yaml",
+            "status": "pass" if ctt_configs else "fail",
+            "detail": "CTT loss weights and variants are config-driven.",
+            "files": [_display_path(repo_root, path) for path in ctt_configs],
+        }
+    )
+    return checks
+
+
+def _run_artifact_checks(repo_root: Path, run_dirs: list[str]) -> list[dict[str, Any]]:
+    rows = []
+    for rel_run_dir in sorted(run_dirs):
+        run_dir = repo_root / rel_run_dir
+        required = list(REQUIRED_RUN_FILES)
+        if rel_run_dir == "runs/data_accounting":
+            required = ["table.json", "table.tex"]
+        missing_required = [
+            name
+            for name in required
+            if not (run_dir / name).exists()
+            and not (rel_run_dir == "runs/paper_ctt_audit" and name == "metrics.json")
+        ]
+        missing_advisor = [name for name in ADVISOR_RUN_FILES if not (run_dir / name).exists()]
+        status = "pass" if not missing_required else "fail"
+        rows.append(
+            {
+                "run_dir": rel_run_dir,
+                "status": status,
+                "missing_required": missing_required,
+                "missing_advisor_contract": missing_advisor,
+                "markdown_report_policy": (
+                    "warning_only_consolidated_readme"
+                    if "report.md" in missing_advisor
+                    else "present"
+                ),
+            }
+        )
+    return rows
+
+
+def _summary(
+    forbidden: list[dict[str, Any]],
+    phrase_checks: list[dict[str, Any]],
+    paper_inputs: dict[str, Any],
+    required_paths: list[dict[str, Any]],
+    run_artifacts: list[dict[str, Any]],
+) -> dict[str, Any]:
+    failures = (
+        len(forbidden)
+        + paper_inputs["num_missing"]
+        + sum(1 for item in phrase_checks if item["status"] == "fail")
+        + sum(1 for item in required_paths if item["status"] == "fail")
+        + sum(1 for item in run_artifacts if item["status"] == "fail")
+    )
+    warnings = sum(len(item.get("missing_advisor_contract", [])) for item in run_artifacts)
+    return {
+        "status": "fail" if failures else "pass",
+        "num_failures": failures,
+        "num_warnings": warnings,
+        "num_forbidden_matches": len(forbidden),
+        "num_paper_inputs": paper_inputs["num_inputs"],
+        "num_run_dirs_in_paper": len(paper_inputs["run_dirs"]),
+    }
+
+
+def _latex_table(payload: dict[str, Any]) -> str:
+    summary = payload["summary"]
+    lines = [
+        "% Auto-generated by scripts/audit_ctt_paper_artifacts.py",
+        "\\begin{tabular}{lrr}",
+        "\\toprule",
+        "Audit item & Count & Status \\\\",
+        "\\midrule",
+        f"Forbidden language matches & {summary['num_forbidden_matches']} & {_latex_status(summary['num_forbidden_matches'] == 0)} \\\\",
+        f"Paper inputs & {summary['num_paper_inputs']} & {_latex_status(payload['paper_inputs']['num_missing'] == 0)} \\\\",
+        f"Run dirs in paper & {summary['num_run_dirs_in_paper']} & {_latex_status(all(row['status'] == 'pass' for row in payload['run_artifacts']))} \\\\",
+        f"Implementation paths & {len(payload['required_paths'])} & {_latex_status(all(row['status'] == 'pass' for row in payload['required_paths']))} \\\\",
+        f"Advisor-contract warnings & {summary['num_warnings']} & warn \\\\",
+        "\\bottomrule",
+        "\\end{tabular}",
+    ]
+    return "\n".join(lines)
+
+
+def _resolve(repo_root: Path, path: Path) -> Path:
+    return path if path.is_absolute() else repo_root / path
+
+
+def _resolve_tex_input(base_dir: Path, raw: str) -> Path:
+    candidate = (base_dir / raw).resolve()
+    if candidate.exists():
+        return candidate
+    if candidate.suffix:
+        return candidate
+    return candidate.with_suffix(".tex")
+
+
+def _run_dir_for_input(repo_root: Path, resolved: Path) -> Path | None:
+    try:
+        rel = resolved.relative_to(repo_root / "runs")
+    except ValueError:
+        return None
+    if not rel.parts:
+        return None
+    return repo_root / "runs" / rel.parts[0]
+
+
+def _display_path(repo_root: Path, path: Path) -> str:
+    try:
+        return str(path.relative_to(repo_root))
+    except ValueError:
+        return str(path)
+
+
+def _line_for_offset(text: str, offset: int) -> int:
+    return text.count("\n", 0, offset) + 1
+
+
+def _git_hash(repo_root: Path) -> str:
+    try:
+        return subprocess.check_output(
+            ["git", "rev-parse", "HEAD"],
+            cwd=repo_root,
+            text=True,
+            stderr=subprocess.DEVNULL,
+        ).strip()
+    except Exception:
+        return "unknown"
+
+
+def _first_existing_hash(repo_root: Path, rel_path: str) -> str:
+    path = repo_root / rel_path
+    if not path.exists():
+        return ""
+    import hashlib
+
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _split_hash(repo_root: Path) -> str:
+    path = repo_root / "runs/data_accounting/table.json"
+    if not path.exists():
+        return ""
+    try:
+        payload = json.loads(path.read_text())
+    except json.JSONDecodeError:
+        return ""
+    return str(payload.get("split_hash", ""))
+
+
+def _latex_status(ok: bool) -> str:
+    return "pass" if ok else "fail"
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())