Recording & datasets¶

from strands_robots import Robot

sim = Robot("so100")
sim.start_recording(repo_id="user/my_dataset", task="pick up the cube", fps=50)
sim.run_policy(robot_name="so100", instruction="pick up the cube",
               policy_provider="mock", duration=10.0)
sim.stop_recording()
# LeRobot v3 dataset written to ~/.strands_robots/datasets/

start_recording requires [lerobot]. Without it, use start_cameras_recording for plain MP4.

`fps` must equal the rollout's `control_frequency`¶

The recorder captures one frame per control step and never decimates, so the rate frames arrive at is the rollout's control_frequency - and LeRobot derives every timestamp from the dataset's declared fps positionally (timestamp = frame_index / fps). A differing fps therefore cannot be honored, only mislabelled, so it is refused:

sim.start_recording(repo_id="user/my_dataset", task="t", fps=30)
sim.run_policy(robot_name="so100", policy_provider="mock")   # default 50.0 Hz
# -> "run_policy: the active recording declares 30 fps but this rollout captures
#     at control_frequency=50 Hz. [...] a 1.667x distortion of the episode
#     duration [...] Align the two rates: pass control_frequency=30 to
#     run_policy(), or record at the rollout's rate"

The refusal lands before any frame is written, so nothing is lost - pass either rate. It matters beyond the label: that per-frame interval is the control period a policy trains on, and replay_episode derives its per-frame physics budget from the dataset rate, so a mislabelled episode also replays at the wrong speed. To record at a lower rate than you control at, run the rollout at that rate - there is no decimating recorder.

PolicyRunner is drivable directly, and the engine's check is not on that path, so PolicyRunner.run / PolicyRunner.evaluate apply the same rule themselves. They raise ValueError rather than returning an error dict, because a direct caller has no tool envelope to read:

runner = PolicyRunner(sim)
runner.run("so100", policy, control_frequency=50.0, on_frame=hook)
# -> ValueError: PolicyRunner.run: the active recording declares 30 fps but this
#    rollout captures at control_frequency=50 Hz. [...] pass control_frequency=30
#    to PolicyRunner.run()

The rule holds whichever call comes first. start_policy returns while its rollout keeps running, so a recording can be opened against a rollout already in flight - and on the defaults (fps=30 against control_frequency=50.0) that recorded a 1.667x mislabelled episode with every call reporting success. start_recording refuses the same disagreement, before creating the dataset:

sim.start_policy(robot_name="so100", policy_provider="mock")   # default 50.0 Hz
sim.start_recording(repo_id="user/my_dataset", task="t", fps=30)
# -> "start_recording: this recording would declare 30 fps but a rollout is
#     already running ('so100' at 50 Hz). [...] Align the two rates: record at
#     the rollout's rate (start_recording(fps=50)), or restart the rollout at the
#     recording's rate (stop_policy(robot_name='so100'), then
#     start_policy(..., control_frequency=30))"

Rollouts running at different rates are refused outright, even when fps matches one of them: their frames interleave into one episode whose single declared rate cannot describe both, so there is no fps to pass instead. Stop all but one, or restart them at a common control_frequency.

Selecting which cameras to record¶

By default every camera in the scene is recorded into the dataset - including the implicit default free camera that exists even before you call add_camera. A policy that declares a fixed set of image features (e.g. SmolVLA expects exactly observation.images.camera1/camera2/camera3) then trains against a dataset that carries a stray observation.images.default view it never asked for, and the extra MP4 stream bloats every episode.

Pass cameras= to record exactly the views the policy expects:

sim.add_camera(name="camera1", ...)
sim.add_camera(name="camera2", ...)
sim.add_camera(name="camera3", ...)
sim.start_recording(
    repo_id="user/my_dataset", task="pick up the cube", fps=50,
    cameras=["camera1", "camera2", "camera3"],   # drops the implicit 'default'
)

The dataset schema then declares only those three image features. Names may be given in raw MuJoCo form (arm0/wrist_cam) or schema-safe form (arm0__wrist_cam); an unknown name fails loudly and lists the available cameras rather than silently recording the wrong set. Omit cameras= to keep the legacy behavior of recording every camera - in that case a one-time warning is logged when the implicit default overview camera is swept in alongside your real sensor cameras, so the stray view is never recorded silently.

