graflo.architecture.evolution.canonical

Canonical vocabulary maps and merge-time cluster resolution.

Two declarations say how two manifests' names relate. A :class:~graflo.architecture.evolution.ops.CanonicalMap translates one side's vocabulary into canonical names — a partial function on names, identity where unmapped, and idempotent: a canonical name is a fixed point nothing maps away from. An equivalence cluster on a :class:~graflo.architecture.evolution.ops.MergeManifestsOp says which classes across the two sides are one, and may leave what they are called to the map. Renames merge, so "canonicalize, then declare equivalences in canonical names" and "declare equivalences in raw names, then canonicalize" are the same function; :func:resolve_clusters computes it directly — one composite relabel per side, applied as a single :class:~graflo.architecture.evolution.ops.CanonicalizeOp — so there is no intermediate vocabulary an author has to write in.

Vocabulary

• declared map — a CanonicalMap the author wrote: op.canonical_maps[scope] and the canonical_maps= pairs handed to merge. Scoped left / right (that side's own names) or both (either side's names, and merged names). Folded per side by :func:compose_canonical_maps into a :class:DeclaredMaps.
• cluster (:class:~graflo.architecture.evolution.equivalence.Cluster, resolved from a :class:~graflo.architecture.evolution.equivalence.ClusterSpec) — one equivalence declaration, resolved: its members per side, in the manifests' own spelling, and its merged name — into translated through the declared maps, else the canonical name a map gives a member, else the one spelling every member shares.
• cluster map — per side, every member onto its merged name, the merged name itself included as a self entry so the op merges into it rather than refusing an occupied target.
• composite map (:class:SideMaps, one CanonicalizeOp per side) — the cluster map plus every declared entry that applies to a non-member: what merge applies to that side before the union by name. A relabel, not a vocabulary — two clusters may legitimately chain (one merged name renamed away by another declaration), which a CanonicalMap refuses.
• fixed point — a canonical target. No declared map and no cluster may move it.
• opinion — what the declared maps say a member's canonical name is: the target it maps to, or itself when it is a fixed point.
• satisfied entry — a declared entry whose source is absent from a side and whose target is present: taken as already applied by the caller. A heuristic — it cannot tell that from a target that never had that source — so it is logged.
• dangling entry — a declared entry that matches nothing on any side it could apply to. A typo, refused: one refusal names every one of them on a side, each with a near-miss candidate where another spelling denotes the same concept. allow_dangling_entries drops them instead, for a shared vocabulary deliberately broader than the manifest it is applied to.
• synthesized cluster — a cluster merge declares itself under name_conflict="union_right" for a name both sides carry after their composite maps, or two spellings of one name, so that a union by name goes through the same identity and property reconciliation as a declared one.
• completion (:class:Completion) — the extension that would make an incomplete declaration consistent, carried by :class:MergeIncompleteError as declaration payloads.

One rule

The declared maps and the equivalences must agree on where every name goes, and a canonical target is a fixed point neither may re-map. Every refusal is an instance of it, in one of four classes:

class	error	a trigger
contradiction	MergeCanonicalConflictError	the map says Firm → Company, the cluster names the merged class Party
ambiguity	MergeCanonicalConflictError	one canonical name denotes two members of one cluster
incomplete	MergeIncompleteError (carries a Completion)	a map entry sends a non-member onto a merged name
dangling	MergeCanonicalConflictError	an entry matching no name on any side it could apply to

Plus the cluster-shape checks of :mod:~graflo.architecture.evolution.equivalence, which run before any rename. Every case a declared entry and a cluster can stand in — agreement, naming, translation and each refusal — is tabulated in docs/concepts/schema/manifest_evolution.md under "Canonical maps".

Identity is nominal: a class is the same class across two manifests only by name (or by declared equivalence) — nothing structural fingerprints it, and merge never infers a match.

Attributes

Scope = Literal['left', 'right', 'both'] module-attribute

