"""Unit tests for ``strands_robots.dataset_recorder.DatasetRecorder``.

These tests exercise the wrapper logic that does NOT require a real LeRobot
dataset by injecting a fake dataset object, so they run on a minimal env
(``lerobot`` not installed). They cover the partial-episode discard behaviour
and the add_frame() control-loop transform (schema-ordered flattening, camera
normalization, drop accounting), plus episode/finalize/push lifecycle.
"""

import logging
from pathlib import Path

import numpy as np
import pytest

from strands_robots.dataset_recorder import DatasetRecorder


class _FakeDatasetWithClear:
    """Fake LeRobot dataset exposing ``clear_episode_buffer`` (preferred path)."""

    def __init__(self):
        self.repo_id = "local/fake"
        self.cleared = 0

    def clear_episode_buffer(self):
        self.cleared += 1


class _FakeDatasetWithCreate:
    """Fake dataset exposing only ``create_episode_buffer`` (fallback path)."""

    def __init__(self):
        self.repo_id = "local/fake"
        self.episode_buffer = {"frames": [1, 2, 3]}
        self.created = 0

    def create_episode_buffer(self):
        self.created += 1
        return {}


class _FakeDatasetNoClear:
    """Fake dataset exposing no buffer-reset surface (warn-only path)."""

    def __init__(self):
        self.repo_id = "local/fake"


def _recorder_for(dataset) -> DatasetRecorder:
    rec = DatasetRecorder(dataset=dataset)
    rec.episode_frame_count = 5  # simulate 5 frames buffered for the open episode
    rec.frame_count = 5
    return rec


def test_clear_episode_buffer_prefers_native_clear():
    ds = _FakeDatasetWithClear()
    rec = _recorder_for(ds)

    assert rec.clear_episode_buffer() is True
    assert ds.cleared == 1
    # Next episode starts at frame 0; cumulative frame_count is untouched
    # (those frames were only ever in the open buffer, not flushed to disk).
    assert rec.episode_frame_count == 0


def test_clear_episode_buffer_falls_back_to_create_buffer():
    ds = _FakeDatasetWithCreate()
    rec = _recorder_for(ds)

    assert rec.clear_episode_buffer() is True
    assert ds.created == 1
    assert ds.episode_buffer == {}
    assert rec.episode_frame_count == 0


def test_clear_episode_buffer_warns_when_no_surface(caplog):
    ds = _FakeDatasetNoClear()
    rec = _recorder_for(ds)

    import logging

    with caplog.at_level(logging.WARNING, logger="strands_robots.dataset_recorder"):
        result = rec.clear_episode_buffer()

    assert result is False
    # Counter still resets so reporting does not carry over the discarded frames.
    assert rec.episode_frame_count == 0
    assert any("partial episode" in r.message for r in caplog.records)


def test_clear_episode_buffer_swallows_dataset_error(caplog):
    """A failure inside the dataset's clear must not mask the original abort."""

    class _Boom:
        repo_id = "local/fake"

        def clear_episode_buffer(self):
            raise RuntimeError("buffer is wedged")

    rec = _recorder_for(_Boom())

    import logging

    with caplog.at_level(logging.WARNING, logger="strands_robots.dataset_recorder"):
        result = rec.clear_episode_buffer()

    assert result is False
    assert rec.episode_frame_count == 0


def test_clear_episode_buffer_ascii_only_warnings(caplog):
    """Recorder log strings must be plain ASCII (project string-hygiene rule)."""
    rec = _recorder_for(_FakeDatasetNoClear())

    import logging

    with caplog.at_level(logging.WARNING, logger="strands_robots.dataset_recorder"):
        rec.clear_episode_buffer()

    for record in caplog.records:
        record.getMessage().encode("ascii")  # raises if any non-ASCII glyph leaked


# add_frame() control-loop transformation behaviour
#
# These tests inject a fake LeRobot dataset that captures every frame handed to
# add_frame(), so they assert the recorder's observable output (the frame dict
# shape, vector ordering, camera normalization, drop accounting) without a real
# lerobot install.


class _CapturingDataset:
    """Fake LeRobot dataset that records frames and exposes a feature schema."""

    def __init__(self, features: dict, *, fail_add: bool = False):
        self.repo_id = "local/fake"
        self.root = "/tmp/local-fake"
        self.features = features
        self.frames: list[dict] = []
        self.saved = 0
        self.finalized = 0
        self.pushed: dict | None = None
        self._fail_add = fail_add

    def add_frame(self, frame):
        if self._fail_add:
            raise RuntimeError("disk full")
        self.frames.append(frame)

    def save_episode(self):
        self.saved += 1

    def finalize(self):
        self.finalized += 1

    def push_to_hub(self, tags=None, private=False):
        self.pushed = {"tags": tags, "private": private}


def _state_action_features(state_names, action_names) -> dict:
    return {
        "observation.state": {"dtype": "float32", "names": state_names},
        "action": {"dtype": "float32", "names": action_names},
    }


def test_add_frame_orders_state_and_action_by_schema():
    """State/action vectors follow the feature-schema order, not dict order."""
    ds = _CapturingDataset(_state_action_features(["j1", "j2", "j3"], ["j1", "j2", "j3"]))
    rec = DatasetRecorder(dataset=ds, task="pick")

    # Pass observation/action keys in a deliberately scrambled order.
    rec.add_frame(
        observation={"j3": 3.0, "j1": 1.0, "j2": 2.0},
        action={"j2": 0.2, "j3": 0.3, "j1": 0.1},
    )

    assert rec.frame_count == 1
    assert rec.episode_frame_count == 1
    frame = ds.frames[0]
    assert np.allclose(frame["observation.state"], [1.0, 2.0, 3.0])
    assert np.allclose(frame["action"], [0.1, 0.2, 0.3])
    assert frame["observation.state"].dtype == np.float32
    assert frame["task"] == "pick"


def test_add_frame_fills_missing_keys_with_zero():
    """A joint absent from the observation contributes 0.0 at its schema slot."""
    ds = _CapturingDataset(_state_action_features(["j1", "j2", "j3"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)

    rec.add_frame(observation={"j1": 1.0, "j3": 3.0}, action={"j1": 0.5}, task="t")

    frame = ds.frames[0]
    assert np.allclose(frame["observation.state"], [1.0, 0.0, 3.0])
    assert np.allclose(frame["action"], [0.5])


def test_add_frame_flattens_vector_valued_entries():
    """List / ndarray observation values are flattened into the state vector."""
    ds = _CapturingDataset(_state_action_features(["pose", "grip"], ["cmd"]))
    rec = DatasetRecorder(dataset=ds)

    rec.add_frame(
        observation={"pose": [1.0, 2.0, 3.0], "grip": 0.5},
        action={"cmd": np.array([0.1, 0.2])},
        task="t",
    )

    frame = ds.frames[0]
    assert np.allclose(frame["observation.state"], [1.0, 2.0, 3.0, 0.5])
    assert np.allclose(frame["action"], [0.1, 0.2])


def test_add_frame_records_numpy_scalar_state_and_action():
    """numpy scalars (np.float32 from indexing a qpos/ctrl array) are recorded
    as scalar columns, not silently dropped.

    MuJoCo observations and policy actions routinely hand back numpy scalar
    types (the element type you get from ``np.asarray(qpos)[i]``), which are
    neither Python ``float`` nor 0-dim ``np.ndarray``. They must still land in
    the flattened state/action vectors - dropping them would record
    wrong-length (or entirely missing) vectors with no error raised.
    """
    ds = _CapturingDataset(_state_action_features(["j1", "j2", "j3"], ["j1", "j2", "j3"]))
    rec = DatasetRecorder(dataset=ds, task="pick")

    rec.add_frame(
        observation={"j1": np.float32(1.0), "j2": np.float64(2.0), "j3": np.int32(3)},
        action={"j1": np.float32(0.1), "j2": np.float32(0.2), "j3": np.float32(0.3)},
    )

    frame = ds.frames[0]
    assert "observation.state" in frame, "numpy-scalar state was dropped"
    assert "action" in frame, "numpy-scalar action was dropped"
    assert np.allclose(frame["observation.state"], [1.0, 2.0, 3.0])
    assert np.allclose(frame["action"], [0.1, 0.2, 0.3])
    assert frame["observation.state"].dtype == np.float32
    assert frame["action"].dtype == np.float32


def test_add_frame_converts_float_images_to_uint8():
    """A float image in [0, 1] is scaled to uint8 HWC for LeRobot."""
    feats = {"observation.images.cam": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(dataset=ds)

    img = np.ones((4, 4, 3), dtype=np.float32)  # all-white in [0, 1]
    rec.add_frame(observation={"cam": img}, action={}, task="t")

    out = ds.frames[0]["observation.images.cam"]
    assert out.dtype == np.uint8
    assert out.shape == (4, 4, 3)
    assert np.array_equal(out, np.full((4, 4, 3), 255, dtype=np.uint8))


def test_add_frame_normalizes_namespaced_camera_keys():
    """A 'arm0/wrist' camera key is rewritten to the declared 'arm0__wrist'."""
    feats = {"observation.images.arm0__wrist": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(dataset=ds)

    img = np.zeros((2, 2, 3), dtype=np.uint8)
    rec.add_frame(observation={"arm0/wrist": img}, action={}, task="t")

    frame = ds.frames[0]
    assert "observation.images.arm0__wrist" in frame
    assert "observation.images.arm0/wrist" not in frame


def test_add_frame_strips_undeclared_cameras():
    """A camera not in the feature schema is dropped to avoid 'extra features'."""
    feats = {"observation.images.declared": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(dataset=ds)

    img = np.zeros((2, 2, 3), dtype=np.uint8)
    rec.add_frame(
        observation={"declared": img, "ghost": img},
        action={},
        task="t",
    )

    frame = ds.frames[0]
    assert "observation.images.declared" in frame
    assert "observation.images.ghost" not in frame


def test_add_frame_strict_reraises_on_dataset_error():
    """strict=True (default) propagates a dataset add_frame failure."""
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]), fail_add=True)
    rec = DatasetRecorder(dataset=ds, strict=True)

    with pytest.raises(RuntimeError, match="disk full"):
        rec.add_frame(observation={"j1": 1.0}, action={"j1": 0.1}, task="t")
    assert rec.frame_count == 0


def test_add_frame_non_strict_counts_drops_without_raising(caplog):
    """strict=False swallows the error, counts the drop, and logs ASCII-only."""
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]), fail_add=True)
    rec = DatasetRecorder(dataset=ds, strict=False)

    with caplog.at_level(logging.WARNING, logger="strands_robots.dataset_recorder"):
        rec.add_frame(observation={"j1": 1.0}, action={"j1": 0.1}, task="t")

    assert rec.dropped_frame_count == 1
    assert rec.frame_count == 0
    for record in caplog.records:
        record.getMessage().encode("ascii")


def test_add_frame_noop_when_closed():
    """A closed recorder ignores add_frame instead of corrupting the dataset."""
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)
    rec._closed = True

    rec.add_frame(observation={"j1": 1.0}, action={"j1": 0.1}, task="t")

    assert ds.frames == []
    assert rec.frame_count == 0


