# Phase 2 contamination audit (§3.3): mechanical 13-token overlap between the mix and the eval tasks. # # What §3.3 permits, precisely: "benchmark validation/dev material may be used mechanically: a script # computes overlap counts and you never read or learn from the items themselves. Test splits stay # untouched until Phase 6." This script obeys that literally, and it is written so that violating it is # hard: the only values that ever leave the process are counts and rates per source. No item text, no # lengths, no samples, nothing a human could read as a question. The print surface is a dict of integers. # # Which splits are the reference, and why it is broader than the minimum: for most of these tasks the # harness draws its few-shot exemplars from TRAIN (ARC, HellaSwag, PIQA, Winogrande) or from MMLU's `dev`, # so overlap with those is contamination by construction. The reference set is {train, validation, dev} # where present, per task, and NEVER `test`. # # Why this file does not use Python sets of hashed grams (it did, and it was wrong): 1.23 B tokens in # 13-token windows is ~1.2e9 windows, and `hashlib.sha256` per window in CPython is ~1-2 us each, so the # audit would have taken hours to days and simply never finished inside a CPU session. Two changes make it # tractable without weakening it: # 1. Gram fingerprints are a vectorised multiplicative rolling hash (mod 2^64) computed per document with # numpy, not a Python loop. Cost is ~20 uint64 ops per token, in C. # 2. The reference is ONE sorted uint64 array plus a uint16 bitmask giving the tasks each gram belongs to, # queried with np.searchsorted. Memory is ~1 GB, not the ~7 GB a Python set of 90M ints needs, and # per-reference sets are unnecessary. # Collision behaviour is stated rather than hidden: 64-bit fingerprints over ~1.2e9 mix grams and ~1e8 # reference grams give an expected spurious-match count of mix_grams x ref_grams / 2^64, reported as # `expected_false_positives`. A raw count below that is noise, not evidence of contamination; above it, # the offending source is investigated. # # Read-only over the mix. CPU session. No credentials in this file. import argparse import hashlib import json import os import struct import numpy as np K = 13 P = np.uint64(0x100000001B3) # FNV-1a prime, odd, so invertible mod 2^64 PINV = np.uint64(pow(int(P), -1, 1 << 64)) MODMASK = np.uint64((1 << 64) - 1) # task -> (repo, config(s) to try in order, splits treated as reference material; 'test' deliberately # absent). Split names were read off the dataset server on 2026-09-19 rather than assumed: gsm8k has # train/test only, truthfulqa's repo is `truthfulqa/truthful_qa` and exposes `validation` only, and # `sylinrl/TruthfulQA` -- the name the first version of this file used -- no longer exists. REFS = [ ("arc_easy", "allenai/ai2_arc", ["ARC-Easy"], ["train", "validation"]), ("arc_challenge", "allenai/ai2_arc", ["ARC-Challenge"], ["train", "validation"]), ("hellaswag", "Rowan/hellaswag", [None], ["train", "validation"]), ("piqa", "baber/piqa", [None], ["train", "validation"]), ("winogrande", "allenai/winogrande", ["winogrande_xl"], ["train", "validation"]), ("mmlu", "cais/mmlu", ["all"], ["dev", "validation"]), ("truthfulqa", "truthfulqa/truthful_qa", ["multiple_choice", "generation"], ["validation"]), ("gsm8k", "openai/gsm8k", ["main"], ["train"]), ] TEXT_KEYS = ("question", "text", "choices", "story", "sentence", "prompt", "ctx", "input", "query", "answer", "solution", "reasoning", "explanation", "goal", "sol1", "sol2", "abstractions", "nonabstractions", "mc1_targets", "mc2_targets") # A reference split that yields fewer rows than this did not load, it merely failed to raise: the first # run of this audit reported `piqa::default::train: 0 items -> 0 grams` with no exception, which a # coverage-blind verdict would have scored as "no contamination found in PIQA". MIN_ITEMS_PER_SPLIT = 200 def window_hashes(ids, k=K): """uint64 fingerprint of every k-length window of `ids` (uint16 array), vectorised. h_i = sum_{j=i}^{i+k-1} x_j * P^(i+k-1-j) mod 2^64. Via the prefix recurrence C_i = C_{i-1}*P + x_i and the identity h_i = C_{i+k-1} - C_{i-1} * P^k. C is computed as P^i * cumsum(x_j * PINV^j), which is what makes it a few numpy passes instead of a Python loop. """ n = int(ids.shape[0]) if n < k: return np.empty(0, dtype=np.uint64) x = ids.astype(np.uint64) def powers(base, m): # [base^0, base^1, ..., base^(m-1)] mod 2^64. np.cumprod of a constant vector would start at # base^1, which shifts every window hash and silently breaks the match against the reference. out = np.empty(m, dtype=np.uint64) out[0] = np.uint64(1) if m > 1: out[1:] = np.cumprod(np.full(m - 1, base, dtype=np.uint64)) return out pow_p, pow_pi = powers(P, n), powers(PINV, n) s = np.cumsum(x * pow_pi, dtype=np.uint64) c = s * pow_p # c[i] = C_i, prefix hash ending at i pk = np.uint64(pow(int(P), k, 1 << 64)) prev = np.concatenate((np.zeros(1, dtype=np.uint64), (c[:n - k] * pk))) h = (c[k - 1:] - prev[:n - k + 1]) & MODMASK 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) elif isinstance(v, dict): for x in v.values(): flatten(x, out) elif isinstance(v, (list, tuple)): for x in v: flatten(x, out) SERVER = "https://datasets-server.huggingface.co" _SIZE_CACHE = {} def server_sizes(repo, cfg): """Row counts the dataset server reports for each split of one config, or {} if it cannot be asked. This exists because a streaming reader can stop early without raising, and because a loader that yields 0 rows of a 16,113-row split (exactly what `baber/piqa` did under the Kaggle image's `datasets`) is indistinguishable from a clean result unless the expected count is checked. Never trust the count from memory: the first version of this file expected 14,042 MMLU validation rows, and 14,042 is MMLU's *test* split -- the one §3.3 forbids reading. `validation` is 1,531. """ import urllib.parse import urllib.request ck = (repo, cfg) if ck in _SIZE_CACHE: return _SIZE_CACHE[ck] q = {"dataset": repo} if cfg: q["config"] = cfg out = {} try: with urllib.request.urlopen(SERVER + "/size?" + urllib.parse.urlencode(q), timeout=90) as r: d = json.load(r) out = {s["split"]: int(s.get("num_rows", 0)) for s in d.get("size", {}).get("splits", [])} except Exception as e: print(f" sizes API {repo}/{cfg or ''}: {type(e).__name__}", flush=True) _SIZE_CACHE[ck] = out return out def api_rows(repo, cfg, sp, cap): """Page the public rows endpoint (100 per request, full=true so no cell is silently truncated).""" import urllib.parse import urllib.request off = 0 while off < cap: q = {"dataset": repo, "split": sp, "offset": off, "limit": 100, "full": "true"} if cfg: q["config"] = cfg try: with urllib.request.urlopen(SERVER + "/rows?" + urllib.parse.urlencode(q), timeout=120) as r: d = json.load(r) except Exception as e: print(f" rows API {repo}/{cfg or ''}/{sp} @{off}: {type(e).__name__}", flush=True) return batch = d.get("rows") or [] if not batch: return for item in batch: yield item.get("row", item) off += len(batch) def ingest(tk, rows, cap): """Encode each row's text fields into gram windows. Returns (items kept, list of gram arrays).""" g, n = [], 0 for row in rows: if n >= cap: break parts = [] for k in TEXT_KEYS: if k in row: flatten(row[k], parts) txt = " ".join(parts) if not txt.strip(): continue n += 1 ids = np.asarray(tk.encode(txt, add_special_tokens=False).ids, dtype=np.uint16) h = window_hashes(ids) if h.size: g.append(h) return n, g def build_reference(tk, max_items_per_split): """Returns (sorted uint64 reference grams, parallel uint16 task bitmask, per-split item counts, errors, per-split expected row counts). Only counts and hashes are ever held or printed. A split is accepted only if the rows actually read reach 90 % of what the dataset server says the split holds; short reads are retried through the HTTP rows endpoint and, if still short, recorded as loader problems -- which fail MEASURED rather than scoring as "no contamination found". """ grams, masks = [], [] counts, errors, srv = {}, {}, {} try: import datasets except Exception as e: datasets = None errors["import datasets"] = f"{type(e).__name__}: {str(e)[:120]}" for ti, (task, repo, configs, splits) in enumerate(REFS): bit = np.uint16(1 << ti) for sp in splits: key = f"{task}::{repo}::{sp}" best = None for cfg in configs: ckey = f"{key}[{cfg or 'default'}]" want = server_sizes(repo, cfg).get(sp) srv[key] = want n, g = 0, [] if datasets is not None: try: ds = (datasets.load_dataset(repo, cfg, split=sp, streaming=True) if cfg else datasets.load_dataset(repo, split=sp, streaming=True)) n, g = ingest(tk, ds, max_items_per_split) except Exception as e: 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", flush=True) n2, g2 = ingest(tk, api_rows(repo, cfg, sp, max_items_per_split), max_items_per_split) if n2 > n: n, g = n2, g2 counts[ckey] = n print(f" ref {ckey}: {n} items of {want} -> {sum(a.size for a in g):,} grams", flush=True) if best is None or n > best[1]: best = (cfg, n, g, want) if n >= MIN_ITEMS_PER_SPLIT and (not want or n >= int(0.9 * want)): break if best is None or best[1] < MIN_ITEMS_PER_SPLIT: errors[key] = (f"no config of {repo} yielded >= {MIN_ITEMS_PER_SPLIT} rows for {sp} " f"(best {(best[1] if best else 0)} of {srv.get(key)})") continue cfg, n, g, want = best if want and n < int(0.9 * want): errors[key] = f"covered {n} of {want} rows in {sp} -- under-read, not calling it clean" continue if g: arr = np.concatenate(g) grams.append(arr) masks.append(np.full(arr.shape, bit, dtype=np.uint16)) if not grams: raise SystemExit("no reference grams built at all -- the audit would be a pass over nothing") allg = np.concatenate(grams) allm = np.concatenate(masks) order = np.argsort(allg, kind="stable") return allg[order], allm[order], counts, errors, srv def gram_lookup(g, ref, refmask): """(number of hits, the task bitmask of the matched reference grams) for one document. numpy only, because this runs ~1.1 M times. `searchsorted` into the sorted reference table plus the parallel bitmask is what makes per-task attribution possible without holding eight sets. """ if g.size == 0: return 0, None pos = np.clip(np.searchsorted(ref, g), 0, ref.size - 1) eq = ref[pos] == g if not eq.any(): return 0, None return int(eq.sum()), refmask[pos[eq]] 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, 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": []} if seen is None: seen = set() names = [t[0] for t in REFS] for idx, raw in docs: b = bytes(raw) ids = np.frombuffer(b, dtype=np.uint16) s["docs"] += 1 s["tokens"] += int(ids.size) dg = hashlib.blake2b(b, digest_size=8).digest() if dg in seen: # counted, never removed: de-duplication is out of scope (D-014) s["dup_docs"] += 1 s["dup_tokens"] += int(ids.size) else: seen.add(dg) g = window_hashes(ids) s["grams"] += int(g.size) h, bits = gram_lookup(g, ref, refmask) if h: s["hit_docs"] += 1 s["hits"] += h s["hit_tokens"] += int(ids.size) s["per_task"] = [a + c for a, c in zip(s["per_task"], per_task_counts(bits))] s["ordinals"].append(idx) if h * 2 >= int(g.size): s["hit_docs_over_half"] += 1 s["seconds"] = round(time.time() - t0, 1) s["per_task_named"] = {n: c for n, c in zip(names, s["per_task"]) if c} return s def doc_stream(BM, shard_paths): """(ordinal, raw document bytes) over a fixed list of shards. Ordinal order is the shard order in the source's own record.json, which is exactly what the merge replays -- so an index computed here still means the same document after a session restart.""" n = 0 for p in shard_paths: for d in BM.iter_shard_docs(p): yield n, d n += 1 def main(): ap = argparse.ArgumentParser() ap.add_argument("--root", default="/kaggle/working/mixroot") ap.add_argument("--stage-subdir", default="stage", help="per-source staged shards live here under --root; scanning them is what makes " "attribution possible, because every merged shard contains all sources at once") ap.add_argument("--max-items-per-split", type=int, default=50000) ap.add_argument("--write-filter", default="", help="directory for .u8 exclusion masks + filter.json (read back by " "build_mix --exclude-dir)") ap.add_argument("--out", default="", help="write the report JSON here (publish_mix expects " "mixroot/audit.json)") ap.add_argument("--ref-cache", default="", help="directory to keep the built reference gram table in. Streaming the eight " "reference sets is the slow half of this script, and a re-audit after a filter " "pass rebuilds exactly the same table -- so it is reused only if the task list, " "item cap, k and tokenizer sha all still match.") a = ap.parse_args() from tokenizers import Tokenizer from huggingface_hub import hf_hub_download tjp = hf_hub_download("HuggingFaceTB/SmolLM2-135M", "tokenizer.json") tk = Tokenizer.from_file(tjp) tk.no_truncation() import hashlib tk_sha = hashlib.sha256(open(tjp, "rb").read()).hexdigest() want_meta = {"k": K, "max_items": a.max_items_per_split, "tokenizer_sha256": tk_sha, "text_keys": list(TEXT_KEYS), "refs": [[t[0], t[1], list(t[2]), list(t[3])] for t in REFS]} ref = refmask = None refcounts = errors = srv = {} if a.ref_cache and os.path.exists(os.path.join(a.ref_cache, "refmeta.json")): try: if json.load(open(os.path.join(a.ref_cache, "refmeta.json"))) == want_meta: ref = np.load(os.path.join(a.ref_cache, "ref.npy")) refmask = np.load(os.path.join(a.ref_cache, "refmask.npy")) refcounts = json.load(open(os.path.join(a.ref_cache, "refcounts.json"))) errors = json.load(open(os.path.join(a.ref_cache, "errors.json"))) srv = json.load(open(os.path.join(a.ref_cache, "srv.json"))) print(f"reference: loaded from cache, {ref.size:,} grams", flush=True) except Exception as e: print(f"reference cache unusable ({type(e).__name__}: {str(e)[:120]}); rebuilding", flush=True) ref = None if ref is None: print("building reference gram table (only counts ever leave this process)", flush=True) ref, refmask, refcounts, errors, srv = build_reference(tk, a.max_items_per_split) if a.ref_cache: os.makedirs(a.ref_cache, exist_ok=True) np.save(os.path.join(a.ref_cache, "ref.npy"), ref) np.save(os.path.join(a.ref_cache, "refmask.npy"), refmask) json.dump(want_meta, open(os.path.join(a.ref_cache, "refmeta.json"), "w")) json.dump(refcounts, open(os.path.join(a.ref_cache, "refcounts.json"), "w")) json.dump(errors, open(os.path.join(a.ref_cache, "errors.json"), "w")) 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} - {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) import build_mix as BM # ---- per-source attribution, over the staged (pre-merge) shards ------------------------------ # The first version attributed merged hits as `c // len(rec["sources"])` to every source named in the # shard. All 141 shards name all 15 sources, so every source received an almost identical number and # the table looked like a finding while being arithmetically meaningless. stage_root = os.path.join(a.root, a.stage_subdir) by_source, masks = {}, {} if os.path.isdir(stage_root): 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): 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", [])] absent = [p for p in paths if not os.path.exists(p)] if absent: errors[f"stage::{key}"] = f"{len(absent)} staged shard(s) missing on disk" print(f" stage {key}: {len(absent)} shard(s) missing, skipped", flush=True) continue s = scan_docs(ref, refmask, doc_stream(BM, paths)) s["shards"] = len(paths) s["stage_complete"] = bool(rec.get("complete")) by_source[key] = s masks[key] = (np.asarray(sorted(s["ordinals"]), dtype=np.int64), s["docs"]) print(f" stage {key}: {s['docs']:,} docs {s['tokens']:,} tok -> hits={s['hits']:,} in " f"{s['hit_docs']:,} docs ({s['hit_docs_over_half']:,} of them >50 % matched) " f"[{s['seconds']}s] {json.dumps(s['per_task_named'])}", flush=True) s.pop("ordinals") else: print(f"no staged sources under {stage_root}: attribution unavailable, totals only", flush=True) # ---- the verdict is always measured on the bytes that will actually train --------------------- man_path = os.path.join(a.root, "manifest.json") hits = {} 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]), seen=seen_all) docs += s["docs"] mix_tokens += s["tokens"] grams_seen += s["grams"] hit_docs += s["hit_docs"] hit_docs_over_half += s["hit_docs_over_half"] dup_docs += s["dup_docs"] dup_tokens += s["dup_tokens"] for n, c in s["per_task_named"].items(): hits[n] = hits.get(n, 0) + c # 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) 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", "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: m[ord_sorted] = 1 m.tofile(os.path.join(a.write_filter, key + ".u8")) s = by_source[key] summary["sources"][key] = {"docs": s["docs"], "tokens": s["tokens"], "dropped_docs": int(ord_sorted.size), "dropped_tokens": int(s["hit_tokens"]), "kept_docs": int(n_docs - ord_sorted.size)} with open(os.path.join(a.write_filter, "filter.json"), "w") as f: json.dump(summary, f, indent=1, default=str) print(f"filter masks written to {a.write_filter} for {len(masks)} source(s)", flush=True) contaminated_sources = {k: v["hit_docs"] / max(1, v["docs"]) for k, v in by_source.items() if v["hit_docs"]} report = { "k": K, "fingerprint": "multiplicative rolling hash mod 2^64, P=0x100000001B3", "reference_items_by_split": refcounts, "reference_rows_expected_by_split": srv, "reference_loader_problems": errors, "reference_grams_distinct": int(np.unique(ref).size), "reference_grams_total": int(ref.size), "tasks_covered": tasks_covered, "shards_missing": missing, "mix_documents_audited": docs, "mix_tokens_audited": mix_tokens, "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, "source_hit_document_fraction": {k: round(v, 6) for k, v in sorted( contaminated_sources.items(), key=lambda kv: -kv[1])}, "duplicate_documents": dup_docs, "duplicate_tokens": dup_tokens, "expected_false_positives": round(expected_fp, 3), "noise_allowance": round(noise_allow, 3), "overlap_exceeds_noise": bool(total_overlap > noise_allow), "note": ("counts only; no item text read or printed. Reference splits are train/validation/dev " "per task -- never test, per §3.3. overlap_by_task sums gram hits, so one verbatim " "benchmark document contributes as many hits as it has windows; read " "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 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) print(f"VERDICT AUDIT_PASSED={report['AUDIT_PASSED']} MEASURED={report['MEASURED']} " f"docs={docs:,} overlap={total_overlap} hit_docs={hit_docs:,} " f"majority_hit_docs={hit_docs_over_half:,} expected_fp={expected_fp:.3f} " f"tasks={len(tasks_covered)}/8 problems={len(errors)}") print("AUDIT_JSON_BEGIN") print(txt) print("AUDIT_JSON_END") if a.out: with open(a.out, "w") as f: f.write(txt) raise SystemExit(0 if report["AUDIT_PASSED"] else (6 if report["MEASURED"] else 7)) if __name__ == "__main__": main()