fix: validate chunk CRCs when reading canonical episodes after sync

src/hflow/episode.py

@@ -308,7 +308,11 @@ def workdir(self) -> Path:
 
     @cached_property
     def _reader(self) -> EpisodeReader:
-        return open_reader(self.path)
+        # validate_crcs=True: a content hash proved this file at sync time,
+        # not at read time. Every post-sync lane (META, relabel, re-check)
+        # consumes this reader, so a canonical episode that decayed on disk
+        # must be diagnosed here rather than re-certified (#474).
+        return open_reader(self.path, validate_crcs=True)
 
     def close(self) -> None:
         if "_reader" in self.__dict__:

src/hflow/reader.py

@@ -298,9 +298,10 @@ def open_reader(path: Path | str, *, validate_crcs: bool = False) -> EpisodeRead
 
     ``validate_crcs`` checks each chunk's CRC as it is decoded, catching
     payload damage that magic-byte and summary checks alone cannot see. It
-    defaults to ``False`` because most callers re-read a canonical file HFlow
-    already produced and already identifies by content hash; pass ``True``
-    only when reading a source that has not been trusted yet (see
-    ``hflow.transform``).
+    defaults to ``False`` for reads that stay out of the message data
+    (summary, metadata, provenance), which chunk CRCs cannot vouch for
+    anyway. Any read that consumes messages should pass ``True``: a content
+    hash or receipt proved the file when it was written, not the bytes on
+    disk at read time (see ``hflow.transform`` and ``hflow.episode``, #474).
     """
     return PythonMcapEpisodeReader(path, validate_crcs=validate_crcs)

tests/reuse_test_helpers.py

@@ -44,6 +44,30 @@ def flip_chunk_payload_bytes(episode_path: Path, *, count: int = 4) -> None:
     episode_path.write_bytes(bytes(data))
 
 
+def flip_first_chunk_stored_crc(episode_path: Path) -> None:
+    """Flip one bit of the first chunk's stored ``uncompressed_crc`` in place.
+
+    The complement of :func:`flip_chunk_payload_bytes` for compressed files:
+    a canonical episode's chunk records region is compressed, so its payload
+    bytes are not addressable in place -- but the CRC field in the chunk
+    header is. The payload still decompresses fine; the bytes simply no
+    longer match the file's own integrity stamp (real-world header rot),
+    which only a CRC-validated read can tell.
+    """
+    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_start = summary.chunk_indexes[0].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), so the stored CRC begins at offset 33.
+    data[chunk_start + 33] ^= 0x01
+    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_processing_regressions.py

@@ -465,6 +465,38 @@ def _state_only_episode(tmp_path: Path) -> Path:
     )
 
 
+def test_check_lanes_refuse_a_canonical_episode_that_decayed_after_sync(tmp_path: Path) -> None:
+    """A canonical episode whose stored chunk CRC no longer matches its bytes
+    must not be re-certified by a post-sync check run (#474): every check that
+    consumes message data errors with the CRC diagnosis instead of stamping
+    fresh findings over bytes the file's own integrity stamp disowns. #429
+    closed this hole on the LeRobot resume path and #462 on the primary ingest
+    read; the post-sync ``Episode`` read is the same trust boundary.
+    """
+    from reuse_test_helpers import flip_first_chunk_stored_crc
+
+    source = _state_only_episode(tmp_path)
+    app = hflow.App("post-sync-crc", data_root=tmp_path / "data")
+    app.process(source, record=False)
+
+    # The control: a healthy canonical runs the check lane unchanged.
+    healthy = app.process(source, record=False, stages={hflow.Stage.META})
+    assert {run.status for run in healthy.checks} == {hflow.CheckStatus.MEASURED}
+
+    flip_first_chunk_stored_crc(healthy.canonical_path)
+
+    damaged = app.process(source, record=False, stages={hflow.Stage.META})
+    digest_run = damaged.check("content_digest")
+    assert digest_run.status == hflow.CheckStatus.ERROR
+    assert digest_run.error is not None
+    assert "CRCValidationError" in digest_run.error
+    assert damaged.has_errors
+    # No fresh evidence lands over the damaged bytes: the only checks still
+    # measuring are the camera ones, which read nothing on a camera-less
+    # episode and record no keys.
+    assert all(not run.result.measurements for run in damaged.checks if run.result is not None)
+
+
 def test_two_checks_recording_one_measurement_key_are_refused(tmp_path: Path) -> None:
     """Every step of one run shares its fingerprint and timestamp, so a shared
     key is a tie the catalog resolves arbitrarily -- one step's value silently