fix(snapshot): enforce the receipt inventory content_id at verify time

src/hflow/snapshot.py

@@ -828,7 +828,11 @@ def verify_dataset_snapshot(
 
     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.
+    snapshot, it is the wrong input entirely. The same applies to a receipt
+    that is internally inconsistent with itself: a missing or malformed
+    ``content_id``, or a recomputed inventory hash that disagrees with the
+    stored one, means the receipt no longer describes the delivered set
+    (tampering or a truncated write), which is exit 2, not damaged bytes.
 
     Extra files under ``assets/`` that the receipt does not name are ignored:
     the receipt covers what was exported, not everything a recipient may add.
@@ -882,6 +886,29 @@ def verify_dataset_snapshot(
         *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
+    # 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)
+    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"])
         delivered_path = resolved_directory / relative_path

tests/test_snapshot_verify.py

@@ -89,6 +89,81 @@ def test_missing_file_reports_missing_alone(tmp_path: Path) -> None:
     assert marker["integrity"]["tables"]["samples"]["path"] in report.findings[0].uri
 
 
+def test_removed_receipt_entry_and_file_raise_inventory_mismatch(tmp_path: Path) -> None:
+    """#473's deleted-member case: when a receipt entry and its file are both
+    gone, the surviving entries agree with each other and every per-file
+    check passes; only the stored inventory content_id, computed over the
+    original set, can witness the loss. The marker is internally
+    inconsistent, so verify raises (CLI exit 2) instead of certifying."""
+
+    def strip_measurements(output_directory: Path) -> str:
+        marker_path = output_directory / "format.json"
+        marker = json.loads(marker_path.read_text())
+        entry = marker["integrity"]["tables"].pop("measurements")
+        marker_path.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n")
+        (output_directory / entry["path"]).unlink()
+        return entry["path"]
+
+    output_directory, _ = _export_two_episode_snapshot(tmp_path, "references")
+    removed = strip_measurements(output_directory)
+
+    with pytest.raises(ValueError, match="content_id"):
+        verify_dataset_snapshot(output_directory)
+
+    # A fresh export for the CLI path: the raise must map to exit 2, the
+    # unreadable-input code, not to a findings-based exit.
+    output_directory, _ = _export_two_episode_snapshot(tmp_path / "cli", "references")
+    strip_measurements(output_directory)
+    assert cli_main(["verify", "snapshot", str(output_directory)]) == 2
+    assert removed
+
+
+def test_deleted_file_with_intact_receipt_reports_missing(tmp_path: Path) -> None:
+    """Negative control for #473: delete the file but keep its receipt entry.
+    This is the ordinary ``missing`` path and must keep reporting DAMAGED
+    with or without the inventory gate; it exercises the per-file loop, not
+    the gate."""
+    output_directory, marker = _export_two_episode_snapshot(tmp_path, "references")
+    (output_directory / marker["integrity"]["tables"]["measurements"]["path"]).unlink()
+
+    report = verify_dataset_snapshot(output_directory)
+
+    assert not report.ok
+    assert [f.reason for f in report.findings] == ["missing"]
+
+
+@pytest.mark.parametrize(
+    ("replacement", "label"),
+    [(None, "absent"), ("", "empty"), (0, "not-a-string"), ([], "wrong-type")],
+    ids=["absent", "empty", "not-a-string", "wrong-type"],
+)
+def test_receipt_without_a_usable_content_id_is_refused(
+    tmp_path: Path, replacement: object, label: str
+) -> None:
+    """The other half of the #473 gate, which the mismatch test cannot reach.
+
+    A receipt whose ``content_id`` is missing or unusable cannot witness a
+    deleted member at all, so certifying it would be certifying that the
+    check ran. Deleting this branch left the whole suite green, so it needs
+    its own case. An ``integrity`` block with no ``content_id`` is not
+    something hflow writes (both arrived in #401), which is exactly why a
+    marker carrying one is unreadable input rather than damaged bytes.
+    """
+    output_directory, _ = _export_two_episode_snapshot(tmp_path, "references")
+    marker_path = output_directory / "format.json"
+    marker = json.loads(marker_path.read_text())
+    if replacement is None:
+        marker["integrity"].pop("content_id")
+    else:
+        marker["integrity"]["content_id"] = replacement
+    marker_path.write_text(json.dumps(marker, indent=2, sort_keys=True) + "\n")
+
+    with pytest.raises(ValueError, match="no usable content_id"):
+        verify_dataset_snapshot(output_directory)
+
+    assert cli_main(["verify", "snapshot", str(output_directory)]) == 2, label
+
+
 def test_truncated_file_reports_size_mismatch_and_skips_the_hash(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None: