Bandwidth A-Z: hash-only rescue, theta-only miner, lazy trace, H=512, real FineWeb mix

scripts/launch_cuda_run.sh

@@ -17,7 +17,6 @@ umask 077
 
 TAG=${TAG:-v1}
 RD=${RD:-/data/role-dispatch}
-T=${TOOLS:-$RD/tools}
 E2E=${E2E:-/root/e2e-deploy-wt/hypertrain-e2e-b/.e2e}
 OUT=${OUT:-$RD/manifest/out-$TAG}
 JOB=${JOB:-$RD/manifest/job-$TAG}
@@ -29,6 +28,7 @@ PY=${PY:-$RD/runtime/venv/bin/python}
 MINER_SRC=${MINER_SRC:-/data/pydeps/hypertrain}
 REPO=${REPO:-$(cd "$(dirname "$0")/.." && pwd)}
 FIX_BASE=${FIX_BASE:-1a8f91e}
+T=${TOOLS:-$REPO/scripts/ops}
 MINER_SSH=${MINER_SSH:-}
 MINER_KEY=${MINER_KEY:-}
 # Signed floor: cutoff >= admitted + staging 1800 s + qualification 9600 s + hard margin 1800 s;
@@ -77,7 +77,8 @@ export PYTHONPATH=$HT_TREE/src:$HT_TREE/scripts:$HT_TREE:/data/pydeps:$RD/runtim
 mkdir -p "$STATE"
 
 STAGE=preflight
-for f in "$T"/{prepare_record.py,task8-deps.py,rescue_parallel.py,task8-materialize-roots.py} \
+for f in "$T"/{prepare_record.py,task8-deps.py,hf_seed.py,rescue_parallel.py,rescue_dedup.py} \
+  "$T"/{task8-materialize-roots.py,prepare_runtime.py} \
   "$T"/{backend_bootstrap.py,live_operator.py,auditors.sh} \
   "$E2E/beacon/pusher.py" "$E2E/relays/build_relay_configs.py" "$E2E/relays/deploy_relay.sh" \
   "$OUT/manifest-envelope.json" "$OUT/manifest.json" "$OUT/profile.json" "$JOB/job.json" \
@@ -120,14 +121,28 @@ case $rc in 0) BACKEND=unregistered ;; 3) BACKEND=cuda ;; 4) die "run ${RID:0:8}
 BUNDLE=$STATE/bundle
 rm -rf "$BUNDLE" && mkdir -p "$BUNDLE" && cp -a "$MINER_SRC" "$BUNDLE/hypertrain"
 find "$BUNDLE" -name __pycache__ -prune -exec rm -rf {} +
+# Diff FIX_BASE against the working tree (not HEAD): uncommitted miner changes ship too.
 if ! grep -q '_retrying("PUT object' "$BUNDLE/hypertrain/miner/core.py"; then
-  git -C "$REPO" diff "$FIX_BASE" HEAD -- src/hypertrain/miner/core.py src/hypertrain/data/store.py \
+  git -C "$REPO" diff "$FIX_BASE" -- src/hypertrain/miner/core.py src/hypertrain/data/store.py \
     | patch -p3 --forward --quiet -d "$BUNDLE/hypertrain" || die "transport fixes do not apply onto $MINER_SRC"
 fi
 grep -q '_retrying("PUT object' "$BUNDLE/hypertrain/miner/core.py" || die "bundle lacks the PUT retry fix"
 grep -q 'def _parts' "$BUNDLE/hypertrain/data/store.py" || die "bundle lacks the parallel part fetch"
+grep -q '^UPLOAD_WORKERS = ' "$BUNDLE/hypertrain/miner/core.py" || die "bundle lacks the parallel part PUT"
+grep -q 'def _local_input' "$BUNDLE/hypertrain/miner/core.py" || die "bundle lacks the theta-only input reuse"
+# store.py has no master-side divergence: the bundle copy must be the working tree's bytes.
+cmp -s "$REPO/src/hypertrain/data/store.py" "$BUNDLE/hypertrain/data/store.py" \
+  || die "bundle data/store.py differs from the working tree"
 log "PREFLIGHT_OK run=$RID image=$IMAGE job_deadline=$JD backend=$BACKEND miner=${MINER_SSH:-unset}"
 if [ "$MODE" = --dry-run ]; then log "DRY_RUN_DONE no provider call, nothing spent"; exit 0; fi
