graflo.architecture.evolution.alignment

Identity alignment: merge an equivalence identity from fundamental ops.

An :class:IdentityAlignment states, for one canonical class, which canonical attributes carry cross-source entity equivalence and how each resource derives them. It is a composer, not a mechanism: :func:alignment_to_ops emits only fundamental ops —

1. AddVertexPropertiesOp — declare the canonical attributes on the class;
2. AddResourceTransformsOp — per-resource derivation steps (gating, normalization, local-key namespacing) appended to the pipelines;
3. ReplaceIdentityOp — a priority funnel over the canonical attributes, in declared order, with the namespaced local_key as the last branch;
4. AddSecondaryIdentitiesOp — the retired side keys as lookup-only secondary identities.

The division of labor is deliberate: a primary identity is a property of the class, so the funnel references only canonical attributes; how a given source populates them is resource knowledge and lives in that resource's pipeline. Derivation inputs are RAW source-doc field names — property renames rewrite vertex.from maps so documents keep their original keys, and transform.call.input is never rewritten.

The member is the unit of derivation. Every record that becomes the canonical class was produced as one member of the equivalence cluster by one resource — by a vertex: Shop step, or by a vertex_router key whose value was Shop. When a resource produces several members, its derivations may be keyed by member; the lowering then reads the side manifest (the merge has already rewritten router type_map values to the canonical name, so the union no longer knows which key was which member) to learn how the resource produces each member, and guards the step with when on the router's discriminator. A guarded step that does not fire writes nothing, so each member's derivation is the single writer of the attribute for its own documents.

A derivation that is not keyed by member is guarded the same way whenever a router produces the class: when admits the discriminator values that route onto it — the type_map keys mapping to it, or its own name for pass-through — so the step runs for no other class's documents. Only a level where a plain vertex step also produces the class, or whose routers read different discriminators, lowers unguarded; there a sibling class declaring a canonical attribute name is refused, since the router would hand it the derived value.

Attributes

AlignmentRow = AlignmentAttribute module-attribute

ClusterMembers = Mapping[str, Collection[str]] module-attribute

MemberProductions = dict[str, dict[str, _MemberProduction]] module-attribute

SideManifests = Mapping[str, GraphManifest] module-attribute

VocabularyMap = CanonicalMap | CanonicalizeOp module-attribute