cameras= is a list of distinct camera names, and every surface that accepts one - start_recording, render_all, and the plain-MP4 start_cameras_recording / start_cameras_recording_synchronous - enforces that shape up front. Two mistakes are refused rather than guessed at, because neither can be honored as written:

A single name passed as a bare string. cameras="wrist" is iterable per character, so it would be read as five cameras, one per letter. Wrap it in a list: cameras=["wrist"].
A repeated name. cameras=["wrist", "wrist"] cannot mean one camera and cannot mean two. In a dataset schema it collapses to a single column, so the recording declares fewer views than were asked for; in render_all it renders the same view twice; and in a plain-MP4 recording it opens a second encoder on the one output path, so the camera is rendered and appended twice per capture tick and the artifact ledger reports two files where one exists.

A Mapping is refused for the same reason - it is iterable over its keys, so its values would be silently discarded. cameras=None keeps its "every camera" meaning.

Where the dataset is written (`root` / `overwrite`)¶

root is the on-disk directory for the dataset (defaults to the LeRobot cache under repo_id when omitted). Passing an existing empty directory - for example one returned by tempfile.mkdtemp() - is accepted and recorded into:

import tempfile
root = tempfile.mkdtemp()                       # existing empty dir
sim.start_recording(repo_id="user/my_dataset", root=root, fps=30)   # records here

When root already contains a LeRobotDataset (a meta/ directory), start_recording resumes it and appends new episodes unless overwrite=True, which wipes and recreates it. A root that exists, is not a LeRobotDataset, and is not empty is left untouched and reported as an error rather than clobbered - pass overwrite=True or choose a new/empty root.

A resume inherits the existing dataset's schema, so fps must match the rate it was created at - a resume cannot change it. Requesting a different rate is refused, naming the on-disk value, rather than appending frames timestamped on the dataset's timebase instead of the one they were captured at:

sim.start_recording(repo_id="user/my_dataset", root=root, fps=60)  # dataset is 30 fps
# -> error: "dataset fps differs: on-disk=30 vs requested=60
#            (a resumed dataset keeps its on-disk rate; pass fps=30 to append at it)"

Pass fps=30 to append at the dataset's rate, or overwrite=True to record a fresh dataset at 60.

When you drive recording through the run_policy tool (which owns the start_recording -> rollout -> stop_recording cycle), forward the same subset with dataset_cameras=:

from strands_robots.tools.run_policy import run_policy

run_policy(
    simulation=sim,
    robot_name="so101",
    policy_provider="lerobot_local",
    instruction="pick up the cube",
    n_episodes=1,
    dataset_root="/tmp/my_dataset",
    dataset_cameras=["camera1", "camera2", "camera3"],  # drops the implicit 'default'
)

When set, dataset_cameras is forwarded as start_recording(cameras=...), which both the MuJoCo and Newton backends support, so the subset is applied identically on either engine. Omit it (the default None) to record every scene camera - the default path forwards no cameras kwarg at all.

To also capture a rollout MP4 (the visual artifact for review or VLM defect-checking), pass the same video={...} config the Simulation.run_policy facade accepts - the tool forwards it per episode:

run_policy(
    simulation=sim,
    robot_name="so101",
    policy_provider="lerobot_local",
    instruction="pick up the cube",
    n_episodes=3,
    dataset_root="/tmp/my_dataset",
    video={"path": "/tmp/rollout.mp4", "fps": 30, "camera": "camera1",
           "width": 640, "height": 480},
)

video["path"] is required to enable recording; a falsy/absent path disables it. Only the keys above are accepted: any other key (filename, resolution, ...) is rejected with an error listing the accepted set, so a mistyped option cannot silently produce a rollout with no video or a video at the wrong resolution. For n_episodes > 1 an _ep{i} suffix is inserted before the extension (/tmp/rollout_ep0.mp4, _ep1, ...) so episodes do not overwrite one another - matching the facade's own multi-episode naming. The returned {"json": {...}} block carries video_paths, the list of MP4s that actually landed on disk (only existing files are reported, never the requested paths).

run_policy tool rollout video (SmolVLA on a simulated SO-101, MuJoCo headless)

A SmolVLA-on-SO-101 rollout recorded through the run_policy tool's video= config (MuJoCo headless, MUJOCO_GL=egl).

Multi-episode recording¶

