fix(hub): pin pretrained artifacts to one commit

* 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) <noreply@anthropic.com>

* chore(scripts): add multiprocessing_context safeguards

* chore(config): add libs note

---------

Co-authored-by: 0o8o0-blip <0o8o0-blip@users.noreply.github.com>

.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: ["*"]

src/lerobot/configs/eval.py

@@ -48,8 +48,13 @@ class EvalPipelineConfig:
         if policy_path:
             yaml_overrides = parser.get_yaml_overrides("policy")
             cli_overrides = parser.get_cli_overrides("policy") or []
+            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, cli_overrides=yaml_overrides + cli_overrides
+                policy_path,
+                revision=pretrained_revision,
+                cli_overrides=yaml_overrides + cli_overrides,
             )
             self.policy.pretrained_path = Path(policy_path)
 

src/lerobot/configs/policies.py

@@ -29,7 +29,7 @@ from huggingface_hub.errors import HfHubHTTPError
 from lerobot.optim import LRSchedulerConfig, OptimizerConfig
 from lerobot.utils.constants import ACTION, OBS_STATE
 from lerobot.utils.device_utils import auto_select_torch_device, is_amp_available, is_torch_device_available
-from lerobot.utils.hub import HubMixin
+from lerobot.utils.hub import HubMixin, extract_commit_hash
 
 from .types import FeatureType, PolicyFeature
 
@@ -82,6 +82,26 @@ class PreTrainedConfig(draccus.ChoiceRegistry, HubMixin, abc.ABC):  # type: igno
     # Optional Hub revision (commit hash, branch, or tag) to pin the pretrained model version.
     pretrained_revision: str | None = None
 
+    @property
+    def _commit_hash(self) -> str | None:
+        """Resolved Hub commit for this runtime load; never serialized."""
+        return self.__dict__.get("_runtime_commit_hash")
+
+    @property
+    def _commit_hash_source(self) -> str | None:
+        """Hub repo whose revision resolved to ``_commit_hash``."""
+        return self.__dict__.get("_runtime_commit_hash_source")
+
+    def _set_hub_commit_hash(self, commit_hash: str | None, source: str | None) -> None:
+        self.__dict__["_runtime_commit_hash"] = commit_hash
+        self.__dict__["_runtime_commit_hash_source"] = source if commit_hash is not None else None
+
+    def get_hub_revision(self, source: str | Path | None, revision: str | None = None) -> str | None:
+        """Return the pinned revision when ``source`` owns the resolved commit."""
+        if self._commit_hash is not None and self._commit_hash_source == str(source):
+            return self._commit_hash
+        return revision
+
     def __post_init__(self) -> None:
         if not self.device or not is_torch_device_available(self.device):
             auto_device = auto_select_torch_device()
@@ -182,7 +202,8 @@ class PreTrainedConfig(draccus.ChoiceRegistry, HubMixin, abc.ABC):  # type: igno
     ) -> T:
         model_id = str(pretrained_name_or_path)
         config_file: str | None = None
-        if Path(model_id).is_dir():
+        is_local = Path(model_id).is_dir()
+        if is_local:
             if CONFIG_NAME in os.listdir(model_id):
                 config_file = os.path.join(model_id, CONFIG_NAME)
             else:
@@ -208,8 +229,12 @@ class PreTrainedConfig(draccus.ChoiceRegistry, HubMixin, abc.ABC):  # type: igno
         if config_file is None:
             raise FileNotFoundError(f"{CONFIG_NAME} not found in {model_id}")
 
+        commit_hash = None if is_local else extract_commit_hash(config_file, revision)
         with open(config_file) as f:
             config = json.load(f)
+        # Runtime Hub metadata must never become part of the serialized config schema.
+        config.pop("_commit_hash", None)
+        config.pop("_commit_hash_source", None)
 
         # Resolve the concrete config subclass from the serialized "type" tag, then parse
         # the config (with CLI overrides) directly for that class. The "type" key is
