feat(dataset): integrate episode streaming into training

src/lerobot/common/train_utils.py

@@ -53,7 +53,11 @@ def get_step_checkpoint_dir(output_dir: Path, total_steps: int, step: int) -> Pa
 
 
 def save_training_step(
-    step: int, save_dir: Path, num_processes: int | None = None, batch_size: int | None = None
+    step: int,
+    save_dir: Path,
+    num_processes: int | None = None,
+    batch_size: int | None = None,
+    num_workers: int | None = None,
 ) -> None:
     state: dict = {"step": step}
     # num_processes and batch_size are recorded so a resumed run can detect a changed world size or
@@ -64,6 +68,8 @@ def save_training_step(
         state["num_processes"] = num_processes
     if batch_size is not None:
         state["batch_size"] = batch_size
+    if num_workers is not None:
+        state["num_workers"] = num_workers
     write_json(state, save_dir / TRAINING_STEP)
 
 
@@ -82,6 +88,11 @@ def load_training_batch_size(checkpoint_dir: Path) -> int | None:
     return load_json(checkpoint_dir / TRAINING_STATE_DIR / TRAINING_STEP).get("batch_size")
 
 
+def load_training_num_workers(checkpoint_dir: Path) -> int | None:
+    """DataLoader worker count recorded at checkpoint time, or None for older checkpoints."""
+    return load_json(checkpoint_dir / TRAINING_STATE_DIR / TRAINING_STEP).get("num_workers")
+
+
 def update_last_checkpoint(checkpoint_dir: Path) -> Path:
     last_checkpoint_dir = checkpoint_dir.parent / LAST_CHECKPOINT_LINK
     if last_checkpoint_dir.is_symlink():
@@ -101,6 +112,7 @@ def save_checkpoint(
     postprocessor: PolicyProcessorPipeline | None = None,
     num_processes: int | None = None,
     batch_size: int | None = None,
+    num_workers: int | None = None,
     model_state_dict: dict | None = None,
     optim_state_dict: dict | None = None,
 ) -> None:
@@ -132,6 +144,8 @@ def save_checkpoint(
             resume. Defaults to None (not recorded).
         batch_size (int | None, optional): Per-process batch size to record for sample-exact
             resume. Defaults to None (not recorded).
+        num_workers (int | None, optional): Per-process DataLoader worker count to record for
+            sample-exact streaming resume. Defaults to None (not recorded).
         model_state_dict: Pre-gathered full (unsharded) model state dict. Required under FSDP,
             where `policy.state_dict()` would return sharded tensors; the caller gathers it via a
             cross-rank collective and passes it here so rank 0 can write it directly. It holds
@@ -160,6 +174,7 @@ def save_checkpoint(
         scheduler,
         num_processes=num_processes,
         batch_size=batch_size,
+        num_workers=num_workers,
         optim_state_dict=optim_state_dict,
     )
 
@@ -171,6 +186,7 @@ def save_training_state(
     scheduler: LRScheduler | None = None,
     num_processes: int | None = None,
     batch_size: int | None = None,
+    num_workers: int | None = None,
     optim_state_dict: dict | None = None,
 ) -> None:
     """
@@ -185,12 +201,20 @@ def save_training_state(
             Defaults to None.
         num_processes (int | None, optional): Distributed world size to record. Defaults to None.
         batch_size (int | None, optional): Per-process batch size to record. Defaults to None.
+        num_workers (int | None, optional): Per-process DataLoader worker count to record.
+            Defaults to None.
         optim_state_dict: Pre-gathered full optimizer state dict (for FSDP). Saved instead of
             `optimizer.state_dict()` when provided. Defaults to None.
     """
     save_dir = checkpoint_dir / TRAINING_STATE_DIR
     save_dir.mkdir(parents=True, exist_ok=True)
-    save_training_step(train_step, save_dir, num_processes=num_processes, batch_size=batch_size)
+    save_training_step(
+        train_step,
+        save_dir,
+        num_processes=num_processes,
+        batch_size=batch_size,
+        num_workers=num_workers,
+    )
     save_rng_state(save_dir)
     if optimizer is not None:
         save_optimizer_state(optimizer, save_dir, optim_state_dict=optim_state_dict)

src/lerobot/configs/default.py

@@ -44,6 +44,14 @@ class DatasetConfig:
     # Has no effect on datasets without depth cameras.
     depth_output_unit: str = DEFAULT_DEPTH_UNIT
     streaming: bool = False
+    # Optional data-plane root used by training-time streaming. Metadata still resolves from repo_id/root.
+    streaming_data_root: str | None = None
+    # Number of complete episodes mixed by the exact-coverage sampler in each DataLoader worker.
+    streaming_episode_pool_size: int = 32
+    # Complete episodes fetched ahead of the current admission frontier.
+    streaming_prefetch_episodes: int = 8
+    # Hard per-worker cap for synthesized episode-video bytes.
+    streaming_byte_budget_gb: float = 8.0
     # Fraction of episodes held out per task for offline evaluation (0.0 = disabled).
     eval_split: float = 0.0
 
@@ -54,6 +62,12 @@ class DatasetConfig:
             )
         if not (0.0 <= self.eval_split < 1.0):
             raise ValueError(f"eval_split must be in [0.0, 1.0), got {self.eval_split}")
+        if self.streaming_episode_pool_size <= 0:
+            raise ValueError("streaming_episode_pool_size must be positive")
+        if self.streaming_prefetch_episodes < 0:
+            raise ValueError("streaming_prefetch_episodes must be non-negative")
+        if self.streaming_byte_budget_gb <= 0:
+            raise ValueError("streaming_byte_budget_gb must be positive")
         if self.episodes is not None:
             if any(ep < 0 for ep in self.episodes):
                 raise ValueError(

src/lerobot/datasets/episode_parquet.py

@@ -0,0 +1,106 @@
+# 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
+
+"""Episode-scoped Parquet reads for training-time dataset streaming."""
+
+from __future__ import annotations
+
+import posixpath
+from collections.abc import Sequence
+from pathlib import Path
+
+import fsspec
+import pyarrow as pa
+import pyarrow.compute as pc
+import pyarrow.parquet as pq
+
+
+class EpisodeParquetReader:
+    """Read complete episodes with column projection from local or fsspec roots."""
+
+    def __init__(self, data_root: str | Path, *, columns: Sequence[str]):
+        if not columns:
+            raise ValueError("EpisodeParquetReader requires at least one projected column")
+        self.columns = tuple(dict.fromkeys(columns))
+        self._read_columns = (
+            self.columns if "episode_index" in self.columns else (*self.columns, "episode_index")
+        )
+        self._filesystem, self._root_path = fsspec.core.url_to_fs(str(data_root))
+
+    def read_episode(
+        self,
+        relative_path: str | Path,
+        *,
+        episode_index: int,
+        expected_rows: int,
+    ) -> pa.Table:
+        if expected_rows <= 0:
+            raise ValueError(f"Episode {episode_index} must contain at least one row")
+
+        path = posixpath.join(self._root_path.rstrip("/"), str(relative_path).lstrip("/"))
+        with self._filesystem.open(path, "rb") as source:
+            parquet = pq.ParquetFile(source)
+            available = set(parquet.schema_arrow.names)
+            missing = sorted(set(self._read_columns) - available)
+            if missing:
+                raise ValueError(f"Parquet file {relative_path} is missing projected columns: {missing}")
+            row_group = self._matching_row_group(parquet, episode_index)
+            table = (
+                parquet.read_row_group(row_group, columns=list(self._read_columns))
+                if row_group is not None
+                else parquet.read(columns=list(self._read_columns))
+            )
+
+        if row_group is None:
+            table = table.filter(pc.equal(table.column("episode_index"), episode_index))
+        self._validate_complete_episode(table, episode_index, expected_rows, relative_path)
+        if "episode_index" not in self.columns:
+            table = table.drop_columns(["episode_index"])
+        return table
+
+    @staticmethod
+    def _matching_row_group(parquet: pq.ParquetFile, episode_index: int) -> int | None:
+        episode_column = next(
+            index
+            for index in range(parquet.metadata.num_columns)
+            if parquet.metadata.schema.column(index).path == "episode_index"
+        )
+        matches = []
+        for row_group in range(parquet.metadata.num_row_groups):
+            statistics = parquet.metadata.row_group(row_group).column(episode_column).statistics
+            if (
+                statistics is not None
+                and statistics.has_min_max
+                and int(statistics.min) == episode_index
+                and int(statistics.max) == episode_index
+            ):
+                matches.append(row_group)
+        return matches[0] if len(matches) == 1 else None
+
+    @staticmethod
+    def _validate_complete_episode(
+        table: pa.Table,
+        episode_index: int,
+        expected_rows: int,
+        relative_path: str | Path,
+    ) -> None:
+        actual_rows = len(table)
+        if actual_rows != expected_rows:
+            raise ValueError(
+                f"Parquet episode {episode_index} in {relative_path}: "
+                f"expected {expected_rows} rows, found {actual_rows}"
+            )
+        episodes = table.column("episode_index").to_pylist()
+        if any(int(value) != episode_index for value in episodes):
+            raise ValueError(f"Parquet file {relative_path} returned rows outside episode {episode_index}")
+        if "frame_index" in table.column_names:
+            frame_indices = [int(value) for value in table.column("frame_index").to_pylist()]
+            if frame_indices != list(range(expected_rows)):
+                raise ValueError(
+                    f"Parquet episode {episode_index} in {relative_path} has non-contiguous frame indices"
+                )

src/lerobot/datasets/factory.py

@@ -66,7 +66,9 @@ def resolve_delta_timestamps(
     return delta_timestamps
 
 
-def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDataset:
+def make_dataset(
+    cfg: TrainPipelineConfig,
+) -> LeRobotDataset | StreamingLeRobotDataset | MultiLeRobotDataset:
     """Handles the logic of setting up delta timestamps and image transforms before creating a dataset.
 
     Args:
@@ -108,9 +110,15 @@ def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDatas
                 delta_timestamps=delta_timestamps,
                 image_transforms=image_transforms,
                 revision=cfg.dataset.revision,
-                max_num_shards=cfg.num_workers,
+                max_num_shards=max(1, cfg.num_workers),
                 tolerance_s=cfg.tolerance_s,
                 return_uint8=True,
+                depth_output_unit=cfg.dataset.depth_output_unit,
+                data_root=cfg.dataset.streaming_data_root,
+                episode_pool_size=cfg.dataset.streaming_episode_pool_size,
+                prefetch_episodes=cfg.dataset.streaming_prefetch_episodes,
+                byte_budget_gb=cfg.dataset.streaming_byte_budget_gb,
+                repeat=True,
             )
     else:
         raise NotImplementedError("The MultiLeRobotDataset isn't supported for now.")
@@ -138,7 +146,10 @@ def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDatas
 
 def make_train_eval_datasets(
     cfg: TrainPipelineConfig,
-) -> tuple[LeRobotDataset | MultiLeRobotDataset, LeRobotDataset | None]:
+) -> tuple[
+    LeRobotDataset | StreamingLeRobotDataset | MultiLeRobotDataset,
+    LeRobotDataset | None,
+]:
     """Create train and optional eval datasets by splitting episodes based on eval_split.
 
     The last ceil(n_episodes * eval_split) episodes per task are held out for evaluation.
@@ -181,17 +192,37 @@ def make_train_eval_datasets(
         ImageTransforms(cfg.dataset.image_transforms) if cfg.dataset.image_transforms.enable else None
     )
 
-    train_dataset = LeRobotDataset(
-        cfg.dataset.repo_id,
-        root=cfg.dataset.root,
-        episodes=train_episodes,
-        delta_timestamps=delta_timestamps,
-        image_transforms=train_image_transforms,
-        revision=cfg.dataset.revision,
-        video_backend=cfg.dataset.video_backend,
-        return_uint8=True,
-        tolerance_s=cfg.tolerance_s,
-    )
+    if cfg.dataset.streaming:
+        train_dataset = StreamingLeRobotDataset(
+            cfg.dataset.repo_id,
+            root=cfg.dataset.root,
+            episodes=train_episodes,
+            delta_timestamps=delta_timestamps,
+            image_transforms=train_image_transforms,
+            revision=cfg.dataset.revision,
+            max_num_shards=max(1, cfg.num_workers),
+            tolerance_s=cfg.tolerance_s,
+            return_uint8=True,
+            depth_output_unit=cfg.dataset.depth_output_unit,
+            data_root=cfg.dataset.streaming_data_root,
+            episode_pool_size=cfg.dataset.streaming_episode_pool_size,
+            prefetch_episodes=cfg.dataset.streaming_prefetch_episodes,
+            byte_budget_gb=cfg.dataset.streaming_byte_budget_gb,
+            repeat=True,
+        )
+    else:
+        train_dataset = LeRobotDataset(
+            cfg.dataset.repo_id,
+            root=cfg.dataset.root,
+            episodes=train_episodes,
+            delta_timestamps=delta_timestamps,
+            image_transforms=train_image_transforms,
+            revision=cfg.dataset.revision,
+            video_backend=cfg.dataset.video_backend,
+            return_uint8=True,
+            depth_output_unit=cfg.dataset.depth_output_unit,
+            tolerance_s=cfg.tolerance_s,
+        )
 
     eval_dataset = LeRobotDataset(
         cfg.dataset.repo_id,
@@ -202,6 +233,7 @@ def make_train_eval_datasets(
         revision=cfg.dataset.revision,
         video_backend=cfg.dataset.video_backend,
         return_uint8=True,
+        depth_output_unit=cfg.dataset.depth_output_unit,
         tolerance_s=cfg.tolerance_s,
     )
 

src/lerobot/datasets/streaming_sidecar.py

@@ -0,0 +1,141 @@
+# 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
+
+"""LeRobot metadata adapter for automatic MP4 sidecar resolution."""
+
+from __future__ import annotations
+
+import logging
+import shutil
+from pathlib import Path
+
+import fsspec
+
+from lerobot.datasets.dataset_metadata import LeRobotDatasetMetadata
+from lerobot.streaming.episode_video import EpisodeVideoManifest
+from lerobot.streaming.sidecar import SidecarSpec, ensure_mp4_sidecar, sidecar_cache_path
+from lerobot.utils.constants import HF_LEROBOT_HOME
+
+DEFAULT_SIDECAR_CACHE = HF_LEROBOT_HOME / "streaming-sidecars"
+
+
+def range_backend_for_root(data_root: str) -> str:
+    """Use direct HTTP only for HF roots; all local and other fsspec protocols stay generic."""
+    return "native-http" if data_root.startswith("hf://") else "fsspec"
+
+
+def streaming_data_root(
+    meta: LeRobotDatasetMetadata,
+    *,
+    requested_root: str | Path | None,
+    configured_data_root: str | None,
+) -> str:
+    if configured_data_root is not None:
+        return configured_data_root.rstrip("/")
+    if requested_root is not None:
+        return str(Path(requested_root).expanduser())
+    return f"hf://datasets/{meta.repo_id}@{meta.revision}"
+
+
+def make_sidecar_spec(meta: LeRobotDatasetMetadata, data_root: str) -> SidecarSpec:
+    relative_paths = sorted(
+        {
+            str(meta.get_video_file_path(episode_index, video_key))
+            for episode_index in range(int(meta.total_episodes))
+            for video_key in meta.video_keys
+        }
+    )
+    root = Path(data_root).expanduser()
+    source_files: tuple[tuple[str, int | None], ...]
+    if root.is_dir():
+        source_files = tuple((path, (root / path).stat().st_size) for path in relative_paths)
+    else:
+        source_files = tuple((path, None) for path in relative_paths)
+    return SidecarSpec(
+        repo_id=meta.repo_id,
+        revision=str(meta.revision),
+        data_root=data_root.rstrip("/"),
+        source_files=source_files,
+    )
+
+
+def build_mp4_sidecar(
+    destination: str | Path,
+    spec: SidecarSpec,
+    *,
+    workers: int = 8,
+    range_backend: str = "native-http",
+    max_probe_bytes: int = 64 * 1024 * 1024,
+) -> None:
+    EpisodeVideoManifest.write_file_sidecar(
+        destination,
+        [path for path, _size in spec.source_files],
+        spec.data_root,
+        spec=spec,
+        range_backend=range_backend,
+        workers=workers,
+        max_probe_bytes=max_probe_bytes,
+    )
+
+
+def published_sidecar_url(spec: SidecarSpec, cache_root: str | Path = DEFAULT_SIDECAR_CACHE) -> str:
+    name = sidecar_cache_path(cache_root, spec).name
+    return f"{spec.data_root}/meta/mp4-sidecars/{name}"
+
+
+def download_published_sidecar(
+    destination: Path,
+    spec: SidecarSpec,
+    *,
+    cache_root: str | Path = DEFAULT_SIDECAR_CACHE,
+) -> bool:
+    if Path(spec.data_root).expanduser().is_dir():
+        return False
+    filesystem, source = fsspec.core.url_to_fs(published_sidecar_url(spec, cache_root))
+    if not filesystem.exists(source):
+        return False
+    with filesystem.open(source, "rb") as remote, destination.open("wb") as local:
+        shutil.copyfileobj(remote, local)
+    return True
+
+
+def ensure_dataset_mp4_sidecar(
+    meta: LeRobotDatasetMetadata,
+    data_root: str,
+    *,
+    cache_root: str | Path = DEFAULT_SIDECAR_CACHE,
+    workers: int = 8,
+    range_backend: str = "native-http",
+    lock_timeout_s: float = 30 * 60,
+) -> Path | None:
+    if not meta.video_keys:
+        return None
+
+    spec = make_sidecar_spec(meta, data_root)
+    logging.info(
+        "Resolving training-time MP4 sidecar for %s@%s (%d files)",
+        spec.repo_id,
+        spec.revision,
+        len(spec.source_files),
+    )
+    return ensure_mp4_sidecar(
+        spec,
+        cache_root,
+        build=lambda path, target_spec: build_mp4_sidecar(
+            path,
+            target_spec,
+            workers=workers,
+            range_backend=range_backend,
+        ),
+        download=lambda path, target_spec: download_published_sidecar(
+            path,
+            target_spec,
+            cache_root=cache_root,
+        ),
+        lock_timeout_s=lock_timeout_s,
+    )

src/lerobot/datasets/video_utils.py

@@ -28,7 +28,7 @@ from dataclasses import asdict, dataclass, field
 from fractions import Fraction
 from pathlib import Path
 from threading import Lock
-from typing import Any, ClassVar
+from typing import Any, BinaryIO, ClassVar
 
 import av
 import fsspec
@@ -105,7 +105,7 @@ def decode_video_frames(
 
 
 def decode_video_frames_pyav(
-    video_path: Path | str,
+    video_path: Path | str | BinaryIO,
     timestamps: list[float],
     tolerance_s: float,
     log_loaded_timestamps: bool = False,
@@ -124,7 +124,7 @@ def decode_video_frames_pyav(
     video can be adjusted at encoding time to trade off decoding speed against file size.
 
     Args:
-        video_path: Path to the video file.
+        video_path: Path to the video file or a seekable binary file object.
         timestamps: List of timestamps (in seconds) to extract frames for.
         tolerance_s: Allowed deviation in seconds between a queried timestamp and the closest
             decoded frame.
@@ -137,7 +137,8 @@ def decode_video_frames_pyav(
         torch.Tensor of shape (len(timestamps), C, H, W).
     """
     # TODO(rcadene): also load audio stream at the same time
-    video_path = str(video_path)
+    video_source = str(video_path) if isinstance(video_path, (Path, str)) else video_path
+    video_label = str(video_path) if isinstance(video_path, (Path, str)) else "<in-memory video>"
 
     # set the first and last requested timestamps
     # Note: previous timestamps are usually loaded, since we need to access the previous key frame
@@ -151,7 +152,7 @@ def decode_video_frames_pyav(
     # av.time_base units (microseconds). `backward=True` lands us on the nearest keyframe at or
     # before `first_ts`, so we can then decode forward until we cover `last_ts`. See:
     # https://pyav.basswood-io.com/docs/stable/api/container.html#av.container.InputContainer.seek
-    with av.open(video_path) as container:
+    with av.open(video_source) as container:
         stream = container.streams.video[0]
         # Seek to the nearest keyframe at or before `first_ts` with a 1 frame margin
         container.seek(
@@ -180,7 +181,7 @@ def decode_video_frames_pyav(
 
     if not loaded_frames:
         raise FrameTimestampError(
-            f"No frames could be decoded from {video_path} in the timestamp range [{first_ts}, {last_ts}]."
+            f"No frames could be decoded from {video_label} in the timestamp range [{first_ts}, {last_ts}]."
         )
 
     query_ts = torch.tensor(timestamps)
@@ -199,7 +200,7 @@ def decode_video_frames_pyav(
             " To be safe, we advise to ignore this item during training."
             f"\nqueried timestamps: {query_ts}"
             f"\nloaded timestamps: {loaded_ts_t}"
-            f"\nvideo: {video_path}"
+            f"\nvideo: {video_label}"
             f"\nbackend: pyav"
         )
 

src/lerobot/scripts/lerobot_train.py

@@ -41,6 +41,7 @@ from lerobot.common.train_utils import (
     load_fsdp_optimizer_state,
     load_training_batch_size,
     load_training_num_processes,
+    load_training_num_workers,
     load_training_state,
     push_checkpoint_to_hub,
     save_checkpoint,
@@ -201,7 +202,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
 
     require_package("accelerate", extra="training")
     from accelerate import Accelerator
-    from accelerate.utils import DistributedDataParallelKwargs, DistributedType
+    from accelerate.utils import DistributedDataParallelKwargs, DistributedType, send_to_device
 
     cfg.validate()
 
@@ -459,6 +460,59 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
     else:
         shuffle = True
         sampler = None
+        train_num_workers = cfg.num_workers
+        if train_num_workers > 0:
+            max_nonempty_workers = dataset.num_episodes // accelerator.num_processes
+            if max_nonempty_workers == 0:
+                raise ValueError(
+                    "At least one distributed rank owns no streaming episode. "
+                    "Reduce the distributed world size."
+                )
+            train_num_workers = min(train_num_workers, max_nonempty_workers)
+            if train_num_workers != cfg.num_workers and is_main_process:
+                logging.warning(
+                    "Reducing streaming DataLoader workers from %d to %d so every rank/worker "
+                    "owns at least one complete episode.",
+                    cfg.num_workers,
+                    train_num_workers,
+                )
+        rank_frames = dataset.num_frames_for_rank(
+            accelerator.process_index,
+            accelerator.num_processes,
+            train_num_workers,
+        )
+        if rank_frames == 0:
+            raise ValueError("This rank owns no streaming episodes. Reduce the distributed world size.")
+        if cfg.resume and step > 0:
+            saved_num_processes = load_training_num_processes(cfg.checkpoint_path)
+            saved_batch_size = load_training_batch_size(cfg.checkpoint_path)
+            saved_num_workers = load_training_num_workers(cfg.checkpoint_path)
+            if saved_num_processes not in (None, accelerator.num_processes):
+                raise ValueError(
+                    "Sample-exact streaming resume requires the checkpoint world size "
+                    f"({saved_num_processes}) to match the current world size "
+                    f"({accelerator.num_processes})."
+                )
+            if saved_batch_size not in (None, cfg.batch_size):
+                raise ValueError(
+                    "Sample-exact streaming resume requires the checkpoint batch size "
+                    f"({saved_batch_size}) to match the current batch size ({cfg.batch_size})."
+                )
+            if saved_num_workers not in (None, train_num_workers):
+                raise ValueError(
+                    "Sample-exact streaming resume requires the checkpoint DataLoader worker count "
+                    f"({saved_num_workers}) to match the current count ({train_num_workers})."
+                )
+            stream_offset = step * (saved_batch_size or cfg.batch_size)
+            dataset.load_state_dict(
+                {
+                    "epoch": 0,
+                    "offset": stream_offset,
+                    "batch_size": cfg.batch_size,
+                }
+            )
+            if is_main_process:
+                logging.info(f"Resuming streaming data order at local sample {stream_offset}")
 
     # Only swap in the language-aware collate when the dataset actually
     # declares language columns; otherwise stay on PyTorch's default
@@ -466,15 +520,20 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
     collate_fn = lerobot_collate_fn if dataset.meta.has_language_columns else None
     dataloader = torch.utils.data.DataLoader(
         dataset,
-        num_workers=cfg.num_workers,
+        num_workers=train_num_workers if cfg.dataset.streaming else cfg.num_workers,
         batch_size=cfg.batch_size,
         shuffle=shuffle and not cfg.dataset.streaming,
         sampler=sampler,
         pin_memory=device.type == "cuda",
         drop_last=False,
         collate_fn=collate_fn,
-        prefetch_factor=cfg.prefetch_factor if cfg.num_workers > 0 else None,
-        persistent_workers=cfg.persistent_workers and cfg.num_workers > 0,
+        prefetch_factor=(
+            cfg.prefetch_factor
+            if (train_num_workers if cfg.dataset.streaming else cfg.num_workers) > 0
+            else None
+        ),
+        persistent_workers=cfg.persistent_workers
+        and (train_num_workers if cfg.dataset.streaming else cfg.num_workers) > 0,
     )
 
     # Build eval dataloader if a held-out split exists
@@ -506,7 +565,16 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
 
     # Prepare everything with accelerator
     accelerator.wait_for_everyone()
-    if eval_dataloader is not None:
+    if cfg.dataset.streaming:
+        # StreamingLeRobotDataset already assigns complete episodes disjointly to ranks and workers.
+        # Preparing this loader would make Accelerate shard its batches a second time.
+        if eval_dataloader is not None:
+            policy, optimizer, lr_scheduler, eval_dataloader = accelerator.prepare(
+                policy, optimizer, lr_scheduler, eval_dataloader
+            )
+        else:
+            policy, optimizer, lr_scheduler = accelerator.prepare(policy, optimizer, lr_scheduler)
+    elif eval_dataloader is not None:
         policy, optimizer, dataloader, lr_scheduler, eval_dataloader = accelerator.prepare(
             policy, optimizer, dataloader, lr_scheduler, eval_dataloader
         )
@@ -569,6 +637,8 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
     for _ in range(step, cfg.steps):
         start_time = time.perf_counter()
         batch = next(dl_iter)
+        if cfg.dataset.streaming:
+            batch = send_to_device(batch, device, non_blocking=device.type == "cuda")
         for cam_key in dataset.meta.camera_keys:
             if cam_key in batch and batch[cam_key].dtype == torch.uint8:
                 batch[cam_key] = batch[cam_key].to(dtype=torch.float32) / 255.0
@@ -665,6 +735,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
                     postprocessor=postprocessor,
                     num_processes=accelerator.num_processes,
                     batch_size=cfg.batch_size,
+                    num_workers=train_num_workers if cfg.dataset.streaming else cfg.num_workers,
                     model_state_dict=model_state_dict,
                     optim_state_dict=optim_state_dict,
                 )

src/lerobot/streaming/__init__.py

@@ -0,0 +1,9 @@
+# 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
+
+"""Low-level primitives for training-time dataset streaming."""

src/lerobot/streaming/episode_video.py

@@ -11,17 +11,19 @@ from __future__ import annotations
 import contextlib
 import io
 import json
+import logging
 import os
+import posixpath
 import threading
 import time
 from collections import OrderedDict
-from concurrent.futures import Future, ThreadPoolExecutor
+from collections.abc import Sequence
+from concurrent.futures import Future, ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from datetime import UTC, datetime
-from importlib import metadata
 from pathlib import Path
 from types import MethodType
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from urllib.parse import quote, urljoin, urlparse
 from uuid import uuid4
 
@@ -31,8 +33,11 @@ import numpy as np
 from huggingface_hub import HfApi, HfFileSystem, constants
 from huggingface_hub.utils import get_session, hf_raise_for_status
 
-from lerobot.datasets.dataset_metadata import LeRobotDatasetMetadata
-from lerobot.datasets.mp4 import Mp4Index, Mp4SampleSlice, fetch_mp4_index, synthesize_mp4
+from lerobot.streaming.mp4 import Mp4Index, Mp4SampleSlice, fetch_mp4_index, synthesize_mp4
+
+if TYPE_CHECKING:
+    from lerobot.datasets.dataset_metadata import LeRobotDatasetMetadata
+    from lerobot.streaming.sidecar import SidecarSpec
 
 _HTTP_FAILURE_LOG_LOCK = threading.Lock()
 
@@ -134,11 +139,15 @@ class ThreadLocalRangeFetcher:
 
     def __init__(self, data_root: str | Path, *, block_size: int = 2**20, cache_type: str = "none"):
         self.data_root = str(data_root).rstrip("/")
-        protocol = "hf" if self.data_root.startswith("hf://") else "file"
-        self.fs = fsspec.filesystem(protocol)
+        self.fs, self._root_path = fsspec.core.url_to_fs(self.data_root)
+        self._is_local = self.fs.protocol in ("file", "local") or (
+            isinstance(self.fs.protocol, tuple) and "file" in self.fs.protocol
+        )
         self.block_size = block_size
         self.cache_type = cache_type
         self._local = threading.local()
+        self._handles_lock = threading.Lock()
+        self._all_handles: dict[int, Any] = {}
         self._timing_lock = threading.Lock()
         self._timing_totals = {
             "range_jobs": 0.0,
@@ -149,9 +158,9 @@ class ThreadLocalRangeFetcher:
         }
 
     def _url(self, relative_path: str) -> str:
-        if self.data_root.startswith("hf://"):
-            return f"{self.data_root}/{relative_path}"
-        return str(Path(self.data_root) / relative_path)
+        if self._is_local:
+            return str(Path(self._root_path) / relative_path)
+        return posixpath.join(self._root_path.rstrip("/"), relative_path.lstrip("/"))
 
     def _handle(self, relative_path: str):
         handles = getattr(self._local, "handles", None)
@@ -165,6 +174,8 @@ class ThreadLocalRangeFetcher:
             )
             self._instrument_hf_handle(handle)
             handles[relative_path] = handle
+            with self._handles_lock:
+                self._all_handles[id(handle)] = handle
         return handle
 
     def info_size(self, relative_path: str) -> int:
@@ -332,13 +343,15 @@ class ThreadLocalRangeFetcher:
             return dict(self._timing_totals)
 
     def close(self) -> None:
-        handles = getattr(self._local, "handles", None)
-        if handles is None:
-            return
-        for handle in handles.values():
+        with self._handles_lock:
+            handles = list(self._all_handles.values())
+            self._all_handles.clear()
+        for handle in handles:
             with contextlib.suppress(Exception):
                 handle.close()
-        handles.clear()
+        local_handles = getattr(self._local, "handles", None)
+        if local_handles is not None:
+            local_handles.clear()
 
 
 class NativeHTTPRangeFetcher:
@@ -751,7 +764,7 @@ def make_range_fetcher(
 
 
 class EpisodeVideoManifest:
-    _FILE_SIDECAR_CACHE: dict[str, dict[str, VideoFileRecord]] = {}
+    _FILE_SIDECAR_CACHE: dict[str, tuple[tuple[int, int], dict[str, VideoFileRecord]]] = {}
     _FILE_SIDECAR_CACHE_LOCK = threading.Lock()
 
     def __init__(
@@ -873,7 +886,14 @@ class EpisodeVideoManifest:
 
         try:
             with ThreadPoolExecutor(max_workers=workers) as pool:
-                return list(pool.map(build_file, rel_paths))
+                futures = {pool.submit(build_file, path): path for path in rel_paths}
+                records = []
+                progress_interval = max(1, len(futures) // 20)
+                for completed, future in enumerate(as_completed(futures), start=1):
+                    records.append(future.result())
+                    if completed == len(futures) or completed % progress_interval == 0:
+                        logging.info("Indexed %d/%d MP4 files for streaming sidecar", completed, len(futures))
+                return sorted(records, key=lambda record: record.file_path)
         finally:
             fetcher.close()
 
@@ -884,6 +904,7 @@ class EpisodeVideoManifest:
         rel_paths: list[str],
         data_root: str | Path,
         *,
+        spec: SidecarSpec,
         range_backend: str = "native-http",
         workers: int = 8,
         header_probe_bytes: int = 4 * 1024 * 1024,
@@ -897,14 +918,22 @@ class EpisodeVideoManifest:
             header_probe_bytes=header_probe_bytes,
             max_probe_bytes=max_probe_bytes,
         )
-        cls.save_file_sidecar(sidecar_path, records)
+        cls.save_file_sidecar(sidecar_path, records, spec=spec)
 
     @staticmethod
-    def save_file_sidecar(sidecar_path: str | Path, records: list[VideoFileRecord]) -> None:
+    def save_file_sidecar(
+        sidecar_path: str | Path,
+        records: list[VideoFileRecord],
+        *,
+        spec: SidecarSpec,
+    ) -> None:
         sidecar_path = Path(sidecar_path)
         sidecar_path.parent.mkdir(parents=True, exist_ok=True)
         payload = {
-            "version": 1,
+            "version": 2,
+            "sidecar": spec.with_source_files(
+                tuple((record.file_path, record.file_size) for record in records)
+            ).to_dict(),
             "files": [
                 {"file_path": record.file_path, "file_size": record.file_size, "mp4": record.mp4.to_dict()}
                 for record in records
@@ -918,17 +947,49 @@ class EpisodeVideoManifest:
             arrays[f"{file_idx}/sample_offsets"] = record.mp4.sample_offsets
             arrays[f"{file_idx}/sync_samples"] = record.mp4.sync_samples
         np.savez_compressed(sidecar_path, manifest_json=json.dumps(payload).encode("utf-8"), **arrays)
+        cache_key = str(sidecar_path.expanduser())
+        with EpisodeVideoManifest._FILE_SIDECAR_CACHE_LOCK:
+            EpisodeVideoManifest._FILE_SIDECAR_CACHE.pop(cache_key, None)
+
+    @staticmethod
+    def load_file_sidecar_metadata(sidecar_path: str | Path) -> dict[str, Any]:
+        with np.load(sidecar_path, allow_pickle=False) as data:
+            payload = json.loads(bytes(data["manifest_json"]).decode("utf-8"))
+        if payload.get("version") != 2 or not isinstance(payload.get("sidecar"), dict):
+            raise ValueError(f"Unsupported MP4 sidecar schema in {sidecar_path}")
+        return payload["sidecar"]
+
+    @staticmethod
+    def validate_file_sidecar(sidecar_path: str | Path, spec: SidecarSpec) -> bool:
+        try:
+            from lerobot.streaming.sidecar import SidecarSpec
+
+            candidate = SidecarSpec.from_dict(EpisodeVideoManifest.load_file_sidecar_metadata(sidecar_path))
+            if not spec.matches(candidate):
+                return False
+            records = EpisodeVideoManifest.load_file_sidecar(sidecar_path)
+        except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
+            return False
+
+        expected = dict(candidate.source_files)
+        actual = {path: record.file_size for path, record in records.items()}
+        return actual == expected
 
     @staticmethod
     def load_file_sidecar(sidecar_path: str | Path) -> dict[str, VideoFileRecord]:
-        cache_key = str(Path(sidecar_path).expanduser())
+        path = Path(sidecar_path).expanduser()
+        cache_key = str(path)
+        stat = path.stat()
+        signature = (stat.st_mtime_ns, stat.st_size)
         with EpisodeVideoManifest._FILE_SIDECAR_CACHE_LOCK:
             cached = EpisodeVideoManifest._FILE_SIDECAR_CACHE.get(cache_key)
-        if cached is not None:
-            return cached
+        if cached is not None and cached[0] == signature:
+            return cached[1]
 
-        with np.load(sidecar_path, allow_pickle=False) as data:
+        with np.load(path, allow_pickle=False) as data:
             payload = json.loads(bytes(data["manifest_json"]).decode("utf-8"))
+            if payload.get("version") != 2:
+                raise ValueError(f"Unsupported MP4 sidecar schema in {path}")
             records = {}
             for file_idx, item in enumerate(payload["files"]):
                 arrays = {
@@ -944,7 +1005,7 @@ class EpisodeVideoManifest:
                 mp4 = Mp4Index.from_dict(item["mp4"], arrays)
                 records[item["file_path"]] = VideoFileRecord(item["file_path"], int(item["file_size"]), mp4)
         with EpisodeVideoManifest._FILE_SIDECAR_CACHE_LOCK:
-            EpisodeVideoManifest._FILE_SIDECAR_CACHE[cache_key] = records
+            EpisodeVideoManifest._FILE_SIDECAR_CACHE[cache_key] = (signature, records)
         return records
 
     def camera_id(self, camera_key: str) -> int:
@@ -1058,7 +1119,7 @@ class ExactCoveragePool:
     def resident(self) -> list[int]:
         return list(self._remaining)
 
-    def __iter__(self) -> "ExactCoveragePool":
+    def __iter__(self) -> ExactCoveragePool:
         return self
 
     def __next__(self) -> tuple[int, int]:
@@ -1097,7 +1158,12 @@ class EpisodeByteCache:
         native_http_retries: int = 4,
         native_http_subranges: int = 1,
         open_decoders: bool = True,
+        max_open_decoders: int = 64,
     ):
+        if byte_budget <= 0:
+            raise ValueError("byte_budget must be positive")
+        if max_open_decoders <= 0:
+            raise ValueError("max_open_decoders must be positive")
         self.manifest = manifest
         self.fetcher = make_range_fetcher(
             data_root,
@@ -1110,9 +1176,12 @@ class EpisodeByteCache:
         )
         self.byte_budget = byte_budget
         self.open_decoders = open_decoders
+        self.max_open_decoders = max_open_decoders
         self._pool = ThreadPoolExecutor(max_workers=workers)
         self._cache: OrderedDict[tuple[int, str], dict[str, Any]] = OrderedDict()
+        self._decoders: OrderedDict[tuple[int, str], Any] = OrderedDict()
         self._futures: dict[tuple[int, str], Future[dict[str, Any]]] = {}
+        self._retained_episodes: dict[int, int] = {}
         self._bytes = 0
         self._lock = threading.Lock()
         self._timing_totals = {
@@ -1124,10 +1193,12 @@ class EpisodeByteCache:
         }
 
     def close(self) -> None:
-        self._pool.shutdown(wait=True)
+        self._pool.shutdown(wait=True, cancel_futures=True)
         with self._lock:
             self._cache.clear()
+            self._decoders.clear()
             self._futures.clear()
+            self._retained_episodes.clear()
             self._bytes = 0
         self.fetcher.close()
 
@@ -1141,9 +1212,34 @@ class EpisodeByteCache:
         for camera_key in self.manifest.video_keys:
             self._submit(episode_index, camera_key)
 
+    def retain_episode(self, episode_index: int) -> None:
+        with self._lock:
+            self._retained_episodes[episode_index] = self._retained_episodes.get(episode_index, 0) + 1
+
+    def release_episode(self, episode_index: int) -> None:
+        with self._lock:
+            count = self._retained_episodes.get(episode_index, 0)
+            if count <= 1:
+                self._retained_episodes.pop(episode_index, None)
+            else:
+                self._retained_episodes[episode_index] = count - 1
+            self._evict_locked()
+
+    @property
+    def resident_bytes(self) -> int:
+        with self._lock:
+            return self._bytes
+
+    @property
+    def open_decoder_count(self) -> int:
+        with self._lock:
+            return len(self._decoders)
+
     def ensure_ready(self, episode_index: int) -> None:
         for camera_key in self.manifest.video_keys:
             self.get_bytes(episode_index, camera_key)
+            if self.open_decoders:
+                self.get_decoder(episode_index, camera_key)
 
     def is_ready(self, episode_index: int) -> bool:
         """Non-blocking: True when every camera of the episode is fetched (cached or future done).
@@ -1164,20 +1260,30 @@ class EpisodeByteCache:
     def get_bytes(self, episode_index: int, camera_key: str) -> bytes:
         return self._get_entry(episode_index, camera_key)["bytes"]
 
-    def get_decoder(self, episode_index: int, camera_key: str):
+    def get_decoder(self, episode_index: int, camera_key: str) -> Any:
+        key = (episode_index, camera_key)
         entry = self._get_entry(episode_index, camera_key)
-        decoder = entry.get("decoder")
-        if decoder is None:
-            decoder = open_video_decoder(io.BytesIO(entry["bytes"]))
-            entry["decoder"] = decoder
+        with self._lock:
+            decoder = self._decoders.get(key)
+            if decoder is not None:
+                self._decoders.move_to_end(key)
+                return decoder
+
+        decoder = open_video_decoder(io.BytesIO(entry["bytes"]))
+        with self._lock:
+            existing = self._decoders.get(key)
+            if existing is not None:
+                self._decoders.move_to_end(key)
+                return existing
+            self._decoders[key] = decoder
+            while len(self._decoders) > self.max_open_decoders:
+                self._decoders.popitem(last=False)
         return decoder
 
     def get_frames(self, episode_index: int, camera_key: str, timestamps: list[float]):
         span = self.manifest.lookup(episode_index, camera_key)
         local_ts = [ts - span.source_start_pts for ts in timestamps]
         decoder = self.get_decoder(episode_index, camera_key)
-        if hasattr(decoder, "get_frames_played_at"):
-            return decoder.get_frames_played_at(local_ts).data
         metadata = decoder.metadata
         fps = getattr(metadata, "average_fps", None)
         if fps is None:
@@ -1224,7 +1330,13 @@ class EpisodeByteCache:
                 return existing
             self._cache[key] = entry
             self._bytes += len(entry["bytes"])
-            self._evict_locked()
+            try:
+                self._evict_locked()
+            except MemoryError:
+                failed_entry = self._cache.pop(key, None)
+                if failed_entry is not None:
+                    self._bytes -= len(failed_entry["bytes"])
+                raise
             timings = entry.pop("_timings", None)
             if timings is not None:
                 self._timing_totals["lookup_s"] += timings["lookup_s"]
@@ -1235,9 +1347,18 @@ class EpisodeByteCache:
             return entry
 
     def _evict_locked(self) -> None:
-        while self._bytes > self.byte_budget and self._cache:
-            _key, entry = self._cache.popitem(last=False)
+        while self._bytes > self.byte_budget:
+            key = next(
+                (candidate for candidate in self._cache if candidate[0] not in self._retained_episodes),
+                None,
+            )
+            if key is None:
+                raise MemoryError(
+                    f"Retained episode bytes exceed byte budget ({self._bytes} > {self.byte_budget})"
+                )
+            entry = self._cache.pop(key)
             self._bytes -= len(entry["bytes"])
+            self._decoders.pop(key, None)
 
     def _fetch_and_synthesize(self, episode_index: int, camera_key: str) -> dict[str, Any]:
         lookup_start = time.perf_counter()
@@ -1263,15 +1384,12 @@ class EpisodeByteCache:
         synthesize_s = time.perf_counter() - synthesize_start
         entry: dict[str, Any] = {
             "bytes": mp4_bytes,
-            "decoder": None,
             "_timings": {
                 "lookup_s": lookup_s,
                 "fetch_s": fetch_s,
                 "synthesize_s": synthesize_s,
             },
         }
-        if self.open_decoders:
-            entry["decoder"] = open_video_decoder(io.BytesIO(mp4_bytes))
         return entry
 
 
@@ -1283,32 +1401,6 @@ def open_video_decoder(file_like_or_bytesio, frame_mappings=None):
     return VideoDecoder(file_like_or_bytesio, seek_mode="approximate")
 
 
-def assert_hf_hub_range_cache_branch() -> None:
-    """Fail unless huggingface_hub was installed from the required range-cache branch."""
-
-    try:
-        dist = metadata.distribution("huggingface_hub")
-    except metadata.PackageNotFoundError as exc:
-        raise AssertionError("huggingface_hub is not installed") from exc
-
-    candidates = []
-    direct_url = dist.read_text("direct_url.json")
-    if direct_url:
-        candidates.append(direct_url)
-        with contextlib.suppress(json.JSONDecodeError):
-            parsed = json.loads(direct_url)
-            candidates.append(str(parsed.get("url", "")))
-            candidates.append(str(parsed.get("vcs_info", {}).get("requested_revision", "")))
-            candidates.append(str(parsed.get("vcs_info", {}).get("commit_id", "")))
-
-    text = "\n".join(candidates)
-    if "feat/hffs-cache-cdn-range-reads" not in text:
-        raise AssertionError(
-            "huggingface_hub must be installed from "
-            "git+https://github.com/huggingface/huggingface_hub.git@feat/hffs-cache-cdn-range-reads"
-        )
-
-
 @dataclass
 class StageTimer:
     fetch_ms: float = 0.0

src/lerobot/streaming/mp4.py

@@ -6,6 +6,8 @@
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 
+"""MP4 indexing and in-memory episode synthesis primitives."""
+
 from __future__ import annotations
 
 import struct

src/lerobot/streaming/sidecar.py

@@ -0,0 +1,176 @@
+# 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
+
+"""Revision-safe lifecycle for locally cached MP4 byte-index sidecars."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import re
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from uuid import uuid4
+
+from filelock import FileLock, Timeout
+
+from lerobot.streaming.episode_video import EpisodeVideoManifest
+
+SIDECAR_SCHEMA_VERSION = 2
+
+
+class SidecarLockTimeoutError(TimeoutError):
+    """Raised when another process does not finish a sidecar build in time."""
+
+
+@dataclass(frozen=True)
+class SidecarSpec:
+    repo_id: str
+    revision: str
+    data_root: str
+    source_files: tuple[tuple[str, int | None], ...]
+    schema_version: int = SIDECAR_SCHEMA_VERSION
+
+    def __post_init__(self) -> None:
+        if not self.repo_id:
+            raise ValueError("repo_id must not be empty")
+        if not self.revision:
+            raise ValueError("revision must not be empty")
+        normalized = tuple(
+            sorted((str(path), None if size is None else int(size)) for path, size in self.source_files)
+        )
+        if any(not path or size is not None and size < 0 for path, size in normalized):
+            raise ValueError("source file paths must be non-empty and sizes must be non-negative")
+        object.__setattr__(self, "source_files", normalized)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "schema_version": self.schema_version,
+            "repo_id": self.repo_id,
+            "revision": self.revision,
+            "data_root": self.data_root,
+            "source_files": [{"path": path, "size": size} for path, size in self.source_files],
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, object]) -> SidecarSpec:
+        source_files = data.get("source_files")
+        if not isinstance(source_files, list):
+            raise ValueError("MP4 sidecar source_files must be a list")
+        parsed_files: list[tuple[str, int | None]] = []
+        for item in source_files:
+            if not isinstance(item, dict) or not isinstance(item.get("path"), str):
+                raise ValueError("Invalid MP4 sidecar source file entry")
+            size = item.get("size")
+            parsed_files.append((item["path"], None if size is None else int(size)))
+        return cls(
+            schema_version=int(data["schema_version"]),
+            repo_id=str(data["repo_id"]),
+            revision=str(data["revision"]),
+            data_root=str(data["data_root"]),
+            source_files=tuple(parsed_files),
+        )
+
+    def with_source_files(self, source_files: tuple[tuple[str, int], ...]) -> SidecarSpec:
+        return SidecarSpec(
+            repo_id=self.repo_id,
+            revision=self.revision,
+            data_root=self.data_root,
+            source_files=source_files,
+            schema_version=self.schema_version,
+        )
+
+    def matches(self, candidate: SidecarSpec) -> bool:
+        if (
+            self.schema_version != candidate.schema_version
+            or self.repo_id != candidate.repo_id
+            or self.revision != candidate.revision
+            or self.data_root != candidate.data_root
+        ):
+            return False
+        expected = dict(self.source_files)
+        actual = dict(candidate.source_files)
+        if expected.keys() != actual.keys():
+            return False
+        return all(size is None or actual[path] == size for path, size in expected.items())
+
+
+SidecarBuilder = Callable[[Path, SidecarSpec], None]
+SidecarDownloader = Callable[[Path, SidecarSpec], bool]
+
+
+def sidecar_cache_path(cache_root: str | Path, spec: SidecarSpec) -> Path:
+    identity = json.dumps(
+        {
+            "schema_version": spec.schema_version,
+            "repo_id": spec.repo_id,
+            "revision": spec.revision,
+            "data_root": spec.data_root,
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    digest = hashlib.sha256(identity.encode()).hexdigest()[:16]
+    repo_slug = re.sub(r"[^A-Za-z0-9_.-]+", "--", spec.repo_id).strip("-") or "dataset"
+    revision_slug = re.sub(r"[^A-Za-z0-9_.-]+", "-", spec.revision).strip("-")[:32] or "revision"
+    return (
+        Path(cache_root).expanduser() / repo_slug / f"mp4-v{spec.schema_version}-{revision_slug}-{digest}.npz"
+    )
+
+
+def ensure_mp4_sidecar(
+    spec: SidecarSpec,
+    cache_root: str | Path,
+    *,
+    build: SidecarBuilder,
+    download: SidecarDownloader | None = None,
+    lock_timeout_s: float = 30 * 60,
+) -> Path:
+    """Return a valid local sidecar, downloading or building it exactly once.
+
+    This function never uploads. ``download`` and ``build`` must write only to the temporary path
+    provided to them; a validated file becomes visible at the cache path through ``os.replace``.
+    """
+
+    destination = sidecar_cache_path(cache_root, spec)
+    if EpisodeVideoManifest.validate_file_sidecar(destination, spec):
+        return destination
+
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    lock_path = destination.with_suffix(f"{destination.suffix}.lock")
+    try:
+        with FileLock(lock_path, timeout=lock_timeout_s):
+            if EpisodeVideoManifest.validate_file_sidecar(destination, spec):
+                return destination
+
+            temporary = destination.parent / f".{destination.name}.{uuid4().hex}.tmp.npz"
+            try:
+                if download is not None:
+                    logging.info("Looking for published MP4 sidecar for %s@%s", spec.repo_id, spec.revision)
+                    if download(temporary, spec) and EpisodeVideoManifest.validate_file_sidecar(
+                        temporary, spec
+                    ):
+                        os.replace(temporary, destination)
+                        return destination
+                    temporary.unlink(missing_ok=True)
+
+                logging.info("Building MP4 sidecar for %s@%s", spec.repo_id, spec.revision)
+                build(temporary, spec)
+                if not EpisodeVideoManifest.validate_file_sidecar(temporary, spec):
+                    raise ValueError("Built MP4 sidecar failed revision and source validation")
+                os.replace(temporary, destination)
+                return destination
+            finally:
+                temporary.unlink(missing_ok=True)
+    except Timeout as exc:
+        raise SidecarLockTimeoutError(
+            f"Timed out waiting {lock_timeout_s:g}s for MP4 sidecar lock {lock_path}"
+        ) from exc