perm_sha: identify the visit order by hashing the permutation, not just its seed (E-035 residual) -- cursor gains shuffle_perm_sha, asserted on resume; session repins the two files

train/train_ounce100m.py

@@ -125,10 +125,11 @@ class HubPush(TrainerCallback):
     """
 
     def __init__(self, repo, api, token, seq_len, seed, fingerprint, seqs_per_step,
-                 prune=True, min_free_gb=6.0, repo_type="dataset", stop_after_steps=0):
+                 prune=True, min_free_gb=6.0, repo_type="dataset", stop_after_steps=0, order_sha=None):
         self.repo, self.api, self.token = repo, api, token
         self.seq_len, self.seed, self.fingerprint = seq_len, seed, fingerprint
         self.seqs_per_step, self.repo_type = seqs_per_step, repo_type
+        self.order_sha = order_sha
         self.prune, self.min_free_gb = prune, min_free_gb
         self.stop_after = int(stop_after_steps or 0)
         self.stopped_at = None
@@ -155,7 +156,7 @@ class HubPush(TrainerCallback):
             raise RuntimeError(f"Trainer reported a save but {d} does not exist")
         samples = samples_at(step, self.seqs_per_step)
         SD.save_cursor(d, SD.cursor_dict(self.seq_len, samples, self.seed, self.fingerprint,
-                                         step=step))
+                                         step=step, order_sha=self.order_sha))
         t0 = time.time()
         # Upload and verify *without* pruning, roll the pointer and read it back, and only then delete the
         # local copy. The old order (prune inside push_and_prune, pointer written afterwards) meant one
@@ -409,9 +410,13 @@ def main():
         say(f"resume from {args.resume} cursor={cur}")
     start_sample = (cur or {}).get("samples_consumed", 0)
     start_step = (cur or {}).get("step") or 0
+    # Identity of the visit order, derived from the corpus size and the seed rather than trusted to the
+    # seed: numpy's permutation is only stable within a version, and Kaggle updates its image underneath
+    # long runs. ~50 ms once per session, and it is what makes "resumed onto a different order" loud.
+    order_sha = SD.perm_sha(store.total_tokens // args.seq_len, args.data_seed)
     if cur:
         for k, v in (("seq_len", args.seq_len), ("shuffle_seed", args.data_seed),
-                     ("dataset_files_sha", fingerprint)):
+                     ("shuffle_perm_sha", order_sha), ("dataset_files_sha", fingerprint)):
             if cur.get(k) != v:
                 raise SystemExit(f"resume refused: cursor {k}={cur.get(k)} != current {v}. "
                                  "A resumed run must read the same data in the same order (§3.1).")
@@ -425,6 +430,7 @@ def main():
             start_step = start_sample // seqs_per_step
 
     train_ds = SD.make_dataset(store, args.seq_len, args.data_seed, start_sample=start_sample)
+    assert getattr(train_ds, "order_sha", None) == order_sha, "two derivations of the order disagree"
     # Validation comes from the mix's held-out val/ shards, never from the training stream: scoring on
     # tokens the model already consumed would be a number nobody could reproduce (and §3.12).
     vs = [s["file"] for s in man.get("val_shards", [])]
@@ -546,7 +552,7 @@ def main():
         hub_cb = HubPush(args.hub_repo, api, os.environ.get("HF_TOKEN"), args.seq_len,
                          args.data_seed, fingerprint, seqs_per_step,
                          prune=args.prune, repo_type=args.hub_repo_type,
-                         stop_after_steps=args.stop_after_steps)
+                         stop_after_steps=args.stop_after_steps, order_sha=order_sha)
         tr.add_callback(hub_cb)
 
     # Precision must be ASSERTED, not configured: bf16 "works" silently on Turing and fp32 master
@@ -637,10 +643,11 @@ def main():
                 shipped.append(fn)
         say(f"tokenizer files shipped with the export: {shipped or 'NONE -- Phase 5 must add them'}")
         SD.save_cursor(exp, SD.cursor_dict(args.seq_len, samples, args.data_seed, fingerprint,
-                                           step=step))
+                                           step=step, order_sha=order_sha))
         json.dump({"step": step, "start_step": start_step, "tokens_consumed": consumed,
                    "tokens_this_run": consumed_this_run, "seconds": round(elapsed, 1),
-                   "params": pc, "fingerprint": fingerprint, "final_loss": final_loss,
+                   "params": pc, "fingerprint": fingerprint, "shuffle_perm_sha": order_sha,
+                   "final_loss": final_loss,
                    "prior_loss": prior, "samples_consumed": samples,
                    "steps_planned": steps_planned},
                   open(os.path.join(exp, "run_summary.json"), "w"), indent=1, sort_keys=True)