Skip to content

ontocast.tool.facts_validation

Deterministic validation, findings, and LLM-free repair for rendered facts.

Split by concern: terms (catalog inventory, namespace closure, ValidationPolicy), literal_repair (parse-time rewrites), unit_findings (per-unit findings for repair renders), shacl (execution, autofix, catalog lint), gate (document-level validation). This package is the public surface; import from here.

FactsAcceptancePolicy

Bases: BaseModel

Which defects block a rendered unit from leaving the loop.

Attributes:

Name Type Description
blocking_finding_kinds frozenset[FactsUnitFindingKind] | None

Finding kinds that block. None -- the default -- blocks on every finding carrying mandatory=True, which is the deterministic validator's own judgement. The explicit set exists so one lane can be silenced without silencing its telemetry, and so a lane found to emit false positives can be switched off without a release. That escape hatch is not optional: binding acceptance to findings means a systematically unfixable finding becomes a permanent per-unit tax, and this codebase has already shipped one (a false mandatory qudt:numericValue UNKNOWN_TERM that ordered renders to destroy correct values).
blocking_fix_severity BlockingFixSeverity

The cut applied to critic-proposed fixes. critical is the default because it is the only severity the critic applies selectively enough to discriminate on. It labels most fixes important, so gating there accepts almost nothing -- worse than the score gate it replaces.
Source code in ontocast/tool/facts_validation/acceptance.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class FactsAcceptancePolicy(BaseModel):
    """Which defects block a rendered unit from leaving the loop.

    Attributes:
        blocking_finding_kinds: Finding kinds that block. ``None`` -- the
            default -- blocks on every finding carrying ``mandatory=True``,
            which is the deterministic validator's own judgement. The explicit
            set exists so one lane can be silenced without silencing its
            telemetry, and so a lane found to emit false positives can be
            switched off without a release. That escape hatch is not optional:
            binding acceptance to findings means a systematically unfixable
            finding becomes a permanent per-unit tax, and this codebase has
            already shipped one (a false mandatory ``qudt:numericValue``
            ``UNKNOWN_TERM`` that ordered renders to destroy correct values).
        blocking_fix_severity: The cut applied to critic-proposed fixes.
            ``critical`` is the default because it is the only severity the
            critic applies selectively enough to discriminate on. It labels
            most fixes ``important``, so gating there accepts almost nothing --
            worse than the score gate it replaces.
    """

    blocking_finding_kinds: frozenset[FactsUnitFindingKind] | None = None
    blocking_fix_severity: BlockingFixSeverity = "critical"

    def blocks_finding(self, finding: FactsUnitFinding) -> bool:
        """True when this deterministic finding must be repaired before exit."""
        if self.blocking_finding_kinds is None:
            return finding.mandatory
        return finding.kind in self.blocking_finding_kinds

    def blocks_fix(self, fix: TripleFix) -> bool:
        """True when this critic-proposed fix must be applied before exit.

        A ``REMOVE`` fix never blocks, whatever its severity. The repair prompt
        it would be rendered into states that a finding is never resolved by
        deleting the statement, so a mandatory REMOVE would contradict the block
        it sits in -- the same shape of contradiction that
        ``shacl_catalog_contradictions`` exists to catch, and one that has
        already caused repair renders to delete valid values wholesale.
        """
        if self.blocking_fix_severity == "never":
            return False
        if fix.action == "REMOVE":
            return False
        if self.blocking_fix_severity == "critical":
            return fix.severity == "critical"
        return fix.severity in ("critical", "important")

blocks_finding(finding)

True when this deterministic finding must be repaired before exit.

Source code in ontocast/tool/facts_validation/acceptance.py 
67
68
69
70
71
def blocks_finding(self, finding: FactsUnitFinding) -> bool:
    """True when this deterministic finding must be repaired before exit."""
    if self.blocking_finding_kinds is None:
        return finding.mandatory
    return finding.kind in self.blocking_finding_kinds

blocks_fix(fix)

True when this critic-proposed fix must be applied before exit.

A REMOVE fix never blocks, whatever its severity. The repair prompt it would be rendered into states that a finding is never resolved by deleting the statement, so a mandatory REMOVE would contradict the block it sits in -- the same shape of contradiction that shacl_catalog_contradictions exists to catch, and one that has already caused repair renders to delete valid values wholesale.

Source code in ontocast/tool/facts_validation/acceptance.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def blocks_fix(self, fix: TripleFix) -> bool:
    """True when this critic-proposed fix must be applied before exit.

    A ``REMOVE`` fix never blocks, whatever its severity. The repair prompt
    it would be rendered into states that a finding is never resolved by
    deleting the statement, so a mandatory REMOVE would contradict the block
    it sits in -- the same shape of contradiction that
    ``shacl_catalog_contradictions`` exists to catch, and one that has
    already caused repair renders to delete valid values wholesale.
    """
    if self.blocking_fix_severity == "never":
        return False
    if fix.action == "REMOVE":
        return False
    if self.blocking_fix_severity == "critical":
        return fix.severity == "critical"
    return fix.severity in ("critical", "important")

FactsValidationReport

Bases: BaseModel

Invariant findings over one aggregated facts graph.

Source code in ontocast/tool/facts_validation/gate.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class FactsValidationReport(BaseModel):
    """Invariant findings over one aggregated facts graph."""

    model_config = {"arbitrary_types_allowed": True}

    findings: list[FactsValidationFinding] = Field(default_factory=list)
    shacl_evaluated: bool | None = Field(
        default=None,
        description=(
            "True when SHACL ran, False when shapes were configured but it "
            "could not (pyshacl missing, graph over the size guard), None when "
            "no shapes were in play. 'No SHACL findings' means nothing without "
            "this: it reads identically for 'conforms' and 'never checked'."
        ),
    )
    shacl_violations: list["ShaclViolation"] = Field(
        default_factory=list,
        exclude=True,
        repr=False,
        description=(
            "Raw, unfiltered pyshacl violations, kept so the autofix pass can "
            "reuse them instead of re-running validation. Internal: never "
            "serialized."
        ),
    )

    @property
    def error_findings(self) -> list[FactsValidationFinding]:
        """Error-severity findings, whatever their kind."""
        return [finding for finding in self.findings if finding.severity == "error"]

error_findings property

Error-severity findings, whatever their kind.

MaterialDefect

Bases: BaseModel

One reason a rendered unit is not acceptable as it stands.

Source code in ontocast/tool/facts_validation/acceptance.py 
35
36
37
38
39
40
class MaterialDefect(BaseModel):
    """One reason a rendered unit is not acceptable as it stands."""

    source: Literal["finding", "critic_fix"]
    kind: str = Field(description="Finding kind, or the fix's action for a critic fix.")
    message: str

ShaclRepairResult

Bases: BaseModel

Outcome of the LLM-free SHACL repair pass.

Source code in ontocast/tool/facts_validation/shacl.py 
285
286
287
288
289
290
291
292
293
294
295
296
class ShaclRepairResult(BaseModel):
    """Outcome of the LLM-free SHACL repair pass."""

    model_config = {"arbitrary_types_allowed": True}

    graph: RDFGraph
    records: list[GraphRepairRecord] = Field(default_factory=list)
    violations_before: int = 0
    violations_after: int = 0
    passes_applied: int = 0
    reverted: bool = False
    ran: bool = False

ShaclViolation

Bases: BaseModel

One SHACL validation result, in the form the repair pass needs.

FactsValidationFinding is the reporting shape and deliberately flat; this keeps the RDF terms (focus node, path, offending value, constraint component) so a repair can act on them.

Source code in ontocast/tool/facts_validation/shacl.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class ShaclViolation(BaseModel):
    """One SHACL validation result, in the form the repair pass needs.

    ``FactsValidationFinding`` is the reporting shape and deliberately flat;
    this keeps the RDF terms (focus node, path, offending value, constraint
    component) so a repair can act on them.
    """

    model_config = {"arbitrary_types_allowed": True}

    focus: Node | None = None
    path: URIRef | None = None
    value: Node | None = None
    component: URIRef | None = None
    # Node, not URIRef: the common authoring style is an inline
    # ``sh:property [ sh:path … ; sh:datatype … ]``, whose shape is a blank
    # node. Narrowing to URIRef dropped it and left every such violation
    # unrepairable. pyshacl reports the same BNode the shapes graph holds, so
    # the datatype lookup resolves.
    source_shape: Node | None = None
    severity: TypingLiteral["error", "warning"] = "error"
    message: str = "SHACL constraint violated."

    def as_finding(self) -> FactsValidationFinding:
        """Project onto the reported finding shape."""
        return FactsValidationFinding(
            kind=FactsValidationFindingKind.SHACL,
            severity=self.severity,
            message=self.message,
            subject=str(self.focus) if self.focus is not None else "",
            predicate=str(self.path) if self.path is not None else "",
            values=[str(self.value)] if self.value is not None else [],
            component=str(self.component) if self.component is not None else "",
            source_shape=(
                str(self.source_shape) if self.source_shape is not None else ""
            ),
        )

as_finding()

Project onto the reported finding shape.

Source code in ontocast/tool/facts_validation/shacl.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def as_finding(self) -> FactsValidationFinding:
    """Project onto the reported finding shape."""
    return FactsValidationFinding(
        kind=FactsValidationFindingKind.SHACL,
        severity=self.severity,
        message=self.message,
        subject=str(self.focus) if self.focus is not None else "",
        predicate=str(self.path) if self.path is not None else "",
        values=[str(self.value)] if self.value is not None else [],
        component=str(self.component) if self.component is not None else "",
        source_shape=(
            str(self.source_shape) if self.source_shape is not None else ""
        ),
    )

ValidationPolicy

Bases: BaseModel

Deployment-level exemptions and vocabulary for deterministic validation.

One object instead of a parameter per concern: the namespaces a deployment shares across catalogs, the sanctioned quantity fallback vocabulary, and the code predicates — everything the term checks must never flag, because configuration explicitly blessed it.

Source code in ontocast/tool/facts_validation/terms.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class ValidationPolicy(BaseModel):
    """Deployment-level exemptions and vocabulary for deterministic validation.

    One object instead of a parameter per concern: the namespaces a deployment
    shares across catalogs, the sanctioned quantity fallback vocabulary, and
    the code predicates — everything the term checks must never flag, because
    configuration explicitly blessed it.
    """

    additional_standard_namespaces: tuple[str, ...] = ()
    quantity_fallback_vocabulary: dict[str, str] | None = None
    code_predicates: tuple[str, ...] = ()

    def standard_namespaces(self) -> tuple[str, ...]:
        """Built-in meta-vocabulary namespaces plus the configured ones."""
        return (*_STANDARD_NAMESPACES, *self.additional_standard_namespaces)

    def exempt_terms(self, *graphs: RDFGraph | None) -> set[str]:
        """Exact IRIs configuration blessed: fallback vocabulary + code predicates."""
        terms = expand_vocabulary_terms(self.quantity_fallback_vocabulary, *graphs)
        terms.update(self.code_predicates)
        return terms

exempt_terms(*graphs)

Exact IRIs configuration blessed: fallback vocabulary + code predicates.

Source code in ontocast/tool/facts_validation/terms.py 
154
155
156
157
158
def exempt_terms(self, *graphs: RDFGraph | None) -> set[str]:
    """Exact IRIs configuration blessed: fallback vocabulary + code predicates."""
    terms = expand_vocabulary_terms(self.quantity_fallback_vocabulary, *graphs)
    terms.update(self.code_predicates)
    return terms

standard_namespaces()

Built-in meta-vocabulary namespaces plus the configured ones.

Source code in ontocast/tool/facts_validation/terms.py 
150
151
152
def standard_namespaces(self) -> tuple[str, ...]:
    """Built-in meta-vocabulary namespaces plus the configured ones."""
    return (*_STANDARD_NAMESPACES, *self.additional_standard_namespaces)

accept_reason(defects)

A short, aggregatable label for why the unit was accepted or not.

Source code in ontocast/tool/facts_validation/acceptance.py 
131
132
133
134
135
136
137
def accept_reason(defects: Sequence[MaterialDefect]) -> str:
    """A short, aggregatable label for why the unit was accepted or not."""
    if not defects:
        return "clean"
    if any(defect.source == "finding" for defect in defects):
        return "mandatory_findings"
    return "critic_critical"

apply_shacl_repairs(graph, shapes_graph, ontology_graph, *, mode='prune', passes=1, fact_namespaces=(), code_predicates=(), inference='rdfs', advanced=True, max_triples=0, initial_violations=None)

