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

AttributeLayer = Literal['equivalence', 'vocabulary', 'rename', 'demotion', 'fusion'] module-attribute

EdgeKind = Literal['member', 'map', 'property_map', 'property_equivalence', 'suggested', 'fusion', 'demotion', 'property_rename', 'derivation', 'local_key'] module-attribute

FindingKind = Literal['cluster-overlap', 'shared-into', 'occupied-into', 'unknown-member', 'disagreement', 'ambiguous-reference', 'unnamed-cluster', 'incomplete', 'dangling-entry', 'satisfied', 'name-collision', 'near-collision', 'prefixed', 'property-disagreement', 'property-rename-collision', 'property-retarget', 'unknown-property', 'identity-disagreement', 'identity-coverage', 'identity-collision', 'origin', 'key-space', 'field-type-conflict', 'field-units-conflict', 'identity-mode-conflict', 'identity-funnel-conflict', 'secondary-identity-conflict', 'edge-type-disagreement', 'lookup_demotion', 'attached', 'shared_key_space', 'attach_order', 'double-home', 'vocabulary_override', 'auto_local_key', 'reference_only', 'self-relation', 'observation-fusion'] module-attribute

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

RoutingRole = Literal['owner', 'attached', 'reference'] module-attribute

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

__all__ = ['AttributeLayer', 'BranchSource', 'EdgeKind', 'FindingKind', 'Merge3Preview', 'MergeFinding', 'MergeOutcome', 'MergePreview', 'NodeKind', 'PreviewBranch', 'PreviewCluster', 'PreviewDemotion', 'PreviewEdge', 'PreviewIdentity', 'PreviewNode', 'RoutingRole', 'RoutingRow', 'Severity', 'SlotNode', 'build_merge3_preview', 'expected_kinds', 'kind_for_check', 'merge_notes', 'outcome_from_exception', 'outcome_from_manifest', 'preview_merge', 'routing_rows', 'routing_table', 'with_merge_result'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

BranchSource

Bases: ConfigBaseModel

One thing that feeds a branch: a resource's derivation or local key, or a member that carries a property branch.

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

class BranchSource(ConfigBaseModel):
    """One thing that feeds a branch: a resource's derivation or local key, or a member that carries a property branch."""

    resource: str | None = PydanticField(
        default=None, description="Union resource name; None for a property branch."
    )
    side: Side | None = PydanticField(
        default=None, description="The side the resource or member comes from."
    )
    exists: bool = PydanticField(
        default=True, description="False when no side has a resource of this name."
    )
    member: str | None = PydanticField(
        default=None,
        description="The member a member-keyed entry is for, or the member carrying a property branch.",
    )
    produces: list[str] = PydanticField(
        default_factory=list,
        description="Members of the group the resource produces on its side.",
    )
    input: list[str] = PydanticField(
        default_factory=list, description="Raw fields a derivation reads."
    )
    module: str | None = PydanticField(default=None, description="Transform module.")
    foo: str | None = PydanticField(default=None, description="Transform function.")
    params: dict[str, Any] = PydanticField(
        default_factory=dict, description="Transform keyword arguments."
    )
    field: str | None = PydanticField(
        default=None, description="Raw field a local key reads."
    )
    tag: str | None = PydanticField(
        default=None, description="The tag a local key value carries; '' is untagged."
    )
    tag_omitted: bool = PydanticField(
        default=False, description="The tag was left out and defaults to the origin."
    )
    auto: bool = PydanticField(
        default=False, description="Merge added this local key: nothing declared one."
    )
    when: dict[str, Any] | None = PydanticField(
        default=None, description="The declared guard, as written."
    )
    derive_at: list[int] | None = PydanticField(
        default=None, description="The declared descend step path for the resource."
    )
    label: str = PydanticField(
        default="", description="One-line caption: transform and guard, or tag."
    )

Attributes

auto = PydanticField(default=False, description='Merge added this local key: nothing declared one.') class-attribute instance-attribute

derive_at = PydanticField(default=None, description='The declared descend step path for the resource.') class-attribute instance-attribute

exists = PydanticField(default=True, description='False when no side has a resource of this name.') class-attribute instance-attribute

field = PydanticField(default=None, description='Raw field a local key reads.') class-attribute instance-attribute

foo = PydanticField(default=None, description='Transform function.') class-attribute instance-attribute

input = PydanticField(default_factory=list, description='Raw fields a derivation reads.') class-attribute instance-attribute

label = PydanticField(default='', description='One-line caption: transform and guard, or tag.') class-attribute instance-attribute

member = PydanticField(default=None, description='The member a member-keyed entry is for, or the member carrying a property branch.') class-attribute instance-attribute

module = PydanticField(default=None, description='Transform module.') class-attribute instance-attribute

params = PydanticField(default_factory=dict, description='Transform keyword arguments.') class-attribute instance-attribute

produces = PydanticField(default_factory=list, description='Members of the group the resource produces on its side.') class-attribute instance-attribute

resource = PydanticField(default=None, description='Union resource name; None for a property branch.') class-attribute instance-attribute

side = PydanticField(default=None, description='The side the resource or member comes from.') class-attribute instance-attribute

tag = PydanticField(default=None, description="The tag a local key value carries; '' is untagged.") class-attribute instance-attribute

tag_omitted = PydanticField(default=False, description='The tag was left out and defaults to the origin.') class-attribute instance-attribute

when = PydanticField(default=None, description='The declared guard, as written.') class-attribute instance-attribute

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."
    )
    kinds: list[FindingKind] = PydanticField(
        default_factory=list,
        description=(
            "For a naming refusal, the kind of every problem it lists, in order."
        ),
    )
    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

