build_mix: hub-init failure is fatal (not silent), empty merge exits 4; add token-scope probe

build/build_mix.py

@@ -219,7 +219,8 @@ class State:
 
 # --------------------------------------------------------------------- stage
 def stage(args, st, tokenizer):
-    remote = HS.remote_stage_files(args.hub_repo) if (args.hub_repo and not args.merge_only) else set()
+    remote = HS.remote_stage_files(args.hub_repo) if (HS and args.hub_repo
+                                    and not args.merge_only) else set()
     if args.hub_repo:
         print(f"hub: {len(remote)} staged files already on {args.hub_repo}", flush=True)
     for src in SOURCES:
@@ -228,7 +229,7 @@ def stage(args, st, tokenizer):
         if rec.get("complete"):
             print(f"stage: skip {key} (complete)", flush=True)
             continue
-        if args.hub_repo:
+        if HS and args.hub_repo:
             # A source whose shards AND record are already on the Hub is done; pulling the record is
             # cheaper than re-staging and keeps the merge's proportional weights honest.
             have = [p for p in remote if p.startswith(f"stage/{key}/") and p.endswith(".bin")]
@@ -352,7 +353,7 @@ def stage(args, st, tokenizer):
         st.commit(new_h, new_k)
         # /kaggle/working is destroyed when the session ends, so a source that took 40 minutes to stage
         # must not exist only there. Published per source, because that is the unit a resume can skip.