@@ -231,4 +256,6 @@ class PreTrainedConfig(draccus.ChoiceRegistry, HubMixin, abc.ABC):  # type: igno
 
         cli_overrides = policy_kwargs.pop("cli_overrides", [])
         with draccus.config_type("json"):
-            return draccus.parse(config_cls, config_file, args=cli_overrides)
+            parsed_config = draccus.parse(config_cls, config_file, args=cli_overrides)
+        parsed_config._set_hub_commit_hash(commit_hash, model_id)
+        return parsed_config

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
@@ -165,8 +172,16 @@ class TrainPipelineConfig(HubMixin):
             )
             self.reward_model.pretrained_path = str(Path(reward_model_path))
         elif policy_path:
-            overrides = parser.get_yaml_overrides("policy") + (parser.get_cli_overrides("policy") or [])
-            self.policy = PreTrainedConfig.from_pretrained(policy_path, cli_overrides=overrides)
+            yaml_overrides = parser.get_yaml_overrides("policy")
+            cli_overrides = parser.get_cli_overrides("policy") or []
+            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=yaml_overrides + cli_overrides,
+            )
             self.policy.pretrained_path = Path(policy_path)
         elif self.resume:
             self._resolve_resume_checkpoint()
@@ -212,6 +227,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:

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()
 

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)

src/lerobot/policies/factory.py

