Merge branch 'main' into codex/language-recipes

docker/Dockerfile.internal

@@ -68,16 +68,17 @@ ENV HOME=/home/user_lerobot \
 # issues with MuJoCo and OpenGL drivers.
 RUN uv venv --python python${PYTHON_VERSION}
 
-# Install third-party dependencies separately for layer caching
+# Install Python dependencies for caching
 COPY --chown=user_lerobot:user_lerobot setup.py pyproject.toml uv.lock README.md MANIFEST.in ./
-RUN uv sync --locked --extra all --no-install-project --no-cache
+COPY --chown=user_lerobot:user_lerobot src/ src/
+
+RUN uv sync --locked --extra all --no-cache
 
 RUN chmod +x /lerobot/.venv/lib/python${PYTHON_VERSION}/site-packages/triton/backends/nvidia/bin/ptxas
 
-# Copy the application source code and install the local project
+# Copy the rest of the application source code
 # Make sure to have the git-LFS files for testing
 COPY --chown=user_lerobot:user_lerobot . .
-RUN uv sync --locked --extra all --no-cache
 
 # Set the default command
 CMD ["/bin/bash"]

docker/Dockerfile.user

@@ -60,14 +60,15 @@ ENV HOME=/home/user_lerobot \
 # run other Python projects in the same container without dependency conflicts.
 RUN uv venv
 
-# Install third-party dependencies separately for layer caching
+# Install Python dependencies for caching
 COPY --chown=user_lerobot:user_lerobot setup.py pyproject.toml uv.lock README.md MANIFEST.in ./
-RUN uv sync --locked --extra all --no-install-project --no-cache
+COPY --chown=user_lerobot:user_lerobot src/ src/
 
-# Copy the application code and install the local project
+RUN uv sync --locked --extra all --no-cache
+
+# Copy the rest of the application code
 # Make sure to have the git-LFS files for testing
 COPY --chown=user_lerobot:user_lerobot . .
-RUN uv sync --locked --extra all --no-cache
 
 # Set the default command
 CMD ["/bin/bash"]

docs/source/backwardcomp.mdx

@@ -58,7 +58,7 @@ final_action = postprocessor(action)
 
 ## Hardware API redesign
 
