Upload build/audit_contamination.py with huggingface_hub

build/audit_contamination.py

@@ -28,6 +28,7 @@
 # Read-only over the mix. CPU session. No credentials in this file.
 
 import argparse
+import hashlib
 import json
 import os
 import struct
@@ -93,6 +94,17 @@ def window_hashes(ids, k=K):
     return h
 
 
+def brief_error(e):
+    """Short, publishable description of a loader failure.
+
+    This report is shipped with the mix, and an exception string from a dataset reader can quote a cell of
+    benchmark text -- exactly the thing §3.3 keeps out of every artifact. Quoted spans are replaced.
+    """
+    import re
+    msg = re.sub(r"'[^']*'|\"[^\"]*\"", "'…'", str(e).replace("\n", " "))
+    return f"{type(e).__name__}: {msg[:140]}"
+
+
 def flatten(v, out):
     if isinstance(v, str):
         out.append(v)
@@ -212,8 +224,8 @@ def build_reference(tk, max_items_per_split):
                               else datasets.load_dataset(repo, split=sp, streaming=True))
                         n, g = ingest(tk, ds, max_items_per_split)
                     except Exception as e:
-                        errors[ckey] = f"{type(e).__name__}: {str(e)[:140]}"
-                        print(f"  ref {ckey}: UNREADABLE {type(e).__name__}: {str(e)[:80]}", flush=True)
+                        errors[ckey] = brief_error(e)
+                        print(f"  ref {ckey}: UNREADABLE {brief_error(e)}", flush=True)
                         continue
                 if want and n < int(0.9 * want):
                     print(f"  ref {ckey}: {n} of {want} rows via datasets -> rows-API fallback",
@@ -270,20 +282,23 @@ def per_task_counts(bits):
     return [int((bits & np.uint16(1 << ti)).astype(np.uint16).sum()) for ti in range(len(REFS))]
 
 
-def scan_docs(ref, refmask, docs):
+def scan_docs(ref, refmask, docs, seen=None):
     """Fold an iterator of (ordinal_within_source, raw_bytes) into counts.
 
     The ordinal is the only identity a document ever gets here: a filter pass can drop it by index later
     without anything in this process knowing what it said. `hit_docs_over_half` is the column that decides
     interpretation -- contamination by verbatim inclusion shows up as documents whose grams are *mostly*
     matched, while one shared boilerplate sentence shows up as a single gram in a handful of documents.
+    Pass one `seen` set across a whole mix, otherwise the duplicate count is per-shard and only a lower
+    bound (found in review).
     """
     import time
     import hashlib
     t0 = time.time()
     s = {"docs": 0, "tokens": 0, "grams": 0, "hit_docs": 0, "hits": 0, "hit_docs_over_half": 0,
          "hit_tokens": 0, "dup_docs": 0, "dup_tokens": 0, "per_task": [0] * len(REFS), "ordinals": []}
-    seen = set()
+    if seen is None:
+        seen = set()
     names = [t[0] for t in REFS]
     for idx, raw in docs:
         b = bytes(raw)
@@ -381,7 +396,9 @@ def main():
             json.dump(srv, open(os.path.join(a.ref_cache, "srv.json"), "w"))
             print(f"reference cached under {a.ref_cache}", flush=True)
     tasks_covered = sorted({k.split("::")[0] for k, v in refcounts.items()
-                            if v >= MIN_ITEMS_PER_SPLIT})
+                            if v >= MIN_ITEMS_PER_SPLIT} -
+                           {k.split("::")[0] for k in errors
+                            if "::" in k and "[" not in k and not k.startswith("stage::")})
     print(f"reference: {ref.size:,} grams, {len(tasks_covered)}/8 tasks covered, "
           f"{len(errors)} loader problems", flush=True)
 
@@ -397,7 +414,13 @@ def main():
         for key in sorted(os.listdir(stage_root)):
             sd = os.path.join(stage_root, key)
             rd = os.path.join(sd, "record.json")
-            if not os.path.isdir(sd) or not os.path.exists(rd):
+            if not os.path.isdir(sd):
+                continue
+            if not os.path.exists(rd):
+                # Absent record means the source never finished staging or was built without a Hub
+                # repo. Treating it as "nothing to scan" would merge it unfiltered (review finding).
+                errors[f"stage::{key}"] = "no record.json -- source is incomplete or unpublished"
+                print(f"  stage {key}: NO record.json, source left unscanned", flush=True)
                 continue
             rec = json.load(open(rd))
             paths = [os.path.join(a.root, s["file"]) for s in rec.get("shards", [])]
@@ -424,15 +447,25 @@ def main():
     docs = grams_seen = mix_tokens = hit_docs = hit_docs_over_half = 0
     dup_docs = dup_tokens = 0
     missing = []
+    val = {"docs": 0, "tokens": 0, "grams": 0, "hit_docs": 0, "hits": 0, "hit_docs_over_half": 0}
+    manifest_digest = None
     if os.path.exists(man_path):
         man = json.load(open(man_path))
