graflo.architecture.evolution.autogenerate

Derive a contract change set from two manifests.

Before this, nothing produced :data:~graflo.architecture.evolution.ops.ManifestOp values: every op in the codebase was hand-built. SchemaDiff (migrate) produces description records on a different plane — it reports what changed but emits nothing that can be applied — and it only ever looks at Schema, never at ingestion_model or bindings.

:func:diff_manifests closes that gap for the mechanically derivable ops, and is explicit about the rest. Its contract is the replay invariant::

ops, warnings = diff_manifests(base, target)
manifest_hash(apply_evolution(base, ops)) == manifest_hash(target)

Where that does not hold, :func:diff_manifests_verified reports the residual rather than claiming success. Silence about an incomplete diff is the one failure mode a change-set generator must not have — it produces a revision that looks applied and is not.

Renames are ambiguous by construction: a dropped mail plus an added email is indistinguishable from a rename. Pass :class:RenameHints when the intent is known; otherwise the pair is emitted as a drop and an add.

Attributes

__all__ = ['RenameHints', 'diff_manifests', 'diff_manifests_verified'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

RenameHints

Bases: ConfigBaseModel

Renames the differ cannot infer, supplied by the caller.

A drop plus an add is structurally identical to a rename. Guessing would turn a data-preserving rename into a destructive drop (or the reverse), so the differ never guesses.

Source code in graflo/architecture/evolution/autogenerate.py

class RenameHints(ConfigBaseModel):
    """Renames the differ cannot infer, supplied by the caller.

    A drop plus an add is structurally identical to a rename. Guessing would
    turn a data-preserving rename into a destructive drop (or the reverse), so
    the differ never guesses.
    """

    vertices: dict[str, str] = PydanticField(
        default_factory=dict, description="``{old_vertex_name: new_vertex_name}``."
    )
    relations: dict[str, str] = PydanticField(
        default_factory=dict, description="``{old_relation: new_relation}``."
    )
    resources: dict[str, str] = PydanticField(
        default_factory=dict, description="``{old_resource: new_resource}``."
    )
    vertex_properties: dict[str, dict[str, str]] = PydanticField(
        default_factory=dict,
        description="``{vertex_name: {old_field: new_field}}``.",
    )
    edge_properties: dict[str, dict[str, str]] = PydanticField(
        default_factory=dict,
        description="``{relation: {old_field: new_field}}``.",
    )

    @model_validator(mode="after")
    def _reject_collapsing_maps(self) -> RenameHints:
        """Reject hints that would collapse two names onto one.

        The hints are handed to the rename ops verbatim, and the differ itself keys
        by the renamed name (``hints.vertices.get(name, name)``), so a collapsing
        hint corrupts the *diff* before any op is applied.
        """
        validate_rename_map_is_injective(
            self.vertices,
            kind="rename hint: vertices",
            merge_hint="MergeVerticesOp(sources=[...], into=...)",
        )
        validate_rename_map_is_injective(
            self.relations,
            kind="rename hint: relations",
            merge_hint="MergeEdgesOp(sources=[...], into=...)",
        )
        validate_rename_map_is_injective(
            self.resources,
            kind="rename hint: resources",
            merge_hint="MergeManifestsOp renames.<side>.resources",
        )
        for vertex_name, field_renames in self.vertex_properties.items():
            validate_rename_map_is_injective(
                field_renames,
                kind=f"rename hint: vertex_properties[{vertex_name!r}]",
                merge_hint="RemoveVertexPropertiesOp to drop the redundant field first",
            )
        for relation, field_renames in self.edge_properties.items():
            validate_rename_map_is_injective(
                field_renames,
                kind=f"rename hint: edge_properties[{relation!r}]",
                merge_hint="RemoveEdgePropertiesOp to drop the redundant field first",
            )
        return self

    def to_ops(self) -> list[ManifestOp]:
        """The rename ops these hints stand for, in the order a diff emits them."""
        ops: list[ManifestOp] = []
        if self.vertices:
            ops.append(RenameVerticesOp(renames=dict(self.vertices)))
        if self.relations:
            ops.append(RenameRelationsOp(renames=dict(self.relations)))
        if self.resources:
            ops.append(RenameResourcesOp(renames=dict(self.resources)))
        if self.vertex_properties:
            ops.append(
                RenameVertexPropertiesOp(
                    renames={k: dict(v) for k, v in self.vertex_properties.items()}
                )
            )
        if self.edge_properties:
            ops.append(
                RenameEdgePropertiesOp(
                    renames={k: dict(v) for k, v in self.edge_properties.items()}
                )
            )
        return ops

    @classmethod
    def from_ops(cls, ops: list[ManifestOp], *, base: GraphManifest) -> RenameHints:
        """The renames *ops* perform on *base*, as hints for diffing it.

        Renames chain (``A → B`` then ``B → C`` is ``A → C``), and a property
        rename follows its owner's later renames. A name *base* does not have is
        left out: renaming an element the ops themselves added is part of that
        addition, not a rename of anything in *base*.
        """
        vertices: dict[str, str] = {}
        relations: dict[str, str] = {}
        resources: dict[str, str] = {}
        vertex_properties: dict[str, dict[str, str]] = {}
        edge_properties: dict[str, dict[str, str]] = {}
        for op in ops:
            if isinstance(op, RenameVerticesOp):
                vertices = _chain_renames(vertices, op.renames)
                vertex_properties = {
                    op.renames.get(owner, owner): fields
                    for owner, fields in vertex_properties.items()
                }
            elif isinstance(op, RenameRelationsOp):
                relations = _chain_renames(relations, op.renames)
                edge_properties = {
                    op.renames.get(owner, owner): fields
                    for owner, fields in edge_properties.items()
                }
            elif isinstance(op, RenameResourcesOp):
                resources = _chain_renames(resources, op.renames)
            elif isinstance(op, RenameVertexPropertiesOp):
                for owner, renames in op.renames.items():
                    vertex_properties[owner] = _chain_renames(
                        vertex_properties.get(owner, {}), renames
                    )
            elif isinstance(op, RenameEdgePropertiesOp):
                # Hints are per relation; a per_edge rename is diffed as a
                # removal plus an addition.
                for owner, renames in op.renames.items():
                    edge_properties[owner] = _chain_renames(
                        edge_properties.get(owner, {}), renames
                    )

        base_vertices = _vertices(base)
        base_relation_fields: dict[str, set[str]] = {}
        for (_s, _t, relation), edge in _edges(base).items():
            if relation:
                base_relation_fields.setdefault(relation, set()).update(
                    edge.property_names
                )
        base_resources = (
            {r.name for r in base.ingestion_model.resources}
            if base.ingestion_model is not None
            else set()
        )
        vertex_origin = {new: old for old, new in vertices.items()}
        relation_origin = {new: old for old, new in relations.items()}
        return cls(
            vertices={
                old: new
                for old, new in vertices.items()
                if old in base_vertices and old != new
            },
            relations={
                old: new
                for old, new in relations.items()
                if old in base_relation_fields and old != new
            },
            resources={
                old: new
                for old, new in resources.items()
                if old in base_resources and old != new
            },
            vertex_properties={
                owner: kept
                for owner, fields in vertex_properties.items()
                if (vertex := base_vertices.get(vertex_origin.get(owner, owner)))
                is not None
                and (
                    kept := {
                        old: new
                        for old, new in fields.items()
                        if old in vertex.property_names and old != new
                    }
                )
            },
            edge_properties={
                owner: kept
                for owner, fields in edge_properties.items()
                if (
                    kept := {
                        old: new
                        for old, new in fields.items()
                        if old
                        in base_relation_fields.get(
                            relation_origin.get(owner, owner), set()
                        )
                        and old != new
                    }
                )
            },
        )

Attributes

edge_properties = PydanticField(default_factory=dict, description='``{relation: {old_field: new_field}}``.') class-attribute instance-attribute

relations = PydanticField(default_factory=dict, description='``{old_relation: new_relation}``.') class-attribute instance-attribute

resources = PydanticField(default_factory=dict, description='``{old_resource: new_resource}``.') class-attribute instance-attribute

vertex_properties = PydanticField(default_factory=dict, description='``{vertex_name: {old_field: new_field}}``.') class-attribute instance-attribute

vertices = PydanticField(default_factory=dict, description='``{old_vertex_name: new_vertex_name}``.') class-attribute instance-attribute

Methods:

from_ops(ops, *, base) classmethod

The renames ops perform on base, as hints for diffing it.

Renames chain (A → B then B → C is A → C), and a property rename follows its owner's later renames. A name base does not have is left out: renaming an element the ops themselves added is part of that addition, not a rename of anything in base.

Source code in graflo/architecture/evolution/autogenerate.py

@classmethod
def from_ops(cls, ops: list[ManifestOp], *, base: GraphManifest) -> RenameHints:
    """The renames *ops* perform on *base*, as hints for diffing it.

    Renames chain (``A → B`` then ``B → C`` is ``A → C``), and a property
    rename follows its owner's later renames. A name *base* does not have is
    left out: renaming an element the ops themselves added is part of that
    addition, not a rename of anything in *base*.
    """
    vertices: dict[str, str] = {}
    relations: dict[str, str] = {}
    resources: dict[str, str] = {}
    vertex_properties: dict[str, dict[str, str]] = {}
    edge_properties: dict[str, dict[str, str]] = {}
    for op in ops:
        if isinstance(op, RenameVerticesOp):
            vertices = _chain_renames(vertices, op.renames)
            vertex_properties = {
                op.renames.get(owner, owner): fields
                for owner, fields in vertex_properties.items()
            }
        elif isinstance(op, RenameRelationsOp):
            relations = _chain_renames(relations, op.renames)
            edge_properties = {
                op.renames.get(owner, owner): fields
                for owner, fields in edge_properties.items()
            }
        elif isinstance(op, RenameResourcesOp):
            resources = _chain_renames(resources, op.renames)
        elif isinstance(op, RenameVertexPropertiesOp):
            for owner, renames in op.renames.items():
                vertex_properties[owner] = _chain_renames(
                    vertex_properties.get(owner, {}), renames
                )
        elif isinstance(op, RenameEdgePropertiesOp):
            # Hints are per relation; a per_edge rename is diffed as a
            # removal plus an addition.
            for owner, renames in op.renames.items():
                edge_properties[owner] = _chain_renames(
                    edge_properties.get(owner, {}), renames
                )

    base_vertices = _vertices(base)
    base_relation_fields: dict[str, set[str]] = {}
    for (_s, _t, relation), edge in _edges(base).items():
        if relation:
            base_relation_fields.setdefault(relation, set()).update(
                edge.property_names
            )
    base_resources = (
        {r.name for r in base.ingestion_model.resources}
        if base.ingestion_model is not None
        else set()
    )
    vertex_origin = {new: old for old, new in vertices.items()}
    relation_origin = {new: old for old, new in relations.items()}
    return cls(
        vertices={
            old: new
            for old, new in vertices.items()
            if old in base_vertices and old != new
        },
        relations={
            old: new
            for old, new in relations.items()
            if old in base_relation_fields and old != new
        },
        resources={
            old: new
            for old, new in resources.items()
            if old in base_resources and old != new
        },
        vertex_properties={
            owner: kept
            for owner, fields in vertex_properties.items()
            if (vertex := base_vertices.get(vertex_origin.get(owner, owner)))
            is not None
            and (
                kept := {
                    old: new
                    for old, new in fields.items()
                    if old in vertex.property_names and old != new
                }
            )
        },
        edge_properties={
            owner: kept
            for owner, fields in edge_properties.items()
            if (
                kept := {
                    old: new
                    for old, new in fields.items()
                    if old
                    in base_relation_fields.get(
                        relation_origin.get(owner, owner), set()
                    )
                    and old != new
                }
            )
        },
    )

to_ops()

The rename ops these hints stand for, in the order a diff emits them.

Source code in graflo/architecture/evolution/autogenerate.py

def to_ops(self) -> list[ManifestOp]:
    """The rename ops these hints stand for, in the order a diff emits them."""
    ops: list[ManifestOp] = []
    if self.vertices:
        ops.append(RenameVerticesOp(renames=dict(self.vertices)))
    if self.relations:
        ops.append(RenameRelationsOp(renames=dict(self.relations)))
    if self.resources:
        ops.append(RenameResourcesOp(renames=dict(self.resources)))
    if self.vertex_properties:
        ops.append(
            RenameVertexPropertiesOp(
                renames={k: dict(v) for k, v in self.vertex_properties.items()}
            )
        )
    if self.edge_properties:
        ops.append(
            RenameEdgePropertiesOp(
                renames={k: dict(v) for k, v in self.edge_properties.items()}
            )
        )
    return ops

Functions:

diff_manifests(base, target, *, hints=None)

Ops turning base into target, plus warnings for what was not expressed.

Ops are ordered so each one's preconditions hold when it runs: renames first (so later ops address the new names), then additions, then property and identity changes, then removals last. Edge property changes precede edge additions: they address a whole relation, and must not reach a sibling edge the diff adds with properties of its own.

Source code in graflo/architecture/evolution/autogenerate.py

def diff_manifests(
    base: GraphManifest,
    target: GraphManifest,
    *,
    hints: RenameHints | None = None,
) -> tuple[list[ManifestOp], list[str]]:
    """Ops turning *base* into *target*, plus warnings for what was not expressed.

    Ops are ordered so each one's preconditions hold when it runs: renames
    first (so later ops address the new names), then additions, then property
    and identity changes, then removals last. Edge property changes precede edge
    additions: they address a whole relation, and must not reach a sibling edge
    the diff adds with properties of its own.
    """
    hints = hints or RenameHints()
    warnings: list[str] = []
    ops: list[ManifestOp] = []
    original = base

    rename_ops = _rename_ops(hints)
    renamed = _apply_renames(base, rename_ops)
    if renamed is not None:
        # Every later stage compares against the manifest the renames actually
        # produce, rather than re-deriving each rename's reach by hand: a rename
        # rewrites resource pipelines, secondary identities and index keys, and
        # a stage that compared against the pre-rename base would report each
        # of those as an edit -- or re-add what the rename already moved.
        base, hints = renamed, RenameHints()

    profile_replaced = _profile_needs_replacing(base, target)
    ops += rename_ops
    ops += _inverse_withdrawal_ops(base, target, hints)
    ops += _vertex_structure_ops(base, target, hints, warnings)
    ops += _edge_property_ops(base, target, hints)
    ops += _edge_structure_ops(base, target, hints, warnings)
    ops += _vertex_property_ops(base, target, hints)
    ops += _identity_ops(base, target, hints, warnings)
    ops += _semantics_ops(base, target, hints)
    ops += _inverse_declaration_ops(
        base, target, hints, profile_replaced=profile_replaced
    )
    if not profile_replaced:
        # A profile replacement carries the indexes too, so emitting both would
        # be redundant and order-sensitive.
        ops += _index_ops(base, target, hints)
    edited: list[str] = []
    resource_ops, pending_transforms = _resource_ops(
        base, target, hints, warnings, edited
    )
    ops += resource_ops
    removals = _removal_ops(base, target, hints)
    ops += _replacement_ops(
        original, ops, removals, target, edited, pending_transforms, warnings
    )
    ops += removals
    # The removal cascades prune bindings, so whether the target's block still
    # has to be set is read off the manifest the ops so far produce.
    replayed = _apply_quietly(original, ops) if ops else original
    ops += _block_ops(replayed or base, target, profile_replaced=profile_replaced)

    _warn_unexpressed(base, target, warnings, hints)
    return ops, warnings

diff_manifests_verified(base, target, *, hints=None)

:func:diff_manifests, with the replay invariant actually checked.

Applies the derived ops to a copy of base and compares the result's hash to target's. A mismatch appends a warning naming the residual difference instead of letting an incomplete change set pass as complete.

Source code in graflo/architecture/evolution/autogenerate.py

def diff_manifests_verified(
    base: GraphManifest,
    target: GraphManifest,
    *,
    hints: RenameHints | None = None,
) -> tuple[list[ManifestOp], list[str]]:
    """:func:`diff_manifests`, with the replay invariant actually checked.

    Applies the derived ops to a copy of *base* and compares the result's hash
    to *target*'s. A mismatch appends a warning naming the residual difference
    instead of letting an incomplete change set pass as complete.
    """
    from .apply import apply_evolution
    from .hashing import manifest_hash

    ops, warnings = diff_manifests(base, target, hints=hints)
    if not ops:
        if manifest_hash(base) != manifest_hash(target):
            warnings.append(
                "manifests differ but no operation was derived; the change is "
                "not expressible in the current op vocabulary"
            )
        return ops, warnings

    try:
        # No version bump: the hash comparison is about content, and a bump
        # would make every verified diff report a spurious difference.
        replayed = apply_evolution(base, ops, bump_version=False, finish_init=False)
    except Exception as exc:
        warnings.append(f"derived operations do not apply cleanly: {exc}")
        return ops, warnings

    if manifest_hash(replayed) != manifest_hash(target):
        warnings.append(
            "replaying the derived operations does not reproduce the target "
            f"manifest; residual: {_residual(replayed, target)}"
        )
    return ops, warnings