Skip to content

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.

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 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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="ComposeManifestsOp 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

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 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
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] = []

    ops += _rename_ops(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)
    ops += _identity_ops(base, target, hints, warnings)
    ops += _index_ops(base, target, hints)
    ops += _removal_ops(base, target, hints)

    _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 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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