Skip to content

graflo.architecture.evolution.revision

Forward-only revision chains over a base manifest.

Alembic's core abstraction is a reversible upgrade() / downgrade() pair. GraFlo cannot offer that: several ops are lossy (see :data:~graflo.architecture.evolution.inverse.IRREVERSIBLE), and a downgrade that silently produces a different manifest is worse than none.

So the model here is a git log, not an Alembic script: an ordered chain of content-hashed change sets over a base manifest. Going back means replaying the chain from the base up to the revision you want, which is always correct. Inverses are used only when no base is available, and only when every op on the path has one.

Each revision records the manifest hash before and after it, so replay is verified rather than assumed. That is the property the existing MigrationRecord cannot offer — it stores bare op-type names, dropping the targets and values a replay would need.

FileRevisionStore

Revisions on disk, one YAML file per revision.

Mirrors migrate.store.FileMigrationStore in shape. Filenames are <index>_<revision>_<slug>.yaml so a directory listing reads in order.

Source code in graflo/architecture/evolution/revision.py 
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
class FileRevisionStore:
    """Revisions on disk, one YAML file per revision.

    Mirrors ``migrate.store.FileMigrationStore`` in shape. Filenames are
    ``<index>_<revision>_<slug>.yaml`` so a directory listing reads in order.
    """

    def __init__(self, root: str | Path = ".graflo/revisions") -> None:
        self.root = Path(root)

    def load(self) -> RevisionChain:
        """Read every stored revision and link them into a chain."""
        if not self.root.exists():
            return RevisionChain()
        payloads = [FileHandle.load(path) for path in sorted(self.root.glob("*.yaml"))]
        revisions = [_revision_from_dict(payload) for payload in payloads]
        return RevisionChain(revisions=_link_order(revisions))

    def save(self, chain: RevisionChain) -> list[Path]:
        """Write *chain*, replacing whatever was there."""
        self.root.mkdir(parents=True, exist_ok=True)
        for stale in self.root.glob("*.yaml"):
            stale.unlink()
        written: list[Path] = []
        for index, revision in enumerate(chain.revisions):
            path = self.root / f"{index:04d}_{revision.revision}_{revision.slug}.yaml"
            FileHandle.dump(_revision_to_dict(revision), path)
            written.append(path)
        return written

    def append(self, revision: Revision) -> Path:
        """Add one revision to the stored chain, validating the link."""
        chain = self.load().extend(revision)
        return self.save(chain)[-1]

append(revision)

Add one revision to the stored chain, validating the link.

Source code in graflo/architecture/evolution/revision.py 
381
382
383
384
def append(self, revision: Revision) -> Path:
    """Add one revision to the stored chain, validating the link."""
    chain = self.load().extend(revision)
    return self.save(chain)[-1]

load()

Read every stored revision and link them into a chain.

Source code in graflo/architecture/evolution/revision.py 
361
362
363
364
365
366
367
def load(self) -> RevisionChain:
    """Read every stored revision and link them into a chain."""
    if not self.root.exists():
        return RevisionChain()
    payloads = [FileHandle.load(path) for path in sorted(self.root.glob("*.yaml"))]
    revisions = [_revision_from_dict(payload) for payload in payloads]
    return RevisionChain(revisions=_link_order(revisions))

save(chain)

Write chain, replacing whatever was there.

Source code in graflo/architecture/evolution/revision.py 
369
370
371
372
373
374
375
376
377
378
379
def save(self, chain: RevisionChain) -> list[Path]:
    """Write *chain*, replacing whatever was there."""
    self.root.mkdir(parents=True, exist_ok=True)
    for stale in self.root.glob("*.yaml"):
        stale.unlink()
    written: list[Path] = []
    for index, revision in enumerate(chain.revisions):
        path = self.root / f"{index:04d}_{revision.revision}_{revision.slug}.yaml"
        FileHandle.dump(_revision_to_dict(revision), path)
        written.append(path)
    return written

Revision

Bases: ConfigBaseModel

One change set, its parent, and the hashes it moves between.

Source code in graflo/architecture/evolution/revision.py 
 68
 69
 70
 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
