graflo.architecture.evolution.apply

Apply manifest evolution operations to a copy of a :class:~graflo.architecture.contract.manifest.GraphManifest.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

apply_add_edge_properties(manifest, op)

Append new edge properties to existing relations or edge triples.

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

def apply_add_edge_properties(manifest: GraphManifest, op: AddEdgePropertiesOp) -> None:
    """Append new edge properties to existing relations or edge triples."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("add_edge_properties requires graph_schema")
    edges = schema.core_schema.edge_config.edges
    known = {edge.relation for edge in edges if edge.relation is not None}
    unknown = sorted(set(op.additions) - known)
    if unknown:
        raise ValueError(f"add_edge_properties: unknown relations: {unknown}")
    edge_ids = {edge.edge_id for edge in edges}
    unknown_edges = [e.edge_id() for e in op.per_edge if e.edge_id() not in edge_ids]
    if unknown_edges:
        raise ValueError(f"add_edge_properties: unknown edges: {unknown_edges}")
    for edge in edges:
        additions = op.additions_for(edge.edge_id)
        if not additions:
            continue
        existing = {field.name: field for field in edge.properties}
        for entry in additions:
            field = Field(name=entry, type=None) if isinstance(entry, str) else entry
            prior = existing.get(field.name)
            if prior is not None:
                _refuse_redeclaration(prior, field, owner=f"edge {edge.edge_id!r}")
                continue
            edge.properties.append(field.model_copy(deep=True))
            existing[field.name] = field
    schema.finish_init()

apply_add_inverse_edges(manifest, op)

Realize declared inverses as explicit edges across schema, profile and ingestion.

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

def apply_add_inverse_edges(manifest: GraphManifest, op: AddInverseEdgesOp) -> None:
    """Realize declared inverses as explicit edges across schema, profile and ingestion."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("add_inverse_edges requires graph_schema")
    edge_config = schema.core_schema.edge_config

    relation_map, added = plan_inverse_edges(schema, op.relations)
    created = {edge.edge_id for edge in added}
    new_edges = [*edge_config.edges, *added]

    _replace_edge_config(schema, edge_config.with_edges(new_edges))
    apply_inverse_edges_to_db_profile(schema.db_profile, relation_map, created)
    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.finish_init()

    if manifest.ingestion_model is None or not created:
        return

    from graflo.architecture.contract.ingestion.resource import Resource

    from .ingestion import set_emit_inverse_flags

    # Which steps mirror is decided against the pipelines as they are; the
    # selectors are extended first because they do not move step positions.
    emission = plan_inverse_emission(manifest, relation_map, created)
    manifest.ingestion_model.resources = [
        Resource.model_validate(
            mirror_resource_selectors(
                resource.to_dict(skip_defaults=False), relation_map, created
            )
        )
        for resource in manifest.ingestion_model.resources
    ]
    manifest.ingestion_model = IngestionModel.model_validate(
        manifest.ingestion_model.to_dict(skip_defaults=False)
    )
    if emission:
        set_emit_inverse_flags(manifest, emission, True)

apply_add_vertex_properties(manifest, op)

Append new vertex properties to existing vertices.

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

