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 with explicit resource_renames",
        )
        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

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

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.

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.
    """
    hints = hints or RenameHints()
    warnings: list[str] = []
    ops: list[ManifestOp] = []

    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_structure_ops(base, target, hints, warnings)
    ops += _vertex_property_ops(base, target, hints)
    ops += _edge_property_ops(base, target, hints, warnings)
    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)
    ops += _resource_ops(base, target, hints, warnings)
    ops += _removal_ops(base, target, hints)
    ops += _block_ops(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