graflo.architecture.evolution.merge3

Three-way merge over manifest change sets, and tracked re-merges.

Merging two world models is not diffing them. Both sides descend from a common ancestor, so the question is never "what is different" but "what did each side change, and do those changes collide". That is the three-way shape: diff base→left, diff base→right, and reconcile.

Slots

Reconciliation happens per slot -- the addressable location an op touches, such as ("vertex", "person", "field", "age"). Two sides that touch disjoint slots, and do not change what the other depends on, merge automatically. Two sides that make the same change to one slot merge to that change, once. Two sides that make different changes to one slot are a :class:MergeConflict, reported rather than guessed at.

Three things make the slot the right unit:

• An order-significant sequence is one slot. A resource pipeline is an ordered program, and half-merging two edits to a program produces something neither author wrote. It conflicts as a unit or it merges as a unit.
• A rename occupies both names. Renaming person → customer on one side while the other side adds a field to person is a genuine collision, and it is invisible unless the rename is understood to touch the old slot too.
• An op touching several slots is atomic: if any one of its slots is contested, the whole op is held back. Applying half an op is not a merge.
• An op reads as well as writes. An edge added onto company writes the edge and depends on company; the other side removing company writes a different slot and still cannot be merged with it (:func:op_reads). A read is disturbed by a write at or above it, never by one beneath: the edge does not care which fields company carries.

A side whose change no operation expresses cannot be merged at all, because the merge is assembled from each side's ops: :func:merge_three_way raises rather than return a clean result that silently lacks it.

Three-way merge is not merge

Three-way merge (this module, commit kind merge3) reconciles two descendants of a common ancestor: names are expected to agree because both sides inherited them, so disagreement is a conflict. Merge (merge_manifests, commit kind merge) joins unrelated lineages by declared equivalence: names are expected to disagree, and the declaration is what reconciles them. Both produce multi-parent commits; they are not the same operation and must not be conflated.

Determinism is a contract

The same inputs produce the same merged manifest, the same conflicts in the same order, and -- through canonical hashing -- the same content hash. Auto-merged ops are applied left-side-first in their diff order, then right-side. Nothing here consults a set iteration order or a dict insertion order.

Attributes

Slot = tuple[str, ...] module-attribute

_ = op_to_dict module-attribute