-        if complete and args.hub_repo:
+        if complete and HS and args.hub_repo:
             with open(os.path.join(st.stage_dir, key, "record.json"), "w") as f:
                 json.dump({"key": key, "src": {k: v for k, v in src.items()},
                            "shards": shards, "tokens": got, "docs_seen": docs_seen,
@@ -392,8 +393,11 @@ def merge(args, st, tokenizer):
         streams.append({"key": key, "weight": float(weight), "credit": 0.0, "it": gen(iters),
                         "pending": None, "emitted_tokens": 0, "exhausted": False})
     if not streams:
-        print("merge: nothing staged", flush=True)
-        return
+        # The cold-resume rehearsal exited 0 here while producing an empty mix, and verify_mix then
+        # reported recount_matches_manifest: true because 0 == 0. An empty mix is the worst possible
+        # silent success in this pipeline, so it is a failure and not a return.
+        print("FATAL merge: nothing staged. Refusing to emit an empty mix.", flush=True)
+        sys.exit(4)
     total_w = sum(s["weight"] for s in streams)
     print(f"merge: {len(streams)} sources, "
           f"{sum(s['weight'] for s in streams) / 1e6:,.0f}M staged tokens", flush=True)
@@ -491,15 +495,21 @@ def main():
 
     global HS
     HS = None
+    # Sync is best-effort to SET UP but not to USE: if the Hub cannot be reached or written, turning it
+    # off silently would reintroduce the exact failure this exists to prevent (a killed session losing
+    # hours of streaming). So a configured repo that cannot be initialised is fatal unless --no-hub.
     if args.hub_repo:
         try:
             import hubsync as HS_mod
             HS = HS_mod
             HS.ensure_repo(args.hub_repo)
         except Exception as e:
-            print(f"!! hub sync unavailable: {type(e).__name__}: {str(e)[:160]}; "
-                  f"build proceeds but is NOT interruption-safe", flush=True)
-            HS = None
+            print(f"FATAL hub sync could not initialise: {type(e).__name__}: {str(e)[:200]}",
+                  flush=True)
+            print("Refusing to run a multi-hour build that is not interruption-safe. Resolve the Hub "
+                  "write path, or pass --hub-repo '' only if you accept losing all staged work when the "
+                  "session ends.", flush=True)
+            sys.exit(3)
 
     st = State(args.root)
     if args.hub_repo and not st.d["staged"]:
@@ -522,7 +532,8 @@ def main():
             print(f"hub: restored {len(st.d['staged'])} staged source(s) "
                   f"({st.staged_tokens_total():,} tokens)", flush=True)
         except Exception as e:
-            print(f"!! hub restore failed: {type(e).__name__}: {str(e)[:160]}", flush=True)
+            print(f"FATAL hub restore failed: {type(e).__name__}: {str(e)[:200]}", flush=True)
+            sys.exit(3)
 
     t0 = time.monotonic()
     if not args.merge_only:

probes/p2_token_scope.py

@@ -0,0 +1,130 @@
+# Phase 2 blocker check: can this token write DATASETS, or only models?
+#
+# Why this exists. `p1-credential-roundtrip` proved a job can create and delete a **model** repo. The
+# cold-resume rehearsal then failed to create a **dataset** repo with `401 Unauthorized` from
+# POST /api/repos/create, using the same token and the same code path. Phase 2 requires publishing the mix
+# as a public `ounce100m-*` *dataset*, and the stage repo is a dataset too -- so if the hypothesis is
+# right, this is a hard blocker on Gate 2 that only the user can clear by adjusting the token.
+#
+# The token is fine-grained (`repo.write` scoped to the user). HF splits "write to models" and "write to
+# datasets" into separate scopes, and `repo.write` has historically meant the former, so the hypothesis is
+# plausible rather than a mystery.
+#
+# Test design: for each repo type, create → upload → read back anonymously → delete, and report the
+# outcome per step. Anything created here is deleted in the same run; nothing lingers in the account.
+# It also dumps the token's own declared capabilities, which is the authoritative answer rather than an
+# inference from a 401.
+#
+# CPU session, zero GPU quota. No credentials in this file (D-006 helper).
+
+import json
+import os
+import urllib.error
+import urllib.request
+
+import ounce100m_credentials
+
+R = {}
+STAMP = "scopetest"
+NAMES = {"model": f"Cion-lab/ounce100m-{STAMP}-model-DELETEME",
+         "dataset": f"Cion-lab/ounce100m-{STAMP}-dataset-DELETEME"}
+
+
+def guard(name, fn):
+    try:
+        R[name] = fn()
+    except Exception as e:
+        detail = ""
+        resp = getattr(e, "response", None)
+        if resp is not None:
+            try:
+                detail = resp.text[:400]
+            except Exception:
+                pass
+        R[name] = {"error": f"{type(e).__name__}: {str(e)[:280]}", "body": detail}
+
+
+guard("install", lambda: ounce100m_credentials.install(verify=True))
+TOKEN = os.environ.get("HF_TOKEN", "")
+
+
+def whoami_full():
+    req = urllib.request.Request("https://huggingface.co/whoami-v2",
+                                 headers={"Authorization": f"Bearer {TOKEN}"})
+    with urllib.request.urlopen(req, timeout=45) as r:
+        d = json.loads(r.read().decode())
+    # report capability *shape*, never the token itself
+    out = {"name": d.get("name"), "type": d.get("type")}
+    full = d.get("full") or {}
+    out["full_keys"] = sorted(full.keys())
+    auth = full.get("auth") or d.get("auth") or {}
+    out["auth"] = {k: v for k, v in auth.items() if k != "accessToken"}
+    for k in ("capabilities", "scopes", "permissions"):
+        if k in full:
+            out[k] = full[k]
+        elif k in d:
+            out[k] = d[k]
+    if "accessToken" in full:
+        at = full["accessToken"]
+        out["accessToken_type"] = type(at).__name__
+        if isinstance(at, dict):
+            out["accessToken_scopes"] = at.get("scopes") or at.get("permission")
+    return out
+
+
+guard("whoami_v2", whoami_full)
+
+
+def trial(kind):
+    from huggingface_hub import HfApi, create_repo, delete_repo
+
+    repo = NAMES[kind]
+    api = HfApi(token=TOKEN)
+    steps = {}
+    create_repo(repo_id=repo, repo_type=kind, private=False, exist_ok=True)
+    steps["create"] = "ok"
+    payload = f"ounce100m scope probe {kind}\n".encode()
+    api.upload_file(path_or_fileobj=payload, path_in_repo="probe.txt", repo_id=repo,
+                    repo_type=kind, commit_message="scope probe")
+    steps["upload"] = "ok"
+    url = (f"https://huggingface.co/{repo}/resolve/main/probe.txt" if kind == "model"
+           else f"https://huggingface.co/datasets/{repo}/resolve/main/probe.txt")
+    try:
+        with urllib.request.urlopen(url, timeout=60) as r:
+            steps["anonymous_readback_matches"] = r.read().strip() == payload.strip()
+    except urllib.error.HTTPError as e:
+        steps["anonymous_readback"] = f"HTTP {e.code}"
+    delete_repo(repo_id=repo, repo_type=kind)
+    steps["delete"] = "ok"
+    return steps
+
+
+for kind in ("model", "dataset"):
+    guard(f"trial_{kind}", lambda k=kind: trial(k))
+
+# confirm the deletes really took, so nothing is left behind in the account
+def listing():
+    out = {}
+    for kind, q in (("model", "models"), ("dataset", "datasets")):
+        try:
+            with urllib.request.urlopen(
+                    f"https://huggingface.co/api/{q}?author=Cion-lab", timeout=45) as r:
+                out[kind] = sorted(m["id"] for m in json.loads(r.read().decode()))
+        except Exception as e:
+            out[kind] = f"{type(e).__name__}"
+    return out
+
+
+guard("namespace_after", listing)
+
+R["verdict"] = {
+    "model_write": "ok" if isinstance(R.get("trial_model"), dict) and
+    R["trial_model"].get("create") == "ok" else R.get("trial_model"),
+    "dataset_write": "ok" if isinstance(R.get("trial_dataset"), dict) and
+    R["trial_dataset"].get("create") == "ok" else R.get("trial_dataset"),
+}
+R["BLOCKS_GATE_2"] = R["verdict"]["dataset_write"] != "ok"
+
+print("SCOPE_JSON_BEGIN")
+print(json.dumps(R, indent=1, default=str)[:5000])
+print("SCOPE_JSON_END")