def apply_add_vertex_properties(
    manifest: GraphManifest, op: AddVertexPropertiesOp
) -> None:
    """Append new vertex properties to existing vertices."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("add_vertex_properties requires graph_schema")
    unknown = set(op.additions) - schema.core_schema.vertex_config.vertex_set
    if unknown:
        raise ValueError(f"add_vertex_properties: unknown vertices: {sorted(unknown)}")
    for vertex in schema.core_schema.vertex_config.vertices:
        additions = op.additions.get(vertex.name, [])
        if not additions:
            continue
        existing = {field.name: field for field in vertex.properties}
        for entry in additions:
            # A bare name keeps its original meaning (untyped property); a Field
            # is appended as authored, carrying its type and grounding.
            field = Field(name=entry, type=None) if isinstance(entry, str) else entry
            prior = existing.get(field.name)
            if prior is not None:
                _refuse_redeclaration(prior, field, owner=f"vertex {vertex.name!r}")
                continue
            vertex.properties.append(field.model_copy(deep=True))
            existing[field.name] = field
    schema.finish_init()

apply_canonicalize(manifest, op)

Mutate manifest in place: relabel by op's vocabulary map in one step.

Every refusal — unknown names, an undeclared occupied target, an attribute collision, an unacknowledged self-relation or observation fusion — leaves manifest untouched: the work happens on a copy that is swapped in only once it is accepted.

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

def apply_canonicalize(manifest: GraphManifest, op: CanonicalizeOp) -> None:
    """Mutate *manifest* in place: relabel by *op*'s vocabulary map in one step.

    Every refusal — unknown names, an undeclared occupied target, an attribute
    collision, an unacknowledged self-relation or observation fusion — leaves
    *manifest* untouched: the work happens on a copy that is swapped in only
    once it is accepted.
    """
    vertex_map = {s: t for s, t in op.vertices.items() if s != t}
    relation_map = {s: t for s, t in op.relations.items() if s != t}
    property_renames = {
        cls: {old: new for old, new in attrs.items() if old != new}
        for cls, attrs in op.properties.items()
    }
    property_renames = {cls: attrs for cls, attrs in property_renames.items() if attrs}

    work = manifest.model_copy(deep=True)
    schema = work.graph_schema
    if schema is None:
        if property_renames:
            raise ValueError("canonicalize: properties require graph_schema")
        if work.ingestion_model is not None and (vertex_map or relation_map):
            _rewrite_ingestion_for_canonicalize(
                work.ingestion_model, vertex_map, relation_map, declared=None
            )
            manifest.ingestion_model = IngestionModel.model_validate(
                work.ingestion_model.to_dict(skip_defaults=False)
            )
        return

    core = schema.core_schema
    _check_property_renames(core.vertex_config, property_renames)
    vertex_groups = _check_vocabulary_map(
        op.vertices, set(core.vertex_config.vertex_set), noun="vertices"
    )
    relation_groups = _check_vocabulary_map(
        op.relations,
        {edge.relation for edge in core.edge_config.edges if edge.relation is not None},
        noun="relations",
    )
    merged_targets = sorted(
        target for target, members in vertex_groups.items() if len(members) > 1
    )

    if property_renames:
        apply_rename_vertex_properties(
            work, RenameVertexPropertiesOp(renames=property_renames)
        )
        core = schema.core_schema

    relabels = (
        bool(vertex_map)
        or bool(relation_map)
        or bool(merged_targets)
        or any(len(members) > 1 for members in relation_groups.values())
    )
    if relabels:
        before_edges = list(core.edge_config.edges)
        new_vc = _canonicalize_vertex_config(
            core.vertex_config, op.vertices, vertex_groups
        )
        edges = redirect_and_merge_edges(core.edge_config.edges, vertex_map)
        edges = remap_relation_and_merge_edges(edges, relation_map)

        # The profile is remapped ahead of the assignment because assigning
        # `core_schema` revalidates the profile against it.
        remap_vertices_in_db_profile(schema.db_profile, vertex_map)
        apply_relation_rename_to_db_profile(schema.db_profile, relation_map)
        merge_relation_entries_in_db_profile(schema.db_profile)
        inverses, symmetric = remap_inverses(
            core.edge_config.inverses,
            core.edge_config.symmetric,
            relation_map,
            kind="canonicalize",
        )
        remapped = EdgeConfig(
            edges=list(core.edge_config.edges),
            inverses=inverses,
            symmetric=symmetric,
        )
        schema.core_schema = CoreSchema(
            vertex_config=new_vc, edge_config=remapped.with_edges(edges)
        )
        schema.db_profile = _revalidate_db_profile(schema.db_profile)
        schema.finish_init()

        if work.ingestion_model is not None:
            _rewrite_ingestion_for_canonicalize(
                work.ingestion_model,
                vertex_map,
                relation_map,
                declared=core.vertex_config.vertex_set,
            )
            work.ingestion_model = IngestionModel.model_validate(
                work.ingestion_model.to_dict(skip_defaults=False)
            )

        advisories: list[str] = []
        # `manifest` is untouched until the swap below, so its resources are
        # the pre-relabel view the fusion check needs.
        before_resources: list[Any] = (
            list(manifest.ingestion_model.resources)
            if manifest.ingestion_model is not None
            else []
        )
        for target in merged_targets:
            self_relations, fused_levels, target_advisories = _describe_merge_impact(
                work,
                before_edges=before_edges,
                before_resources=before_resources,
                merged=target,
                mapping=vertex_map,
            )
            sources = sorted(m for m in vertex_groups[target] if m != target)
            if self_relations and not op.allow_self_relations:
                raise ValueError(
                    f"canonicalize: merging {sources} into {target!r} turns edges "
                    f"into self-relations: {self_relations}. Both endpoints then "
                    "share one accumulator slot, so assembly merges observations "
                    "that were separate nodes. Remove or retarget those edges "
                    "first, or set allow_self_relations=true to accept the "
                    "self-relation."
                )
            if fused_levels and not op.allow_observation_fusion:
                raise ValueError(
                    f"canonicalize: merging {sources} into {target!r} leaves "
                    f"pipeline slots producing {target!r} more than once: "
                    f"{fused_levels}. Steps at one level share an accumulator "
                    "slot unless they carry distinct `role`s, so one document "
                    "that yields both fuses them into a single node. Give each "
                    "step its own `role` (and address it from the edge with "
                    "`source_role` / `target_role`), split the resource, or set "
                    "allow_observation_fusion=true if fusing them is the intent."
                )
            advisories.extend(a for a in target_advisories if a not in advisories)
        for advisory in advisories:
            logger.warning("canonicalize: %s", advisory)

    manifest.graph_schema = work.graph_schema
    manifest.ingestion_model = work.ingestion_model
    manifest.bindings = work.bindings

apply_declare_edge_inverses(manifest, op)

Record inverse pairs and symmetric relations; creates no edge.

Restating a declaration, in either order, is a no-op; giving a relation a second inverse is refused by the one table rule.

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

def apply_declare_edge_inverses(
    manifest: GraphManifest, op: DeclareEdgeInversesOp
) -> None:
    """Record inverse pairs and symmetric relations; creates no edge.

    Restating a declaration, in either order, is a no-op; giving a relation a
    second inverse is refused by the one table rule.
    """
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("declare_edge_inverses requires graph_schema")
    edge_config = schema.core_schema.edge_config
    inverses, symmetric = normalize_inverse_table(
        [(p.relation, p.inverse) for p in edge_config.inverses]
        + list(op.inverses.items()),
        [*edge_config.symmetric, *op.symmetric],
        kind="declare_edge_inverses",
    )
    _replace_edge_config(
        schema,
        EdgeConfig(
            edges=list(edge_config.edges), inverses=inverses, symmetric=symmetric
        ),
    )
    schema.finish_init()

apply_evolution(manifest, ops, *, bump_version='minor', finish_init=True, strict_references=False, dynamic_edge_feedback=False)

Return a deep copy of manifest with ops applied and optionally re-initialized.

Compare before/after contract identity with :func:graflo.migrate.io.manifest_hash (stable hash over schema, ingestion_model, and bindings blocks).

MergeManifestsOp is rejected at dispatch — use :func:~graflo.architecture.evolution.merge.merge_manifests instead.

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

def apply_evolution(
    manifest: GraphManifest,
    ops: Sequence[ManifestOp],
    *,
    bump_version: bool | Literal["minor"] = "minor",
    finish_init: bool = True,
    strict_references: bool = False,
    dynamic_edge_feedback: bool = False,
) -> GraphManifest:
    """Return a deep copy of *manifest* with *ops* applied and optionally re-initialized.

    Compare before/after contract identity with :func:`graflo.migrate.io.manifest_hash`
    (stable hash over schema, ingestion_model, and bindings blocks).

    ``MergeManifestsOp`` is rejected at dispatch — use
    :func:`~graflo.architecture.evolution.merge.merge_manifests` instead.
    """
    out = manifest.model_copy(deep=True)
    advisories_before = _inverse_advisories(manifest)

    for op in ops:
        _dispatch_op(out, op)

    for advisory in _inverse_advisories(out) - advisories_before:
        logger.warning("declared inverses: %s", advisory)

    _bump_schema_version(out, bump_version)

    if finish_init:
        out.finish_init(
            strict_references=strict_references,
            dynamic_edge_feedback=dynamic_edge_feedback,
        )
    return out

apply_manifest_ops_inplace(manifest, ops)

Apply each evolution op to manifest in place.

Does not copy the manifest, bump schema version, or call :meth:GraphManifest.finish_init. Callers that need re-validation after mutation should invoke finish_init themselves.

MergeManifestsOp is rejected at dispatch — use :func:~graflo.architecture.evolution.merge.merge_manifests instead.

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

def apply_manifest_ops_inplace(
    manifest: GraphManifest,
    ops: Sequence[ManifestOp],
) -> None:
    """Apply each evolution op to *manifest* in place.

    Does not copy the manifest, bump schema version, or call :meth:`GraphManifest.finish_init`.
    Callers that need re-validation after mutation should invoke ``finish_init`` themselves.

    ``MergeManifestsOp`` is rejected at dispatch — use
    :func:`~graflo.architecture.evolution.merge.merge_manifests` instead.
    """
    for op in ops:
        _dispatch_op(manifest, op)

apply_merge_edges(manifest, op)

Merge edge relation names into one canonical relation.

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

def apply_merge_edges(manifest: GraphManifest, op: MergeEdgesOp) -> None:
    """Merge edge relation names into one canonical relation."""
    if op.into in set(op.sources):
        raise ValueError("merge_edges: `sources` must not include `into`")
    relation_map = {source: op.into for source in op.sources}
    schema = manifest.graph_schema
    if schema is not None:
        # Collapse the schema edges *first*, on the model, so `merge_edge_pair` unions
        # properties and identities. Renaming first would leave two edges under one
        # edge_id, which `EdgeConfig` now rejects — and silently shadowed one of them
        # before it did. The profile is renamed in place ahead of the assignment
        # because assigning `core_schema` revalidates the profile against it.
        apply_relation_rename_to_db_profile(schema.db_profile, relation_map)
        merge_relation_entries_in_db_profile(schema.db_profile)
        edge_config = schema.core_schema.edge_config
        merged_edges = remap_relation_and_merge_edges(edge_config.edges, relation_map)
        inverses, symmetric = remap_inverses(
            edge_config.inverses,
            edge_config.symmetric,
            relation_map,
            kind="merge_edges",
        )
        remapped = EdgeConfig(
            edges=list(edge_config.edges),
            inverses=inverses,
            symmetric=symmetric,
        )
        schema.core_schema = CoreSchema(
            vertex_config=schema.core_schema.vertex_config,
            edge_config=remapped.with_edges(merged_edges),
        )
        schema.db_profile = _revalidate_db_profile(schema.db_profile)
    # Deliberately non-injective, so this takes the unguarded internal path. Schema and
    # profile are already collapsed; this carries the rename into ingestion. Reapplying
    # the map to the profile is a no-op because `into` is never one of `sources`.
    _rename_relations_inplace(manifest, relation_map)

apply_merge_vertices(manifest, op)

Mutate manifest in place: merge source vertices into into.

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

def apply_merge_vertices(
    manifest: GraphManifest,
    op: MergeVerticesOp,
) -> None:
    """Mutate *manifest* in place: merge source vertices into ``into``."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("merge_vertices requires graph_schema")

    sources = list(op.sources)
    into = op.into
    sset = set(sources)
    if into in sset:
        raise ValueError("merge_vertices: `into` must not appear in `sources`")

    core = schema.core_schema
    before_edges = list(core.edge_config.edges)
    new_vc = _build_merged_vertex_config(core.vertex_config, sources, into)
    m = {s: into for s in sources}
    merged_edges = redirect_and_merge_edges(core.edge_config.edges, m)

    schema.core_schema = CoreSchema(
        vertex_config=new_vc,
        edge_config=core.edge_config.with_edges(merged_edges),
    )
    apply_vertex_merge_to_db_profile(schema.db_profile, sset, into)
    schema.db_profile = _revalidate_db_profile(schema.db_profile)

    before_resources: list[Any] = []
    if manifest.ingestion_model is not None:
        # The rewrite rebuilds the resource list from dicts and never mutates
        # the old Resource objects, so holding them keeps the pre-merge view.
        before_resources = list(manifest.ingestion_model.resources)
        _rewrite_ingestion_for_merge(
            manifest.ingestion_model, m, declared=core.vertex_config.vertex_set
        )
        manifest.ingestion_model = IngestionModel.model_validate(
            manifest.ingestion_model.to_dict(skip_defaults=False)
        )

    self_relations, fused_levels, advisories = _describe_merge_impact(
        manifest,
        before_edges=before_edges,
        before_resources=before_resources,
        merged=into,
        mapping=m,
    )
    if self_relations and not op.allow_self_relations:
        raise ValueError(
            f"merge_vertices: merging {sorted(sset)} into {into!r} turns edges into "
            f"self-relations: {self_relations}. Both endpoints then share one "
            "accumulator slot, so assembly merges observations that were separate "
            "nodes. "
            "Remove or retarget those edges first, or set allow_self_relations=true "
            "to accept the self-relation."
        )
    if fused_levels and not op.allow_observation_fusion:
        raise ValueError(
            f"merge_vertices: merging {sorted(sset)} into {into!r} leaves pipeline "
            f"slots producing {into!r} more than once: {fused_levels}. Steps at "
            "one level share an accumulator slot unless they carry distinct "
            "`role`s, so one document that yields both fuses them into a single "
            "node. Give each step its own `role` (and address it from the edge "
            "with `source_role` / `target_role`), split the resource, or set "
            "allow_observation_fusion=true if fusing them is the intent."
        )
    for advisory in advisories:
        logger.warning("merge_vertices: %s", advisory)

