feat(import): per-episode receipts in prepared-manifest.json

.gitignore

@@ -26,4 +26,4 @@ packages/hflow-server/src/hflow_server/static/
 # Transient: `pnpm gen:api` dumps the schema here on its way to src/apiSchema.ts.
 .openapi.json
 
-**/node_modules/*
+**/node_modules/*

docs/how-to/import-lerobot-v3.md

@@ -26,7 +26,7 @@ data/lerobot_pusht/
 ├── _lerobot_cache/                 # downloaded metadata, Parquet, and video
 ├── landing/
 │   └── lerobot_episode_0001.mcap   # canonical episode
-└── prepared-manifest.json          # source commit and import summary
+└── prepared-manifest.json          # source commit, import summary, per-episode receipts
 ```
 
 Re-running against the same output directory reuses downloaded source files.
@@ -46,7 +46,12 @@ uv run hflow import lerobot \
 ```
 
 Durable outputs land as `landing/*.mcap` and `prepared-manifest.json` under
-that prefix. Hugging Face downloads stay in the local mirror under
+that prefix. The manifest (schema version 3) carries a `episodes` receipt
+list: every delivered episode's published URI, its `content_id` (the sha256
+content address of the canonical bytes, the same value the catalog uses for
+dedupe), and its `size_bytes`, so a delivery can be checked against the
+manifest without re-running the import. Hugging Face downloads stay in the
+local mirror under
 `_lerobot_cache/` (`HFLOW_MIRROR_DIR`, or `$XDG_CACHE_HOME/hflow/mirrors`) and
 are never uploaded into the bucket. The success manifest is published only
 after every selected episode object has been written.

src/hflow/importers/lerobot.py

@@ -31,6 +31,7 @@
 
 from mcap.writer import Writer as McapWriter
 
+from hflow.catalog import content_episode_id
 from hflow.ffmpeg import ffmpeg_path, ffmpeg_version, ffprobe_path
 from hflow.storage import LocalStorageRoot, StorageRoot, parse_storage_root
 from hflow.transform import TransformConfig, write_canonical_episode
@@ -75,6 +76,21 @@ class _VideoWindow(TypedDict):
     to_timestamp: float
 
 
+class _PublishedEpisode(TypedDict):
+    """One delivered episode as the manifest receipt describes it.
+
+    ``uri`` is the published object URI (a bucket root has no local paths),
+    ``content_id`` is ``content_episode_id`` over the canonical file taken
+    while it is still on local disk, and ``size_bytes`` is that same file's
+    size. Together they carry everything a delivery-verification reader of
+    the manifest needs without a second schema bump.
+    """
+
+    uri: str
+    content_id: str
+    size_bytes: int
+
+
 @dataclass(frozen=True)
 class DatasetSource:
     repo_id: str
@@ -710,11 +726,11 @@ def import_lerobot_dataset(
     selected_episode_indexes = (
         [episode_index] if episode_index is not None else list(range(len(episode_rows)))
     )
-    published_episode_uris: list[str] = []
+    published_episodes: list[_PublishedEpisode] = []
 
     dataset_source = source_archive["dataset"]
     for selected_episode_index in selected_episode_indexes:
-        published_episode_uris.append(
+        published_episodes.append(
             _convert_single_episode(
                 source_archive=source_archive,
                 dataset_source=dataset_source,
@@ -728,14 +744,15 @@ def import_lerobot_dataset(
 
     manifest_contents = json.dumps(
         {
-            "schema_version": 2,
+            "schema_version": 3,
             "dataset": {
                 "repo_id": dataset_source.repo_id,
                 "revision": dataset_source.revision,
                 "license": dataset_source.license,
             },
             "camera_keys": list(resolved_camera_keys),
-            "episodes_converted": len(published_episode_uris),
+            "episodes_converted": len(published_episodes),
+            "episodes": list(published_episodes),
             "converter_version": CONVERTER_VERSION,
         },
         indent=2,
@@ -746,7 +763,7 @@ def import_lerobot_dataset(
         temporary_manifest_path.write_text(manifest_contents + "\n", encoding="utf-8")
         published_manifest_uri = storage.publish(temporary_manifest_path, "prepared-manifest.json")
     logger.info("wrote LeRobot import manifest %s", published_manifest_uri)
-    return published_episode_uris
+    return [episode["uri"] for episode in published_episodes]
 
 
 def _convert_single_episode(
@@ -757,10 +774,11 @@ def _convert_single_episode(
     camera_keys: tuple[str, ...],
     numeric_schemas: dict[str, _NumericSchema],
     frames_per_second: int,
-) -> str:
+) -> _PublishedEpisode:
     """Convert a single episode to canonical MCAP and publish it.
 
-    Returns the published episode URI under ``landing/``.
+    Returns the published episode's manifest receipt: the published object
+    URI, the content id of the canonical bytes, and the byte size.
     """
     import duckdb
 
@@ -1039,15 +1057,23 @@ def _feature_rows(feature_name: str) -> list | None:
             TransformConfig(gop_seconds=1.0),
             source_uri=source_uri,
         )
-        published_uri = storage.publish(canonical_episode_path, landing_relative_key)
+        # Hash and size the canonical file while it is still on local disk,
+        # before storage.publish: for a bucket root, reading the content id
+        # back from the published object means downloading our own upload.
+        episode_content_id = content_episode_id(canonical_episode_path)
         episode_size_bytes = canonical_episode_path.stat().st_size
+        published_uri = storage.publish(canonical_episode_path, landing_relative_key)
 
     logger.info(
         "wrote canonical LeRobot episode %s (%.2f MB)",
         published_uri,
         episode_size_bytes / 1_000_000,
     )
-    return published_uri
+    return {
+        "uri": published_uri,
+        "content_id": episode_content_id,
+        "size_bytes": episode_size_bytes,
+    }
 
 
 __all__ = ["import_lerobot_dataset"]

tests/test_lerobot_converter.py

@@ -427,8 +427,9 @@ def fake_transcode(mp4_path: Path, *_args: object, **_kwargs: object) -> list[by
     )
 
     published_uris: list[str] = []
+    receipts: list[prep._PublishedEpisode] = []
     for episode_index in (0, 1, 0):
-        uri = prep._convert_single_episode(
+        receipt = prep._convert_single_episode(
             source_archive=source_archive,
             dataset_source=dataset_source,
             storage=LocalStorageRoot(tmp_path / "output"),
@@ -437,11 +438,20 @@ def fake_transcode(mp4_path: Path, *_args: object, **_kwargs: object) -> list[by
             numeric_schemas=numeric_schemas,
             frames_per_second=30,
         )
-        published_uris.append(uri)
+        published_uris.append(receipt["uri"])
+        receipts.append(receipt)
 
     assert published_uris[0].endswith("landing/lerobot_episode_0001.mcap")
     assert published_uris[1].endswith("landing/lerobot_episode_0002.mcap")
 
+    # The manifest tests build their receipts in the convert stub, so this is
+    # the only place the real function's receipt is checked against the object
+    # it published rather than against a value the test wrote itself.
+    for receipt in receipts:
+        landed_path = Path(receipt["uri"])
+        assert receipt["content_id"] == prep.content_episode_id(landed_path)
+        assert receipt["size_bytes"] == landed_path.stat().st_size
+
     video_urls = [
         "https://huggingface.co/datasets/fake/repo/resolve/abc/videos/"
         "observation.images.up/chunk-000/file-000.mp4",
@@ -1074,13 +1084,18 @@ def fake_convert(
         camera_keys: object,
         numeric_schemas: object,
         frames_per_second: object,
-    ) -> str:
+    ) -> prep._PublishedEpisode:
         del source_archive, dataset_source, camera_keys, numeric_schemas, frames_per_second
         relative_key = f"landing/lerobot_episode_{episode_index + 1:04d}.mcap"
         staged = tmp_path / f"staged-{episode_index}.mcap"
         staged.write_bytes(f"episode-{episode_index}".encode())
         published_keys.append(relative_key)
-        return storage.publish(staged, relative_key)
+        published_uri = storage.publish(staged, relative_key)
+        return {
+            "uri": published_uri,
+            "content_id": prep.content_episode_id(staged),
+            "size_bytes": staged.stat().st_size,
+        }
 
     monkeypatch.setattr(prep, "_convert_single_episode", fake_convert)
     return published_keys
@@ -1107,15 +1122,23 @@ def test_import_returns_local_uris_and_keeps_cache_beside_landing(
     assert Path(episode_uris[0]).is_file()
     assert (output_dir / "prepared-manifest.json").is_file()
     manifest_payload = json.loads((output_dir / "prepared-manifest.json").read_text())
+    landed_episode_path = output_dir / "landing" / "lerobot_episode_0001.mcap"
     assert manifest_payload == {
-        "schema_version": 2,
+        "schema_version": 3,
         "dataset": {
             "repo_id": "fake/repo",
             "revision": "abc",
             "license": "apache-2.0",
         },
         "camera_keys": [prep.DEFAULT_CAMERA_KEY],
         "episodes_converted": 1,
+        "episodes": [
+            {
+                "uri": episode_uris[0],
+                "content_id": prep.content_episode_id(landed_episode_path),
+                "size_bytes": landed_episode_path.stat().st_size,
+            }
+        ],
         "converter_version": prep.CONVERTER_VERSION,
     }
     assert (output_dir / "_lerobot_cache" / "abc").is_dir()
@@ -1186,7 +1209,7 @@ def fake_download(url: str, destination_path: Path, **_kwargs: object) -> None:
         lambda source_path, output_path, *args, **kwargs: shutil.copy(source_path, output_path),
     )
 
-    uri = prep._convert_single_episode(
+    receipt = prep._convert_single_episode(
         source_archive=source_archive,
         dataset_source=dataset_source,
         storage=LocalStorageRoot(tmp_path / "output"),
@@ -1196,7 +1219,7 @@ def fake_download(url: str, destination_path: Path, **_kwargs: object) -> None:
         frames_per_second=30,
     )
 
-    episode_metadata = open_reader(uri).metadata()
+    episode_metadata = open_reader(receipt["uri"]).metadata()
     assert episode_metadata["episode/v1"]["converter_version"] == prep.CONVERTER_VERSION
     assert episode_metadata["source-provenance/v1"]["converter_version"] == prep.CONVERTER_VERSION
 
@@ -1236,6 +1259,88 @@ def test_import_publishes_into_a_bucket_data_root_without_uploading_cache(
     assert (data_root.mirror / "_lerobot_cache" / "abc").is_dir()
 
 
+def test_manifest_records_per_episode_receipts(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The manifest lists every delivered episode with its content id and size.
+
+    A recipient of a prepared corpus gets a receipt that can be checked
+    against the landing directory without re-running the import; the
+    content id is the same ``content_episode_id`` the catalog dedupes on.
+    """
+    output_dir = tmp_path / "out"
+    monkeypatch.setattr(
+        prep, "_hf_repo_info", lambda repo, revision: {"sha": "abc", "license": "apache-2.0"}
+    )
+    monkeypatch.setattr(prep, "_ensure_source_archive", _stub_single_episode_source_archive)
+    _install_publish_through_convert(monkeypatch, tmp_path)
+
+    episode_uris = prep.import_lerobot_dataset(
+        dataset_repo="fake/repo",
+        revision="main",
+        output_dir=output_dir,
+        episode_index=0,
+    )
+
+    manifest = json.loads((output_dir / "prepared-manifest.json").read_text())
+    assert manifest["schema_version"] == 3
+    # v2 top-level keys survive: readers of the old schema keep working.
+    assert manifest["episodes_converted"] == 1
+    assert manifest["dataset"] == {
+        "repo_id": "fake/repo",
+        "revision": "abc",
+        "license": "apache-2.0",
+    }
+    assert manifest["converter_version"] == prep.CONVERTER_VERSION
+    assert manifest["camera_keys"] == [prep.DEFAULT_CAMERA_KEY]
+
+    entries = manifest["episodes"]
+    assert len(entries) == 1
+    entry = entries[0]
+    assert entry["uri"] == episode_uris[0]
+    assert entry["size_bytes"] == len(b"episode-0")
+    assert entry["content_id"] == prep.content_episode_id(
+        output_dir / "landing" / "lerobot_episode_0001.mcap"
+    )
+    # The receipt describes the published landing object, not a local
+    # staging path: for a bucket root this entry is an object URI that a
+    # recipient of the bucket prefix can resolve without our filesystem.
+    assert entry["uri"].endswith("landing/lerobot_episode_0001.mcap")
+    assert "canonical-" not in entry["uri"]
+    assert "staged-" not in entry["uri"]
+
+
+def test_manifest_content_id_detects_a_truncated_episode(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The #379 controlled result as a test: truncating one episode to zero
+    bytes is detectable from the delivery by re-hashing against the manifest."""
+    output_dir = tmp_path / "out"
+    monkeypatch.setattr(
+        prep, "_hf_repo_info", lambda repo, revision: {"sha": "abc", "license": "apache-2.0"}
+    )
+    monkeypatch.setattr(prep, "_ensure_source_archive", _stub_single_episode_source_archive)
+    _install_publish_through_convert(monkeypatch, tmp_path)
+
+    prep.import_lerobot_dataset(
+        dataset_repo="fake/repo",
+        revision="main",
+        output_dir=output_dir,
+        episode_index=0,
+    )
+
+    manifest = json.loads((output_dir / "prepared-manifest.json").read_text())
+    entry = manifest["episodes"][0]
+    episode_path = output_dir / "landing" / "lerobot_episode_0001.mcap"
+    original_size = episode_path.stat().st_size
+    assert original_size == entry["size_bytes"]
+    assert prep.content_episode_id(episode_path) == entry["content_id"]
+
+    episode_path.write_bytes(b"")
+    assert episode_path.stat().st_size != entry["size_bytes"]
+    assert prep.content_episode_id(episode_path) != entry["content_id"]
+
+
 def test_import_skips_bucket_manifest_when_an_episode_publish_fails(
     tmp_path: Path,
     bucket_over_tmp: tuple[object, Path],