feat(snapshot): verify delivered snapshots against integrity receipts

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

@@ -44,15 +44,15 @@ The observable result is:
 
 ```text
 dataset-snapshot/
-├── format.json
-├── samples.parquet
-├── measurements.parquet
-├── observations.parquet
-├── media.parquet
-├── check_runs.parquet
-├── tags.parquet
-├── intervals.parquet
-└── assets/                 # copy mode only, when artifacts exist
+|-- format.json
+|-- samples.parquet
+|-- measurements.parquet
+|-- observations.parquet
+|-- media.parquet
+|-- check_runs.parquet
+|-- tags.parquet
+|-- intervals.parquet
+`-- assets/                 # copy mode only, when artifacts exist
 ```
 
 ## Format contract
@@ -79,11 +79,20 @@ external readers; receipts live only under `integrity`:
 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`.
+Verify the delivery after transfer with
+`hflow verify snapshot <directory>` or
+`verify_dataset_snapshot(<directory>)`: every table and copied asset named
+in the receipt is re-read and compared by size and sha256, and every finding
+is returned in one report built from the shared verification types in
+`hflow.verification` (a `VerificationReport` with status `ok`, `damaged`,
+or `unverifiable`). Recorded paths are joined only onto the handed
+directory, so a copied delivery verifies in place. Exit `0` clean,
+`1` damaged, `3` unverifiable, `2` unreadable input. A pre-#401
+`format.json` with no `integrity` key is reported as `no-receipt`,
+unverifiable, not corrupt. 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

src/hflow/__init__.py

@@ -72,6 +72,7 @@
     RetainedDatasetSnapshotBackup,
     SnapshotMediaMode,
     export_dataset_snapshot,
+    verify_dataset_snapshot,
 )
 from hflow.steps import (
     RUN_PROFILES,
@@ -208,5 +209,6 @@
     "step_version_from_contract",
     "testing",
     "to_grid",
+    "verify_dataset_snapshot",
     "write_canonical_episode",
 ]

src/hflow/cli.py

@@ -377,6 +377,31 @@ def _build_parser() -> argparse.ArgumentParser:
         help="atomically replace an existing export directory",
     )
 
