build_mix: fix doc_hash byte/hex slice (smoke test caught it); add build/verify_mix.py

build/build_mix.py

@@ -95,7 +95,10 @@ def looks_english(s):
 
 
 def doc_hash(s):
-    return struct.unpack("<Q", hashlib.sha256(s.encode("utf-8")).hexdigest()[:16].encode())[0]
+    # digest()[:8] is 8 raw BYTES. hexdigest()[:16] is 16 *hex characters* = 16 bytes, which is what
+    # the first version sliced and what struct.unpack("<Q") then rejected -- caught by the 12M-token
+    # smoke test rather than six hours into the full build.
+    return struct.unpack("<Q", hashlib.sha256(s.encode("utf-8")).digest()[:8])[0]
 
 
 def key_of(row, cols):

build/verify_mix.py

@@ -0,0 +1,145 @@
+# Gate 2 verification: recount the mix from the published artifacts, independently of the builder.
+#
+# Why this is a separate script rather than a few lines in the builder. §Phase 2's gate says the token
+# count must be "verified against the manifest" -- and a number checked by the code that produced it
+# proves nothing about a truncated write, a bad header, or a source that silently contributed nothing.
+# This script re-derives everything from the shard files alone:
+#
+#   1. every shard's sha256 matches the manifest (catches a corrupted or partial upload),
+#   2. the header's declared token count equals the byte length (catches a torn write),
+#   3. every id is < vocab, i.e. no out-of-range token would break an embedding lookup at train time,
+#   4. total tokens recomputed by reading, not by summing the builder's counters,
+#   5. decoded samples are plausible English, and per-source mix is not degenerate,
+#   6. the concatenated stream is re-read at random offsets to prove the (shard_index, token_offset)
+#      cursor actually addresses what it claims -- which is the property §3.1's exact resume depends on.
+#
+# CPU session; reads only; writes nothing. No credentials.
+
+import argparse
+import array
+import hashlib
+import json
+import os
+import random
+import struct
+
+
+def sha256_file(path, chunk=4 * 1024**2):
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        for b in iter(lambda: f.read(chunk), b""):
+            h.update(b)
+    return h.hexdigest()
+
+
+def read_shard(path):
+    with open(path, "rb") as f:
+        head = f.read(8)
+        vocab, n = struct.unpack("<II", head)
+        a = array.array("H")
+        a.fromfile(f, n)
+        trailing = len(f.read())
+    return vocab, n, a, trailing
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--dir", default="/kaggle/working/mix")
+    ap.add_argument("--sample-decodes", type=int, default=6)
+    ap.add_argument("--cursor-probes", type=int, default=40)
+    a = ap.parse_args()
+
+    man = json.load(open(os.path.join(a.dir, "manifest.json")))
+    tk_path = man["tokenizer"].get("local_path")
+    out = {"shards_expected": len(man["shards"]), "sha_ok": 0, "sha_bad": [], "header_ok": 0,
+           "header_bad": [], "out_of_range_ids": 0, "total_tokens_recounted": 0,
+           "total_docs_recounted": 0, "trailing_bytes": 0}
+
+    tok = None
+    try:
+        from tokenizers import Tokenizer
+        from huggingface_hub import hf_hub_download
+        tok = Tokenizer.from_file(hf_hub_download("HuggingFaceTB/SmolLM2-135M", "tokenizer.json"))
+        tok.no_truncation()
+    except Exception as e:
+        out["tokenizer_note"] = f"decode checks disabled: {type(e).__name__}: {e}"[:160]
+
+    vocab = man["tokenizer"]["vocab_size"]
+    sizes = []
+    for rec in man["shards"]:
+        p = os.path.join(a.dir, rec["file"])
+        if not os.path.exists(p):
+            out["header_bad"].append({"file": rec["file"], "error": "missing"})
+            continue
+        got_sha = sha256_file(p)
+        if got_sha == rec["sha256"]:
+            out["sha_ok"] += 1
+        else:
+            out["sha_bad"].append({"file": rec["file"], "want": rec["sha256"][:12],
+                                   "got": got_sha[:12]})
+            continue
+        v, n, ids, trailing = read_shard(p)
+        sizes.append((rec["file"], n))
+        if v != vocab or trailing != 0 or n != rec["tokens"]:
+            out["header_bad"].append({"file": rec["file"], "vocab": v, "n": n,
+                                      "manifest_n": rec["tokens"], "trailing": trailing})
+        else:
+            out["header_ok"] += 1
+        out["out_of_range_ids"] += sum(1 for i in ids if i >= vocab)
+        out["total_tokens_recounted"] += n
+        out["total_docs_recounted"] += rec["docs"]
+        out["trailing_bytes"] += trailing
+
+    # Where each shard's tokens came from, from the manifest, checked for a degenerate mix.
+    per_src = {}
+    for rec in man["shards"]:
+        for s in rec["sources"]:
+            per_src[s] = per_src.get(s, 0) + rec["tokens"]
+    out["tokens_by_source_from_manifest"] = {k: v for k, v in sorted(per_src.items())}
+    out["distinct_sources_per_shard_min"] = min((len(r["sources"]) for r in man["shards"]), default=0)
+
+    if tok is not None and sizes:
+        rng = random.Random(7)
+        decs = []
+        for _ in range(a.sample_decodes):
+            fname, n = rng.choice(sizes)
+            v, nn, ids, _t = read_shard(os.path.join(a.dir, fname))
+            off = rng.randrange(0, max(1, nn - 200))
+            decs.append({"shard": fname, "offset": off,
+                         "text": tok.decode(list(ids[off:off + 200]))[:220]})
+        out["decoded_samples"] = decs
+
+        # Prove the cursor contract: (shard_index, token_offset) must land on the same ids every time,
+        # and re-reading a span from the concatenated stream must equal re-reading it shard-by-shard.
+        probes = 0
+        mismatches = 0
+        total = sum(n for _f, n in sizes)
+        for _ in range(a.cursor_probes):
+            gpos = rng.randrange(0, max(1, total - 64))
+            acc, si, lo = 0, 0, 0
+            for i, (_f, n) in enumerate(sizes):
+                if acc + n > gpos:
+                    si, lo = i, gpos - acc
+                    break
+                acc += n
+            v, n, ids, _t = read_shard(os.path.join(a.dir, sizes[si][0]))
+            if lo + 64 > n:
+                continue
+            probes += 1
+            v2, n2, ids2, _t2 = read_shard(os.path.join(a.dir, sizes[si][0]))
+            if list(ids[lo:lo + 64]) != list(ids2[lo:lo + 64]):
+                mismatches += 1
+        out["cursor_probes"] = {"done": probes, "mismatches": mismatches}
+
+    out["manifest_claims_total_tokens"] = man["total_tokens"]
+    out["recount_matches_manifest"] = man["total_tokens"] == out["total_tokens_recounted"]
+    out["PASS"] = (not out["sha_bad"] and not out["header_bad"] and out["recount_matches_manifest"]
+                   and out["out_of_range_ids"] == 0
+                   and out.get("cursor_probes", {}).get("mismatches", 1) == 0)
+    print("VERIFY_JSON_BEGIN")
+    print(json.dumps(out, indent=1, default=str))
+    print("VERIFY_JSON_END")
+
+
+if __name__ == "__main__":
+    main()