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.

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."""
    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}")
    for edge in edges:
        additions = (
            op.additions.get(edge.relation, []) if edge.relation is not None else []
        )
        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
            )
            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
            )
            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)
        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.

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."""
    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 op.keep_resources is not None:
        _apply_keep_resources(manifest, set(op.keep_resources))

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")

    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 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 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()}
    _refuse_unknown_edge_properties("remove_edge_properties", schema, removals)
    for edge in schema.core_schema.edge_config.edges:
        remove_fields = (
            removals.get(edge.relation, set()) if edge.relation is not None else set()
        )
        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 relation {edge.relation}: {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)
    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
        ),
    )

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
    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
            ),
        )
        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)}")

    core.vertex_config.remove_vertices(removed)
    filtered_edges = [
        e
        for e in core.edge_config.edges
        if e.source not in removed and e.target not in removed
    ]
    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:
        _prune_ingestion_for_removed_vertices(
            manifest.ingestion_model,
            removed,
            surviving=set(core.vertex_config.vertex_set),
        )
        manifest.ingestion_model = IngestionModel.model_validate(
            manifest.ingestion_model.to_dict(skip_defaults=False)
        )
        surviving = {r.name for r in manifest.ingestion_model.resources}
        _filter_bindings_for_resources(manifest, surviving)

apply_rename_edge_properties(manifest, op)

Rename edge properties by relation 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 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()},
    )
    for edge in schema.core_schema.edge_config.edges:
        if edge.relation is not None and edge.relation in op.renames:
            _refuse_folding_renames(
                "rename_edge_properties",
                f"edge {edge.edge_id}",
                [field.name for field in edge.properties],
                op.renames[edge.relation],
            )
    for edge in schema.core_schema.edge_config.edges:
        per_relation = (
            op.renames.get(edge.relation, {}) if edge.relation is not None else {}
        )
        if not per_relation:
            continue
        seen: set[str] = set()
        new_properties: list[Field] = []
        for field in edge.properties:
            new_name = per_relation.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 = [
            [
                per_relation.get(token, token)
                if token not in {"source", "target", "relation"}
                else token
                for token in identity
            ]
            for identity in edge.identities
        ]
    apply_edge_property_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_edge_properties_in_pipeline(
            pipeline, renames_by_relation=op.renames
        ),
    )

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),
        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)

Apply DB-flavor-specific sanitization to manifest in place.

Merges:

1. Storage-name sanitization on :class:DatabaseProfile.
2. Reserved-word vertex field renames (via apply_rename_vertex_properties).
3. TigerGraph identity normalization (cross-relation), propagated to ingestion via the same field-rename code path.

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

def apply_sanitize(manifest: GraphManifest, op: SanitizeOp) -> None:
    """Apply DB-flavor-specific sanitization to *manifest* in place.

    Merges:

    1. Storage-name sanitization on :class:`DatabaseProfile`.
    2. Reserved-word vertex field renames (via ``apply_rename_vertex_properties``).
    3. TigerGraph identity normalization (cross-relation), propagated to
       ingestion via the same field-rename code path.
    """
    from graflo.db.util import load_reserved_words
    from graflo.onto import DBType

    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)

    run_name_sanitization = bool(reserved_words) or op.db_flavor == DBType.TIGERGRAPH
    if run_name_sanitization:
        apply_storage_name_sanitization_to_db_profile(
            schema.db_profile,
            schema,
            reserved_words,
            db_flavor=op.db_flavor,
        )
        schema.db_profile = _revalidate_db_profile(schema.db_profile)

        field_renames = compute_vertex_field_renames(
            schema, reserved_words, db_flavor=op.db_flavor
        )
        if field_renames:
            apply_rename_vertex_properties(
                manifest,
                RenameVertexPropertiesOp(renames=field_renames),
            )

    identity_renames = normalize_relation_identity(schema, op.db_flavor)
    if identity_renames:
        apply_field_rename_to_db_profile(schema.db_profile, identity_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, identity_renames
            ),
            vertex_field_renames=identity_renames,
        )

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, and each secondary identity's fields.

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``, and each secondary identity's ``fields``.

    **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
        ]

    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)