class Revision(ConfigBaseModel):
    """One change set, its parent, and the hashes it moves between."""

    revision: str = PydanticField(
        ..., description="Content-derived id of this revision."
    )
    down_revision: str | None = PydanticField(
        default=None,
        description="Parent revision id; ``None`` marks the first revision.",
    )
    label: str | None = PydanticField(
        default=None, description="Short human-readable name."
    )
    created_at: str | None = PydanticField(
        default=None, description="ISO-8601 timestamp, supplied by the caller."
    )
    ops: list[RevisionOp] = PydanticField(
        ..., min_length=1, description="Ordered operations this revision applies."
    )
    manifest_hash_before: str = PydanticField(
        ..., description="Manifest hash this revision expects to start from."
    )
    manifest_hash_after: str = PydanticField(
        ..., description="Manifest hash this revision produces."
    )
    reversible: bool = PydanticField(
        default=True,
        description="Whether every op in this revision has a total inverse.",
    )
    notes: str | None = None

    @property
    def slug(self) -> str:
        """Filesystem-safe label fragment used in the stored filename."""
        base = _SLUG_STRIP.sub("_", (self.label or "revision").lower()).strip("_")
        return base or "revision"

slug property

Filesystem-safe label fragment used in the stored filename.

RevisionChain

Bases: ConfigBaseModel

An ordered, singly-linked chain of revisions.

Linear in v1: branching would need a merge policy for two change sets over the same parent, which is a separate design question.

Source code in graflo/architecture/evolution/revision.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
class RevisionChain(ConfigBaseModel):
    """An ordered, singly-linked chain of revisions.

    Linear in v1: branching would need a merge policy for two change sets over
    the same parent, which is a separate design question.
    """

    revisions: list[Revision] = PydanticField(default_factory=list)

    @model_validator(mode="after")
    def _validate_links(self) -> RevisionChain:
        if not self.revisions:
            return self

        ids = [revision.revision for revision in self.revisions]
        duplicates = {rev for rev in ids if ids.count(rev) > 1}
        if duplicates:
            raise ValueError(f"duplicate revision ids: {sorted(duplicates)}")

        if self.revisions[0].down_revision is not None:
            raise ValueError(
                f"the first revision '{ids[0]}' points at parent "
                f"'{self.revisions[0].down_revision}', which is not in this chain"
            )
        for previous, current in zip(self.revisions, self.revisions[1:]):
            if current.down_revision != previous.revision:
                raise ValueError(
                    f"revision '{current.revision}' points at parent "
                    f"'{current.down_revision}', but follows '{previous.revision}'; "
                    "the chain must be linear"
                )
            if current.manifest_hash_before != previous.manifest_hash_after:
                raise ValueError(
                    f"revision '{current.revision}' expects to start from hash "
                    f"{current.manifest_hash_before[:12]} but '{previous.revision}' "
                    f"produces {previous.manifest_hash_after[:12]}"
                )
        return self

    def head(self) -> Revision | None:
        """The most recent revision, or ``None`` for an empty chain."""
        return self.revisions[-1] if self.revisions else None

    def get(self, revision_id: str) -> Revision | None:
        """The revision with *revision_id*, or ``None``."""
        for revision in self.revisions:
            if revision.revision == revision_id:
                return revision
        return None

    def path_to(self, revision_id: str | None) -> list[Revision]:
        """Revisions from the base up to and including *revision_id*.

        ``None`` means the whole chain.
        """
        if revision_id is None:
            return list(self.revisions)
        path: list[Revision] = []
        for revision in self.revisions:
            path.append(revision)
            if revision.revision == revision_id:
                return path
        raise RevisionError(f"unknown revision '{revision_id}'")

    def extend(self, revision: Revision) -> RevisionChain:
        """A new chain with *revision* appended; validation re-runs on the copy."""
        return RevisionChain(revisions=[*self.revisions, revision])

    @property
    def reversible(self) -> bool:
        """Whether every revision in the chain can be inverted."""
        return all(revision.reversible for revision in self.revisions)

reversible property

Whether every revision in the chain can be inverted.

extend(revision)

A new chain with revision appended; validation re-runs on the copy.

Source code in graflo/architecture/evolution/revision.py 
214
215
216
def extend(self, revision: Revision) -> RevisionChain:
    """A new chain with *revision* appended; validation re-runs on the copy."""
    return RevisionChain(revisions=[*self.revisions, revision])

