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.

Modules:

Name Description
acceptance

What counts as a blocking defect in a rendered unit graph.
critic_patch

Compile critic-proposed fixes into a validated graph patch, with no LLM call.
gate

Document-level validation for the post-aggregation facts gate.
literal_repair

LLM-free parse-time repairs on rendered facts graphs.
shacl

SHACL execution, shape assembly, autofix repairs, and the catalog lint.
terms

Catalog term inventory, namespace closure rules, and alias candidates.
unit_findings

Deterministic per-unit findings on a rendered facts graph.

Attributes

__all__ = ['CompiledFixes', 'CriticPatchPolicy', 'FactsAcceptancePolicy', 'FactsValidationReport', 'MaterialDefect', 'ShaclRepairResult', 'ShaclViolation', 'ValidationPolicy', 'accept_reason', 'apply_compiled_patch', 'apply_shacl_repairs', 'build_surface_index', 'collect_catalog_terms', 'collect_declared_namespaces', 'collect_shacl_shapes', 'collect_unit_findings', 'compile_critic_fixes', 'dedupe_literal_variants', 'domain_violation_findings', 'expand_vocabulary_terms', 'material_defects', 'normalize_literals_against_schema', 'promote_degenerate_bounds', 'promote_degenerate_bounds_from_vocabulary', 'record_facts_gate_metrics', 'repair_compact_iri_literals', 'repair_literal_type_objects', 'repair_property_aliases', 'resolve_code_literals', 'resolve_unique_surface', 'run_shacl', 'shacl_catalog_contradictions', 'count_shacl_focus_nodes', 'summarize_conformance', 'unit_numeric_inventory', 'validate_aggregated_facts'] module-attribute

Classes

CompiledFixes dataclass

The mechanical half of a critique, split from the half needing a render.

Source code in ontocast/tool/facts_validation/critic_patch.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
@dataclass(frozen=True)
class CompiledFixes:
    """The mechanical half of a critique, split from the half needing a render."""

    #: Delete-then-insert patch over every fix in ``applied``, or ``None``
    #: when nothing compiled. The whole-pass view; ``patches`` is the same
    #: content one fix at a time.
    update: GraphUpdate | None = None
    #: One patch per fix in ``applied``, in the order proposed.
    patches: list[FixPatch] = field(default_factory=list)
    #: Fixes folded into ``update``.
    applied: list[TripleFix] = field(default_factory=list)
    #: Fixes that could not be applied; counted and reported, not retried.
    residual: list[TripleFix] = field(default_factory=list)
    #: Fixes whose delete set and insert set are the same statements. They ask
    #: for no change, so they are neither applied nor sent back as work -- but
    #: they are counted, because a critique made mostly of these is a critic
    #: producing motion rather than corrections.
    noop: list[TripleFix] = field(default_factory=list)
    #: Ids cited that the index never issued. Counted, never guessed at.
    bad_index_refs: int = 0
    #: True when the delete-share cap fired: every fix that removes something
    #: went back as residual and only the pure additions were kept.
    delete_capped: bool = False
    #: Delete halves withheld by screening, in fixes.
    deletes_refused: int = 0
    #: Fixes whose payload named a prefix nothing declares -- neither the
    #: payload's own ``@context`` nor the unit graph's bindings -- so the
    #: statement it meant could not be identified. Sent back as residual and
    #: counted here rather than applied with ``prefix:local`` as the IRI.
    unresolved_prefix: int = 0
    #: Fixes whose inserts would mint a placeholder: a subject named for an
    #: ignored token or artifact, or a new node carrying only annotations
    #: and no type. The critic's way of "resolving" a coverage finding for a
    #: number it could not place; refused and sent back rather than applied.
    junk_refused: int = 0
    #: Fixes whose payload typed a literal its lexical form cannot hold --
    #: the same defect the render path quarantines. Sent back as residual
    #: rather than written into the unit graph, where every later
    #: serialization and validation walk would re-derive the failed value.
    quarantined_literal: int = 0

Attributes

applied = field(default_factory=list) class-attribute instance-attribute
bad_index_refs = 0 class-attribute instance-attribute
delete_capped = False class-attribute instance-attribute
deletes_refused = 0 class-attribute instance-attribute
junk_refused = 0 class-attribute instance-attribute
noop = field(default_factory=list) class-attribute instance-attribute
patches = field(default_factory=list) class-attribute instance-attribute
quarantined_literal = 0 class-attribute instance-attribute
residual = field(default_factory=list) class-attribute instance-attribute
unresolved_prefix = 0 class-attribute instance-attribute
update = None class-attribute instance-attribute

Methods:

__init__(update=None, patches=list(), applied=list(), residual=list(), noop=list(), bad_index_refs=0, delete_capped=False, deletes_refused=0, unresolved_prefix=0, junk_refused=0, quarantined_literal=0)

CriticPatchPolicy dataclass

How much destruction one critic pass is allowed to do.

A compiled patch is transparent -- both halves are known before anything is touched -- so the limits are enforced by withholding the delete half rather than by inspecting the wreckage afterwards. Every rule here drops something and counts it; none of them raises.

Source code in ontocast/tool/facts_validation/critic_patch.py 
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
@dataclass(frozen=True)
class CriticPatchPolicy:
    """How much destruction one critic pass is allowed to do.

    A compiled patch is transparent -- both halves are known before anything is
    touched -- so the limits are enforced by withholding the delete half rather
    than by inspecting the wreckage afterwards. Every rule here drops something
    and counts it; none of them raises.
    """

    #: Largest share of the target graph one pass may remove. A critique that
    #: wants to delete more than this has stopped correcting and started
    #: rewriting, which is the operation the loop deliberately does not offer.
    #: Over the cap the fixes that remove anything go back whole -- a REPLACE
    #: stripped of its delete half is an ADD of the new value beside the old
    #: one, a different edit from the one proposed -- and only pure additions
    #: are kept.
    max_delete_share: float = 0.25
    #: Deletions always permitted regardless of share. Without a floor the cap
    #: is strictest exactly where it should be loosest: on a short unit a single
    #: legitimate correction is already a large fraction of the graph.
    min_deletes: int = 5
    #: Whether a REPLACE may delete statements about one subject while writing
    #: about another. That is a rename, not a correction, and it leaves the new
    #: subject untyped and unlabelled while orphaning the old one.
    allow_subject_rename: bool = False

Attributes

allow_subject_rename = False class-attribute instance-attribute
max_delete_share = 0.25 class-attribute instance-attribute
min_deletes = 5 class-attribute instance-attribute

Methods:

__init__(max_delete_share=0.25, min_deletes=5, allow_subject_rename=False)

FactsAcceptancePolicy

Bases: BaseModel

Which defects block a rendered unit from leaving the loop.

Attributes:

Name Type Description
blocking_finding_kinds frozenset[str] | 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
90
91
92
93
94
95
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[str] | None = None
    blocking_fix_severity: BlockingFixSeverity = "critical"

    def blocks_finding(self, finding: UnitFinding) -> bool:
        """True when this deterministic finding must be repaired before exit.

        Typed on the shared base, and matched on the kind's *value*, so one
        policy serves both phases: the facts and ontology finding kinds are
        separate enums with no member in common, and the alternative was a
        second policy class differing only in an annotation.
        """
        if self.blocking_finding_kinds is None:
            return finding.mandatory
        return str(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. Acceptance is
        about whether the unit may leave the loop, and a removal that the patch
        screening refused -- because it would empty a subject, or exceed the
        delete cap -- is precisely a removal that should not hold the unit back.
        The screening decides what may be deleted; this decides what is worth
        another pass, and a deletion is never the thing worth insisting on.
        """
        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")

Attributes

blocking_finding_kinds = None class-attribute instance-attribute
blocking_fix_severity = 'critical' class-attribute instance-attribute

Methods:

blocks_finding(finding)

True when this deterministic finding must be repaired before exit.

Typed on the shared base, and matched on the kind's value, so one policy serves both phases: the facts and ontology finding kinds are separate enums with no member in common, and the alternative was a second policy class differing only in an annotation.

Source code in ontocast/tool/facts_validation/acceptance.py 
67
68
69
70
71
72
73
74
75
76
77
def blocks_finding(self, finding: UnitFinding) -> bool:
    """True when this deterministic finding must be repaired before exit.

    Typed on the shared base, and matched on the kind's *value*, so one
    policy serves both phases: the facts and ontology finding kinds are
    separate enums with no member in common, and the alternative was a
    second policy class differing only in an annotation.
    """
    if self.blocking_finding_kinds is None:
        return finding.mandatory
    return str(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. Acceptance is about whether the unit may leave the loop, and a removal that the patch screening refused -- because it would empty a subject, or exceed the delete cap -- is precisely a removal that should not hold the unit back. The screening decides what may be deleted; this decides what is worth another pass, and a deletion is never the thing worth insisting on.

Source code in ontocast/tool/facts_validation/acceptance.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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. Acceptance is
    about whether the unit may leave the loop, and a removal that the patch
    screening refused -- because it would empty a subject, or exceed the
    delete cap -- is precisely a removal that should not hold the unit back.
    The screening decides what may be deleted; this decides what is worth
    another pass, and a deletion is never the thing worth insisting on.
    """
    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"]

Attributes

error_findings property

Error-severity findings, whatever their kind.

findings = Field(default_factory=list) class-attribute instance-attribute
model_config = {'arbitrary_types_allowed': True} class-attribute instance-attribute
shacl_evaluated = 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'.") class-attribute instance-attribute
shacl_violations = 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.') class-attribute instance-attribute

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

Attributes

kind = Field(description="Finding kind, or the fix's action for a critic fix.") class-attribute instance-attribute
message instance-attribute
source instance-attribute

ShaclRepairResult

Bases: BaseModel

Outcome of the LLM-free SHACL repair pass.

Source code in ontocast/tool/facts_validation/shacl.py 
300
301
302
303
304
305
306
307
308
309
310
311
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

Attributes

graph instance-attribute
model_config = {'arbitrary_types_allowed': True} class-attribute instance-attribute
passes_applied = 0 class-attribute instance-attribute
ran = False class-attribute instance-attribute
records = Field(default_factory=list) class-attribute instance-attribute
reverted = False class-attribute instance-attribute
violations_after = 0 class-attribute instance-attribute
violations_before = 0 class-attribute instance-attribute

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

Attributes

component = None class-attribute instance-attribute
focus = None class-attribute instance-attribute
message = 'SHACL constraint violated.' class-attribute instance-attribute
model_config = {'arbitrary_types_allowed': True} class-attribute instance-attribute
path = None class-attribute instance-attribute
severity = 'error' class-attribute instance-attribute
source_shape = None class-attribute instance-attribute
value = None class-attribute instance-attribute

Methods:

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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
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, ...] = ()
    #: IRIs the shapes-derived conformance chapter instructs the renderer to
    #: emit. Same class of exemption as the fallback vocabulary: the prompt
    #: recommends these terms, so flagging them UNKNOWN_TERM would have the
    #: repair lane delete exactly what the prompt required. Threaded per unit
    #: from the tenancy's shapes catalog.
    contract_exempt_terms: tuple[str, ...] = ()
    #: Keep identifier digit groups out of the numeric-coverage inventory.
    #: Same class as the exemptions above -- something configuration says the
    #: deterministic checks must not put in front of the renderer.
    numeric_identifier_guard: bool = False
    #: Floor on the share of schema terms a render must take from the
    #: catalog before DOMAIN_ADHERENCE fires. 0 disables the check.
    domain_adherence_min_share: float = 0.15
    #: Fewest distinct schema terms a render must use before its catalog
    #: share is judged at all. A share over a handful of terms is noise: a
    #: front-matter unit typing an identifier and an author with generic
    #: vocabulary is not a render that abandoned the catalog.
    domain_adherence_min_terms: int = 4

    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, and the shapes-contract terms."""
        terms = expand_vocabulary_terms(self.quantity_fallback_vocabulary, *graphs)
        terms.update(self.code_predicates)
        terms.update(self.contract_exempt_terms)
        return terms

Attributes

additional_standard_namespaces = () class-attribute instance-attribute
code_predicates = () class-attribute instance-attribute
contract_exempt_terms = () class-attribute instance-attribute
domain_adherence_min_share = 0.15 class-attribute instance-attribute
domain_adherence_min_terms = 4 class-attribute instance-attribute
numeric_identifier_guard = False class-attribute instance-attribute
quantity_fallback_vocabulary = None class-attribute instance-attribute

Methods:

exempt_terms(*graphs)

Exact IRIs configuration blessed: fallback vocabulary, code predicates, and the shapes-contract terms.

Source code in ontocast/tool/facts_validation/terms.py 
172
173
174
175
176
177
178
def exempt_terms(self, *graphs: RDFGraph | None) -> set[str]:
    """Exact IRIs configuration blessed: fallback vocabulary, code
    predicates, and the shapes-contract terms."""
    terms = expand_vocabulary_terms(self.quantity_fallback_vocabulary, *graphs)
    terms.update(self.code_predicates)
    terms.update(self.contract_exempt_terms)
    return terms
standard_namespaces()

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

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

Functions:

accept_reason(defects)

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

Source code in ontocast/tool/facts_validation/acceptance.py 
138
139
140
141
142
143
144
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_compiled_patch(graph, update)

Apply a compiled patch to graph in place, deletes before inserts.

The ordering is the one :meth:GraphUpdateRenderReport.to_graph_update fixes, and the operations are the same TripleOps a render produces -- this is the render's apply step over an in-memory graph, not a second way to mutate one.

Parameters:

Name Type Description Default
graph RDFGraph

The unit graph to patch.

 required
update GraphUpdate

The compiled patch.

 required
Source code in ontocast/tool/facts_validation/critic_patch.py 
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
def apply_compiled_patch(graph: RDFGraph, update: GraphUpdate) -> None:
    """Apply a compiled patch to ``graph`` in place, deletes before inserts.

    The ordering is the one :meth:`GraphUpdateRenderReport.to_graph_update`
    fixes, and the operations are the same ``TripleOp``s a render produces --
    this is the render's apply step over an in-memory graph, not a second way
    to mutate one.

    Args:
        graph: The unit graph to patch.
        update: The compiled patch.
    """
    for operation in update.triple_operations:
        if operation.type == "delete":
            for triple in operation.graph:
                graph.remove(triple)
        else:
            for triple in operation.graph:
                graph.add(triple)

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 
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
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
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 
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
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, coverage_mandatory='off', policy=None, is_citation_metadata=False, is_non_content=False, full_catalog_terms=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 by default: numeric mentions of the source text absent from the graph, as two findings -- measurements (a number with its unit, in text order with context) and bare numbers -- whose mandatory flag coverage_mandatory sets (off | measurements | all; the booleans are off/all).

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.

A citation-metadata unit is rendered with the bibliographic vocabulary and told to keep away from the domain ontology, so its catalog share is zero by instruction; DOMAIN_ADHERENCE is not judged there, nor on a non-content unit (front matter, author notes, identifiers), which has no domain facts for the catalog to describe.

full_catalog_terms is the whole catalog's inventory, as opposed to the unit's retrieved snapshot in ontology_graph. A term the snapshot did not retrieve is still a term, so it is never reported as unknown -- and, for a term that is genuinely unknown, replacement candidates are offered only under token containment, never on similarity alone.

Source code in ontocast/tool/facts_validation/unit_findings.py 
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
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
def collect_unit_findings(
    *,
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    quarantined: list[RejectedLiteralTriple],
    extraction_text: str,
    fact_namespaces: list[str],
    coverage_limit: int = 30,
    coverage_mandatory: str | bool = "off",
    policy: ValidationPolicy | None = None,
    is_citation_metadata: bool = False,
    is_non_content: bool = False,
    full_catalog_terms: set[str] | 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 by default: numeric
    mentions of the source text absent from the graph, as two findings --
    measurements (a number with its unit, in text order with context) and
    bare numbers -- whose mandatory flag ``coverage_mandatory`` sets
    (``off`` | ``measurements`` | ``all``; the booleans are ``off``/``all``).

    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.

    A citation-metadata unit is rendered with the bibliographic vocabulary
    and told to keep away from the domain ontology, so its catalog share is
    zero by instruction; ``DOMAIN_ADHERENCE`` is not judged there, nor on a
    non-content unit (front matter, author notes, identifiers), which has no
    domain facts for the catalog to describe.

    ``full_catalog_terms`` is the whole catalog's inventory, as opposed to
    the unit's retrieved snapshot in ``ontology_graph``. A term the snapshot
    did not retrieve is still a term, so it is never reported as unknown --
    and, for a term that is genuinely unknown, replacement candidates are
    offered only under token containment, never on similarity alone.
    """
    policy = policy or ValidationPolicy()
    known_terms = set(full_catalog_terms or ())
    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=_containment_qualified(
                            term,
                            _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=_containment_qualified(
                            term,
                            _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 known_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=_containment_qualified(
                            term,
                            _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(
        _unit_symbol_case_findings(
            graph,
            ontology_graph,
            extraction_text,
            normalized_fact_namespaces,
            policy,
        )
    )
    findings.extend(domain_violation_findings(graph, ontology_graph))
    if not is_citation_metadata and not is_non_content:
        findings.extend(
            _domain_adherence_findings(
                graph,
                catalog_terms,
                normalized_fact_namespaces,
                policy.domain_adherence_min_share,
                min_terms=policy.domain_adherence_min_terms,
            )
        )
    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,
        )
    )

    if coverage_limit > 0:
        findings.extend(
            _numeric_coverage_findings(
                unit_numeric_inventory(
                    graph=graph,
                    ontology_graph=ontology_graph,
                    extraction_text=extraction_text,
                    policy=policy,
                    limit=coverage_limit,
                ),
                coverage_mode(coverage_mandatory),
            )
        )

    return findings

compile_critic_fixes(fixes, graph, *, index=None, policy=None)

Split a critique into a mechanical patch and the fixes needing a render.

Parameters:

Name Type Description Default
fixes Sequence[TripleFix]

Fixes from the critique report, in the order proposed.

 required
graph RDFGraph

The rendered unit graph the fixes refer to.

 required
index TripleIndex | None

The ids handed to the critic for this graph. When present, a fix that cites ids is resolved by lookup; the requoting path is the fallback for a fix that cites none.

 None
policy CriticPatchPolicy | None

Limits on what the patch may destroy. None uses the defaults, which are the facts-side ones.

 None

Returns:

Name Type Description
CompiledFixes CompiledFixes

update is delete-then-insert over the fixes in
CompiledFixes

applied; residual holds what needs judgement, noop what asks
CompiledFixes

for nothing.
Source code in ontocast/tool/facts_validation/critic_patch.py 
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
def compile_critic_fixes(
    fixes: Sequence[TripleFix],
    graph: RDFGraph,
    *,
    index: TripleIndex | None = None,
    policy: CriticPatchPolicy | None = None,
) -> CompiledFixes:
    """Split a critique into a mechanical patch and the fixes needing a render.

    Args:
        fixes: Fixes from the critique report, in the order proposed.
        graph: The rendered unit graph the fixes refer to.
        index: The ids handed to the critic for this graph. When present, a fix
            that cites ids is resolved by lookup; the requoting path is the
            fallback for a fix that cites none.
        policy: Limits on what the patch may destroy. ``None`` uses the
            defaults, which are the facts-side ones.

    Returns:
        CompiledFixes: ``update`` is delete-then-insert over the fixes in
        ``applied``; ``residual`` holds what needs judgement, ``noop`` what asks
        for nothing.
    """
    active = policy if policy is not None else CriticPatchPolicy()
    candidates: list[_Candidate] = []
    residual: list[TripleFix] = []
    noop: list[TripleFix] = []
    bad_index_refs = 0
    unresolved_prefix = 0
    junk_refused = 0
    quarantined_literal = 0

    for fix in fixes:
        parsed_correct = _parse_payload(fix.correct_value, graph)
        correct = parsed_correct.graph
        # A payload that named an unexpandable prefix is counted once per fix,
        # and only when the fix needed that payload: a REMOVE cited by id is
        # carried out whatever its unused correct_value says.
        unresolved = parsed_correct.unresolved_prefix and fix.action != "REMOVE"
        # A payload holding an invalid typed literal yields no graph, so an
        # ADD or REPLACE built on it has nothing to insert and falls through
        # to residual below. Counted here, once, where the reason is still
        # known -- and, as with an unresolved prefix, only when the fix needed
        # the payload at all.
        if parsed_correct.quarantined_literal and fix.action != "REMOVE":
            logger.info(
                "Critic fix refused (quarantined_literal): %s", fix.correct_value
            )
            quarantined_literal += 1
        insert_triples = (
            [triple for triple in correct if triple not in graph] if correct else []
        )
        if fix.action in ("ADD", "REPLACE") and insert_triples:
            junk = _junk_reason(insert_triples, graph)
            if junk is not None:
                # Whole fix back, delete half included: carrying out the
                # removal alone would be the delete-only edit the rollback
                # exists to undo.
                logger.info("Critic fix refused (%s): %s", junk, fix.correct_value)
                junk_refused += 1
                residual.append(fix)
                continue

        matched: list[Triple] = []
        if fix.action in ("REMOVE", "REPLACE"):
            if index is not None and fix.triple_ids:
                by_id, bad = _deletes_by_id(fix, graph, index)
                bad_index_refs += bad
                if by_id is None:
                    unresolved_prefix += int(unresolved)
                    residual.append(fix)
                    continue
                matched = by_id
            else:
                # Fallback for a fix that cites no id. The quoted statements must
                # all be present: a misquote has misunderstood the graph, and
                # acting on it would delete something the critic never looked at.
                parsed_incorrect = _parse_payload(fix.incorrect_value, graph)
                incorrect = parsed_incorrect.graph
                unresolved = unresolved or parsed_incorrect.unresolved_prefix
                quoted = (
                    [triple for triple in incorrect if triple in graph]
                    if incorrect
                    else []
                )
                if not quoted or len(quoted) != len(incorrect or []):
                    unresolved_prefix += int(unresolved)
                    residual.append(fix)
                    continue
                matched = quoted

        # A fix that removes exactly what it re-adds changes nothing. Left in
        # `residual` it would ask the next pass to redo nothing; applied, it
        # would inflate the count of fixes that "landed".
        if matched and set(matched) == set(correct or []):
            noop.append(fix)
            continue

        if fix.action == "REMOVE":
            if not matched:
                residual.append(fix)
                continue
            candidates.append(_Candidate(fix=fix, deletes=matched))
        elif fix.action == "ADD":
            if not insert_triples:
                unresolved_prefix += int(unresolved)
                residual.append(fix)
                continue
            candidates.append(_Candidate(fix=fix, inserts=insert_triples))
        elif fix.action == "REPLACE":
            if not matched or not correct:
                unresolved_prefix += int(unresolved)
                residual.append(fix)
                continue
            candidates.append(
                _Candidate(fix=fix, deletes=matched, inserts=list(correct))
            )
        else:
            residual.append(fix)

    kept, screened_out, refused, capped = _screen(candidates, graph, active)
    residual.extend(screened_out)

    deletes = RDFGraph()
    inserts = RDFGraph()
    applied: list[TripleFix] = []
    patches: list[FixPatch] = []
    for candidate in kept:
        if not candidate.deletes and not candidate.inserts:
            residual.append(candidate.fix)
            continue
        fix_deletes = RDFGraph()
        fix_inserts = RDFGraph()
        for triple in candidate.deletes:
            fix_deletes.add(triple)
        for triple in candidate.inserts:
            fix_inserts.add(triple)
        # Delete-then-insert leaves a triple on both sides present either way,
        # so dropping it from the delete side changes nothing and keeps the
        # patch honest about what it removes.
        for triple in fix_inserts:
            fix_deletes.remove(triple)
        operations: list[TripleOp] = []
        if len(fix_deletes):
            operations.append(TripleOp(type="delete", graph=fix_deletes))
        if len(fix_inserts):
            operations.append(TripleOp(type="insert", graph=fix_inserts))
        if not operations:
            noop.append(candidate.fix)
            continue
        patches.append(
            FixPatch(
                fix=candidate.fix,
                update=GraphUpdate(triple_operations=operations),
                deletes=len(fix_deletes),
                inserts=len(fix_inserts),
            )
        )
        applied.append(candidate.fix)
        for triple in fix_deletes:
            deletes.add(triple)
        for triple in fix_inserts:
            inserts.add(triple)

    for triple in inserts:
        deletes.remove(triple)

    if not patches:
        return CompiledFixes(
            residual=residual,
            noop=noop,
            bad_index_refs=bad_index_refs,
            delete_capped=capped,
            deletes_refused=refused,
            unresolved_prefix=unresolved_prefix,
            junk_refused=junk_refused,
            quarantined_literal=quarantined_literal,
        )

    combined: list[TripleOp] = []
    if len(deletes):
        combined.append(TripleOp(type="delete", graph=deletes))
    if len(inserts):
        combined.append(TripleOp(type="insert", graph=inserts))
    logger.info(
        "Critic fixes: %d compiled to patches (-%d/+%d triples), "
        "%d need a render, %d asked for no change, %d delete half(s) refused, "
        "%d refused as placeholders, %d with an unresolved prefix, "
        "%d with a quarantined literal%s",
        len(applied),
        len(deletes),
        len(inserts),
        len(residual),
        len(noop),
        refused,
        junk_refused,
        unresolved_prefix,
        quarantined_literal,
        " (delete-share cap fired)" if capped else "",
    )
    return CompiledFixes(
        update=GraphUpdate(triple_operations=combined),
        patches=patches,
        applied=applied,
        residual=residual,
        noop=noop,
        bad_index_refs=bad_index_refs,
        delete_capped=capped,
        deletes_refused=refused,
        unresolved_prefix=unresolved_prefix,
        junk_refused=junk_refused,
        quarantined_literal=quarantined_literal,
    )

count_shacl_focus_nodes(data_graph, shapes_graph)

Count data-graph nodes any shape actually targets.

This is the denominator conforms is silent about. A validation run over zero focus nodes reports no violations for the same reason an empty query returns no rows -- nothing was examined. Reported alongside conforms so a clean result cannot be read as a passing one when the shapes and the data never met.

Class targeting follows rdfs:subClassOf because SHACL's sh:targetClass is subclass-aware.

Parameters:

Name Type Description Default
data_graph RDFGraph

The graph that was validated.

 required
shapes_graph RDFGraph | None

The shapes it was validated against; None when SHACL did not run.

 required

Returns:

Type Description
int | None

int | None: Number of distinct focus nodes, or None if no shapes.
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
def count_shacl_focus_nodes(
    data_graph: RDFGraph, shapes_graph: RDFGraph | None
) -> int | None:
    """Count data-graph nodes any shape actually targets.

    This is the denominator ``conforms`` is silent about. A validation run over
    zero focus nodes reports no violations for the same reason an empty query
    returns no rows -- nothing was examined. Reported alongside ``conforms`` so
    a clean result cannot be read as a passing one when the shapes and the data
    never met.

    Class targeting follows ``rdfs:subClassOf`` because SHACL's
    ``sh:targetClass`` is subclass-aware.

    Args:
        data_graph: The graph that was validated.
        shapes_graph: The shapes it was validated against; ``None`` when SHACL
            did not run.

    Returns:
        int | None: Number of distinct focus nodes, or ``None`` if no shapes.
    """
    if shapes_graph is None or not len(shapes_graph):
        return None
    focus: set = set()
    for target_class in shapes_graph.objects(None, SH.targetClass):
        classes = {target_class}
        # Walk down the hierarchy: a shape on a superclass targets instances
        # of every subclass too.
        pending = [target_class]
        while pending:
            current = pending.pop()
            for sub in data_graph.subjects(RDFS.subClassOf, current):
                if sub not in classes:
                    classes.add(sub)
                    pending.append(sub)
        for klass in classes:
            focus.update(data_graph.subjects(RDF.type, klass))
    focus.update(shapes_graph.objects(None, SH.targetNode))
    for predicate in shapes_graph.objects(None, SH.targetSubjectsOf):
        focus.update(data_graph.subjects(predicate, None))
    for predicate in shapes_graph.objects(None, SH.targetObjectsOf):
        focus.update(data_graph.objects(None, predicate))
    return len(focus)

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 
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
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 
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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
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[UnitFinding]

Deterministic findings collected against the current graph.

 required
fixes Sequence[TripleFix]

Fixes the LLM critic proposed, if it ran. Empty is normal -- with FACTS_CRITIC_PASSES=0 the critic never runs and acceptance rests entirely on the findings. (The critic budget is independent of MAX_VISITS, which bounds failed renders only.)

 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 
 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
def material_defects(
    findings: Sequence[UnitFinding],
    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 --
            with ``FACTS_CRITIC_PASSES=0`` the critic never runs and
            acceptance rests entirely on the findings. (The critic budget is
            independent of ``MAX_VISITS``, which bounds failed renders only.)
        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 
 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
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 
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
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
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 
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
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 
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
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_compact_iri_literals(graph, ontology_context_graph, known_terms)

Coerce plain-string compact IRIs on IRI-valued positions into IRIs.

A JSON-LD bare string ("qudt:unit": "unit:NanoM") parses as a literal, so the node loses its link and reads as missing the property. A literal is rewritten only when every condition holds: it is plain (no language tag, no datatype or xsd:string); the predicate is not rdf:type (that is :func:repair_literal_type_objects); the lexical form is a compact IRI whose prefix the graph or the known-prefix context binds; and either the schema says the predicate takes an IRI or the expanded IRI is a known term. Prose such as "time: 10 minutes" never matches the compact-IRI form.

Parameters:

Name Type Description Default
graph RDFGraph

Facts graph, repaired in place.

 required
ontology_context_graph RDFGraph

Schema that says which predicates take IRIs.

 required
known_terms set[str]

IRIs the catalog and fallback vocabulary declare; covers predicates whose range the vendored vocabulary leaves undeclared.

 required

Returns:

Type Description
tuple[int, list[GraphRepairRecord]]

Tuple of (number of rewritten triples, applied-repair records).
Source code in ontocast/tool/facts_validation/literal_repair.py 
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
def repair_compact_iri_literals(
    graph: RDFGraph,
    ontology_context_graph: RDFGraph,
    known_terms: set[str],
) -> tuple[int, list[GraphRepairRecord]]:
    """Coerce plain-string compact IRIs on IRI-valued positions into IRIs.

    A JSON-LD bare string (``"qudt:unit": "unit:NanoM"``) parses as a literal,
    so the node loses its link and reads as missing the property. A literal is
    rewritten only when every condition holds: it is plain (no language tag,
    no datatype or ``xsd:string``); the predicate is not ``rdf:type`` (that is
    :func:`repair_literal_type_objects`); the lexical form is a compact IRI
    whose prefix the graph or the known-prefix context binds; and either the
    schema says the predicate takes an IRI or the expanded IRI is a known term.
    Prose such as ``"time: 10 minutes"`` never matches the compact-IRI form.

    Args:
        graph: Facts graph, repaired in place.
        ontology_context_graph: Schema that says which predicates take IRIs.
        known_terms: IRIs the catalog and fallback vocabulary declare; covers
            predicates whose range the vendored vocabulary leaves undeclared.

    Returns:
        Tuple of (number of rewritten triples, applied-repair records).
    """
    prefix_map = dict(RDFGraph.get_known_prefixes() or {})
    prefix_map.update(
        {prefix: str(namespace) for prefix, namespace in graph.namespaces() if prefix}
    )
    replacements: list[tuple[tuple[Node, Node, Node], URIRef, str]] = []
    expects_iri: dict[URIRef, bool] = {}
    for subject, predicate, obj in graph:
        if predicate == RDF.type or not isinstance(obj, Literal):
            continue
        if obj.language or obj.datatype not in (None, XSD.string):
            continue
        lexical = str(obj).strip()
        match = _COMPACT_IRI.match(lexical)
        if match is None:
            continue
        namespace = prefix_map.get(match.group(1))
        if not namespace:
            continue
        expanded = f"{namespace}{match.group(2)}"
        if expanded not in known_terms:
            if not isinstance(predicate, URIRef):
                continue
            if predicate not in expects_iri:
                expects_iri[predicate] = _predicate_expects_iri_object(
                    predicate, ontology_context_graph
                )[0]
            if not expects_iri[predicate]:
                continue
        replacements.append(((subject, predicate, obj), URIRef(expanded), lexical))

    applied: list[GraphRepairRecord] = []
    for (subject, predicate, obj), iri, lexical in replacements:
        graph.remove((subject, predicate, obj))
        graph.add((subject, predicate, iri))
        applied.append(
            GraphRepairRecord(
                kind=FactsUnitFindingKind.COMPACT_IRI_LITERAL,
                source=lexical,
                target=str(iri),
            )
        )
        logger.info("Repaired compact-IRI literal %r -> <%s>", lexical, iri)
    if applied:
        record_active_count("repair/compact_iri_literal", len(applied))
    return len(applied), applied

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 
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
223
224
225
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.95, exempt_terms=None, full_catalog_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). The rewrite is deterministic and fires only when exactly one candidate qualifies: its name tokens contain the alias's, the alias's contain its, or the two spell the same once case and separators are folded. Several qualifying candidates are a tie that the SequenceMatcher ratio may break -- the best-scoring one wins when it clears min_ratio and nothing ties it. Similarity alone never licenses a rewrite: a high ratio also joins names that differ by exactly one token, and rewriting across that gap substitutes one property for another. Anything else becomes a mandatory finding carrying 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, and neither is a predicate in full_catalog_terms: a term the whole catalog declares is real even when this unit's snapshot did not retrieve it, and the snapshot's look-alike is then a different property, not the intended one. Such a predicate is left untouched here; whether it is reported is the term checks' business.

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 
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
def repair_property_aliases(
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    *,
    min_ratio: float = 0.95,
    exempt_terms: set[str] | None = None,
    full_catalog_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``). The rewrite is deterministic and fires only when
    exactly one candidate *qualifies*: its name tokens contain the alias's,
    the alias's contain its, or the two spell the same once case and
    separators are folded. Several qualifying candidates are a tie that the
    ``SequenceMatcher`` ratio may break -- the best-scoring one wins when it
    clears ``min_ratio`` and nothing ties it. Similarity alone never licenses
    a rewrite: a high ratio also joins names that differ by exactly one token,
    and rewriting across that gap substitutes one property for another.
    Anything else becomes a mandatory finding carrying 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, and neither is a predicate
    in ``full_catalog_terms``: a term the whole catalog declares is real even
    when this unit's snapshot did not retrieve it, and the snapshot's
    look-alike is then a different property, not the intended one. Such a
    predicate is left untouched here; whether it is reported is the term
    checks' business.

    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()
    known_elsewhere = full_catalog_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 str(predicate) not in known_elsewhere
        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
        )
        target = _select_alias_target(str(alias), candidates, min_ratio)
        if target is not None:
            replacement = URIRef(target)
            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 
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
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 
337
338
339
340
341
342
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, catalog_terms=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.

Parameters:

Name Type Description Default
shapes_graph RDFGraph | None

The shapes the gate validates against.

 required
ontology_graph RDFGraph | None

The ontology context the facts were rendered against.

 required
policy ValidationPolicy | None

The exemptions the unit validator applies. Pass the same policy the unit loop uses -- including its shapes-contract exemptions -- or the check reports contradictions the validator never raises.

 None
catalog_terms Collection[str] | None

The whole catalog's term inventory, when ontology_graph is only a retrieved subset of it. The unit validator judges unknown terms against the whole catalog, so a term the snapshot omitted is not a contradiction.

 None
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
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def shacl_catalog_contradictions(
    shapes_graph: RDFGraph | None,
    ontology_graph: RDFGraph | None,
    *,
    policy: ValidationPolicy | None = None,
    catalog_terms: Collection[str] | 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.

    Args:
        shapes_graph: The shapes the gate validates against.
        ontology_graph: The ontology context the facts were rendered against.
        policy: The exemptions the unit validator applies. Pass the same
            policy the unit loop uses -- including its shapes-contract
            exemptions -- or the check reports contradictions the validator
            never raises.
        catalog_terms: The whole catalog's term inventory, when
            ``ontology_graph`` is only a retrieved subset of it. The unit
            validator judges unknown terms against the whole catalog, so a
            term the snapshot omitted is not a contradiction.
    """
    if shapes_graph is None or ontology_graph is None:
        return []
    catalog_terms = set(collect_catalog_terms(ontology_graph)) | set(
        catalog_terms or ()
    )
    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=(), focus_nodes=None)

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.

 ()
focus_nodes int | None

Data-graph nodes the shapes actually target, from :func:count_shacl_focus_nodes. Zero means the shapes examined nothing, so a violation-free result says nothing about the data.

 None

Returns:

Type Description
dict

conforms (None when SHACL did not run or when it ran over an
dict

empty focus set), counts by severity, by finding kind, by SHACL
dict

constraint component and shape, and the applied repair counts by kind.
dict

shacl_vacuous flags the empty-focus-set case, which is a
dict

measurement failure rather than a passing grade.
Source code in ontocast/tool/facts_validation/gate.py 
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
188
189
190
191
192
193
194
def summarize_conformance(
    findings: Sequence[FactsValidationFinding],
    *,
    shacl_evaluated: bool | None = None,
    repairs: Sequence[GraphRepairRecord] = (),
    focus_nodes: int | None = None,
) -> 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.
        focus_nodes: Data-graph nodes the shapes actually target, from
            :func:`count_shacl_focus_nodes`. Zero means the shapes examined
            nothing, so a violation-free result says nothing about the data.

    Returns:
        ``conforms`` (None when SHACL did not run **or** when it ran over an
        empty focus set), counts by severity, by finding kind, by SHACL
        constraint component and shape, and the applied repair counts by kind.
        ``shacl_vacuous`` flags the empty-focus-set case, which is a
        measurement failure rather than a passing grade.
    """
    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

    vacuous = bool(shacl_evaluated) and focus_nodes == 0
    return {
        "shacl_evaluated": shacl_evaluated,
        "conforms": (None if not shacl_evaluated or vacuous else not shacl_findings),
        "shacl_focus_nodes": focus_nodes,
        "shacl_vacuous": vacuous,
        "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())),
    }

unit_numeric_inventory(*, graph, ontology_graph, extraction_text, policy=None, limit=30)

Numbers of the unit text absent from its graph, measurements first.

The unit surfaces come from the unit individuals of the ontology context (found through the configured unit-role property), so a catalog-specific unit counts as a measurement once the catalog declares it. Measurements are judged against structured (number, unit) pairs in the graph — a bare numeric literal does not clear a unit-adjacent mention. Shared by the coverage findings and the completion pass, which must agree on what is missing.

Source code in ontocast/tool/facts_validation/unit_findings.py 
 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
def unit_numeric_inventory(
    *,
    graph: RDFGraph,
    ontology_graph: RDFGraph | None,
    extraction_text: str,
    policy: ValidationPolicy | None = None,
    limit: int = 30,
) -> NumericInventory:
    """Numbers of the unit text absent from its graph, measurements first.

    The unit surfaces come from the unit individuals of the ontology context
    (found through the configured unit-role property), so a catalog-specific
    unit counts as a measurement once the catalog declares it. Measurements
    are judged against structured ``(number, unit)`` pairs in the graph —
    a bare numeric literal does not clear a unit-adjacent mention. Shared by
    the coverage findings and the completion pass, which must agree on what
    is missing.
    """
    policy = policy or ValidationPolicy()
    unit_properties = expand_vocabulary_terms(
        _vocabulary_role_subset(policy.quantity_fallback_vocabulary, "unit"),
        graph,
        ontology_graph,
    )
    numeric_value_properties = expand_vocabulary_terms(
        _vocabulary_role_subset(policy.quantity_fallback_vocabulary, "numeric_value"),
        graph,
        ontology_graph,
    )
    return missing_numeric_inventory(
        extraction_text,
        graph,
        unit_surfaces=unit_surfaces_in_ontology(ontology_graph, unit_properties),
        ontology_graph=ontology_graph,
        numeric_value_properties=numeric_value_properties,
        unit_properties=unit_properties,
        ignore_identifier_fragments=policy.numeric_identifier_guard,
        limit=limit,
    )

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, cross_unit_pairs=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. The IRI branch additionally accepts cross_unit_pairs, which separates the two by provenance rather than by frequency.
  • 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
cross_unit_pairs Sequence[tuple[str, str]] | None

Canonical (subject, predicate) pairs whose IRI objects came from more than one unit. When supplied, an IRI-branch SUSPECT_MULTI_VALUE finding on a pair not listed here is reported as a warning and never vetoes a cluster: a single unit asserting two objects on one predicate is reading one sentence, not the residue of a bad identity decision. None disables the distinction.

 None
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: a registry number and a full title can be two names for one key-confirmed record, 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 
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
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
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,
    cross_unit_pairs: Sequence[tuple[str, 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. The
          IRI branch additionally accepts ``cross_unit_pairs``, which
          separates the two by provenance rather than by frequency.
        - ``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.
        cross_unit_pairs: Canonical (subject, predicate) pairs whose IRI
            objects came from more than one unit. When supplied, an
            IRI-branch SUSPECT_MULTI_VALUE finding on a pair *not* listed
            here is reported as a warning and never vetoes a cluster: a
            single unit asserting two objects on one predicate is reading
            one sentence, not the residue of a bad identity decision.
            None disables the distinction.
        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: a registry number
            and a full title can be two names for one key-confirmed record,
            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()
    cross_unit_object_keys = (
        set(cross_unit_pairs) if cross_unit_pairs is not None else None
    )

    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
        ):
            # Frequency says this predicate is usually single-valued, which on
            # its own is evidence of nothing: a genuinely multi-valued
            # statement is rare by construction. Provenance is what separates
            # the two cases -- only objects arriving from different units could
            # have been brought together by an identity decision.
            merge_created = (
                cross_unit_object_keys is None
                or (str(subject), str(predicate)) in cross_unit_object_keys
            )
            if merge_created:
                flagged_pairs.add((subject, predicate))
            findings.append(
                FactsValidationFinding(
                    kind=FactsValidationFindingKind.SUSPECT_MULTI_VALUE,
                    severity=(
                        "error"
                        if suspect_multi_value_severity == "error" and merge_created
                        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,
    )