+    verify_parser = subparsers.add_parser(
+        "verify",
+        help="verify a delivery against its receipt",
+        description=(
+            "Group commands for verifying deliveries against their recorded "
+            "receipts. Use `snapshot` to re-check a delivered dataset snapshot "
+            "directory against the integrity receipt inside its format.json."
+        ),
+    )
+    verify_subparsers = verify_parser.add_subparsers(dest="verify_command", required=True)
+    verify_snapshot_parser = verify_subparsers.add_parser(
+        "snapshot",
+        help="verify a delivered dataset snapshot against its integrity receipt",
+        description=(
+            "Re-reads every table and copied asset named in the snapshot's "
+            "integrity receipt and reports bytes changed, files missing, and "
+            "size mismatches. Exit 0 clean, 1 damaged, 2 unreadable input, "
+            "3 no receipt (unverifiable)."
+        ),
+    )
+    verify_snapshot_parser.add_argument(
+        "directory",
+        help="the delivered dataset snapshot directory to verify",
+    )
+
     stale_parser = subparsers.add_parser(
         "stale",
         help="list episodes whose latest cataloged run predates the current pipeline version",
@@ -1472,6 +1497,25 @@ def _command_export_snapshot(arguments: argparse.Namespace) -> int:
     return 0
 
 
+def _command_verify_snapshot(arguments: argparse.Namespace) -> int:
+    from hflow.snapshot import verify_dataset_snapshot
+    from hflow.verification import exit_code_for
+
+    try:
+        report = verify_dataset_snapshot(Path(arguments.directory))
+    except (ValueError, FileNotFoundError, NotADirectoryError, OSError) as error:
+        print(f"verify snapshot: {error}", file=sys.stderr)
+        return 2
+    if report.ok:
+        print("verified: every receipted file matches its size and sha256")
+        return exit_code_for(report)
+    print(f"not verified: {len(report.findings)} finding(s)")
+    for finding in report.findings:
+        print(f"  [{finding.reason}] {finding.uri}")
+        print(f"    {finding.detail}")
+    return exit_code_for(report)
+
+
 def _command_doctor(arguments: argparse.Namespace) -> int:
     # Findings, not exceptions, across files as well: an unreadable path is a
     # finding about the corpus, reported in place, so a batch run never loses
@@ -1644,6 +1688,10 @@ def main(argv: list[str] | None = None) -> int:
         if arguments.export_command == "snapshot":
             return _command_export_snapshot(arguments)
         raise AssertionError(f"unhandled export command {arguments.export_command!r}")
+    if arguments.command == "verify":
+        if arguments.verify_command == "snapshot":
+            return _command_verify_snapshot(arguments)
+        raise AssertionError(f"unhandled verify command {arguments.verify_command!r}")
     if arguments.command == "stale":
         return _command_stale(arguments)
     if arguments.command == "doctor":

src/hflow/snapshot.py

@@ -15,6 +15,7 @@
 from dataclasses import dataclass
 from enum import StrEnum
 from pathlib import Path, PurePosixPath
+from typing import TYPE_CHECKING
 from uuid import uuid4
 
 import duckdb
@@ -24,6 +25,9 @@
 from hflow.curation import open_catalog_connection
 from hflow.storage import StorageRoot, fetch_uri
 
+if TYPE_CHECKING:
+    from hflow.verification import VerificationReport
+
 DATASET_SNAPSHOT_FORMAT_NAME = "hflow-dataset-snapshot"
 DATASET_SNAPSHOT_FORMAT_VERSION = "1"
 
@@ -797,3 +801,129 @@ def export_dataset_snapshot(
         media_mode=resolved_media_mode,
         retained_backup=retained_backup,
     )
+
+
+def verify_dataset_snapshot(
+    snapshot_directory: Path | str,
+) -> "VerificationReport":
+    """Check a delivered dataset snapshot against the receipt inside it.
+
+    Re-reads every table and copied asset named in the ``integrity`` block of
+    ``format.json`` and compares sizes and SHA-256 digests, resolving every
+    recorded path strictly relative to the handed snapshot directory. A
+    recorded path is never read as absolute, so a snapshot moved or copied
+    to another root verifies in place. Returns every finding in one report
+    instead of raising at the first mismatch, because a partial transfer
+    usually damages more than one file.
+
+    Findings use the shared reason vocabulary from ``hflow.verification``
+    (#432 contract, #454 shape), each meaning something different:
+
+    - ``content-id-mismatch``: the sha256 of the file under this root does
+      not match the sha256 recorded in the receipt.
+    - ``missing``: named in the receipt, absent under this root.
+    - ``size-mismatch``: size differs first; the hash read is skipped.
+    - ``no-receipt``: a valid pre-#401 ``format.json`` with no
+      ``integrity`` key. That snapshot is unverifiable, not corrupt.
+
+    Unreadable input (a missing directory, a missing or unparsable
+    ``format.json``) raises instead: that is not a finding about a delivered
+    snapshot, it is the wrong input entirely.
+
+    Extra files under ``assets/`` that the receipt does not name are ignored:
+    the receipt covers what was exported, not everything a recipient may add.
+
+    This catches corruption, truncation, and partial transfer. It is not a
+    tamper defence: the receipt travels unsigned inside the same
+    ``format.json`` it describes, so anyone who can rewrite a table can
+    recompute the hashes to match.
+    """
+    from hflow.verification import (
+        REASON_CONTENT_ID_MISMATCH,
+        REASON_MISSING,
+        REASON_NO_RECEIPT,
+        REASON_SIZE_MISMATCH,
+        VerificationFinding,
+        VerificationReport,
+        VerificationStatus,
+    )
+
+    resolved_directory = Path(snapshot_directory)
+    findings: list[VerificationFinding] = []
+
+    if not resolved_directory.is_dir():
+        raise NotADirectoryError(f"snapshot directory does not exist: {resolved_directory}")
+    marker_path = resolved_directory / "format.json"
+    if not marker_path.is_file():
+        raise FileNotFoundError(f"no format.json in {resolved_directory}; not a dataset snapshot")
+    try:
+        format_marker = json.loads(marker_path.read_text())
+    except (json.JSONDecodeError, UnicodeDecodeError) as error:
+        raise ValueError(f"format.json is unreadable: {error}") from error
+
+    integrity = format_marker.get("integrity")
+    if not isinstance(integrity, dict):
+        # A valid v1 snapshot from before #401: verifiable nothing, corrupt nothing.
+        return VerificationReport(
+            status=VerificationStatus.UNVERIFIABLE,
+            findings=[
+                VerificationFinding(
+                    uri="format.json",
+                    reason=REASON_NO_RECEIPT,
+                    detail=(
+                        "format.json carries no integrity receipt "
+                        "(pre-#401 snapshot); the delivery is unverifiable, not corrupt"
+                    ),
+                ),
+            ],
+        )
+
+    receipt_entries: list[dict[str, str | int]] = [
+        *integrity.get("tables", {}).values(),
+        *integrity.get("assets", []),
+    ]
+    for entry in receipt_entries:
+        relative_path = str(entry["path"])
+        delivered_path = resolved_directory / relative_path
+        if not delivered_path.is_file():
+            findings.append(
+                VerificationFinding(
+                    uri=relative_path,
+                    reason=REASON_MISSING,
+                    detail=(
+                        f"receipted file missing at {relative_path!r} "
+                        "under the verified snapshot root"
+                    ),
+                )
+            )
+            continue
+        delivered_size = delivered_path.stat().st_size
+        receipt_size = int(entry["size_bytes"])
+        if delivered_size != receipt_size:
+            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"
+                    ),
+                )
+            )
+            continue
+        delivered_sha256 = _sha256_hex(delivered_path)
+        if delivered_sha256 != entry["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}"
+                    ),
+                )
+            )
+
+    if findings:
+        return VerificationReport(status=VerificationStatus.DAMAGED, findings=findings)
+    return VerificationReport(status=VerificationStatus.OK)

