migration: make per-rank resume prefetch resilient to Hub timeouts

Retry the destination list_repo_files with backoff (and fall back to an
empty resume set) so a single transient huggingface.co read timeout no
longer fails an entire SLURM rank and skips its dataset slice.

community_v3_migration/classify.py

@@ -0,0 +1,155 @@
+"""Classify each sub-dataset: is it SO-100/101, and what joint encoding is it in?
+
+Detection = robot_type string (recording-time signal) cross-checked against the
+per-episode stats min/max (magnitude + exact-boundary saturation). Mismatches are
+flagged as `ambiguous` for manual review rather than silently converted.
+"""
+import json
+from pathlib import Path
+import numpy as np
+
+# so100/so101 (+ _follower/_bimanual), so_follower, and bimanual bi_so* (bi_so_follower,
+# bi_so100_follower, ...; 12-dim).
+SO_PREFIXES = ("so100", "so101", "so_", "bi_so")
+SO_EXACT: set[str] = set()
+# Robots that superficially look SO-like but are NOT in scope for the joint fix:
+NEVER_FIX = {"koch", "koch_follower", "koch_bimanual", "moss", "moss_follower"}
+
+RAD_MAX = 3.5     # |val| below this => radians
+DEG_MIN = 105.0   # |val| above this => old-convention degrees
+SAT_ATOL = 0.5    # closeness to +/-100 / 0 / 100 counted as normalization saturation
+
+
+def is_so_robot_type(rt: str) -> bool:
+    """True if the recorded ``robot_type`` denotes an in-scope SO-100/101 arm."""
+    return bool(rt) and (rt.startswith(SO_PREFIXES) or rt in SO_EXACT) and rt not in NEVER_FIX
+
+
+SO_JOINTS = ("shoulder_pan", "shoulder_lift", "elbow_flex", "wrist_flex", "wrist_roll", "gripper")
+
+
+def is_end_effector(info: dict) -> bool:
+    """True if action/observation.state are task-space end-effector features (e.g. ``ee_x``,
+    ``ee_roll``) rather than joint angles. Such datasets are out of scope for the joint fix."""
+    feats = info.get("features", {})
+    for key in ("action", "observation.state"):
+        names = [str(n).lower() for n in (feats.get(key, {}).get("names") or [])]
+        if any(n.startswith("ee_") or "end_effector" in n or "eef" in n for n in names):
+            return True
+        if {"x", "y", "z"} <= set(names):
+            return True
+    return False
+
+
+def _leading_so_joints(names: list[str], dim: int) -> int:
+    """Number of LEADING joints (a multiple of 6) that match the SO joint order in blocks of 6.
+    When names are absent, fall back to the full dim if it's already a multiple of 6, else 0."""
+    if not names:
+        return dim if dim and dim % 6 == 0 else 0
+    k = 0
+    while (k + 1) * 6 <= len(names) and all(SO_JOINTS[i] in names[k * 6 + i] for i in range(6)):
+        k += 1
+    return k * 6
+
+
+def so_joint_count(info: dict, key: str) -> int:
+    """Leading SO-arm joint count for one feature (``action`` / ``observation.state``). Trailing
+    non-SO columns (bbox, appended EE pose, ...) are excluded so only the genuine SO joints are
+    ever degrees-converted."""
+    feat = info.get("features", {}).get(key, {})
+    dim = (feat.get("shape") or [0])[0]
+    names = [str(n).lower() for n in (feat.get("names") or [])]
+    return _leading_so_joints(names, dim)
+
+
+def is_mislabeled_so(info: dict) -> bool:
+    """True when ``robot_type`` claims SO but no leading 6-DOF SO joint block can be substantiated
+    from action/observation.state (wrong dim, or names that don't match the SO set). When the first
+    6 joint names DO match, the SO block is honored (and processed) even if extra columns follow."""
+    return max(so_joint_count(info, "action"), so_joint_count(info, "observation.state")) == 0
+
+
+def load_info(root: Path) -> dict:
+    return json.loads((Path(root) / "meta" / "info.json").read_text())
+
+
+def _global_bounds(root: Path):
+    """Per-joint global min/max over action (fallback observation.state), across episodes."""
+    lo = hi = None
+    key_used = None
+    with open(Path(root) / "meta" / "episodes_stats.jsonl") as f:
+        for line in f:
+            s = json.loads(line)["stats"]
+            key = "action" if "action" in s else ("observation.state" if "observation.state" in s else None)
+            if key is None:
+                continue
+            key_used = key
+            mn = np.asarray(s[key]["min"], dtype=float)
+            mx = np.asarray(s[key]["max"], dtype=float)
+            lo = mn if lo is None else np.minimum(lo, mn)
+            hi = mx if hi is None else np.maximum(hi, mx)
+    return lo, hi, key_used
+
+
+def encoding_from_bounds(lo, hi, rt: str) -> dict:
+    """Detect the SO-arm joint encoding from per-joint global min/max and the robot_type name.
+
+    Layout-agnostic (v2.1 episodes_stats or v3.0 stats.json both reduce to lo/hi here), so it is
+    the single source of truth for the degrees_old / degrees_new / normalized / radians decision.
+    """
+    lo = np.asarray(lo, dtype=float)
+    hi = np.asarray(hi, dtype=float)
+    maxabs = float(np.nanmax(np.abs(np.concatenate([lo, hi]))))
+    # saturation on any arm joint (index != gripper) at +/-100, or gripper at 0/100
+    n = 6
+    sat = False
+    for a in range(len(hi) // n):
+        arm_hi, arm_lo = hi[a * n:a * n + n], lo[a * n:a * n + n]
+        joints_hi, joints_lo = arm_hi[:5], arm_lo[:5]
+        grip_hi, grip_lo = arm_hi[5], arm_lo[5]
+        sat |= bool(np.any(np.isclose(joints_hi, 100, atol=SAT_ATOL)) or
+                    np.any(np.isclose(joints_lo, -100, atol=SAT_ATOL)) or
+                    np.isclose(grip_hi, 100, atol=SAT_ATOL) or np.isclose(grip_lo, 0, atol=SAT_ATOL))
+
+    if maxabs <= RAD_MAX:
+        enc = "radians"
+    elif maxabs > DEG_MIN:
+        enc = "degrees_old"
+    elif sat:
+        enc = "normalized"
+    else:
+        enc = "degrees_new"
+
+    name_says_new = rt.endswith(("_follower", "_bimanual"))
+    ambiguous = (enc == "degrees_old" and name_says_new) or (enc in ("normalized", "degrees_new") and not name_says_new)
+    return {"encoding": enc, "maxabs": round(maxabs, 2), "saturates": sat, "ambiguous": ambiguous}
+
+
+def classify(root) -> dict:
+    root = Path(root)
+    info = load_info(root)
+    rt = info.get("robot_type", "") or ""
+    dim = (info.get("features", {}).get("action", {}).get("shape") or [None])[0]
+    out = {"root": str(root), "robot_type": rt, "action_dim": dim,
+           "codebase_version": info.get("codebase_version"), "ambiguous": False}
+
+    if is_end_effector(info):
+        return {**out, "is_so": False, "encoding": "end_effector",
+                "note": "task-space end-effector features"}
+
+    if is_so_robot_type(rt) and is_mislabeled_so(info):
+        return {**out, "is_so": False, "encoding": "non_so", "mislabeled_so": True,
+                "note": "robot_type claims SO but joint dim/names don't match a 6-DOF SO arm"}
+
+    is_so = is_so_robot_type(rt)
+    if not is_so:
+        return {**out, "is_so": False, "encoding": "non_so"}
+
+    lo, hi, key_used = _global_bounds(root)
+    if lo is None:
+        return {**out, "is_so": True, "encoding": "unknown", "ambiguous": True,
+                "note": "no action/state stats found"}
+
+    n = so_joint_count(info, key_used) or len(hi)  # ignore trailing non-joint columns
+    return {**out, "is_so": True, "stats_key": key_used, "so_dim": n,
+            **encoding_from_bounds(lo[:n], hi[:n], rt)}

community_v3_migration/extract_dataset.py

@@ -0,0 +1,55 @@
+"""Extract one sub-dataset from a LeRobotDataset monorepo into a standalone HF dataset repo.
+
+Downloads only ``{src-repo}/{folder}/...`` (scoped listing, no whole-repo walk) and re-uploads
+its contents at the ROOT of a NEW dataset repo, so the result is a self-contained LeRobotDataset.
+
+  python extract_dataset.py --folder wannrrr/etnai --dst-repo CarolinePascal/etnai
+"""
+import argparse
+import shutil
+import sys
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from run_migration import download_subfolder  # noqa: E402
+
+
+def main():
+    ap = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    ap.add_argument("--folder", required=True, metavar="USER/DATASET",
+                    help="Sub-dataset path within the source monorepo, e.g. 'wannrrr/etnai'.")
+    ap.add_argument("--dst-repo", required=True, metavar="ORG/NAME",
+                    help="New standalone destination dataset repo (must differ from the source).")
+    ap.add_argument("--src-repo", default="lerobot/community_dataset_v3", metavar="ORG/NAME",
+                    help="Source monorepo to pull the sub-dataset from.")
+    ap.add_argument("--work-dir", default="./extract_work", help="Local scratch directory.")
+    ap.add_argument("--private", action="store_true", help="Create the destination repo as private.")
+    args = ap.parse_args()
+
+    if args.dst_repo in (args.src_repo, args.folder):
+        ap.error("--dst-repo must be a new repo name, distinct from the source repo/folder.")
+
+    local = Path(args.work_dir) / args.folder
+    if local.parent.exists():
+        shutil.rmtree(local.parent, ignore_errors=True)
+
+    print(f"downloading {args.src_repo}/{args.folder} ...", file=sys.stderr)
+    download_subfolder(args.folder, args.work_dir, repo=args.src_repo)
+    if not (local / "meta" / "info.json").exists():
+        ap.error(f"'{args.folder}' is not a LeRobotDataset (no meta/info.json) in {args.src_repo}")
+
+    api = HfApi()
+    api.create_repo(args.dst_repo, repo_type="dataset", private=args.private, exist_ok=True)
+    print(f"uploading -> {args.dst_repo} ...", file=sys.stderr)
+    api.upload_folder(repo_id=args.dst_repo, repo_type="dataset", folder_path=str(local),
+                      commit_message=f"Standalone copy of {args.folder} from {args.src_repo}")
+    shutil.rmtree(Path(args.work_dir) / args.folder.split("/")[0], ignore_errors=True)
+    print(f"done: https://huggingface.co/datasets/{args.dst_repo}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

community_v3_migration/fix_dataset.py

@@ -0,0 +1,260 @@
+"""Rewrite observation.state / action to degrees in a LOCAL v2.1 SO-arm dataset, then
+regenerate meta/episodes_stats.jsonl (action & state only; other features preserved).
+Run this BEFORE the stock v2.1->v3.0 converter so its stats aggregation stays correct.
+"""
+import json
+from pathlib import Path
+import numpy as np
+import pandas as pd
+
+import so_arm_frame
+from classify import classify, load_info, so_joint_count
+
+VALUE_COLS = ("observation.state", "action")
+
+
+def _stack(col_values) -> np.ndarray:
+    return np.stack([np.asarray(v, dtype=np.float64) for v in col_values])  # (N, D)
+
+
+def _set_robot_type(root: Path, robot_type: str) -> None:
+    info_path = root / "meta" / "info.json"
+    info = json.loads(info_path.read_text())
+    info["robot_type"] = robot_type
+    info_path.write_text(json.dumps(info, indent=4))
+
+
+def _rewrite_parquet(root: Path, encoding: str, so_dims: dict) -> None:
+    for pq in sorted((root / "data").glob("*/*.parquet")):
+        df = pd.read_parquet(pq)
+        changed = False
+        for col in VALUE_COLS:
+            n = so_dims.get(col, 0)
+            if col in df.columns and n:
+                full = _stack(df[col].values)  # (N, D)
+                full[:, :n] = so_arm_frame.to_degrees(full[:, :n], encoding, n_joints_per_arm=6)
+                df[col] = list(full.astype(np.float32))
+                changed = True
+        if changed:
+            df.to_parquet(pq, index=False)
+
+
+def _regen_episode_stats(root: Path) -> None:
+    stats_path = root / "meta" / "episodes_stats.jsonl"
+    orig = {}
+    with open(stats_path) as f:
+        for line in f:
+            e = json.loads(line)
+            orig[e["episode_index"]] = e
+    for pq in sorted((root / "data").glob("*/*.parquet")):
+        df = pd.read_parquet(pq)
+        for ep in np.unique(df["episode_index"].values):
+            ep = int(ep)
+            sub = df[df["episode_index"] == ep]
+            entry = orig.get(ep)
+            if entry is None:
+                continue
+            for col in VALUE_COLS:
+                if col in sub.columns:
+                    a = _stack(sub[col].values)  # (n, D)
+                    entry["stats"][col] = {
+                        "min": a.min(0).tolist(), "max": a.max(0).tolist(),
+                        "mean": a.mean(0).tolist(), "std": a.std(0).tolist(),
+                        "count": [int(a.shape[0])],
+                    }
+    with open(stats_path, "w") as f:
+        for ep in sorted(orig):
+            f.write(json.dumps(orig[ep]) + "\n")
+
+
+def _read_jsonl(path: Path) -> list[dict]:
+    with open(path) as f:
+        return [json.loads(line) for line in f if line.strip()]
+
+
+def _write_jsonl(path: Path, rows: list[dict]) -> None:
+    with open(path, "w") as f:
+        for r in rows:
+            f.write(json.dumps(r) + "\n")
+
+
+def data_video_episode_mismatch(root) -> str | None:
+    """Return a description when the data files and any camera's video files disagree on the
+    episode count (dataset can't be migrated, e.g. 'All cams dont have same number of episodes'),
+    else None. Datasets without videos never mismatch here."""
+    root = Path(root)
+    info = json.loads((root / "meta" / "info.json").read_text())
+    counts = {"data": len(list((root / "data").glob("*/episode_*.parquet")))}
+    for k, f in info.get("features", {}).items():
+        if f.get("dtype") == "video":
+            counts[k] = len(list((root / "videos").glob(f"*/{k}/episode_*.mp4")))
+    if len(counts) > 1 and len(set(counts.values())) > 1:
+        return f"data/video episode counts disagree: {counts}"
+    return None
+
+
+def _file_ep_indices(root: Path, pattern: str) -> list[int]:
+    return sorted(int(p.stem.split("_")[-1]) for p in root.glob(pattern))
+
+
+def reindex_episodes(root) -> str | None:
+    """Compact non-contiguous episode indices to 0..N-1 when every source agrees on the set.
+
+    Some datasets (e.g. '*_clean' variants) had episodes deleted, leaving gaps in the episode
+    numbering (data, videos, and metadata all skip the same indices, e.g. {20, 37, 38, 39}). The
+    stock v2.1->v3.0 converter renumbers data/videos by sorted file order (0..N-1) but reads the
+    original gapped indices from episodes.jsonl, so the two disagree and it raises
+    "Number of episodes is not the same". When the data files, every camera's videos, and both
+    metadata files list the *exact same* episode index set, remap it to 0..N-1 everywhere so the
+    converter's positional alignment holds. Returns a note if remapped, else None (already
+    contiguous, or the sources disagree -> unsafe to touch)."""
+    root = Path(root)
+    info = json.loads((root / "meta" / "info.json").read_text())
+
+    ref = _file_ep_indices(root, "data/*/episode_*.parquet")
+    if not ref:
+        return None
+    sources = {"data": ref}
+    vkeys = [k for k, f in info.get("features", {}).items() if f.get("dtype") == "video"]
+    for k in vkeys:
+        sources[k] = _file_ep_indices(root, f"videos/*/{k}/episode_*.mp4")
+    eps = _read_jsonl(root / "meta" / "episodes.jsonl")
+    stats = _read_jsonl(root / "meta" / "episodes_stats.jsonl")
+    sources["episodes"] = sorted(e["episode_index"] for e in eps)
+    sources["episodes_stats"] = sorted(s["episode_index"] for s in stats)
+
+    if any(v != ref for v in sources.values()):
+        return None  # sources disagree on the episode set -> not safe to reindex here
+    n = len(ref)
+    if ref == list(range(n)):
+        return None  # already contiguous
+
+    remap = {old: new for new, old in enumerate(ref)}
+
+    # Data: rewrite episode_index (and rebuild the global 'index'), then rename the file. Ascending
+    # order is collision-free because new <= old for every episode.
+    running = 0
+    for old in ref:
+        matches = list((root / "data").glob(f"*/episode_{old:06d}.parquet"))
+        if not matches:
+            return None
+        pq = matches[0]
+        df = pd.read_parquet(pq)
+        if "episode_index" in df.columns:
+            df["episode_index"] = remap[old]
+        if "index" in df.columns:
+            df["index"] = np.arange(running, running + len(df), dtype=df["index"].dtype)
+        running += len(df)
+        df.to_parquet(pq, index=False)
+        dst = pq.with_name(f"episode_{remap[old]:06d}.parquet")
+        if dst != pq:
+            pq.rename(dst)
+
+    # Videos: rename per camera (ascending -> collision-free).
+    for k in vkeys:
+        for old in ref:
+            for mp4 in (root / "videos").glob(f"*/{k}/episode_{old:06d}.mp4"):
+                dst = mp4.with_name(f"episode_{remap[old]:06d}.mp4")
+                if dst != mp4:
+                    mp4.rename(dst)
+
+    for e in eps:
+        e["episode_index"] = remap[e["episode_index"]]
+    for s in stats:
+        s["episode_index"] = remap[s["episode_index"]]
+    _write_jsonl(root / "meta" / "episodes.jsonl", sorted(eps, key=lambda e: e["episode_index"]))
+    _write_jsonl(root / "meta" / "episodes_stats.jsonl", sorted(stats, key=lambda s: s["episode_index"]))
+
+    info["total_episodes"] = n
+    info["total_frames"] = int(running)
+    if "total_videos" in info:
+        info["total_videos"] = n * len(vkeys)
+    info["splits"] = {"train": f"0:{n}"}
+    (root / "meta" / "info.json").write_text(json.dumps(info, indent=4))
+    return f"episode indices compacted to 0..{n - 1} (dropped gaps {sorted(set(range(ref[-1] + 1)) - set(ref))})"
+
+
+def reconcile_episode_count(root) -> str | None:
+    """When the data files and video files agree on an episode count N but the metadata lists a
+    different count, rewrite the metadata (episodes.jsonl, episodes_stats.jsonl, info.json) to N.
+
+    Only the safe direction is handled: trimming metadata that lists MORE episodes than actually
+    exist. If the data itself is non-contiguous, the videos disagree with the data, or the metadata
+    lists FEWER episodes than the data (which would require fabricating per-episode stats), nothing
+    is changed and the stock converter's mismatch error is left to surface. Returns a note on fix."""
+    root = Path(root)
+    info = json.loads((root / "meta" / "info.json").read_text())
+
+    data_idx = sorted(int(p.stem.split("_")[-1]) for p in (root / "data").glob("*/episode_*.parquet"))
+    n = len(data_idx)
+    if n == 0 or data_idx != list(range(n)):
+        return None
+
+    vkeys = [k for k, f in info.get("features", {}).items() if f.get("dtype") == "video"]
+    for k in vkeys:
+        if len(list((root / "videos").glob(f"*/{k}/episode_*.mp4"))) != n:
+            return None  # data and videos disagree -> out of scope for this fix
+
+    eps_path = root / "meta" / "episodes.jsonl"
+    stats_path = root / "meta" / "episodes_stats.jsonl"
+    eps, stats = _read_jsonl(eps_path), _read_jsonl(stats_path)
+    if len(eps) == n and len(stats) == n:
+        return None
+
+    eps_keep = [e for e in eps if e.get("episode_index", -1) < n]
+    stats_keep = [s for s in stats if s.get("episode_index", -1) < n]
+    if len(eps_keep) != n or len(stats_keep) != n:
+        return None  # metadata is missing episodes present in the data -> can't safely fabricate
+
+    dropped = max(len(eps), len(stats)) - n
+    _write_jsonl(eps_path, eps_keep)
+    _write_jsonl(stats_path, stats_keep)
+    info["total_episodes"] = n
+    info["total_frames"] = int(sum(e.get("length", 0) for e in eps_keep))
+    if "total_videos" in info:
+        info["total_videos"] = n * len(vkeys)
+    info["splits"] = {"train": f"0:{n}"}
+    (root / "meta" / "info.json").write_text(json.dumps(info, indent=4))
+    return f"metadata episode count reconciled to {n} (data & videos agree; dropped {dropped} stale meta entries)"
+
+
+def fix_dataset_in_place(root) -> dict:
+    """Returns the classification dict augmented with the action taken."""
+    root = Path(root)
+    cls = classify(root)
+    if cls.get("mislabeled_so"):
+        # robot_type claims SO but the joints prove otherwise (wrong dim or non-SO names).
+        # Relabel to 'unknown' and migrate structurally rather than degrees-converting on a
+        # false assumption; the joint values are left exactly as recorded.
+        _set_robot_type(root, "unknown")
+        return {**cls, "robot_type": "unknown", "converted": False,
+                "action": f"structural v2.1->v3.0 only; robot_type relabeled '{cls.get('robot_type')}'"
+                          "->'unknown' (joints don't match a 6-DOF SO arm), joint values left unchanged"}
+    enc = cls.get("encoding")
+    if not cls.get("is_so") or enc in ("radians", "unknown", "non_so"):
+        reason = {
+            "non_so": "not an SO-100/101 dataset",
+            "radians": "SO-arm joints already in radians",
+            "unknown": "SO-arm but joint encoding could not be determined",
+        }.get(enc, "no joint conversion applicable")
+        return {**cls, "converted": False,
+                "action": f"structural v2.1->v3.0 only ({reason}); joint values left unchanged"}
+    if enc == "normalized" and not so_arm_frame.CANON_IS_CALIBRATED:
+        # Without per-robot calibration the un-normalization is an identity (placeholder
+        # spans == 100), so rewriting is pointless. Keep the normalized values as-is and let
+        # the dataset card flag them APPROXIMATE instead.
+        return {**cls, "converted": False,
+                "action": "structural v2.1->v3.0 only; joint values kept in normalized units "
+                          "(-100..100 / 0..100), NOT converted to degrees (uncalibrated -> APPROXIMATE)"}
+    # drop stray files that would otherwise be uploaded
+    for junk in (root / "meta").glob("info.json.bak"):
+        junk.unlink()
+    info = load_info(root)
+    so_dims = {c: so_joint_count(info, c) for c in VALUE_COLS}
+    _rewrite_parquet(root, enc, so_dims)
+    _regen_episode_stats(root)
+    full_dims = {c: (info.get("features", {}).get(c, {}).get("shape") or [0])[0] for c in VALUE_COLS}
+    partial = any(0 < so_dims[c] < full_dims[c] for c in VALUE_COLS)
+    tail = " (leading SO joints only; trailing non-joint columns left unchanged)" if partial else ""
+    return {**cls, "converted": True,
+            "action": f"structural v2.1->v3.0 + joint values converted ({enc} -> degrees){tail}"}

community_v3_migration/migrate_molmoact.py

@@ -0,0 +1,133 @@
+"""Migrate the SO-100/101 datasets referenced by ``allenai/MolmoAct2-SO100_101-Dataset``.
+
+That repo does NOT store the datasets themselves; it lists them. Each
+``language_annotations/{user}/{dataset}/...`` folder name is the HF repo id of a *standalone*
+LeRobotDataset. This script derives those repo ids, drops any already present in
+``lerobot/community_dataset_v3`` (already migrated) and in the destination (resume), then runs
+the exact same per-dataset pipeline as ``run_migration.py`` on each remaining standalone repo
+(download whole repo -> SO-arm joint fix -> v2.1->v3.0 convert -> card -> upload -> cleanup).
+
+  python migrate_molmoact.py --dst-repo lerobot/community_dataset_v3 --work-dir ./molmo_work
+Flags mirror run_migration.py: --only-classify, --no-push, --folder-name USER/DATASET [...],
+--limit N, --reference-repo (the "already migrated" set to skip against).
+"""
+import argparse
+import csv
+import shutil
+import sys
+import traceback
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from classify import classify  # noqa: E402
+from run_migration import already_done, list_datasets, migrate_one  # noqa: E402
+
+LIST_REPO = "allenai/MolmoAct2-SO100_101-Dataset"
+REFERENCE_REPO = "lerobot/community_dataset_v3"  # the "already migrated" set to skip against
+ANNOTATIONS_PREFIX = "language_annotations/"
+
+
+def list_molmoact_datasets(api: HfApi, repo: str = LIST_REPO) -> list[str]:
+    """Standalone dataset repo ids (``{user}/{dataset}``) derived from the folder names under
+    ``language_annotations/`` in the MolmoAct listing repo."""
+    files = api.list_repo_files(repo, repo_type="dataset")
+    return sorted({"/".join(f.split("/")[1:3]) for f in files
+                   if f.startswith(ANNOTATIONS_PREFIX) and len(f.split("/")) >= 3})
+
+
+def pending_datasets(api: HfApi, subs: list[str], dst_repo: str | None,
+                     reference_repo: str, no_upload: bool, only_classify: bool) -> list[str]:
+    """Drop ids already in the reference repo (already migrated) and, unless classify/no-push,
+    ids already in the destination repo (resume)."""
+    skip = set(list_datasets(api, reference_repo))
+    if not only_classify and not no_upload and dst_repo and dst_repo != reference_repo:
+        skip |= {p[: -len("/meta/info.json")] for p in api.list_repo_files(dst_repo, repo_type="dataset")
+                 if p.endswith("/meta/info.json")}
+    return [s for s in subs if s not in skip]
+
+
+def main():
+    ap = argparse.ArgumentParser(
+        description="Migrate the standalone SO-100/101 datasets listed by "
+                    f"{LIST_REPO} to LeRobotDataset v3.0 (degrees), skipping any already present "
+                    f"in --reference-repo. One dataset at a time (download -> fix -> convert -> "
+                    "upload -> cleanup); resumable.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+    ap.add_argument("--dst-repo", default=REFERENCE_REPO, metavar="ORG/NAME",
+                    help="Destination HF dataset repo to push the converted v3.0 datasets to "
+                         "(created if missing).")
+    ap.add_argument("--reference-repo", default=REFERENCE_REPO, metavar="ORG/NAME",
+                    help="Repo whose datasets are considered already migrated and skipped.")
+    ap.add_argument("--work-dir", default="./molmo_work", metavar="DIR",
+                    help="Local scratch directory (one dataset lives here at a time on a push run).")
+    ap.add_argument("--manifest", default="manifest_molmoact.csv", metavar="CSV",
+                    help="CSV log appended to as datasets are processed. Reused across resumed runs.")
+    ap.add_argument("--limit", type=int, default=None, metavar="N",
+                    help="Process only the first N pending datasets (alphabetical). Ignored with "
+                         "--folder-name.")
+    ap.add_argument("--folder-name", nargs="+", default=None, metavar="USER/DATASET",
+                    help="One or more specific standalone repo ids to process (must appear in the "
+                         f"{LIST_REPO} listing).")
+    ap.add_argument("--only-classify", action="store_true",
+                    help="Detect robot type + joint encoding and write the manifest only; no "
+                         "download of data, convert, or push.")
+    ap.add_argument("--no-push", action="store_true",
+                    help="Fix + convert locally but do NOT upload; output kept under --work-dir.")
+    args = ap.parse_args()
+    no_upload = args.no_push
+
+    api = HfApi()
+    all_ids = list_molmoact_datasets(api)
+    if args.folder_name:
+        wanted = {n.strip("/") for n in args.folder_name}
+        subs = [s for s in all_ids if s in wanted]
+        missing = wanted - set(subs)
+        if missing:
+            print(f"warning: not in {LIST_REPO} listing: {', '.join(sorted(missing))}", file=sys.stderr)
+    else:
+        subs = pending_datasets(api, all_ids, args.dst_repo, args.reference_repo, no_upload, args.only_classify)
+        if args.limit:
+            subs = subs[: args.limit]
+    print(f"{len(subs)} dataset(s) to process (of {len(all_ids)} listed)", file=sys.stderr)
+    if not subs:
+        return
+
+    if not args.only_classify and not no_upload:
+        api.create_repo(args.dst_repo, repo_type="dataset", exist_ok=True)
+    dst_files = set() if (args.only_classify or no_upload) else set(
+        api.list_repo_files(args.dst_repo, repo_type="dataset"))
+
+    first = not Path(args.manifest).exists()
+    with open(args.manifest, "a", newline="") as mf:
+        w = None
+        for i, sub in enumerate(subs):
+            try:
+                if args.only_classify:
+                    from huggingface_hub import snapshot_download
+                    local = Path(args.work_dir) / sub
+                    snapshot_download(repo_id=sub, repo_type="dataset", local_dir=str(local),
+                                      allow_patterns=["meta/*"])
+                    row = {"root": sub, **classify(local)}
+                    shutil.rmtree(Path(args.work_dir) / sub.split("/")[0], ignore_errors=True)
+                elif not no_upload and already_done(api, args.dst_repo, sub, dst_files):
+                    row = {"root": sub, "action": "skipped: already present in destination repo"}
+                else:
+                    row = migrate_one(api, args.dst_repo, sub, args.work_dir, no_upload, standalone=True)
+            except Exception as e:
+                row = {"root": sub, "action": f"ERROR: {e}"}
+                traceback.print_exc()
+            if w is None:
+                w = csv.DictWriter(mf, fieldnames=sorted(
+                    {"root", "robot_type", "is_so", "encoding", "action_dim",
+                     "maxabs", "ambiguous", "action", "codebase_version", "note"}))
+                if first:
+                    w.writeheader()
+            w.writerow({k: row.get(k) for k in w.fieldnames})
+            mf.flush()
+            print(f"[{i+1}/{len(subs)}] {sub}: {row.get('action')}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

community_v3_migration/prune_destination.py

@@ -0,0 +1,137 @@
+"""Prune datasets from the migrated destination repo based on the run manifest(s).
+
+Selects, from the manifest CSV(s) written by run_migration.py / slurm_migrate.py, the datasets to
+remove and deletes their folders from the destination repo. Two independent filters:
+
+  --errored        rows whose migration action starts with "ERROR:" (failed to convert; usually
+                   absent from the repo, but any partial upload left behind is cleaned up).
+  --mislabeled-so  rows whose robot_type claims SO but the dataset isn't a real 6-DOF SO arm
+                   (action_dim not a multiple of 6, or a classification note saying so).
+
+Only folders actually present in the destination repo are touched. Dry-run by default; pass --yes
+to perform the deletions.
+
+  python prune_destination.py --manifest /fsx/$USER/cdv3_manifests/manifest_*.csv \
+      --dst-repo lerobot/community_dataset_v3 --errored --mislabeled-so        # dry-run
+  python prune_destination.py --manifest manifest_*.csv --errored --mislabeled-so --yes
+"""
+import argparse
+import sys
+import time
+
+import pandas as pd
+from huggingface_hub import HfApi
+
+from classify import is_so_robot_type
+
+DST_REPO = "lerobot/community_dataset_v3"
+
+
+def _is_errored(row) -> bool:
+    return str(row.get("action") or "").strip().upper().startswith("ERROR")
+
+
+def _is_mislabeled_so(row) -> bool:
+    if "claims SO but" in str(row.get("note") or ""):
+        return True
+    if not is_so_robot_type(str(row.get("robot_type") or "")):
+        return False
+    try:
+        return int(float(row["action_dim"])) % 6 != 0
+    except (TypeError, ValueError, KeyError):
+        return False
+
+
+def select(df: pd.DataFrame, present: set[str], errored: bool, mislabeled: bool) -> dict[str, str]:
+    """Map each dataset root that is present in the repo AND matches an enabled filter to a reason.
+    'errored' takes precedence over 'mislabeled-so' when a row matches both."""
+    out: dict[str, str] = {}
+    for _, row in df.iterrows():
+        root = str(row.get("root") or "").strip()
+        if not root or root not in present or root in out:
+            continue
+        if errored and _is_errored(row):
+            out[root] = "errored"
+        elif mislabeled and _is_mislabeled_so(row):
+            out[root] = "mislabeled-so"
+    return out
+
+
+def main():
+    ap = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--manifest", nargs="+", required=True, metavar="CSV",
+                    help="One or more manifest CSVs from the migration run (per-rank files ok).")
+    ap.add_argument("--dst-repo", default=DST_REPO, metavar="ORG/NAME",
+                    help="Destination HF dataset repo to prune.")
+    ap.add_argument("--errored", action="store_true",
+                    help="Delete datasets that errored during migration.")
+    ap.add_argument("--mislabeled-so", action="store_true",
+                    help="Delete datasets labelled SO but not a real 6-DOF SO arm.")
+    ap.add_argument("--yes", action="store_true",
+                    help="Actually delete. Without it, only prints what would be deleted (dry-run).")
+    ap.add_argument("--list-missing", action="store_true",
+                    help="Report datasets in the manifest that are NOT present in the destination "
+                         "repo (grouped by their migration action), then exit without deleting.")
+    ap.add_argument("--report", action="store_true",
+                    help="List all three categories (errored, mislabeled-so, missing) without "
+                         "deleting anything, then exit. '*' marks datasets absent from the repo.")
+    args = ap.parse_args()
+    if not (args.errored or args.mislabeled_so or args.list_missing or args.report):
+        ap.error("enable at least one of: --errored, --mislabeled-so, --list-missing, --report")
+
+    df = pd.concat([pd.read_csv(p) for p in args.manifest], ignore_index=True)
+    if "root" not in df.columns:
+        ap.error("manifest has no 'root' column; is this a run_migration.py manifest?")
+    df = df.drop_duplicates(subset="root", keep="last")
+
+    api = HfApi()
+    present = {p[: -len("/meta/info.json")]
+               for p in api.list_repo_files(args.dst_repo, repo_type="dataset")
+               if p.endswith("/meta/info.json")}
+
+    if args.report or args.list_missing:
+        def _dump(title, sub):
+            print(f"== {title} ({len(sub)}) ==")
+            for root in sorted(sub):
+                print(f"  {' ' if root in present else '*'} {root}")
+            print()
+        missing = set(df.loc[~df["root"].isin(present), "root"])
+        if args.report:
+            _dump("errored", set(df.loc[df.apply(_is_errored, axis=1), "root"]))
+            _dump("mislabeled-so", set(df.loc[df.apply(_is_mislabeled_so, axis=1), "root"]))
+        _dump("missing from repo", missing)
+        print(f"{df['root'].nunique()} unique manifest rows, {len(present)} datasets in "
+              f"{args.dst_repo}. '*' = absent from repo.", file=sys.stderr)
+        return
+
+    to_delete = select(df, present, args.errored, args.mislabeled_so)
+    for root, reason in sorted(to_delete.items()):
+        print(f"{reason:14s} {root}")
+    print(f"\n{len(to_delete)} dataset(s) present in {args.dst_repo} match "
+          f"({df['root'].nunique()} rows in manifest, {len(present)} datasets in repo).", file=sys.stderr)
+
+    if not args.yes:
+        print("dry-run: nothing deleted. re-run with --yes to delete.", file=sys.stderr)
+        return
+    failed = []
+    for root, reason in sorted(to_delete.items()):
+        for attempt in range(1, 4):  # transient Hub ReadTimeouts are common; retry with backoff
+            try:
+                api.delete_folder(path_in_repo=root, repo_id=args.dst_repo, repo_type="dataset",
+                                  commit_message=f"Prune {root} ({reason})")
+                print(f"deleted {root} ({reason})", file=sys.stderr)
+                break
+            except Exception as e:
+                if attempt == 3:
+                    failed.append(root)
+                    print(f"FAILED {root}: {e}", file=sys.stderr)
+                else:
+                    time.sleep(2 ** attempt)
+    if failed:
+        print(f"\n{len(failed)} deletion(s) failed (likely transient); safe to re-run to retry: "
+              f"{failed}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

community_v3_migration/run_migration.py

@@ -0,0 +1,300 @@
+"""End-to-end migration of the community_dataset_v3 monorepo to v3.0 + SO-arm degrees.
+
+For each `{user}/{dataset}` sub-dataset: stream-download it, fix SO-arm joint values
+(if applicable), run the stock v2.1->v3.0 structural converter locally, upload the v3.0
+result under the same path into a NEW repo, then delete the local copy. Resumable.
+
+  uv run python run_migration.py --dst-repo HuggingFaceVLA/community_dataset_v3_degrees \
+      --work-dir /big/disk/cdv3_work --manifest manifest.csv
+Flags: --only-classify (just write manifest), --no-push (fix+convert locally, keep output,
+       no upload), --folder-name A [B ...] (target specific dataset folders), --limit N.
+Uncalibrated `normalized` datasets keep their normalized joint units (flagged APPROXIMATE on
+the card); paste fitted CANON ranges in so_arm_frame.py to convert them to degrees instead.
+"""
+import argparse, csv, json, shutil, sys, traceback
+from pathlib import Path
+from huggingface_hub import HfApi
+
+import so_arm_frame
+from classify import classify, is_end_effector, load_info
+from fix_dataset import (
+    data_video_episode_mismatch,
+    fix_dataset_in_place,
+    reconcile_episode_count,
+    reindex_episodes,
+)
+
+SRC_REPO = "HuggingFaceVLA/community_dataset_v3"
+NORMALIZED_TAG = "normalized"  # card tag for datasets whose SO joints stay in normalized units
+
+
+def download_subfolder(sub: str, work_dir: str, patterns: list[str] | None = None, repo: str = SRC_REPO) -> None:
+    """Download only ``{repo}/{sub}/...`` into ``work_dir``.
+
+    ``snapshot_download`` walks the entire repo tree (``list_repo_tree(recursive=True)``
+    with no path scope) before applying ``allow_patterns``. On this 791-dataset monorepo
+    that whole-repo enumeration is pathologically slow and looks like a hang. Listing the
+    scoped ``path_in_repo=sub`` subtree and fetching its files directly avoids it.
+    """
+    from fnmatch import fnmatch
+
+    from huggingface_hub import hf_hub_download
+    from huggingface_hub.hf_api import RepoFile
+
+    api = HfApi()
+    for entry in api.list_repo_tree(repo, path_in_repo=sub, repo_type="dataset", recursive=True):
+        if not isinstance(entry, RepoFile):
+            continue
+        if patterns and not any(fnmatch(entry.path, pat) for pat in patterns):
+            continue
+        hf_hub_download(repo, filename=entry.path, repo_type="dataset", local_dir=work_dir)
+
+
+def list_datasets(api: HfApi, repo: str) -> list[str]:
+    files = api.list_repo_files(repo, repo_type="dataset")
+    roots = {p[: -len("/meta/info.json")] for p in files if p.endswith("/meta/info.json")}
+    return sorted(roots)
+
+
+def resolve_folders(api: HfApi, repo: str, names: list[str]) -> list[str]:
+    """Expand each --folder-name into concrete dataset roots (folders that contain
+    meta/info.json). A name may be a full dataset path (returned as-is) or a namespace/prefix
+    like 'Beegbrain' (expanded to every dataset beneath it). Unknown names pass through so
+    they surface as a clear per-item error instead of a confusing FileNotFoundError."""
+    out: list[str] = []
+    for name in names:
+        name = name.strip("/")
+        try:
+            paths = [e.path for e in api.list_repo_tree(
+                repo, path_in_repo=name, recursive=True, repo_type="dataset")]
+        except Exception:
+            out.append(name)  # let it fail loudly downstream
+            continue
+        roots = sorted({p[: -len("/meta/info.json")] for p in paths if p.endswith("/meta/info.json")})
+        out.extend(roots or [name])
+    seen: set[str] = set()
+    return [r for r in out if not (r in seen or seen.add(r))]
+
+
+def already_done(api: HfApi, dst: str, sub: str, dst_files: set[str]) -> bool:
+    return f"{sub}/meta/info.json" in dst_files  # present in target => skip (resume)
+
+
+def _write_dataset_card(local: Path, sub: str, result: dict, standalone: bool = False) -> None:
+    """Regenerate the sub-dataset's card the way LeRobot does (create_lerobot_dataset_card
+    from meta/info.json), then append a migration section documenting provenance and the
+    joint-encoding fix. When ``standalone`` is set, ``sub`` is itself the source dataset's HF
+    repo id (rather than a folder inside the ``SRC_REPO`` monorepo)."""
+    enc = result.get("encoding")
+    converted_degrees = bool(result.get("converted"))
+    approx = enc == "normalized" and not so_arm_frame.CANON_IS_CALIBRATED
+
+    enc_labels = {
+        "degrees_old": "legacy degrees (old community frame, pre-#777 convention)",
+        "degrees_new": "degrees (recorded with `use_degrees=True`)",
+        "normalized": "normalized units (joints -100..100, gripper 0..100)",
+        "radians": "radians",
+        "unknown": "undetermined",
+    }
+    joint_actions = {
+        "degrees_old": "per-joint offsets and axis directions corrected to the post-#777 frame (values stay in degrees)",
+        "degrees_new": "already in the post-#777 degrees frame; values unchanged",
+        "normalized": ("un-normalized to physical degrees using calibrated joint ranges"
+                       if converted_degrees else
+                       "left in normalized units (-100..100 joints, 0..100 gripper); NOT converted to degrees"),
+        "radians": "left unchanged (already in radians)",
+        "unknown": "left unchanged (encoding could not be determined)",
+    }
+
+    lines = [
+        "## Migration to LeRobotDataset v3.0",
+        "",
+        "Migrated to LeRobotDataset **v3.0**"
+        + (" with SO-100/101 joint state/action mapped to the post-#777 physical frame (in degrees)."
+           if converted_degrees else "."),
+        "",
+        (f"- Source: [`{sub}`](https://huggingface.co/datasets/{sub})" if standalone else
+         f"- Source: [`{SRC_REPO}`](https://huggingface.co/datasets/{SRC_REPO}/tree/main/{sub}) (`{sub}`)"),
+        "- Codebase version: v2.1 -> v3.0",
+    ]
+    if result.get("is_so"):
+        lines += [
+            f"- Original joint encoding: {enc_labels.get(enc, enc)}",
+            f"- Joint values: {joint_actions.get(enc, 'left unchanged')}",
+            f"- Robot type: `{result.get('robot_type')}`",
+            f"- Action dimension: {result.get('action_dim')}",
+        ]
+    else:
+        lines += ["- Joint values: not applicable (not an SO-100/101 dataset)"]
+    if approx:
+        lines += ["", "> **Note:** per-robot calibration was unavailable, so joint state/action were "
+                  "left in their original *normalized* units (-100..100 joints, 0..100 gripper) rather "
+                  "than converted to physical degrees. Treat these joint values as APPROXIMATE."]
+    if result.get("ambiguous"):
+        lines += ["", "> **Note:** joint-encoding detection was flagged ambiguous; conversion used the "
+                  "best-guess encoding above and may warrant manual review."]
+
+    section = "\n".join(lines) + "\n"
+
+    readme = local / "README.md"
+    try:
+        try:
+            from lerobot.datasets.utils import create_lerobot_dataset_card
+        except ImportError:
+            from lerobot.common.datasets.utils import create_lerobot_dataset_card
+
+        class _Info(dict):            # satisfies both the dict and .to_dict() card variants
+            def to_dict(self):
+                return dict(self)
+
+        rt = result.get("robot_type") or None
+        tags = [rt] if rt else []
+        if enc == "normalized" and not converted_degrees:
+            tags.append(NORMALIZED_TAG)
+        card = create_lerobot_dataset_card(
+            tags=tags or None,
+            dataset_info=_Info(load_info(local)),
+            license="apache-2.0",
+            repo_id=sub,
+        )
+        card.text = card.text.rstrip() + "\n\n" + section
+        card.save(str(readme))
+    except Exception:
+        # LeRobot card generator unavailable at runtime: keep the standalone migration note.
+        if readme.exists():
+            readme.write_text(readme.read_text().rstrip() + "\n\n" + section)
+        else:
+            readme.write_text(f"# {sub}\n\n" + section)
+
+
+def migrate_one(api, dst_repo, sub, work_dir, no_upload, src_repo: str = SRC_REPO,
+                standalone: bool = False) -> dict:
+    local = Path(work_dir) / sub
+    if local.parent.exists():
+        shutil.rmtree(local.parent, ignore_errors=True)  # clean any partial
+    if standalone:
+        # ``sub`` is a self-contained HF dataset repo (not a monorepo folder): pull it whole.
+        from huggingface_hub import snapshot_download
+        snapshot_download(repo_id=sub, repo_type="dataset", local_dir=str(local))
+    else:
+        download_subfolder(sub, work_dir, repo=src_repo)
+
+    info = load_info(local)
+    if info.get("codebase_version") != "v2.1":
+        return {"root": sub, "action": f"skipped: source codebase is {info.get('codebase_version')} (expected v2.1)"}
+    if is_end_effector(info):
+        return {"root": sub, "robot_type": info.get("robot_type"),
+                "action": "skipped: end-effector (task-space) dataset, out of scope"}
+    mismatch = data_video_episode_mismatch(local)
+    if mismatch:
+        return {"root": sub, "robot_type": info.get("robot_type"),
+                "action": f"skipped: {mismatch}"}
+
+    result = fix_dataset_in_place(local)          # SO-arm value fix (or structural_only)
+
+    reconciled = reconcile_episode_count(local)   # align stale meta counts to data+video files
+    if reconciled:
+        result["action"] = f"{result['action']}; {reconciled}"
+
+    reindexed = reindex_episodes(local)           # compact non-contiguous episode indices to 0..N-1
+    if reindexed:
+        result["action"] = f"{result['action']}; {reindexed}"
+
+    from lerobot.scripts.convert_dataset_v21_to_v30 import convert_dataset
+    convert_dataset(repo_id=sub, root=str(local), push_to_hub=False)  # v2.1 -> v3.0, in place
+
+    _write_dataset_card(local, sub, result, standalone=standalone)  # document conversion in the card
+
+    base = {k: result.get(k) for k in
+            ("robot_type", "is_so", "encoding", "action_dim", "maxabs", "ambiguous", "action")}
+    if no_upload:
+        # keep the converted output on disk for inspection; do NOT delete or push
+        base["action"] = f"{base['action']}; not pushed (kept locally at {local})"
+        return {"root": sub, **base}
+    api.upload_folder(repo_id=dst_repo, repo_type="dataset", folder_path=str(local),
+                      path_in_repo=sub, commit_message=f"Add {sub} (v3.0, {result['action']})")
+    shutil.rmtree(Path(work_dir) / sub.split("/")[0], ignore_errors=True)  # drop after successful push
+    return {"root": sub, **base}
+
+
+def main():
+    ap = argparse.ArgumentParser(
+        description="Migrate the HuggingFaceVLA/community_dataset_v3 monorepo to LeRobotDataset "
+                    "v3.0, converting SO-100/101 joint state/action to physical degrees along "
+                    "the way. Processes one sub-dataset at a time (download -> fix -> convert -> "
+                    "upload -> cleanup) and is resumable.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+    ap.add_argument("--dst-repo", default=None, metavar="ORG/NAME",
+                    help="Destination HF dataset repo to push the converted v3.0 datasets to "
+                         "(created if missing). Required unless --no-push or --only-classify.")
+    ap.add_argument("--work-dir", default="./cdv3_work", metavar="DIR",
+                    help="Local scratch directory used to download, convert, and (unless pushing) "
+                         "retain each sub-dataset. Only one dataset lives here at a time on a push run.")
+    ap.add_argument("--manifest", default="manifest.csv", metavar="CSV",
+                    help="CSV log appended to as datasets are processed (robot_type, detected "
+                         "encoding, action taken, errors). Reused across resumed runs.")
+    ap.add_argument("--limit", type=int, default=None, metavar="N",
+                    help="Process only the first N sub-datasets (alphabetical). Ignored when "
+                         "--folder-name is given. Useful for a quick end-to-end smoke test.")
+    ap.add_argument("--folder-name", nargs="+", default=None, metavar="USER/DATASET",
+                    help=f"One or more folders WITHIN the {SRC_REPO} monorepo to process. Either a "
+                         "full dataset path ('Beegbrain/draw_pixel_art') or a whole namespace "
+                         "('Beegbrain'), which expands to every dataset under it. Skips the full "
+                         "791-dataset listing.")
+    ap.add_argument("--only-classify", action="store_true",
+                    help="Detect each dataset's robot type and joint encoding and write the "
+                         "manifest, without downloading data, converting, or pushing. Run this "
+                         "first to review scope (especially rows flagged ambiguous=True).")
+    ap.add_argument("--no-push", action="store_true",
+                    help="Fix + convert locally but do NOT upload; the converted v3.0 output is "
+                         "kept under --work-dir for inspection instead of being deleted.")
+    args = ap.parse_args()
+    no_upload = args.no_push
+    if not no_upload and not args.only_classify and not args.dst_repo:
+        ap.error("--dst-repo is required unless --no-push or --only-classify is set.")
+
+    api = HfApi()
+    if args.folder_name:
+        subs = resolve_folders(api, SRC_REPO, args.folder_name)
+        print(f"targeting {len(subs)} sub-dataset(s): {', '.join(subs)}", file=sys.stderr)
+    else:
+        subs = list_datasets(api, SRC_REPO)
+        if args.limit:
+            subs = subs[: args.limit]
+        print(f"{len(subs)} sub-datasets found", file=sys.stderr)
+
+    if not args.only_classify and not no_upload:
+        api.create_repo(args.dst_repo, repo_type="dataset", exist_ok=True)
+    dst_files = set() if (args.only_classify or no_upload) else set(
+        api.list_repo_files(args.dst_repo, repo_type="dataset"))
+
+    first = not Path(args.manifest).exists()
+    with open(args.manifest, "a", newline="") as mf:
+        w = None
+        for i, sub in enumerate(subs):
+            try:
+                if args.only_classify:
+                    # classify without full download: fetch just the meta/ of this sub
+                    download_subfolder(sub, args.work_dir, patterns=[f"{sub}/meta/*"])
+                    row = {"root": sub, **classify(Path(args.work_dir) / sub)}
+                    shutil.rmtree(Path(args.work_dir) / sub.split("/")[0], ignore_errors=True)
+                elif not no_upload and already_done(api, args.dst_repo, sub, dst_files):
+                    row = {"root": sub, "action": "skipped: already present in destination repo"}
+                else:
+                    row = migrate_one(api, args.dst_repo, sub, args.work_dir, no_upload)
+            except Exception as e:
+                row = {"root": sub, "action": f"ERROR: {e}"}
+                traceback.print_exc()
+            if w is None:
+                w = csv.DictWriter(mf, fieldnames=sorted(
+                    {"root", "robot_type", "is_so", "encoding", "action_dim",
+                     "maxabs", "ambiguous", "action", "codebase_version", "note"}))
+                if first:
+                    w.writeheader()
+            w.writerow({k: row.get(k) for k in w.fieldnames})
+            mf.flush()
+            print(f"[{i+1}/{len(subs)}] {sub}: {row.get('action')}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

community_v3_migration/slurm_migrate.py

@@ -0,0 +1,285 @@
+"""SLURM-distributed driver for run_migration.py.
+
+Fans the ``community_dataset_v3`` -> v3.0 migration out across SLURM workers, mirroring the
+datatrove pattern used by ``examples/dataset/slurm_recompute_stats.py``. This is a *map-only*
+job: each worker owns a stride ``subs[rank::world_size]`` of the sub-datasets and runs the
+exact same per-dataset pipeline as ``run_migration.py`` (download -> fix -> v2.1->v3.0 convert
+-> upload -> cleanup). There is no aggregate step.
+
+Resume is twofold and free: (1) each worker skips any sub-dataset already present in the
+destination repo (``already_done``), and (2) datatrove skips ranks whose completion marker
+exists. Re-run the identical command to mop up failures.
+
+Example (numeric smoke test on one namespace, no SLURM):
+    python slurm_migrate.py --slurm 0 --workers 1 \
+        --dst-repo HuggingFaceVLA/community_dataset_v3_degrees \
+        --work-dir ./cdv3_work --manifest-dir ./cdv3_manifests \
+        --folder-name Beegbrain
+
+Full run on the cluster:
+    python slurm_migrate.py \
+        --dst-repo HuggingFaceVLA/community_dataset_v3_degrees \
+        --work-dir /fsx/$USER/cdv3_work \
+        --manifest-dir /fsx/$USER/cdv3_manifests \
+        --logs-dir /fsx/$USER/logs/cdv3_migrate \
+        --workers 64 --partition hopper-cpu --qos normal \
+        --cpus-per-task 4 --mem-per-cpu 4G \
+        --env-command "source /fsx/$USER/venvs/lerobot/bin/activate; export HF_TOKEN=<token>"
+
+IMPORTANT: workers must reach the internet (HF download + upload) and have a write-scoped
+HF token (HF_TOKEN) in --env-command. Keep --workers modest (many concurrent commits to one
+destination repo contend); rely on resume passes to clear transient upload failures.
+"""
+
+import argparse
+from pathlib import Path
+
+from datatrove.executor import LocalPipelineExecutor
+from datatrove.executor.slurm import SlurmPipelineExecutor
+from datatrove.pipeline.base import PipelineStep
+
+MIGRATION_DIR = str(Path(__file__).resolve().parent)
+
+
+class MigrateShard(PipelineStep):
+    """Each worker migrates its ``subs[rank::world_size]`` slice of sub-datasets."""
+
+    def __init__(
+        self,
+        subs,
+        dst_repo,
+        work_dir,
+        manifest_dir,
+        migration_dir,
+        no_push=False,
+        only_classify=False,
+        standalone=False,
+    ):
+        super().__init__()
+        self.subs = subs
+        self.dst_repo = dst_repo
+        self.work_dir = work_dir
+        self.manifest_dir = manifest_dir
+        self.migration_dir = migration_dir
+        self.no_push = no_push
+        self.only_classify = only_classify
+        self.standalone = standalone
+
+    def run(self, data=None, rank: int = 0, world_size: int = 1):
+        # Pickled onto the worker: keep self-contained. The migration package dir must be on
+        # sys.path so ``run_migration`` and its siblings (classify/fix_dataset/so_arm_frame)
+        # import.
+        import csv
+        import logging
+        import shutil
+        import sys
+        import time
+        import traceback
+        from pathlib import Path
+
+        if self.migration_dir not in sys.path:
+            sys.path.insert(0, self.migration_dir)
+
+        from classify import classify
+        from huggingface_hub import HfApi
+        from run_migration import already_done, download_subfolder, migrate_one
+
+        from lerobot.utils.utils import init_logging
+
+        init_logging()
+
+        my_subs = self.subs[rank::world_size]
+        if not my_subs:
+            logging.info(f"Rank {rank}: no sub-datasets assigned")
+            return
+        logging.info(f"Rank {rank}: {len(my_subs)} / {len(self.subs)} sub-datasets")
+
+        # Per-rank scratch and manifest so workers never collide (migrate_one wipes
+        # work_dir/<namespace> around each dataset).
+        work_dir = str(Path(self.work_dir) / f"rank_{rank:05d}")
+        Path(work_dir).mkdir(parents=True, exist_ok=True)
+        Path(self.manifest_dir).mkdir(parents=True, exist_ok=True)
+        manifest = Path(self.manifest_dir) / f"manifest_{rank:05d}.csv"
+
+        api = HfApi()
+        # Resume prefetch. A single transient Hub read timeout here must NOT kill the whole
+        # rank (and skip its entire dataset slice), so retry with backoff and, as a last
+        # resort, fall back to an empty set (already-present datasets are re-checked per item
+        # and, for --source molmoact, were already filtered out on the submit node).
+        dst_files: set = set()
+        if not self.only_classify and not self.no_push:
+            for attempt in range(5):
+                try:
+                    dst_files = set(api.list_repo_files(self.dst_repo, repo_type="dataset"))
+                    break
+                except Exception as e:
+                    if attempt == 4:
+                        logging.warning(f"Rank {rank}: could not list {self.dst_repo} after 5 "
+                                        f"tries ({e}); proceeding without a resume set.")
+                    else:
+                        time.sleep(5 * (attempt + 1))
+
+        fieldnames = sorted(
+            {"root", "robot_type", "is_so", "encoding", "action_dim", "maxabs", "ambiguous", "action"}
+        )
+        write_header = not manifest.exists()
+        with open(manifest, "a", newline="") as mf:
+            w = csv.DictWriter(mf, fieldnames=fieldnames)
+            if write_header:
+                w.writeheader()
+            for i, sub in enumerate(my_subs):
+                try:
+                    if self.only_classify:
+                        if self.standalone:
+                            from huggingface_hub import snapshot_download
+
+                            snapshot_download(repo_id=sub, repo_type="dataset",
+                                              local_dir=str(Path(work_dir) / sub),
+                                              allow_patterns=["meta/*"])
+                        else:
+                            download_subfolder(sub, work_dir, patterns=[f"{sub}/meta/*"])
+                        row = {"root": sub, **classify(Path(work_dir) / sub)}
+                        shutil.rmtree(Path(work_dir) / sub.split("/")[0], ignore_errors=True)
+                    elif not self.no_push and already_done(api, self.dst_repo, sub, dst_files):
+                        row = {"root": sub, "action": "skipped: already present in destination repo"}
+                    else:
+                        row = migrate_one(api, self.dst_repo, sub, work_dir, self.no_push,
+                                          standalone=self.standalone)
+                except Exception as e:
+                    row = {"root": sub, "action": f"ERROR: {e}"}
+                    traceback.print_exc()
+                w.writerow({k: row.get(k) for k in fieldnames})
+                mf.flush()
+                logging.info(f"Rank {rank} [{i + 1}/{len(my_subs)}] {sub}: {row.get('action')}")
+
+
+def _mem_gb(mem: str) -> int:
+    s = str(mem).strip().lower().rstrip("b").rstrip("g")
+    return int(float(s))
+
+
+def _make_executor(pipeline, logs_dir, job_name, slurm, workers, time, partition, cpus, mem, qos, env_command, venv_path):
+    kwargs = {"pipeline": pipeline, "logging_dir": str(Path(logs_dir) / job_name)}
+    if slurm:
+        kwargs.update(
+            {
+                "job_name": job_name,
+                "tasks": workers,
+                "workers": workers,
+                "time": time,
+                "partition": partition,
+                "cpus_per_task": cpus,
+                "mem_per_cpu_gb": _mem_gb(mem),
+                "sbatch_args": {},
+            }
+        )
+        if qos:
+            kwargs["qos"] = qos
+        if venv_path:
+            kwargs["venv_path"] = venv_path
+        if env_command:
+            kwargs["env_command"] = env_command
+        return SlurmPipelineExecutor(**kwargs)
+    kwargs.update({"tasks": workers, "workers": 1})
+    return LocalPipelineExecutor(**kwargs)
+
+
+def main():
+    import sys
+
+    if MIGRATION_DIR not in sys.path:
+        sys.path.insert(0, MIGRATION_DIR)
+    from huggingface_hub import HfApi
+    from migrate_molmoact import REFERENCE_REPO, list_molmoact_datasets, pending_datasets
+    from run_migration import SRC_REPO, list_datasets, resolve_folders
+
+    p = argparse.ArgumentParser(
+        description="SLURM-distributed migration to LeRobotDataset v3.0 (map-only). Source is "
+                    "either the community_dataset_v3 monorepo or the standalone datasets listed "
+                    "by allenai/MolmoAct2-SO100_101-Dataset.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    p.add_argument("--source", choices=("monorepo", "molmoact"), default="monorepo",
+                   help="'monorepo': HuggingFaceVLA/community_dataset_v3 subfolders. "
+                        "'molmoact': standalone datasets listed by allenai/MolmoAct2-SO100_101-Dataset.")
+    p.add_argument("--reference-repo", default=REFERENCE_REPO, metavar="ORG/NAME",
+                   help="(--source molmoact) Repo whose datasets are already migrated and skipped.")
+    p.add_argument("--dst-repo", default=None, metavar="ORG/NAME", help="Destination HF dataset repo.")
+    p.add_argument("--work-dir", default="./cdv3_work", help="Scratch root; each rank gets a subdir.")
+    p.add_argument("--manifest-dir", default="./cdv3_manifests", help="Per-rank manifest CSVs land here.")
+    p.add_argument("--logs-dir", type=Path, default=Path("logs"), help="datatrove logs dir.")
+    p.add_argument("--job-name", default="cdv3_migrate", help="SLURM job name.")
+    p.add_argument("--workers", type=int, default=64, help="Number of parallel SLURM tasks.")
+    p.add_argument("--slurm", type=int, default=1, help="1 = submit via SLURM; 0 = run locally.")
+    p.add_argument("--partition", default=None, help="SLURM partition, e.g. 'hopper-cpu'.")
+    p.add_argument("--qos", default=None, help="SLURM QoS, e.g. 'normal'.")
+    p.add_argument("--cpus-per-task", type=int, default=4, help="CPUs per SLURM task.")
+    p.add_argument("--mem-per-cpu", default="4G", help="Memory per CPU, e.g. '4G'.")
+    p.add_argument("--time", default="24:00:00", help="Wall-clock limit per task.")
+    p.add_argument("--venv-path", default=None, help="venv activate script sourced on each worker.")
+    p.add_argument("--env-command", default=None, help="Raw shell snippet run before python (export HF_TOKEN, etc.).")
+    p.add_argument("--folder-name", nargs="+", default=None, help="Target specific folders/namespaces instead of all.")
+    p.add_argument("--limit", type=int, default=None, help="Only the first N sub-datasets (ignored with --folder-name).")
+    p.add_argument("--only-classify", action="store_true", help="Only classify + write manifest; no convert/upload.")
+    p.add_argument("--no-push", action="store_true", help="Fix + convert locally, keep output, do not upload.")
+    args = p.parse_args()
+
+    if not args.no_push and not args.only_classify and not args.dst_repo:
+        p.error("--dst-repo is required unless --no-push or --only-classify is set.")
+
+    api = HfApi()
+    standalone = args.source == "molmoact"
+    if standalone:
+        all_ids = list_molmoact_datasets(api)
+        if args.folder_name:
+            wanted = {n.strip("/") for n in args.folder_name}
+            subs = [s for s in all_ids if s in wanted]
+        else:
+            subs = pending_datasets(api, all_ids, args.dst_repo, args.reference_repo,
+                                    args.no_push, args.only_classify)
+            if args.limit:
+                subs = subs[: args.limit]
+    elif args.folder_name:
+        subs = resolve_folders(api, SRC_REPO, args.folder_name)
+    else:
+        subs = list_datasets(api, SRC_REPO)
+        if args.limit:
+            subs = subs[: args.limit]
+    print(f"{len(subs)} sub-datasets targeted", file=sys.stderr)
+    if not subs:
+        p.error("no sub-datasets resolved")
+
+    # Create the destination repo once on the submit node so workers don't race on it.
+    if not args.only_classify and not args.no_push:
+        api.create_repo(args.dst_repo, repo_type="dataset", exist_ok=True)
+
+    executor = _make_executor(
+        pipeline=[
+            MigrateShard(
+                subs,
+                args.dst_repo,
+                args.work_dir,
+                args.manifest_dir,
+                MIGRATION_DIR,
+                no_push=args.no_push,
+                only_classify=args.only_classify,
+                standalone=standalone,
+            )
+        ],
+        logs_dir=args.logs_dir,
+        job_name=args.job_name,
+        slurm=args.slurm == 1,
+        workers=args.workers,
+        time=args.time,
+        partition=args.partition,
+        cpus=args.cpus_per_task,
+        mem=args.mem_per_cpu,
+        qos=args.qos,
+        env_command=args.env_command,
+        venv_path=args.venv_path,
+    )
+    executor.run()
+
+
+if __name__ == "__main__":
+    main()

community_v3_migration/so_arm_frame.py

@@ -0,0 +1,69 @@
+"""SO-100/101 joint-frame conversion to physical degrees (post-#777 convention).
+
+Two calibration-free branches + one that needs an assumed canonical range:
+
+  * degrees_old  (bare robot_type `so100`/`so101`, |vals|>~180): PR #3879 old->new
+                 convention (sign flip shoulder_lift, +90 deg shoulder_lift/elbow_flex).
+                 EXACT.
+  * degrees_new  (`*_follower` recorded with use_degrees=True, not saturated): already
+                 degrees. EXACT.
+  * normalized   (`*_follower`, -100..100 joints / 0..100 gripper, saturates at bounds):
+                 5 arm joints are mid-range-zero, only the SCALE is missing (per-robot
+                 range_min/max not stored) -> use assumed canonical spans below. APPROXIMATE.
+                 The gripper (0..100) is kept in its native frame, matching degrees_new.
+  * radians      -> untouched.
+
+Joint order per arm: shoulder_pan, shoulder_lift, elbow_flex, wrist_flex, wrist_roll, gripper.
+Bimanual (12-dim) tiles the 6-joint block twice.
+"""
+import numpy as np
+
+JOINT_ORDER = ["shoulder_pan", "shoulder_lift", "elbow_flex", "wrist_flex", "wrist_roll", "gripper"]
+
+# --- PR #3879 (degrees). old(community frame) <-> new(v3.0 / post-#777) frame. ---
+SIGNS = np.array([1.0, -1.0, 1.0, 1.0, 1.0, 1.0], dtype=np.float64)
+OFFSETS_DEG = np.array([0.0, 90.0, 90.0, 0.0, 0.0, 0.0], dtype=np.float64)
+
+# --- Canonical per-joint spans (DEGREES) used ONLY to invert the -100..100 normalization of
+#     the 5 arm joints (RANGE_M100_100) when per-robot calibration is unavailable: normalized
+#     +/-100 -> +/-HALF_RANGE. The gripper (RANGE_0_100) is left in its native 0..100 frame in
+#     every SO dataset, so it needs no canonical span. THESE ARE PLACEHOLDERS — run
+#     calibrate_canonical_ranges.py and paste the fitted values here before a production run. ---
+CANON_HALF_RANGE_DEG = np.array([100.0, 100.0, 100.0, 100.0, 100.0], dtype=np.float64)  # 5 arm joints
+CANON_IS_CALIBRATED = False  # flipped to True once you paste fitted values
+
+
+def _convert_arm(x: np.ndarray, encoding: str) -> np.ndarray:
+    """x: (..., 6) for a single SO arm -> degrees (..., 6)."""
+    x = np.asarray(x, dtype=np.float64)
+    if encoding == "radians":
+        return x
+    if encoding == "degrees_old":
+        return SIGNS * (x - OFFSETS_DEG)
+    if encoding == "degrees_new":
+        return x
+    if encoding == "normalized":
+        new_deg = np.array(x, dtype=np.float64)
+        new_deg[..., :5] = (x[..., :5] / 100.0) * CANON_HALF_RANGE_DEG
+        # gripper is RANGE_0_100 in every SO dataset (including use_degrees=True / degrees_new),
+        # so it is already frame-consistent and must be left untouched, not remapped to +/-deg.
+        return new_deg
+    raise ValueError(f"unknown encoding: {encoding!r}")
+
+
+def to_degrees(arr, encoding: str, n_joints_per_arm: int = 6) -> np.ndarray:
+    """arr: (..., D) with D a multiple of 6. Returns float32 degrees, same shape."""
+    arr = np.asarray(arr, dtype=np.float64)
+    d = arr.shape[-1]
+    if d % n_joints_per_arm != 0:
+        raise ValueError(f"action/state dim {d} is not a multiple of {n_joints_per_arm}")
+    if encoding == "normalized" and not CANON_IS_CALIBRATED:
+        raise RuntimeError(
+            "CANON ranges are placeholders. Run calibrate_canonical_ranges.py and set "
+            "CANON_* + CANON_IS_CALIBRATED=True before converting 'normalized' datasets to degrees."
+        )
+    out = np.empty_like(arr)
+    for a in range(d // n_joints_per_arm):
+        sl = slice(a * n_joints_per_arm, (a + 1) * n_joints_per_arm)
+        out[..., sl] = _convert_arm(arr[..., sl], encoding)
+    return out.astype(np.float32)

community_v3_migration/tag_normalized.py

@@ -0,0 +1,117 @@
+"""Tag the migrated (v3.0) datasets whose SO-arm joints are still in normalized units.
+
+Walks every `{user}/{dataset}` sub-dataset in the destination repo, re-classifies it from its
+v3.0 metadata (meta/info.json + meta/stats.json), and adds the `normalized` card tag to any whose
+joint state/action are in normalized units (-100..100 / 0..100) rather than physical degrees. These
+are the datasets run_migration.py left un-converted (uncalibrated -> APPROXIMATE). Idempotent:
+already-tagged datasets are skipped. Dry-run by default; pass --yes to actually push the edited card.
+
+  python tag_normalized.py --dst-repo lerobot/community_dataset_v3            # dry-run
+  python tag_normalized.py --dst-repo lerobot/community_dataset_v3 --yes
+"""
+import argparse
+import json
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+import numpy as np
+from huggingface_hub import DatasetCard, HfApi, hf_hub_download
+
+from classify import (
+    encoding_from_bounds,
+    is_end_effector,
+    is_mislabeled_so,
+    is_so_robot_type,
+    load_info,
+    so_joint_count,
+)
+from run_migration import NORMALIZED_TAG
+
+DST_REPO = "lerobot/community_dataset_v3"
+
+
+def is_normalized(root: Path) -> bool:
+    """True when the SO-arm joints of a v3.0 sub-dataset at ``root`` are still normalized."""
+    info = load_info(root)
+    rt = info.get("robot_type", "") or ""
+    if is_end_effector(info) or not is_so_robot_type(rt) or is_mislabeled_so(info):
+        return False
+    stats = json.loads((root / "meta" / "stats.json").read_text())
+    key = next((k for k in ("action", "observation.state") if k in stats), None)
+    if key is None:
+        return False
+    lo = np.asarray(stats[key]["min"], dtype=float)
+    hi = np.asarray(stats[key]["max"], dtype=float)
+    n = so_joint_count(info, key) or len(hi)
+    return encoding_from_bounds(lo[:n], hi[:n], rt)["encoding"] == "normalized"
+
+
+def main():
+    ap = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--dst-repo", default=DST_REPO, metavar="ORG/NAME",
+                    help="Destination HF dataset repo whose sub-datasets are inspected and tagged.")
+    ap.add_argument("--limit", type=int, default=None, metavar="N",
+                    help="Inspect only the first N sub-datasets (alphabetical). Useful for a smoke test.")
+    ap.add_argument("--yes", action="store_true",
+                    help="Actually push the tag. Without it, only prints what would be tagged (dry-run).")
+    args = ap.parse_args()
+
+    api = HfApi()
+    subs = sorted(p[: -len("/meta/info.json")]
+                  for p in api.list_repo_files(args.dst_repo, repo_type="dataset")
+                  if p.endswith("/meta/info.json"))
+    if args.limit:
+        subs = subs[: args.limit]
+    print(f"{len(subs)} sub-datasets in {args.dst_repo}", file=sys.stderr)
+
+    tagged, already, failed = [], [], []
+    for i, sub in enumerate(subs):
+        try:
+            with tempfile.TemporaryDirectory() as tmp:
+                for f in ("meta/info.json", "meta/stats.json"):
+                    hf_hub_download(args.dst_repo, f"{sub}/{f}", repo_type="dataset", local_dir=tmp)
+                if not is_normalized(Path(tmp) / sub):
+                    continue
+                readme = hf_hub_download(args.dst_repo, f"{sub}/README.md", repo_type="dataset", local_dir=tmp)
+                card = DatasetCard.load(readme)
+                card_tags = list(card.data.tags or [])
+                if NORMALIZED_TAG in card_tags:
+                    already.append(sub)
+                    continue
+                if not args.yes:
+                    tagged.append(sub)
+                    continue
+                card.data.tags = card_tags + [NORMALIZED_TAG]
+                card.save(readme)
+                for attempt in range(1, 4):  # transient Hub ReadTimeouts are common; retry w/ backoff
+                    try:
+                        api.upload_file(path_or_fileobj=readme, path_in_repo=f"{sub}/README.md",
+                                        repo_id=args.dst_repo, repo_type="dataset",
+                                        commit_message=f"Tag {sub} '{NORMALIZED_TAG}'")
+                        tagged.append(sub)
+                        break
+                    except Exception as e:
+                        if attempt == 3:
+                            failed.append(sub)
+                            print(f"FAILED {sub}: {e}", file=sys.stderr)
+                        else:
+                            time.sleep(2 ** attempt)
+        except Exception as e:
+            failed.append(sub)
+            print(f"ERROR {sub}: {e}", file=sys.stderr)
+        print(f"[{i + 1}/{len(subs)}] {sub}", file=sys.stderr)
+
+    verb = "tagged" if args.yes else "would tag"
+    for sub in tagged:
+        print(f"{verb}: {sub}")
+    print(f"\n{len(tagged)} {verb} '{NORMALIZED_TAG}', {len(already)} already tagged, "
+          f"{len(failed)} failed.", file=sys.stderr)
+    if not args.yes and tagged:
+        print("dry-run: nothing pushed. re-run with --yes to apply.", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

docs/source/annotation_pipeline.mdx

@@ -81,6 +81,12 @@ merged. Both prompts also carry a causal **event-boundary** definition (a
 new event starts when an object becomes held / is released / reaches a new
 location / a lid changes state / contents move) to sharpen where cuts land.
 
+Optionally, a third **seeded-relabel** pass (`--plan.subtask_seeded_relabel`)
+revisits each span with its previous/current/next segment contact sheets and
+minimally corrects the label, using the first label as a prior — it keeps the
+boundaries fixed and only sharpens wording, at the cost of one extra call per
+subtask.
+
 The resulting spans are then stitched into a gap-free, full-episode
 cover, so **every frame has exactly one active subtask**. See
 [`run_hf_job.py`](https://github.com/huggingface/lerobot/blob/main/examples/annotations/run_hf_job.py)
@@ -157,30 +163,33 @@ Every module is on by default and can be toggled independently (set to
 
 ### The VLM (`--vlm.*`)
 
-| Flag                       | Default            | What it does                                                                        |
+| Flag                       | Default            | What it does                                                                         |
-| -------------------------- | ------------------ | ----------------------------------------------------------------------------------- |
+| -------------------------- | ------------------ | ------------------------------------------------------------------------------------ |
-| `--vlm.model_id`           | `Qwen/Qwen3.6-27B` | The model to serve and prompt.                                                      |
+| `--vlm.model_id`           | `Qwen/Qwen3.6-27B` | The model to serve and prompt.                                                       |
-| `--vlm.camera_key`         | first `images.*`   | Which camera every prompt is grounded on.                                           |
+| `--vlm.camera_key`         | first `images.*`   | Which camera every prompt is grounded on.                                            |
-| `--vlm.serve_command`      | auto               | The exact `vllm serve …` command (set TP size, GPU memory, `--max-model-len` here). |
+| `--vlm.serve_command`      | auto               | The exact `vllm serve …` command (set TP size, GPU memory, `--max-model-len` here).  |
-| `--vlm.parallel_servers`   | `1`                | Independent servers for round-robin routing (one per GPU).                          |
+| `--vlm.parallel_servers`   | `1`                | Independent servers for round-robin routing (one per GPU).                           |
-| `--vlm.num_gpus`           | `0`                | GPUs per server (`0` = one each).                                                   |
+| `--vlm.num_gpus`           | `0`                | GPUs per server (`0` = one each).                                                    |
-| `--vlm.client_concurrency` | `16`               | In-flight requests across all servers.                                              |
+| `--vlm.client_concurrency` | `16`               | In-flight requests across all servers.                                               |
-| `--vlm.max_new_tokens`     | `512`              | Generation cap per call.                                                            |
+| `--vlm.max_new_tokens`     | `512`              | Generation cap per call.                                                             |
-| `--vlm.temperature`        | `0.2`              | Sampling temperature.                                                               |
+| `--vlm.temperature`        | `0.2`              | Sampling temperature.                                                                |
+| `--vlm.reasoning_effort`   | `null`             | Thinking-budget hint (`low`/`medium`/`high`) forwarded to OpenAI-compatible servers. |
 
 ### Subtasks / plan / memory (`--plan.*`)
 
-| Flag                            | Default    | What it does                                                                                                              |
+| Flag                            | Default    | What it does                                                                                                                 |
-| ------------------------------- | ---------- | ------------------------------------------------------------------------------------------------------------------------- |
+| ------------------------------- | ---------- | ---------------------------------------------------------------------------------------------------------------------------- |
-| `--plan.frames_per_second`      | `2.0`      | Frame sampling rate for the contact sheets (`2.0` = one frame every 0.5s).                                                |
+| `--plan.frames_per_second`      | `2.0`      | Frame sampling rate for the contact sheets (`2.0` = one frame every 0.5s).                                                   |
-| `--plan.max_frames_per_prompt`  | `60`       | Frame budget per VLM call. Episodes whose sampling exceeds this are auto-windowed at the same density, then stitched.     |
+| `--plan.max_frames_per_prompt`  | `60`       | Frame budget per VLM call. Episodes whose sampling exceeds this are auto-windowed at the same density, then stitched.        |
-| `--plan.contact_sheet_columns`  | `5`        | Columns per contact-sheet grid (`contact_sheet_frames_per_sheet` tiles, time row-major).                                  |
+| `--plan.contact_sheet_columns`  | `5`        | Columns per contact-sheet grid (`contact_sheet_frames_per_sheet` tiles, time row-major).                                     |
-| `--plan.plan_max_steps`         | `8`        | Upper bound on subtasks per episode.                                                                                      |
+| `--plan.plan_max_steps`         | `8`        | Upper bound on subtasks per episode.                                                                                         |
-| `--plan.subtask_describe_first` | `true`     | Run the describe→segment grounding pass (best subtask quality; +1 call/episode).                                          |
+| `--plan.subtask_describe_first` | `true`     | Run the describe→segment grounding pass (best subtask quality; +1 call/episode).                                             |
-| `--plan.emit_plan`              | `true`     | Emit the numbered `plan` rows (`false` = subtasks + memory only).                                                         |
+| `--plan.subtask_seeded_relabel` | `false`    | Second pass: re-label each subtask from its prev/current/next contact sheets, seeded with the first label (+1 call/subtask). |
-| `--plan.emit_memory`            | `true`     | Emit the `memory` rows (`false` = subtasks + plan only); symmetric to `emit_plan`.                                        |
+| `--plan.subtask_relabel_frames` | `5`        | Frames sampled uniformly per segment sheet in the relabel pass (only used when `subtask_seeded_relabel=true`).               |
-| `--plan.n_task_rephrasings`     | `10`       | How many `task_aug` rephrasings to emit (`0` disables).                                                                   |
+| `--plan.emit_plan`              | `true`     | Emit the numbered `plan` rows (`false` = subtasks + memory only).                                                            |
-| `--plan.derive_task_from_video` | `if_short` | Use the dataset task as-is (`off`), only when it's missing/short (`if_short`), or always re-derive from video (`always`). |
+| `--plan.emit_memory`            | `true`     | Emit the `memory` rows (`false` = subtasks + plan only); symmetric to `emit_plan`.                                           |
+| `--plan.n_task_rephrasings`     | `10`       | How many `task_aug` rephrasings to emit (`0` disables).                                                                      |
+| `--plan.derive_task_from_video` | `if_short` | Use the dataset task as-is (`off`), only when it's missing/short (`if_short`), or always re-derive from video (`always`).    |
 
 ### Interjections + VQA
 

src/lerobot/annotations/steerable_pipeline/config.py

@@ -65,6 +65,14 @@ class PlanConfig:
     # invented from the task text (+1 VLM call/episode).
     subtask_describe_first: bool = True
 
+    # Seeded relabeling: after segmentation, re-label each span with a focused
+    # pass that sees the previous / current / next segment contact sheets and
+    # minimally corrects the seed label (macrodata's best end-to-end labeling
+    # step). Costs +1 VLM call per subtask; off by default.
+    subtask_seeded_relabel: bool = False
+    # Frames sampled uniformly per segment sheet in the relabel pass.
+    subtask_relabel_frames: int = 5
+
     # Emit ``style="plan"`` rows at each boundary; False = subtasks + memory only.
     emit_plan: bool = True
 
@@ -160,6 +168,11 @@ class VlmConfig:
     # Forwarded as extra_body.chat_template_kwargs (e.g. {"enable_thinking": false}).
     chat_template_kwargs: dict[str, Any] | None = None
 
+    # OpenAI-style thinking budget hint ("low"/"medium"/"high"); forwarded to
+    # the server when set. Used to cap a thinking model's reasoning so it
+    # leaves tokens for the actual JSON answer on OpenAI-compatible endpoints.
+    reasoning_effort: str | None = None
+
 
 @dataclass
 class ExecutorConfig:

src/lerobot/annotations/steerable_pipeline/frames.py

@@ -413,7 +413,16 @@ def _draw_timestamp_badge(image: PIL.Image.Image, timestamp: float) -> PIL.Image
 
     result = image.copy()
     draw = ImageDraw.Draw(result)
-    font = ImageFont.load_default()
+    # Scale the timestamp to the tile so it stays legible after the model
+    # downsamples the full sheet into 768px tiles — a tiny bitmap font blurs
+    # at contact-sheet resolution and the VLM can no longer read the exact
+    # source time, which is what the boundary score depends on. ``size=`` is
+    # supported by Pillow's bitmap default since 10.1; fall back otherwise.
+    badge_px = max(14, round(image.height * 0.12))
+    try:
+        font = ImageFont.load_default(size=badge_px)
+    except TypeError:
+        font = ImageFont.load_default()
     label = f"{timestamp:06.2f}s"
     left, top, right, bottom = draw.textbbox((0, 0), label, font=font)
     text_w, text_h = right - left, bottom - top

src/lerobot/annotations/steerable_pipeline/modules/plan_subtasks_memory.py

@@ -116,6 +116,8 @@ class PlanSubtasksMemoryModule:
             rows.extend(self._task_aug_rows([effective_task, *variants], t0))
 
         subtask_spans = self._generate_subtasks(record, task=effective_task)
+        if self.config.subtask_seeded_relabel and subtask_spans:
+            subtask_spans = self._seeded_relabel(record, subtask_spans, effective_task)
 
         # subtask rows
         for span in subtask_spans:
@@ -509,6 +511,51 @@ class PlanSubtasksMemoryModule:
 
         return cleaned
 
+    def _seeded_relabel(
+        self, record: EpisodeRecord, spans: list[dict[str, Any]], task: str
+    ) -> list[dict[str, Any]]:
+        """Re-label each span using prev/current/next segment contact sheets.
+
+        Boundaries are kept fixed; only ``text`` is refined. The original
+        ("seed") label is passed as a strong prior so the model verifies and
+        minimally corrects it rather than re-describing from scratch — the
+        macrodata seeded-relabeling step. One VLM call per span.
+        """
+        n = len(spans)
+        out: list[dict[str, Any]] = []
+        for i, span in enumerate(spans):
+            content: list[dict[str, Any]] = []
+            if i > 0:
+                content += self._segment_sheet(record, spans[i - 1])
+            content += self._segment_sheet(record, span)
+            if i < n - 1:
+                content += self._segment_sheet(record, spans[i + 1])
+            prompt = load_prompt("plan_subtask_relabel").format(
+                episode_task=task,
+                seed_label=span["text"],
+                segment_index=i + 1,
+                segment_count=n,
+                start=float(span["start"]),
+                end=float(span["end"]),
+            )
+            content.append({"type": "text", "text": prompt})
+            label = self._vlm_field([{"role": "user", "content": content}], "label")
+            text = label.strip() if isinstance(label, str) and label.strip() else span["text"]
+            out.append({**span, "text": text})
+        return out
+
+    def _segment_sheet(self, record: EpisodeRecord, span: dict[str, Any]) -> list[dict[str, Any]]:
+        """Contact-sheet block(s) for one span: up to N frames sampled uniformly."""
+        s, e = float(span["start"]), float(span["end"])
+        n = max(1, int(self.config.subtask_relabel_frames))
+        if e <= s or n == 1:
+            timestamps = [s]
+        else:
+            step = (e - s) / (n - 1)
+            timestamps = [s + i * step for i in range(n)]
+        frames = self.frame_provider.frames_at(record, timestamps)
+        return self._contact_sheet_blocks(frames, timestamps[: len(frames)])
+
     def _generate_subtasks_windowed(
         self, record: EpisodeRecord, task: str, window_s: float
     ) -> list[dict[str, Any]]:

src/lerobot/annotations/steerable_pipeline/prompts/__init__.py

@@ -22,12 +22,23 @@ plain editors and roundtrip cleanly through ``ruff format``.
 
 from __future__ import annotations
 
+import os
 from pathlib import Path
 
 _DIR = Path(__file__).parent
 
 
 def load(name: str) -> str:
-    """Read prompt template ``name.txt`` from the ``prompts/`` directory."""
+    """Read prompt template ``name.txt`` from the ``prompts/`` directory.
+
+    A ``LEROBOT_PROMPT_OVERRIDE_<name>`` environment variable, when set to a
+    non-empty value, takes precedence over the packaged file. This lets prompt
+    search (e.g. GEPA) inject candidate templates into a remote job without
+    rebuilding the package; the override must keep the same ``{placeholder}``
+    fields the call site formats in.
+    """
+    override = os.environ.get(f"LEROBOT_PROMPT_OVERRIDE_{name}")
+    if override and override.strip():
+        return override
     path = _DIR / f"{name}.txt"
     return path.read_text(encoding="utf-8")

src/lerobot/annotations/steerable_pipeline/prompts/plan_subtask_relabel.txt

@@ -0,0 +1,35 @@
+Annotate one fixed segment from a longer robot demonstration.
+
+Return only JSON:
+  {{"label": "<short descriptive subtask label>"}}
+
+You are shown up to three timestamped contact sheets, in order:
+- The FIRST sheet is the PREVIOUS segment (context only); it may be absent.
+- The SECOND sheet is the CURRENT target segment.
+- The THIRD sheet is the NEXT segment (context only); it may be absent.
+Each tile has its timestamp (seconds, absolute video time) burned into its
+top-left corner.
+
+Episode instruction: "{episode_task}"
+Target segment: {segment_index} of {segment_count}
+Target time: {start:.2f}s to {end:.2f}s
+Original predicted label for this exact segment: "{seed_label}"
+
+Rules:
+- Label ONLY the current target segment (the second sheet). Use the
+  previous/next sheets only to disambiguate what changed.
+- Treat the original predicted label as a STRONG PRIOR, not ground truth:
+  verify it against the current segment and correct it minimally.
+- If it already names the right action and main object, keep it; only fix
+  grammar or add a clearly visible essential detail.
+- If it is vague but directionally correct, make it more specific.
+- If it describes the previous/next segment, the wrong action, wrong
+  object, wrong destination, or a wrong state change, replace it.
+- Do not describe the previous or next segment, and do not split, merge,
+  or move the fixed segment.
+- Do not introduce an action that is not clearly visible in the current
+  target segment.
+- Use one concise imperative phrase. Name the manipulated object and the
+  action / state change. Include source, destination, side, direction,
+  final placement, or opened/closed state when visible and central.
+- Do not mention timestamps, frame numbers, uncertainty, or intent.

src/lerobot/annotations/steerable_pipeline/prompts/plan_subtasks.txt

@@ -1,112 +1,68 @@
-You are labeling a teleoperated robot demonstration.
+You are annotating a teleoperated robot demonstration shown as
+timestamped contact sheets (each tile has its time in seconds burned
+into the top-left corner). The operator's goal was: "{episode_task}"
 
-The user originally asked: "{episode_task}"
+{observation_block}Reconstruct the sequence of COMPLETED manipulation events the robot
+performs, in chronological order. Output one segment per event with a
+[start, end] time in seconds and a short action label.
 
-You are shown the entire demonstration as a single video. Watch the
+GROUNDING — read first, it overrides everything below:
-whole clip, then segment it into a list of consecutive atomic subtasks
+- Label ONLY events you can SEE in the frames. The instruction is the
-the robot performs.
+  goal; the VIDEO is the ground truth for what actually happened.
+- Do NOT invent, anticipate, or pad steps that are not shown.
 
-{observation_block}GROUNDING — read this first, it overrides everything below:
+Granularity — segment by completed events, not by motion:
-- Label ONLY what the robot actually does in the video. Every subtask
+- Start a NEW segment whenever the world state changes: an object is
-  you emit must correspond to motion you can SEE in specific frames.
+  grasped, lifted, transported, placed, or released; a held object
-- Do NOT invent, anticipate, or pad. If the robot only does one thing
+  changes; a drawer/door/lid/container opens or closes; contents move
-  (e.g. it just navigates to a location and the clip ends), emit
+  between containers (poured); a tool starts or stops acting on a
-  EXACTLY ONE subtask. Many demonstrations are a single atomic skill.
+  surface. Watch the gripper open/close transitions — they usually mark
-- ``max_steps`` below is a hard CEILING, not a target. Emitting fewer
+  boundaries.
-  subtasks than the ceiling is not just allowed, it is expected for
+- Do NOT split approach, reach, grasp adjustment, small repositioning,
-  short / atomic demonstrations. One correct subtask is far better
+  hesitation, or retreat into their own segments. Fold each into the
-  than several invented ones.
+  event it belongs to (the approach is part of the pick; the retreat is
-- If the video does not clearly show the action implied by the task,
+  part of the place).
-  describe what you actually see — do NOT fabricate the task's steps
+- Do NOT merge separate completed events. Each distinct pick, place,
-  from the instruction text. The instruction tells you the goal; the
+  open, close, pour, push, wipe, or insert is its own segment, even when
-  VIDEO is the ground truth for what happened.
+  they repeat on different objects or locations.
+- Most segments last 2-10 seconds. Shorter segments are okay ONLY for
+  fast pick / place / open / close / release events. Never emit a
+  segment shorter than {min_subtask_seconds} seconds; merge a too-short
+  candidate into its neighbour instead.
+- Skip idle time, pure camera motion, and tiny hand jitter.
 
-Authoring rules — Hi Robot atom granularity, pi0.7-style short prompts:
+Labels — short imperative phrases:
+- One concise command naming the action and the manipulated object, e.g.
+  "pick up the red cup", "put the cup on the shelf", "open the top
+  drawer", "pour water into the glass", "insert the plug into the
+  socket".
+- Include source, destination, side, direction, or the final
+  open/closed state when it is visible and central to the event.
+- Prefer these verbs (extend only when none fits): pick up, put, place,
+  push, pull, turn, press, open, close, pour, insert, wipe, stack.
+  Disambiguate by what you SEE:
+    * STACK vs PUT: object placed ON TOP OF another object -> "stack".
+    * INSERT vs PUT: object pushed INTO a fitted slot/hole/socket -> "insert".
+    * PICK UP vs PUT (direction): gripper CLOSES and object moves WITH
+      the hand -> "pick up"; gripper OPENS and object stays -> "put".
+    * POUR vs PUT: source is tilted and contents flow -> "pour".
+- Use the exact object nouns implied by the task; stay consistent across
+  the episode (don't switch "cube" to "block").
+- Write imperative commands, never third person ("the robot ..."), and
+  drop articles/adverbs.
 
-- Each subtask = one COMPOSITE atomic skill the low-level policy can
+Timing:
-  execute end-to-end. A "skill" bundles its own approach motion with
+- Use the burned-in timestamps to set start and end. Boundaries should
-  its terminal action — do NOT split the approach off as its own
+  land on or near a printed time, and every [start, end] must lie within
-  subtask. The whole-arm policy already learns to reach as part of
+  [0.0, {episode_duration}] seconds, be non-overlapping, and cover the
-  every manipulation primitive.
+  episode in order.
-- Write each subtask as an IMPERATIVE COMMAND, starting with one of
+- Emit at most {max_steps} segments.
-  these verbs (extend only when none fits):
-    pick up <obj>           — approach + grasp + lift in one subtask
-    put <obj> on/in <loc>   — transport + release in one subtask
-    place <obj> on/in <loc> — synonym of "put"; pick one and stay consistent
-    push <obj>              — contact + linear shove
-    pull <obj>              — contact + linear retract
-    turn <knob/dial/handle> — rotary actuation
-    press <button>          — single-press contact
-    open <drawer/door/lid>  — full open motion
-    close <drawer/door/lid> — full close motion
-    pour <src> into <dst>   — tilt + flow
-    insert <obj> into <slot>— alignment + push-fit
-    go to <loc>             — ONLY when no grasp / actuation follows
-                             (e.g. a pure relocation between phases).
-                             If the next subtask grasps something at
-                             that location, drop "go to ..." and just
-                             write "pick up ..." instead.
-- Forbidden ultra-fine splits — the VLM is NOT allowed to emit these
-  as standalone subtasks; fold them into the parent composite:
-    "move to X"   → fold into "pick up X" (or whatever follows)
-    "reach for X" → fold into "pick up X"
-    "grasp X"     → fold into "pick up X"
-    "lift X"      → fold into "pick up X" (or "put X on Y" if it's
-                    the transport phase of a place)
-    "release X"   → fold into "put X on Y" (or "place X in Y")
-- Keep it SHORT — a verb phrase, not a sentence. Drop articles
-  ("the", "a") and adverbs ("carefully", "slowly"). Add a "how"
-  detail (which hand, which grasp point) ONLY when it is needed to
-  disambiguate. Every subtask must begin with one of the verbs
-  above (no leading nouns, no "then", no "first").
-- NEVER use third person. Never write "the robot", "the arm", "the
-  gripper moves", "it picks up" — the robot is implied. Command it,
-  do not describe it.
-- Use the exact object nouns from the task above. If the task says
-  "cube", every subtask says "cube" — never switch to "block". If it
-  says "box", never switch to "bin"/"container". Keep vocabulary
-  consistent across the whole episode.
-- Good: "pick up blue cube", "put blue cube in box", "open drawer",
-  "turn red knob", "press start button", "go to sink".
-- Bad: "move to blue cube" (approach as its own subtask — forbidden,
-  must be folded into "pick up blue cube"); "the robot arm moves
-  towards the blue cube" (third person, too long); "carefully pick
-  up the cube" (adverb, article); "release the yellow block"
-  ("block" when the task said "cube", and "release" must be folded
-  into a "put"/"place" subtask).
-- Subtasks are non-overlapping and cover the full episode in order.
-  Choose the cut points yourself based on what you see in the video
-  (gripper open/close events, contact, regrasps, transitions).
-- Each subtask spans at least {min_subtask_seconds} seconds. If a
-  candidate span would be shorter, merge it into its neighbour
-  rather than emitting it.
-- Do not exceed {max_steps} subtasks total. Fewer, larger composites
-  are preferred over many micro-steps.
-- Every subtask's [start_time, end_time] must lie within
-  [0.0, {episode_duration}] seconds.
-
-SPECIAL CASES — verb disambiguation (each rule is narrowly visual and
-fires ONLY on the spatial situation it names; it must not change how you
-label any other situation):
-- STACK vs PUT: if an object is placed ON TOP OF another specific object
-  (not on a flat table / shelf / counter), use "stack ... on ...", not
-  "put". "stack blue book on green book", NOT "put blue book on table".
-- INSERT vs PUT: if an object goes INTO a fitted slot / hole / socket /
-  receptacle (push-fit), use "insert ... into ...", not "put".
-- RETRIEVE/PICK-UP vs PUT (direction): watch the gripper. If it CLOSES
-  on the object and the object moves WITH the hand, it is "pick up" /
-  "retrieve" (object leaves its location). If the gripper OPENS and the
-  object stays where the hand left it, it is "put" / "place" (object
-  arrives at a location). Decide by which way the object moves, not by
-  where the hand ends up.
-- POUR vs PUT: only use "pour" when the source is tilted and contents
-  flow out; moving a full container without tilting is "put"/"place".
 
 Output strictly valid JSON of shape:
 
   {{
     "subtasks": [
-      {{"text": "<short imperative verb phrase>", "start": <float>, "end": <float>}},
+      {{"text": "<short imperative action label>", "start": <float>, "end": <float>}},
       ...
     ]
   }}

src/lerobot/annotations/steerable_pipeline/vlm_client.py

@@ -285,6 +285,8 @@ def _make_openai_client(config: VlmConfig) -> VlmClient:
             "max_tokens": max_tok,
             "temperature": temp,
         }
+        if config.reasoning_effort:
+            kwargs["reasoning_effort"] = config.reasoning_effort
         extra_body: dict[str, Any] = {}
         if send_mm_kwargs and mm_kwargs:
             extra_body["mm_processor_kwargs"] = {**mm_kwargs, "do_sample_frames": True}
@@ -296,7 +298,13 @@ def _make_openai_client(config: VlmConfig) -> VlmClient:
             chosen = clients[rr_counter["i"] % len(clients)]
             rr_counter["i"] += 1
         response = chosen.chat.completions.create(**kwargs)
-        return response.choices[0].message.content or ""
+        # Some OpenAI-compatible servers can return a choice with no message
+        # (safety filter, or a "thinking" model that spends the whole budget
+        # before emitting content). Treat that as an empty reply so the
+        # JSON-retry path handles it instead of crashing the run.
+        choice = response.choices[0] if response.choices else None
+        message = choice.message if choice is not None else None
+        return (message.content if message is not None else None) or ""
 
     def _gen(batch: Sequence[Sequence[dict[str, Any]]], max_tok: int, temp: float) -> list[str]:
         if len(batch) <= 1 or config.client_concurrency <= 1:

src/lerobot/datasets/video_utils.py

@@ -729,6 +729,7 @@ def concatenate_video_files(
             stream_map[input_stream.index].time_base = input_stream.time_base
 
     # Demux + remux packets (no re-encode)
+    last_dts: dict[int, int] = {}
     for packet in input_container.demux():
         # Skip packets from un-mapped streams
         if packet.stream.index not in stream_map:
@@ -738,6 +739,19 @@ def concatenate_video_files(
         if packet.dts is None:
             continue
 
+        # Enforce strictly increasing DTS. Clips encoded with B-frames start at a negative DTS, so the
+        # concat demuxer can emit the first packet of a clip with a DTS equal to the last of the previous
+        # one; the MP4 muxer rejects duplicate/decreasing DTS with [Errno 22]. Nudge such packets forward.
+        prev_dts = last_dts.get(packet.stream.index)
+        if prev_dts is not None and packet.dts <= prev_dts:
+            shift = prev_dts + 1 - packet.dts
+            packet.dts += shift
+            if packet.pts is not None:
+                packet.pts += shift
+        if packet.pts is not None and packet.pts < packet.dts:
+            packet.pts = packet.dts
+        last_dts[packet.stream.index] = packet.dts
+
         output_stream = stream_map[packet.stream.index]
         packet.stream = output_stream
         output_container.mux(packet)

tests/annotations/test_modules.py

@@ -85,7 +85,7 @@ def _spy_responder(captured: list[list[dict[str, Any]]], reply: Any):
 def test_module1_plan_memory_subtask_smoke(fixture_dataset_root: Path, tmp_path: Path) -> None:
     vlm = make_canned_responder(
         {
-            "atomic subtasks": {
+            "COMPLETED manipulation events": {
                 "subtasks": [
                     {"text": "grasp the handle of the sponge", "start": 0.0, "end": 0.4},
                     {"text": "wipe the counter from left to right", "start": 0.4, "end": 0.8},
@@ -126,7 +126,7 @@ def test_module1_emit_memory_false_skips_memory_keeps_subtasks_and_plan(
     leaving subtask + plan generation intact — symmetric to ``emit_plan``."""
     vlm = make_canned_responder(
         {
-            "atomic subtasks": {
+            "COMPLETED manipulation events": {
                 "subtasks": [
                     {"text": "grasp the handle of the sponge", "start": 0.0, "end": 0.4},
                     {"text": "wipe the counter from left to right", "start": 0.4, "end": 0.8},
@@ -318,7 +318,7 @@ def test_module1_attaches_contact_sheets_to_subtask_prompt(
                     return block.get("text", "")
         return ""
 
-    subtask_calls = [m for m in captured if "atomic subtasks" in _prompt_text(m)]
+    subtask_calls = [m for m in captured if "COMPLETED manipulation events" in _prompt_text(m)]
     assert len(subtask_calls) == 1, "expected exactly one subtask-prompt VLM call"
     content = subtask_calls[0][0]["content"]
     video_blocks = [b for b in content if isinstance(b, dict) and b.get("type") == "video"]