apply_project_manifest(manifest, op)

Project manifest to surviving vertices/edges with consistent cascade.

A resource the cascade shortened is kept trimmed, with a warning naming it, or dropped under partial_resources="drop".

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

def apply_project_manifest(manifest: GraphManifest, op: ProjectManifestOp) -> None:
    """Project manifest to surviving vertices/edges with consistent cascade.

    A resource the cascade shortened is kept trimmed, with a warning naming it,
    or dropped under ``partial_resources="drop"``.
    """
    from .canonicalize import canonical_payload

    before = (
        {r.name: canonical_payload(r) for r in manifest.ingestion_model.resources}
        if manifest.ingestion_model is not None
        else {}
    )
    plan = compute_projection(manifest, op)
    if plan.removed_edge_ids:
        apply_remove_edge_ids(manifest, plan.removed_edge_ids)
    if plan.removed_vertices:
        apply_remove_vertices(
            manifest,
            RemoveVerticesOp(names=sorted(plan.removed_vertices)),
        )
    if manifest.ingestion_model is None:
        return
    present = [r.name for r in manifest.ingestion_model.resources]
    allowed = set(op.keep_resources) if op.keep_resources is not None else set(present)
    trimmed = sorted(
        r.name
        for r in manifest.ingestion_model.resources
        if r.name in allowed
        and r.name in before
        and canonical_payload(r) != before[r.name]
    )
    if trimmed and op.partial_resources == "drop":
        allowed -= set(trimmed)
    elif trimmed:
        logger.warning(
            "project_manifest trimmed resources %s to the kept types; they still "
            "read their sources. Set partial_resources='drop' to remove them.",
            trimmed,
        )
    if allowed != set(present):
        _apply_keep_resources(manifest, allowed)

apply_remove_edge_ids(manifest, removed_edge_ids)

Remove edges by logical triple and prune related references.

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