__all__ = ['CanonicalMap', 'ClusterResolution', 'Completion', 'DanglingEntry', 'DeclaredMaps', 'MergeCanonicalConflictError', 'MergeIncompleteError', 'Scope', 'SideMaps', 'SideNames', 'canonical_map_to_ops', 'canonical_near_collisions', 'canonicalize_ops', 'clusters_to_side_maps', 'compose_canonical_maps', 'dangling_entries', 'fold_declared_maps', 'resolve_clusters', 'same_name_groups', 'trim_canonical_map', 'validate_and_complete_canonical_map'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

CanonicalMap

Bases: ConfigBaseModel

Declared translation of a source vocabulary into canonical names.

A partial function on names, identity where unmapped: vertices maps source class names to canonical class names, relations does the same for relation names, and properties maps, per source class name, source attribute names to canonical attribute names — including for classes whose name does not change. Two sources sharing a target is a merge and must be acknowledged with allow_merges.

It is a vocabulary, so it is idempotent: a canonical name is a fixed point that no entry maps away from. A chain ({X: Z, Z: Q}) or a swap is refused at construction — that shape is a relabel, which :class:CanonicalizeOp expresses directly. The rule is what lets two maps, or a map and an equivalence, be checked for agreement without asking in which order they were written.

Used on its own through :func:~graflo.architecture.evolution.canonical.canonical_map_to_ops, and on :attr:MergeManifestsOp.canonical_maps where it names the merged classes and is checked against the declared equivalences.

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

class CanonicalMap(ConfigBaseModel):
    """Declared translation of a source vocabulary into canonical names.

    A partial function on names, identity where unmapped: ``vertices`` maps
    source class names to canonical class names, ``relations`` does the same
    for relation names, and ``properties`` maps, per *source* class name,
    source attribute names to canonical attribute names — including for
    classes whose name does not change. Two sources sharing a target is a
    merge and must be acknowledged with ``allow_merges``.

    It is a **vocabulary**, so it is idempotent: a canonical name is a fixed
    point that no entry maps away from. A chain (``{X: Z, Z: Q}``) or a swap
    is refused at construction — that shape is a relabel, which
    :class:`CanonicalizeOp` expresses directly. The rule is what lets two
    maps, or a map and an equivalence, be checked for agreement without
    asking in which order they were written.

    Used on its own through :func:`~graflo.architecture.evolution.canonical.canonical_map_to_ops`,
    and on :attr:`MergeManifestsOp.canonical_maps` where it names the
    merged classes and is checked against the declared equivalences.
    """

    vertices: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Class rename map: ``{source_class: canonical_class}``.",
    )
    properties: dict[str, dict[str, str]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-source-class attribute rename map: "
            "``{source_class: {source_attr: canonical_attr}}``."
        ),
    )
    relations: dict[str, str] = PydanticField(
        default_factory=dict,
        description="Relation rename map: ``{source_relation: canonical_relation}``.",
    )
    allow_merges: bool = PydanticField(
        default=False,
        description=(
            "Accept a non-injective ``vertices`` / ``relations`` map. Two "
            "sources sharing a canonical target is a *merge*, not a rename; "
            "it must be a stated intent because merging fuses entities and "
            "can create self-relations."
        ),
    )
    allow_dangling_entries: bool = PydanticField(
        default=False,
        description=(
            "Accept entries that name nothing in the manifest the map is "
            "applied to, dropping and logging each one. A shared vocabulary "
            "map is legitimately broader than any single manifest. Off by "
            "default, because a misspelt class has exactly the same shape, "
            "and dropping it silently narrows the rename to less than the "
            "author asked for."
        ),
    )

    @model_validator(mode="after")
    def _validate_maps(self) -> CanonicalMap:
        if not self.allow_merges:
            # Identity entries (source == target) are excluded: a lowered
            # cluster map deliberately carries one for every member, including
            # the merged name itself, to declare it a member of its own group
            # -- that self entry must not read as a collision here. The op the
            # map lowers to counts it, which is where the merge is acknowledged.
            validate_rename_map_is_injective(
                {s: t for s, t in self.vertices.items() if s != t},
                kind="canonical vertex",
                merge_hint="CanonicalMap(allow_merges=True)",
            )
            validate_rename_map_is_injective(
                {s: t for s, t in self.relations.items() if s != t},
                kind="canonical relation",
                merge_hint="CanonicalMap(allow_merges=True)",
            )
        validate_vocabulary_is_idempotent(self.vertices, kind="canonical vertex")
        validate_vocabulary_is_idempotent(self.relations, kind="canonical relation")
        for source_class, attr_map in self.properties.items():
            validate_rename_map_is_injective(
                attr_map,
                kind=f"canonical property (class {source_class!r})",
                merge_hint="a transform that combines the fields upstream",
            )
            validate_vocabulary_is_idempotent(
                attr_map, kind=f"canonical property (class {source_class!r})"
            )
        return self

    def canonical_class(self, source_class: str) -> str:
        """Canonical name of *source_class* (itself when unmapped)."""
        return self.vertices.get(source_class, source_class)

    def canonical_relation(self, source_relation: str) -> str:
        """Canonical name of *source_relation* (itself when unmapped)."""
        return self.relations.get(source_relation, source_relation)

    @property
    def vertex_targets(self) -> set[str]:
        """Canonical class names this map establishes (targets of a real rename)."""
        return {t for s, t in self.vertices.items() if s != t}

    @property
    def relation_targets(self) -> set[str]:
        """Canonical relation names this map establishes."""
        return {t for s, t in self.relations.items() if s != t}

    def canonical_property_names(self, canonical_class: str) -> set[str]:
        """Canonical attribute names the map establishes on *canonical_class*."""
        names: set[str] = set()
        for source_class, attr_map in self.properties.items():
            if self.canonical_class(source_class) == canonical_class:
                names.update(new for old, new in attr_map.items() if old != new)
        return names

Attributes

allow_dangling_entries = PydanticField(default=False, description='Accept entries that name nothing in the manifest the map is applied to, dropping and logging each one. A shared vocabulary map is legitimately broader than any single manifest. Off by default, because a misspelt class has exactly the same shape, and dropping it silently narrows the rename to less than the author asked for.') class-attribute instance-attribute

allow_merges = PydanticField(default=False, description='Accept a non-injective ``vertices`` / ``relations`` map. Two sources sharing a canonical target is a *merge*, not a rename; it must be a stated intent because merging fuses entities and can create self-relations.') class-attribute instance-attribute

properties = PydanticField(default_factory=dict, description='Per-source-class attribute rename map: ``{source_class: {source_attr: canonical_attr}}``.') class-attribute instance-attribute

relation_targets property

Canonical relation names this map establishes.

relations = PydanticField(default_factory=dict, description='Relation rename map: ``{source_relation: canonical_relation}``.') class-attribute instance-attribute

vertex_targets property

Canonical class names this map establishes (targets of a real rename).

vertices = PydanticField(default_factory=dict, description='Class rename map: ``{source_class: canonical_class}``.') class-attribute instance-attribute

Methods:

canonical_class(source_class)

Canonical name of source_class (itself when unmapped).

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

def canonical_class(self, source_class: str) -> str:
    """Canonical name of *source_class* (itself when unmapped)."""
    return self.vertices.get(source_class, source_class)

canonical_property_names(canonical_class)

Canonical attribute names the map establishes on canonical_class.

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

def canonical_property_names(self, canonical_class: str) -> set[str]:
    """Canonical attribute names the map establishes on *canonical_class*."""
    names: set[str] = set()
    for source_class, attr_map in self.properties.items():
        if self.canonical_class(source_class) == canonical_class:
            names.update(new for old, new in attr_map.items() if old != new)
    return names

canonical_relation(source_relation)

Canonical name of source_relation (itself when unmapped).

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

def canonical_relation(self, source_relation: str) -> str:
    """Canonical name of *source_relation* (itself when unmapped)."""
    return self.relations.get(source_relation, source_relation)

ClusterResolution dataclass

What merge applies: the resolved clusters and one composite relabel per side.

declared is the folded declared vocabulary as seen from each side; side_maps is the composite — every cluster member onto its merged name, every applicable declared entry as written — one :class:~graflo.architecture.evolution.ops.CanonicalizeOp per side. index includes any cluster merge synthesized.

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

@dataclass(frozen=True)
class ClusterResolution:
    """What merge applies: the resolved clusters and one composite relabel per side.

    ``declared`` is the folded declared vocabulary as seen from each side;
    ``side_maps`` is the composite — every cluster member onto its merged
    name, every applicable declared entry as written — one
    :class:`~graflo.architecture.evolution.ops.CanonicalizeOp` per side.
    ``index`` includes any cluster merge synthesized.
    """

    index: ClusterIndex
    side_maps: SideMaps
    declared: DeclaredMaps

Attributes

declared instance-attribute

index instance-attribute

side_maps instance-attribute

Methods:

__init__(index, side_maps, declared)

Completion dataclass

The extension that would make an incomplete merge declaration consistent.

kind says what to do: extend_cluster — replace one declared cluster by the payload carried here (the same declaration with one more member); declare_equivalences — add the carried declarations to the op (or set name_conflict="union_right", which declares exactly these itself). Payloads are VertexEquivalence / RelationEquivalence documents, so a CLI can print them and an author can paste them.

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

@dataclass(frozen=True)
class Completion:
    """The extension that would make an incomplete merge declaration consistent.

    ``kind`` says what to do: ``extend_cluster`` — replace one declared cluster
    by the payload carried here (the same declaration with one more member);
    ``declare_equivalences`` — add the carried declarations to the op (or set
    ``name_conflict="union_right"``, which declares exactly these itself).
    Payloads are ``VertexEquivalence`` / ``RelationEquivalence`` documents, so
    a CLI can print them and an author can paste them.
    """

    kind: Literal["extend_cluster", "declare_equivalences"]
    side: Side | None = None
    vertex_equivalences: tuple[dict[str, Any], ...] = ()
    relation_equivalences: tuple[dict[str, Any], ...] = ()

    def to_dict(self) -> dict[str, Any]:
        """The completion as a plain document."""
        out: dict[str, Any] = {"kind": self.kind}
        if self.side is not None:
            out["side"] = self.side
        if self.vertex_equivalences:
            out["vertex_equivalences"] = [dict(p) for p in self.vertex_equivalences]
        if self.relation_equivalences:
            out["relation_equivalences"] = [dict(p) for p in self.relation_equivalences]
        return out

Attributes

kind instance-attribute

relation_equivalences = () class-attribute instance-attribute

side = None class-attribute instance-attribute

vertex_equivalences = () class-attribute instance-attribute

Methods:

__init__(kind, side=None, vertex_equivalences=(), relation_equivalences=())

to_dict()

The completion as a plain document.

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

def to_dict(self) -> dict[str, Any]:
    """The completion as a plain document."""
    out: dict[str, Any] = {"kind": self.kind}
    if self.side is not None:
        out["side"] = self.side
    if self.vertex_equivalences:
        out["vertex_equivalences"] = [dict(p) for p in self.vertex_equivalences]
    if self.relation_equivalences:
        out["relation_equivalences"] = [dict(p) for p in self.relation_equivalences]
    return out

DanglingEntry dataclass

A canonical-map entry whose source matches nothing on its side.

The source names no class or relation the side declares, and the entry is none of the four ways an absent source is still meaningful: a cluster member, an already-applied rename, a merged name as the author spelled it, or a both-scoped entry that applies to the other side.

Carried as data rather than refused one at a time, so that authoring a map against a schema of hundreds of classes is not one refusal per mistake.

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

