From d63e6e67a5284d05807c209b162242a9106ab12d Mon Sep 17 00:00:00 2001 From: Kohei SENDAI <80389896+k1000dai@users.noreply.github.com> Date: Mon, 27 Jul 2026 18:54:09 +0900 Subject: [PATCH 01/10] fix convverstion err (#3656) Co-authored-by: Steven Palma --- src/lerobot/scripts/convert_dataset_v21_to_v30.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/lerobot/scripts/convert_dataset_v21_to_v30.py b/src/lerobot/scripts/convert_dataset_v21_to_v30.py index f516aa8fc..ca7d1e91b 100644 --- a/src/lerobot/scripts/convert_dataset_v21_to_v30.py +++ b/src/lerobot/scripts/convert_dataset_v21_to_v30.py @@ -61,6 +61,7 @@ import pyarrow as pa import tqdm from datasets import Dataset, Features, Image from huggingface_hub import HfApi, snapshot_download +from huggingface_hub.errors import RevisionNotFoundError from requests import HTTPError from lerobot.datasets import CODEBASE_VERSION, LeRobotDataset, aggregate_stats @@ -521,7 +522,7 @@ def convert_dataset( hub_api = HfApi() try: hub_api.delete_tag(repo_id, tag=CODEBASE_VERSION, repo_type="dataset") - except HTTPError as e: + except (HTTPError, RevisionNotFoundError) as e: print(f"tag={CODEBASE_VERSION} probably doesn't exist. Skipping exception ({e})") pass hub_api.delete_files( From 6c57dfd2eeb9bd40be364ffd329562d990fa1de7 Mon Sep 17 00:00:00 2001 From: "hf-dependantbot-rollout[bot]" <285970069+hf-dependantbot-rollout[bot]@users.noreply.github.com> Date: Mon, 27 Jul 2026 13:22:04 +0200 Subject: [PATCH 02/10] chore: enable Dependabot weekly GitHub Actions bumps (#3677) Co-authored-by: hf-dependantbot-rollout[bot] <285970069+hf-dependantbot-rollout[bot]@users.noreply.github.com> Co-authored-by: Steven Palma --- .github/dependabot.yml | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 .github/dependabot.yml diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 000000000..15f7bdd79 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,11 @@ +version: 2 +updates: + - package-ecosystem: "github-actions" + directory: "/" + schedule: + interval: "weekly" + cooldown: + default-days: 7 + groups: + actions: + patterns: ["*"] From ab87fd97646f323d5267b029b2c2f2181e9da886 Mon Sep 17 00:00:00 2001 From: MihaiAnca13 Date: Mon, 27 Jul 2026 12:44:32 +0100 Subject: [PATCH 03/10] fix(datasets): clear video frame staging on episode reset (#3683) * fix video frame staging cleanup on episode reset * linting --------- Co-authored-by: Steven Palma --- src/lerobot/datasets/dataset_writer.py | 37 ++++++++++++++++---------- tests/datasets/test_dataset_writer.py | 32 ++++++++++++++++++++++ 2 files changed, 55 insertions(+), 14 deletions(-) diff --git a/src/lerobot/datasets/dataset_writer.py b/src/lerobot/datasets/dataset_writer.py index a6049312f..5693e9bb2 100644 --- a/src/lerobot/datasets/dataset_writer.py +++ b/src/lerobot/datasets/dataset_writer.py @@ -172,6 +172,23 @@ class DatasetWriter: def _get_image_file_dir(self, episode_index: int, image_key: str) -> Path: return self._get_image_file_path(episode_index, image_key, frame_index=0).parent + def _get_episode_buffer_index(self) -> int: + episode_index = self.episode_buffer["episode_index"] + # episode_index is `int` when freshly created, but becomes `np.ndarray` after + # save_episode() mutates the buffer. Handle both types here. + if isinstance(episode_index, np.ndarray): + episode_index = episode_index.item() if episode_index.size == 1 else episode_index[0] + return int(episode_index) + + def _delete_camera_frame_dirs(self, camera_keys: list[str]) -> None: + if self.image_writer is not None: + self._wait_image_writer() + episode_index = self._get_episode_buffer_index() + for camera_key in camera_keys: + img_dir = self._get_image_file_dir(episode_index, camera_key) + if img_dir.is_dir(): + shutil.rmtree(img_dir) + def _save_image( self, image: torch.Tensor | np.ndarray | PIL.Image.Image, fpath: Path, compress_level: int = 1 ) -> None: @@ -369,7 +386,9 @@ class DatasetWriter: self._episodes_since_last_encoding = 0 if episode_data is None: - self.clear_episode_buffer(delete_images=len(self._meta.image_keys) > 0) + if len(self._meta.image_keys) > 0: + self._delete_camera_frame_dirs(self._meta.image_keys) + self.episode_buffer = self._create_episode_buffer() def _batch_save_episode_video(self, start_episode: int, end_episode: int | None = None) -> None: """Batch save videos for multiple episodes.""" @@ -561,10 +580,10 @@ class DatasetWriter: return metadata def clear_episode_buffer(self, delete_images: bool = True) -> None: - """Discard the current episode buffer and optionally delete temp images. + """Discard the current episode buffer and optionally delete temp camera frames. Args: - delete_images: If ``True``, remove temporary image directories + delete_images: If ``True``, remove temporary camera frame directories written for the current episode. """ # Cancel streaming encoder if active @@ -572,17 +591,7 @@ class DatasetWriter: self._streaming_encoder.cancel_episode() if delete_images: - if self.image_writer is not None: - self._wait_image_writer() - episode_index = self.episode_buffer["episode_index"] - # episode_index is `int` when freshly created, but becomes `np.ndarray` after - # save_episode() mutates the buffer. Handle both types here. - if isinstance(episode_index, np.ndarray): - episode_index = episode_index.item() if episode_index.size == 1 else episode_index[0] - for cam_key in self._meta.image_keys: - img_dir = self._get_image_file_dir(episode_index, cam_key) - if img_dir.is_dir(): - shutil.rmtree(img_dir) + self._delete_camera_frame_dirs(self._meta.camera_keys) self.episode_buffer = self._create_episode_buffer() diff --git a/tests/datasets/test_dataset_writer.py b/tests/datasets/test_dataset_writer.py index 17785ad74..bd51a2687 100644 --- a/tests/datasets/test_dataset_writer.py +++ b/tests/datasets/test_dataset_writer.py @@ -204,6 +204,38 @@ def test_clear_resets_buffer(tmp_path): assert dataset.writer.episode_buffer["size"] == 0 +def test_clear_removes_video_frame_staging_dir(tmp_path): + """clear_episode_buffer() removes PNG staging dirs for video features.""" + video_key = "observation.images.cam" + features = { + video_key: { + "dtype": "video", + "shape": (64, 96, 3), + "names": ["height", "width", "channels"], + }, + "action": {"dtype": "float32", "shape": (2,), "names": None}, + } + dataset = LeRobotDataset.create( + repo_id=DUMMY_REPO_ID, + fps=DEFAULT_FPS, + features=features, + root=tmp_path / "ds", + use_videos=True, + ) + + dataset.add_frame(_make_frame(features)) + video_staging_dir = ( + dataset.root + / Path(DEFAULT_IMAGE_PATH.format(image_key=video_key, episode_index=0, frame_index=0)).parent + ) + assert video_staging_dir.is_dir() + + dataset.clear_episode_buffer() + + assert dataset.writer.episode_buffer["size"] == 0 + assert not video_staging_dir.exists() + + def test_finalize_is_idempotent(tmp_path): """Calling finalize() twice does not raise.""" dataset = LeRobotDataset.create( From 801346e18cea12999f095be5b1caa9c5a6842607 Mon Sep 17 00:00:00 2001 From: Steven Palma Date: Mon, 27 Jul 2026 14:08:11 +0200 Subject: [PATCH 04/10] fix(scripts): restore policy training mode after eval_policy() in lerobot-eval (#4162) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(scripts/eval): restore policy training mode after eval_policy() `eval_policy` calls `policy.eval()` before the rollout but never restores the prior mode on return. When called from the training loop (`lerobot_train.py`'s `eval_policy_all -> run_one -> eval_one -> eval_policy` chain), the policy is left in eval mode for every subsequent training step, which silently: * disables Dropout (no regularisation), * freezes BatchNorm running stats (no further EMA updates). Under DDP only `is_main_process` runs eval (lerobot_train.py:527), so the main rank ends up in eval mode while workers stay in train mode — the all-reduced gradients then combine forward passes computed with different dropout masks and different BN behaviour, a real DDP-correctness issue. Scope of impact: * Affects every policy with Dropout in its forward path. In-tree, that includes the default ACT (6 Dropout layers at p=0.1), Diffusion (vision backbone), VQ-BeT, Multi-Task DiT, X-VLA, plus all VLA policies that inherit Dropout from their pretrained HF backbone (PI0/PI0.5/PI0-FAST, SmolVLA, GR00T-N1.5, EO1, Wall-X). * Triggers from the first eval onward. On the default config (steps=100k, eval_freq=20k) that's 80% of training; on the LIBERO / RoboCasa / VLABench example commands in docs/ (eval_freq=1k–5k) it's 95–99% of training. * Policies using only LayerNorm/GroupNorm and no Dropout (TDMPC, RTC) are unaffected. Policies using `FrozenBatchNorm2d` (ACT's ResNet backbone) are immune to the BN-stat half; the Dropout half still bites. Fix: * Snapshot `policy.training` on entry to `eval_policy`. * Restore it on normal return. * Save-and-restore is a strict no-op for callers that pass an already-eval-mode policy (e.g. the standalone `lerobot-eval` script loading a frozen checkpoint). * Restoration is placed before the normal return only, not in a try/finally — exception paths leave the policy in eval mode, same as today. A try/finally upgrade would require re-indenting ~165 lines and can land as a separate cleanup if desired. Tests (tests/scripts/test_eval.py, 7 tests total, ~1.6s): * Regression gates on the lerobot_eval fix itself: training-mode preservation, eval-mode preservation, dropout-active behavioural check, non-crash for both entry modes. * Quantitative mechanism demonstration (`test_missing_mode_restoration_hurts_generalisation`): trains a tiny Dropout+BatchNorm MLP under both the bug pattern and the fix pattern on identical data and seed, then asserts the buggy variant generalises at least 5% worse on a held-out val set. In repeated runs we see 10-25% deltas on this toy problem; real policies (more layers, more Dropout, longer training) generally see larger gaps. Lives alongside the regression tests so the empirical proof is reproducible from the repo without adding a separate benchmarks/ directory. * fix(scripts): keep policy train/eval --------- Co-authored-by: ModeEric --- src/lerobot/scripts/lerobot_eval.py | 69 +++++++++++++++++------------ 1 file changed, 41 insertions(+), 28 deletions(-) diff --git a/src/lerobot/scripts/lerobot_eval.py b/src/lerobot/scripts/lerobot_eval.py index 722763d6e..c4ed35145 100644 --- a/src/lerobot/scripts/lerobot_eval.py +++ b/src/lerobot/scripts/lerobot_eval.py @@ -453,6 +453,9 @@ def eval_policy( raise exc from None start = time.time() + # Preserve the mode for direct callers. eval_policy_all scopes the mode + # around all tasks so parallel evaluations cannot race with each other. + was_training = policy.training policy.eval() # Determine how many batched rollouts we need to get n_episodes. Note that if n_episodes is not evenly @@ -674,6 +677,8 @@ def eval_policy( if save_predicted_video: info["predicted_video_paths"] = predicted_video_paths + policy.train(was_training) + return info @@ -1010,40 +1015,48 @@ def eval_policy_all( recording_private=recording_private, ) - if max_parallel_tasks <= 1: - prefetch_thread: threading.Thread | None = None - for i, (task_group, task_id, env) in enumerate(tasks): - if prefetch_thread is not None: - prefetch_thread.join() - prefetch_thread = None + # Set the shared policy's mode before launching any workers. Restoring it + # inside individual tasks would let one task enable training mode while + # another task is still evaluating. + was_training = policy.training + policy.eval() + try: + if max_parallel_tasks <= 1: + prefetch_thread: threading.Thread | None = None + for i, (task_group, task_id, env) in enumerate(tasks): + if prefetch_thread is not None: + prefetch_thread.join() + prefetch_thread = None - try: - tg, tid, metrics = task_runner(task_group, task_id, env) - _accumulate_to(tg, metrics) - per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics}) - finally: - env.close() - # Prefetch next task's workers *after* closing current env to prevent - # GPU memory overlap between consecutive tasks. - if i + 1 < len(tasks): - next_env = tasks[i + 1][2] - if hasattr(next_env, "_ensure"): - prefetch_thread = threading.Thread(target=next_env._ensure, daemon=True) - prefetch_thread.start() - else: - with cf.ThreadPoolExecutor(max_workers=max_parallel_tasks) as executor: - fut2meta = {} - for task_group, task_id, env in tasks: - fut = executor.submit(task_runner, task_group, task_id, env) - fut2meta[fut] = (task_group, task_id, env) - for fut in cf.as_completed(fut2meta): - tg, tid, env = fut2meta[fut] try: - tg, tid, metrics = fut.result() + tg, tid, metrics = task_runner(task_group, task_id, env) _accumulate_to(tg, metrics) per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics}) finally: env.close() + # Prefetch next task's workers *after* closing current env to prevent + # GPU memory overlap between consecutive tasks. + if i + 1 < len(tasks): + next_env = tasks[i + 1][2] + if hasattr(next_env, "_ensure"): + prefetch_thread = threading.Thread(target=next_env._ensure, daemon=True) + prefetch_thread.start() + else: + with cf.ThreadPoolExecutor(max_workers=max_parallel_tasks) as executor: + fut2meta = {} + for task_group, task_id, env in tasks: + fut = executor.submit(task_runner, task_group, task_id, env) + fut2meta[fut] = (task_group, task_id, env) + for fut in cf.as_completed(fut2meta): + tg, tid, env = fut2meta[fut] + try: + tg, tid, metrics = fut.result() + _accumulate_to(tg, metrics) + per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics}) + finally: + env.close() + finally: + policy.train(was_training) # compute aggregated metrics helper (robust to lists/scalars) def _agg_from_list(xs): From bbeacfe57d0db2dc08a009d3e474a741c65adc00 Mon Sep 17 00:00:00 2001 From: Steven Palma Date: Mon, 27 Jul 2026 14:09:58 +0200 Subject: [PATCH 05/10] fix(record): connect teleoperator before robot to avoid watchdog jump (#4166) * fix(record): connect teleoperator before robot to avoid watchdog jump lerobot-record connected the robot before the teleoperator. A robot's connect()/reset() can leave it holding a default pose under a firmware watchdog (e.g. Unitree G1); if teleop.connect() (model loading, IK init, network setup) then takes longer than that watchdog, the joints drop to damping and the first send_action() makes the robot jump. Swap the order so the teleoperator connects first, matching the ordering already used in lerobot_teleoperate.py. Pure ordering fix, no API change. Fixes #3684 * fix(record): trim comment and add connect-order regression test Address review feedback on #3684: - Trim the verbose ordering comment down to two lines. - Add test_record_connects_teleop_before_robot to tests/test_control_robot.py, asserting teleop.connect() runs before robot.connect() in record(). * chore(test): remove test --------- Co-authored-by: Jaimin Patel Co-authored-by: Martino Russi <77496684+nepyope@users.noreply.github.com> --- src/lerobot/scripts/lerobot_record.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/lerobot/scripts/lerobot_record.py b/src/lerobot/scripts/lerobot_record.py index 4233f0bc2..532a8c602 100644 --- a/src/lerobot/scripts/lerobot_record.py +++ b/src/lerobot/scripts/lerobot_record.py @@ -453,9 +453,11 @@ def record( encoder_queue_maxsize=cfg.dataset.encoder_queue_maxsize, ) - robot.connect() + # Connect the teleoperator before the robot so the robot isn't left idle (and possibly + # tripping a firmware watchdog) during teleop init. Matches lerobot_teleoperate.py. if teleop is not None: teleop.connect() + robot.connect() listener, events = init_keyboard_listener() From acd42b4d85f95ccfa2d6563fde41c47598f455f7 Mon Sep 17 00:00:00 2001 From: WOLIKIMCHENG <35391914+WOLIKIMCHENG@users.noreply.github.com> Date: Mon, 27 Jul 2026 21:53:31 +0800 Subject: [PATCH 06/10] fix(processor): keep missing local state resolution local (#3715) Co-authored-by: root Co-authored-by: Steven Palma --- src/lerobot/processor/pipeline.py | 22 ++- .../test_pipeline_from_pretrained_helpers.py | 145 +++++++++++++++++- 2 files changed, 162 insertions(+), 5 deletions(-) diff --git a/src/lerobot/processor/pipeline.py b/src/lerobot/processor/pipeline.py index b9b9c6c43..e40a7c479 100644 --- a/src/lerobot/processor/pipeline.py +++ b/src/lerobot/processor/pipeline.py @@ -713,6 +713,8 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): ProcessorMigrationError: If the model requires migration to processor format. """ model_id = str(pretrained_model_name_or_path) + model_path = Path(model_id) + is_local_source = model_path.is_dir() or model_path.is_file() hub_download_kwargs = { "force_download": force_download, "resume_download": resume_download, @@ -731,7 +733,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): # 3. Build steps with overrides steps, validated_overrides = cls._build_steps_with_overrides( - loaded_config, overrides or {}, model_id, base_path, hub_download_kwargs + loaded_config, overrides or {}, model_id, base_path, hub_download_kwargs, is_local_source ) # 4. Validate that all overrides were used @@ -921,6 +923,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): model_id: str, base_path: Path | None, hub_download_kwargs: dict[str, Any], + is_local_source: bool = False, ) -> tuple[list[ProcessorStep], set[str]]: """Build all processor steps with overrides and state loading. @@ -944,7 +947,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): 3. **State Loading** (via _load_step_state): - **If step has "state_file"**: Load tensor state from .safetensors - **Local first**: Check base_path/state_file.safetensors - - **Hub fallback**: Download state file if not found locally + - **Hub fallback**: Download state file if the pipeline was loaded from the Hub - **Optional**: Only load if step has load_state_dict method 4. **Override Tracking**: @@ -962,6 +965,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): model_id: The model identifier (needed for Hub state file downloads) base_path: Local directory path for finding state files hub_download_kwargs: Parameters for hf_hub_download (tokens, cache, etc.) + is_local_source: Whether model_id resolved to a local directory or config file. Returns: Tuple of (instantiated_steps_list, unused_override_keys) @@ -975,7 +979,9 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): steps, remaining_override_keys = cls._build_steps_from_config(loaded_config, overrides) for step_instance, step_entry in zip(steps, loaded_config["steps"], strict=True): - cls._load_step_state(step_instance, step_entry, model_id, base_path, hub_download_kwargs) + cls._load_step_state( + step_instance, step_entry, model_id, base_path, hub_download_kwargs, is_local_source + ) return steps, remaining_override_keys @@ -1139,6 +1145,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): model_id: str, base_path: Path | None, hub_download_kwargs: dict[str, Any], + is_local_source: bool = False, ) -> None: """Load state dictionary for a processor step if available. @@ -1157,7 +1164,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): - **Use case**: Loading from local saved model directory 2. **Hub download fallback**: Download state file from repository - - **When triggered**: Local file not found or base_path is None + - **When triggered**: Local file not found and the pipeline source is a Hub repo - **Process**: Use hf_hub_download with same parameters as config - **Example**: Download "normalize_step_0.safetensors" from "user/repo" - **Result**: Downloaded to local cache, path returned @@ -1178,6 +1185,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): model_id: The model identifier (used for Hub downloads if needed) base_path: Local directory path for finding state files (None for Hub-only) hub_download_kwargs: Parameters for hf_hub_download (tokens, cache, etc.) + is_local_source: Whether model_id resolved to a local directory or config file. Note: This method modifies step_instance in-place and returns None. @@ -1191,6 +1199,12 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin): # Try local file first if base_path and (base_path / state_filename).exists(): state_path = str(base_path / state_filename) + elif is_local_source: + state_path = base_path / state_filename if base_path else Path(state_filename) + raise FileNotFoundError( + f"State file '{state_filename}' was not found for local processor pipeline " + f"'{model_id}' at '{state_path}'." + ) else: # Download from Hub state_path = hf_hub_download( diff --git a/tests/processor/test_pipeline_from_pretrained_helpers.py b/tests/processor/test_pipeline_from_pretrained_helpers.py index 89d45cbad..36f9a8fad 100644 --- a/tests/processor/test_pipeline_from_pretrained_helpers.py +++ b/tests/processor/test_pipeline_from_pretrained_helpers.py @@ -26,8 +26,17 @@ import tempfile from pathlib import Path import pytest +import torch +from safetensors.torch import save_file -from lerobot.processor.pipeline import DataProcessorPipeline, ProcessorMigrationError +from lerobot.configs import PipelineFeatureType, PolicyFeature +from lerobot.processor.pipeline import ( + DataProcessorPipeline, + ProcessorMigrationError, + ProcessorStep, + ProcessorStepRegistry, +) +from lerobot.types import EnvTransition # Simplified Config Loading Tests @@ -98,6 +107,140 @@ def test_load_config_nonexistent_path_tries_hub(): DataProcessorPipeline._load_config("nonexistent/path", "processor.json", {}) +def test_from_pretrained_local_directory_missing_state_does_not_call_hub(monkeypatch): + """Local processor dirs must fail locally when a state file is missing.""" + + @ProcessorStepRegistry.register("local_missing_state_step") + class LocalMissingStateStep(ProcessorStep): + def __call__(self, transition: EnvTransition) -> EnvTransition: + return transition + + def transform_features( + self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]] + ) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]: + return features + + def load_state_dict(self, state: dict[str, torch.Tensor]) -> None: + pass + + try: + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + config = { + "name": "LocalMissingStatePipeline", + "steps": [{"registry_name": "local_missing_state_step", "state_file": "missing.safetensors"}], + } + (tmp_path / "processor.json").write_text(json.dumps(config)) + + def fail_hub_download(*args, **kwargs): + pytest.fail("local missing processor state should not call hf_hub_download") + + monkeypatch.setattr("lerobot.processor.pipeline.hf_hub_download", fail_hub_download) + + with pytest.raises(FileNotFoundError, match="missing.safetensors.*local processor pipeline"): + DataProcessorPipeline.from_pretrained(tmp_path, config_filename="processor.json") + finally: + ProcessorStepRegistry.unregister("local_missing_state_step") + + +def test_from_pretrained_local_config_file_missing_state_does_not_call_hub(monkeypatch): + """Local single-file processor configs must also keep missing state resolution local.""" + + @ProcessorStepRegistry.register("local_file_missing_state_step") + class LocalFileMissingStateStep(ProcessorStep): + def __call__(self, transition: EnvTransition) -> EnvTransition: + return transition + + def transform_features( + self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]] + ) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]: + return features + + def load_state_dict(self, state: dict[str, torch.Tensor]) -> None: + pass + + try: + with tempfile.TemporaryDirectory() as tmp_dir: + tmp_path = Path(tmp_dir) + config_path = tmp_path / "processor.json" + config = { + "name": "LocalFileMissingStatePipeline", + "steps": [ + {"registry_name": "local_file_missing_state_step", "state_file": "missing.safetensors"} + ], + } + config_path.write_text(json.dumps(config)) + + def fail_hub_download(*args, **kwargs): + pytest.fail("local missing processor state should not call hf_hub_download") + + monkeypatch.setattr("lerobot.processor.pipeline.hf_hub_download", fail_hub_download) + + with pytest.raises(FileNotFoundError, match="missing.safetensors.*local processor pipeline"): + DataProcessorPipeline.from_pretrained(config_path, config_filename="ignored.json") + finally: + ProcessorStepRegistry.unregister("local_file_missing_state_step") + + +def test_from_pretrained_hub_source_missing_local_state_still_calls_hub(monkeypatch, tmp_path): + """Hub sources still fall back to hf_hub_download for state files.""" + + @ProcessorStepRegistry.register("hub_state_step") + class HubStateStep(ProcessorStep): + def __init__(self): + self.value = torch.tensor(0) + + def __call__(self, transition: EnvTransition) -> EnvTransition: + return transition + + def transform_features( + self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]] + ) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]: + return features + + def load_state_dict(self, state: dict[str, torch.Tensor]) -> None: + self.value = state["value"] + + try: + state_path = tmp_path / "downloaded.safetensors" + save_file({"value": torch.tensor(7)}, state_path) + loaded_config = { + "name": "HubStatePipeline", + "steps": [{"registry_name": "hub_state_step", "state_file": "hub_state.safetensors"}], + } + calls = [] + + def fake_load_config(cls, model_id, config_filename, hub_download_kwargs): + return loaded_config, tmp_path / "hub_cache" + + def fake_hub_download(**kwargs): + calls.append(kwargs) + return str(state_path) + + monkeypatch.setattr(DataProcessorPipeline, "_load_config", classmethod(fake_load_config)) + monkeypatch.setattr("lerobot.processor.pipeline.hf_hub_download", fake_hub_download) + + pipeline = DataProcessorPipeline.from_pretrained("user/repo", config_filename="processor.json") + + assert calls == [ + { + "repo_id": "user/repo", + "filename": "hub_state.safetensors", + "repo_type": "model", + "force_download": False, + "resume_download": None, + "proxies": None, + "token": None, + "cache_dir": None, + "local_files_only": False, + "revision": None, + } + ] + assert pipeline.steps[0].value.item() == 7 + finally: + ProcessorStepRegistry.unregister("hub_state_step") + + # Config Validation Tests From a96540a2c4c443f5e5098ffcd21d0eb7d0be29eb Mon Sep 17 00:00:00 2001 From: Steven Palma Date: Mon, 27 Jul 2026 18:20:39 +0200 Subject: [PATCH 07/10] fix rollout policy revision loading (#4161) Co-authored-by: RaviTeja-Kondeti --- src/lerobot/policies/factory.py | 1 + src/lerobot/policies/groot/processor_groot.py | 5 + src/lerobot/rollout/configs.py | 13 ++- src/lerobot/rollout/context.py | 45 +++++--- tests/test_rollout.py | 105 ++++++++++++++++++ 5 files changed, 154 insertions(+), 15 deletions(-) diff --git a/src/lerobot/policies/factory.py b/src/lerobot/policies/factory.py index 36a0de7ca..1848a6ffd 100644 --- a/src/lerobot/policies/factory.py +++ b/src/lerobot/policies/factory.py @@ -177,6 +177,7 @@ def make_pre_post_processors( return make_groot_pre_post_processors_from_pretrained( config=policy_cfg, pretrained_path=pretrained_path, + revision=pretrained_revision, dataset_stats=kwargs.get("dataset_stats"), dataset_meta=kwargs.get("dataset_meta"), preprocessor_overrides=kwargs.get("preprocessor_overrides"), diff --git a/src/lerobot/policies/groot/processor_groot.py b/src/lerobot/policies/groot/processor_groot.py index 20b3518a3..0bd976a85 100644 --- a/src/lerobot/policies/groot/processor_groot.py +++ b/src/lerobot/policies/groot/processor_groot.py @@ -475,6 +475,7 @@ def make_groot_pre_post_processors_from_pretrained( config: GrootConfig, pretrained_path: str, *, + revision: str | None = None, dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None, dataset_meta: Any | None = None, preprocessor_overrides: dict[str, Any] | None = None, @@ -511,6 +512,7 @@ def make_groot_pre_post_processors_from_pretrained( preprocessor, postprocessor = _load_groot_processor_pipelines( pretrained_path, + revision=revision, preprocessor_overrides=preprocessor_overrides, postprocessor_overrides=postprocessor_overrides, preprocessor_config_filename=preprocessor_config_filename, @@ -526,6 +528,7 @@ def make_groot_pre_post_processors_from_pretrained( def _load_groot_processor_pipelines( pretrained_path: str, *, + revision: str | None, preprocessor_overrides: dict[str, Any], postprocessor_overrides: dict[str, Any], preprocessor_config_filename: str, @@ -540,6 +543,7 @@ def _load_groot_processor_pipelines( preprocessor = PolicyProcessorPipeline.from_pretrained( pretrained_model_name_or_path=pretrained_path, config_filename=preprocessor_config_filename, + revision=revision, overrides=preprocessor_overrides, to_transition=batch_to_transition, to_output=transition_to_batch, @@ -547,6 +551,7 @@ def _load_groot_processor_pipelines( postprocessor = PolicyProcessorPipeline.from_pretrained( pretrained_model_name_or_path=pretrained_path, config_filename=postprocessor_config_filename, + revision=revision, overrides=postprocessor_overrides, to_transition=policy_action_to_transition, to_output=transition_to_policy_action, diff --git a/src/lerobot/rollout/configs.py b/src/lerobot/rollout/configs.py index 639e2ba29..c0b5b345f 100644 --- a/src/lerobot/rollout/configs.py +++ b/src/lerobot/rollout/configs.py @@ -326,8 +326,17 @@ class RolloutConfig: policy_path = parser.get_path_arg("policy") if policy_path: - cli_overrides = parser.get_cli_overrides("policy") - self.policy = PreTrainedConfig.from_pretrained(policy_path, cli_overrides=cli_overrides) + yaml_overrides = parser.get_yaml_overrides("policy") + cli_overrides = parser.get_cli_overrides("policy") or [] + policy_overrides = yaml_overrides + cli_overrides + pretrained_revision = parser.parse_arg("pretrained_revision", cli_overrides) + if pretrained_revision is None: + pretrained_revision = parser.parse_arg("pretrained_revision", yaml_overrides) + self.policy = PreTrainedConfig.from_pretrained( + policy_path, + revision=pretrained_revision, + cli_overrides=policy_overrides, + ) self.policy.pretrained_path = policy_path if self.policy is None: raise ValueError("--policy.path is required for rollout") diff --git a/src/lerobot/rollout/context.py b/src/lerobot/rollout/context.py index 20a7d715a..2bada502e 100644 --- a/src/lerobot/rollout/context.py +++ b/src/lerobot/rollout/context.py @@ -27,7 +27,7 @@ from threading import Event import torch -from lerobot.configs import FeatureType +from lerobot.configs import FeatureType, PreTrainedConfig from lerobot.datasets import ( LeRobotDataset, aggregate_pipeline_dataset_features, @@ -159,6 +159,35 @@ class RolloutContext: # --------------------------------------------------------------------------- +def _load_pretrained_policy(policy_config: PreTrainedConfig) -> PreTrainedPolicy: + """Load policy weights, keeping adapter and base-model revisions independent.""" + pretrained_revision = policy_config.pretrained_revision + policy_class = get_policy_class(policy_config.type) + + if not policy_config.use_peft: + return policy_class.from_pretrained( + policy_config.pretrained_path, + config=policy_config, + revision=pretrained_revision, + ) + + from peft import PeftConfig, PeftModel + + peft_path = policy_config.pretrained_path + peft_config = PeftConfig.from_pretrained(peft_path, revision=pretrained_revision) + policy = policy_class.from_pretrained( + pretrained_name_or_path=peft_config.base_model_name_or_path, + config=policy_config, + revision=peft_config.revision, + ) + return PeftModel.from_pretrained( + policy, + peft_path, + config=peft_config, + revision=pretrained_revision, + ) + + def build_rollout_context( cfg: RolloutConfig, shutdown_event: Event, @@ -176,7 +205,6 @@ def build_rollout_context( # --- 1. Policy (heavy I/O, but no hardware yet) ------------------- logger.info("Loading policy from '%s'...", cfg.policy.pretrained_path) policy_config = cfg.policy - policy_class = get_policy_class(policy_config.type) if hasattr(policy_config, "compile_model"): policy_config.compile_model = cfg.use_torch_compile @@ -187,17 +215,7 @@ def build_rollout_context( "Please use `cpu` or `cuda` backend." ) - if policy_config.use_peft: - from peft import PeftConfig, PeftModel - - peft_path = policy_config.pretrained_path - peft_config = PeftConfig.from_pretrained(peft_path) - policy = policy_class.from_pretrained( - pretrained_name_or_path=peft_config.base_model_name_or_path, config=policy_config - ) - policy = PeftModel.from_pretrained(policy, peft_path, config=peft_config) - else: - policy = policy_class.from_pretrained(policy_config.pretrained_path, config=policy_config) + policy = _load_pretrained_policy(policy_config) if is_rtc: policy.config.rtc_config = cfg.inference.rtc @@ -392,6 +410,7 @@ def build_rollout_context( preprocessor, postprocessor = make_pre_post_processors( policy_cfg=policy_config, pretrained_path=cfg.policy.pretrained_path, + pretrained_revision=policy_config.pretrained_revision, dataset_stats=dataset_stats, preprocessor_overrides={ "device_processor": {"device": cfg.device}, diff --git a/tests/test_rollout.py b/tests/test_rollout.py index 85a29ff4c..c247ff3c1 100644 --- a/tests/test_rollout.py +++ b/tests/test_rollout.py @@ -17,6 +17,8 @@ from __future__ import annotations import dataclasses +import sys +from types import SimpleNamespace from unittest.mock import MagicMock import pytest @@ -106,6 +108,109 @@ def test_sentry_config_defaults(): assert cfg.target_video_file_size_mb is None +def test_rollout_config_passes_policy_pretrained_revision(monkeypatch): + from lerobot.configs import PreTrainedConfig, parser + from lerobot.rollout import RolloutConfig + from tests.mocks.mock_robot import MockRobotConfig + + captured = {} + + def fake_from_pretrained(cls, pretrained_name_or_path, **kwargs): + captured["pretrained_name_or_path"] = pretrained_name_or_path + captured.update(kwargs) + return SimpleNamespace(device="cpu", pretrained_revision=kwargs["revision"]) + + monkeypatch.setattr(parser, "get_yaml_overrides", lambda _: ["--pretrained_revision=yaml-sha"]) + monkeypatch.setattr( + sys, + "argv", + ["lerobot-rollout", "--policy.path=user/policy", "--policy.pretrained_revision=cli-sha"], + ) + monkeypatch.setattr(PreTrainedConfig, "from_pretrained", classmethod(fake_from_pretrained)) + + cfg = RolloutConfig(robot=MockRobotConfig()) + + assert captured["pretrained_name_or_path"] == "user/policy" + assert captured["revision"] == "cli-sha" + assert captured["cli_overrides"] == [ + "--pretrained_revision=yaml-sha", + "--pretrained_revision=cli-sha", + ] + assert cfg.policy.pretrained_path == "user/policy" + assert cfg.policy.pretrained_revision == "cli-sha" + + +def test_load_pretrained_policy_passes_revision(monkeypatch): + import lerobot.rollout.context as rollout_context + + policy_config = SimpleNamespace( + type="mock", + use_peft=False, + pretrained_path="user/policy", + pretrained_revision="policy-sha", + ) + policy_class = MagicMock() + loaded_policy = MagicMock() + policy_class.from_pretrained.return_value = loaded_policy + monkeypatch.setattr(rollout_context, "get_policy_class", lambda _: policy_class) + + policy = rollout_context._load_pretrained_policy(policy_config) + + assert policy is loaded_policy + policy_class.from_pretrained.assert_called_once_with( + "user/policy", + config=policy_config, + revision="policy-sha", + ) + + +def test_load_pretrained_peft_policy_keeps_adapter_and_base_revisions_separate(monkeypatch): + import lerobot.rollout.context as rollout_context + + policy_config = SimpleNamespace( + type="mock", + use_peft=True, + pretrained_path="user/adapter", + pretrained_revision="adapter-sha", + ) + policy_class = MagicMock() + base_policy = MagicMock() + policy_class.from_pretrained.return_value = base_policy + monkeypatch.setattr(rollout_context, "get_policy_class", lambda _: policy_class) + + peft_config = SimpleNamespace( + base_model_name_or_path="user/base-policy", + revision="base-sha", + ) + peft_config_from_pretrained = MagicMock(return_value=peft_config) + adapted_policy = MagicMock() + peft_model_from_pretrained = MagicMock(return_value=adapted_policy) + monkeypatch.setitem( + sys.modules, + "peft", + SimpleNamespace( + PeftConfig=SimpleNamespace(from_pretrained=peft_config_from_pretrained), + PeftModel=SimpleNamespace(from_pretrained=peft_model_from_pretrained), + ), + ) + + policy = rollout_context._load_pretrained_policy(policy_config) + + assert policy is adapted_policy + peft_config_from_pretrained.assert_called_once_with("user/adapter", revision="adapter-sha") + policy_class.from_pretrained.assert_called_once_with( + pretrained_name_or_path="user/base-policy", + config=policy_config, + revision="base-sha", + ) + peft_model_from_pretrained.assert_called_once_with( + base_policy, + "user/adapter", + config=peft_config, + revision="adapter-sha", + ) + + # --------------------------------------------------------------------------- # RolloutRingBuffer # --------------------------------------------------------------------------- From fd53716688fc19d62369c6397d7682a0f0386208 Mon Sep 17 00:00:00 2001 From: Thomas Landeg <63191808+landegt@users.noreply.github.com> Date: Mon, 27 Jul 2026 09:42:17 -0700 Subject: [PATCH 08/10] fix(envs): make metaworld seeding reproducible (#3727) Co-authored-by: Pepijn <138571049+pkooij@users.noreply.github.com> --- src/lerobot/envs/metaworld.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/lerobot/envs/metaworld.py b/src/lerobot/envs/metaworld.py index bffcf6b6e..a32acbff1 100644 --- a/src/lerobot/envs/metaworld.py +++ b/src/lerobot/envs/metaworld.py @@ -155,6 +155,7 @@ class MetaworldEnv(gym.Env): env.model.cam_pos[2] = [0.75, 0.075, 0.7] env.reset() env._freeze_rand_vec = False # otherwise no randomization + env.seeded_rand_vec = True # use seeded RNG so reset(seed=X) controls object positions self._env = env def render(self) -> np.ndarray: @@ -220,6 +221,8 @@ class MetaworldEnv(gym.Env): self._ensure_env() super().reset(seed=seed) + if seed is not None: + self._env.seed(seed) raw_obs, info = self._env.reset(seed=seed) observation = self._format_raw_obs(raw_obs) From 95256d766d7dd686e1e7e99d42d05cf207a9abc2 Mon Sep 17 00:00:00 2001 From: Xingdong Zuo Date: Tue, 28 Jul 2026 02:29:44 +0900 Subject: [PATCH 09/10] feat(lekiwi): support LeKiwi in lerobot-replay CLI (#3739) Register the `lekiwi` robot module in `lerobot_replay.py` so episodes can be replayed on a LeKiwi via `--robot.type=lekiwi_client`. The module is already registered in `lerobot_calibrate.py` and `lerobot_setup_motors.py`; this fills the gap so the replay CLI recognizes the same robot. Replayed actions are loaded from the dataset as torch tensors, which `json.dumps` cannot serialize when `LeKiwiClient.send_action` ships them over ZMQ. Coerce each action value to a plain float before sending. This is scoped to the LeKiwi network client and does not affect any other robot. Co-authored-by: Steven Palma --- src/lerobot/robots/lekiwi/lekiwi_client.py | 4 ++++ src/lerobot/scripts/lerobot_replay.py | 1 + 2 files changed, 5 insertions(+) diff --git a/src/lerobot/robots/lekiwi/lekiwi_client.py b/src/lerobot/robots/lekiwi/lekiwi_client.py index 1bc3dadc4..2850842a2 100644 --- a/src/lerobot/robots/lekiwi/lekiwi_client.py +++ b/src/lerobot/robots/lekiwi/lekiwi_client.py @@ -323,6 +323,10 @@ class LeKiwiClient(Robot): np.ndarray: the action sent to the motors, potentially clipped. """ + # Action values may be torch tensors (e.g. replayed from a dataset) or numpy + # scalars; json.dumps only serializes Python primitives, so coerce each value to a + # plain float before sending. + action = {key: float(value) for key, value in action.items()} self.zmq_cmd_socket.send_string(json.dumps(action)) # action is in motor space # TODO(Steven): Remove the np conversion when it is possible to record a non-numpy array value diff --git a/src/lerobot/scripts/lerobot_replay.py b/src/lerobot/scripts/lerobot_replay.py index 1851f7c2b..b2ab2b731 100644 --- a/src/lerobot/scripts/lerobot_replay.py +++ b/src/lerobot/scripts/lerobot_replay.py @@ -61,6 +61,7 @@ from lerobot.robots import ( # noqa: F401 earthrover_mini_plus, hope_jr, koch_follower, + lekiwi, make_robot_from_config, omx_follower, openarm_follower, From 95211b98f1cd6b638bda84a8d28f9e41323229dd Mon Sep 17 00:00:00 2001 From: Steven Palma Date: Tue, 28 Jul 2026 00:42:55 +0200 Subject: [PATCH 10/10] feat(config): add `multiprocessing` option to `DataLoader` context and sets `spawn` as default (#4139) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Add dataloader_multiprocessing_context, default to spawn Make the DataLoader multiprocessing start method configurable on TrainPipelineConfig and default it to 'spawn'. The previous default (fork on Linux) is unsafe with libraries that hold non-fork-safe state in the parent process — common ones in this codebase are PyAV, torchcodec, and the ffmpeg shared libs they wrap. Symptoms reported in #2488, #2209, and observed locally include: - multiprocessing.context.AuthenticationError: digest received was wrong - RuntimeError: Pin memory thread exited unexpectedly - RuntimeError: DataLoader worker exited unexpectedly - Random SIGSEGV inside worker processes during video decode Switching to spawn re-imports modules cleanly in each worker and eliminates these failure modes. Added the setting as a config field rather than hard-coding so users on platforms where fork is preferred can opt back in via --dataloader-multiprocessing-context=fork. * Address review: shorten config comment, note spawn startup tradeoff Per @jashshah999, mention that spawn workers re-import modules and so add some startup time vs fork. Also trim the failure-mode dump from the inline comment — the linked issue covers the symptoms in detail. Co-Authored-By: Claude Opus 4.7 (1M context) * chore(scripts): add multiprocessing_context safeguards * chore(config): add libs note --------- Co-authored-by: 0o8o0-blip <0o8o0-blip@users.noreply.github.com> --- src/lerobot/configs/train.py | 18 ++++++++++++++++++ src/lerobot/scripts/lerobot_train.py | 16 ++++++++++++---- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/src/lerobot/configs/train.py b/src/lerobot/configs/train.py index e3d354691..e92247188 100644 --- a/src/lerobot/configs/train.py +++ b/src/lerobot/configs/train.py @@ -14,6 +14,7 @@ import builtins import datetime as dt import json +import multiprocessing import os import tempfile from dataclasses import dataclass, field @@ -101,6 +102,12 @@ class TrainPipelineConfig(HubMixin): batch_size: int = 8 prefetch_factor: int = 4 persistent_workers: bool = True + # DataLoader worker start method. "spawn" is safer than "fork" with + # non-fork-safe libs (PyAV / torchcodec / ffmpeg), but adds some + # worker-startup time per run since workers re-import modules instead + # of inheriting parent state. Override with `--dataloader_multiprocessing_context=fork` + # when appropriate, or set it to `null` to use Python's platform default. + dataloader_multiprocessing_context: str | None = "spawn" steps: int = 100_000 # Run policy in the simulation environment every N steps to measure reward/success (0 = disabled). env_eval_freq: int = 20_000 @@ -212,6 +219,17 @@ class TrainPipelineConfig(HubMixin): self.reward_model.pretrained_path = str(policy_dir) def validate(self) -> None: + available_contexts = multiprocessing.get_all_start_methods() + if ( + self.dataloader_multiprocessing_context is not None + and self.dataloader_multiprocessing_context not in available_contexts + ): + raise ValueError( + "`dataloader_multiprocessing_context` must be None or one of " + f"{available_contexts} on this platform, got " + f"{self.dataloader_multiprocessing_context!r}." + ) + self._resolve_pretrained_from_cli() if self.policy is None and self.reward_model is None: diff --git a/src/lerobot/scripts/lerobot_train.py b/src/lerobot/scripts/lerobot_train.py index ec5565cf4..8bfa16a98 100644 --- a/src/lerobot/scripts/lerobot_train.py +++ b/src/lerobot/scripts/lerobot_train.py @@ -71,6 +71,16 @@ from lerobot.utils.utils import ( from .lerobot_eval import eval_policy_all +def _dataloader_worker_kwargs(cfg: TrainPipelineConfig) -> dict[str, Any]: + """Return worker-only DataLoader options, disabling them for single-process loading.""" + workers_enabled = cfg.num_workers > 0 + return { + "prefetch_factor": cfg.prefetch_factor if workers_enabled else None, + "persistent_workers": cfg.persistent_workers and workers_enabled, + "multiprocessing_context": cfg.dataloader_multiprocessing_context if workers_enabled else None, + } + + def update_policy( train_metrics: MetricsTracker, policy: PreTrainedPolicy, @@ -473,8 +483,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None): pin_memory=device.type == "cuda", drop_last=False, collate_fn=collate_fn, - prefetch_factor=cfg.prefetch_factor if cfg.num_workers > 0 else None, - persistent_workers=cfg.persistent_workers and cfg.num_workers > 0, + **_dataloader_worker_kwargs(cfg), ) # Build eval dataloader if a held-out split exists @@ -500,8 +509,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None): pin_memory=device.type == "cuda", drop_last=False, collate_fn=eval_collate_fn, - prefetch_factor=cfg.prefetch_factor if cfg.num_workers > 0 else None, - persistent_workers=cfg.persistent_workers and cfg.num_workers > 0, + **_dataloader_worker_kwargs(cfg), ) # Prepare everything with accelerator