graflo.architecture.evolution.preview

What a merge would do, and every way it could refuse — without refusing.

:func:~graflo.architecture.evolution.merge.merge_manifests raises at the first refusal. That is right for a function that returns a manifest — a half-merged model is worse than none — but it makes authoring a merge a game of whack-a-mole: fix the contradiction the message names, run again, learn about the next one. Three declarations that each refuse take three runs to discover.

This module is the other view. :func:preview_merge walks the same declarations and reports every problem it finds, as data:

• the declaration graph — classes and their attributes on each side, the clusters that collapse them, the canonical names the maps establish, and the edges between them. This is the class_A - attr_a - attr_b - class_B picture the authoring model is actually about;
• findings, each naming the nodes it is about, at one of three severities — possible (found structurally, by this module), refusal (what merge actually raised, if it was asked to try) and note (an acknowledged heuristic, such as an entry taken as already applied);
• an outcome, from a real merge attempt.

Nothing here is a second implementation of the resolution rules. Every check calls the function in :mod:~graflo.architecture.evolution.canonical that merge itself calls, in units small enough — one declaration, one map entry, one member — that a refusal on one unit does not hide the others. A refusal carries its check and its subjects, so the finding it becomes is pinned to the same nodes the message names, and there is no parallel copy of the rules to fall out of date.

The consistency invariant, asserted in the tests: whatever merge refuses, the structural pass has a finding of a matching kind for, and a merge that succeeds leaves no refusal finding behind.

Merging two branches of one lineage needs none of this. :func:~graflo.architecture.evolution.merge3.merge_three_way already returns its conflicts rather than raising them, one record per contested slot, so :func:build_merge3_preview only has to put them in the shape they already have: slots form a tree — vertex/person contains vertex/person/field/age — and the tree is what shows where in the model two branches collided.

Attributes

EdgeKind = Literal['member', 'map', 'property_map', 'property_equivalence', 'suggested'] module-attribute

FindingKind = Literal['cluster_overlap', 'shared_into', 'occupied_into', 'unknown_member', 'disagreement', 'ambiguity', 'unnamed_cluster', 'incomplete', 'dangling', 'satisfied', 'name_collision', 'near_collision', 'prefixed', 'property_disagreement', 'property_collision', 'property_retarget', 'unknown_property', 'identity_disagreement', 'type_conflict', 'unit_conflict', 'identity_mode_conflict', 'identity_funnel_conflict', 'secondary_identity_conflict', 'edge_conflict'] module-attribute

NodeKind = Literal['class', 'relation', 'attribute', 'merged', 'canonical', 'ghost'] module-attribute

Severity = Literal['refusal', 'possible', 'note'] module-attribute