kinds = PydanticField(default_factory=list, description='For a naming refusal, the kind of every problem it lists, in order.') 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")
    )
    routing: list[RoutingRow] = PydanticField(
        default_factory=list,
        description="Every resource producing a merged class, once merge went through.",
    )

    @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 edges_between(self, source: str, target: str) -> list[PreviewEdge]:
        """Every edge from node *source* to node *target*."""
        return [e for e in self.edges if e.source == source and e.target == target]

    def focus(self, target: str | int) -> MergePreview:
        """The part of this preview one cluster, merged attribute or ``rest`` is about.

        An integer names the cluster ``op.vertex_equivalences[target]`` folded into.

        See :func:`~graflo.architecture.evolution.preview_focus.focus`.
        """
        from .preview_focus import focus

        return focus(self, target)

    def focus_targets(
        self, attributes: Literal["nontrivial", "all"] = "nontrivial"
    ) -> list[str]:
        """Every target :meth:`focus` takes, in reading order.

        See :func:`~graflo.architecture.evolution.preview_focus.focus_targets`.
        """
        from .preview_focus import focus_targets

        return focus_targets(self, attributes)

    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=outcome.kinds[0]
            if outcome.kinds
            else 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,
        )
        message = outcome.message or ""
        listed = {
            position
            for position, finding in enumerate(self.findings)
            if finding.severity == "possible"
            and finding.message
            and finding.message != message
            and finding.message in message
        }
        if listed:
            # A naming refusal that lists several problems: each one the
            # structural pass found is part of it, and none is reported twice.
            return self.model_copy(
                update={
                    "outcome": outcome,
                    "findings": [
                        finding.model_copy(
                            update={
                                "severity": "refusal",
                                "source": "merge",
                                "error_type": outcome.error_type,
                            }
                        )
                        if position in listed
                        else finding
                        for position, finding in enumerate(self.findings)
                    ],
                }
            )
        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

routing = PydanticField(default_factory=list, description='Every resource producing a merged class, once merge went through.') 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]

edges_between(source, target)

Every edge from node source to node target.

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

def edges_between(self, source: str, target: str) -> list[PreviewEdge]:
    """Every edge from node *source* to node *target*."""
    return [e for e in self.edges if e.source == source and e.target == target]

focus(target)

The part of this preview one cluster, merged attribute or rest is about.

An integer names the cluster op.vertex_equivalences[target] folded into.

See :func:~graflo.architecture.evolution.preview_focus.focus.

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