__all__ = ['ConflictResolution', 'MergeConflict', 'MergeError', 'MergeRecipe', 'MergeResult', 'Slot', 'build_merge_recipe', 'build_recipe', 'describe_slot', 'find_merge_base', 'merge_three_way', 'op_slots', 're_merge', 'take_left', 'take_right'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

ConflictResolution

Bases: ConfigBaseModel

The decision for one contested slot.

Take-left and take-right are not special cases: they are this model holding the corresponding side's op list. A hand-written third answer is expressed the same way, which is what lets a recorded resolution replay on a re-merge.

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

class ConflictResolution(ConfigBaseModel):
    """The decision for one contested slot.

    Take-left and take-right are not special cases: they are this model holding
    the corresponding side's op list. A hand-written third answer is expressed
    the same way, which is what lets a recorded resolution replay on a re-merge.
    """

    slot: list[str] = PydanticField(..., description="The slot being resolved.")
    ops: list[RevisionOp] = PydanticField(
        default_factory=list,
        description="Ops to apply for this slot; empty means 'neither side'.",
    )
    rationale: str | None = PydanticField(
        default=None, description="Why, for the next reader."
    )

    @property
    def slot_key(self) -> Slot:
        return tuple(self.slot)

Attributes

ops = PydanticField(default_factory=list, description="Ops to apply for this slot; empty means 'neither side'.") class-attribute instance-attribute

rationale = PydanticField(default=None, description='Why, for the next reader.') class-attribute instance-attribute

slot = PydanticField(..., description='The slot being resolved.') class-attribute instance-attribute

slot_key property

MergeConflict

Bases: ConfigBaseModel

One slot both sides changed, differently.

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

class MergeConflict(ConfigBaseModel):
    """One slot both sides changed, differently."""

    slot: list[str] = PydanticField(
        ..., description="The contested location, as path segments."
    )
    left_ops: list[RevisionOp] = PydanticField(
        default_factory=list, description="What the left side did here."
    )
    right_ops: list[RevisionOp] = PydanticField(
        default_factory=list, description="What the right side did here."
    )
    base_excerpt: dict[str, Any] = PydanticField(
        default_factory=dict,
        description="The ancestor's state at this slot, for a human deciding.",
    )
    reason: str = PydanticField(
        default="both sides changed this slot",
        description="Why this could not be merged automatically.",
    )

    @property
    def slot_key(self) -> Slot:
        return tuple(self.slot)

Attributes

base_excerpt = PydanticField(default_factory=dict, description="The ancestor's state at this slot, for a human deciding.") class-attribute instance-attribute

left_ops = PydanticField(default_factory=list, description='What the left side did here.') class-attribute instance-attribute

reason = PydanticField(default='both sides changed this slot', description='Why this could not be merged automatically.') class-attribute instance-attribute

right_ops = PydanticField(default_factory=list, description='What the right side did here.') class-attribute instance-attribute

slot = PydanticField(..., description='The contested location, as path segments.') class-attribute instance-attribute

slot_key property

MergeError

Bases: RuntimeError

A merge cannot proceed: unmergeable inputs, or an invalid resolution.

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

class MergeError(RuntimeError):
    """A merge cannot proceed: unmergeable inputs, or an invalid resolution."""

MergeRecipe

Bases: ConfigBaseModel

How a merge was performed, recorded so it can be performed again.

This is the rerere analogue: when the left side advances and the same merge is run again, the recorded resolutions are re-applied to any slot that conflicts again, and only genuinely new conflicts reach a human. That is what makes a tracked merge cheap enough to keep re-running, which is what makes an overlay maintainable rather than a one-time fork.

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

class MergeRecipe(ConfigBaseModel):
    """How a merge was performed, recorded so it can be performed again.

    This is the ``rerere`` analogue: when the left side advances and the same
    merge is run again, the recorded resolutions are re-applied to any slot that
    conflicts *again*, and only genuinely new conflicts reach a human. That is
    what makes a tracked merge cheap enough to keep re-running, which is what
    makes an overlay maintainable rather than a one-time fork.
    """

    kind: str = PydanticField(
        default="merge3", description="merge3 (common ancestor) or merge (unrelated)."
    )
    left: str = PydanticField(..., description="Content hash of the left state.")
    right: str = PydanticField(..., description="Content hash of the right state.")
    base: str | None = PydanticField(
        default=None, description="Content hash of the merge base; merge3 only."
    )
    resolutions: list[ConflictResolution] = PydanticField(
        default_factory=list, description="Slot-keyed decisions, replayable."
    )
    equivalences: dict[str, Any] = PydanticField(
        default_factory=dict,
        description="Declared alignment for a merge; empty for merge3.",
    )
    name_conflict: str | None = PydanticField(
        default=None, description="Merge's name-conflict policy, when applicable."
    )

    def content_hash(self) -> str:
        """Content address of this recipe.

        Resolutions are hashed in slot order, not in the order a human happened
        to supply them, so the same decisions always address the same recipe.
        """
        payload = {
            "kind": self.kind,
            "left": self.left,
            "right": self.right,
            "base": self.base,
            "equivalences": self.equivalences,
            "name_conflict": self.name_conflict,
            "resolutions": sorted(
                (
                    {
                        "slot": list(resolution.slot),
                        "ops": ops_to_dicts(list(resolution.ops)),
                    }
                    for resolution in self.resolutions
                ),
                key=lambda entry: json.dumps(entry["slot"]),
            ),
        }
        return suthing.stable_hash(payload)

Attributes

base = PydanticField(default=None, description='Content hash of the merge base; merge3 only.') class-attribute instance-attribute

equivalences = PydanticField(default_factory=dict, description='Declared alignment for a merge; empty for merge3.') class-attribute instance-attribute

kind = PydanticField(default='merge3', description='merge3 (common ancestor) or merge (unrelated).') class-attribute instance-attribute

left = PydanticField(..., description='Content hash of the left state.') class-attribute instance-attribute

name_conflict = PydanticField(default=None, description="Merge's name-conflict policy, when applicable.") class-attribute instance-attribute

resolutions = PydanticField(default_factory=list, description='Slot-keyed decisions, replayable.') class-attribute instance-attribute

right = PydanticField(..., description='Content hash of the right state.') class-attribute instance-attribute

Methods:

content_hash()

Content address of this recipe.

Resolutions are hashed in slot order, not in the order a human happened to supply them, so the same decisions always address the same recipe.

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

def content_hash(self) -> str:
    """Content address of this recipe.

    Resolutions are hashed in slot order, not in the order a human happened
    to supply them, so the same decisions always address the same recipe.
    """
    payload = {
        "kind": self.kind,
        "left": self.left,
        "right": self.right,
        "base": self.base,
        "equivalences": self.equivalences,
        "name_conflict": self.name_conflict,
        "resolutions": sorted(
            (
                {
                    "slot": list(resolution.slot),
                    "ops": ops_to_dicts(list(resolution.ops)),
                }
                for resolution in self.resolutions
            ),
            key=lambda entry: json.dumps(entry["slot"]),
        ),
    }
    return suthing.stable_hash(payload)

MergeResult

Bases: ConfigBaseModel

What a merge produced, or could not.

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

class MergeResult(ConfigBaseModel):
    """What a merge produced, or could not."""

    ops: list[RevisionOp] = PydanticField(
        default_factory=list,
        description="Ops applied to the base to reach the merged manifest.",
    )
    conflicts: list[MergeConflict] = PydanticField(
        default_factory=list, description="Slots needing a decision, slot-sorted."
    )
    warnings: list[str] = PydanticField(default_factory=list)
    merged_hash: str | None = PydanticField(
        default=None, description="Content hash of the merged manifest, if one exists."
    )

    @property
    def clean(self) -> bool:
        """Whether the merge completed with no decisions left to make."""
        return not self.conflicts

Attributes

clean property

Whether the merge completed with no decisions left to make.

conflicts = PydanticField(default_factory=list, description='Slots needing a decision, slot-sorted.') class-attribute instance-attribute

merged_hash = PydanticField(default=None, description='Content hash of the merged manifest, if one exists.') class-attribute instance-attribute

ops = PydanticField(default_factory=list, description='Ops applied to the base to reach the merged manifest.') class-attribute instance-attribute

warnings = PydanticField(default_factory=list) class-attribute instance-attribute

Functions:

build_merge_recipe(left, right, op)

Record how a merge was declared, addressed by content.

The merge counterpart to :func:build_recipe. Two things differ, and both follow from merge joining unrelated lineages rather than reconciling related ones: there is no merge base, so base is None; and there are no conflicts to resolve, because merge refuses rather than resolving, so resolutions stays empty.

What takes their place is the declaration itself. The whole op is recorded -- equivalences, canonical maps, identity alignments, resource renames and the name-conflict policy -- because all of it is "how these two were joined", and a re-merge that had only the equivalences would reconstruct a different manifest.

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

def build_merge_recipe(
    left: GraphManifest,
    right: GraphManifest,
    op: ops.MergeManifestsOp,
) -> MergeRecipe:
    """Record how a merge was *declared*, addressed by content.

    The merge counterpart to :func:`build_recipe`. Two things differ, and both
    follow from merge joining unrelated lineages rather than reconciling
    related ones: there is no merge base, so ``base`` is ``None``; and there are
    no conflicts to resolve, because merge refuses rather than resolving, so
    ``resolutions`` stays empty.

    What takes their place is the declaration itself. The whole op is recorded
    -- equivalences, canonical maps, identity alignments, resource renames and
    the name-conflict policy -- because all of it is "how these two were
    joined", and a re-merge that had only the equivalences would reconstruct a
    different manifest.
    """
    return MergeRecipe(
        kind="merge",
        left=manifest_hash(left),
        right=manifest_hash(right),
        base=None,
        resolutions=[],
        equivalences=op.to_dict(),
        name_conflict=op.name_conflict,
    )

build_recipe(base, left, right, *, resolutions=None, kind='merge3')

Record how this merge was resolved, addressed by content.

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

def build_recipe(
    base: GraphManifest | None,
    left: GraphManifest,
    right: GraphManifest,
    *,
    resolutions: list[ConflictResolution] | None = None,
    kind: str = "merge3",
) -> MergeRecipe:
    """Record how this merge was resolved, addressed by content."""
    return MergeRecipe(
        kind=kind,
        left=manifest_hash(left),
        right=manifest_hash(right),
        base=manifest_hash(base) if base is not None else None,
        resolutions=list(resolutions or []),
    )

describe_slot(slot)

A slot as a human reads it: vertex/person/field/age.

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

def describe_slot(slot: Slot) -> str:
    """A slot as a human reads it: ``vertex/person/field/age``."""
    return "/".join(slot)

find_merge_base(history, left, right)

The best common ancestor of left and right, or None.

"Best" is the common ancestor furthest from the roots, ties broken on commit id so the choice is deterministic. Multiple genuinely-incomparable bases (a criss-cross history) are picked between with a warning rather than handled properly: recursive merge is a known upgrade path and is explicitly out of scope here, because real criss-cross histories do not arise until people are merging merges routinely.

Parameters:

Name	Type	Description	Default
history	Any	A History (kept structural to avoid an import cycle).	required
left	str	One commit id.	required
right	str	The other commit id.	required

Returns:

Type	Description
str | None	The merge-base commit id, or None when the two share no ancestor --
str | None	which means they are unrelated lineages, and the operation you want is
str | None	merge, not merge.

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

def find_merge_base(history: Any, left: str, right: str) -> str | None:
    """The best common ancestor of *left* and *right*, or ``None``.

    "Best" is the common ancestor furthest from the roots, ties broken on
    commit id so the choice is deterministic. Multiple genuinely-incomparable
    bases (a criss-cross history) are picked between with a warning rather than
    handled properly: recursive merge is a known upgrade path and is explicitly
    out of scope here, because real criss-cross histories do not arise until
    people are merging merges routinely.

    Args:
        history: A ``History`` (kept structural to avoid an import cycle).
        left: One commit id.
        right: The other commit id.

    Returns:
        The merge-base commit id, or ``None`` when the two share no ancestor --
        which means they are unrelated lineages, and the operation you want is
        merge, not merge.
    """
    left_ancestors = history.ancestors(left, include_self=True)
    right_ancestors = history.ancestors(right, include_self=True)
    common = left_ancestors & right_ancestors
    if not common:
        return None

    generation = _generations(history)
    candidates = sorted(common, key=lambda cid: (-generation.get(cid, 0), cid))
    best = candidates[0]

    # A candidate that is an ancestor of the chosen one is subsumed, not rival.
    rivals = [
        cid
        for cid in candidates[1:]
        if generation.get(cid, 0) == generation.get(best, 0)
    ]
    if rivals:
        logger.warning(
            "multiple merge bases for %s and %s (%s); picking %s deterministically. "
            "Recursive merge is not implemented",
            left[:8],
            right[:8],
            ", ".join(cid[:8] for cid in [best, *rivals]),
            best[:8],
        )
    return best

merge_three_way(base, left, right, *, resolutions=None, hints=None)

Reconcile left and right, both descended from base.

Parameters:

Name	Type	Description	Default
base	GraphManifest	The common ancestor.	required
left	GraphManifest	One descendant. Its ops are applied first.	required
right	GraphManifest	The other descendant.	required
resolutions	list[ConflictResolution] | None	Decisions for contested slots. A slot with a resolution is no longer a conflict; its ops replace both sides' at that slot.	None
hints	RenameHints | None	Rename hints for the two diffs -- renames are never inferred (a drop plus an add is not a rename), so a rename on either side needs a hint to be seen as one.	None

Returns:

Type	Description
GraphManifest | None	(merged_manifest_or_None, result). The manifest is None exactly
MergeResult	when unresolved conflicts remain.

Raises:

Type	Description
MergeError	A side changed something no operation expresses -- a property gained by one of a relation's edges and not its siblings, an edited pipeline -- so the merged manifest would silently lack it. Also when the merged change set does not apply to base.

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

def merge_three_way(
    base: GraphManifest,
    left: GraphManifest,
    right: GraphManifest,
    *,
    resolutions: list[ConflictResolution] | None = None,
    hints: RenameHints | None = None,
) -> tuple[GraphManifest | None, MergeResult]:
    """Reconcile *left* and *right*, both descended from *base*.

    Args:
        base: The common ancestor.
        left: One descendant. Its ops are applied first.
        right: The other descendant.
        resolutions: Decisions for contested slots. A slot with a resolution is
            no longer a conflict; its ops replace both sides' at that slot.
        hints: Rename hints for the two diffs -- renames are never *inferred*
            (a drop plus an add is not a rename), so a rename on either side
            needs a hint to be seen as one.

    Returns:
        ``(merged_manifest_or_None, result)``. The manifest is ``None`` exactly
        when unresolved conflicts remain.

    Raises:
        MergeError: A side changed something no operation expresses -- a
            property gained by one of a relation's edges and not its siblings,
            an edited pipeline -- so the merged manifest would silently lack
            it. Also when the merged change set does not apply to *base*.
    """
    from .apply import apply_evolution

    resolved: dict[Slot, ConflictResolution] = {
        resolution.slot_key: resolution for resolution in (resolutions or [])
    }

    left_ops, left_warnings = diff_manifests_verified(base, left, hints=hints)
    right_ops, right_warnings = diff_manifests_verified(base, right, hints=hints)
    warnings = [f"left: {w}" for w in left_warnings]
    warnings += [f"right: {w}" for w in right_warnings]
    if warnings:
        # The merge is assembled from each side's ops. A change the ops do not
        # carry is a change the merged manifest will not have, and a result
        # that is clean and incomplete is the one outcome a merge may not have.
        raise MergeError(
            "a side changed something no operation expresses, so merging would "
            "drop it: " + "; ".join(warnings)
        )

    left_slots = {slot for op in left_ops for slot in op_slots(op)}
    right_slots = {slot for op in right_ops for slot in op_slots(op)}

    # A slot is contested when both sides reach it and disagree about how.
    contested: set[Slot] = set()
    for slot in sorted(_contact_points(left_slots, right_slots)):
        if _canonical_ops(_ops_touching(left_ops, slot)) != _canonical_ops(
            _ops_touching(right_ops, slot)
        ):
            contested.add(slot)

    # A slot is contested, too, when one side changes what the other side's
    # change depends on: an edge added onto a vertex the other side removed.
    # The two ops write different slots, so nothing above sees them meet.
    depends: dict[str, set[Slot]] = _dependencies(left_ops, right_ops, base)
    for key, slots in _dependencies(right_ops, left_ops, base).items():
        depends.setdefault(key, set()).update(slots)
    depended_on = {slot for slots in depends.values() for slot in slots}
    written_by_both = set(contested)
    contested |= depended_on

    def reaches(op: ManifestOp, slot: Slot) -> bool:
        """Whether *op* writes at *slot* or depends on what is written there."""
        return _touches(op_slots(op), slot) or slot in depends.get(
            _canonical_ops([op]), ()
        )

    def reason_for(slot: Slot) -> str:
        if slot == ("manifest",):
            return "a whole-manifest op cannot be merged with another change"
        if slot in written_by_both:
            return "both sides changed this slot differently"
        return "one side changed this slot and the other side's change depends on it"

    unresolved = sorted(contested - set(resolved))
    conflicts = [
        MergeConflict(
            slot=list(slot),
            left_ops=ops_from_dicts(
                ops_to_dicts([op for op in left_ops if reaches(op, slot)])
            ),
            right_ops=ops_from_dicts(
                ops_to_dicts([op for op in right_ops if reaches(op, slot)])
            ),
            base_excerpt=_base_excerpt(base, slot),
            reason=reason_for(slot),
        )
        for slot in unresolved
    ]

    # An op is held back if *any* slot it reaches is contested: applying half an
    # op is not a merge, and neither is applying an op whose ground moved.
    def blocked(op: ManifestOp) -> bool:
        return any(reaches(op, slot) for slot in contested)

    # Assemble in diff order, with each resolution taking the *place* of the
    # ops it replaces rather than being appended at the end.
    #
    # Order is a precondition, not a presentation detail: `diff_manifests`
    # emits renames before additions before identity changes before removals,
    # so that each op's preconditions hold when it runs. Appending resolutions
    # last inverts that -- a re-key resolution would land *after* the demoted
    # key was added as a secondary identity, and the apply fails on a duplicate
    # that only exists because of the reordering.
    merged_ops: list[ManifestOp] = []
    applied_canonical: set[str] = set()
    placed_slots: set[Slot] = set()

    def emit(op: ManifestOp) -> None:
        # Both sides making the identical change is agreement, not duplication.
        canonical = _canonical_ops([op])
        if canonical in applied_canonical:
            return
        applied_canonical.add(canonical)
        merged_ops.append(op)

    def place_resolutions_for(op: ManifestOp) -> None:
        """Emit the decisions for whichever contested slots *op* reaches."""
        for contested_slot in sorted(contested):
            if contested_slot in placed_slots:
                continue
            if not reaches(op, contested_slot):
                continue
            placed_slots.add(contested_slot)
            resolution = resolved.get(contested_slot)
            if resolution is not None:
                for resolution_op in resolution.ops:
                    emit(resolution_op)

    for op in [*left_ops, *right_ops]:
        if blocked(op):
            place_resolutions_for(op)
        else:
            emit(op)

    # A decision for a slot no op reached (both sides' ops were deduplicated
    # away, say) still belongs in the change set.
    for slot in sorted(contested - placed_slots):
        resolution = resolved.get(slot)
        if resolution is not None:
            for resolution_op in resolution.ops:
                emit(resolution_op)
    if conflicts:
        return None, MergeResult(ops=[], conflicts=conflicts, warnings=warnings)

    if not merged_ops:
        # Both sides are the base, or their changes cancelled out.
        return base, MergeResult(
            ops=[], conflicts=[], warnings=warnings, merged_hash=manifest_hash(base)
        )

    try:
        merged = apply_evolution(
            base, merged_ops, bump_version=False, finish_init=False
        )
    except Exception as exc:
        raise MergeError(
            "the merged change set does not apply to the base: "
            f"{type(exc).__name__}: {exc}"
        ) from exc

    return merged, MergeResult(
        ops=ops_from_dicts(ops_to_dicts(merged_ops)),
        conflicts=[],
        warnings=warnings,
        merged_hash=manifest_hash(merged),
    )

op_reads(op, base=None)

What op needs to be there and does not itself change.

:func:op_slots is what an op writes. That is not enough to tell whether two ops are independent: add_edges writes an edge and reads its endpoint vertices, so a remove_vertices on the other side -- which cascades over that vertex's edges -- shares no written slot with it and still cannot be merged with it. One order drops the new edge without a word; the other does not apply.

A read conflicts with a write at or above it, never below: an edge onto company depends on company existing under that name, not on which fields it carries.

Ops addressed by relation depend on whatever that relation connects, which the op does not say; that is read from base. A relation one side renamed is looked up under the name base knows, so its endpoints are not seen -- the rename itself occupies both names and conflicts with the other side's edits to the relation, which covers the common case.

An op not listed reads nothing beyond what it writes: a property op lives under its vertex's slot, so the containment of written slots already ties it to that vertex.

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

def op_reads(op: ManifestOp, base: GraphManifest | None = None) -> set[Slot]:
    """What *op* needs to be there and does not itself change.

    :func:`op_slots` is what an op writes. That is not enough to tell whether
    two ops are independent: ``add_edges`` writes an edge and *reads* its
    endpoint vertices, so a ``remove_vertices`` on the other side -- which
    cascades over that vertex's edges -- shares no written slot with it and
    still cannot be merged with it. One order drops the new edge without a
    word; the other does not apply.

    A read conflicts with a write **at or above** it, never below: an edge onto
    ``company`` depends on ``company`` existing under that name, not on which
    fields it carries.

    Ops addressed by relation depend on whatever that relation connects, which
    the op does not say; that is read from *base*. A relation one side renamed
    is looked up under the name *base* knows, so its endpoints are not seen --
    the rename itself occupies both names and conflicts with the other side's
    edits to the relation, which covers the common case.

    An op not listed reads nothing beyond what it writes: a property op lives
    under its vertex's slot, so the containment of written slots already ties
    it to that vertex.
    """
    reads: set[Slot] = set()

    # ── edges, addressed by endpoints ───────────────────────────────────────
    if isinstance(op, ops.AddEdgesOp):
        for edge in op.edges:
            reads |= {_vertex_slot(edge.source), _vertex_slot(edge.target)}
            if edge.by is not None:
                reads.add(_vertex_slot(edge.by))
    elif isinstance(op, ops.RetargetEdgesOp):
        for entry in op.edges:
            reads |= {_vertex_slot(entry.source), _vertex_slot(entry.target)}
            reads |= {
                _vertex_slot(name)
                for name in (entry.new_source, entry.new_target)
                if name is not None
            }
    elif isinstance(op, (ops.SetEdgeDirectedOp, ops.SetEdgeSemanticsOp)):
        for selector in op.edges:
            reads |= {_vertex_slot(selector.source), _vertex_slot(selector.target)}
    elif isinstance(op, ops.ReplaceEdgeIdentitiesOp):
        for entry in op.edges:
            reads |= {_vertex_slot(entry.source), _vertex_slot(entry.target)}
            tokens = {token for key in entry.identities for token in key}
            if entry.relation is not None and tokens - _ENDPOINT_TOKENS:
                reads.add((*_relation_slot(entry.relation), "field"))
    elif isinstance(op, (ops.AddEdgeIndexesOp, ops.RemoveEdgeIndexesOp)):
        for entry in op.edges:
            reads |= {_vertex_slot(entry.source), _vertex_slot(entry.target)}
            if entry.relation is not None:
                reads.add((*_relation_slot(entry.relation), "field"))
    elif isinstance(op, ops.SetFieldSemanticsOp):
        for target in op.targets:
            if isinstance(target, ops.EdgeFieldSemanticsTarget):
                reads |= {_vertex_slot(target.source), _vertex_slot(target.target)}

    # ── edges, addressed by relation name ───────────────────────────────────
    elif isinstance(op, ops.RemoveEdgesOp):
        for selector in op.edges:
            reads |= {_vertex_slot(selector.source), _vertex_slot(selector.target)}
        for relation in op.relations:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, ops.AddEdgePropertiesOp):
        for relation in op.additions:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, ops.RemoveEdgePropertiesOp):
        for relation in op.removals:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, (ops.RenameEdgePropertiesOp, ops.RenameRelationsOp)):
        for relation in op.renames:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, ops.MergeEdgesOp):
        for relation in [*op.sources, op.into]:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, (ops.AddInverseEdgesOp, ops.SetNativeInversesOp)):
        for relation in op.relations or []:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, ops.RetractEdgeInversesOp):
        for relation in op.relations:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, ops.DeclareEdgeInversesOp):
        for relation in [*op.inverses, *op.inverses.values(), *op.symmetric]:
            reads |= _relation_endpoints(base, relation)
    elif isinstance(op, ops.ChangeFieldTypesOp):
        for relation in op.edges:
            reads |= _relation_endpoints(base, relation)

    # ── keys ────────────────────────────────────────────────────────────────
    elif isinstance(op, ops.ReplaceIdentityOp):
        for vertex, replacement in op.replacements.items():
            reads |= _key_field_reads(vertex, _identity_target_fields(replacement.to))
    elif isinstance(op, ops.AddSecondaryIdentitiesOp):
        for vertex, entries in op.additions.items():
            for entry in entries:
                reads |= _key_field_reads(vertex, list(entry.fields))
    elif isinstance(op, ops.AddVertexIndexesOp):
        for vertex, indexes in op.indexes.items():
            for index in indexes:
                reads |= {_field_slot(vertex, field) for field in index.fields}

    return reads

