graflo.architecture.evolution.inverse_plan

Planning what to do about declared inverses: realize, repair, switch, withdraw.

Pure planners in the manner of :mod:graflo.architecture.evolution.state_core.plan: each reads a manifest, emits primitive :data:~graflo.architecture.evolution.ops.ManifestOp values and applies nothing. The op list is the reviewable artifact, apply_evolution applies it, invert_ops undoes it, and a commit records exactly those primitives -- so a plan replays without this module.

What a planner adds over calling the ops by hand is the part that is awkward by hand: deciding which relations an op may name, in which order the ops must run, and saying what was left out and why. Each plan is checked against a copy of the manifest before it is returned, so an op that would be refused is reported as skipped rather than handed to the caller to trip over.

The line the planners hold is the one the audit draws (:func:graflo.architecture.profile.inverses.audit_inverses): what one place merely under-reports is propagated, and what two places contradict is listed and left alone.

Attributes

Realization = Literal['native', 'materialized'] module-attribute

The two ways a declared pair can be stored. A pair with neither is simply declared, which is the default and needs nothing.

RealizeStrategy = Literal['auto', 'native', 'materialized'] module-attribute

__all__ = ['InversePlan', 'Realization', 'RealizeStrategy', 'Skipped', 'plan_declare_symmetric', 'plan_realize_inverses', 'plan_repair_inverses', 'plan_switch_realization', 'plan_withdraw_realization'] module-attribute

Classes

InversePlan

Bases: ConfigBaseModel

Ops that change how declared inverses are realized, with what was left out.

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

class InversePlan(ConfigBaseModel):
    """Ops that change how declared inverses are realized, with what was left out."""

    ops: list[ManifestOp] = PydanticField(default_factory=list)
    selected: list[str] = PydanticField(
        default_factory=list, description="Relations the ops act on."
    )
    skipped: list[Skipped] = PydanticField(default_factory=list)
    before: InverseReport = PydanticField(
        default_factory=InverseReport,
        description="The audit the plan was made from.",
    )
    after: InverseReport = PydanticField(
        default_factory=InverseReport,
        description="The audit of the manifest with the ops applied.",
    )

    @property
    def remaining(self) -> list[InverseFinding]:
        """Conflicts and unrepaired findings still present once the ops are applied."""
        return [f for f in self.after.findings if f.severity != "note"]

    def to_lines(self) -> list[str]:
        """The plan as text: what it does, what it skipped, what is left."""
        lines = [f"{len(self.ops)} op(s)"]
        lines += [f"  {op.op}: {_summary(op)}" for op in self.ops]
        if self.skipped:
            lines += ["", f"skipped ({len(self.skipped)}):"]
            lines += [f"  - {s.relation}: [{s.code}] {s.reason}" for s in self.skipped]
        if self.remaining:
            lines += ["", f"left untouched ({len(self.remaining)}):"]
            lines += [
                f"  - [{f.severity}/{f.kind}] {f.message}" for f in self.remaining
            ]
        return lines

Attributes

after = PydanticField(default_factory=InverseReport, description='The audit of the manifest with the ops applied.') class-attribute instance-attribute

before = PydanticField(default_factory=InverseReport, description='The audit the plan was made from.') class-attribute instance-attribute

ops = PydanticField(default_factory=list) class-attribute instance-attribute

remaining property

Conflicts and unrepaired findings still present once the ops are applied.

selected = PydanticField(default_factory=list, description='Relations the ops act on.') class-attribute instance-attribute

skipped = PydanticField(default_factory=list) class-attribute instance-attribute

Methods:

to_lines()

The plan as text: what it does, what it skipped, what is left.

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

def to_lines(self) -> list[str]:
    """The plan as text: what it does, what it skipped, what is left."""
    lines = [f"{len(self.ops)} op(s)"]
    lines += [f"  {op.op}: {_summary(op)}" for op in self.ops]
    if self.skipped:
        lines += ["", f"skipped ({len(self.skipped)}):"]
        lines += [f"  - {s.relation}: [{s.code}] {s.reason}" for s in self.skipped]
    if self.remaining:
        lines += ["", f"left untouched ({len(self.remaining)}):"]
        lines += [
            f"  - [{f.severity}/{f.kind}] {f.message}" for f in self.remaining
        ]
    return lines

Skipped

Bases: ConfigBaseModel