__all__ = ['AlignmentAttribute', 'AlignmentConflictError', 'AlignmentRow', 'DerivationSpec', 'IdentityAlignment', 'LocalKeySource', 'LocalKeySpec', 'SharedDerivation', 'alignment_to_ops', 'validate_alignment'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

AlignmentAttribute

Bases: ConfigBaseModel

One aligned canonical attribute; list position = funnel priority.

Each entry lowers to one :class:~graflo.architecture.schema.identity_funnel.IdentityBranch over name, so the list order is the funnel order. name is what the derived attribute is called — nothing collapses onto it, which is why it is not into; into is accepted as a legacy alias.

sources is keyed by resource because derivation inputs are that resource's raw column names. An entry takes one of three shapes:

• a single :class:DerivationSpec — the resource produces one member of the cluster, or all its members share the key column;
• a list of specs — the resource produces several members, each with its own key column, and at most one spec yields a value for any document (the others read an empty column). Lowers to scratch fields plus a coalesce_fields step;
• a dict keyed by member class — the resource produces several members and which one a document is must decide the derivation (the members share a column, or each carries its own marker). The lowering asks the side manifest how the resource produces each member and guards the step accordingly (when on the router's discriminator, or nothing for a plain vertex step). A member is keyed by its own name on its side or, through merge_manifests, by its canonical name;
• a :class:SharedDerivation — the same dict, spelled once: one call shared by the listed members, with only the parameters that differ.

Behind a vertex_router the first two shapes are guarded as well: the lowering reads which discriminator values route onto the class and puts them in when, so the steps run for no other class's documents.

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

class AlignmentAttribute(ConfigBaseModel):
    """One aligned canonical attribute; list position = funnel priority.

    Each entry lowers to one
    :class:`~graflo.architecture.schema.identity_funnel.IdentityBranch` over
    ``name``, so the list order *is* the funnel order. ``name`` is what the
    derived attribute is called — nothing collapses onto it, which is why it is
    not ``into``; ``into`` is accepted as a legacy alias.

    ``sources`` is keyed by resource because derivation inputs are that
    resource's raw column names. An entry takes one of three shapes:

    * a single :class:`DerivationSpec` — the resource produces one member of
      the cluster, or all its members share the key column;
    * a **list** of specs — the resource produces several members, each with
      its **own key column**, and at most one spec yields a value for any
      document (the others read an empty column). Lowers to scratch fields
      plus a ``coalesce_fields`` step;
    * a **dict keyed by member class** — the resource produces several members
      and which one a document *is* must decide the derivation (the members
      share a column, or each carries its own marker). The lowering asks the
      side manifest how the resource produces each member and guards the step
      accordingly (``when`` on the router's discriminator, or nothing for a
      plain ``vertex`` step). A member is keyed by its own name on its
      side or, through ``merge_manifests``, by its canonical name;
    * a :class:`SharedDerivation` — the same dict, spelled once: one call
      shared by the listed members, with only the parameters that differ.

    Behind a ``vertex_router`` the first two shapes are guarded as well: the
    lowering reads which discriminator values route onto the class and puts
    them in ``when``, so the steps run for no other class's documents.
    """

    name: str = PydanticField(
        ...,
        validation_alias=AliasChoices("name", "into"),
        description=(
            "Canonical attribute name on the class; funnel branch id. ``into`` "
            "is accepted as a legacy alias."
        ),
    )
    sources: dict[
        str,
        DerivationSpec
        | SharedDerivation
        | list[DerivationSpec]
        | dict[str, DerivationSpec],
    ] = PydanticField(
        ...,
        min_length=1,
        description=(
            "Per-resource derivation: ``{resource: spec}``; ``{resource: "
            "[spec, ...]}`` when the members the resource produces carry "
            "different key columns; ``{resource: {member_class: spec}}`` when "
            "the member a document is must decide the derivation, or a "
            "``SharedDerivation`` spelling that dict once."
        ),
    )

    def _by_member(self, resource: str) -> dict[str, DerivationSpec] | None:
        spec = self.sources.get(resource)
        if isinstance(spec, SharedDerivation):
            return spec.expand()
        return spec if isinstance(spec, dict) else None

    def specs_for(self, resource: str) -> list[DerivationSpec]:
        """Derivations *resource* contributes to this attribute, in order."""
        spec = self.sources.get(resource)
        if spec is None:
            return []
        if isinstance(spec, DerivationSpec):
            return [spec]
        if isinstance(spec, list):
            return list(spec)
        by_member = spec.expand() if isinstance(spec, SharedDerivation) else spec
        return list(by_member.values())

    def members_for(self, resource: str) -> list[str] | None:
        """Member classes keying *resource*'s specs, or ``None`` if unkeyed."""
        by_member = self._by_member(resource)
        return list(by_member) if by_member is not None else None

Attributes

name = PydanticField(..., validation_alias=AliasChoices('name', 'into'), description='Canonical attribute name on the class; funnel branch id. ``into`` is accepted as a legacy alias.') class-attribute instance-attribute

sources = PydanticField(..., min_length=1, description='Per-resource derivation: ``{resource: spec}``; ``{resource: [spec, ...]}`` when the members the resource produces carry different key columns; ``{resource: {member_class: spec}}`` when the member a document is must decide the derivation, or a ``SharedDerivation`` spelling that dict once.') class-attribute instance-attribute

Methods:

members_for(resource)

Member classes keying resource's specs, or None if unkeyed.

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

def members_for(self, resource: str) -> list[str] | None:
    """Member classes keying *resource*'s specs, or ``None`` if unkeyed."""
    by_member = self._by_member(resource)
    return list(by_member) if by_member is not None else None

specs_for(resource)

Derivations resource contributes to this attribute, in order.

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

def specs_for(self, resource: str) -> list[DerivationSpec]:
    """Derivations *resource* contributes to this attribute, in order."""
    spec = self.sources.get(resource)
    if spec is None:
        return []
    if isinstance(spec, DerivationSpec):
        return [spec]
    if isinstance(spec, list):
        return list(spec)
    by_member = spec.expand() if isinstance(spec, SharedDerivation) else spec
    return list(by_member.values())

AlignmentConflictError

Bases: ValueError

An identity alignment contradicts the union manifest or canonical maps.

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

class AlignmentConflictError(ValueError):
    """An identity alignment contradicts the union manifest or canonical maps."""

DerivationSpec

Bases: ConfigBaseModel

How one resource derives a canonical attribute from its raw doc fields.

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

class DerivationSpec(ConfigBaseModel):
    """How one resource derives a canonical attribute from its raw doc fields."""

    input: list[str] = PydanticField(
        ...,
        min_length=1,
        description=(
            "RAW source-doc field names fed to the function, in order. "
            "Documents keep their original keys after property renames, so "
            "canonical property names are usually wrong here."
        ),
    )
    module: str = PydanticField(
        default="graflo.util.transform",
        description="Module holding the derivation function.",
    )
    foo: str = PydanticField(
        default="gated_normalized_key",
        description="Function name; called as ``foo(*values, **params)``.",
    )
    params: dict[str, Any] = PydanticField(
        default_factory=dict,
        description="Keyword parameters for the function (gate prefix, ...).",
    )

Attributes

foo = PydanticField(default='gated_normalized_key', description='Function name; called as ``foo(*values, **params)``.') class-attribute instance-attribute

input = PydanticField(..., min_length=1, description='RAW source-doc field names fed to the function, in order. Documents keep their original keys after property renames, so canonical property names are usually wrong here.') class-attribute instance-attribute

module = PydanticField(default='graflo.util.transform', description='Module holding the derivation function.') class-attribute instance-attribute

params = PydanticField(default_factory=dict, description='Keyword parameters for the function (gate prefix, ...).') class-attribute instance-attribute

IdentityAlignment

Bases: ConfigBaseModel

Cross-source identity alignment for one canonical class.

attributes order is funnel priority: a record keys by the highest-priority aligned attribute it carries. Two records fuse when their strongest present attribute coincides — a match on a lower-priority attribute does NOT fuse records when one of them also carries a higher-priority one.

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

class IdentityAlignment(ConfigBaseModel):
    """Cross-source identity alignment for one canonical class.

    ``attributes`` order is funnel priority: a record keys by the highest-priority
    aligned attribute it carries. Two records fuse when their strongest
    present attribute coincides — a match on a lower-priority attribute does
    NOT fuse records when one of them also carries a higher-priority one.
    """

    vertex: str = PydanticField(
        ...,
        description="The canonical class whose identity is being aligned.",
    )
    attributes: list[AlignmentAttribute] = PydanticField(
        default_factory=list,
        validation_alias=AliasChoices("attributes", "rows"),
        description=(
            "Aligned canonical attributes, in priority order. ``rows`` is "
            "accepted as a legacy alias."
        ),
    )
    local_key: LocalKeySpec | None = PydanticField(
        default=None,
        description=(
            "Fallback identity for records carrying no aligned attribute. "
            "Without it such records get no identity and are dropped."
        ),
    )
    secondary_identities: dict[str, list[str]] = PydanticField(
        default_factory=dict,
        description=(
            "Retired side keys kept as lookup-only secondary identities: "
            "``{name: [field, ...]}``."
        ),
    )
    at: dict[str, list[int]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-resource pipeline level to derive at, as ``descend`` step "
            "indices. Omitted resources resolve to the single level producing "
            "``vertex`` (for member-keyed sources: the level producing the "
            "member on its side); supply a path only when a resource produces "
            "it at more than one level."
        ),
    )

    @model_validator(mode="after")
    def _validate_shape(self) -> IdentityAlignment:
        if not self.attributes and self.local_key is None:
            raise ValueError(
                "IdentityAlignment requires at least one attribute or a local_key"
            )
        into_names = [attribute.name for attribute in self.attributes]
        if self.local_key is not None:
            into_names.append(self.local_key.name)
        duplicates = {n for n in into_names if into_names.count(n) > 1}
        if duplicates:
            raise ValueError(
                f"IdentityAlignment: duplicate target attributes {sorted(duplicates)}"
            )
        return self

Attributes

at = PydanticField(default_factory=dict, description='Per-resource pipeline level to derive at, as ``descend`` step indices. Omitted resources resolve to the single level producing ``vertex`` (for member-keyed sources: the level producing the member on its side); supply a path only when a resource produces it at more than one level.') class-attribute instance-attribute

attributes = PydanticField(default_factory=list, validation_alias=AliasChoices('attributes', 'rows'), description='Aligned canonical attributes, in priority order. ``rows`` is accepted as a legacy alias.') class-attribute instance-attribute

local_key = PydanticField(default=None, description='Fallback identity for records carrying no aligned attribute. Without it such records get no identity and are dropped.') class-attribute instance-attribute

secondary_identities = PydanticField(default_factory=dict, description='Retired side keys kept as lookup-only secondary identities: ``{name: [field, ...]}``.') class-attribute instance-attribute

vertex = PydanticField(..., description='The canonical class whose identity is being aligned.') class-attribute instance-attribute

LocalKeySource

Bases: ConfigBaseModel

Where one resource's side-local key comes from, and its namespace tag.

The tag is what keeps records of different sources apart once they fail to fuse: f2 from one source and f2 from another are different entities, and a:f2 / b:f2 say so. It is required so that opting out is a statement, not an omission: tag=None (stored as "", the neutral element, so it survives serialization) keeps the raw value as the local key with no separator — the author's claim that the values are already unique across every source of the class (UUIDs, IRIs, ids the source itself prefixes).

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

class LocalKeySource(ConfigBaseModel):
    """Where one resource's side-local key comes from, and its namespace tag.

    The tag is what keeps records of different sources apart once they fail to
    fuse: ``f2`` from one source and ``f2`` from another are different
    entities, and ``a:f2`` / ``b:f2`` say so. It is required so that opting
    out is a statement, not an omission: ``tag=None`` (stored as ``""``, the
    neutral element, so it survives serialization) keeps the raw value as the
    local key with no separator — the author's claim that the values are
    already unique across every source of the class (UUIDs, IRIs, ids the
    source itself prefixes).
    """

    field: str = PydanticField(
        ...,
        description="RAW doc field carrying the side-local key.",
    )
    tag: str = PydanticField(
        ...,
        description=(
            "Namespace tag: tag 'a' turns 'f2' into 'a:f2'. ``None`` or ``\"\"`` "
            "keeps the raw value, no separator — only for values already "
            "unique across every source of the class."
        ),
    )

    @field_validator("tag", mode="before")
    @classmethod
    def _none_is_the_empty_tag(cls, value: Any) -> Any:
        return "" if value is None else value

    gate: str | None = PydanticField(
        default=None,
        description=(
            "Optional RAW doc field deciding whether this source applies — the "
            "router's discriminator when one resource contributes several "
            "local keys. Omit when ``field`` is empty for the other branches, "
            "which already selects."
        ),
    )
    gate_prefix: str = PydanticField(
        default="",
        description=(
            'Required prefix of the ``gate`` value; ``""`` always passes. '
            "Meaningless without ``gate``."
        ),
    )

    @model_validator(mode="after")
    def _validate_gate(self) -> LocalKeySource:
        if self.gate is None and self.gate_prefix:
            raise ValueError(
                "LocalKeySource: gate_prefix is meaningless without a gate field"
            )
        return self

Attributes

field = PydanticField(..., description='RAW doc field carrying the side-local key.') class-attribute instance-attribute

gate = PydanticField(default=None, description="Optional RAW doc field deciding whether this source applies — the router's discriminator when one resource contributes several local keys. Omit when ``field`` is empty for the other branches, which already selects.") class-attribute instance-attribute

gate_prefix = PydanticField(default='', description='Required prefix of the ``gate`` value; ``""`` always passes. Meaningless without ``gate``.') class-attribute instance-attribute

tag = PydanticField(..., description='Namespace tag: tag \'a\' turns \'f2\' into \'a:f2\'. ``None`` or ``""`` keeps the raw value, no separator — only for values already unique across every source of the class.') class-attribute instance-attribute

LocalKeySpec

Bases: ConfigBaseModel

The canonical fallback identity attribute for non-aligned records.

sources takes the same three shapes as :attr:AlignmentAttribute.sources: one source, a list (one per member, each reading its own column), or a dict keyed by member class (the member decides; the gate is derived from how the resource produces it, so a member-keyed source must not set gate).

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

class LocalKeySpec(ConfigBaseModel):
    """The canonical fallback identity attribute for non-aligned records.

    ``sources`` takes the same three shapes as
    :attr:`AlignmentAttribute.sources`: one source, a list (one per member,
    each reading its own column), or a dict keyed by member class (the member
    decides; the gate is derived from how the resource produces it, so a
    member-keyed source must not set ``gate``).
    """

    name: str = PydanticField(
        default="local_key",
        validation_alias=AliasChoices("name", "into"),
        description=(
            "Canonical fallback property name on the class. ``into`` is "
            "accepted as a legacy alias."
        ),
    )
    sep: str = PydanticField(
        default=":",
        description="Separator between tag and key.",
    )
    sources: dict[
        str, LocalKeySource | list[LocalKeySource] | dict[str, LocalKeySource]
    ] = PydanticField(
        ...,
        min_length=1,
        description=(
            "Per-resource local-key wiring: ``{resource: source}``; ``{resource: "
            "[source, ...]}`` when the members carry different key columns; "
            "``{resource: {member_class: source}}`` when the member decides."
        ),
    )

    @model_validator(mode="after")
    def _validate_member_sources(self) -> LocalKeySpec:
        for resource, entry in self.sources.items():
            if not isinstance(entry, dict):
                continue
            gated = sorted(m for m, src in entry.items() if src.gate is not None)
            if gated:
                raise ValueError(
                    f"LocalKeySpec: member-keyed sources for resource {resource!r} "
                    f"set a gate on {gated}; the member already decides, and the "
                    "gate is derived from how the resource produces it"
                )
        return self

    def sources_for(self, resource: str) -> list[LocalKeySource]:
        """Local-key sources *resource* contributes, in order."""
        source = self.sources.get(resource)
        if source is None:
            return []
        if isinstance(source, LocalKeySource):
            return [source]
        return list(source.values()) if isinstance(source, dict) else list(source)

    def members_for(self, resource: str) -> list[str] | None:
        """Member classes keying *resource*'s sources, or ``None`` if unkeyed."""
        source = self.sources.get(resource)
        return list(source) if isinstance(source, dict) else None

Attributes

name = PydanticField(default='local_key', validation_alias=AliasChoices('name', 'into'), description='Canonical fallback property name on the class. ``into`` is accepted as a legacy alias.') class-attribute instance-attribute

sep = PydanticField(default=':', description='Separator between tag and key.') class-attribute instance-attribute

sources = PydanticField(..., min_length=1, description='Per-resource local-key wiring: ``{resource: source}``; ``{resource: [source, ...]}`` when the members carry different key columns; ``{resource: {member_class: source}}`` when the member decides.') class-attribute instance-attribute

Methods:

members_for(resource)

Member classes keying resource's sources, or None if unkeyed.

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

def members_for(self, resource: str) -> list[str] | None:
    """Member classes keying *resource*'s sources, or ``None`` if unkeyed."""
    source = self.sources.get(resource)
    return list(source) if isinstance(source, dict) else None

sources_for(resource)

Local-key sources resource contributes, in order.

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

def sources_for(self, resource: str) -> list[LocalKeySource]:
    """Local-key sources *resource* contributes, in order."""
    source = self.sources.get(resource)
    if source is None:
        return []
    if isinstance(source, LocalKeySource):
        return [source]
    return list(source.values()) if isinstance(source, dict) else list(source)

SharedDerivation

Bases: ConfigBaseModel

One derivation shared by several members, varying only in parameters.

The compact spelling of the member-keyed form for the common case: the call is the same for every member and only a parameter changes — a marker prefix per class — or nothing does. members is a list of member classes, or a dict from member to the parameters that differ; each member's derivation is spec with those parameters laid over spec.params. Anything else that differs between members — the input columns, the function — is a different derivation: spell it with the explicit {member: spec} dict.

Expands to that dict; the lowering never sees this model.

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

class SharedDerivation(ConfigBaseModel):
    """One derivation shared by several members, varying only in parameters.

    The compact spelling of the member-keyed form for the common case: the
    call is the same for every member and only a parameter changes — a marker
    prefix per class — or nothing does. ``members`` is a list of member
    classes, or a dict from member to the parameters that differ; each
    member's derivation is ``spec`` with those parameters laid over
    ``spec.params``. Anything else that differs between members — the input
    columns, the function — is a different derivation: spell it with the
    explicit ``{member: spec}`` dict.

    Expands to that dict; the lowering never sees this model.
    """

    spec: DerivationSpec = PydanticField(
        ...,
        description="The derivation every member shares.",
    )
    members: list[str] | dict[str, dict[str, Any]] = PydanticField(
        ...,
        description=(
            "Member classes sharing ``spec``: a list when nothing varies, or "
            "``{member: {param: value}}`` naming what does."
        ),
    )

    @model_validator(mode="after")
    def _validate_members(self) -> SharedDerivation:
        if not self.members:
            raise ValueError("SharedDerivation: members must name at least one class")
        if isinstance(self.members, list) and len(set(self.members)) != len(
            self.members
        ):
            raise ValueError("SharedDerivation: members lists a class twice")
        return self

    def expand(self) -> dict[str, DerivationSpec]:
        """The explicit ``{member: spec}`` dict this stands for."""
        if isinstance(self.members, list):
            return {member: self.spec for member in self.members}
        return {
            member: self.spec.model_copy(
                update={"params": {**self.spec.params, **overrides}}
            )
            for member, overrides in self.members.items()
        }

Attributes

members = PydanticField(..., description='Member classes sharing ``spec``: a list when nothing varies, or ``{member: {param: value}}`` naming what does.') class-attribute instance-attribute

spec = PydanticField(..., description='The derivation every member shares.') class-attribute instance-attribute

Methods:

expand()

The explicit {member: spec} dict this stands for.

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

def expand(self) -> dict[str, DerivationSpec]:
    """The explicit ``{member: spec}`` dict this stands for."""
    if isinstance(self.members, list):
        return {member: self.spec for member in self.members}
    return {
        member: self.spec.model_copy(
            update={"params": {**self.spec.params, **overrides}}
        )
        for member, overrides in self.members.items()
    }

Functions:

alignment_to_ops(alignment, *, manifest=None, canonical_maps=(), sides=None, cluster_members=None)

Merge the alignment into an ordered list of fundamental ops.

Apply the result to the merged union with :func:~graflo.architecture.evolution.apply.apply_evolution. When manifest is given, :func:validate_alignment runs first. Member-keyed sources need sides (the pre-merge manifests) to resolve how each resource produces each member; merge_manifests passes them.

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

def alignment_to_ops(
    alignment: IdentityAlignment,
    *,
    manifest: GraphManifest | None = None,
    canonical_maps: Sequence[VocabularyMap] = (),
    sides: SideManifests | None = None,
    cluster_members: ClusterMembers | None = None,
) -> list[ManifestOp]:
    """Merge the alignment into an ordered list of fundamental ops.

    Apply the result to the merged union with
    :func:`~graflo.architecture.evolution.apply.apply_evolution`. When
    *manifest* is given, :func:`validate_alignment` runs first. Member-keyed
    sources need *sides* (the pre-merge manifests) to resolve how each
    resource produces each member; ``merge_manifests`` passes them.
    """
    if manifest is not None:
        validate_alignment(
            alignment,
            manifest,
            canonical_maps=canonical_maps,
            sides=sides,
            cluster_members=cluster_members,
        )
    sides = _require_sides(alignment, sides)
    productions = (
        resolve_member_productions(alignment, sides) if sides is not None else {}
    )

    ops: list[ManifestOp] = []

    into_names = [attribute.name for attribute in alignment.attributes]
    if alignment.local_key is not None:
        into_names.append(alignment.local_key.name)
    ops.append(AddVertexPropertiesOp(additions={alignment.vertex: list(into_names)}))

    if manifest is not None:
        levels = resolve_derivation_levels(alignment, manifest, productions=productions)
    else:
        levels = {
            resource: list(next(iter(per_member.values())).level)
            for resource, per_member in productions.items()
        }

    # Behind a router, an unkeyed derivation is guarded by the class as a
    # whole; without the manifest there is no router to read, so nothing is.
    class_guards: dict[str, dict[str, Any] | None] = {}
    if manifest is not None:
        for resource, path in levels.items():
            steps = _producing_steps(manifest, resource, path, alignment.vertex)
            class_guards[resource] = _class_guard(steps, alignment.vertex)

    additions: dict[str, list[dict[str, Any]]] = {}
    for attribute in alignment.attributes:
        for resource in attribute.sources:
            additions.setdefault(resource, []).extend(
                _derivation_steps(
                    attribute.name,
                    attribute.specs_for(resource),
                    guards=_guards(
                        productions, resource, attribute.members_for(resource)
                    ),
                    when=class_guards.get(resource),
                )
            )
    if alignment.local_key is not None:
        local_key = alignment.local_key
        for resource in local_key.sources:
            additions.setdefault(resource, []).extend(
                _local_key_steps(
                    local_key,
                    local_key.sources_for(resource),
                    guards=_guards(
                        productions, resource, local_key.members_for(resource)
                    ),
                    when=class_guards.get(resource),
                )
            )
    ops.append(
        AddResourceTransformsOp(
            additions=additions,
            at={
                resource: path
                for resource, path in levels.items()
                if path and resource in additions
            },
        )
    )

    if manifest is not None:
        ensure = _ensure_extracted_fields_op(alignment, manifest, levels, into_names)
        if ensure is not None:
            ops.append(ensure)

    # Always a funnel, even for a single attribute: gating means presence is never
    # guaranteed, and include_branch_id keeps branches collision-free.
    branches = [IdentityBranch(id=name, fields=[name]) for name in into_names]
    ops.append(
        ReplaceIdentityOp(
            replacements={
                alignment.vertex: IdentityReplacement(
                    to=FunnelIdentityTarget(funnel=IdentityFunnel(branches=branches)),
                    # The pre-alignment identity on a merged class is the
                    # merged union of the side keys — a field-set no record
                    # carries. Demoting it would index nothing; the per-side
                    # keys are demoted explicitly below instead.
                    retire="keep",
                )
            }
        )
    )

    if alignment.secondary_identities:
        ops.append(
            AddSecondaryIdentitiesOp(
                additions={
                    alignment.vertex: [
                        SecondaryIdentity(name=name, fields=list(fields))
                        for name, fields in sorted(
                            alignment.secondary_identities.items()
                        )
                    ]
                }
            )
        )

    return ops

rekey_members(alignment, *, sides, resolve)

Name every member key as its side names the member.

resolve maps (side, key) to the member it names — merge_manifests passes the aligned cluster's resolution, so a member may be keyed by its own name or its canonical one. Unresolved keys pass through for the validator to report. Two keys naming one member under one resource are refused.

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

def rekey_members(
    alignment: IdentityAlignment,
    *,
    sides: SideManifests,
    resolve: Callable[[Side, str], str],
) -> IdentityAlignment:
    """Name every member key as its side names the member.

    *resolve* maps ``(side, key)`` to the member it names — ``merge_manifests``
    passes the aligned cluster's resolution, so a member may be keyed by its
    own name or its canonical one. Unresolved keys pass through for the
    validator to report. Two keys naming one member under one resource are
    refused.
    """

    def _names(resource: str, keys: Iterable[str]) -> dict[str, str]:
        side_name, _ = _side_of(resource, sides)
        side: Side = "left" if side_name == "left" else "right"
        out: dict[str, str] = {}
        for key in keys:
            member = resolve(side, key)
            prior = next((k for k, m in out.items() if m == member), None)
            if prior is not None:
                raise _conflict(
                    "member keyed twice",
                    f"resource {resource!r} keys {side} member {member!r} as both "
                    f"{prior!r} and {key!r}",
                    "Key each member once.",
                )
            out[key] = member
        return out

    attributes: list[AlignmentAttribute] = []
    for attribute in alignment.attributes:
        sources: dict[
            str,
            DerivationSpec
            | SharedDerivation
            | list[DerivationSpec]
            | dict[str, DerivationSpec],
        ] = {}
        for resource, spec in attribute.sources.items():
            if isinstance(spec, SharedDerivation):
                names = _names(resource, spec.members)
                members: list[str] | dict[str, dict[str, Any]] = (
                    [names[m] for m in spec.members]
                    if isinstance(spec.members, list)
                    else {names[m]: p for m, p in spec.members.items()}
                )
                sources[resource] = spec.model_copy(update={"members": members})
            elif isinstance(spec, dict):
                names = _names(resource, spec)
                sources[resource] = {names[m]: v for m, v in spec.items()}
            else:
                sources[resource] = spec
        attributes.append(attribute.model_copy(update={"sources": sources}))

    local_key = alignment.local_key
    if local_key is not None:
        local_sources: dict[
            str, LocalKeySource | list[LocalKeySource] | dict[str, LocalKeySource]
        ] = {}
        for resource, entry in local_key.sources.items():
            if isinstance(entry, dict):
                names = _names(resource, entry)
                local_sources[resource] = {names[m]: v for m, v in entry.items()}
            else:
                local_sources[resource] = entry
        local_key = local_key.model_copy(update={"sources": local_sources})

    return alignment.model_copy(
        update={"attributes": attributes, "local_key": local_key}
    )

resolve_derivation_levels(alignment, manifest, *, productions=None)

Pipeline level each referenced resource derives at, keyed by resource.

A derivation must land at the level that produces the aligned class: an actor reads its transform buffer at its own LocationIndex with no ancestor fallback, and a descend subtree runs before its own level's transforms. Placing it anywhere else derives nothing, silently.

IdentityAlignment.at overrides the lookup. A member-keyed resource takes the level its members are produced at (see :func:resolve_member_productions). Otherwise a resource must produce the class at exactly one level — zero and several are both :class:AlignmentConflictError, because either answer the resolver could pick would be a guess about where the source fields live.

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

def resolve_derivation_levels(
    alignment: IdentityAlignment,
    manifest: GraphManifest,
    *,
    productions: MemberProductions | None = None,
) -> dict[str, list[int]]:
    """Pipeline level each referenced resource derives at, keyed by resource.

    A derivation must land at the level that produces the aligned class: an
    actor reads its transform buffer at its own ``LocationIndex`` with no
    ancestor fallback, and a ``descend`` subtree runs before its own level's
    transforms. Placing it anywhere else derives nothing, silently.

    ``IdentityAlignment.at`` overrides the lookup. A member-keyed resource
    takes the level its members are produced at (see
    :func:`resolve_member_productions`). Otherwise a resource must produce the
    class at exactly one level — zero and several are both
    :class:`AlignmentConflictError`, because either answer the resolver could
    pick would be a guess about where the source fields live.
    """
    pipelines = _resource_pipelines(manifest)
    known = _vertex_set(manifest)
    levels: dict[str, list[int]] = {}
    for resource in sorted(_referenced_resources(alignment)):
        pipeline = pipelines.get(resource, [])
        if resource in alignment.at:
            path = list(alignment.at[resource])
            try:
                level = resolve_pipeline_level(list(pipeline), path)
            except ValueError as exc:
                raise _conflict(
                    "unresolvable level",
                    f"`at` for resource {resource!r}: {exc}",
                    "Each index must address a descend step; [] is the root level.",
                ) from exc
            # An override that resolves but produces nothing is the failure this
            # resolution exists to prevent: the derivations would be appended,
            # run, find no inputs, and skip without a word.
            if not any(
                isinstance(step, dict)
                and alignment.vertex
                in step_produces_vertices(step, known_vertices=known)
                for step in level
            ):
                candidates = find_vertex_producing_levels(
                    pipeline, alignment.vertex, known_vertices=known
                )
                raise _conflict(
                    "level produces nothing",
                    f"`at` sends resource {resource!r} derivations to level "
                    f"{path or 'root'}, which produces no {alignment.vertex!r}",
                    (
                        f"A transform is only visible to actors at its own "
                        f"level; {alignment.vertex!r} is produced at {candidates}."
                    )
                    if candidates
                    else f"This resource never produces {alignment.vertex!r}.",
                )
            levels[resource] = path
            continue

        if productions and resource in productions:
            # The members resolved to one level on the side; the union's
            # pipeline for this resource is the same object graph.
            levels[resource] = list(next(iter(productions[resource].values())).level)
            continue

        candidates = find_vertex_producing_levels(
            pipeline, alignment.vertex, known_vertices=known
        )
        if not candidates:
            raise _conflict(
                "resource does not produce the class",
                f"resource {resource!r} has no pipeline step producing "
                f"{alignment.vertex!r}",
                "An alignment derives canonical attributes for the documents "
                "that become this class; a resource that never produces it has "
                "nothing to derive.",
            )
        if len(candidates) > 1:
            raise _conflict(
                "ambiguous level",
                f"resource {resource!r} produces {alignment.vertex!r} at "
                f"levels {candidates}",
                f"Derivation inputs live at one level. Pick it with "
                f"IdentityAlignment(at={{{resource!r}: {candidates[0]}}}).",
            )
        levels[resource] = candidates[0]
    return levels

resolve_member_productions(alignment, sides)

How each resource produces every member its sources are keyed by.

Read off the sides — the pre-merge manifests — because the merge has already rewritten each router's type_map values to the canonical name, so the union cannot say which key produced which member.

All members a resource keys must resolve to one pipeline level: the derivations are appended per resource at one level, and a member produced under a different descend would not see them.

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

def resolve_member_productions(
    alignment: IdentityAlignment, sides: SideManifests
) -> MemberProductions:
    """How each resource produces every member its sources are keyed by.

    Read off the *sides* — the pre-merge manifests — because the merge has
    already rewritten each router's ``type_map`` values to the canonical name,
    so the union cannot say which key produced which member.

    All members a resource keys must resolve to one pipeline level: the
    derivations are appended per resource at one level, and a member produced
    under a different ``descend`` would not see them.
    """
    out: MemberProductions = {}
    for resource, members in sorted(_member_keyed_resources(alignment).items()):
        per_member = {
            member: _resolve_member_production(alignment, resource, member, sides)
            for member in sorted(members)
        }
        levels = {tuple(p.level) for p in per_member.values()}
        if len(levels) > 1:
            raise _conflict(
                "members at different levels",
                f"resource {resource!r} produces "
                f"{ {m: p.level for m, p in per_member.items()} }",
                "Derivations are appended per resource at one level; produce "
                "the members at one level or align them through separate "
                "resources.",
            )
        out[resource] = per_member
    return out

validate_alignment(alignment, manifest, *, canonical_maps=(), sides=None, cluster_members=None)

Fail loudly when alignment contradicts manifest or the canonical maps.

manifest is the merged union the alignment ops will be applied to. Pass the maps used to canonicalize the sides — declared :class:CanonicalMap s or the composite :class:~graflo.architecture.evolution.ops.CanonicalizeOp merge applied — to catch derivation inputs written in canonical vocabulary: renamed documents still carry their raw field names, so a rename target used as a derivation input reads an absent field and silently derives nothing.

sides are the pre-merge side manifests, required by member-keyed sources; cluster_members are the aligned cluster's members per side, which lets a member key be checked against the cluster it claims.

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

def validate_alignment(
    alignment: IdentityAlignment,
    manifest: GraphManifest,
    *,
    canonical_maps: Sequence[VocabularyMap] = (),
    sides: SideManifests | None = None,
    cluster_members: ClusterMembers | None = None,
) -> None:
    """Fail loudly when *alignment* contradicts *manifest* or the canonical maps.

    *manifest* is the merged union the alignment ops will be applied to.
    Pass the maps used to canonicalize the sides — declared
    :class:`CanonicalMap`\\ s or the composite
    :class:`~graflo.architecture.evolution.ops.CanonicalizeOp` merge applied — to catch
    derivation inputs written in canonical vocabulary: renamed documents still
    carry their raw field names, so a rename *target* used as a derivation
    input reads an absent field and silently derives nothing.

    *sides* are the pre-merge side manifests, required by member-keyed
    sources; *cluster_members* are the aligned cluster's members per side,
    which lets a member key be checked against the cluster it claims.
    """
    schema = manifest.graph_schema
    if schema is None:
        raise AlignmentConflictError("identity alignment requires graph_schema")
    vertex_config = schema.core_schema.vertex_config
    if alignment.vertex not in vertex_config.vertex_set:
        raise _conflict(
            "unknown vertex",
            f"{alignment.vertex!r} is not defined in the manifest",
            f"Defined: {sorted(vertex_config.vertex_set)}.",
        )
    if manifest.ingestion_model is None:
        raise AlignmentConflictError(
            "identity alignment requires ingestion_model — derivations are "
            "resource pipeline steps"
        )
    known_resources = {r.name for r in manifest.ingestion_model.resources}

    resource_refs: set[str] = set()
    raw_inputs: dict[str, list[str]] = {}
    for attribute in alignment.attributes:
        resource_refs.update(attribute.sources)
        for resource in attribute.sources:
            for spec in attribute.specs_for(resource):
                raw_inputs.setdefault(resource, []).extend(spec.input)
    if alignment.local_key is not None:
        resource_refs.update(alignment.local_key.sources)
        for resource in alignment.local_key.sources:
            for src in alignment.local_key.sources_for(resource):
                raw_inputs.setdefault(resource, []).append(src.field)
                if src.gate is not None:
                    raw_inputs.setdefault(resource, []).append(src.gate)
    resource_refs.update(alignment.at)

    missing = sorted(resource_refs - known_resources)
    if missing:
        raise _conflict(
            "unknown resources",
            f"{missing} are not defined in the manifest",
            f"Defined: {sorted(known_resources)}.",
        )

    sides = _require_sides(alignment, sides)
    productions = (
        resolve_member_productions(alignment, sides) if sides is not None else {}
    )
    for resource, per_member in productions.items():
        assert sides is not None
        side, _ = _side_of(resource, sides)
        if cluster_members is not None:
            allowed = set(cluster_members.get(side, ()))
            outside = sorted(set(per_member) - allowed)
            if outside:
                raise _conflict(
                    "member outside the cluster",
                    f"resource {resource!r} ({side}) keys derivations by "
                    f"{outside}, which are not members of the "
                    f"{alignment.vertex!r} cluster on that side",
                    f"Members on the {side}: {sorted(allowed)}.",
                )
        for production in per_member.values():
            if production.type_field is not None:
                raw_inputs.setdefault(resource, []).append(production.type_field)

    current_identity = set(vertex_config.identity_fields(alignment.vertex))
    into_names = [attribute.name for attribute in alignment.attributes]
    if alignment.local_key is not None:
        into_names.append(alignment.local_key.name)
    colliding = sorted(set(into_names) & current_identity)
    if colliding:
        raise _conflict(
            "identity collision",
            f"target attributes {colliding} are already primary-identity "
            f"fields of {alignment.vertex!r}",
            "Pick canonical attribute names distinct from the current key; "
            "the alignment replaces the identity wholesale.",
        )

    declared = set(vertex_config.property_names(alignment.vertex))
    for name, fields in alignment.secondary_identities.items():
        undeclared = sorted(set(fields) - declared)
        if undeclared:
            raise _conflict(
                "undeclared secondary fields",
                f"secondary identity {name!r} references {undeclared}, not "
                f"declared on {alignment.vertex!r}",
                "Secondary identities index existing properties.",
            )

    rename_targets = _canonical_rename_targets(tuple(canonical_maps))
    if rename_targets:
        for resource, fields in raw_inputs.items():
            canonical_used = sorted(set(fields) & rename_targets)
            if canonical_used:
                raise _conflict(
                    "canonical name as derivation input",
                    f"resource {resource!r} derivations read {canonical_used}, "
                    "which are canonical rename targets — documents still "
                    "carry the RAW source field names",
                    "Use the raw field names the source documents actually "
                    "carry (property renames rewrite vertex.from maps, not "
                    "transform inputs).",
                )

    levels = resolve_derivation_levels(alignment, manifest, productions=productions)

    # Scratch fields exist only for the column-presence (list) form; a
    # member-keyed derivation is the single writer of its attribute.
    scratch_names = {
        _scratch_name(attribute.name, index)
        for attribute in alignment.attributes
        for resource in attribute.sources
        if attribute.members_for(resource) is None
        and len(attribute.specs_for(resource)) > 1
        for index in range(len(attribute.specs_for(resource)))
    }
    if alignment.local_key is not None:
        scratch_names |= {
            _scratch_name(alignment.local_key.name, index)
            for resource in alignment.local_key.sources
            if alignment.local_key.members_for(resource) is None
            and len(alignment.local_key.sources_for(resource)) > 1
            for index in range(len(alignment.local_key.sources_for(resource)))
        }
    colliding_scratch = sorted(scratch_names & declared)
    if colliding_scratch:
        raise _conflict(
            "scratch name collision",
            f"multi-branch derivations would write {colliding_scratch}, which "
            f"are declared properties of {alignment.vertex!r}",
            "Rename the property, or the canonical attribute the scratch "
            "names are derived from.",
        )

    for resource, path in sorted(levels.items()):
        steps = _producing_steps(manifest, resource, path, alignment.vertex)
        unguarded = (
            _unkeyed_names(alignment, resource)
            if _class_guard(steps, alignment.vertex) is None
            else []
        )
        for step in steps:
            if unguarded:
                _check_sibling_classes(alignment, manifest, step, resource, unguarded)
            _warn_on_one_derivation_for_several_members(alignment, step, resource)
    if sides is not None and cluster_members is not None:
        for resource in productions:
            _warn_on_partial_member_coverage(
                alignment, resource, sides, cluster_members
            )

    if alignment.local_key is None:
        logger.warning(
            "identity alignment for %r has no local_key: records matching no "
            "aligned attribute complete no funnel branch and are dropped",
            alignment.vertex,
        )