def test_save_episode_reports_per_episode_frame_count():
    """save_episode reports frames since the last save and resets the counter."""
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)

    rec.add_frame(observation={"j1": 1.0}, action={"j1": 0.1}, task="t")
    rec.add_frame(observation={"j1": 2.0}, action={"j1": 0.2}, task="t")
    result = rec.save_episode()

    assert result["status"] == "success"
    assert result["episode"] == 1
    assert result["episode_frames"] == 2
    assert result["total_frames"] == 2
    assert rec.episode_frame_count == 0  # reset for the next episode

    # A second episode reports only its own frames; total stays monotonic.
    rec.add_frame(observation={"j1": 3.0}, action={"j1": 0.3}, task="t")
    result2 = rec.save_episode()
    assert result2["episode_frames"] == 1
    assert result2["total_frames"] == 3


def test_save_episode_poisons_recorder_on_failure():
    """A failed save closes the recorder so later frames cannot corrupt data."""

    class _BadSave(_CapturingDataset):
        def save_episode(self):
            raise RuntimeError("encode failed")

    ds = _BadSave(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)
    rec.add_frame(observation={"j1": 1.0}, action={"j1": 0.1}, task="t")

    result = rec.save_episode()

    assert result["status"] == "error"
    assert rec._closed is True
    # save_episode on a closed recorder returns a clean error, not a crash.
    assert rec.save_episode()["status"] == "error"


def test_finalize_is_idempotent_and_closes():
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)

    rec.finalize()
    rec.finalize()  # second call is a no-op

    assert rec._closed is True
    assert ds.finalized == 1


def test_push_to_hub_success_and_failure():
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)
    rec.episode_count = 2
    rec.frame_count = 50

    ok = rec.push_to_hub(tags=["sim"], private=True)
    assert ok == {
        "status": "success",
        "repo_id": "local/fake",
        "episodes": 2,
        "frames": 50,
    }
    assert ds.pushed == {"tags": ["sim"], "private": True}

    class _BadPush(_CapturingDataset):
        def push_to_hub(self, tags=None, private=False):
            raise RuntimeError("network down")

    bad = DatasetRecorder(dataset=_BadPush(_state_action_features(["j1"], ["j1"])))
    bad.episode_count = 1
    bad.frame_count = 10
    err = bad.push_to_hub()
    assert err["status"] == "error"
    assert "network down" in err["message"]


def test_push_to_hub_refuses_empty_dataset_no_frames():
    """A recorder that captured no frames must not hit the Hub.

    Regression: push_to_hub used to publish unconditionally, so a rollout that
    never fed the recorder (e.g. driven by eval_policy / a bare step loop)
    produced a Hub repo containing only meta/info.json. The guard now returns a
    structured error and makes no Hub call.
    """
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)
    # frame_count and episode_count are 0 by construction.
    result = rec.push_to_hub(tags=["sim"])

    assert result["status"] == "error"
    assert "empty dataset" in result["message"]
    assert ds.pushed is None  # the underlying dataset.push_to_hub was never called


def test_push_to_hub_refuses_frames_without_saved_episode():
    """Frames buffered but no episode flushed is still an empty dataset on disk."""
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)
    rec.frame_count = 30
    rec.episode_count = 0  # save_episode was never called (or it failed)

    result = rec.push_to_hub()

    assert result["status"] == "error"
    assert "empty dataset" in result["message"]
    assert ds.pushed is None


def test_repo_id_root_and_repr_properties():
    ds = _CapturingDataset(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)
    rec.episode_count = 1
    rec.frame_count = 7

    assert rec.repo_id == "local/fake"
    assert rec.root == "/tmp/local-fake"
    rep = repr(rec)
    assert "local/fake" in rep and "episodes=1" in rep and "frames=7" in rep


# ---------------------------------------------------------------------------
# load_lerobot_episode: frame-range resolution for replay
# ---------------------------------------------------------------------------
# load_lerobot_episode() resolves the [start, start+length) frame window for a
# given episode index across three LeRobot dataset shapes:
#   1. episode_data_index present (the fast path: from/to index tensors),
#   2. no episode_data_index but meta.episodes carries per-episode lengths,
#   3. neither usable -> last-resort frame scan keyed on episode_index.
# It guards out-of-range and empty episodes with ValueError. These tests inject
# a fake LeRobotDataset so the real body runs without lerobot installed.


class _IndexCell:
    """Mimics a tensor cell exposing ``.item()`` (as torch index tensors do)."""

    def __init__(self, value: int) -> None:
        self._value = value

    def item(self) -> int:
        return self._value


class _IndexColumn:
    """Indexable column of _IndexCell, mimicking dataset.episode_data_index['from']."""

    def __init__(self, values: list[int]) -> None:
        self._values = values

    def __getitem__(self, idx: int) -> _IndexCell:
        return _IndexCell(self._values[idx])


class _FakeMeta:
    def __init__(self, total_episodes=None, episodes=None) -> None:
        if total_episodes is not None:
            self.total_episodes = total_episodes
        if episodes is not None:
            self.episodes = episodes


class _FakeDatasetWithIndex:
    """Fast path: exposes episode_data_index with from/to columns."""

    def __init__(self, repo_id, root=None) -> None:
        self.repo_id = repo_id
        self.root = root
        self.meta = _FakeMeta(total_episodes=3)
        self.episode_data_index = {
            "from": _IndexColumn([0, 10, 25]),
            "to": _IndexColumn([10, 25, 40]),
        }


class _FakeDatasetMetaLengths:
    """No episode_data_index; lengths live in meta.episodes."""

    def __init__(self, repo_id, root=None) -> None:
        self.repo_id = repo_id
        self.root = root
        self.meta = _FakeMeta(
            episodes=[{"length": 10}, {"length": 15}, {"length": 5}],
        )


class _RaisingIndex:
    """Subscriptable that raises on lookup, simulating an unusable index.

    Raises ``KeyError`` (the idiomatic ``LookupError`` for a failed
    subscription) so the helper's broad ``except Exception`` still catches it
    and falls through to the frame-scan path, mirroring a real dataset whose
    ``episode_data_index`` is present but missing the expected columns.
    """

    def __getitem__(self, key):
        raise KeyError(key)


class _FakeDatasetScan:
    """Neither fast path usable: forces the frame-scan fallback.

    episode_data_index access raises, so the helper falls through to scanning
    frames by their episode_index field.
    """

    def __init__(self, repo_id, root=None) -> None:
        self.repo_id = repo_id
        self.root = root
        self.meta = _FakeMeta(total_episodes=2)
        # episode 0 -> frames [0,1,2]; episode 1 -> frames [3,4]
        self._frames = [
            {"episode_index": _IndexCell(0)},
            {"episode_index": _IndexCell(0)},
            {"episode_index": _IndexCell(0)},
            {"episode_index": _IndexCell(1)},
            {"episode_index": _IndexCell(1)},
        ]

        # episode_data_index exists (so hasattr passes) but subscripting it
        # raises, forcing the helper into the frame-scan fallback.
        self.episode_data_index = _RaisingIndex()

    def __len__(self) -> int:
        return len(self._frames)

    def __getitem__(self, idx):
        return self._frames[idx]


def _patch_lerobot_dataset(monkeypatch, fake_cls) -> None:
    """Inject a fake LeRobotDataset into the import target used by the helper."""
    import sys
    import types

    module = types.ModuleType("lerobot.datasets.lerobot_dataset")
    setattr(module, "LeRobotDataset", fake_cls)
    monkeypatch.setitem(sys.modules, "lerobot.datasets.lerobot_dataset", module)


def test_load_episode_uses_episode_data_index_fast_path(monkeypatch):
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetWithIndex)
    ds, start, length = dr.load_lerobot_episode("user/data", episode=1)

    assert start == 10
    assert length == 15
    assert ds.repo_id == "user/data"


def test_load_episode_first_episode_window(monkeypatch):
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetWithIndex)
    _, start, length = dr.load_lerobot_episode("user/data", episode=0)

    assert start == 0
    assert length == 10


def test_load_episode_falls_back_to_meta_lengths(monkeypatch):
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetMetaLengths)
    # episode 2 starts after episodes 0 (10) + 1 (15) = 25, with length 5.
    _, start, length = dr.load_lerobot_episode("user/data", episode=2)

    assert start == 25
    assert length == 5


def test_load_episode_scans_frames_as_last_resort(monkeypatch):
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetScan)
    _, start, length = dr.load_lerobot_episode("user/data", episode=1)

    # episode 1 occupies frames 3 and 4.
    assert start == 3
    assert length == 2


def test_load_episode_rejects_out_of_range(monkeypatch):
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetWithIndex)
    with pytest.raises(ValueError, match="out of range"):
        dr.load_lerobot_episode("user/data", episode=3)


def test_load_episode_rejects_empty_episode(monkeypatch):
    from strands_robots import dataset_recorder as dr

    class _EmptyEpisode:
        def __init__(self, repo_id, root=None) -> None:
            self.meta = _FakeMeta(episodes=[{"length": 0}])

    _patch_lerobot_dataset(monkeypatch, _EmptyEpisode)
    with pytest.raises(ValueError, match="no frames"):
        dr.load_lerobot_episode("user/data", episode=0)


def test_load_episode_scan_breaks_after_target_episode(monkeypatch):
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetScan)
    # episode 0 occupies frames 0,1,2; the scan must stop at frame 3 (episode 1).
    _, start, length = dr.load_lerobot_episode("user/data", episode=0)

    assert start == 0
    assert length == 3