def focus(self, target: str | int) -> MergePreview:
    """The part of this preview one cluster, merged attribute or ``rest`` is about.

    An integer names the cluster ``op.vertex_equivalences[target]`` folded into.

    See :func:`~graflo.architecture.evolution.preview_focus.focus`.
    """
    from .preview_focus import focus

    return focus(self, target)

focus_targets(attributes='nontrivial')

Every target :meth:focus takes, in reading order.

See :func:~graflo.architecture.evolution.preview_focus.focus_targets.

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

def focus_targets(
    self, attributes: Literal["nontrivial", "all"] = "nontrivial"
) -> list[str]:
    """Every target :meth:`focus` takes, in reading order.

    See :func:`~graflo.architecture.evolution.preview_focus.focus_targets`.
    """
    from .preview_focus import focus_targets

    return focus_targets(self, attributes)

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=outcome.kinds[0]
        if outcome.kinds
        else 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,
    )
    message = outcome.message or ""
    listed = {
        position
        for position, finding in enumerate(self.findings)
        if finding.severity == "possible"
        and finding.message
        and finding.message != message
        and finding.message in message
    }
    if listed:
        # A naming refusal that lists several problems: each one the
        # structural pass found is part of it, and none is reported twice.
        return self.model_copy(
            update={
                "outcome": outcome,
                "findings": [
                    finding.model_copy(
                        update={
                            "severity": "refusal",
                            "source": "merge",
                            "error_type": outcome.error_type,
                        }
                    )
                    if position in listed
                    else finding
                    for position, finding in enumerate(self.findings)
                ],
            }
        )
    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})

PreviewBranch

Bases: ConfigBaseModel

One funnel branch, or one derive attribute.

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

class PreviewBranch(ConfigBaseModel):
    """One funnel branch, or one ``derive`` attribute."""

    position: int = PydanticField(
        ..., description="1-based order a record tries it in; 0 for a `derive` entry."
    )
    id: str = PydanticField(..., description="Branch id, as the funnel names it.")
    kind: BranchKind = PydanticField(..., description="What the branch keys on.")
    fields: list[str] = PydanticField(
        default_factory=list, description="Merged attributes the branch reads."
    )
    sources: list[BranchSource] = PydanticField(default_factory=list)
    sep: str | None = PydanticField(
        default=None,
        description="What joins a local key's tag and value; None for other kinds.",
    )

Attributes

fields = PydanticField(default_factory=list, description='Merged attributes the branch reads.') class-attribute instance-attribute

id = PydanticField(..., description='Branch id, as the funnel names it.') class-attribute instance-attribute

kind = PydanticField(..., description='What the branch keys on.') class-attribute instance-attribute

position = PydanticField(..., description='1-based order a record tries it in; 0 for a `derive` entry.') class-attribute instance-attribute

