ontocast.tool.ontology_validation

Deterministic validation of ontology deltas.

Two lanes with different authorities: unit_findings validates a unit's delta against its own (possibly partial) prompt snapshot inside the loop; reconcile checks the merged document delta against the FULL catalog terminals at reduce time, where duplicates invisible to a retrieved snapshot become detectable.

MintedDuplicate

Bases: BaseModel

One minted term that exactly matches an existing catalog term.

Source code in ontocast/tool/ontology_validation/reconcile.py

class MintedDuplicate(BaseModel):
    """One minted term that exactly matches an existing catalog term."""

    minted_iri: str
    catalog_iri: str
    #: The exact surface form (label/prefLabel/notation) both terms share.
    surface: str
    #: ``"property"`` / ``"class"`` / ``"unknown"`` — the minted term's role
    #: as evidenced by the delta itself.
    role: str

apply_minted_duplicate_rewrites(merged_inserts, duplicates)

Rewrite minted IRIs to their catalog IRIs, in place.

Substitutes in subject and object position — a second minted term referencing the duplicate must end up pointing at the catalog term, or the rewrite would strand it. Predicate position is included for completeness (a minted property duplicate used as a predicate elsewhere in the delta).

Returns:

Type	Description
int	Number of triples rewritten.

Source code in ontocast/tool/ontology_validation/reconcile.py

def apply_minted_duplicate_rewrites(
    merged_inserts: RDFGraph,
    duplicates: list[MintedDuplicate],
) -> int:
    """Rewrite minted IRIs to their catalog IRIs, in place.

    Substitutes in subject **and** object position — a second minted term
    referencing the duplicate must end up pointing at the catalog term, or the
    rewrite would strand it. Predicate position is included for completeness
    (a minted property duplicate used as a predicate elsewhere in the delta).

    Returns:
        Number of triples rewritten.
    """
    if not duplicates:
        return 0
    mapping = {
        URIRef(duplicate.minted_iri): URIRef(duplicate.catalog_iri)
        for duplicate in duplicates
    }
    rewritten = 0
    for triple in list(merged_inserts):
        subject, predicate, obj = triple
        replaced = (
            mapping.get(subject, subject) if isinstance(subject, URIRef) else subject,
            mapping.get(predicate, predicate)
            if isinstance(predicate, URIRef)
            else predicate,
            mapping.get(obj, obj) if isinstance(obj, URIRef) else obj,
        )
        if replaced == triple:
            continue
        merged_inserts.remove(triple)
        merged_inserts.add(replaced)
        rewritten += 1
    return rewritten

collect_ontology_unit_findings(*, inserts, deletes, snapshot_graph, merged_graph=None, fact_namespaces=(), policy=None)

Assemble all deterministic findings for one unit's ontology delta.

Parameters:

Name	Type	Description	Default
inserts	RDFGraph	Net new triples this unit adds (build_delta().inserts).	required
deletes	RDFGraph	Snapshot triples this unit removes.	required
snapshot_graph	RDFGraph | None	The prompt ontology context; None/empty means the fresh-create path, where every catalog-relative check is skipped.	required
merged_graph	RDFGraph | None	snapshot + delta when the caller already has it (working_graph after updates apply) — passing it avoids a snapshot copy. Built here when omitted.	None
fact_namespaces	Sequence[str]	Facts/document namespaces; ontology terms minted there are mandatory findings.	()
policy	ValidationPolicy | None	Deployment namespace exemptions; None (tests) uses the built-in standard namespaces only.	None

Returns:

Type	Description
list[OntologyUnitFinding]	Findings, mandatory first is not guaranteed — order follows the
list[OntologyUnitFinding]	check sequence; callers filter on mandatory.

Source code in ontocast/tool/ontology_validation/unit_findings.py

def collect_ontology_unit_findings(
    *,
    inserts: RDFGraph,
    deletes: RDFGraph,
    snapshot_graph: RDFGraph | None,
    merged_graph: RDFGraph | None = None,
    fact_namespaces: Sequence[str] = (),
    policy: ValidationPolicy | None = None,
) -> list[OntologyUnitFinding]:
    """Assemble all deterministic findings for one unit's ontology delta.

    Args:
        inserts: Net new triples this unit adds (``build_delta().inserts``).
        deletes: Snapshot triples this unit removes.
        snapshot_graph: The prompt ontology context; ``None``/empty means the
            fresh-create path, where every catalog-relative check is skipped.
        merged_graph: ``snapshot + delta`` when the caller already has it
            (``working_graph`` after updates apply) — passing it avoids a
            snapshot copy. Built here when omitted.
        fact_namespaces: Facts/document namespaces; ontology terms minted
            there are mandatory findings.
        policy: Deployment namespace exemptions; ``None`` (tests) uses the
            built-in standard namespaces only.

    Returns:
        Findings, mandatory first is *not* guaranteed — order follows the
        check sequence; callers filter on ``mandatory``.
    """
    active_policy = policy or ValidationPolicy()
    if merged_graph is None:
        merged_graph = RDFGraph()
        if snapshot_graph is not None:
            for triple in snapshot_graph:
                merged_graph.add(triple)
        for triple in inserts:
            merged_graph.add(triple)

    declared = collect_declared_namespaces(snapshot_graph)
    findings: list[OntologyUnitFinding] = [
        *_namespace_findings(
            inserts, declared, fact_namespaces, active_policy.standard_namespaces()
        ),
        *_degenerate_restriction_findings(inserts),
        *_missing_label_findings(inserts, snapshot_graph),
        *_subclass_cycle_findings(inserts, merged_graph),
        *_role_confusion_findings(inserts, snapshot_graph),
        *_cardinality_contradiction_findings(inserts, merged_graph),
        *_foreign_delete_findings(deletes, snapshot_graph, inserts),
        *_label_collision_findings(inserts, snapshot_graph),
    ]
    return findings