+# Runtime uid 65532 and the master container (sees /data only) cannot read a /root checkout:
+# run a frozen copy of the tools from /data.
+if [ "${T#"$RD"/}" = "$T" ]; then
+  TS=$RD/ops-$TAG
+  install -d -o 65532 -g 65532 -m 0700 "$TS"
+  for f in "$T"/*.py "$T"/*.sh; do install -o 65532 -g 65532 -m 0700 "$f" "$TS/"; done
+  T=$TS
+fi
 
 if [ ! -e "$STATE/window" ]; then
   A=$(date +%s); CUT=$((A + WINDOW)); [ "$CUT" -le "$JD" ] || CUT=$JD

scripts/ops/auditors.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+# Two independent contract-2 auditor clients (auditor-0 -> manifest slot 0, auditor-1 -> slot 1).
+# CUDA replay runs master-side (store.execute_audit_v2 -> _role_launch "audit" -> owned host).
+# Usage: auditors.sh stage | check RUN_ID | once RUN_ID | start RUN_ID | stop | cleanup
+set -euo pipefail
+IMG=${HT_IMAGE:-ghcr.io/cortexlm/hypertrain@sha256:58e2b42713a9065758796408e7aea4fb3260ccfaa5ad7bad4859fd87320c93d6}
+API=${HT_API:-http://cortex-challenge-hypertrain:8000}
+NET=${HT_NET:-cortex-challenges}
+SRC=${HT_AUDITOR_SEEDS:-/root/.config/cortex-secrets/od-large-e2e}
+BASE=/data/role-dispatch/auditors
+NAMES=(hypertrain-auditor-0 hypertrain-auditor-1)
+cmd=${1:?stage|check|once|start|stop|cleanup}; RID=${2:-}
+need_rid() { [[ $RID =~ ^[0-9a-f]{64}$ ]] || { echo "RUN_ID (64 hex) required" >&2; exit 2; }; }
+flags() { echo --network "$NET" --read-only --user 65532:65532 --cap-drop ALL --security-opt no-new-privileges \
+  --tmpfs /tmp:size=64m -e PYTHONPATH=/data/pydeps -e HOME=/tmp -v /data/pydeps:/data/pydeps:ro \
+  -v "$BASE/auditor-$1":/state:ro --entrypoint sh; }
+cli() { echo "python -m hypertrain.auditor --api $API --run-id $RID --contract 2 --keyfile /state/auditor.seed"; }
+present() { # exact task-owned containers that exist now; docker failure propagates
+  local all n; all=$(docker ps -a --format '{{.Names}}') || exit 1
+  for n in "${NAMES[@]}"; do grep -qxF "$n" <<<"$all" && echo "$n"; done; return 0; }
+remove_owned() {
+  local live; live=$(present) || exit 1
+  [[ -z $live ]] || docker rm -f $live || exit 1
+  live=$(present) || exit 1
+  [[ -z $live ]] || { echo "auditor containers still present: $live" >&2; exit 1; }
+  echo "auditor containers absent"; }
+case $cmd in
+  stage) install -d -o root -g root -m 0755 $BASE
+         for i in 0 1; do install -d -o 65532 -g 65532 -m 0700 $BASE/auditor-$i
+           install -o 65532 -g 65532 -m 0600 $SRC/auditor-$i.seed $BASE/auditor-$i/auditor.seed; done
+         stat -c '%a %u:%g %n' $BASE/auditor-*/auditor.seed ;;
+  check) need_rid   # GET only: contract 2 and this identity at its exact manifest slot
+         for i in 0 1; do docker run --rm $(flags $i) "$IMG" -c "python - <<'EOF'
+import hypertrain.trainer, httpx
+from pathlib import Path
+from hypertrain.miner.core import load_keyfile
+k=load_keyfile(Path('/state/auditor.seed')); r=httpx.get('$API/v2/runs/$RID',timeout=30); r.raise_for_status(); j=r.json()
+aud=j['manifest_envelope']['body']['training']['auditors']
+assert j['contract']==2, j.get('contract')
+assert len(aud)>$i and aud[$i]==k.ss58, ('slot $i', k.ss58, aud)
+print('READY auditor-$i', k.ss58, 'slot $i')
+EOF"; done ;;
+  once)  need_rid   # proof: each identity must settle one job with result MATCH; empty queue fails
+         for i in 0 1; do out=$(docker run --rm $(flags $i) "$IMG" -c "$(cli) --once"); echo "$out"
+           grep -q '"result": "MATCH"' <<<"$out" || { echo "auditor-$i: no MATCH verdict" >&2; exit 1; }
+           echo "auditor-$i MATCH"; done ;;
+  start) need_rid   # production loop: re-poll every 20s in-container; a failed poll (transient 5xx while the
+         # master is busy) is logged with its exit code and retried, never a container restart
+         for i in 0 1; do docker run -d --name ${NAMES[$i]} --restart always --label io.cortex.task=auditors \
+           --log-opt max-size=20m $(flags $i) "$IMG" -c "while :; do $(cli) || echo AUDITOR_POLL_EXIT=\$?; sleep 20; done"; done ;;
+  stop)  remove_owned ;;
+  cleanup) remove_owned
+         rm -f $BASE/auditor-0/auditor.seed $BASE/auditor-1/auditor.seed
+         rmdir $BASE/auditor-0 $BASE/auditor-1 $BASE
+         test ! -e $BASE && echo "auditors cleanup: seeds+state absent" ;;
+  *) echo "unknown $cmd" >&2; exit 2 ;;
+esac