def apply_remove_edge_ids(
    manifest: GraphManifest, removed_edge_ids: set[EdgeId]
) -> None:
    """Remove edges by logical triple and prune related references."""
    if not removed_edge_ids:
        return
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("remove_edge_ids requires graph_schema")
    _splice_superproperties(schema.core_schema, removed_edge_ids)

    surviving = [
        edge
        for edge in schema.core_schema.edge_config.edges
        if edge.edge_id not in removed_edge_ids
    ]
    edge_config_before = schema.core_schema.edge_config
    apply_edge_id_removal_to_db_profile(schema.db_profile, removed_edge_ids)
    retain_native_inverses(schema.db_profile, _relations_of(surviving))
    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.core_schema = CoreSchema(
        vertex_config=schema.core_schema.vertex_config,
        edge_config=schema.core_schema.edge_config.with_edges(surviving),
    )
    schema.finish_init()

    if manifest.ingestion_model is None:
        return

    from graflo.architecture.contract.ingestion.resource import Resource

    from .ingestion import set_emit_inverse_flags, stranded_emission_flags

    # A step that mirrored into a removed inverse edge has nothing left to write
    # into. Cleared before steps are dropped, while step positions still hold.
    stranded = stranded_emission_flags(
        manifest, edge_config_before, schema.core_schema.edge_config
    )
    if stranded:
        set_emit_inverse_flags(manifest, stranded, False)

    resources: list[Resource] = []
    for resource in manifest.ingestion_model.resources:
        payload = resource.to_dict(skip_defaults=False)
        pipeline = payload.get("pipeline")
        if isinstance(pipeline, list):
            payload["pipeline"] = rewrite_remove_edge_ids_in_pipeline(
                pipeline, removed_edge_ids
            )
        for key in ("infer_edge_only", "infer_edge_except"):
            specs = payload.get(key)
            if isinstance(specs, list):
                payload[key] = [
                    spec
                    for spec in specs
                    if _edge_id_from_resource_spec(spec) not in removed_edge_ids
                ]
        extra_weights = payload.get("extra_weights")
        if isinstance(extra_weights, list):
            payload["extra_weights"] = [
                entry
                for entry in extra_weights
                if _edge_id_from_resource_spec(entry) not in removed_edge_ids
            ]
        resources.append(Resource.model_validate(payload))
    manifest.ingestion_model.resources = resources
    manifest.ingestion_model = IngestionModel.model_validate(
        manifest.ingestion_model.to_dict(skip_defaults=False)
    )

apply_remove_edge_properties(manifest, op)

Remove edge properties by relation or by triple and clean references.

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

def apply_remove_edge_properties(
    manifest: GraphManifest, op: RemoveEdgePropertiesOp
) -> None:
    """Remove edge properties by relation or by triple and clean references."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("remove_edge_properties requires graph_schema")
    removals = {relation: set(fields) for relation, fields in op.removals.items()}
    removals_by_edge = {entry.edge_id(): set(entry.fields) for entry in op.per_edge}
    _refuse_unknown_edge_properties(
        "remove_edge_properties", schema, removals, removals_by_edge
    )
    for edge in schema.core_schema.edge_config.edges:
        remove_fields = op.removals_for(edge.edge_id)
        if not remove_fields:
            continue
        blocked_tokens = set().union(
            *[
                set(identity) - {"source", "target", "relation"}
                for identity in edge.identities
            ]
        )
        overlap = sorted(blocked_tokens & remove_fields)
        if overlap:
            raise ValueError(
                "remove_edge_properties cannot remove identity fields "
                f"for edge {edge.edge_id}: {overlap}"
            )
        edge.properties = [
            field for field in edge.properties if field.name not in remove_fields
        ]
    apply_edge_property_removal_to_db_profile(
        schema.db_profile, removals, removals_by_edge=removals_by_edge
    )
    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.finish_init()
    _rebuild_ingestion_with_pipeline_rewrite(
        manifest,
        lambda pipeline: rewrite_edge_properties_in_pipeline(
            pipeline, removals_by_relation=removals, removals_by_edge=removals_by_edge
        ),
    )

apply_remove_edges(manifest, op)

Remove edges by relation name and/or by triple, pruning related references.

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

def apply_remove_edges(manifest: GraphManifest, op: RemoveEdgesOp) -> None:
    """Remove edges by relation name and/or by triple, pruning related references."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("remove_edges requires graph_schema")
    if op.edges:
        wanted = {selector.edge_id() for selector in op.edges}
        existing = {edge.edge_id for edge in schema.core_schema.edge_config.edges}
        unknown = sorted(wanted - existing, key=str)
        if unknown:
            raise ValueError(f"remove_edges: unknown edges: {unknown}")
        apply_remove_edge_ids(manifest, wanted)
    removed = set(op.relations)
    if not removed:
        return
    _splice_superproperties(
        schema.core_schema,
        {
            edge.edge_id
            for edge in schema.core_schema.edge_config.edges
            if edge.relation in removed
        },
    )
    apply_relation_removal_to_db_profile(schema.db_profile, removed)
    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.core_schema = CoreSchema(
        vertex_config=schema.core_schema.vertex_config,
        edge_config=schema.core_schema.edge_config.with_edges(
            [
                edge
                for edge in schema.core_schema.edge_config.edges
                if edge.relation not in removed
            ]
        ),
    )
    schema.finish_init()

    if manifest.ingestion_model is not None:
        from graflo.architecture.contract.ingestion.resource import Resource

        resources: list[Resource] = []
        for resource in manifest.ingestion_model.resources:
            payload = resource.to_dict(skip_defaults=False)
            pipeline = payload.get("pipeline")
            if isinstance(pipeline, list):
                payload["pipeline"] = rewrite_remove_relations_in_pipeline(
                    pipeline, removed
                )
            for key in ("infer_edge_only", "infer_edge_except"):
                specs = payload.get(key)
                if isinstance(specs, list):
                    payload[key] = [
                        spec
                        for spec in specs
                        if not (
                            isinstance(spec, dict) and spec.get("relation") in removed
                        )
                    ]
            extra_weights = payload.get("extra_weights")
            if isinstance(extra_weights, list):
                payload["extra_weights"] = [
                    entry
                    for entry in extra_weights
                    if not (
                        isinstance(entry, dict)
                        and isinstance(entry.get("edge"), dict)
                        and entry["edge"].get("relation") in removed
                    )
                ]
            resources.append(Resource.model_validate(payload))
        manifest.ingestion_model.resources = resources
        manifest.ingestion_model = IngestionModel.model_validate(
            manifest.ingestion_model.to_dict(skip_defaults=False)
        )

apply_remove_vertex_properties(manifest, op)

Remove vertex properties and clean up ingestion/db profile references.

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