A recording session is one dataset. The simplest way to collect N episodes in one session is run_policy(n_episodes=N) - it runs N rollouts back-to-back, flushes a dataset episode boundary after each, and resets the sim between episodes for you:

sim.start_recording(repo_id="user/my_dataset", task="pick up the cube", fps=50)
sim.run_policy(robot_name="so100", instruction="pick up the cube",
               policy_provider="mock", n_steps=60, n_episodes=20)
sim.stop_recording()
# -> 20 episodes, each with its own episode_index / length / from_index / to_index

n_steps (or duration) is the per-episode horizon. reset_between=False chains episodes from the previous end state instead of resetting. When a seed is given it is offset per episode (seed + i) for reproducible-yet-distinct rollouts, and a video={...} config is written per episode to a path with _ep{i} inserted before the extension so episodes do not overwrite one another. The aggregate result carries n_episodes_completed, episodes_saved, total_steps, and a per-episode list in its {"json": {...}} block.

If you need full control over each rollout (different instructions, custom randomization, conditional logic between episodes), drive the loop yourself and call save_episode() after each rollout to flush it as its own episode:

sim.start_recording(repo_id="user/my_dataset", task="pick up the cube", fps=50)
for _ in range(20):
    sim.reset()
    sim.run_policy(robot_name="so100", instruction="pick up the cube",
                   policy_provider="mock", n_steps=60)
    sim.save_episode()        # flush this rollout as one episode
sim.stop_recording()          # flushes any trailing rollout automatically

save_episode is idempotent on an empty buffer, so it is safe to call unconditionally inside a loop. LeRobot computes stats.json per episode and then aggregates, so per-rollout boundaries keep dataset statistics correct across the reset() teleport between rollouts.

reset() is itself an episode boundary during a recording: if frames are buffered when you call it, reset() flushes them as their own episode before re-initializing the world (it reports the saved episode in its result text). This means a bare run_policy + reset collection loop already produces one episode per rollout - the explicit save_episode() is optional when you reset between rollouts:

sim.start_recording(repo_id="user/my_dataset", task="pick up the cube", fps=50)
for _ in range(20):
    sim.run_policy(robot_name="so100", instruction="pick up the cube",
                   policy_provider="mock", n_steps=60)
    sim.reset()               # flushes this rollout as one episode, then resets
sim.stop_recording()          # flushes any trailing rollout automatically
# -> 20 episodes

Without n_episodes, an explicit save_episode(), or a reset() between rollouts, all 20 rollouts append to the same buffer and stop_recording flushes them as a single episode_index=0 (1200 steps in one episode). To DISCARD a partial rollout instead of flushing it on the next reset(), call clear_episode_buffer() first.

Verifying episode count¶

An LLM agent narrating "20 episodes recorded" is not proof: a single run_policy(n_episodes=1) (or 20 looped tool calls into one open buffer) produces one merged episode_index=0 mega-episode while the agent believes it recorded 20. Never trust agent narration for dataset structure - verify against the on-disk metadata. After stop_recording, call verify_dataset_episodes:

sim.stop_recording()
result = sim.verify_dataset_episodes(expected=20)
assert result["status"] == "success"   # else MISMATCH, fail loud

It checks two independent sources of truth and requires them to AGREE:

the parquet under meta/episodes/**/*.parquet (the distinct episode_index set - the ground truth), and
the total_episodes header in meta/info.json.

status is "error" when the parquet count differs from expected OR when the parquet disagrees with info.json (an internally inconsistent dataset, e.g. an interrupted finalize - sources_agree is then False), so a dataset that happens to match expected on one source but not the other still fails. The {"json": {...}} block carries expected, actual, info_total_episodes, sources_agree, episode_indices, and total_frames for programmatic CI gating. The pure-pyarrow read_dataset_episode_indices(root) exposes the same facts without instantiating a LeRobotDataset.

The same check runs from the shell against any LeRobot dataset on disk, with an exit code suitable for CI:

strands-robots verify-dataset /path/to/dataset --expected 20   # exit 0 pass, 1 fail
strands-robots verify-dataset /path/to/dataset --json          # machine-readable report
strands-robots verify-dataset /path/to/dataset --no-check-videos  # skip the per-episode MP4 checks