Repair SHACL violations in code, with no LLM round-trip.

Bounded validate -> repair -> revalidate loop. A pass is kept only when it strictly reduces the violation count: a repair that trades triples for no conformance gain is reverted, the same discipline the un-merge repair uses.

Repairs by constraint component
  • sh:datatype: retype a literal that parses as the declared datatype ("2019"^^xsd:string -> "2019"^^xsd:gYear).
  • sh:class / sh:nodeKind: replace a string literal with the one catalog IRI declaring it as a surface form (qudt:unit "meV" -> unit:MilliElectronVolt). Ambiguous forms are left reported.
  • sh:minCount (mode prune only): drop a focus node that asserts nothing beyond rdf:type/rdfs:label and is referenced by at most one subject, together with that reference.

Everything else -- sh:maxCount (owned by the functional-violation and un-merge machinery), sh:not, sh:qualifiedValueShape, SPARQL constraints -- is reported, never repaired.

Parameters:

Name Type Description Default
graph RDFGraph

Aggregated facts graph, repaired in place: it may be oxigraph-backed and carry RDF 1.2 triple terms, which a copied rdflib graph would silently drop. A pass that fails the accept test is rolled back triple-for-triple instead.

 required
shapes_graph RDFGraph | None

Shapes to validate against; None disables the pass.

 required
ontology_graph RDFGraph | None

Merged ontology context, indexed for surface forms.

 required
mode str

off | rewrite (rewrites only) | prune (also prunes).

 'prune'
passes int

Maximum repair rounds.

 1
fact_namespaces Sequence[str]

Only nodes under these namespaces are repaired.

 ()
code_predicates Sequence[str]

Code-bearing predicates for surface resolution.

 ()
inference str

pyshacl pre-inference mode.

 'rdfs'
advanced bool

Enable SHACL Advanced Features.

 True
max_triples int

Skip validation above this graph size; 0 disables.

 0
initial_violations Sequence[ShaclViolation] | None

Violations already computed for graph with the same parameters (e.g. by the reporting pass), reused to skip the redundant first validation.

 None

Returns:

Type Description
ShaclRepairResult

The repaired graph, the applied repair records, and fact-scoped
ShaclRepairResult