One relation a plan left alone, and why.

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

class Skipped(ConfigBaseModel):
    """One relation a plan left alone, and why."""

    relation: str
    code: str = PydanticField(..., description="Stable identifier of the reason.")
    reason: str

Attributes

code = PydanticField(..., description='Stable identifier of the reason.') class-attribute instance-attribute

reason instance-attribute

relation instance-attribute

Functions:

plan_declare_symmetric(manifest, relations)

Ops that make relations symmetric: their edges undirected, then the declaration.

directed: false and symmetric state one fact at two granularities and the schema refuses either without the other, so they are two ops in a fixed order. Keeping them two ops -- rather than one op that does both -- is what lets each be undone exactly.

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

def plan_declare_symmetric(
    manifest: GraphManifest, relations: Sequence[str]
) -> InversePlan:
    """Ops that make relations symmetric: their edges undirected, then the declaration.

    ``directed: false`` and ``symmetric`` state one fact at two granularities and
    the schema refuses either without the other, so they are two ops in a fixed
    order. Keeping them two ops -- rather than one op that does both -- is what
    lets each be undone exactly.
    """
    before = audit_inverses(manifest)
    if manifest.graph_schema is None:
        return InversePlan(before=before, after=before)
    edge_config = manifest.graph_schema.core_schema.edge_config
    skipped: list[Skipped] = []
    candidates: list[tuple[str, list[ManifestOp]]] = []
    for name in dict.fromkeys(relations):
        edges = [edge for edge in edge_config.edges if edge.relation == name]
        if not edges:
            skipped.append(
                Skipped(relation=name, code="no_edge", reason="labels no declared edge")
            )
            continue
        if before.pair(name) is not None:
            skipped.append(
                Skipped(
                    relation=name,
                    code="paired",
                    reason=(
                        "has a declared inverse; a relation is either paired with "
                        "another or its own inverse (retract_edge_inverses first)"
                    ),
                )
            )
            continue
        ops: list[ManifestOp] = []
        directed = [edge.edge_id for edge in edges if edge.directed]
        if directed:
            ops.append(SetEdgeDirectedOp(edges=_selectors(directed), directed=False))
        if name not in edge_config.symmetric:
            ops.append(DeclareEdgeInversesOp(symmetric=[name]))
        if not ops:
            skipped.append(
                Skipped(relation=name, code="already", reason="already symmetric")
            )
            continue
        candidates.append((name, ops))
    return _finish(manifest, before, candidates, skipped)

plan_realize_inverses(manifest, *, strategy='auto', relations=None)

Ops that realize declared pairs, and every pair left as it is with the reason.

Parameters:

Name	Type	Description	Default
manifest	GraphManifest	The manifest to plan against; not changed.	required
strategy	RealizeStrategy	native has the database maintain each eligible pair, materialized stores the inverse as declared edges fed by the same rows, and auto picks per target backend from what a reverse read costs there -- which, on most backends, is nothing, so auto realizes nothing, leaves the pair declared, and says so.	'auto'
relations	Sequence[str] | None	Restrict to the pairs these relations belong to (either side). Omitted: every declared pair.	None

A pair is never realized two ways: one that is already realized the other way is skipped, and :func:plan_switch_realization moves it.

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

def plan_realize_inverses(
    manifest: GraphManifest,
    *,
    strategy: RealizeStrategy = "auto",
    relations: Sequence[str] | None = None,
) -> InversePlan:
    """Ops that realize declared pairs, and every pair left as it is with the reason.

    Args:
        manifest: The manifest to plan against; not changed.
        strategy: ``native`` has the database maintain each eligible pair,
            ``materialized`` stores the inverse as declared edges fed by the same
            rows, and ``auto`` picks per target backend from what a reverse read
            costs there -- which, on most backends, is *nothing*, so ``auto``
            realizes nothing, leaves the pair declared, and says so.
        relations: Restrict to the pairs these relations belong to (either
            side). Omitted: every declared pair.

    A pair is never realized two ways: one that is already realized the other
    way is skipped, and :func:`plan_switch_realization` moves it.
    """
    before = audit_inverses(manifest)
    if manifest.graph_schema is None:
        return InversePlan(before=before, after=before)
    chosen, skipped = _pairs_for(before, relations)
    target, why = _auto_target(manifest) if strategy == "auto" else (strategy, "")

    candidates: list[tuple[str, list[ManifestOp]]] = []
    for pair, named in chosen:
        forward = _stored_forward(pair, named)
        label = named or forward or pair.relation
        if pair.state == "conflicting":
            kinds = sorted(
                {
                    f.kind
                    for f in before.conflicts()
                    if set(f.relations) & {pair.relation, pair.inverse}
                }
            )
            skipped.append(
                Skipped(
                    relation=label,
                    code="conflicting",
                    reason=f"the pair has unresolved conflicts: {kinds}",
                )
            )
        elif forward is None:
            skipped.append(
                Skipped(
                    relation=label,
                    code="no_edge",
                    reason="neither relation of the pair labels a declared edge",
                )
            )
        elif target is None and pair.state == "declared":
            skipped.append(
                Skipped(relation=label, code="declaration_suffices", reason=why)
            )
        elif target is None:
            # `auto` never withdraws what is already stored: whoever stored it may
            # have had a reason the backend's read cost does not show.
            skipped.append(
                Skipped(
                    relation=label,
                    code="stored_not_needed",
                    reason=(
                        f"is already {pair.state}, which this backend does not "
                        f"need ({why}); left as it is -- withdrawing the "
                        "realization (plan_withdraw_realization; `graflo inverses "
                        "withdraw`) drops the stored inverse and keeps the "
                        "declaration"
                    ),
                )
            )
        elif target == "native":
            candidates += _native_candidates(pair, label, skipped)
        else:
            candidates += _materialized_candidates(pair, forward, label, skipped)
    return _finish(manifest, before, candidates, skipped)

plan_repair_inverses(manifest)

Ops that propagate what the manifest under-reports; contradictions are left alone.

Works through the repairable findings of the audit in a fixed order, and keeps a repair only if, applied to a working copy, the finding it answers is gone and no new finding has appeared. A repair that is refused, or that trades one finding for another, is reported as skipped. Conflicts are never touched: they are in remaining.

The manifest need not load -- one assembled by a merge often does not -- see :func:graflo.architecture.profile.inverses.manifest_for_audit.

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

def plan_repair_inverses(manifest: GraphManifest) -> InversePlan:
    """Ops that propagate what the manifest under-reports; contradictions are left alone.

    Works through the ``repairable`` findings of the audit in a fixed order, and
    keeps a repair only if, applied to a working copy, the finding it answers is
    gone and no new finding has appeared. A repair that is refused, or that
    trades one finding for another, is reported as skipped. Conflicts are never
    touched: they are in ``remaining``.

    The manifest need not load -- one assembled by a merge often does not --
    see :func:`graflo.architecture.profile.inverses.manifest_for_audit`.
    """
    before = audit_inverses(manifest)
    working = manifest.model_copy(deep=True)
    ops: list[ManifestOp] = []
    selected: list[str] = []
    skipped: list[Skipped] = []

    for kind in _REPAIR_ORDER:
        for finding in [f for f in before.repairable() if f.kind == kind]:
            current = audit_inverses(working)
            if finding.key not in {f.key for f in current.findings}:
                continue  # an earlier repair already took care of it
            label = finding.relations[0] if finding.relations else kind
            candidate = _repair_ops(finding)
            if not candidate:
                continue
            trial = working.model_copy(deep=True)
            refusal = _try(trial, candidate)
            if refusal is not None:
                skipped.append(Skipped(relation=label, code="refused", reason=refusal))
                continue
            after = audit_inverses(trial)
            introduced = [
                f for f in after.introduced_since(current) if f.severity != "note"
            ]
            still_there = finding.key in {f.key for f in after.findings}
            if still_there or introduced:
                skipped.append(
                    Skipped(
                        relation=label,
                        code="not_a_clean_repair",
                        reason=(
                            f"propagating [{finding.kind}] "
                            + (
                                "left the finding in place"
                                if still_there
                                else "introduced "
                                + ", ".join(sorted({f.kind for f in introduced}))
                            )
                        ),
                    )
                )
                continue
            working = trial
            ops += candidate
            selected.append(label)

    return InversePlan(
        ops=ops,
        selected=list(dict.fromkeys(selected)),
        skipped=skipped,
        before=before,
        after=audit_inverses(working),
    )

plan_switch_realization(manifest, relations, *, to)

Ops that move pairs from the realization they have to to.