verify-dataset reuses the same pure-pyarrow read_dataset_episode_indices helper (no lerobot import) and flags four failure modes: the mega-episode (fewer distinct episodes than --expected), meta/info.json total_episodes / total_frames drifting from the parquet ground truth (caught even without --expected), any episode below --min-frames (default 1), and - unless --no-check-videos is passed - any per-episode video file that is missing or empty on disk. The last check is the video-modality sibling of the mega-episode class: a dataset can carry the right episode count yet have no pixels because the recorder's video encoder failed or the MP4 streams were never written. It resolves each camera's MP4 from meta/info.json's video_path template and the episode parquet's videos/<key>/chunk_index / file_index columns, and reports the count it checked in video_files_checked. The programmatic form is strands_robots.verify_dataset.verify_dataset(root, expected=None, min_frames=1, check_videos=True), which returns the same report dict.

verify-dataset always produces a report - it never crashes on the corruption it exists to flag. A corrupt or foreign meta/episodes parquet, a non-v3 video_path template, or a truncated / unreadable MP4 is reported as a problem string in the report (and a non-zero exit code), not surfaced as a raw traceback.

Corruption confined to SOME episode parquet shards (the usual outcome of an interrupted rsync or hub download) is localised rather than fatal: each unreadable shard is named as a problem, the readable shards still supply total_episodes / frames_per_episode, and the info.json, video, and dead-column checks still run against them. Only a meta/episodes tree with no readable shard at all reports zero episodes. The same holds for verify_dataset_episodes, which additionally refuses to certify a dataset with unreadable shards even when the readable count matches expected - the count is then only a lower bound, reported in the unreadable_files diagnostics.

Recording paths¶

Method	Extra needed	Output
`start_recording` / `stop_recording`	`[lerobot]`	LeRobot v3 (parquet + MP4)
`save_episode`	`[lerobot]`	Close current rollout as one episode (call once per `run_policy` for N episodes)
`start_cameras_recording` / `stop_cameras_recording`	`[sim-mujoco]` alone	Plain MP4, no parquet

fps, width, height and max_frames_per_camera on the plain-MP4 recorders must be positive whole numbers - the same domain run_policy(video={...}), start_recording(fps=...) and the shared encoder strands_robots.rendering.encode_clip enforce. An unusable value (fps=0, a negative frame cap) is a structured error naming the parameter rather than a recording that reports success and writes no file. Omit width/height to use each camera's configured resolution.

Encoding frames directly through encode_clip(frames, path, fps=...) follows the same rule and raises ValueError for a rate it cannot honor - a fractional or non-positive fps previously produced a clip at some other rate (the GIF writer clamped it to 1 fps, ffmpeg substituted its own default) or no file at all, with the output path still returned. encode_clip also raises RuntimeError when the encoder wrote no clip despite accepting the frames, so a returned path always names a clip that exists.

Video codec (H.264 default, AV1 opt-in)¶

start_recording (and DatasetRecorder.create/resume) default to vcodec="h264". H.264 is universally decodable - including by OpenCV's cv2.VideoCapture, which is what many downstream VLM video readers use to watch a recorded episode. The recorded per-camera MP4s therefore reopen and decode everywhere without a transcode step:

import cv2
cap = cv2.VideoCapture(".../videos/observation.images.base/chunk-000/file-000.mp4")
ok, frame = cap.read()   # H.264: ok is True, frame is a real (H, W, 3) array

Pass vcodec="libsvtav1" to opt into AV1 for smaller files in storage-constrained training pipelines. AV1 reads back fine through LeRobot's own loader (torchcodec/pyav), but OpenCV wheels commonly lack an AV1 decoder and silently yield 0 frames (VideoCapture.read() returns False immediately even though the stream has frames) - so avoid AV1 if anything in your pipeline reads the videos through OpenCV.

The flat vcodec value accepts either codec names (h264, hevc, libsvtav1) or ffmpeg encoder names (libx264, libx265); it is normalized to whatever the installed LeRobot version's encoder config expects. An unsupported codec fails loudly rather than silently reverting to the default.

DatasetRecorder direct API¶

from strands_robots.dataset_recorder import DatasetRecorder

recorder = DatasetRecorder.create(
    repo_id="user/my_dataset",
    fps=30,
    robot_type="so100",
    # When recording from a real LeRobot hardware robot pass the schema dicts
    # straight through:
    #   robot_features=robot.observation_features,
    #   action_features=robot.action_features,
    # When recording from a sim Robot (no `observation_features` attr), pass
    # `joint_names=[...]` instead - the recorder builds the schema for you.
    camera_keys=["default"],
    joint_names=["joint_1", "joint_2", "joint_3", "joint_4", "joint_5", "joint_6"],
    task="pick up the red cube",
    # root=None → ~/.strands_robots/datasets/
    # vcodec="h264", streaming_encoding=True, image_writer_threads=4
)