def apply_remove_vertex_properties(
    manifest: GraphManifest, op: RemoveVertexPropertiesOp
) -> None:
    """Remove vertex properties and clean up ingestion/db profile references."""
    if not op.removals:
        return
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("remove_vertex_properties requires graph_schema")

    unknown_vertices = sorted(
        set(op.removals) - schema.core_schema.vertex_config.vertex_set
    )
    if unknown_vertices:
        raise ValueError(
            f"remove_vertex_properties: unknown vertices in removals: {unknown_vertices}"
        )

    removals = {vertex_name: set(fields) for vertex_name, fields in op.removals.items()}

    # A name the vertex never declared is a mistake in the change set, as an
    # unknown vertex is: skipping it would record a removal that removed nothing.
    unknown_fields = {
        vertex.name: missing
        for vertex in schema.core_schema.vertex_config.vertices
        if (
            missing := sorted(
                removals.get(vertex.name, set()) - {f.name for f in vertex.properties}
            )
        )
    }
    if unknown_fields:
        raise ValueError(
            f"remove_vertex_properties: unknown properties in removals: {unknown_fields}"
        )

    for vertex in schema.core_schema.vertex_config.vertices:
        remove_fields = removals.get(vertex.name, set())
        if not remove_fields:
            continue
        # Every identity plane, not only the primary key: a hash property, a
        # funnel branch field, or a secondary key that loses a field leaves a
        # key that can no longer be computed or selected.
        for plane, fields in _identity_planes(vertex):
            overlap = sorted(fields & remove_fields)
            if overlap:
                raise ValueError(
                    f"remove_vertex_properties cannot remove {plane} fields "
                    f"for vertex {vertex.name}: {overlap}"
                )
        vertex.properties = [
            field for field in vertex.properties if field.name not in remove_fields
        ]

    for vertex_name, indexes in list(schema.db_profile.vertex_indexes.items()):
        remove_fields = removals.get(vertex_name, set())
        if not remove_fields:
            continue
        updated_indexes = []
        for index in indexes:
            fields = [field for field in index.fields if field not in remove_fields]
            if fields:
                updated_indexes.append(index.model_copy(update={"fields": fields}))
        schema.db_profile.vertex_indexes[vertex_name] = updated_indexes

    for edge_spec in schema.db_profile.edge_specs:
        updated_indexes = []
        for index in edge_spec.indexes:
            fields = list(index.fields)
            source_removals = removals.get(edge_spec.source, set())
            target_removals = removals.get(edge_spec.target, set())
            if source_removals:
                fields = [field for field in fields if field not in source_removals]
            if target_removals:
                fields = [field for field in fields if field not in target_removals]
            if fields:
                updated_indexes.append(index.model_copy(update={"fields": fields}))
        edge_spec.indexes = updated_indexes

    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.finish_init()

    if manifest.ingestion_model is not None:
        _rebuild_ingestion_with_pipeline_rewrite(
            manifest,
            lambda pipeline: rewrite_remove_vertex_properties_in_pipeline(
                pipeline,
                removals,
                declared=schema.core_schema.vertex_config.vertex_set,
            ),
        )
        for resource in manifest.ingestion_model.resources:
            if resource.extra_weights:
                for entry in resource.extra_weights:
                    for weight in entry.vertex_weights:
                        if not isinstance(weight.name, str):
                            continue
                        remove_fields = removals.get(weight.name, set())
                        if not remove_fields:
                            continue
                        weight.fields = [
                            field
                            for field in weight.fields
                            if field not in remove_fields
                        ]
                        weight.map = {
                            key: value
                            for key, value in weight.map.items()
                            if key not in remove_fields
                        }
                        weight.filter = {
                            key: value
                            for key, value in weight.filter.items()
                            if key not in remove_fields
                        }

apply_remove_vertices(manifest, op)

Mutate manifest in place: cascade-remove vertices (schema, ingestion, bindings).

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

def apply_remove_vertices(manifest: GraphManifest, op: RemoveVerticesOp) -> None:
    """Mutate *manifest* in place: cascade-remove vertices (schema, ingestion, bindings)."""
    removed = set(op.names)
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("remove_vertices requires graph_schema")

    core = schema.core_schema
    missing = removed - core.vertex_config.vertex_set
    if missing:
        raise ValueError(f"Unknown vertices to remove: {sorted(missing)}")
    # A removed class or relation is spliced out of the hierarchy: what was
    # under it moves under its parents, so the survivors' subsumptions hold.
    parents = spliced_superclasses(core.vertex_config.vertices, removed)
    cascaded = {
        e.edge_id
        for e in core.edge_config.edges
        if e.source in removed or e.target in removed
    }
    edge_parents = spliced_superproperties(core.edge_config.edges, cascaded)

    core.vertex_config.remove_vertices(removed)
    for vertex in core.vertex_config.vertices:
        if vertex.name in parents:
            vertex.subclass_of = parents[vertex.name]
    filtered_edges = [
        e
        for e in core.edge_config.edges
        if e.source not in removed and e.target not in removed
    ]
    for edge in filtered_edges:
        if edge.edge_id in edge_parents:
            edge.subproperty_of = edge_parents[edge.edge_id]
    retain_native_inverses(schema.db_profile, _relations_of(filtered_edges))
    schema.core_schema = CoreSchema(
        vertex_config=core.vertex_config,
        edge_config=core.edge_config.with_edges(filtered_edges),
    )

    apply_vertex_removal_to_db_profile(schema.db_profile, removed)
    schema.db_profile = _revalidate_db_profile(schema.db_profile)

    if manifest.ingestion_model is not None:
        surviving_vertices = set(core.vertex_config.vertex_set)
        prune_connectors_for_removed_vertices(
            manifest, removed, declared=surviving_vertices | removed
        )
        _prune_ingestion_for_removed_vertices(
            manifest.ingestion_model,
            removed,
            surviving=surviving_vertices,
        )
        manifest.ingestion_model = IngestionModel.model_validate(
            manifest.ingestion_model.to_dict(skip_defaults=False)
        )
        surviving = {r.name for r in manifest.ingestion_model.resources}
        prune_bindings_for_resources(manifest, surviving)

apply_rename_edge_properties(manifest, op)

Rename edge properties by relation or by triple and propagate references.

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