class TestBuildFeaturesSchema:
    """Behavioral tests for ``DatasetRecorder._build_features`` -- the function
    that turns robot/action feature descriptors into the LeRobot v3 ``features``
    schema every recording is created with.

    The contract under test is the *shape of the emitted schema dict*: which
    keys appear, their ``dtype``/``shape``/``names``, and how the state and
    action dimensions are derived from the several mutually-exclusive input
    sources (explicit feature dicts, a flat ``joint_names`` list, or the
    action-mirrors-state fallback). These are pure-logic assertions -- no
    LeRobot install is required because ``_build_features`` is a classmethod
    that only manipulates plain dicts.
    """

    def test_camera_keys_emit_video_features_with_default_dims(self):
        """Each camera key becomes an ``observation.images.<name>`` video
        feature; absent per-camera dims fall back to the global video size."""
        features = DatasetRecorder._build_features(
            camera_keys=["top", "wrist"],
            video_height=480,
            video_width=640,
        )

        assert features["observation.images.top"] == {
            "dtype": "video",
            "shape": (3, 480, 640),
            "names": ["channels", "height", "width"],
        }
        assert features["observation.images.wrist"]["shape"] == (3, 480, 640)

    def test_camera_dims_override_per_camera_resolution(self):
        """A per-camera entry in ``camera_dims`` overrides the global size for
        that camera only; others keep the fallback."""
        features = DatasetRecorder._build_features(
            camera_keys=["top", "wrist"],
            camera_dims={"top": (240, 320)},
            video_height=480,
            video_width=640,
        )

        assert features["observation.images.top"]["shape"] == (3, 240, 320)
        assert features["observation.images.wrist"]["shape"] == (3, 480, 640)

    def test_use_videos_false_emits_image_dtype(self):
        """``use_videos=False`` records still frames (``image`` dtype) rather
        than encoded video."""
        features = DatasetRecorder._build_features(camera_keys=["top"], use_videos=False)

        assert features["observation.images.top"]["dtype"] == "image"

    def test_robot_features_drive_state_excluding_cameras(self):
        """Scalar entries in ``robot_features`` become the ``observation.state``
        vector; image/video entries are excluded from the state count."""
        features = DatasetRecorder._build_features(
            robot_features={
                "shoulder.pos": {"dtype": "float32"},
                "elbow.pos": {"dtype": "float32"},
                "front_cam": {"dtype": "video"},
            },
        )

        state = features["observation.state"]
        assert state["dtype"] == "float32"
        assert state["shape"] == (2,)
        assert state["names"] == ["shoulder.pos", "elbow.pos"]

    def test_joint_names_drive_state_when_no_robot_features(self):
        """With no ``robot_features``, a flat ``joint_names`` list defines the
        state dimension and names."""
        features = DatasetRecorder._build_features(joint_names=["j1", "j2", "j3"])

        assert features["observation.state"]["shape"] == (3,)
        assert features["observation.state"]["names"] == ["j1", "j2", "j3"]

    def test_action_features_drive_action_excluding_cameras(self):
        """Explicit ``action_features`` define the action vector, excluding any
        image/video columns."""
        features = DatasetRecorder._build_features(
            action_features={
                "shoulder.pos": {"dtype": "float32"},
                "gripper.pos": {"dtype": "float32"},
                "debug_cam": {"dtype": "image"},
            },
        )

        action = features["action"]
        assert action["shape"] == (2,)
        assert action["names"] == ["shoulder.pos", "gripper.pos"]

    def test_action_names_define_action_columns_independent_of_joints(self):
        """An explicit ``action_names`` list defines the action columns even
        when it diverges from ``joint_names``. This is the actuator-keyed sim
        path: a policy emits actuator short-names (``shoulder_pan_act``) while
        observation.state stays joint-keyed (``shoulder_pan``), so the recorded
        action columns must follow the actuator keys add_frame receives -- not
        the joint names, which would never match and record all zeros."""
        features = DatasetRecorder._build_features(
            joint_names=["shoulder_pan", "shoulder_lift", "elbow"],
            action_names=["shoulder_pan_act", "shoulder_lift_act", "elbow_act"],
        )

        # State stays joint-keyed.
        assert features["observation.state"]["names"] == ["shoulder_pan", "shoulder_lift", "elbow"]
        # Action follows the actuator keys.
        assert features["action"]["shape"] == (3,)
        assert features["action"]["names"] == ["shoulder_pan_act", "shoulder_lift_act", "elbow_act"]

    def test_action_features_take_precedence_over_action_names(self):
        """``action_features`` (explicit typed schema) wins over the
        ``action_names`` fallback when both are supplied."""
        features = DatasetRecorder._build_features(
            action_features={"a": {"dtype": "float32"}, "b": {"dtype": "float32"}},
            action_names=["x", "y", "z"],
            joint_names=["j1", "j2"],
        )

        assert features["action"]["names"] == ["a", "b"]

    def test_action_names_fall_back_to_joint_names_when_absent(self):
        """With no ``action_names`` the action columns mirror ``joint_names``
        (the actuator-mirrors-joints case for stock arms, unchanged behaviour)."""
        features = DatasetRecorder._build_features(joint_names=["j1", "j2", "j3"])

        assert features["action"]["names"] == ["j1", "j2", "j3"]

    def test_action_mirrors_state_when_only_robot_features_given(self):
        """With state from ``robot_features`` but no action source, the action
        feature mirrors the state dimension and names."""
        features = DatasetRecorder._build_features(
            robot_features={"a": {"dtype": "float32"}, "b": {"dtype": "float32"}},
        )

        assert features["action"]["shape"] == (2,)
        assert features["action"]["names"] == ["a", "b"]
        # The mirror is a copy, not an alias of the state names list.
        assert features["action"]["names"] is not features["observation.state"]["names"]

    def test_non_dict_feature_values_count_as_scalar_state(self):
        """A feature whose value is not a dict (e.g. a bare descriptor) is
        treated as a scalar state column, not skipped."""
        features = DatasetRecorder._build_features(
            robot_features={"raw_scalar": "float32", "cam": {"dtype": "video"}},
        )

        assert features["observation.state"]["shape"] == (1,)
        assert features["observation.state"]["names"] == ["raw_scalar"]

    def test_empty_inputs_emit_empty_schema(self):
        """With no cameras, features, or joint names there is nothing to record
        -- the schema is empty rather than carrying zero-dim entries."""
        features = DatasetRecorder._build_features()

        assert features == {}


# ---------------------------------------------------------------------------
# DatasetRecorder.resume() -- the multi-episode append entry point.
#
# resume() opens an EXISTING on-disk LeRobotDataset for appending (the plain
# constructor returns a read-only dataset). Its body must:
#   1. hard-fail with a clear RuntimeError on LeRobot versions lacking resume(),
#   2. route the requested vcodec version-tolerantly -- pass ``vcodec=`` when the
#      installed resume() takes it, else wrap it in a ``VideoEncoderConfig`` for
#      the ``camera_encoder=`` kwarg (0.5.2+), warning if that config is absent,
#   3. forward the optional streaming/threads/backend kwargs only when supported,
#   4. seed episode/frame counters from the existing dataset so totals report
#      correctly, swallowing a malformed meta rather than crashing.
# These tests inject fake LeRobotDataset classes whose ``resume`` classmethods
# expose different signatures, so the real body runs without lerobot installed.


class _ResumeMeta:
    def __init__(self, total_episodes, total_frames) -> None:
        self.total_episodes = total_episodes
        self.total_frames = total_frames


class _FakeDatasetVcodecResume:
    """resume() accepts the legacy ``vcodec=`` kwarg directly."""

    last_resume_kwargs: dict = {}

    def __init__(self, repo_id, root=None, meta=None) -> None:
        self.repo_id = repo_id
        self.root = root
        self.meta = meta or _ResumeMeta(total_episodes=2, total_frames=40)

    @classmethod
    def resume(cls, repo_id, root=None, vcodec="libsvtav1", streaming_encoding=True):
        cls.last_resume_kwargs = {
            "repo_id": repo_id,
            "root": root,
            "vcodec": vcodec,
            "streaming_encoding": streaming_encoding,
        }
        return cls(repo_id, root=root)


class _FakeDatasetCameraEncoderResume:
    """resume() takes ``camera_encoder=`` (0.5.2+) plus thread/backend kwargs."""

    last_resume_kwargs: dict = {}

    def __init__(self, repo_id, root=None, meta=None) -> None:
        self.repo_id = repo_id
        self.root = root
        self.meta = meta or _ResumeMeta(total_episodes=5, total_frames=123)

    @classmethod
    def resume(
        cls,
        repo_id,
        root=None,
        camera_encoder=None,
        image_writer_threads=4,
        video_backend="auto",
    ):
        cls.last_resume_kwargs = {
            "repo_id": repo_id,
            "root": root,
            "camera_encoder": camera_encoder,
            "image_writer_threads": image_writer_threads,
            "video_backend": video_backend,
        }
        return cls(repo_id, root=root)


def _install_video_encoder_config(monkeypatch):
    """Provide a stub ``lerobot.configs.video.VideoEncoderConfig`` and return a
    list that captures each instance constructed, so a test can assert the
    recorder wrapped its vcodec into the config."""
    import sys
    import types

    constructed = []

    class _VideoEncoderConfig:
        def __init__(self, vcodec) -> None:
            self.vcodec = vcodec
            constructed.append(self)

    module = types.ModuleType("lerobot.configs.video")
    module.VideoEncoderConfig = _VideoEncoderConfig
    monkeypatch.setitem(sys.modules, "lerobot.configs.video", module)
    return constructed


def test_resume_raises_clear_error_when_lerobot_lacks_resume(monkeypatch):
    """Pre-0.5.2 LeRobot has no resume(); the recorder must say so explicitly."""
    from strands_robots import dataset_recorder as dr

    class _NoResumeDataset:
        def __init__(self, repo_id, root=None) -> None:
            self.repo_id = repo_id

    _patch_lerobot_dataset(monkeypatch, _NoResumeDataset)
    with pytest.raises(RuntimeError, match="no LeRobotDataset.resume"):
        dr.DatasetRecorder.resume("user/data")


def test_resume_passes_vcodec_directly_when_supported(monkeypatch):
    """When resume() takes ``vcodec=``, the recorder forwards it as-is and only
    adds the optional kwargs the signature actually declares."""
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetVcodecResume)
    recorder = dr.DatasetRecorder.resume("user/data", root="/tmp/ds", vcodec="libx264", task="pick")

    sent = _FakeDatasetVcodecResume.last_resume_kwargs
    # The flat ``vcodec`` surface validates against the codec-name allowlist
    # (it rejects "libx264"), so the ffmpeg name is normalized to "h264".
    assert sent["vcodec"] == "h264"
    assert sent["repo_id"] == "user/data"
    assert sent["root"] == "/tmp/ds"
    assert sent["streaming_encoding"] is True
    # video_backend / image_writer_threads are NOT in this signature -> not sent.
    assert "video_backend" not in sent
    assert "image_writer_threads" not in sent
    assert recorder.default_task == "pick"


def test_resume_wraps_vcodec_in_camera_encoder_on_052(monkeypatch):
    """On 0.5.2+ (camera_encoder= kwarg), the vcodec is wrapped in a
    VideoEncoderConfig, and thread/backend kwargs are forwarded."""
    from strands_robots import dataset_recorder as dr

    constructed = _install_video_encoder_config(monkeypatch)
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetCameraEncoderResume)
    dr.DatasetRecorder.resume("user/data", vcodec="libsvtav1", video_backend="pyav")

    sent = _FakeDatasetCameraEncoderResume.last_resume_kwargs
    assert sent["video_backend"] == "pyav"
    assert sent["image_writer_threads"] == 4
    # vcodec must have been wrapped, not passed raw.
    assert "vcodec" not in sent
    assert len(constructed) == 1
    assert sent["camera_encoder"] is constructed[0]
    assert constructed[0].vcodec == "libsvtav1"


def test_resume_warns_when_video_encoder_config_missing(monkeypatch, caplog):
    """If resume() wants camera_encoder= but VideoEncoderConfig can't be
    imported, the recorder warns and proceeds with camera_encoder unset."""
    import sys

    from strands_robots import dataset_recorder as dr

    # Ensure the import target is absent so the import raises ImportError.
    monkeypatch.setitem(sys.modules, "lerobot.configs.video", None)
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetCameraEncoderResume)

    with caplog.at_level("WARNING"):
        dr.DatasetRecorder.resume("user/data", vcodec="libsvtav1")

    sent = _FakeDatasetCameraEncoderResume.last_resume_kwargs
    assert sent["camera_encoder"] is None
    assert any("VideoEncoderConfig" in rec.message for rec in caplog.records)