@dataclass(frozen=True)
class DanglingEntry:
    """A canonical-map entry whose source matches nothing on its side.

    The source names no class or relation the side declares, and the entry is
    none of the four ways an absent source is still meaningful: a cluster
    member, an already-applied rename, a merged name as the author spelled
    it, or a ``both``-scoped entry that applies to the other side.

    Carried as data rather than refused one at a time, so that authoring a map
    against a schema of hundreds of classes is not one refusal per mistake.
    """

    side: Side
    kind: Kind | Literal["property"]
    source: str
    target: str | None = None
    suggestion: str = ""

    def describe(self) -> str:
        """The entry as it reads in a refusal, without naming the side.

        The near-miss :attr:`suggestion` is left to the caller to place: it is
        a trailing clause, and only a listing has somewhere to put one.
        """
        if self.kind == "property":
            return f"attribute map for {self.source!r}"
        return f"{self.kind} {self.source!r} -> {self.target!r}"

Attributes

kind instance-attribute

side instance-attribute

source instance-attribute

suggestion = '' class-attribute instance-attribute

target = None class-attribute instance-attribute

Methods:

__init__(side, kind, source, target=None, suggestion='')

describe()

The entry as it reads in a refusal, without naming the side.

The near-miss :attr:suggestion is left to the caller to place: it is a trailing clause, and only a listing has somewhere to put one.

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

def describe(self) -> str:
    """The entry as it reads in a refusal, without naming the side.

    The near-miss :attr:`suggestion` is left to the caller to place: it is
    a trailing clause, and only a listing has somewhere to put one.
    """
    if self.kind == "property":
        return f"attribute map for {self.source!r}"
    return f"{self.kind} {self.source!r} -> {self.target!r}"

DeclaredMaps dataclass

The declared vocabulary as each side sees it.

left / right are the both-scoped map folded under that side's own map; both is kept apart because its entries may legitimately apply to one side only.

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

@dataclass(frozen=True)
class DeclaredMaps:
    """The declared vocabulary as each side sees it.

    ``left`` / ``right`` are the ``both``-scoped map folded under that side's
    own map; ``both`` is kept apart because its entries may legitimately apply
    to one side only.
    """

    left: CanonicalMap
    right: CanonicalMap
    both: CanonicalMap = field(default_factory=CanonicalMap)

    def __getitem__(self, side: Side) -> CanonicalMap:
        return self.left if side == "left" else self.right

Attributes

both = field(default_factory=CanonicalMap) class-attribute instance-attribute

left instance-attribute

right instance-attribute

Methods:

__getitem__(side)

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

def __getitem__(self, side: Side) -> CanonicalMap:
    return self.left if side == "left" else self.right

__init__(left, right, both=CanonicalMap())

MergeCanonicalConflictError

Bases: Refusal

A merge op's clusters and its declared maps contradict each other.

A contradiction (one name, two targets; a fixed point moved), an ambiguity (a canonical name denoting two members), or a dangling entry. The subclass :class:MergeIncompleteError is the one refusal an extension resolves.

check names the rule that refused — the parenthesised phrase in the message — and subjects the names it is about, as :func:~graflo.architecture.evolution.equivalence.subject ids; see :class:.Refusal.

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

class MergeCanonicalConflictError(Refusal):
    """A merge op's clusters and its declared maps contradict each other.

    A contradiction (one name, two targets; a fixed point moved), an ambiguity
    (a canonical name denoting two members), or a dangling entry. The
    subclass :class:`MergeIncompleteError` is the one refusal an extension
    resolves.

    ``check`` names the rule that refused — the parenthesised phrase in the
    message — and ``subjects`` the names it is about, as
    :func:`~graflo.architecture.evolution.equivalence.subject` ids; see
    :class:`.Refusal`.
    """

MergeIncompleteError

Bases: MergeCanonicalConflictError

The declarations are consistent but do not cover a name; an extension would.

Distinct from a contradiction: nothing has to be retracted, something has to be added, and :attr:completion says what.

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

class MergeIncompleteError(MergeCanonicalConflictError):
    """The declarations are consistent but do not cover a name; an extension would.

    Distinct from a contradiction: nothing has to be retracted, something has
    to be added, and :attr:`completion` says what.
    """

    def __init__(
        self,
        message: str,
        completion: Completion,
        *,
        check: str = "",
        subjects: tuple[str, ...] = (),
    ) -> None:
        super().__init__(message, check=check, subjects=subjects)
        self.completion = completion

Attributes

completion = completion instance-attribute

Methods:

__init__(message, completion, *, check='', subjects=())

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

def __init__(
    self,
    message: str,
    completion: Completion,
    *,
    check: str = "",
    subjects: tuple[str, ...] = (),
) -> None:
    super().__init__(message, check=check, subjects=subjects)
    self.completion = completion

SideMaps dataclass

The composite relabel per side: the one op merge applies to each.

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

@dataclass(frozen=True)
class SideMaps:
    """The composite relabel per side: the one op merge applies to each."""

    left: CanonicalizeOp
    right: CanonicalizeOp

    def __getitem__(self, side: Side) -> CanonicalizeOp:
        return self.left if side == "left" else self.right

Attributes

left instance-attribute

right instance-attribute

Methods:

