fix(snapshot): refuse markers whose format identity the exporter would refuse

src/hflow/snapshot.py

@@ -126,6 +126,25 @@ def _sha256_hex(path: Path) -> str:
     return digest.hexdigest()
 
 
+def _marker_identifies_dataset_snapshot(format_marker: dict) -> bool:
+    """Whether ``format_marker`` claims to be a snapshot this version handles.
+
+    Two callers ask this and must never disagree: the exporter deciding
+    whether a destination is one of ours to replace, and the verifier
+    deciding whether a directory is one of ours to certify (#472). If the
+    verifier were the looser of the two, exit 0 would mean "some directory
+    with an integrity-shaped key matched".
+
+    Deliberately strict about the version's type. The writer records the
+    string ``"1"``, so a JSON number ``1`` is a different value and is
+    refused. Callers phrase their own message; only the predicate is shared.
+    """
+    return (
+        format_marker.get("format") == DATASET_SNAPSHOT_FORMAT_NAME
+        and format_marker.get("format_version") == DATASET_SNAPSHOT_FORMAT_VERSION
+    )
+
+
 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 {
@@ -614,10 +633,7 @@ def _parse_dataset_snapshot_destination(
             f"snapshot export destination {output_directory} cannot be replaced because "
             f"{_FORMAT_MARKER_FILE_NAME} is not a JSON object"
         )
-    if (
-        format_marker.get("format") != DATASET_SNAPSHOT_FORMAT_NAME
-        or format_marker.get("format_version") != DATASET_SNAPSHOT_FORMAT_VERSION
-    ):
+    if not _marker_identifies_dataset_snapshot(format_marker):
         raise ValueError(
             f"snapshot export destination {output_directory} cannot be replaced because "
             f"{_FORMAT_MARKER_FILE_NAME} does not identify supported "
@@ -865,6 +881,23 @@ def verify_dataset_snapshot(
     except (json.JSONDecodeError, UnicodeDecodeError) as error:
         raise ValueError(f"format.json is unreadable: {error}") from error
 
+    # Format identity gate (#472), sharing the exporter's predicate so the two
+    # can never drift apart. Exit 0 then means "this is an HFlow snapshot and
+    # the receipt matched", never "some directory with an integrity-shaped key
+    # matched". The message names the values found and calls out the version's
+    # type, because a JSON number 1 is the easy mistake to make.
+    found_format = format_marker.get("format")
+    found_version = format_marker.get("format_version")
+    if not _marker_identifies_dataset_snapshot(format_marker):
+        raise ValueError(
+            f"format.json is not a {DATASET_SNAPSHOT_FORMAT_NAME!r} format version "
+            f"{DATASET_SNAPSHOT_FORMAT_VERSION!r} dataset snapshot: found format "
+            f"{found_format!r}, format_version {found_version!r}. Both must match "
+            "the exporter exactly, including the version's type: the writer records "
+            f"it as the string {DATASET_SNAPSHOT_FORMAT_VERSION!r}, so a JSON number "
+            "1 is refused"
+        )
+
     integrity = format_marker.get("integrity")
     if not isinstance(integrity, dict):
         # A valid v1 snapshot from before #401: verifiable nothing, corrupt nothing.

tests/test_dataset_snapshot.py

@@ -437,6 +437,45 @@ def test_dataset_snapshot_overwrite_refuses_a_symlinked_format_marker(tmp_path:
     assert format_marker.is_symlink()
 
 
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("format", "someone-elses-dataset-snapshot"),
+        ("format_version", "2"),
+        ("format_version", 1),
+    ],
+    ids=["foreign-format-name", "future-version", "version-as-a-json-number"],
+)
+def test_dataset_snapshot_overwrite_refuses_a_marker_without_our_format_identity(
+    tmp_path: Path, field: str, value: object
+) -> None:
+    """Overwrite is destructive, so identity is what makes it safe.
+
+    This guard shares its predicate with the verifier's #472 gate, and it had
+    no test of its own: disabling it left the whole suite green while the
+    verifier's tests kept passing. A shared predicate needs a case on both
+    sides or it can be loosened from one and noticed by neither.
+
+    The directory keeps its contents, which is the part that matters: a
+    refused overwrite must not have deleted anything first.
+    """
+    catalog = Catalog(tmp_path / "catalog")
+    output_directory = tmp_path / "dataset-snapshot"
+    hflow.export_dataset_snapshot(catalog.location, output_directory)
+    format_marker = output_directory / "format.json"
+    marker = json.loads(format_marker.read_text())
+    marker[field] = value
+    format_marker.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n")
+    sentinel = output_directory / "samples.parquet"
+    sentinel_bytes = sentinel.read_bytes()
+
+    with pytest.raises(ValueError, match="does not identify supported"):
+        hflow.export_dataset_snapshot(catalog.location, output_directory, overwrite=True)
+
+    assert sentinel.read_bytes() == sentinel_bytes
+    assert json.loads(format_marker.read_text())[field] == value
+
+
 def test_dataset_snapshot_excludes_check_runs_without_a_committed_episode(tmp_path: Path) -> None:
     catalog = Catalog(tmp_path / "catalog")
     canonical_episode = tmp_path / "committed.canonical.mcap"

tests/test_snapshot_verify.py

@@ -6,6 +6,7 @@
 nothing less.
 """
 
+import hashlib
 import json
 import shutil
 from pathlib import Path
@@ -223,6 +224,81 @@ def test_pre_401_format_json_is_unverifiable_not_corrupt(tmp_path: Path) -> None
     assert [f.reason for f in report.findings] == ["no-receipt"]
 
 
+def test_foreign_marker_is_refused_at_the_boundary(tmp_path: Path) -> None:
+    """#472: a directory the exporter would refuse cannot be certified. A
+    marker with an integrity-shaped key but no format identity never reaches
+    the receipt logic; verify raises and the CLI maps to exit 2."""
+    foreign = tmp_path / "some-other-tools-output"
+    foreign.mkdir()
+    payload = b"not-a-hflow-snapshot-at-all"
+    (foreign / "data.parquet").write_bytes(payload)
+    (foreign / "format.json").write_text(
+        json.dumps(
+            {
+                "producer": "not-hflow",
+                "integrity": {
+                    "tables": {
+                        "data": {
+                            "path": "data.parquet",
+                            "size_bytes": len(payload),
+                            "sha256": hashlib.sha256(payload).hexdigest(),
+                        }
+                    }
+                },
+            }
+        )
+    )
+
+    with pytest.raises(ValueError, match="not a 'hflow-dataset-snapshot'"):
+        verify_dataset_snapshot(foreign)
+
+    assert cli_main(["verify", "snapshot", str(foreign)]) == 2
+
+
+def test_unsupported_or_mistyped_version_is_refused(tmp_path: Path) -> None:
+    """#472: version 1 is the only version there has ever been, and the
+    comparison is deliberately identical to the writer, which records the
+    version as a string. A future version raises, and so does a JSON number
+    1: an easy honest mistake, so the error says exactly why."""
+    output_directory, _ = _export_two_episode_snapshot(tmp_path, "references")
+    marker_path = output_directory / "format.json"
+
+    marker = json.loads(marker_path.read_text())
+    marker["format_version"] = "2"
+    marker_path.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n")
+    with pytest.raises(ValueError, match="format_version '2'"):
+        verify_dataset_snapshot(output_directory)
+
+    marker["format_version"] = 1
+    marker_path.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n")
+    with pytest.raises(ValueError, match="JSON number 1 is refused"):
+        verify_dataset_snapshot(output_directory)
+
+    marker["format_version"] = "1"
+    marker_path.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n")
+    assert cli_main(["verify", "snapshot", str(output_directory)]) == 0
+
+
+def test_a_right_version_with_a_foreign_format_name_is_refused(tmp_path: Path) -> None:
+    """The other half of the identity predicate.
+
+    `test_foreign_marker_is_refused_at_the_boundary` uses a marker carrying
+    neither field, so the version check alone refuses it and the format-name
+    check is never the thing that fires. Dropping the name comparison from
+    the predicate left the whole suite green. This pins it: a marker claiming
+    version 1 of somebody else's format is still not ours to certify.
+    """
+    output_directory, _ = _export_two_episode_snapshot(tmp_path, "references")
+    marker_path = output_directory / "format.json"
+    marker = json.loads(marker_path.read_text())
+    marker["format"] = "someone-elses-dataset-snapshot"
+    marker_path.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n")
+
+    with pytest.raises(ValueError, match="someone-elses-dataset-snapshot"):
+        verify_dataset_snapshot(output_directory)
+    assert cli_main(["verify", "snapshot", str(output_directory)]) == 2
+
+
 def test_extra_files_under_assets_are_ignored(tmp_path: Path) -> None:
     """Files the receipt does not name produce no finding and no warning:
     unlisted extras are outside the receipt's contract."""