for step in control_loop:
    recorder.add_frame(observation, action, task="pick up the red cube")
recorder.save_episode()
recorder.finalize()
recorder.push_to_hub(tags=["so100", "sim"], private=False)

Append to existing dataset (requires lerobot>=0.5.2):

recorder = DatasetRecorder.resume(repo_id="user/my_dataset", task="pick up the blue cube")
recorder.add_frame(observation, action)
recorder.save_episode()
recorder.finalize()

Schema column names must be distinct¶

camera_keys, joint_names and action_names each declare the recorded dataset's column names, so each must be a list of distinct, non-blank names. create() refuses anything else before it touches the on-disk target, so a refused overwrite=True call leaves an existing dataset intact:

DatasetRecorder.create(repo_id="user/d", joint_names="gripper")     # ValueError
DatasetRecorder.create(repo_id="user/d", camera_keys=["front", "front"])  # ValueError

Both mistakes used to be accepted and only surfaced in the recorded data:

A single name passed as a bare string is iterable per character, so joint_names="gripper" declared seven columns (g, r, i, p, p, e, r). add_frame reads each declared name out of the observation, and none of those names is in it, so every column recorded 0.0 for the whole episode - create(), add_frame(), save_episode() and finalize() all succeeded. Wrap a single name in a list: joint_names=["gripper"].
A repeated name collapses where it keys a dict and doubles where it indexes a position. camera_keys=["front", "front"] declared one camera column for the two the caller asked for; joint_names=["j1", "j2", "j2"] recorded j2 twice and the joint the caller meant not at all.

None and [] still mean "not supplied" - the schema is then derived from robot_features / action_features, or from joint_names for the action columns.

Re-recording into an existing `repo_id`¶

DatasetRecorder.create() builds a fresh dataset. If the resolved dataset directory already exists, LeRobot's create() would raise a bare FileExistsError (its mkdir uses exist_ok=False). create() resolves this up front, matching the start_recording facade:

overwrite=True wipes the existing directory and creates a fresh dataset.
overwrite=False (default) on an existing dataset (a dir containing meta/) raises a clear FileExistsError naming overwrite=True (fresh) and resume() (append) - not a cryptic LeRobot error, and never a silent clobber.
An existing empty directory (e.g. from tempfile.mkdtemp()) is cleared so create() does not dead-end on its own existence guard.
A non-empty non-dataset directory raises ValueError rather than deleting unrelated files.

# Re-run a capture script into the same repo_id, replacing the old dataset:
recorder = DatasetRecorder.create(
    repo_id="user/my_dataset", fps=30, joint_names=[...], overwrite=True,
)

Use resume() (not create(overwrite=True)) when you want to append episodes to the existing dataset instead of replacing it.

A frame the recorder cannot write fails the rollout¶

DatasetRecorder is fail-fast by default (strict=True): if the underlying LeRobotDataset write fails, it raises strands_robots.dataset_recorder.RecordingFrameError instead of continuing. When the recorder is driven by run_policy, that ends the rollout with status="error" naming the frame the recording stopped being complete at:

Policy failed: dataset add_frame failed after 7 frame(s) written; the recording
is incomplete from this frame on (strict=True, so it is not dropped silently):
<the underlying error>

Continuing past a lost frame is not a smaller failure. Timestamps are derived positionally from the declared fps, so the surviving frames are re-stamped into a shorter span than they were captured over - a rollout that loses every other frame at 50 Hz produces an episode labelled at 2x speed, with no gap to detect.

Construct the recorder with strict=False to trade that for best-effort recording: a failed write is counted in dropped_frame_count, warned about at WARNING (on the 1st, 2nd, 4th, 8th ... failure so a 50 Hz loop cannot flood the log), and the rollout continues.

Instance methods¶