count_fixes_targeting_snapshot(fixes, snapshot_graph, insert_subjects)

Critic fixes aimed at catalog content this unit's delta never touched.

The ontology critic is shown snapshot + delta and can reject a unit for pre-existing catalog defects the renderer cannot own. This counts the proposed fixes whose incorrect_value names a snapshot-declared subject absent from the unit's inserts — by full-IRI or prefixed-name substring match, so it is a lower bound, recorded as telemetry rather than used for control flow.

Source code in ontocast/tool/ontology_validation/unit_findings.py

def count_fixes_targeting_snapshot(
    fixes: Sequence[TripleFix],
    snapshot_graph: RDFGraph | None,
    insert_subjects: set[str],
) -> int:
    """Critic fixes aimed at catalog content this unit's delta never touched.

    The ontology critic is shown ``snapshot + delta`` and can reject a unit
    for pre-existing catalog defects the renderer cannot own. This counts the
    proposed fixes whose ``incorrect_value`` names a snapshot-declared subject
    absent from the unit's inserts — by full-IRI or prefixed-name substring
    match, so it is a lower bound, recorded as telemetry rather than used for
    control flow.
    """
    if snapshot_graph is None or not fixes:
        return 0
    prefix_map = [
        (prefix, str(namespace))
        for prefix, namespace in snapshot_graph.namespaces()
        if prefix
    ]
    snapshot_only: set[str] = set()
    qnames: set[str] = set()
    for subject in snapshot_graph.subjects():
        if not isinstance(subject, URIRef):
            continue
        text = str(subject)
        if text in insert_subjects or text in snapshot_only:
            continue
        snapshot_only.add(text)
        for prefix, namespace in prefix_map:
            if text.startswith(namespace) and len(text) > len(namespace):
                qnames.add(f"{prefix}:{text[len(namespace) :]}")
    count = 0
    for fix in fixes:
        haystack = fix.incorrect_value or ""
        if not haystack:
            continue
        if any(iri in haystack for iri in snapshot_only) or any(
            qname in haystack for qname in qnames
        ):
            count += 1
    return count

detect_minted_duplicates(merged_inserts, terminal_graphs)

Find minted terms whose surface form the full catalog already declares.

Parameters:

Name	Type	Description	Default
merged_inserts	RDFGraph	Document-level merged insert delta.	required
terminal_graphs	dict[str, RDFGraph]	Writable IRI -> the freshest full terminal graph (the same graphs the apply step writes onto).	required

Returns:

Type	Description
list[MintedDuplicate]	One record per (minted term, catalog term) unique-surface match with a
list[MintedDuplicate]	compatible role, ordered by minted IRI. Detection only — the caller
list[MintedDuplicate]	decides whether to rewrite.

Source code in ontocast/tool/ontology_validation/reconcile.py

def detect_minted_duplicates(
    merged_inserts: RDFGraph,
    terminal_graphs: dict[str, RDFGraph],
) -> list[MintedDuplicate]:
    """Find minted terms whose surface form the full catalog already declares.

    Args:
        merged_inserts: Document-level merged insert delta.
        terminal_graphs: Writable IRI -> the freshest full terminal graph
            (the same graphs the apply step writes onto).

    Returns:
        One record per (minted term, catalog term) unique-surface match with a
        compatible role, ordered by minted IRI. Detection only — the caller
        decides whether to rewrite.
    """
    if not terminal_graphs or len(merged_inserts) == 0:
        return []

    terminal_subjects: set[str] = set()
    for terminal in terminal_graphs.values():
        for subject in terminal.subjects():
            if isinstance(subject, URIRef):
                terminal_subjects.add(str(subject))

    roles = _minted_roles(merged_inserts)
    indexed_terminals = [
        (terminal, build_surface_index(terminal))
        for terminal in terminal_graphs.values()
    ]

    def first_match(subject: URIRef, role: str) -> MintedDuplicate | None:
        for predicate in _SURFACE_PREDICATES:
            for value in merged_inserts.objects(subject, predicate):
                if not isinstance(value, Literal):
                    continue
                surface = str(value).strip()
                if not surface:
                    continue
                for terminal, index in indexed_terminals:
                    catalog_iri = resolve_unique_surface(index, surface)
                    if catalog_iri is None or str(catalog_iri) == str(subject):
                        continue
                    if not _role_compatible(role, str(catalog_iri), terminal):
                        continue
                    return MintedDuplicate(
                        minted_iri=str(subject),
                        catalog_iri=str(catalog_iri),
                        surface=surface,
                        role=role,
                    )
        return None

    duplicates: list[MintedDuplicate] = []
    for subject in sorted(
        {s for s in merged_inserts.subjects() if isinstance(s, URIRef)}, key=str
    ):
        if str(subject) in terminal_subjects:
            continue
        match = first_match(subject, roles.get(subject, "unknown"))
        if match is not None:
            duplicates.append(match)
    return duplicates