def test_resume_seeds_counters_from_existing_dataset(monkeypatch):
    """Counters are seeded from the dataset meta so reported totals include the
    episodes/frames already on disk."""
    from strands_robots import dataset_recorder as dr

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetVcodecResume)
    recorder = dr.DatasetRecorder.resume("user/data")

    assert recorder.episode_count == 2
    assert recorder.frame_count == 40


def test_resume_tolerates_unreadable_meta_counters(monkeypatch):
    """A dataset whose meta lacks numeric totals must not crash resume(); the
    counters simply stay at their zero defaults."""
    from strands_robots import dataset_recorder as dr

    class _BadMeta:
        @property
        def total_episodes(self):
            raise AttributeError("no totals on this meta")

    class _FakeDatasetBadMeta(_FakeDatasetVcodecResume):
        def __init__(self, repo_id, root=None, meta=None) -> None:
            super().__init__(repo_id, root=root, meta=_BadMeta())

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetBadMeta)
    recorder = dr.DatasetRecorder.resume("user/data")

    assert recorder.episode_count == 0
    assert recorder.frame_count == 0


# ---------------------------------------------------------------------------
# DatasetRecorder.create() -- the new-dataset entry point.
#
# create() builds a LeRobotDataset from auto-detected features and must route
# the requested vcodec across LeRobot versions exactly as resume() does:
#   * 0.5.0/0.5.1: create(..., vcodec="libsvtav1")
#   * 0.5.2+:      create(..., camera_encoder=VideoEncoderConfig(vcodec=...))
# It also forwards the optional streaming_encoding / image_writer_threads /
# video_backend kwargs only when the installed create() signature declares them,
# and returns a recorder bound to the created dataset and default task. These
# tests inject fake LeRobotDataset classes whose ``create`` classmethods expose
# different signatures, so the real body runs without lerobot installed.


class _FakeDatasetVcodecCreate:
    """create() accepts the legacy ``vcodec=`` kwarg directly (0.5.0/0.5.1)."""

    last_create_kwargs: dict = {}

    def __init__(self, repo_id, root=None) -> None:
        self.repo_id = repo_id
        self.root = root

    @classmethod
    def create(
        cls,
        repo_id,
        fps=30,
        root=None,
        robot_type="unknown",
        features=None,
        use_videos=True,
        image_writer_threads=4,
        vcodec="libsvtav1",
        streaming_encoding=True,
    ):
        cls.last_create_kwargs = {
            "repo_id": repo_id,
            "fps": fps,
            "root": root,
            "robot_type": robot_type,
            "features": features,
            "use_videos": use_videos,
            "image_writer_threads": image_writer_threads,
            "vcodec": vcodec,
            "streaming_encoding": streaming_encoding,
        }
        return cls(repo_id, root=root)


class _FakeDatasetCameraEncoderCreate:
    """create() takes ``camera_encoder=`` (0.5.2+) plus backend kwargs."""

    last_create_kwargs: dict = {}

    def __init__(self, repo_id, root=None) -> None:
        self.repo_id = repo_id
        self.root = root

    @classmethod
    def create(
        cls,
        repo_id,
        fps=30,
        root=None,
        robot_type="unknown",
        features=None,
        use_videos=True,
        image_writer_threads=4,
        camera_encoder=None,
        streaming_encoding=True,
        video_backend="auto",
    ):
        cls.last_create_kwargs = {
            "repo_id": repo_id,
            "features": features,
            "image_writer_threads": image_writer_threads,
            "camera_encoder": camera_encoder,
            "streaming_encoding": streaming_encoding,
            "video_backend": video_backend,
        }
        return cls(repo_id, root=root)


def test_create_passes_vcodec_directly_when_supported(monkeypatch):
    """When create() takes ``vcodec=``, the recorder forwards it as-is and does
    not synthesize a camera_encoder; backend kwargs absent from the signature
    are not sent."""
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetVcodecCreate)
    recorder = DatasetRecorder.create(
        "user/data", fps=50, root="/tmp/ds", joint_names=["j1", "j2"], vcodec="libx264", task="pick"
    )

    sent = _FakeDatasetVcodecCreate.last_create_kwargs
    # The flat ``vcodec`` surface validates against the codec-name allowlist
    # (it rejects "libx264"), so the ffmpeg name is normalized to "h264".
    assert sent["vcodec"] == "h264"
    assert sent["repo_id"] == "user/data"
    assert sent["fps"] == 50
    assert sent["streaming_encoding"] is True
    # camera_encoder / video_backend are NOT in this signature -> not sent.
    assert "camera_encoder" not in sent
    assert "video_backend" not in sent
    assert recorder.default_task == "pick"
    assert recorder.dataset.repo_id == "user/data"


def test_create_wraps_vcodec_in_camera_encoder_on_052(monkeypatch):
    """On 0.5.2+ (camera_encoder= kwarg), the vcodec is wrapped in a
    VideoEncoderConfig and the backend kwargs are forwarded."""
    constructed = _install_video_encoder_config(monkeypatch)
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetCameraEncoderCreate)
    DatasetRecorder.create("user/data", joint_names=["j1"], vcodec="libsvtav1", video_backend="pyav")

    sent = _FakeDatasetCameraEncoderCreate.last_create_kwargs
    assert sent["video_backend"] == "pyav"
    assert sent["image_writer_threads"] == 4
    # vcodec must have been wrapped, not passed raw.
    assert "vcodec" not in sent
    assert len(constructed) == 1
    assert sent["camera_encoder"] is constructed[0]
    assert constructed[0].vcodec == "libsvtav1"


class _VideoBackendProbeCreate:
    """create() declares video_backend with a distinctive sentinel default so a
    test can tell whether the recorder forwarded a value or left it untouched."""

    _UNSET = "__unset__"
    last_create_kwargs: dict = {}

    def __init__(self, repo_id, root=None) -> None:
        self.repo_id = repo_id
        self.root = root

    @classmethod
    def create(
        cls,
        repo_id,
        fps=30,
        root=None,
        robot_type="unknown",
        features=None,
        use_videos=True,
        image_writer_threads=4,
        camera_encoder=None,
        streaming_encoding=True,
        video_backend="__unset__",
    ):
        cls.last_create_kwargs = {"video_backend": video_backend}
        return cls(repo_id, root=root)


class _VideoBackendProbeResume:
    """resume() counterpart of _VideoBackendProbeCreate."""

    _UNSET = "__unset__"
    last_resume_kwargs: dict = {}

    def __init__(self, repo_id, root=None, meta=None) -> None:
        self.repo_id = repo_id
        self.root = root
        self.meta = meta or _ResumeMeta(total_episodes=0, total_frames=0)

    @classmethod
    def resume(
        cls,
        repo_id,
        root=None,
        camera_encoder=None,
        image_writer_threads=4,
        video_backend="__unset__",
    ):
        cls.last_resume_kwargs = {"video_backend": video_backend}
        return cls(repo_id, root=root)


def test_create_omits_video_backend_by_default(monkeypatch):
    """Regression: video_backend defaults to None and must NOT be forwarded to
    LeRobot.create() unless explicitly set.

    ``video_backend`` is LeRobot's video *decode* backend; valid values are
    "torchcodec"/"pyav" ("video_reader" is a deprecated alias). The old default
    of "auto" was never a valid decode backend, so read-back through the created
    dataset (e.g. dataset[0]) raised ``ValueError: Unsupported video backend:
    auto``. With the fix, the default is None and the kwarg is left off so
    LeRobot picks its own platform default. The fake's sentinel default proves
    the recorder sent nothing (pre-fix it forwarded "auto" and this fails).
    """
    _install_video_encoder_config(monkeypatch)
    _patch_lerobot_dataset(monkeypatch, _VideoBackendProbeCreate)

    DatasetRecorder.create("user/data", joint_names=["j1"], vcodec="libsvtav1")

    assert _VideoBackendProbeCreate.last_create_kwargs["video_backend"] == "__unset__"


def test_resume_omits_video_backend_by_default(monkeypatch):
    """Regression: resume() must not forward video_backend when left at its
    None default (see test_create_omits_video_backend_by_default)."""
    _install_video_encoder_config(monkeypatch)
    _patch_lerobot_dataset(monkeypatch, _VideoBackendProbeResume)

    DatasetRecorder.resume("user/data", vcodec="libsvtav1")

    assert _VideoBackendProbeResume.last_resume_kwargs["video_backend"] == "__unset__"


def test_create_forwards_video_backend_when_explicitly_set(monkeypatch):
    """When a caller explicitly passes a valid decode backend, it IS forwarded."""
    _install_video_encoder_config(monkeypatch)
    _patch_lerobot_dataset(monkeypatch, _VideoBackendProbeCreate)

    DatasetRecorder.create("user/data", joint_names=["j1"], vcodec="libsvtav1", video_backend="pyav")

    assert _VideoBackendProbeCreate.last_create_kwargs["video_backend"] == "pyav"


def test_create_warns_when_video_encoder_config_missing(monkeypatch, caplog):
    """If create() wants camera_encoder= but VideoEncoderConfig can't be
    imported, the recorder warns and proceeds with camera_encoder unset rather
    than crashing the recording setup."""
    import sys

    monkeypatch.setitem(sys.modules, "lerobot.configs.video", None)
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetCameraEncoderCreate)

    with caplog.at_level("WARNING"):
        DatasetRecorder.create("user/data", joint_names=["j1"], vcodec="libsvtav1")

    sent = _FakeDatasetCameraEncoderCreate.last_create_kwargs
    assert sent["camera_encoder"] is None
    assert any("VideoEncoderConfig" in rec.message for rec in caplog.records)


def test_create_forwards_optional_kwargs_only_when_supported(monkeypatch):
    """A minimal create() lacking streaming_encoding / video_backend must not be
    handed those kwargs (version-tolerant forwarding, no TypeError)."""

    class _FakeDatasetMinimalCreate:
        last_create_kwargs: dict = {}

        def __init__(self, repo_id, root=None) -> None:
            self.repo_id = repo_id
            self.root = root

        @classmethod
        def create(
            cls,
            repo_id,
            fps=30,
            root=None,
            robot_type="unknown",
            features=None,
            use_videos=True,
            image_writer_threads=4,
            vcodec="libsvtav1",
        ):
            cls.last_create_kwargs = {"repo_id": repo_id, "vcodec": vcodec}
            return cls(repo_id, root=root)

    _patch_lerobot_dataset(monkeypatch, _FakeDatasetMinimalCreate)
    recorder = DatasetRecorder.create("user/data", joint_names=["j1"])

    sent = _FakeDatasetMinimalCreate.last_create_kwargs
    # The default codec "h264" is already an allowlist-valid codec name and is
    # forwarded unchanged onto the flat ``vcodec`` surface.
    assert sent == {"repo_id": "user/data", "vcodec": "h264"}
    assert recorder.dataset.repo_id == "user/data"


