graflo.architecture.evolution.alignment

Derived identity: lower a merged class's identity branches to fundamental ops.

A :class:~graflo.architecture.evolution.ops.VertexEquivalence declares the merged key as ordered funnel branches. A branch over properties the members already carry needs nothing but the funnel. A :class:~graflo.architecture.evolution.ops.DerivedBranch or :class:~graflo.architecture.evolution.ops.LocalKeyBranch needs each source to compute its attribute, and :func:identity_to_ops emits only fundamental ops for that —

1. AddVertexPropertiesOp — declare the derived attributes on the class;
2. AddResourceTransformsOp — per-resource derivation steps (normalization, local-key namespacing) appended to the pipelines;
3. EnsureExtractedFieldsOp — when a producing router restricts what it extracts;
4. ReplaceIdentityOp — a priority funnel over the branches, in declared order. merge_manifests then demotes each member's own key to a secondary identity against that final funnel.

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. An explicit when on the spec replaces that derived guard. Only a level where a plain vertex step also produces the class lowers unguarded; there a sibling class declaring a derived attribute name is refused, since the router would hand it the derived value.

A level producing the class through routers that read different discriminators — the two roles of an edge resource — hosts no derivation at all, keyed or not: the level's transform buffer is shared by every router at it, so one derived attribute cannot hold one value per role. Such a resource stays out of the sources and references the class instead; :func:hosts_member_derivation tells the two apart.

Attributes

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

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

SideManifests = Mapping[str, GraphManifest] module-attribute

SteppedBranch = DerivedBranch | LocalKeyBranch module-attribute

VocabularyMap = CanonicalMap | CanonicalizeOp module-attribute