-PR [#777](https://github.com/huggingface/lerobot/pull/777) improves the LeRobot calibration but is **not backward-compatible**. Below is an overview of what changed and how you can continue to work with datasets created before this pull request.
+PR [#777](https://github.com/huggingface/lerobot/pull/777) improves the LeRobot calibration but is **not backward-compatible**. Below is a overview of what changed and how you can continue to work with datasets created before this pull request.
 
 ### What changed?
 
@@ -129,8 +129,8 @@ python examples/backward_compatibility/replay.py \
 
 Policies output actions in the same format as the datasets (`torch.Tensors`). Therefore, the same transformations should be applied.
 
-To find these transformations, we recommend first replaying an episode of the dataset your policy was trained on using the section above.
-Then, add these same transformations to your inference script (shown here in the `record.py` script):
+To find these transformations, we recommend to first try and and replay an episode of the dataset your policy was trained on using the section above.
+Then, add these same transformations on your inference script (shown here in the `record.py` script):
 
 ```diff
 action_values = predict_action(

docs/source/cameras.mdx

@@ -136,10 +136,6 @@ config = RealSenseCameraConfig(
     height=480,
     color_mode=ColorMode.RGB,
     use_depth=True,
-    # Optional fixed color controls. Omit them to leave the current sensor settings unchanged.
-    exposure=120,
-    gain=64,
-    white_balance=4600,
     rotation=Cv2Rotation.NO_ROTATION
 )
 
@@ -158,15 +154,6 @@ finally:
 ```
 <!-- prettier-ignore-end -->
 
-Manual color controls disable the corresponding automatic exposure or white-balance mode. Their
-supported ranges vary by camera model; an invalid value raises an error at connection time that
-includes the range reported by the sensor. Requesting an unsupported control also raises an error.
-Omitted controls leave the sensor's existing automatic or manual setting unchanged. These options
-require `use_rgb=True`.
-
-On the RealSense D405, the color stream is provided by the Stereo Module, so changing manual
-exposure or gain also affects the depth stream.
-
 </hfoption>
 </hfoptions>
 

docs/source/feetech.mdx

@@ -40,10 +40,10 @@ This tutorial guides you through updating the firmware of Feetech motors using t
 For each motor you want to update:
 
 1. **Select the motor** from the list by clicking on it
-2. **Click the Upgrade tab**:
-3. **Click the Online button**:
-   - If a potential firmware update is found, it will be displayed in the box
-4. **Click the Upgrade button**:
+2. **Click on Upgrade tab**:
+3. **Click on Online button**:
+   - If an potential firmware update is found, it will be displayed in the box
+4. **Click on Upgrade button**:
    - The update progress will be displayed
 
 ## Step 6: Verify Update

docs/source/language_and_recipes.mdx

@@ -141,6 +141,17 @@ sample["target_message_indices"]
 
 The renderer does not apply a tokenizer chat template. Policy processors decide how to serialize the messages for their backbone, which keeps the same dataset usable across SmolVLA, Pi0.5, and any future VLM that expects OpenAI-style chat messages.
 
+## Blends
+
+Blend recipes select one weighted sub-recipe deterministically from the sample index.
+`recipes/subtask_mem.yaml` trains the compact core blend — high-level subtask prediction, low-level execution, and memory. `recipes/subtask_mem_vqa_speech.yaml` is the fuller variant that also adds VQA and spoken interjection responses.
+
+A message recipe with a supervised assistant turn on the `low_level` stream trains
+the π0.5 paper's joint sequence instead of a blend: the target span gets text CE
+while also conditioning the action losses in the same forward.
+`recipes/subtask_joint.yaml` is the provided example; pair it with
+`--policy.joint_subtask_conditioning=true` at inference.
+
 ## Graceful absence
 
 If both language columns are missing, `None`, or empty, `RenderMessagesStep` is a no-op.

docs/source/processors_robots_teleop.mdx

@@ -22,7 +22,7 @@ With processors, you choose the learning features you want to use for your polic
 ## Three pipelines
 
 We often compose three pipelines. Depending on your setup, some can be empty if action and observation spaces already match.
-Each of these pipelines handles different conversions between different action and observation spaces. Below is a quick explanation of each pipeline.
+Each of these pipelines handle different conversions between different action and observation spaces. Below is a quick explanation of each pipeline.
 
 1. Pipeline 1: Teleop action space → dataset action space (phone pose → EE targets)
 2. Pipeline 2: Dataset action space → robot command space (EE targets → joints)
@@ -74,15 +74,15 @@ In the phone to SO-100 follower examples we use the following adapters:
 - `robot_action_to_transition`: transforms the teleop action dict to a pipeline transition.
 - `transition_to_robot_action`: transforms the pipeline transition to a robot action dict.
 - `observation_to_transition`: transforms the robot observation dict to a pipeline transition.
-- `transition_to_observation`: transforms the pipeline transition to an observation dict.
+- `transition_to_observation`: transforms the pipeline transition to a observation dict.
 
-Check out [src/lerobot/processor/converters.py](https://github.com/huggingface/lerobot/blob/main/src/lerobot/processor/converters.py) for more details.
+Checkout [src/lerobot/processor/converters.py](https://github.com/huggingface/lerobot/blob/main/src/lerobot/processor/converters.py) for more details.
 
 ## Dataset feature contracts
 
 Dataset features are determined by the keys saved in the dataset. Each step can declare what features it modifies in a contract called `transform_features(...)`. Once you build a processor, the processor can then aggregate all of these features with `aggregate_pipeline_dataset_features()` and merge multiple feature dicts with `combine_feature_dicts(...)`.
 
-Below is an example of how we declare features with the `transform_features` method in the phone to SO-100 follower examples:
+Below is and example of how we declare features with the `transform_features` method in the phone to SO-100 follower examples:
 
 ```python
     def transform_features(

docs/source/rtc.mdx

@@ -57,7 +57,7 @@ policy_cfg.rtc_config = RTCConfig(
 policy = PI0Policy.from_pretrained("lerobot/pi0_base", policy_cfg=policy_cfg, device="cuda")
 
 # Now use predict_action_chunk with RTC parameters
-inference_delay = 4  # How many steps of inference latency, this value should be calculated based on the inference latency of the policy
+inference_delay = 4  # How many steps of inference latency, this values should be calculated based on the inference latency of the policy
 
 # Initialize the action queue
 action_queue = ActionQueue(policy_cfg.rtc_config)
@@ -100,7 +100,7 @@ Typical values: 8-12 steps
 RTCConfig(execution_horizon=10)
 ```
 
-**`max_guidance_weight`**: How strongly to enforce consistency with the previous chunk. This is a hyperparameter that can be tuned to balance the smoothness of the transitions and the reactivity of the policy. For 10 steps flow matching (SmolVLA, Pi0, Pi0.5), a value of 10.0 is an optimal value.
+**`max_guidance_weight`**: How strongly to enforce consistency with the previous chunk. This is a hyperparameter that can be tuned to balance the smoothness of the transitions and the reactivity of the policy. For 10 steps flow matching (SmolVLA, Pi0, Pi0.5), a value of 10.0 is a optimal value.
 
 **`prefix_attention_schedule`**: How to weight consistency across the overlap region.
 

docs/source/using_dataset_tools.mdx

@@ -50,11 +50,11 @@ lerobot-edit-dataset \
 Divide a dataset into multiple subsets.
 
 ```bash
-# Split by fractions (e.g. 60% train, 20% val, 20% test)
+# Split by fractions (e.g. 80% train, 20% test, 20% val)
 lerobot-edit-dataset \
     --repo_id lerobot/pusht \
     --operation.type split \
-    --operation.splits '{"train": 0.6, "val": 0.2, "test": 0.2}'
+    --operation.splits '{"train": 0.8, "test": 0.2, "val": 0.2}'
 
 # Split by specific episode indices
 lerobot-edit-dataset \

pyproject.toml

@@ -494,19 +494,6 @@ ignore_errors = true
 module = "lerobot.envs.*"
 ignore_errors = false
 
-[[tool.mypy.overrides]]
-module = "lerobot.annotations.*"
-ignore_errors = false
-disallow_untyped_defs = true
-disallow_incomplete_defs = true
-check_untyped_defs = true
-
-[[tool.mypy.overrides]]
-module = "lerobot.transforms.*"
-ignore_errors = false
-disallow_untyped_defs = true
-disallow_incomplete_defs = true
-check_untyped_defs = true
 
 # [[tool.mypy.overrides]]
 # module = "lerobot.utils.*"

src/lerobot/cameras/opencv/camera_opencv.py

@@ -120,22 +120,14 @@ class OpenCVCamera(Camera):
         self.rotation: int | None = get_cv2_rotation(config.rotation)
         self.backend: int = config.backend
 
-        self.capture_width: int | None = None
-        self.capture_height: int | None = None
-        self._reset_connection_settings()
+        if self.height and self.width:
+            self.capture_width, self.capture_height = self.width, self.height
+            if self.rotation in [cv2.ROTATE_90_CLOCKWISE, cv2.ROTATE_90_COUNTERCLOCKWISE]:
+                self.capture_width, self.capture_height = self.height, self.width
 
     def __str__(self) -> str:
         return f"{self.__class__.__name__}({self.index_or_path})"
 
-    def _reset_connection_settings(self) -> None:
-        """Restore settings that may have been auto-detected during a failed connection."""
-        self.fps = self.config.fps
-        self.width = self.config.width
-        self.height = self.config.height
-        self.capture_width, self.capture_height = self.width, self.height
-        if self.rotation in [cv2.ROTATE_90_CLOCKWISE, cv2.ROTATE_90_COUNTERCLOCKWISE]:
-            self.capture_width, self.capture_height = self.height, self.width
-
     @property
     def is_connected(self) -> bool:
         """Checks if the camera is currently connected and opened."""
@@ -172,25 +164,17 @@ class OpenCVCamera(Camera):
                 f"Failed to open {self}.Run `lerobot-find-cameras opencv` to find available cameras."
             )
 
-        try:
-            self._configure_capture_settings()
-            self._start_read_thread()
+        self._configure_capture_settings()
+        self._start_read_thread()
 
-            if warmup and self.warmup_s > 0:
-                start_time = time.time()
-                while time.time() - start_time < self.warmup_s:
-                    self.async_read(timeout_ms=self.warmup_s * 1000)
-                    time.sleep(0.1)
-                with self.frame_lock:
-                    if self.latest_frame is None:
-                        raise ConnectionError(f"{self} failed to capture frames during warmup.")
-        except BaseException:
-            try:
-                self._cleanup_resources()
-            except Exception:
-                logger.exception(f"Failed to fully clean up {self} after connect() failed.")
-            self._reset_connection_settings()
-            raise
+        if warmup and self.warmup_s > 0:
+            start_time = time.time()
+            while time.time() - start_time < self.warmup_s:
+                self.async_read(timeout_ms=self.warmup_s * 1000)
+                time.sleep(0.1)
+            with self.frame_lock:
+                if self.latest_frame is None:
+                    raise ConnectionError(f"{self} failed to capture frames during warmup.")
 
         logger.info(f"{self} connected.")
 
@@ -328,36 +312,32 @@ class OpenCVCamera(Camera):
 
         for target in targets_to_scan:
             camera = cv2.VideoCapture(target)
-            try:
-                if camera.isOpened():
-                    default_width = int(camera.get(cv2.CAP_PROP_FRAME_WIDTH))
-                    default_height = int(camera.get(cv2.CAP_PROP_FRAME_HEIGHT))
-                    default_fps = camera.get(cv2.CAP_PROP_FPS)
-                    default_format = camera.get(cv2.CAP_PROP_FORMAT)
+            if camera.isOpened():
+                default_width = int(camera.get(cv2.CAP_PROP_FRAME_WIDTH))
+                default_height = int(camera.get(cv2.CAP_PROP_FRAME_HEIGHT))
+                default_fps = camera.get(cv2.CAP_PROP_FPS)
+                default_format = camera.get(cv2.CAP_PROP_FORMAT)
 
-                    # Get FOURCC code and convert to string
-                    default_fourcc_code = camera.get(cv2.CAP_PROP_FOURCC)
-                    default_fourcc_code_int = int(default_fourcc_code)
-                    default_fourcc = "".join(
-                        [chr((default_fourcc_code_int >> 8 * i) & 0xFF) for i in range(4)]
-                    )
+                # Get FOURCC code and convert to string
+                default_fourcc_code = camera.get(cv2.CAP_PROP_FOURCC)
+                default_fourcc_code_int = int(default_fourcc_code)
+                default_fourcc = "".join([chr((default_fourcc_code_int >> 8 * i) & 0xFF) for i in range(4)])
 
-                    camera_info = {
-                        "name": f"OpenCV Camera @ {target}",
-                        "type": "OpenCV",
-                        "id": target,
-                        "backend_api": camera.getBackendName(),
-                        "default_stream_profile": {
-                            "format": default_format,
-                            "fourcc": default_fourcc,
-                            "width": default_width,
-                            "height": default_height,
-                            "fps": default_fps,
-                        },
-                    }
+                camera_info = {
+                    "name": f"OpenCV Camera @ {target}",
+                    "type": "OpenCV",
+                    "id": target,
+                    "backend_api": camera.getBackendName(),
+                    "default_stream_profile": {
+                        "format": default_format,
+                        "fourcc": default_fourcc,
+                        "width": default_width,
+                        "height": default_height,
+                        "fps": default_fps,
+                    },
+                }
 
-                    found_cameras_info.append(camera_info)
-            finally:
+                found_cameras_info.append(camera_info)
                 camera.release()
 
         return found_cameras_info
@@ -516,26 +496,6 @@ class OpenCVCamera(Camera):
             self.latest_timestamp = None
             self.new_frame_event.clear()
 
-    def _cleanup_resources(self) -> None:
-        """Stop background reads and release the capture, including after partial setup."""
-        read_thread = self.thread
-        videocapture = self.videocapture
-
-        try:
-            self._stop_read_thread()
-        finally:
-            self.videocapture = None
-            try:
-                if videocapture is not None:
-                    videocapture.release()
-            finally:
-                # Releasing the device may unblock a hardware read that outlived
-                # the first bounded join in _stop_read_thread().
-                if read_thread is not None and read_thread.is_alive():
-                    read_thread.join(timeout=2.0)
-                    if read_thread.is_alive():  # pragma: no cover
-                        logger.warning(f"{self} read thread remained alive after releasing the capture.")
-
     @check_if_not_connected
     def async_read(self, timeout_ms: float = 200) -> NDArray[Any]:
         """
@@ -626,6 +586,16 @@ class OpenCVCamera(Camera):
         if not self.is_connected and self.thread is None:
             raise DeviceNotConnectedError(f"{self} not connected.")
 
-        self._cleanup_resources()
+        if self.thread is not None:
+            self._stop_read_thread()
+
+        if self.videocapture is not None:
+            self.videocapture.release()
+            self.videocapture = None
+
+        with self.frame_lock:
+            self.latest_frame = None
+            self.latest_timestamp = None
+            self.new_frame_event.clear()
 
         logger.info(f"{self} disconnected.")

src/lerobot/cameras/realsense/camera_realsense.py

@@ -121,9 +121,6 @@ class RealSenseCamera(Camera):
 
         self.config = config
 
-        self.width: int | None = config.width
-        self.height: int | None = config.height
-
         if config.serial_number_or_name.isdigit():
             self.serial_number = config.serial_number_or_name
         else:
@@ -134,9 +131,6 @@ class RealSenseCamera(Camera):
         self.use_rgb = config.use_rgb
         self.use_depth = config.use_depth
         self.warmup_s = config.warmup_s
-        self.exposure: int | None = config.exposure
-        self.gain: int | None = config.gain
-        self.white_balance: int | None = config.white_balance
 
         self.rs_pipeline: rs.pipeline | None = None
         self.rs_profile: rs.pipeline_profile | None = None
@@ -151,23 +145,14 @@ class RealSenseCamera(Camera):
 
         self.rotation: int | None = get_cv2_rotation(config.rotation)
 
-        self.capture_width: int | None = None
-        self.capture_height: int | None = None
-        self._reset_connection_settings()
+        if self.height and self.width:
+            self.capture_width, self.capture_height = self.width, self.height
+            if self.rotation in [cv2.ROTATE_90_CLOCKWISE, cv2.ROTATE_90_COUNTERCLOCKWISE]:
+                self.capture_width, self.capture_height = self.height, self.width
 
     def __str__(self) -> str:
         return f"{self.__class__.__name__}({self.serial_number})"
 
-    def _reset_connection_settings(self) -> None:
-        """Restore settings that may have been auto-detected during a failed connection."""
-        self.fps = self.config.fps
-        self.width = self.config.width
-        self.height = self.config.height
-        self.warmup_s = self.config.warmup_s
-        self.capture_width, self.capture_height = self.width, self.height
-        if self.rotation in [cv2.ROTATE_90_CLOCKWISE, cv2.ROTATE_90_COUNTERCLOCKWISE]:
-            self.capture_width, self.capture_height = self.height, self.width
-
     @property
     def is_connected(self) -> bool:
         """Checks if the camera pipeline is started and streams are active."""
@@ -187,8 +172,7 @@ class RealSenseCamera(Camera):
 
         Raises:
             DeviceAlreadyConnectedError: If the camera is already connected.
-            ValueError: If the configuration is invalid, a requested sensor option is unsupported,
-                or a requested sensor value is invalid.
+            ValueError: If the configuration is invalid (e.g., missing serial/name, name not unique).
             ConnectionError: If the camera is found but fails to start the pipeline or no RealSense devices are detected at all.
             RuntimeError: If the pipeline starts but fails to apply requested settings.
         """
@@ -206,31 +190,22 @@ class RealSenseCamera(Camera):
                 f"Failed to open {self}.Run `lerobot-find-cameras realsense` to find available cameras."
             ) from e
 
-        try:
-            self._configure_capture_settings()
-            self._configure_sensor_options()
-            self._start_read_thread()
+        self._configure_capture_settings()
+        self._start_read_thread()
 
-            # NOTE(Steven/Caroline): Enforcing at least one second of warmup as RS cameras need a bit of time before the first read. If we don't wait, the first read from the warmup will raise.
-            self.warmup_s = max(self.warmup_s, 1)
+        # NOTE(Steven/Caroline): Enforcing at least one second of warmup as RS cameras need a bit of time before the first read. If we don't wait, the first read from the warmup will raise.
+        self.warmup_s = max(self.warmup_s, 1)
 
-            warmup_read = self.async_read if self.use_rgb else self.async_read_depth
-            start_time = time.time()
-            while time.time() - start_time < self.warmup_s:
-                warmup_read(timeout_ms=self.warmup_s * 1000)
-                time.sleep(0.1)
-            with self.frame_lock:
-                if (self.use_rgb and self.latest_color_frame is None) or (
-                    self.use_depth and self.latest_depth_frame is None
-                ):
-                    raise ConnectionError(f"{self} failed to capture frames during warmup.")
-        except BaseException:
-            try:
-                self._cleanup_resources()
-            except Exception:
-                logger.exception(f"Failed to fully clean up {self} after connect() failed.")
-            self._reset_connection_settings()
-            raise
+        warmup_read = self.async_read if self.use_rgb else self.async_read_depth
+        start_time = time.time()
+        while time.time() - start_time < self.warmup_s:
+            warmup_read(timeout_ms=self.warmup_s * 1000)
+            time.sleep(0.1)
+        with self.frame_lock:
+            if (self.use_rgb and self.latest_color_frame is None) or (
+                self.use_depth and self.latest_depth_frame is None
+            ):
+                raise ConnectionError(f"{self} failed to capture frames during warmup.")
 
         logger.info(f"{self} connected.")
 
@@ -364,111 +339,6 @@ class RealSenseCamera(Camera):
         self.new_frame_event.clear()
         return self._async_read(timeout_ms=10000, read_depth=read_depth)
 
-    def _get_color_sensor(self) -> "rs.sensor":
-        """Returns the sensor that controls the color stream.
-
-        Most RealSense cameras expose "RGB Camera" for color. The D405 has no
-        separate RGB module — its color stream comes from "Stereo Module".
-        We try RGB Camera first, then fall back to Stereo Module.
-        """
-        if self.rs_profile is None:
-            raise RuntimeError(f"{self}: rs_profile must be initialized before use.")
-
-        device = self.rs_profile.get_device()
-        sensors = {s.get_info(rs.camera_info.name): s for s in device.query_sensors()}
-
-        for name in ("RGB Camera", "Stereo Module"):
-            if name in sensors:
-                return sensors[name]
-
-        available = list(sensors.keys())
-        raise RuntimeError(f"{self}: no color sensor found. Available sensors: {available}")
-
-    def _set_sensor_option(self, sensor: "rs.sensor", option: "rs.option", value: float, label: str) -> None:
-        """Sets a sensor option, re-raising range errors with actionable diagnostics."""
-        try:
-            sensor.set_option(option, value)
-        except Exception as e:
-            range_info = ""
-            try:
-                option_range = sensor.get_option_range(option)
-                range_info = (
-                    f" (supported range: min={option_range.min}, max={option_range.max}, "
-                    f"step={option_range.step}, default={option_range.default})"
-                )
-            except Exception:
-                range_info = " (option range unavailable)"
-            raise ValueError(
-                f"{self}: failed to set {label} to {value}{range_info}. Original error: {e}"
-            ) from e
-
-    def _configure_sensor_options(self) -> None:
-        """Applies manual sensor options (exposure, gain, white balance) to the color sensor.
-
-        When exposure or gain is set, auto-exposure is disabled first. When white_balance
-        is set, auto white balance is disabled first. An omitted option is left unchanged,
-        and configuration is skipped entirely if all options are omitted.
-
-        Raises:
-            ValueError: If the sensor does not support a requested option or a requested
-                value is invalid. Invalid-value errors include the option name, requested
-                value, and supported range when available.
-        """
-        if self.exposure is None and self.gain is None and self.white_balance is None:
-            return
-
-        color_sensor = self._get_color_sensor()
-
-        requested_options = (
-            (rs.option.exposure, self.exposure, "exposure"),
-            (rs.option.gain, self.gain, "gain"),
-            (rs.option.white_balance, self.white_balance, "white balance"),
-        )
-        unsupported_options = [
-            label
-            for option, value, label in requested_options
-            if value is not None and not color_sensor.supports(option)
-        ]
-        if unsupported_options:
-            raise ValueError(
-                f"{self}: color sensor does not support requested manual options: {unsupported_options}."
-            )
-
-        manual_exposure_requested = self.exposure is not None or self.gain is not None
-        if manual_exposure_requested:
-            if color_sensor.supports(rs.option.enable_auto_exposure):
-                self._set_sensor_option(color_sensor, rs.option.enable_auto_exposure, 0, "auto-exposure")
-                logger.info(f"{self} auto-exposure disabled.")
-            else:
-                logger.warning(
-                    f"{self} sensor does not support disabling auto-exposure; "
-                    "applying manual exposure/gain directly."
-                )
-
-        if self.exposure is not None:
-            self._set_sensor_option(color_sensor, rs.option.exposure, self.exposure, "exposure")
-            logger.info(f"{self} exposure set to {self.exposure}.")
-
-        if self.gain is not None:
-            self._set_sensor_option(color_sensor, rs.option.gain, self.gain, "gain")
-            logger.info(f"{self} gain set to {self.gain}.")
-
-        if self.white_balance is not None:
-            if color_sensor.supports(rs.option.enable_auto_white_balance):
-                self._set_sensor_option(
-                    color_sensor, rs.option.enable_auto_white_balance, 0, "auto white balance"
-                )
-                logger.info(f"{self} auto white balance disabled.")
-            else:
-                logger.warning(
-                    f"{self} sensor does not support disabling auto white balance; "
-                    "applying manual white balance directly."
-                )
-            self._set_sensor_option(
-                color_sensor, rs.option.white_balance, self.white_balance, "white balance"
-            )
-            logger.info(f"{self} white balance set to {self.white_balance}.")
-
     @check_if_not_connected
     def read_depth(self, timeout_ms: int = 200) -> NDArray[Any]:
         """
@@ -671,27 +541,6 @@ class RealSenseCamera(Camera):
             self.latest_timestamp = None
             self.new_frame_event.clear()
 
-    def _cleanup_resources(self) -> None:
-        """Stop background reads and stop the pipeline, including after partial setup."""
-        read_thread = self.thread
-        rs_pipeline = self.rs_pipeline
-
-        try:
-            self._stop_read_thread()
-        finally:
-            self.rs_pipeline = None
-            self.rs_profile = None
-            try:
-                if rs_pipeline is not None:
-                    rs_pipeline.stop()
-            finally:
-                # Stopping the pipeline may unblock a hardware read that outlived
-                # the first bounded join in _stop_read_thread().
-                if read_thread is not None and read_thread.is_alive():
-                    read_thread.join(timeout=2.0)
-                    if read_thread.is_alive():  # pragma: no cover
-                        logger.warning(f"{self} read thread remained alive after stopping the pipeline.")
-
     def _async_read(self, timeout_ms: float, read_depth: bool = False) -> NDArray[Any]:
         """Shared helper for :meth:`async_read`/:meth:`async_read_depth`: return the latest buffered frame."""
         if self.thread is None or not self.thread.is_alive():
@@ -835,5 +684,18 @@ class RealSenseCamera(Camera):
                 f"Attempted to disconnect {self}, but it appears already disconnected."
             )
 
-        self._cleanup_resources()
+        if self.thread is not None:
+            self._stop_read_thread()
+
+        if self.rs_pipeline is not None:
+            self.rs_pipeline.stop()
+            self.rs_pipeline = None
+            self.rs_profile = None
+
+        with self.frame_lock:
+            self.latest_color_frame = None
+            self.latest_depth_frame = None
+            self.latest_timestamp = None
+            self.new_frame_event.clear()
+
         logger.info(f"{self} disconnected.")

src/lerobot/cameras/realsense/configuration_realsense.py

@@ -46,17 +46,6 @@ class RealSenseCameraConfig(CameraConfig):
         use_depth: Whether to enable depth stream. Defaults to False.
         rotation: Image rotation setting (0°, 90°, 180°, or 270°). Defaults to no rotation.
         warmup_s: Time reading frames before returning from connect (in seconds)
-        exposure: Manual exposure value for the color sensor. When set, auto-exposure is
-            disabled and this fixed value is used. Valid ranges are camera-model specific
-            and reported if the value is rejected. Defaults to None (leave unchanged).
-        gain: Manual gain value for the color sensor. When set, auto-exposure is disabled
-            and this fixed gain is used, which also freezes exposure at its current value
-            when no exposure is configured. Valid ranges are camera-model specific and
-            reported if the value is rejected. Defaults to None (leave unchanged).
-        white_balance: Manual white balance value for the color sensor. When set, auto
-            white balance is disabled and this fixed value is used. Valid ranges are
-            camera-model specific and reported if the value is rejected. Defaults to None
-            (leave unchanged).
 
     Note:
         - Either name or serial_number must be specified.
@@ -72,9 +61,6 @@ class RealSenseCameraConfig(CameraConfig):
     use_depth: bool = False
     rotation: Cv2Rotation = Cv2Rotation.NO_ROTATION
     warmup_s: int = 1
-    exposure: int | None = None
-    gain: int | None = None
-    white_balance: int | None = None
 
     def __post_init__(self) -> None:
         self.color_mode = ColorMode(self.color_mode)
@@ -83,18 +69,6 @@ class RealSenseCameraConfig(CameraConfig):
         if not self.use_rgb and not self.use_depth:
             raise ValueError("At least one of `use_rgb` or `use_depth` must be enabled.")
 
-        manual_color_options = {
-            "exposure": self.exposure,
-            "gain": self.gain,
-            "white_balance": self.white_balance,
-        }
-        configured_color_options = [name for name, value in manual_color_options.items() if value is not None]
-        if configured_color_options and not self.use_rgb:
-            raise ValueError(
-                "Manual color sensor options require `use_rgb=True`. "
-                f"Configured options: {configured_color_options}."
-            )
-
         values = (self.fps, self.width, self.height)
         if any(v is not None for v in values) and any(v is None for v in values):
             raise ValueError(

src/lerobot/configs/dataset.py

@@ -71,19 +71,13 @@ class DatasetRecordConfig:
     # Number of threads per encoder instance. None = auto (codec default).
     # Lower values reduce CPU usage, maps to 'lp' (via svtav1-params) for libsvtav1 and 'threads' for h264/hevc..
     encoder_threads: int | None = None
-    # Skip appending the date-time tag to repo_id, keeping the user-provided name as-is
-    # (e.g. self-managed versioned names intended for a later `lerobot-edit-dataset merge`).
-    no_stamp: bool = False
 
     def stamp_repo_id(self) -> None:
         """Append a date-time tag to ``repo_id`` so each recording session gets a unique name.
 
         Must be called explicitly at dataset *creation* time — not on resume,
         where the existing ``repo_id`` (already stamped) must be preserved.
-        No-op when ``no_stamp`` is set, preserving a user-managed ``repo_id``.
         """
-        if self.no_stamp:
-            return
         if self.repo_id:
             timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
             self.repo_id = f"{self.repo_id}_{timestamp}"

src/lerobot/configs/default.py

@@ -33,6 +33,8 @@ class DatasetConfig:
     # looked up under $HF_LEROBOT_HOME/repo_id and Hub downloads use a revision-safe cache under $HF_LEROBOT_HOME/hub.
     root: str | None = None
     episodes: list[int] | None = None
+    # Episode indices to drop (e.g. corrupt or heterogeneous ones). Applied on top of `episodes`.
+    exclude_episodes: list[int] | None = None
     image_transforms: ImageTransformsConfig = field(default_factory=ImageTransformsConfig)
     revision: str | None = None
     use_imagenet_stats: bool = True
@@ -62,6 +64,10 @@ class DatasetConfig:
             if len(self.episodes) != len(set(self.episodes)):
                 duplicates = sorted({ep for ep in self.episodes if self.episodes.count(ep) > 1})
                 raise ValueError(f"Episode indices contain duplicates: {duplicates}")
+        if self.exclude_episodes is not None and any(ep < 0 for ep in self.exclude_episodes):
+            raise ValueError(
+                f"exclude_episodes must be non-negative, got: {[ep for ep in self.exclude_episodes if ep < 0]}"
+            )
 
 
 @dataclass

src/lerobot/configs/recipe.py

@@ -78,7 +78,7 @@ class MessageTurn:
             raise ValueError(f"Unsupported message stream: {self.stream!r}")
         if self.content is None and self.tool_calls_from is None:
             raise ValueError("MessageTurn.content is required unless tool_calls_from is set.")
-        if self.content is not None and not isinstance(self.content, (str, list)):
+        if self.content is not None and not isinstance(self.content, str | list):
             raise TypeError("MessageTurn.content must be a string, a list of HF-style blocks, or None.")
         if isinstance(self.content, list):
             for block in self.content:
@@ -147,7 +147,7 @@ class TrainingRecipe:
         return cls.from_dict(data)
 
     def _validate_message_recipe(self) -> None:
-        """Ensure every templated binding is known and at least one turn is a target."""
+        """Validate bindings and require text or low-level action supervision."""
         assert self.messages is not None
         known_bindings = set(DEFAULT_BINDINGS) | set(self.bindings or {}) | {"task"}
 
@@ -156,8 +156,14 @@ class TrainingRecipe:
             if missing:
                 raise ValueError(f"MessageTurn references unknown binding(s): {sorted(missing)}")
 
-        if not any(turn.target for turn in self.messages):
-            raise ValueError("Message recipes must contain at least one target turn.")
+        has_target = any(turn.target for turn in self.messages)
+        has_low_level = any(turn.stream == "low_level" for turn in self.messages)
+        if not (has_target or has_low_level):
+            raise ValueError(
+                "Message recipes must contain at least one supervised turn — "
+                "either ``target: true`` (text CE) or ``stream: low_level`` "
+                "(flow/action loss)."
+            )
 
     def _validate_blend_recipe(self) -> None:
         """Ensure each blend component is a non-empty, weighted message recipe."""

src/lerobot/configs/recipes/subtask.yaml

@@ -0,0 +1,16 @@
+# Predicts subtasks from tasks and trains subtask-conditioned action flow without memory or plans.
+# Requires `subtask` annotations; samples with missing `if_present` bindings do not render.
+
+blend:
+
+  high_level_subtask:
+    weight: 0.30
+    messages:
+      - {role: user, content: "${task}", stream: high_level}
+      - {role: assistant, content: "${subtask}", stream: high_level, target: true, if_present: subtask}
+
+  low_level_execution:
+    weight: 0.70
+    messages:
+      # The low-level stream trains action flow on the generated or annotated subtask.
+      - {role: user, content: "${subtask}", stream: low_level, if_present: subtask}

src/lerobot/configs/recipes/subtask_joint.yaml

@@ -0,0 +1,13 @@
+# Paper-style joint sequence (pi0.5 §IV-B): one sample supervises the subtask
+# text with CE and, because the assistant turn is part of the prefix, conditions
+# the FAST and flow action losses on the same annotated subtask in one forward.
+# The supervised span is attended causally; the action losses see task + subtask.
+#
+# Pair with `--policy.joint_subtask_conditioning=true` at inference so the flow
+# prefix reproduces this layout (task turn with state + causal generated subtask).
+# Samples without a `subtask` annotation fall back to a plain task-prompt
+# low-level sample via `if_present`.
+
+messages:
+  - {role: user, content: "${task}", stream: low_level}
+  - {role: assistant, content: "${subtask}", stream: low_level, target: true, if_present: subtask}

src/lerobot/configs/recipes/subtask_mem.yaml

@@ -0,0 +1,30 @@
+# Trains subtask prediction, subtask-conditioned action flow, and memory updates without plans.
+# Requires `subtask` and `memory`; missing `if_present` bindings skip the affected sub-recipe.
+
+blend:
+
+  high_level_subtask:
+    weight: 0.25
+    messages:
+      - {role: user, content: "${task}", stream: high_level}
+      - {role: assistant, content: "${subtask}", stream: high_level, target: true, if_present: subtask}
+
+  low_level_execution:
+    weight: 0.60
+    messages:
+      # The low-level stream trains action flow on the generated or annotated subtask.
+      - {role: user, content: "${subtask}", stream: low_level, if_present: subtask}
+
+  memory_update:
+    # `active_at` densifies sparse boundaries while preserving the prior-memory/subtask mapping.
+    # Inference controls update timing through `subtask_change` events.
+    weight: 0.15
+    bindings:
+      prior_memory: "nth_prev(style=memory, offset=1)"
+      current_memory: "active_at(t, style=memory)"
+      completed_subtask: "nth_prev(style=subtask, offset=1)"
+    messages:
+      - {role: user, content: "${task}", stream: high_level}
+      - {role: assistant, content: "Previous memory: ${prior_memory}", stream: high_level, if_present: prior_memory}
+      - {role: user, content: "Completed subtask: ${completed_subtask}", stream: high_level, if_present: completed_subtask}
+      - {role: assistant, content: "${current_memory}", stream: high_level, target: true, if_present: current_memory}

src/lerobot/configs/recipes/subtask_mem_vqa_speech.yaml

@@ -0,0 +1,70 @@
+# Adds memory, spoken interjection responses, and camera-grounded VQA to subtask/action training.
+# Missing optional annotations skip only their sub-recipe; `say` tool calls tokenize as `<say>...</say>`.
+
+blend:
+
+  high_level_subtask:
+    weight: 0.25
+    messages:
+      - {role: user, content: "${task}", stream: high_level}
+      - {role: assistant, content: "${subtask}", stream: high_level, target: true, if_present: subtask}
+
+  low_level_execution:
+    weight: 0.40
+    messages:
+      # The low-level stream trains action flow on the generated or annotated subtask.
+      - {role: user, content: "${subtask}", stream: low_level, if_present: subtask}
+
+  memory_update:
+    # `active_at` densifies sparse boundaries while preserving the prior-memory/subtask mapping.
+    # Inference controls update timing through `subtask_change` events.
+    weight: 0.10
+    bindings:
+      prior_memory: "nth_prev(style=memory, offset=1)"
+      current_memory: "active_at(t, style=memory)"
+      completed_subtask: "nth_prev(style=subtask, offset=1)"
+    messages:
+      - {role: user, content: "${task}", stream: high_level}
+      - {role: assistant, content: "Previous memory: ${prior_memory}", stream: high_level, if_present: prior_memory}
+      - {role: user, content: "Completed subtask: ${completed_subtask}", stream: high_level, if_present: completed_subtask}
+      - {role: assistant, content: "${current_memory}", stream: high_level, target: true, if_present: current_memory}
+
+  user_interjection_response:
+    weight: 0.10
+    bindings:
+      interjection: "emitted_at(t, style=interjection)"
+      speech: "emitted_at(t, role=assistant, tool_name=say)"
+    messages:
+      - {role: user, content: "${task}", stream: high_level}
+      - {role: user, content: "${interjection}", stream: high_level, if_present: interjection}
+      # The assistant target is a `say` tool call flattened to a `<say>...</say>` marker.
+      - {role: assistant, stream: high_level, target: true, if_present: speech, tool_calls_from: speech}
+
+  # Each camera uses a separate VQA sub-recipe for view-specific binding.
+  ask_vqa_top:
+    weight: 0.075
+    bindings:
+      vqa_query: "emitted_at(t, style=vqa, role=user, camera=observation.images.front)"
+      vqa: "emitted_at(t, style=vqa, role=assistant, camera=observation.images.front)"
+    messages:
+      - role: user
+        stream: high_level
+        if_present: vqa_query
+        content:
+          - {type: image, feature: observation.images.front}
+          - {type: text, text: "${vqa_query}"}
+      - {role: assistant, content: "${vqa}", stream: high_level, target: true, if_present: vqa}
+
+  ask_vqa_wrist:
+    weight: 0.075
+    bindings:
+      vqa_query: "emitted_at(t, style=vqa, role=user, camera=observation.images.wrist)"
+      vqa: "emitted_at(t, style=vqa, role=assistant, camera=observation.images.wrist)"
+    messages:
+      - role: user
+        stream: high_level
+        if_present: vqa_query
+        content:
+          - {type: image, feature: observation.images.wrist}
+          - {type: text, text: "${vqa_query}"}
+      - {role: assistant, content: "${vqa}", stream: high_level, target: true, if_present: vqa}

src/lerobot/datasets/aggregate.py

@@ -19,7 +19,6 @@ import copy
 import logging
 import shutil
 from pathlib import Path
-from typing import Any, NotRequired, TypedDict
 
 import datasets
 import pandas as pd
@@ -50,32 +49,8 @@ from .utils import (
 )
 from .video_utils import concatenate_video_files, get_video_duration_in_s
 
-logger = logging.getLogger(__name__)
 
-type FeatureDict = dict[str, dict[str, Any]]
-type ChunkFile = tuple[int, int]
-
-
-class IndexState(TypedDict):
-    chunk: int
-    file: int
-    src_to_dst: NotRequired[dict[ChunkFile, ChunkFile]]
-
-
-class VideoIndex(TypedDict):
-    chunk: int
-    file: int
-    latest_duration: float
-    episode_duration: float
-    src_to_offset: NotRequired[dict[ChunkFile, float]]
-    src_to_dst: NotRequired[dict[ChunkFile, ChunkFile]]
-    dst_file_durations: NotRequired[dict[ChunkFile, float]]
-
-
-type VideoIndexState = dict[str, VideoIndex]
-
-
-def merge_video_feature_info_for_aggregate(all_metadata: list[LeRobotDatasetMetadata]) -> FeatureDict:
+def merge_video_feature_info_for_aggregate(all_metadata: list[LeRobotDatasetMetadata]) -> dict[str, dict]:
     """Create a merged video feature info dictionary for aggregation. The video encoder info is merged field-by-field: each key is kept only when every source agrees; otherwise that key is set to ``null`` (or ``{}`` for ``video.extra_options``) and a warning is logged.
 
     Args:
@@ -84,14 +59,14 @@ def merge_video_feature_info_for_aggregate(all_metadata: list[LeRobotDatasetMeta
     Returns:
         dict: A dictionary of merged video feature info.
     """
-    merged_info: FeatureDict = copy.deepcopy(all_metadata[0].features)
+    merged_info = copy.deepcopy(all_metadata[0].features)
     video_keys = [k for k in merged_info if merged_info[k].get("dtype") == "video"]
 
     for vk in video_keys:
         video_infos = [m.features.get(vk, {}).get("info") or {} for m in all_metadata]
         base_video_info = video_infos[0]
 
-        merged_encoder_info: dict[str, Any] = {}
+        merged_encoder_info: dict = {}
         fallback_keys: list[str] = []
         for info_key in VIDEO_ENCODER_INFO_KEYS:
             values = [info.get(info_key, None) for info in video_infos]
@@ -105,7 +80,7 @@ def merge_video_feature_info_for_aggregate(all_metadata: list[LeRobotDatasetMeta
                 merged_encoder_info[info_key] = {} if info_key == "video.extra_options" else None
 
         if fallback_keys:
-            logger.warning(
+            logging.warning(
                 f"Merging heterogeneous or incomplete video encoder metadata for feature {vk}. "
                 f"Setting these keys to null: {fallback_keys}.",
             )
@@ -117,7 +92,7 @@ def merge_video_feature_info_for_aggregate(all_metadata: list[LeRobotDatasetMeta
     return merged_info
 
 
-def validate_all_metadata(all_metadata: list[LeRobotDatasetMetadata]) -> tuple[int, str | None, FeatureDict]:
+def validate_all_metadata(all_metadata: list[LeRobotDatasetMetadata]):
     """Validates that all dataset metadata have consistent properties.
 
     Ensures all datasets have the same fps, robot_type, and features to guarantee
@@ -154,9 +129,7 @@ def validate_all_metadata(all_metadata: list[LeRobotDatasetMetadata]) -> tuple[i
     return fps, robot_type, features
 
 
-def update_data_df(
-    df: pd.DataFrame, src_meta: LeRobotDatasetMetadata, dst_meta: LeRobotDatasetMetadata
-) -> pd.DataFrame:
+def update_data_df(df, src_meta, dst_meta):
     """Updates a data DataFrame with new indices and task mappings for aggregation.
 
     Adjusts episode indices, frame indices, and task indices to account for
@@ -181,12 +154,12 @@ def update_data_df(
 
 
 def update_meta_data(
-    df: pd.DataFrame,
-    dst_meta: LeRobotDatasetMetadata,
-    meta_idx: IndexState,
-    data_idx: IndexState,
-    videos_idx: VideoIndexState,
-) -> pd.DataFrame:
+    df,
+    dst_meta,
+    meta_idx,
+    data_idx,
+    videos_idx,
+):
     """Updates metadata DataFrame with new chunk, file, and timestamp indices.
 
     Adjusts all indices and timestamps to account for previously aggregated
@@ -316,7 +289,7 @@ def aggregate_datasets(
     chunk_size: int | None = None,
     concatenate_videos: bool = True,
     concatenate_data: bool = True,
-) -> None:
+):
     """Aggregates multiple LeRobot datasets into a single unified dataset.
 
     This is the main function that orchestrates the aggregation process by:
@@ -336,7 +309,7 @@ def aggregate_datasets(
         concatenate_videos: When False, keep one mp4 per source file instead of packing into shards.
         concatenate_data: When False, keep one parquet per source file instead of packing into shards.
     """
-    logger.info("Start aggregate_datasets")
+    logging.info("Start aggregate_datasets")
 
     if data_files_size_in_mb is None:
         data_files_size_in_mb = DEFAULT_DATA_FILE_SIZE_IN_MB
@@ -368,15 +341,15 @@ def aggregate_datasets(
         video_files_size_in_mb=video_files_size_in_mb,
     )
 
-    logger.info("Find all tasks")
+    logging.info("Find all tasks")
     unique_tasks = pd.concat([m.tasks for m in all_metadata]).index.unique()
     dst_meta.tasks = pd.DataFrame(
         {"task_index": range(len(unique_tasks))}, index=pd.Index(unique_tasks, name="task")
     )
 
-    meta_idx: IndexState = {"chunk": 0, "file": 0}
-    data_idx: IndexState = {"chunk": 0, "file": 0}
-    videos_idx: VideoIndexState = {
+    meta_idx = {"chunk": 0, "file": 0}
+    data_idx = {"chunk": 0, "file": 0}
+    videos_idx = {
         key: {"chunk": 0, "file": 0, "latest_duration": 0, "episode_duration": 0} for key in video_keys
     }
 
@@ -400,17 +373,12 @@ def aggregate_datasets(
         dst_meta.info.total_frames += src_meta.total_frames
 
     finalize_aggregation(dst_meta, all_metadata)
-    logger.info("Aggregation complete.")
+    logging.info("Aggregation complete.")
 
 
 def aggregate_videos(
-    src_meta: LeRobotDatasetMetadata,
-    dst_meta: LeRobotDatasetMetadata,
-    videos_idx: VideoIndexState,
-    video_files_size_in_mb: float,
-    chunk_size: int,
-    concatenate_videos: bool = True,
-) -> VideoIndexState:
+    src_meta, dst_meta, videos_idx, video_files_size_in_mb, chunk_size, concatenate_videos=True
+):
     """Aggregates video chunks from a source dataset into the destination dataset.
 
     Handles video file concatenation and rotation based on file size limits.
@@ -438,16 +406,15 @@ def aggregate_videos(
             videos_idx[key]["dst_file_durations"] = {}
 
     for key, video_idx in videos_idx.items():
-        unique_chunk_file_pairs: list[ChunkFile] = sorted(
-            {
-                (chunk, file)
-                for chunk, file in zip(
-                    src_meta.episodes[f"videos/{key}/chunk_index"],
-                    src_meta.episodes[f"videos/{key}/file_index"],
-                    strict=False,
-                )
-            }
-        )
+        unique_chunk_file_pairs = {
+            (chunk, file)
+            for chunk, file in zip(
+                src_meta.episodes[f"videos/{key}/chunk_index"],
+                src_meta.episodes[f"videos/{key}/file_index"],
+                strict=False,
+            )
+        }
+        unique_chunk_file_pairs = sorted(unique_chunk_file_pairs)
 
         chunk_idx = video_idx["chunk"]
         file_idx = video_idx["file"]
@@ -522,14 +489,7 @@ def aggregate_videos(
     return videos_idx
 
 
-def aggregate_data(
-    src_meta: LeRobotDatasetMetadata,
-    dst_meta: LeRobotDatasetMetadata,
-    data_idx: IndexState,
-    data_files_size_in_mb: float,
-    chunk_size: int,
-    concatenate_data: bool = True,
-) -> IndexState:
+def aggregate_data(src_meta, dst_meta, data_idx, data_files_size_in_mb, chunk_size, concatenate_data=True):
     """Aggregates data chunks from a source dataset into the destination dataset.
 
     Reads source data files, updates indices to match the aggregated dataset,
@@ -550,16 +510,14 @@ def aggregate_data(
     Returns:
         dict: Updated data_idx with current chunk and file indices.
     """
-    unique_chunk_file_ids: list[ChunkFile] = sorted(
-        {
-            (c, f)
-            for c, f in zip(
-                src_meta.episodes["data/chunk_index"],
-                src_meta.episodes["data/file_index"],
-                strict=False,
-            )
-        }
-    )
+    unique_chunk_file_ids = {
+        (c, f)
+        for c, f in zip(
+            src_meta.episodes["data/chunk_index"], src_meta.episodes["data/file_index"], strict=False
+        )
+    }
+
+    unique_chunk_file_ids = sorted(unique_chunk_file_ids)
     contains_images = len(dst_meta.image_keys) > 0
 
     # retrieve features schema for proper image typing in parquet
@@ -567,7 +525,7 @@ def aggregate_data(
 
     # Track source to destination file mapping for metadata update
     # This is critical for handling datasets that are already results of a merge
-    src_to_dst: dict[ChunkFile, ChunkFile] = {}
+    src_to_dst: dict[tuple[int, int], tuple[int, int]] = {}
 
     for src_chunk_idx, src_file_idx in unique_chunk_file_ids:
         src_path = src_meta.root / DEFAULT_DATA_PATH.format(
@@ -606,13 +564,7 @@ def aggregate_data(
     return data_idx
 
 
-def aggregate_metadata(
-    src_meta: LeRobotDatasetMetadata,
-    dst_meta: LeRobotDatasetMetadata,
-    meta_idx: IndexState,
-    data_idx: IndexState,
-    videos_idx: VideoIndexState,
-) -> IndexState:
+def aggregate_metadata(src_meta, dst_meta, meta_idx, data_idx, videos_idx):
     """Aggregates metadata from a source dataset into the destination dataset.
 
     Reads source metadata files, updates all indices and timestamps,
@@ -628,16 +580,16 @@ def aggregate_metadata(
     Returns:
         dict: Updated meta_idx with current chunk and file indices.
     """
-    chunk_file_ids: list[ChunkFile] = sorted(
-        {
-            (c, f)
-            for c, f in zip(
-                src_meta.episodes["meta/episodes/chunk_index"],
-                src_meta.episodes["meta/episodes/file_index"],
-                strict=False,
-            )
-        }
-    )
+    chunk_file_ids = {
+        (c, f)
+        for c, f in zip(
+            src_meta.episodes["meta/episodes/chunk_index"],
+            src_meta.episodes["meta/episodes/file_index"],
+            strict=False,
+        )
+    }
+
+    chunk_file_ids = sorted(chunk_file_ids)
     for chunk_idx, file_idx in chunk_file_ids:
         src_path = src_meta.root / DEFAULT_EPISODES_PATH.format(chunk_index=chunk_idx, file_index=file_idx)
         df = pd.read_parquet(src_path)
@@ -670,16 +622,16 @@ def aggregate_metadata(
 def append_or_create_parquet_file(
     df: pd.DataFrame,
     src_path: Path,
-    idx: IndexState,
+    idx: dict[str, int],
     max_mb: float,
     chunk_size: int,
     default_path: str,
     contains_images: bool = False,
-    aggr_root: Path | None = None,
+    aggr_root: Path = None,
     hf_features: datasets.Features | None = None,
     concatenate: bool = True,
     one_row_group_per_episode: bool = False,
-) -> tuple[IndexState, ChunkFile]:
+) -> tuple[dict[str, int], tuple[int, int]]:
     """Appends data to an existing parquet file or creates a new one based on size constraints.
 
     Manages file rotation when size limits are exceeded to prevent individual files
@@ -702,13 +654,7 @@ def append_or_create_parquet_file(
     Returns:
         tuple: (updated_idx, (dst_chunk, dst_file)) where updated_idx is the index dict
                and (dst_chunk, dst_file) is the actual destination file the data was written to.
-
-    Raises:
-        ValueError: If aggr_root is not provided.
     """
-    if aggr_root is None:
-        raise ValueError("aggr_root must be provided.")
-
     dst_chunk, dst_file = idx["chunk"], idx["file"]
     dst_path = aggr_root / default_path.format(chunk_index=dst_chunk, file_index=dst_file)
 
@@ -752,9 +698,7 @@ def append_or_create_parquet_file(
     return idx, (dst_chunk, dst_file)
 
 
-def finalize_aggregation(
-    aggr_meta: LeRobotDatasetMetadata, all_metadata: list[LeRobotDatasetMetadata]
-) -> None:
+def finalize_aggregation(aggr_meta, all_metadata):
     """Finalizes the dataset aggregation by writing summary files and statistics.
 
     Writes the tasks file, info file with total counts and splits, and
@@ -764,16 +708,16 @@ def finalize_aggregation(
         aggr_meta: Aggregated dataset metadata.
         all_metadata: List of all source dataset metadata objects.
     """
-    logger.info("write tasks")
+    logging.info("write tasks")
     write_tasks(aggr_meta.tasks, aggr_meta.root)
 
-    logger.info("write info")
+    logging.info("write info")
     aggr_meta.info.total_tasks = len(aggr_meta.tasks)
     aggr_meta.info.total_episodes = sum(m.total_episodes for m in all_metadata)
     aggr_meta.info.total_frames = sum(m.total_frames for m in all_metadata)
     aggr_meta.info.splits = {"train": f"0:{sum(m.total_episodes for m in all_metadata)}"}
     write_info(aggr_meta.info, aggr_meta.root)
 
-    logger.info("write stats")
+    logging.info("write stats")
     aggr_meta.stats = aggregate_stats([m.stats for m in all_metadata])
     write_stats(aggr_meta.stats, aggr_meta.root)

src/lerobot/datasets/dataset_metadata.py

@@ -188,8 +188,8 @@ class LeRobotDatasetMetadata:
     def _load_metadata(self):
         self.info = load_info(self.root)
         check_version_compatibility(self.repo_id, self._version, CODEBASE_VERSION)
-        self.tasks = load_tasks(self.root) if self.total_tasks > 0 else None
-        self.episodes = load_episodes(self.root) if self.total_episodes > 0 else None
+        self.tasks = load_tasks(self.root)
+        self.episodes = load_episodes(self.root)
         self.stats = load_stats(self.root)
 
     def ensure_readable(self) -> None:

src/lerobot/datasets/dataset_reader.py

@@ -163,10 +163,40 @@ class DatasetReader:
     def _load_hf_dataset(self) -> datasets.Dataset:
         """hf_dataset contains all the observations, states, actions, rewards, etc."""
         features = get_hf_features_from_features(self._meta.features)
+        # Annotated datasets may have language columns absent from metadata.
+        # Extend the schema before the strict Parquet cast.
+        features = self._extend_features_with_language_columns(features)
         hf_dataset = load_nested_dataset(self.root / "data", features=features, episodes=self.episodes)
         hf_dataset.set_transform(hf_transform_to_torch)
         return hf_dataset
 
+    def _extend_features_with_language_columns(self, features: datasets.Features) -> datasets.Features:
+        """Register language columns found in Parquet but missing from metadata."""
+        # Leave empty datasets to fail through the normal loading path.
+        try:
+            sample = next((self.root / "data").glob("*/*.parquet"))
+        except StopIteration:
+            return features
+
+        from pyarrow import parquet as _pq  # noqa: PLC0415
+
+        schema_names = set(_pq.read_schema(sample).names)
+        from .language import (  # noqa: PLC0415
+            LANGUAGE_EVENTS,
+            LANGUAGE_PERSISTENT,
+            language_events_column_feature,
+            language_persistent_column_feature,
+        )
+
+        extra: dict[str, object] = {}
+        if LANGUAGE_PERSISTENT in schema_names and LANGUAGE_PERSISTENT not in features:
+            extra[LANGUAGE_PERSISTENT] = language_persistent_column_feature()
+        if LANGUAGE_EVENTS in schema_names and LANGUAGE_EVENTS not in features:
+            extra[LANGUAGE_EVENTS] = language_events_column_feature()
+        if not extra:
+            return features
+        return datasets.Features({**features, **extra})
+
     def _check_cached_episodes_sufficient(self) -> bool:
         """Check if the cached dataset contains all requested episodes and their video files."""
         if self.hf_dataset is None or len(self.hf_dataset) == 0:

src/lerobot/datasets/factory.py

@@ -66,6 +66,17 @@ def resolve_delta_timestamps(
     return delta_timestamps
 
 
+def _resolve_episodes(
+    episodes: list[int] | None, exclude_episodes: list[int] | None, total_episodes: int
+) -> list[int] | None:
+    """Apply an episode exclusion list on top of an optional allowlist."""
+    if not exclude_episodes:
+        return episodes
+    base = episodes if episodes is not None else list(range(total_episodes))
+    excluded = set(exclude_episodes)
+    return [episode for episode in base if episode not in excluded]
+
+
 def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDataset:
     """Handles the logic of setting up delta timestamps and image transforms before creating a dataset.
 
@@ -87,11 +98,14 @@ def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDatas
             cfg.dataset.repo_id, root=cfg.dataset.root, revision=cfg.dataset.revision
         )
         delta_timestamps = resolve_delta_timestamps(cfg.trainable_config, ds_meta)
+        episodes = _resolve_episodes(
+            cfg.dataset.episodes, cfg.dataset.exclude_episodes, ds_meta.total_episodes
+        )
         if not cfg.dataset.streaming:
             dataset = LeRobotDataset(
                 cfg.dataset.repo_id,
                 root=cfg.dataset.root,
-                episodes=cfg.dataset.episodes,
+                episodes=episodes,
                 delta_timestamps=delta_timestamps,
                 image_transforms=image_transforms,
                 revision=cfg.dataset.revision,
@@ -104,7 +118,7 @@ def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDatas
             dataset = StreamingLeRobotDataset(
                 cfg.dataset.repo_id,
                 root=cfg.dataset.root,
-                episodes=cfg.dataset.episodes,
+                episodes=episodes,
                 delta_timestamps=delta_timestamps,
                 image_transforms=image_transforms,
                 revision=cfg.dataset.revision,

src/lerobot/datasets/language_render.py

@@ -162,14 +162,28 @@ def render_sample(
     task: str | None = None,
     dataset_ctx: Any | None = None,
 ) -> RenderedMessages | None:
-    """Render the chat-style messages for a single dataset sample.
+    """Resolve one sample's bindings and render its message recipe.
 
-    Resolves the recipe's bindings against ``persistent`` and ``events`` rows
-    at frame timestamp ``t``, then expands the recipe's message templates.
-    Returns ``None`` if the resolved sample contains no target message.
+    Returns ``None`` when no text or low-level action supervision applies.
     """
     persistent_rows = _normalize_rows(persistent or [])
     event_rows = _normalize_rows(events or [])
+
+    # Route sparse VQA frames to a matching view-specific component before weighted selection.
+    # This avoids dropping annotated frames or selecting VQA without annotations.
+    if recipe.blend is not None:
+        vqa_rendered = _render_vqa_if_present(
+            recipe,
+            persistent=persistent_rows,
+            events=event_rows,
+            t=t,
+            sample_idx=sample_idx,
+            task=task,
+            dataset_ctx=dataset_ctx,
+        )
+        if vqa_rendered is not None:
+            return vqa_rendered
+
     selected_recipe = _select_recipe(recipe, sample_idx)
     bindings = _resolve_bindings(
         selected_recipe,
@@ -183,6 +197,55 @@ def render_sample(
     return _render_message_recipe(selected_recipe, bindings)
 
 
+def _render_vqa_if_present(
+    recipe: TrainingRecipe,
+    *,
+    persistent: Sequence[LanguageRow],
+    events: Sequence[LanguageRow],
+    t: float,
+    sample_idx: int,
+    task: str | None,
+    dataset_ctx: Any | None,
+) -> RenderedMessages | None:
+    """Render a matching VQA component, or return ``None`` for normal selection.
+
+    Multiple matching views are selected deterministically by relative weight.
+    """
+    assert recipe.blend is not None
+    renderable: list[tuple[float, RenderedMessages]] = []
+    for name, component in recipe.blend.items():
+        if not name.startswith("ask_vqa"):
+            continue
+        bindings = _resolve_bindings(
+            component,
+            persistent=persistent,
+            events=events,
+            t=t,
+            sample_idx=sample_idx,
+            task=task,
+            dataset_ctx=dataset_ctx,
+        )
+        rendered = _render_message_recipe(component, bindings)
+        if rendered is not None:
+            renderable.append((float(component.weight or 0.0), rendered))
+
+    if not renderable:
+        return None
+    if len(renderable) == 1:
+        return renderable[0][1]
+
+    # Choose among matching cameras by relative weight, or uniformly when all weights are zero.
+    total = sum(w for w, _ in renderable) or float(len(renderable))
+    digest = hashlib.blake2b(f"vqa:{sample_idx}".encode(), digest_size=8).digest()
+    draw = int.from_bytes(digest, "big") / 2**64 * total
+    cumulative = 0.0
+    for w, rendered in renderable:
+        cumulative += w or (total / len(renderable))
+        if draw < cumulative:
+            return rendered
+    return renderable[-1][1]
+
+
 def _select_recipe(recipe: TrainingRecipe, sample_idx: int) -> TrainingRecipe:
     """Pick a deterministic blend component for ``sample_idx`` (or return ``recipe``)."""
     if recipe.blend is None:
@@ -346,7 +409,9 @@ def _render_message_recipe(
         if turn.target:
             target_indices.append(message_idx)
 
-    if not target_indices:
+    # Keep samples with either text targets or low-level action supervision.
+    has_low_level = any(stream == "low_level" for stream in streams)
+    if not target_indices and not has_low_level:
         return None
 
     rendered = {
@@ -403,14 +468,12 @@ def _validate_rendered(rendered: RenderedMessages) -> None:
 
     if len(streams) != len(messages):
         raise ValueError("message_streams must be aligned with messages.")
-    if not target_indices:
-        raise ValueError("Rendered samples must contain at least one target message.")
+    # Require text or low-level action supervision.
+    if not target_indices and not any(s == "low_level" for s in streams):
+        raise ValueError("Rendered samples must contain a target message or a low_level-stream message.")
     for idx in target_indices:
         if idx < 0 or idx >= len(messages):
             raise ValueError(f"Target message index {idx} is out of bounds.")
-    # ``stream`` is enforced non-None at MessageTurn construction time
-    # (see ``MessageTurn.__post_init__``), so a missing stream here would
-    # mean the dataclass invariant was bypassed; no need to re-check.
 
 
 def _nth_relative(

src/lerobot/envs/libero.py

@@ -384,12 +384,7 @@ class LiberoEnv(gym.Env):
 
     def close(self):
         if self._env is not None:
-            try:
-                self._env.close()
-            finally:
-                # LIBERO deletes its inner env on close, so this wrapper must
-                # be recreated before the next reset.
-                self._env = None
+            self._env.close()
 
 
 def _make_env_fns(

src/lerobot/envs/robotwin.py

@@ -384,9 +384,7 @@ class RoboTwinEnv(gym.Env):
 
         self._env: Any | None = None  # deferred — created on first reset() inside worker
         self._step_count: int = 0
-        self._black_frame: np.ndarray = np.zeros(
-            (self.observation_height, self.observation_width, 3), dtype=np.uint8
-        )
+        self._black_frame = np.zeros((self.observation_height, self.observation_width, 3), dtype=np.uint8)
 
         image_spaces = {
             cam: spaces.Box(

src/lerobot/envs/vlabench.py

@@ -373,7 +373,7 @@ class VLABenchEnv(gym.Env):
 
         if action.shape[0] != 7:
             # Unknown layout — fall back to zero-pad so the sim doesn't crash.
-            padded: np.ndarray = np.zeros(ctrl_dim, dtype=np.float64)
+            padded = np.zeros(ctrl_dim, dtype=np.float64)
             padded[: min(action.shape[0], ctrl_dim)] = action[:ctrl_dim]
             return padded
 

src/lerobot/motors/dynamixel/tables.py

@@ -122,9 +122,6 @@ MODEL_ENCODING_TABLE = {
     "xm430-w350": X_SERIES_ENCODINGS_TABLE,
     "xm540-w270": X_SERIES_ENCODINGS_TABLE,
     "xc430-w150": X_SERIES_ENCODINGS_TABLE,
-    "xh540-w150": X_SERIES_ENCODINGS_TABLE,
-    "xc330-t288": X_SERIES_ENCODINGS_TABLE,
-    "xc330-t181": X_SERIES_ENCODINGS_TABLE,
 }
 
 # {model: model_resolution}
@@ -137,9 +134,6 @@ MODEL_RESOLUTION = {
     "xm430-w350": 4096,
     "xm540-w270": 4096,
     "xc430-w150": 4096,
-    "xh540-w150": 4096,
-    "xc330-t288": 4096,
-    "xc330-t181": 4096,
 }
 
 # {model: model_number}
@@ -151,9 +145,6 @@ MODEL_NUMBER_TABLE = {
     "xm430-w350": 1020,
     "xm540-w270": 1120,
     "xc430-w150": 1070,
-    "xh540-w150": 1110,
-    "xc330-t288": 1220,
-    "xc330-t181": 1210,
 }
 
 # {model: available_operating_modes}
@@ -165,9 +156,6 @@ MODEL_OPERATING_MODES = {
     "xm430-w350": [0, 1, 3, 4, 5, 16],
     "xm540-w270": [0, 1, 3, 4, 5, 16],
     "xc430-w150": [1, 3, 4, 16],
-    "xh540-w150": [0, 1, 3, 4, 5, 16],
-    "xc330-t288": [0, 1, 3, 4, 5, 16],
-    "xc330-t181": [0, 1, 3, 4, 5, 16],
 }
 
 MODEL_CONTROL_TABLE = {
@@ -178,9 +166,6 @@ MODEL_CONTROL_TABLE = {
     "xm430-w350": X_SERIES_CONTROL_TABLE,
     "xm540-w270": X_SERIES_CONTROL_TABLE,
     "xc430-w150": X_SERIES_CONTROL_TABLE,
-    "xh540-w150": X_SERIES_CONTROL_TABLE,
-    "xc330-t288": X_SERIES_CONTROL_TABLE,
-    "xc330-t181": X_SERIES_CONTROL_TABLE,
 }
 
 MODEL_BAUDRATE_TABLE = {
@@ -191,9 +176,6 @@ MODEL_BAUDRATE_TABLE = {
     "xm430-w350": X_SERIES_BAUDRATE_TABLE,
     "xm540-w270": X_SERIES_BAUDRATE_TABLE,
     "xc430-w150": X_SERIES_BAUDRATE_TABLE,
-    "xh540-w150": X_SERIES_BAUDRATE_TABLE,
-    "xc330-t288": X_SERIES_BAUDRATE_TABLE,
-    "xc330-t181": X_SERIES_BAUDRATE_TABLE,
 }
 
 AVAILABLE_BAUDRATES = [

src/lerobot/policies/evo1/processor_evo1.py

@@ -302,33 +302,6 @@ def _pad_evo1_stats(
     return padded_stats
 
 
-def _refresh_evo1_normalization_steps(
-    config: Evo1Config,
-    preprocessor: PolicyProcessorPipeline,
-    postprocessor: PolicyProcessorPipeline,
-) -> None:
-    """Re-pad checkpoint-loaded (un)normalizer stats/features to EVO1's fixed widths.
-
-    Loading a checkpoint injects the raw dataset stats (unpadded to max_state_dim/max_action_dim)
-    into the (un)normalizer via the generic override path in make_pre_post_processors. Those stats
-    and their declared features must be re-padded/reshaped to EVO1's fixed widths, otherwise
-    normalization fails against the padded state/action tensors (e.g. state padded to 24 vs. 8-dim
-    LIBERO stats). Padding is a no-op when stats are already at the target width.
-    """
-    normalization_features = _evo1_normalization_features(config)
-    action_features = _evo1_action_features(config)
-    for step in preprocessor.steps:
-        if isinstance(step, NormalizerProcessorStep):
-            step.features = normalization_features
-            step.stats = _pad_evo1_stats(config, step.stats)
-            step.to(device=step.device, dtype=step.dtype)
-    for step in postprocessor.steps:
-        if isinstance(step, UnnormalizerProcessorStep):
-            step.features = action_features
-            step.stats = _pad_evo1_stats(config, step.stats)
-            step.to(device=step.device, dtype=step.dtype)
-
-
 def reconcile_evo1_processors(
     config: Evo1Config,
     preprocessor: PolicyProcessorPipeline,
@@ -336,19 +309,16 @@ def reconcile_evo1_processors(
 ) -> tuple[PolicyProcessorPipeline, PolicyProcessorPipeline]:
     """Reconcile checkpoint-loaded pipelines with the current EVO1 config.
 
-    Three things cannot be restored from a serialized pipeline alone: the EVO1 batch converter
-    (converters are plain functions and are never serialized), eval-time CLI overrides of the
-    action postprocessing flags (`postprocess_action_dim`, `binarize_gripper`, `gripper_*`), and the
-    (un)normalizer stats/features when the generic override path injects raw, unpadded dataset
-    stats. This restores the converter, re-pads the normalization stats to EVO1's fixed widths, and
-    rebuilds the action step from the current config so those overrides take effect.
+    Two things cannot be restored from a serialized pipeline alone: the EVO1 batch converter
+    (converters are plain functions and are never serialized), and eval-time CLI overrides of the
+    action postprocessing flags (`postprocess_action_dim`, `binarize_gripper`, `gripper_*`). This
+    restores the converter and rebuilds the action step from the current config so those overrides
+    take effect.
     """
     # Pipelines reloaded from a checkpoint come back with the default batch converter, which drops
     # non-observation extras (embodiment_id, state_mask, custom task fields) needed by EVO1.
     preprocessor.to_transition = evo1_batch_to_transition
 
-    _refresh_evo1_normalization_steps(config, preprocessor, postprocessor)
-
     action_step = Evo1ActionProcessorStep(
         action_dim=_evo1_action_dim(config),
         binarize_gripper=config.binarize_gripper,

src/lerobot/policies/factory.py

@@ -44,19 +44,12 @@ from lerobot.utils.constants import (
     POLICY_PREPROCESSOR_DEFAULT_NAME,
 )
 from lerobot.utils.feature_utils import dataset_to_policy_features
-from lerobot.utils.import_utils import _peft_available, require_package
 
 from .evo1.configuration_evo1 import Evo1Config
 from .groot.configuration_groot import GrootConfig
 from .pretrained import PreTrainedPolicy
 from .utils import validate_visual_features_consistency
 
-if TYPE_CHECKING or _peft_available:
-    from peft import PeftConfig, PeftModel
-else:
-    PeftConfig = None
-    PeftModel = None
-
 
 def _reconnect_relative_absolute_steps(
     preprocessor: PolicyProcessorPipeline, postprocessor: PolicyProcessorPipeline
@@ -341,15 +334,12 @@ def make_policy(
         # Load a pretrained PEFT model on top of the policy. The pretrained path points to the folder/repo
         # of the adapter and the adapter's config contains the path to the base policy. So we need the
         # adapter config first, then load the correct policy and then apply PEFT.
-        require_package("peft", extra="peft")
+        from peft import PeftConfig, PeftModel
 
         logging.info("Loading policy's PEFT adapter.")
 
         peft_pretrained_path = str(cfg.pretrained_path)
-        peft_config = PeftConfig.from_pretrained(
-            peft_pretrained_path,
-            revision=cfg.pretrained_revision,
-        )
+        peft_config = PeftConfig.from_pretrained(peft_pretrained_path)
 
         kwargs["pretrained_name_or_path"] = peft_config.base_model_name_or_path
         if not kwargs["pretrained_name_or_path"]:
@@ -360,14 +350,9 @@ def make_policy(
                 "the adapter was trained."
             )
 
-        kwargs["revision"] = peft_config.revision
         policy = policy_cls.from_pretrained(**kwargs)
         policy = PeftModel.from_pretrained(
-            policy,
-            peft_pretrained_path,
-            config=peft_config,
-            revision=cfg.pretrained_revision,
-            is_trainable=True,
+            policy, peft_pretrained_path, config=peft_config, is_trainable=True
         )
 
     else:

src/lerobot/policies/molmoact2/modeling_molmoact2.py

@@ -43,22 +43,11 @@ from torch.distributions import Beta
 
 from lerobot.policies.pretrained import PreTrainedPolicy
 from lerobot.utils.constants import ACTION
-from lerobot.utils.import_utils import (
-    _peft_available,
-    _scipy_available,
-    _transformers_available,
-    require_package,
-)
+from lerobot.utils.import_utils import _scipy_available, _transformers_available, require_package
 
 from ..rtc.modeling_rtc import RTCProcessor
 from .configuration_molmoact2 import MolmoAct2Config
 
-if TYPE_CHECKING or _peft_available:
-    from peft import LoraConfig, get_peft_model
-else:
-    LoraConfig = None
-    get_peft_model = None
-
 logger = logging.getLogger(__name__)
 
 
@@ -1742,11 +1731,13 @@ class MolmoAct2Policy(PreTrainedPolicy):
 
     def _build_inner_lora_config(self):
         require_package("peft", extra="molmoact2")
+        from peft import LoraConfig
 
         return LoraConfig(**self._get_inner_peft_targets())
 
     def _apply_lora_adapters(self) -> None:
         require_package("peft", extra="molmoact2")
+        from peft import get_peft_model
 
         peft_config = self._build_inner_lora_config()
         self._validate_peft_config(peft_config)

src/lerobot/policies/pretrained.py

@@ -34,22 +34,14 @@ 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.import_utils import _peft_available, require_package
 
 from .utils import log_model_loading_keys
 
-if TYPE_CHECKING or _peft_available:
-    from peft import PEFT_TYPE_TO_CONFIG_MAPPING, PeftType, get_peft_model
-else:
-    PEFT_TYPE_TO_CONFIG_MAPPING = None
-    PeftType = None
-    get_peft_model = None
+T = TypeVar("T", bound="PreTrainedPolicy")
 
 if TYPE_CHECKING:
     from lerobot.datasets.dataset_metadata import LeRobotDatasetMetadata
 
-T = TypeVar("T", bound="PreTrainedPolicy")
-
 
 def _build_card_context(
     cfg: TrainPipelineConfig | None,
@@ -392,7 +384,7 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
             peft_cli_overrides: Optional dict of CLI overrides (method_type, target_modules, r, etc.)
                 These are merged with policy defaults to build the final config.
         """
-        require_package("peft", extra="peft")
+        from peft import get_peft_model
 
         # If user provided a complete config, use it directly (with overrides)
         if peft_config is not None:
@@ -463,7 +455,7 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
         Returns:
             Preprocessed dict with renamed keys and init_type mapped to method-specific key.
         """
-        require_package("peft", extra="peft")
+        from peft import PeftType
 
         cli_overrides = cli_overrides.copy()
 
@@ -488,7 +480,7 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
 
     def _build_peft_config(self, cli_overrides: dict):
         """Build a PEFT config from policy defaults and CLI overrides."""
-        require_package("peft", extra="peft")
+        from peft import PEFT_TYPE_TO_CONFIG_MAPPING, PeftType
 
         # Determine PEFT method type (default to LORA)
         method_type_str = cli_overrides.get("method_type") or "lora"
@@ -515,7 +507,7 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
 
     def _apply_peft_cli_overrides(self, peft_config, cli_overrides: dict):
         """Apply CLI overrides to an existing PEFT config."""
-        require_package("peft", extra="peft")
+        from peft import PEFT_TYPE_TO_CONFIG_MAPPING, PeftType
 
         # Get method type from existing config or CLI override
         method_type_str = cli_overrides.get("method_type")

src/lerobot/processor/batch_processor.py

@@ -175,9 +175,6 @@ class AddBatchDimensionComplementaryDataStep(ComplementaryDataProcessorStep):
             if isinstance(task_index_value, Tensor) and task_index_value.dim() == 0:
                 complementary_data["task_index"] = task_index_value.unsqueeze(0)
 
-        complementary_data.pop("language_persistent", None)
-        complementary_data.pop("language_events", None)
-
         if "messages" in complementary_data:
             messages = complementary_data["messages"]
             if isinstance(messages, list) and (not messages or isinstance(messages[0], dict)):

src/lerobot/processor/pipeline.py

@@ -41,7 +41,7 @@ from pathlib import Path
 from typing import Any, TypedDict, TypeVar, cast
 
 import torch
-from huggingface_hub import hf_hub_download
+from huggingface_hub import hf_hub_download, snapshot_download
 from safetensors.torch import load_file, save_file
 
 from lerobot.configs import PipelineFeatureType, PolicyFeature
@@ -205,6 +205,10 @@ class ProcessorStep(ABC):
         """
         return None
 
+    def save_artifacts(self, save_directory: Path) -> dict[str, str]:
+        """Save non-tensor assets and map constructor arguments to relative paths."""
+        return {}
+
     def reset(self) -> None:
         """Resets the internal state of the processor step, if any."""
         return None
@@ -549,6 +553,22 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
         pipeline_config = self.get_config()
         pipeline_state_dict = self.state_dict()
 
+        for processor_step, step_entry in zip(self.steps, pipeline_config["steps"], strict=True):
+            artifacts = processor_step.save_artifacts(save_directory)
+            if artifacts:
+                for config_key, relative_path in artifacts.items():
+                    artifact_path = Path(relative_path)
+                    if artifact_path.is_absolute() or ".." in artifact_path.parts:
+                        raise ValueError(
+                            f"Processor artifact path must be relative to the checkpoint: {relative_path!r}"
+                        )
+                    if not (save_directory / artifact_path).exists():
+                        raise FileNotFoundError(
+                            f"Processor step did not save declared artifact '{relative_path}'"
+                        )
+                    step_entry["config"][config_key] = artifact_path.as_posix()
+                step_entry["artifacts"] = artifacts
+
         for state_key, step_state_dict in pipeline_state_dict.items():
             state_filename = f"{state_key}.safetensors"
             save_file(step_state_dict, save_directory / state_filename)
@@ -733,7 +753,13 @@ 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, is_local_source
+            loaded_config,
+            overrides or {},
+            model_id,
+            base_path,
+            config_filename,
+            hub_download_kwargs,
+            is_local_source,
         )
 
         # 4. Validate that all overrides were used
@@ -922,6 +948,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
         overrides: dict[str, Any],
         model_id: str,
         base_path: Path | None,
+        config_filename: str,
         hub_download_kwargs: dict[str, Any],
         is_local_source: bool = False,
     ) -> tuple[list[ProcessorStep], set[str]]:
@@ -976,15 +1003,68 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
             ImportError: If a step class cannot be imported or found in registry
             ValueError: If a step cannot be instantiated with its configuration
         """
+        loaded_config = deepcopy(loaded_config)
+        cls._resolve_artifact_paths(
+            loaded_config,
+            model_id,
+            base_path,
+            config_filename,
+            hub_download_kwargs,
+        )
         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, is_local_source
+                step_instance,
+                step_entry,
+                model_id,
+                base_path,
+                config_filename,
+                hub_download_kwargs,
+                is_local_source,
             )
 
         return steps, remaining_override_keys
 
+    @classmethod
+    def _resolve_artifact_paths(
+        cls,
+        loaded_config: dict[str, Any],
+        model_id: str,
+        base_path: Path | None,
+        config_filename: str,
+        hub_download_kwargs: dict[str, Any],
+    ) -> None:
+        """Resolve declared relative processor artifacts before step construction."""
+        is_local = Path(model_id).is_dir() or Path(model_id).is_file()
+
+        for step_entry in loaded_config["steps"]:
+            artifacts = step_entry.get("artifacts", {})
+            for config_key, relative_path in artifacts.items():
+                artifact_path = Path(relative_path)
+                if artifact_path.is_absolute() or ".." in artifact_path.parts:
+                    raise ValueError(
+                        f"Processor artifact path must be relative to the checkpoint: {relative_path!r}"
+                    )
+
+                resolved_path = base_path / artifact_path if base_path is not None else artifact_path
+                if not resolved_path.exists() and not is_local:
+                    repository_path = Path(config_filename).parent / artifact_path
+                    snapshot_download(
+                        repo_id=model_id,
+                        repo_type="model",
+                        allow_patterns=f"{repository_path.as_posix()}/**",
+                        **hub_download_kwargs,
+                    )
+
+                if not resolved_path.exists():
+                    step_name = step_entry.get("registry_name", step_entry.get("class", "unknown"))
+                    raise FileNotFoundError(
+                        f"Missing processor artifact '{relative_path}' for step '{step_name}' "
+                        f"next to '{config_filename}'. Checkpoint artifacts are incomplete."
+                    )
+                step_entry["config"][config_key] = str(resolved_path)
+
     @classmethod
     def _build_steps_from_config(
         cls,
@@ -1144,6 +1224,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
         step_entry: dict[str, Any],
         model_id: str,
         base_path: Path | None,
+        config_filename: str,
         hub_download_kwargs: dict[str, Any],
         is_local_source: bool = False,
     ) -> None:
@@ -1209,7 +1290,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
             # Download from Hub
             state_path = hf_hub_download(
                 repo_id=model_id,
-                filename=state_filename,
+                filename=(Path(config_filename).parent / state_filename).as_posix(),
                 repo_type="model",
                 **hub_download_kwargs,
             )

src/lerobot/processor/render_messages_processor.py

@@ -16,7 +16,7 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 from typing import Any
 
 from lerobot.configs import PipelineFeatureType, PolicyFeature
@@ -32,17 +32,18 @@ from .pipeline import ProcessorStep, ProcessorStepRegistry
 @dataclass
 @ProcessorStepRegistry.register(name="render_messages_processor")
 class RenderMessagesStep(ProcessorStep):
-    """Processor step that turns raw language columns into rendered chat messages.
-
-    Reads ``language_persistent`` and ``language_events`` from the transition's
-    complementary data, renders them through ``recipe`` at the sample timestamp,
-    and replaces the raw columns with the resulting ``messages`` /
-    ``message_streams`` / ``target_message_indices`` keys.
-    """
+    """Render language columns into recipe-defined messages and supervision metadata."""
 
     recipe: TrainingRecipe
     dataset_ctx: Any | None = None
 
+    def __post_init__(self) -> None:
+        if isinstance(self.recipe, dict):
+            self.recipe = TrainingRecipe.from_dict(self.recipe)
+
+    def get_config(self) -> dict[str, Any]:
+        return {"recipe": asdict(self.recipe)}
+
     def __call__(self, transition: EnvTransition) -> EnvTransition | None:
         """Render messages for a single transition; return ``None`` to drop it."""
         complementary_data = transition.get(TransitionKey.COMPLEMENTARY_DATA) or {}
@@ -50,7 +51,17 @@ class RenderMessagesStep(ProcessorStep):
         events = complementary_data.get(LANGUAGE_EVENTS) or []
 
         if not persistent and not events:
-            return transition
+            rendered = _fallback_low_level_render(complementary_data.get("task"))
+            if rendered is None:
+                return transition
+            new_transition = transition.copy()
+            new_complementary_data = dict(new_transition.get(TransitionKey.COMPLEMENTARY_DATA) or {})
+            new_complementary_data.update(rendered)
+            new_transition[TransitionKey.COMPLEMENTARY_DATA] = new_complementary_data
+            return new_transition
+
+        if _is_batched_language(persistent) or _is_batched_language(events):
+            return self._call_batch(transition, complementary_data, persistent, events)
 
         timestamp = complementary_data.get("timestamp")
         if timestamp is None:
@@ -67,18 +78,147 @@ class RenderMessagesStep(ProcessorStep):
             dataset_ctx=self.dataset_ctx,
         )
         if rendered is None:
-            return None
+            rendered = _fallback_low_level_render(complementary_data.get("task"))
+            if rendered is None:
+                return None
 
         new_transition = transition.copy()
-        new_complementary_data = dict(complementary_data)
+        new_complementary_data = dict(new_transition.get(TransitionKey.COMPLEMENTARY_DATA) or {})
         new_complementary_data.pop(LANGUAGE_PERSISTENT, None)
         new_complementary_data.pop(LANGUAGE_EVENTS, None)
         new_complementary_data.update(rendered)
         new_transition[TransitionKey.COMPLEMENTARY_DATA] = new_complementary_data
         return new_transition
 
+    def _call_batch(
+        self,
+        transition: EnvTransition,
+        complementary_data: dict[str, Any],
+        persistent_batch: list,
+        events_batch: list,
+    ) -> EnvTransition | None:
+        timestamp = complementary_data.get("timestamp")
+        if timestamp is None:
+            raise KeyError("RenderMessagesStep requires sample timestamp in complementary data.")
+
+        batch_size = max(len(persistent_batch), len(events_batch))
+        messages: list[list[dict[str, Any]]] = []
+        message_streams: list[list[str | None]] = []
+        target_message_indices: list[list[int]] = []
+        keep_indices: list[int] = []
+
+        for i in range(batch_size):
+            rendered = render_sample(
+                recipe=self.recipe,
+                persistent=persistent_batch[i] if i < len(persistent_batch) else [],
+                events=events_batch[i] if i < len(events_batch) else [],
+                t=_batch_value(timestamp, i),
+                sample_idx=int(_batch_value(complementary_data.get("index", 0), i)),
+                task=_batch_value(complementary_data.get("task"), i),
+                dataset_ctx=self.dataset_ctx,
+            )
+            if rendered is None:
+                rendered = _fallback_low_level_render(_batch_value(complementary_data.get("task"), i))
+                if rendered is None:
+                    continue
+            keep_indices.append(i)
+            messages.append(rendered["messages"])
+            message_streams.append(rendered["message_streams"])
+            target_message_indices.append(rendered["target_message_indices"])
+
+        if not messages:
+            return None
+
+        new_transition = (
+            _select_batch_indices(transition, keep_indices)
+            if len(keep_indices) != batch_size
+            else transition.copy()
+        )
+        new_complementary_data = dict(new_transition.get(TransitionKey.COMPLEMENTARY_DATA) or {})
+        new_complementary_data.pop(LANGUAGE_PERSISTENT, None)
+        new_complementary_data.pop(LANGUAGE_EVENTS, None)
+        new_complementary_data["messages"] = messages
+        new_complementary_data["message_streams"] = message_streams
+        new_complementary_data["target_message_indices"] = target_message_indices
+        new_transition[TransitionKey.COMPLEMENTARY_DATA] = new_complementary_data
+        return new_transition
+
     def transform_features(
         self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
     ) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
         """Pass features through unchanged; rendering only touches complementary data."""
         return features
+
+
+def _scalar(value: Any) -> float | int:
+    """Unwrap a tensor/array/single-element list into a Python scalar."""
+    if hasattr(value, "item"):
+        return value.item()
+    if isinstance(value, list):
+        if len(value) != 1:
+            raise ValueError(f"Expected a scalar, got list of length {len(value)}: {value!r}")
+        return _scalar(value[0])
+    return value
+
+
+def _is_batched_language(value: Any) -> bool:
+    return isinstance(value, list) and bool(value) and isinstance(value[0], list)
+
+
+def _batch_value(value: Any, index: int) -> Any:
+    if value is None:
+        return None
+    if isinstance(value, list):
+        return value[index]
+    if hasattr(value, "ndim") and value.ndim > 0:
+        return _scalar(value[index])
+    return _scalar(value)
+
+
+def _select_batch_indices(transition: EnvTransition, indices: list[int]) -> EnvTransition:
+    selected = transition.copy()
+    for key in (TransitionKey.OBSERVATION, TransitionKey.COMPLEMENTARY_DATA):
+        data = selected.get(key)
+        if isinstance(data, dict):
+            selected[key] = {k: _select_value(v, indices) for k, v in data.items()}
+    action = selected.get(TransitionKey.ACTION)
+    if action is not None:
+        selected[TransitionKey.ACTION] = _select_value(action, indices)
+    return selected
+
+
+def _select_value(value: Any, indices: list[int]) -> Any:
+    if isinstance(value, list) and len(value) >= len(indices):
+        return [value[i] for i in indices]
+    if hasattr(value, "index_select") and hasattr(value, "new_tensor") and getattr(value, "ndim", 0) > 0:
+        return value.index_select(0, value.new_tensor(indices).long())
+    return value
+
+
+def _fallback_low_level_render(task: Any) -> dict[str, Any] | None:
+    """Keep action-only samples trainable when no recipe branch matches."""
+    if hasattr(task, "item"):
+        task = task.item()
+    if isinstance(task, list):
+        messages = []
+        message_streams = []
+        target_message_indices = []
+        for t in task:
+            rendered = _fallback_low_level_render(t)
+            if rendered is None:
+                return None
+            messages.append(rendered["messages"])
+            message_streams.append(rendered["message_streams"])
+            target_message_indices.append(rendered["target_message_indices"])
+        return {
+            "messages": messages,
+            "message_streams": message_streams,
+            "target_message_indices": target_message_indices,
+        }
+    if not isinstance(task, str) or not task:
+        return None
+    return {
+        "messages": [{"role": "user", "content": task}],
+        "message_streams": ["low_level"],
+        "target_message_indices": [],
+    }

src/lerobot/processor/tokenizer_processor.py

@@ -25,6 +25,7 @@ from __future__ import annotations
 
 import logging
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 import torch
@@ -32,6 +33,7 @@ import torch
 from lerobot.configs import FeatureType, PipelineFeatureType, PolicyFeature
 from lerobot.types import EnvTransition, RobotObservation, TransitionKey
 from lerobot.utils.constants import (
+    ACTION_CODE_TOKEN_MASK,
     ACTION_TOKEN_MASK,
     ACTION_TOKENS,
     OBS_LANGUAGE_ATTENTION_MASK,
@@ -136,7 +138,7 @@ class TokenizerProcessorStep(ObservationProcessorStep):
         # Standardize to a list of strings for the tokenizer
         if isinstance(task, str):
             return [task]
-        elif isinstance(task, (list, tuple)) and all(isinstance(t, str) for t in task):
+        elif isinstance(task, list | tuple) and all(isinstance(t, str) for t in task):
             return list(task)
 
         return None
@@ -349,6 +351,7 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
     max_action_tokens: int = 256
     fast_skip_tokens: int = 128
     paligemma_tokenizer_name: str = "google/paligemma-3b-pt-224"
+    allow_truncation: bool = True
     # Internal tokenizer instance (not part of the config)
     action_tokenizer: Any = field(default=None, init=False, repr=False)
     _paligemma_tokenizer: Any = field(default=None, init=False, repr=False)
@@ -412,14 +415,15 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
             # During inference, no action is available, skip tokenization
             return new_transition
 
-        # Tokenize and get both tokens and mask
-        tokens, mask = self._tokenize_action(action)
+        # Tokenize and get masks for the full formatted sequence and the discrete action codes.
+        tokens, mask, code_mask = self._tokenize_action(action)
 
         # Store mask in complementary data
         complementary_data = new_transition.get(TransitionKey.COMPLEMENTARY_DATA, {})
         if complementary_data is None:
             complementary_data = {}
         complementary_data[ACTION_TOKEN_MASK] = mask
+        complementary_data[ACTION_CODE_TOKEN_MASK] = code_mask
         complementary_data[ACTION_TOKENS] = tokens
         new_transition[TransitionKey.COMPLEMENTARY_DATA] = complementary_data
         return new_transition
@@ -430,7 +434,7 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
         """
         return self._paligemma_tokenizer.vocab_size - 1 - self.fast_skip_tokens - tokens
 
-    def _tokenize_action(self, action: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
+    def _tokenize_action(self, action: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
         """
         Tokenizes the action tensor and creates a mask.
 
@@ -459,6 +463,7 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
         # The fast tokenizer expects action data and returns token IDs
         tokens_list = []
         masks_list = []
+        code_masks_list = []
 
         for i in range(batch_size):
             # Tokenize single action (move to CPU first as tokenizer uses scipy which requires numpy)
@@ -476,65 +481,82 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
             if tokens.dim() > 1:
                 tokens = tokens.flatten()
 
+            action_code_tokens = self._act_tokens_to_paligemma_tokens(tokens)
             bos_id = self._paligemma_tokenizer.bos_token_id
-            # add bos
+            prompt_tokens = torch.tensor(
+                self._paligemma_tokenizer.encode("Action: ", add_special_tokens=False),
+                device=action.device,
+            )
+            end_tokens = torch.tensor(self._paligemma_tokenizer.encode("|"), device=action.device)
+
+            code_start = 1 + len(prompt_tokens)
+            code_end = code_start + len(action_code_tokens)
             tokens = torch.cat(
                 [
                     torch.tensor([bos_id], device=action.device),
-                    torch.tensor(
-                        self._paligemma_tokenizer.encode("Action: ", add_special_tokens=False),
-                        device=action.device,
-                    ),
-                    self._act_tokens_to_paligemma_tokens(tokens),
-                    torch.tensor(self._paligemma_tokenizer.encode("|"), device=action.device),
+                    prompt_tokens,
+                    action_code_tokens,
+                    end_tokens,
                 ]
             )
+            code_mask = torch.zeros(len(tokens), dtype=torch.bool, device=action.device)
+            code_mask[code_start:code_end] = True
 
             # Truncate or pad to max_action_tokens
             if len(tokens) > self.max_action_tokens:
+                if not self.allow_truncation:
+                    raise ValueError(
+                        f"FAST action sequence has {len(tokens)} tokens, exceeding "
+                        f"max_action_tokens={self.max_action_tokens}."
+                    )
                 logging.warning(
                     f"Token length ({len(tokens)}) exceeds max length ({self.max_action_tokens}), truncating. "
                     "Consider increasing the `max_action_tokens` in your model config if this happens frequently."
                 )
                 tokens = tokens[: self.max_action_tokens]
+                code_mask = code_mask[: self.max_action_tokens]
                 mask = torch.ones(self.max_action_tokens, dtype=torch.bool, device=action.device)
             else:
+                pad_len = self.max_action_tokens - len(tokens)
                 mask = torch.cat(
                     [
                         torch.ones(len(tokens), dtype=torch.bool, device=action.device),
-                        torch.zeros(
-                            self.max_action_tokens - len(tokens), dtype=torch.bool, device=action.device
-                        ),
+                        torch.zeros(pad_len, dtype=torch.bool, device=action.device),
                     ]
                 )
+                code_mask = torch.nn.functional.pad(code_mask, (0, pad_len), value=False)
                 # Pad tokens with zeros
-                tokens = torch.nn.functional.pad(tokens, (0, self.max_action_tokens - len(tokens)), value=0)
+                tokens = torch.nn.functional.pad(tokens, (0, pad_len), value=0)
 
             tokens_list.append(tokens)
             masks_list.append(mask)
+            code_masks_list.append(code_mask)
 
         # Stack into batched tensors
         tokens_batch = torch.stack(tokens_list, dim=0)  # (B, max_action_tokens)
         masks_batch = torch.stack(masks_list, dim=0)  # (B, max_action_tokens)
+        code_masks_batch = torch.stack(code_masks_list, dim=0)  # (B, max_action_tokens)
 
         # Remove batch dimension if input was single sample
         if single_sample:
             tokens_batch = tokens_batch.squeeze(0)
             masks_batch = masks_batch.squeeze(0)
+            code_masks_batch = code_masks_batch.squeeze(0)
 
         # Move to the same device as the input
         if device is not None:
             tokens_batch = tokens_batch.to(device)
             masks_batch = masks_batch.to(device)
+            code_masks_batch = code_masks_batch.to(device)
 
-        return tokens_batch, masks_batch
+        return tokens_batch, masks_batch, code_masks_batch
 
     def action(self, action: torch.Tensor) -> torch.Tensor:
         """
         This method is not used since we override __call__.
         Required by ActionProcessorStep ABC.
         """
-        tokens, _ = self._tokenize_action(action)
+        tokens, _, _ = self._tokenize_action(action)
         return tokens
 
     def get_config(self) -> dict[str, Any]:
@@ -550,6 +572,9 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
         config = {
             "trust_remote_code": self.trust_remote_code,
             "max_action_tokens": self.max_action_tokens,
+            "fast_skip_tokens": self.fast_skip_tokens,
+            "paligemma_tokenizer_name": self.paligemma_tokenizer_name,
+            "allow_truncation": self.allow_truncation,
         }
 
         # Only save tokenizer_name if it was used to create the tokenizer
@@ -558,6 +583,14 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
 
         return config
 
+    def save_artifacts(self, save_directory: Path) -> dict[str, str]:
+        artifact_path = Path("action_tokenizer")
+        save_pretrained = getattr(self.action_tokenizer, "save_pretrained", None)
+        if save_pretrained is None:
+            raise TypeError("Action tokenizer must implement save_pretrained() to save a portable pipeline.")
+        save_pretrained(save_directory / artifact_path)
+        return {"action_tokenizer_name": artifact_path.as_posix()}
+
     def transform_features(
         self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
     ) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:

src/lerobot/robots/so_follower/config_so_follower.py

@@ -46,12 +46,6 @@ class SOFollowerConfig:
     position_i_coefficient: int = 0
     position_d_coefficient: int = 32
 
-    # Number of extra attempts when a `sync_read` of the motors fails. Feetech buses can occasionally
-    # return a corrupted status packet ("Incorrect status packet!"), especially when several joints move
-    # at once, which otherwise aborts the control loop. Retries are immediate (no sleep) and only happen on
-    # failure, so the steady-state read cost is unchanged.
-    num_read_retries: int = 2
-
 
 @RobotConfig.register_subclass("so101_follower")
 @RobotConfig.register_subclass("so100_follower")

src/lerobot/robots/so_follower/so_follower.py

@@ -142,25 +142,11 @@ class SOFollower(Robot):
         range_mins[full_turn_motor] = 0
         range_maxes[full_turn_motor] = 4095
 
-        drive_modes = dict.fromkeys(self.bus.motors, 0)
-        input(f"Fully close the gripper of {self} and press ENTER....")
-        gripper_closed_pos = self.bus.read(
-            "Present_Position", "gripper", normalize=False, num_retry=self.config.num_read_retries
-        )
-        distance_to_min = abs(gripper_closed_pos - range_mins["gripper"])
-        distance_to_max = abs(gripper_closed_pos - range_maxes["gripper"])
-        if min(distance_to_min, distance_to_max) > (range_maxes["gripper"] - range_mins["gripper"]) * 0.2:
-            raise ValueError("Gripper is not fully closed. Run calibration again.")
-
-        drive_modes["gripper"] = int(distance_to_max < distance_to_min)
-        if drive_modes["gripper"]:
-            logger.info("Gripper motor is inverted, setting drive_mode=1 to compensate.")
-
         self.calibration = {}
         for motor, m in self.bus.motors.items():
             self.calibration[motor] = MotorCalibration(
                 id=m.id,
-                drive_mode=drive_modes[motor],
+                drive_mode=0,
                 homing_offset=homing_offsets[motor],
                 range_min=range_mins[motor],
                 range_max=range_maxes[motor],
@@ -194,7 +180,7 @@ class SOFollower(Robot):
     def get_observation(self) -> RobotObservation:
         # Read arm position
         start = time.perf_counter()
-        obs_dict = self.bus.sync_read("Present_Position", num_retry=self.config.num_read_retries)
+        obs_dict = self.bus.sync_read("Present_Position")
         obs_dict = {f"{motor}.pos": val for motor, val in obs_dict.items()}
         dt_ms = (time.perf_counter() - start) * 1e3
         logger.debug(f"{self} read state: {dt_ms:.1f}ms")
@@ -235,7 +221,7 @@ class SOFollower(Robot):
         # Cap goal position when too far away from present position.
         # /!\ Slower fps expected due to reading from the follower.
         if self.config.max_relative_target is not None:
-            present_pos = self.bus.sync_read("Present_Position", num_retry=self.config.num_read_retries)
+            present_pos = self.bus.sync_read("Present_Position")
             goal_present_pos = {key: (g_pos, present_pos[key]) for key, g_pos in goal_pos.items()}
             goal_pos = ensure_safe_goal_position(goal_present_pos, self.config.max_relative_target)
 

src/lerobot/rollout/context.py

@@ -24,7 +24,6 @@ from __future__ import annotations
 import logging
 from dataclasses import dataclass, field
 from threading import Event
-from typing import TYPE_CHECKING
 
 import torch
 
@@ -48,7 +47,6 @@ from lerobot.processor.relative_action_processor import RelativeActionsProcessor
 from lerobot.robots import make_robot_from_config
 from lerobot.teleoperators import Teleoperator, make_teleoperator_from_config
 from lerobot.utils.feature_utils import combine_feature_dicts, hw_to_dataset_features
-from lerobot.utils.import_utils import _peft_available, require_package
 
 from .configs import BaseStrategyConfig, DAggerStrategyConfig, RolloutConfig
 from .inference import (
@@ -59,12 +57,6 @@ from .inference import (
 )
 from .robot_wrapper import ThreadSafeRobot
 
-if TYPE_CHECKING or _peft_available:
-    from peft import PeftConfig, PeftModel
-else:
-    PeftConfig = None
-    PeftModel = None
-
 logger = logging.getLogger(__name__)
 
 
@@ -179,7 +171,7 @@ def _load_pretrained_policy(policy_config: PreTrainedConfig) -> PreTrainedPolicy
             revision=pretrained_revision,
         )
 
-    require_package("peft", extra="peft")
+    from peft import PeftConfig, PeftModel
 
     peft_path = policy_config.pretrained_path
     peft_config = PeftConfig.from_pretrained(peft_path, revision=pretrained_revision)
@@ -302,22 +294,12 @@ def build_rollout_context(
     # ``observation_features`` values are either a tuple (camera shape) or the
     # ``float`` type itself used as a sentinel for scalar motor features —
     # see ``dict[str, type | tuple]`` annotation on ``Robot.observation_features``.
-    # Keep cameras (tuple) plus both joint-position (.pos) and base-velocity (.vel)
-    # scalar state features. LeKiwi's observation.state is 9-dim (6 arm .pos +
-    # x/y/theta.vel) and the policy was trained/normalized on all 9; the old .pos-only
-    # filter fed a 6-dim state into a 9-dim normalizer → RuntimeError (size 6 vs 9).
-    # Pure-arm robots have no .vel state keys, so this is a no-op for them.
     observation_features_hw = {
         k: v
         for k, v in all_obs_features.items()
-        if isinstance(v, tuple) or (v is float and k.endswith((".pos", ".vel")))
+        if isinstance(v, tuple) or (v is float and k.endswith(".pos"))
     }
-    # Keep both joint-position (.pos) and base-velocity (.vel) action features so
-    # mobile manipulators command the base too (e.g. LeKiwi: 6 arm .pos +
-    # x/y/theta.vel = 9-dim action). Pure-arm robots have no .vel keys, so this is
-    # a no-op for them. Without the .vel keys the base velocities are silently
-    # dropped from dataset_features[ACTION]/ordered_action_keys and the base never moves.
-    action_features_hw = {k: v for k, v in robot.action_features.items() if k.endswith((".pos", ".vel"))}
+    action_features_hw = {k: v for k, v in robot.action_features.items() if k.endswith(".pos")}
 
     # The action side is always needed: sync inference reads action names from
     # ``dataset_features[ACTION]`` to map policy tensors back to robot actions.

src/lerobot/scripts/augment_dataset_quantile_stats.py

@@ -36,7 +36,6 @@ python src/lerobot/scripts/augment_dataset_quantile_stats.py \
 import argparse
 import concurrent.futures
 import logging
-import os
 from pathlib import Path
 
 import numpy as np
@@ -53,7 +52,6 @@ from lerobot.datasets import (
     get_feature_stats,
     write_stats,
 )
-from lerobot.datasets.compute_stats import sample_indices
 from lerobot.utils.utils import init_logging
 
 
@@ -79,14 +77,12 @@ def has_quantile_stats(stats: dict[str, dict] | None, quantile_list_keys: list[s
     return False
 
 
-def process_single_episode(dataset: LeRobotDataset, episode_idx: int, use_sampling: bool = True) -> dict:
+def process_single_episode(dataset: LeRobotDataset, episode_idx: int) -> dict:
     """Process a single episode and return its statistics.
 
     Args:
         dataset: The LeRobot dataset
         episode_idx: Index of the episode to process
-        use_sampling: If True, sub-sample image/video frames per episode to bound
-            memory. If False, use every frame (exact, higher memory).
 
     Returns:
         Dictionary containing episode statistics
@@ -96,31 +92,16 @@ def process_single_episode(dataset: LeRobotDataset, episode_idx: int, use_sampli
     start_idx = dataset.meta.episodes[episode_idx]["dataset_from_index"]
     end_idx = dataset.meta.episodes[episode_idx]["dataset_to_index"]
 
-    episode_len = end_idx - start_idx
-
-    # Images/video are the memory hog, so sub-sample those frames per episode;
-    # numeric columns are cheap, so read them in full (exact).
-    image_keys = [k for k in dataset.features if dataset.features[k]["dtype"] in ("image", "video")]
-    numeric_keys = [
-        k for k in dataset.features if dataset.features[k]["dtype"] not in ("image", "video", "string")
-    ]
-
     collected_data: dict[str, list] = {}
+    for idx in range(start_idx, end_idx):
+        item = dataset[idx]
+        for key, value in item.items():
+            if key not in dataset.features:
+                continue
 
-    # Numeric features: every frame, read directly from the underlying table.
-    if numeric_keys:
-        numeric_cols = dataset.hf_dataset.select_columns(numeric_keys)[start_idx:end_idx]
-        for key in numeric_keys:
-            collected_data[key] = [torch.as_tensor(v) for v in numeric_cols[key]]
-
-    # Image/video features: decode only a sampled subset of frames.
-    if image_keys:
-        sampled_offsets = sample_indices(episode_len) if use_sampling else list(range(episode_len))
-        for offset in sampled_offsets:
-            item = dataset[start_idx + offset]
-            for key in image_keys:
-                if key in item:
-                    collected_data.setdefault(key, []).append(item[key])
+            if key not in collected_data:
+                collected_data[key] = []
+            collected_data[key].append(value)
 
     ep_stats = {}
     for key, data_list in collected_data.items():
@@ -150,13 +131,11 @@ def process_single_episode(dataset: LeRobotDataset, episode_idx: int, use_sampli
     return ep_stats
 
 
-def compute_quantile_stats_for_dataset(dataset: LeRobotDataset, use_sampling: bool = True) -> dict[str, dict]:
+def compute_quantile_stats_for_dataset(dataset: LeRobotDataset) -> dict[str, dict]:
     """Compute quantile statistics for all episodes in the dataset.
 
     Args:
         dataset: The LeRobot dataset to compute statistics for
-        use_sampling: If True, sub-sample image/video frames per episode to bound
-            memory. If False, use every frame (exact, higher memory).
 
     Returns:
         Dictionary containing aggregated statistics with quantiles
@@ -174,15 +153,15 @@ def compute_quantile_stats_for_dataset(dataset: LeRobotDataset, use_sampling: bo
     if has_videos:
         logging.info("Dataset contains video keys - using sequential processing for thread safety")
         for episode_idx in tqdm(range(dataset.num_episodes), desc="Processing episodes"):
-            ep_stats = process_single_episode(dataset, episode_idx, use_sampling)
+            ep_stats = process_single_episode(dataset, episode_idx)
             episode_stats_list.append(ep_stats)
     else:
         logging.info("Dataset has no video keys - using parallel processing for better performance")
-        max_workers = min(dataset.num_episodes, int(os.environ.get("LEROBOT_STATS_MAX_WORKERS", 16)))
+        max_workers = min(dataset.num_episodes, 16)
 
         with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
             future_to_episode = {
-                executor.submit(process_single_episode, dataset, episode_idx, use_sampling): episode_idx
+                executor.submit(process_single_episode, dataset, episode_idx): episode_idx
                 for episode_idx in range(dataset.num_episodes)
             }
 
@@ -209,7 +188,6 @@ def augment_dataset_with_quantile_stats(
     repo_id: str,
     root: str | Path | None = None,
     overwrite: bool = False,
-    use_sampling: bool = True,
 ) -> None:
     """Augment a dataset with quantile statistics if they are missing.
 
@@ -217,8 +195,6 @@ def augment_dataset_with_quantile_stats(
         repo_id: Repository ID of the dataset
         root: Local root directory for the dataset
         overwrite: Overwrite existing quantile statistics if they already exist
-        use_sampling: If True, sub-sample image/video frames per episode to bound
-            memory. If False, use every frame (exact, higher memory).
     """
     logging.info(f"Loading dataset: {repo_id}")
     dataset = LeRobotDataset(
@@ -232,7 +208,7 @@ def augment_dataset_with_quantile_stats(
 
     logging.info("Dataset does not contain quantile statistics. Computing them now...")
 
-    new_stats = compute_quantile_stats_for_dataset(dataset, use_sampling=use_sampling)
+    new_stats = compute_quantile_stats_for_dataset(dataset)
 
     logging.info("Updating dataset metadata with new quantile statistics")
     dataset.meta.stats = new_stats
@@ -272,14 +248,6 @@ def main():
         action="store_true",
         help="Overwrite existing quantile statistics if they already exist",
     )
-    parser.add_argument(
-        "--no-sampling",
-        action="store_true",
-        help=(
-            "Compute stats over every frame (exact, higher memory). By default, "
-            "image/video frames are sub-sampled per episode to bound memory."
-        ),
-    )
 
     args = parser.parse_args()
     root = Path(args.root) if args.root else None
@@ -290,7 +258,6 @@ def main():
         repo_id=args.repo_id,
         root=root,
         overwrite=args.overwrite,
-        use_sampling=not args.no_sampling,
     )
 
 

src/lerobot/scripts/convert_dataset_v21_to_v30.py

@@ -94,8 +94,6 @@ from lerobot.datasets.video_utils import concatenate_video_files, get_video_dura
 from lerobot.utils.constants import HF_LEROBOT_HOME
 from lerobot.utils.utils import flatten_dict, init_logging
 
-logger = logging.getLogger(__name__)
-
 V21 = "v2.1"
 V30 = "v3.0"
 
@@ -478,11 +476,11 @@ def convert_dataset(
     # First check if the dataset already has a v3.0 version
     if root is None and not force_conversion:
         try:
-            logger.info("Trying to download v3.0 version of the dataset from the hub...")
+            print("Trying to download v3.0 version of the dataset from the hub...")
             snapshot_download(repo_id, repo_type="dataset", revision=V30, local_dir=HF_LEROBOT_HOME / repo_id)
             return
         except Exception:
-            logger.info("Dataset does not have an uploaded v3.0 version. Continuing with conversion.")
+            print("Dataset does not have an uploaded v3.0 version. Continuing with conversion.")
 
     # Set root based on whether local dataset path is provided
     use_local_dataset = False
@@ -490,7 +488,7 @@ def convert_dataset(
     if root.exists():
         validate_local_dataset_version(root)
         use_local_dataset = True
-        logger.info(f"Using local dataset at {root}")
+        print(f"Using local dataset at {root}")
 
     old_root = root.parent / f"{root.name}_old"
     new_root = root.parent / f"{root.name}_v30"
@@ -525,7 +523,7 @@ def convert_dataset(
         try:
             hub_api.delete_tag(repo_id, tag=CODEBASE_VERSION, repo_type="dataset")
         except (HTTPError, RevisionNotFoundError) as e:
-            logger.warning(f"tag={CODEBASE_VERSION} probably doesn't exist. Skipping exception ({e})")
+            print(f"tag={CODEBASE_VERSION} probably doesn't exist. Skipping exception ({e})")
             pass
         hub_api.delete_files(
             delete_patterns=["data/chunk*/episode_*", "meta/*.jsonl", "videos/chunk*"],

src/lerobot/scripts/lerobot_annotate.py

@@ -154,14 +154,14 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
     repo_id = cfg.new_repo_id or cfg.repo_id
     commit_message = cfg.push_commit_message or "Add steerable annotations (lerobot-annotate)"
     api = HfApi()
-    logger.info(f"[lerobot-annotate] creating/locating dataset repo {repo_id}...")
+    print(f"[lerobot-annotate] creating/locating dataset repo {repo_id}...", flush=True)
     api.create_repo(
         repo_id=repo_id,
         repo_type="dataset",
         private=cfg.push_private,
         exist_ok=True,
     )
-    logger.info(f"[lerobot-annotate] uploading {root} -> {repo_id}...")
+    print(f"[lerobot-annotate] uploading {root} -> {repo_id}...", flush=True)
     commit_info = api.upload_folder(
         folder_path=str(root),
         repo_id=repo_id,
@@ -172,7 +172,7 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
         # at the source dataset; a fresh card is generated below instead.
         ignore_patterns=[".annotate_staging/**", "**/.DS_Store", "README.md"],
     )
-    logger.info(f"[lerobot-annotate] uploaded to https://huggingface.co/datasets/{repo_id}")
+    print(f"[lerobot-annotate] uploaded to https://huggingface.co/datasets/{repo_id}", flush=True)
 
     dataset_info = load_info(root)
     card = create_lerobot_dataset_card(dataset_info=dataset_info, license="apache-2.0", repo_id=repo_id)
@@ -200,13 +200,14 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
         with suppress(RevisionNotFoundError):
             api.delete_tag(repo_id, tag=version_tag, repo_type="dataset")
         api.create_tag(**tag_kwargs)
-        logger.info(f"[lerobot-annotate] tagged {repo_id} as {version_tag}")
+        print(f"[lerobot-annotate] tagged {repo_id} as {version_tag}", flush=True)
     except Exception as exc:  # noqa: BLE001
-        logger.warning(
+        print(
             f"[lerobot-annotate] WARNING: could not create tag {version_tag!r} on {repo_id}: {exc}. "
             "Dataset is uploaded but ``LeRobotDataset`` won't be able to load it until it's tagged. "
             "Run: from huggingface_hub import HfApi; "
-            f"HfApi().create_tag({repo_id!r}, tag={version_tag!r}, repo_type='dataset', exist_ok=True)"
+            f"HfApi().create_tag({repo_id!r}, tag={version_tag!r}, repo_type='dataset', exist_ok=True)",
+            flush=True,
         )
 
 

src/lerobot/scripts/lerobot_dataset_viz.py

@@ -89,8 +89,6 @@ from lerobot.datasets import LeRobotDataset
 from lerobot.utils.constants import ACTION, DONE, OBS_STATE, REWARD, SUCCESS
 from lerobot.utils.utils import init_logging
 
-logger = logging.getLogger(__name__)
-
 DEFAULT_FOXGLOVE_PORT = 8765
 DEFAULT_RERUN_PORT = 9090
 
@@ -301,7 +299,7 @@ def visualize_dataset(
             while True:
                 time.sleep(1)
         except KeyboardInterrupt:
-            logger.info("Ctrl-C received. Exiting.")
+            print("Ctrl-C received. Exiting.")
 
 
 def main():

src/lerobot/scripts/lerobot_eval.py

@@ -62,7 +62,7 @@ from dataclasses import asdict
 from functools import partial
 from pathlib import Path
 from pprint import pformat
-from typing import TYPE_CHECKING, Any, TypedDict
+from typing import Any, TypedDict
 
 import einops
 import gymnasium as gym
@@ -87,7 +87,7 @@ from lerobot.processor import PolicyProcessorPipeline
 from lerobot.types import PolicyAction
 from lerobot.utils.constants import ACTION, DONE, OBS_IMAGE, OBS_IMAGES, OBS_STR, REWARD
 from lerobot.utils.device_utils import get_safe_torch_device
-from lerobot.utils.import_utils import _peft_available, register_third_party_plugins, require_package
+from lerobot.utils.import_utils import register_third_party_plugins
 from lerobot.utils.io_utils import write_video
 from lerobot.utils.random_utils import set_seed
 from lerobot.utils.utils import (
@@ -95,14 +95,6 @@ from lerobot.utils.utils import (
     inside_slurm,
 )
 
-if TYPE_CHECKING or _peft_available:
-    from peft import PeftModel
-else:
-    PeftModel = None
-
-
-logger = logging.getLogger(__name__)
-
 
 def _env_features_to_dataset_features(env_features: dict) -> dict:
     """Convert EnvConfig.features to the dict format expected by LeRobotDataset.create()."""
@@ -452,11 +444,13 @@ def eval_policy(
         exc = ValueError(
             f"Policy of type 'PreTrainedPolicy' is expected, but type '{type(policy)}' was provided."
         )
-        if not _peft_available:
-            raise exc
-        require_package("peft", extra="peft")
-        if not isinstance(policy, PeftModel):
-            raise exc
+        try:
+            from peft import PeftModel
+
+            if not isinstance(policy, PeftModel):
+                raise exc
+        except ImportError:
+            raise exc from None
 
     start = time.time()
     # Preserve the mode for direct callers. eval_policy_all scopes the mode
@@ -564,7 +558,7 @@ def eval_policy(
         if seeds:
             all_seeds.extend(seeds)
         else:
-            all_seeds.extend([None] * env.num_envs)
+            all_seeds.append(None)
 
         # FIXME: episode_data is either None or it doesn't exist
         if return_episode_data:
@@ -802,13 +796,13 @@ def eval_main(cfg: EvalPipelineConfig):
             recording_repo_id=cfg.eval.recording_repo_id,
             recording_private=cfg.eval.recording_private,
         )
-        logger.info("Overall Aggregated Metrics:")
-        logger.info(info["overall"])
+        print("Overall Aggregated Metrics:")
+        print(info["overall"])
 
         # Print per-suite stats
         for task_group, task_group_info in info.items():
-            logger.info(f"\nAggregated Metrics for {task_group}:")
-            logger.info(task_group_info)
+            print(f"\nAggregated Metrics for {task_group}:")
+            print(task_group_info)
     # Close all vec envs
     close_envs(envs)
 

src/lerobot/scripts/lerobot_find_cameras.py

@@ -40,7 +40,6 @@ from PIL import Image
 from lerobot.cameras import ColorMode
 from lerobot.cameras.opencv import OpenCVCamera, OpenCVCameraConfig
 from lerobot.cameras.realsense import RealSenseCamera, RealSenseCameraConfig
-from lerobot.utils.utils import init_logging
 
 logger = logging.getLogger(__name__)
 
@@ -286,8 +285,6 @@ def save_images_from_all_cameras(
 
 
 def main():
-    init_logging()
-
     parser = argparse.ArgumentParser(
         description="Unified camera utility script for listing cameras and capturing images."
     )

src/lerobot/scripts/lerobot_rollout.py

@@ -165,7 +165,6 @@ from lerobot.robots import (  # noqa: F401
     earthrover_mini_plus,
     hope_jr,
     koch_follower,
-    lekiwi,
     omx_follower,
     openarm_follower,
     reachy2,

src/lerobot/scripts/lerobot_train.py

@@ -22,8 +22,7 @@ import dataclasses
 import logging
 import sys
 import time
-from collections.abc import Iterator
-from contextlib import contextmanager, nullcontext
+from contextlib import nullcontext
 from pprint import pformat
 from typing import TYPE_CHECKING, Any
 
@@ -58,7 +57,7 @@ from lerobot.optim.factory import make_optimizer_and_scheduler
 from lerobot.policies import PreTrainedPolicy, make_policy, make_pre_post_processors
 from lerobot.rewards import make_reward_pre_post_processors
 from lerobot.utils.collate import lerobot_collate_fn
-from lerobot.utils.import_utils import _peft_available, register_third_party_plugins, require_package
+from lerobot.utils.import_utils import register_third_party_plugins
 from lerobot.utils.logging_utils import AverageMeter, MetricsTracker
 from lerobot.utils.random_utils import set_seed
 from lerobot.utils.utils import (
@@ -69,28 +68,9 @@ from lerobot.utils.utils import (
     inside_slurm,
 )
 
-if TYPE_CHECKING or _peft_available:
-    from peft import PeftModel
-else:
-    PeftModel = None
-
 from .lerobot_eval import eval_policy_all
 
 
-@contextmanager
-def _make_eval_envs(cfg: TrainPipelineConfig) -> Iterator[dict[str, dict[int, Any]]]:
-    """Create evaluation environments for one run and always dispose of them."""
-    envs = make_env(
-        cfg.env,
-        n_envs=cfg.eval.batch_size,
-        use_async_envs=cfg.eval.use_async_envs,
-    )
-    try:
-        yield envs
-    finally:
-        close_envs(envs)
-
-
 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
@@ -227,6 +207,8 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
     if cfg.job.is_remote:
         return submit_to_hf(cfg)
 
+    from lerobot.utils.import_utils import require_package
+
     require_package("accelerate", extra="training")
     from accelerate import Accelerator
     from accelerate.utils import DistributedDataParallelKwargs, DistributedType
@@ -295,6 +277,14 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
     if not is_main_process:
         dataset, eval_dataset = make_train_eval_datasets(cfg)
 
+    # Create environment used for evaluating checkpoints during training on simulation data.
+    # On real-world data, no need to create an environment as evaluations are done outside train.py,
+    # using the eval.py instead, with gym_dora environment and dora-rs.
+    eval_env = None
+    if cfg.env_eval_freq > 0 and cfg.env is not None and is_main_process:
+        logging.info("Creating env")
+        eval_env = make_env(cfg.env, n_envs=cfg.eval.batch_size, use_async_envs=cfg.eval.use_async_envs)
+
     if cfg.is_reward_model_training:
         if is_main_process:
             logging.info("Creating reward model")
@@ -322,7 +312,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
     if cfg.peft is not None:
         if cfg.is_reward_model_training:
             raise ValueError("PEFT is only supported for policy training. ")
-        require_package("peft", extra="peft")
+        from peft import PeftModel
 
         if isinstance(policy, PeftModel):
             logging.info("PEFT adapter already loaded from checkpoint, skipping wrap_with_peft.")
@@ -702,7 +692,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
             if is_main_process:
                 step_id = get_step_identifier(step, cfg.steps)
                 logging.info(f"Eval policy at step {step}")
-                with _make_eval_envs(cfg) as eval_env, torch.no_grad(), accelerator.autocast():
+                with torch.no_grad(), accelerator.autocast():
                     eval_info = eval_policy_all(
                         envs=eval_env,  # dict[suite][task_id] -> vec_env
                         policy=accelerator.unwrap_model(policy),
@@ -750,6 +740,9 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
     if is_main_process:
         progbar.close()
 
+    if eval_env:
+        close_envs(eval_env)
+
     is_fsdp = accelerator.distributed_type == DistributedType.FSDP
     model_state_dict = accelerator.get_state_dict(policy) if is_fsdp else None
     if is_main_process:

src/lerobot/scripts/lerobot_train_tokenizer.py

@@ -45,7 +45,6 @@ lerobot-train-tokenizer \
 """
 
 import json
-import logging
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -64,9 +63,6 @@ else:
 from lerobot.configs import NormalizationMode, parser
 from lerobot.datasets import LeRobotDataset
 from lerobot.utils.constants import ACTION, OBS_STATE
-from lerobot.utils.utils import init_logging
-
-logger = logging.getLogger(__name__)
 
 
 @dataclass
@@ -278,8 +274,11 @@ def process_episode(args):
 
         return action_chunks
 
-    except Exception:
-        logger.exception("Error processing episode %s", ep_idx)
+    except Exception as e:
+        print(f"Error processing episode {ep_idx}: {e}")
+        import traceback
+
+        traceback.print_exc()
         return None
 
 
@@ -301,10 +300,10 @@ def train_fast_tokenizer(
     Returns:
         Trained FAST tokenizer
     """
-    logger.info(f"Training FAST tokenizer on {len(action_chunks)} action chunks...")
-    logger.info(f"Action chunk shape: {action_chunks.shape}")
-    logger.info(f"Vocab size: {vocab_size}")
-    logger.info(f"DCT scale: {scale}")
+    print(f"Training FAST tokenizer on {len(action_chunks)} action chunks...")
+    print(f"Action chunk shape: {action_chunks.shape}")
+    print(f"Vocab size: {vocab_size}")
+    print(f"DCT scale: {scale}")
 
     # download the tokenizer source code (not pretrained weights)
     # we'll train a new tokenizer on our own data
@@ -315,7 +314,7 @@ def train_fast_tokenizer(
 
     # train the new tokenizer on our action data using .fit()
     # this trains the BPE tokenizer on DCT coefficients
-    logger.info("Training new tokenizer (this may take a few minutes)...")
+    print("Training new tokenizer (this may take a few minutes)...")
     tokenizer = base_tokenizer.fit(
         action_data_list,
         scale=scale,
@@ -323,21 +322,21 @@ def train_fast_tokenizer(
         time_horizon=action_chunks.shape[1],  # action_horizon
         action_dim=action_chunks.shape[2],  # encoded dimensions
     )
-    logger.info("✓ Tokenizer training complete!")
+    print("✓ Tokenizer training complete!")
 
     # validate it works
     sample_chunk = action_chunks[0]
     encoded = tokenizer(sample_chunk[None])[0]
     if isinstance(encoded, list):
         encoded = np.array(encoded)
-    logger.info(f"Sample encoding: {len(encoded)} tokens for chunk shape {sample_chunk.shape}")
+    print(f"Sample encoding: {len(encoded)} tokens for chunk shape {sample_chunk.shape}")
 
     return tokenizer
 
 
 def compute_compression_stats(tokenizer, action_chunks: np.ndarray):
     """Compute compression statistics."""
-    logger.info("\nComputing compression statistics...")
+    print("\nComputing compression statistics...")
 
     # sample for stats (use max 1000 chunks for speed)
     sample_size = min(1000, len(action_chunks))
@@ -367,12 +366,12 @@ def compute_compression_stats(tokenizer, action_chunks: np.ndarray):
         "max_token_length": float(np.max(token_lengths)),
     }
 
-    logger.info("Compression Statistics:")
-    logger.info(f"  Average compression ratio: {stats['compression_ratio']:.2f}x")
-    logger.info(f"  Mean token length: {stats['mean_token_length']:.1f}")
-    logger.info(f"  P99 token length: {stats['p99_token_length']:.0f}")
-    logger.info(f"  Min token length: {stats['min_token_length']:.0f}")
-    logger.info(f"  Max token length: {stats['max_token_length']:.0f}")
+    print("Compression Statistics:")
+    print(f"  Average compression ratio: {stats['compression_ratio']:.2f}x")
+    print(f"  Mean token length: {stats['mean_token_length']:.1f}")
+    print(f"  P99 token length: {stats['p99_token_length']:.0f}")
+    print(f"  Min token length: {stats['min_token_length']:.0f}")
+    print(f"  Max token length: {stats['max_token_length']:.0f}")
 
     return stats
 
@@ -386,9 +385,9 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
         cfg: TokenizerTrainingConfig dataclass with all configuration parameters
     """
     # load dataset
-    logger.info(f"Loading dataset: {cfg.repo_id}")
+    print(f"Loading dataset: {cfg.repo_id}")
     dataset = LeRobotDataset(repo_id=cfg.repo_id, root=cfg.root)
-    logger.info(f"Dataset loaded: {dataset.num_episodes} episodes, {dataset.num_frames} frames")
+    print(f"Dataset loaded: {dataset.num_episodes} episodes, {dataset.num_frames} frames")
 
     # parse normalization mode
     try:
@@ -398,7 +397,7 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
             f"Invalid normalization_mode: {cfg.normalization_mode}. "
             f"Must be one of: {', '.join([m.value for m in NormalizationMode])}"
         ) from err
-    logger.info(f"Normalization mode: {norm_mode.value}")
+    print(f"Normalization mode: {norm_mode.value}")
 
     # parse encoded dimensions
     encoded_dim_ranges = []
@@ -407,38 +406,38 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
         encoded_dim_ranges.append((start, end))
 
     total_encoded_dims = sum(end - start for start, end in encoded_dim_ranges)
-    logger.info(f"Encoding {total_encoded_dims} dimensions: {cfg.encoded_dims}")
+    print(f"Encoding {total_encoded_dims} dimensions: {cfg.encoded_dims}")
 
     # parse relative dimensions
     relative_dim_list = None
     if cfg.relative_dims is not None and cfg.relative_dims.strip():
         relative_dim_list = [int(d.strip()) for d in cfg.relative_dims.split(",")]
-        logger.info(f"Relative dimensions: {relative_dim_list}")
+        print(f"Relative dimensions: {relative_dim_list}")
     else:
-        logger.info("No relative dimensions specified")
+        print("No relative dimensions specified")
 
-    logger.info(f"Use relative transform: {cfg.use_relative_transform}")
+    print(f"Use relative transform: {cfg.use_relative_transform}")
     if cfg.use_relative_transform and (relative_dim_list is None or len(relative_dim_list) == 0):
-        logger.warning(
+        print(
             "Warning: use_relative_transform=True but no relative_dims specified. "
             "No relative transform will be applied."
         )
 
-    logger.info(f"Action horizon: {cfg.action_horizon}")
-    logger.info(f"State key: {cfg.state_key}")
+    print(f"Action horizon: {cfg.action_horizon}")
+    print(f"State key: {cfg.state_key}")
 
     # determine episodes to process
     num_episodes = dataset.num_episodes
     if cfg.max_episodes is not None:
         num_episodes = min(cfg.max_episodes, num_episodes)
 
-    logger.info(f"Processing {num_episodes} episodes...")
+    print(f"Processing {num_episodes} episodes...")
 
     # process episodes sequentially (to avoid pickling issues with dataset)
     all_chunks = []
     for ep_idx in range(num_episodes):
         if ep_idx % 10 == 0:
-            logger.info(f"  Processing episode {ep_idx}/{num_episodes}...")
+            print(f"  Processing episode {ep_idx}/{num_episodes}...")
 
         chunks = process_episode(
             (
@@ -456,19 +455,19 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
 
     # concatenate all chunks
     all_chunks = np.concatenate(all_chunks, axis=0)
-    logger.info(f"Collected {len(all_chunks)} action chunks")
+    print(f"Collected {len(all_chunks)} action chunks")
 
     # extract only encoded dimensions FIRST (before normalization)
     encoded_chunks = []
     for start, end in encoded_dim_ranges:
         encoded_chunks.append(all_chunks[:, :, start:end])
     encoded_chunks = np.concatenate(encoded_chunks, axis=-1)  # [N, H, D_encoded]
-    logger.info(f"Extracted {encoded_chunks.shape[-1]} encoded dimensions")
+    print(f"Extracted {encoded_chunks.shape[-1]} encoded dimensions")
 
     # apply normalization to encoded dimensions
-    logger.info("\nBefore normalization - overall stats:")
-    logger.info(f"  Min: {np.min(encoded_chunks):.4f}, Max: {np.max(encoded_chunks):.4f}")
-    logger.info(f"  Mean: {np.mean(encoded_chunks):.4f}, Std: {np.std(encoded_chunks):.4f}")
+    print("\nBefore normalization - overall stats:")
+    print(f"  Min: {np.min(encoded_chunks):.4f}, Max: {np.max(encoded_chunks):.4f}")
+    print(f"  Mean: {np.mean(encoded_chunks):.4f}, Std: {np.std(encoded_chunks):.4f}")
 
     # get normalization stats from dataset
     norm_stats = dataset.meta.stats
@@ -490,9 +489,9 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
                     encoded_stats[stat_name] = stat_array[encoded_dim_indices]
 
         if encoded_stats:
-            logger.info(f"\nNormalization stats for encoded dimensions (mode: {norm_mode.value}):")
+            print(f"\nNormalization stats for encoded dimensions (mode: {norm_mode.value}):")
             for stat_name, stat_values in encoded_stats.items():
-                logger.info(
+                print(
                     f"  {stat_name}: shape={stat_values.shape}, "
                     f"range=[{np.min(stat_values):.4f}, {np.max(stat_values):.4f}]"
                 )
@@ -500,27 +499,27 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
             # apply normalization based on mode
             try:
                 encoded_chunks = apply_normalization(encoded_chunks, encoded_stats, norm_mode, eps=1e-8)
-                logger.info(f"\nApplied {norm_mode.value} normalization")
+                print(f"\nApplied {norm_mode.value} normalization")
             except ValueError as e:
-                logger.warning(f"Warning: {e}. Using raw actions without normalization.")
+                print(f"Warning: {e}. Using raw actions without normalization.")
 
-            logger.info("\nAfter normalization - overall stats:")
-            logger.info(f"  Min: {np.min(encoded_chunks):.4f}, Max: {np.max(encoded_chunks):.4f}")
-            logger.info(f"  Mean: {np.mean(encoded_chunks):.4f}, Std: {np.std(encoded_chunks):.4f}")
+            print("\nAfter normalization - overall stats:")
+            print(f"  Min: {np.min(encoded_chunks):.4f}, Max: {np.max(encoded_chunks):.4f}")
+            print(f"  Mean: {np.mean(encoded_chunks):.4f}, Std: {np.std(encoded_chunks):.4f}")
 
-            logger.info("\nPer-dimension stats (after normalization):")
+            print("\nPer-dimension stats (after normalization):")
             for d in range(encoded_chunks.shape[-1]):
                 dim_data = encoded_chunks[:, :, d]
-                logger.info(
+                print(
                     f"  Dim {d}: min={np.min(dim_data):7.4f}, max={np.max(dim_data):7.4f}, "
                     f"mean={np.mean(dim_data):7.4f}, std={np.std(dim_data):7.4f}"
                 )
         else:
-            logger.warning("Warning: Could not extract stats for encoded dimensions, using raw actions")
+            print("Warning: Could not extract stats for encoded dimensions, using raw actions")
     else:
-        logger.warning("Warning: No normalization stats found in dataset, using raw actions")
+        print("Warning: No normalization stats found in dataset, using raw actions")
 
-    logger.info(f"Encoded chunks shape: {encoded_chunks.shape}")
+    print(f"Encoded chunks shape: {encoded_chunks.shape}")
 
     # train FAST tokenizer
     tokenizer = train_fast_tokenizer(
@@ -562,8 +561,8 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
     with open(output_path / "metadata.json", "w") as f:
         json.dump(metadata, f, indent=2)
 
-    logger.info(f"\nSaved FAST tokenizer to {output_path}")
-    logger.info(f"Metadata: {json.dumps(metadata, indent=2)}")
+    print(f"\nSaved FAST tokenizer to {output_path}")
+    print(f"Metadata: {json.dumps(metadata, indent=2)}")
 
     # push to Hugging Face Hub if requested
     if cfg.push_to_hub:
@@ -571,10 +570,10 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
         hub_repo_id = cfg.hub_repo_id
         if hub_repo_id is None:
             hub_repo_id = output_path.name
-            logger.info(f"\nNo hub_repo_id provided, using: {hub_repo_id}")
+            print(f"\nNo hub_repo_id provided, using: {hub_repo_id}")
 
-        logger.info(f"\nPushing tokenizer to Hugging Face Hub: {hub_repo_id}")
-        logger.info(f"   Private: {cfg.hub_private}")
+        print(f"\nPushing tokenizer to Hugging Face Hub: {hub_repo_id}")
+        print(f"   Private: {cfg.hub_private}")
 
         try:
             # use the tokenizer's push_to_hub method
@@ -594,15 +593,14 @@ def train_tokenizer(cfg: TokenizerTrainingConfig):
                 commit_message="Upload tokenizer metadata",
             )
 
-            logger.info(f"Successfully pushed tokenizer to: https://huggingface.co/{hub_repo_id}")
+            print(f"Successfully pushed tokenizer to: https://huggingface.co/{hub_repo_id}")
         except Exception as e:
-            logger.error(f"Error pushing to hub: {e}")
-            logger.error("   Make sure you're logged in with `huggingface-cli login`")
+            print(f"Error pushing to hub: {e}")
+            print("   Make sure you're logged in with `huggingface-cli login`")
 
 
 def main():
     """CLI entry point that parses arguments and runs the tokenizer training."""
-    init_logging()
     train_tokenizer()
 
 

src/lerobot/teleoperators/so_leader/config_so_leader.py

@@ -29,12 +29,6 @@ class SOLeaderConfig:
     # Whether to use degrees for angles
     use_degrees: bool = True
 
-    # Number of extra attempts when a `sync_read` of the motors fails. Feetech buses can occasionally
-    # return a corrupted status packet ("Incorrect status packet!"), especially when several joints move
-    # at once, which otherwise aborts the teleoperation loop. Retries are immediate (no sleep) and only
-    # happen on failure, so the steady-state read cost is unchanged.
-    num_read_retries: int = 2
-
 
 @TeleoperatorConfig.register_subclass("so101_leader")
 @TeleoperatorConfig.register_subclass("so100_leader")

src/lerobot/teleoperators/so_leader/so_leader.py

@@ -110,25 +110,11 @@ class SOLeader(Teleoperator):
         range_mins[full_turn_motor] = 0
         range_maxes[full_turn_motor] = 4095
 
-        drive_modes = dict.fromkeys(self.bus.motors, 0)
-        input(f"Fully close the gripper of {self} and press ENTER....")
-        gripper_closed_pos = self.bus.read(
-            "Present_Position", "gripper", normalize=False, num_retry=self.config.num_read_retries
-        )
-        distance_to_min = abs(gripper_closed_pos - range_mins["gripper"])
-        distance_to_max = abs(gripper_closed_pos - range_maxes["gripper"])
-        if min(distance_to_min, distance_to_max) > (range_maxes["gripper"] - range_mins["gripper"]) * 0.2:
-            raise ValueError("Gripper is not fully closed. Run calibration again.")
-
-        drive_modes["gripper"] = int(distance_to_max < distance_to_min)
-        if drive_modes["gripper"]:
-            logger.info("Gripper motor is inverted, setting drive_mode=1 to compensate.")
-
         self.calibration = {}
         for motor, m in self.bus.motors.items():
             self.calibration[motor] = MotorCalibration(
                 id=m.id,
-                drive_mode=drive_modes[motor],
+                drive_mode=0,
                 homing_offset=homing_offsets[motor],
                 range_min=range_mins[motor],
                 range_max=range_maxes[motor],
@@ -159,7 +145,7 @@ class SOLeader(Teleoperator):
     @check_if_not_connected
     def get_action(self) -> dict[str, float]:
         start = time.perf_counter()
-        action = self.bus.sync_read("Present_Position", num_retry=self.config.num_read_retries)
+        action = self.bus.sync_read("Present_Position")
         action = {f"{motor}.pos": val for motor, val in action.items()}
         dt_ms = (time.perf_counter() - start) * 1e3
         logger.debug(f"{self} read action: {dt_ms:.1f}ms")

src/lerobot/transforms/transforms.py

@@ -41,7 +41,7 @@ class RandomSubsetApply(Transform):
 
     def __init__(
         self,
-        transforms: Sequence[Callable[..., Any]],
+        transforms: Sequence[Callable],
         p: list[float] | None = None,
         n_subset: int | None = None,
         random_order: bool = False,
@@ -50,7 +50,7 @@ class RandomSubsetApply(Transform):
         if not isinstance(transforms, Sequence):
             raise TypeError("Argument transforms should be a sequence of callables")
         if p is None:
-            p = [1.0] * len(transforms)
+            p = [1] * len(transforms)
         elif len(p) != len(transforms):
             raise ValueError(
                 f"Length of p doesn't match the number of transforms: {len(p)} != {len(transforms)}"
@@ -69,7 +69,7 @@ class RandomSubsetApply(Transform):
         self.n_subset = n_subset
         self.random_order = random_order
 
-        self.selected_transforms: list[Callable[..., Any]] = []
+        self.selected_transforms = None
 
     def forward(self, *inputs: Any) -> Any:
         needs_unpacking = len(inputs) > 1
@@ -119,7 +119,7 @@ class SharpnessJitter(Transform):
         super().__init__()
         self.sharpness = self._check_input(sharpness)
 
-    def _check_input(self, sharpness: float | Sequence[float]) -> tuple[float, float]:
+    def _check_input(self, sharpness):
         if isinstance(sharpness, (int | float)):
             if sharpness < 0:
                 raise ValueError("If sharpness is a single number, it must be non negative.")
@@ -215,7 +215,7 @@ class ImageTransformsConfig:
     )
 
 
-def make_transform_from_config(cfg: ImageTransformConfig) -> Transform:
+def make_transform_from_config(cfg: ImageTransformConfig):
     if cfg.type == "SharpnessJitter":
         return SharpnessJitter(**cfg.kwargs)
 
@@ -236,8 +236,8 @@ class ImageTransforms(Transform):
         super().__init__()
         self._cfg = cfg
 
-        self.weights: list[float] = []
-        self.transforms: dict[str, Transform] = {}
+        self.weights = []
+        self.transforms = {}
         for tf_name, tf_cfg in cfg.tfs.items():
             if tf_cfg.weight <= 0.0:
                 continue

src/lerobot/utils/collate.py

@@ -22,7 +22,7 @@ from torch.utils.data._utils.collate import default_collate
 
 from lerobot.datasets.language import LANGUAGE_COLUMNS
 
-_PYTHON_LIST_KEYS = {"messages", "message_streams", "target_message_indices"}
+_PYTHON_LIST_KEYS = {"messages", "message_streams", "target_message_indices", *LANGUAGE_COLUMNS}
 
 
 def lerobot_collate_fn(batch: list[dict[str, Any] | None]) -> dict[str, Any] | None:

src/lerobot/utils/constants.py

@@ -26,6 +26,7 @@ OBS_IMAGES = OBS_IMAGE + "s"
 OBS_LANGUAGE = OBS_STR + ".language"
 OBS_LANGUAGE_TOKENS = OBS_LANGUAGE + ".tokens"
 OBS_LANGUAGE_ATTENTION_MASK = OBS_LANGUAGE + ".attention_mask"
+OBS_LANGUAGE_CAUSAL_MARKS = OBS_LANGUAGE + ".causal_marks"
 OBS_LANGUAGE_SUBTASK = OBS_STR + ".subtask"
 OBS_LANGUAGE_SUBTASK_TOKENS = OBS_LANGUAGE_SUBTASK + ".tokens"
 OBS_LANGUAGE_SUBTASK_ATTENTION_MASK = OBS_LANGUAGE_SUBTASK + ".attention_mask"
@@ -34,6 +35,7 @@ ACTION = "action"
 ACTION_PREFIX = ACTION + "."
 ACTION_TOKENS = ACTION + ".tokens"
 ACTION_TOKEN_MASK = ACTION + ".token_mask"
+ACTION_CODE_TOKEN_MASK = ACTION + ".code_token_mask"
 REWARD = "next.reward"
 TRUNCATED = "next.truncated"
 DONE = "next.done"

src/lerobot/utils/utils.py

@@ -133,13 +133,10 @@ def say(text: str, blocking: bool = False):
     else:
         raise RuntimeError("Unsupported operating system for text-to-speech.")
 
-    try:
-        if blocking:
-            subprocess.run(cmd, check=True, timeout=5)
-        else:
-            subprocess.Popen(cmd, creationflags=subprocess.CREATE_NO_WINDOW if system == "Windows" else 0)
-    except (FileNotFoundError, subprocess.TimeoutExpired) as e:
-        logging.warning("Text-to-speech command failed: %s | Error: %s", cmd, e)
+    if blocking:
+        subprocess.run(cmd, check=True)
+    else:
+        subprocess.Popen(cmd, creationflags=subprocess.CREATE_NO_WINDOW if system == "Windows" else 0)
 
 
 def log_say(text: str, play_sounds: bool = True, blocking: bool = False):

tests/cameras/test_opencv.py

@@ -20,7 +20,7 @@
 # ```
 
 from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import cv2
 import numpy as np
@@ -123,73 +123,6 @@ def test_invalid_width_connect():
         camera.connect(warmup=False)
 
 
-def test_connect_cleans_up_after_settings_failure_and_allows_retry():
-    config = OpenCVCameraConfig(index_or_path=DEFAULT_PNG_FILE_PATH, warmup_s=0)
-    camera = OpenCVCamera(config)
-    opened_captures = []
-
-    def fail_settings():
-        opened_captures.append(camera.videocapture)
-        raise RuntimeError("settings failed")
-
-    with (
-        patch.object(camera, "_configure_capture_settings", side_effect=fail_settings),
-        pytest.raises(RuntimeError, match="settings failed"),
-    ):
-        camera.connect(warmup=False)
-
-    assert camera.videocapture is None
-    assert camera.thread is None
-    assert not camera.is_connected
-    assert opened_captures[0] is not None
-    assert not opened_captures[0].isOpened()
-
-    camera.connect(warmup=False)
-    assert camera.is_connected
-    camera.disconnect()
-
-
-def test_connect_cleans_up_after_warmup_failure_and_allows_retry():
-    config = OpenCVCameraConfig(index_or_path=DEFAULT_PNG_FILE_PATH, warmup_s=1)
-    camera = OpenCVCamera(config)
-    read_threads = []
-
-    def fail_warmup(*_args, **_kwargs):
-        read_threads.append(camera.thread)
-        raise TimeoutError("no frame")
-
-    with (
-        patch.object(camera, "async_read", side_effect=fail_warmup),
-        pytest.raises(TimeoutError, match="no frame"),
-    ):
-        camera.connect()
-
-    assert camera.videocapture is None
-    assert camera.thread is None
-    assert not camera.is_connected
-    assert read_threads[0] is not None
-    assert not read_threads[0].is_alive()
-
-    camera.connect(warmup=False)
-    assert camera.is_connected
-    camera.disconnect()
-
-
-def test_find_cameras_releases_unopened_handles():
-    module_path = OpenCVCamera.__module__
-    unopened_capture = MagicMock()
-    unopened_capture.isOpened.return_value = False
-
-    with (
-        patch(f"{module_path}.platform.system", return_value="Darwin"),
-        patch(f"{module_path}.MAX_OPENCV_INDEX", 1),
-        patch(f"{module_path}.cv2.VideoCapture", return_value=unopened_capture),
-    ):
-        assert OpenCVCamera.find_cameras() == []
-
-    unopened_capture.release.assert_called_once_with()
-
-
 @pytest.mark.parametrize("index_or_path", TEST_IMAGE_PATHS, ids=TEST_IMAGE_SIZES)
 def test_read(index_or_path):
     config = OpenCVCameraConfig(index_or_path=index_or_path, warmup_s=0)

tests/cameras/test_realsense.py

@@ -20,7 +20,7 @@
 # ```
 
 from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import numpy as np
 import pytest
@@ -30,8 +30,6 @@ from lerobot.utils.errors import DeviceAlreadyConnectedError, DeviceNotConnected
 
 pytest.importorskip("pyrealsense2")
 
-import pyrealsense2 as rs
-
 from lerobot.cameras.realsense import RealSenseCamera, RealSenseCameraConfig
 
 TEST_ARTIFACTS_DIR = Path(__file__).parent.parent / "artifacts" / "cameras"
@@ -63,17 +61,6 @@ def test_abc_implementation():
     _ = RealSenseCamera(config)
 
 
-@pytest.mark.parametrize("option", ["exposure", "gain", "white_balance"])
-def test_manual_color_option_requires_rgb(option):
-    with pytest.raises(ValueError, match="use_rgb=True"):
-        RealSenseCameraConfig(
-            serial_number_or_name="042",
-            use_rgb=False,
-            use_depth=True,
-            **{option: 100},
-        )
-
-
 def test_connect():
     config = RealSenseCameraConfig(serial_number_or_name="042", warmup_s=0)
 
@@ -96,27 +83,6 @@ def test_connect_invalid_camera_path(patch_realsense):
         camera.connect(warmup=False)
 
 
-def test_connect_cleans_up_when_sensor_configuration_fails():
-    config = RealSenseCameraConfig(serial_number_or_name="042", exposure=120)
-    camera = RealSenseCamera(config)
-    pipeline = MagicMock()
-    pipeline.start.return_value = MagicMock()
-
-    with (
-        patch("lerobot.cameras.realsense.camera_realsense.rs.pipeline", return_value=pipeline),
-        patch.object(camera, "_configure_rs_pipeline_config"),
-        patch.object(camera, "_configure_capture_settings"),
-        patch.object(camera, "_configure_sensor_options", side_effect=ValueError("invalid exposure")),
-        pytest.raises(ValueError, match="invalid exposure"),
-    ):
-        camera.connect(warmup=False)
-
-    pipeline.stop.assert_called_once_with()
-    assert camera.rs_pipeline is None
-    assert camera.rs_profile is None
-    assert not camera.is_connected
-
-
 def test_invalid_width_connect():
     config = RealSenseCameraConfig(serial_number_or_name="042", width=99999, height=480, fps=30)
     camera = RealSenseCamera(config)
@@ -125,33 +91,6 @@ def test_invalid_width_connect():
         camera.connect(warmup=False)
 
 
-def test_connect_cleans_up_after_warmup_failure_and_allows_retry():
-    config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30)
-    camera = RealSenseCamera(config)
-    read_threads = []
-
-    def fail_warmup(*_args, **_kwargs):
-        read_threads.append(camera.thread)
-        raise TimeoutError("no frame")
-
-    with (
-        patch.object(camera, "async_read", side_effect=fail_warmup),
-        pytest.raises(TimeoutError, match="no frame"),
-    ):
-        camera.connect()
-
-    assert camera.rs_pipeline is None
-    assert camera.rs_profile is None
-    assert camera.thread is None
-    assert not camera.is_connected
-    assert read_threads[0] is not None
-    assert not read_threads[0].is_alive()
-
-    camera.connect(warmup=False)
-    assert camera.is_connected
-    camera.disconnect()
-
-
 def test_read():
     config = RealSenseCameraConfig(serial_number_or_name="042", width=640, height=480, fps=30, warmup_s=0)
     with RealSenseCamera(config) as camera:
@@ -289,203 +228,6 @@ def test_read_latest_too_old():
             _ = camera.read_latest(max_age_ms=0)  # immediately too old
 
 
-def _make_mock_sensor(name: str, supported_options: set | None = None) -> MagicMock:
-    """Build a fake rs.sensor that reports a name and a configurable supported-options set."""
-    supported = supported_options if supported_options is not None else set()
-    sensor = MagicMock()
-    sensor.get_info.return_value = name
-    sensor.supports.side_effect = lambda opt: opt in supported
-    return sensor
-
-
-def _attach_mock_color_sensor(camera: RealSenseCamera, sensor: MagicMock) -> None:
-    """Wire camera.rs_profile so _get_color_sensor finds the given sensor."""
-    profile = MagicMock()
-    device = MagicMock()
-    device.query_sensors.return_value = [sensor]
-    profile.get_device.return_value = device
-    camera.rs_profile = profile
-
-
-def test_get_color_sensor_prefers_rgb_camera():
-    config = RealSenseCameraConfig(serial_number_or_name="042")
-    camera = RealSenseCamera(config)
-
-    rgb = _make_mock_sensor("RGB Camera")
-    stereo = _make_mock_sensor("Stereo Module")
-    profile = MagicMock()
-    device = MagicMock()
-    device.query_sensors.return_value = [stereo, rgb]
-    profile.get_device.return_value = device
-    camera.rs_profile = profile
-
-    assert camera._get_color_sensor() is rgb
-
-
-def test_get_color_sensor_falls_back_to_stereo_module():
-    """D405 has no separate RGB module; color comes from Stereo Module."""
-    config = RealSenseCameraConfig(serial_number_or_name="042")
-    camera = RealSenseCamera(config)
-
-    stereo = _make_mock_sensor("Stereo Module")
-    _attach_mock_color_sensor(camera, stereo)
-
-    assert camera._get_color_sensor() is stereo
-
-
-def test_get_color_sensor_raises_with_available_sensors():
-    config = RealSenseCameraConfig(serial_number_or_name="042")
-    camera = RealSenseCamera(config)
-
-    other = _make_mock_sensor("Motion Module")
-    _attach_mock_color_sensor(camera, other)
-
-    with pytest.raises(RuntimeError, match="Motion Module"):
-        camera._get_color_sensor()
-
-
-def test_configure_sensor_options_skipped_when_none():
-    config = RealSenseCameraConfig(serial_number_or_name="042")
-    camera = RealSenseCamera(config)
-
-    with patch.object(RealSenseCamera, "_get_color_sensor") as mock_get:
-        camera._configure_sensor_options()
-        mock_get.assert_not_called()
-
-
-def test_configure_sensor_options_applies_all_values():
-    config = RealSenseCameraConfig(serial_number_or_name="042", exposure=120, gain=64, white_balance=4600)
-    camera = RealSenseCamera(config)
-
-    sensor = _make_mock_sensor(
-        "RGB Camera",
-        supported_options={
-            rs.option.enable_auto_exposure,
-            rs.option.exposure,
-            rs.option.gain,
-            rs.option.enable_auto_white_balance,
-            rs.option.white_balance,
-        },
-    )
-    _attach_mock_color_sensor(camera, sensor)
-
-    camera._configure_sensor_options()
-
-    sensor.set_option.assert_any_call(rs.option.enable_auto_exposure, 0)
-    sensor.set_option.assert_any_call(rs.option.exposure, 120)
-    sensor.set_option.assert_any_call(rs.option.gain, 64)
-    sensor.set_option.assert_any_call(rs.option.enable_auto_white_balance, 0)
-    sensor.set_option.assert_any_call(rs.option.white_balance, 4600)
-
-
-@pytest.mark.parametrize(
-    ("config_field", "option", "label"),
-    [
-        ("exposure", rs.option.exposure, "exposure"),
-        ("gain", rs.option.gain, "gain"),
-        ("white_balance", rs.option.white_balance, "white balance"),
-    ],
-)
-def test_configure_sensor_options_raises_when_requested_option_is_unsupported(config_field, option, label):
-    config = RealSenseCameraConfig(serial_number_or_name="042", **{config_field: 100})
-    camera = RealSenseCamera(config)
-
-    sensor = _make_mock_sensor("RGB Camera", supported_options=set())
-    _attach_mock_color_sensor(camera, sensor)
-
-    with pytest.raises(ValueError, match=label):
-        camera._configure_sensor_options()
-
-    sensor.supports.assert_any_call(option)
-    sensor.set_option.assert_not_called()
-
-
-@pytest.mark.parametrize(
-    ("config_field", "option", "value"),
-    [
-        ("exposure", rs.option.exposure, 120),
-        ("gain", rs.option.gain, 64),
-    ],
-)
-def test_configure_sensor_options_exposure_or_gain_disables_auto_exposure(config_field, option, value):
-    """white_balance=None should not touch auto white balance."""
-    config = RealSenseCameraConfig(serial_number_or_name="042", **{config_field: value})
-    camera = RealSenseCamera(config)
-
-    sensor = _make_mock_sensor(
-        "RGB Camera",
-        supported_options={rs.option.enable_auto_exposure, option},
-    )
-    _attach_mock_color_sensor(camera, sensor)
-
-    camera._configure_sensor_options()
-
-    calls = [call.args for call in sensor.set_option.call_args_list]
-    assert (rs.option.enable_auto_exposure, 0) in calls
-    assert (option, value) in calls
-    for opt, _ in calls:
-        assert opt != rs.option.enable_auto_white_balance
-        assert opt != rs.option.white_balance
-
-
-def test_configure_sensor_options_warns_when_auto_exposure_control_is_unsupported(caplog):
-    config = RealSenseCameraConfig(serial_number_or_name="042", exposure=120)
-    camera = RealSenseCamera(config)
-
-    sensor = _make_mock_sensor("RGB Camera", supported_options={rs.option.exposure})
-    _attach_mock_color_sensor(camera, sensor)
-
-    with caplog.at_level("WARNING"):
-        camera._configure_sensor_options()
-
-    sensor.set_option.assert_called_once_with(rs.option.exposure, 120)
-    assert "does not support disabling auto-exposure" in caplog.text
-
-
-def test_configure_sensor_options_warns_when_auto_white_balance_control_is_unsupported(caplog):
-    config = RealSenseCameraConfig(serial_number_or_name="042", white_balance=4600)
-    camera = RealSenseCamera(config)
-
-    sensor = _make_mock_sensor("RGB Camera", supported_options={rs.option.white_balance})
-    _attach_mock_color_sensor(camera, sensor)
-
-    with caplog.at_level("WARNING"):
-        camera._configure_sensor_options()
-
-    sensor.set_option.assert_called_once_with(rs.option.white_balance, 4600)
-    assert "does not support disabling auto white balance" in caplog.text
-
-
-def test_configure_sensor_options_out_of_range_raises_value_error():
-    """set_option errors should be re-raised as ValueError with range diagnostics."""
-    config = RealSenseCameraConfig(serial_number_or_name="042", exposure=999999)
-    camera = RealSenseCamera(config)
-
-    sensor = _make_mock_sensor(
-        "RGB Camera",
-        supported_options={rs.option.enable_auto_exposure, rs.option.exposure},
-    )
-
-    def fake_set_option(option, value):
-        if option == rs.option.exposure:
-            raise RuntimeError("value out of range")
-
-    sensor.set_option.side_effect = fake_set_option
-
-    option_range = MagicMock(min=1, max=10000, step=1, default=156)
-    sensor.get_option_range.return_value = option_range
-
-    _attach_mock_color_sensor(camera, sensor)
-
-    with pytest.raises(ValueError, match="exposure") as exc_info:
-        camera._configure_sensor_options()
-
-    msg = str(exc_info.value)
-    assert "999999" in msg
-    assert "min=1" in msg
-    assert "max=10000" in msg
-
-
 @pytest.mark.parametrize(
     "rotation",
     [

tests/configs/test_recipe.py

@@ -29,6 +29,13 @@ def test_message_recipe_validates_unknown_binding():
         )
 
 
+def test_canonical_recipe_loads():
+    """The canonical PI052 blend YAML loads + validates."""
+    recipe = TrainingRecipe.from_yaml(Path("src/lerobot/configs/recipes/subtask_mem_vqa_speech.yaml"))
+    assert recipe.blend is not None
+    assert sum(c.weight for c in recipe.blend.values()) == pytest.approx(1.0)
+
+
 def test_message_turn_requires_a_stream():
     """Every turn must declare a stream — None is rejected at construction.
 

tests/datasets/test_augment_quantile_stats.py

@@ -1,104 +0,0 @@
-# Copyright 2025 The HuggingFace Inc. team. All rights reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import numpy as np
-import pytest
-
-pytest.importorskip("datasets", reason="datasets is required (install lerobot[dataset])")
-
-from lerobot.scripts.augment_dataset_quantile_stats import (
-    compute_quantile_stats_for_dataset,
-    has_quantile_stats,
-)
-
-
-def _numeric_keys(dataset):
-    return [k for k, v in dataset.features.items() if v["dtype"] not in ("image", "video", "string")]
-
-
-def _image_keys(dataset):
-    return [k for k, v in dataset.features.items() if v["dtype"] in ("image", "video")]
-
-
-def test_numeric_stats_are_unaffected_by_sampling(tmp_path, lerobot_dataset_factory):
-    """Sampling only touches image/video frames; numeric features are read in
-    full either way, so their stats must be identical with and without sampling."""
-    dataset = lerobot_dataset_factory(
-        root=tmp_path / "ds", total_episodes=2, total_frames=400, use_videos=False
-    )
-
-    exact = compute_quantile_stats_for_dataset(dataset, use_sampling=False)
-    sampled = compute_quantile_stats_for_dataset(dataset, use_sampling=True)
-
-    numeric_keys = _numeric_keys(dataset)
-    assert numeric_keys, "fixture should expose numeric features"
-    for key in numeric_keys:
-        if key not in exact:
-            continue
-        for stat in ("mean", "std", "q01", "q50", "q99"):
-            if stat in exact[key]:
-                np.testing.assert_allclose(
-                    sampled[key][stat],
-                    exact[key][stat],
-                    rtol=1e-6,
-                    atol=1e-6,
-                    err_msg=f"numeric feature '{key}' stat '{stat}' changed under sampling",
-                )
-
-
-def test_image_sampling_reduces_data_but_keeps_stats_close(tmp_path, lerobot_dataset_factory):
-    """For images, sampling should reduce the number of samples considered while
-    keeping the resulting statistics close to the exact ones."""
-    dataset = lerobot_dataset_factory(
-        root=tmp_path / "ds", total_episodes=2, total_frames=400, use_videos=False
-    )
-
-    exact = compute_quantile_stats_for_dataset(dataset, use_sampling=False)
-    sampled = compute_quantile_stats_for_dataset(dataset, use_sampling=True)
-
-    image_keys = _image_keys(dataset)
-    assert image_keys, "fixture should expose at least one image feature"
-    for key in image_keys:
-        # sampling actually looked at fewer pixels
-        assert sampled[key]["count"][0] < exact[key]["count"][0]
-        # but per-channel mean stays close
-        np.testing.assert_allclose(
-            sampled[key]["mean"],
-            exact[key]["mean"],
-            rtol=0.15,
-            err_msg=f"image feature '{key}' mean drifted too far under sampling",
-        )
-
-
-def test_short_episodes_use_all_frames(tmp_path, lerobot_dataset_factory):
-    """With episodes shorter than the sampling floor, sampling is a no-op and
-    must produce exactly the same stats as the exact path."""
-    dataset = lerobot_dataset_factory(
-        root=tmp_path / "ds", total_episodes=2, total_frames=40, use_videos=False
-    )
-
-    exact = compute_quantile_stats_for_dataset(dataset, use_sampling=False)
-    sampled = compute_quantile_stats_for_dataset(dataset, use_sampling=True)
-
-    for key in _image_keys(dataset):
-        assert sampled[key]["count"][0] == exact[key]["count"][0]
-
-
-def test_quantile_stats_present_after_compute(tmp_path, lerobot_dataset_factory):
-    """The computed stats should contain quantile keys for the dataset."""
-    dataset = lerobot_dataset_factory(
-        root=tmp_path / "ds", total_episodes=2, total_frames=200, use_videos=False
-    )
-    stats = compute_quantile_stats_for_dataset(dataset, use_sampling=True)
-    assert has_quantile_stats(stats)

tests/datasets/test_language_render.py

@@ -343,6 +343,84 @@ def test_resolve_task_explicit_override_beats_rephrasings():
     assert rendered["messages"][0]["content"] == "explicit override wins"
 
 
+def test_flow_only_low_level_recipe_renders_without_target():
+    """Regression: a flow-only ``low_level`` recipe has no ``target`` turn —
+    its supervision is the action-expert flow loss, not text-CE. It must
+    still render (not ``None``), otherwise every blend draw of it is dropped
+    and the action expert never receives a flow loss."""
+    recipe = TrainingRecipe(
+        messages=[
+            MessageTurn(
+                role="user",
+                content="${subtask}",
+                stream="low_level",
+                if_present="subtask",
+            ),
+        ],
+        bindings={"subtask": "active_at(t, style=subtask)"},
+    )
+
+    rendered = render_sample(
+        recipe=recipe,
+        persistent=PERSISTENT,
+        events=[],
+        t=0.5,
+        sample_idx=0,
+        task="clean kitchen",
+    )
+
+    assert rendered is not None
+    assert rendered["messages"] == [{"role": "user", "content": "subtask 0"}]
+    assert rendered["message_streams"] == ["low_level"]
+    assert rendered["target_message_indices"] == []
+
+
+def test_vqa_frame_is_consumed_over_the_weighted_blend():
+    """A frame carrying a VQA annotation renders the ``ask_vqa*`` sub-recipe
+    even when its blend weight is tiny — VQA annotations are sparse and must
+    never be wasted on a subtask/action draw."""
+    recipe = TrainingRecipe(
+        blend={
+            "high_level_subtask": TrainingRecipe(
+                weight=0.99,
+                messages=[
+                    MessageTurn(role="user", content="${task}", stream="high_level"),
+                    MessageTurn(role="assistant", content="a subtask", stream="high_level", target=True),
+                ],
+            ),
+            "ask_vqa_top": TrainingRecipe(
+                weight=0.01,
+                bindings={
+                    "vqa_query": "emitted_at(t, style=vqa, role=user, camera=observation.images.top)",
+                    "vqa": "emitted_at(t, style=vqa, role=assistant, camera=observation.images.top)",
+                },
+                messages=[
+                    MessageTurn(
+                        role="user", content="${vqa_query}", stream="high_level", if_present="vqa_query"
+                    ),
+                    MessageTurn(
+                        role="assistant",
+                        content="${vqa}",
+                        stream="high_level",
+                        target=True,
+                        if_present="vqa",
+                    ),
+                ],
+            ),
+        }
+    )
+    # A frame WITH a vqa event renders VQA on every sample_idx, despite the
+    # ask_vqa weight being only 0.01.
+    for sample_idx in range(20):
+        rendered = render_sample(
+            recipe=recipe, persistent=PERSISTENT, events=EVENTS_AT_1, t=1.0, sample_idx=sample_idx, task="x"
+        )
+        assert rendered["messages"][-1]["content"] == '{"count": 2}', sample_idx
+    # A frame WITHOUT a vqa event falls back to the normal weighted blend.
+    rendered = render_sample(recipe=recipe, persistent=PERSISTENT, events=[], t=1.0, sample_idx=0, task="x")
+    assert rendered["messages"][-1]["content"] == "a subtask"
+
+
 def test_emitted_at_persistent_tolerates_small_timestamp_drift():
     """Persistent ``emitted_at`` should match within EMITTED_AT_TOLERANCE_S
     so callers that derive ``t`` arithmetically (``frame_idx / fps``) still

tests/datasets/test_lerobot_dataset.py

@@ -482,20 +482,6 @@ def test_add_frame_works_in_write_mode(tmp_path):
 # ── Resume mode ──────────────────────────────────────────────────────
 
 
-def test_resume_freshly_created_empty_dataset(tmp_path):
-    """resume() accepts a local dataset created before any episode was recorded."""
-    root = tmp_path / "resume_empty_ds"
-    LeRobotDataset.create(repo_id=DUMMY_REPO_ID, fps=DEFAULT_FPS, features=SIMPLE_FEATURES, root=root)
-
-    resumed = LeRobotDataset.resume(repo_id=DUMMY_REPO_ID, root=root)
-
-    assert isinstance(resumed.writer, DatasetWriter)
-    assert resumed.meta.total_episodes == 0
-    assert resumed.meta.total_frames == 0
-    assert resumed.meta.tasks is None
-    assert resumed.meta.episodes is None
-
-
 def test_resume_creates_writer(tmp_path):
     """After resume(), writer is a DatasetWriter."""
     root = tmp_path / "resume_ds"

tests/datasets/test_sampler.py

@@ -25,7 +25,7 @@ from datasets import Dataset  # noqa: E402
 from lerobot.datasets.io_utils import (
     hf_transform_to_torch,
 )
-from lerobot.datasets.sampler import EpisodeAwareSampler
+from lerobot.datasets.sampler import EpisodeAwareSampler, compute_sampler_state
 
 
 def calculate_episode_data_index(hf_dataset: Dataset) -> dict[str, torch.Tensor]:
@@ -154,8 +154,6 @@ def test_partial_episode_drop_warns(caplog):
 
 # --- seeded (seed, epoch) shuffling, resume, and state ---
 
-from lerobot.datasets.sampler import compute_sampler_state  # noqa: E402
-
 EPISODE_BOUNDS = ([0, 2, 3], [2, 3, 6])  # episodes of 2, 1 and 3 frames
 
 

tests/motors/test_feetech.py

@@ -294,19 +294,6 @@ def test__sync_read(addr, length, ids_values, mock_motors, dummy_motors):
     assert read_values == ids_values
 
 
-def test__sync_read_retries_after_transient_failure(mock_motors, dummy_motors):
-    addr, length, ids_values = (10, 4, {1: 1337})
-    stub = mock_motors.build_sync_read_stub(addr, length, ids_values, num_invalid_try=1)
-    bus = FeetechMotorsBus(port=mock_motors.port, motors=dummy_motors)
-    bus.connect(handshake=False)
-
-    read_values, read_comm = bus._sync_read(addr, length, list(ids_values), num_retry=1)
-
-    assert read_comm == scs.COMM_SUCCESS
-    assert read_values == ids_values
-    assert mock_motors.stubs[stub].calls == 2
-
-
 @pytest.mark.parametrize("raise_on_error", (True, False))
 def test__sync_read_comm(raise_on_error, mock_motors, dummy_motors):
     addr, length, ids_values = (10, 4, {1: 1337})

tests/policies/evo1/test_evo1.py

@@ -496,60 +496,6 @@ def test_evo1_processor_save_load_round_trip_applies_config_overrides(tmp_path):
     assert "embodiment_id" in processed
 
 
-def test_reconcile_evo1_processors_repads_overridden_stats(tmp_path):
-    """Loading a checkpoint and injecting raw (unpadded) dataset stats must be re-padded.
-
-    Regression test: lerobot-train passes the raw dataset stats as normalizer/unnormalizer
-    overrides when resuming from a checkpoint (e.g. stage2 from a stage1 checkpoint). Those stats
-    are at the dataset dims (e.g. LIBERO state=8/action=7), but EVO1 pads state/action to
-    max_state_dim/max_action_dim before normalization, so reconcile_evo1_processors must re-pad the
-    stats or normalization crashes with a shape mismatch.
-    """
-    config = make_config()
-    preprocessor, postprocessor = make_evo1_pre_post_processors(config, dataset_stats=make_stats())
-    preprocessor.save_pretrained(tmp_path)
-    postprocessor.save_pretrained(tmp_path)
-
-    # Reload with the generic override path injecting raw, unpadded dataset stats.
-    raw_stats = make_stats()
-    loaded_pre = PolicyProcessorPipeline.from_pretrained(
-        tmp_path,
-        config_filename=f"{POLICY_PREPROCESSOR_DEFAULT_NAME}.json",
-        overrides={"normalizer_processor": {"stats": raw_stats}},
-        to_transition=batch_to_transition,
-        to_output=transition_to_batch,
-    )
-    loaded_post = PolicyProcessorPipeline.from_pretrained(
-        tmp_path,
-        config_filename=f"{POLICY_POSTPROCESSOR_DEFAULT_NAME}.json",
-        overrides={"unnormalizer_processor": {"stats": raw_stats}},
-        to_transition=policy_action_to_transition,
-        to_output=transition_to_policy_action,
-    )
-
-    # Sanity: the override really injected unpadded stats before reconciliation.
-    normalizer = next(step for step in loaded_pre.steps if isinstance(step, NormalizerProcessorStep))
-    assert normalizer._tensor_stats[OBS_STATE]["min"].shape == (STATE_DIM,)
-
-    loaded_pre, loaded_post = reconcile_evo1_processors(config, loaded_pre, loaded_post)
-
-    normalizer = next(step for step in loaded_pre.steps if isinstance(step, NormalizerProcessorStep))
-    unnormalizer = next(step for step in loaded_post.steps if isinstance(step, UnnormalizerProcessorStep))
-    assert normalizer._tensor_stats[OBS_STATE]["min"].shape == (MAX_STATE_DIM,)
-    assert normalizer._tensor_stats[ACTION]["min"].shape == (MAX_ACTION_DIM,)
-    assert unnormalizer._tensor_stats[ACTION]["min"].shape == (MAX_ACTION_DIM,)
-
-    # Normalizing a padded state must not raise (this is the exact runtime path that crashed).
-    processed = loaded_pre(
-        {
-            "task": "pick the block",
-            OBS_STATE: torch.zeros(STATE_DIM),
-            f"{OBS_IMAGES}.front": torch.rand(3, 16, 16),
-        }
-    )
-    assert processed[OBS_STATE].shape == (1, MAX_STATE_DIM)
-
-
 def test_evo1_policy_forward_and_inference_use_batched_embedding(monkeypatch):
     monkeypatch.setattr(modeling_evo1, "Evo1Model", DummyEvo1Model)
     policy = modeling_evo1.Evo1Policy(make_config())

tests/policies/test_factory_peft_revision.py

@@ -1,83 +0,0 @@
-# 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.
-
-from types import SimpleNamespace
-from unittest.mock import MagicMock
-
-import torch
-
-import lerobot.policies.factory as policy_factory
-
-
-def test_make_policy_keeps_peft_adapter_and_base_revisions_separate(monkeypatch):
-    cfg = SimpleNamespace(
-        type="mock",
-        device="cpu",
-        pretrained_path="user/adapter",
-        pretrained_revision="adapter-sha",
-        use_peft=True,
-        input_features={},
-        output_features={},
-    )
-    dataset_meta = SimpleNamespace(features={}, stats={})
-
-    base_policy = torch.nn.Linear(1, 1)
-    policy_from_pretrained = MagicMock(return_value=base_policy)
-    policy_class = SimpleNamespace(from_pretrained=policy_from_pretrained)
-    monkeypatch.setattr(policy_factory, "get_policy_class", lambda _: policy_class)
-    monkeypatch.setattr(policy_factory, "dataset_to_policy_features", lambda _: {})
-    monkeypatch.setattr(policy_factory, "validate_visual_features_consistency", lambda *args: None)
-
-    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 = torch.nn.Linear(1, 1)
-    peft_model_from_pretrained = MagicMock(return_value=adapted_policy)
-    require_package = MagicMock()
-    monkeypatch.setattr(policy_factory, "require_package", require_package)
-    monkeypatch.setattr(
-        policy_factory,
-        "PeftConfig",
-        SimpleNamespace(from_pretrained=peft_config_from_pretrained),
-    )
-    monkeypatch.setattr(
-        policy_factory,
-        "PeftModel",
-        SimpleNamespace(from_pretrained=peft_model_from_pretrained),
-    )
-
-    policy = policy_factory.make_policy(cfg, ds_meta=dataset_meta)
-
-    assert policy is adapted_policy
-    require_package.assert_called_once_with("peft", extra="peft")
-    peft_config_from_pretrained.assert_called_once_with(
-        "user/adapter",
-        revision="adapter-sha",
-    )
-    policy_from_pretrained.assert_called_once_with(
-        config=cfg,
-        dataset_stats=dataset_meta.stats,
-        dataset_meta=dataset_meta,
-        pretrained_name_or_path="user/base-policy",
-        revision="base-sha",
-    )
-    peft_model_from_pretrained.assert_called_once_with(
-        base_policy,
-        "user/adapter",
-        config=peft_config,
-        revision="adapter-sha",
-        is_trainable=True,
-    )

tests/processor/test_render_messages_processor.py

@@ -12,7 +12,9 @@ from lerobot.processor.render_messages_processor import RenderMessagesStep  # no
 from lerobot.types import TransitionKey  # noqa: E402
 
 
-def test_render_messages_step_noops_without_language_columns():
+def test_render_messages_step_renders_task_fallback_without_language_columns():
+    """No language columns + a task string → low-level task fallback render,
+    matching what the policy sees at eval time on unannotated observations."""
     recipe = TrainingRecipe(
         messages=[
             MessageTurn(role="user", content="${task}", stream="high_level"),
@@ -21,6 +23,24 @@ def test_render_messages_step_noops_without_language_columns():
     )
     transition = create_transition(complementary_data={"task": "do it"})
 
+    out = RenderMessagesStep(recipe)(transition)
+    data = out[TransitionKey.COMPLEMENTARY_DATA]
+
+    assert data["messages"] == [{"role": "user", "content": "do it"}]
+    assert data["message_streams"] == ["low_level"]
+    assert data["target_message_indices"] == []
+    assert data["task"] == "do it"
+
+
+def test_render_messages_step_noops_without_language_columns_or_task():
+    recipe = TrainingRecipe(
+        messages=[
+            MessageTurn(role="user", content="${task}", stream="high_level"),
+            MessageTurn(role="assistant", content="${subtask}", stream="low_level", target=True),
+        ]
+    )
+    transition = create_transition(complementary_data={})
+
     assert RenderMessagesStep(recipe)(transition) == transition
 
 
@@ -58,3 +78,70 @@ def test_render_messages_step_renders_and_drops_raw_language():
     assert data["messages"][-1]["content"] == "reach carefully"
     assert data["message_streams"] == ["high_level", "low_level"]
     assert data["target_message_indices"] == [1]
+
+
+def test_render_messages_step_falls_back_to_low_level_task_when_recipe_misses():
+    recipe = TrainingRecipe(
+        messages=[
+            MessageTurn(
+                role="assistant",
+                content="${subtask}",
+                stream="high_level",
+                target=True,
+                if_present="subtask",
+            ),
+        ]
+    )
+    transition = create_transition(
+        complementary_data={
+            "task": "pick the cube",
+            "timestamp": torch.tensor(0.0),
+            "index": torch.tensor(7),
+            "language_persistent": [],
+            "language_events": [{"style": "unmatched", "timestamp": 0.0}],
+        }
+    )
+
+    out = RenderMessagesStep(recipe)(transition)
+    data = out[TransitionKey.COMPLEMENTARY_DATA]
+
+    assert data["messages"] == [{"role": "user", "content": "pick the cube"}]
+    assert data["message_streams"] == ["low_level"]
+    assert data["target_message_indices"] == []
+
+
+def test_render_messages_step_falls_back_per_sample_in_batched_language():
+    recipe = TrainingRecipe(
+        messages=[
+            MessageTurn(
+                role="assistant",
+                content="${subtask}",
+                stream="high_level",
+                target=True,
+                if_present="subtask",
+            ),
+        ]
+    )
+    transition = create_transition(
+        action=torch.arange(4).reshape(2, 2),
+        complementary_data={
+            "task": ["pick the cube", "open the drawer"],
+            "timestamp": torch.tensor([0.0, 1.0]),
+            "index": torch.tensor([7, 8]),
+            "language_persistent": [[], []],
+            "language_events": [
+                [{"style": "unmatched", "timestamp": 0.0}],
+                [{"style": "unmatched", "timestamp": 1.0}],
+            ],
+        },
+    )
+
+    out = RenderMessagesStep(recipe)(transition)
+    data = out[TransitionKey.COMPLEMENTARY_DATA]
+
+    assert data["messages"] == [
+        [{"role": "user", "content": "pick the cube"}],
+        [{"role": "user", "content": "open the drawer"}],
+    ]
+    assert data["message_streams"] == [["low_level"], ["low_level"]]
+    assert data["target_message_indices"] == [[], []]

tests/processor/test_tokenizer_processor.py

@@ -25,7 +25,7 @@ import pytest
 import torch
 
 from lerobot.configs.types import FeatureType, PipelineFeatureType, PolicyFeature
-from lerobot.processor import DataProcessorPipeline, TokenizerProcessorStep
+from lerobot.processor import ActionTokenizerProcessorStep, DataProcessorPipeline, TokenizerProcessorStep
 from lerobot.processor.converters import create_transition, identity_transition
 from lerobot.types import TransitionKey
 from lerobot.utils.constants import (
@@ -88,6 +88,46 @@ class MockTokenizer:
         return result
 
 
+def test_action_tokenizer_config_preserves_token_mapping():
+    processor = object.__new__(ActionTokenizerProcessorStep)
+    processor.trust_remote_code = True
+    processor.max_action_tokens = 384
+    processor.fast_skip_tokens = 64
+    processor.paligemma_tokenizer_name = "custom/paligemma"
+    processor.allow_truncation = False
+    processor.action_tokenizer_name = "custom/fast"
+    processor.action_tokenizer_input_object = None
+
+    assert processor.get_config() == {
+        "trust_remote_code": True,
+        "max_action_tokens": 384,
+        "fast_skip_tokens": 64,
+        "paligemma_tokenizer_name": "custom/paligemma",
+        "allow_truncation": False,
+        "action_tokenizer_name": "custom/fast",
+    }
+
+
+def test_action_tokenizer_can_reject_truncated_sequences():
+    processor = object.__new__(ActionTokenizerProcessorStep)
+    processor.max_action_tokens = 4
+    processor.fast_skip_tokens = 128
+    processor.allow_truncation = False
+    processor.action_tokenizer = lambda _actions: [1, 2, 3]
+    processor._paligemma_tokenizer = type(
+        "Tokenizer",
+        (),
+        {
+            "vocab_size": 1000,
+            "bos_token_id": 2,
+            "encode": lambda _self, text, **_kwargs: [10, 11] if text == "Action: " else [12, 1],
+        },
+    )()
+
+    with pytest.raises(ValueError, match="max_action_tokens=4"):
+        processor._tokenize_action(torch.zeros(1, 2, 1))
+
+
 @pytest.fixture
 def mock_tokenizer():
     """Provide a mock tokenizer for testing."""

tests/robots/test_so100_follower.py

@@ -49,7 +49,7 @@ def _make_bus_mock() -> MagicMock:
 
 
 @pytest.fixture
-def follower(tmp_path):
+def follower():
     bus_mock = _make_bus_mock()
 
     def _bus_side_effect(*_args, **kwargs):
@@ -71,7 +71,7 @@ def follower(tmp_path):
         ),
         patch.object(SO100Follower, "configure", lambda self: None),
     ):
-        cfg = SO100FollowerConfig(port="/dev/null", calibration_dir=tmp_path)
+        cfg = SO100FollowerConfig(port="/dev/null")
         robot = SO100Follower(cfg)
         yield robot
         if robot.is_connected:
@@ -99,27 +99,6 @@ def test_get_observation(follower):
         assert obs[f"{motor}.pos"] == idx
 
 
-def test_get_observation_uses_read_retries(follower):
-    # Feetech buses can intermittently fail a sync_read; the follower should forward the configured
-    # retry count so transient failures don't abort the control loop (see #3131).
-    follower.config.num_read_retries = 7
-    follower.connect()
-    follower.get_observation()
-
-    follower.bus.sync_read.assert_called_once_with("Present_Position", num_retry=7)
-
-
-def test_send_action_uses_read_retries(follower):
-    follower.config.max_relative_target = 10.0
-    follower.config.num_read_retries = 7
-    follower.connect()
-
-    action = {f"{motor}.pos": value * 10 for value, motor in enumerate(follower.bus.motors, 1)}
-    follower.send_action(action)
-
-    follower.bus.sync_read.assert_called_once_with("Present_Position", num_retry=7)
-
-
 def test_send_action(follower):
     follower.connect()
 
@@ -149,51 +128,3 @@ def test_configure_writes_position_pid_coefficients():
     bus_mock.write.assert_any_call("P_Coefficient", "shoulder_pan", 32)
     bus_mock.write.assert_any_call("I_Coefficient", "shoulder_pan", 1)
     bus_mock.write.assert_any_call("D_Coefficient", "shoulder_pan", 16)
-
-
-@pytest.mark.parametrize(
-    "gripper_closed_pos, expected_drive_mode",
-    [
-        (2035, 0),  # closed position at range_min -> raw increases when opening -> not inverted
-        (3528, 1),  # closed position at range_max -> raw increases when closing -> inverted
-        (2781, None),  # not near either end stop -> unsafe to infer
-    ],
-)
-def test_calibrate_detects_gripper_drive_mode(follower, gripper_closed_pos, expected_drive_mode):
-    """Regression test for #3942: the follower gripper can be mounted mirrored with respect to the
-    leader's, in which case its raw position increases when closing. Calibration must detect this
-    and set drive_mode=1 so that normalized values follow the 0=closed/100=open convention."""
-    follower.connect()
-
-    motors = list(follower.bus.motors)
-    follower.bus.set_half_turn_homings.return_value = dict.fromkeys(motors, 0)
-    follower.bus.record_ranges_of_motion.return_value = (
-        dict.fromkeys(motors, 2035),
-        dict.fromkeys(motors, 3528),
-    )
-    follower.bus.read.return_value = gripper_closed_pos
-
-    with (
-        patch("builtins.input", return_value=""),
-        patch.object(type(follower), "_save_calibration", lambda self: None),
-    ):
-        follower.calibration = {}
-        if expected_drive_mode is None:
-            with pytest.raises(ValueError, match="Gripper is not fully closed"):
-                follower.calibrate()
-        else:
-            follower.calibrate()
-
-    follower.bus.read.assert_called_with(
-        "Present_Position",
-        "gripper",
-        normalize=False,
-        num_retry=follower.config.num_read_retries,
-    )
-    if expected_drive_mode is None:
-        return
-
-    assert follower.calibration["gripper"].drive_mode == expected_drive_mode
-    for motor in motors:
-        if motor != "gripper":
-            assert follower.calibration[motor].drive_mode == 0

tests/test_rollout.py

@@ -185,25 +185,18 @@ def test_load_pretrained_peft_policy_keeps_adapter_and_base_revisions_separate(m
     peft_config_from_pretrained = MagicMock(return_value=peft_config)
     adapted_policy = MagicMock()
     peft_model_from_pretrained = MagicMock(return_value=adapted_policy)
-    require_package = MagicMock()
-    monkeypatch.setattr(rollout_context, "require_package", require_package)
-    monkeypatch.setattr(
-        rollout_context,
-        "PeftConfig",
-        SimpleNamespace(from_pretrained=peft_config_from_pretrained),
-        raising=False,
-    )
-    monkeypatch.setattr(
-        rollout_context,
-        "PeftModel",
-        SimpleNamespace(from_pretrained=peft_model_from_pretrained),
-        raising=False,
+    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
-    require_package.assert_called_once_with("peft", extra="peft")
     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",

tests/utils/test_collate.py

@@ -9,7 +9,7 @@ import torch  # noqa: E402
 from lerobot.utils.collate import lerobot_collate_fn  # noqa: E402
 
 
-def test_lerobot_collate_preserves_messages_and_drops_raw_language():
+def test_lerobot_collate_preserves_messages_and_raw_language():
     batch = [
         {
             "index": torch.tensor(0),
@@ -17,14 +17,14 @@ def test_lerobot_collate_preserves_messages_and_drops_raw_language():
             "message_streams": ["low_level"],
             "target_message_indices": [0],
             "language_persistent": [{"content": "raw"}],
-            "language_events": [],
+            "language_events": [{"content": "event a"}],
         },
         {
             "index": torch.tensor(1),
             "messages": [{"role": "assistant", "content": "b"}],
             "message_streams": ["low_level"],
             "target_message_indices": [0],
-            "language_persistent": [{"content": "raw"}],
+            "language_persistent": [{"content": "raw b"}],
             "language_events": [],
         },
     ]
@@ -36,8 +36,8 @@ def test_lerobot_collate_preserves_messages_and_drops_raw_language():
     assert out["messages"][1][0]["content"] == "b"
     assert out["message_streams"] == [["low_level"], ["low_level"]]
     assert out["target_message_indices"] == [[0], [0]]
-    assert "language_persistent" not in out
-    assert "language_events" not in out
+    assert out["language_persistent"] == [[{"content": "raw"}], [{"content": "raw b"}]]
+    assert out["language_events"] == [[{"content": "event a"}], []]
 
 
 def test_lerobot_collate_passes_through_standard_batch():