fix(import): CRC-validate reused landing episodes before stamping receipts

src/hflow/importers/lerobot.py

@@ -29,6 +29,7 @@
 from typing import NotRequired, TypedDict
 from urllib.parse import urlsplit
 
+from mcap.reader import make_reader
 from mcap.writer import Writer as McapWriter
 
 from hflow.catalog import content_episode_id
@@ -155,7 +156,13 @@ def _episode_identity_matches(
     episode_index: int,
     camera_keys: tuple[str, ...],
 ) -> bool:
-    """True when a published landing file belongs to this exact import."""
+    """True when a published landing file belongs to this exact import.
+
+    Identity over the metadata records, then a CRC-validated full message
+    pass: a reused episode must be whole, not merely labeled. Metadata-only
+    reads never touch chunk payloads, so without this pass a payload-damaged
+    file would match identity and be stamped with a fresh receipt.
+    """
     from mcap.exceptions import McapError
 
     reader = None
@@ -180,7 +187,7 @@ def _episode_identity_matches(
     if recorded_camera_keys is None:
         return False
     expected_gop = f"{IMPORT_GOP_SECONDS:g}"
-    return (
+    if not (
         episode_metadata.get("source_dataset") == dataset_source.repo_id
         and episode_metadata.get("source_revision") == dataset_source.revision
         and episode_metadata.get("source_episode_index") == str(episode_index)
@@ -189,7 +196,29 @@ def _episode_identity_matches(
         and recorded_camera_keys == camera_keys
         and episode_metadata.get("gop_seconds") == expected_gop
         and provenance_metadata.get("gop_seconds") == expected_gop
-    )
+    ):
+        return False
+    # Metadata matching is not integrity: the metadata records live outside
+    # the chunks, so payload damage never reaches them. Reuse must hold the
+    # file to the same standard hflow doctor applies to any canonical file.
+    try:
+        with local_episode_path.open("rb") as stream:
+            validated_reader = make_reader(stream, validate_crcs=True)
+            for _ in validated_reader.iter_messages(log_time_order=False):
+                pass
+    except (OSError, McapError, ValueError) as error:
+        # Reached only when identity already matched, so this is our own prior
+        # output found damaged, not a stranger's file. Re-conversion repairs it
+        # silently otherwise, which would hide bit rot in a landing tree for as
+        # long as the imports keep succeeding.
+        logger.warning(
+            "re-converting landing episode %s: identity matches but the payload "
+            "failed CRC validation (%s)",
+            local_episode_path,
+            error,
+        )
+        return False
+    return True
 
 
 def _try_reuse_completed_episode(
@@ -923,7 +952,11 @@ def import_lerobot_dataset(
             camera_keys=resolved_camera_keys,
         )
         if reused_episode is not None:
-            logger.info("reusing completed LeRobot episode %s", reused_episode["uri"])
+            logger.info(
+                "reusing verified completed LeRobot episode %s (content_id %s)",
+                reused_episode["uri"],
+                reused_episode["content_id"],
+            )
             published_episodes.append(reused_episode)
             continue
         published_episodes.append(

tests/reuse_test_helpers.py

@@ -0,0 +1,49 @@
+"""Shared helpers for reusing-episode and delivery-corruption tests."""
+
+from __future__ import annotations
+
+import io
+import struct
+from pathlib import Path
+
+from hflow.catalog import content_episode_id
+
+
+def flip_chunk_payload_bytes(episode_path: Path, *, count: int = 4) -> None:
+    """Corrupt message-data bytes in place, leaving the file valid.
+
+    Locates the first chunk record and flips ``count`` bytes inside its
+    message-data records (the tail of the chunk's uncompressed records
+    region). The MCAP summary, indexes, and metadata records stay intact;
+    the chunk CRC no longer matches the damaged bytes, which is exactly what
+    a CRC-validated read catches and what on-disk bit rot or a partial
+    external copy looks like when the container survives. The fixture
+    writer emits uncompressed chunks (``compression=""``), so the records
+    region is plaintext.
+    """
+    from mcap.reader import make_reader
+
+    data = bytearray(episode_path.read_bytes())
+    summary = make_reader(io.BytesIO(bytes(data))).get_summary()
+    if summary is None or not summary.chunk_indexes:
+        raise ValueError(f"{episode_path} has no chunk records to corrupt")
+    chunk_index_record = summary.chunk_indexes[0]
+    chunk_start = chunk_index_record.chunk_start_offset
+    # Chunk record per the MCAP spec: opcode(1) length(8) then
+    # message_start_time(8) message_end_time(8) uncompressed_size(8)
+    # uncompressed_crc(4) compression_length(4) compression records_length(8)
+    # records.
+    compression_length = struct.unpack_from("<I", data, chunk_start + 37)[0]
+    record_length = struct.unpack_from("<Q", data, chunk_start + 1)[0]
+    records_start = chunk_start + 9 + 8 + 8 + 8 + 4 + 4 + compression_length + 8
+    records_end = chunk_start + 9 + record_length
+    if records_end - records_start < count:
+        raise ValueError("chunk records region too small for the requested corruption")
+    for offset in range(records_end - count, records_end):
+        data[offset] ^= 0xFF
+    episode_path.write_bytes(bytes(data))
+
+
+def content_id_differs_from_delivery_receipt(episode_path: Path, receipt_content_id: str) -> bool:
+    """True when the file on disk no longer matches the recorded content id."""
+    return content_episode_id(episode_path) != receipt_content_id

tests/test_lerobot_converter.py

@@ -7,6 +7,7 @@
 
 import io
 import json
+import logging
 import shutil
 import subprocess
 import urllib.request
@@ -1499,14 +1500,24 @@ def _write_identity_matching_landing_mcap(
     field and leave the rest matching, which is what separates the individual
     comparisons in ``_episode_identity_matches`` from each other.
     """
-    from mcap.writer import Writer
+    from mcap.writer import CompressionType, Writer
 
     from hflow.format import METADATA_RECORD_EPISODE, METADATA_RECORD_PROVENANCE
 
     destination.parent.mkdir(parents=True, exist_ok=True)
     with destination.open("wb") as stream:
-        writer = Writer(stream)
+        # Uncompressed chunks: the payload-corruption test flips bytes inside
+        # the chunk records region, which is only addressable in place when
+        # the records are plaintext.
+        writer = Writer(stream, compression=CompressionType.NONE)
         writer.start(profile="", library="test-lerobot-resume")
+        schema_id = writer.register_schema(
+            name="test.Pointer", encoding="ros2msg", data=b"int32 x\n"
+        )
+        channel_id = writer.register_channel(
+            topic="/pointer", message_encoding="ros2msg", schema_id=schema_id
+        )
+        writer.add_message(channel_id, log_time=10**9, data=b"\x01\x00\x00\x00", publish_time=10**9)
         writer.add_metadata(
             METADATA_RECORD_EPISODE,
             {
@@ -2110,3 +2121,90 @@ def test_converter_version_bumped_with_the_label_support() -> None:
     """The label changes episode/v1 bytes, which content_episode_id hashes:
     the converter version moves with the change, not after it."""
     assert prep.CONVERTER_VERSION == "lerobot-converter-v7"
+
+
+def test_reuse_refuses_a_landing_episode_with_damaged_payload(
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Payload damage that leaves the MCAP structure and metadata intact is
+    still damaged work: the reuse path CRC-validates before stamping the
+    receipt, and refuses rather than laundering the bytes (#426)."""
+    from reuse_test_helpers import flip_chunk_payload_bytes
+
+    data_root = LocalStorageRoot(tmp_path / "out")
+    landing = tmp_path / "out" / "landing" / "lerobot_episode_0001.mcap"
+    _write_identity_matching_landing_mcap(
+        landing,
+        dataset_source=_MATCHING_SOURCE,
+        episode_index=0,
+        camera_keys=_MATCHING_CAMERA_KEYS,
+        marker="payload-damaged",
+    )
+    receipt_before = prep._try_reuse_completed_episode(
+        data_root,
+        dataset_source=_MATCHING_SOURCE,
+        episode_index=0,
+        camera_keys=_MATCHING_CAMERA_KEYS,
+    )
+    assert receipt_before is not None
+    intact_content_id = receipt_before["content_id"]
+
+    flip_chunk_payload_bytes(landing)
+
+    with caplog.at_level(logging.WARNING, logger="hflow.importers.lerobot"):
+        assert (
+            prep._try_reuse_completed_episode(
+                data_root,
+                dataset_source=_MATCHING_SOURCE,
+                episode_index=0,
+                camera_keys=_MATCHING_CAMERA_KEYS,
+            )
+            is None
+        )
+    # Silent re-conversion would repair the damage and hide it. The operator
+    # gets told which file failed, so bit rot in a landing tree is visible
+    # rather than absorbed by the next successful import.
+    warning_messages = [record.getMessage() for record in caplog.records]
+    assert any("failed CRC validation" in message for message in warning_messages), warning_messages
+    assert any(str(landing) in message for message in warning_messages), warning_messages
+    # The damaged bytes must not be laundered: the damaged file's hash
+    # differs from the receipt the intact file earned.
+    from reuse_test_helpers import content_id_differs_from_delivery_receipt
+
+    assert content_id_differs_from_delivery_receipt(landing, intact_content_id)
+
+
+def test_reuse_accepts_an_intact_episode_after_the_crc_pass(
+    tmp_path: Path,
+) -> None:
+    """The control: the CRC pass adds no refusal for undamaged bytes."""
+    from reuse_test_helpers import flip_chunk_payload_bytes
+
+    data_root = LocalStorageRoot(tmp_path / "out")
+    landing = tmp_path / "out" / "landing" / "lerobot_episode_0001.mcap"
+    _write_identity_matching_landing_mcap(
+        landing,
+        dataset_source=_MATCHING_SOURCE,
+        episode_index=0,
+        camera_keys=_MATCHING_CAMERA_KEYS,
+        marker="intact",
+    )
+
+    reused = prep._try_reuse_completed_episode(
+        data_root,
+        dataset_source=_MATCHING_SOURCE,
+        episode_index=0,
+        camera_keys=_MATCHING_CAMERA_KEYS,
+    )
+    assert reused is not None
+
+    flip_chunk_payload_bytes(landing)
+
+    damaged = prep._try_reuse_completed_episode(
+        data_root,
+        dataset_source=_MATCHING_SOURCE,
+        episode_index=0,
+        camera_keys=_MATCHING_CAMERA_KEYS,
+    )
+    assert damaged is None