Add bounded generation transition receipts

src/docforge/generation_diff.py

@@ -0,0 +1,911 @@
+"""Bounded latest-generation transition receipts for disposable graph indexes."""
+
+from __future__ import annotations
+
+import json
+import os
+import secrets
+import stat
+from collections.abc import Mapping
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal, cast
+
+from ._fs_safety import open_bound_directory, require_bound_directory
+from .errors import DocForgeError
+from .models import Edge, Node, ProjectDescriptor, ProjectSnapshot
+from .pagination import canonical_hash
+
+GENERATION_DIFF_SCHEMA_VERSION = 1
+GENERATION_DIFF_SEMANTICS_VERSION = 1
+MAX_GENERATION_DIFF_ITEMS = 1_000
+MAX_GENERATION_DIFF_BYTES = 1_048_576
+GENERATION_DIFF_FILENAME = "generation-diff.json"
+
+IndexSignature = tuple[int, int, int, int, int]
+PredecessorReason = Literal[
+    "no_predecessor",
+    "predecessor_unsafe",
+    "predecessor_unsupported_schema",
+    "predecessor_foreign",
+    "predecessor_policy_incompatible",
+    "predecessor_corrupt",
+    "predecessor_unattested",
+    "predecessor_changed",
+    "no_meaningful_transition",
+]
+
+_GENERATION_KEYS = frozenset(
+    {
+        "revision",
+        "source_hash",
+        "node_count",
+        "node_hash",
+        "edge_count",
+        "edge_hash",
+        "index_schema_version",
+    }
+)
+_SUMMARY_KEYS = frozenset(
+    {
+        "nodes_added",
+        "nodes_removed",
+        "nodes_changed",
+        "edges_added",
+        "edges_removed",
+        "total_changes",
+    }
+)
+_SIGNATURE_KEYS = frozenset({"device", "inode", "size", "mtime_ns", "ctime_ns"})
+_RECEIPT_KEYS = frozenset(
+    {
+        "schema_version",
+        "diff_semantics_version",
+        "project_id",
+        "project_root_fingerprint",
+        "adapter",
+        "kind",
+        "reason",
+        "from_generation",
+        "to_generation",
+        "summary",
+        "items",
+        "full_item_count",
+        "retained_item_count",
+        "details_truncated",
+        "truncation_reason",
+        "full_collection_hash",
+        "retained_collection_hash",
+        "index_signature",
+        "receipt_hash",
+    }
+)
+_BASELINE_REASONS = frozenset(
+    {
+        "no_predecessor",
+        "predecessor_unsafe",
+        "predecessor_unsupported_schema",
+        "predecessor_foreign",
+        "predecessor_policy_incompatible",
+        "predecessor_corrupt",
+        "predecessor_unattested",
+        "predecessor_changed",
+        "no_meaningful_transition",
+    }
+)
+_NODE_ITEM_KEYS = frozenset(
+    {
+        "entity",
+        "change",
+        "node_id",
+        "before_content_hash",
+        "after_content_hash",
+        "before_source_path",
+        "after_source_path",
+        "before_node_hash",
+        "after_node_hash",
+        "changed_fields",
+        "item_hash",
+    }
+)
+_NODE_CHANGED_FIELDS = frozenset(
+    {
+        "title",
+        "family",
+        "authority",
+        "status",
+        "tags",
+        "summary",
+        "content",
+        "source_path",
+        "source_anchor",
+        "content_hash",
+    }
+)
+_EDGE_ITEM_KEYS = frozenset(
+    {
+        "entity",
+        "change",
+        "source_id",
+        "relation",
+        "target_id",
+        "item_hash",
+    }
+)
+
+
+@dataclass(frozen=True)
+class PublishedGraph:
+    """One completely verified predecessor publication."""
+
+    generation: dict[str, object]
+    nodes: tuple[Node, ...]
+    edges: tuple[Edge, ...]
+    logic_hash: str
+    signature: IndexSignature
+
+
+@dataclass(frozen=True)
+class GenerationDiffDraft:
+    """A receipt body awaiting the committed index file identity."""
+
+    fields: dict[str, object]
+    items: tuple[dict[str, object], ...]
+
+
+def generation_diff_path(descriptor: ProjectDescriptor) -> Path:
+    """Return the fixed project-confined latest-transition receipt path."""
+
+    return descriptor.cache_root / GENERATION_DIFF_FILENAME
+
+
+def index_signature(path: Path) -> IndexSignature:
+    """Return the exact identity of one safe regular index publication."""
+
+    try:
+        status = path.lstat()
+    except OSError as error:
+        raise DocForgeError("missing_index", "Derived index does not exist") from error
+    if stat.S_ISLNK(status.st_mode) or not stat.S_ISREG(status.st_mode):
+        raise DocForgeError("path_escape", "Derived index path is not a safe regular file")
+    return (
+        status.st_dev,
+        status.st_ino,
+        status.st_size,
+        status.st_mtime_ns,
+        status.st_ctime_ns,
+    )
+
+
+def signature_payload(signature: IndexSignature) -> dict[str, int]:
+    """Convert one stat identity to its versioned JSON representation."""
+
+    return {
+        "device": signature[0],
+        "inode": signature[1],
+        "size": signature[2],
+        "mtime_ns": signature[3],
+        "ctime_ns": signature[4],
+    }
+
+
+def generation_identity(status: Mapping[str, object]) -> dict[str, object]:
+    """Select the primary-graph identity stored in a public diff receipt."""
+
+    return {
+        "revision": status["revision"],
+        "source_hash": status["source_hash"],
+        "node_count": status["node_count"],
+        "node_hash": status["node_hash"],
+        "edge_count": status["edge_count"],
+        "edge_hash": status["edge_hash"],
+        "index_schema_version": status["index_schema_version"],
+    }
+
+
+def prepare_generation_diff(
+    descriptor: ProjectDescriptor,
+    *,
+    predecessor: PublishedGraph | None,
+    predecessor_reason: PredecessorReason | None,
+    current_snapshot: ProjectSnapshot,
+    current_status: Mapping[str, object],
+    preserved_receipt: Mapping[str, object] | None = None,
+) -> GenerationDiffDraft:
+    """Prepare one deterministic latest transition without publishing it."""
+
+    current_generation = generation_identity(current_status)
+    if predecessor is None:
+        return _baseline_draft(
+            descriptor,
+            current_generation,
+            predecessor_reason or "no_predecessor",
+        )
+
+    if predecessor.generation == current_generation:
+        if preserved_receipt is not None and _receipt_targets(
+            preserved_receipt,
+            descriptor,
+            current_generation,
+            predecessor.signature,
+        ):
+            fields = {
+                key: value
+                for key, value in preserved_receipt.items()
+                if key not in {"index_signature", "receipt_hash", "items"}
+            }
+            preserved_items = preserved_receipt.get("items")
+            if not isinstance(preserved_items, list):
+                raise DocForgeError(
+                    "invalid_generation_diff",
+                    "Preserved generation diff has no item collection",
+                )
+            items = tuple(
+                cast(dict[str, object], item)
+                for item in cast(list[object], preserved_items)
+                if isinstance(item, dict)
+            )
+            return GenerationDiffDraft(fields=fields, items=items)
+        return _baseline_draft(
+            descriptor,
+            current_generation,
+            "no_meaningful_transition",
+        )
+
+    items = _change_items(
+        predecessor.nodes,
+        predecessor.edges,
+        current_snapshot.nodes,
+        current_snapshot.edges,
+    )
+    summary = _summary(items)
+    return GenerationDiffDraft(
+        fields={
+            "schema_version": GENERATION_DIFF_SCHEMA_VERSION,
+            "diff_semantics_version": GENERATION_DIFF_SEMANTICS_VERSION,
+            "project_id": descriptor.project_id,
+            "project_root_fingerprint": _root_fingerprint(descriptor),
+            "adapter": descriptor.adapter,
+            "kind": "transition",
+            "reason": None,
+            "from_generation": predecessor.generation,
+            "to_generation": current_generation,
+            "summary": summary,
+            "full_item_count": len(items),
+            "full_collection_hash": canonical_hash([item["item_hash"] for item in items]),
+        },
+        items=items,
+    )
+
+
+def finalize_generation_diff(
+    draft: GenerationDiffDraft,
+    *,
+    signature: IndexSignature,
+) -> dict[str, object]:
+    """Bind a draft to the committed index and enforce fixed receipt ceilings."""
+
+    full_count = cast(int, draft.fields["full_item_count"])
+    retained = list(draft.items[:MAX_GENERATION_DIFF_ITEMS])
+    item_limited = full_count > len(retained)
+    byte_limited = draft.fields.get("truncation_reason") == "receipt_byte_limit"
+    while True:
+        receipt = _final_receipt(
+            draft.fields,
+            retained,
+            signature=signature,
+            item_limited=item_limited,
+            byte_limited=byte_limited,
+        )
+        if len(_receipt_bytes(receipt)) <= MAX_GENERATION_DIFF_BYTES:
+            return receipt
+        if not retained:
+            raise DocForgeError(
+                "generation_diff_failure",
+                "Generation diff identity exceeds the fixed receipt byte limit",
+            )
+        retained.pop()
+        byte_limited = True
+
+
+def publish_generation_diff(
+    descriptor: ProjectDescriptor,
+    receipt: Mapping[str, object],
+) -> None:
+    """Atomically replace the single latest-generation receipt."""
+
+    root = descriptor.cache_root
+    if generation_diff_path(descriptor).parent != root:
+        raise DocForgeError("path_escape", "Generation diff path is not confined")
+    raw = _receipt_bytes(receipt)
+    if len(raw) > MAX_GENERATION_DIFF_BYTES:
+        raise DocForgeError(
+            "generation_diff_failure",
+            "Generation diff receipt exceeds the fixed byte limit",
+        )
+    root_fd = open_bound_directory(root)
+    temporary_name = f".generation-diff-{secrets.token_hex(12)}"
+    temporary_created = False
+    try:
+        try:
+            status = os.stat(
+                GENERATION_DIFF_FILENAME,
+                dir_fd=root_fd,
+                follow_symlinks=False,
+            )
+        except FileNotFoundError:
+            status = None
+        if status is not None and (
+            stat.S_ISLNK(status.st_mode) or not stat.S_ISREG(status.st_mode)
+        ):
+            raise DocForgeError(
+                "path_escape",
+                "Generation diff receipt path is not a safe regular file",
+            )
+        descriptor_fd = os.open(
+            temporary_name,
+            os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+            0o600,
+            dir_fd=root_fd,
+        )
+        temporary_created = True
+        with os.fdopen(descriptor_fd, "wb") as handle:
+            handle.write(raw)
+            handle.flush()
+            os.fsync(handle.fileno())
+        require_bound_directory(root, root_fd)
+        os.replace(
+            temporary_name,
+            GENERATION_DIFF_FILENAME,
+            src_dir_fd=root_fd,
+            dst_dir_fd=root_fd,
+        )
+        temporary_created = False
+        os.fsync(root_fd)
+        require_bound_directory(root, root_fd)
+    except Exception:
+        if temporary_created:
+            with suppress(OSError):
+                os.unlink(temporary_name, dir_fd=root_fd)
+        raise
+    finally:
+        os.close(root_fd)
+
+
+def load_generation_diff(
+    descriptor: ProjectDescriptor,
+) -> tuple[dict[str, object] | None, str | None, IndexSignature | None]:
+    """Read and validate one receipt without repairing any derived state."""
+
+    try:
+        root_fd = open_bound_directory(descriptor.cache_root)
+    except DocForgeError as error:
+        reason = "missing_receipt" if error.code == "missing_index" else "unsafe_receipt"
+        return None, reason, None
+    try:
+        try:
+            descriptor_fd = os.open(
+                GENERATION_DIFF_FILENAME,
+                os.O_RDONLY | os.O_NOFOLLOW,
+                dir_fd=root_fd,
+            )
+        except FileNotFoundError:
+            return None, "missing_receipt", None
+        except OSError:
+            return None, "unsafe_receipt", None
+        with os.fdopen(descriptor_fd, "rb") as handle:
+            status_before = os.fstat(handle.fileno())
+            if stat.S_ISLNK(status_before.st_mode) or not stat.S_ISREG(status_before.st_mode):
+                return None, "unsafe_receipt", None
+            before: IndexSignature = (
+                status_before.st_dev,
+                status_before.st_ino,
+                status_before.st_size,
+                status_before.st_mtime_ns,
+                status_before.st_ctime_ns,
+            )
+            if before[2] > MAX_GENERATION_DIFF_BYTES:
+                return None, "oversized_receipt", before
+            raw = handle.read(MAX_GENERATION_DIFF_BYTES + 1)
+            status_after = os.fstat(handle.fileno())
+        after: IndexSignature = (
+            status_after.st_dev,
+            status_after.st_ino,
+            status_after.st_size,
+            status_after.st_mtime_ns,
+            status_after.st_ctime_ns,
+        )
+        if len(raw) > MAX_GENERATION_DIFF_BYTES:
+            return None, "oversized_receipt", before
+        try:
+            path_status = os.stat(
+                GENERATION_DIFF_FILENAME,
+                dir_fd=root_fd,
+                follow_symlinks=False,
+            )
+        except OSError:
+            return None, "receipt_changed", before
+        path_signature: IndexSignature = (
+            path_status.st_dev,
+            path_status.st_ino,
+            path_status.st_size,
+            path_status.st_mtime_ns,
+            path_status.st_ctime_ns,
+        )
+        try:
+            require_bound_directory(descriptor.cache_root, root_fd)
+        except DocForgeError:
+            return None, "unsafe_receipt", before
+    finally:
+        os.close(root_fd)
+    if before != after or before != path_signature or len(raw) != before[2]:
+        return None, "receipt_changed", before
+    try:
+        parsed: object = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return None, "corrupt_receipt", before
+    if not isinstance(parsed, dict):
+        return None, "corrupt_receipt", before
+    receipt = cast(dict[str, object], parsed)
+    if not validate_generation_diff_receipt(receipt):
+        return None, "corrupt_receipt", before
+    if (
+        receipt.get("project_id") != descriptor.project_id
+        or receipt.get("project_root_fingerprint") != _root_fingerprint(descriptor)
+        or receipt.get("adapter") != descriptor.adapter
+    ):
+        return None, "foreign_receipt", before
+    return receipt, None, before
+
+
+def valid_generation_identity(value: object) -> bool:
+    """Return whether a primary-graph generation has the exact version-1 shape."""
+
+    return _valid_generation(value)
+
+
+def validate_generation_diff_receipt(
+    receipt: Mapping[str, object],
+    *,
+    descriptor: ProjectDescriptor | None = None,
+) -> bool:
+    """Strictly validate the complete version-1 receipt and its hashes."""
+
+    if frozenset(receipt) != _RECEIPT_KEYS:
+        return False
+    if (
+        receipt.get("schema_version") != GENERATION_DIFF_SCHEMA_VERSION
+        or receipt.get("diff_semantics_version") != GENERATION_DIFF_SEMANTICS_VERSION
+        or not _nonempty(receipt.get("project_id"))
+        or not _fingerprint(receipt.get("project_root_fingerprint"))
+        or not _nonempty(receipt.get("adapter"))
+    ):
+        return False
+    if descriptor is not None and (
+        receipt["project_id"] != descriptor.project_id
+        or receipt["project_root_fingerprint"] != _root_fingerprint(descriptor)
+        or receipt["adapter"] != descriptor.adapter
+    ):
+        return False
+    kind = receipt.get("kind")
+    reason = receipt.get("reason")
+    from_generation = receipt.get("from_generation")
+    if kind == "baseline":
+        if reason not in _BASELINE_REASONS or from_generation is not None:
+            return False
+    elif kind == "transition":
+        if (
+            reason is not None
+            or not _valid_generation(from_generation)
+            or from_generation == receipt.get("to_generation")
+        ):
+            return False
+    else:
+        return False
+    if not _valid_generation(receipt.get("to_generation")):
+        return False
+    summary_value = receipt.get("summary")
+    items_value = receipt.get("items")
+    if not isinstance(summary_value, dict) or not isinstance(items_value, list):
+        return False
+    summary = cast(dict[str, object], summary_value)
+    items = cast(list[object], items_value)
+    if frozenset(summary) != _SUMMARY_KEYS or any(
+        type(value) is not int or value < 0 for value in summary.values()
+    ):
+        return False
+    total = sum(
+        cast(int, summary[key])
+        for key in (
+            "nodes_added",
+            "nodes_removed",
+            "nodes_changed",
+            "edges_added",
+            "edges_removed",
+        )
+    )
+    if summary.get("total_changes") != total:
+        return False
+    if kind == "baseline" and (total != 0 or items):
+        return False
+    if len(items) > MAX_GENERATION_DIFF_ITEMS or not all(_valid_item(item) for item in items):
+        return False
+    typed_items = tuple(cast(dict[str, object], item) for item in items)
+    if list(typed_items) != sorted(typed_items, key=_item_sort_key):
+        return False
+    identities = tuple(_item_identity(item) for item in typed_items)
+    if len(identities) != len(set(identities)):
+        return False
+    retained_summary = _summary(typed_items)
+    if any(
+        retained_summary[key] > cast(int, summary[key])
+        for key in (
+            "nodes_added",
+            "nodes_removed",
+            "nodes_changed",
+            "edges_added",
+            "edges_removed",
+        )
+    ):
+        return False
+    full_count = receipt.get("full_item_count")
+    retained_count = receipt.get("retained_item_count")
+    truncated = receipt.get("details_truncated")
+    truncation_reason = receipt.get("truncation_reason")
+    if (
+        type(full_count) is not int
+        or type(retained_count) is not int
+        or full_count != total
+        or retained_count != len(items)
+        or retained_count > full_count
+        or type(truncated) is not bool
+        or truncated != (retained_count < full_count)
+    ):
+        return False
+    if truncation_reason is None:
+        if truncated or full_count > MAX_GENERATION_DIFF_ITEMS:
+            return False
+    elif truncation_reason == "receipt_item_limit":
+        if (
+            not truncated
+            or full_count <= MAX_GENERATION_DIFF_ITEMS
+            or retained_count != MAX_GENERATION_DIFF_ITEMS
+        ):
+            return False
+    elif truncation_reason == "receipt_byte_limit":
+        if not truncated or retained_count >= min(full_count, MAX_GENERATION_DIFF_ITEMS):
+            return False
+    else:
+        return False
+    item_hashes = [cast(dict[str, object], item)["item_hash"] for item in items]
+    retained_hash = canonical_hash(item_hashes)
+    if receipt.get("retained_collection_hash") != retained_hash:
+        return False
+    full_hash = receipt.get("full_collection_hash")
+    if not _sha256(full_hash):
+        return False
+    if not truncated and full_hash != retained_hash:
+        return False
+    signature_value = receipt.get("index_signature")
+    if not isinstance(signature_value, dict):
+        return False
+    signature = cast(dict[str, object], signature_value)
+    if frozenset(signature) != _SIGNATURE_KEYS:
+        return False
+    if any(type(value) is not int or value < 0 for value in signature.values()):
+        return False
+    receipt_hash = receipt.get("receipt_hash")
+    if not _sha256(receipt_hash):
+        return False
+    unhashed = {key: value for key, value in receipt.items() if key != "receipt_hash"}
+    return receipt_hash == canonical_hash(unhashed)
+
+
+def _baseline_draft(
+    descriptor: ProjectDescriptor,
+    current_generation: Mapping[str, object],
+    reason: PredecessorReason,
+) -> GenerationDiffDraft:
+    empty_hash = canonical_hash([])
+    return GenerationDiffDraft(
+        fields={
+            "schema_version": GENERATION_DIFF_SCHEMA_VERSION,
+            "diff_semantics_version": GENERATION_DIFF_SEMANTICS_VERSION,
+            "project_id": descriptor.project_id,
+            "project_root_fingerprint": _root_fingerprint(descriptor),
+            "adapter": descriptor.adapter,
+            "kind": "baseline",
+            "reason": reason,
+            "from_generation": None,
+            "to_generation": dict(current_generation),
+            "summary": {
+                "nodes_added": 0,
+                "nodes_removed": 0,
+                "nodes_changed": 0,
+                "edges_added": 0,
+                "edges_removed": 0,
+                "total_changes": 0,
+            },
+            "full_item_count": 0,
+            "full_collection_hash": empty_hash,
+        },
+        items=(),
+    )
+
+
+def _change_items(
+    before_nodes: tuple[Node, ...],
+    before_edges: tuple[Edge, ...],
+    after_nodes: tuple[Node, ...],
+    after_edges: tuple[Edge, ...],
+) -> tuple[dict[str, object], ...]:
+    before_by_id = {node.node_id: node for node in before_nodes}
+    after_by_id = {node.node_id: node for node in after_nodes}
+    items: list[dict[str, object]] = []
+    for node_id in sorted(before_by_id.keys() | after_by_id.keys()):
+        before = before_by_id.get(node_id)
+        after = after_by_id.get(node_id)
+        if before == after:
+            continue
+        if before is None:
+            change = "added"
+        elif after is None:
+            change = "removed"
+        else:
+            change = "changed"
+        before_payload = before.as_dict() if before is not None else None
+        after_payload = after.as_dict() if after is not None else None
+        changed_fields = (
+            []
+            if before_payload is None or after_payload is None
+            else sorted(key for key in before_payload if before_payload[key] != after_payload[key])
+        )
+        payload: dict[str, object] = {
+            "entity": "node",
+            "change": change,
+            "node_id": node_id,
+            "before_content_hash": None if before is None else before.content_hash,
+            "after_content_hash": None if after is None else after.content_hash,
+            "before_source_path": None if before is None else before.source_path,
+            "after_source_path": None if after is None else after.source_path,
+            "before_node_hash": (
+                None if before_payload is None else canonical_hash(before_payload)
+            ),
+            "after_node_hash": (None if after_payload is None else canonical_hash(after_payload)),
+            "changed_fields": changed_fields,
+        }
+        payload["item_hash"] = canonical_hash(payload)
+        items.append(payload)
+    before_edge_set = {(edge.source_id, edge.relation, edge.target_id) for edge in before_edges}
+    after_edge_set = {(edge.source_id, edge.relation, edge.target_id) for edge in after_edges}
+    for change, values in (
+        ("removed", sorted(before_edge_set - after_edge_set)),
+        ("added", sorted(after_edge_set - before_edge_set)),
+    ):
+        for source_id, relation, target_id in values:
+            payload = {
+                "entity": "edge",
+                "change": change,
+                "source_id": source_id,
+                "relation": relation,
+                "target_id": target_id,
+            }
+            payload["item_hash"] = canonical_hash(payload)
+            items.append(payload)
+    return tuple(sorted(items, key=_item_sort_key))
+
+
+def _item_sort_key(item: Mapping[str, object]) -> tuple[str, str, str, str, str]:
+    return (
+        cast(str, item["entity"]),
+        cast(str, item.get("node_id", item.get("source_id", ""))),
+        cast(str, item.get("relation", "")),
+        cast(str, item.get("target_id", "")),
+        cast(str, item["change"]),
+    )
+
+
+def _item_identity(item: Mapping[str, object]) -> tuple[str, ...]:
+    if item["entity"] == "node":
+        return ("node", cast(str, item["node_id"]))
+    return (
+        "edge",
+        cast(str, item["source_id"]),
+        cast(str, item["relation"]),
+        cast(str, item["target_id"]),
+    )
+
+
+def _summary(items: tuple[dict[str, object], ...]) -> dict[str, int]:
+    result = {
+        "nodes_added": 0,
+        "nodes_removed": 0,
+        "nodes_changed": 0,
+        "edges_added": 0,
+        "edges_removed": 0,
+        "total_changes": len(items),
+    }
+    for item in items:
+        entity = cast(str, item["entity"])
+        change = cast(str, item["change"])
+        key = f"{entity}s_{change}"
+        result[key] += 1
+    return result
+
+
+def _final_receipt(
+    fields: Mapping[str, object],
+    retained: list[dict[str, object]],
+    *,
+    signature: IndexSignature,
+    item_limited: bool,
+    byte_limited: bool,
+) -> dict[str, object]:
+    full_count = cast(int, fields["full_item_count"])
+    truncated = len(retained) < full_count
+    if byte_limited:
+        reason: str | None = "receipt_byte_limit"
+    elif item_limited:
+        reason = "receipt_item_limit"
+    else:
+        reason = None
+    receipt = {
+        **fields,
+        "items": retained,
+        "retained_item_count": len(retained),
+        "details_truncated": truncated,
+        "truncation_reason": reason,
+        "retained_collection_hash": canonical_hash([item["item_hash"] for item in retained]),
+        "index_signature": signature_payload(signature),
+    }
+    receipt["receipt_hash"] = canonical_hash(receipt)
+    return receipt
+
+
+def _receipt_targets(
+    receipt: Mapping[str, object],
+    descriptor: ProjectDescriptor,
+    generation: Mapping[str, object],
+    signature: IndexSignature,
+) -> bool:
+    return (
+        validate_generation_diff_receipt(receipt, descriptor=descriptor)
+        and receipt.get("to_generation") == dict(generation)
+        and receipt.get("index_signature") == signature_payload(signature)
+    )
+
+
+def _valid_generation(value: object) -> bool:
+    if not isinstance(value, dict):
+        return False
+    generation = cast(dict[str, object], value)
+    if frozenset(generation) != _GENERATION_KEYS:
+        return False
+    source_hash = generation.get("source_hash")
+    node_hash = generation.get("node_hash")
+    edge_hash = generation.get("edge_hash")
+    return (
+        _nonempty(generation.get("revision"))
+        and _sha256(source_hash)
+        and _sha256(node_hash)
+        and _sha256(edge_hash)
+        and type(generation.get("node_count")) is int
+        and cast(int, generation["node_count"]) >= 0
+        and type(generation.get("edge_count")) is int
+        and cast(int, generation["edge_count"]) >= 0
+        and type(generation.get("index_schema_version")) is int
+        and cast(int, generation["index_schema_version"]) >= 1
+    )
+
+
+def _valid_item(value: object) -> bool:
+    if not isinstance(value, dict):
+        return False
+    payload = cast(dict[str, object], value)
+    entity = payload.get("entity")
+    if entity == "node":
+        if frozenset(payload) != _NODE_ITEM_KEYS:
+            return False
+        changed_fields_value = payload.get("changed_fields")
+        if not isinstance(changed_fields_value, list):
+            return False
+        changed_fields = cast(list[object], changed_fields_value)
+        if (
+            payload.get("change") not in {"added", "removed", "changed"}
+            or not _nonempty(payload.get("node_id"))
+            or any(
+                not _nonempty(field) or field not in _NODE_CHANGED_FIELDS
+                for field in changed_fields
+            )
+            or changed_fields != sorted(set(cast(list[str], changed_fields)))
+        ):
+            return False
+        for key in (
+            "before_content_hash",
+            "after_content_hash",
+            "before_node_hash",
+            "after_node_hash",
+        ):
+            candidate = payload.get(key)
+            if candidate is not None and not _sha256(candidate):
+                return False
+        for key in ("before_source_path", "after_source_path"):
+            candidate = payload.get(key)
+            if candidate is not None and not _nonempty(candidate):
+                return False
+        change = payload["change"]
+        before_values = (
+            payload["before_content_hash"],
+            payload["before_source_path"],
+            payload["before_node_hash"],
+        )
+        after_values = (
+            payload["after_content_hash"],
+            payload["after_source_path"],
+            payload["after_node_hash"],
+        )
+        if change == "added":
+            if (
+                any(value is not None for value in before_values)
+                or not all(value is not None for value in after_values)
+                or changed_fields
+            ):
+                return False
+        elif change == "removed":
+            if (
+                not all(value is not None for value in before_values)
+                or any(value is not None for value in after_values)
+                or changed_fields
+            ):
+                return False
+        elif (
+            not all(value is not None for value in (*before_values, *after_values))
+            or not changed_fields
+            or payload["before_node_hash"] == payload["after_node_hash"]
+        ):
+            return False
+    elif entity == "edge":
+        if (
+            frozenset(payload) != _EDGE_ITEM_KEYS
+            or payload.get("change") not in {"added", "removed"}
+            or not _nonempty(payload.get("source_id"))
+            or not _nonempty(payload.get("relation"))
+            or not _nonempty(payload.get("target_id"))
+        ):
+            return False
+    else:
+        return False
+    item_hash = payload.get("item_hash")
+    unhashed = {key: item for key, item in payload.items() if key != "item_hash"}
+    return _sha256(item_hash) and item_hash == canonical_hash(unhashed)
+
+
+def _receipt_bytes(receipt: Mapping[str, object]) -> bytes:
+    return json.dumps(receipt, sort_keys=True, indent=2, ensure_ascii=True).encode("utf-8") + b"\n"
+
+
+def _root_fingerprint(descriptor: ProjectDescriptor) -> str:
+    from .project import project_root_fingerprint
+
+    return project_root_fingerprint(descriptor.root)
+
+
+def _nonempty(value: object) -> bool:
+    return isinstance(value, str) and bool(value)
+
+
+def _sha256(value: object) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) == 64
+        and all(character in "0123456789abcdef" for character in value)
+    )
+
+
+def _fingerprint(value: object) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) == 16
+        and all(character in "0123456789abcdef" for character in value)
+    )