violation counts before and after (the population conforms is
ShaclRepairResult

judged on; the loop's accept test uses the raw count internally).
Source code in ontocast/tool/facts_validation/shacl.py 
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
def apply_shacl_repairs(
    graph: RDFGraph,
    shapes_graph: RDFGraph | None,
    ontology_graph: RDFGraph | None,
    *,
    mode: str = "prune",
    passes: int = 1,
    fact_namespaces: Sequence[str] = (),
    code_predicates: Sequence[str] = (),
    inference: str = "rdfs",
    advanced: bool = True,
    max_triples: int = 0,
    initial_violations: Sequence[ShaclViolation] | None = None,
) -> ShaclRepairResult:
    """Repair SHACL violations in code, with no LLM round-trip.

    Bounded ``validate -> repair -> revalidate`` loop. A pass is kept only when
    it strictly reduces the violation count: a repair that trades triples for
    no conformance gain is reverted, the same discipline the un-merge repair
    uses.

    Repairs by constraint component:
        - ``sh:datatype``: retype a literal that parses as the declared
          datatype (``"2019"^^xsd:string`` -> ``"2019"^^xsd:gYear``).
        - ``sh:class`` / ``sh:nodeKind``: replace a string literal with the one
          catalog IRI declaring it as a surface form (``qudt:unit "meV"`` ->
          ``unit:MilliElectronVolt``). Ambiguous forms are left reported.
        - ``sh:minCount`` (mode ``prune`` only): drop a focus node that asserts
          nothing beyond ``rdf:type``/``rdfs:label`` and is referenced by at
          most one subject, together with that reference.

    Everything else -- ``sh:maxCount`` (owned by the functional-violation and
    un-merge machinery), ``sh:not``, ``sh:qualifiedValueShape``, SPARQL
    constraints -- is reported, never repaired.

    Args:
        graph: Aggregated facts graph, repaired **in place**: it may be
            oxigraph-backed and carry RDF 1.2 triple terms, which a copied
            rdflib graph would silently drop. A pass that fails the accept
            test is rolled back triple-for-triple instead.
        shapes_graph: Shapes to validate against; ``None`` disables the pass.
        ontology_graph: Merged ontology context, indexed for surface forms.
        mode: ``off`` | ``rewrite`` (rewrites only) | ``prune`` (also prunes).
        passes: Maximum repair rounds.
        fact_namespaces: Only nodes under these namespaces are repaired.
        code_predicates: Code-bearing predicates for surface resolution.
        inference: pyshacl pre-inference mode.
        advanced: Enable SHACL Advanced Features.
        max_triples: Skip validation above this graph size; 0 disables.
        initial_violations: Violations already computed for ``graph`` with the
            same parameters (e.g. by the reporting pass), reused to skip the
            redundant first validation.

    Returns:
        The repaired graph, the applied repair records, and fact-scoped
        violation counts before and after (the population ``conforms`` is
        judged on; the loop's accept test uses the raw count internally).
    """
    if mode == "off" or shapes_graph is None or not len(shapes_graph) or passes <= 0:
        return ShaclRepairResult(graph=graph)

    def _validate(target: RDFGraph) -> list[ShaclViolation] | None:
        return run_shacl(
            target,
            shapes_graph,
            ontology_graph=ontology_graph,
            inference=inference,
            advanced=advanced,
            max_triples=max_triples,
        )

    violations = (
        list(initial_violations) if initial_violations is not None else _validate(graph)
    )
    if violations is None:
        return ShaclRepairResult(graph=graph)

    def _scoped_count(candidates: Sequence[ShaclViolation]) -> int:
        return len(_fact_scope_violations(graph, candidates, fact_namespaces))

    result = ShaclRepairResult(
        graph=graph,
        violations_before=_scoped_count(violations),
        violations_after=_scoped_count(violations),
        ran=True,
    )
    surface_index = build_surface_index(ontology_graph, code_predicates)

    def _rollback(added: Sequence[tuple], removed: Sequence[tuple]) -> None:
        for triple in added:
            graph.remove(triple)
        for triple in removed:
            graph.add(triple)

    for _ in range(passes):
        if not violations:
            break
        plan = _shacl_repairs_for(
            graph,
            shapes_graph,
            violations,
            mode=mode,
            surface_index=surface_index,
            fact_namespaces=fact_namespaces,
        )
        records = plan.records
        if not records:
            break

        applied_removals: list[tuple] = []
        applied_removal_set: set[tuple] = set()
        seen_removals: set[tuple] = set()
        for triple in plan.removals:
            if triple in seen_removals:
                continue
            seen_removals.add(triple)
            if triple in graph:
                graph.remove(triple)
                applied_removals.append(triple)
                applied_removal_set.add(triple)
        applied_additions: list[tuple] = []
        for triple in plan.additions:
            if triple not in graph:
                graph.add(triple)
                applied_additions.append(triple)

        candidate_violations = _validate(graph)
        if candidate_violations is None:
            _rollback(applied_additions, applied_removals)
            break
        if len(candidate_violations) >= len(violations):
            logger.warning(
                "SHACL autofix: pass did not reduce violations (%d -> %d); "
                "keeping the pre-repair graph",
                len(violations),
                len(candidate_violations),
            )
            _rollback(applied_additions, applied_removals)
            result.reverted = True
            break

        # Only once the pass is accepted, and deliberately after the accept test
        # rather than alongside the removals: reification quads cannot travel
        # through _rollback (rdflib cannot add or remove a triple-term triple),
        # and validation runs on a copy with triple terms stripped, so this
        # changes no count and needs no undo.
        #
        # Retarget before sweeping. A statement that is retyped *and* then
        # pruned in the same pass has to be swept at its new triple term, which
        # the retarget has already installed -- so prune still wins, by the
        # sweep matching rather than by the retarget happening to miss.
        retargeted = retarget_reifiers(
            graph,
            {
                removed: replacement
                for removed, replacement in plan.retargets.items()
                if removed in applied_removal_set and replacement in graph
            },
        )
        swept = drop_reifiers_mentioning(graph, plan.pruned)

        provenance_note = ", ".join(
            note
            for note in (
                f"{retargeted} provenance quad(s) retargeted" if retargeted else "",
                f"{swept} orphaned provenance quad(s) swept" if swept else "",
            )
            if note
        )
        logger.info(
            "SHACL autofix: %d repair(s) applied, violations %d -> %d%s",
            len(records),
            len(violations),
            len(candidate_violations),
            f", {provenance_note}" if provenance_note else "",
        )
        violations = candidate_violations
        result.records.extend(records)
        result.passes_applied += 1
        result.violations_after = _scoped_count(candidate_violations)

    return result

build_surface_index(ontology_graph, code_predicates=())

Map exact catalog surface forms to the IRIs declaring them.

Case-sensitive and exact: these are codes and names a model may have transcribed verbatim ("d", "meV", "CsPbBr3"), not free text to be fuzzy-matched. A form claimed by more than one IRI stays in the index and is rejected at lookup time — an ambiguous code is not a repairable one.

Parameters:

Name Type Description Default
ontology_graph RDFGraph | None

Merged ontology context to index.

 required
code_predicates Sequence[str]

Extra code-bearing predicates (UCUM codes, symbols, notations) on top of the standard name predicates.

 ()

Returns:

Type Description
dict[str, set[str]]

Surface form -> set of IRIs declaring it.
Source code in ontocast/tool/facts_validation/terms.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def build_surface_index(
    ontology_graph: RDFGraph | None,
    code_predicates: Sequence[str] = (),
) -> dict[str, set[str]]:
    """Map exact catalog surface forms to the IRIs declaring them.

    Case-sensitive and exact: these are codes and names a model may have
    transcribed verbatim (``"d"``, ``"meV"``, ``"CsPbBr3"``), not free text to
    be fuzzy-matched. A form claimed by more than one IRI stays in the index and
    is rejected at lookup time — an ambiguous code is not a repairable one.

    Args:
        ontology_graph: Merged ontology context to index.
        code_predicates: Extra code-bearing predicates (UCUM codes, symbols,
            notations) on top of the standard name predicates.

    Returns:
        Surface form -> set of IRIs declaring it.
    """
    index: dict[str, set[str]] = {}
    if ontology_graph is None:
        return index
    predicates: list[URIRef] = [RDFS.label, SKOS.prefLabel, SKOS.notation]
    predicates.extend(URIRef(predicate) for predicate in code_predicates)
    for predicate in predicates:
        for subject, value in ontology_graph.subject_objects(predicate):
            if not isinstance(subject, URIRef) or not isinstance(value, Literal):
                continue
            text = str(value).strip()
            if text:
                index.setdefault(text, set()).add(str(subject))
    return index

collect_catalog_terms(ontology_graph)

All IRIs appearing anywhere in the ontology context.

Source code in ontocast/tool/facts_validation/terms.py 
72
73
74
75
76
77
78
79
80
81
def collect_catalog_terms(ontology_graph: RDFGraph | None) -> set[str]:
    """All IRIs appearing anywhere in the ontology context."""
    terms: set[str] = set()
    if ontology_graph is None:
        return terms
    for triple in ontology_graph:
        for term in triple:
            if isinstance(term, URIRef):
                terms.add(str(term))
    return terms

collect_declared_namespaces(ontology_graph)

Namespaces the catalog declares terms in (subject-position IRIs).

The UNKNOWN_TERM check treats a namespace as closed — flagging members the catalog does not list — only when the catalog actually declares terms there. A namespace the catalog merely references (qudt:QuantityValue in an rdfs:subClassOf, qudt:unit in an owl:onProperty) is an external vocabulary the catalog borrows from, and the catalog is not an authority on its membership. Treating referenced-only namespaces as closed produced mandatory findings against canonical external properties (qudt:numericValue), which repair renders then obeyed by deleting correct data.

Source code in ontocast/tool/facts_validation/terms.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def collect_declared_namespaces(ontology_graph: RDFGraph | None) -> set[str]:
    """Namespaces the catalog *declares* terms in (subject-position IRIs).

    The UNKNOWN_TERM check treats a namespace as closed — flagging members the
    catalog does not list — only when the catalog actually declares terms
    there. A namespace the catalog merely *references* (``qudt:QuantityValue``
    in an ``rdfs:subClassOf``, ``qudt:unit`` in an ``owl:onProperty``) is an
    external vocabulary the catalog borrows from, and the catalog is not an
    authority on its membership. Treating referenced-only namespaces as closed
    produced mandatory findings against canonical external properties
    (``qudt:numericValue``), which repair renders then obeyed by deleting
    correct data.
    """
    namespaces: set[str] = set()
    if ontology_graph is None:
        return namespaces
    for subject in ontology_graph.subjects():
        if isinstance(subject, URIRef):
            namespaces.add(_namespace_of(str(subject)))
    return namespaces

collect_shacl_shapes(ontology_graph, stored_shapes)

Assemble the SHACL shapes graph for the validation gate.

Sources: the deployment's shapes partition (stored_shapes, resolved by :class:~ontocast.tool.shapes_catalog.ShapesCatalog -- seeded from FACTS_SHAPES_DIR and mutable over /shapes), plus the ontology context itself when it already carries sh:NodeShape declarations inline -- the zero-config path for catalogs that ship shapes next to their schema.

Parameters:

Name Type Description Default
ontology_graph RDFGraph | None

Ontology context offered to the renderer.

 required
stored_shapes RDFGraph | None

Merged shapes graph from the shapes partition.

 required

Returns:

Type Description
RDFGraph | None

RDFGraph | None: The shapes to validate against, or None when there
RDFGraph | None

are none -- which is what keeps shacl_evaluated at None
RDFGraph | None

("never checked") rather than reporting a clean run.
Source code in ontocast/tool/facts_validation/shacl.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def collect_shacl_shapes(
    ontology_graph: RDFGraph | None, stored_shapes: RDFGraph | None
) -> RDFGraph | None:
    """Assemble the SHACL shapes graph for the validation gate.

    Sources: the deployment's shapes partition (``stored_shapes``, resolved by
    :class:`~ontocast.tool.shapes_catalog.ShapesCatalog` -- seeded from
    ``FACTS_SHAPES_DIR`` and mutable over ``/shapes``), plus the ontology
    context itself when it already carries ``sh:NodeShape`` declarations inline
    -- the zero-config path for catalogs that ship shapes next to their schema.

    Args:
        ontology_graph: Ontology context offered to the renderer.
        stored_shapes: Merged shapes graph from the shapes partition.

    Returns:
        RDFGraph | None: The shapes to validate against, or ``None`` when there
        are none -- which is what keeps ``shacl_evaluated`` at ``None``
        ("never checked") rather than reporting a clean run.
    """
    shapes = RDFGraph()
    if stored_shapes is not None and len(stored_shapes):
        shapes += stored_shapes
    node_shape = SH.NodeShape
    if ontology_graph is not None and (None, RDF.type, node_shape) in ontology_graph:
        shapes += ontology_graph
    return shapes if len(shapes) else None

collect_unit_findings(*, graph, ontology_graph, quarantined, extraction_text, fact_namespaces, coverage_limit=30, policy=None)

Assemble all deterministic findings for one rendered unit graph.

Mandatory: quarantined literals (with closed-range individual suggestions), forbidden-namespace terms (example.org), doc-namespace predicates, unresolved catalog near-misses, predicates asserted on a subject whose type contradicts their rdfs:domain, and value nodes whose only numeric content sits in a label. Advisory-strong: numeric mentions of the source text absent from the graph — the renderer decides per item whether each is an extractable quantity or an artifact.

The policy's exempt terms (the sanctioned fallback vocabulary the facts prompt itself names, plus code predicates) never raise UNKNOWN_TERM: flagging the vocabulary the prompt recommends produced mandatory findings that repair renders obeyed by deleting correct data.

Source code in ontocast/tool/facts_validation/unit_findings.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
def collect_unit_findings(
    *,
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    quarantined: list[RejectedLiteralTriple],
    extraction_text: str,
    fact_namespaces: list[str],
    coverage_limit: int = 30,
    policy: ValidationPolicy | None = None,
) -> list[FactsUnitFinding]:
    """Assemble all deterministic findings for one rendered unit graph.

    Mandatory: quarantined literals (with closed-range individual
    suggestions), forbidden-namespace terms (``example.org``), doc-namespace
    predicates, unresolved catalog near-misses, predicates asserted on a
    subject whose type contradicts their ``rdfs:domain``, and value nodes
    whose only numeric content sits in a label. Advisory-strong: numeric
    mentions of the source text absent from the graph — the renderer decides
    per item whether each is an extractable quantity or an artifact.

    The policy's exempt terms (the sanctioned fallback vocabulary the facts
    prompt itself names, plus code predicates) never raise UNKNOWN_TERM:
    flagging the vocabulary the prompt recommends produced mandatory findings
    that repair renders obeyed by deleting correct data.
    """
    policy = policy or ValidationPolicy()
    findings: list[FactsUnitFinding] = []
    standard_namespaces = policy.standard_namespaces()
    fallback_terms = policy.exempt_terms(graph, ontology_graph)
    quantity_fallback_vocabulary = policy.quantity_fallback_vocabulary

    for rejected in quarantined:
        findings.append(
            FactsUnitFinding(
                kind=FactsUnitFindingKind.QUARANTINED_LITERAL,
                message=(
                    f"Triple excluded ({rejected.reason}): the object of "
                    f"<{rejected.predicate}> must be an IRI/valid literal, got "
                    f"'{rejected.object_lexical}'."
                ),
                subject=rejected.subject,
                predicate=rejected.predicate,
                value=rejected.object_lexical,
                suggestions=_closed_range_suggestions(rejected, ontology_graph),
            )
        )

    catalog_terms = collect_catalog_terms(ontology_graph)
    declared_namespaces = collect_declared_namespaces(ontology_graph)
    normalized_fact_namespaces = [ns for ns in fact_namespaces if ns]

    prefix_map = {
        prefix: str(namespace) for prefix, namespace in graph.namespaces() if prefix
    }
    flagged_terms: set[str] = set()
    for subject, predicate, obj in graph:
        if predicate == RDF.type and isinstance(obj, Literal):
            lexical = str(obj).strip()
            if lexical in flagged_terms:
                continue
            flagged_terms.add(lexical)
            resolved = _resolve_type_literal(lexical, prefix_map)
            findings.append(
                FactsUnitFinding(
                    kind=FactsUnitFindingKind.LITERAL_TYPE_OBJECT,
                    message=(
                        f"rdf:type object '{lexical}' is a string literal, not "
                        "an IRI; assert the type as a catalog class IRI "
                        "(`a prefix:Class`), never as a quoted string."
                    ),
                    subject=str(subject),
                    value=lexical,
                    suggestions=[resolved]
                    if resolved and resolved in catalog_terms
                    else [],
                )
            )
            continue
        for position, term in (("predicate", predicate), ("type", obj)):
            if not isinstance(term, URIRef):
                continue
            if position == "type" and predicate != RDF.type:
                continue
            text = str(term)
            if text in flagged_terms:
                continue
            if text.startswith(_FORBIDDEN_NAMESPACES):
                flagged_terms.add(text)
                findings.append(
                    FactsUnitFinding(
                        kind=FactsUnitFindingKind.UNKNOWN_TERM,
                        message=(
                            f"<{text}> uses the example.org placeholder namespace; "
                            "replace it with a catalog term or express the "
                            "statement with catalog/standard vocabulary."
                        ),
                        predicate=text,
                        suggestions=_alias_candidates(
                            term,
                            graph,
                            catalog_terms,
                            ontology_graph=ontology_graph,
                            position=position,
                        )
                        if catalog_terms
                        else [],
                    )
                )
                continue
            if any(text.startswith(ns) for ns in normalized_fact_namespaces):
                flagged_terms.add(text)
                role_message = (
                    f"Predicate <{text}> is minted in the facts/document "
                    "namespace; facts namespaces hold instances only — "
                    "use a catalog or standard-vocabulary property."
                    if position == "predicate"
                    else f"rdf:type object <{text}> is a class minted in the "
                    "facts/document namespace; facts namespaces hold instances, "
                    "not classes — type the instance with a catalog or "
                    "standard-vocabulary class."
                )
                findings.append(
                    FactsUnitFinding(
                        kind=FactsUnitFindingKind.UNKNOWN_TERM,
                        message=role_message,
                        predicate=text,
                        suggestions=_alias_candidates(
                            term,
                            graph,
                            catalog_terms,
                            ontology_graph=ontology_graph,
                            position=position,
                        )
                        if catalog_terms
                        else [],
                    )
                )
                continue
            namespace = _namespace_of(text)
            if (
                catalog_terms
                and namespace in declared_namespaces
                and text not in catalog_terms
                and text not in fallback_terms
                and not namespace.startswith(standard_namespaces)
            ):
                flagged_terms.add(text)
                findings.append(
                    FactsUnitFinding(
                        kind=FactsUnitFindingKind.UNKNOWN_TERM,
                        message=(
                            f"<{text}> does not exist in its ontology; rewrite "
                            "the term IN PLACE to the closest correct term from "
                            "the ontology chapter (or a suggested candidate), "
                            "keeping the statement and its value. Do NOT delete "
                            "the statement."
                        ),
                        predicate=text,
                        suggestions=_alias_candidates(
                            term,
                            graph,
                            catalog_terms,
                            ontology_graph=ontology_graph,
                            position=position,
                        ),
                    )
                )

    findings.extend(
        _scalar_as_bounds_findings(graph, ontology_graph, normalized_fact_namespaces)
    )
    findings.extend(domain_violation_findings(graph, ontology_graph))
    findings.extend(
        _label_only_number_findings(
            graph,
            unit_properties=expand_vocabulary_terms(
                _vocabulary_role_subset(quantity_fallback_vocabulary, "unit"),
                graph,
                ontology_graph,
            ),
            numeric_value_properties=expand_vocabulary_terms(
                _vocabulary_role_subset(quantity_fallback_vocabulary, "numeric_value"),
                graph,
                ontology_graph,
            ),
            fact_namespaces=normalized_fact_namespaces,
        )
    )

    missing = missing_numeric_mentions(extraction_text, graph, limit=coverage_limit)
    if missing:
        findings.append(
            FactsUnitFinding(
                kind=FactsUnitFindingKind.NUMERIC_COVERAGE,
                mandatory=False,
                message=(
                    "These numbers appear in the source text but not in the "
                    "graph. For each: extract it as a typed literal on an "
                    "appropriate node (verbatim value and source unit — never "
                    "convert units) if it is a factual quantity, or ignore it "
                    "if it is a page/citation/figure artifact: " + ", ".join(missing)
                ),
                value=", ".join(missing),
            )
        )

    return findings

dedupe_literal_variants(graph, fact_namespaces=None)

Collapse duplicate literals differing only in language tag or datatype.

The renderer emits the same value inconsistently across chunks — "X"@en in one unit, "X"^^xsd:string in another, a plain "X" in a third — and after aggregation one (subject, predicate) carries all three as distinct RDF terms. One survives per lexical form: the language-tagged form (each distinct language kept — those are distinct assertions), else the plain form, else the xsd:string form. Reified provenance follows the survivor.

Parameters:

Name Type Description Default
graph RDFGraph

Aggregated facts graph, mutated in place.

 required
fact_namespaces Sequence[str] | None

When set, only subjects under these namespaces are touched.

 None

Returns:

Name Type Description
One list[GraphRepairRecord]
Source code in ontocast/tool/facts_validation/literal_repair.py 
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
def dedupe_literal_variants(
    graph: RDFGraph,
    fact_namespaces: Sequence[str] | None = None,
) -> list[GraphRepairRecord]:
    """Collapse duplicate literals differing only in language tag or datatype.

    The renderer emits the same value inconsistently across chunks —
    ``"X"@en`` in one unit, ``"X"^^xsd:string`` in another, a plain ``"X"``
    in a third — and after aggregation one ``(subject, predicate)`` carries
    all three as distinct RDF terms. One survives per lexical form: the
    language-tagged form (each distinct language kept — those are distinct
    assertions), else the plain form, else the ``xsd:string`` form. Reified
    provenance follows the survivor.

    Args:
        graph: Aggregated facts graph, mutated in place.
        fact_namespaces: When set, only subjects under these namespaces are
            touched.

    Returns:
        One :class:`GraphRepairRecord` per removed literal variant.
    """
    from ontocast.onto.rdfgraph import retarget_reifiers
    from ontocast.tool.facts_validation.terms import _in_fact_scope

    namespaces = [ns for ns in (fact_namespaces or []) if ns]
    groups: dict[tuple[URIRef, URIRef, str], list[Literal]] = {}
    for subject, predicate, obj in graph:
        if not isinstance(subject, URIRef) or not isinstance(predicate, URIRef):
            continue
        if not isinstance(obj, Literal):
            continue
        if namespaces and not _in_fact_scope(subject, namespaces):
            continue
        groups.setdefault((subject, predicate, str(obj)), []).append(obj)

    records: list[GraphRepairRecord] = []
    replacements: dict[tuple[Node, Node, Node], tuple[Node, Node, Node]] = {}
    for (subject, predicate, _lexical), literals in sorted(
        groups.items(), key=lambda item: (str(item[0][0]), str(item[0][1]), item[0][2])
    ):
        if len(literals) < 2:
            continue
        tagged = sorted(
            (lit for lit in literals if lit.language is not None),
            key=lambda lit: str(lit.language),
        )
        plain = [
            lit for lit in literals if lit.language is None and lit.datatype is None
        ]
        typed = [
            lit
            for lit in literals
            if lit.language is None and lit.datatype == XSD.string
        ]
        if tagged:
            keep = tagged[0]
            drop = plain + typed
        elif plain:
            keep = plain[0]
            drop = typed
        else:
            # Distinct datatypes beyond xsd:string are not variants of one
            # value; leave them to the datatype-aware repairs.
            continue
        for variant in drop:
            graph.remove((subject, predicate, variant))
            replacements[(subject, predicate, variant)] = (subject, predicate, keep)
            records.append(
                GraphRepairRecord(
                    kind=FactsGateRepairKind.LITERAL_VARIANT_PRUNED,
                    source=f"{subject} {predicate} {variant.n3()}",
                    target=keep.n3(),
                )
            )
    if replacements:
        retarget_reifiers(graph, replacements)
        logger.info(
            "Collapsed %d literal variant(s) differing only in language tag "
            "or datatype",
            len(records),
        )
    return records

domain_violation_findings(graph, ontology_graph)

Report subjects whose asserted type contradicts a predicate's domain.

Asserting a triple whose predicate declares an rdfs:domain entails that the subject belongs to that domain, so an untyped subject is never a violation -- the type is simply left to inference. It becomes one when the subject carries an asserted type that is unrelated to the declared domain: inference then adds the domain class on top of an incompatible one, and the contradiction surfaces later as a confusing failure somewhere else (SHACL reporting a missing property on a class the graph never meant to assert) rather than at the triple that caused it.

Conservative by construction, since a false accusation costs a render pass. A subject is reported only when it has at least one asserted type and every asserted type is unrelated to every declared domain -- neither a subtype nor a supertype of it, following rdfs:subClassOf and owl:equivalentClass intersections in both directions. Typing a subject with a supertype of the domain (sosa:Observation where the domain is obs:QuantitativeObservation) is consistent: inference specializes it, it contradicts nothing, and flagging it would bury the real violations.

Parameters:

Name Type Description Default
graph RDFGraph

Rendered facts graph for one unit.

 required
ontology_graph RDFGraph | None

Ontology context the renderer was given.

 required

Returns:

Name Type Description
list list[FactsUnitFinding]

One mandatory finding per offending (subject, predicate) pair,
list[FactsUnitFinding]

ordered by subject then predicate.
Source code in ontocast/tool/facts_validation/unit_findings.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def domain_violation_findings(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
) -> list[FactsUnitFinding]:
    """Report subjects whose asserted type contradicts a predicate's domain.

    Asserting a triple whose predicate declares an ``rdfs:domain`` *entails*
    that the subject belongs to that domain, so an untyped subject is never a
    violation -- the type is simply left to inference. It becomes one when the
    subject carries an asserted type that is unrelated to the declared domain:
    inference then adds the domain class on top of an incompatible one, and
    the contradiction surfaces later as a confusing failure somewhere else
    (SHACL reporting a missing property on a class the graph never meant to
    assert) rather than at the triple that caused it.

    Conservative by construction, since a false accusation costs a render pass.
    A subject is reported only when it has at least one asserted type and every
    asserted type is *unrelated* to every declared domain -- neither a subtype
    nor a supertype of it, following ``rdfs:subClassOf`` and
    ``owl:equivalentClass`` intersections in both directions. Typing a subject
    with a supertype of the domain (``sosa:Observation`` where the domain is
    ``obs:QuantitativeObservation``) is consistent: inference specializes it,
    it contradicts nothing, and flagging it would bury the real violations.

    Args:
        graph: Rendered facts graph for one unit.
        ontology_graph: Ontology context the renderer was given.

    Returns:
        list: One mandatory finding per offending (subject, predicate) pair,
        ordered by subject then predicate.
    """
    if ontology_graph is None or not len(ontology_graph):
        return []
    domains = _declared_domains(ontology_graph)
    if not domains:
        return []
    described = _described_classes(ontology_graph)

    closures: dict[URIRef, set[URIRef]] = {}
    findings: list[FactsUnitFinding] = []
    reported: set[tuple[str, str]] = set()

    for subject, predicate, _ in sorted(graph, key=lambda t: (str(t[0]), str(t[1]))):
        declared = domains.get(predicate)
        if declared is None or not isinstance(subject, URIRef):
            continue
        # Only domains the context places in a hierarchy can be argued about.
        declared = {value for value in declared if value in described}
        if not declared:
            continue
        asserted = {
            value
            for value in graph.objects(subject, RDF.type)
            if isinstance(value, URIRef)
        }
        if not asserted or not asserted <= described:
            continue

        def closure(class_iri: URIRef) -> set[URIRef]:
            if class_iri not in closures:
                closures[class_iri] = _superclass_closure(class_iri, ontology_graph)
            return closures[class_iri]

        # Compatible in either direction: the asserted type specializes a
        # declared domain, or a declared domain specializes the asserted type.
        domain_closure = set().union(*(closure(value) for value in declared))
        if any(
            closure(asserted_type) & declared or asserted_type in domain_closure
            for asserted_type in asserted
        ):
            continue
        key = (str(subject), str(predicate))
        if key in reported:
            continue
        reported.add(key)
        expected = ", ".join(f"<{value}>" for value in sorted(declared, key=str))
        actual = ", ".join(f"<{value}>" for value in sorted(asserted, key=str))
        findings.append(
            FactsUnitFinding(
                kind=FactsUnitFindingKind.DOMAIN_VIOLATION,
                message=(
                    f"<{subject}> is typed {actual} but carries <{predicate}>, "
                    f"whose rdfs:domain is {expected}. Either type the subject "
                    "as the declared domain, or use the property that fits the "
                    "type it has."
                ),
                subject=str(subject),
                predicate=str(predicate),
                suggestions=sorted(str(value) for value in declared),
            )
        )
    return findings

expand_vocabulary_terms(vocabulary, *graphs)

Expand configured vocabulary terms (CURIEs or full IRIs) to IRI strings.

CURIEs are expanded against the prefix bindings of every graph given, in order; a CURIE whose prefix no graph binds is dropped rather than guessed.

Source code in ontocast/tool/facts_validation/terms.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def expand_vocabulary_terms(
    vocabulary: dict[str, str] | None,
    *graphs: RDFGraph | None,
) -> set[str]:
    """Expand configured vocabulary terms (CURIEs or full IRIs) to IRI strings.

    CURIEs are expanded against the prefix bindings of every graph given, in
    order; a CURIE whose prefix no graph binds is dropped rather than guessed.
    """
    terms: set[str] = set()
    if not vocabulary:
        return terms
    bindings: dict[str, str] = {}
    for graph in graphs:
        if graph is None:
            continue
        for prefix, uri in graph.namespaces():
            if prefix:
                bindings.setdefault(prefix, str(uri))
    for term in vocabulary.values():
        if not term:
            continue
        if term.startswith("http://") or term.startswith("https://"):
            terms.add(term)
            continue
        prefix, separator, local = term.partition(":")
        if separator and local and prefix in bindings:
            terms.add(bindings[prefix] + local)
    return terms

material_defects(findings, fixes, policy=None)

Every reason the unit is not acceptable, deterministic evidence first.

Parameters:

Name Type Description Default
findings Sequence[FactsUnitFinding]

Deterministic findings collected against the current graph.

 required
fixes Sequence[TripleFix]

Fixes the LLM critic proposed, if it ran. Empty is normal -- at MAX_VISITS=1 the critic never runs and acceptance rests entirely on the findings.

 required
policy FactsAcceptancePolicy | None

The deployment's cut. None uses the defaults.

 None

Returns:

Type Description
list[MaterialDefect]

Material defects; empty means accept. The list is returned rather than
list[MaterialDefect]

a bool so the caller can record why a unit was rejected, which the
list[MaterialDefect]

score gate never made recordable.
Source code in ontocast/tool/facts_validation/acceptance.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def material_defects(
    findings: Sequence[FactsUnitFinding],
    fixes: Sequence[TripleFix],
    policy: FactsAcceptancePolicy | None = None,
) -> list[MaterialDefect]:
    """Every reason the unit is not acceptable, deterministic evidence first.

    Args:
        findings: Deterministic findings collected against the current graph.
        fixes: Fixes the LLM critic proposed, if it ran. Empty is normal --
            at ``MAX_VISITS=1`` the critic never runs and acceptance rests
            entirely on the findings.
        policy: The deployment's cut. ``None`` uses the defaults.

    Returns:
        Material defects; empty means accept. The list is returned rather than
        a bool so the caller can record *why* a unit was rejected, which the
        score gate never made recordable.
    """
    active = policy if policy is not None else FactsAcceptancePolicy()
    defects = [
        MaterialDefect(
            source="finding", kind=str(finding.kind), message=finding.message
        )
        for finding in findings
        if active.blocks_finding(finding)
    ]
    defects.extend(
        MaterialDefect(
            source="critic_fix",
            kind=fix.action,
            message=fix.explanation,
        )
        for fix in fixes
        if active.blocks_fix(fix)
    )
    return defects

normalize_literals_against_schema(graph, ontology_graph)

Retype literals whose predicate declares a compatible rdfs:range.

Fixes the qudt:numericValue 230 vs "230"^^xsd:decimal drift at parse time, and the same drift for the date-like datatypes: when the schema declares a range in :data:_RETYPABLE_RANGE_DATATYPES and the lexical form parses as that datatype, the literal is rewritten with it.

A literal is only retyped from an untyped, xsd:string, or numeric source -- a string range must never be able to clobber a correctly typed value -- and language-tagged literals are left alone, since they are rdf:langString and retyping would discard the tag.

Returns:

Type Description
int

Number of retyped literals.
Source code in ontocast/tool/facts_validation/literal_repair.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def normalize_literals_against_schema(
    graph: RDFGraph, ontology_graph: RDFGraph | None
) -> int:
    """Retype literals whose predicate declares a compatible ``rdfs:range``.

    Fixes the ``qudt:numericValue 230`` vs ``"230"^^xsd:decimal`` drift at parse
    time, and the same drift for the date-like datatypes: when the schema
    declares a range in :data:`_RETYPABLE_RANGE_DATATYPES` and the lexical form
    parses as that datatype, the literal is rewritten with it.

    A literal is only retyped from an untyped, ``xsd:string``, or numeric source
    -- a string range must never be able to clobber a correctly typed value --
    and language-tagged literals are left alone, since they are
    ``rdf:langString`` and retyping would discard the tag.

    Returns:
        Number of retyped literals.
    """
    if ontology_graph is None:
        return 0
    declared_ranges: dict[URIRef, URIRef] = {}
    for predicate, range_iri in ontology_graph.subject_objects(RDFS.range):
        if (
            isinstance(predicate, URIRef)
            and isinstance(range_iri, URIRef)
            and range_iri in _RETYPABLE_RANGE_DATATYPES
        ):
            declared_ranges[predicate] = range_iri

    if not declared_ranges:
        return 0

    replacements: list[tuple[tuple, tuple]] = []
    for subject, predicate, obj in graph:
        if not isinstance(obj, Literal) or not isinstance(predicate, URIRef):
            continue
        target_datatype = declared_ranges.get(predicate)
        if target_datatype is None or obj.datatype == target_datatype:
            continue
        if obj.language is not None:
            continue
        numeric_target = target_datatype in _NUMERIC_RANGE_DATATYPES
        source_admissible = obj.datatype is None or obj.datatype == XSD.string
        if numeric_target:
            # Keep the pre-existing numeric->numeric promotion (integer to
            # decimal, say), which a source-side "untyped or string" rule alone
            # would silently drop.
            source_admissible = (
                source_admissible or obj.datatype in _NUMERIC_RANGE_DATATYPES
            )
        if not source_admissible:
            continue
        lexical = str(obj).strip()
        gregorian = _GREGORIAN_RANGE_PATTERNS.get(target_datatype)
        if numeric_target:
            parses = canonical_number(lexical) is not None
        elif gregorian is not None:
            parses = gregorian.match(lexical) is not None
        else:
            parses = _literal_parses_as(lexical, target_datatype)
        if not parses:
            continue
        replacements.append(
            (
                (subject, predicate, obj),
                (subject, predicate, Literal(lexical, datatype=target_datatype)),
            )
        )

    for old, new in replacements:
        graph.remove(old)
        graph.add(new)
    return len(replacements)

promote_degenerate_bounds(graph, *, numeric_value_property, lower_bound_property, upper_bound_property, inclusive_flag_properties=())

Rewrite equal lower/upper bounds into a single scalar value, in place.

A node whose lower and upper bounds carry the same canonical numeric value encodes an exact scalar as a fake range. The rewrite fires only when the encoding is unambiguous: exactly one literal per bound property, equal canonical values, no existing scalar on the node, and no exclusive-bound flag (an exclusive equal bound denotes an empty interval — malformed, and left for findings). Property IRIs are injected by the caller from configuration; nothing is hardcoded.

Returns:

Type Description
int

Number of nodes rewritten.
Source code in ontocast/tool/facts_validation/literal_repair.py 
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
def promote_degenerate_bounds(
    graph: RDFGraph,
    *,
    numeric_value_property: str,
    lower_bound_property: str,
    upper_bound_property: str,
    inclusive_flag_properties: Sequence[str] = (),
) -> int:
    """Rewrite equal lower/upper bounds into a single scalar value, in place.

    A node whose lower and upper bounds carry the same canonical numeric value
    encodes an exact scalar as a fake range. The rewrite fires only when the
    encoding is unambiguous: exactly one literal per bound property, equal
    canonical values, no existing scalar on the node, and no exclusive-bound
    flag (an exclusive equal bound denotes an empty interval — malformed, and
    left for findings). Property IRIs are injected by the caller from
    configuration; nothing is hardcoded.

    Returns:
        Number of nodes rewritten.
    """
    lower_ref = URIRef(lower_bound_property)
    upper_ref = URIRef(upper_bound_property)
    value_ref = URIRef(numeric_value_property)
    flag_refs = [URIRef(term) for term in inclusive_flag_properties]
    promoted = 0
    for subject in sorted(set(graph.subjects(lower_ref, None)), key=str):
        lowers = [obj for obj in graph.objects(subject, lower_ref)]
        uppers = [obj for obj in graph.objects(subject, upper_ref)]
        if len(lowers) != 1 or len(uppers) != 1:
            continue
        if not isinstance(lowers[0], Literal) or not isinstance(uppers[0], Literal):
            continue
        if (subject, value_ref, None) in graph:
            continue
        low = canonical_number(str(lowers[0]).strip())
        high = canonical_number(str(uppers[0]).strip())
        if low is None or low != high:
            continue
        if any(
            str(flag_value).strip().lower() == "false"
            for flag_ref in flag_refs
            for flag_value in graph.objects(subject, flag_ref)
        ):
            continue
        graph.remove((subject, lower_ref, lowers[0]))
        graph.remove((subject, upper_ref, uppers[0]))
        for flag_ref in flag_refs:
            for flag_value in list(graph.objects(subject, flag_ref)):
                graph.remove((subject, flag_ref, flag_value))
        graph.add((subject, value_ref, Literal(low, datatype=XSD.decimal)))
        promoted += 1
    if promoted:
        logger.info(
            "Promoted %d degenerate bound pair(s) to <%s>",
            promoted,
            numeric_value_property,
        )
    return promoted

promote_degenerate_bounds_from_vocabulary(graph, ontology_graph, vocabulary)

Run :func:promote_degenerate_bounds with properties from configuration.

Active only when the quantity vocabulary names all three roles — numeric_value, lower_bound, upper_bound (roles containing inclusive supply the optional bound flags). The default vocabulary carries no bound roles, so this is off unless a deployment configures its range encoding.

Source code in ontocast/tool/facts_validation/literal_repair.py 
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
def promote_degenerate_bounds_from_vocabulary(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    vocabulary: dict[str, str] | None,
) -> int:
    """Run :func:`promote_degenerate_bounds` with properties from configuration.

    Active only when the quantity vocabulary names all three roles —
    ``numeric_value``, ``lower_bound``, ``upper_bound`` (roles containing
    ``inclusive`` supply the optional bound flags). The default vocabulary
    carries no bound roles, so this is off unless a deployment configures its
    range encoding.
    """
    vocabulary = vocabulary or {}
    numeric_terms = expand_vocabulary_terms(
        {"numeric_value": vocabulary.get("numeric_value", "")}, graph, ontology_graph
    )
    lower_terms = expand_vocabulary_terms(
        {"lower_bound": vocabulary.get("lower_bound", "")}, graph, ontology_graph
    )
    upper_terms = expand_vocabulary_terms(
        {"upper_bound": vocabulary.get("upper_bound", "")}, graph, ontology_graph
    )
    inclusive_terms = expand_vocabulary_terms(
        _vocabulary_role_subset(vocabulary, "inclusive"), graph, ontology_graph
    )
    if len(numeric_terms) != 1 or len(lower_terms) != 1 or len(upper_terms) != 1:
        return 0
    return promote_degenerate_bounds(
        graph,
        numeric_value_property=next(iter(numeric_terms)),
        lower_bound_property=next(iter(lower_terms)),
        upper_bound_property=next(iter(upper_terms)),
        inclusive_flag_properties=sorted(inclusive_terms),
    )

record_facts_gate_metrics(metrics, *, report, repair_result, ontology_context_empty=False)

Write the validation-gate metrics both entry paths share.

The graph pipeline's VALIDATE_FACTS node and the single-unit gate behind /process_unit run the same checks minus the un-merge repair, and had drifted into two hand-maintained copies of these writes — so a metric added to one path was silently absent from the other, and batch dumps stopped being comparable across entry paths, which is the one thing they exist for. Merge-specific counters stay with the graph pipeline: they have no meaning for a single unit.

Takes a plain mapping rather than AgentState so the tool layer stays ignorant of the state graph.

Parameters:

Name Type Description Default
metrics MutableMapping[str, int | float | str | dict]

AgentState.retrieval_metrics, mutated in place.

 required
report FactsValidationReport

Validation report describing the graph that will be served.

 required
repair_result ShaclRepairResult

Outcome of :func:apply_shacl_repairs. Its counters are written only when the pass actually ran, so "SHACL did not run" stays distinguishable from "ran and found nothing".

 required
ontology_context_empty bool

Whether the facts were validated with no catalog vocabulary at all. The per-term non-catalog check cannot see this — with no context there is nothing to compare against — so it is reported here, where an empty context is known to be unexpected. Only the document path used to report it, which left /process_unit silently unable to say the same thing.

 False
Source code in ontocast/tool/facts_validation/gate.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def record_facts_gate_metrics(
    metrics: MutableMapping[str, int | float | str | dict],
    *,
    report: FactsValidationReport,
    repair_result: ShaclRepairResult,
    ontology_context_empty: bool = False,
) -> None:
    """Write the validation-gate metrics both entry paths share.

    The graph pipeline's ``VALIDATE_FACTS`` node and the single-unit gate behind
    ``/process_unit`` run the same checks minus the un-merge repair, and had
    drifted into two hand-maintained copies of these writes — so a metric added
    to one path was silently absent from the other, and batch dumps stopped
    being comparable across entry paths, which is the one thing they exist for.
    Merge-specific counters stay with the graph pipeline: they have no meaning
    for a single unit.

    Takes a plain mapping rather than ``AgentState`` so the tool layer stays
    ignorant of the state graph.

    Args:
        metrics: ``AgentState.retrieval_metrics``, mutated in place.
        report: Validation report describing the graph that will be served.
        repair_result: Outcome of :func:`apply_shacl_repairs`. Its counters are
            written only when the pass actually ran, so "SHACL did not run"
            stays distinguishable from "ran and found nothing".
        ontology_context_empty: Whether the facts were validated with no
            catalog vocabulary at all. The per-term non-catalog check cannot
            see this — with no context there is nothing to compare against — so
            it is reported here, where an empty context is known to be
            unexpected. Only the document path used to report it, which left
            ``/process_unit`` silently unable to say the same thing.
    """
    if ontology_context_empty:
        reason = metrics.get(
            RetrievalMetric.EMPTY_SNAPSHOT_REASON, "no ontology context was assembled"
        )
        logger.warning(
            "Validating facts against an empty ontology context (%s); every "
            "extracted term is outside the catalog.",
            reason,
        )
        metrics[RetrievalMetric.VALIDATED_WITHOUT_ONTOLOGY_CONTEXT] = True
    if repair_result.ran:
        metrics[RetrievalMetric.FACTS_SHACL_VIOLATIONS_BEFORE] = (
            repair_result.violations_before
        )
        metrics[RetrievalMetric.FACTS_SHACL_VIOLATIONS_AFTER] = (
            repair_result.violations_after
        )
        metrics[RetrievalMetric.FACTS_SHACL_REPAIRS] = len(repair_result.records)
        metrics[RetrievalMetric.FACTS_SHACL_AUTOFIX_PASSES] = (
            repair_result.passes_applied
        )
        metrics[RetrievalMetric.FACTS_SHACL_AUTOFIX_REVERTED] = repair_result.reverted
    metrics[RetrievalMetric.FACTS_VALIDATION_FINDINGS] = len(report.findings)
    metrics[RetrievalMetric.FACTS_VALIDATION_ERRORS] = len(report.error_findings)

repair_literal_type_objects(graph)

Coerce literal rdf:type objects into IRIs.

The renderer sometimes emits a "prefix:Class"^^xsd:string instead of a prefix:Class (JSON-LD bare-string type values parse the same way). A literal-typed node is invisible to SPARQL class queries, reasoning, and the aggregator's URI minting/entity matching, all of which guard on isinstance(obj, URIRef). Absolute IRIs and compact IRIs bound in the graph are rewritten deterministically; unresolvable forms become MANDATORY findings.

Returns:

Type Description
int

Tuple of (number of rewritten triples, unresolved findings,
list[FactsUnitFinding]

applied-repair records).
Source code in ontocast/tool/facts_validation/literal_repair.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def repair_literal_type_objects(
    graph: RDFGraph,
) -> tuple[int, list[FactsUnitFinding], list[GraphRepairRecord]]:
    """Coerce literal ``rdf:type`` objects into IRIs.

    The renderer sometimes emits ``a "prefix:Class"^^xsd:string`` instead of
    ``a prefix:Class`` (JSON-LD bare-string type values parse the same way).
    A literal-typed node is invisible to SPARQL class queries, reasoning, and
    the aggregator's URI minting/entity matching, all of which guard on
    ``isinstance(obj, URIRef)``. Absolute IRIs and compact IRIs bound in the
    graph are rewritten deterministically; unresolvable forms become MANDATORY
    findings.

    Returns:
        Tuple of (number of rewritten triples, unresolved findings,
        applied-repair records).
    """
    prefix_map = {
        prefix: str(namespace) for prefix, namespace in graph.namespaces() if prefix
    }
    rewritten = 0
    findings: list[FactsUnitFinding] = []
    applied: list[GraphRepairRecord] = []
    for subject, predicate, obj in list(graph.triples((None, RDF.type, None))):
        if not isinstance(obj, Literal):
            continue
        lexical = str(obj).strip()
        resolved = _resolve_type_literal(lexical, prefix_map)
        if resolved is not None:
            graph.remove((subject, predicate, obj))
            graph.add((subject, RDF.type, URIRef(resolved)))
            rewritten += 1
            applied.append(
                GraphRepairRecord(
                    kind=FactsUnitFindingKind.LITERAL_TYPE_OBJECT,
                    source=lexical,
                    target=resolved,
                )
            )
            logger.info(
                "Repaired literal rdf:type object %r -> <%s>", lexical, resolved
            )
            continue
        findings.append(
            FactsUnitFinding(
                kind=FactsUnitFindingKind.LITERAL_TYPE_OBJECT,
                message=(
                    f"rdf:type object '{lexical}' is a string literal, not an "
                    "IRI; assert the type as a catalog class IRI "
                    "(`a prefix:Class`), never as a quoted string."
                ),
                subject=str(subject),
                value=lexical,
            )
        )
    return rewritten, findings, applied

repair_property_aliases(graph, ontology_graph, *, min_ratio=0.85, exempt_terms=None)

Rewrite near-miss predicates in catalog namespaces; report ambiguity.

A predicate whose namespace belongs to the ontology context but which is not itself a catalog term is a near-miss (qqval:lowerBound for qqval:hasLowerBound). When exactly one candidate scores above min_ratio (token containment counts as 1.0) the rewrite is applied deterministically; otherwise a mandatory finding carries the top suggestions.

Only namespaces the catalog declares terms in are eligible (see :func:collect_declared_namespaces); exempt_terms (expanded fallback vocabulary) are never treated as near-misses.

Returns:

Type Description
int

Tuple of (number of rewritten triples, unresolved findings,
list[FactsUnitFinding]

applied-repair records).
Source code in ontocast/tool/facts_validation/literal_repair.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def repair_property_aliases(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    *,
    min_ratio: float = 0.85,
    exempt_terms: set[str] | None = None,
) -> tuple[int, list[FactsUnitFinding], list[GraphRepairRecord]]:
    """Rewrite near-miss predicates in catalog namespaces; report ambiguity.

    A predicate whose namespace belongs to the ontology context but which is
    not itself a catalog term is a near-miss (``qqval:lowerBound`` for
    ``qqval:hasLowerBound``). When exactly one candidate scores above
    ``min_ratio`` (token containment counts as 1.0) the rewrite is applied
    deterministically; otherwise a mandatory finding carries the top
    suggestions.

    Only namespaces the catalog *declares* terms in are eligible (see
    :func:`collect_declared_namespaces`); ``exempt_terms`` (expanded fallback
    vocabulary) are never treated as near-misses.

    Returns:
        Tuple of (number of rewritten triples, unresolved findings,
        applied-repair records).
    """
    catalog_terms = collect_catalog_terms(ontology_graph)
    if not catalog_terms:
        return 0, [], []
    declared_namespaces = collect_declared_namespaces(ontology_graph)
    exempt = exempt_terms or set()

    findings: list[FactsUnitFinding] = []
    applied: list[GraphRepairRecord] = []
    rewritten = 0
    predicates = {
        predicate
        for predicate in graph.predicates()
        if isinstance(predicate, URIRef)
        and str(predicate) not in catalog_terms
        and str(predicate) not in exempt
        and _namespace_of(str(predicate)) in declared_namespaces
    }
    for alias in sorted(predicates, key=str):
        candidates = _alias_candidates(
            alias, graph, catalog_terms, ontology_graph=ontology_graph
        )
        strong = [
            candidate
            for candidate in candidates
            if _name_tokens(_local_name(str(alias)))
            and (
                _name_tokens(_local_name(str(alias)))
                <= _name_tokens(_local_name(candidate))
                or _name_tokens(_local_name(candidate))
                <= _name_tokens(_local_name(str(alias)))
                or SequenceMatcher(
                    None,
                    _local_name(str(alias)).lower(),
                    _local_name(candidate).lower(),
                ).ratio()
                >= min_ratio
            )
        ]
        if len(strong) == 1:
            replacement = URIRef(strong[0])
            alias_triples = 0
            for subject, predicate, obj in list(graph.triples((None, alias, None))):
                graph.remove((subject, predicate, obj))
                graph.add((subject, replacement, obj))
                rewritten += 1
                alias_triples += 1
            applied.append(
                GraphRepairRecord(
                    kind=FactsUnitFindingKind.PROPERTY_ALIAS,
                    source=str(alias),
                    target=str(replacement),
                    triple_count=alias_triples,
                )
            )
            logger.info("Repaired property alias %s -> %s", alias, replacement)
            continue
        findings.append(
            FactsUnitFinding(
                kind=FactsUnitFindingKind.PROPERTY_ALIAS,
                message=(
                    f"Predicate <{alias}> is not defined in its ontology; "
                    "replace it with the correct catalog property."
                ),
                predicate=str(alias),
                suggestions=candidates,
            )
        )
    return rewritten, findings, applied

resolve_code_literals(graph, ontology_graph, code_predicates=())

Link nodes to the catalog individual whose code they already carry.

A renderer that reads 4-15 days often annotates the value node with the code it saw — qudt:ucumCode "d" — instead of the object property that points at the individual — qudt:unit unit:DAY. The graph is well-formed, so no range check fires, but every query reading the object property gets an unbound result. The code came from the text and the individual is in the catalog, so the link is recoverable without asking the model again.

Fully schema-driven, no vocabulary compiled in: the connecting property is whichever object property the ontology context declares with a range the resolved individual is typed as, and a domain the subject satisfies. If the schema offers several such properties, or none, nothing is added.

Parameters:

Name Type Description Default
graph RDFGraph

Rendered facts graph, repaired in place.

 required
ontology_graph RDFGraph | None

Merged ontology context, read-only.

 required
code_predicates Sequence[str]

Predicates carrying machine-resolvable codes.

 ()

Returns:

Type Description
tuple[int, list[GraphRepairRecord]]

Tuple of (number of added triples, applied-repair records).
Source code in ontocast/tool/facts_validation/literal_repair.py 
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
def resolve_code_literals(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    code_predicates: Sequence[str] = (),
) -> tuple[int, list[GraphRepairRecord]]:
    """Link nodes to the catalog individual whose code they already carry.

    A renderer that reads ``4-15 days`` often annotates the value node with the
    code it saw — ``qudt:ucumCode "d"`` — instead of the object property that
    points at the individual — ``qudt:unit unit:DAY``. The graph is well-formed,
    so no range check fires, but every query reading the object property gets
    an unbound result. The code came from the text and the individual is in the
    catalog, so the link is recoverable without asking the model again.

    Fully schema-driven, no vocabulary compiled in: the connecting property is
    whichever object property the ontology context declares with a range the
    resolved individual is typed as, and a domain the subject satisfies. If the
    schema offers several such properties, or none, nothing is added.

    Args:
        graph: Rendered facts graph, repaired in place.
        ontology_graph: Merged ontology context, read-only.
        code_predicates: Predicates carrying machine-resolvable codes.

    Returns:
        Tuple of (number of added triples, applied-repair records).
    """
    if ontology_graph is None or not code_predicates:
        return 0, []
    code_terms = [URIRef(predicate) for predicate in code_predicates]
    # Only the code predicates themselves resolve here: a label match is a
    # different, much weaker signal and belongs to the shapes-driven pass.
    code_index: dict[str, set[str]] = {}
    for predicate in code_terms:
        for subject, value in ontology_graph.subject_objects(predicate):
            if isinstance(subject, URIRef) and isinstance(value, Literal):
                text = str(value).strip()
                if text:
                    code_index.setdefault(text, set()).add(str(subject))
    if not code_index:
        return 0, []

    domains = _declared_domains(ontology_graph)
    ranges: dict[URIRef, set[URIRef]] = {}
    for predicate, _, range_iri in ontology_graph.triples((None, RDFS.range, None)):
        if isinstance(predicate, URIRef) and isinstance(range_iri, URIRef):
            ranges.setdefault(predicate, set()).add(range_iri)

    # Superclass closures repeat heavily across literals; memoise per call.
    closures: dict[URIRef, set[URIRef]] = {}

    def closure(class_iri: URIRef) -> set[URIRef]:
        if class_iri not in closures:
            closures[class_iri] = _superclass_closure(class_iri, ontology_graph)
        return closures[class_iri]

    # The usage-evidence scan is a full graph walk; build it lazily, once,
    # only if some literal actually needs the no-declared-range fallback.
    linking_evidence: list[tuple[set[URIRef], set[URIRef], URIRef]] | None = None

    added = 0
    records: list[GraphRepairRecord] = []
    for code_predicate in code_terms:
        for subject, value in list(graph.subject_objects(code_predicate)):
            if not isinstance(subject, URIRef) or not isinstance(value, Literal):
                continue
            resolved = resolve_unique_surface(code_index, str(value))
            if resolved is None:
                continue
            resolved_types: set[URIRef] = set()
            for type_iri in ontology_graph.objects(resolved, RDF.type):
                if isinstance(type_iri, URIRef):
                    resolved_types |= closure(type_iri)
            if not resolved_types:
                continue
            subject_types: set[URIRef] = set()
            for type_iri in graph.objects(subject, RDF.type):
                if isinstance(type_iri, URIRef):
                    subject_types |= closure(type_iri)

            candidates = [
                predicate
                for predicate, range_set in ranges.items()
                if range_set & resolved_types
                and (
                    predicate not in domains
                    or not domains[predicate]
                    or domains[predicate] & subject_types
                )
            ]
            if not candidates:
                # Vendored vocabulary projections often declare individuals and
                # their codes but no rdfs:range (the shipped QUDT unit subset is
                # one). Fall back to how the graph already links this kind of
                # subject to this kind of individual -- the same
                # induce-from-usage move the functional-predicate harvest makes.
                if linking_evidence is None:
                    linking_evidence = _collect_linking_evidence(
                        graph, ontology_graph, closure
                    )
                candidates = _observed_linking_predicates(
                    linking_evidence, subject_types, resolved_types
                )
            # Already linked, ambiguous, or unsupported by the schema.
            candidates = [
                predicate
                for predicate in candidates
                if (subject, predicate, None) not in graph
            ]
            if len(candidates) != 1:
                continue
            predicate = candidates[0]
            graph.add((subject, predicate, resolved))
            added += 1
            records.append(
                GraphRepairRecord(
                    kind=FactsGateRepairKind.CODE_RESOLVED,
                    source=f"{code_predicate} {value.n3()}",
                    target=f"{predicate} {resolved}",
                )
            )
            logger.info(
                "Resolved code %s on <%s> to <%s %s>",
                value.n3(),
                subject,
                predicate,
                resolved,
            )
    return added, records

resolve_unique_surface(index, text)

The single IRI declaring text as a surface form, if exactly one does.

Source code in ontocast/tool/facts_validation/terms.py 
317
318
319
320
321
322
def resolve_unique_surface(index: dict[str, set[str]], text: str) -> URIRef | None:
    """The single IRI declaring ``text`` as a surface form, if exactly one does."""
    candidates = index.get(text.strip(), set())
    if len(candidates) != 1:
        return None
    return URIRef(next(iter(candidates)))

run_shacl(graph, shapes_graph, *, ontology_graph=None, inference='rdfs', advanced=True, max_triples=0)

Validate graph against shapes_graph, returning the violations.

Reaching here means shapes were found, so the caller expects validation to happen: a missing extra or a skipped run is reported at warning level, not debug. Silently returning "no violations" is indistinguishable from "conforms", so those cases return None.

The ontology context is mixed in (ont_graph) rather than left out. A facts graph states that a value uses unit:DAY; that the individual is a qudt:Unit is stated only in the catalog. Validating the facts alone therefore fails every sh:class constraint pointing at a catalog individual — violations that describe the missing schema, not the data.

RDFS inference is the default for the same reason. SHACL resolves class targets through rdfs:subClassOf on its own, but property paths carry no entailment: a shape on obs:hasResult does not see the life:hasStorageResult the renderer emitted, and reports the more specific statement as a missing one, so turning inference off raises the violation count rather than lowering it.

Parameters:

Name Type Description Default
graph RDFGraph

Data graph to validate.

 required
shapes_graph RDFGraph

Shapes to validate against.

 required
ontology_graph RDFGraph | None

Schema mixed into the data graph for validation.

 None
inference str

pyshacl pre-inference (none / rdfs / owlrl).

 'rdfs'
advanced bool

Enable SHACL Advanced Features.

 True
max_triples int

Skip validation above this graph size; 0 disables.

 0

Returns:

Type Description
list[ShaclViolation] | None

Violations in report order, or None when validation did not run.
Source code in ontocast/tool/facts_validation/shacl.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def run_shacl(
    graph: RDFGraph,
    shapes_graph: RDFGraph,
    *,
    ontology_graph: RDFGraph | None = None,
    inference: str = "rdfs",
    advanced: bool = True,
    max_triples: int = 0,
) -> list[ShaclViolation] | None:
    """Validate ``graph`` against ``shapes_graph``, returning the violations.

    Reaching here means shapes were found, so the caller expects validation to
    happen: a missing extra or a skipped run is reported at warning level, not
    debug. Silently returning "no violations" is indistinguishable from
    "conforms", so those cases return ``None``.

    The ontology context is mixed in (``ont_graph``) rather than left out. A
    facts graph states that a value uses ``unit:DAY``; that the individual *is*
    a ``qudt:Unit`` is stated only in the catalog. Validating the facts alone
    therefore fails every ``sh:class`` constraint pointing at a catalog
    individual — violations that describe the missing schema, not the data.

    RDFS inference is the default for the same reason. SHACL resolves class
    targets through ``rdfs:subClassOf`` on its own, but property paths carry no
    entailment: a shape on ``obs:hasResult`` does not see the
    ``life:hasStorageResult`` the renderer emitted, and reports the more
    specific statement as a missing one, so turning inference off raises the
    violation count rather than lowering it.

    Args:
        graph: Data graph to validate.
        shapes_graph: Shapes to validate against.
        ontology_graph: Schema mixed into the data graph for validation.
        inference: pyshacl pre-inference (``none`` / ``rdfs`` / ``owlrl``).
        advanced: Enable SHACL Advanced Features.
        max_triples: Skip validation above this graph size; 0 disables.

    Returns:
        Violations in report order, or ``None`` when validation did not run.
    """
    try:
        import pyshacl
    except ImportError:
        logger.warning(
            "SHACL shapes are configured but pyshacl is not installed; "
            "skipping SHACL validation. Install the extra: uv sync --extra shacl"
        )
        return None

    if max_triples and len(graph) > max_triples:
        logger.warning(
            "Skipping SHACL validation: %d triples exceeds "
            "FACTS_SHACL_MAX_TRIPLES=%d. The graph is unvalidated, not conformant.",
            len(graph),
            max_triples,
        )
        return None

    # pyshacl clones and mixes the data graph through plain rdflib graphs,
    # which cannot hold the RDF 1.2 triple terms an oxigraph-backed aggregated
    # graph carries (rdflib ``Graph.add`` asserts on them). Hand pyshacl a
    # sanitised copy; the dropped reification provenance carries no shape
    # targets, so validation loses nothing.
    data_graph = RDFGraph()
    copy_triples(graph, data_graph, origin="run_shacl")
    for prefix, namespace in graph.namespaces():
        data_graph.bind(prefix, namespace, override=True)

    conforms, results_graph, _ = pyshacl.validate(
        data_graph,
        shacl_graph=shapes_graph,
        ont_graph=(
            ontology_graph
            if ontology_graph is not None and len(ontology_graph)
            else None
        ),
        inference=inference,
        advanced=advanced,
        abort_on_first=False,
    )
    if conforms:
        return []

    violations: list[ShaclViolation] = []
    for result in results_graph.subjects(RDF.type, SH.ValidationResult):
        severity_iri = results_graph.value(result, SH.resultSeverity)
        message = results_graph.value(result, SH.resultMessage)
        path = results_graph.value(result, SH.resultPath)
        component = results_graph.value(result, SH.sourceConstraintComponent)
        source_shape = results_graph.value(result, SH.sourceShape)
        violations.append(
            ShaclViolation(
                focus=results_graph.value(result, SH.focusNode),
                path=path if isinstance(path, URIRef) else None,
                value=results_graph.value(result, SH.value),
                component=component if isinstance(component, URIRef) else None,
                source_shape=source_shape,
                severity=("error" if severity_iri == SH.Violation else "warning"),
                message=str(message) if message else "SHACL constraint violated.",
            )
        )
    return violations

shacl_catalog_contradictions(shapes_graph, ontology_graph, *, policy=None)

Property paths the shapes require but the unit validator would flag.

A SHACL property shape with sh:minCount >= 1 demands a property that the deterministic UNKNOWN_TERM check — same closure rules, same exemptions — would report as not existing. Data cannot satisfy both: the renderer is ordered to remove exactly what validation requires. Found live in practice, where shapes required qudt:numericValue while the validator's mandatory findings drove repair renders to delete it. Callers log the returned IRIs as configuration errors.

Source code in ontocast/tool/facts_validation/shacl.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def shacl_catalog_contradictions(
    shapes_graph: RDFGraph | None,
    ontology_graph: RDFGraph | None,
    *,
    policy: ValidationPolicy | None = None,
) -> list[str]:
    """Property paths the shapes require but the unit validator would flag.

    A SHACL property shape with ``sh:minCount >= 1`` demands a property that
    the deterministic UNKNOWN_TERM check — same closure rules, same
    exemptions — would report as not existing. Data cannot satisfy both: the
    renderer is ordered to remove exactly what validation requires. Found live
    in practice, where shapes required ``qudt:numericValue``
    while the validator's mandatory findings drove repair renders to delete
    it. Callers log the returned IRIs as configuration errors.
    """
    if shapes_graph is None or ontology_graph is None:
        return []
    catalog_terms = collect_catalog_terms(ontology_graph)
    if not catalog_terms:
        return []
    policy = policy or ValidationPolicy()
    declared_namespaces = collect_declared_namespaces(ontology_graph)
    standard_namespaces = policy.standard_namespaces()
    fallback_terms = policy.exempt_terms(shapes_graph, ontology_graph)
    required: set[str] = set()
    for shape, path in shapes_graph.subject_objects(SH.path):
        if not isinstance(path, URIRef):
            continue
        min_count = next(shapes_graph.objects(shape, SH.minCount), None)
        try:
            if min_count is None or int(str(min_count)) < 1:
                continue
        except ValueError:
            continue
        required.add(str(path))
    contradictions = [
        term
        for term in sorted(required)
        if _namespace_of(term) in declared_namespaces
        and term not in catalog_terms
        and term not in fallback_terms
        and not _namespace_of(term).startswith(standard_namespaces)
    ]
    return contradictions

summarize_conformance(findings, *, shacl_evaluated=None, repairs=())

Roll findings up into the shape a report or a client can read.

Counting by constraint component is what separates "168 violations" from "two systematic defects": 71 missing-qualifier violations on one shape are one modelling gap, not 71 problems to triage.

Parameters:

Name Type Description Default
findings Sequence[FactsValidationFinding]

Residual findings after any repair.

 required
shacl_evaluated bool | None

Whether SHACL actually ran (see :class:FactsValidationReport).

 None
repairs Sequence[GraphRepairRecord]

LLM-free repairs the gate applied.

 ()

Returns:

Type Description
dict

conforms (None when SHACL did not run), counts by severity, by
dict

finding kind, by SHACL constraint component and shape, and the applied
dict

repair counts by kind.
Source code in ontocast/tool/facts_validation/gate.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def summarize_conformance(
    findings: Sequence[FactsValidationFinding],
    *,
    shacl_evaluated: bool | None = None,
    repairs: Sequence[GraphRepairRecord] = (),
) -> dict:
    """Roll findings up into the shape a report or a client can read.

    Counting by constraint component is what separates "168 violations" from
    "two systematic defects": 71 missing-qualifier violations on one shape are
    one modelling gap, not 71 problems to triage.

    Args:
        findings: Residual findings after any repair.
        shacl_evaluated: Whether SHACL actually ran (see
            :class:`FactsValidationReport`).
        repairs: LLM-free repairs the gate applied.

    Returns:
        ``conforms`` (None when SHACL did not run), counts by severity, by
        finding kind, by SHACL constraint component and shape, and the applied
        repair counts by kind.
    """
    shacl_findings = [
        finding
        for finding in findings
        if finding.kind == FactsValidationFindingKind.SHACL
    ]
    by_kind: dict[str, int] = {}
    for finding in findings:
        by_kind[str(finding.kind)] = by_kind.get(str(finding.kind), 0) + 1
    by_component: dict[str, int] = {}
    by_shape: dict[str, int] = {}
    for finding in shacl_findings:
        if finding.component:
            key = _local_name(finding.component) or finding.component
            by_component[key] = by_component.get(key, 0) + 1
        if finding.source_shape:
            by_shape[finding.source_shape] = by_shape.get(finding.source_shape, 0) + 1
    repairs_by_kind: dict[str, int] = {}
    for record in repairs:
        repairs_by_kind[str(record.kind)] = repairs_by_kind.get(str(record.kind), 0) + 1

    return {
        "shacl_evaluated": shacl_evaluated,
        "conforms": None if not shacl_evaluated else not shacl_findings,
        "findings": len(findings),
        "errors": sum(1 for finding in findings if finding.severity == "error"),
        "warnings": sum(1 for finding in findings if finding.severity == "warning"),
        "by_kind": dict(sorted(by_kind.items())),
        "shacl_violations": len(shacl_findings),
        "shacl_by_constraint": dict(
            sorted(by_component.items(), key=lambda item: (-item[1], item[0]))
        ),
        "shacl_by_shape": dict(
            sorted(by_shape.items(), key=lambda item: (-item[1], item[0]))
        ),
        "repairs_applied": dict(sorted(repairs_by_kind.items())),
    }

validate_aggregated_facts(graph, ontology_graph, *, shapes_graph=None, fact_namespaces=None, suspect_multi_value_severity='error', functional_min_single_support=3, quantity_fallback_vocabulary=None, shacl_inference='rdfs', shacl_advanced=True, shacl_max_triples=0, key_supported_subjects=None)

Check post-merge invariants over the aggregated facts graph.

Deterministic defense-in-depth behind the merge guards: merge-signature violations here are almost always a bad identity merge, and error-severity findings of those kinds on merged subjects drive the un-merge repair. SHACL findings are reported but never drive it: a constraint violation says a node is under-specified, not that two entities were wrongly identified.

Checks
  • FUNCTIONAL_VIOLATION: >= 2 distinct objects on a predicate the schema constrains to at most one value (owl:FunctionalProperty or an OWL max-cardinality-1 restriction).
  • SUSPECT_MULTI_VALUE: >= 2 distinct canonical numeric values on one (subject, predicate); >= 2 mutually irreconcilable short string values on a predicate that is string-single-valued for a dominant majority (distinct names collapsed into one node); or >= 2 IRI objects on a predicate that is single-valued for a dominant majority of other subjects. Severity is configurable — legitimate multi-value modeling exists, bad merges are far more common.
  • DEGENERATE_COREFERENCE: one IRI object shared by >= 2 distinct functional-ish predicates of one subject (collapsed range bounds).
  • SHACL: optional, when pyshacl is installed and shapes exist.
  • NON_CATALOG_VOCABULARY: warning-only telemetry for terms the ontology context never supplied, which mark a retrieval miss the renderer papered over with a documented fallback.
  • MIXED_OBJECT_KINDS: warning-only telemetry for predicates used with both IRI and literal objects across the graph.

Parameters:

Name Type Description Default
graph RDFGraph

Aggregated facts graph.

 required
ontology_graph RDFGraph | None

Merged ontology context (functionality harvest).

 required
shapes_graph RDFGraph | None

Optional SHACL shapes graph.

 None
fact_namespaces list[str] | None

When set, only subjects under these namespaces are reported (ontology entities are not the gate's business).

 None
suspect_multi_value_severity str

"error" or "warning" for SUSPECT_MULTI_VALUE findings.

 'error'
functional_min_single_support int

Minimum single-valued subjects before a predicate counts as dominantly single-valued.

 3
shacl_inference str

pyshacl pre-inference mode (see :func:run_shacl).

 'rdfs'
shacl_advanced bool

Enable SHACL Advanced Features.

 True
shacl_max_triples int

Skip SHACL above this graph size; 0 disables.

 0
key_supported_subjects Sequence[str] | None

Final URIs of merge clusters backed by natural-key evidence. Irreconcilable string values on these subjects are reported as warnings, not errors: "Application no. 36760/06" and "Case of Stanev v. Bulgaria" are two names for one key-confirmed case, and an error here would drive the un-merge repair to split a correct merge.

 None

Returns:

Type Description
FactsValidationReport

Report with all findings, ordered by subject.
Source code in ontocast/tool/facts_validation/gate.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
def validate_aggregated_facts(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    *,
    shapes_graph: RDFGraph | None = None,
    fact_namespaces: list[str] | None = None,
    suspect_multi_value_severity: str = "error",
    functional_min_single_support: int = 3,
    quantity_fallback_vocabulary: dict[str, str] | None = None,
    shacl_inference: str = "rdfs",
    shacl_advanced: bool = True,
    shacl_max_triples: int = 0,
    key_supported_subjects: Sequence[str] | None = None,
) -> FactsValidationReport:
    """Check post-merge invariants over the aggregated facts graph.

    Deterministic defense-in-depth behind the merge guards: merge-signature
    violations here are almost always a bad identity merge, and error-severity
    findings of those kinds on merged subjects drive the un-merge repair.
    SHACL findings are reported but never drive it: a constraint violation
    says a node is under-specified, not that two entities were wrongly
    identified.

    Checks:
        - ``FUNCTIONAL_VIOLATION``: >= 2 distinct objects on a predicate the
          schema constrains to at most one value (``owl:FunctionalProperty``
          or an OWL max-cardinality-1 restriction).
        - ``SUSPECT_MULTI_VALUE``: >= 2 distinct canonical numeric values on
          one (subject, predicate); >= 2 mutually irreconcilable short string
          values on a predicate that is string-single-valued for a dominant
          majority (distinct names collapsed into one node); or >= 2 IRI
          objects on a predicate that is single-valued for a dominant
          majority of other subjects. Severity is configurable — legitimate
          multi-value modeling exists, bad merges are far more common.
        - ``DEGENERATE_COREFERENCE``: one IRI object shared by >= 2 distinct
          functional-ish predicates of one subject (collapsed range bounds).
        - ``SHACL``: optional, when ``pyshacl`` is installed and shapes exist.
        - ``NON_CATALOG_VOCABULARY``: warning-only telemetry for terms the
          ontology context never supplied, which mark a retrieval miss the
          renderer papered over with a documented fallback.
        - ``MIXED_OBJECT_KINDS``: warning-only telemetry for predicates used
          with both IRI and literal objects across the graph.

    Args:
        graph: Aggregated facts graph.
        ontology_graph: Merged ontology context (functionality harvest).
        shapes_graph: Optional SHACL shapes graph.
        fact_namespaces: When set, only subjects under these namespaces are
            reported (ontology entities are not the gate's business).
        suspect_multi_value_severity: ``"error"`` or ``"warning"`` for
            SUSPECT_MULTI_VALUE findings.
        functional_min_single_support: Minimum single-valued subjects before a
            predicate counts as dominantly single-valued.
        shacl_inference: pyshacl pre-inference mode (see :func:`run_shacl`).
        shacl_advanced: Enable SHACL Advanced Features.
        shacl_max_triples: Skip SHACL above this graph size; 0 disables.
        key_supported_subjects: Final URIs of merge clusters backed by
            natural-key evidence. Irreconcilable *string* values on these
            subjects are reported as warnings, not errors: "Application no.
            36760/06" and "Case of Stanev v. Bulgaria" are two names for one
            key-confirmed case, and an error here would drive the un-merge
            repair to split a correct merge.

    Returns:
        Report with all findings, ordered by subject.
    """
    namespaces = [ns for ns in (fact_namespaces or []) if ns]
    key_supported = set(key_supported_subjects or ())
    functional = harvest_max_one_predicates(ontology_graph)

    # Provenance machinery (chunk nodes, derivation annotations) is
    # legitimately multi-valued and never the gate's business.
    provenance_subjects = {
        subject
        for subject in graph.subjects(RDF.type, PROV.Entity)
        if isinstance(subject, URIRef)
    }

    object_groups: dict[tuple[URIRef, URIRef], set] = {}
    iri_groups: dict[tuple[URIRef, URIRef], set[URIRef]] = {}
    string_groups: dict[tuple[URIRef, URIRef], set[str]] = {}
    predicate_object_kinds: dict[URIRef, dict[str, int]] = {}
    for subject, predicate, obj in graph:
        if (
            not isinstance(subject, URIRef)
            or not isinstance(predicate, URIRef)
            or predicate == RDF.type
            # owl:sameAs is the aggregator's own merge bookkeeping: the rewriter
            # emits `canonical owl:sameAs original` per remapped entity, so it
            # carries 1 object for an unmerged entity and N-1 for a cluster.
            # Left in, it reads as "dominantly single-valued" and every large
            # cluster becomes an error that drives the repair to un-merge it.
            or predicate == OWL.sameAs
            or subject in provenance_subjects
            or str(predicate).startswith(str(PROV))
        ):
            continue
        object_groups.setdefault((subject, predicate), set()).add(obj)
        if isinstance(obj, URIRef):
            iri_groups.setdefault((subject, predicate), set()).add(obj)
        elif isinstance(obj, Literal) and canonical_literal(obj) is None:
            normalized = normalize_string_value(str(obj))
            if 0 < len(normalized) <= _NAME_LIKE_MAX_VALUE_LENGTH:
                string_groups.setdefault((subject, predicate), set()).add(normalized)
        if _in_fact_scope(subject, namespaces) and predicate != RDFS.label:
            kinds = predicate_object_kinds.setdefault(predicate, {})
            kind = "iri" if isinstance(obj, URIRef) else "literal"
            kinds[kind] = kinds.get(kind, 0) + 1

    dominant_single = _dominant_single_valued_predicates(
        iri_groups, min_single_support=functional_min_single_support
    )
    # String-valued analogue, over short name-like values: the signature of
    # distinct entities collapsed into one node is a naming predicate that is
    # single-valued everywhere else suddenly carrying several irreconcilable
    # names.
    dominant_single_strings = _dominant_single_valued_predicates(
        string_groups, min_single_support=functional_min_single_support
    )
    functional_ish = functional | dominant_single

    findings: list[FactsValidationFinding] = []
    flagged_pairs: set[tuple[URIRef, URIRef]] = set()

    for (subject, predicate), objects in sorted(
        object_groups.items(), key=lambda item: (str(item[0][0]), str(item[0][1]))
    ):
        if not _in_fact_scope(subject, namespaces):
            continue
        distinct = _distinct_object_keys(objects)
        if predicate in functional and len(distinct) >= 2:
            flagged_pairs.add((subject, predicate))
            findings.append(
                FactsValidationFinding(
                    kind=FactsValidationFindingKind.FUNCTIONAL_VIOLATION,
                    message=(
                        f"<{subject}> holds {len(distinct)} distinct values for "
                        f"<{predicate}>, which the schema constrains to at most "
                        "one."
                    ),
                    subject=str(subject),
                    predicate=str(predicate),
                    values=sorted(str(obj) for obj in objects),
                )
            )
            continue

        numeric_values = {
            canonical[0]
            for obj in objects
            if isinstance(obj, Literal)
            and (canonical := canonical_literal(obj)) is not None
            and canonical[1] == "numeric"
        }
        if len(numeric_values) >= 2:
            flagged_pairs.add((subject, predicate))
            findings.append(
                FactsValidationFinding(
                    kind=FactsValidationFindingKind.SUSPECT_MULTI_VALUE,
                    severity=(
                        "error"
                        if suspect_multi_value_severity == "error"
                        else "warning"
                    ),
                    message=(
                        f"<{subject}> carries {len(numeric_values)} distinct "
                        f"numeric values on <{predicate}> — the signature of "
                        "distinct quantities collapsed into one node."
                    ),
                    subject=str(subject),
                    predicate=str(predicate),
                    values=sorted(numeric_values),
                )
            )
            continue

        string_forms = sorted(string_groups.get((subject, predicate), set()))
        if len(string_forms) >= 2 and predicate in dominant_single_strings:
            # Name variants of one entity are alias-compatible ("mr beer" /
            # "mr karlheinz beer"); irreconcilable values ("mrs e palm" /
            # "mrs w thomassen") mark distinct entities collapsed into one
            # node — the failure the numeric branch cannot see. On a merge
            # backed by a shared identifier value, disagreement is name
            # variance of one confirmed entity: warning, never a veto.
            if any(
                not string_values_compatible(left, right)
                for left, right in combinations(string_forms, 2)
            ):
                subject_key_supported = str(subject) in key_supported
                if not subject_key_supported:
                    flagged_pairs.add((subject, predicate))
                findings.append(
                    FactsValidationFinding(
                        kind=FactsValidationFindingKind.SUSPECT_MULTI_VALUE,
                        severity=(
                            "error"
                            if suspect_multi_value_severity == "error"
                            and not subject_key_supported
                            else "warning"
                        ),
                        message=(
                            f"<{subject}> carries {len(string_forms)} "
                            f"mutually irreconcilable string values on "
                            f"<{predicate}>, which is single-valued for a "
                            "dominant majority of other subjects — the "
                            "signature of distinct entities collapsed into "
                            "one node."
                        ),
                        subject=str(subject),
                        predicate=str(predicate),
                        values=string_forms,
                    )
                )
                continue

        iri_objects = iri_groups.get((subject, predicate), set())
        if (
            len(iri_objects) >= 2
            and predicate not in functional
            and predicate in dominant_single
        ):
            flagged_pairs.add((subject, predicate))
            findings.append(
                FactsValidationFinding(
                    kind=FactsValidationFindingKind.SUSPECT_MULTI_VALUE,
                    severity=(
                        "error"
                        if suspect_multi_value_severity == "error"
                        else "warning"
                    ),
                    message=(
                        f"<{subject}> points at {len(iri_objects)} objects via "
                        f"<{predicate}>, which is single-valued for every other "
                        "subject in this graph."
                    ),
                    subject=str(subject),
                    predicate=str(predicate),
                    values=sorted(str(obj) for obj in iri_objects),
                )
            )

    coreference: dict[tuple[URIRef, URIRef], set[URIRef]] = {}
    for (subject, predicate), objects in iri_groups.items():
        if predicate not in functional_ish or not _in_fact_scope(subject, namespaces):
            continue
        for obj in objects:
            coreference.setdefault((subject, obj), set()).add(predicate)
    for (subject, obj), predicates in sorted(
        coreference.items(), key=lambda item: (str(item[0][0]), str(item[0][1]))
    ):
        if len(predicates) < 2:
            continue
        findings.append(
            FactsValidationFinding(
                kind=FactsValidationFindingKind.DEGENERATE_COREFERENCE,
                message=(
                    f"<{subject}> reaches <{obj}> through "
                    f"{len(predicates)} single-valued predicates "
                    f"({', '.join(sorted(str(p) for p in predicates))}) — "
                    "distinct endpoints (e.g. range bounds) likely merged."
                ),
                subject=str(subject),
                predicate=", ".join(sorted(str(p) for p in predicates)),
                values=[str(obj)],
            )
        )

    shacl_evaluated: bool | None = None
    shacl_violations: list[ShaclViolation] = []
    if shapes_graph is not None and len(shapes_graph):
        violations = run_shacl(
            graph,
            shapes_graph,
            ontology_graph=ontology_graph,
            inference=shacl_inference,
            advanced=shacl_advanced,
            max_triples=shacl_max_triples,
        )
        shacl_evaluated = violations is not None
        shacl_violations = list(violations or [])
        # Filter on the violation, which still holds the focus as an RDF term.
        # Filtering the projected finding instead compared a stringified blank
        # node against namespace prefixes, so it matched nothing and every
        # blank-node violation was dropped from the report — including the ones
        # the repair pass had just acted on.
        findings.extend(
            violation.as_finding()
            for violation in shacl_violations
            if _violation_in_fact_scope(graph, violation.focus, namespaces)
        )

    findings.extend(
        _non_catalog_vocabulary_findings(
            graph, ontology_graph, namespaces, quantity_fallback_vocabulary
        )
    )

    findings.extend(_dangling_reference_findings(graph, namespaces))

    # Object-kind self-consistency: a predicate carrying IRI objects on some
    # subjects and literal objects on others ("worksFor <org>" here, "worksFor
    # 'Ministry of Justice'" there) is un-queryable by shape. Warning-only
    # telemetry — never a merge signature.
    for predicate in sorted(predicate_object_kinds, key=str):
        kinds = predicate_object_kinds[predicate]
        iri_count = kinds.get("iri", 0)
        literal_count = kinds.get("literal", 0)
        if iri_count and literal_count:
            findings.append(
                FactsValidationFinding(
                    kind=FactsValidationFindingKind.MIXED_OBJECT_KINDS,
                    severity="warning",
                    message=(
                        f"<{predicate}> is used with {iri_count} IRI object(s) "
                        f"and {literal_count} literal object(s) — the same "
                        "relation is asserted as a link on some subjects and "
                        "as a string on others, so no single query shape "
                        "matches both."
                    ),
                    predicate=str(predicate),
                )
            )

    return FactsValidationReport(
        findings=findings,
        shacl_evaluated=shacl_evaluated,
        shacl_violations=shacl_violations,
    )