Method	What
`add_frame(observation, action, task=None, camera_keys=None)`	Append one timestep
`save_episode()`	Flush buffer as a new episode
`clear_episode_buffer()`	Discard current episode
`finalize()`	Write metadata, stats, close writers
`push_to_hub(tags=None, private=False)`	Upload to a versioned HF dataset repo
`sync_to_bucket(bucket, run_id=None, private=True)`	Sync to a mutable HF Storage Bucket (`hf://buckets/...`) — Xet-deduped collection target; needs the `hf` CLI. `bucket` (`name` or `org/name`) and `run_id` (single segment) are allowlist-validated (`[A-Za-z0-9._-]`, no traversal) before the sync

Read back¶

Fully materialized (downloads everything):

from lerobot.datasets.lerobot_dataset import LeRobotDataset

ds = LeRobotDataset(repo_id="user/my_dataset", root="/tmp/my_dataset")
print(len(ds), ds[0].keys())

Replay an episode¶

sim.replay_episode(repo_id, robot_name=..., episode=0, root=None, speed=1.0) plays a recorded episode back through the sim: each recorded frame is one control step, applied via send_action and integrated for a full control period derived from the dataset fps, so a position-servo robot reproduces the recorded trajectory. speed scales only the wall-clock playback rate.

Each recorded action index is bound to an action key. By default those are robot_action_keys(robot_name) — the robot's actuator keys, which is the ordering the recorder writes the action column in. Pass action_key_map only when the dataset's action ordering differs:

sim.replay_episode(
    "user/my_dataset",
    robot_name="so101",
    root="/tmp/my_dataset",
    action_key_map=["1", "2", "3", "4", "5", "6"],  # one key per action index
)

action_key_map must be a non-empty list/tuple of unique strings whose length equals the recorded action vector's width. A bare string (consumed one key per character), a non-string entry, a duplicate key, or a width mismatch is rejected with an actionable error before the dataset is fetched — never truncated to fit.

A "success" status means every frame reached the actuators. If a recorded action cannot be applied — e.g. the mapped keys resolve to no actuator on this robot — the replay aborts at that frame and returns status="error" with the frame index, how many frames were applied, and the unresolved keys:

result = sim.replay_episode("user/my_dataset", robot_name="so101", action_key_map=["wrong"] * 6)
result["status"]                                  # "error"
result["content"][1]["json"]["unresolved_keys"]   # ['wrong', ...]
result["content"][1]["json"]["frames_applied"]    # 0

Stream back (no full download)¶

sim.stream_dataset() is the in-process counterpart to start_recording / stop_recording — it reads frames lazily from the Hub (or a local root) via LeRobot's StreamingLeRobotDataset. Camera frames are decoded on the fly from the MP4 shards; state/action come from the parquet shards.

from strands_robots import Robot

sim = Robot("so100")
reader = sim.stream_dataset(
    "user/my_dataset",                 # or a local repo_id + root=
    root="/tmp/my_dataset",
    delta_timestamps={                 # optional: stacked time windows + *_is_pad masks
        "observation.state": [-0.0667, -0.0333, 0.0],
        "action": [0.0, 0.0333, 0.0667],
    },
    shuffle=False,                     # chronological for replay/eval
)
print(reader.num_episodes, reader.num_frames, reader.fps)
for frame in reader:
    ...

# torch DataLoader (shuffles INTERNALLY — do not pass shuffle=True):
for batch in reader.dataloader(batch_size=64, num_workers=4):
    ...

Equivalently, the standalone reader: from strands_robots import StreamingDatasetReader.

Useful kwargs (forwarded to StreamingLeRobotDataset, version-tolerant): episodes=[...] (subset without download), buffer_size, max_num_shards, return_uint8=True (default; halves frame bandwidth), and drop_videos=True (proprio-only — skips video decode entirely, so it works on edge devices without a torchcodec wheel; requires delta_timestamps with at least one non-video key, otherwise open() raises ValueError rather than silently streaming video anyway).

One kwarg is not tolerant-forwarded because its absence changes semantics: repo_type="bucket" requires lerobot>=0.6.1 — on older versions open() raises RuntimeError instead of silently streaming from the versioned dataset namespace (a different storage system).

For training, the upstream trainer uses the same engine:

python -m lerobot.scripts.lerobot_train --policy.type=act \
  --dataset.repo_id=user/my_dataset --dataset.streaming=true --num_workers=4

macOS: video streaming needs Homebrew ffmpeg on the dyld path. import strands_robots auto-fixes this (zero-touch); disable with STRANDS_ROBOTS_NO_DYLD_SHIM=1. See the README "Recording & streaming datasets" section.