graflo.architecture.evolution.merge_core

Pure merge helpers for logical vertices and edges.

Attributes

Classes

EdgeMergeError

Bases: Refusal

Two declarations of one logical edge that disagree on what it is.

edge_id leaves type and by out so two sources can describe one logical edge, which is what makes the disagreement expressible at all.

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

class EdgeMergeError(Refusal):
    """Two declarations of one logical edge that disagree on what it is.

    ``edge_id`` leaves ``type`` and ``by`` out so two sources can describe one
    logical edge, which is what makes the disagreement expressible at all.
    """

    def __init__(self, message: str, *, check: str, edge_id: EdgeId) -> None:
        super().__init__(message, check=check)
        self.edge_id = edge_id

Attributes

edge_id = edge_id instance-attribute

Methods:

__init__(message, *, check, edge_id)

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

def __init__(self, message: str, *, check: str, edge_id: EdgeId) -> None:
    super().__init__(message, check=check)
    self.edge_id = edge_id

VertexMergeError

Bases: Refusal

Two vertex declarations that cannot be unioned into one.

Identity is the subject of every rule here. The four modes are mutually exclusive by construction, a funnel's branch order is its key, and a secondary identity's name is what an edge step selects by -- so none of them has a weaker-wins ordering the union could apply on the author's behalf.