op_slots(op)

Every addressable location op touches.

Dispatch is on the op class, not on its op string literal. Two reasons: a type checker can narrow it, so a field read against the wrong op model is caught at check time rather than at merge time; and a literal renamed in ops.py cannot silently fall through to the catch-all here.

Total over the op vocabulary. An op reaching the fallback is treated as touching the whole manifest, which conflicts with everything -- the safe direction, since an unrecognised op silently merging with anything is how a merge quietly corrupts a schema. test_merge3.py asserts every member of the union resolves to a real slot, so the fallback is a backstop rather than a policy.

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

def op_slots(op: ManifestOp) -> set[Slot]:
    """Every addressable location *op* touches.

    Dispatch is on the op *class*, not on its ``op`` string literal. Two
    reasons: a type checker can narrow it, so a field read against the wrong op
    model is caught at check time rather than at merge time; and a literal
    renamed in ``ops.py`` cannot silently fall through to the catch-all here.

    Total over the op vocabulary. An op reaching the fallback is treated as
    touching the whole manifest, which conflicts with everything -- the safe
    direction, since an unrecognised op silently merging with anything is how a
    merge quietly corrupts a schema. ``test_merge3.py`` asserts every member of
    the union resolves to a real slot, so the fallback is a backstop rather
    than a policy.
    """
    slots: set[Slot] = set()

    # ── vertices ────────────────────────────────────────────────────────────
    if isinstance(op, ops.AddVerticesOp):
        slots |= {_vertex_slot(v.name) for v in op.vertices}
    elif isinstance(op, ops.RemoveVerticesOp):
        slots |= {_vertex_slot(name) for name in op.names}
    elif isinstance(op, ops.RenameVerticesOp):
        # Both names: a rename collides with any edit to either side of it.
        for old, new in op.renames.items():
            slots |= {_vertex_slot(old), _vertex_slot(new)}
    elif isinstance(op, ops.MergeVerticesOp):
        slots |= {_vertex_slot(name) for name in op.sources}
        slots.add(_vertex_slot(op.into))
    elif isinstance(op, ops.CanonicalizeOp):
        # Every name on either side of the map is occupied, as for a rename.
        for old, new in op.vertices.items():
            slots |= {_vertex_slot(old), _vertex_slot(new)}
        for vertex, renames in op.properties.items():
            for old, new in renames.items():
                slots |= {_field_slot(vertex, old), _field_slot(vertex, new)}
        for old, new in op.relations.items():
            slots |= {_relation_slot(old), _relation_slot(new)}

    # ── vertex properties ───────────────────────────────────────────────────
    elif isinstance(op, ops.AddVertexPropertiesOp):
        for vertex in op.additions:
            slots |= {_field_slot(vertex, name) for name in op.field_names(vertex)}
    elif isinstance(op, ops.RemoveVertexPropertiesOp):
        for vertex, fields in op.removals.items():
            slots |= {_field_slot(vertex, field) for field in fields}
    elif isinstance(op, ops.RenameVertexPropertiesOp):
        for vertex, renames in op.renames.items():
            for old, new in renames.items():
                slots |= {_field_slot(vertex, old), _field_slot(vertex, new)}

    # ── identity ────────────────────────────────────────────────────────────
    elif isinstance(op, ops.ReplaceIdentityOp):
        # Identity is a property of the vertex as a whole, not of one field, so
        # two sides re-keying the same vertex must collide even when they name
        # entirely different fields.
        slots |= {(*_vertex_slot(vertex), "identity") for vertex in op.replacements}
    elif isinstance(op, ops.AddSecondaryIdentitiesOp):
        slots |= {(*_vertex_slot(vertex), "secondary") for vertex in op.additions}
    elif isinstance(op, ops.RemoveSecondaryIdentitiesOp):
        slots |= {(*_vertex_slot(vertex), "secondary") for vertex in op.removals}

    # ── edges, addressed by endpoints ───────────────────────────────────────
    elif isinstance(
        op,
        (
            ops.AddEdgesOp,
            ops.RetargetEdgesOp,
            ops.ReplaceEdgeIdentitiesOp,
            ops.AddEdgeIndexesOp,
            ops.RemoveEdgeIndexesOp,
        ),
    ):
        # `Edge.edge_id` is a property and the entry models' is a method, so
        # read the triple itself, which every edge-ish model carries.
        slots |= {
            _edge_slot(entry.source, entry.target, entry.relation) for entry in op.edges
        }
    elif isinstance(op, ops.SetEdgeDirectedOp):
        slots |= {(*_edge_slot(*entry.edge_id()), "directed") for entry in op.edges}

    # Block-level setters. One slot for the whole block, because that is what
    # the op replaces -- two independent bindings edits therefore conflict,
    # which is the price of the block having any op at all.
    elif isinstance(op, ops.SetBindingsOp):
        slots |= {("bindings",)}
    elif isinstance(op, ops.SetDbProfileOp):
        slots |= {("db_profile",)}
    elif isinstance(op, ops.SetEdgeSemanticsOp):
        slots |= {(*_edge_slot(*entry.edge_id()), "semantics") for entry in op.edges}

    # ── grounding ───────────────────────────────────────────────────────────
    #
    # A narrower slot than the element itself: grounding a type and renaming a
    # property of it are independent edits, and merging them is the ordinary
    # case rather than a conflict.
    elif isinstance(op, ops.SetVertexSemanticsOp):
        slots |= {(*_vertex_slot(name), "semantics") for name in op.semantics}
    elif isinstance(op, ops.SetVertexDescriptionsOp):
        slots |= {(*_vertex_slot(name), "description") for name in op.descriptions}
    elif isinstance(op, ops.SetFieldSemanticsOp):
        for target in op.targets:
            if isinstance(target, ops.FieldSemanticsTarget):
                slots.add((*_field_slot(target.vertex, target.field), "semantics"))
            else:
                slots.add(
                    (*_edge_slot(*target.edge_id()), "field", target.field, "semantics")
                )

    # ── edges, addressed by relation name ───────────────────────────────────
    elif isinstance(op, ops.RemoveEdgesOp):
        slots |= {_relation_slot(relation) for relation in op.relations}
        slots |= {_edge_slot(*selector.edge_id()) for selector in op.edges}
    elif isinstance(op, ops.RenameRelationsOp):
        # `{old: new}` -- both names are occupied.
        for old, new in op.renames.items():
            slots |= {_relation_slot(old), _relation_slot(new)}
    elif isinstance(op, ops.AddInverseEdgesOp):
        if op.relations is None:
            # Realizes whatever is declared at replay time, so it depends on the
            # whole table.
            slots.add(("edge_inverses",))
        else:
            slots |= {_relation_slot(relation) for relation in op.relations}
    elif isinstance(op, ops.DeclareEdgeInversesOp):
        for relation, inverse in op.inverses.items():
            slots |= {
                (*_relation_slot(relation), "inverse"),
                (*_relation_slot(inverse), "inverse"),
            }
        slots |= {(*_relation_slot(name), "inverse") for name in op.symmetric}
    elif isinstance(op, ops.RetractEdgeInversesOp):
        slots |= {(*_relation_slot(relation), "inverse") for relation in op.relations}
    elif isinstance(op, ops.SetNativeInversesOp):
        slots |= {
            (*_relation_slot(relation), "native_inverse") for relation in op.relations
        }
    elif isinstance(op, ops.MergeEdgesOp):
        slots |= {_relation_slot(relation) for relation in op.sources}
        slots.add(_relation_slot(op.into))

    # ── edge properties ─────────────────────────────────────────────────────
    elif isinstance(op, ops.AddEdgePropertiesOp):
        slots |= {(*_relation_slot(rel), "field") for rel in op.additions}
    elif isinstance(op, ops.RemoveEdgePropertiesOp):
        slots |= {(*_relation_slot(rel), "field") for rel in op.removals}
    elif isinstance(op, ops.RenameEdgePropertiesOp):
        slots |= {(*_relation_slot(rel), "field") for rel in op.renames}

    # ── vertex indexes ──────────────────────────────────────────────────────
    elif isinstance(op, (ops.AddVertexIndexesOp, ops.RemoveVertexIndexesOp)):
        slots |= {(*_vertex_slot(vertex), "index") for vertex in op.indexes}

    # ── types ───────────────────────────────────────────────────────────────
    elif isinstance(op, ops.ChangeFieldTypesOp):
        for vertex, fields in op.vertices.items():
            slots |= {(*_field_slot(vertex, field), "type") for field in fields}
        for relation, fields in op.edges.items():
            slots |= {
                (*_relation_slot(relation), "field", str(field), "type")
                for field in fields
            }

    # ── ingestion ───────────────────────────────────────────────────────────
    elif isinstance(op, ops.SetInverseEmissionOp):
        # Steps are addressed by position, and a position is only meaningful
        # against the pipeline it was read from: one slot per resource.
        slots |= {_resource_slot(name) for name in op.steps}
    elif isinstance(op, (ops.AddResourceTransformsOp, ops.EnsureExtractedFieldsOp)):
        # A pipeline is one slot per resource: an ordered program cannot be
        # half-merged, so it conflicts as a unit or merges as a unit.
        slots |= {_resource_slot(name) for name in op.additions}
    elif isinstance(op, ops.RenameResourcesOp):
        for old, new in op.renames.items():
            slots |= {_resource_slot(old), _resource_slot(new)}
    elif isinstance(op, ops.AddResourcesOp):
        slots |= {_resource_slot(resource.name) for resource in op.resources}
    elif isinstance(op, ops.RemoveResourcesOp):
        slots |= {_resource_slot(name) for name in op.names}

    # ── whole-manifest ops ──────────────────────────────────────────────────
    elif isinstance(op, (ops.ProjectManifestOp, ops.SanitizeOp, ops.MergeManifestsOp)):
        # These rewrite everything, so they conflict with any other change.
        # That is the honest answer: there is no way to merge "keep only these
        # vertices" with an unrelated edit and be sure of the result.
        slots.add(("manifest",))

    if not slots:
        logger.warning(
            "op '%s' has no slot mapping; treating it as touching the whole "
            "manifest, so it will conflict with any other change",
            op.op,
        )
        slots.add(("manifest",))
    return slots