__getitem__(side)

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

def __getitem__(self, side: Side) -> CanonicalizeOp:
    return self.left if side == "left" else self.right

__init__(left, right)

SideNames dataclass

The class and relation names one side actually declares.

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

@dataclass(frozen=True)
class SideNames:
    """The class and relation names one side actually declares."""

    vertices: frozenset[str]
    relations: frozenset[str]

    @classmethod
    def of(cls, manifest: GraphManifest) -> SideNames:
        schema = manifest.graph_schema
        if schema is None:
            return cls(vertices=frozenset(), relations=frozenset())
        return cls(
            vertices=frozenset(schema.core_schema.vertex_config.vertex_set),
            relations=frozenset(
                edge.relation
                for edge in schema.core_schema.edge_config.edges
                if edge.relation is not None
            ),
        )

    def of_kind(self, kind: Kind) -> frozenset[str]:
        return self.vertices if kind == "vertex" else self.relations

Attributes

relations instance-attribute

vertices instance-attribute

Methods:

__init__(vertices, relations)

of(manifest) classmethod

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

@classmethod
def of(cls, manifest: GraphManifest) -> SideNames:
    schema = manifest.graph_schema
    if schema is None:
        return cls(vertices=frozenset(), relations=frozenset())
    return cls(
        vertices=frozenset(schema.core_schema.vertex_config.vertex_set),
        relations=frozenset(
            edge.relation
            for edge in schema.core_schema.edge_config.edges
            if edge.relation is not None
        ),
    )

of_kind(kind)

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

def of_kind(self, kind: Kind) -> frozenset[str]:
    return self.vertices if kind == "vertex" else self.relations

Functions:

canonical_map_to_ops(cm, *, allow_self_relations=False, allow_observation_fusion=False)

Lower a declared map to its single op.

A canonical map is a function on names, and :class:~graflo.architecture.evolution.ops.CanonicalizeOp applies exactly that function in one step: attribute renames (keyed by the source class) first, then classes and relations simultaneously, so no op order can leak into the result. A map with no effective entry lowers to no op at all. A group of more than one class or relation is a merge and is refused unless allow_merges is set.

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

def canonical_map_to_ops(
    cm: CanonicalMap,
    *,
    allow_self_relations: bool = False,
    allow_observation_fusion: bool = False,
) -> list[ManifestOp]:
    """Lower a declared map to its single op.

    A canonical map is a function on names, and
    :class:`~graflo.architecture.evolution.ops.CanonicalizeOp` applies exactly
    that function in one step: attribute renames (keyed by the source class)
    first, then classes and relations simultaneously, so no op order can leak
    into the result. A map with no effective entry lowers to no op at all. A
    group of more than one class or relation is a merge and is refused unless
    ``allow_merges`` is set.
    """
    return canonicalize_ops(
        CanonicalizeOp(
            vertices=dict(cm.vertices),
            properties={cls: dict(attrs) for cls, attrs in cm.properties.items()},
            relations=dict(cm.relations),
            allow_merges=cm.allow_merges,
            allow_self_relations=allow_self_relations,
            allow_observation_fusion=allow_observation_fusion,
        )
    )

canonical_near_collisions(left_names, right_names, *, exempt)

(left, right) pairs that key alike under canonical_slug but differ.

Exact matches are excluded: those are the same-name path, so the two checks can never report one pair twice.

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

def canonical_near_collisions(
    left_names: Iterable[str],
    right_names: Iterable[str],
    *,
    exempt: Collection[str],
) -> list[tuple[str, str]]:
    """``(left, right)`` pairs that key alike under ``canonical_slug`` but differ.

    Exact matches are excluded: those are the same-name path, so the two
    checks can never report one pair twice.
    """
    by_key: dict[str, list[str]] = {}
    for name in left_names:
        by_key.setdefault(canonical_slug(name), []).append(name)
    pairs: list[tuple[str, str]] = []
    for right_name in right_names:
        if right_name in exempt:
            continue
        for left_name in by_key.get(canonical_slug(right_name), []):
            if left_name != right_name:
                pairs.append((left_name, right_name))
    return sorted(set(pairs))

canonicalize_ops(op)

op as the op list to apply: empty when it has no effective entry.

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

def canonicalize_ops(op: CanonicalizeOp) -> list[ManifestOp]:
    """*op* as the op list to apply: empty when it has no effective entry."""
    if not _effective(op.vertices, op.relations, op.properties):
        return []
    return [op]

clusters_to_side_maps(index, *, allow_merges)

Lower every cluster of index into a pair of per-side relabels, and nothing else.

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

def clusters_to_side_maps(index: ClusterIndex, *, allow_merges: bool) -> SideMaps:
    """Lower every cluster of *index* into a pair of per-side relabels, and nothing else."""
    ops: dict[Side, CanonicalizeOp] = {}
    for side in _SIDES:
        vertices, relations, properties = _cluster_maps(index, side)
        ops[side] = CanonicalizeOp(
            vertices=vertices,
            relations=relations,
            properties=properties,
            allow_merges=allow_merges,
        )
    return SideMaps(left=ops["left"], right=ops["right"])

compose_canonical_maps(base, extension)

Partial-function union of two declared maps; a target of either is a fixed point.