into_name is the merged vertex the union was heading for, and properties the offending names where the rule has them (the hash properties, a secondary identity's field-set). subjects is empty: this layer has no notion of left, right or merged -- it is reached from merge, from merge_vertices and from a per-side canonicalize alike -- so the caller that knows the scope builds the ids.

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

class VertexMergeError(Refusal):
    """Two vertex declarations that cannot be unioned into one.

    Identity is the subject of every rule here. The four modes are mutually
    exclusive by construction, a funnel's branch order *is* its key, and a
    secondary identity's name is what an edge step selects by -- so none of
    them has a weaker-wins ordering the union could apply on the author's
    behalf.

    ``into_name`` is the merged vertex the union was heading for, and
    ``properties`` the offending names where the rule has them (the hash
    properties, a secondary identity's field-set). ``subjects`` is empty: this
    layer has no notion of left, right or merged -- it is reached from
    merge, from ``merge_vertices`` and from a per-side canonicalize alike --
    so the caller that knows the scope builds the ids.
    """

    def __init__(
        self,
        message: str,
        *,
        check: str,
        into_name: str,
        properties: tuple[str, ...] = (),
    ) -> None:
        super().__init__(message, check=check)
        self.into_name = into_name
        self.properties = properties

Attributes

into_name = into_name instance-attribute

properties = properties instance-attribute

Methods:

__init__(message, *, check, into_name, properties=())

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

def __init__(
    self,
    message: str,
    *,
    check: str,
    into_name: str,
    properties: tuple[str, ...] = (),
) -> None:
    super().__init__(message, check=check)
    self.into_name = into_name
    self.properties = properties

Functions:

merge_edge_pair(a, b)

Merge two edges with the same :attr:~graflo.architecture.schema.edge.Edge.edge_id.

type / by must agree: edge_id leaves them out so two sources can describe one logical edge, but a DIRECT edge and an INDIRECT one via some vertex are different physical things with no weaker-wins ordering between them (unlike directed), so disagreement raises rather than keeping one side's silently.

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

def merge_edge_pair(a: Edge, b: Edge) -> Edge:
    """Merge two edges with the same :attr:`~graflo.architecture.schema.edge.Edge.edge_id`.

    ``type`` / ``by`` must agree: ``edge_id`` leaves them out so two sources can
    describe one logical edge, but a ``DIRECT`` edge and an ``INDIRECT`` one via
    some vertex are different physical things with no weaker-wins ordering
    between them (unlike ``directed``), so disagreement raises rather than
    keeping one side's silently.
    """
    if (a.type, a.by) != (b.type, b.by):
        raise EdgeMergeError(
            f"Cannot merge edge {a.edge_id!r}: sources disagree on type/by "
            f"({a.type!r}, {a.by!r}) vs ({b.type!r}, {b.by!r})",
            check="edge type disagreement",
            edge_id=a.edge_id,
        )
    props = union_field_lists(a.properties + b.properties, owner=f"edge {a.edge_id!r}")

    identities_out: list[list[str]] = []
    seen_identities: set[tuple[str, ...]] = set()
    for identity in a.identities + b.identities:
        t = tuple(identity)
        if t not in seen_identities:
            seen_identities.add(t)
            identities_out.append(list(identity))

    descriptions = [a.description, b.description]
    descriptions = [d for d in descriptions if d]
    desc_out: str | None = None
    if len(descriptions) == 1:
        desc_out = descriptions[0]
    elif len(descriptions) > 1:
        desc_out = " / ".join(descriptions)

    return Edge(
        source=a.source,
        target=a.target,
        relation=a.relation,
        description=desc_out,
        identities=identities_out,
        properties=props,
        type=a.type,
        by=a.by,
        # Undirected wins: it is the weaker assertion, and treating a merged
        # undirected edge as directed would let AddInverseEdgesOp synthesize an
        # inverse that duplicates it.
        directed=a.directed and b.directed,
        semantics=merge_semantics(a.semantics, b.semantics),
    )

merge_vertex_models(vertices, into_name)

Merge vertex definitions into a single :class:Vertex.

Identity mode is carried through the merge: blank / assigned propagate when any source declares them, and hash_identity_properties / secondary_identities are unioned. The mutual exclusions enforced by :meth:Vertex.set_identity are checked here so the failure names the merge rather than surfacing from pydantic.

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

def merge_vertex_models(vertices: list[Vertex], into_name: str) -> Vertex:
    """Merge vertex definitions into a single :class:`Vertex`.

    Identity mode is carried through the merge: ``blank`` / ``assigned`` propagate when
    any source declares them, and ``hash_identity_properties`` / ``secondary_identities``
    are unioned. The mutual exclusions enforced by :meth:`Vertex.set_identity` are
    checked here so the failure names the merge rather than surfacing from pydantic.
    """
    if not vertices:
        raise ValueError("merge_vertex_models requires at least one vertex")

    props = union_field_lists(
        (f for v in vertices for f in v.properties), owner=f"vertex {into_name!r}"
    )

    identity_out: list[str] = []
    seen_id: set[str] = set()
    for v in vertices:
        for x in v.identity:
            if x not in seen_id:
                identity_out.append(x)
                seen_id.add(x)

    # Deduplicated like every other list field; merge runs this merge twice
    # (per side, then at union), so a repeated filter would otherwise compound.
    filters_out: list[FilterExpression] = []
    seen_filters: set[str] = set()
    for v in vertices:
        for f in v.filters:
            key = json.dumps(
                f.to_dict(skip_defaults=False), sort_keys=True, default=str
            )
            if key not in seen_filters:
                seen_filters.add(key)
                filters_out.append(f)

    descriptions = [v.description for v in vertices if v.description]
    if not descriptions:
        desc_out: str | None = None
    elif len(descriptions) == 1:
        desc_out = descriptions[0]
    else:
        desc_out = " / ".join(descriptions)

    blank_out = any(v.blank for v in vertices)
    assigned_out = any(v.assigned for v in vertices)
    if blank_out and assigned_out:
        raise VertexMergeError(
            f"Cannot merge into vertex '{into_name}': sources mix blank and assigned "
            "identity modes, which are mutually exclusive",
            check="vertex identity mode conflict",
            into_name=into_name,
        )

    hash_out: list[str] = []
    seen_hash: set[str] = set()
    for v in vertices:
        for name in v.hash_identity_properties:
            if name not in seen_hash:
                hash_out.append(name)
                seen_hash.add(name)
    if assigned_out and hash_out:
        raise VertexMergeError(
            f"Cannot merge into vertex '{into_name}': an assigned source cannot be "
            f"merged with hash-identity sources (hash properties: {hash_out})",
            check="vertex identity mode conflict",
            into_name=into_name,
            properties=tuple(hash_out),
        )
    if blank_out and hash_out:
        # The one pair neither this kernel nor ``Vertex.set_identity`` used to
        # refuse. It validates, and ``identity_mode`` reads ``blank`` first, so
        # the merged vertex keys on a generated id and the declared digest is
        # never consulted -- rows that should have deduplicated silently do not.
        raise VertexMergeError(
            f"Cannot merge into vertex '{into_name}': a blank source cannot be "
            f"merged with hash-identity sources (hash properties: {hash_out}). A "
            "blank vertex keys on a generated id, so the digest would never be "
            "consulted.",
            check="vertex identity mode conflict",
            into_name=into_name,
            properties=tuple(hash_out),
        )

    funnel_out = _merge_identity_funnels(vertices, into_name)
    if funnel_out is not None:
        if hash_out:
            raise VertexMergeError(
                f"Cannot merge into vertex '{into_name}': sources mix an identity "
                f"funnel with flat hash properties {hash_out}. Express the flat key "
                "as a funnel branch first, then merge.",
                check="vertex identity funnel conflict",
                into_name=into_name,
                properties=tuple(hash_out),
            )
        if assigned_out or blank_out:
            raise VertexMergeError(
                f"Cannot merge into vertex '{into_name}': an identity funnel cannot "
                "be merged with assigned or blank sources",
                check="vertex identity funnel conflict",
                into_name=into_name,
            )

    secondary_out = _union_secondary_identities(vertices, into_name, identity_out)
    if blank_out and secondary_out:
        raise VertexMergeError(
            f"Cannot merge into vertex '{into_name}': a blank source cannot be merged "
            "with sources declaring secondary_identities — a blank vertex has no "
            "source-visible key to match on",
            check="vertex secondary identity conflict",
            into_name=into_name,
            properties=tuple(
                sorted({f for entry in secondary_out for f in entry.fields})
            ),
        )

    return Vertex(
        name=into_name,
        properties=props,
        identity=identity_out,
        filters=filters_out,
        description=desc_out,
        blank=blank_out,
        assigned=assigned_out,
        hash_identity_properties=hash_out,
        identity_funnel=funnel_out,
        secondary_identities=secondary_out,
        semantics=reduce(merge_semantics, (v.semantics for v in vertices), None),
    )

redirect_and_merge_edges(edges, mapping)

Apply vertex mapping to endpoints, then merge duplicate edge identities.

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

def redirect_and_merge_edges(edges: list[Edge], mapping: dict[str, str]) -> list[Edge]:
    """Apply vertex *mapping* to endpoints, then merge duplicate edge identities."""

    def _map_endpoint(n: str) -> str:
        return mapping.get(n, n)

    redirected: list[Edge] = []
    for e in edges:
        redirected.append(
            e.model_copy(
                update={
                    "source": _map_endpoint(e.source),
                    "target": _map_endpoint(e.target),
                }
            )
        )

    by_id: dict[EdgeId, Edge] = {}
    for e in redirected:
        eid = e.edge_id
        if eid not in by_id:
            by_id[eid] = e
        else:
            by_id[eid] = merge_edge_pair(by_id[eid], e)
    return list(by_id.values())

remap_relation_and_merge_edges(edges, relation_map)

Remap edge relation names and merge duplicate edge identities.

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

def remap_relation_and_merge_edges(
    edges: list[Edge], relation_map: dict[str, str]
) -> list[Edge]:
    """Remap edge relation names and merge duplicate edge identities."""
    if not relation_map:
        return list(edges)
    remapped = [
        edge.model_copy(
            update={"relation": relation_map.get(edge.relation, edge.relation)}
        )
        if edge.relation is not None
        else edge
        for edge in edges
    ]
    by_id: dict[EdgeId, Edge] = {}
    for edge in remapped:
        if edge.edge_id in by_id:
            by_id[edge.edge_id] = merge_edge_pair(by_id[edge.edge_id], edge)
            continue
        by_id[edge.edge_id] = edge
    return list(by_id.values())