feat(runtime): add grounded VQA command

AGENT_GUIDE.md

@@ -61,20 +61,16 @@ Full details in [`docs/source/so101.mdx`](./docs/source/so101.mdx) and [`docs/so
 **4.1 Install**
 
 ```bash
-# uv (recommended — see AGENTS.md and CLAUDE.md)
-uv sync --locked --extra feetech          # SO-100/SO-101 motor stack
-# uv sync --locked --extra all            # everything
-# uv sync --locked --extra smolvla        # add SmolVLA deps
-
-# pip (alternative, e.g. when not working from source)
-# pip install 'lerobot[feetech]'
-# pip install 'lerobot[all]'
-# pip install 'lerobot[smolvla]'
-
+pip install 'lerobot[feetech]'              # SO-100/SO-101 motor stack
+# pip install 'lerobot[all]'                # everything
+# pip install 'lerobot[aloha,pusht]'        # specific features
+# pip install 'lerobot[smolvla]'            # add SmolVLA deps
 git lfs install && git lfs pull
 hf auth login                               # required to push datasets/policies
 ```
 
+Contributors can alternatively use `uv sync --locked --extra feetech` (see `AGENTS.md`).
+
 **4.2 Find USB ports** — run once per arm, unplug when prompted.
 
 ```bash

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/bring_your_own_policies.mdx

@@ -191,6 +191,162 @@ def make_my_policy_pre_post_processors(
 
 ---
 
+## Adding high- and low-level language control
+
+The policy API above is sufficient for training and standard evaluation. To use a language-conditioned policy with interactive `lerobot-rollout`, also register a runtime adapter. The adapter keeps policy-specific prompting and tokenization out of the generic control loop.
+
+The runtime supports two policy shapes:
+
+| Policy shape     | Behavior                                                                | Adapter                                        |
+| ---------------- | ----------------------------------------------------------------------- | ---------------------------------------------- |
+| Low-level / flat | The operator's task or subtask directly conditions action prediction.   | Reuse `DirectTaskPolicyAdapter`.               |
+| High + low level | The policy generates subtasks or memory, then conditions actions on it. | Subclass `BaseLanguageAdapter`, as PI052 does. |
+
+During a rollout, `RuntimeState` stores the high-level task and the active language context:
+
+```text
+task ──> adapter.generate_text("subtask", ...) ──> state.language_context["subtask"]
+                                                       │
+observation ──> processors ──> adapter.select_action() ─┴─> action chunk ──> robot
+```
+
+The generic runtime handles generation frequency, pause/resume, prompt replacement, action queues, and dispatch. The adapter only translates between that runtime contract and your policy.
+
+### Low-level policies
+
+If your policy already consumes the live task through its normal preprocessor and implements `predict_action_chunk`, register the shared direct adapter. PI0.5 and MolmoAct2 use this path:
+
+```python
+# src/lerobot/runtime/registry.py
+_ADAPTERS = {
+    # ...
+    "my_policy": "lerobot.runtime.adapter:DirectTaskPolicyAdapter",
+}
+```
+
+Run it with direct-subtask mode so the operator supplies the instruction used by the action policy:
+
+```bash
+lerobot-rollout \
+    --language \
+    --policy.path=user/my_policy_checkpoint \
+    --robot.type=so101_follower \
+    --robot.port=/dev/ttyACM0 \
+    --direct_subtask
+```
+
+The rollout context builds the observation batch with the current instruction before `DirectTaskPolicyAdapter` calls `policy.predict_action_chunk(observation)`. No text-generation method is required.
+
+### Hierarchical policies
+
+For a policy that generates language and actions, subclass [`BaseLanguageAdapter`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/runtime/adapter.py) and implement two methods:
+
+- `generate_text(kind, observation, state, user_text=None) -> str` generates a `subtask`, `memory`, or interjection response.
+- `select_action(observation, state)` builds the low-level prompt from the active context and returns an action chunk.
+
+This abbreviated adapter follows [`PI052PolicyAdapter`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/inference/pi052_adapter.py):
+
+```python
+# inference/my_policy_adapter.py
+from typing import Any
+
+from lerobot.runtime import RuntimeState
+from lerobot.runtime.adapter import BaseLanguageAdapter
+from lerobot.utils.constants import (
+    OBS_LANGUAGE_ATTENTION_MASK,
+    OBS_LANGUAGE_TOKENS,
+)
+
+
+class MyPolicyAdapter(BaseLanguageAdapter):
+    def select_action(self, observation: dict[str, Any], state: RuntimeState):
+        instruction = state.language_context.get("subtask") or state.task or ""
+        tokens, attention_mask = tokenize_instruction(instruction)
+
+        batch = dict(observation)
+        batch[OBS_LANGUAGE_TOKENS] = tokens
+        batch[OBS_LANGUAGE_ATTENTION_MASK] = attention_mask
+        return self.policy.predict_action_chunk(batch)
+
+    def generate_text(
+        self,
+        kind: str,
+        observation: dict[str, Any] | None,
+        state: RuntimeState,
+        user_text: str | None = None,
+    ) -> str:
+        messages = self.build_messages(kind, state, user_text)
+        batch, tokenizer = tokenize_messages(messages, observation)
+        return self.policy.select_message(
+            batch,
+            tokenizer=tokenizer,
+            min_new_tokens=self.gen.min_new_tokens,
+            temperature=self.gen.temperature,
+            top_p=self.gen.top_p,
+        )
+
+    def build_messages(
+        self, kind: str, state: RuntimeState, user_text: str | None
+    ) -> list[dict[str, str]]:
+        if kind == "subtask":
+            return [{"role": "user", "content": state.task or ""}]
+        if kind == "memory":
+            return [
+                {"role": "user", "content": state.task or ""},
+                {
+                    "role": "user",
+                    "content": f"Completed subtask: {state.extra.get('prior_subtask', '')}",
+                },
+            ]
+        if kind == "interjection":
+            return [
+                {"role": "user", "content": state.task or ""},
+                {"role": "user", "content": user_text or ""},
+            ]
+        raise ValueError(f"Unsupported text kind: {kind}")
+```
+
+`tokenize_instruction` and `tokenize_messages` are policy-specific helpers. They must reproduce the prompt format used during training; PI052, for example, adds the discretized robot state to its low-level subtask prompt and uses the same PaliGemma formatting for `select_message`.
+
+`BaseLanguageAdapter` provides the default hierarchy: regenerate a subtask at action-chunk boundaries, update memory when the subtask changes, and handle user interjections. Override `_regenerate_context` only if your policy uses a different hierarchy.
+
+Register the adapter with a lazy import so importing LeRobot does not load the model or its optional dependencies:
+
+```python
+# src/lerobot/runtime/registry.py
+_ADAPTERS = {
+    # ...
+    "my_policy": "lerobot.policies.my_policy.inference.my_policy_adapter:MyPolicyAdapter",
+}
+```
+
+The key must match the policy's registered type. Once registered, the same checkpoint works through the shared entry point:
+
+```bash
+lerobot-rollout \
+    --language \
+    --policy.path=user/my_hierarchical_checkpoint \
+    --robot.type=so101_follower \
+    --robot.port=/dev/ttyACM0 \
+    --task="put the cup in the sink"
+```
+
+For RoboCasa-compatible policies, replace the robot arguments with `--sim --sim.task=<task>`. Without `--direct_subtask`, the adapter generates the low-level subtask; with it, the operator bypasses high-level generation and supplies each subtask.
+
+### Keep training and deployment aligned
+
+The adapter is intentionally small, but its prompts are part of the model contract:
+
+- Use the same tokenizer, role formatting, special tokens, image ordering, and state encoding as training.
+- Condition `select_action` on `state.language_context["subtask"]`, falling back to `state.task` for direct or not-yet-generated prompts.
+- Return a full action chunk from `select_action`; the runtime handles control-rate dispatch.
+- Keep optional model dependencies inside lazy imports.
+- Test adapter selection, generated-message routing, action-batch construction, and direct-subtask behavior with a lightweight fake policy.
+
+PI052 is the complete in-tree reference: its [processor](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/processor_pi052.py) renders the training recipe, its [policy](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/modeling_pi052.py) exposes text and action generation, and its [adapter](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/inference/pi052_adapter.py) reconstructs those same prompts at deployment.
+
+---
+
 ## Path A: Out-of-tree plugin
 
 The fastest way to ship a policy: package it as a standalone Python distribution and install it alongside LeRobot. No PR required, you own the release cycle, and you can publish to PyPI under your own namespace.

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/env_processor.mdx

@@ -88,6 +88,20 @@ policy_preprocessor = NormalizerProcessorStep(stats=dataset_stats)
 
 The same policy can work with different environment processors, and the same environment processor can work with different policies:
 
+````python
+# Use SmolVLA policy with LIBERO environment
+# Use SmolVLA policy with LIBERO environment
+libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(
+    env_cfg=libero_cfg,
+    policy_cfg=smolvla_cfg,
+)
+smolvla_preprocessor, smolvla_postprocessor = make_pre_post_processors(smolvla_cfg)
+# Or use ACT policy with the same LIBERO environment
+libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(
+    env_cfg=libero_cfg,
+    policy_cfg=act_cfg,
+)
+act_preprocessor, act_postprocessor = make_pre_post_processors(act_cfg)
 ```python
 # Use SmolVLA policy with LIBERO environment
 libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(
@@ -102,7 +116,6 @@ libero_preprocessor, libero_postprocessor = make_env_pre_post_processors(
     policy_cfg=act_cfg,
 )
 act_preprocessor, act_postprocessor = make_pre_post_processors(act_cfg)
-```
 
 ### 3. **Easier Experimentation**
 
@@ -132,7 +145,7 @@ class LiberoVelocityProcessorStep(ObservationProcessorStep):
         state = torch.cat([eef_pos, eef_axisangle, eef_vel,
                           gripper_pos, gripper_vel], dim=-1)  # 14D
         return state
-```
+````
 
 ### 4. **Cleaner Environment Code**
 

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/hope_jr.mdx

@@ -211,7 +211,7 @@ Record, Replay and Train with Hope-JR is still experimental.
 
 ### Record
 
-This step records the dataset, which can be seen as an example [here](https://huggingface.co/datasets/nepyope/hand_record_test_with_video_data).
+This step records the dataset, which can be seen as an example [here](https://huggingface.co/datasets/nepyope/hand_record_test_with_video_data/settings).
 
 ```bash
 lerobot-record \

docs/source/inference.mdx

@@ -1,6 +1,6 @@
 # Policy Deployment (lerobot-rollout)
 
-`lerobot-rollout` is the single CLI for deploying trained policies on real robots. It supports multiple execution strategies and inference backends, from quick evaluation to continuous recording and human-in-the-loop data collection.
+`lerobot-rollout` is the single CLI for deploying trained policies on real robots or in an interactive simulator. It supports multiple execution strategies and inference backends, from quick evaluation to continuous recording, language-driven control, and human-in-the-loop data collection.
 
 ## Quick Start
 
@@ -197,6 +197,52 @@ Teleop is optional — if omitted the robot holds its position during the reset
 
 ---
 
+## Interactive language control
+
+Language-conditioned policies can expose a high-level text head in addition to
+their action head. Add `--language` to open-prompt one of these policies on a
+real robot. Language-only flags such as `--direct_subtask` select this mode
+automatically.
+
+MolmoAct2 has no high-level planner, so use direct-subtask mode and type each
+next low-level instruction yourself:
+
+```bash
+lerobot-rollout \
+    --policy.path=lerobot/MolmoAct2-SO100_101-LeRobot \
+    --policy.device=cuda \
+    --robot.type=so101_follower \
+    --robot.port=/dev/ttyACM1 \
+    --robot.cameras='{"cam0":{"type":"opencv","index_or_path":"/dev/video0","width":640,"height":480,"fps":30,"fourcc":"MJPG","backend":200},"cam1":{"type":"opencv","index_or_path":"/dev/video2","width":640,"height":480,"fps":30,"fourcc":"MJPG","backend":200}}' \
+    --direct_subtask \
+    --robot.max_relative_target='{"shoulder_pan":5,"shoulder_lift":5,"elbow_flex":5,"wrist_flex":5,"wrist_roll":5,"gripper":5}'
+```
+
+The robot starts paused. Type a subtask, then use `/resume` and `/pause` to
+control action dispatch. Check the workspace and motion limits before resuming.
+Without `--direct_subtask`, a policy such as PI052 generates its active subtask
+from the high-level `--task` itself.
+
+RoboCasa uses the same runtime and processor path. `--sim` selects it
+automatically, so no robot configuration is needed:
+
+```bash
+MUJOCO_GL=egl lerobot-rollout \
+    --policy.path=lerobot/pi052_robocasa \
+    --sim --sim.task=CloseFridge --sim.split=pretrain \
+    --task="close the fridge" \
+    --disable_memory \
+    --sim.render_size=384 \
+    --sim.views=robot0_agentview_left,robot0_eye_in_hand,robot0_agentview_right \
+    --mode=action --ctrl_hz=20
+```
+
+Open `http://localhost:8010` for the live simulator view. Add
+`--sim.direct_subtask` to bypass the language planner and make each typed prompt
+the action policy's current subtask.
+
+---
+
 ## Inference Backends
 
 Select a backend with `--inference.type=<name>`. All strategies work with both backends.

docs/source/integrate_hardware.mdx

@@ -18,7 +18,7 @@ If you're using Feetech or Dynamixel motors, LeRobot provides built-in bus inter
 - [`DynamixelMotorsBus`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/motors/dynamixel/dynamixel.py) – for controlling Dynamixel servos
 
 Please refer to the [`MotorsBus`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/motors/motors_bus.py) abstract class to learn about its API.
-For a good example of how it can be used, you can have a look at our own [SO101 follower implementation](https://github.com/huggingface/lerobot/blob/main/src/lerobot/robots/so_follower/so_follower.py)
+For a good example of how it can be used, you can have a look at our own [SO101 follower implementation](https://github.com/huggingface/lerobot/blob/main/src/lerobot/robots/so_follower/so101_follower/so101_follower.py)
 
 Use these if compatible. Otherwise, you'll need to find or write a Python interface (not covered in this tutorial):
 

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/lekiwi.mdx

@@ -51,7 +51,7 @@ In addition to these instructions, you need to install the Feetech SDK & ZeroMQ
 pip install -e ".[lekiwi]"
 ```
 
-Great 🤗! You are now done installing LeRobot, and we can begin assembling the SO100/SO101 arms and the mobile base 🤖.
+Great :hugs:! You are now done installing LeRobot, and we can begin assembling the SO100/SO101 arms and the mobile base :robot:.
 Every time you now want to use LeRobot, you can go to the `~/lerobot` folder where we installed LeRobot and run one of the commands.
 
 # Step-by-Step Assembly Instructions

docs/source/pi0fast.mdx

@@ -174,7 +174,7 @@ The model takes images, text instructions, and robot state as input, and outputs
 
 ## Reproducing π₀Fast results
 
-We reproduce the results of π₀Fast on the LIBERO benchmark using the LeRobot implementation. We take the LeRobot PiFast base model [lerobot/pi0fast-base](https://huggingface.co/lerobot/pi0fast-base) and finetune for an additional 40k steps in bfloat16, with batch size of 256 on 8 H100 GPUs using the [HuggingFace LIBERO dataset](https://huggingface.co/datasets/HuggingFaceVLA/libero).
+We reproduce the results of π₀Fast on the LIBERO benchmark using the LeRobot implementation. We take the LeRobot PiFast base model [lerobot/pi0fast-base](https://huggingface.co/lerobot/pi0fast-base) and finetune for an additional 40kk steps in bfloat16, with batch size of 256 on 8 H100 GPUs using the [HuggingFace LIBERO dataset](https://huggingface.co/datasets/HuggingFaceVLA/libero).
 
 The finetuned model can be found here:
 

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/smolvla.mdx

@@ -93,7 +93,7 @@ lerobot-train --help
 
 ## Evaluate the finetuned model and run it in real-time
 
-Similarly for when recording an episode, it is recommended that you are logged in to the HuggingFace Hub. You can follow the corresponding steps: [Record a dataset](./il_robots#record-a-dataset).
+Similarly for when recording an episode, it is recommended that you are logged in to the HuggingFace Hub. You can follow the corresponding steps: [Record a dataset](./il_robots).
 Once you are logged in, you can run inference in your setup by doing:
 
 ```bash

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()
-
-    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
+        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})"
+
     @property
     def is_connected(self) -> bool:
         """Checks if the camera is currently connected and opened."""
@@ -172,7 +164,6 @@ 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()
 
@@ -184,13 +175,6 @@ class OpenCVCamera(Camera):
             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
 
         logger.info(f"{self} connected.")
 
@@ -328,7 +312,6 @@ 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))
@@ -338,9 +321,7 @@ class OpenCVCamera(Camera):
                 # 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)]
