ctt artifacts 2026-07-02 workspace/scripts/audit_cil_charts.py

workspace/scripts/audit_cil_charts.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+
+REQUIRED_SPLITS = ("train", "val", "test")
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Audit CIL chart split indexes for leakage.")
+    parser.add_argument("--chart-root", type=Path, default=Path("data/cil_charts"))
+    parser.add_argument("--run-root", type=Path, default=Path("runs"))
+    parser.add_argument("--out-dir", type=Path, default=Path("runs/leakage_audit"))
+    args = parser.parse_args(argv)
+
+    indexes = _load_indexes(args.chart_root)
+    violations: list[str] = []
+    warnings: list[str] = []
+    for split in REQUIRED_SPLITS:
+        if split not in indexes:
+            violations.append(f"missing index for split {split}: {args.chart_root / split / 'index.json'}")
+    if not violations:
+        _audit_split_flags(indexes, violations)
+        _audit_overlap(indexes, violations)
+        _audit_shards(indexes, violations)
+        _audit_run_hashes(args.run_root, warnings)
+
+    report = {
+        "status": "pass" if not violations else "fail",
+        "chart_root": str(args.chart_root),
+        "indexes": {
+            split: {
+                "path": str(path),
+                "num_groups_exported": payload.get("num_groups_exported"),
+                "num_rows": payload.get("num_rows"),
+                "include_outcomes": payload.get("include_outcomes"),
+                "retrieval_index_allowed": payload.get("retrieval_index_allowed"),
+                "audience": payload.get("audience"),
+                "content_hash": payload.get("content_hash"),
+                "split_hash": payload.get("split_hash"),
+            }
+            for split, (path, payload) in sorted(indexes.items())
+        },
+        "violations": violations,
+        "warnings": warnings,
+    }
+    out_dir = args.out_dir
+    out_dir.mkdir(parents=True, exist_ok=True)
+    (out_dir / "report.json").write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
+    (out_dir / "report.md").write_text(_markdown(report) + "\n")
+    print(json.dumps({"status": report["status"], "violations": len(violations)}, indent=2))
+    return 1 if violations else 0
+
+
+def _load_indexes(chart_root: Path) -> dict[str, tuple[Path, dict[str, Any]]]:
+    output: dict[str, tuple[Path, dict[str, Any]]] = {}
+    for split in REQUIRED_SPLITS:
+        path = chart_root / split / "index.json"
+        if path.exists():
+            output[split] = (path, json.loads(path.read_text()))
+    return output
+
+
+def _audit_split_flags(indexes: dict[str, tuple[Path, dict[str, Any]]], violations: list[str]) -> None:
+    for split, (path, payload) in indexes.items():
+        include_outcomes = bool(payload.get("include_outcomes"))
+        retrieval = bool(payload.get("retrieval_index_allowed"))
+        audience = str(payload.get("audience"))
+        if split == "train":
+            if not retrieval:
+                violations.append(f"{path}: train split is not retrieval_index_allowed")
+            if not include_outcomes:
+                violations.append(f"{path}: train split should include outcomes for training")
+        else:
+            if retrieval:
+                violations.append(f"{path}: non-train split is retrieval_index_allowed")
+            if include_outcomes:
+                violations.append(f"{path}: non-train split exposes outcomes in chart DB")
+            if audience != "evaluator_only":
+                violations.append(f"{path}: non-train split audience must be evaluator_only")
+        if split != str(payload.get("split")):
+            violations.append(f"{path}: index split field does not match directory")
+        if not payload.get("deployment_candidate_excludes_expert", False):
+            violations.append(f"{path}: missing deployment_candidate_excludes_expert=true")
+
+
+def _audit_overlap(indexes: dict[str, tuple[Path, dict[str, Any]]], violations: list[str]) -> None:
+    for left in REQUIRED_SPLITS:
+        for right in REQUIRED_SPLITS:
+            if left >= right or left not in indexes or right not in indexes:
+                continue
+            left_payload = indexes[left][1]
+            right_payload = indexes[right][1]
+            group_overlap = set(left_payload.get("group_hashes", [])) & set(
+                right_payload.get("group_hashes", [])
+            )
+            state_overlap = set(left_payload.get("state_hashes", [])) & set(
+                right_payload.get("state_hashes", [])
+            )
+            if group_overlap:
+                violations.append(f"{left}/{right}: {len(group_overlap)} overlapping chart_ids")
+            if state_overlap:
+                violations.append(f"{left}/{right}: {len(state_overlap)} overlapping state_hashes")
+
+
+def _audit_shards(indexes: dict[str, tuple[Path, dict[str, Any]]], violations: list[str]) -> None:
+    for split, (path, payload) in indexes.items():
+        for shard in payload.get("shards", []):
+            shard_path = path.parent / str(shard.get("path", ""))
+            if not shard_path.exists():
+                violations.append(f"{path}: missing shard {shard_path}")
+
+
+def _audit_run_hashes(run_root: Path, warnings: list[str]) -> None:
+    metrics_files = sorted(run_root.glob("*/metrics.json"))
+    if not metrics_files:
+        warnings.append(f"{run_root}: no run metrics.json files found for hash audit")
+        return
+    for path in metrics_files:
+        try:
+            payload = json.loads(path.read_text())
+        except json.JSONDecodeError:
+            warnings.append(f"{path}: invalid JSON; skipped hash audit")
+            continue
+        if "data_hash" not in payload:
+            warnings.append(f"{path}: missing data_hash")
+        if "split_hash" not in payload:
+            warnings.append(f"{path}: missing split_hash")
+
+
+def _markdown(report: dict[str, Any]) -> str:
+    lines = [
+        "# CIL Chart Leakage Audit",
+        "",
+        f"Status: `{report['status']}`",
+        "",
+        "| Split | Rows | Charts | Outcomes in DB | Retrieval allowed | Audience |",
+        "| --- | ---: | ---: | --- | --- | --- |",
+    ]
+    for split, payload in sorted(report["indexes"].items()):
+        lines.append(
+            f"| {split} | {payload.get('num_rows')} | {payload.get('num_groups_exported')} | "
+            f"{payload.get('include_outcomes')} | {payload.get('retrieval_index_allowed')} | "
+            f"{payload.get('audience')} |"
+        )
+    lines.extend(["", "## Violations", ""])
+    lines.extend(f"- {item}" for item in report["violations"]) if report["violations"] else lines.append("- None")
+    lines.extend(["", "## Warnings", ""])
+    lines.extend(f"- {item}" for item in report["warnings"]) if report["warnings"] else lines.append("- None")
+    return "\n".join(lines)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())