Every source named by base or extension maps to exactly one target; a source the two disagree on raises :class:MergeCanonicalConflictError. A target of either map is a fixed point the other may not move, checked in both directions so the result does not depend on which map is base. properties union the same way per source class.

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

def compose_canonical_maps(base: CanonicalMap, extension: CanonicalMap) -> CanonicalMap:
    """Partial-function union of two declared maps; a target of either is a fixed point.

    Every source named by *base* or *extension* maps to exactly one target; a
    source the two disagree on raises :class:`MergeCanonicalConflictError`.
    A target of either map is a fixed point the other may not move, checked
    in both directions so the result does not depend on which map is *base*.
    ``properties`` union the same way per source class.
    """
    vertices = _compose_name_maps(base.vertices, extension.vertices, noun="vertex")
    relations = _compose_name_maps(base.relations, extension.relations, noun="relation")
    properties: dict[str, dict[str, str]] = {
        cls: dict(attrs) for cls, attrs in base.properties.items()
    }
    for source_class, attr_map in extension.properties.items():
        bucket = properties.setdefault(source_class, {})
        for old, new in attr_map.items():
            existing = bucket.get(old)
            if existing is not None and existing != new:
                raise _conflict(
                    "canonical property clash",
                    f"{source_class}.{old} maps to both {existing!r} and {new!r}",
                    "Reconcile the canonical maps.",
                )
            bucket[old] = new
    return CanonicalMap(
        vertices=vertices,
        relations=relations,
        properties=properties,
        allow_merges=base.allow_merges or extension.allow_merges,
        allow_dangling_entries=(
            base.allow_dangling_entries or extension.allow_dangling_entries
        ),
    )

dangling_entries(cm, manifest, *, side='left')

The map's entries that match nothing in manifest, with near-miss candidates.

A canonical map is authored against one manifest long before it is merged against another, and this is that check on its own: no equivalences, no other side, no merge. With no clusters declared there are no merged names and no both scope, so the classification merge uses collapses to its two surviving cases — a source the manifest declares is applicable, a source it does not but whose target it does is already applied — and everything else dangles.

Parameters:

Name	Type	Description	Default
cm	CanonicalMap	The declared map.	required
manifest	GraphManifest	The manifest the map is meant to apply to.	required
side	Side	Which side the map is scoped to; names the entries in the result.	'left'

Returns:

Name	Type	Description
One	DanglingEntry
	...	relations, then attribute maps. Empty means the map applies as written.

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

def dangling_entries(
    cm: CanonicalMap, manifest: GraphManifest, *, side: Side = "left"
) -> tuple[DanglingEntry, ...]:
    """The map's entries that match nothing in *manifest*, with near-miss candidates.

    A canonical map is authored against one manifest long before it is merged
    against another, and this is that check on its own: no equivalences, no
    other side, no merge. With no clusters declared there are no merged
    names and no ``both`` scope, so the classification merge uses collapses
    to its two surviving cases — a source the manifest declares is applicable,
    a source it does not but whose target it does is already applied — and
    everything else dangles.

    Args:
        cm: The declared map.
        manifest: The manifest the map is meant to apply to.
        side: Which side the map is scoped to; names the entries in the result.

    Returns:
        One :class:`DanglingEntry` per unmatched entry, vertices first, then
        relations, then attribute maps. Empty means the map applies as written.
    """
    names = SideNames.of(manifest)
    out: list[DanglingEntry] = []
    for kind, mapping, known in (
        ("vertex", cm.vertices, names.vertices),
        ("relation", cm.relations, names.relations),
    ):
        for source, target in mapping.items():
            if source in known or target in known:
                continue
            out.append(
                DanglingEntry(
                    side=side,
                    kind=kind,  # type: ignore[arg-type]
                    source=source,
                    target=target,
                    suggestion=did_you_mean(source, known),
                )
            )
    for cls in cm.properties:
        if cls in names.vertices or cm.canonical_class(cls) in names.vertices:
            continue
        out.append(
            DanglingEntry(
                side=side,
                kind="property",
                source=cls,
                suggestion=did_you_mean(cls, names.vertices),
            )
        )
    return tuple(out)

fold_declared_maps(op, extra)

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

def fold_declared_maps(
    op: MergeManifestsOp, extra: Sequence[tuple[Side, CanonicalMap]]
) -> DeclaredMaps:
    scoped: dict[str, CanonicalMap] = {
        "left": CanonicalMap(),
        "right": CanonicalMap(),
        "both": CanonicalMap(),
    }
    for scope, cm in op.canonical_maps.items():
        scoped[scope] = compose_canonical_maps(scoped[scope], cm)
    for side, cm in extra:
        scoped[side] = compose_canonical_maps(scoped[side], cm)
    return DeclaredMaps(
        left=compose_canonical_maps(scoped["both"], scoped["left"]),
        right=compose_canonical_maps(scoped["both"], scoped["right"]),
        both=scoped["both"],
    )

resolve_clusters(op, *, left, right, canonical_maps=())

Resolve op's clusters against its declared maps and build the per-side composite.

canonical_maps are extra (side, map) pairs folded into op.canonical_maps. left / right are the manifests about to be merged, in whatever vocabulary they are in: a declared entry whose source is absent on its side but whose target is present is satisfied and is a no-op; one matching nothing is refused as a typo.