sep = PydanticField(default=None, description="What joins a local key's tag and value; None for other kinds.") class-attribute instance-attribute

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

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 wrote it, if any."
    )
    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`."
    )
    declarations: list[int] = PydanticField(
        default_factory=list,
        description=(
            "Positions, in the op's vertex or relation equivalences, of the "
            "declarations folded into this group; empty when merge synthesized it."
        ),
    )
    declared_left: list[str] = PydanticField(
        default_factory=list,
        description="Left members the declarations name; the rest a vocabulary joined.",
    )
    declared_right: list[str] = PydanticField(
        default_factory=list,
        description="Right members the declarations name; the rest a vocabulary joined.",
    )
    allow: list[str] = PydanticField(
        default_factory=list, description="The declaration's `allow` entries."
    )
    declared_identity: bool = PydanticField(
        default=False, description="Whether the declaration states an `identity`."
    )
    derived_identity: bool = PydanticField(
        default=False,
        description="Whether its `identity` has a derived or `local_key` branch.",
    )
    identity: PreviewIdentity | None = PydanticField(
        default=None,
        description="How the merged class is keyed; None for a relation group.",
    )

Attributes

allow = PydanticField(default_factory=list, description="The declaration's `allow` entries.") class-attribute instance-attribute

declarations = PydanticField(default_factory=list, description="Positions, in the op's vertex or relation equivalences, of the declarations folded into this group; empty when merge synthesized 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 wrote it, if any.') class-attribute instance-attribute

declared_left = PydanticField(default_factory=list, description='Left members the declarations name; the rest a vocabulary joined.') class-attribute instance-attribute

declared_right = PydanticField(default_factory=list, description='Right members the declarations name; the rest a vocabulary joined.') class-attribute instance-attribute

derived_identity = PydanticField(default=False, description='Whether its `identity` has a derived or `local_key` branch.') class-attribute instance-attribute

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

identity = PydanticField(default=None, description='How the merged class is keyed; None for a relation group.') 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

PreviewDemotion

Bases: ConfigBaseModel

A member key the new identity demotes to a lookup-only secondary.

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

class PreviewDemotion(ConfigBaseModel):
    """A member key the new identity demotes to a lookup-only secondary."""

    side: Side
    member: str
    fields: list[str] = PydanticField(
        default_factory=list,
        description="The key fields, as the member spells them after its vocabulary renames.",
    )
    merged_fields: list[str] = PydanticField(
        default_factory=list, description="Their names on the merged class."
    )
    key_space: str = PydanticField(..., description="What names the demoted key.")
    tags: list[str] = PydanticField(
        default_factory=list, description="Every key space the op's tags name for it."
    )

Attributes

fields = PydanticField(default_factory=list, description='The key fields, as the member spells them after its vocabulary renames.') class-attribute instance-attribute

key_space = PydanticField(..., description='What names the demoted key.') class-attribute instance-attribute

member instance-attribute

merged_fields = PydanticField(default_factory=list, description='Their names on the merged class.') class-attribute instance-attribute

side instance-attribute

tags = PydanticField(default_factory=list, description="Every key space the op's tags name for it.") 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.",
    )
    layer: AttributeLayer | None = PydanticField(
        default=None,
        description="For an attribute edge, which declaration it comes from.",
    )
    effective: bool = PydanticField(
        default=True,
        description="False when a later declaration sent the attribute elsewhere.",
    )

Attributes

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

effective = PydanticField(default=True, description='False when a later declaration sent the attribute elsewhere.') 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

layer = PydanticField(default=None, description='For an attribute edge, which declaration it comes from.') 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

PreviewIdentity

Bases: ConfigBaseModel

How a merged class is keyed, and where every key value comes from.

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

class PreviewIdentity(ConfigBaseModel):
    """How a merged class is keyed, and where every key value comes from."""

    mode: IdentityMode
    branches: list[PreviewBranch] = PydanticField(default_factory=list)
    derive: list[PreviewBranch] = PydanticField(
        default_factory=list,
        description="`derive` entries: attributes a branch keys on, computed per resource.",
    )
    digest_field: str | None = PydanticField(
        default=None, description="Where a funnel stores its digest."
    )
    retire: str | None = PydanticField(
        default=None, description="What happens to each member's own key."
    )
    derive_at: dict[str, list[int]] = PydanticField(default_factory=dict)
    demoted: list[PreviewDemotion] = PydanticField(default_factory=list)

    def role_of(self, attribute: str) -> list[int]:
        """Positions of the branches that key on merged *attribute*."""
        return [b.position for b in self.branches if attribute in b.fields]

Attributes

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

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

derive = PydanticField(default_factory=list, description='`derive` entries: attributes a branch keys on, computed per resource.') class-attribute instance-attribute

derive_at = PydanticField(default_factory=dict) class-attribute instance-attribute

digest_field = PydanticField(default=None, description='Where a funnel stores its digest.') class-attribute instance-attribute

mode instance-attribute

retire = PydanticField(default=None, description="What happens to each member's own key.") class-attribute instance-attribute

Methods:

role_of(attribute)

Positions of the branches that key on merged attribute.

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

def role_of(self, attribute: str) -> list[int]:
    """Positions of the branches that key on merged *attribute*."""
    return [b.position for b in self.branches if attribute in b.fields]

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

RoutingRow

Bases: ConfigBaseModel

One resource producing a merged class, and the key its records reach it by.

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

class RoutingRow(ConfigBaseModel):
    """One resource producing a merged class, and the key its records reach it by."""

    vertex: str = PydanticField(..., description="The merged class.")
    resource: str = PydanticField(..., description="The union resource.")
    side: Side = PydanticField(..., description="The side the resource came from.")
    members: tuple[str, ...] = PydanticField(
        ..., description="The members of that side it produces, in source names."
    )
    role: RoutingRole = PydanticField(..., description="How it reaches the class.")
    key: str = PydanticField(
        ...,
        description=(
            "The class's primary identity fields for an owner; the secondary "
            "identity (key space) it finds the class by otherwise."
        ),
    )

Attributes

key = PydanticField(..., description="The class's primary identity fields for an owner; the secondary identity (key space) it finds the class by otherwise.") class-attribute instance-attribute

members = PydanticField(..., description='The members of that side it produces, in source names.') class-attribute instance-attribute

resource = PydanticField(..., description='The union resource.') class-attribute instance-attribute

role = PydanticField(..., description='How it reaches the class.') class-attribute instance-attribute

side = PydanticField(..., description='The side the resource came from.') class-attribute instance-attribute

vertex = PydanticField(..., description='The merged class.') 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.kinds:
        return frozenset(outcome.kinds)
    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 is an instance of.

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`` is an instance of.

    Falls back to the exception type, and finally to ``disagreement`` -- the
    most general of the four classes.
    """
    if check and is_refusal_kind(check) and check in _KIND_BY_CHECK:
        return _KIND_BY_CHECK[check]
    if error_type:
        kinds = _KINDS_BY_TYPE_NAME.get(error_type)
        if kinds:
            return min(kinds)
    return "disagreement"

merge_notes(report)

What a merge that went through changed about what resources do, as notes.

A member key demoted beside a funnel of several branches no longer deduplicates the member's own records; a resource is attached to a merged class; members of one side share a key space; a resource runs before the one whose nodes it needs.

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

def merge_notes(report: MergeReport) -> list[MergeFinding]:
    """What a merge that went through changed about what resources do, as notes.

    A member key demoted beside a funnel of several branches no longer
    deduplicates the member's own records; a resource is attached to a merged
    class; members of one side share a key space; a resource runs before the
    one whose nodes it needs.
    """
    return [
        *_demotion_notes(report.demoted),
        *_attached_notes(report.attached),
        *_shared_key_space_notes(report.shared_key_spaces),
        *_attach_order_notes(report.order_cycles),
    ]

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."""
    refusal = exc if isinstance(exc, Refusal) else None
    completion = _completion_of(exc)
    kinds: list[FindingKind] = (
        [cast(FindingKind, finding.kind) for finding in exc.findings]
        if isinstance(exc, MergeNamingError)
        else []
    )
    return (
        MergeOutcome(
            status="refused",
            error_type=type(exc).__name__,
            message=str(exc),
            check=(refusal.check if refusal is not None else "") or None,
            kinds=kinds,
            completion=completion.to_dict() if completion is not None else None,
        ),
        refusal.subjects if refusal is not None else (),
    )

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 (with their identities) and canonical maps.	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 (with their identities) and canonical maps.
        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.
    """
    manifests: dict[Side, GraphManifest] = {"left": left, "right": right}
    names: dict[Side, SideNames] = {
        "left": SideNames.of(left),
        "right": SideNames.of(right),
    }
    preview = _Builder(
        op=op, manifests=manifests, names=names, canonical_maps=canonical_maps
    ).build()
    if not attempt:
        return preview
    from .alignment import AlignmentConflictError
    from .merge import _merge_manifests

    try:
        merged, report = _merge_manifests(
            left, right, op, canonical_maps=canonical_maps, finish_init=False
        )
    except (
        MergeCanonicalConflictError,
        MergeIdentityError,
        MergeNameConflictError,
        AlignmentConflictError,
        ValueError,
    ) as exc:
        outcome, subjects = outcome_from_exception(exc)
        return preview.with_outcome(outcome, subjects=subjects)
    return with_merge_result(preview, merged, report)

routing_rows(manifest, report)

Every resource producing a merged class, with its role and key there.

Rows are grouped by merged class and follow the union's resource order.

Parameters:

Name	Type	Description	Default
manifest	GraphManifest	The merged manifest report describes.	required
report	MergeReport	The report returned beside it by :func:~graflo.architecture.evolution.merge.merge_manifests_with_report.	required

Returns:

Name	Type	Description
One	list[RoutingRow]
	list[RoutingRow]	reference.

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

def routing_rows(manifest: GraphManifest, report: MergeReport) -> list[RoutingRow]:
    """Every resource producing a merged class, with its role and key there.

    Rows are grouped by merged class and follow the union's resource order.

    Args:
        manifest: The merged manifest *report* describes.
        report: The report returned beside it by
            :func:`~graflo.architecture.evolution.merge.merge_manifests_with_report`.

    Returns:
        One :class:`RoutingRow` per key owner, attached producer and pinned
        reference.
    """
    schema = manifest.graph_schema
    ingestion = manifest.ingestion_model
    position = (
        {resource.name: i for i, resource in enumerate(ingestion.resources)}
        if ingestion is not None
        else {}
    )

    def primary(vertex: str) -> str:
        if schema is None:
            return "-"
        return ", ".join(schema.core_schema.vertex_config.identity_fields(vertex))

    rows = [
        *(
            RoutingRow(
                vertex=owner.vertex,
                resource=owner.resource,
                side=owner.side,
                members=owner.members,
                role="owner",
                key=primary(owner.vertex),
            )
            for owner in report.owners
        ),
        *(
            RoutingRow(
                vertex=producer.vertex,
                resource=producer.resource,
                side=producer.side,
                members=producer.members,
                role="attached",
                key=producer.key,
            )
            for producer in report.attached
        ),
        *(
            RoutingRow(
                vertex=reference.vertex,
                resource=reference.resource,
                side=reference.side,
                members=reference.members,
                role="reference",
                key=reference.key,
            )
            for reference in report.references
        ),
    ]
    return sorted(
        rows,
        key=lambda row: (row.vertex, position.get(row.resource, len(position))),
    )

routing_table(rows)

The routing rows as text: one block per merged class, named on its first row.

Columns: merged resource side members role key. No rows, no table.

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

def routing_table(rows: Sequence[RoutingRow]) -> list[str]:
    """The routing rows as text: one block per merged class, named on its first row.

    Columns: ``merged resource side members role key``. No rows, no table.
    """
    if not rows:
        return []
    header = ("merged", "resource", "side", "members", "role", "key")
    cells: list[tuple[str, ...]] = []
    previous: str | None = None
    for row in sorted(rows, key=lambda r: r.vertex):
        cells.append(
            (
                row.vertex if row.vertex != previous else "",
                row.resource,
                row.side,
                ", ".join(row.members) or "-",
                row.role,
                row.key,
            )
        )
        previous = row.vertex
    widths = [max(len(row[i]) for row in (header, *cells)) for i in range(len(header))]
    return [
        "  ".join(
            cell.ljust(width) for cell, width in zip(row, widths, strict=True)
        ).rstrip()
        for row in (header, *cells)
    ]

with_merge_result(preview, merged, report)

preview with the outcome, routing and notes of a merge that went through.

For a caller that ran the merge itself -- the CLI does, once -- on a preview built with attempt=False; :func:preview_merge folds its own attempt in the same way.

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

def with_merge_result(
    preview: MergePreview, merged: GraphManifest, report: MergeReport
) -> MergePreview:
    """*preview* with the outcome, routing and notes of a merge that went through.

    For a caller that ran the merge itself -- the CLI does, once -- on a
    preview built with ``attempt=False``; :func:`preview_merge` folds its own
    attempt in the same way.
    """
    preview = preview.with_outcome(outcome_from_manifest(merged))
    known = {n.id for n in preview.nodes}
    return preview.model_copy(
        update={
            "routing": routing_rows(merged, report),
            "findings": [
                *preview.findings,
                *(
                    note.model_copy(
                        update={"nodes": [n for n in note.nodes if n in known]}
                    )
                    for note in merge_notes(report)
                ),
            ],
        }
    )