def test_create_builds_features_from_joints_and_cameras(monkeypatch):
    """create() derives the LeRobot ``features`` schema from joint_names and
    camera_keys and hands it to the underlying dataset constructor."""
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetVcodecCreate)
    DatasetRecorder.create(
        "user/data",
        joint_names=["shoulder", "elbow"],
        camera_keys=["top"],
        video_height=240,
        video_width=320,
    )

    features = _FakeDatasetVcodecCreate.last_create_kwargs["features"]
    assert features["observation.state"]["names"] == ["shoulder", "elbow"]
    assert features["observation.images.top"]["shape"] == (3, 240, 320)


# camera_key_map remap + camera-key-mismatch diagnostic
#
# Regression coverage for the silent data-loss mode where a policy declares
# image_keys (e.g. "image"/"wrist_image") that never match the names the
# sim/hardware camera streams emit (e.g. "front_camera"/"wrist_camera"): every
# image frame is stripped and the dataset records zero video columns with no
# error. The recorder now (a) accepts a camera_key_map remap and (b) warns once
# when all observed cameras are dropped.


def test_add_frame_camera_key_map_remaps_observed_to_declared():
    """camera_key_map rewrites an observed stream name to the declared schema key."""
    feats = {"observation.images.image": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(dataset=ds, camera_key_map={"front_camera": "image"})

    img = np.zeros((2, 2, 3), dtype=np.uint8)
    rec.add_frame(observation={"front_camera": img}, action={}, task="t")

    frame = ds.frames[0]
    assert "observation.images.image" in frame
    assert "observation.images.front_camera" not in frame


def test_add_frame_camera_key_map_accepts_fully_qualified_keys():
    """camera_key_map entries may be written as observation.images.* keys."""
    feats = {"observation.images.wrist_image": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(
        dataset=ds,
        camera_key_map={"observation.images.wrist_camera": "observation.images.wrist_image"},
    )

    img = np.zeros((2, 2, 3), dtype=np.uint8)
    rec.add_frame(observation={"wrist_camera": img}, action={}, task="t")

    frame = ds.frames[0]
    assert "observation.images.wrist_image" in frame
    assert "observation.images.wrist_camera" not in frame


def test_add_frame_warns_once_when_no_camera_matches(caplog):
    """All observed cameras stripped (none declared) -> one loud diagnostic."""
    feats = {"observation.images.image": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(dataset=ds)

    img = np.zeros((2, 2, 3), dtype=np.uint8)
    with caplog.at_level(logging.WARNING):
        rec.add_frame(observation={"front_camera": img}, action={}, task="t")
        rec.add_frame(observation={"front_camera": img}, action={}, task="t")

    mismatch = [r for r in caplog.records if "match the declared image features" in r.message]
    assert len(mismatch) == 1  # warned once, not per frame
    # The message names the offending observed and declared keys + the remedy.
    assert "front_camera" in mismatch[0].getMessage()
    assert "image" in mismatch[0].getMessage()
    assert "camera_key_map" in mismatch[0].getMessage()
    # The image data was still dropped (no remap supplied).
    assert "observation.images.front_camera" not in ds.frames[0]


def test_add_frame_no_mismatch_warning_on_partial_strip(caplog):
    """A partial strip (one camera matches) is the normal 'ignore extra' path."""
    feats = {"observation.images.declared": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(dataset=ds)

    img = np.zeros((2, 2, 3), dtype=np.uint8)
    with caplog.at_level(logging.WARNING):
        rec.add_frame(observation={"declared": img, "ghost": img}, action={}, task="t")

    assert not [r for r in caplog.records if "match the declared image features" in r.message]
    assert "observation.images.declared" in ds.frames[0]
    assert "observation.images.ghost" not in ds.frames[0]


def test_add_frame_no_mismatch_warning_when_remap_resolves_it(caplog):
    """With a remap that resolves the mismatch, no diagnostic is emitted."""
    feats = {"observation.images.image": {"dtype": "video"}}
    ds = _CapturingDataset(feats)
    rec = DatasetRecorder(dataset=ds, camera_key_map={"front_camera": "image"})

    img = np.zeros((2, 2, 3), dtype=np.uint8)
    with caplog.at_level(logging.WARNING):
        rec.add_frame(observation={"front_camera": img}, action={}, task="t")

    assert not [r for r in caplog.records if "match the declared image features" in r.message]


def test_has_lerobot_dataset_does_not_negatively_cache(monkeypatch):
    """A failed lerobot-availability probe must NOT be cached.

    lerobot availability is a process capability that can transiently fail to
    resolve (a slow/locked import, or a temporarily monkeypatched
    ``sys.modules``). Freezing the first ``False`` permanently disabled
    recording for the rest of the process - ``start_recording`` would report
    ``requires strands-robots[lerobot]`` forever even after the condition
    cleared. The probe must re-attempt the import on the next call.
    """
    import importlib
    import sys

    from strands_robots import dataset_recorder as dr

    # Capture the real module so we can swap it back in a fully reversible way
    # (monkeypatch.setitem restores the original entry on teardown; no delitem
    # that could leave the parent package attribute stale for later tests).
    real_module = importlib.import_module("lerobot.datasets.lerobot_dataset")

    # Start from a clean, un-probed state regardless of implementation.
    monkeypatch.setattr(dr, "_HAS_LEROBOT_DATASET", [], raising=False)

    # Transient failure: the lerobot dataset module does not resolve.
    monkeypatch.setitem(sys.modules, "lerobot.datasets.lerobot_dataset", None)
    assert dr.has_lerobot_dataset() is False

    # Failure clears: the very next call must re-probe and resolve True.
    monkeypatch.setitem(sys.modules, "lerobot.datasets.lerobot_dataset", real_module)
    assert dr.has_lerobot_dataset() is True


def test_load_lerobot_episode_rejects_negative_index():
    """A negative episode index is a programming error, not episode N-1.

    ``load_lerobot_episode`` previously only guarded the upper bound
    (``episode >= num_episodes``); a negative index fell through to Python's
    negative list indexing, silently resolving (and replaying) the LAST
    episode while reporting the negative number. Reject it up front, before
    any dataset construction, so the contract holds even without lerobot
    installed and without a network round-trip.
    """
    from strands_robots.dataset_recorder import load_lerobot_episode

    with pytest.raises(ValueError, match="non-negative"):
        load_lerobot_episode("any/repo", episode=-1)


class TestCodecRouting:
    """Cover ``_codec_create_kwargs`` version-tolerant codec routing.

    LeRobot routes the codec onto different surfaces across releases (flat
    ``vcodec`` on 0.5.0/0.5.1; an encoder-config object on later minors), but
    every one validates against the same codec-name allowlist and rejects the
    ffmpeg names "libx264"/"libx265". Routing the codec onto the wrong (or no)
    surface silently drops the caller's request and falls back to the AV1
    default. These tests assert the requested codec actually reaches the
    surface this LeRobot exposes, with name spellings normalized per surface.
    """

    def test_flat_vcodec_uses_codec_name(self):
        from strands_robots.dataset_recorder import _codec_create_kwargs

        # The flat ``vcodec`` surface validates against the codec-name allowlist
        # in every supported LeRobot (>=0.5.0,<0.6.0), so the codec name is
        # forwarded as-is - NOT mapped to the ffmpeg name "libx264".
        assert _codec_create_kwargs({"vcodec": None}, "h264") == {"vcodec": "h264"}
        # An ffmpeg name is normalized to its allowlist codec name.
        assert _codec_create_kwargs({"vcodec": None}, "libx264") == {"vcodec": "h264"}
        assert _codec_create_kwargs({"vcodec": None}, "libx265") == {"vcodec": "hevc"}
        # AV1 and HW encoders are already codec names and pass through.
        assert _codec_create_kwargs({"vcodec": None}, "libsvtav1") == {"vcodec": "libsvtav1"}
        assert _codec_create_kwargs({"vcodec": None}, "h264_nvenc") == {"vcodec": "h264_nvenc"}

    def test_no_known_codec_surface_returns_empty(self):
        from strands_robots.dataset_recorder import _codec_create_kwargs

        # A signature exposing none of the known codec kwargs -> no routing
        # (recorder falls back to the LeRobot default codec).
        assert _codec_create_kwargs({"repo_id": None, "fps": None}, "h264") == {}

    def test_rgb_encoder_normalizes_ffmpeg_name_to_allowlist(self):
        pytest.importorskip("lerobot")
        rgb = pytest.importorskip("lerobot.configs.video")
        if not hasattr(rgb, "RGBEncoderConfig"):
            pytest.skip("this LeRobot has no RGBEncoderConfig surface")
        from strands_robots.dataset_recorder import _codec_create_kwargs

        # ffmpeg "libx264" must be normalized to the allowlist name "h264",
        # otherwise RGBEncoderConfig rejects it and the codec is dropped.
        out = _codec_create_kwargs({"rgb_encoder": None}, "libx264")
        assert set(out) == {"rgb_encoder"}
        assert out["rgb_encoder"].vcodec == "h264"
        # Canonical "h264" is already allowlist-valid and passes through.
        out2 = _codec_create_kwargs({"rgb_encoder": None}, "h264")
        assert out2["rgb_encoder"].vcodec == "h264"
        # AV1 opt-in survives onto the encoder config.
        out3 = _codec_create_kwargs({"rgb_encoder": None}, "libsvtav1")
        assert out3["rgb_encoder"].vcodec == "libsvtav1"

    def test_rgb_encoder_rejects_unknown_codec_loudly(self):
        pytest.importorskip("lerobot")
        rgb = pytest.importorskip("lerobot.configs.video")
        if not hasattr(rgb, "RGBEncoderConfig"):
            pytest.skip("this LeRobot has no RGBEncoderConfig surface")
        from strands_robots.dataset_recorder import _codec_create_kwargs

        # An unsupported codec must surface LeRobot's ValueError rather than
        # silently reverting to the default codec.
        with pytest.raises(ValueError):
            _codec_create_kwargs({"rgb_encoder": None}, "not_a_codec")

    def test_rgb_encoder_falls_back_when_config_class_missing(self, monkeypatch, caplog):
        """A LeRobot exposing ``rgb_encoder=`` but whose ``RGBEncoderConfig`` is
        unimportable must warn and route no codec surface (falling back to the
        default) rather than crash dataset setup.

        This mirrors the ``camera_encoder`` fallback contract: the two encoder-
        config surfaces are equivalent version-drift shims, so a build that
        renamed/moved ``RGBEncoderConfig`` should degrade identically to one that
        dropped ``VideoEncoderConfig`` - a warning, an empty routing dict, and no
        exception.
        """
        import sys

        from strands_robots.dataset_recorder import _codec_create_kwargs

        # Setting the import target to None makes
        # ``from lerobot.configs.video import RGBEncoderConfig`` raise
        # ImportError, exercising the fallback branch without needing a LeRobot
        # build that actually lacks the class.
        monkeypatch.setitem(sys.modules, "lerobot.configs.video", None)

        with caplog.at_level("WARNING"):
            out = _codec_create_kwargs({"rgb_encoder": None}, "libsvtav1", context="create")

        # No codec surface routed -> recorder uses the LeRobot default codec.
        assert out == {}
        # The warning names the missing config class so the cause is actionable.
        assert any("RGBEncoderConfig" in rec.message for rec in caplog.records)


def test_add_frame_fills_missing_action_key_with_zero():
    """An action key absent from the action dict contributes 0.0 at its slot.

    ``add_frame`` flattens the action into feature-schema order. A control step
    that does not populate every declared action key (e.g. a gripper channel the
    policy left unset) must still emit a full-width action vector with the
    missing slots zeroed - a short vector would silently misalign the recorded
    ``action`` column against its declared names.
    """
    ds = _CapturingDataset(_state_action_features(["j1"], ["a", "b", "c"]))
    rec = DatasetRecorder(dataset=ds)

    # "b" is absent from the action dict entirely; the schema still has 3 slots.
    rec.add_frame(observation={"j1": 1.0}, action={"a": 0.5, "c": 0.7}, task="t")

    frame = ds.frames[0]
    assert np.allclose(frame["action"], [0.5, 0.0, 0.7])
    assert frame["action"].dtype == np.float32


def test_finalize_swallows_dataset_error_and_still_closes(caplog):
    """A dataset ``finalize()`` that raises is logged, not propagated.

    Finalization runs at the end of a recording (often during teardown); a
    writer that has already released its parquet handle must not turn a
    best-effort flush into a crash. The recorder logs a warning, marks itself
    closed, and a subsequent ``finalize()`` is a clean no-op.
    """

    class _FinalizeRaises(_CapturingDataset):
        def finalize(self):
            raise RuntimeError("parquet writer already gone")

    ds = _FinalizeRaises(_state_action_features(["j1"], ["j1"]))
    rec = DatasetRecorder(dataset=ds)

    with caplog.at_level("WARNING"):
        rec.finalize()  # must not raise

    assert rec._closed is True
    assert any("finalize warning" in rec_.message for rec_ in caplog.records)

    # Idempotent after a failed finalize: the second call early-returns.
    rec.finalize()


def test_get_lerobot_dataset_class_raises_clean_importerror_when_unavailable(monkeypatch):
    """``_get_lerobot_dataset_class`` surfaces a clean, actionable ImportError.

    When lerobot's dataset module does not resolve and no test has injected a
    mock class, the resolver must raise an ImportError that names the install
    remedy rather than leaking the raw import failure to the caller.
    """
    import sys

    from strands_robots import dataset_recorder as dr

    # No test-injected mock class, and the real import target does not resolve.
    monkeypatch.setattr(dr, "LeRobotDataset", None, raising=False)
    monkeypatch.setitem(sys.modules, "lerobot.datasets.lerobot_dataset", None)

    with pytest.raises(ImportError, match="pip install lerobot"):
        dr._get_lerobot_dataset_class()


def test_read_dataset_episode_indices_dedups_and_skips_columnless_parquet(tmp_path):
    """Episode ground truth dedups repeated ``episode_index`` rows and skips a
    shard carrying no ``episode_index`` column.

    A finalized dataset can spread episodes over several parquet shards; a shard
    may repeat an episode row (first length wins) or be a metadata-only shard
    with no ``episode_index`` column at all. The reader must produce a single
    deduplicated, ordered episode list regardless.
    """
    pa = pytest.importorskip("pyarrow")
    pq = pytest.importorskip("pyarrow.parquet")

    from strands_robots.dataset_recorder import read_dataset_episode_indices

    ep_dir = tmp_path / "meta" / "episodes"
    (ep_dir / "chunk-000").mkdir(parents=True)
    (ep_dir / "chunk-001").mkdir(parents=True)

    # Shard 0 (sorts first): episode 0 appears twice - first length (3) wins.
    pq.write_table(
        pa.table({"episode_index": [0, 0, 1], "length": [3, 99, 5]}),
        ep_dir / "chunk-000" / "episodes_000.parquet",
    )
    # Shard 1: a metadata-only shard with no episode_index column is skipped.
    pq.write_table(
        pa.table({"some_other_stat": [1.0, 2.0]}),
        ep_dir / "chunk-001" / "episodes_001.parquet",
    )

    result = read_dataset_episode_indices(tmp_path)

    assert result["episode_indices"] == [0, 1]
    assert result["total_episodes"] == 2
    assert result["frames_per_episode"] == [3, 5]  # first-seen length for ep 0
    assert result["total_frames"] == 8
    assert result["info_total_episodes"] is None  # no meta/info.json written
    assert result["unreadable_files"] == []  # every shard was readable


def test_read_dataset_episode_indices_keeps_readable_shards_when_one_is_corrupt(tmp_path):
    """One corrupt shard must not erase the episode truth of the readable ones.

    Partial damage is the common case (an interrupted rsync or hub download
    truncates a file or two of many). The reader reports the broken shard by
    relative path in ``unreadable_files`` and still returns the episodes it
    could read, so a caller can localise the damage instead of seeing an empty
    dataset.
    """
    pa = pytest.importorskip("pyarrow")
    pq = pytest.importorskip("pyarrow.parquet")

    from strands_robots.dataset_recorder import read_dataset_episode_indices

    ep_dir = tmp_path / "meta" / "episodes" / "chunk-000"
    ep_dir.mkdir(parents=True)
    pq.write_table(pa.table({"episode_index": [0, 1], "length": [10, 12]}), ep_dir / "file-000.parquet")
    (ep_dir / "file-001.parquet").write_bytes(b"truncated, not a parquet file")

    result = read_dataset_episode_indices(tmp_path)

    assert result["episode_indices"] == [0, 1]
    assert result["total_frames"] == 22
    assert len(result["unreadable_files"]) == 1
    assert result["unreadable_files"][0].startswith("meta/episodes/chunk-000/file-001.parquet: ")


def test_read_dataset_episode_indices_raises_when_no_shard_is_readable(tmp_path):
    """With no readable shard there is no ground truth, so the read fails loud.

    Returning an empty episode list would be indistinguishable from a genuinely
    empty dataset, so a wholly unreadable ``meta/episodes`` tree raises with
    every file and its read error named.
    """
    pytest.importorskip("pyarrow")

    from strands_robots.dataset_recorder import read_dataset_episode_indices

    ep_dir = tmp_path / "meta" / "episodes" / "chunk-000"
    ep_dir.mkdir(parents=True)
    (ep_dir / "file-000.parquet").write_bytes(b"not a parquet file")

    with pytest.raises(ValueError, match=r"No readable meta/episodes parquet"):
        read_dataset_episode_indices(tmp_path)


class _FakeSyncDataset:
    """Minimal fake dataset exposing the surface ``sync_to_bucket`` reads.

    ``sync_to_bucket`` only touches ``dataset.root`` (to locate ``meta/``) and
    ``dataset.repo_id`` (to derive a default ``run_id``); nothing else is
    required, so no real LeRobotDataset is needed.
    """

    def __init__(self, root, repo_id="my-org/robot-fave"):
        self.root = str(root)
        self.repo_id = repo_id


def _sync_recorder(root, repo_id="my-org/robot-fave") -> DatasetRecorder:
    rec = DatasetRecorder(dataset=_FakeSyncDataset(root, repo_id))
    rec.episode_count = 2
    rec.frame_count = 40
    return rec


def _write_meta(root) -> None:
    """Create the ``meta/`` directory that a finalized dataset must have."""
    (root / "meta").mkdir(parents=True, exist_ok=True)


def test_sync_to_bucket_missing_hf_cli_errors(tmp_path, monkeypatch):
    """No ``hf`` CLI on PATH -> actionable error, never a subprocess call."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: None)

    def _boom(*_a, **_k):
        raise AssertionError("subprocess must not run when hf CLI is absent")

    monkeypatch.setattr(subprocess, "run", _boom)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "error"
    assert "hf" in result["message"]


def test_sync_to_bucket_old_huggingface_hub_errors(tmp_path, monkeypatch):
    """A hub too old for the subcommands -> upgrade instruction, never a subprocess call.

    Regression test for the version gate: on every release before 1.5 the ``hf``
    binary exists but lacks the ``buckets``/``sync`` subcommands, so without the
    gate the user gets CLI usage noise piped verbatim into the status dict.
    """
    import subprocess

    import huggingface_hub

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")
    monkeypatch.setattr(huggingface_hub, "__version__", "0.36.2")

    def _boom(*_a, **_k):
        raise AssertionError("subprocess must not run when huggingface_hub is too old")

    monkeypatch.setattr(subprocess, "run", _boom)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "error"
    assert "huggingface_hub>=1.5" in result["message"]
    assert "0.36.2" in result["message"]
    assert "pip install -U 'huggingface_hub>=1.5'" in result["message"]


def test_sync_to_bucket_new_huggingface_hub_passes_version_gate(tmp_path, monkeypatch):
    """The oldest hub that ships the subcommands passes the gate and proceeds to sync."""
    import subprocess

    import huggingface_hub

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")
    monkeypatch.setattr(huggingface_hub, "__version__", "1.5.0")

    def _fake_run(cmd, *_a, **_k):
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "success"


def test_sync_to_bucket_tolerates_a_bucket_that_already_exists(tmp_path, monkeypatch):
    """Re-syncing into an existing bucket must succeed, not fail on create.

    ``create=True`` runs ``hf buckets create`` before every sync, so the second
    sync of a run - the daily re-sync the bucket exists for - always hits an
    already-created bucket. The hub reports that as a 409 whose body reads "You
    already created this bucket repo", which does not contain "exists"; matching
    only that one substring turned the normal case into
    ``bucket create failed: ... 409 Conflict``.
    """
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    calls = []

    def _fake_run(cmd, *_a, **_k):
        calls.append(cmd)
        if "create" in cmd:
            return subprocess.CompletedProcess(
                cmd,
                1,
                stdout="",
                stderr=(
                    "Error: Client error '409 Conflict' for url "
                    "'https://huggingface.co/api/buckets/my-org/robot-fave'\n"
                    "You already created this bucket repo: my-org/robot-fave"
                ),
            )
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "success", result
    assert any("sync" in cmd for cmd in calls), "the sync must still run after create is skipped"


def test_sync_to_bucket_still_reports_a_genuine_create_failure(tmp_path, monkeypatch):
    """A create failure that is not an already-exists conflict still errors.

    Guards the already-exists tolerance above from widening into "ignore every
    create failure", which would report success for a sync that never ran.
    """
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _fake_run(cmd, *_a, **_k):
        if "create" in cmd:
            return subprocess.CompletedProcess(
                cmd, 1, stdout="", stderr="Error: 403 Forbidden - insufficient permissions"
            )
        raise AssertionError("sync must not run when bucket creation genuinely failed")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "error"
    assert "403" in result["message"]


def test_sync_to_bucket_unimportable_huggingface_hub_skips_version_gate(tmp_path, monkeypatch):
    """No importable huggingface_hub package -> the gate fails open.

    The ``hf`` binary can come from a different environment on PATH (e.g. a
    pipx install) whose version this interpreter cannot see; blocking on a
    missing *package* would break a working CLI. The normal subprocess error
    path still surfaces genuine failures.
    """
    import subprocess
    import sys

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")
    # A None entry in sys.modules makes `import huggingface_hub` raise ImportError.
    monkeypatch.setitem(sys.modules, "huggingface_hub", None)

    def _fake_run(cmd, *_a, **_k):
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "success"


def test_sync_to_bucket_unparseable_huggingface_hub_version_skips_gate(tmp_path, monkeypatch):
    """An unparseable version string fails open rather than blocking the CLI."""
    import subprocess

    import huggingface_hub

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")
    monkeypatch.setattr(huggingface_hub, "__version__", "unknown")

    def _fake_run(cmd, *_a, **_k):
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "success"


@pytest.mark.parametrize(
    "bad_bucket",
    [
        "my-org/robot; rm -rf /",  # shell metacharacters
        "my-org/../etc",  # path traversal
        "org/name/extra",  # more than one path segment
        "-leading-dash",  # must start alphanumeric
        "org/ name",  # embedded space
    ],
)
def test_sync_to_bucket_rejects_unsafe_bucket(tmp_path, monkeypatch, bad_bucket):
    """Bucket names failing the allowlist are rejected before any subprocess.

    This pins the injection-prevention contract: ``sync_to_bucket`` is reachable
    from an agent tool (``stop_recording(bucket=...)``), so a weakened bucket
    regex would let shell metacharacters / path traversal reach ``hf``.
    """
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _boom(*_a, **_k):
        raise AssertionError(f"subprocess must not run for unsafe bucket {bad_bucket!r}")

    monkeypatch.setattr(subprocess, "run", _boom)

    result = _sync_recorder(tmp_path).sync_to_bucket(bad_bucket)

    assert result["status"] == "error"
    assert "invalid bucket" in result["message"]


def test_sync_to_bucket_requires_finalized_meta_dir(tmp_path, monkeypatch):
    """A dataset with no ``meta/`` (never finalized) is refused."""
    from strands_robots import dataset_recorder as dr

    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")
    # No _write_meta: meta/ is absent.

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "error"
    assert "meta/" in result["message"]
    assert "finalize()" in result["message"]


@pytest.mark.parametrize("bad_run_id", ["bad/id", "run;id", ".."])
def test_sync_to_bucket_rejects_unsafe_run_id(tmp_path, monkeypatch, bad_run_id):
    """run_id must be a single safe path segment (no '/', traversal, metachars)."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _boom(*_a, **_k):
        raise AssertionError(f"subprocess must not run for unsafe run_id {bad_run_id!r}")

    monkeypatch.setattr(subprocess, "run", _boom)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave", run_id=bad_run_id)

    assert result["status"] == "error"
    assert "invalid run_id" in result["message"]


def test_sync_to_bucket_success_creates_and_syncs(tmp_path, monkeypatch):
    """Happy path: creates the bucket then syncs to the derived HF URI."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    calls: list[list[str]] = []

    def _fake_run(cmd, *_a, **_k):
        calls.append(cmd)
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path, repo_id="my-org/robot-fave").sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "success"
    # run_id defaults to the dataset repo_id's last path segment.
    assert result["bucket_uri"] == "hf://buckets/my-org/robot-fave/robot-fave"
    assert result["episodes"] == 2
    assert result["frames"] == 40
    # First call creates the bucket, second syncs the local root to the URI.
    assert calls[0][:3] == ["hf", "buckets", "create"]
    assert calls[1][0] == "hf" and calls[1][1] == "sync"
    assert calls[1][-1] == "hf://buckets/my-org/robot-fave/robot-fave"


def test_sync_to_bucket_preexisting_bucket_is_not_an_error(tmp_path, monkeypatch):
    """`hf buckets create` failing because the bucket already exists is tolerated."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _fake_run(cmd, *_a, **_k):
        if cmd[:3] == ["hf", "buckets", "create"]:
            return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="bucket already exists")
        return subprocess.CompletedProcess(cmd, 0, stdout="synced", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave", run_id="nightly")

    assert result["status"] == "success"
    assert result["bucket_uri"] == "hf://buckets/my-org/robot-fave/nightly"


def test_sync_to_bucket_create_failure_surfaced(tmp_path, monkeypatch):
    """A genuine `hf buckets create` failure is surfaced, not swallowed."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _fake_run(cmd, *_a, **_k):
        if cmd[:3] == ["hf", "buckets", "create"]:
            return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="permission denied")
        raise AssertionError("sync must not run after a fatal create failure")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "error"
    assert "permission denied" in result["message"]


def test_sync_to_bucket_sync_failure_surfaced(tmp_path, monkeypatch):
    """A non-zero `hf sync` return code is surfaced with its stderr."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _fake_run(cmd, *_a, **_k):
        if cmd[:3] == ["hf", "buckets", "create"]:
            return subprocess.CompletedProcess(cmd, 0, stdout="created", stderr="")
        return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="network unreachable")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result["status"] == "error"
    assert "network unreachable" in result["message"]


def test_sync_to_bucket_delete_flag_forwards_to_sync(tmp_path, monkeypatch):
    """delete=True appends --delete to the `hf sync` command (mirror semantics)."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    calls: list[list[str]] = []

    def _fake_run(cmd, *_a, **_k):
        calls.append(cmd)
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave", run_id="nightly", create=False, delete=True)

    assert result["status"] == "success"
    # create=False -> only the sync call happens.
    assert len(calls) == 1
    assert calls[0][:2] == ["hf", "sync"]
    assert "--delete" in calls[0]


# ── sync_dataset_to_bucket (module-level, lifecycle-independent) ───────────
#
# Issue #1502: syncing an on-disk dataset must not require a live recording
# session. These tests exercise the free function directly - no DatasetRecorder
# is constructed - and pin that the method is a thin delegate.


def test_sync_dataset_to_bucket_works_without_a_recorder(tmp_path, monkeypatch):
    """The free function syncs a bare directory: no recorder, no lerobot.

    run_id defaults to the directory name (there is no repo_id to derive from),
    and the success dict carries the derived bucket URI.
    """
    import subprocess

    from strands_robots import dataset_recorder as dr

    root = tmp_path / "robot-fave"
    (root / "meta").mkdir(parents=True)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    calls: list[list[str]] = []

    def _fake_run(cmd, *_a, **_k):
        calls.append(cmd)
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = dr.sync_dataset_to_bucket(root, "my-org/robot-fave")

    assert result["status"] == "success"
    # run_id derived from Path(root).name, not from any repo_id.
    assert result["bucket_uri"] == "hf://buckets/my-org/robot-fave/robot-fave"
    # No recorder in scope: episodes/frames counters are a recorder concept.
    assert "episodes" not in result
    assert "frames" not in result
    assert calls[0][:3] == ["hf", "buckets", "create"]
    assert calls[1][:2] == ["hf", "sync"]
    assert calls[1][2] == str(root)
    assert calls[1][3] == "hf://buckets/my-org/robot-fave/robot-fave"


def test_sync_dataset_to_bucket_standalone_needs_no_recorder(tmp_path, monkeypatch):
    """The module-level ``sync_dataset_to_bucket`` syncs an on-disk dataset with
    no live DatasetRecorder (lifecycle-independent daily-sync path used by
    ``stop_recording(bucket=...)`` on an idle sim)."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    root = tmp_path / "robot-fave"
    root.mkdir()
    _write_meta(root)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    calls: list[list[str]] = []

    def _fake_run(cmd, *_a, **_k):
        calls.append(cmd)
        return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="")

    monkeypatch.setattr(subprocess, "run", _fake_run)

    result = dr.sync_dataset_to_bucket(str(root), "my-org/robot-fave")

    assert result["status"] == "success"
    # run_id defaults to the dataset directory name when no recorder exists.
    assert result["bucket_uri"] == "hf://buckets/my-org/robot-fave/robot-fave"
    assert calls[0][:3] == ["hf", "buckets", "create"]
    assert calls[1][:2] == ["hf", "sync"] and calls[1][2] == str(root)


def test_sync_dataset_to_bucket_standalone_requires_finalized_meta(tmp_path, monkeypatch):
    """The standalone sync refuses a dataset root without ``meta/``."""
    from strands_robots import dataset_recorder as dr

    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    result = dr.sync_dataset_to_bucket(str(tmp_path), "my-org/robot-fave")

    assert result["status"] == "error"
    assert "meta/" in result["message"]


def test_sync_dataset_to_bucket_accepts_str_root(tmp_path, monkeypatch):
    """``root`` may be a plain string (the docstring advertises str | Path)."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    root = tmp_path / "run-021"
    (root / "meta").mkdir(parents=True)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")
    monkeypatch.setattr(
        subprocess,
        "run",
        lambda cmd, *_a, **_k: subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr=""),
    )

    result = dr.sync_dataset_to_bucket(str(root), "my-org/robot-fave", create=False)

    assert result["status"] == "success"
    assert result["bucket_uri"] == "hf://buckets/my-org/robot-fave/run-021"


def test_sync_dataset_to_bucket_missing_hf_cli_errors(tmp_path, monkeypatch):
    """No ``hf`` CLI -> actionable error, never a subprocess call."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: None)

    def _boom(*_a, **_k):
        raise AssertionError("subprocess must not run when hf CLI is absent")

    monkeypatch.setattr(subprocess, "run", _boom)

    result = dr.sync_dataset_to_bucket(tmp_path, "my-org/robot-fave")

    assert result["status"] == "error"
    assert "hf" in result["message"]


@pytest.mark.parametrize(
    "bad_bucket",
    [
        "my-org/robot; rm -rf /",  # shell metacharacters
        "my-org/../etc",  # path traversal
        "org/name/extra",  # more than one path segment
        "-leading-dash",  # must start alphanumeric
        "org/ name",  # embedded space
    ],
)
def test_sync_dataset_to_bucket_rejects_unsafe_bucket(tmp_path, monkeypatch, bad_bucket):
    """The lifted free function keeps the bucket allowlist (LLM-input safety).

    The refactor must not weaken the injection-prevention contract that the
    method carried: values failing ``_BUCKET_RE`` are rejected before any
    subprocess or URI interpolation.
    """
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _boom(*_a, **_k):
        raise AssertionError(f"subprocess must not run for unsafe bucket {bad_bucket!r}")

    monkeypatch.setattr(subprocess, "run", _boom)

    result = dr.sync_dataset_to_bucket(tmp_path, bad_bucket)

    assert result["status"] == "error"
    assert "invalid bucket" in result["message"]


@pytest.mark.parametrize("bad_run_id", ["bad/id", "run;id", ".."])
def test_sync_dataset_to_bucket_rejects_unsafe_run_id(tmp_path, monkeypatch, bad_run_id):
    """The lifted free function keeps the run_id allowlist (LLM-input safety)."""
    import subprocess

    from strands_robots import dataset_recorder as dr

    _write_meta(tmp_path)
    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")

    def _boom(*_a, **_k):
        raise AssertionError(f"subprocess must not run for unsafe run_id {bad_run_id!r}")

    monkeypatch.setattr(subprocess, "run", _boom)

    result = dr.sync_dataset_to_bucket(tmp_path, "my-org/robot-fave", run_id=bad_run_id)

    assert result["status"] == "error"
    assert "invalid run_id" in result["message"]


def test_sync_dataset_to_bucket_requires_meta_dir(tmp_path, monkeypatch):
    """A directory with no ``meta/`` (never finalized) is refused."""
    from strands_robots import dataset_recorder as dr

    monkeypatch.setattr(dr, "_hf_executable", lambda: "hf")
    # No _write_meta: meta/ is absent.

    result = dr.sync_dataset_to_bucket(tmp_path, "my-org/robot-fave")

    assert result["status"] == "error"
    assert "meta/" in result["message"]
    assert "finalize()" in result["message"]


def test_recorder_sync_to_bucket_delegates_to_module_helper(tmp_path, monkeypatch):
    """The method is a thin delegate: dataset root forwarded, run_id derived
    from repo_id, and episodes/frames counters added on success."""
    from strands_robots import dataset_recorder as dr

    seen: dict = {}

    def _fake_sync(root, bucket, run_id=None, *, create=True, private=True, delete=False):
        seen.update(root=root, bucket=bucket, run_id=run_id, create=create, private=private, delete=delete)
        return {"status": "success", "bucket_uri": f"hf://buckets/{bucket}/{run_id}"}

    monkeypatch.setattr(dr, "sync_dataset_to_bucket", _fake_sync)

    result = _sync_recorder(tmp_path, repo_id="my-org/robot-fave").sync_to_bucket(
        "my-org/robot-fave", create=False, private=False, delete=True
    )

    assert seen["root"] == str(tmp_path)
    assert seen["bucket"] == "my-org/robot-fave"
    # run_id defaults to the dataset repo_id's last segment, not the dir name.
    assert seen["run_id"] == "robot-fave"
    assert (seen["create"], seen["private"], seen["delete"]) == (False, False, True)
    # The method augments the helper's success dict with recorder counters.
    assert result["status"] == "success"
    assert result["episodes"] == 2
    assert result["frames"] == 40


def test_recorder_sync_to_bucket_error_passthrough_has_no_counters(tmp_path, monkeypatch):
    """On helper failure the method returns the error dict untouched."""
    from strands_robots import dataset_recorder as dr

    monkeypatch.setattr(
        dr,
        "sync_dataset_to_bucket",
        lambda *_a, **_k: {"status": "error", "message": "boom"},
    )

    result = _sync_recorder(tmp_path).sync_to_bucket("my-org/robot-fave")

    assert result == {"status": "error", "message": "boom"}


def test_sync_dataset_to_bucket_exported_top_level():
    """``strands_robots.sync_dataset_to_bucket`` lazy-resolves to the helper."""
    from strands_robots import __all__ as strands_robots_all
    from strands_robots import dataset_recorder as dr
    from strands_robots import sync_dataset_to_bucket

    assert "sync_dataset_to_bucket" in strands_robots_all
    # The from-import above goes through the package's lazy ``__getattr__``,
    # so identity with the helper proves the lazy resolution works.
    assert sync_dataset_to_bucket is dr.sync_dataset_to_bucket


def test_hf_executable_prefers_interpreter_env_over_path(tmp_path, monkeypatch):
    """`_hf_executable` finds the ``hf`` next to the running interpreter before
    falling back to PATH, so sync works from a venv whose bin is not activated.
    """
    import shutil

    from strands_robots import dataset_recorder as dr

    # A fake venv layout: <venv>/bin/python and <venv>/bin/hf.
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    (bin_dir / "hf").write_text("#!/bin/sh\n")
    monkeypatch.setattr(dr.sys, "executable", str(bin_dir / "python"))
    # PATH lookup would resolve elsewhere; the interpreter-env hit must win.
    monkeypatch.setattr(shutil, "which", lambda _name: "/somewhere/else/hf")

    assert dr._hf_executable() == str(bin_dir / "hf")


def test_hf_executable_falls_back_to_path(tmp_path, monkeypatch):
    """When no ``hf`` sits next to the interpreter, fall back to PATH lookup."""
    import shutil

    from strands_robots import dataset_recorder as dr

    empty = tmp_path / "bin"
    empty.mkdir()
    monkeypatch.setattr(dr.sys, "executable", str(empty / "python"))
    monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/hf")

    assert dr._hf_executable() == "/usr/bin/hf"


# create(overwrite=...) makes re-recording into an existing repo_id honest:
# LeRobotDataset.create() mkdir()s with exist_ok=False and raises a bare
# FileExistsError on an existing dir. create() now resolves that up front -
# overwrite=True wipes and recreates; overwrite=False raises a clear error
# naming overwrite= / resume() (never a cryptic FileExistsError, never a silent
# clobber). resolve_dataset_dir() is the shared root resolver the sim facade
# also uses.


class _FakeDatasetOverwriteCreate:
    """create() records that it ran; it does not touch the filesystem."""

    created = 0

    def __init__(self, repo_id, root=None) -> None:
        self.repo_id = repo_id
        self.root = root

    @classmethod
    def create(
        cls, repo_id, fps=30, root=None, robot_type="unknown", features=None, use_videos=True, image_writer_threads=4
    ):
        cls.created += 1
        return cls(repo_id, root=root)


def test_resolve_dataset_dir_prefers_explicit_root():
    from strands_robots.dataset_recorder import resolve_dataset_dir

    assert resolve_dataset_dir("user/data", root="/tmp/somewhere") == Path("/tmp/somewhere")


def test_resolve_dataset_dir_treats_bare_repo_id_as_local_path():
    from strands_robots.dataset_recorder import resolve_dataset_dir

    # No owner/name slash -> a local directory path, not $HF_LEROBOT_HOME/<id>.
    assert resolve_dataset_dir("my_local_dataset") == Path("my_local_dataset")


def test_create_raises_clear_error_on_existing_dataset_without_overwrite(tmp_path, monkeypatch):
    """create() on an existing LeRobotDataset dir (has meta/) and no overwrite
    raises a clear FileExistsError naming overwrite= and resume() - not a bare
    LeRobot FileExistsError and not a silent append."""
    _FakeDatasetOverwriteCreate.created = 0
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetOverwriteCreate)
    root = tmp_path / "ds"
    (root / "meta").mkdir(parents=True)

    with pytest.raises(FileExistsError, match="overwrite=True.*resume|resume.*overwrite"):
        DatasetRecorder.create("user/data", root=str(root), joint_names=["j1"])
    # create() must NOT have run, and the existing dataset is left untouched.
    assert _FakeDatasetOverwriteCreate.created == 0
    assert (root / "meta").exists()


def test_create_overwrite_removes_existing_dataset_then_creates_fresh(tmp_path, monkeypatch):
    """overwrite=True wipes an existing dataset dir before create() runs."""
    _FakeDatasetOverwriteCreate.created = 0
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetOverwriteCreate)
    root = tmp_path / "ds"
    (root / "meta").mkdir(parents=True)
    (root / "meta" / "info.json").write_text("{}")

    recorder = DatasetRecorder.create("user/data", root=str(root), joint_names=["j1"], overwrite=True)

    # The stale dataset content was removed before create() ran fresh.
    assert not (root / "meta" / "info.json").exists()
    assert _FakeDatasetOverwriteCreate.created == 1
    assert recorder.dataset.repo_id == "user/data"