get(revision_id)

The revision with revision_id, or None.

Source code in graflo/architecture/evolution/revision.py 
193
194
195
196
197
198
def get(self, revision_id: str) -> Revision | None:
    """The revision with *revision_id*, or ``None``."""
    for revision in self.revisions:
        if revision.revision == revision_id:
            return revision
    return None

head()

The most recent revision, or None for an empty chain.

Source code in graflo/architecture/evolution/revision.py 
189
190
191
def head(self) -> Revision | None:
    """The most recent revision, or ``None`` for an empty chain."""
    return self.revisions[-1] if self.revisions else None

path_to(revision_id)

Revisions from the base up to and including revision_id.

None means the whole chain.

Source code in graflo/architecture/evolution/revision.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
def path_to(self, revision_id: str | None) -> list[Revision]:
    """Revisions from the base up to and including *revision_id*.

    ``None`` means the whole chain.
    """
    if revision_id is None:
        return list(self.revisions)
    path: list[Revision] = []
    for revision in self.revisions:
        path.append(revision)
        if revision.revision == revision_id:
            return path
    raise RevisionError(f"unknown revision '{revision_id}'")

RevisionError

Bases: RuntimeError

A chain is malformed, or a replay did not reproduce a recorded hash.

Source code in graflo/architecture/evolution/revision.py 
49
50
class RevisionError(RuntimeError):
    """A chain is malformed, or a replay did not reproduce a recorded hash."""

apply_revisions(base, chain, *, upto=None, verify=True, finish_init=False)

Replay chain onto base, verifying each recorded hash on the way.

Verification is the point: a chain that no longer describes the manifest it was generated from fails here, loudly, instead of producing a plausible but wrong result.

Source code in graflo/architecture/evolution/revision.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def apply_revisions(
    base: GraphManifest,
    chain: RevisionChain,
    *,
    upto: str | None = None,
    verify: bool = True,
    finish_init: bool = False,
) -> GraphManifest:
    """Replay *chain* onto *base*, verifying each recorded hash on the way.

    Verification is the point: a chain that no longer describes the manifest it
    was generated from fails here, loudly, instead of producing a plausible but
    wrong result.
    """
    from .apply import apply_evolution

    current = base
    for revision in chain.path_to(upto):
        if verify:
            actual = manifest_hash(current)
            if actual != revision.manifest_hash_before:
                raise RevisionError(
                    f"revision '{revision.revision}' expects to start from hash "
                    f"{revision.manifest_hash_before[:12]} but the manifest hashes "
                    f"{actual[:12]}; the base or an earlier revision has drifted"
                )
        current = apply_evolution(
            current, revision.ops, bump_version=False, finish_init=finish_init
        )
        if verify:
            produced = manifest_hash(current)
            if produced != revision.manifest_hash_after:
                raise RevisionError(
                    f"replaying revision '{revision.revision}' produced hash "
                    f"{produced[:12]}, not the recorded {revision.manifest_hash_after[:12]}"
                )
    return current

build_revision(base, ops, *, down_revision=None, label=None, created_at=None, notes=None)

Apply ops to base and record the result as a :class:Revision.

The ops are applied here rather than trusted, so both hashes describe a transition that actually happened.

Source code in graflo/architecture/evolution/revision.py 
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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def build_revision(
    base: GraphManifest,
    ops: list[ManifestOp],
    *,
    down_revision: str | None = None,
    label: str | None = None,
    created_at: str | None = None,
    notes: str | None = None,
) -> Revision:
    """Apply *ops* to *base* and record the result as a :class:`Revision`.

    The ops are applied here rather than trusted, so both hashes describe a
    transition that actually happened.
    """
    from .apply import apply_evolution

    if not ops:
        raise RevisionError("a revision needs at least one operation")

    before = manifest_hash(base)
    applied = apply_evolution(base, ops, bump_version=False, finish_init=False)
    after = manifest_hash(applied)
    if after == before:
        raise RevisionError(
            "the operations leave the manifest unchanged; there is nothing to record"
        )

    return Revision(
        revision=compute_revision_id(ops, down_revision),
        down_revision=down_revision,
        label=label,
        created_at=created_at,
        # Round-tripped through the codec on the way in, so a revision can only
        # ever hold ops that survive serialization — the property replay
        # depends on. Also rejects the binary compose op, which no single
        # manifest transition can express.
        ops=ops_from_dicts(ops_to_dicts(list(ops))),
        manifest_hash_before=before,
        manifest_hash_after=after,
        reversible=all(irreversible_reason(op) is None for op in ops),
        notes=notes,
    )