-                    )
+                default_fourcc = "".join([chr((default_fourcc_code_int >> 8 * i) & 0xFF) for i in range(4)])
 
                 camera_info = {
                     "name": f"OpenCV Camera @ {target}",
@@ -357,7 +338,6 @@ class OpenCVCamera(Camera):
                 }
 
                 found_cameras_info.append(camera_info)
-            finally:
                 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()
-
-    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
+        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})"
+
     @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,9 +190,7 @@ 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()
 
         # 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.
@@ -224,13 +206,6 @@ class RealSenseCamera(Camera):
                 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
 
         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,8 +406,7 @@ def aggregate_videos(
             videos_idx[key]["dst_file_durations"] = {}
 
     for key, video_idx in videos_idx.items():
-        unique_chunk_file_pairs: list[ChunkFile] = sorted(
-            {
+        unique_chunk_file_pairs = {
             (chunk, file)
             for chunk, file in zip(
                 src_meta.episodes[f"videos/{key}/chunk_index"],
@@ -447,7 +414,7 @@ def aggregate_videos(
                 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(
-        {
+    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,
+            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,8 +580,7 @@ def aggregate_metadata(
     Returns:
         dict: Updated meta_idx with current chunk and file indices.
     """
-    chunk_file_ids: list[ChunkFile] = sorted(
-        {
+    chunk_file_ids = {
         (c, f)
         for c, f in zip(
             src_meta.episodes["meta/episodes/chunk_index"],
@@ -637,7 +588,8 @@ def aggregate_metadata(
             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/configs.py

@@ -560,7 +560,13 @@ class RoboCasaEnv(EnvConfig):
             kwargs["split"] = self.split
         return kwargs
 
-    def create_envs(self, n_envs: int, use_async_envs: bool = False):
+    def create_envs(
+        self,
+        n_envs: int,
+        use_async_envs: bool = False,
+        terminate_on_success: bool = True,
+        horizon: int | None = None,
+    ):
         from .robocasa import create_robocasa_envs
 
         if self.task is None:
@@ -574,6 +580,8 @@ class RoboCasaEnv(EnvConfig):
             env_cls=env_cls,
             episode_length=self.episode_length,
             obj_registries=tuple(self.obj_registries),
+            terminate_on_success=terminate_on_success,
+            horizon=horizon,
         )
 
 

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
 
 
 def _make_env_fns(

src/lerobot/envs/robocasa.py

@@ -33,8 +33,8 @@ logger = logging.getLogger(__name__)
 
 # Dimensions for the flat action/state vectors used by the LeRobot wrapper.
 # These correspond to the PandaOmron robot in RoboCasa365.
-OBS_STATE_DIM = 16  # base_pos(3) + base_quat(4) + ee_pos_rel(3) + ee_quat_rel(4) + gripper_qpos(2)
-ACTION_DIM = 12  # base_motion(4) + control_mode(1) + ee_pos(3) + ee_rot(3) + gripper(1)
+OBS_STATE_DIM = 16  # ee_pos_rel(3) + ee_quat_rel(4) + base_pos(3) + base_quat(4) + gripper_qpos(2)
+ACTION_DIM = 12  # ee_pos(3) + ee_rot(3) + gripper(1) + base_motion(4) + control_mode(1)
 ACTION_LOW = -1.0
 ACTION_HIGH = 1.0
 
@@ -101,14 +101,15 @@ def _resolve_tasks(task: str) -> tuple[list[str], str | None]:
 def convert_action(flat_action: np.ndarray) -> dict[str, Any]:
     """Split a flat (12,) action vector into a RoboCasa action dict.
 
-    Layout: base_motion(4) + control_mode(1) + ee_pos(3) + ee_rot(3) + gripper(1)
+    Layout (openpi / robocasa.utils.env_utils.convert_action order):
+        ee_pos(3) + ee_rot(3) + gripper(1) + base_motion(4) + control_mode(1)
     """
     return {
-        "action.base_motion": flat_action[0:4],
-        "action.control_mode": flat_action[4:5],
-        "action.end_effector_position": flat_action[5:8],
-        "action.end_effector_rotation": flat_action[8:11],
-        "action.gripper_close": flat_action[11:12],
+        "action.end_effector_position": flat_action[0:3],
+        "action.end_effector_rotation": flat_action[3:6],
+        "action.gripper_close": flat_action[6:7],
+        "action.base_motion": flat_action[7:11],
+        "action.control_mode": flat_action[11:12],
     }
 
 
@@ -136,9 +137,16 @@ class RoboCasaEnv(gym.Env):
         episode_length: int | None = None,
         obj_registries: Sequence[str] = DEFAULT_OBJ_REGISTRIES,
         episode_index: int = 0,
+        terminate_on_success: bool = True,
+        horizon: int | None = None,
     ):
         super().__init__()
         self.task = task
+        # When False, a task-success does NOT end/reset the episode — used by the
+        # interactive sim so one kitchen persists across sequential prompts.
+        self.terminate_on_success = terminate_on_success
+        # Underlying robosuite horizon (steps before truncation). None -> default.
+        self.horizon = horizon
         self.obs_type = obs_type
         self.render_mode = render_mode
         self.observation_width = observation_width
@@ -210,12 +218,16 @@ class RoboCasaEnv(gym.Env):
         # (only None/"all"/"pretrain"/"target" are valid). Always pass a
         # valid value so we don't hit that default. Extra kwargs are
         # forwarded to the underlying kitchen env via create_env/robosuite.make.
+        extra_kwargs: dict[str, Any] = {}
+        if self.horizon is not None:
+            extra_kwargs["horizon"] = int(self.horizon)
         self._env = RoboCasaGymEnv(
             env_name=self.task,
             camera_widths=self.observation_width,
             camera_heights=self.observation_height,
             split=self.split if self.split is not None else "all",
             obj_registries=self.obj_registries,
+            **extra_kwargs,
         )
 
         ep_meta = self._env.env.get_ep_meta()
@@ -230,12 +242,14 @@ class RoboCasaEnv(gym.Env):
             return {"pixels": images}
 
         # `state.*` keys come from PandaOmronKeyConverter inside the wrapper.
+        # openpi state order: ee first, then base, then gripper (matches the
+        # openpi robocasa pipeline / examples/robocasa/main.py state layout).
         agent_pos = np.concatenate(
             [
-                raw_obs.get("state.base_position", np.zeros(3)),
-                raw_obs.get("state.base_rotation", np.zeros(4)),
                 raw_obs.get("state.end_effector_position_relative", np.zeros(3)),
                 raw_obs.get("state.end_effector_rotation_relative", np.zeros(4)),
+                raw_obs.get("state.base_position", np.zeros(3)),
+                raw_obs.get("state.base_rotation", np.zeros(4)),
                 raw_obs.get("state.gripper_qpos", np.zeros(2)),
             ],
             axis=-1,
@@ -280,7 +294,7 @@ class RoboCasaEnv(gym.Env):
         raw_obs, reward, done, truncated, info = self._env.step(action_dict)
 
         is_success = bool(info.get("success", False))
-        terminated = done or is_success
+        terminated = done or (is_success and self.terminate_on_success)
         info.update({"task": self.task, "done": done, "is_success": is_success})
 
         observation = self._format_raw_obs(raw_obs)
@@ -313,6 +327,8 @@ def _make_env_fns(
     split: str | None,
     episode_length: int | None,
     obj_registries: Sequence[str],
+    terminate_on_success: bool = True,
+    horizon: int | None = None,
 ) -> list[Callable[[], RoboCasaEnv]]:
     """Build n_envs factory callables for a single task.
 
@@ -335,6 +351,8 @@ def _make_env_fns(
             episode_length=episode_length,
             obj_registries=obj_registries,
             episode_index=episode_index,
+            terminate_on_success=terminate_on_success,
+            horizon=horizon,
         )
 
     return [partial(_make_env, i) for i in range(n_envs)]
@@ -348,6 +366,8 @@ def create_robocasa_envs(
     env_cls: Callable[[Sequence[Callable[[], Any]]], Any] | None = None,
     episode_length: int | None = None,
     obj_registries: Sequence[str] = DEFAULT_OBJ_REGISTRIES,
+    terminate_on_success: bool = True,
+    horizon: int | None = None,
 ) -> dict[str, dict[int, Any]]:
     """Create vectorized RoboCasa365 environments with a consistent return shape.
 
@@ -409,6 +429,8 @@ def create_robocasa_envs(
             split=split,
             episode_length=episode_length,
             obj_registries=obj_registries,
+            terminate_on_success=terminate_on_success,
+            horizon=horizon,
         )
 
         if is_async:

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/gaussian_actor/configuration_gaussian_actor.py

@@ -37,19 +37,13 @@ def is_image_feature(key: str) -> bool:
 @dataclass
 class ConcurrencyConfig:
     """Configuration for the concurrency of the actor and learner.
-
     Possible values are:
     - "threads": Use threads for the actor and learner.
     - "processes": Use processes for the actor and learner.
-
-    ``multiprocessing_context`` selects the process-wide start method when
-    processes are used. Set it to ``None`` to preserve Python's default or a
-    method already selected by the embedding application.
     """
 
     actor: str = "threads"
     learner: str = "threads"
-    multiprocessing_context: str | None = "spawn"
 
 
 @dataclass

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/delta_action_processor.py

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

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:
+            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:
@@ -66,19 +77,148 @@ class RenderMessagesStep(ProcessorStep):
             task=complementary_data.get("task"),
             dataset_ctx=self.dataset_ctx,
         )
+        if rendered is 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/rl/actor.py

@@ -91,7 +91,7 @@ from lerobot.robots import so_follower  # noqa: F401
 from lerobot.teleoperators import gamepad, so_leader  # noqa: F401
 from lerobot.teleoperators.utils import TeleopEvents
 from lerobot.utils.device_utils import get_safe_torch_device
-from lerobot.utils.process import ProcessSignalHandler, ensure_multiprocessing_start_method
+from lerobot.utils.process import ProcessSignalHandler
 from lerobot.utils.random_utils import set_seed
 from lerobot.utils.robot_utils import precise_sleep
 from lerobot.utils.transition import (
@@ -124,7 +124,9 @@ def actor_cli(cfg: TrainRLServerPipelineConfig):
     cfg.validate()
     display_pid = False
     if not use_threads(cfg):
-        ensure_multiprocessing_start_method(cfg.policy.concurrency.multiprocessing_context)
+        import torch.multiprocessing as mp
+
+        mp.set_start_method("spawn")
         display_pid = True
 
     # Create logs directory to ensure it exists

src/lerobot/rl/buffer.py

@@ -18,7 +18,7 @@ import functools
 import threading
 from collections.abc import Callable, Sequence
 from contextlib import suppress
-from typing import NotRequired, TypedDict
+from typing import TypedDict
 
 import torch
 import torch.nn.functional as F  # noqa: N812
@@ -36,7 +36,7 @@ class BatchTransition(TypedDict):
     next_state: dict[str, torch.Tensor]
     done: torch.Tensor
     truncated: torch.Tensor
-    complementary_info: NotRequired[dict[str, torch.Tensor | float | int] | None]
+    complementary_info: dict[str, torch.Tensor | float | int] | None = None
 
 
 def random_crop_vectorized(images: torch.Tensor, output_size: tuple) -> torch.Tensor:

src/lerobot/rl/learner.py

@@ -102,7 +102,7 @@ from lerobot.utils.constants import (
 )
 from lerobot.utils.device_utils import get_safe_torch_device
 from lerobot.utils.io_utils import load_json, write_json
-from lerobot.utils.process import ProcessSignalHandler, ensure_multiprocessing_start_method
+from lerobot.utils.process import ProcessSignalHandler
 from lerobot.utils.random_utils import set_seed
 from lerobot.utils.utils import (
     format_big_number,
@@ -123,7 +123,9 @@ def train_cli(cfg: TrainRLServerPipelineConfig):
     # Fail fast with a friendly error if the optional ``hilserl`` extra is missing.
     require_package("grpcio", extra="hilserl", import_name="grpc")
     if not use_threads(cfg):
-        ensure_multiprocessing_start_method(cfg.policy.concurrency.multiprocessing_context)
+        import torch.multiprocessing as mp
+
+        mp.set_start_method("spawn")
 
     # Use the job_name from the config
     train(

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/robot_kinematic_processor.py

@@ -510,10 +510,10 @@ class ForwardKinematicsJointsToEEAction(RobotActionProcessorStep):
         # We only use the ee pose in the dataset, so we don't need the joint positions
         for n in self.motor_names:
             features[PipelineFeatureType.ACTION].pop(f"{n}.pos", None)
-        # Store end-effector features as actions in the dataset schema
+        # We specify the dataset features of this step that we want to be stored in the dataset
         for k in ["x", "y", "z", "wx", "wy", "wz", "gripper_pos"]:
             features[PipelineFeatureType.ACTION][f"ee.{k}"] = PolicyFeature(
-                type=FeatureType.ACTION, shape=(1,)
+                type=FeatureType.STATE, shape=(1,)
             )
         return features
 

src/lerobot/robots/so_follower/so_follower.py

@@ -180,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")
@@ -221,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/robots/unitree_g1/config_unitree_g1.py

@@ -68,10 +68,6 @@ class UnitreeG1Config(RobotConfig):
     # Compensates for gravity on the unitree's arms using the arm ik solver
     gravity_compensation: bool = False
 
-    # Locomotion controller class name, e.g. "GrootLocomotionController",
-    # "HolosomaLocomotionController", or "SonicWholeBodyController". None disables it.
-    # Selecting "SonicWholeBodyController" implicitly switches the robot to the 64-D
-    # latent-token action/observation interface (``motion_token.{i}.pos`` action and a
-    # ``motion_token_state.{i}.pos`` state echo) so ``lerobot-rollout`` can drive a
-    # policy trained on SONIC motion tokens (e.g. nepyope/sonic_walk).
+    # Lower-body controller class name, e.g. "GrootLocomotionController" or
+    # "HolosomaLocomotionController". None disables it.
     controller: str | None = None

src/lerobot/robots/unitree_g1/controllers/__init__.py

@@ -1,27 +0,0 @@
-#!/usr/bin/env python
-
-# Copyright 2025 The HuggingFace Inc. team. All rights reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""Unitree G1 locomotion controllers (Groot, Holosoma, SONIC)."""
-
-from .gr00t_locomotion import GrootLocomotionController
-from .holosoma_locomotion import HolosomaLocomotionController
-from .sonic_whole_body import SonicWholeBodyController
-
-__all__ = [
-    "GrootLocomotionController",
-    "HolosomaLocomotionController",
-    "SonicWholeBodyController",
-]

src/lerobot/robots/unitree_g1/controllers/sonic_whole_body.py

@@ -1,360 +0,0 @@
-#!/usr/bin/env python
-
-# Copyright 2025 The HuggingFace Inc. team. All rights reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""SONIC decoder whole-body controller for the Unitree G1 (token-only).
-
-Pure-Python/ONNX re-implementation of the *decode* half of NVIDIA's SONIC deploy stack.
-The encoder is intentionally absent: a token-output VLA (e.g. ``nepyope/sonic_walk``)
-supplies the 64-D latent ``motion_token`` directly each tick, and the SONIC **decoder**
-maps ``token + recent proprioception history`` to a residual action that is scaled and
-added onto the standing pose (``default_angles``) to produce 50 Hz joint-position targets
-for the robot's PD controller.
-
-Index spaces: joints exist in two orderings — **IsaacLab** (policy/training order) and
-**MuJoCo** (deploy order). ``ISAACLAB_TO_MUJOCO`` / ``MUJOCO_TO_ISAACLAB`` (in g1_utils)
-convert between them. Quaternions are scalar-first ``(w, x, y, z)``.
-"""
-
-from __future__ import annotations
-
-import json
-import logging
-
-import numpy as np
-import onnx
-import onnxruntime as ort
-from huggingface_hub import hf_hub_download
-
-from ..g1_utils import (
-    ISAACLAB_TO_MUJOCO,
-    MUJOCO_TO_ISAACLAB,
-    G1_29_JointIndex,
-    get_gravity_orientation,
-)
-
-logger = logging.getLogger(__name__)
-
-# ── Constants (hardware-validated; see the NVIDIA SONIC deploy reference) ──────
-CONTROL_DT = 0.02  # 50 Hz control period (s)
-TOKEN_DIM = 64  # decoder latent size
-
-# SONIC decoder checkpoint: NVIDIA's decoder ONNX re-packaged with its deploy constants
-# (kp/kd PD gains, the standing pose default_angles, and the residual action_scale) embedded
-# in the ONNX metadata; see upload_sonic_decoder.py for provisioning. The runtime loads the
-# model *and* all of these straight from the checkpoint (the Holosoma convention), so no
-# motor-physics math happens at deploy time.
-DEFAULT_SONIC_REPO_ID = "lerobot/sonic_decoder"
-DECODER_FILENAME = "model_decoder.onnx"
-DECODER_INPUT_DIM = 994  # token(64) + 10-frame proprio history + gravity
-
-
-def load_sonic_decoder(repo_id: str = DEFAULT_SONIC_REPO_ID):
-    """Load the SONIC decoder ONNX and its baked-in deploy constants from the checkpoint.
-
-    Returns ``(decoder_session, kp, kd, default_angles, action_scale, neutral_token)``. The
-    gains/pose/scale are (29,) float32 in IsaacLab joint order and ``neutral_token`` is the
-    (64,) float32 idle latent -- all read from the ONNX ``metadata_props`` rather than
-    recomputed/hardcoded at deploy time (mirrors ``holosoma_locomotion.load_policy``).
-    """
-    decoder_path = hf_hub_download(repo_id=repo_id, filename=DECODER_FILENAME)
-    so = ort.SessionOptions()
-    so.log_severity_level = 3  # quiet ORT logs
-    session = ort.InferenceSession(decoder_path, sess_options=so)
-    dec_dim = int(session.get_inputs()[0].shape[1])
-    if dec_dim != DECODER_INPUT_DIM:
-        raise RuntimeError(f"Unexpected decoder input dim {dec_dim} (expected {DECODER_INPUT_DIM})")
-
-    meta = {p.key: p.value for p in onnx.load(decoder_path, load_external_data=False).metadata_props}
-    required = ("kp", "kd", "default_angles", "action_scale", "neutral_token")
-    missing = [k for k in required if k not in meta]
-    if missing:
-        raise ValueError(
-            f"SONIC decoder ONNX at {repo_id} is missing metadata {missing}; "
-            "re-run upload_sonic_decoder.py to (re)provision the checkpoint."
-        )
-    arr = {k: np.array(json.loads(meta[k]), dtype=np.float32) for k in required}
-    logger.info("Loaded SONIC deploy constants from %s (%d joints)", repo_id, len(arr["kp"]))
-    return session, arr["kp"], arr["kd"], arr["default_angles"], arr["action_scale"], arr["neutral_token"]
-
-
-# Action-feature prefix for the latent-token interface (see _extract_token_from_action).
-TOKEN_ACTION_PREFIX = "motion_token"  # nosec B105 - feature-key prefix, not a secret
-# Proprio-state prefix for the token interface: the robot echoes the last commanded token
-# here so ``lerobot-rollout`` aggregates it into a 64-D ``observation.state``.
-TOKEN_STATE_PREFIX = "motion_token_state"  # nosec B105 - feature-key prefix, not a secret
-
-
-def token_action_key(i: int) -> str:
-    """Action-dict key for the i-th component of the 64-D SONIC latent token.
-
-    The ``.pos`` suffix is required so the value flows through ``lerobot-rollout``, which
-    only routes ``.pos`` scalar features onto the policy action vector.
-    """
-    return f"{TOKEN_ACTION_PREFIX}.{i}.pos"
-
-
-def token_state_key(i: int) -> str:
-    """Observation key for the i-th component of the 64-D SONIC latent token state."""
-    return f"{TOKEN_STATE_PREFIX}.{i}.pos"
-
-
-# Startup blend duration: over the first control ticks, linearly interpolate every joint
-# from the robot's initial measured pose into the policy's commanded target, so control
-# eases in without a snap on the first command.
-INIT_RAMP_S = 3.0
-
-
-def _extract_token_from_action(action: dict | None) -> np.ndarray | None:
-    """Reassemble a dense (64,) latent token from ``motion_token.{i}`` keys, or None.
-
-    The token-only interface: the caller supplies the 64-D encoder latent directly (e.g. a
-    token-output VLA's action), which the decoder consumes with the encoder bypassed.
-    Requires the full dense token; a partial one is ignored (returns None).
-    """
-    if not action:
-        return None
-    keys = [token_action_key(i) for i in range(TOKEN_DIM)]
-    if any(key not in action for key in keys):
-        return None
-    return np.fromiter((float(action[key]) for key in keys), dtype=np.float32, count=TOKEN_DIM)
-
-
-class SonicDecoder:
-    """Runs the SONIC decoder ONNX model and owns the proprioception history.
-
-    Each tick it appends the latest robot state to 10-frame history buffers, then maps the
-    supplied 64-D ``token`` + that history to a residual action added onto ``default_angles``.
-    The encoder is bypassed entirely (token supplied by the policy). ``default_angles`` and
-    ``action_scale`` are (29,) float32 in IsaacLab order, loaded from the checkpoint.
-    """
-
-    def __init__(self, decoder, default_angles, action_scale):
-        self.decoder = decoder
-        self.decoder_input = decoder.get_inputs()[0].name
-        self.default_angles = np.asarray(default_angles, np.float32)
-        self.action_scale = np.asarray(action_scale, np.float32)
-        self.default_angles_mj = self.default_angles[MUJOCO_TO_ISAACLAB]
-        self.token = np.zeros(TOKEN_DIM, np.float32)
-        self.last_action_mj = np.zeros(29, np.float32)
-        self.h_q_mj = [np.zeros(29, np.float32)] * 10
-        self.h_dq_mj = [np.zeros(29, np.float32)] * 10
-        self.h_ang = [np.zeros(3, np.float32)] * 10
-        self.h_act_mj = [np.zeros(29, np.float32)] * 10
-        self.h_quat = [np.array([1, 0, 0, 0], np.float32)] * 10
-
-    def reset(self):
-        """Clear the token and 10-frame proprioception history.
-
-        ``UnitreeG1.reset()`` relies on this so the first decoder outputs of a new episode
-        are not contaminated by the previous episode's state.
-        """
-        self.token = np.zeros(TOKEN_DIM, np.float32)
-        self.last_action_mj = np.zeros(29, np.float32)
-        self.h_q_mj = [np.zeros(29, np.float32)] * 10
-        self.h_dq_mj = [np.zeros(29, np.float32)] * 10
-        self.h_ang = [np.zeros(3, np.float32)] * 10
-        self.h_act_mj = [np.zeros(29, np.float32)] * 10
-        self.h_quat = [np.array([1, 0, 0, 0], np.float32)] * 10
-
-    def update_history(self, q, dq, ang, quat):
-        """Push the latest proprioception (pos/vel/gyro/orientation) into the 10-frame buffers."""
-        quat = quat / (np.linalg.norm(quat) + 1e-8)
-        # Reorder IsaacLab-order state into the MuJoCo order the decoder consumes. This
-        # permutation direction is validated against the deployed SONIC ONNX; don't flip it.
-        q_mj = q[MUJOCO_TO_ISAACLAB]
-        dq_mj = dq[MUJOCO_TO_ISAACLAB]
-        self.h_q_mj = [q_mj - self.default_angles_mj] + self.h_q_mj[:-1]
-        self.h_dq_mj = [dq_mj] + self.h_dq_mj[:-1]
-        self.h_ang = [ang.copy()] + self.h_ang[:-1]
-        self.h_act_mj = [self.last_action_mj.copy()] + self.h_act_mj[:-1]
-        self.h_quat = [quat.copy()] + self.h_quat[:-1]
-
-    def build_decoder_obs(self):
-        """Assemble the 994-D decoder input: token + 10-frame proprioception history + gravity."""
-        obs = np.zeros(994, np.float32)
-        off = 0
-        obs[off : off + 64] = self.token
-        off += 64
-        for h, sz in [
-            (list(reversed(self.h_ang)), 3),
-            (list(reversed(self.h_q_mj)), 29),
-            (list(reversed(self.h_dq_mj)), 29),
-            (list(reversed(self.h_act_mj)), 29),
-        ]:
-            for f in range(10):
-                obs[off : off + sz] = h[f]
-                off += sz
-        for q in reversed(self.h_quat):
-            obs[off : off + 3] = get_gravity_orientation(q)
-            off += 3
-        assert off == 994, f"Decoder obs mismatch: {off}"
-        return obs
-
-    def step(self, robot_obs, token, debug=False):
-        """One control tick: read robot obs, decode the supplied token -> joint targets.
-
-        Args:
-            robot_obs: dict with ``<joint>.q``/``.dq`` and ``imu.*`` fields.
-            token: 64-D latent supplied by the policy (encoder bypassed).
-            debug: log action/delta norms.
-
-        Returns:
-            dict of ``<joint>.q`` target positions (rad) in IsaacLab joint order.
-        """
-        self.token = np.asarray(token, np.float32)
-        jnames = [m.name for m in G1_29_JointIndex]
-        q = np.array(
-            [
-                robot_obs.get(f"{n}.q", self.default_angles[m.value])
-                for m, n in zip(G1_29_JointIndex, jnames, strict=False)
-            ],
-            np.float32,
-        )
-        dq = np.array([robot_obs.get(f"{n}.dq", 0.0) for n in jnames], np.float32)
-        quat = np.array(
-            [
-                robot_obs.get("imu.quat.w", 1),
-                robot_obs.get("imu.quat.x", 0),
-                robot_obs.get("imu.quat.y", 0),
-                robot_obs.get("imu.quat.z", 0),
-            ],
-            np.float32,
-        )
-        ang = np.array([robot_obs.get(f"imu.gyro.{a}", 0) for a in "xyz"], np.float32)
-        self.update_history(q, dq, ang, quat)
-        action_mj = (
-            self.decoder.run(None, {self.decoder_input: self.build_decoder_obs().reshape(1, -1)})[0]
-            .squeeze()
-            .astype(np.float32)
-        )
-        self.last_action_mj = action_mj.copy()
-        target = self.default_angles + action_mj[ISAACLAB_TO_MUJOCO] * self.action_scale
-        if debug:
-            delta = target - q
-            logger.debug(
-                "token_norm=%.4f action_norm=%.4f delta_max=%.4f delta_rms=%.4f",
-                np.linalg.norm(self.token),
-                np.linalg.norm(action_mj),
-                np.max(np.abs(delta)),
-                np.sqrt(np.mean(delta**2)),
-            )
-        return {f"{m.name}.q": float(target[m.value]) for m in G1_29_JointIndex}
-
-
-class SonicRuntime:
-    """Loads the SONIC decoder ONNX model and owns the decode controller.
-
-    Token-only deploy: the encoder is bypassed; each tick the decoder consumes a 64-D
-    latent token supplied directly by the policy.
-    """
-
-    def __init__(self):
-        decoder_sess, self.kp, self.kd, default_angles, action_scale, neutral_token = load_sonic_decoder()
-        self.default_angles = default_angles
-        self.neutral_token = neutral_token
-        self.controller = SonicDecoder(decoder_sess, default_angles, action_scale)
-
-    @property
-    def pipeline(self):
-        return self.controller
-
-    def reset(self):
-        self.controller.reset()
-
-    def shutdown(self):
-        pass
-
-
-class SonicWholeBodyController:
-    """Full-body SONIC controller for UnitreeG1's background controller thread."""
-
-    control_dt = CONTROL_DT
-    full_body = True
-
-    def __init__(self):
-        logger.info("Loading SONIC whole-body controller...")
-        self._runtime = SonicRuntime()
-        self.kp = self._runtime.kp
-        self.kd = self._runtime.kd
-        self.controller = self._runtime.controller
-        self._default_angles = self._runtime.default_angles
-        self._neutral_token = self._runtime.neutral_token
-
-        # Startup blend: ease from the robot's initial pose into the first commanded policy
-        # targets over INIT_RAMP_S (captured on the first control tick).
-        self._init_ramp_steps = max(1, round(INIT_RAMP_S / CONTROL_DT))
-        self._init_step = 0
-        self._start_pose: dict[str, float] = {}
-
-        # Token-interface state. The controller holds a stable *neutral* token until the first
-        # real token arrives, and afterwards holds the *last* token received between ticks (the
-        # async controller runs ~50 Hz while a token VLA streams ~30 Hz).
-        self._last_token: np.ndarray | None = None
-
-        logger.info("SONIC ready (decoder, 64-D token command path)")
-
-    def _startup_blend(self, obs: dict, out: dict) -> dict:
-        """Ease into policy control at startup: for the first ``INIT_RAMP_S`` seconds,
-        interpolate between the robot's pose captured on the first tick and the policy's
-        live commanded target, so the handoff has no snap.
-
-        ``out`` is the policy's ``<joint>.q`` target dict for this tick; the blend ratio
-        climbs 0->1 over the ramp, after which the raw policy target passes through.
-        """
-        if self._init_step >= self._init_ramp_steps or not out:
-            return out
-        if self._init_step == 0:
-            # Capture the robot's actual pose as the interpolation start point.
-            self._start_pose = {
-                f"{m.name}.q": float(obs.get(f"{m.name}.q", self._default_angles[m.value]))
-                for m in G1_29_JointIndex
-            }
-        self._init_step += 1
-        ratio = min(1.0, self._init_step / self._init_ramp_steps)
-        blended = {
-            k: self._start_pose.get(k, float(tgt)) * (1.0 - ratio) + float(tgt) * ratio
-            for k, tgt in out.items()
-        }
-        if self._init_step >= self._init_ramp_steps:
-            logger.info("SONIC startup blend complete -> full policy control")
-        return blended
-
-    def run_step(self, action: dict, obs: dict) -> dict:
-        if not obs:
-            return {}
-
-        # Token-only interface (token-output VLA): a dense 64-D ``motion_token.{i}`` command
-        # is decoded directly, encoder bypassed.
-        token = _extract_token_from_action(action)
-        if token is not None:
-            self._last_token = token
-        elif self._last_token is None:
-            # No token has arrived yet: hold the checkpoint's neutral token, which the decoder
-            # maps to a stable, natural standing pose.
-            self._last_token = self._neutral_token.copy()
-        # Either a fresh token this tick or the last one received (held between the ~30 Hz
-        # token stream and the ~50 Hz control loop).
-        return self._startup_blend(obs, self.controller.step(obs, self._last_token))
-
-    def reset(self):
-        self._runtime.reset()
-        self._init_step = 0  # re-run the startup blend after a reset
-        self._start_pose = {}
-        # Drop the held token so the neutral token is re-seeded after a reset.
-        self._last_token = None
-
-    def shutdown(self):
-        self._runtime.shutdown()

src/lerobot/robots/unitree_g1/g1_utils.py

@@ -23,47 +23,6 @@ import numpy as np
 
 NUM_MOTORS = 29
 
-# Joint-order permutations between the two 29-DoF layouts used across the G1 stack:
-# IsaacLab (policy/training order) and MuJoCo (deploy order). ``a[ISAACLAB_TO_MUJOCO]``
-# reorders an IsaacLab-ordered vector into MuJoCo order, and vice-versa.
-ISAACLAB_TO_MUJOCO = np.array(
-    [
-        0,
-        3,
-        6,
-        9,
-        13,
-        17,
-        1,
-        4,
-        7,
-        10,
-        14,
-        18,
-        2,
-        5,
-        8,
-        11,
-        15,
-        19,
-        21,
-        23,
-        25,
-        27,
-        12,
-        16,
-        20,
-        22,
-        24,
-        26,
-        28,
-    ],
-    dtype=np.int32,
-)
-# The two orderings are inverses of each other, so derive one from the other (argsort) to
-# guarantee they can never drift out of sync.
-MUJOCO_TO_ISAACLAB = np.argsort(ISAACLAB_TO_MUJOCO).astype(np.int32)
-
 REMOTE_AXES = ("remote.lx", "remote.ly", "remote.rx", "remote.ry")
 REMOTE_BUTTONS = tuple(f"remote.button.{i}" for i in range(16))
 REMOTE_KEYS = REMOTE_AXES + REMOTE_BUTTONS
@@ -109,9 +68,8 @@ def make_locomotion_controller(name: str | None):
     if name is None:
         return None
     controllers = {
-        "GrootLocomotionController": "lerobot.robots.unitree_g1.controllers.gr00t_locomotion",
-        "HolosomaLocomotionController": "lerobot.robots.unitree_g1.controllers.holosoma_locomotion",
-        "SonicWholeBodyController": "lerobot.robots.unitree_g1.controllers.sonic_whole_body",
+        "GrootLocomotionController": "lerobot.robots.unitree_g1.gr00t_locomotion",
+        "HolosomaLocomotionController": "lerobot.robots.unitree_g1.holosoma_locomotion",
     }
     module_path = controllers.get(name)
     if module_path is None:

src/lerobot/robots/unitree_g1/gr00t_locomotion.py

@@ -21,7 +21,7 @@ import numpy as np
 import onnxruntime as ort
 from huggingface_hub import hf_hub_download
 
-from ..g1_utils import (
+from .g1_utils import (
     REMOTE_AXES,
     REMOTE_BUTTONS,
     G1_29_JointIndex,

src/lerobot/robots/unitree_g1/holosoma_locomotion.py

@@ -22,7 +22,7 @@ import onnx
 import onnxruntime as ort
 from huggingface_hub import hf_hub_download
 
-from ..g1_utils import (
+from .g1_utils import (
     REMOTE_AXES,
     G1_29_JointArmIndex,
     G1_29_JointIndex,

src/lerobot/robots/unitree_g1/unitree_g1.py

@@ -34,6 +34,7 @@ from .config_unitree_g1 import UnitreeG1Config
 from .g1_kinematics import G1_29_ArmIK
 from .g1_utils import (
     REMOTE_AXES,
+    REMOTE_KEYS,
     G1_29_JointArmIndex,
     G1_29_JointIndex,
     default_remote_input,
@@ -147,51 +148,22 @@ class UnitreeG1(Robot):
 
         self.arm_ik = G1_29_ArmIK() if config.gravity_compensation else None
 
-        # Lower-body / whole-body controller loaded dynamically
+        # Lower-body controller loaded dynamically
         self.controller: LocomotionController | None = make_locomotion_controller(config.controller)
+
         # Controller thread state
         self._controller_thread = None
         self._controller_action_lock = threading.Lock()
         self.controller_input = default_remote_input()
         self.controller_output = {}
 
-        # Token-mode state: last 64-D SONIC latent token commanded by the policy,
-        # echoed back as ``observation.state`` so a token-output VLA closes the loop
-        # on its own previous token. Implicit whenever the SONIC whole-body controller
-        # is active. Seeded to zeros; the controller's startup blend eases joints in.
-        self._last_token: np.ndarray | None = None
-        if self._sonic_token:
-            from .controllers.sonic_whole_body import TOKEN_DIM
-
-            self._last_token = np.zeros(TOKEN_DIM, dtype=np.float32)
-
-    @property
-    def _sonic_token(self) -> bool:
-        """Whether the SONIC whole-body decoder is active.
-
-        A SONIC controller consumes a 64-D latent motion token as its action and echoes
-        the last commanded token as ``observation.state``. Keyed purely off the selected
-        controller so the token interface is implicit -- no separate config flag.
-        """
-        return self.config.controller == "SonicWholeBodyController"
-
     def _subscribe_lowstate(self):  # polls robot state @ 250Hz
         while not self._shutdown_event.is_set():
             start_time = time.time()
 
             # Step simulation if in simulation mode
             if self.config.is_simulation and self.sim_env is not None:
-                try:
                 self.sim_env.step()
-                except ValueError as e:
-                    # Startup race: the sim thread can step once before reset() has
-                    # written a valid base pose, giving a zero-norm pelvis quaternion
-                    # (scipy>=1.11 raises instead of normalizing). Skip and retry so
-                    # the thread survives instead of dying and freezing the sim.
-                    if "zero norm" not in str(e).lower():
-                        raise
-                    time.sleep(self.control_dt)
-                    continue
 
             msg = self.lowstate_subscriber.Read()
             if msg is not None:
@@ -259,38 +231,15 @@ class UnitreeG1(Robot):
                 features[f"{cam}_depth"] = (cfg.height, cfg.width, 1)
         return features
 
-    @property
-    def _token_state_ft(self) -> dict[str, type]:
-        """64-D SONIC latent-token proprio state (``motion_token_state.{i}.pos``).
-
-        Exposed only when a SONIC whole-body controller is active; aggregated by the
-        rollout into a 64-D ``observation.state`` (the last token the policy commanded).
-        """
-        if not self._sonic_token:
-            return {}
-        from .controllers.sonic_whole_body import TOKEN_DIM, token_state_key
-
-        return {token_state_key(i): float for i in range(TOKEN_DIM)}
-
     @cached_property
     def observation_features(self) -> dict[str, type | tuple]:
-        return {**self._motors_ft, **self._token_state_ft, **self._cameras_ft}
+        return {**self._motors_ft, **self._cameras_ft}
 
     @cached_property
     def action_features(self) -> dict[str, type]:
-        # No controller configured at all: raw 29-DoF joint teleop.
         if self.controller is None:
             return {f"{G1_29_JointIndex(motor).name}.q": float for motor in G1_29_JointIndex}
 
-        # Token-output VLA (SONIC decoder): advertise a 64-D latent-token action space
-        # (``motion_token.{i}.pos``) so ``lerobot-rollout`` maps a 64-D policy output
-        # straight onto the decoder, bypassing the encoder.
-        if self._sonic_token:
-            from .controllers.sonic_whole_body import TOKEN_DIM, token_action_key
-
-            return {token_action_key(i): float for i in range(TOKEN_DIM)}
-
-        # Locomotion controllers (GR00T / Holosoma): arm joint targets + joystick axes.
         arm_features = {f"{G1_29_JointArmIndex(motor).name}.q": float for motor in G1_29_JointArmIndex}
         remote_features = dict.fromkeys(REMOTE_AXES, float)
         return {**arm_features, **remote_features}
@@ -322,12 +271,8 @@ class UnitreeG1(Robot):
                 with self._controller_action_lock:
                     controller_input = dict(self.controller_input)
 
-                # Full-body controllers (SONIC) consume the full observation dict; others
-                # take the raw lowstate. get_observation() is the single lowstate -> obs builder.
-                controller_state = (
-                    self.get_observation() if getattr(self.controller, "full_body", False) else lowstate
-                )
-                controller_action = self.controller.run_step(controller_input, controller_state)
+                # Run controller step
+                controller_action = self.controller.run_step(controller_input, lowstate)
 
                 # Write controller output snapshot
                 with self._controller_action_lock:
@@ -398,9 +343,6 @@ class UnitreeG1(Robot):
 
         self.kp = np.array(self.config.kp, dtype=np.float32)
         self.kd = np.array(self.config.kd, dtype=np.float32)
-        if self.controller is not None and hasattr(self.controller, "kp"):
-            self.kp = np.array(self.controller.kp, dtype=np.float32)
-            self.kd = np.array(self.controller.kd, dtype=np.float32)
 
         for joint in G1_29_JointIndex:
             self.msg.motor_cmd[joint].mode = 1
@@ -449,10 +391,6 @@ class UnitreeG1(Robot):
             if self._controller_thread.is_alive():
                 logger.warning("Controller thread did not stop cleanly")
 
-        # Release controller resources (e.g. SONIC decoder sessions).
-        if self.controller is not None and hasattr(self.controller, "shutdown"):
-            self.controller.shutdown()
-
         # Close simulation environment
         if self.config.is_simulation and self.sim_env is not None:
             try:
@@ -523,15 +461,6 @@ class UnitreeG1(Robot):
         if lowstate.wireless_remote:
             obs["wireless_remote"] = lowstate.wireless_remote
 
-        # Token mode: echo the last commanded latent token as observation.state so a
-        # token-output VLA closes the loop on its own previous token.
-        if self._sonic_token:
-            from .controllers.sonic_whole_body import token_state_key
-
-            token = self._last_token if self._last_token is not None else []
-            for i, v in enumerate(token):
-                obs[token_state_key(i)] = float(v)
-
         # Cameras - read images from ZMQ cameras
         for cam_name, cam in self._cameras.items():
             if getattr(cam, "use_rgb", True):
@@ -544,22 +473,9 @@ class UnitreeG1(Robot):
     def send_action(self, action: RobotAction) -> RobotAction:
         action_to_publish = action
         if self.controller is not None:
-            # SONIC decoder: pull the 64-D latent token out of the action and remember it
-            # for the observation.state echo. The controller thread reads it back from
-            # controller_input (populated below) and decodes it into a 29-DoF command.
-            if self._sonic_token:
-                from .controllers.sonic_whole_body import _extract_token_from_action
-
-                token = _extract_token_from_action(action)
-                if token is not None:
-                    self._last_token = token
-            self._update_controller_action(action)
-            # Full-body controllers (SONIC) own the whole 29-DoF command; nothing to
-            # publish here (the controller thread is the sole publisher).
-            if getattr(self.controller, "full_body", False):
-                return action
             # Controller thread owns legs/waist. Here we only update joystick inputs
             # and publish arm targets from the teleoperator.
+            self._update_controller_action(action)
             arm_prefixes = tuple(j.name for j in G1_29_JointArmIndex)
             action_to_publish = {
                 key: value
@@ -587,17 +503,11 @@ class UnitreeG1(Robot):
         return action
 
     def _update_controller_action(self, action: RobotAction) -> None:
-        """Update controller input state from an incoming teleop action.
-
-        Controller-agnostic: every value-carrying key (locomotion ``remote.*`` axes or
-        SONIC ``motion_token.*`` values) is forwarded verbatim into ``controller_input``
-        and each controller extracts only the keys it understands. The robot deliberately
-        does not enumerate any controller's key schema here.
-        """
+        """Update controller input state from incoming teleop action."""
         with self._controller_action_lock:
-            for key, value in action.items():
-                if isinstance(key, str) and value is not None:
-                    self.controller_input[key] = value
+            for key in REMOTE_KEYS:
+                if key in action:
+                    self.controller_input[key] = action[key]
 
     @property
     def is_calibrated(self) -> bool:

src/lerobot/robots/utils.py

@@ -21,6 +21,8 @@ from lerobot.utils.import_utils import make_device_from_device_class
 from .config import RobotConfig
 from .robot import Robot
 
+logger = logging.getLogger(__name__)
+
 
 def make_robot_from_config(config: RobotConfig) -> Robot:
     # TODO(Steven): Consider just using the make_device_from_device_class for all types
@@ -118,7 +120,7 @@ def ensure_safe_goal_position(
             }
 
     if warnings_dict:
-        logging.warning(
+        logger.warning(
             "Relative goal position magnitude had to be clamped to be safe.\n"
             f"{pformat(warnings_dict, indent=4)}"
         )

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/runtime/__init__.py

@@ -0,0 +1,38 @@
+# 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.
+
+"""Policy-agnostic runtime for language-conditioned policies.
+
+Adapters registered in :mod:`lerobot.runtime.registry` are served by ``lerobot-rollout --language``.
+"""
+
+from .adapter import BaseLanguageAdapter, GenerationConfig, LanguageDiagnostics
+from .language_runtime import (
+    LanguageConditionedPolicyAdapter,
+    LanguageConditionedRuntime,
+    RuntimeState,
+    Tick,
+    TickClock,
+)
+
+__all__ = [
+    "BaseLanguageAdapter",
+    "GenerationConfig",
+    "LanguageConditionedPolicyAdapter",
+    "LanguageConditionedRuntime",
+    "LanguageDiagnostics",
+    "RuntimeState",
+    "Tick",
+    "TickClock",
+]

src/lerobot/runtime/adapter.py

@@ -0,0 +1,165 @@
+# 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.
+
+"""Policy adapters for the language runtime.
+
+The base adapter owns generation control and diagnostics while subclasses provide policy-specific actions and text.
+"""
+
+from __future__ import annotations
+
+import re
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any
+
+from .language_runtime import RuntimeState
+
+_SAY_RE = re.compile(r"<\s*say\s*>(.*?)<\s*/\s*say\s*>", re.IGNORECASE | re.DOTALL)
+
+
+@dataclass
+class GenerationConfig:
+    """Text-generation settings fixed for the adapter's lifetime."""
+
+    min_new_tokens: int = 0
+    temperature: float = 0.0
+    top_p: float = 1.0
+    chunks_per_regen: int = 1  # regenerate the language context every N action chunks
+    enable_memory: bool = True  # generate a running memory note on subtask change
+    enable_subtask: bool = True  # generate the low-level subtask (off => use the given text directly)
+
+
+@dataclass
+class LanguageDiagnostics:
+    """Runtime-panel generation counters keyed by text kind."""
+
+    last_raw: dict[str, str] = field(default_factory=dict)
+    empty: dict[str, int] = field(default_factory=dict)
+    repeat: int = 0
+
+    def _bump(self, table: dict[str, int], kind: str) -> int:
+        table[kind] = table.get(kind, 0) + 1
+        return table[kind]
+
+
+class BaseLanguageAdapter(ABC):
+    """Batteries-included adapter: generic high-level control, policy primitives abstract."""
+
+    def __init__(self, policy: Any, gen: GenerationConfig | None = None) -> None:
+        self.policy = policy
+        self.gen = gen or GenerationConfig()
+        self.diag = LanguageDiagnostics()
+        self._chunks_until_regen = 0
+
+    @abstractmethod
+    def select_action(self, observation: dict[str, Any], state: RuntimeState) -> Any:
+        """Produce an action chunk from the observation + current language context."""
+
+    @abstractmethod
+    def generate_text(
+        self,
+        kind: str,
+        observation: dict[str, Any] | None,
+        state: RuntimeState,
+        user_text: str | None = None,
+    ) -> str:
+        """Generate one text stream (``kind``) and return the decoded string."""
+
+    def update_language_state(self, observation: dict[str, Any] | None, state: RuntimeState) -> None:
+        """Throttled regeneration of the language context (subtask / memory / ...)."""
+        if self._chunks_until_regen > 0:
+            self._chunks_until_regen -= 1
+            return
+        self._chunks_until_regen = max(1, self.gen.chunks_per_regen) - 1
+        self._regenerate_context(observation, state)
+
+    def handle_interjection(
+        self, user_text: str, observation: dict[str, Any] | None, state: RuntimeState
+    ) -> None:
+        """React to a mid-run user message by regenerating the plan."""
+        out = self.generate_text("interjection", observation, state, user_text=user_text)
+        plan = self.plan_from_text(out)
+        if plan:
+            state.set_context("plan", plan, label="plan")
+
+    def plan_from_text(self, text: str) -> str:
+        """Strip ``<say>`` speech markers from a generated plan."""
+        plan, _speech = split_plan_and_say(text)
+        return plan
+
+    def _regenerate_context(self, observation: dict[str, Any] | None, state: RuntimeState) -> None:
+        """Default hierarchy: regenerate the subtask, then memory when it changes.
+
+        Override for a policy with a different language hierarchy.
+        """
+        if not self.gen.enable_subtask:
+            # Preserve operator-provided subtasks in direct mode.
+            return
+        subtask = self._generate_filtered("subtask", observation, state)
+        if subtask is None:
+            return
+        previous = state.language_context.get("subtask")
+        if not state.set_context("subtask", subtask, label="subtask"):
+            self.diag.repeat += 1
+            return
+        self.diag.repeat = 0
+        if previous:
+            state.extra["prior_subtask"] = previous
+        if not self.gen.enable_memory:
+            return
+        memory = self._generate_filtered("memory", observation, state)
+        if memory is not None:
+            state.set_context("memory", memory, label="memory")
+
+    def _generate_filtered(
+        self, kind: str, observation: dict[str, Any] | None, state: RuntimeState
+    ) -> str | None:
+        """Generate one ``kind``, record diagnostics, and drop empty output."""
+        text = self.generate_text(kind, observation, state)
+        self.diag.last_raw[kind] = text or ""
+        if not text:
+            count = self.diag._bump(self.diag.empty, kind)
+            if count == 1 or count % 5 == 0:
+                state.log(f"  [info] {kind} gen returned empty (x{count})")
+            return None
+        return text
+
+
+class DirectTaskPolicyAdapter(BaseLanguageAdapter):
+    """Adapter for flat policies whose preprocessors condition actions on the operator's task."""
+
+    def select_action(self, observation: dict[str, Any], state: RuntimeState) -> Any:
+        return self.policy.predict_action_chunk(observation)
+
+    def generate_text(
+        self,
+        kind: str,
+        observation: dict[str, Any] | None,
+        state: RuntimeState,
+        user_text: str | None = None,
+    ) -> str:
+        return ""
+
+
+def split_plan_and_say(text: str) -> tuple[str, str]:
+    """Split ``plan <say>speech</say>`` into ``(plan, speech)``."""
+    if not text:
+        return "", ""
+    match = _SAY_RE.search(text)
+    if not match:
+        return text.strip(), ""
+    speech = match.group(1).strip().strip('"').strip("'")
+    plan = (text[: match.start()] + text[match.end() :]).strip()
+    return plan, speech

src/lerobot/runtime/language_runtime.py

@@ -0,0 +1,349 @@
+# 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.
+
+"""Small reusable runtime for language-conditioned robot policies."""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from collections import deque
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any, Protocol
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class RuntimeState:
+    """Explicit state shared by the runtime and policy adapter."""
+
+    task: str = ""
+    language_context: dict[str, str] = field(default_factory=dict)
+    action_queue: deque[Any] = field(default_factory=deque)
+    events: set[str] = field(default_factory=set)
+    log_lines: list[str] = field(default_factory=list)
+    mode: str = "action"
+    stop: bool = False
+    tick: Tick | None = None
+    actions_dispatched: int = 0
+    action_deadline: float | None = None
+    extra: dict[str, Any] = field(default_factory=dict)
+    revision: int = 0
+    lock: Any = field(default_factory=threading.RLock, repr=False)
+
+    def emit(self, event_name: str) -> None:
+        self.events.add(event_name)
+
+    def take_event(self, event_name: str) -> bool:
+        if event_name not in self.events:
+            return False
+        self.events.remove(event_name)
+        return True
+
+    def log(self, line: str) -> None:
+        self.log_lines.append(line)
+
+    def set_context(self, key: str, value: str | None, *, label: str | None = None) -> bool:
+        with self.lock:
+            previous = self.language_context.get(key)
+            if previous == value:
+                return False
+            if value is None:
+                self.language_context.pop(key, None)
+            else:
+                self.language_context[key] = value
+            self.revision += 1
+            if label is not None and value:
+                self.log(f"  {label}: {value}")
+            return True
+
+    def get(self, key: str, default: Any = None) -> Any:
+        try:
+            return self[key]
+        except KeyError:
+            return default
+
+    def setdefault(self, key: str, default: Any = None) -> Any:
+        current = self.get(key, None)
+        if current is not None:
+            return current
+        self[key] = default
+        return default
+
+    def __getitem__(self, key: str) -> Any:
+        if hasattr(self, key):
+            return getattr(self, key)
+        if key in self.extra:
+            return self.extra[key]
+        raise KeyError(key)
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        with self.lock:
+            if hasattr(self, key):
+                if key == "mode" and self.mode != value:
+                    self.revision += 1
+                setattr(self, key, value)
+            else:
+                self.extra[key] = value
+
+
+class LanguageConditionedPolicyAdapter(Protocol):
+    """Runtime policy contract, implemented directly or through ``BaseLanguageAdapter``."""
+
+    def select_action(self, observation: dict[str, Any], state: RuntimeState) -> Any: ...
+
+    def update_language_state(self, observation: dict[str, Any] | None, state: RuntimeState) -> None: ...
+
+    def handle_interjection(
+        self, user_text: str, observation: dict[str, Any] | None, state: RuntimeState
+    ) -> None: ...
+
+
+@dataclass
+class Tick:
+    index: int
+    monotonic_seconds: float
+
+
+@dataclass
+class TickClock:
+    max_rate_hz: float = 50.0
+    _index: int = field(default=0, init=False)
+    _last_seconds: float | None = field(default=None, init=False)
+
+    def advance(self) -> Tick:
+        period = 1.0 / max(self.max_rate_hz, 0.1)
+        now = time.monotonic()
+        if self._last_seconds is not None:
+            sleep_for = (self._last_seconds + period) - now
+            if sleep_for > 0:
+                time.sleep(sleep_for)
+                now = time.monotonic()
+        self._last_seconds = now
+        self._index += 1
+        return Tick(index=self._index, monotonic_seconds=now)
+
+
+@dataclass
+class _RateGate:
+    hz: float
+    _last_seconds: float | None = None
+
+    def due(self, tick: Tick, *, force: bool = False) -> bool:
+        if force:
+            self._last_seconds = tick.monotonic_seconds
+            return True
+        period = 1.0 / max(self.hz, 1e-6)
+        if self._last_seconds is None or tick.monotonic_seconds - self._last_seconds >= period:
+            self._last_seconds = tick.monotonic_seconds
+            return True
+        return False
+
+    def rearm(self) -> None:
+        self._last_seconds = None
+
+
+@dataclass
+class LanguageConditionedRuntime:
+    """Generic tick loop for language-conditioned robot policies."""
+
+    policy_adapter: LanguageConditionedPolicyAdapter
+    observation_provider: Callable[[], dict[str, Any] | None] | None = None
+    action_executor: Callable[[Any], None] | None = None
+    event_collector: Callable[[RuntimeState], None] | None = None
+    chunk_hz: float = 4.0
+    ctrl_hz: float = 50.0
+    high_level_hz: float = 1.0
+    max_rate_hz: float = 50.0
+
+    state: RuntimeState = field(default_factory=RuntimeState)
+    _chunk_gate: _RateGate = field(init=False)
+    _ctrl_gate: _RateGate = field(init=False)
+    _language_gate: _RateGate = field(init=False)
+    _stop: bool = field(default=False, init=False)
+    _last_dispatch_seconds: float | None = field(default=None, init=False)
+
+    def __post_init__(self) -> None:
+        self._chunk_gate = _RateGate(self.chunk_hz)
+        self._ctrl_gate = _RateGate(self.ctrl_hz)
+        self._language_gate = _RateGate(self.high_level_hz)
+
+    @property
+    def policy(self) -> Any:
+        return getattr(self.policy_adapter, "policy", self.policy_adapter)
+
+    def set_task(self, task: str) -> None:
+        with self.state.lock:
+            if self.state.task != task:
+                self.state.revision += 1
+            self.state.task = task
+            self.state.log(f"Task: {task}")
+
+    def stop(self) -> None:
+        self._stop = True
+        self.state.stop = True
+
+    def run(self, *, max_ticks: int | None = None) -> None:
+        clock = TickClock(max_rate_hz=self.max_rate_hz)
+        while not self._stop:
+            tick = clock.advance()
+            self._run_tick(tick)
+            self._flush_logs()
+            if self.state.stop:
+                self._stop = True
+            if max_ticks is not None and tick.index >= max_ticks:
+                break
+        self._on_shutdown()
+
+    def step_once(self) -> list[str]:
+        previous = self.state.tick.index if self.state.tick is not None else 0
+        tick = Tick(index=previous + 1, monotonic_seconds=time.monotonic())
+        self._run_tick(tick, force_rates=True)
+        return list(self.state.log_lines)
+
+    def _run_tick(self, tick: Tick, *, force_rates: bool = False) -> None:
+        self.state.tick = tick
+        self.state.log_lines = []
+        if self.event_collector is not None:
+            self.event_collector(self.state)
+        self._handle_action_deadline()
+        if self.state.stop:
+            return
+        self.maybe_update_language_state(force=force_rates)
+        self.maybe_handle_user_events()
+        self.maybe_enqueue_action_chunk(force=force_rates)
+        self.dispatch_action(force=force_rates)
+        self.state.events.clear()
+
+    def _current_observation(self) -> dict[str, Any] | None:
+        if self.observation_provider is None:
+            return None
+        try:
+            return self.observation_provider()
+        except Exception as exc:  # noqa: BLE001
+            logger.debug("observation_provider failed: %s", exc)
+            return None
+
+    def maybe_update_language_state(self, *, force: bool = False) -> None:
+        if self.state.mode != "action" or not self.state.task:
+            return
+        if self.state.action_queue:
+            self._language_gate.rearm()
+            return
+        if self.state.tick is None or not self._language_gate.due(self.state.tick, force=force):
+            return
+        observation = self._current_observation()
+        try:
+            self.policy_adapter.update_language_state(observation, self.state)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("language update failed: %s", exc, exc_info=logger.isEnabledFor(logging.DEBUG))
+            self.state.log(f"  [warn] language update failed: {type(exc).__name__}: {exc}")
+
+    def maybe_handle_user_events(self) -> None:
+        if self.state.take_event("user_interjection"):
+            self._handle_user_interjection()
+
+    def _handle_user_interjection(self) -> None:
+        text = str(self.state.extra.get("recent_interjection") or "")
+        if not text:
+            return
+        observation = self._current_observation()
+        self.policy_adapter.handle_interjection(text, observation, self.state)
+        self.state.extra["recent_interjection"] = None
+
+    def maybe_enqueue_action_chunk(self, *, force: bool = False) -> None:
+        with self.state.lock:
+            if self.state.mode != "action" or not self.state.task:
+                return
+            if self.state.action_queue:
+                return
+            if self.state.tick is None or not self._chunk_gate.due(self.state.tick, force=force):
+                return
+            revision = self.state.revision
+        observation = self._current_observation()
+        if observation is None:
+            return
+        try:
+            chunk = self.policy_adapter.select_action(observation, self.state)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("select_action failed: %s", exc, exc_info=logger.isEnabledFor(logging.DEBUG))
+            self.state.log(f"  [warn] select_action failed: {type(exc).__name__}: {exc}")
+            return
+        with self.state.lock:
+            if (
+                self.state.revision != revision
+                or self.state.mode != "action"
+                or self.state.stop
+                or self._stop
+            ):
+                logger.info("Discarded an action chunk invalidated during inference.")
+                return
+            self._enqueue_chunk(chunk)
+
+    def _enqueue_chunk(self, chunk: Any) -> None:
+        if chunk is None:
+            return
+        chunk_iter = chunk[0] if getattr(chunk, "ndim", None) == 3 else chunk
+        if getattr(chunk_iter, "ndim", None) == 1:
+            chunk_iter = chunk_iter.unsqueeze(0)
+        for step in chunk_iter:
+            self.state.action_queue.append(step.unsqueeze(0) if hasattr(step, "unsqueeze") else step)
+        try:
+            self.state.extra["last_chunk_size"] = int(chunk_iter.shape[0])
+        except Exception:  # noqa: BLE001
+            self.state.extra["last_chunk_size"] = len(self.state.action_queue)
+
+    def dispatch_action(self, *, force: bool = False) -> None:
+        if self.state.mode != "action":
+            self._last_dispatch_seconds = None
+            return
+        if self.state.tick is None or not self._ctrl_gate.due(self.state.tick, force=force):
+            return
+        queue = self.state.action_queue
+        if not queue:
+            self._last_dispatch_seconds = None
+            return
+        now = time.monotonic()
+        if self._last_dispatch_seconds is None or self.ctrl_hz <= 0:
+            n_to_pop = 1
+        else:
+            n_to_pop = max(1, min(len(queue), int(round((now - self._last_dispatch_seconds) * self.ctrl_hz))))
+        self._last_dispatch_seconds = now
+        latest = None
+        for _ in range(n_to_pop):
+            if not queue:
+                break
+            latest = queue.popleft()
+            self.state.actions_dispatched += 1
+        if latest is not None and self.action_executor is not None:
+            self.action_executor(latest)
+
+    def _handle_action_deadline(self) -> None:
+        deadline = self.state.action_deadline
+        if self.state.mode == "action" and deadline is not None and time.monotonic() >= deadline:
+            self.state.mode = "paused"
+            self.state.action_deadline = None
+            self.state.action_queue.clear()
+            self.state.log("timed action elapsed — paused")
+
+    def _flush_logs(self) -> None:
+        for line in self.state.log_lines:
+            print(f"[runtime] {line}", flush=True)
+
+    def _on_shutdown(self) -> None:
+        self.state.action_queue.clear()
+        print("[runtime] stopped", flush=True)

src/lerobot/runtime/registry.py

@@ -0,0 +1,39 @@
+# 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.
+
+"""Lazy mapping from policy types to language-runtime adapters."""
+
+from __future__ import annotations
+
+import importlib
+from collections.abc import Callable
+from typing import Any
+
+_ADAPTERS: dict[str, str] = {
+    "pi052": "lerobot.policies.pi052.inference.pi052_adapter:PI052PolicyAdapter",
+    "pi05": "lerobot.runtime.adapter:DirectTaskPolicyAdapter",
+    "molmoact2": "lerobot.runtime.adapter:DirectTaskPolicyAdapter",
+}
+
+
+def get_language_adapter_factory(policy_type: str) -> Callable[..., Any]:
+    """Return the adapter class registered for ``policy_type``."""
+    spec = _ADAPTERS.get(policy_type)
+    if spec is None:
+        raise ValueError(
+            f"No language-runtime adapter registered for policy type {policy_type!r}. "
+            f"Registered: {sorted(_ADAPTERS)}. Add an entry to lerobot.runtime.registry."
+        )
+    module_path, class_name = spec.split(":")
+    return getattr(importlib.import_module(module_path), class_name)

src/lerobot/runtime/sim_robocasa.py

@@ -0,0 +1,406 @@
+# 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.
+
+"""RoboCasa backend for interactive language-conditioned rollouts.
+
+It reuses the eval observation/action pipeline while prompts control a persistent selected scene.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import torch
+
+from lerobot.utils.io_utils import StreamingVideoWriter
+from lerobot.utils.video_annotation import annotate_frame
+
+logger = logging.getLogger(__name__)
+
+
+def _short_cam_name(cam: str) -> str:
+    """Human-friendly view label for a RoboCasa camera name."""
+    c = cam.replace("robot0_", "")
+    return {
+        "agentview_left": "left",
+        "agentview_right": "right",
+        "eye_in_hand": "wrist",
+    }.get(c, c)
+
+
+def _label_panel(img: np.ndarray, label: str) -> np.ndarray:
+    """Draw a small camera-view label in the bottom-left corner of a panel."""
+    try:
+        import cv2  # noqa: PLC0415
+    except ImportError:
+        return img
+    y = img.shape[0] - 6
+    cv2.putText(img, label, (5, y), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 0, 0), 3, cv2.LINE_AA)
+    cv2.putText(img, label, (5, y), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 255, 0), 1, cv2.LINE_AA)
+    return img
+
+
+# Two workers avoid broken single-worker EGL rendering; only env 0 is displayed.
+_SIM_N_ENVS = 2
+
+
+def create_sim_env(
+    *,
+    task: str,
+    split: str | None,
+    obj_registries: list[str],
+    seed: int | None,
+    render_size: int = 384,
+) -> tuple[Any, dict]:
+    """Create and reset the vectorized RoboCasa environment before CUDA initializes.
+
+    Two workers keep EGL stable, while only env 0 is driven and displayed.
+    """
+    from lerobot.envs.configs import RoboCasaEnv as RoboCasaEnvConfig  # noqa: PLC0415
+
+    # The policy resizes inputs, so render_size only affects display quality and cost.
+    env_cfg = RoboCasaEnvConfig(
+        task=task,
+        split=split,
+        obj_registries=list(obj_registries),
+        observation_height=render_size,
+        observation_width=render_size,
+    )
+    # Keep one kitchen alive across sequential prompts.
+    envs = env_cfg.create_envs(
+        n_envs=_SIM_N_ENVS,
+        use_async_envs=True,
+        terminate_on_success=False,
+        horizon=100_000,
+    )
+    env = envs[next(iter(envs))][0]
+    logger.info("[sim] resetting RoboCasa scene task=%r split=%r (n_envs=%d)", task, split, _SIM_N_ENVS)
+    seeds = None if seed is None else [seed + i for i in range(_SIM_N_ENVS)]
+    obs, _ = env.reset(seed=seeds)
+    return env, obs
+
+
+def start_mjpeg_server(port: int, get_frame: Callable[[], np.ndarray | None]) -> Any:
+    """Start an MJPEG server that shows a placeholder until ``get_frame`` returns frames."""
+    import io  # noqa: PLC0415
+    import threading  # noqa: PLC0415
+    import time  # noqa: PLC0415
+    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer  # noqa: PLC0415
+
+    from PIL import Image  # noqa: PLC0415
+
+    _placeholder = Image.new("RGB", (256, 256), (17, 17, 17))
+
+    class _Handler(BaseHTTPRequestHandler):
+        def log_message(self, *args):  # silence per-request logging
+            pass
+
+        def do_GET(self):  # noqa: N802
+            if self.path in ("/", "/index.html"):
+                self.send_response(200)
+                self.send_header("Content-Type", "text/html")
+                self.end_headers()
+                self.wfile.write(
+                    b"<html><body style='margin:0;background:#111;text-align:center'>"
+                    b"<img src='/stream' style='max-width:100vw;max-height:100vh;"
+                    b"image-rendering:pixelated'></body></html>"
+                )
+                return
+            if self.path != "/stream":
+                self.send_response(404)
+                self.end_headers()
+                return
+            self.send_response(200)
+            self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=frame")
+            self.end_headers()
+            try:
+                while True:
+                    frame = get_frame()
+                    buf = io.BytesIO()
+                    img = Image.fromarray(frame) if frame is not None else _placeholder
+                    img.save(buf, format="JPEG", quality=80)
+                    data = buf.getvalue()
+                    self.wfile.write(
+                        b"--frame\r\nContent-Type: image/jpeg\r\nContent-Length: "
+                        + str(len(data)).encode()
+                        + b"\r\n\r\n"
+                        + data
+                        + b"\r\n"
+                    )
+                    time.sleep(0.05)
+            except (BrokenPipeError, ConnectionResetError):
+                pass
+
+    try:
+        # Bind all interfaces intentionally so the viewer remains reachable
+        # through the documented SSH port-forwarding workflow.
+        server = ThreadingHTTPServer(("0.0.0.0", port), _Handler)  # nosec B104
+    except OSError as exc:
+        logger.warning("[sim] could not start live stream on port %d: %s", port, exc)
+        print(f"[runtime] WARNING: live stream port {port} unavailable ({exc})", flush=True)
+        return None
+    threading.Thread(target=server.serve_forever, daemon=True, name="sim-mjpeg").start()
+    print(
+        f"[runtime] live view: http://localhost:{port}  "
+        f"(over SSH: ssh -L {port}:localhost:{port} <host>) — loading until scene is ready",
+        flush=True,
+    )
+    return server
+
+
+class RoboCasaSimBackend:
+    """Expose a RoboCasa environment through the runtime observation/action contract.
+
+    The environment must be created before the policy initializes CUDA.
+    """
+
+    def __init__(
+        self,
+        *,
+        env: Any,
+        last_obs: dict,
+        task: str,
+        seed: int | None,
+        device: str,
+        preprocessor: Any,
+        postprocessor: Any,
+        record: bool = True,
+        output_dir: str | None = None,
+        view_cams: list[str] | None = None,
+    ) -> None:
+        self.env = env
+        self._last_obs = last_obs
+        self._scene_task = task
+        self._view_cams = view_cams or [
+            "robot0_agentview_left",
+            "robot0_eye_in_hand",
+            "robot0_agentview_right",
+        ]
+        self.device = torch.device(device) if isinstance(device, str) else device
+        self.preprocessor = preprocessor
+        self.postprocessor = postprocessor
+        self.seed = seed
+        self.record = record
+        self.output_dir = Path(output_dir) if output_dir else Path("outputs/runtime_sim")
+
+        self._video_writer: StreamingVideoWriter | None = None
+        self._video_path: Path | None = None
+        self._live_counter = 0
+        self._latest_frame: np.ndarray | None = None
+        self._stream_server: Any = None
+        self._reset_count = 0
+        # Bind these after runtime construction for live annotations.
+        self._task_getter: Callable[[], str | None] | None = None
+        self._subtask_getter: Callable[[], str | None] | None = None
+        self._memory_getter: Callable[[], str | None] | None = None
+        logger.info("[sim] scene ready — task_description=%r", self._scene_description())
+
+    def bind_runtime(self, runtime: Any) -> None:
+        """Wire live task/subtask/memory getters from the runtime state."""
+        self._task_getter = lambda: runtime.state.get("task")
+        self._subtask_getter = lambda: runtime.state.language_context.get("subtask")
+        self._memory_getter = lambda: (runtime.state.get("language_context") or {}).get("memory")
+
+    def _scene_description(self) -> str:
+        try:
+            return str(self.env.get_attr("task_description")[0]) or self._scene_task
+        except Exception:  # noqa: BLE001
+            return self._scene_task
+
+    def _current_task(self) -> str:
+        task = self._task_getter() if self._task_getter else None
+        return task or self._scene_description() or self._scene_task
+
+    def reset_scene(self) -> None:
+        """Re-roll the kitchen: reset the env to a fresh scene (new layout/style).
+
+        Uses a new seed each call so ``/reset`` explores different kitchens.
+        """
+        self._reset_count += 1
+        n = self.env.num_envs
+        if self.seed is None:
+            seeds = None
+        else:
+            base = self.seed + self._reset_count * 1000
+            seeds = [base + i for i in range(n)]
+        obs, _ = self.env.reset(seed=seeds)
+        self._last_obs = obs
+        logger.info("[sim] scene reset (#%d)", self._reset_count)
+
+    def _env0_obs(self) -> dict:
+        """Slice env 0 out of the batched vec-env observation (batch of 1)."""
+        raw = self._last_obs or {}
+        pixels = raw.get("pixels")
+        out: dict[str, Any] = {}
+        if isinstance(pixels, dict):
+            out["pixels"] = {k: np.asarray(v)[0:1] for k, v in pixels.items()}
+        agent_pos = raw.get("agent_pos")
+        if agent_pos is not None:
+            out["agent_pos"] = np.asarray(agent_pos)[0:1]
+        return out
+
+    def observation_provider(self) -> dict | None:
+        from lerobot.envs.utils import preprocess_observation  # noqa: PLC0415
+
+        try:
+            obs = preprocess_observation(self._env0_obs())
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("[sim] preprocess_observation failed: %s", exc)
+            return None
+        # The adapter later replaces this recipe input with its generated subtask.
+        obs["task"] = [self._current_task()]
+        if self.preprocessor is not None:
+            try:
+                obs = self.preprocessor(obs)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("[sim] preprocessor failed: %s", exc)
+                return None
+        return {
+            k: (v.to(self.device) if isinstance(v, torch.Tensor) else v)
+            for k, v in obs.items()
+            if isinstance(k, str) and k.startswith("observation.")
+        }
+
+    def action_executor(self, action: Any) -> None:
+        try:
+            if self.postprocessor is not None:
+                action = self.postprocessor(action)
+            if isinstance(action, torch.Tensor):
+                if action.ndim > 1 and action.shape[0] == 1:
+                    action = action.squeeze(0)
+                action = action.detach().to("cpu").numpy()
+            # Tile env 0's action because the extra workers exist only for EGL stability.
+            action_row = np.asarray(action, dtype=np.float32).reshape(-1)
+            action_np = np.tile(action_row, (self.env.num_envs, 1))
+            obs, _reward, terminated, truncated, _info = self.env.step(action_np)
+            self._last_obs = obs
+            self._capture_frame()
+            # AsyncVectorEnv resets terminated sub-environments automatically.
+            if bool(np.any(terminated)) or bool(np.any(truncated)):
+                logger.info("[sim] episode ended — scene auto-reset")
+        except Exception as exc:  # noqa: BLE001
+            logger.error("[sim] env.step failed: %s", exc, exc_info=True)
+
+    def _multiview_frame(self) -> np.ndarray | None:
+        """Label and compose env 0's existing observation views without extra rendering."""
+        pixels = (self._last_obs or {}).get("pixels")
+        if not isinstance(pixels, dict) or not pixels:
+            return None
+        panels: list[np.ndarray] = []
+        for cam in self._view_cams:
+            v = pixels.get(cam)
+            if v is None:
+                continue
+            img = np.asarray(v)
+            if img.ndim == 4:  # (n_envs, H, W, C) -> env 0
+                img = img[0]
+            if img.ndim != 3 or img.shape[-1] != 3:
+                continue
+            panels.append(_label_panel(np.ascontiguousarray(img.astype(np.uint8)), _short_cam_name(cam)))
+        if not panels:
+            return None
+        h = min(p.shape[0] for p in panels)
+        panels = [p[:h] for p in panels]
+        return np.concatenate(panels, axis=1)
+
+    def _capture_frame(self) -> None:
+        frame = self._multiview_frame()
+        if frame is None:  # fallback to single env.render()
+            try:
+                rendered = self.env.call("render")[0]
+                if isinstance(rendered, np.ndarray) and rendered.ndim == 3:
+                    frame = rendered
+            except Exception as exc:  # noqa: BLE001
+                logger.debug("[sim] render failed: %s", exc)
+        if frame is None:
+            return
+        subtask = self._subtask_getter() if self._subtask_getter else None
+        memory = self._memory_getter() if self._memory_getter else None
+        annotated = annotate_frame(
+            frame,
+            (("Task", self._current_task()), ("Subtask", subtask), ("Memory", memory)),
+        )
+        self._latest_frame = annotated  # served by the live MJPEG stream
+        self._write_live_frame(annotated)
+        if self.record:
+            self._write_recording_frame(annotated)
+
+    def _write_recording_frame(self, frame: np.ndarray) -> None:
+        try:
+            if self._video_writer is None:
+                self.output_dir.mkdir(parents=True, exist_ok=True)
+                stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+                self._video_path = self.output_dir / f"sim_{stamp}.mp4"
+                fps = int((getattr(self.env, "metadata", None) or {}).get("render_fps", 20))
+                self._video_writer = StreamingVideoWriter(self._video_path, fps)
+            self._video_writer.add_frame(frame)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("[sim] video encoding failed: %s", exc)
+            self.record = False
+
+    def _write_live_frame(self, frame: np.ndarray) -> None:
+        """Write a rolling latest.png every few frames for live viewing over SSH.
+
+        Open ``{output_dir}/latest.png`` in an editor/viewer and refresh to watch
+        the rollout in near-real-time without a GUI window. Written atomically
+        (temp + replace) so a reader never sees a half-written file.
+        """
+        self._live_counter += 1
+        if self._live_counter % 3 != 0:
+            return
+        try:
+            import os  # noqa: PLC0415
+
+            from PIL import Image  # noqa: PLC0415
+
+            self.output_dir.mkdir(parents=True, exist_ok=True)
+            tmp = self.output_dir / ".latest.tmp.png"
+            Image.fromarray(frame).save(tmp)
+            os.replace(tmp, self.output_dir / "latest.png")
+        except Exception as exc:  # noqa: BLE001
+            logger.debug("[sim] live frame write failed: %s", exc)
+
+    def _flush_video(self) -> None:
+        if self._video_writer is None:
+            return
+        writer = self._video_writer
+        self._video_writer = None
+        try:
+            writer.close()
+            logger.info("[sim] wrote video (%d frames) to %s", writer.frames_written, self._video_path)
+            print(f"[runtime] sim video saved to {self._video_path}", flush=True)
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("[sim] video close failed: %s", exc)
+
+    def attach_stream_server(self, server: Any) -> None:
+        """Attach an already-running MJPEG server so disconnect() can stop it."""
+        self._stream_server = server
+
+    def disconnect(self) -> None:
+        """Match the robot backend's cleanup contract."""
+        if self._stream_server is not None:
+            try:
+                self._stream_server.shutdown()
+            except Exception as exc:  # noqa: BLE001
+                logger.debug("[sim] stream server shutdown raised %s", exc)
+        self._flush_video()
+        try:
+            self.env.close()
+        except Exception as exc:  # noqa: BLE001
+            logger.debug("[sim] env.close raised %s", exc)

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,21 +87,26 @@ 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 (
     init_logging,
     inside_slurm,
 )
-
-if TYPE_CHECKING or _peft_available:
-    from peft import PeftModel
-else:
-    PeftModel = None
+from lerobot.utils.video_annotation import annotate_frame
 
 
-logger = logging.getLogger(__name__)
+def _annotate_eval_frames(frames: np.ndarray, task: str | None, subtask: str | None) -> np.ndarray:
+    """Overlay the high-level task and predicted subtask onto rendered frames.
+
+    ``frames`` is ``(n_envs, H, W, C)`` uint8. Best-effort: if OpenCV isn't
+    available the frames are returned unchanged so eval never fails over a
+    visualization concern.
+    """
+    if frames.ndim != 4 or frames.shape[-1] != 3:
+        return frames
+    return np.stack([annotate_frame(frame, (("Task", task), ("Subtask", subtask))) for frame in frames])
 
 
 def _env_features_to_dataset_features(env_features: dict) -> dict:
@@ -452,11 +457,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")
+        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
@@ -483,11 +490,36 @@ def eval_policy(
             return
         n_to_render_now = min(max_episodes_rendered - n_episodes_rendered, env.num_envs)
         if isinstance(env, gym.vector.SyncVectorEnv):
-            ep_frames.append(np.stack([env.envs[i].render() for i in range(n_to_render_now)]))  # noqa: B023
+            frames = np.stack([env.envs[i].render() for i in range(n_to_render_now)])  # noqa: B023
         elif hasattr(env, "call"):
             # Here we must render all frames and discard any we don't need.
             # Covers AsyncVectorEnv and _LazyAsyncVectorEnv (which wraps one).
-            ep_frames.append(np.stack(env.call("render")[:n_to_render_now]))
+            frames = np.stack(env.call("render")[:n_to_render_now])
+        else:
+            return
+
+        # Overlay the high-level task and (for hierarchical policies like
+        # pi052) the predicted low-level subtask onto each frame. Both are
+        # best-effort: missing values just skip that line.
+        try:
+            tasks = list(env.call("task_description"))
+        except (AttributeError, NotImplementedError):
+            try:
+                tasks = list(env.call("task"))
+            except (AttributeError, NotImplementedError):
+                tasks = None
+        subtasks = getattr(policy, "last_subtasks", None)
+        annotated = []
+        for i in range(frames.shape[0]):
+            subtask_i = subtasks[i] if subtasks is not None and i < len(subtasks) else None
+            annotated.append(
+                _annotate_eval_frames(
+                    frames[i : i + 1],
+                    tasks[i] if tasks is not None and i < len(tasks) else None,
+                    subtask_i,
+                )[0]
+            )
+        ep_frames.append(np.stack(annotated))
 
     if max_episodes_rendered > 0:
         video_paths: list[str] = []
@@ -564,7 +596,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 +834,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

@@ -28,6 +28,7 @@ lerobot-find-cameras
 # NOTE(Steven): macOS cameras sometimes report different FPS at init time, not an issue here as we don't specify FPS when opening the cameras, but the information displayed might not be truthful.
 
 import argparse
+import concurrent.futures
 import logging
 import time
 from pathlib import Path
@@ -39,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__)
 
@@ -132,7 +132,7 @@ def save_image(
     camera_identifier: str | int,
     images_dir: Path,
     camera_type: str,
-) -> None:
+):
     """
     Saves a single image to disk using Pillow. Handles color conversion if necessary.
     """
@@ -151,7 +151,7 @@ def save_image(
         logger.error(f"Failed to save image for camera {camera_identifier} (type {camera_type}): {e}")
 
 
-def create_camera_instance(cam_meta: dict[str, Any], *, warmup_s: int = 1) -> dict[str, Any] | None:
+def create_camera_instance(cam_meta: dict[str, Any]) -> dict[str, Any] | None:
     """Create and connect to a camera instance based on metadata."""
     cam_type = cam_meta.get("type")
     cam_id = cam_meta.get("id")
@@ -164,14 +164,12 @@ def create_camera_instance(cam_meta: dict[str, Any], *, warmup_s: int = 1) -> di
             cv_config = OpenCVCameraConfig(
                 index_or_path=cam_id,
                 color_mode=ColorMode.RGB,
-                warmup_s=warmup_s,
             )
             instance = OpenCVCamera(cv_config)
         elif cam_type == "RealSense":
             rs_config = RealSenseCameraConfig(
                 serial_number_or_name=cam_id,
                 color_mode=ColorMode.RGB,
-                warmup_s=warmup_s,
             )
             instance = RealSenseCamera(rs_config)
         else:
@@ -189,7 +187,9 @@ def create_camera_instance(cam_meta: dict[str, Any], *, warmup_s: int = 1) -> di
         return None
 
 
-def process_camera_image(cam_dict: dict[str, Any], output_dir: Path, current_time: float) -> None:
+def process_camera_image(
+    cam_dict: dict[str, Any], output_dir: Path, current_time: float
+) -> concurrent.futures.Future | None:
     """Capture and process an image from a single camera."""
     cam = cam_dict["instance"]
     meta = cam_dict["meta"]
@@ -199,7 +199,7 @@ def process_camera_image(cam_dict: dict[str, Any], output_dir: Path, current_tim
     try:
         image_data = cam.read()
 
-        save_image(
+        return save_image(
             image_data,
             cam_id_str,
             output_dir,
@@ -214,9 +214,10 @@ def process_camera_image(cam_dict: dict[str, Any], output_dir: Path, current_tim
     return None
 
 
-def cleanup_camera(cam_dict: dict[str, Any]) -> None:
+def cleanup_cameras(cameras_to_use: list[dict[str, Any]]):
     """Disconnect all cameras."""
-    logger.info(f"Disconnecting camera with ID {cam_dict['meta'].get('id')}...")
+    logger.info(f"Disconnecting {len(cameras_to_use)} cameras...")
+    for cam_dict in cameras_to_use:
         try:
             if cam_dict["instance"] and cam_dict["instance"].is_connected:
                 cam_dict["instance"].disconnect()
@@ -228,7 +229,6 @@ def save_images_from_all_cameras(
     output_dir: Path,
     record_time_s: float = 2.0,
     camera_type: str | None = None,
-    warmup_s: int = 1,
 ):
     """
     Connects to detected cameras (optionally filtered by type) and saves images from each.
@@ -239,7 +239,6 @@ def save_images_from_all_cameras(
         record_time_s: Duration in seconds to record images.
         camera_type: Optional string to filter cameras ("realsense" or "opencv").
                             If None, uses all detected cameras.
-        warmup_s: Duration in seconds to warmup camera before recording images.
     """
     output_dir.mkdir(parents=True, exist_ok=True)
     logger.info(f"Saving images to {output_dir}")
@@ -249,32 +248,47 @@ def save_images_from_all_cameras(
         logger.warning("No cameras detected matching the criteria. Cannot save images.")
         return
 
-    logger.info(
-        f"Starting image capture for {record_time_s} seconds from {len(all_camera_metadata)} cameras."
-    )
-
-    try:
+    cameras_to_use = []
     for cam_meta in all_camera_metadata:
-            cam_dict = create_camera_instance(cam_meta, warmup_s=warmup_s)
-            if cam_dict is None:
-                continue
+        camera_instance = create_camera_instance(cam_meta)
+        if camera_instance:
+            cameras_to_use.append(camera_instance)
+
+    if not cameras_to_use:
+        logger.warning("No cameras could be connected. Aborting image save.")
+        return
+
+    logger.info(f"Starting image capture for {record_time_s} seconds from {len(cameras_to_use)} cameras.")
     start_time = time.perf_counter()
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=len(cameras_to_use) * 2) as executor:
+        try:
             while time.perf_counter() - start_time < record_time_s:
+                futures = []
                 current_capture_time = time.perf_counter()
-                process_camera_image(cam_dict, output_dir, current_capture_time)
-            cleanup_camera(cam_dict)
+
+                for cam_dict in cameras_to_use:
+                    future = process_camera_image(cam_dict, output_dir, current_capture_time)
+                    if future:
+                        futures.append(future)
+
+                if futures:
+                    concurrent.futures.wait(futures)
+
         except KeyboardInterrupt:
             logger.info("Capture interrupted by user.")
         finally:
+            print("\nFinalizing image saving...")
+            executor.shutdown(wait=True)
+            cleanup_cameras(cameras_to_use)
             print(f"Image capture finished. Images saved to {output_dir}")
 
 
 def main():
-    init_logging()
-
     parser = argparse.ArgumentParser(
         description="Unified camera utility script for listing cameras and capturing images."
     )
+
     parser.add_argument(
         "camera_type",
         type=str,
@@ -292,14 +306,8 @@ def main():
     parser.add_argument(
         "--record-time-s",
         type=float,
-        default=2.0,
-        help="Time duration to attempt capturing frames. Default: 2 seconds.",
-    )
-    parser.add_argument(
-        "--warmup-s",
-        type=int,
-        default=1,
-        help="Time duration to warmup camera before attempting to capture frames. Default: 1 second.",
+        default=6.0,
+        help="Time duration to attempt capturing frames. Default: 6 seconds.",
     )
     args = parser.parse_args()
     save_images_from_all_cameras(**vars(args))

src/lerobot/scripts/lerobot_rollout.py

@@ -151,6 +151,7 @@ Usage examples
 """
 
 import logging
+import sys
 
 from lerobot.cameras.opencv import OpenCVCameraConfig  # noqa: F401
 from lerobot.cameras.realsense import RealSenseCameraConfig  # noqa: F401
@@ -165,7 +166,6 @@ from lerobot.robots import (  # noqa: F401
     earthrover_mini_plus,
     hope_jr,
     koch_follower,
-    lekiwi,
     omx_follower,
     openarm_follower,
     reachy2,
@@ -242,10 +242,69 @@ def rollout(cfg: RolloutConfig):
     logger.info("Rollout finished")
 
 
-def main():
-    """CLI entry point for ``lerobot-rollout``."""
+_LANGUAGE_RUNTIME_FLAGS = {
+    "--language",
+    "--no_robot",
+    "--sim",
+    "--direct_subtask",
+    "--sim.direct_subtask",
+    "--disable_memory",
+    "--fp8",
+}
+_LANGUAGE_RUNTIME_PREFIXES = (
+    "--sim.",
+    "--chunk_hz",
+    "--ctrl_hz",
+    "--high_level_hz",
+    "--subtask_chunks_per_gen",
+    "--text_min_new_tokens",
+    "--text_temperature",
+    "--text_top_p",
+)
+
+
+def _uses_language_runtime(argv: list[str]) -> bool:
+    """Return whether *argv* selects the interactive language runtime.
+
+    ``--language`` is the explicit selector for real-robot runs whose other
+    options overlap with the standard rollout CLI. Language-only options also
+    select it automatically, which keeps the former language-runtime examples
+    working after replacing their command name with ``lerobot-rollout``.
+    """
+    return any(
+        arg.split("=", 1)[0] in _LANGUAGE_RUNTIME_FLAGS or arg.startswith(_LANGUAGE_RUNTIME_PREFIXES)
+        for arg in argv
+    )
+
+
+def main(argv: list[str] | None = None):
+    """CLI entry point for ``lerobot-rollout``.
+
+    Standard policy deployment continues through :class:`RolloutConfig`.
+    Interactive language-conditioned and RoboCasa runs share this entry point
+    and are selected with ``--language`` or any language-runtime-only option.
+    """
     register_third_party_plugins()
-    rollout()
+    cli_args = list(sys.argv[1:] if argv is None else argv)
+    if _uses_language_runtime(cli_args):
+        from lerobot.runtime.cli import run as run_language_runtime
+
+        # ``--language`` is a dispatcher flag, not part of the runtime's own
+        # argparse surface. All other arguments pass through unchanged.
+        runtime_args = [arg for arg in cli_args if arg != "--language"]
+        return run_language_runtime(runtime_args, prog="lerobot-rollout")
+
+    if argv is None:
+        return rollout()
+
+    # draccus reads sys.argv. Supporting an explicit argv keeps this entry
+    # point easy to smoke-test and mirrors the language-runtime branch above.
+    previous_argv = sys.argv
+    try:
+        sys.argv = [previous_argv[0], *cli_args]
+        return rollout()
+    finally:
+        sys.argv = previous_argv
 
 
 if __name__ == "__main__":

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/phone/teleop_phone.py

@@ -171,13 +171,7 @@ class IOSPhone(BasePhone, Teleoperator):
         # HEBI provides orientation in w, x, y, z format.
         # Scipy's Rotation expects x, y, z, w.
         quat_xyzw = np.concatenate((ar_quat[1:], [ar_quat[0]]))  # wxyz to xyzw
-        # ARKit can emit zero/NaN quaternions before tracking is ready or on a
-        # dropped packet. Rotation.from_quat now rejects those; degrade the same
-        # way as a missing pose so teleop stays alive mid-session.
-        try:
         rot = Rotation.from_quat(quat_xyzw)
-        except ValueError:
-            return False, None, None, None
         pos = ar_pos - rot.apply(self.config.camera_offset)
         return True, pos, rot, pose
 

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

@@ -145,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/device_utils.py

@@ -37,25 +37,16 @@ def auto_select_torch_device() -> torch.device:
 
 # TODO(Steven): Remove log. log shouldn't be an argument, this should be handled by the logger level
 def get_safe_torch_device(try_device: str, log: bool = False) -> torch.device:
-    """Given a string, return a torch.device with checks on whether the device is available.
-
-    Raises:
-        ValueError: If the requested device family is known but not available on
-            this machine (``AssertionError`` was previously used and is easy to
-            mistake for a programmer bug under ``python -O`` where asserts vanish).
-    """
+    """Given a string, return a torch.device with checks on whether the device is available."""
     try_device = str(try_device)
     if try_device.startswith("cuda"):
-        if not torch.cuda.is_available():
-            raise ValueError(f"Requested device {try_device!r} but CUDA is not available.")
+        assert torch.cuda.is_available()
         device = torch.device(try_device)
     elif try_device == "mps":
-        if not torch.backends.mps.is_available():
-            raise ValueError("Requested device 'mps' but MPS is not available.")
+        assert torch.backends.mps.is_available()
         device = torch.device("mps")
     elif try_device == "xpu":
-        if not torch.xpu.is_available():
-            raise ValueError("Requested device 'xpu' but XPU is not available.")
+        assert torch.xpu.is_available()
         device = torch.device("xpu")
     elif try_device == "cpu":
         device = torch.device("cpu")

src/lerobot/utils/io_utils.py

@@ -23,6 +23,46 @@ logger = logging.getLogger(__name__)
 JsonLike = str | int | float | bool | None | list["JsonLike"] | dict[str, "JsonLike"] | tuple["JsonLike", ...]
 
 
+class StreamingVideoWriter:
+    """Incrementally encode RGB frames to an MP4 without retaining them in memory."""
+
+    def __init__(self, video_path: str | Path, fps: int) -> None:
+        from .import_utils import require_package
+
+        require_package("av", extra="av-dep")
+        import av
+
+        self._av = av
+        self._container = av.open(str(video_path), mode="w")
+        self._stream = self._container.add_stream("libx264", rate=fps)
+        self._shape: tuple[int, int] | None = None
+        self.frames_written = 0
+
+    def add_frame(self, frame_array) -> None:
+        orig_height, orig_width = frame_array.shape[:2]
+        height = orig_height - orig_height % 2
+        width = orig_width - orig_width % 2
+        if self._shape is None:
+            self._shape = (height, width)
+            self._stream.width = width
+            self._stream.height = height
+            self._stream.pix_fmt = "yuv420p"
+        elif self._shape != (height, width):
+            raise ValueError(f"Video frame shape changed from {self._shape} to {(height, width)}")
+        frame = self._av.VideoFrame.from_ndarray(frame_array[:height, :width], format="rgb24")
+        for packet in self._stream.encode(frame):
+            self._container.mux(packet)
+        self.frames_written += 1
+
+    def close(self) -> None:
+        if self._container is None:
+            return
+        for packet in self._stream.encode():
+            self._container.mux(packet)
+        self._container.close()
+        self._container = None
+
+
 def load_json(fpath: Path) -> Any:
     """Load data from a JSON file.
 
@@ -32,21 +72,21 @@ def load_json(fpath: Path) -> Any:
     Returns:
         Any: The data loaded from the JSON file.
     """
-    with open(fpath, encoding="utf-8") as f:
+    with open(fpath) as f:
         return json.load(f)
 
 
-def write_json(data: JsonLike, fpath: Path) -> None:
-    """Write JSON-serializable data to a file.
+def write_json(data: dict, fpath: Path) -> None:
+    """Write data to a JSON file.
 
     Creates parent directories if they don't exist.
 
     Args:
-        data: JSON-serializable data to write.
+        data (dict): The dictionary to write.
         fpath (Path): The path to the output JSON file.
     """
     fpath.parent.mkdir(exist_ok=True, parents=True)
-    with open(fpath, "w", encoding="utf-8") as f:
+    with open(fpath, "w") as f:
         json.dump(data, f, indent=4, ensure_ascii=False)
 
 
@@ -58,36 +98,12 @@ def write_video(video_path: str | Path, stacked_frames: list, fps: int) -> None:
         stacked_frames: List of HWC uint8 numpy arrays (RGB).
         fps: Frames per second for the output video.
     """
-    from .import_utils import require_package
-
-    require_package("av", extra="av-dep")
-    import av
-
-    with av.open(str(video_path), mode="w") as container:
-        orig_height, orig_width = stacked_frames[0].shape[:2]
-        # yuv420p requires even dimensions; crop by one pixel if needed
-        height = orig_height if orig_height % 2 == 0 else orig_height - 1
-        width = orig_width if orig_width % 2 == 0 else orig_width - 1
-        if height != orig_height or width != orig_width:
-            logger.warning(
-                "Frame dimensions %dx%d are not even; cropping to %dx%d for yuv420p compatibility.",
-                orig_width,
-                orig_height,
-                width,
-                height,
-            )
-        stream = container.add_stream("libx264", rate=fps)
-        stream.width = width
-        stream.height = height
-        stream.pix_fmt = "yuv420p"
+    writer = StreamingVideoWriter(video_path, fps)
+    try:
         for frame_array in stacked_frames:
-            if height != orig_height or width != orig_width:
-                frame_array = frame_array[:height, :width]
-            frame = av.VideoFrame.from_ndarray(frame_array, format="rgb24")
-            for packet in stream.encode(frame):
-                container.mux(packet)
-        for packet in stream.encode():
-            container.mux(packet)
+            writer.add_frame(frame_array)
+    finally:
+        writer.close()
 
 
 def deserialize_json_into_object[T: JsonLike](fpath: Path, obj: T) -> T:

src/lerobot/utils/process.py

@@ -16,39 +16,11 @@
 # limitations under the License.
 
 import logging
-import multiprocessing
 import os
 import signal
 import sys
 
 
-def ensure_multiprocessing_start_method(start_method: str | None) -> None:
-    """Set a multiprocessing start method once, or verify the existing method matches.
-
-    Passing ``None`` leaves Python's process-wide default untouched. This is useful
-    when LeRobot is embedded in an application that owns multiprocessing setup.
-    """
-    if start_method is None:
-        return
-
-    available_methods = multiprocessing.get_all_start_methods()
-    if start_method not in available_methods:
-        raise ValueError(
-            f"Multiprocessing start method must be one of {available_methods} on this platform, "
-            f"got {start_method!r}."
-        )
-
-    current_method = multiprocessing.get_start_method(allow_none=True)
-    if current_method is None:
-        multiprocessing.set_start_method(start_method)
-    elif current_method != start_method:
-        raise RuntimeError(
-            f"Multiprocessing start method is already {current_method!r}; cannot change it to "
-            f"{start_method!r}. Set the configured multiprocessing context to null to keep the "
-            "application's existing method, or launch LeRobot in a fresh process."
-        )
-
-
 class ProcessSignalHandler:
     """Utility class to attach graceful shutdown signal handlers.
 

src/lerobot/utils/rerun_visualization.py

@@ -38,7 +38,10 @@ def _is_scalar(x):
 
 
 def init_rerun(
-    session_name: str = "lerobot_control_loop", ip: str | None = None, port: int | None = None
+    session_name: str = "lerobot_control_loop",
+    ip: str | None = None,
+    port: int | None = None,
+    web_port: int | None = None,
 ) -> None:
     """
     Initializes the Rerun SDK for visualizing the control loop.
@@ -47,6 +50,7 @@ def init_rerun(
         session_name: Name of the Rerun session.
         ip: Optional IP for connecting to a Rerun server.
         port: Optional port for connecting to a Rerun server.
+        web_port: Serve a headless web viewer on this port, using ``port`` for gRPC.
     """
 
     require_package("rerun-sdk", extra="viz", import_name="rerun")
@@ -60,6 +64,10 @@ def init_rerun(
     memory_limit = os.getenv("LEROBOT_RERUN_MEMORY_LIMIT", "10%")
     if ip and port:
         rr.connect_grpc(url=f"rerun+http://{ip}:{port}/proxy")
+    elif web_port is not None:
+        grpc_port = port or 9876
+        url = rr.serve_grpc(grpc_port=grpc_port)
+        rr.serve_web_viewer(web_port=web_port, open_browser=False, connect_to=url)
     else:
         rr.spawn(memory_limit=memory_limit)
 

src/lerobot/utils/robot_utils.py

@@ -30,10 +30,6 @@ def precise_sleep(seconds: float, spin_threshold: float = 0.010, sleep_margin: f
     """
     if seconds <= 0:
         return
-    if spin_threshold < 0:
-        raise ValueError(f"spin_threshold must be >= 0, got {spin_threshold}")
-    if sleep_margin < 0:
-        raise ValueError(f"sleep_margin must be >= 0, got {sleep_margin}")
 
     system = platform.system()
     # On macOS and Windows the scheduler / sleep granularity can make

src/lerobot/utils/rotation.py

@@ -29,12 +29,9 @@ class Rotation:
     def __init__(self, quat: np.ndarray) -> None:
         """Initialize rotation from quaternion [x, y, z, w]."""
         self._quat = np.asarray(quat, dtype=float)
-        if self._quat.shape != (4,):
-            raise ValueError(f"Quaternion must have shape (4,), got {self._quat.shape}")
-        # Normalize quaternion. Reject the zero vector — it has no orientation.
+        # Normalize quaternion
         norm = np.linalg.norm(self._quat)
-        if norm <= 0.0 or not np.isfinite(norm):
-            raise ValueError(f"Quaternion must be a non-zero finite vector; got {self._quat} (norm={norm})")
+        if norm > 0:
             self._quat = self._quat / norm
 
     @classmethod

src/lerobot/utils/transition.py

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import NotRequired, TypedDict
+from typing import TypedDict
 
 import torch
 
@@ -28,7 +28,7 @@ class Transition(TypedDict):
     next_state: dict[str, torch.Tensor]
     done: bool
     truncated: bool
-    complementary_info: NotRequired[dict[str, torch.Tensor | float | int] | None]
+    complementary_info: dict[str, torch.Tensor | float | int] | None = None
 
 
 def move_transition_to_device(transition: Transition, device: str = "cpu") -> Transition:

src/lerobot/utils/utils.py

@@ -24,6 +24,7 @@ import sys
 import time
 from collections.abc import Iterator
 from copy import copy, deepcopy
+from datetime import datetime
 from pathlib import Path
 from statistics import mean
 from typing import TYPE_CHECKING, Any
@@ -60,16 +61,14 @@ def init_logging(
         accelerator: Optional Accelerator instance (for multi-GPU detection)
     """
 
-    class LeRobotFormatter(logging.Formatter):
-        def format(self, record: logging.LogRecord) -> str:
-            record.lerobot_location = f"{record.pathname}:{record.lineno}"[-15:]
-            record.lerobot_pid = f"[PID: {os.getpid()}] " if display_pid else ""
-            return super().format(record)
+    def custom_format(record: logging.LogRecord) -> str:
+        dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        fnameline = f"{record.pathname}:{record.lineno}"
+        pid_str = f"[PID: {os.getpid()}] " if display_pid else ""
+        return f"{record.levelname} {pid_str}{dt} {fnameline[-15:]:>15} {record.getMessage()}"
 
-    formatter = LeRobotFormatter(
-        "%(levelname)s %(lerobot_pid)s%(asctime)s %(lerobot_location)15s %(message)s",
-        datefmt="%Y-%m-%d %H:%M:%S",
-    )
+    formatter = logging.Formatter()
+    formatter.format = custom_format
 
     logger = logging.getLogger()
     logger.setLevel(logging.NOTSET)
@@ -134,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)
+        subprocess.run(cmd, check=True)
     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)
 
 
 def log_say(text: str, play_sounds: bool = True, blocking: bool = False):

src/lerobot/utils/video_annotation.py

@@ -0,0 +1,71 @@
+# 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.
+
+"""Best-effort text overlays shared by evaluation and interactive rollouts."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+import numpy as np
+
+
+def annotate_frame(frame: np.ndarray, fields: Iterable[tuple[str, str | None]]) -> np.ndarray:
+    """Return an RGB frame annotated with the non-empty labeled ``fields``."""
+    if frame.ndim != 3 or frame.shape[-1] != 3:
+        return frame
+    try:
+        import cv2  # noqa: PLC0415
+    except ImportError:
+        return frame
+
+    text_rows = [f"{label}: {value}" for label, value in fields if value]
+    if not text_rows:
+        return frame
+
+    image = np.ascontiguousarray(frame).copy()
+    font, scale, thickness, margin = cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1, 6
+    max_width = image.shape[1] - 2 * margin
+    lines: list[str] = []
+    for text in text_rows:
+        current = ""
+        for word in text.split():
+            candidate = f"{current} {word}".strip()
+            width = cv2.getTextSize(candidate, font, scale, thickness)[0][0]
+            if width > max_width and current:
+                lines.append(current)
+                current = word
+            else:
+                current = candidate
+        if current:
+            lines.append(current)
+
+    line_height = 20
+    header_height = min(image.shape[0], len(lines) * line_height + 6)
+    backdrop = image.copy()
+    cv2.rectangle(backdrop, (0, 0), (image.shape[1], header_height), (0, 0, 0), -1)
+    cv2.addWeighted(backdrop, 0.55, image, 0.45, 0, dst=image)
+
+    for index, line in enumerate(lines):
+        cv2.putText(
+            image,
+            line,
+            (margin, 18 + index * line_height),
+            font,
+            scale,
+            (255, 255, 255),
+            thickness,
+            cv2.LINE_AA,
+        )
+    return image

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