perf(datasets): bound memory of visual episode stats via batched streaming

Accumulate image/video per-episode stats into a RunningQuantileStats in
frame batches instead of materialising every sampled frame at once. Peak
memory is now bounded by one batch (frame_batch_size x C x H x W) regardless
of episode length, preventing OOM on long, high-resolution episodes. count
keeps the per-frame convention of get_feature_stats.

docs/source/annotation_pipeline.mdx

@@ -81,12 +81,6 @@ 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)
@@ -163,33 +157,30 @@ Every module is on by default and can be toggled independently (set to
 
 ### The VLM (`--vlm.*`)
 
-| Flag                       | Default            | What it does                                                                         |
-| -------------------------- | ------------------ | ------------------------------------------------------------------------------------ |
-| `--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.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.num_gpus`           | `0`                | GPUs per server (`0` = one each).                                                    |
-| `--vlm.client_concurrency` | `16`               | In-flight requests across all servers.                                               |
-| `--vlm.max_new_tokens`     | `512`              | Generation cap per call.                                                             |
-| `--vlm.temperature`        | `0.2`              | Sampling temperature.                                                                |
-| `--vlm.reasoning_effort`   | `null`             | Thinking-budget hint (`low`/`medium`/`high`) forwarded to OpenAI-compatible servers. |
+| Flag                       | Default            | What it does                                                                        |
+| -------------------------- | ------------------ | ----------------------------------------------------------------------------------- |
+| `--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.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.num_gpus`           | `0`                | GPUs per server (`0` = one each).                                                   |
+| `--vlm.client_concurrency` | `16`               | In-flight requests across all servers.                                              |
+| `--vlm.max_new_tokens`     | `512`              | Generation cap per call.                                                            |
+| `--vlm.temperature`        | `0.2`              | Sampling temperature.                                                               |
 
 ### Subtasks / plan / memory (`--plan.*`)
 
-| 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.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.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_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.subtask_relabel_frames` | `5`        | Frames sampled uniformly per segment sheet in the relabel pass (only used when `subtask_seeded_relabel=true`).               |
-| `--plan.emit_plan`              | `true`     | Emit the numbered `plan` rows (`false` = subtasks + memory only).                                                            |
-| `--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`).    |
+| 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.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.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.emit_plan`              | `true`     | Emit the numbered `plan` rows (`false` = subtasks + memory only).                                                         |
+| `--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
 

docs/source/lingbot_va.mdx

@@ -111,26 +111,17 @@ Requirements:
 - The block-causal masks use PyTorch **flex-attention**, so build the policy with
   `--policy.attn_mode=flex` for training (the default `torch` SDPA is inference-only).
 - The full 5B DiT does not fit a single 24–32 GB GPU under AdamW; fine-tune with **LoRA**
-  (`--peft.method_type=LORA`) and/or optimizer offload. `get_optim_params` returns only the
-  trainable (e.g. adapter) parameters; the VAE + UMT5 text encoder stay frozen. Install the
-  `lerobot[peft]` extra to enable PEFT support (see the [PEFT training guide](./peft_training)).
+  (`--policy.use_peft=true`) and/or optimizer offload. `get_optim_params` returns only the
+  trainable (e.g. adapter) parameters; the VAE + UMT5 text encoder stay frozen.
 
 ```bash
 lerobot-train \
   --policy.path=lerobot/lingbot_va_libero_long --policy.attn_mode=flex \
-  --peft.method_type=LORA --peft.r=32 --peft.lora_alpha=32 \
-  --peft.target_modules='transformer\.blocks\.\d+\.attn[12]\.(to_q|to_v)' \
+  --policy.use_peft=true \
   --dataset.repo_id=<your LeRobot-format dataset> \
   --batch_size=1 --steps=... --output_dir=outputs/train/lingbot_va
 ```
 
-Unlike SmolVLA / π₀, LingBot-VA does not ship built-in default LoRA targets, so you must pass
-`--peft.target_modules` explicitly. Only `self.transformer` (the dual-stream Wan transformer) is
-trainable; the example above adapts the query/value projections of both its self-attention
-(`attn1`) and cross-attention (`attn2`) blocks — the standard LoRA target set. Broaden it (e.g.
-add `to_k`/`to_out`, or the `ffn` layers) if you need a higher-capacity adapter. Passing
-`--peft.method_type` implies PEFT, so `--policy.use_peft=true` is not required.
-
 The dataset must provide camera clips (a temporal window per camera, VAE-encoded to
 `frame_chunk_size` latent frames) and `frame_chunk_size * action_per_frame` action steps per item.
 

docs/source/peft_training.mdx

@@ -62,10 +62,3 @@ to the `--peft.full_training_modules` parameter:
 
 The learning rate and the scheduled target learning rate can usually be scaled by a factor of 10 compared to the
 learning rate used for full fine-tuning (e.g., 1e-4 normal, so 1e-3 using LoRA).
-
-## Other policies
-
-The same `--peft.*` flags work for any pre-trained policy. Some policies (SmolVLA, π₀, π₀.₅) ship
-built-in default `target_modules`, so `--peft.method_type=LORA` is enough. Others do not, and will
-ask you to pass `--peft.target_modules` explicitly — for example LingBot-VA, whose recommended
-targets are documented in its [dedicated guide](./lingbot_va#training--fine-tuning).

examples/annotations/run_hf_job.py

@@ -46,11 +46,8 @@ CMD = (
     "apt-get update -qq && apt-get install -y -qq git ffmpeg && "
     "pip install --no-deps "
     "'lerobot @ git+https://github.com/huggingface/lerobot.git@main' && "
-    # Pins mirror pyproject.toml — unpinned installs pull av 18 / datasets 5 /
-    # draccus 0.11, which break lerobot at import time.
     "pip install --upgrade-strategy only-if-needed "
-    "'datasets>=4.7.0,<5.0.0' 'pyarrow>=21.0.0,<30.0.0' 'av>=15.0.0,<16.0.0' 'draccus==0.10.0' "
-    "'pandas>=2.0.0,<3.0.0' jsonlines gymnasium torchcodec mergedeep pyyaml-include toml typing-inspect "
+    "datasets pyarrow av jsonlines draccus gymnasium torchcodec mergedeep pyyaml-include toml typing-inspect "
     "openai && "
     "export VLLM_MEMORY_PROFILER_ESTIMATE_CUDAGRAPHS=0 && "
     "export VLLM_VIDEO_BACKEND=pyav && "

examples/dataset/slurm_recompute_stats.py

@@ -0,0 +1,489 @@
+#!/usr/bin/env python
+
+# Copyright 2025 The HuggingFace Inc. team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+SLURM-distributed recomputation of a LeRobotDataset's ``meta/stats.json``.
+
+Modified copy of lerobot's examples/dataset/slurm_recompute_stats.py
+(feat/recompute-stats-readonly-and-visual branch) with cluster-friendly additions:
+
+  1. --qos              : pass a SLURM QoS through to every worker's sbatch.
+  2. --venv-path        : activate a venv on each worker before the python step.
+  3. --env-command      : raw shell snippet injected before the python step (e.g. to
+                          export HF_LEROBOT_HOME). Runs in addition to --venv-path.
+  4. --chain-aggregate  : submit ``aggregate`` with an afterok dependency on
+                          ``compute`` so it only runs once all shards exist
+                          (no manual squeue-wait, no gap/overlap race).
+  5. --update-episode-stats : in ``aggregate``, also rewrite the per-episode stats in the
+                          episodes parquet so they stay consistent with meta/stats.json
+                          (default: only stats.json is written).
+
+Data access: no filesystem mount. Point HF_LEROBOT_HOME at a node-visible shared
+cache (e.g. /fsx/$USER/.cache) so the dataset downloads once and all workers read
+it. This is the download route; the source dataset is fetched from the Hub on the
+CPU workers.
+
+IMPORTANT — how to run (do NOT sbatch this file):
+  Run it as a normal python process on the LOGIN node. datatrove submits the
+  workers for you. The reference copy (--new-root) is built on the login node and
+  references the shared HF cache, so /fsx must be visible there (it is).
+
+Requires: pip install 'lerobot[dataset]' datatrove
+
+Example (single command, compute then dependent aggregate):
+
+    export HF_LEROBOT_HOME=/fsx/$USER/.cache
+
+    python slurm_recompute_stats_patched.py compute \
+        --repo-id behavior-1k/2026-challenge-demos \
+        --new-root /fsx/$USER/behavior-1k_recomputed \
+        --shard-dir /fsx/$USER/behavior-1k_recomputed/stats_shards \
+        --logs-dir /fsx/$USER/logs/recompute \
+        --skip-image-video 0 \
+        --workers 250 \
+        --partition hopper-cpu \
+        --qos normal \
+        --cpus-per-task 8 --mem-per-cpu 4G \
+        --venv-path /fsx/$USER/venvs/lerobot/bin/activate \
+        --env-command 'export HF_LEROBOT_HOME=/fsx/'"$USER"'/.cache' \
+        --chain-aggregate
+
+REHEARSE FIRST with --workers 2 --skip-image-video 1 and inspect one worker's log
+under --logs-dir to confirm QoS was accepted and a numeric stats.json is written.
+"""
+
+import argparse
+from pathlib import Path
+
+from datatrove.executor import LocalPipelineExecutor
+from datatrove.executor.slurm import SlurmPipelineExecutor
+from datatrove.pipeline.base import PipelineStep
+
+class ComputeEpisodeStatsShards(PipelineStep):
+    """Each worker computes per-episode stats for its ``episodes[rank::world_size]`` shard."""
+
+    def __init__(self, repo_id, root, new_root, skip_image_video, shard_dir, video_backend=None):
+        super().__init__()
+        self.repo_id = repo_id
+        self.root = root
+        self.new_root = new_root
+        self.skip_image_video = skip_image_video
+        self.shard_dir = shard_dir
+        self.video_backend = video_backend
+
+    def run(self, data=None, rank: int = 0, world_size: int = 1):
+        # NOTE: this method is pickled and executed on a worker, where this script's module
+        # globals are NOT available. Keep it self-contained: import locally and don't reference
+        # module-level helpers/constants.
+        import logging
+        import pickle
+        from pathlib import Path
+
+        from lerobot.datasets import LeRobotDataset, compute_dataset_episode_stats
+        from lerobot.utils.utils import init_logging
+
+        init_logging()
+        load_kwargs = {"video_backend": self.video_backend} if self.video_backend else {}
+        root = self.new_root if self.new_root and Path(self.new_root).exists() else self.root
+        dataset = LeRobotDataset(self.repo_id, root=root, **load_kwargs)
+
+        my_episodes = list(range(dataset.meta.total_episodes))[rank::world_size]
+        if not my_episodes:
+            logging.info(f"Rank {rank}: no episodes assigned")
+            return
+        logging.info(f"Rank {rank}: {len(my_episodes)} / {dataset.meta.total_episodes} episodes")
+
+        episode_stats = compute_dataset_episode_stats(
+            dataset,
+            episode_indices=my_episodes,
+            skip_image_video=self.skip_image_video,
+        )
+
+        shard_dir = Path(self.shard_dir)
+        shard_dir.mkdir(parents=True, exist_ok=True)
+        out = shard_dir / f"episode_stats_{rank:05d}.pkl"
+        with open(out, "wb") as f:
+            pickle.dump(episode_stats, f)
+        logging.info(f"Rank {rank}: saved {len(episode_stats)} episode stats to {out}")
+
+
+class AggregateEpisodeStats(PipelineStep):
+    """Merge all per-episode stat shards into meta/stats.json."""
+
+    def __init__(
+        self,
+        repo_id,
+        root,
+        new_root,
+        shard_dir,
+        push_to_hub=False,
+        video_backend=None,
+        update_episode_stats=False,
+    ):
+        super().__init__()
+        self.repo_id = repo_id
+        self.root = root
+        self.new_root = new_root
+        self.shard_dir = shard_dir
+        self.push_to_hub = push_to_hub
+        self.video_backend = video_backend
+        self.update_episode_stats = update_episode_stats
+
+    def run(self, data=None, rank: int = 0, world_size: int = 1):
+        # NOTE: pickled and executed on a worker; keep self-contained (see ComputeEpisodeStatsShards.run).
+        import logging
+        import pickle
+        from pathlib import Path
+
+        from lerobot.datasets import LeRobotDataset, aggregate_episode_stats
+        from lerobot.utils.utils import init_logging
+
+        init_logging()
+        if rank != 0:
+            return
+
+        shard_dir = Path(self.shard_dir)
+        shards = sorted(shard_dir.glob("episode_stats_*.pkl"))
+        if not shards:
+            raise FileNotFoundError(f"No episode stat shards found in {shard_dir}")
+
+        # Shards map episode_index -> stats; merging by key makes a dropped shard show up as a
+        # missing episode and a re-run shard overwrite rather than double-count.
+        all_episode_stats = {}
+        for shard in shards:
+            with open(shard, "rb") as f:
+                all_episode_stats.update(pickle.load(f))
+        logging.info(f"Aggregating {len(all_episode_stats)} episode stats from {len(shards)} shards")
+
+        load_kwargs = {"video_backend": self.video_backend} if self.video_backend else {}
+        root = self.new_root if self.new_root and Path(self.new_root).exists() else self.root
+        dataset = LeRobotDataset(self.repo_id, root=root, **load_kwargs)
+
+        # Aggregation is order-independent, so the only way sharding changes the result is a
+        # gap (dropped shard) or an overlap (episode counted twice). Verify the shards cover
+        # every episode exactly once before writing stats.json.
+        expected_episodes = dataset.meta.total_episodes
+        if len(all_episode_stats) != expected_episodes:
+            raise ValueError(
+                f"Expected {expected_episodes} per-episode stats (one per episode) but got "
+                f"{len(all_episode_stats)} across {len(shards)} shards. A compute shard is likely "
+                "missing or was written more than once; re-run the failed shards before aggregating."
+            )
+
+        # Frame-count check catches the case where a duplicate and a gap cancel out in the
+        # episode count: summed per-episode frame counts must equal the dataset's total frames.
+        stats_values = list(all_episode_stats.values())
+        numeric_key = next(
+            (
+                k
+                for k, v in dataset.meta.features.items()
+                if v["dtype"] not in ("image", "video", "string") and stats_values and k in stats_values[0]
+            ),
+            None,
+        )
+        if numeric_key is not None:
+            total_frames = sum(int(s[numeric_key]["count"][0]) for s in stats_values)
+            if total_frames != dataset.meta.total_frames:
+                raise ValueError(
+                    f"Summed frame count from shards ({total_frames}) != dataset total_frames "
+                    f"({dataset.meta.total_frames}); episodes are double-counted or missing."
+                )
+
+        new_stats = aggregate_episode_stats(
+            dataset, all_episode_stats, update_episode_stats=self.update_episode_stats
+        )
+        if new_stats is None:
+            raise RuntimeError("Aggregation produced no stats")
+        logging.info(f"Wrote stats for features: {list(new_stats.keys())} to {dataset.root}")
+
+        if self.push_to_hub:
+            logging.info(f"Pushing {self.repo_id} to hub")
+            dataset.push_to_hub()
+
+
+def _mem_gb(mem: str) -> int:
+    """Parse '4G' / '4GB' / '4' into an int number of GB for datatrove's mem_per_cpu_gb."""
+    s = str(mem).strip().lower().rstrip("b").rstrip("g")
+    return int(float(s))
+
+
+def _make_executor(
+    pipeline,
+    logs_dir,
+    job_name,
+    slurm,
+    workers,
+    tasks,
+    time,
+    partition,
+    cpus,
+    mem,
+    qos=None,
+    env_command=None,
+    venv_path=None,
+    depends=None,
+):
+    kwargs = {"pipeline": pipeline, "logging_dir": str(Path(logs_dir) / job_name)}
+    if slurm:
+        kwargs.update(
+            {
+                "job_name": job_name,
+                "tasks": tasks,
+                "workers": workers,
+                "time": time,
+                "partition": partition,
+                "cpus_per_task": cpus,
+                "mem_per_cpu_gb": _mem_gb(mem),  # datatrove's native field (int GB)
+                "sbatch_args": {},
+            }
+        )
+        if qos:
+            kwargs["qos"] = qos  # -> "#SBATCH --qos=<qos>" on every worker
+        if venv_path:
+            kwargs["venv_path"] = venv_path  # datatrove sources this before the python step
+        if env_command:
+            kwargs["env_command"] = env_command  # extra raw snippet before python (composes with venv_path)
+        if depends is not None:
+            kwargs["depends"] = depends  # chains --dependency=afterok:<compute jobid>
+        return SlurmPipelineExecutor(**kwargs)
+    kwargs.update({"tasks": tasks, "workers": 1})
+    return LocalPipelineExecutor(**kwargs)
+
+
+def _maybe_reference_copy(repo_id, root, new_root, download_videos):
+    """Create the read-only-safe reference copy once, before submitting workers.
+
+    Loads metadata only (to resolve the source root and revision) instead of a full
+    ``LeRobotDataset``, which would also memory-map the entire frame index just to read a
+    path. Fetches the source into the shared cache so the copy's symlinks point at real
+    files and workers don't each re-download, pulling videos only when the run needs them
+    (i.e. when image/video stats are being recomputed).
+    """
+    if not new_root:
+        return
+    from huggingface_hub import snapshot_download
+
+    from lerobot.datasets.dataset_metadata import LeRobotDatasetMetadata
+    from lerobot.scripts.lerobot_edit_dataset import _reference_copy_dataset
+    from lerobot.utils.constants import HF_LEROBOT_HUB_CACHE
+
+    new_root_path = Path(new_root)
+    if new_root_path.exists():
+        return
+
+    meta = LeRobotDatasetMetadata(repo_id, root=Path(root) if root else None)
+    ignore_patterns = None if download_videos else "videos/"
+    if root:
+        snapshot_download(
+            repo_id,
+            repo_type="dataset",
+            revision=meta.revision,
+            local_dir=meta.root,
+            ignore_patterns=ignore_patterns,
+        )
+        src_root = Path(meta.root)
+    else:
+        src_root = Path(
+            snapshot_download(
+                repo_id,
+                repo_type="dataset",
+                revision=meta.revision,
+                cache_dir=HF_LEROBOT_HUB_CACHE,
+                ignore_patterns=ignore_patterns,
+            )
+        )
+    _reference_copy_dataset(src_root, new_root_path)
+
+
+def _add_shared_args(p):
+    p.add_argument("--repo-id", type=str, required=True, help="Dataset identifier, e.g. 'user/dataset'.")
+    p.add_argument("--root", type=str, default=None, help="Source dataset root (defaults to the Hub cache).")
+    p.add_argument(
+        "--new-root",
+        type=str,
+        default=None,
+        help="Writable output root; a read-only-safe reference copy of --root. If omitted, stats "
+        "are written in place at --root.",
+    )
+    p.add_argument("--shard-dir", type=Path, default=Path("stats_shards"), help="Per-rank shard dir.")
+    p.add_argument("--logs-dir", type=Path, default=Path("logs"), help="datatrove logs dir.")
+    p.add_argument("--job-name", type=str, default=None, help="SLURM job name.")
+    p.add_argument("--slurm", type=int, default=1, help="1 = submit via SLURM; 0 = run locally.")
+    p.add_argument("--partition", type=str, default=None, help="SLURM partition, e.g. 'hopper-cpu'.")
+    p.add_argument("--qos", type=str, default=None, help="SLURM QoS, e.g. 'normal'. Passed to every worker.")
+    p.add_argument("--cpus-per-task", type=int, default=4, help="CPUs per SLURM task.")
+    p.add_argument("--mem-per-cpu", type=str, default="4G", help="Memory per CPU, e.g. '4G'.")
+    p.add_argument(
+        "--video-backend",
+        type=str,
+        default=None,
+        help="Video decoding backend (e.g. 'pyav', 'torchcodec'). Defaults to the dataset's default; "
+        "use 'pyav' if torchcodec fails to load locally.",
+    )
+    p.add_argument("--venv-path", type=str, default=None, help="venv activate script sourced on each worker.")
+    p.add_argument(
+        "--env-command",
+        type=str,
+        default=None,
+        help="Raw shell snippet injected into each worker's sbatch before the python step "
+        "(e.g. to export HF_LEROBOT_HOME). Runs in addition to --venv-path.",
+    )
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="PATCHED SLURM-distributed LeRobotDataset stats recomputation",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    cp = sub.add_parser("compute", help="Distribute per-episode stats across SLURM workers.")
+    _add_shared_args(cp)
+    cp.add_argument("--workers", type=int, default=50, help="Number of parallel SLURM tasks.")
+    cp.add_argument(
+        "--skip-image-video",
+        type=int,
+        default=1,
+        help="1 = numeric features only (fast); 0 = also recompute image/video stats (decodes frames).",
+    )
+    cp.add_argument(
+        "--chain-aggregate",
+        action="store_true",
+        help="After building compute, submit aggregate with an afterok dependency (single command).",
+    )
+    cp.add_argument("--push-to-hub", action="store_true", help="For the chained aggregate: push after done.")
+    cp.add_argument(
+        "--update-episode-stats",
+        action="store_true",
+        help="For the chained aggregate: also rewrite per-episode stats in the episodes parquet.",
+    )
+
+    ap = sub.add_parser("aggregate", help="Merge shards into meta/stats.json.")
+    _add_shared_args(ap)
+    ap.add_argument("--push-to-hub", action="store_true", help="Push the dataset after aggregation.")
+    ap.add_argument(
+        "--update-episode-stats",
+        action="store_true",
+        help="Also rewrite per-episode stats in the episodes parquet to match stats.json.",
+    )
+    ap.add_argument(
+        "--depends-job-id",
+        type=str,
+        default=None,
+        help="Optional SLURM job id; aggregate waits for it (afterok) before running.",
+    )
+
+    args = parser.parse_args()
+    slurm = args.slurm == 1
+
+    if args.command == "compute":
+        # The reference copy (if any) is created once on the submitting node so workers
+        # can all load --new-root without racing to build it. Videos are only fetched when
+        # image/video stats are being recomputed.
+        _maybe_reference_copy(
+            args.repo_id, args.root, args.new_root, download_videos=not bool(args.skip_image_video)
+        )
+
+        compute_exec = _make_executor(
+            pipeline=[
+                ComputeEpisodeStatsShards(
+                    args.repo_id,
+                    args.root,
+                    args.new_root,
+                    bool(args.skip_image_video),
+                    str(args.shard_dir),
+                    args.video_backend,
+                )
+            ],
+            logs_dir=args.logs_dir,
+            job_name=args.job_name or "recompute_stats_compute",
+            slurm=slurm,
+            workers=args.workers,
+            tasks=args.workers,
+            time="24:00:00",
+            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,
+        )
+
+        if args.chain_aggregate and slurm:
+            # Build aggregate depending on compute. datatrove launches the dependency
+            # (compute) first, then submits aggregate with --dependency=afterok:<jobid>.
+            aggregate_exec = _make_executor(
+                pipeline=[
+                    AggregateEpisodeStats(
+                        args.repo_id,
+                        args.root,
+                        args.new_root,
+                        str(args.shard_dir),
+                        args.push_to_hub,
+                        args.video_backend,
+                        args.update_episode_stats,
+                    )
+                ],
+                logs_dir=args.logs_dir,
+                job_name="recompute_stats_aggregate",
+                slurm=slurm,
+                workers=1,
+                tasks=1,
+                time="02:00:00",
+                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,
+                depends=compute_exec,
+            )
+            aggregate_exec.run()
+        else:
+            compute_exec.run()
+    else:
+        aggregate_exec = _make_executor(
+            pipeline=[
+                AggregateEpisodeStats(
+                    args.repo_id,
+                    args.root,
+                    args.new_root,
+                    str(args.shard_dir),
+                    args.push_to_hub,
+                    args.video_backend,
+                    args.update_episode_stats,
+                )
+            ],
+            logs_dir=args.logs_dir,
+            job_name=args.job_name or "recompute_stats_aggregate",
+            slurm=slurm,
+            workers=1,
+            tasks=1,
+            time="02:00:00",
+            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,
+        )
+        if args.depends_job_id is not None:
+            aggregate_exec.depends_job_id = args.depends_job_id
+        aggregate_exec.run()
+
+
+if __name__ == "__main__":
+    main()

src/lerobot/annotations/steerable_pipeline/config.py

@@ -65,14 +65,6 @@ 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
 
@@ -168,11 +160,6 @@ 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,16 +413,7 @@ def _draw_timestamp_badge(image: PIL.Image.Image, timestamp: float) -> PIL.Image
 
     result = image.copy()
     draw = ImageDraw.Draw(result)
-    # 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()
+    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,8 +116,6 @@ 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:
@@ -511,51 +509,6 @@ 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,23 +22,12 @@ 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.
-
-    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
+    """Read prompt template ``name.txt`` from the ``prompts/`` directory."""
     path = _DIR / f"{name}.txt"
     return path.read_text(encoding="utf-8")

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

@@ -1,35 +0,0 @@
-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,68 +1,112 @@
-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}"
+You are labeling a teleoperated robot demonstration.
 
-{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.
+The user originally asked: "{episode_task}"
 
-GROUNDING — read first, it overrides everything below:
-- Label ONLY events you can SEE in the frames. The instruction is the
-  goal; the VIDEO is the ground truth for what actually happened.
-- Do NOT invent, anticipate, or pad steps that are not shown.
+You are shown the entire demonstration as a single video. Watch the
+whole clip, then segment it into a list of consecutive atomic subtasks
+the robot performs.
 
-Granularity — segment by completed events, not by motion:
-- Start a NEW segment whenever the world state changes: an object is
-  grasped, lifted, transported, placed, or released; a held object
-  changes; a drawer/door/lid/container opens or closes; contents move
-  between containers (poured); a tool starts or stops acting on a
-  surface. Watch the gripper open/close transitions — they usually mark
-  boundaries.
-- Do NOT split approach, reach, grasp adjustment, small repositioning,
-  hesitation, or retreat into their own segments. Fold each into the
-  event it belongs to (the approach is part of the pick; the retreat is
-  part of the place).
-- Do NOT merge separate completed events. Each distinct pick, place,
-  open, close, pour, push, wipe, or insert is its own segment, even when
-  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.
+{observation_block}GROUNDING — read this first, it overrides everything below:
+- Label ONLY what the robot actually does in the video. Every subtask
+  you emit must correspond to motion you can SEE in specific frames.
+- Do NOT invent, anticipate, or pad. If the robot only does one thing
+  (e.g. it just navigates to a location and the clip ends), emit
+  EXACTLY ONE subtask. Many demonstrations are a single atomic skill.
+- ``max_steps`` below is a hard CEILING, not a target. Emitting fewer
+  subtasks than the ceiling is not just allowed, it is expected for
+  short / atomic demonstrations. One correct subtask is far better
+  than several invented ones.
+- If the video does not clearly show the action implied by the task,
+  describe what you actually see — do NOT fabricate the task's steps
+  from the instruction text. The instruction tells you the goal; the
+  VIDEO is the ground truth for what happened.
 
-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.
+Authoring rules — Hi Robot atom granularity, pi0.7-style short prompts:
 
-Timing:
-- Use the burned-in timestamps to set start and end. Boundaries should
-  land on or near a printed time, and every [start, end] must lie within
-  [0.0, {episode_duration}] seconds, be non-overlapping, and cover the
-  episode in order.
-- Emit at most {max_steps} segments.
+- Each subtask = one COMPOSITE atomic skill the low-level policy can
+  execute end-to-end. A "skill" bundles its own approach motion with
+  its terminal action — do NOT split the approach off as its own
+  subtask. The whole-arm policy already learns to reach as part of
+  every manipulation primitive.
+- Write each subtask as an IMPERATIVE COMMAND, starting with one of
+  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 action label>", "start": <float>, "end": <float>}},
+      {{"text": "<short imperative verb phrase>", "start": <float>, "end": <float>}},
       ...
     ]
   }}

src/lerobot/annotations/steerable_pipeline/vlm_client.py

@@ -285,8 +285,6 @@ 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}
@@ -298,13 +296,7 @@ def _make_openai_client(config: VlmConfig) -> VlmClient:
             chosen = clients[rr_counter["i"] % len(clients)]
             rr_counter["i"] += 1
         response = chosen.chat.completions.create(**kwargs)
-        # 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 ""
+        return response.choices[0].message.content 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/configs/train.py

@@ -221,11 +221,6 @@ class TrainPipelineConfig(HubMixin):
             )
 
         active_cfg = self.trainable_config
-
-        # Keep the policy-level `use_peft` flag in sync with the presence of a `--peft.*` config.
-        if self.peft is not None and self.policy is not None:
-            self.policy.use_peft = True
-
         if self.rename_map and active_cfg.pretrained_path is None:
             raise ValueError(
                 "`rename_map` requires a pretrained policy checkpoint. "

src/lerobot/datasets/__init__.py

@@ -25,6 +25,8 @@ from .compute_stats import DEFAULT_QUANTILES, aggregate_stats, get_feature_stats
 from .dataset_metadata import CODEBASE_VERSION, LeRobotDatasetMetadata
 from .dataset_tools import (
     add_features,
+    aggregate_episode_stats,
+    compute_dataset_episode_stats,
     convert_image_to_video_dataset,
     delete_episodes,
     merge_datasets,
@@ -34,6 +36,7 @@ from .dataset_tools import (
     reencode_dataset,
     remove_feature,
     split_dataset,
+    write_episode_stats,
 )
 from .factory import make_dataset, make_train_eval_datasets, resolve_delta_timestamps
 from .image_writer import safe_stop_image_writer
@@ -78,8 +81,10 @@ __all__ = [
     "detect_available_encoders_pyav",
     "add_features",
     "aggregate_datasets",
+    "aggregate_episode_stats",
     "aggregate_pipeline_dataset_features",
     "aggregate_stats",
+    "compute_dataset_episode_stats",
     "convert_image_to_video_dataset",
     "create_initial_features",
     "compute_sampler_state",
@@ -99,5 +104,6 @@ __all__ = [
     "resolve_delta_timestamps",
     "safe_stop_image_writer",
     "split_dataset",
+    "write_episode_stats",
     "write_stats",
 ]

src/lerobot/datasets/dataset_tools.py

@@ -33,11 +33,13 @@ from pathlib import Path
 import datasets
 import numpy as np
 import pandas as pd
+import pyarrow as pa
 import pyarrow.parquet as pq
 import torch
 from tqdm import tqdm
 
 from lerobot.configs import (
+    DEFAULT_DEPTH_UNIT,
     DepthEncoderConfig,
     RGBEncoderConfig,
     VideoEncoderConfig,
@@ -51,11 +53,15 @@ from lerobot.utils.utils import flatten_dict
 
 from .aggregate import aggregate_datasets
 from .compute_stats import (
+    RunningQuantileStats,
     aggregate_stats,
+    auto_downsample_height_width,
     compute_episode_stats,
     compute_relative_action_stats,
+    sample_indices,
 )
 from .dataset_metadata import LeRobotDatasetMetadata
+from .depth_utils import dequantize_depth
 from .image_writer import write_image
 from .io_utils import (
     get_parquet_file_size_in_mb,
@@ -77,6 +83,7 @@ from .utils import (
     update_chunk_file_indices,
 )
 from .video_utils import (
+    decode_video_frames,
     encode_video_frames,
     reencode_video,
 )
@@ -1559,6 +1566,191 @@ def modify_tasks(
     return dataset
 
 
+def _load_episode_image_frames(
+    dataset: LeRobotDataset,
+    key: str,
+    ep_idx: int,
+    frame_offsets: list[int],
+    is_depth: bool,
+) -> np.ndarray:
+    """Load sampled frames of an image feature for one episode as a (N, C, H, W) array."""
+    ep = dataset.meta.episodes[ep_idx]
+    from_idx = ep["dataset_from_index"]
+    column = dataset.hf_dataset.with_format(None).select_columns(key)
+
+    frames = []
+    for offset in frame_offsets:
+        img = column[from_idx + offset][key]
+        if is_depth:
+            arr = np.array(img)
+            if arr.ndim == 2:
+                arr = arr[np.newaxis, ...]
+        else:
+            arr = np.transpose(np.array(img.convert("RGB"), dtype=np.uint8), (2, 0, 1))
+        frames.append(auto_downsample_height_width(arr))
+    return np.stack(frames)
+
+
+def _load_episode_video_frames(
+    dataset: LeRobotDataset,
+    key: str,
+    ep_idx: int,
+    frame_offsets: list[int],
+    is_depth: bool,
+) -> np.ndarray:
+    """Load sampled frames of a video feature for one episode as a (N, C, H, W) array."""
+    ep = dataset.meta.episodes[ep_idx]
+    video_path = dataset.root / dataset.meta.get_video_file_path(ep_idx, key)
+    from_timestamp = ep[f"videos/{key}/from_timestamp"]
+    timestamps = [from_timestamp + offset / dataset.meta.fps for offset in frame_offsets]
+
+    frames = decode_video_frames(
+        video_path,
+        timestamps,
+        dataset.tolerance_s,
+        backend=dataset._video_backend,
+        return_uint8=not is_depth,
+        is_depth=is_depth,
+    )
+    if is_depth:
+        # ``decode_video_frames`` returns raw 12-bit codec values; dequantize back to
+        # the recorded depth unit so stats match record-time stats (which are stored in
+        # ``info.depth_unit`` and only rescaled to the output unit on read).
+        info = dataset.meta.features[key].get("info") or {}
+        depth_encoder = DepthEncoderConfig.from_video_info(info)
+        frames = dequantize_depth(
+            frames,
+            depth_min=depth_encoder.depth_min,
+            depth_max=depth_encoder.depth_max,
+            shift=depth_encoder.shift,
+            use_log=depth_encoder.use_log,
+            output_unit=info.get("depth_unit") or DEFAULT_DEPTH_UNIT,
+        )
+    return np.stack([auto_downsample_height_width(frame) for frame in frames.numpy()])
+
+
+def _compute_visual_episode_stats(
+    dataset: LeRobotDataset,
+    ep_idx: int,
+    visual_keys: list[str],
+    frame_batch_size: int = 32,
+) -> dict:
+    """Compute per-episode statistics for image/video features by sampling frames.
+
+    Mirrors the image/video branch of :func:`compute_episode_stats`: per-channel stats
+    are computed on downsampled sampled frames, then RGB stats are rescaled to [0, 1]
+    (depth maps keep their native units).
+
+    Frames are decoded and accumulated into a :class:`RunningQuantileStats` in batches of
+    ``frame_batch_size`` rather than materialising every sampled frame at once. Peak memory
+    is bounded by one batch (``frame_batch_size x C x H x W``) regardless of episode length,
+    which keeps long, high-resolution episodes from exhausting memory.
+    """
+    ep_length = dataset.meta.episodes[ep_idx]["length"]
+    frame_offsets = sample_indices(ep_length)
+
+    ep_stats = {}
+    for key in visual_keys:
+        is_depth = key in dataset.meta.depth_keys
+        is_video = dataset.meta.features[key]["dtype"] == "video"
+
+        running = RunningQuantileStats()
+        for start in range(0, len(frame_offsets), frame_batch_size):
+            batch_offsets = frame_offsets[start : start + frame_batch_size]
+            if is_video:
+                frames = _load_episode_video_frames(dataset, key, ep_idx, batch_offsets, is_depth)
+            else:
+                frames = _load_episode_image_frames(dataset, key, ep_idx, batch_offsets, is_depth)
+            # (N, C, H, W) -> (N * H * W, C) so stats are accumulated per channel.
+            running.update(np.moveaxis(frames, 1, -1).reshape(-1, frames.shape[1]))
+
+        stats = running.get_statistics()
+        normalization_factor = 1.0 if is_depth else 255.0
+        num_channels = stats["mean"].shape[0]
+        # ``count`` follows the per-frame convention of ``get_feature_stats`` (number of
+        # sampled frames), not the per-pixel count tracked internally by RunningQuantileStats.
+        ep_stats[key] = {
+            k: np.array([len(frame_offsets)])
+            if k == "count"
+            else v.reshape(num_channels, 1, 1) / normalization_factor
+            for k, v in stats.items()
+        }
+
+    return ep_stats
+
+
+def compute_dataset_episode_stats(
+    dataset: LeRobotDataset,
+    episode_indices: list[int] | None = None,
+    skip_image_video: bool = True,
+    drop_keys: list[str] | None = None,
+) -> dict[int, dict]:
+    """Compute per-episode statistics for a subset of episodes.
+
+    This is the shardable unit of work behind :func:`recompute_stats`: distribute
+    ``episode_indices`` across workers (e.g. ``list(range(n))[rank::world_size]``),
+    then combine the results with :func:`aggregate_episode_stats`.
+
+    Args:
+        dataset: The LeRobotDataset to compute stats for.
+        episode_indices: Episodes to process. When ``None``, all episodes are processed.
+        skip_image_video: If True (default), only numeric features are computed. If False,
+            image/video stats are also computed by sampling and decoding frames.
+        drop_keys: Feature keys to exclude (e.g. ``action`` when it is computed separately
+            in relative-action space).
+
+    Returns:
+        A mapping of episode index to its per-episode stat dict. Keeping the episode index
+        (rather than a bare list) lets callers write the stats back to the correct episode
+        row, and survives sharding since shards can be merged by key.
+    """
+    features = dataset.meta.features
+    meta_keys = {"index", "episode_index", "task_index", "frame_index", "timestamp"}
+    drop = set(drop_keys or [])
+    features_to_compute = {
+        k: v
+        for k, v in features.items()
+        if v["dtype"] != "string"
+        and k not in meta_keys
+        and k not in drop
+        and (not skip_image_video or v["dtype"] not in ["image", "video"])
+    }
+    numeric_keys = [k for k, v in features_to_compute.items() if v["dtype"] not in ["image", "video"]]
+    visual_keys = [k for k, v in features_to_compute.items() if v["dtype"] in ["image", "video"]]
+
+    if dataset.meta.episodes is None:
+        dataset.meta.episodes = load_episodes(dataset.meta.root)
+
+    if episode_indices is None:
+        episode_indices = list(range(dataset.meta.total_episodes))
+
+    # Group requested episodes by their data parquet file so each file is read once.
+    file_to_episodes: dict[Path, list[int]] = {}
+    for ep_idx in episode_indices:
+        file_to_episodes.setdefault(dataset.meta.get_data_file_path(ep_idx), []).append(ep_idx)
+
+    all_episode_stats = {}
+    for src_path, eps in tqdm(sorted(file_to_episodes.items()), desc="Computing stats from data files"):
+        df = pd.read_parquet(dataset.root / src_path) if numeric_keys else None
+        for ep_idx in sorted(eps):
+            episode_data = {}
+            if numeric_keys:
+                ep_df = df[df["episode_index"] == ep_idx]
+                for key in numeric_keys:
+                    if key in ep_df.columns:
+                        values = ep_df[key].values
+                        episode_data[key] = (
+                            np.stack(values) if hasattr(values[0], "__len__") else np.array(values)
+                        )
+
+            ep_stats = compute_episode_stats(episode_data, features_to_compute)
+            if visual_keys:
+                ep_stats.update(_compute_visual_episode_stats(dataset, int(ep_idx), visual_keys))
+            all_episode_stats[int(ep_idx)] = ep_stats
+
+    return all_episode_stats
+
+
 def recompute_stats(
     dataset: LeRobotDataset,
     skip_image_video: bool = True,
@@ -1566,13 +1758,21 @@ def recompute_stats(
     relative_exclude_joints: list[str] | None = None,
     chunk_size: int = 50,
     num_workers: int = 0,
+    update_episode_stats: bool = False,
 ) -> LeRobotDataset:
     """Recompute stats.json from scratch by iterating all episodes.
 
     Args:
         dataset: The LeRobotDataset to recompute stats for.
         skip_image_video: If True (default), only recompute stats for numeric features
-            (action, state, etc.) and keep existing image/video stats unchanged.
+            (action, state, etc.) and keep existing image/video stats unchanged. If False,
+            image/video stats are also recomputed by sampling and decoding frames from each
+            episode (this reads the image/video files, unlike the numeric-only path).
+        update_episode_stats: If True, also rewrite the per-episode ``stats/*`` columns in the
+            episodes parquet files so they stay consistent with the aggregated ``stats.json``.
+            Defaults to False (only ``stats.json`` is rewritten). Requires a writable
+            ``dataset.root``. Note that relative-action stats are aggregate-only and are not
+            written per-episode.
         relative_action: If True, compute action stats in relative space by
             iterating all valid action chunks and subtracting the current state.
             This matches the normalization distribution the model sees during
@@ -1588,24 +1788,12 @@ def recompute_stats(
         The same dataset with updated stats.
     """
     features = dataset.meta.features
-    meta_keys = {"index", "episode_index", "task_index", "frame_index", "timestamp"}
-    numeric_features = {
-        k: v
-        for k, v in features.items()
-        if v["dtype"] not in ["image", "video", "string"] and k not in meta_keys
-    }
-
-    if skip_image_video:
-        features_to_compute = numeric_features
-    else:
-        features_to_compute = {
-            k: v for k, v in features.items() if v["dtype"] != "string" and k not in meta_keys
-        }
 
     # When relative_action is enabled, compute action stats via chunk-based sampling
     # (matching what the model sees during training) and skip action in the
     # per-episode pass below.
     relative_action_stats = None
+    drop_keys = None
     if relative_action and ACTION in features and OBS_STATE in features:
         if relative_exclude_joints is None:
             relative_exclude_joints = ["gripper"]
@@ -1616,56 +1804,105 @@ def recompute_stats(
             exclude_joints=relative_exclude_joints,
             num_workers=num_workers,
         )
-        features_to_compute.pop(ACTION, None)
+        drop_keys = [ACTION]
 
-    logging.info(f"Recomputing stats for features: {list(features_to_compute.keys())}")
+    all_episode_stats = compute_dataset_episode_stats(
+        dataset, skip_image_video=skip_image_video, drop_keys=drop_keys
+    )
 
-    data_dir = dataset.root / DATA_DIR
-    parquet_files = sorted(data_dir.glob("*/*.parquet"))
-    if not parquet_files:
-        raise ValueError(f"No parquet files found in {data_dir}")
-
-    all_episode_stats = []
-    # TODO: enable image and video stats re-computation
-    numeric_keys = [k for k, v in features_to_compute.items() if v["dtype"] not in ["image", "video"]]
-
-    for parquet_path in tqdm(parquet_files, desc="Computing stats from data files"):
-        df = pd.read_parquet(parquet_path)
-
-        for ep_idx in sorted(df["episode_index"].unique()):
-            ep_df = df[df["episode_index"] == ep_idx]
-            episode_data = {}
-            for key in numeric_keys:
-                if key in ep_df.columns:
-                    values = ep_df[key].values
-                    if hasattr(values[0], "__len__"):
-                        episode_data[key] = np.stack(values)
-                    else:
-                        episode_data[key] = np.array(values)
-
-            ep_stats = compute_episode_stats(episode_data, features_to_compute)
-            all_episode_stats.append(ep_stats)
-
-    if features_to_compute and not all_episode_stats:
+    new_stats = aggregate_episode_stats(
+        dataset,
+        all_episode_stats,
+        extra_stats={ACTION: relative_action_stats} if relative_action_stats else None,
+        update_episode_stats=update_episode_stats,
+    )
+    if new_stats is None:
         logging.warning("No episode stats computed")
-        return dataset
+    else:
+        logging.info("Stats recomputed successfully")
+    return dataset
 
-    new_stats = aggregate_stats(all_episode_stats) if all_episode_stats else {}
 
-    if relative_action_stats is not None:
-        new_stats[ACTION] = relative_action_stats
+def write_episode_stats(dataset: LeRobotDataset, episode_stats: dict[int, dict]) -> None:
+    """Overwrite the per-episode ``stats/*`` columns in the episodes parquet files in place.
 
-    # Merge: keep existing stats for features we didn't recompute
+    Only the features present in ``episode_stats[ep_idx]`` are rewritten; stats columns for
+    features that were not recomputed are left untouched. Every other episode column (tasks,
+    length, chunk/file indices, frame ranges, …) is preserved. ``dataset.root`` must be
+    writable (e.g. the reference copy created for read-only sources).
+    """
+    if not episode_stats:
+        return
+
+    meta = dataset.meta
+    if meta.episodes is None:
+        meta.episodes = load_episodes(meta.root)
+
+    # Group episodes by the parquet file that holds them so each file is rewritten once.
+    file_to_episodes: dict[tuple[int, int], list[int]] = {}
+    for ep_idx in episode_stats:
+        ep = meta.episodes[ep_idx]
+        key = (ep["meta/episodes/chunk_index"], ep["meta/episodes/file_index"])
+        file_to_episodes.setdefault(key, []).append(ep_idx)
+
+    for (chunk_idx, file_idx), eps in file_to_episodes.items():
+        path = meta.root / DEFAULT_EPISODES_PATH.format(chunk_index=chunk_idx, file_index=file_idx)
+        table = pq.read_table(path)
+        rows = table.to_pylist()
+        row_by_ep = {row["episode_index"]: row for row in rows}
+        for ep_idx in eps:
+            row = row_by_ep[ep_idx]
+            for feature, feature_stats in episode_stats[ep_idx].items():
+                for stat_name, value in feature_stats.items():
+                    col = f"stats/{feature}/{stat_name}"
+                    if col in row:
+                        row[col] = np.asarray(value).tolist()
+        # Reuse the source schema so the rewritten stats keep the exact on-disk types.
+        new_table = pa.Table.from_pylist(rows, schema=table.schema)
+        pq.write_table(new_table, path, compression="snappy", use_dictionary=True)
+
+
+def aggregate_episode_stats(
+    dataset: LeRobotDataset,
+    episode_stats: dict[int, dict],
+    extra_stats: dict | None = None,
+    update_episode_stats: bool = False,
+) -> dict | None:
+    """Aggregate per-episode stats, merge with existing stats, and write ``stats.json``.
+
+    Companion to :func:`compute_dataset_episode_stats` for the distributed workflow: pass the
+    merged ``{episode_index: stats}`` mapping of every worker's per-episode stats. ``extra_stats``
+    lets callers inject feature stats computed outside the per-episode pass (e.g. relative-action
+    stats).
+
+    Args:
+        dataset: The dataset whose ``meta/stats.json`` (and optionally episode stats) is updated.
+        episode_stats: Mapping of episode index to its per-episode stat dict.
+        extra_stats: Feature stats to inject into the aggregate (not written per-episode).
+        update_episode_stats: If True, also rewrite the per-episode ``stats/*`` columns in the
+            episodes parquet files via :func:`write_episode_stats`.
+
+    Returns the written stats dict, or ``None`` if there was nothing to aggregate.
+    """
+    if not episode_stats and not extra_stats:
+        return None
+
+    new_stats = aggregate_stats(list(episode_stats.values())) if episode_stats else {}
+    if extra_stats:
+        new_stats.update(extra_stats)
+
+    # Merge: keep existing stats for features we didn't recompute.
     if dataset.meta.stats:
         for key, value in dataset.meta.stats.items():
-            if key not in new_stats:
-                new_stats[key] = value
+            new_stats.setdefault(key, value)
 
     write_stats(new_stats, dataset.root)
     dataset.meta.stats = new_stats
 
-    logging.info("Stats recomputed successfully")
-    return dataset
+    if update_episode_stats:
+        write_episode_stats(dataset, episode_stats)
+
+    return new_stats
 
 
 def convert_image_to_video_dataset(

src/lerobot/policies/evo1/internvl3_embedder.py

@@ -27,11 +27,9 @@ from lerobot.utils.import_utils import _transformers_available, require_package
 
 if TYPE_CHECKING or _transformers_available:
     from transformers import AutoModel, AutoTokenizer
-    from transformers.utils import is_flash_attn_2_available
 else:
     AutoModel = None
     AutoTokenizer = None
-    is_flash_attn_2_available = None
 
 IMAGENET_MEAN = (0.485, 0.456, 0.406)
 IMAGENET_STD = (0.229, 0.224, 0.225)
@@ -137,13 +135,9 @@ class InternVL3Embedder(nn.Module):
                 raise ValueError(f"Unsupported EVO1 vlm_dtype '{model_dtype}'") from exc
         self.model_dtype = model_dtype
 
-        attn_implementation = (
-            "flash_attention_2" if (use_flash_attn and is_flash_attn_2_available()) else "eager"
-        )
+        attn_implementation = "flash_attention_2" if (use_flash_attn and _flash_attn_available()) else "eager"
         if use_flash_attn and attn_implementation == "eager":
-            logger.warning(
-                "Flash Attention 2 is unavailable on this runtime. Falling back to eager attention."
-            )
+            logger.warning("flash_attn is not installed. Falling back to eager attention.")
 
         self.model = AutoModel.from_pretrained(
             model_name,
@@ -365,3 +359,11 @@ class InternVL3Embedder(nn.Module):
     @property
     def device(self) -> torch.device:
         return next(self.model.parameters()).device
+
+
+def _flash_attn_available() -> bool:
+    try:
+        import flash_attn  # noqa: F401
+    except ModuleNotFoundError:
+        return False
+    return True

src/lerobot/policies/factory.py

@@ -513,37 +513,6 @@ def make_pre_post_processors(
     return processors
 
 
-def _has_peft_adapter_config(
-    pretrained_path: str,
-    revision: str | None = None,
-) -> bool:
-    """Return whether ``pretrained_path`` points to an existing PEFT adapter.
-
-    A PEFT adapter checkpoint always ships an ``adapter_config.json``.
-    A plain base-model checkpoint does not. This distinction lets us tell apart
-    two very different ``use_peft=True`` scenarios that both set ``pretrained_path``:
-
-    * loading/resuming a previously trained adapter (config lives at ``pretrained_path``)
-    * starting a *fresh* PEFT fine-tune on top of a base model
-
-    Works for both local directories and Hub repo ids.
-    """
-    import os
-
-    adapter_config_name = "adapter_config.json"
-
-    if os.path.isdir(pretrained_path):
-        return os.path.isfile(os.path.join(pretrained_path, adapter_config_name))
-
-    from huggingface_hub import file_exists
-    from huggingface_hub.errors import HfHubHTTPError
-
-    try:
-        return file_exists(pretrained_path, adapter_config_name, revision=revision)
-    except (HfHubHTTPError, OSError):
-        return False
-
-
 def make_policy(
     cfg: PreTrainedConfig,
     ds_meta: LeRobotDatasetMetadata | None = None,
@@ -638,24 +607,13 @@ def make_policy(
             "the PEFT config parameters to be set. For training with PEFT, see `lerobot_train.py` on how to do that."
         )
 
-    # When `use_peft=True` and a checkpoint is given, the checkpoint can be one of two things:
-    # 1. A base model checkpoint (e.g., a pretrained policy) on which we want to start a fresh PEFT fine-tune.
-    # 2. A PEFT adapter checkpoint (e.g., a previously trained PEFT adapter)
-    # We distinguish between these two cases
-    load_existing_adapter = (
-        cfg.pretrained_path
-        and cfg.use_peft
-        and _has_peft_adapter_config(str(cfg.pretrained_path), cfg.pretrained_revision)
-    )
-
-    if cfg.pretrained_path and not load_existing_adapter:
+    if cfg.pretrained_path and not cfg.use_peft:
         # Load a pretrained policy and override the config if needed (for example, if there are inference-time
-        # hyperparameters that we want to vary). This also covers starting a fresh PEFT fine-tune on top of a
-        # base model: the base weights are loaded here and `wrap_with_peft` builds the adapter afterwards.
+        # hyperparameters that we want to vary).
         kwargs["pretrained_name_or_path"] = cfg.pretrained_path
         kwargs["revision"] = cfg.pretrained_revision
         policy = policy_cls.from_pretrained(**kwargs)
-    elif load_existing_adapter:
+    elif cfg.pretrained_path and cfg.use_peft:
         # Load a pretrained PEFT model on top of the policy. The pretrained path points to the folder/repo
         # of the adapter and the adapter's config contains the path to the base policy. So we need the
         # adapter config first, then load the correct policy and then apply PEFT.

src/lerobot/policies/pretrained.py

@@ -23,6 +23,8 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING, TypedDict, TypeVar, Unpack
 
+import packaging
+import safetensors
 from huggingface_hub import HfApi, ModelCard, ModelCardData, hf_hub_download, save_torch_state_dict
 from huggingface_hub.constants import SAFETENSORS_SINGLE_FILE
 from huggingface_hub.errors import HfHubHTTPError
@@ -32,7 +34,6 @@ from torch import Tensor, nn
 from lerobot.__version__ import __version__
 from lerobot.configs import PreTrainedConfig
 from lerobot.configs.train import TrainPipelineConfig
-from lerobot.utils.device_utils import resolve_safetensors_device
 from lerobot.utils.hub import HubMixin
 
 from .utils import log_model_loading_keys
@@ -220,10 +221,26 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
 
     @classmethod
     def _load_as_safetensor(cls, model: T, model_file: str, map_location: str, strict: bool) -> T:
-        missing_keys, unexpected_keys = load_model_as_safetensor(
-            model, model_file, strict=strict, device=resolve_safetensors_device(map_location)
-        )
+        # Create base kwargs
+        kwargs = {"strict": strict}
+
+        # Add device parameter for newer versions that support it
+        if packaging.version.parse(safetensors.__version__) >= packaging.version.parse("0.4.3"):
+            kwargs["device"] = map_location
+
+        # Load the model with appropriate kwargs
+        missing_keys, unexpected_keys = load_model_as_safetensor(model, model_file, **kwargs)
         log_model_loading_keys(missing_keys, unexpected_keys)
+
+        # For older versions, manually move to device if needed
+        if "device" not in kwargs and map_location != "cpu":
+            logging.warning(
+                "Loading model weights on other devices than 'cpu' is not supported natively in your version of safetensors."
+                " This means that the model is loaded on 'cpu' first and then copied to the device."
+                " This leads to a slower loading time."
+                " Please update safetensors to version 0.4.3 or above for improved performance."
+            )
+            model.to(map_location)
         return model
 
     @abc.abstractmethod

src/lerobot/rewards/pretrained.py

@@ -21,6 +21,8 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING, Any, TypeVar
 
+import packaging
+import safetensors
 from huggingface_hub import HfApi, ModelCard, ModelCardData, hf_hub_download
 from huggingface_hub.constants import SAFETENSORS_SINGLE_FILE
 from huggingface_hub.errors import HfHubHTTPError
@@ -28,7 +30,6 @@ from safetensors.torch import load_model as load_model_as_safetensor, save_model
 from torch import Tensor, nn
 
 from lerobot.configs.rewards import RewardModelConfig
-from lerobot.utils.device_utils import resolve_safetensors_device
 from lerobot.utils.hub import HubMixin
 
 if TYPE_CHECKING:
@@ -128,13 +129,29 @@ class PreTrainedRewardModel(nn.Module, HubMixin, abc.ABC):
 
     @classmethod
     def _load_as_safetensor(cls, model: T, model_file: str, map_location: str, strict: bool) -> T:
-        missing_keys, unexpected_keys = load_model_as_safetensor(
-            model, model_file, strict=strict, device=resolve_safetensors_device(map_location)
-        )
+        # Create base kwargs
+        kwargs = {"strict": strict}
+
+        # Add device parameter for newer versions that support it
+        if packaging.version.parse(safetensors.__version__) >= packaging.version.parse("0.4.3"):
+            kwargs["device"] = map_location
+
+        # Load the model with appropriate kwargs
+        missing_keys, unexpected_keys = load_model_as_safetensor(model, model_file, **kwargs)
         if missing_keys:
             logging.warning(f"Missing key(s) when loading model: {missing_keys}")
         if unexpected_keys:
             logging.warning(f"Unexpected key(s) when loading model: {unexpected_keys}")
+
+        # For older versions, manually move to device if needed
+        if "device" not in kwargs and map_location != "cpu":
+            logging.warning(
+                "Loading model weights on other devices than 'cpu' is not supported natively in your version of safetensors."
+                " This means that the model is loaded on 'cpu' first and then copied to the device."
+                " This leads to a slower loading time."
+                " Please update safetensors to version 0.4.3 or above for improved performance."
+            )
+            model.to(map_location)
         return model
 
     def get_optim_params(self):

src/lerobot/scripts/lerobot_annotate.py

@@ -28,12 +28,7 @@ For distributed runs, see ``examples/annotations/run_hf_job.py``.
 """
 
 import logging
-from contextlib import suppress
 from pathlib import Path
-from typing import TYPE_CHECKING
-
-from huggingface_hub import HfApi, snapshot_download
-from huggingface_hub.errors import RevisionNotFoundError
 
 from lerobot.annotations.steerable_pipeline.config import AnnotationPipelineConfig
 from lerobot.annotations.steerable_pipeline.executor import Executor
@@ -47,12 +42,6 @@ from lerobot.annotations.steerable_pipeline.validator import StagingValidator
 from lerobot.annotations.steerable_pipeline.vlm_client import make_vlm_client
 from lerobot.annotations.steerable_pipeline.writer import LanguageColumnsWriter
 from lerobot.configs import parser
-from lerobot.utils.import_utils import _datasets_available, require_package
-
-if TYPE_CHECKING or _datasets_available:
-    from lerobot.datasets.dataset_metadata import CODEBASE_VERSION
-    from lerobot.datasets.io_utils import load_info
-    from lerobot.datasets.utils import create_lerobot_dataset_card
 
 logger = logging.getLogger(__name__)
 
@@ -61,6 +50,8 @@ def _resolve_root(cfg: AnnotationPipelineConfig) -> Path:
     if cfg.root is not None:
         return Path(cfg.root)
     if cfg.repo_id is not None:
+        from huggingface_hub import snapshot_download
+
         return Path(snapshot_download(repo_id=cfg.repo_id, repo_type="dataset"))
     raise ValueError("Either --root or --repo_id must be provided.")
 
@@ -134,7 +125,7 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
 
     Pushes to ``cfg.new_repo_id`` when set, otherwise back to ``cfg.repo_id``.
     """
-    require_package("datasets", "dataset")
+    from huggingface_hub import HfApi  # noqa: PLC0415
 
     repo_id = cfg.new_repo_id or cfg.repo_id
     commit_message = cfg.push_commit_message or "Add steerable annotations (lerobot-annotate)"
@@ -152,26 +143,33 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
         repo_id=repo_id,
         repo_type="dataset",
         commit_message=commit_message,
-        # README.md is excluded because when pushing to ``new_repo_id`` the
-        # source card's links (e.g. the visualize badge) would keep pointing
-        # at the source dataset; a fresh card is generated below instead.
-        ignore_patterns=[".annotate_staging/**", "**/.DS_Store", "README.md"],
+        ignore_patterns=[".annotate_staging/**", "**/.DS_Store"],
     )
     print(f"[lerobot-annotate] uploaded to https://huggingface.co/datasets/{repo_id}", flush=True)
 
-    dataset_info = load_info(root)
-    card = create_lerobot_dataset_card(dataset_info=dataset_info, license="apache-2.0", repo_id=repo_id)
-    card.push_to_hub(repo_id=repo_id, repo_type="dataset")
-
     # Tag the upload with the codebase version. ``LeRobotDatasetMetadata``
     # resolves the dataset revision via ``get_safe_version`` which scans
     # for tags like ``v3.0``; without a tag it raises
     # ``RevisionNotFoundError``. Read the version straight from the
     # dataset's own ``meta/info.json`` so we tag whatever the writer
     # actually wrote (no accidental drift if the codebase floor moves).
-    version_tag = (
-        dataset_info.codebase_version if dataset_info.codebase_version.startswith("v") else CODEBASE_VERSION
-    )
+    from lerobot.datasets.dataset_metadata import CODEBASE_VERSION  # noqa: PLC0415
+
+    info_path = root / "meta" / "info.json"
+    version_tag = CODEBASE_VERSION
+    if info_path.exists():
+        try:
+            from lerobot.utils.io_utils import load_json  # noqa: PLC0415
+
+            info = load_json(info_path)
+            ds_version = info.get("codebase_version")
+            if isinstance(ds_version, str) and ds_version.startswith("v"):
+                version_tag = ds_version
+        except Exception as exc:  # noqa: BLE001
+            print(
+                f"[lerobot-annotate] could not read codebase_version from info.json ({exc}); falling back to {version_tag}",
+                flush=True,
+            )
     revision = getattr(commit_info, "oid", None)
     tag_kwargs = {
         "repo_id": repo_id,
@@ -182,6 +180,10 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
         tag_kwargs["revision"] = revision
 
     try:
+        from contextlib import suppress  # noqa: PLC0415
+
+        from huggingface_hub.errors import RevisionNotFoundError  # noqa: PLC0415
+
         with suppress(RevisionNotFoundError):
             api.delete_tag(repo_id, tag=version_tag, repo_type="dataset")
         api.create_tag(**tag_kwargs)

src/lerobot/scripts/lerobot_edit_dataset.py

@@ -167,7 +167,9 @@ Show dataset information without feature details:
         --operation.type info \
         --operation.show_features false
 
-Recompute dataset statistics (saves to lerobot/pusht_recomputed_stats by default):
+Recompute dataset statistics (saves to lerobot/pusht_recomputed_stats by default). The source
+dataset is never modified: large files are symlinked and only meta/ is copied, so this also works
+on read-only source datasets:
     lerobot-edit-dataset \
         --repo_id lerobot/pusht \
         --operation.type recompute_stats
@@ -178,6 +180,19 @@ Recompute stats and save to a specific new repo_id:
         --new_repo_id lerobot/pusht_new_stats \
         --operation.type recompute_stats
 
+Recompute stats including image/video features (samples and decodes frames from each episode):
+    lerobot-edit-dataset \
+        --repo_id lerobot/pusht \
+        --operation.type recompute_stats \
+        --operation.skip_image_video false
+
+Recompute stats and also rewrite the per-episode stats in the episodes parquet (keeps
+meta/stats.json and the per-episode stats consistent):
+    lerobot-edit-dataset \
+        --repo_id lerobot/pusht \
+        --operation.type recompute_stats \
+        --operation.update_episode_stats true
+
 Recompute stats in-place (overwrites original dataset stats):
     lerobot-edit-dataset \
         --repo_id lerobot/pusht \
@@ -325,6 +340,7 @@ class RecomputeStatsConfig(OperationConfig):
     relative_exclude_joints: list[str] | None = None
     chunk_size: int = 50
     num_workers: int = 0
+    update_episode_stats: bool = False
     overwrite: bool = False
 
 
@@ -377,6 +393,30 @@ def _resolve_io_paths(
     return output_repo_id, input_path, output_path
 
 
+def _reference_copy_dataset(input_root: Path, output_root: Path) -> None:
+    """Create a lightweight copy of a dataset that never modifies the source.
+
+    The directory tree is recreated with real directories, and every file is
+    symlinked to its source counterpart so no data is duplicated and the source is
+    only ever read. Files under ``meta/`` are instead copied as real, writable files
+    so that stats/info can be rewritten without touching the original. Symlinking
+    individual files (rather than whole directories) keeps ``push_to_hub`` working,
+    since ``Path.glob`` follows file symlinks but does not descend into symlinked
+    directories. This makes the operation safe on read-only source datasets.
+    """
+    for src in input_root.rglob("*"):
+        rel = src.relative_to(input_root)
+        dst = output_root / rel
+        if src.is_dir():
+            dst.mkdir(parents=True, exist_ok=True)
+        elif rel.parts[0] == "meta":
+            dst.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copyfile(src, dst)  # copyfile ignores source perms, so dst is writable
+        else:
+            dst.parent.mkdir(parents=True, exist_ok=True)
+            dst.symlink_to(src.resolve())
+
+
 def get_output_path(
     repo_id: str,
     new_repo_id: str | None,
@@ -674,14 +714,18 @@ def handle_recompute_stats(cfg: EditDatasetConfig) -> None:
         )
         dataset = LeRobotDataset(cfg.repo_id, root=input_root)
     else:
-        logging.info(f"Copying dataset from {input_root} to {output_root}")
+        logging.info(f"Referencing dataset from {input_root} into {output_root} (source is left untouched)")
         if output_root.exists():
             backup_path = output_root.with_name(output_root.name + "_old")
             logging.warning(f"Output directory {output_root} already exists. Moving to {backup_path}")
             if backup_path.exists():
                 shutil.rmtree(backup_path)
             shutil.move(output_root, backup_path)
-        shutil.copytree(input_root, output_root)
+        # recompute_stats only reads data/ and rewrites files under meta/ (stats.json, and
+        # the episodes parquet when update_episode_stats is set), so symlink the large
+        # immutable files and copy only meta/. This avoids duplicating the dataset and works
+        # even when the source dataset is read-only.
+        _reference_copy_dataset(input_root, output_root)
         dataset = LeRobotDataset(output_repo_id, root=output_root)
 
     logging.info(f"Recomputing stats for {cfg.repo_id}")
@@ -698,6 +742,7 @@ def handle_recompute_stats(cfg: EditDatasetConfig) -> None:
         relative_exclude_joints=cfg.operation.relative_exclude_joints,
         chunk_size=cfg.operation.chunk_size,
         num_workers=cfg.operation.num_workers,
+        update_episode_stats=cfg.operation.update_episode_stats,
     )
 
     logging.info(f"Stats written to {dataset.root}")

src/lerobot/scripts/lerobot_train.py

@@ -171,9 +171,6 @@ def update_policy(
     train_metrics.update_s = time.perf_counter() - start_time
     if torch.cuda.is_available():
         train_metrics.gpu_mem_gb = torch.cuda.max_memory_allocated() / (1024**3)
-    # Aggregate the policy's scalar outputs for logging and rank-reduction across the log window.
-    if output_dict:
-        train_metrics.update_metrics(output_dict)
     return train_metrics, output_dict
 
 
@@ -575,7 +572,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
         batch = preprocessor(batch)
         train_tracker.dataloading_s = time.perf_counter() - start_time
 
-        train_tracker, _ = update_policy(
+        train_tracker, output_dict = update_policy(
             train_tracker,
             policy,
             batch,
@@ -608,10 +605,9 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
                     train_tracker.samples_per_s = effective_batch_size / step_time
                 logging.info(train_tracker)
                 if wandb_logger:
-                    # Policy sub-losses (latent_loss, action_loss, ...) are aggregated into the
-                    # tracker by update_policy, so to_dict() already carries their windowed,
-                    # rank-reduced averages — no per-step output_dict passthrough needed.
                     wandb_log_dict = train_tracker.to_dict()
+                    if output_dict:
+                        wandb_log_dict.update(output_dict)
                     # Log sample weighting statistics if enabled
                     if sample_weighter is not None:
                         weighter_stats = sample_weighter.get_stats()

src/lerobot/utils/device_utils.py

@@ -59,20 +59,6 @@ def get_safe_torch_device(try_device: str, log: bool = False) -> torch.device:
     return device
 
 
-def resolve_safetensors_device(map_location: str | torch.device) -> str:
-    """Resolve a device string for a safetensors load, working around a device-mapping quirk.
-
-    safetensors' load maps the bare string "cuda" to cuda:0 regardless of the current device
-    (unlike torch's .to("cuda"), which honors torch.cuda.current_device()). Under multi-GPU
-    accelerate/FSDP every rank would then load its weights onto GPU 0, OOMing it before sharding.
-    Resolve "cuda" to the concrete current-device index so each rank loads onto its own GPU.
-    """
-    map_location = str(map_location)
-    if map_location == "cuda" and torch.cuda.is_available():
-        return f"cuda:{torch.cuda.current_device()}"
-    return map_location
-
-
 def get_safe_dtype(dtype: torch.dtype, device: str | torch.device):
     """
     mps is currently not compatible with float64

src/lerobot/utils/logging_utils.py

@@ -104,7 +104,6 @@ class MetricsTracker:
         "episodes",
         "epochs",
         "accelerator",
-        "_caller_metrics",
     ]
 
     def __init__(
@@ -130,9 +129,6 @@ class MetricsTracker:
         self.episodes = self.samples / self._avg_samples_per_ep
         self.epochs = self.samples / self._num_frames
         self.accelerator = accelerator
-        # Meter names the caller registered up front. update_metrics() leaves these untouched, so a
-        # policy that echoes e.g. "loss" in its output dict can't clobber the aggregated meter.
-        self._caller_metrics: set[str] = set(self.metrics)
 
     def __getattr__(self, name: str) -> int | dict[str, AverageMeter] | AverageMeter | Any:
         if name in self.__dict__:
@@ -160,21 +156,6 @@ class MetricsTracker:
         self.episodes = self.samples / self._avg_samples_per_ep
         self.epochs = self.samples / self._num_frames
 
-    def update_metrics(self, values: dict[str, Any]) -> None:
-        """Accumulate a dict of scalar metrics, auto-registering a meter for each new key.
-
-        Non-numeric values and bools are ignored.
-        Caller-registered metrics (those passed to the constructor) are never overridden.
-        """
-        for name, value in values.items():
-            if isinstance(value, bool) or not isinstance(value, (int, float)):
-                continue
-            if name in self._caller_metrics:
-                continue
-            if name not in self.metrics:
-                self.metrics[name] = AverageMeter(name, ":.3f", reduction="mean")
-            self.metrics[name].update(float(value))
-
     def reduce_across_ranks(self) -> None:
         """
         Synchronises the running averages of every metric whose ``reduction`` is not ``"none"``

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(
         {
-            "COMPLETED manipulation events": {
+            "atomic subtasks": {
                 "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(
         {
-            "COMPLETED manipulation events": {
+            "atomic subtasks": {
                 "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 "COMPLETED manipulation events" in _prompt_text(m)]
+    subtask_calls = [m for m in captured if "atomic subtasks" 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"]

tests/policies/test_peft_base_model_loading.py

@@ -1,142 +0,0 @@
-# Copyright 2024 The HuggingFace Inc. team. All rights reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-"""Regression tests for PEFT loading when the checkpoint is a base model (see issue #3975).
-
-Starting a fresh LoRA/PEFT fine-tune points ``--policy.path`` at a *base model* (no
-``adapter_config.json``) while also setting ``use_peft=True``. This must NOT be mistaken for
-loading an existing PEFT adapter. These tests lock in the base-model vs. adapter distinction
-made by ``lerobot.policies.factory._has_peft_adapter_config`` and the branch it drives in
-``make_policy``. They are pure/fast (no network, no ``peft``), so they run in CI.
-"""
-
-import json
-from unittest.mock import MagicMock, patch
-
-import torch
-from huggingface_hub.errors import HfHubHTTPError
-
-from lerobot.policies.factory import _has_peft_adapter_config
-
-
-def test_local_base_model_dir_has_no_adapter_config(tmp_path):
-    # A base-model checkpoint directory (only model weights, no adapter config).
-    (tmp_path / "model.safetensors").write_bytes(b"")
-    (tmp_path / "config.json").write_text("{}")
-    assert _has_peft_adapter_config(str(tmp_path)) is False
-
-
-def test_local_adapter_dir_has_adapter_config(tmp_path):
-    (tmp_path / "adapter_config.json").write_text(json.dumps({"peft_type": "LORA"}))
-    (tmp_path / "adapter_model.safetensors").write_bytes(b"")
-    assert _has_peft_adapter_config(str(tmp_path)) is True
-
-
-def test_hub_base_model_repo_has_no_adapter_config():
-    with patch("huggingface_hub.file_exists", return_value=False) as mock_exists:
-        assert _has_peft_adapter_config("lerobot/lingbot_va_base") is False
-    mock_exists.assert_called_once()
-    assert mock_exists.call_args.args[1] == "adapter_config.json"
-
-
-def test_hub_adapter_repo_has_adapter_config():
-    with patch("huggingface_hub.file_exists", return_value=True):
-        assert _has_peft_adapter_config("some/adapter-repo", revision="main") is True
-
-
-def test_hub_lookup_error_falls_back_to_base_model():
-    # Offline / private / transient Hub errors must not crash; treat as "not an adapter".
-    with patch(
-        "huggingface_hub.file_exists",
-        side_effect=HfHubHTTPError("boom", response=MagicMock()),
-    ):
-        assert _has_peft_adapter_config("some/private-repo") is False
-    with patch("huggingface_hub.file_exists", side_effect=OSError("offline")):
-        assert _has_peft_adapter_config("some/repo") is False
-
-
-def _make_dummy_policy_cfg(pretrained_path, use_peft):
-    cfg = MagicMock()
-    cfg.type = "act"
-    cfg.device = "cpu"
-    cfg.pretrained_path = pretrained_path
-    cfg.pretrained_revision = None
-    cfg.use_peft = use_peft
-    cfg.input_features = {}
-    cfg.output_features = {}
-    return cfg
-
-
-@patch("lerobot.policies.factory.validate_visual_features_consistency")
-@patch("lerobot.policies.factory.env_to_policy_features", return_value={})
-@patch("lerobot.policies.factory.get_policy_class")
-def test_make_policy_base_model_with_use_peft_loads_base_not_adapter(
-    mock_get_cls, _mock_features, _mock_validate
-):
-    """`use_peft=True` on a base model must load the base weights, not a PEFT adapter.
-
-    Before the #3975 fix this went down the ``PeftConfig.from_pretrained`` path and failed
-    looking for a non-existent ``adapter_config.json``.
-    """
-    from lerobot.policies import factory
-
-    policy_cls = MagicMock()
-    loaded_policy = torch.nn.Linear(1, 1)  # a real nn.Module so make_policy's assert passes
-    policy_cls.from_pretrained.return_value = loaded_policy
-    mock_get_cls.return_value = policy_cls
-
-    cfg = _make_dummy_policy_cfg(pretrained_path="lerobot/lingbot_va_base", use_peft=True)
-    env_cfg = MagicMock()
-
-    with patch.object(factory, "_has_peft_adapter_config", return_value=False) as mock_has_adapter:
-        policy = factory.make_policy(cfg=cfg, env_cfg=env_cfg)
-
-    mock_has_adapter.assert_called_once()
-    # Base model is loaded via the normal pretrained path...
-    policy_cls.from_pretrained.assert_called_once()
-    assert policy_cls.from_pretrained.call_args.kwargs["pretrained_name_or_path"] == (
-        "lerobot/lingbot_va_base"
-    )
-    # ...and PEFT adapter loading is NOT attempted (would need peft + adapter_config.json).
-    assert policy is loaded_policy
-
-
-@patch("lerobot.policies.factory.validate_visual_features_consistency")
-@patch("lerobot.policies.factory.env_to_policy_features", return_value={})
-@patch("lerobot.policies.factory.get_policy_class")
-def test_make_policy_existing_adapter_uses_peft_loading(mock_get_cls, _mock_features, _mock_validate):
-    """A real adapter checkpoint (has ``adapter_config.json``) must go through PEFT loading."""
-    from lerobot.policies import factory
-
-    policy_cls = MagicMock()
-    mock_get_cls.return_value = policy_cls
-
-    cfg = _make_dummy_policy_cfg(pretrained_path="some/adapter-repo", use_peft=True)
-    env_cfg = MagicMock()
-
-    policy_cls.from_pretrained.return_value = torch.nn.Linear(1, 1)
-
-    fake_peft = MagicMock()
-    fake_peft_config = MagicMock()
-    fake_peft_config.base_model_name_or_path = "lerobot/lingbot_va_base"
-    fake_peft.PeftConfig.from_pretrained.return_value = fake_peft_config
-    fake_peft.PeftModel.from_pretrained.return_value = torch.nn.Linear(1, 1)
-
-    with (
-        patch.object(factory, "_has_peft_adapter_config", return_value=True),
-        patch.dict("sys.modules", {"peft": fake_peft}),
-    ):
-        factory.make_policy(cfg=cfg, env_cfg=env_cfg)
-
-    fake_peft.PeftConfig.from_pretrained.assert_called_once_with("some/adapter-repo")
-    fake_peft.PeftModel.from_pretrained.assert_called_once()

tests/scripts/test_lerobot_annotate.py

@@ -18,8 +18,6 @@ import json
 from types import SimpleNamespace
 
 import pytest
-import requests
-from huggingface_hub.errors import RevisionNotFoundError
 
 # ``lerobot.scripts.lerobot_annotate`` (and the ``_push_to_hub`` path it
 # exercises) imports ``lerobot.datasets``, which only ships under the
@@ -28,13 +26,11 @@ pytest.importorskip("datasets", reason="datasets is required (install lerobot[da
 
 
 def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
-    from lerobot.scripts import lerobot_annotate
+    from lerobot.scripts.lerobot_annotate import _push_to_hub
 
     root = tmp_path / "dataset"
     (root / "meta").mkdir(parents=True)
-    (root / "meta" / "info.json").write_text(
-        json.dumps({"codebase_version": "v3.0", "fps": 30, "features": {}})
-    )
+    (root / "meta" / "info.json").write_text(json.dumps({"codebase_version": "v3.0"}))
 
     calls = {}
 
@@ -47,6 +43,9 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
             return SimpleNamespace(oid="abc123")
 
         def delete_tag(self, repo_id, **kwargs):
+            import requests
+            from huggingface_hub.errors import RevisionNotFoundError
+
             calls["delete_tag"] = {"repo_id": repo_id, **kwargs}
             # Simulate the common case: no stale tag to delete.
             raise RevisionNotFoundError("no such tag", response=requests.Response())
@@ -54,12 +53,7 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
         def create_tag(self, **kwargs):
             calls["create_tag"] = kwargs
 
-    monkeypatch.setattr(lerobot_annotate, "HfApi", FakeHfApi)
-
-    def fake_card_push(self, **kwargs):
-        calls["card_push"] = {"content": str(self), **kwargs}
-
-    monkeypatch.setattr("huggingface_hub.DatasetCard.push_to_hub", fake_card_push)
+    monkeypatch.setattr("huggingface_hub.HfApi", FakeHfApi)
 
     cfg = SimpleNamespace(
         repo_id="source/dataset",
@@ -68,7 +62,7 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
         push_commit_message=None,
     )
 
-    lerobot_annotate._push_to_hub(root, cfg)
+    _push_to_hub(root, cfg)
 
     assert calls["create_repo"] == {
         "repo_id": "annotated/dataset",
@@ -77,13 +71,6 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
         "exist_ok": True,
     }
     assert calls["upload_folder"]["repo_id"] == "annotated/dataset"
-    # The source README must not be copied over: its links (e.g. the
-    # visualize badge) point at the source dataset. A card regenerated for
-    # the target repo is pushed instead.
-    assert "README.md" in calls["upload_folder"]["ignore_patterns"]
-    assert calls["card_push"]["repo_id"] == "annotated/dataset"
-    assert "visualize_dataset?path=annotated/dataset" in calls["card_push"]["content"]
-    assert "source/dataset" not in calls["card_push"]["content"]
     # A stale tag (e.g. from a previous annotation run) is deleted first so
     # the new tag always points at the upload we just made.
     assert calls["delete_tag"] == {

tests/utils/test_logging_utils.py

@@ -233,37 +233,3 @@ def test_metrics_tracker_reduce_across_ranks_invokes_reduce():
     # accumulate against the cluster view rather than the stale per-rank sum.
     meter = tracker.update_s
     assert meter.sum / meter.count == pytest.approx(meter.avg)
-
-
-def test_metrics_tracker_update_metrics_registers_and_averages():
-    tracker = MetricsTracker(batch_size=32, num_frames=1000, num_episodes=50, metrics={})
-    tracker.update_metrics({"latent_loss": 0.2, "action_loss": 0.4})
-    tracker.update_metrics({"latent_loss": 0.4, "action_loss": 0.6})
-
-    # New keys are auto-registered as mean-reduced meters and averaged over the window.
-    assert tracker.metrics["latent_loss"].reduction == "mean"
-    assert tracker.metrics["latent_loss"].avg == pytest.approx(0.3)
-    assert tracker.metrics["action_loss"].avg == pytest.approx(0.5)
-    assert tracker.to_dict()["latent_loss"] == pytest.approx(0.3)
-
-
-def test_metrics_tracker_update_metrics_skips_non_numeric():
-    tracker = MetricsTracker(batch_size=32, num_frames=1000, num_episodes=50, metrics={})
-    tracker.update_metrics({"loss": 0.5, "head_mode": "sparse", "enabled": True})
-
-    # strings and bools ignored
-    assert "loss" in tracker.metrics
-    assert "head_mode" not in tracker.metrics
-    assert "enabled" not in tracker.metrics
-
-
-def test_metrics_tracker_update_metrics_does_not_override_caller_meter():
-    # A policy that echoes "loss" in its output dict must not overwrite the caller-owned,
-    # already-aggregated loss meter.
-    metrics = {"loss": AverageMeter("loss", ":.3f", reduction="mean")}
-    tracker = MetricsTracker(batch_size=32, num_frames=1000, num_episodes=50, metrics=metrics)
-    tracker.loss = 1.0  # caller-set optimized loss
-    tracker.update_metrics({"loss": 99.0, "latent_loss": 0.2})
-
-    assert tracker.metrics["loss"].avg == pytest.approx(1.0)  # snapshot ignored
-    assert tracker.metrics["latent_loss"].avg == pytest.approx(0.2)