+        # Bound this report to the exact bytes it scanned: publish_mix re-computes the digest from the
+        # manifest it is about to ship, so an audit.json carried over from a different mix -- even one of
+        # the same total size -- is refused instead of certifying it.
+        manifest_digest = hashlib.sha256(json.dumps(
+            [[s["file"], s.get("tokens"), s.get("docs")] for s in man["shards"]] +
+            [["VAL:" + s["file"], s.get("tokens"), s.get("docs")] for s in man.get("val_shards", [])],
+            sort_keys=True).encode()).hexdigest()
+        seen_all = set()      # one set across the whole mix, so duplicates are counted, not floored
         for rec in man["shards"]:
             p = os.path.join(a.root, rec["file"])
             if not os.path.exists(p):
                 # A shard not on disk is unmeasured, not clean (E-017's shape).
                 missing.append(rec["file"])
                 continue
-            s = scan_docs(ref, refmask, doc_stream(BM, [p]))
+            s = scan_docs(ref, refmask, doc_stream(BM, [p]), seen=seen_all)
             docs += s["docs"]
             mix_tokens += s["tokens"]
             grams_seen += s["grams"]
@@ -442,21 +475,35 @@ def main():
             dup_tokens += s["dup_tokens"]
             for n, c in s["per_task_named"].items():
                 hits[n] = hits.get(n, 0) + c
-        total_overlap = sum(hits.values())
-        print(f"  mix: {docs:,} docs, {grams_seen:,} grams, {total_overlap:,} hits in {hit_docs:,} docs "
-              f"({hit_docs_over_half:,} majority-hit)", flush=True)
+        # The held-out split is reported on, so it must be scanned too. A benchmark document sitting in
+        # validation would flatter every validation-PPL number this project publishes, and nothing else
+        # in the pipeline reads those bytes (review finding).
+        for rec in man.get("val_shards", []):
+            p = os.path.join(a.root, rec["file"])
+            if not os.path.exists(p):
+                missing.append(rec["file"])
+                continue
+            v = scan_docs(ref, refmask, doc_stream(BM, [p]), seen=seen_all)
+            for k2 in ("docs", "tokens", "grams", "hit_docs", "hits", "hit_docs_over_half"):
+                val[k2] += int(v.get(k2, 0))
+        total_overlap = sum(hits.values()) + val["hits"]
+        print(f"  mix: {docs:,} docs, {grams_seen:,} grams, {sum(hits.values()):,} train hits in "
+              f"{hit_docs:,} docs ({hit_docs_over_half:,} majority-hit); val: {val['docs']:,} docs, "
+              f"{val['hits']:,} hits in {val['hit_docs']:,} docs", flush=True)
     else:
         total_overlap = 0
         print(f"no manifest at {a.root}: merged mix not scanned", flush=True)
 
-    expected_fp = grams_seen * int(np.unique(ref).size) / float(1 << 64)
+    grams_all = grams_seen + val["grams"]
+    expected_fp = grams_all * int(np.unique(ref).size) / float(1 << 64)
     noise_allow = max(10.0, 10.0 * expected_fp)
     per_task_named = {t[0]: hits.get(t[0], 0) for t in REFS}
 
     if a.write_filter:
         os.makedirs(a.write_filter, exist_ok=True)
         summary = {"k": K, "criterion": "any 13-gram of the document appears in a reference "
-                                        "train/validation/dev split", "sources": {}}
+                                        "train/validation/dev split",
+                   "scanned_sources": sorted(masks), "sources": {}}
         for key, (ord_sorted, n_docs) in masks.items():
             m = np.zeros(n_docs, dtype=np.uint8)
             if ord_sorted.size:
@@ -488,6 +535,11 @@ def main():
         "mix_grams_audited": grams_seen,
         "mix_documents_with_any_hit": hit_docs,
         "mix_documents_majority_hit": hit_docs_over_half,
+        "val_documents_audited": val["docs"],
+        "val_tokens_audited": val["tokens"],
+        "val_documents_with_any_hit": val["hit_docs"],
+        "val_hits": val["hits"],
+        "mix_bytes_sha256": manifest_digest,
         "overlap_by_task": per_task_named,
         "overlap_total": total_overlap,
         "by_source": by_source,
@@ -504,8 +556,11 @@ def main():
                  "mix_documents_majority_hit and source_hit_document_fraction to judge severity. "
                  "Duplicate counting is diagnostic only: de-duplication is out of scope (D-014), "
                  "contamination removal is not (§3.3)."),
-        "COVERED_ALL_SHARDS": bool(not missing and docs > 0),
-        "MEASURED": bool(docs > 0 and grams_seen > 0 and not errors and len(tasks_covered) == 8),
+        "COVERED_ALL_SHARDS": bool(not missing and docs > 0 and val["docs"] > 0),
+        "FILTER_MASKS_WRITTEN": sorted(masks) if a.write_filter else None,
+        "MEASURED": bool(docs > 0 and grams_seen > 0 and val["docs"] > 0 and not errors
+                         and len(tasks_covered) == 8 and manifest_digest
+                         and (not a.write_filter or masks)),
     }
     report["AUDIT_PASSED"] = bool(report["MEASURED"] and total_overlap <= noise_allow)
     txt = json.dumps(report, indent=1, default=str)