ontocast.tool.facts_validation.gate

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

Merge-signature findings, SHACL conformance summary, non-catalog vocabulary telemetry, dangling references, and the gate metrics record.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

FactsValidationReport

Bases: BaseModel

Invariant findings over one aggregated facts graph.

Source code in ontocast/tool/facts_validation/gate.py

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

Functions:

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

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)

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

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)

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

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())),
    }

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

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