def test_create_clears_existing_empty_dir(tmp_path, monkeypatch):
    """An existing EMPTY dir (e.g. tempfile.mkdtemp()) is cleared so create()
    does not dead-end on LeRobot's exist_ok=False guard."""
    _FakeDatasetOverwriteCreate.created = 0
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetOverwriteCreate)
    root = tmp_path / "empty_ds"
    root.mkdir()

    recorder = DatasetRecorder.create("user/data", root=str(root), joint_names=["j1"])

    assert _FakeDatasetOverwriteCreate.created == 1
    assert recorder.dataset.repo_id == "user/data"


def test_create_refuses_nonempty_non_dataset_dir(tmp_path, monkeypatch):
    """A non-empty dir that is not a LeRobotDataset is never clobbered."""
    _FakeDatasetOverwriteCreate.created = 0
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetOverwriteCreate)
    root = tmp_path / "notes"
    root.mkdir()
    (root / "important.txt").write_text("do not delete")

    with pytest.raises(ValueError, match="not a LeRobotDataset"):
        DatasetRecorder.create("user/data", root=str(root), joint_names=["j1"])
    assert _FakeDatasetOverwriteCreate.created == 0
    assert (root / "important.txt").exists()


# create() target resolution also handles a repo_id/root that points at an
# existing plain FILE (not a directory) - e.g. an agent passing root= to a path
# that already holds a file. overwrite=False must refuse with a clear "not a
# directory" error rather than dead-ending inside LeRobotDataset.create();
# overwrite=True removes the file so a fresh dataset can be created in its place.