def apply_rename_edge_properties(
    manifest: GraphManifest, op: RenameEdgePropertiesOp
) -> None:
    """Rename edge properties by relation or by triple and propagate references."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("rename_edge_properties requires graph_schema")
    _refuse_unknown_edge_properties(
        "rename_edge_properties",
        schema,
        {relation: set(renames) for relation, renames in op.renames.items()},
        {entry.edge_id(): set(entry.renames) for entry in op.per_edge},
    )
    for edge in schema.core_schema.edge_config.edges:
        if renames := op.renames_for(edge.edge_id):
            _refuse_folding_renames(
                "rename_edge_properties",
                f"edge {edge.edge_id}",
                [field.name for field in edge.properties],
                renames,
            )
    for edge in schema.core_schema.edge_config.edges:
        renames = op.renames_for(edge.edge_id)
        if not renames:
            continue
        seen: set[str] = set()
        new_properties: list[Field] = []
        for field in edge.properties:
            new_name = renames.get(field.name, field.name)
            if new_name in seen:
                continue
            seen.add(new_name)
            new_properties.append(field.model_copy(update={"name": new_name}))
        edge.properties = new_properties
        edge.identities = [
            [
                renames.get(token, token)
                if token not in {"source", "target", "relation"}
                else token
                for token in identity
            ]
            for identity in edge.identities
        ]
    renames_by_edge = {entry.edge_id(): dict(entry.renames) for entry in op.per_edge}
    apply_edge_property_rename_to_db_profile(
        schema.db_profile, op.renames, renames_by_edge=renames_by_edge
    )
    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.finish_init()

    _rebuild_ingestion_with_pipeline_rewrite(
        manifest,
        lambda pipeline: rewrite_edge_properties_in_pipeline(
            pipeline, renames_by_relation=op.renames, renames_by_edge=renames_by_edge
        ),
    )

apply_rename_relations(manifest, op)

Rename logical relation names across schema/ingestion/db profile.

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

def apply_rename_relations(manifest: GraphManifest, op: RenameRelationsOp) -> None:
    """Rename logical relation names across schema/ingestion/db profile."""
    schema = manifest.graph_schema
    if schema is not None:
        known_relations = {
            edge.relation
            for edge in schema.core_schema.edge_config.edges
            if edge.relation is not None
        }
        unknown = sorted(set(op.renames) - known_relations)
        if unknown:
            raise ValueError(f"rename_relations: unknown relations: {unknown}")
        # A relation name is only unique per (source, target) pair, so the
        # collision guard is at the edge-id level -- unlike vertices, a target
        # relation name reused on a *different* pair is not a collision. Only
        # ids that survive the rename can be collided with: the map applies in
        # one step, so a chain like ``{x: z, z: q}`` on one pair is legal.
        edges = schema.core_schema.edge_config.edges
        surviving_ids = {
            edge.edge_id for edge in edges if edge.relation not in op.renames
        }
        collisions = sorted(
            f"({edge.source}, {edge.target}, {edge.relation!r} -> "
            f"{op.renames[edge.relation]!r})"
            for edge in edges
            if edge.relation in op.renames
            and (edge.source, edge.target, op.renames[edge.relation]) in surviving_ids
        )
        if collisions:
            raise ValueError(
                f"rename_relations: renamed relations collide with an existing "
                f"edge: {collisions}. Renaming cannot merge edges — merge them "
                "explicitly instead."
            )
    _rename_relations_inplace(manifest, op.renames)

apply_rename_resources(manifest, op)

Rename ingestion resources and bindings references.

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

def apply_rename_resources(manifest: GraphManifest, op: RenameResourcesOp) -> None:
    """Rename ingestion resources and bindings references."""
    if manifest.ingestion_model is not None:
        _validate_rename_against_existing(
            op.renames,
            {resource.name for resource in manifest.ingestion_model.resources},
            kind="rename_resources",
            noun="resources",
        )
    _apply_rename_entities(manifest, resource_map=op.renames)

apply_rename_vertex_properties(manifest, op)

Rename vertex properties (and their references) across the manifest.

Mutates manifest in place:

• Rewrites schema Field.name and vertex.identity.
• Rewrites :class:DatabaseProfile field references (vertex_indexes, edge_specs.indexes, default_property_values).
• Rewrites resource pipelines so that VertexActor.from covers the rename and TransformActor.rename produces the renamed property (see :func:rewrite_vertex_field_names_in_pipeline).
• Rewrites Resource.extra_weights / vertex_weights (and any vertex_weights embedded in edge pipeline steps).

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

def apply_rename_vertex_properties(
    manifest: GraphManifest, op: RenameVertexPropertiesOp
) -> None:
    """Rename vertex properties (and their references) across the manifest.

    Mutates *manifest* in place:

    - Rewrites schema ``Field.name`` and ``vertex.identity``.
    - Rewrites :class:`DatabaseProfile` field references (vertex_indexes,
      edge_specs.indexes, default_property_values).
    - Rewrites resource pipelines so that ``VertexActor.from`` covers the
      rename and ``TransformActor.rename`` produces the renamed property
      (see :func:`rewrite_vertex_field_names_in_pipeline`).
    - Rewrites ``Resource.extra_weights`` / ``vertex_weights`` (and any
      ``vertex_weights`` embedded in ``edge`` pipeline steps).
    """
    if not op.renames:
        return
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("rename_vertex_properties requires graph_schema")

    unknown = sorted(set(op.renames) - schema.core_schema.vertex_config.vertex_set)
    if unknown:
        raise ValueError(
            f"rename_vertex_properties: unknown vertices in renames: {unknown}"
        )
    for vertex in schema.core_schema.vertex_config.vertices:
        per_vertex = op.renames.get(vertex.name) or {}
        declared = [field.name for field in vertex.properties]
        missing = sorted(set(per_vertex) - set(declared))
        if missing:
            raise ValueError(
                "rename_vertex_properties: unknown properties in renames for "
                f"{vertex.name}: {missing}"
            )
        _refuse_folding_renames(
            "rename_vertex_properties", f"vertex {vertex.name}", declared, per_vertex
        )

    _rename_fields_in_schema(schema, op.renames)
    apply_field_rename_to_db_profile(schema.db_profile, op.renames)
    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.finish_init()

    _rebuild_ingestion_with_pipeline_rewrite(
        manifest,
        lambda pipeline: rewrite_vertex_field_names_in_pipeline(
            pipeline,
            op.renames,
            declared=schema.core_schema.vertex_config.vertex_set,
        ),
        vertex_field_renames=op.renames,
    )

apply_rename_vertices(manifest, op)

Rename logical vertex names across schema/ingestion/bindings.

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

def apply_rename_vertices(manifest: GraphManifest, op: RenameVerticesOp) -> None:
    """Rename logical vertex names across schema/ingestion/bindings."""
    schema = manifest.graph_schema
    if schema is not None:
        _validate_rename_against_existing(
            op.renames,
            set(schema.core_schema.vertex_config.vertex_set),
            kind="rename_vertices",
            noun="vertices",
        )
    _rename_vertices_inplace(manifest, op.renames)

apply_retract_edge_inverses(manifest, op)

Withdraw declarations by relation name; refused while a native inverse realizes one.

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

def apply_retract_edge_inverses(
    manifest: GraphManifest, op: RetractEdgeInversesOp
) -> None:
    """Withdraw declarations by relation name; refused while a native inverse realizes one."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("retract_edge_inverses requires graph_schema")
    edge_config = schema.core_schema.edge_config
    declared = inverse_map(edge_config.inverses, edge_config.symmetric)
    unknown = sorted(set(op.relations) - set(declared))
    if unknown:
        raise ValueError(
            f"retract_edge_inverses: no declared inverse for relations: {unknown}"
        )
    retracted = {name for r in op.relations for name in (r, declared[r])}
    native = sorted(retracted & set(schema.db_profile.native_inverses))
    if native:
        raise ValueError(
            "retract_edge_inverses: pairs are still realized by native inverses of "
            f"{native}; withdraw them first (set_native_inverses enabled=false)"
        )
    feeding = _steps_mirroring(manifest, retracted)
    if feeding:
        raise ValueError(
            "retract_edge_inverses: edge steps still mirror these relations into "
            f"their inverse (emit_inverse): {feeding}; clear the flags first "
            "(set_inverse_emission enabled=false), or remove the inverse edges "
            "(remove_edges), which clears them"
        )
    _replace_edge_config(
        schema,
        EdgeConfig(
            edges=list(edge_config.edges),
            inverses=[
                pair.model_copy(deep=True)
                for pair in edge_config.inverses
                if pair.relation not in retracted
            ],
            symmetric=[name for name in edge_config.symmetric if name not in retracted],
        ),
    )
    schema.finish_init()

