refactor(snapshot): type receipt entries while preserving the content_id invariant

src/hflow/snapshot.py

@@ -145,16 +145,79 @@ def _marker_identifies_dataset_snapshot(format_marker: dict) -> bool:
     )
 
 
-def _file_integrity_record(relative_path: str, absolute_path: Path) -> dict[str, str | int]:
+@dataclass(frozen=True)
+class FileIntegrityRecord:
+    """One receipt entry with known field types.
+
+    Receipt entries arrive from external JSON where every field could be
+    anything; this type is what they become once the boundary has checked
+    them. ``to_dict_for_hashing`` rebuilds exactly the dict shape the
+    exporter has always serialized, so the ``content_id`` digest stays
+    byte-identical with every snapshot ever exported (#489).
+
+    Typing the entries also fixed what the digest covers. Hashing raw dicts
+    made it depend on every key an entry happened to carry; it now depends on
+    these three fields, which are the ones that define a delivery. So an
+    entry with an extra key hashes the same, where it used to hash
+    differently. That is deliberate: a later format revision can add metadata
+    without invalidating the digest of every snapshot already exported. It
+    does mean a marker edited to add a field is not caught here, which costs
+    nothing, because this receipt travels unsigned inside the file it
+    describes and was never a tamper defence.
+    """
+
+    path: str
+    size_bytes: int
+    sha256: str
+
+    def to_dict_for_hashing(self) -> dict[str, str | int]:
+        return {
+            "path": self.path,
+            "size_bytes": self.size_bytes,
+            "sha256": self.sha256,
+        }
+
+
+def _parse_file_integrity_record(entry: object) -> FileIntegrityRecord:
+    """Check one raw marker entry at the boundary and type it.
+
+    Strict, no coercion: a receipt whose ``sha256`` arrived as a JSON number
+    used to fall through to a per-file comparison that can never succeed and
+    was reported as damaged bytes; the truth is that the receipt itself is
+    malformed, which is unreadable input (#489).
+    """
+    if not isinstance(entry, dict):
+        raise ValueError(
+            f"integrity receipt entry must be a JSON object, got {type(entry).__name__}"
+        )
+    for field_name, expected_type in (
+        ("path", str),
+        ("sha256", str),
+        ("size_bytes", int),
+    ):
+        value = entry.get(field_name)
+        if isinstance(value, bool) or not isinstance(value, expected_type):
+            raise ValueError(
+                f"integrity receipt entry field {field_name!r} must be "
+                f"{expected_type.__name__}, got {type(value).__name__}"
+            )
+    return FileIntegrityRecord(
+        path=entry["path"],
+        size_bytes=entry["size_bytes"],
+        sha256=entry["sha256"],
+    )
+
+
+def _file_integrity_record(relative_path: str, absolute_path: Path) -> FileIntegrityRecord:
     """Receipt for one delivered snapshot file (table or copied asset)."""
-    return {
-        "path": relative_path,
-        "size_bytes": absolute_path.stat().st_size,
-        "sha256": _sha256_hex(absolute_path),
-    }
+    return FileIntegrityRecord(
+        path=relative_path,
+        size_bytes=absolute_path.stat().st_size,
+        sha256=_sha256_hex(absolute_path),
+    )
 
 
-def _inventory_content_id(entries: list[dict[str, str | int]]) -> str:
+def _inventory_content_id(entries: list[FileIntegrityRecord]) -> str:
     """Full SHA-256 of the normalized integrity inventory.
 
     Entries are sorted by ``path`` and serialized with stable separators so the
@@ -169,8 +232,12 @@ def _inventory_content_id(entries: list[dict[str, str | int]]) -> str:
     full-length like the per-file hashes and Croissant's SHA-256
     recommendation.
     """