compute_revision_id(ops, down_revision)

Content-derived id: sha256 over the canonical ops plus the parent id.

Deterministic on purpose — the same change set on the same parent always yields the same id, so a re-generated revision is recognisably the same one rather than a duplicate with a fresh random name.

Source code in graflo/architecture/evolution/revision.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def compute_revision_id(ops: list[ManifestOp], down_revision: str | None) -> str:
    """Content-derived id: sha256 over the canonical ops plus the parent id.

    Deterministic on purpose — the same change set on the same parent always
    yields the same id, so a re-generated revision is recognisably the same one
    rather than a duplicate with a fresh random name.
    """
    payload = json.dumps(
        {"ops": ops_to_dicts(list(ops)), "down_revision": down_revision},
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:REVISION_ID_LENGTH]

downgrade_to(chain, target_revision, *, base=None, current=None)

The manifest as of target_revision.

target_revision=None means the state before the first revision — the base. (Deliberately the opposite of :func:apply_revisions's upto=None, which applies the whole chain; here you are naming where to stop going back.)

Prefers replaying from base, which is correct for every chain. Falls back to applying inverses to current only when no base is available, and raises when any op on the path back is irreversible rather than returning a manifest that merely resembles the original.

Source code in graflo/architecture/evolution/revision.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
def downgrade_to(
    chain: RevisionChain,
    target_revision: str | None,
    *,
    base: GraphManifest | None = None,
    current: GraphManifest | None = None,
) -> GraphManifest:
    """The manifest as of *target_revision*.

    ``target_revision=None`` means the state *before* the first revision — the
    base. (Deliberately the opposite of :func:`apply_revisions`'s ``upto=None``,
    which applies the whole chain; here you are naming where to stop going back.)

    Prefers replaying from *base*, which is correct for every chain. Falls back
    to applying inverses to *current* only when no base is available, and raises
    when any op on the path back is irreversible rather than returning a
    manifest that merely resembles the original.
    """
    if base is not None:
        # ``target_revision=None`` means "the state before any revision" — the
        # base itself. Note this is the opposite of ``apply_revisions(upto=None)``,
        # which means "apply the whole chain".
        if target_revision is None:
            return base
        return apply_revisions(base, chain, upto=target_revision)

    if current is None:
        raise RevisionError(
            "downgrade needs either the base manifest (preferred, always exact) "
            "or the current one to invert from"
        )

    undo = _revisions_after(chain, target_revision)
    if not undo:
        return current

    blockers = [
        f"{revision.revision}: {op.op} ({irreversible_reason(op)})"
        for revision in undo
        for op in revision.ops
        if irreversible_reason(op) is not None
    ]
    if blockers:
        raise RevisionError(
            "cannot downgrade by inversion — these operations are irreversible: "
            + "; ".join(blockers)
            + ". Replay from the base manifest instead."
        )

    from .apply import apply_evolution

    manifest = current
    for revision in reversed(undo):
        for op in reversed(list(revision.ops)):
            # Walking backwards, the running manifest is the state *after* this
            # op — which is all that exists without a base. An inverse needing
            # the pre-state (restoring a removed vertex, say) cannot find its
            # data there and returns None, which is reported rather than
            # papered over.
            inverse = invert_op(op, manifest=manifest)
            if inverse is None:
                raise RevisionError(
                    f"revision '{revision.revision}': operation '{op.op}' cannot "
                    "be inverted from the current manifest alone — the data it "
                    "would restore is no longer present. Replay from the base "
                    "manifest instead."
                )
            manifest = apply_evolution(
                manifest, [inverse], bump_version=False, finish_init=False
            )
        if manifest_hash(manifest) != revision.manifest_hash_before:
            raise RevisionError(
                f"inverting revision '{revision.revision}' produced hash "
                f"{manifest_hash(manifest)[:12]}, not the recorded "
                f"{revision.manifest_hash_before[:12]}; the inversion is not exact"
            )
    return manifest