ops_independent(one, other, base=None)

Whether one and other can be applied in either order to one effect.

Neither writes where the other writes, and neither writes what the other reads. Two ops reading the same thing are independent: two edges onto one vertex do not get in each other's way.

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

def ops_independent(
    one: ManifestOp, other: ManifestOp, base: GraphManifest | None = None
) -> bool:
    """Whether *one* and *other* can be applied in either order to one effect.

    Neither writes where the other writes, and neither writes what the other
    reads. Two ops reading the same thing are independent: two edges onto one
    vertex do not get in each other's way.
    """
    one_writes, other_writes = op_slots(one), op_slots(other)
    if any(_covers(a, b) or _covers(b, a) for a in one_writes for b in other_writes):
        return False
    return not any(
        _depends_on(read, written)
        for reader, writes in ((one, other_writes), (other, one_writes))
        for read in op_reads(reader, base)
        for written in writes
    )

re_merge(recipe, base, left, right, *, hints=None)

Re-run a recorded merge with its resolutions pre-applied.

The point of a tracked merge: when the left side advances, this replays the decisions already made and surfaces only conflicts that are genuinely new. Recorded resolutions for slots that no longer conflict are simply not needed and are reported as such, rather than being force-applied -- a stale decision reapplied to a slot nobody contested is how a re-merge quietly reverts someone's work.

