File size: 7,805 Bytes
813de1c 36ff02f 813de1c 36ff02f 813de1c 36ff02f 813de1c 0637c8b 813de1c 49f6940 813de1c 49f6940 813de1c 49f6940 813de1c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | #!/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"))
parser.add_argument(
"--no-markdown-report",
action="store_true",
help="Do not write report.md; persistent prose is consolidated in README.md.",
)
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")
_write_markdown_report(out_dir, report, no_markdown_report=args.no_markdown_report)
print(json.dumps({"status": report["status"], "violations": len(violations)}, indent=2))
return 1 if violations else 0
def _write_markdown_report(
out_dir: Path,
report: dict[str, Any],
*,
no_markdown_report: bool,
) -> None:
report_path = out_dir / "report.md"
if no_markdown_report:
report_path.unlink(missing_ok=True)
return
report_path.write_text(_markdown(report) + "\n")
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 audience != "evaluator_only":
violations.append(f"{path}: non-train split audience must be evaluator_only")
contract = payload.get("leakage_contract", {})
if not bool(contract.get("deployment_must_not_read_outcomes", False)):
violations.append(
f"{path}: non-train split must declare deployment_must_not_read_outcomes=true"
)
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 and not _sidecar_has_content(path.parent / "data_hash.txt"):
warnings.append(f"{path}: missing data_hash")
if "split_hash" not in payload and not _sidecar_has_content(path.parent / "split_hash.txt"):
warnings.append(f"{path}: missing split_hash")
def _sidecar_has_content(path: Path) -> bool:
return path.exists() and bool(path.read_text().strip())
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())
|