__all__ = ['AlignmentConflictError', 'IdentityPlan', 'hosts_member_derivation', 'identity_to_ops', 'validate_identity'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

AlignmentConflictError

Bases: ValueError

A derived identity contradicts the union manifest or canonical maps.

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

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

IdentityPlan dataclass

One class's declared identity branches, and where its sources derive them.

branches are in funnel order, as :attr:VertexEquivalence.identity declares them; at is :attr:VertexEquivalence.derive_at; digest_field is :attr:VertexEquivalence.digest_field; derive holds :attr:VertexEquivalence.derive, one derivation per attribute, which name and composite branches key on.

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

@dataclass(frozen=True)
class IdentityPlan:
    """One class's declared identity branches, and where its sources derive them.

    ``branches`` are in funnel order, as :attr:`VertexEquivalence.identity`
    declares them; ``at`` is :attr:`VertexEquivalence.derive_at`;
    ``digest_field`` is :attr:`VertexEquivalence.digest_field`; ``derive``
    holds :attr:`VertexEquivalence.derive`, one derivation per attribute, which
    name and composite branches key on.
    """

    vertex: str
    branches: tuple[IdentityBranchDecl, ...]
    at: Mapping[str, list[int]] = dataclasses.field(default_factory=dict)
    digest_field: str = "id"
    derive: tuple[DerivedBranch, ...] = ()

    def __post_init__(self) -> None:
        try:
            check_identity_branches(
                self.branches,
                label=f"identity of {self.vertex!r}",
                derived=[d.name for d in self.derive],
            )
        except ValueError as exc:
            raise _conflict(
                "identity branches", str(exc), "Reorder or rename the branches."
            ) from exc

    @property
    def derived(self) -> list[DerivedBranch]:
        """Every derived attribute: the ``derive`` entries, then derived branches."""
        return [
            *self.derive,
            *(b for b in self.branches if isinstance(b, DerivedBranch)),
        ]

    @property
    def local_key(self) -> LocalKeyBranch | None:
        """The ``local_key`` branch, if declared."""
        return next((b for b in self.branches if isinstance(b, LocalKeyBranch)), None)

    @property
    def stepped(self) -> list[SteppedBranch]:
        """Branches whose attribute the pipelines derive, derived ones first."""
        out: list[SteppedBranch] = list(self.derived)
        if self.local_key is not None:
            out.append(self.local_key)
        return out

    @property
    def raw(self) -> list[tuple[str, ...]]:
        """Branches over properties the members carry, as field tuples.

        A name or composite branch over ``derive`` attributes is not one: its
        fields are derived, never all derived and properties at once.
        """
        attributes = {d.name for d in self.derive}
        return [
            tuple(branch_fields(b))
            for b in self.branches
            if isinstance(b, str | list) and branch_fields(b)[0] not in attributes
        ]

    def derived_names(self) -> list[str]:
        """Attributes the pipelines derive: every derived branch, then the local key."""
        return [branch.name for branch in self.stepped]

    def funnel(self) -> IdentityFunnel:
        """The funnel the branches declare."""
        return identity_branches_funnel(self.branches)

Attributes

at = dataclasses.field(default_factory=dict) class-attribute instance-attribute

branches instance-attribute

derive = () class-attribute instance-attribute

derived property

Every derived attribute: the derive entries, then derived branches.

digest_field = 'id' class-attribute instance-attribute

local_key property

The local_key branch, if declared.

raw property

Branches over properties the members carry, as field tuples.

A name or composite branch over derive attributes is not one: its fields are derived, never all derived and properties at once.

stepped property

Branches whose attribute the pipelines derive, derived ones first.

vertex instance-attribute

Methods:

__init__(vertex, branches, at=dict(), digest_field='id', derive=())

__post_init__()

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

def __post_init__(self) -> None:
    try:
        check_identity_branches(
            self.branches,
            label=f"identity of {self.vertex!r}",
            derived=[d.name for d in self.derive],
        )
    except ValueError as exc:
        raise _conflict(
            "identity branches", str(exc), "Reorder or rename the branches."
        ) from exc

derived_names()

Attributes the pipelines derive: every derived branch, then the local key.

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

def derived_names(self) -> list[str]:
    """Attributes the pipelines derive: every derived branch, then the local key."""
    return [branch.name for branch in self.stepped]

funnel()

The funnel the branches declare.

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

def funnel(self) -> IdentityFunnel:
    """The funnel the branches declare."""
    return identity_branches_funnel(self.branches)

Functions:

hosts_member_derivation(manifest, resource, member, *, at=None)

Whether resource can derive an attribute for the records that become member.

False when the level producing member — at, or the one level that does — produces it through routers reading more than one discriminator: the level's transform buffer is shared by every router at it, so one derived attribute cannot hold a different value per role, and a source naming the resource, keyed or not, is refused. Merge gives such a resource no automatic key; it references the class by the members' own keys. True wherever the other validators decide — a resource or level that does not produce the member, or several levels without at.

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

def hosts_member_derivation(
    manifest: GraphManifest,
    resource: str,
    member: str,
    *,
    at: Sequence[int] | None = None,
) -> bool:
    """Whether *resource* can derive an attribute for the records that become *member*.

    ``False`` when the level producing *member* — *at*, or the one level that
    does — produces it through routers reading more than one discriminator:
    the level's transform buffer is shared by every router at it, so one
    derived attribute cannot hold a different value per role, and a source
    naming the resource, keyed or not, is refused. Merge gives such a resource
    no automatic key; it references the class by the members' own keys.
    ``True`` wherever the other validators decide — a resource or level that
    does not produce the member, or several levels without *at*.
    """
    pipelines = _resource_pipelines(manifest)
    if resource not in pipelines:
        return True
    pipeline = pipelines[resource]
    if at is not None:
        path = list(at)
    else:
        candidates = find_vertex_producing_levels(
            pipeline, member, known_vertices=_vertex_set(manifest)
        )
        if len(candidates) != 1:
            return True
        path = candidates[0]
    try:
        steps = _producing_steps(manifest, resource, path, member)
    except ValueError:
        return True
    type_fields = _level_discriminators(steps)
    return type_fields is None or len(type_fields) <= 1

identity_to_ops(plan, *, manifest=None, canonical_maps=(), sides=None, cluster_members=None, member_identity=None, uncovered_producers='refuse', completes_property_branch=None, origins=None)

Lower plan to 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_identity runs first. Member-keyed sources need sides (the pre-merge manifests) to resolve how each resource produces each member; merge_manifests passes them. A local_key source with no tag is tagged with the origin of its resource's side: origins keyed by side, resolved through sides.

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

def identity_to_ops(
    plan: IdentityPlan,
    *,
    manifest: GraphManifest | None = None,
    canonical_maps: Sequence[VocabularyMap] = (),
    sides: SideManifests | None = None,
    cluster_members: ClusterMembers | None = None,
    member_identity: Collection[str] | None = None,
    uncovered_producers: Literal["refuse", "allow"] = "refuse",
    completes_property_branch: Callable[[str], bool] | None = None,
    origins: Mapping[Side, str] | None = None,
) -> list[ManifestOp]:
    """Lower *plan* to 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_identity` runs first. Member-keyed
    sources need *sides* (the pre-merge manifests) to resolve how each
    resource produces each member; ``merge_manifests`` passes them. A
    ``local_key`` source with no ``tag`` is tagged with the origin of its
    resource's side: *origins* keyed by side, resolved through *sides*.
    """
    if manifest is not None:
        validate_identity(
            plan,
            manifest,
            canonical_maps=canonical_maps,
            sides=sides,
            cluster_members=cluster_members,
            member_identity=member_identity,
            uncovered_producers=uncovered_producers,
            completes_property_branch=completes_property_branch,
        )
    sides = _require_sides(plan, sides)
    productions = resolve_member_productions(plan, sides) if sides is not None else {}

    ops: list[ManifestOp] = []
    into_names = plan.derived_names()
    if into_names:
        ops.append(AddVertexPropertiesOp(additions={plan.vertex: list(into_names)}))

    if manifest is not None:
        levels = resolve_derivation_levels(plan, 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, plan.vertex)
            class_guards[resource] = _class_guard(steps, plan.vertex)

    additions: dict[str, list[dict[str, Any]]] = {}
    for branch in plan.stepped:
        for resource in branch.sources:
            additions.setdefault(resource, []).extend(
                _branch_steps(
                    branch,
                    resource,
                    productions=productions,
                    class_guard=class_guards.get(resource),
                    sides=sides,
                    origins=origins,
                )
            )
    if additions:
        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 and into_names:
        ensure = _ensure_extracted_fields_op(plan, manifest, levels, into_names)
        if ensure is not None:
            ops.append(ensure)

    # One property branch is a natural key; anything else is a funnel, because
    # a derived attribute is never guaranteed present and include_branch_id
    # keeps branches over equal values apart.
    if len(plan.branches) == 1 and not plan.stepped:
        target: FunnelIdentityTarget | NaturalIdentityTarget = NaturalIdentityTarget(
            identity=list(plan.raw[0])
        )
    else:
        target = FunnelIdentityTarget(
            funnel=plan.funnel(), digest_field=plan.digest_field
        )
    ops.append(
        ReplaceIdentityOp(
            replacements={
                plan.vertex: IdentityReplacement(
                    to=target,
                    # The pre-lowering identity on a merged class is the
                    # merged union of the side keys — a field-set no record
                    # carries. Demoting it would index nothing; merge demotes
                    # the per-member keys once this funnel is in place.
                    retire="keep",
                )
            }
        )
    )
    return ops

rekey_members(plan, *, 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 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(
    plan: IdentityPlan,
    *,
    sides: SideManifests,
    resolve: Callable[[Side, str], str],
) -> IdentityPlan:
    """Name every member key as its side names the member.

    *resolve* maps ``(side, key)`` to the member it names — ``merge_manifests``
    passes the 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

    def _rekeyed(entries: Mapping[str, Any]) -> dict[str, Any]:
        out: dict[str, Any] = {}
        for resource, entry in entries.items():
            if isinstance(entry, dict):
                names = _names(resource, entry)
                out[resource] = {names[m]: v for m, v in entry.items()}
            else:
                out[resource] = entry
        return out

    branches: list[IdentityBranchDecl] = []
    for branch in plan.branches:
        if isinstance(branch, DerivedBranch):
            branch = branch.model_copy(update={"sources": _rekeyed(branch.sources)})
        elif isinstance(branch, LocalKeyBranch):
            branch = branch.model_copy(update={"local_key": _rekeyed(branch.local_key)})
        branches.append(branch)
    derive = tuple(
        d.model_copy(update={"sources": _rekeyed(d.sources)}) for d in plan.derive
    )
    return dataclasses.replace(plan, branches=tuple(branches), derive=derive)

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

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

A derivation must land at the level that produces the 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.

plan.at (the equivalence's derive_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(
    plan: IdentityPlan,
    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 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.

    ``plan.at`` (the equivalence's ``derive_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(plan)):
        pipeline = pipelines.get(resource, [])
        if resource in plan.at:
            path = list(plan.at[resource])
            try:
                level = resolve_pipeline_level(list(pipeline), path)
            except ValueError as exc:
                raise _conflict(
                    "unresolvable level",
                    f"`derive_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 plan.vertex in step_produces_vertices(step, known_vertices=known)
                for step in level
            ):
                candidates = find_vertex_producing_levels(
                    pipeline, plan.vertex, known_vertices=known
                )
                raise _conflict(
                    "level produces nothing",
                    f"`derive_at` sends resource {resource!r} derivations to level "
                    f"{path or 'root'}, which produces no {plan.vertex!r}",
                    (
                        f"A transform is only visible to actors at its own "
                        f"level; {plan.vertex!r} is produced at {candidates}."
                    )
                    if candidates
                    else f"This resource never produces {plan.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, plan.vertex, known_vertices=known
        )
        if not candidates:
            raise _conflict(
                "resource does not produce the class",
                f"resource {resource!r} has no pipeline step producing {plan.vertex!r}",
                "A derived branch computes an attribute 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 {plan.vertex!r} at "
                f"levels {candidates}",
                f"Derivation inputs live at one level. Pick it with "
                f"derive_at={{{resource!r}: {candidates[0]}}} on the equivalence.",
            )
        levels[resource] = candidates[0]
    return levels

resolve_member_productions(plan, 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(
    plan: IdentityPlan, 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(plan).items()):
        per_member = {
            member: _resolve_member_production(plan, 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 derive them through separate "
                "resources.",
            )
        out[resource] = per_member
    return out

uncovered_producers(plan, manifest, *, completes_property_branch=None)

Resources that upsert the class but can complete none of its key, sorted.

Every record of the class keys on the funnel over the plan's branches. A resource no derived branch names derives none of their attributes, so each record it upserts completes no branch and is dropped -- the whole resource, and every edge it emits to the class -- unless it completes a property branch. completes_property_branch answers that per resource (merge knows each member's properties); without it, a plan with any property branch is taken to cover every resource. A resource whose steps producing the class only look it up (lookup_only) or find it by a secondary identity (find), on a vertex step or a router, creates nothing and is not listed.

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

def uncovered_producers(
    plan: IdentityPlan,
    manifest: GraphManifest,
    *,
    completes_property_branch: Callable[[str], bool] | None = None,
) -> list[str]:
    """Resources that upsert the class but can complete none of its key, sorted.

    Every record of the class keys on the funnel over the plan's branches. A
    resource no derived branch names derives none of their attributes, so
    each record it upserts completes no branch and is dropped -- the whole
    resource, and every edge it emits to the class -- unless it completes a
    property branch. *completes_property_branch* answers that per resource
    (merge knows each member's properties); without it, a plan with any
    property branch is taken to cover every resource. A resource whose steps
    producing the class only look it up (``lookup_only``) or find it by a
    secondary identity (``find``), on a vertex step or a router, creates
    nothing and is not listed.
    """
    covered = _referenced_resources(plan)
    known = _vertex_set(manifest)
    out: list[str] = []
    for resource, pipeline in sorted(_resource_pipelines(manifest).items()):
        if resource in covered:
            continue
        if not any(
            not step_looks_up(step, plan.vertex)
            and step_finds(step, plan.vertex, known_vertices=known) is None
            for step in _steps_producing_anywhere(pipeline, plan.vertex, known)
        ):
            continue
        if plan.raw and (
            completes_property_branch is None or completes_property_branch(resource)
        ):
            continue
        out.append(resource)
    return out

validate_identity(plan, manifest, *, canonical_maps=(), sides=None, cluster_members=None, member_identity=None, uncovered_producers='refuse', completes_property_branch=None)

Fail loudly when plan contradicts manifest or the canonical maps.

manifest is the merged union the 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 cluster's members per side, which lets a member key be checked against the cluster it claims.

member_identity are the identity fields the class's records carry — for a merged class, its members' own pre-merge keys under their canonical names. A derived attribute named like one would overwrite that key's value with the derived one. It defaults to the class's current identity in manifest.

uncovered_producers decides a resource that upserts the class but derives none of its key (see :func:uncovered_producers): refuse it, or allow it for a caller that turns it into a reference -- merge does, once the member keys it can look the class up by exist. completes_property_branch says whether a resource's records complete one of the plan's property branches; see :func:uncovered_producers.

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

def validate_identity(
    plan: IdentityPlan,
    manifest: GraphManifest,
    *,
    canonical_maps: Sequence[VocabularyMap] = (),
    sides: SideManifests | None = None,
    cluster_members: ClusterMembers | None = None,
    member_identity: Collection[str] | None = None,
    uncovered_producers: Literal["refuse", "allow"] = "refuse",
    completes_property_branch: Callable[[str], bool] | None = None,
) -> None:
    """Fail loudly when *plan* contradicts *manifest* or the canonical maps.

    *manifest* is the merged union the 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 cluster's members per side, which lets
    a member key be checked against the cluster it claims.

    *member_identity* are the identity fields the class's records carry — for
    a merged class, its members' own pre-merge keys under their canonical
    names. A derived attribute named like one would overwrite that key's
    value with the derived one. It defaults to the class's current identity
    in *manifest*.

    *uncovered_producers* decides a resource that upserts the class but
    derives none of its key (see :func:`uncovered_producers`): ``refuse`` it,
    or ``allow`` it for a caller that turns it into a reference -- merge does,
    once the member keys it can look the class up by exist.
    *completes_property_branch* says whether a resource's records complete
    one of the plan's property branches; see :func:`uncovered_producers`.
    """
    schema = manifest.graph_schema
    if schema is None:
        raise AlignmentConflictError("a derived identity requires graph_schema")
    vertex_config = schema.core_schema.vertex_config
    if plan.vertex not in vertex_config.vertex_set:
        raise _conflict(
            "unknown vertex",
            f"{plan.vertex!r} is not defined in the manifest",
            f"Defined: {sorted(vertex_config.vertex_set)}.",
        )
    if manifest.ingestion_model is None:
        raise AlignmentConflictError(
            "a derived identity requires ingestion_model — derivations are "
            "resource pipeline steps"
        )
    known_resources = {r.name for r in manifest.ingestion_model.resources}

    resource_refs: set[str] = set(plan.at)
    raw_inputs: dict[str, list[str]] = {}
    for branch in plan.stepped:
        resource_refs.update(branch.sources)
        for resource in branch.sources:
            for spec in _specs(branch, resource):
                fields = raw_inputs.setdefault(resource, [])
                fields.extend(
                    spec.input if isinstance(spec, DerivationSpec) else [spec.field]
                )
                if spec.when is not None:
                    fields.append(spec.when.field)

    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(plan, sides)
    productions = resolve_member_productions(plan, 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"{plan.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(
        member_identity
        if member_identity is not None
        else vertex_config.identity_fields(plan.vertex)
    )
    colliding = sorted(set(plan.derived_names()) & current_identity)
    if colliding:
        raise _conflict(
            "identity collision",
            f"derived attributes {colliding} are already identity fields of "
            f"{plan.vertex!r}'s records",
            "Pick attribute names distinct from the members' own keys; a "
            "derivation would overwrite the key's value. To key on a member's "
            "own key, list it as a property branch.",
        )

    declared = set(vertex_config.property_names(plan.vertex))
    undeclared = sorted({f for fields in plan.raw for f in fields} - declared)
    if undeclared:
        raise _conflict(
            "undeclared property branch",
            f"property branches name {undeclared}, which {plan.vertex!r} does "
            "not declare",
            "A property branch keys on a property the members carry, under "
            "its canonical name.",
        )

    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 - _raw_properties(resource, sides, cluster_members)
            )
            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(plan, manifest, productions=productions)
    if uncovered_producers == "refuse":
        _check_uncovered_producers(plan, manifest, completes_property_branch)
    _check_derivation_signatures(plan)

    for resource, path in sorted(levels.items()):
        steps = _producing_steps(manifest, resource, path, plan.vertex)
        type_fields = _level_discriminators(steps)
        if type_fields is not None and len(type_fields) > 1:
            raise _across_roles(resource, plan.vertex, type_fields)
        unguarded = (
            _unguarded_names(plan, resource)
            if _class_guard(steps, plan.vertex) is None
            else []
        )
        for step in steps:
            if unguarded:
                _check_sibling_classes(plan, manifest, step, resource, unguarded)
            _warn_on_one_derivation_for_several_members(plan, step, resource)
    if sides is not None and cluster_members is not None:
        for resource in productions:
            _warn_on_partial_member_coverage(plan, resource, sides, cluster_members)

    if plan.local_key is None and not plan.raw:
        logger.warning(
            "identity for %r has no local_key branch: records deriving none of "
            "its attributes complete no funnel branch and are dropped",
            plan.vertex,
        )