apply_sanitize(manifest, op)

Record the target flavor's physical names in the profile, in place.

Only schema.db_profile changes: vertex storage names, relation names and vertex/edge property names the flavor cannot store get a stored name there (see :func:~graflo.architecture.evolution.sanitize.assign_physical_names). The logical schema and the ingestion model are untouched, so documents keep their logical keys and the writer translates them at the database call.

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

def apply_sanitize(manifest: GraphManifest, op: SanitizeOp) -> None:
    """Record the target flavor's physical names in the profile, in place.

    Only ``schema.db_profile`` changes: vertex storage names, relation names
    and vertex/edge property names the flavor cannot store get a stored name
    there (see :func:`~graflo.architecture.evolution.sanitize.assign_physical_names`).
    The logical schema and the ingestion model are untouched, so documents keep
    their logical keys and the writer translates them at the database call.
    """
    from graflo.db.util import load_reserved_words

    if manifest.graph_schema is None:
        return

    schema = manifest.graph_schema
    if op.reserved_words is not None:
        reserved_words = {word.upper() for word in op.reserved_words}
    else:
        reserved_words = load_reserved_words(op.db_flavor)

    assign_physical_names(schema, reserved_words, db_flavor=op.db_flavor)
    schema.db_profile = _revalidate_db_profile(schema.db_profile)
    schema.finish_init()

merge_fused_slots(resources, *, merged, mapping)

Accumulator slots of the pre-merge resources where merged would fuse two members.

One source document then yields two members that land in one slot, and assembly folds them into a single node.

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

def merge_fused_slots(
    resources: Sequence[Any], *, merged: str, mapping: Mapping[str, str]
) -> list[str]:
    """Accumulator slots of the pre-merge *resources* where *merged* would fuse two members.

    One source document then yields two members that land in one slot, and
    assembly folds them into a single node.
    """
    out: list[str] = []
    for resource in resources:
        out.extend(
            f"{resource.name}: {hit}"
            for hit in _fused_slots(
                resource.pipeline, merged=merged, mapping=dict(mapping)
            )
        )
    return out

merge_self_relations(edges, mapping)

Edges between two distinct vertices that mapping sends to one name.

edges are the pre-merge schema edges. Each such edge becomes an edge from the merged vertex to itself.

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

def merge_self_relations(
    edges: Sequence[Edge], mapping: Mapping[str, str]
) -> list[str]:
    """Edges between two distinct vertices that *mapping* sends to one name.

    *edges* are the pre-merge schema edges. Each such edge becomes an edge from
    the merged vertex to itself.
    """
    return sorted(
        f"({edge.source}, {edge.target}, {edge.relation}) -> "
        f"({mapping.get(edge.source, edge.source)}, "
        f"{mapping.get(edge.target, edge.target)}, {edge.relation})"
        for edge in edges
        if edge.source != edge.target
        and mapping.get(edge.source, edge.source)
        == mapping.get(edge.target, edge.target)
    )

prune_bindings_for_resources(manifest, surviving)

Drop the bindings that served only resources outside surviving, in place.

resource_connector rows keep only surviving resources. A connector bound to resources (by resource_name or by a row) and to none that survives is dropped with its connector_connection rows; a connector shared with a surviving resource, or bound to no resource at all, stays.

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

def prune_bindings_for_resources(manifest: GraphManifest, surviving: set[str]) -> None:
    """Drop the bindings that served only resources outside *surviving*, in place.

    ``resource_connector`` rows keep only surviving resources. A connector bound
    to resources (by ``resource_name`` or by a row) and to none that survives is
    dropped with its ``connector_connection`` rows; a connector shared with a
    surviving resource, or bound to no resource at all, stays.
    """
    from graflo.architecture.contract.bindings import Bindings
    from graflo.architecture.contract.bindings.core import ResourceConnectorBinding

    bindings = manifest.bindings
    if bindings is None:
        return
    rows = [
        ResourceConnectorBinding.model_validate(row) if isinstance(row, dict) else row
        for row in bindings.resource_connector
    ]
    bound: dict[str, set[str]] = {}
    for connector in bindings.connectors:
        if connector.resource_name is not None:
            bound.setdefault(connector.hash, set()).add(connector.resource_name)
    for row in rows:
        for connector_hash in bindings.resolve_connector_refs_to_hashes(
            [row.connector]
        ):
            bound.setdefault(connector_hash, set()).add(row.resource)
    dropped = {
        connector_hash
        for connector_hash, resources in bound.items()
        if not resources & surviving
    }

    data = bindings.to_dict(skip_defaults=False)
    data["resource_connector"] = [
        row.to_dict() for row in rows if row.resource in surviving
    ]
    data["connectors"] = [
        connector.to_dict(skip_defaults=False)
        for connector in bindings.connectors
        if connector.hash not in dropped
    ]
    data["connector_connection"] = [
        row.to_dict()
        for row in bindings.connector_connection_bindings
        if not bindings.resolve_connector_refs_to_hashes([row.connector]) & dropped
    ]
    manifest.bindings = Bindings.model_validate(data)

prune_connectors_for_removed_vertices(manifest, removed, *, declared)

Drop what the bindings fetch only for classes in removed, in place.

Runs before the resources are trimmed. Each request a connector makes puts constants on its rows (:meth:ResourceConnector.row_constants); routed through the pipeline of every resource the connector is bound to, they reach some classes of declared. A request is dead when it reached some class and every class it reached, in every bound resource, is removed. An enumerated connector loses each value all of whose requests are dead, and is dropped when a variable has no value left; any other connector is dropped when its request is dead. A connector bound to no resource, or named as another connector's parent, is left alone.

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

