ontocast.tool.ontology_validation.reconcile

Reduce-time reconciliation of minted terms against full catalog terminals.

The per-unit lane cannot do this job: under vector-retrieval context the snapshot is a retrieved subset, so a unit that re-mints a concept the catalog already defines is — from inside the unit — inventing a genuinely new term. The duplicate lives precisely in the part of the catalog the snapshot does not contain, which is why _label_collision_findings (indexed on the snapshot) is structurally blind to it. The reduce step is the first place the full terminals are in hand — they are already fetched there for the namespace owner map — so that is where minted terms are checked.

Matching is deliberately the strictest rule that exists in this codebase: exact surface form (build_surface_index — case-sensitive labels, prefLabels, notations), resolving to exactly one catalog IRI (resolve_unique_surface refuses ambiguous surfaces), with a compatible role (a minted property never reconciles onto a catalog class or vice versa). Everything looser is a judgment call and stays out of the deterministic lane.

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

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