A name both sides carry after their composite maps, and no cluster merges, is what op.name_conflict decides: error refuses it as incomplete, naming the equivalences to declare; union_right declares them itself (a synthesized cluster, so the union goes through the same identity and property reconciliation as a declared one — two spellings of one name, canonical_slug alike, are one such cluster under the left spelling); prefix_right leaves them to merge to keep apart.

Raises :class:~graflo.architecture.evolution.equivalence.ClusterConflictError when the declared clusters themselves conflict, :class:MergeCanonicalConflictError when a map and the equivalences disagree — a member the map sends elsewhere than the merged name, a canonical class or attribute re-targeted by a cluster, a cluster with no name, a dangling entry, a property equivalence naming an absent or colliding field — and :class:MergeIncompleteError when an extension would resolve it: a map entry sending a non-member onto a merged name, or a shared name under name_conflict="error".

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

def resolve_clusters(
    op: MergeManifestsOp,
    *,
    left: GraphManifest,
    right: GraphManifest,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
) -> ClusterResolution:
    """Resolve *op*'s clusters against its declared maps and build the per-side composite.

    *canonical_maps* are extra ``(side, map)`` pairs folded into
    ``op.canonical_maps``. *left* / *right* are the manifests about to be
    merged, in whatever vocabulary they are in: a declared entry whose
    source is absent on its side but whose target is present is satisfied and
    is a no-op; one matching nothing is refused as a typo.

    A name both sides carry after their composite maps, and no cluster
    merges, is what ``op.name_conflict`` decides: ``error`` refuses it as
    incomplete, naming the equivalences to declare; ``union_right`` declares
    them itself (a **synthesized** cluster, so the union goes through the
    same identity and property reconciliation as a declared one — two
    spellings of one name, ``canonical_slug`` alike, are one such cluster
    under the left spelling); ``prefix_right`` leaves them to merge to keep
    apart.

    Raises :class:`~graflo.architecture.evolution.equivalence.ClusterConflictError`
    when the declared clusters themselves conflict,
    :class:`MergeCanonicalConflictError` when a map and the equivalences
    disagree — a member the map sends elsewhere than the merged name, a
    canonical class or attribute re-targeted by a cluster, a cluster with no
    name, a dangling entry, a property equivalence naming an absent or
    colliding field — and :class:`MergeIncompleteError` when an extension
    would resolve it: a map entry sending a non-member onto a merged name,
    or a shared name under ``name_conflict="error"``.
    """
    declared = fold_declared_maps(op, canonical_maps)
    names: dict[Side, SideNames] = {
        "left": SideNames.of(left),
        "right": SideNames.of(right),
    }
    manifests: dict[Side, GraphManifest] = {"left": left, "right": right}
    resolution = _resolve(
        op,
        declared=declared,
        names=names,
        manifests=manifests,
        synthesized_from=(len(op.vertex_equivalences), len(op.relation_equivalences)),
    )
    if op.name_conflict == "prefix_right":
        return resolution

    near = op.name_conflict == "union_right"
    vertex_groups = same_name_groups(resolution, names, kind="vertex", near=near)
    relation_groups = same_name_groups(resolution, names, kind="relation", near=near)
    if not vertex_groups and not relation_groups:
        return resolution

    if op.name_conflict == "error":
        kind: Kind = "vertex" if vertex_groups else "relation"
        shared = [into for into, _l, _r in (vertex_groups or relation_groups)]
        raise _incomplete(
            f"{kind} name collision",
            f"{shared} exist on both sides and no equivalence merges them",
            "Declare the equivalences the completion carries, set "
            "name_conflict='union_right' to union by name, or "
            "name_conflict='prefix_right' to keep them apart.",
            Completion(
                kind="declare_equivalences",
                vertex_equivalences=_same_name_payloads(vertex_groups),
                relation_equivalences=_same_name_payloads(relation_groups),
            ),
            subjects=tuple(subject("merged", name) for name in shared),
        )

    nary = any(
        len(l_members) > 1 or len(r_members) > 1
        for _into, l_members, r_members in (*vertex_groups, *relation_groups)
    )
    extended = op.model_copy(
        update={
            "vertex_equivalences": [
                *op.vertex_equivalences,
                *(
                    VertexEquivalence.model_validate(payload)
                    for payload in _same_name_payloads(vertex_groups)
                ),
            ],
            "relation_equivalences": [
                *op.relation_equivalences,
                *(
                    RelationEquivalence.model_validate(payload)
                    for payload in _same_name_payloads(relation_groups)
                ),
            ],
            "allow_merges": op.allow_merges or nary,
        }
    )
    return _resolve(
        extended,
        declared=declared,
        names=names,
        manifests=manifests,
        synthesized_from=(len(op.vertex_equivalences), len(op.relation_equivalences)),
    )

same_name_groups(resolution, names, *, kind, near)

(merged name, left members, right members) for every name no cluster covers.

Names are compared after each side's composite map, since that is what the union sees. With near, two spellings of one name (canonical_slug alike) form one group too, merged under the left spelling.

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