def prune_connectors_for_removed_vertices(
    manifest: GraphManifest, removed: set[str], *, declared: Collection[str]
) -> None:
    """Drop what the bindings fetch only for classes in *removed*, in place.

    Runs before the resources are trimmed. Each request a connector makes puts
    constants on its rows (:meth:`ResourceConnector.row_constants`); routed
    through the pipeline of every resource the connector is bound to, they
    reach some classes of *declared*. A request is dead when it reached some
    class and every class it reached, in every bound resource, is removed. An
    enumerated connector loses each value all of whose requests are dead, and
    is dropped when a variable has no value left; any other connector is
    dropped when its request is dead. A connector bound to no resource, or
    named as another connector's parent, is left alone.
    """
    from graflo.architecture.contract.bindings import (
        APIConnector,
        Bindings,
        TemplatedConnector,
    )
    from graflo.architecture.contract.bindings.core import ResourceConnectorBinding

    bindings = manifest.bindings
    ingestion = manifest.ingestion_model
    if bindings is None or ingestion is None or not removed:
        return
    pipelines = {resource.name: resource.pipeline for resource in ingestion.resources}
    rows = [
        ResourceConnectorBinding.model_validate(row) if isinstance(row, dict) else row
        for row in bindings.resource_connector
    ]
    bound: dict[str, set[str]] = {}
    for connector in bindings.connectors:
        if connector.resource_name is not None:
            bound.setdefault(connector.hash, set()).add(connector.resource_name)
    for row in rows:
        for connector_hash in bindings.resolve_connector_refs_to_hashes(
            [row.connector]
        ):
            bound.setdefault(connector_hash, set()).add(row.resource)
    parents = {
        parent.hash
        for connector in bindings.connectors
        if isinstance(connector, APIConnector)
        and (parent := bindings.get_parent_connector(connector)) is not None
    }

    def dead(constants: Mapping[str, Any], resources: set[str]) -> bool:
        reaches = [
            _row_reach(pipelines[name], constants, declared)
            for name in resources
            if name in pipelines
        ]
        return bool(reaches) and all(reach and reach <= removed for reach in reaches)

    dropped: set[str] = set()
    replaced: dict[str, Any] = {}
    for connector in bindings.connectors:
        resources = bound.get(connector.hash)
        if not resources or connector.hash in parents:
            continue
        if not isinstance(connector, TemplatedConnector) or not connector.values:
            if all(dead(c, resources) for c in connector.row_constants()):
                dropped.add(connector.hash)
            continue
        combinations = connector.row_constants()
        kept: dict[str, list[Any]] = {}
        for name, items in connector.values.items():
            kept[name] = [
                item
                for item in items
                if not all(
                    dead(c, resources)
                    for c in combinations
                    if type(c[name]) is type(item) and c[name] == item
                )
            ]
        if any(not items for items in kept.values()):
            dropped.add(connector.hash)
        elif kept != connector.values:
            payload = connector.to_dict(skip_defaults=False)
            payload["values"] = kept
            replaced[connector.hash] = type(connector).model_validate(payload)

    if not dropped and not replaced:
        return

    def ref(reference: str) -> str | None:
        """*reference* as it should read now; ``None`` when its connector is gone."""
        hashes = bindings.resolve_connector_refs_to_hashes([reference])
        if hashes & dropped:
            return None
        for old_hash in hashes:
            if old_hash in replaced and reference == old_hash:
                return replaced[old_hash].hash
        return reference

    data = bindings.to_dict(skip_defaults=False)
    data["connectors"] = [
        (replaced.get(c.hash) or c).to_dict(skip_defaults=False)
        for c in bindings.connectors
        if c.hash not in dropped
    ]
    data["resource_connector"] = [
        {**row.to_dict(), "connector": new}
        for row in rows
        if (new := ref(row.connector)) is not None
    ]
    data["connector_connection"] = [
        {**row.to_dict(), "connector": new}
        for row in bindings.connector_connection_bindings
        if (new := ref(row.connector)) is not None
    ]
    manifest.bindings = Bindings.model_validate(data)

relabel_vertex_fields(vertex, renames)

One vertex with its attribute names rewritten, as a new validated model.

Every field-set that names a property follows the rename: properties, identity, hash_identity_properties, each funnel branch's fields and when_all_present, each secondary identity's fields, and the leaf fields of filters.

A property whose new name is already taken is dropped, not appended. That is not a shortcut -- _check_property_renames has already refused a genuine rename collision by then, naming it as one. Appending instead would hand :meth:Vertex.set_identity two fields of one name, and the union_field_lists it runs would report a type conflict for what is really a collision, misclassified and wrapped by pydantic besides.

Pure: the argument is untouched, and the result is built in one model_validate rather than by assignment, so there is no intermediate state in which validate_assignment can re-add a stale pre-rename name as an untyped ghost property.

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

def relabel_vertex_fields(vertex: Vertex, renames: Mapping[str, str]) -> Vertex:
    """One vertex with its attribute names rewritten, as a new validated model.

    Every field-set that names a property follows the rename: ``properties``,
    ``identity``, ``hash_identity_properties``, each funnel branch's ``fields``
    and ``when_all_present``, each secondary identity's ``fields``, and the
    leaf fields of ``filters``.

    **A property whose new name is already taken is dropped, not appended.**
    That is not a shortcut -- ``_check_property_renames`` has already refused a
    genuine rename collision by then, naming it as one. Appending instead would
    hand :meth:`Vertex.set_identity` two fields of one name, and the
    ``union_field_lists`` it runs would report a *type conflict* for what is
    really a collision, misclassified and wrapped by pydantic besides.

    Pure: the argument is untouched, and the result is built in one
    ``model_validate`` rather than by assignment, so there is no intermediate
    state in which ``validate_assignment`` can re-add a stale pre-rename name
    as an untyped ghost property.
    """
    if not renames:
        return vertex

    payload = vertex.to_dict(skip_defaults=False)
    payload["identity"] = _rename_field_list(vertex.identity, renames)
    payload["hash_identity_properties"] = _rename_field_list(
        vertex.hash_identity_properties, renames
    )
    if vertex.identity_funnel is not None:
        payload["identity_funnel"] = vertex.identity_funnel.model_copy(
            update={
                "branches": [
                    branch.model_copy(
                        update={
                            "fields": _rename_field_list(branch.fields, renames),
                            "when_all_present": (
                                _rename_field_list(branch.when_all_present, renames)
                                if branch.when_all_present is not None
                                else None
                            ),
                        }
                    )
                    for branch in vertex.identity_funnel.branches
                ]
            }
        ).to_dict(skip_defaults=False)
    if vertex.secondary_identities:
        payload["secondary_identities"] = [
            entry.model_copy(
                update={"fields": _rename_field_list(entry.fields, renames)}
            ).to_dict(skip_defaults=False)
            for entry in vertex.secondary_identities
        ]
    if vertex.filters:
        payload["filters"] = [
            expression.rename_fields(renames) for expression in vertex.filters
        ]

    new_properties: list[Field] = []
    seen_names: set[str] = set()
    for field in vertex.properties:
        new_name = renames.get(field.name, field.name)
        if new_name in seen_names:
            continue
        seen_names.add(new_name)
        new_properties.append(
            field
            if new_name == field.name
            else field.model_copy(update={"name": new_name})
        )
    payload["properties"] = [f.to_dict(skip_defaults=False) for f in new_properties]
    return Vertex.model_validate(payload)