Parameters:

Name	Type	Description	Default
recipe	MergeRecipe	The recorded merge.	required
base	GraphManifest	The (possibly new) merge base.	required
left	GraphManifest	The (possibly advanced) left state.	required
right	GraphManifest	The right state.	required
hints	RenameHints | None	Rename hints for the underlying diffs.	None

Returns:

Type	Description
tuple[GraphManifest | None, MergeResult]	(merged_or_None, result), as :func:merge_three_way.

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

def re_merge(
    recipe: MergeRecipe,
    base: GraphManifest,
    left: GraphManifest,
    right: GraphManifest,
    *,
    hints: RenameHints | None = None,
) -> tuple[GraphManifest | None, MergeResult]:
    """Re-run a recorded merge with its resolutions pre-applied.

    The point of a *tracked* merge: when the left side advances, this replays
    the decisions already made and surfaces only conflicts that are genuinely
    new. Recorded resolutions for slots that no longer conflict are simply not
    needed and are reported as such, rather than being force-applied -- a stale
    decision reapplied to a slot nobody contested is how a re-merge quietly
    reverts someone's work.

    Args:
        recipe: The recorded merge.
        base: The (possibly new) merge base.
        left: The (possibly advanced) left state.
        right: The right state.
        hints: Rename hints for the underlying diffs.

    Returns:
        ``(merged_or_None, result)``, as :func:`merge_three_way`.
    """
    # What *would* conflict this time, before any recorded decision is applied.
    # Comparing against the conflicts that survive the replay instead would
    # report every resolution that did its job as "not needed" -- precisely
    # backwards, and the reading a maintainer would act on by deleting it.
    _unresolved, dry = merge_three_way(base, left, right, hints=hints)
    contested = {conflict.slot_key for conflict in dry.conflicts}

    merged, result = merge_three_way(
        base, left, right, resolutions=recipe.resolutions, hints=hints
    )

    recorded = {resolution.slot_key for resolution in recipe.resolutions}
    replayed = sorted(recorded & contested)
    if replayed:
        result.warnings.append(
            f"{len(replayed)} recorded resolution(s) replayed: "
            + ", ".join(describe_slot(slot) for slot in replayed)
        )
    unused = sorted(recorded - contested)
    if unused:
        result.warnings.append(
            f"{len(unused)} recorded resolution(s) were not needed this time: "
            + ", ".join(describe_slot(slot) for slot in unused)
        )
    genuinely_new = sorted(contested - recorded)
    if genuinely_new:
        result.warnings.append(
            f"{len(genuinely_new)} conflict(s) are new since the recorded merge: "
            + ", ".join(describe_slot(slot) for slot in genuinely_new)
        )
    return merged, result

take_left(conflict, *, rationale=None)

Resolve conflict by keeping the left side's ops.

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

def take_left(
    conflict: MergeConflict, *, rationale: str | None = None
) -> ConflictResolution:
    """Resolve *conflict* by keeping the left side's ops."""
    return ConflictResolution(
        slot=list(conflict.slot),
        ops=ops_from_dicts(ops_to_dicts(list(conflict.left_ops))),
        rationale=rationale or "took left",
    )

take_right(conflict, *, rationale=None)

Resolve conflict by keeping the right side's ops.

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

def take_right(
    conflict: MergeConflict, *, rationale: str | None = None
) -> ConflictResolution:
    """Resolve *conflict* by keeping the right side's ops."""
    return ConflictResolution(
        slot=list(conflict.slot),
        ops=ops_from_dicts(ops_to_dicts(list(conflict.right_ops))),
        rationale=rationale or "took right",
    )