@@ -171,12 +171,16 @@ def make_pre_post_processors(
         ValueError: If no processor factory exists for the given policy configuration type.
     """
     if pretrained_path:
+        revision_resolver = getattr(policy_cfg, "get_hub_revision", None)
+        if callable(revision_resolver):
+            pretrained_revision = revision_resolver(pretrained_path, pretrained_revision)
         if isinstance(policy_cfg, GrootConfig):
             from .groot.processor_groot import make_groot_pre_post_processors_from_pretrained
 
             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"),

src/lerobot/policies/groot/modeling_groot.py

@@ -37,6 +37,7 @@ from torch import Tensor
 
 from lerobot.configs import FeatureType, PolicyFeature
 from lerobot.utils.constants import ACTION, OBS_IMAGES
+from lerobot.utils.hub import extract_commit_hash
 from lerobot.utils.import_utils import _transformers_available, require_package
 
 from ..pretrained import PreTrainedPolicy
@@ -195,6 +196,8 @@ class GrootPolicy(PreTrainedPolicy):
         )
 
         model_id = str(pretrained_name_or_path)
+        if config is not None:
+            revision = config.get_hub_revision(model_id, revision)
         is_finetuned_checkpoint = False
 
         # Check if this is a fine-tuned LeRobot checkpoint (has model.safetensors)
@@ -204,7 +207,7 @@ class GrootPolicy(PreTrainedPolicy):
             else:
                 # Try to download the safetensors file to check if it exists
                 try:
-                    hf_hub_download(
+                    resolved_model_file = hf_hub_download(
                         repo_id=model_id,
                         filename=SAFETENSORS_SINGLE_FILE,
                         revision=revision,
@@ -214,6 +217,10 @@ class GrootPolicy(PreTrainedPolicy):
                         token=token,
                         local_files_only=local_files_only,
                     )
+                    resolved_commit_hash = extract_commit_hash(resolved_model_file, revision)
+                    revision = resolved_commit_hash or revision
+                    if config is not None and config._commit_hash is None:
+                        config._set_hub_commit_hash(resolved_commit_hash, model_id)
                     is_finetuned_checkpoint = True
                 except HfHubHTTPError:
                     is_finetuned_checkpoint = False

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,

src/lerobot/policies/pi0/modeling_pi0.py

@@ -53,6 +53,7 @@ from lerobot.utils.constants import (
     OBS_LANGUAGE_TOKENS,
     OBS_STATE,
 )
+from lerobot.utils.hub import extract_commit_hash
 
 from ..common.flow_matching import euler_integrate, sample_noise, sample_time_beta
 from ..common.vla_utils import (
@@ -814,6 +815,8 @@ class PI0Policy(PreTrainedPolicy):
                 **kwargs,
             )
 
+        revision = config.get_hub_revision(pretrained_name_or_path, revision)
+
         # Initialize model without loading weights
         # Check if dataset_stats were provided in kwargs
         model = cls(config, **kwargs)
@@ -832,9 +835,13 @@ class PI0Policy(PreTrainedPolicy):
                     resume_download=kwargs.get("resume_download"),
                     proxies=kwargs.get("proxies"),
                     token=kwargs.get("token"),
-                    revision=kwargs.get("revision"),
+                    revision=revision,
                     local_files_only=kwargs.get("local_files_only", False),
                 )
+                if config._commit_hash is None:
+                    config._set_hub_commit_hash(
+                        extract_commit_hash(resolved_file, revision), str(pretrained_name_or_path)
+                    )
                 from safetensors.torch import load_file
 
                 original_state_dict = load_file(resolved_file)

src/lerobot/policies/pi05/modeling_pi05.py

@@ -50,6 +50,7 @@ from lerobot.utils.constants import (
     OBS_LANGUAGE_ATTENTION_MASK,
     OBS_LANGUAGE_TOKENS,
 )
+from lerobot.utils.hub import extract_commit_hash
 
 from ..common.flow_matching import euler_integrate, sample_noise, sample_time_beta
 from ..common.vla_utils import (
@@ -779,6 +780,8 @@ class PI05Policy(PreTrainedPolicy):
                 **kwargs,
             )
 
+        revision = config.get_hub_revision(pretrained_name_or_path, revision)
+
         # Initialize model without loading weights
         # Check if dataset_stats were provided in kwargs
         model = cls(config, **kwargs)
@@ -797,9 +800,13 @@ class PI05Policy(PreTrainedPolicy):
                     resume_download=kwargs.get("resume_download"),
                     proxies=kwargs.get("proxies"),
                     token=kwargs.get("token"),
-                    revision=kwargs.get("revision"),
+                    revision=revision,
                     local_files_only=kwargs.get("local_files_only", False),
                 )
+                if config._commit_hash is None:
+                    config._set_hub_commit_hash(
+                        extract_commit_hash(resolved_file, revision), str(pretrained_name_or_path)
+                    )
                 from safetensors.torch import load_file
 
                 original_state_dict = load_file(resolved_file)

src/lerobot/policies/pi0_fast/modeling_pi0_fast.py

@@ -55,6 +55,7 @@ from lerobot.utils.constants import (
     OBS_LANGUAGE_ATTENTION_MASK,
     OBS_LANGUAGE_TOKENS,
 )
+from lerobot.utils.hub import extract_commit_hash
 
 from ..common.vla_utils import pad_vector, prepare_attention_masks_4d, resize_with_pad_torch
 from ..pretrained import PreTrainedPolicy, T
@@ -808,6 +809,8 @@ class PI0FastPolicy(PreTrainedPolicy):
                 **kwargs,
             )
 
+        revision = config.get_hub_revision(pretrained_name_or_path, revision)
+
         # Initialize model without loading weights
         # Check if dataset_stats were provided in kwargs
         model = cls(config, **kwargs)
@@ -826,9 +829,13 @@ class PI0FastPolicy(PreTrainedPolicy):
                     resume_download=kwargs.get("resume_download"),
                     proxies=kwargs.get("proxies"),
                     token=kwargs.get("token"),
-                    revision=kwargs.get("revision"),
+                    revision=revision,
                     local_files_only=kwargs.get("local_files_only", False),
                 )
+                if config._commit_hash is None:
+                    config._set_hub_commit_hash(
+                        extract_commit_hash(resolved_file, revision), str(pretrained_name_or_path)
+                    )
                 from safetensors.torch import load_file
 
                 original_state_dict = load_file(resolved_file)

src/lerobot/policies/pretrained.py

@@ -33,7 +33,7 @@ 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 lerobot.utils.hub import HubMixin, extract_commit_hash
 
 from .utils import log_model_loading_keys
 
@@ -190,6 +190,7 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
                 **kwargs,
             )
         model_id = str(pretrained_name_or_path)
+        revision = config.get_hub_revision(model_id, revision)
         instance = cls(config, **kwargs)
         if os.path.isdir(model_id):
             print("Loading weights from local directory")
@@ -208,6 +209,8 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
                     token=token,
                     local_files_only=local_files_only,
                 )
+                if config._commit_hash is None:
+                    config._set_hub_commit_hash(extract_commit_hash(model_file, revision), model_id)
                 policy = cls._load_as_safetensor(instance, model_file, config.device, strict)
             except HfHubHTTPError as e:
                 raise FileNotFoundError(

src/lerobot/policies/xvla/modeling_xvla.py

@@ -31,6 +31,7 @@ from torch import Tensor, nn
 
 from lerobot.configs import PreTrainedConfig
 from lerobot.utils.constants import ACTION, OBS_LANGUAGE_TOKENS, OBS_STATE
+from lerobot.utils.hub import extract_commit_hash
 from lerobot.utils.import_utils import _transformers_available, require_package
 
 from ..common.vla_utils import pad_vector, resize_with_pad
@@ -459,6 +460,7 @@ class XVLAPolicy(PreTrainedPolicy):
             )
 
         model_id = str(pretrained_name_or_path)
+        revision = config.get_hub_revision(model_id, revision)
         instance = cls(config, **kwargs)
         # step 2: locate model.safetensors
         if os.path.isdir(model_id):
@@ -480,6 +482,8 @@ class XVLAPolicy(PreTrainedPolicy):
                     token=token,
                     local_files_only=local_files_only,
                 )
+                if config._commit_hash is None:
+                    config._set_hub_commit_hash(extract_commit_hash(model_file, revision), model_id)
             except HfHubHTTPError as e:
                 raise FileNotFoundError(f"model.safetensors not found on the Hub at {model_id}") from e
 

src/lerobot/processor/delta_action_processor.py

@@ -132,10 +132,20 @@ class MapDeltaActionToRobotActionStep(RobotActionProcessorStep):
     def transform_features(
         self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
     ) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
-        for axis in ["x", "y", "z", "gripper"]:
+        for axis in ["x", "y", "z"]:
             features[PipelineFeatureType.ACTION].pop(f"delta_{axis}", None)
+        features[PipelineFeatureType.ACTION].pop("gripper", None)
 
-        for feat in ["enabled", "target_x", "target_y", "target_z", "target_wx", "target_wy", "target_wz"]:
+        for feat in [
+            "enabled",
+            "target_x",
+            "target_y",
+            "target_z",
+            "target_wx",
+            "target_wy",
+            "target_wz",
+            "gripper_vel",
+        ]:
             features[PipelineFeatureType.ACTION][f"{feat}"] = PolicyFeature(
                 type=FeatureType.ACTION, shape=(1,)
             )

src/lerobot/processor/pipeline.py

@@ -47,7 +47,7 @@ from safetensors.torch import load_file, save_file
 from lerobot.configs import PipelineFeatureType, PolicyFeature
 from lerobot.types import EnvAction, EnvTransition, PolicyAction, RobotAction, RobotObservation, TransitionKey
 from lerobot.utils.constants import HF_LEROBOT_HOME
-from lerobot.utils.hub import HubMixin
+from lerobot.utils.hub import HubMixin, extract_commit_hash
 
 from .converters import batch_to_transition, create_transition, transition_to_batch
 
@@ -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,
@@ -725,13 +727,17 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
 
         # 1. Load configuration using simplified 3-way logic
         loaded_config, base_path = cls._load_config(model_id, config_filename, hub_download_kwargs)
+        if not is_local_source:
+            commit_hash = extract_commit_hash(base_path, revision)
+            if commit_hash is not None:
+                hub_download_kwargs["revision"] = commit_hash
 
         # 2. Validate configuration and handle migration
         cls._validate_loaded_config(model_id, loaded_config, config_filename)
 
         # 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 +927,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 +951,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 +969,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 +983,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 +1149,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 +1168,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 +1189,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 +1203,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(

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

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")

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,40 @@ class RolloutContext:
 # ---------------------------------------------------------------------------
 
 
+def _load_pretrained_policy(policy_config: PreTrainedConfig) -> PreTrainedPolicy:
+    """Load policy weights, keeping adapter and base-model revisions independent."""
+    revision_resolver = getattr(policy_config, "get_hub_revision", None)
+    pretrained_revision = (
+        revision_resolver(policy_config.pretrained_path, policy_config.pretrained_revision)
+        if callable(revision_resolver)
+        else 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 +210,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 +220,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 +415,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},

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(

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):

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()
 

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,

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

src/lerobot/utils/hub.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import builtins
+import re
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import Any, TypeVar
@@ -23,6 +24,29 @@ from huggingface_hub.utils import validate_hf_hub_args
 from .constants import CHECKPOINTS_DIR
 
 T = TypeVar("T", bound="HubMixin")
+REGEX_COMMIT_HASH = re.compile(r"^[0-9a-f]{40}$")
+
+
+def extract_commit_hash(resolved_file: str | Path | None, revision: str | None = None) -> str | None:
+    """Extract the immutable commit hash backing a resolved Hub file.
+
+    Hub cache paths contain ``snapshots/<commit_hash>/``. If the requested
+    revision is already a full commit hash, use it as a fallback for custom
+    cache layouts that do not expose the standard snapshot path.
+    """
+    if resolved_file is not None:
+        path_parts = Path(resolved_file).parts
+        try:
+            snapshot_index = path_parts.index("snapshots")
+            commit_hash = path_parts[snapshot_index + 1]
+            if REGEX_COMMIT_HASH.fullmatch(commit_hash):
+                return commit_hash
+        except (ValueError, IndexError):
+            pass
+
+    if revision is not None and REGEX_COMMIT_HASH.fullmatch(revision):
+        return revision
+    return None
 
 
 def find_latest_hub_checkpoint(

tests/configs/test_policy_revision_pinning.py

@@ -0,0 +1,77 @@
+# Copyright 2026 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.
+
+import json
+from dataclasses import dataclass
+
+from lerobot.configs import PreTrainedConfig
+
+
+@PreTrainedConfig.register_subclass("revision_pinning_test")
+@dataclass
+class RevisionPinningTestConfig(PreTrainedConfig):
+    @property
+    def observation_delta_indices(self) -> list | None:
+        return None
+
+    @property
+    def action_delta_indices(self) -> list | None:
+        return None
+
+    @property
+    def reward_delta_indices(self) -> list | None:
+        return None
+
+    def get_optimizer_preset(self):
+        raise NotImplementedError
+
+    def get_scheduler_preset(self):
+        raise NotImplementedError
+
+    def validate_features(self) -> None:
+        pass
+
+
+def test_pretrained_config_pins_resolved_hub_commit(monkeypatch, tmp_path):
+    commit_hash = "a" * 40
+    snapshot_dir = tmp_path / "models--user--policy" / "snapshots" / commit_hash
+    RevisionPinningTestConfig(device="cpu").save_pretrained(snapshot_dir)
+    calls = []
+
+    def fake_hub_download(**kwargs):
+        calls.append(kwargs)
+        return str(snapshot_dir / "config.json")
+
+    monkeypatch.setattr("lerobot.configs.policies.hf_hub_download", fake_hub_download)
+
+    config = PreTrainedConfig.from_pretrained("user/policy", revision="main")
+
+    assert calls[0]["revision"] == "main"
+    assert config._commit_hash == commit_hash
+    assert config._commit_hash_source == "user/policy"
+    assert config.get_hub_revision("user/policy", "main") == commit_hash
+    assert config.get_hub_revision("user/base-policy", "base-tag") == "base-tag"
+
+
+def test_runtime_commit_hash_is_not_serialized(tmp_path):
+    config = RevisionPinningTestConfig(device="cpu")
+    config._set_hub_commit_hash("a" * 40, "user/policy")
+
+    config.save_pretrained(tmp_path)
+
+    serialized_config = json.loads((tmp_path / "config.json").read_text())
+    assert "_commit_hash" not in serialized_config
+    assert "_commit_hash_source" not in serialized_config
+    assert "_runtime_commit_hash" not in serialized_config
+    assert "_runtime_commit_hash_source" not in serialized_config

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(

tests/policies/test_pretrained_revision_pinning.py

@@ -0,0 +1,122 @@
+# Copyright 2026 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.
+
+import torch
+
+from lerobot.policies import factory
+from lerobot.policies.act.configuration_act import ACTConfig
+from lerobot.policies.pretrained import PreTrainedPolicy
+from lerobot.processor import PolicyProcessorPipeline
+
+
+class _MinimalPolicy(PreTrainedPolicy):
+    config_class = ACTConfig
+    name = "minimal_revision_test"
+
+    def get_optim_params(self) -> dict:
+        return {}
+
+    def reset(self) -> None:
+        pass
+
+    def forward(self, batch: dict[str, torch.Tensor]) -> tuple[torch.Tensor, dict | None]:
+        return torch.tensor(0), None
+
+    def predict_action_chunk(self, batch: dict[str, torch.Tensor], **kwargs) -> torch.Tensor:
+        return torch.tensor(0)
+
+    def select_action(self, batch: dict[str, torch.Tensor], **kwargs) -> torch.Tensor:
+        return torch.tensor(0)
+
+
+def _skip_safetensor_loading(monkeypatch):
+    monkeypatch.setattr(
+        _MinimalPolicy,
+        "_load_as_safetensor",
+        classmethod(lambda cls, model, model_file, map_location, strict: model),
+    )
+
+
+def test_policy_weights_use_config_commit_hash(monkeypatch):
+    config = ACTConfig(device="cpu")
+    config._set_hub_commit_hash("a" * 40, "user/policy")
+    calls = []
+
+    def fake_hub_download(**kwargs):
+        calls.append(kwargs)
+        return "/unused/model.safetensors"
+
+    monkeypatch.setattr("lerobot.policies.pretrained.hf_hub_download", fake_hub_download)
+    _skip_safetensor_loading(monkeypatch)
+
+    _MinimalPolicy.from_pretrained("user/policy", config=config, revision="main")
+
+    assert calls[0]["revision"] == "a" * 40
+
+
+def test_policy_does_not_reuse_commit_hash_for_another_repo(monkeypatch):
+    config = ACTConfig(device="cpu")
+    config._set_hub_commit_hash("a" * 40, "user/adapter")
+    calls = []
+
+    def fake_hub_download(**kwargs):
+        calls.append(kwargs)
+        return "/unused/model.safetensors"
+
+    monkeypatch.setattr("lerobot.policies.pretrained.hf_hub_download", fake_hub_download)
+    _skip_safetensor_loading(monkeypatch)
+
+    _MinimalPolicy.from_pretrained("user/base-policy", config=config, revision="base-tag")
+
+    assert calls[0]["revision"] == "base-tag"
+
+
+def test_policy_records_weight_commit_for_explicit_config(monkeypatch):
+    commit_hash = "a" * 40
+    config = ACTConfig(device="cpu")
+
+    def fake_hub_download(**kwargs):
+        return f"/cache/models--user--policy/snapshots/{commit_hash}/model.safetensors"
+
+    monkeypatch.setattr("lerobot.policies.pretrained.hf_hub_download", fake_hub_download)
+    _skip_safetensor_loading(monkeypatch)
+
+    _MinimalPolicy.from_pretrained("user/policy", config=config, revision="main")
+
+    assert config._commit_hash == commit_hash
+    assert config._commit_hash_source == "user/policy"
+
+
+def test_processor_factory_uses_config_commit_hash(monkeypatch):
+    config = ACTConfig(device="cpu")
+    config._set_hub_commit_hash("a" * 40, "user/policy")
+    calls = []
+
+    def fake_from_pretrained(cls, **kwargs):
+        calls.append(kwargs)
+        return PolicyProcessorPipeline(steps=[])
+
+    monkeypatch.setattr(
+        factory.PolicyProcessorPipeline,
+        "from_pretrained",
+        classmethod(fake_from_pretrained),
+    )
+
+    factory.make_pre_post_processors(
+        config,
+        pretrained_path="user/policy",
+        pretrained_revision="main",
+    )
+
+    assert [call["revision"] for call in calls] == ["a" * 40, "a" * 40]

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,202 @@ 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")
+
+
+def test_from_pretrained_pins_hub_state_to_config_commit(monkeypatch, tmp_path):
+    """A mutable processor revision is resolved once and reused for state files."""
+
+    @ProcessorStepRegistry.register("pinned_hub_state_step")
+    class PinnedHubStateStep(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:
+        commit_hash = "a" * 40
+        snapshot_dir = tmp_path / "models--user--policy" / "snapshots" / commit_hash
+        snapshot_dir.mkdir(parents=True)
+        config_path = snapshot_dir / "processor.json"
+        config_path.write_text(
+            json.dumps(
+                {
+                    "name": "PinnedHubStatePipeline",
+                    "steps": [
+                        {
+                            "registry_name": "pinned_hub_state_step",
+                            "state_file": "hub_state.safetensors",
+                        }
+                    ],
+                }
+            )
+        )
+        state_path = tmp_path / "downloaded.safetensors"
+        save_file({"value": torch.tensor(7)}, state_path)
+        calls = []
+
+        def fake_hub_download(**kwargs):
+            calls.append(kwargs)
+            if kwargs["filename"] == "processor.json":
+                return str(config_path)
+            return str(state_path)
+
+        monkeypatch.setattr("lerobot.processor.pipeline.hf_hub_download", fake_hub_download)
+
+        pipeline = DataProcessorPipeline.from_pretrained(
+            "user/policy",
+            config_filename="processor.json",
+            revision="main",
+        )
+
+        assert calls[0]["revision"] == "main"
+        assert calls[1]["revision"] == commit_hash
+        assert pipeline.steps[0].value.item() == 7
+    finally:
+        ProcessorStepRegistry.unregister("pinned_hub_state_step")
+
+
 # Config Validation Tests
 
 

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
 # ---------------------------------------------------------------------------

tests/utils/test_hub.py

@@ -14,7 +14,7 @@
 
 from unittest.mock import MagicMock
 
-from lerobot.utils.hub import find_latest_hub_checkpoint
+from lerobot.utils.hub import extract_commit_hash, find_latest_hub_checkpoint
 
 
 def _patch_list_files(monkeypatch, files):
@@ -52,3 +52,20 @@ def test_find_latest_hub_checkpoint_ignores_non_step_entries(monkeypatch):
 def test_find_latest_hub_checkpoint_none_when_no_checkpoints(monkeypatch):
     _patch_list_files(monkeypatch, ["config.json", "model.safetensors"])
     assert find_latest_hub_checkpoint("u/run") is None
+
+
+def test_extract_commit_hash_from_hub_snapshot_path():
+    commit_hash = "a" * 40
+    resolved_file = f"/cache/models--user--policy/snapshots/{commit_hash}/config.json"
+
+    assert extract_commit_hash(resolved_file, revision="main") == commit_hash
+
+
+def test_extract_commit_hash_falls_back_to_full_sha_revision():
+    commit_hash = "b" * 40
+
+    assert extract_commit_hash("/custom/cache/config.json", revision=commit_hash) == commit_hash
+
+
+def test_extract_commit_hash_rejects_mutable_revision_without_snapshot():
+    assert extract_commit_hash("/custom/cache/config.json", revision="main") is None