def same_name_groups(
    resolution: ClusterResolution,
    names: Mapping[Side, SideNames],
    *,
    kind: Kind,
    near: bool,
) -> list[tuple[str, list[str], list[str]]]:
    """``(merged name, left members, right members)`` for every name no cluster covers.

    Names are compared after each side's composite map, since that is what
    the union sees. With *near*, two spellings of one name (``canonical_slug``
    alike) form one group too, merged under the left spelling.
    """
    index = resolution.index
    merged = index.labels if kind == "vertex" else index.relation_labels
    pre = {
        side: _preimages(
            _mapping(resolution.side_maps[side], kind), names[side].of_kind(kind)
        )
        for side in _SIDES
    }
    groups: list[tuple[str, list[str], list[str]]] = []
    if near:
        by_key: dict[str, dict[Side, list[str]]] = {}
        for side in _SIDES:
            for post_name in pre[side]:
                if post_name in merged:
                    continue
                by_key.setdefault(canonical_slug(post_name), {}).setdefault(
                    side, []
                ).append(post_name)
        for _key, sides in sorted(by_key.items()):
            left_post = sorted(sides.get("left", []))
            right_post = sorted(sides.get("right", []))
            if not left_post or not right_post:
                continue
            exact = sorted(set(left_post) & set(right_post))
            into = exact[0] if exact else left_post[0]
            groups.append(
                (
                    into,
                    [m for n in left_post for m in pre["left"][n]],
                    [m for n in right_post for m in pre["right"][n]],
                )
            )
        return groups
    for post_name in sorted(set(pre["left"]) & set(pre["right"])):
        if post_name in merged:
            continue
        groups.append((post_name, pre["left"][post_name], pre["right"][post_name]))
    return groups

trim_canonical_map(cm, manifest, *, side='left')

cm with its entries for manifest only, and the entries dropped.

Trimming is an authoring step, not something merge does on its own: the result is a value to inspect and save, so that a map narrowed to a manifest is a change with a diff rather than a silent omission at merge time. A map that is deliberately broader than any one manifest is better served by allow_dangling_entries, which keeps the map whole.

Returns:

Type	Description
CanonicalMap	The trimmed map and the entries removed, as
tuple[DanglingEntry, ...]

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

def trim_canonical_map(
    cm: CanonicalMap, manifest: GraphManifest, *, side: Side = "left"
) -> tuple[CanonicalMap, tuple[DanglingEntry, ...]]:
    """*cm* with its entries for *manifest* only, and the entries dropped.

    Trimming is an authoring step, not something merge does on its own: the
    result is a value to inspect and save, so that a map narrowed to a manifest
    is a change with a diff rather than a silent omission at merge time. A
    map that is deliberately broader than any one manifest is better served by
    ``allow_dangling_entries``, which keeps the map whole.

    Returns:
        The trimmed map and the entries removed, as
        :func:`dangling_entries` reports them.
    """
    dropped = dangling_entries(cm, manifest, side=side)
    if not dropped:
        return cm, ()
    gone = {(entry.kind, entry.source) for entry in dropped}
    return (
        cm.model_copy(
            update={
                "vertices": {
                    s: t for s, t in cm.vertices.items() if ("vertex", s) not in gone
                },
                "relations": {
                    s: t for s, t in cm.relations.items() if ("relation", s) not in gone
                },
                "properties": {
                    c: dict(a)
                    for c, a in cm.properties.items()
                    if ("property", c) not in gone
                },
            }
        ),
        dropped,
    )

validate_and_complete_canonical_map(op, *, left, right, canonical_maps=())

Validate op against its declared maps and return the completed per-side relabels.

The merged name of every cluster is completed — from into, the declared map, or the members' shared spelling — and every member maps onto it; the declared map's remaining entries are carried as written. Apply the result to each side with :func:canonicalize_ops before the schema/resource union, which is what :func:~graflo.architecture.evolution.merge.merge_manifests does.

Raises :class:MergeCanonicalConflictError for every refusal of :func:resolve_clusters, wrapping a :class:~graflo.architecture.evolution.equivalence.ClusterConflictError when the declared clusters themselves conflict.

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

def validate_and_complete_canonical_map(
    op: MergeManifestsOp,
    *,
    left: GraphManifest,
    right: GraphManifest,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
) -> SideMaps:
    """Validate *op* against its declared maps and return the completed per-side relabels.

    The merged name of every cluster is completed — from ``into``, the
    declared map, or the members' shared spelling — and every member maps
    onto it; the declared map's remaining entries are carried as written.
    Apply the result to each side with :func:`canonicalize_ops` before the
    schema/resource union, which is what
    :func:`~graflo.architecture.evolution.merge.merge_manifests` does.

    Raises :class:`MergeCanonicalConflictError` for every refusal of
    :func:`resolve_clusters`, wrapping a
    :class:`~graflo.architecture.evolution.equivalence.ClusterConflictError`
    when the declared clusters themselves conflict.
    """
    try:
        return resolve_clusters(
            op, left=left, right=right, canonical_maps=canonical_maps
        ).side_maps
    except ClusterConflictError as exc:
        raise MergeCanonicalConflictError(
            f"merge contradicts the canonical map (cluster conflict): {exc}",
            check=exc.check or "cluster conflict",
            subjects=exc.subjects,
        ) from exc