src/hflow/verification.py

@@ -0,0 +1,75 @@
+"""Shared verification types for every ``hflow verify ...`` command.
+
+``VerificationReport`` is the common surface for delivery verifiers: a
+verifier reads the receipt a delivery carries, compares it to the bytes
+under the root being verified, and returns one report. Files nobody
+listed are ignored. The LeRobot import verifier (#454) and the dataset
+snapshot verifier (#428) both return this shape, so one CLI and one
+exit-code mapping cover every delivered artifact.
+
+Reasons are fixed strings so callers can branch on them.
+``exit_code_for`` maps a report to the verify-family exit codes: 0 clean,
+1 damaged, 3 unverifiable. Exit 2 (unreadable input) is raised as an
+exception before a report exists.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import StrEnum
+
+REASON_MISSING = "missing"
+REASON_SIZE_MISMATCH = "size-mismatch"
+REASON_CONTENT_ID_MISMATCH = "content-id-mismatch"
+# #428: a readable snapshot format.json that carries no integrity receipt.
+REASON_NO_RECEIPT = "no-receipt"
+
+
+class VerificationStatus(StrEnum):
+    """Outcome of a verify that successfully read its receipt format."""
+
+    OK = "ok"
+    DAMAGED = "damaged"
+    UNVERIFIABLE = "unverifiable"
+
+
+@dataclass(frozen=True)
+class VerificationFinding:
+    """One claimed object that does not match its receipt."""
+
+    uri: str
+    reason: str
+    detail: str
+
+
+@dataclass(frozen=True)
+class VerificationReport:
+    """Result of verifying a delivered artifact against its receipts.
+
+    ``.ok`` is True when every claimed object still matches, including the
+    empty-claim case (a readable receipt that lists nothing). ``status``
+    distinguishes clean, damaged, and unverifiable (no receipt) so CLI exit
+    codes stay distinct.
+    """
+
+    status: VerificationStatus
+    findings: list[VerificationFinding] = field(default_factory=list)
+
+    @property
+    def ok(self) -> bool:
+        return self.status is VerificationStatus.OK
+
+
+def exit_code_for(report: VerificationReport) -> int:
+    """Map a readable verification outcome to the verify-family exit code.
+
+    ``0`` clean, ``1`` damaged, ``3`` unverifiable. Exit ``2`` (unreadable
+    input) is raised as an exception before a report exists.
+    """
+    if report.status is VerificationStatus.OK:
+        return 0
+    if report.status is VerificationStatus.DAMAGED:
+        return 1
+    if report.status is VerificationStatus.UNVERIFIABLE:
+        return 3
+    raise AssertionError(f"unhandled verification status {report.status!r}")