-    normalized = sorted(entries, key=lambda entry: str(entry["path"]))
-    payload = json.dumps(normalized, sort_keys=True, separators=(",", ":"))
+    normalized = sorted(entries, key=lambda record: record.path)
+    payload = json.dumps(
+        [record.to_dict_for_hashing() for record in normalized],
+        sort_keys=True,
+        separators=(",", ":"),
+    )
     return hashlib.sha256(payload.encode()).hexdigest()
 
 
@@ -186,7 +253,7 @@ def _build_snapshot_integrity_marker_fields(
     fetched for hashing. Copy mode re-reads each copied asset once after the
     copy to compute its hash.
     """
-    tables: dict[str, dict[str, str | int]] = {}
+    tables: dict[str, FileIntegrityRecord] = {}
     for table_name, file_name in _REQUIRED_TABLE_FILES.items():
         absolute_path = staging_directory / file_name
         if not absolute_path.is_file():
@@ -196,7 +263,7 @@ def _build_snapshot_integrity_marker_fields(
             )
         tables[table_name] = _file_integrity_record(file_name, absolute_path)
 
-    assets: list[dict[str, str | int]] = []
+    assets: list[FileIntegrityRecord] = []
     assets_directory = staging_directory / _COPIED_ASSETS_DIRECTORY_NAME
     if assets_directory.is_dir():
         for absolute_path in sorted(assets_directory.rglob("*")):
@@ -208,8 +275,10 @@ def _build_snapshot_integrity_marker_fields(
     inventory = [*tables.values(), *assets]
     return {
         "integrity": {
-            "tables": tables,
-            "assets": assets,
+            "tables": {
+                table_name: record.to_dict_for_hashing() for table_name, record in tables.items()
+            },
+            "assets": [record.to_dict_for_hashing() for record in assets],
             "content_id": _inventory_content_id(inventory),
         }
     }
@@ -915,35 +984,44 @@ def verify_dataset_snapshot(
             ],
         )
 
-    receipt_entries: list[dict[str, str | int]] = [
-        *integrity.get("tables", {}).values(),
-        *integrity.get("assets", []),
+    # Boundary parse (#489): receipt entries arrive from external JSON with
+    # unknown types; they become typed records here or the verify refuses,
+    # naming the field. Refusal is exit 2 unreadable input, not a finding:
+    # a receipt whose sha256 arrived as a number used to fall through to a
+    # per-file comparison that can never succeed and was reported as damaged
+    # bytes.
+    receipt_records = [
+        _parse_file_integrity_record(entry)
+        for entry in [
+            *integrity.get("tables", {}).values(),
+            *integrity.get("assets", []),
+        ]
     ]
 
     # The deleted-member gate (#473): when a receipt entry and its file are
     # both gone, the surviving entries are self-consistent and every per-file
     # check passes; only the stored inventory hash, computed over the original
     # set, differs from the hash of what remains. Recompute it exactly as the
-    # exporter did (:194) and refuse a receipt that no longer describes the
-    # delivered set. Internal inconsistency is unreadable input, not damage,
-    # so it raises to exit 2 like an unparsable marker rather than reporting
+    # exporter did and refuse a receipt that no longer describes the delivered
+    # set. Internal inconsistency is unreadable input, not damage, so it
+    # raises to exit 2 like an unparsable marker rather than reporting
     # findings.
     stored_content_id = integrity.get("content_id")
     if not isinstance(stored_content_id, str) or not stored_content_id:
         raise ValueError(
             "format.json integrity receipt carries no usable content_id; "
             "the delivered member set cannot be checked against the receipt"
         )
-    recomputed_content_id = _inventory_content_id(receipt_entries)
+    recomputed_content_id = _inventory_content_id(receipt_records)
     if recomputed_content_id != stored_content_id:
         raise ValueError(
             "format.json integrity receipt is internally inconsistent: recomputed "
             f"inventory content_id {recomputed_content_id!r} != stored "
             f"{stored_content_id!r}; the receipt no longer describes the delivered set"
         )
 
-    for entry in receipt_entries:
-        relative_path = str(entry["path"])
+    for record in receipt_records:
+        relative_path = record.path
         delivered_path = resolved_directory / relative_path
         if not delivered_path.is_file():
             findings.append(
@@ -958,28 +1036,27 @@ def verify_dataset_snapshot(
             )
             continue
         delivered_size = delivered_path.stat().st_size
-        receipt_size = int(entry["size_bytes"])
-        if delivered_size != receipt_size:
+        if delivered_size != record.size_bytes:
             findings.append(
                 VerificationFinding(
                     uri=relative_path,
                     reason=REASON_SIZE_MISMATCH,
                     detail=(
                         f"size under the verified root {delivered_size} bytes "
-                        f"!= receipt {receipt_size} bytes"
+                        f"!= receipt {record.size_bytes} bytes"
                     ),
                 )
             )
             continue
         delivered_sha256 = _sha256_hex(delivered_path)
-        if delivered_sha256 != entry["sha256"]:
+        if delivered_sha256 != record.sha256:
             findings.append(
                 VerificationFinding(
                     uri=relative_path,
                     reason=REASON_CONTENT_ID_MISMATCH,
                     detail=(
                         f"sha256 under the verified root {delivered_sha256!r} "
-                        f"!= receipt sha256 {entry['sha256']!r}"
+                        f"!= receipt sha256 {record.sha256!r}"
                     ),
                 )
             )

tests/test_dataset_snapshot.py

@@ -193,8 +193,11 @@ def test_dataset_snapshot_is_tool_neutral_and_selected_by_manifest(tmp_path: Pat
         assert receipt["path"] == file_name
         assert receipt["size_bytes"] == (output_directory / file_name).stat().st_size
         assert receipt["sha256"] == snapshot_module._sha256_hex(output_directory / file_name)
-    inventory = [*integrity["tables"].values(), *integrity["assets"]]
-    assert integrity["content_id"] == snapshot_module._inventory_content_id(inventory)
+    inventory_records = [
+        snapshot_module._parse_file_integrity_record(entry)
+        for entry in [*integrity["tables"].values(), *integrity["assets"]]
+    ]
+    assert integrity["content_id"] == snapshot_module._inventory_content_id(inventory_records)
     assert len(integrity["content_id"]) == 64
 
     sample_row = duckdb.execute(
@@ -739,8 +742,11 @@ def test_dataset_snapshot_copy_mode_records_asset_integrity(tmp_path: Path) -> N
     assert asset_receipt["path"].startswith("assets/")
     assert asset_receipt["size_bytes"] == asset_path.stat().st_size
     assert asset_receipt["sha256"] == snapshot_module._sha256_hex(asset_path)
-    inventory = [*integrity["tables"].values(), *integrity["assets"]]
-    assert integrity["content_id"] == snapshot_module._inventory_content_id(inventory)
+    inventory_records = [
+        snapshot_module._parse_file_integrity_record(entry)
+        for entry in [*integrity["tables"].values(), *integrity["assets"]]
+    ]
+    assert integrity["content_id"] == snapshot_module._inventory_content_id(inventory_records)
     assert len(integrity["content_id"]) == 64
 
 

tests/test_snapshot_verify.py

@@ -425,3 +425,97 @@ def test_damage_is_reported_from_the_verified_root_not_the_export_root(
     damaged_report = verify_dataset_snapshot(root_b)
     assert not damaged_report.ok
     assert [f.reason for f in damaged_report.findings] == ["content-id-mismatch"]
+
+
+_KNOWN_RECEIPT_ENTRIES: list[dict[str, str | int]] = [
+    {
+        "path": "samples.parquet",
+        "size_bytes": 164981,
+        "sha256": "a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f90",
+    },
+    {
+        "path": "measurements.parquet",
+        "size_bytes": 5223,
+        "sha256": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+    },
+    {
+        "path": "assets/wrist_cam/frame_0000000001.jpg",
+        "size_bytes": 20481,
+        "sha256": "4444444444444444444444444444444444444444444444444444444444444444",
+    },
+]
+
+# The digest over these exact entries, serialized the way the exporter has
+# always done it (sorted by path, keys sorted, compact separators). If this
+# value changes, every snapshot ever exported fails verification.
+_GOLDEN_INVENTORY_CONTENT_ID = "b4cd6b846051175bbf1c57e5f5fcd5edee479b5cf2afd8294335c071561217bf"
+
+
+def test_inventory_digest_is_byte_identical_through_the_record_bridge() -> None:
+    """#489's hard constraint: typing the receipt entries must not move the
+    content_id hash by one byte. The old path hashes raw dicts straight from
+    the marker; the new path hashes records converted back through
+    ``to_dict_for_hashing``. Both must produce the same string and the same
+    digest, and the digest must equal the golden value."""
+    old_payload = json.dumps(
+        sorted(_KNOWN_RECEIPT_ENTRIES, key=lambda entry: str(entry["path"])),
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    records = [
+        hflow.snapshot._parse_file_integrity_record(entry) for entry in _KNOWN_RECEIPT_ENTRIES
+    ]
+    new_payload = json.dumps(
+        [record.to_dict_for_hashing() for record in sorted(records, key=lambda r: r.path)],
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+
+    assert new_payload == old_payload
+    assert hflow.snapshot._inventory_content_id(records) == hflow.snapshot._inventory_content_id(
+        [hflow.snapshot._parse_file_integrity_record(entry) for entry in _KNOWN_RECEIPT_ENTRIES]
+    )
+    assert hflow.snapshot._inventory_content_id(records) == _GOLDEN_INVENTORY_CONTENT_ID
+
+
+def test_the_digest_covers_the_three_delivery_fields_and_nothing_else() -> None:
+    """Typing the entries narrowed what the digest is computed over, and that
+    is a decision rather than an accident.
+
+    Hashing raw dicts meant the digest depended on every key an entry
+    happened to carry. Hashing records means it depends on exactly ``path``,
+    ``size_bytes`` and ``sha256``, which are the three facts that define a
+    delivery. An entry carrying an extra key therefore hashes the same now
+    and used to hash differently.
+
+    The consequence to keep: additive metadata in a later format revision
+    cannot silently invalidate the digest of every snapshot already
+    exported. The consequence to know: a marker whose entries were edited to
+    add a field is no longer caught here. That is not a loss, because the
+    receipt travels unsigned inside the file it describes and was never a
+    tamper defence, and the guarantee the docs make (a deleted member stays
+    visible) is unaffected. Restoring raw-dict hashing to "tighten" this
+    would trade a real compatibility property for an imaginary one.
+    """
+    entries_with_an_extra_field = [
+        {**entry, "injected_field": "not written by hflow"} for entry in _KNOWN_RECEIPT_ENTRIES
+    ]
+    records = [
+        hflow.snapshot._parse_file_integrity_record(entry) for entry in entries_with_an_extra_field
+    ]
+
+    assert hflow.snapshot._inventory_content_id(records) == _GOLDEN_INVENTORY_CONTENT_ID
+
+    # And the deleted-member guarantee still holds over the narrowed digest.
+    assert hflow.snapshot._inventory_content_id(records[:-1]) != _GOLDEN_INVENTORY_CONTENT_ID
+
+
+def test_receipt_entry_with_numeric_sha256_is_refused_at_the_boundary() -> None:
+    """#489's silent bug: a receipt whose sha256 arrived as a JSON number
+    used to reach a per-file comparison that can never succeed and was
+    reported as damaged bytes. The boundary refuses it instead, naming the
+    field, because a malformed receipt is unreadable input."""
+    from hflow.snapshot import _parse_file_integrity_record
+
+    with pytest.raises(ValueError, match="sha256"):
+        _parse_file_integrity_record({"path": "samples.parquet", "size_bytes": 10, "sha256": 123})