feat(snapshot): record table and copied-asset integrity in format.json (#397)

docs/how-to/export-dataset-snapshot.md

@@ -58,9 +58,37 @@ dataset-snapshot/
 ## Format contract
 
 `format.json` identifies `hflow-dataset-snapshot` format version `1`, names
-each table, records the media mode, and states whether media paths are relative
-to the export directory. The JSON marker is written last and the completed
-directory is activated atomically.
+each table (a name-to-filename map), records the media mode, and states whether
+media paths are relative to the export directory. The JSON marker is written
+last and the completed directory is activated atomically.
+
+Under the same format version `1`, an additive `integrity` block records
+delivery integrity so a later verifier (not shipped here) can tell whether the
+published bytes are still intact. The original `tables` map is unchanged for
+external readers; receipts live only under `integrity`:
+
+| Field | Meaning |
+| --- | --- |
+| `tables.<name>` | Required Parquet filename (unchanged string map) |
+| `integrity.tables.<name>.path` | Same file relative to the export root |
+| `integrity.tables.<name>.size_bytes` | Size of that file at export time |
+| `integrity.tables.<name>.sha256` | Full SHA-256 of that file's bytes |
+| `integrity.assets[]` | Same `path` / `size_bytes` / `sha256` for every regular file under `assets/` in copy mode; empty in references mode (remote media are not fetched) |
+| `integrity.content_id` | Full SHA-256 of the normalized inventory (all table and asset receipts, sorted by path), so a deleted member is visible even when every remaining file still matches |
+
+Copy mode re-reads each copied asset once after the copy to compute its hash
+(a second full read of media bytes on export).
+
+This is a receipt, not a verify command: export does not re-read the
+destination after transfer, and there is no public `verify_dataset_snapshot`
+API or CLI yet. Older HFlow overwrite checks only `format` and
+`format_version`, and readers that only consume the string `tables` map keep
+working; the `integrity` key is purely additive under format version `1`.
+
+The receipt travels unsigned inside the `format.json` it describes, so it
+catches corruption and accidental loss, not tampering: anyone who can edit a
+table can recompute the hashes to match. Use a signature or an out-of-band
+checksum if you need to detect a deliberate change.
 
 The Parquet tables form one snapshot:
 

src/hflow/snapshot.py

@@ -35,6 +35,16 @@
 _TAGS_TABLE_FILE_NAME = "tags.parquet"
 _INTERVALS_TABLE_FILE_NAME = "intervals.parquet"
 _FORMAT_MARKER_FILE_NAME = "format.json"
+_REQUIRED_TABLE_FILES: dict[str, str] = {
+    "samples": _SAMPLES_TABLE_FILE_NAME,
+    "measurements": _MEASUREMENTS_TABLE_FILE_NAME,
+    "observations": _OBSERVATIONS_TABLE_FILE_NAME,
+    "media": _MEDIA_TABLE_FILE_NAME,
+    "check_runs": _CHECK_RUNS_TABLE_FILE_NAME,
+    "tags": _TAGS_TABLE_FILE_NAME,
+    "intervals": _INTERVALS_TABLE_FILE_NAME,
+}
+_COPIED_ASSETS_DIRECTORY_NAME = "assets"
 
 
 class SnapshotMediaMode(StrEnum):
@@ -103,6 +113,85 @@ def _quote_sql_string(value: str) -> str:
     return "'" + value.replace("'", "''") + "'"
 
 
+def _sha256_hex(path: Path) -> str:
+    """Full SHA-256 hex digest of a file's bytes."""
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        while chunk := stream.read(8 * 1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _file_integrity_record(relative_path: str, absolute_path: Path) -> dict[str, str | int]:
+    """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),
+    }
+
+
+def _inventory_content_id(entries: list[dict[str, str | int]]) -> str:
+    """Full SHA-256 of the normalized integrity inventory.
+
+    Entries are sorted by ``path`` and serialized with stable separators so the
+    digest depends only on the delivered set, not write order. It catches
+    corruption and accidental loss: a truncated table, a partial transfer, a
+    member that never arrived. It is not a tamper defence, because it travels
+    unsigned inside the same ``format.json`` it describes, so anyone who edits
+    a table can recompute it.
+
+    This is a delivery receipt, not an episode identity. The field name
+    ``content_id`` matches prepared-manifest receipts (#389) but the value is
+    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=(",", ":"))
+    return hashlib.sha256(payload.encode()).hexdigest()
+
+
+def _build_snapshot_integrity_marker_fields(
+    staging_directory: Path,
+) -> dict[str, object]:
+    """Additive ``integrity`` block for ``format.json`` under format version 1.
+
+    Required Parquet tables and every regular file under ``assets/`` (copy mode)
+    get ``path`` / ``size_bytes`` / ``sha256``. ``content_id`` digests that
+    normalized inventory so a deleted member is detectable without a verifier
+    product yet. References mode leaves ``assets`` empty: remote media are not
+    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]] = {}
+    for table_name, file_name in _REQUIRED_TABLE_FILES.items():
+        absolute_path = staging_directory / file_name
+        if not absolute_path.is_file():
+            raise FileNotFoundError(
+                f"snapshot staging is missing required table {file_name!r} "
+                f"under {staging_directory}"
+            )
+        tables[table_name] = _file_integrity_record(file_name, absolute_path)
+
+    assets: list[dict[str, str | int]] = []
+    assets_directory = staging_directory / _COPIED_ASSETS_DIRECTORY_NAME
+    if assets_directory.is_dir():
+        for absolute_path in sorted(assets_directory.rglob("*")):
+            if not absolute_path.is_file():
+                continue
+            relative_path = absolute_path.relative_to(staging_directory).as_posix()
+            assets.append(_file_integrity_record(relative_path, absolute_path))
+
+    inventory = [*tables.values(), *assets]
+    return {
+        "integrity": {
+            "tables": tables,
+            "assets": assets,
+            "content_id": _inventory_content_id(inventory),
+        }
+    }
+
+
 def _copy_query_to_parquet(
     connection: duckdb.DuckDBPyConnection, query: str, destination: Path
 ) -> int:
@@ -335,7 +424,11 @@ def _copied_media_relative_path(*, episode_id: str, artifact_name: str, artifact
     safe_source_file_name = _safe_file_name(source_file_name)
     identity_digest = hashlib.sha256(f"{artifact_name}\0{artifact_uri}".encode()).hexdigest()[:12]
     safe_episode_id = _safe_file_name(episode_id)
-    return Path("assets") / safe_episode_id / f"{identity_digest}-{safe_source_file_name}"
+    return (
+        Path(_COPIED_ASSETS_DIRECTORY_NAME)
+        / safe_episode_id
+        / f"{identity_digest}-{safe_source_file_name}"
+    )
 
 
 def _write_media_table(
@@ -579,6 +672,14 @@ def export_dataset_snapshot(
     mode every recorded artifact is materialized below ``assets/`` and the
     media table stores a path relative to the export directory.
 
+    ``format.json`` stays format version ``1``. The published ``tables`` map
+    remains a name-to-filename contract. An additive ``integrity`` block
+    records per-file ``path`` / ``size_bytes`` / ``sha256`` for every required
+    table and every copied asset, plus a full-length ``content_id`` over that
+    normalized inventory. Copy mode re-reads each copied asset once after the
+    copy to hash it. This is a delivery receipt only: this export does not
+    verify the destination after transfer.
+
     The completed directory appears atomically. Existing destinations are
     refused unless ``overwrite=True`` and ``format.json`` identifies a
     supported HFlow dataset snapshot; even then, the prior export remains in
@@ -670,15 +771,8 @@ def export_dataset_snapshot(
             "media_uri_base": (
                 "export_directory" if resolved_media_mode is SnapshotMediaMode.COPY else None
             ),
-            "tables": {
-                "samples": _SAMPLES_TABLE_FILE_NAME,
-                "measurements": _MEASUREMENTS_TABLE_FILE_NAME,
-                "observations": _OBSERVATIONS_TABLE_FILE_NAME,
-                "media": _MEDIA_TABLE_FILE_NAME,
-                "check_runs": _CHECK_RUNS_TABLE_FILE_NAME,
-                "tags": _TAGS_TABLE_FILE_NAME,
-                "intervals": _INTERVALS_TABLE_FILE_NAME,
-            },
+            "tables": dict(_REQUIRED_TABLE_FILES),
+            **_build_snapshot_integrity_marker_fields(staging_directory),
         }
         (staging_directory / _FORMAT_MARKER_FILE_NAME).write_text(
             json.dumps(format_marker, indent=2, sort_keys=True) + "\n"

tests/test_dataset_snapshot.py

@@ -1,4 +1,5 @@
 import json
+import shutil
 from pathlib import Path
 
 import duckdb
@@ -175,6 +176,26 @@ def test_dataset_snapshot_is_tool_neutral_and_selected_by_manifest(tmp_path: Pat
     assert format_marker["format"] == "hflow-dataset-snapshot"
     assert format_marker["format_version"] == "1"
     assert format_marker["media_mode"] == "references"
+    assert format_marker["tables"] == {
+        "samples": "samples.parquet",
+        "measurements": "measurements.parquet",
+        "observations": "observations.parquet",
+        "media": "media.parquet",
+        "check_runs": "check_runs.parquet",
+        "tags": "tags.parquet",
+        "intervals": "intervals.parquet",
+    }
+    integrity = format_marker["integrity"]
+    assert integrity["assets"] == []
+    assert set(integrity["tables"]) == set(format_marker["tables"])
+    for table_name, file_name in format_marker["tables"].items():
+        receipt = integrity["tables"][table_name]
+        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)
+    assert len(integrity["content_id"]) == 64
 
     sample_row = duckdb.execute(
         """
@@ -647,3 +668,143 @@ def test_snapshot_samples_report_unverified_for_a_crashed_critical_check(
         [str(output_directory / "samples.parquet")],
     ).fetchone()
     assert sample_row == (append_result.episode_id, "unverified")
+
+
+def test_dataset_snapshot_copy_mode_records_asset_integrity(tmp_path: Path) -> None:
+    catalog = Catalog(tmp_path / "catalog")
+    _append_snapshot_episode(
+        catalog,
+        tmp_path,
+        name="copy-integrity",
+        score=0.5,
+        with_media=True,
+    )
+    output_directory = tmp_path / "dataset-snapshot"
+
+    report = hflow.export_dataset_snapshot(
+        catalog.location,
+        output_directory,
+        media_mode=hflow.SnapshotMediaMode.COPY,
+    )
+
+    assert report.copied_media_count == 1
+    format_marker = json.loads((output_directory / "format.json").read_text())
+    assert format_marker["format_version"] == "1"
+    assert format_marker["media_mode"] == "copy"
+    assert format_marker["tables"]["media"] == "media.parquet"
+    integrity = format_marker["integrity"]
+    assert len(integrity["assets"]) == 1
+    asset_receipt = integrity["assets"][0]
+    asset_path = output_directory / asset_receipt["path"]
+    assert asset_path.is_file()
+    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)
+    assert len(integrity["content_id"]) == 64
+
+
+def test_dataset_snapshot_marker_integrity_detects_post_export_mutations(
+    tmp_path: Path,
+) -> None:
+    catalog = Catalog(tmp_path / "catalog")
+    _append_snapshot_episode(
+        catalog,
+        tmp_path,
+        name="mutation-source",
+        score=0.9,
+        with_media=True,
+    )
+    output_directory = tmp_path / "dataset-snapshot"
+    hflow.export_dataset_snapshot(
+        catalog.location,
+        output_directory,
+        media_mode=hflow.SnapshotMediaMode.COPY,
+    )
+    original_marker = json.loads((output_directory / "format.json").read_text())
+    assert original_marker["format_version"] == "1"
+    original_integrity = original_marker["integrity"]
+
+    media_copy = tmp_path / "media-mutated"
+    tags_copy = tmp_path / "tags-deleted"
+    samples_copy = tmp_path / "samples-truncated"
+    for destination in (media_copy, tags_copy, samples_copy):
+        shutil.copytree(output_directory, destination)
+
+    # Replacing copied media bytes leaves tables readable but mismatches the receipt.
+    media_receipt = original_integrity["assets"][0]
+    media_path = media_copy / media_receipt["path"]
+    media_path.write_bytes(b"replaced media bytes")
+    assert snapshot_module._sha256_hex(media_path) != media_receipt["sha256"]
+    assert media_path.stat().st_size != media_receipt["size_bytes"]
+    media_inventory = [
+        *[
+            snapshot_module._file_integrity_record(receipt["path"], media_copy / receipt["path"])
+            for receipt in original_integrity["tables"].values()
+        ],
+        snapshot_module._file_integrity_record(media_receipt["path"], media_path),
+    ]
+    assert (
+        snapshot_module._inventory_content_id(media_inventory) != original_integrity["content_id"]
+    )
+
+    # Deleting a required table is invisible to format/format_version checks,
+    # but the inventory digest no longer matches the marker.
+    (tags_copy / "tags.parquet").unlink()
+    assert not (tags_copy / "tags.parquet").exists()
+    remaining_entries = [
+        snapshot_module._file_integrity_record(receipt["path"], tags_copy / receipt["path"])
+        for receipt in original_integrity["tables"].values()
+        if (tags_copy / receipt["path"]).is_file()
+    ] + [
+        snapshot_module._file_integrity_record(receipt["path"], tags_copy / receipt["path"])
+        for receipt in original_integrity["assets"]
+    ]
+    assert (
+        snapshot_module._inventory_content_id(remaining_entries) != original_integrity["content_id"]
+    )
+    overwrite_marker = json.loads((tags_copy / "format.json").read_text())
+    assert overwrite_marker["format"] == "hflow-dataset-snapshot"
+    assert overwrite_marker["format_version"] == "1"
+    assert overwrite_marker["tables"]["tags"] == "tags.parquet"
+
+    # Truncation changes size and hash while the destination still looks like a snapshot.
+    samples_path = samples_copy / "samples.parquet"
+    samples_path.write_bytes(samples_path.read_bytes()[:64])
+    samples_receipt = original_integrity["tables"]["samples"]
+    assert samples_path.stat().st_size != samples_receipt["size_bytes"]
+    assert snapshot_module._sha256_hex(samples_path) != samples_receipt["sha256"]
+
+
+def test_dataset_snapshot_overwrite_still_accepts_integrity_enriched_v1_marker(
+    tmp_path: Path,
+) -> None:
+    catalog = Catalog(tmp_path / "catalog")
+    _append_snapshot_episode(
+        catalog,
+        tmp_path,
+        name="overwrite-v1",
+        score=0.2,
+        with_media=False,
+    )
+    output_directory = tmp_path / "dataset-snapshot"
+    hflow.export_dataset_snapshot(catalog.location, output_directory)
+    first_marker = json.loads((output_directory / "format.json").read_text())
+    assert "integrity" in first_marker
+    assert "content_id" in first_marker["integrity"]
+    assert first_marker["format_version"] == "1"
+    assert first_marker["tables"]["samples"] == "samples.parquet"
+
+    report = hflow.export_dataset_snapshot(
+        catalog.location,
+        output_directory,
+        overwrite=True,
+    )
+    assert report.retained_backup is None
+    second_marker = json.loads((output_directory / "format.json").read_text())
+    assert second_marker["format_version"] == "1"
+    assert "integrity" in second_marker
+    assert "content_id" in second_marker["integrity"]
+    assert second_marker["tables"] == first_marker["tables"]
+    assert set(second_marker["integrity"]["tables"]) == set(first_marker["tables"])