def test_create_refuses_file_target_without_overwrite(tmp_path, monkeypatch):
    """A file at the resolved target (not a dir) raises a clear ValueError and
    never runs create() or deletes the file."""
    _FakeDatasetOverwriteCreate.created = 0
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetOverwriteCreate)
    target = tmp_path / "not_a_dir"
    target.write_text("i am a file")

    with pytest.raises(ValueError, match="not a directory"):
        DatasetRecorder.create("user/data", root=str(target), joint_names=["j1"])
    assert _FakeDatasetOverwriteCreate.created == 0
    assert target.is_file()
    assert target.read_text() == "i am a file"


def test_create_overwrite_replaces_file_target_then_creates_fresh(tmp_path, monkeypatch):
    """overwrite=True on a file target unlinks it before create() runs fresh."""
    _FakeDatasetOverwriteCreate.created = 0
    _patch_lerobot_dataset(monkeypatch, _FakeDatasetOverwriteCreate)
    target = tmp_path / "stale"
    target.write_text("stale contents")

    recorder = DatasetRecorder.create("user/data", root=str(target), joint_names=["j1"], overwrite=True)

    assert not target.exists()
    assert _FakeDatasetOverwriteCreate.created == 1
    assert recorder.dataset.repo_id == "user/data"


# resolve_dataset_dir() mirrors LeRobotDataset root resolution. When lerobot is
# not importable it must fall back to the documented default dataset home
# (~/.cache/huggingface/lerobot) instead of raising, so callers can still
# inspect the target path in a lerobot-less environment.


def test_resolve_dataset_dir_falls_back_to_default_home_when_lerobot_absent(monkeypatch):
    import importlib

    from strands_robots.dataset_recorder import resolve_dataset_dir

    # Removing HF_LEROBOT_HOME from lerobot's constants module makes
    # `from lerobot.utils.constants import HF_LEROBOT_HOME` raise ImportError,
    # exercising the documented default-home fallback without disturbing any
    # other lerobot import (the module object itself is left in place).
    # When lerobot is installed, drop the constant so the import fails; when it
    # is absent the fallback is already the natural path - both reach line 216.
    try:
        constants = importlib.import_module("lerobot.utils.constants")
        monkeypatch.delattr(constants, "HF_LEROBOT_HOME", raising=False)
    except ImportError:  # lerobot not installed; the fallback path is what we test
        pass

    resolved = resolve_dataset_dir("user/data")

    assert resolved == Path.home() / ".cache" / "huggingface" / "lerobot" / "user" / "data"