__all__ = ['EdgeKind', 'FindingKind', 'Merge3Preview', 'MergeFinding', 'MergeOutcome', 'MergePreview', 'NodeKind', 'PreviewCluster', 'PreviewEdge', 'PreviewNode', 'Severity', 'SlotNode', 'build_merge3_preview', 'expected_kinds', 'kind_for_check', 'outcome_from_exception', 'outcome_from_manifest', 'preview_merge'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

Merge3Preview

Bases: ConfigBaseModel

A three-way merge as a slot tree: what moved, and where the branches met.

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

class Merge3Preview(ConfigBaseModel):
    """A three-way merge as a slot tree: what moved, and where the branches met."""

    nodes: list[SlotNode] = PydanticField(default_factory=list)
    conflicts: int = PydanticField(default=0, description="Contested slot count.")
    clean: bool = PydanticField(
        default=True, description="Whether the merge left no decision to make."
    )
    merged_hash: str | None = PydanticField(
        default=None,
        description="Content hash of the merged manifest, if there is one.",
    )
    warnings: list[str] = PydanticField(default_factory=list)

    @property
    def contested(self) -> list[SlotNode]:
        """The contested slots, in tree order."""
        return [node for node in self.nodes if node.contested]

    def children_of(self, node_id: str | None) -> list[SlotNode]:
        """Slots directly under *node_id*; pass ``None`` for the roots."""
        return [node for node in self.nodes if node.parent == node_id]

Attributes

clean = PydanticField(default=True, description='Whether the merge left no decision to make.') class-attribute instance-attribute

conflicts = PydanticField(default=0, description='Contested slot count.') class-attribute instance-attribute

contested property

The contested slots, in tree order.

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

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

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

Methods:

children_of(node_id)

Slots directly under node_id; pass None for the roots.

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

def children_of(self, node_id: str | None) -> list[SlotNode]:
    """Slots directly under *node_id*; pass ``None`` for the roots."""
    return [node for node in self.nodes if node.parent == node_id]

MergeFinding

Bases: ConfigBaseModel

One thing wrong with the declarations, or one acknowledged heuristic.

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

class MergeFinding(ConfigBaseModel):
    """One thing wrong with the declarations, or one acknowledged heuristic."""

    kind: FindingKind = PydanticField(
        ..., description="Which rule it is an instance of."
    )
    severity: Severity = PydanticField(..., description="How much it matters.")
    message: str = PydanticField(..., description="What to tell the author.")
    source: Literal["merge", "structure"] = PydanticField(
        ..., description="Whether merge raised it, or this module found it."
    )
    nodes: list[str] = PydanticField(
        default_factory=list, description="Node ids this finding is about."
    )
    edges: list[str] = PydanticField(
        default_factory=list, description="Edge ids this finding is about."
    )
    completion: dict[str, Any] | None = PydanticField(
        default=None,
        description="The declaration that would settle it, when there is one.",
    )
    check: str | None = PydanticField(
        default=None, description="The refusal's own name for the rule."
    )
    error_type: str | None = PydanticField(
        default=None, description="Exception type, for a finding merge raised."
    )

Attributes

check = PydanticField(default=None, description="The refusal's own name for the rule.") class-attribute instance-attribute

completion = PydanticField(default=None, description='The declaration that would settle it, when there is one.') class-attribute instance-attribute

edges = PydanticField(default_factory=list, description='Edge ids this finding is about.') class-attribute instance-attribute

error_type = PydanticField(default=None, description='Exception type, for a finding merge raised.') class-attribute instance-attribute

kind = PydanticField(..., description='Which rule it is an instance of.') class-attribute instance-attribute

message = PydanticField(..., description='What to tell the author.') class-attribute instance-attribute

nodes = PydanticField(default_factory=list, description='Node ids this finding is about.') class-attribute instance-attribute

severity = PydanticField(..., description='How much it matters.') class-attribute instance-attribute

source = PydanticField(..., description='Whether merge raised it, or this module found it.') class-attribute instance-attribute

MergeOutcome

Bases: ConfigBaseModel

What a real merge attempt produced, or refused with.

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

class MergeOutcome(ConfigBaseModel):
    """What a real merge attempt produced, or refused with."""

    status: Literal["merged", "refused", "not_attempted"] = PydanticField(
        ..., description="Whether merge ran, and how it ended."
    )
    error_type: str | None = PydanticField(default=None, description="Exception type.")
    message: str | None = PydanticField(
        default=None, description="The refusal, in full."
    )
    check: str | None = PydanticField(
        default=None, description="The rule that refused."
    )
    completion: dict[str, Any] | None = PydanticField(
        default=None,
        description="The extension that would settle an incomplete refusal.",
    )
    vertices: int | None = PydanticField(
        default=None, description="Vertex count of the merged schema."
    )
    edges: int | None = PydanticField(
        default=None, description="Edge count of the merged schema."
    )
    version: str | None = PydanticField(
        default=None, description="Version of the merged schema."
    )

Attributes

check = PydanticField(default=None, description='The rule that refused.') class-attribute instance-attribute

completion = PydanticField(default=None, description='The extension that would settle an incomplete refusal.') class-attribute instance-attribute

edges = PydanticField(default=None, description='Edge count of the merged schema.') class-attribute instance-attribute

error_type = PydanticField(default=None, description='Exception type.') class-attribute instance-attribute

message = PydanticField(default=None, description='The refusal, in full.') class-attribute instance-attribute

status = PydanticField(..., description='Whether merge ran, and how it ended.') class-attribute instance-attribute

version = PydanticField(default=None, description='Version of the merged schema.') class-attribute instance-attribute

vertices = PydanticField(default=None, description='Vertex count of the merged schema.') class-attribute instance-attribute

MergePreview

Bases: ConfigBaseModel

The declaration graph, everything wrong with it, and what merge did.

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

class MergePreview(ConfigBaseModel):
    """The declaration graph, everything wrong with it, and what merge did."""

    left_name: str = PydanticField(
        default="left", description="Name of the left manifest."
    )
    right_name: str = PydanticField(
        default="right", description="Name of the right manifest."
    )
    name_conflict: str = PydanticField(
        default="error", description="The op's policy for names no equivalence covers."
    )
    nodes: list[PreviewNode] = PydanticField(default_factory=list)
    edges: list[PreviewEdge] = PydanticField(default_factory=list)
    clusters: list[PreviewCluster] = PydanticField(default_factory=list)
    findings: list[MergeFinding] = PydanticField(default_factory=list)
    outcome: MergeOutcome = PydanticField(
        default_factory=lambda: MergeOutcome(status="not_attempted")
    )

    @property
    def refused(self) -> bool:
        """Whether the merge attempt refused."""
        return self.outcome.status == "refused"

    @property
    def blocking(self) -> list[MergeFinding]:
        """Findings that would stop a merge: the refusal and every possible one."""
        return [f for f in self.findings if f.severity != "note"]

    def node(self, node_id: str) -> PreviewNode | None:
        """The node with *node_id*, or ``None``."""
        return next((n for n in self.nodes if n.id == node_id), None)

    def attributes_of(self, node_id: str) -> list[PreviewNode]:
        """Attribute nodes owned by the class node *node_id*, in declared order."""
        return [n for n in self.nodes if n.owner == node_id]

    def with_outcome(
        self, outcome: MergeOutcome, *, subjects: Sequence[str] = ()
    ) -> MergePreview:
        """A copy carrying *outcome*, and the finding a refusal becomes.

        The structural pass usually found the refusal too — it calls the same
        check — so the two are folded into one finding marked ``refusal``
        rather than listed twice. Reporting one problem as two would undercut
        the only number this is for: how many things are actually wrong.
        """
        if outcome.status != "refused":
            return self.model_copy(update={"outcome": outcome})

        known = {n.id for n in self.nodes}
        refusal = MergeFinding(
            kind=kind_for_check(outcome.check, outcome.error_type),
            severity="refusal",
            message=outcome.message or "merge refused",
            source="merge",
            nodes=[s for s in subjects if s in known],
            completion=outcome.completion,
            check=outcome.check,
            error_type=outcome.error_type,
        )
        findings: list[MergeFinding] = []
        folded = False
        for finding in self.findings:
            if not folded and _is_same_problem(finding, refusal):
                findings.append(
                    refusal.model_copy(
                        update={
                            "nodes": list(
                                dict.fromkeys([*refusal.nodes, *finding.nodes])
                            ),
                            "edges": finding.edges,
                        }
                    )
                )
                folded = True
                continue
            findings.append(finding)
        if not folded:
            findings.append(refusal)
        return self.model_copy(update={"outcome": outcome, "findings": findings})

Attributes

blocking property

Findings that would stop a merge: the refusal and every possible one.

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

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

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

left_name = PydanticField(default='left', description='Name of the left manifest.') class-attribute instance-attribute

name_conflict = PydanticField(default='error', description="The op's policy for names no equivalence covers.") class-attribute instance-attribute

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

outcome = PydanticField(default_factory=lambda: MergeOutcome(status='not_attempted')) class-attribute instance-attribute

refused property

Whether the merge attempt refused.

right_name = PydanticField(default='right', description='Name of the right manifest.') class-attribute instance-attribute

Methods:

attributes_of(node_id)

Attribute nodes owned by the class node node_id, in declared order.

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

def attributes_of(self, node_id: str) -> list[PreviewNode]:
    """Attribute nodes owned by the class node *node_id*, in declared order."""
    return [n for n in self.nodes if n.owner == node_id]

node(node_id)

The node with node_id, or None.

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

def node(self, node_id: str) -> PreviewNode | None:
    """The node with *node_id*, or ``None``."""
    return next((n for n in self.nodes if n.id == node_id), None)

with_outcome(outcome, *, subjects=())

A copy carrying outcome, and the finding a refusal becomes.

The structural pass usually found the refusal too — it calls the same check — so the two are folded into one finding marked refusal rather than listed twice. Reporting one problem as two would undercut the only number this is for: how many things are actually wrong.

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

def with_outcome(
    self, outcome: MergeOutcome, *, subjects: Sequence[str] = ()
) -> MergePreview:
    """A copy carrying *outcome*, and the finding a refusal becomes.

    The structural pass usually found the refusal too — it calls the same
    check — so the two are folded into one finding marked ``refusal``
    rather than listed twice. Reporting one problem as two would undercut
    the only number this is for: how many things are actually wrong.
    """
    if outcome.status != "refused":
        return self.model_copy(update={"outcome": outcome})

    known = {n.id for n in self.nodes}
    refusal = MergeFinding(
        kind=kind_for_check(outcome.check, outcome.error_type),
        severity="refusal",
        message=outcome.message or "merge refused",
        source="merge",
        nodes=[s for s in subjects if s in known],
        completion=outcome.completion,
        check=outcome.check,
        error_type=outcome.error_type,
    )
    findings: list[MergeFinding] = []
    folded = False
    for finding in self.findings:
        if not folded and _is_same_problem(finding, refusal):
            findings.append(
                refusal.model_copy(
                    update={
                        "nodes": list(
                            dict.fromkeys([*refusal.nodes, *finding.nodes])
                        ),
                        "edges": finding.edges,
                    }
                )
            )
            folded = True
            continue
        findings.append(finding)
    if not folded:
        findings.append(refusal)
    return self.model_copy(update={"outcome": outcome, "findings": findings})

PreviewCluster

Bases: ConfigBaseModel

One equivalence declaration, resolved as far as it could be.

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

class PreviewCluster(ConfigBaseModel):
    """One equivalence declaration, resolved as far as it could be."""

    id: str = PydanticField(..., description="Stable id of the cluster.")
    kind: Kind = PydanticField(
        ..., description="Whether it collapses classes or relations."
    )
    into: str | None = PydanticField(
        default=None, description="Merged name; None when it could not be resolved."
    )
    declared_into: str | None = PydanticField(
        default=None, description="`into` as the author spelled it, before translation."
    )
    left: list[str] = PydanticField(default_factory=list, description="Left members.")
    right: list[str] = PydanticField(default_factory=list, description="Right members.")
    synthesized: bool = PydanticField(
        default=False, description="Declared by merge itself under `union_right`."
    )
    declared_identity: bool = PydanticField(
        default=False, description="Whether the declaration states an `identity`."
    )
    aligned: bool = PydanticField(
        default=False, description="Whether an `identity_alignments` entry names it."
    )

Attributes

aligned = PydanticField(default=False, description='Whether an `identity_alignments` entry names it.') class-attribute instance-attribute

declared_identity = PydanticField(default=False, description='Whether the declaration states an `identity`.') class-attribute instance-attribute

declared_into = PydanticField(default=None, description='`into` as the author spelled it, before translation.') class-attribute instance-attribute

id = PydanticField(..., description='Stable id of the cluster.') class-attribute instance-attribute

into = PydanticField(default=None, description='Merged name; None when it could not be resolved.') class-attribute instance-attribute

kind = PydanticField(..., description='Whether it collapses classes or relations.') class-attribute instance-attribute

left = PydanticField(default_factory=list, description='Left members.') class-attribute instance-attribute

right = PydanticField(default_factory=list, description='Right members.') class-attribute instance-attribute

synthesized = PydanticField(default=False, description='Declared by merge itself under `union_right`.') class-attribute instance-attribute

PreviewEdge

Bases: ConfigBaseModel

One thing a declaration says about a pair of names.

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

class PreviewEdge(ConfigBaseModel):
    """One thing a declaration says about a pair of names."""

    id: str = PydanticField(..., description="Stable id: `kind:source->target`.")
    source: str = PydanticField(..., description="Node id this edge leaves.")
    target: str = PydanticField(..., description="Node id this edge enters.")
    kind: EdgeKind = PydanticField(..., description="Which declaration said it.")
    label: str = PydanticField(default="", description="Short caption, when useful.")
    declared_by: str = PydanticField(
        default="",
        description="Where it was declared: a map scope, or a cluster's merged name.",
    )

Attributes

declared_by = PydanticField(default='', description="Where it was declared: a map scope, or a cluster's merged name.") class-attribute instance-attribute

id = PydanticField(..., description='Stable id: `kind:source->target`.') class-attribute instance-attribute

kind = PydanticField(..., description='Which declaration said it.') class-attribute instance-attribute

label = PydanticField(default='', description='Short caption, when useful.') class-attribute instance-attribute

source = PydanticField(..., description='Node id this edge leaves.') class-attribute instance-attribute

target = PydanticField(..., description='Node id this edge enters.') class-attribute instance-attribute

PreviewNode

Bases: ConfigBaseModel

One class, relation or attribute in the declaration graph.

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

class PreviewNode(ConfigBaseModel):
    """One class, relation or attribute in the declaration graph."""

    id: str = PydanticField(
        ..., description="Stable id, as `subject()` builds it: `left:Firm.firm_id`."
    )
    kind: NodeKind = PydanticField(..., description="What this node stands for.")
    name: str = PydanticField(..., description="The name as its side spells it.")
    side: Side | None = PydanticField(
        default=None,
        description="Which manifest it comes from; merged names have none.",
    )
    owner: str | None = PydanticField(
        default=None, description="For an attribute, the node id of its class."
    )
    identity: bool = PydanticField(
        default=False,
        description="Whether this attribute takes part in its class's identity.",
    )
    exists: bool = PydanticField(
        default=True,
        description="False when a declaration names it but the manifest does not.",
    )
    field_type: str | None = PydanticField(
        default=None, description="Declared type of an attribute, when it has one."
    )
    identity_mode: str | None = PydanticField(
        default=None, description="natural / hash / blank / assigned, for a class."
    )

Attributes

exists = PydanticField(default=True, description='False when a declaration names it but the manifest does not.') class-attribute instance-attribute

field_type = PydanticField(default=None, description='Declared type of an attribute, when it has one.') class-attribute instance-attribute

id = PydanticField(..., description='Stable id, as `subject()` builds it: `left:Firm.firm_id`.') class-attribute instance-attribute

identity = PydanticField(default=False, description="Whether this attribute takes part in its class's identity.") class-attribute instance-attribute

identity_mode = PydanticField(default=None, description='natural / hash / blank / assigned, for a class.') class-attribute instance-attribute

kind = PydanticField(..., description='What this node stands for.') class-attribute instance-attribute

name = PydanticField(..., description='The name as its side spells it.') class-attribute instance-attribute

owner = PydanticField(default=None, description='For an attribute, the node id of its class.') class-attribute instance-attribute

side = PydanticField(default=None, description='Which manifest it comes from; merged names have none.') class-attribute instance-attribute

SlotNode

Bases: ConfigBaseModel

One addressable location in the manifest, and what each branch did to it.

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

class SlotNode(ConfigBaseModel):
    """One addressable location in the manifest, and what each branch did to it."""

    id: str = PydanticField(
        ..., description="The slot as a path: `vertex/person/field/age`."
    )
    segment: str = PydanticField(..., description="This slot's own last segment.")
    depth: int = PydanticField(..., description="How many segments deep it sits.")
    parent: str | None = PydanticField(
        default=None, description="The containing slot, or None at the root."
    )
    contested: bool = PydanticField(
        default=False, description="Whether both branches changed this slot."
    )
    reason: str | None = PydanticField(
        default=None, description="Why it could not be reconciled automatically."
    )
    left_ops: list[str] = PydanticField(
        default_factory=list, description="What the left branch did here, by op name."
    )
    right_ops: list[str] = PydanticField(
        default_factory=list, description="What the right branch did here."
    )
    clean_ops: list[str] = PydanticField(
        default_factory=list,
        description="Ops the merge applied here without a decision.",
    )
    base_excerpt: dict[str, Any] | None = PydanticField(
        default=None, description="The ancestor's state here, for whoever decides."
    )

Attributes

base_excerpt = PydanticField(default=None, description="The ancestor's state here, for whoever decides.") class-attribute instance-attribute

clean_ops = PydanticField(default_factory=list, description='Ops the merge applied here without a decision.') class-attribute instance-attribute

contested = PydanticField(default=False, description='Whether both branches changed this slot.') class-attribute instance-attribute

depth = PydanticField(..., description='How many segments deep it sits.') class-attribute instance-attribute

id = PydanticField(..., description='The slot as a path: `vertex/person/field/age`.') class-attribute instance-attribute

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

parent = PydanticField(default=None, description='The containing slot, or None at the root.') class-attribute instance-attribute

reason = PydanticField(default=None, description='Why it could not be reconciled automatically.') class-attribute instance-attribute

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

segment = PydanticField(..., description="This slot's own last segment.") class-attribute instance-attribute

Functions:

build_merge3_preview(result, *, base=None)

Project a :class:~graflo.architecture.evolution.merge3.MergeResult onto its slot tree.

Slots are paths and contain one another, so the set of slots a merge touched is already a tree once its prefixes are filled in. Contested slots carry what each branch did and the ancestor's state; slots the merge settled on its own carry the ops it applied, which is the context that makes a conflict legible — a rename colliding with three field edits is one conflict about the vertex, with the field edits visible beneath it.

Parameters:

Name	Type	Description	Default
result	Any	What merge_three_way returned alongside the merged manifest.	required
base	GraphManifest | None	The common ancestor. Unused today; accepted so a caller can pass it without knowing whether the excerpt came from the result.	None

Returns:

Name	Type	Description
A	Merge3Preview
	Merge3Preview	is exactly the case worth drawing.

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

def build_merge3_preview(
    result: Any, *, base: GraphManifest | None = None
) -> Merge3Preview:
    """Project a :class:`~graflo.architecture.evolution.merge3.MergeResult` onto its slot tree.

    Slots are paths and contain one another, so the set of slots a merge
    touched is already a tree once its prefixes are filled in. Contested slots
    carry what each branch did and the ancestor's state; slots the merge
    settled on its own carry the ops it applied, which is the context that makes
    a conflict legible — a rename colliding with three field edits is one
    conflict about the vertex, with the field edits visible beneath it.

    Args:
        result: What ``merge_three_way`` returned alongside the merged manifest.
        base: The common ancestor. Unused today; accepted so a caller can pass
            it without knowing whether the excerpt came from the result.

    Returns:
        A :class:`Merge3Preview`. Never raises: a merge that could not complete
        is exactly the case worth drawing.
    """
    from .merge3 import describe_slot, op_slots

    del base  # excerpts travel on the conflicts themselves

    nodes: dict[str, SlotNode] = {}

    def ensure(slot: tuple[str, ...]) -> SlotNode:
        """The node for *slot*, and every slot that contains it."""
        node_id = describe_slot(slot)
        existing = nodes.get(node_id)
        if existing is not None:
            return existing
        parent = describe_slot(slot[:-1]) if len(slot) > 1 else None
        if len(slot) > 1:
            ensure(slot[:-1])
        node = SlotNode(
            id=node_id,
            segment=slot[-1],
            depth=len(slot) - 1,
            parent=parent,
        )
        nodes[node_id] = node
        return node

    for op in result.ops:
        for slot in op_slots(op):
            if not slot:
                continue
            ensure(slot).clean_ops.append(op.op)

    for conflict in result.conflicts:
        slot = tuple(conflict.slot)
        if not slot:
            continue
        node = ensure(slot)
        nodes[node.id] = node.model_copy(
            update={
                "contested": True,
                "reason": conflict.reason,
                "left_ops": [op.op for op in conflict.left_ops],
                "right_ops": [op.op for op in conflict.right_ops],
                "base_excerpt": conflict.base_excerpt or None,
                # A contested slot is held back whole, so nothing here was
                # applied cleanly; showing both would misread the result.
                "clean_ops": [],
            }
        )

    ordered = sorted(nodes.values(), key=lambda n: (n.depth, n.id))
    return Merge3Preview(
        nodes=ordered,
        conflicts=len(result.conflicts),
        clean=result.clean,
        merged_hash=result.merged_hash,
        warnings=list(result.warnings),
    )

expected_kinds(outcome)

Structural finding kinds that should accompany outcome.

The invariant this module is tested against: whatever merge refused, the structural pass saw something of a matching kind. An empty set means the refusal is one the preview is not asked to anticipate -- a structural merge error from deep inside the union, say -- and asserts nothing.

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

def expected_kinds(outcome: MergeOutcome) -> frozenset[FindingKind]:
    """Structural finding kinds that should accompany *outcome*.

    The invariant this module is tested against: whatever merge refused, the
    structural pass saw something of a matching kind. An empty set means the
    refusal is one the preview is not asked to anticipate -- a structural merge
    error from deep inside the union, say -- and asserts nothing.
    """
    if outcome.status != "refused":
        return frozenset()
    if outcome.check:
        return frozenset({kind_for_check(outcome.check, outcome.error_type)})
    by_type = _KINDS_BY_TYPE_NAME.get(outcome.error_type or "")
    return by_type if by_type is not None else frozenset()

kind_for_check(check, error_type=None)

The finding kind a refusal's check phrase is an instance of.

Longest match wins, so "property rename collision" is not read as the plain "collision" of a name clash. Falls back to the exception type, and finally to disagreement -- the most general of the four classes.

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

def kind_for_check(check: str | None, error_type: str | None = None) -> FindingKind:
    """The finding kind a refusal's ``check`` phrase is an instance of.

    Longest match wins, so ``"property rename collision"`` is not read as the
    plain ``"collision"`` of a name clash. Falls back to the exception type,
    and finally to ``disagreement`` -- the most general of the four classes.
    """
    if check:
        for phrase in sorted(_KIND_BY_CHECK, key=len, reverse=True):
            if phrase in check:
                return _KIND_BY_CHECK[phrase]
    if error_type:
        kinds = _KINDS_BY_TYPE_NAME.get(error_type)
        if kinds:
            return min(kinds)
    return "disagreement"

outcome_from_exception(exc)

The outcome a refusal is, and the node ids it names.

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

def outcome_from_exception(
    exc: BaseException,
) -> tuple[MergeOutcome, tuple[str, ...]]:
    """The outcome a refusal is, and the node ids it names."""
    completion = getattr(exc, "completion", None)
    return (
        MergeOutcome(
            status="refused",
            error_type=type(exc).__name__,
            message=str(exc),
            check=getattr(exc, "check", "") or None,
            completion=completion.to_dict() if completion is not None else None,
        ),
        tuple(getattr(exc, "subjects", ()) or ()),
    )

outcome_from_manifest(manifest)

The outcome a merged manifest is.

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

def outcome_from_manifest(manifest: GraphManifest) -> MergeOutcome:
    """The outcome a merged manifest is."""
    schema = manifest.graph_schema
    if schema is None:
        return MergeOutcome(status="merged")
    core = schema.core_schema
    return MergeOutcome(
        status="merged",
        vertices=len(core.vertex_config.vertices),
        edges=len(core.edge_config.edges),
        version=str(schema.metadata.version) if schema.metadata.version else None,
    )

preview_merge(left, right, op, *, canonical_maps=(), attempt=True)

The declaration graph of a merge, and everything wrong with it.

Walks op's equivalences and canonical maps against left and right without refusing: each declaration, each map entry and each member is put through the same check :func:~graflo.architecture.evolution.merge.merge_manifests uses, one at a time, so a problem with one does not hide the rest. Every refusal becomes a possible finding naming the nodes it is about.

With attempt, merge is then run for real and its result -- the merged schema's shape, or the one refusal it raised, with the completion that would settle it -- is recorded as the outcome and as a single refusal finding. Set it to False to describe the declarations without merging.

Parameters:

Name	Type	Description	Default
left	GraphManifest	The left manifest, in whatever vocabulary it is in.	required
right	GraphManifest	The right manifest.	required
op	MergeManifestsOp	The merge op: equivalences, canonical maps, identity alignments.	required
canonical_maps	Sequence[tuple[Side, CanonicalMap]]	Extra (side, map) pairs, folded into op's.	()
attempt	bool	Whether to run a real merge for the authoritative outcome.	True

Returns:

Name	Type	Description
A	MergePreview
	MergePreview	declarations -- that is the point -- so an empty
	MergePreview
	MergePreview	like.

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

def preview_merge(
    left: GraphManifest,
    right: GraphManifest,
    op: MergeManifestsOp,
    *,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
    attempt: bool = True,
) -> MergePreview:
    """The declaration graph of a merge, and everything wrong with it.

    Walks *op*'s equivalences and canonical maps against *left* and *right*
    without refusing: each declaration, each map entry and each member is put
    through the same check
    :func:`~graflo.architecture.evolution.merge.merge_manifests` uses, one
    at a time, so a problem with one does not hide the rest. Every refusal
    becomes a ``possible`` finding naming the nodes it is about.

    With *attempt*, merge is then run for real and its result -- the merged
    schema's shape, or the one refusal it raised, with the completion that
    would settle it -- is recorded as the outcome and as a single ``refusal``
    finding. Set it to ``False`` to describe the declarations without merging.

    Args:
        left: The left manifest, in whatever vocabulary it is in.
        right: The right manifest.
        op: The merge op: equivalences, canonical maps, identity alignments.
        canonical_maps: Extra ``(side, map)`` pairs, folded into ``op``'s.
        attempt: Whether to run a real merge for the authoritative outcome.

    Returns:
        A :class:`MergePreview`. It never raises for a problem with the
        declarations -- that is the point -- so an empty
        :attr:`~MergePreview.blocking` is what "this would merge" looks
        like.
    """
    declared = fold_declared_maps(op, canonical_maps)
    manifests: dict[Side, GraphManifest] = {"left": left, "right": right}
    names: dict[Side, SideNames] = {
        "left": SideNames.of(left),
        "right": SideNames.of(right),
    }

    builder = _Builder(op=op, manifests=manifests, names=names, declared=declared)
    preview = builder.build()

    if op.name_conflict == "union_right":
        # The names both sides arrive at are only known once the first pass has
        # applied the declared maps, so the clusters merge would synthesize
        # for them are resolved on a second pass over the extended op.
        extended = _extended(op, builder)
        if extended is not None:
            preview = _Builder(
                op=extended,
                manifests=manifests,
                names=names,
                declared=declared,
                synthesized_from=(
                    len(op.vertex_equivalences),
                    len(op.relation_equivalences),
                ),
            ).build()
    if not attempt:
        return preview
    outcome, subjects = _attempt(left, right, op, canonical_maps)
    return preview.with_outcome(outcome, subjects=subjects)