train: hubckpt -- download_checkpoint prefix/landing-dir fix (found by P1 v2)

train/hubckpt.py

@@ -211,15 +211,20 @@ def download_checkpoint(repo, path_in_repo, dest, api, repo_type="dataset", toke
     """
     import time as _t
     prefix = path_in_repo.rstrip("/") + "/"
-    have = {k: v for k, v in hub_listing(repo, repo_type, token=token).items() if k.startswith(prefix)}
-    if not have:
-        raise RuntimeError(f"{repo}:{path_in_repo} lists no files -- refusing to resume from nothing")
+    listing = hub_listing(repo, repo_type, token=token)
+    if not any(k.startswith(prefix) for k in listing):
+        raise RuntimeError(f"{repo} has no files under {prefix} -- refusing to resume from nothing")
+    # Keys come back as full repo paths; strip the prefix once here so `rel` and `filename` agree.
+    # (Passing prefix+full_path downstream asked the Hub for ckpt/x/ckpt/x/cursor.json -- a 404 that
+    # looked like a missing checkpoint, found by preflight P1 on CPU rather than mid-main-run.)
+    have = {k[len(prefix):]: v for k, v in listing.items() if k.startswith(prefix)}
     os.makedirs(dest, exist_ok=True)
     t0 = _t.time()
     tot = 0
+    landed = []
     for rel in sorted(have):
         p = api.hf_hub_download(repo_id=repo, filename=prefix + rel, repo_type=repo_type,
-                               local_dir=dest, token=token)
+                                local_dir=dest, token=token)
         sz = os.path.getsize(p)
         if sz != have[rel][0]:
             raise RuntimeError(f"{rel}: hub lists {have[rel][0]} bytes, {sz} arrived")
@@ -227,8 +232,15 @@ def download_checkpoint(repo, path_in_repo, dest, api, repo_type="dataset", toke
         if sha and len(sha) == 64 and sha256_file(p) != sha:
             raise RuntimeError(f"{rel}: sha256 does not match the hub's lfs oid")
         tot += sz
+        landed.append(p)
+    # local_dir mirrors the repo layout, so a checkpoint at ckpt/checkpoint-7/ lands *inside* dest with
+    # its subdirectories. The caller needs the directory that actually holds the weights, not the one it
+    # asked for, or the resume silently points at an empty prefix and starts from step 0.
+    dirs = {os.path.dirname(p) for p in landed}
+    if len(dirs) != 1:
+        raise RuntimeError(f"checkpoint files landed in {len(dirs)} directories: {sorted(dirs)[:4]}")
     return {"files": len(have), "bytes": tot, "seconds": round(_t.time() - t0, 1),
-            "dest": dest, "path_in_repo": path_in_repo}
+            "dir": dirs.pop(), "requested_dest": dest, "path_in_repo": path_in_repo}
 
 
 def write_latest_pointer(repo, cur, api, repo_type="dataset", token=None):