scripts/ops/backend_bootstrap.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Bootstrap the live CUDA backend from verified two-host qualification, before run POST."""
+
+import argparse
+import hashlib
+import json
+import sys
+from pathlib import Path
+
+import hypertrain.trainer  # noqa: F401  determinism setup before any torch import
+from hypertrain.beacon import parse_round
+from hypertrain.challenge.app import Config, load_coord_key
+from hypertrain.challenge.store import ChallengeStore
+from hypertrain.data.store import LocalFSStore
+from hypertrain.gpu_ops import network_qualification as engine
+from hypertrain.gpu_ops.journal import fsha
+from hypertrain.miner.core import load_keyfile
+from hypertrain.protocol import envelope_v2
+from hypertrain.protocol.jcs import canonicalize
+from hypertrain.protocol.messages_v2 import RunManifestV2
+
+p = argparse.ArgumentParser(description=__doc__)
+p.add_argument("--manifest-envelope", type=Path, required=True)
+p.add_argument("--owner-seed", type=Path, required=True)
+p.add_argument("--config", type=Path, nargs=2, required=True, help="h0 h1: <output>/config.json")
+p.add_argument("--result", type=Path, nargs=2, required=True, help="h0 h1: <output>/result.json")
+a = p.parse_args()
+
+raw = a.manifest_envelope.read_bytes()
+env = envelope_v2.parse_envelope(raw)
+if env.type != "RunManifestV2" or not envelope_v2.verify_envelope(raw):
+    sys.exit("manifest envelope is not a verified RunManifestV2")
+manifest = RunManifestV2.model_validate(env.body)
+cfg = Config.from_env()
+owner = load_keyfile(a.owner_seed)
+if cfg.owner_hotkey is None or owner.ss58 != cfg.owner_hotkey or env.signer != cfg.owner_hotkey:
+    sys.exit("owner seed / manifest signer is not HYPERTRAIN_OWNER_HOTKEY")
+
+# The existing verifier runs before any Receipt is sealed or store opened.
+for config_path in a.config:
+    job = engine.check_inputs(engine.Qualification.model_validate_json(config_path.read_bytes()))
+    if job.manifest != manifest:
+        sys.exit(f"{config_path}: qualification seed job binds another manifest")
+verdict = engine.compare(list(a.result), manifest)
+if verdict != {"two_host_bitwise_match": True, "workload_allowed": False}:
+    sys.exit(f"two-host comparison did not match: {verdict}")
+
+ref = manifest.training.reference_spec
+evidence = {
+    "run_id": manifest.run_id(),
+    "backend": "cuda",
+    "reference_hash": hashlib.sha256(canonicalize(ref.model_dump(mode="json"))).hexdigest(),
+    "layout_hash": ref.layout.model_dump_json(),
+    "configs": [fsha(c) for c in a.config],
+    "results": [fsha(r) for r in a.result],
+    "summaries": [
+        fsha(r.parent / f"round-{w}/published/rank-{k}/summary.json")
+        for r in a.result
+        for w in range(2)
+        for k in range(ref.layout.n_gpus)
+    ],
+}
+authority = hashlib.sha256(canonicalize(evidence)).hexdigest()
+
+store = ChallengeStore(
+    cfg.state_dir,
+    cfg.params,
+    load_coord_key(cfg.coord_key_file),
+    cfg.owner_hotkey,
+    parse_round,
+    LocalFSStore(cfg.state_dir / "objects"),
+)
+try:
+    now = store._now(store._db)
+    receipt = envelope_v2.seal(
+        owner,
+        "Receipt",
+        manifest.run_id(),
+        {"w": 0, "commit_hash": authority, "received_round": now},
+        now + 100,
+    )
+    store._bootstrap_backend_v2(manifest, tuple(a.config), tuple(a.result), canonicalize(receipt))
+    print(
+        json.dumps(
+            {"run_id": manifest.run_id(), "received_round": now, "authority_hash": authority}
+        )
+    )
+finally:
+    store.close()

scripts/ops/hf_seed.py

@@ -0,0 +1,104 @@
+"""Rebuild seed objects on a host from the pinned HF genesis checkpoint instead of uploading them.
+
+A seed object (start_state, ef_in) is a safetensors file whose tensors are either all zero or a
+`theta/<name>` copy of the genesis checkpoint's `<name>`. plan() derives that layout from the
+controller's local file; the host gets only the header (~100 KB) and a copy list, downloads the
+checkpoint with parallel range requests and writes header + zeros + theta. The caller proves the
+result with the usual size+sha256 check and falls back to uploading on any mismatch.
+"""
+
+import json
+import shlex
+import struct
+from pathlib import Path
+
+REPO_ID = "OpenType/neon-1.2-genesis"
+# Resolved 2026-10-15 via https://huggingface.co/api/models/OpenType/neon-1.2-genesis
+REVISION = "1aa51eb5561719044f206061ba63a0d8e94ab5b0"
+URL = f"https://huggingface.co/{REPO_ID}/resolve/{REVISION}/model.safetensors"
+SIZE = 1_415_480_056
+SHA256 = "c2b614d2353af47583d4f0604db006e04430a6e84f16066161e748e7fc0b6aa9"
+RANGES = 8
+CHUNK = 16 << 20
+
+
+def plan(local: Path) -> tuple[bytes, list[list]] | None:
+    """(header bytes, copy list) for `local`, or None when it is not header + zeros + theta."""
+    with local.open("rb") as f:
+        raw = f.read(8)
+        head = raw + f.read(struct.unpack("<Q", raw)[0])
+        tensors = json.loads(head[8:])
+        tensors.pop("__metadata__", None)
+        base = len(head)
+        if base + max(t["data_offsets"][1] for t in tensors.values()) != local.stat().st_size:
+            return None
+        copy = []
+        for name, t in sorted(tensors.items(), key=lambda kv: kv[1]["data_offsets"][0]):
+            lo, hi = t["data_offsets"]
+            f.seek(base + lo)
+            left, zero = hi - lo, True
+            while left and zero:
+                b = f.read(min(left, CHUNK))
+                zero = bool(b) and b.count(0) == len(b)
+                left -= len(b)
+            if zero:
+                continue
+            if not name.startswith("theta/"):
+                return None
+            copy.append([base + lo, name.removeprefix("theta/"), hi - lo, t["dtype"], t["shape"]])
+    return head, copy
+
+
+def payload(head: bytes, copy: list[list], out: str, src: str, size: int) -> bytes:
+    spec = json.dumps({"out": out, "src": src, "size": size, "copy": copy}).encode()
+    return struct.pack("<Q", len(spec)) + spec + head
+
+
+# Host side: stdin = payload(); checks every copied tensor's dtype/shape/size against the
+# downloaded checkpoint header; zeros are a sparse truncate.
+BUILD = r"""
+import json, os, struct, sys
+data = sys.stdin.buffer.read()
+n = struct.unpack("<Q", data[:8])[0]
+spec, head = json.loads(data[8:8 + n]), data[8 + n:]
+part = spec["out"] + ".hf-part"
+with open(part, "wb") as f:
+    f.write(head)
+    f.truncate(spec["size"])
+    if spec["copy"]:
+        src = open(spec["src"], "rb")
+        hl = struct.unpack("<Q", src.read(8))[0]
+        hh = json.loads(src.read(hl))
+        for dst, name, nbytes, dtype, shape in spec["copy"]:
+            t = hh[name]
+            lo, hi = t["data_offsets"]
+            if (hi - lo, t["dtype"], t["shape"]) != (nbytes, dtype, shape):
+                sys.exit("layout mismatch " + name)
+            src.seek(8 + hl + lo)
+            f.seek(dst)
+            while nbytes:
+                b = src.read(min(nbytes, 1 << 24))
+                if not b:
+                    sys.exit("short checkpoint " + name)
+                f.write(b)
+                nbytes -= len(b)
+os.replace(part, spec["out"])
+print("HF_BUILT", spec["out"], spec["size"])
+"""
+
+
+def download_cmd(dst: str, ranges: int = RANGES) -> str:
+    """bash: fetch the pinned checkpoint to `dst` with `ranges` parallel curls; verify size+sha."""
+    q, step = shlex.quote(dst), -(-SIZE // ranges)
+    gets = "".join(
+        f"curl -fsSL --retry 3 --connect-timeout 20 -r {i * step}-{min(SIZE, (i + 1) * step) - 1}"
+        f' -o {q}.r{i} {URL} & p="$p $!"; '
+        for i in range(ranges)
+    )
+    parts = " ".join(f"{q}.r{i}" for i in range(ranges))
+    return (
+        f'set -e; [ "$(sha256sum {q} 2>/dev/null | cut -c1-64)" = {SHA256} ] && exit 0; p=; '
+        f"{gets}for x in $p; do wait $x; done; cat {parts} > {q}.part; rm -f {parts}; "
+        f'[ "$(stat -c %s {q}.part)" = {SIZE} ]; '
+        f'[ "$(sha256sum {q}.part | cut -c1-64)" = {SHA256} ]; mv {q}.part {q}'
+    )