A pair is realized one way, so switching is withdraw-then-add, in that order. Each relation named is the side that stays stored: switching employed_by to native removes the declared employs edges and has the database maintain them instead. Eligibility for native is checked as the schema will be after the withdrawal, before anything is planned, so a pair is never left withdrawn and unrealized. A pair that is only declared has nothing to withdraw, so switching it is the same as realizing it.

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

def plan_switch_realization(
    manifest: GraphManifest, relations: Sequence[str], *, to: Realization
) -> InversePlan:
    """Ops that move pairs from the realization they have to ``to``.

    A pair is realized one way, so switching is withdraw-then-add, in that
    order. Each relation named is the side that stays stored: switching
    ``employed_by`` to ``native`` removes the declared ``employs`` edges and has
    the database maintain them instead. Eligibility for ``native`` is checked
    *as the schema will be after the withdrawal*, before anything is planned, so
    a pair is never left withdrawn and unrealized. A pair that is only declared
    has nothing to withdraw, so switching it is the same as realizing it.
    """
    before = audit_inverses(manifest)
    if manifest.graph_schema is None:
        return InversePlan(before=before, after=before)
    chosen, skipped = _pairs_for(before, relations)
    schema = manifest.graph_schema

    candidates: list[tuple[str, list[ManifestOp]]] = []
    for pair, named in chosen:
        subject = _subject(pair, named, skipped)
        if subject is None:
            continue
        label, forward = subject
        if pair.state == to:
            skipped.append(
                Skipped(relation=label, code="already", reason=f"already {to}")
            )
            continue
        withdraw: list[ManifestOp] = []
        if pair.state == "native" or (
            pair.state in ("materialized", "partial") and to != "materialized"
        ):
            forward, withdraw = _withdrawal(manifest, pair, forward)

        add: list[ManifestOp] = []
        if to == "native":
            trial = manifest.model_copy(deep=True)
            refusal = _try(trial, withdraw)
            violations = (
                []
                if refusal is not None or trial.graph_schema is None
                else native_inverse_violations(
                    trial.graph_schema.db_profile,
                    trial.graph_schema.core_schema.edge_config,
                    {v.name for v in schema.core_schema.vertex_config.vertices},
                    candidates=[forward],
                )
            )
            if violations:
                skipped.extend(
                    Skipped(relation=label, code=v.code, reason=v.message)
                    for v in violations
                    if v.relation == forward
                )
                continue
            add.append(SetNativeInversesOp(relations=[forward]))
        else:
            add.append(AddInverseEdgesOp(relations=[forward]))
        candidates.append((label, [*withdraw, *add]))
    return _finish(manifest, before, candidates, skipped)

plan_withdraw_realization(manifest, relations)

Ops that stop storing the inverse of pairs, keeping their declaration.

The reverse of realizing: a native inverse is handed back to nothing, and the declared edges of a materialized inverse are removed -- which also clears the emit_inverse flags that fed them. Each relation named is the side that stays stored. The pair remains in edge_config.inverses, so the inverse name keeps resolving on reads wherever the backend can follow an edge from its target.

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

def plan_withdraw_realization(
    manifest: GraphManifest, relations: Sequence[str]
) -> InversePlan:
    """Ops that stop storing the inverse of pairs, keeping their declaration.

    The reverse of realizing: a native inverse is handed back to nothing, and the
    declared edges of a materialized inverse are removed -- which also clears the
    ``emit_inverse`` flags that fed them. Each relation named is the side that
    stays stored. The pair remains in ``edge_config.inverses``, so the inverse
    name keeps resolving on reads wherever the backend can follow an edge from
    its target.
    """
    before = audit_inverses(manifest)
    if manifest.graph_schema is None:
        return InversePlan(before=before, after=before)
    chosen, skipped = _pairs_for(before, relations)

    candidates: list[tuple[str, list[ManifestOp]]] = []
    for pair, named in chosen:
        subject = _subject(pair, named, skipped)
        if subject is None:
            continue
        label, forward = subject
        if pair.state == "declared":
            skipped.append(
                Skipped(
                    relation=label,
                    code="already",
                    reason="nothing realizes the pair; it is only declared",
                )
            )
            continue
        _forward, withdraw = _withdrawal(manifest, pair, forward)
        if withdraw:
            candidates.append((label, withdraw))
    return _finish(manifest, before, candidates, skipped)