shifty-engine 0.5.1

//! Reference shape satisfaction `G, v ⊨ φ` and schema validation `G ⊨ S`
//! (doc 00 §3–§4, Table 2). This is the conformance *oracle*: the optimized
//! engines in later layers must agree with it.
//!
//! Two evaluators share the logic: `holds` returns a bare bool (used for
//! target selection and counting), while `explain` returns the specific
//! atomic constraints that failed, with the value node and path at which they
//! failed — enough for per-constraint reporting. The `∀π = ∃≤0 π.¬φ` encoding
//! lets a failed universal drill straight into the offending value node's inner
//! constraint.
//!
//! `ShapeEvaluator` is the deep module behind every validation-facing API. Its
//! callers provide a graph/path backend, one finalized arena, and a SPARQL
//! executor; they ask only whether a `(focus, ShapeId)` holds. The evaluator
//! centralizes path traversal, qualified counting, short-circuit Boolean logic,
//! class/value comparisons, opaque SPARQL leaves, and recursion semantics so
//! these concerns are not reimplemented in reporting, inference, or evidence.
//!
//! Completed judgments are memoized across a whole graph snapshot. The active
//! set is different: encountering an active positive recursive judgment returns
//! the coinductive provisional truth needed for the greatest fixpoint, and that
//! context-dependent answer is deliberately not memoized. This separation is
//! the crucial invariant—cache only conclusions independent of the current
//! recursion proof, while making the recursion rule explicit in one place.

use crate::focus::FocusNodes;
use crate::frozen::FrozenIndexedDataset;
use crate::path::{PathBackend, pred, succ};
use crate::profile::ShapeCacheSample;
use crate::sparql::{SparqlDiagnostic, SparqlExecutor, SparqlViolation};
use crate::value::{compare_terms, value_type_holds};
use oxrdf::{Graph, NamedNode, Term};
use regex::Regex;
use rustc_hash::{FxHashMap, FxHashSet};
use serde::{Deserialize, Serialize};
use shifty_algebra::render::{
    describe_negation_in, describe_shape_in, negated_class_target_shape, path_to_string_in,
    shape_to_string_in, xone_alternatives,
};
use shifty_algebra::{
    ConstraintKind, NodeExpr, Path, Prefixes, Schema, Selector, Severity, Shape, ShapeArena,
    ShapeId, SparqlConstraint,
};
use shifty_opt::{FocusSource, PhysicalPlan, analyze};
use std::cmp::Ordering;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::fmt;
use std::sync::OnceLock;

#[derive(Debug, Clone, Copy)]
struct EvalResult {
    holds: bool,
    cacheable: bool,
}

/// Keyed by shape first so a probe can borrow the term instead of owning it: a
/// flat `(ShapeId, Term)` key cannot be looked up without materializing the
/// tuple, so every hit paid a `Term` clone it then discarded — 75–82% of probes
/// on the Brick corpus. Hashing is `Fx` rather than SipHash because these keys
/// are shape ids and terms from an already-parsed graph, never adversarial
/// input.
///
/// Both are modest: together they are ~3–5% of evidence-materialization time
/// and 4–15% of conformance time. Key handling is *not* where evidence tracing
/// spends itself — the cost is re-deriving the same `(shape, focus)` conclusion
/// through many parents, which needs a cache of evidence values, not a cheaper
/// probe.
#[derive(Default)]
struct EvalState {
    memo: FxHashMap<ShapeId, FxHashMap<Term, bool>>,
    active: FxHashMap<ShapeId, FxHashSet<Term>>,
    telemetry: Option<ShapeCacheSample>,
}

/// Per-graph-snapshot shape evaluator.
///
/// Completed checks are shared across statements and focus nodes. Results that
/// depend on the coinductive recursion back-edge are deliberately not cached:
/// their provisional `true` may only be valid in the active call context.
pub(crate) struct ShapeEvaluator<'a> {
    g: &'a dyn PathBackend,
    arena: &'a ShapeArena,
    /// The vocabulary report messages are rendered in. Paired with `arena`
    /// because every message built here compacts IRIs from it.
    prefixes: &'a Prefixes,
    sparql: &'a SparqlExecutor,
    state: EvalState,
}

impl<'a> ShapeEvaluator<'a> {
    pub(crate) fn new(
        g: &'a dyn PathBackend,
        arena: &'a ShapeArena,
        prefixes: &'a Prefixes,
        sparql: &'a SparqlExecutor,
    ) -> Self {
        Self {
            g,
            arena,
            prefixes,
            sparql,
            state: EvalState {
                telemetry: crate::profile::is_enabled().then(ShapeCacheSample::default),
                ..EvalState::default()
            },
        }
    }

    pub(crate) fn holds(&mut self, node: &Term, id: ShapeId) -> bool {
        holds_memoized(self.g, self.arena, node, id, self.sparql, &mut self.state).holds
    }

    pub(crate) fn sparql(&self) -> &SparqlExecutor {
        self.sparql
    }

    /// The path-evaluation backend, for sibling folds (e.g. witnessing).
    pub(crate) fn backend(&self) -> &dyn PathBackend {
        self.g
    }

    /// The shape arena, for sibling folds (e.g. witnessing).
    pub(crate) fn arena(&self) -> &ShapeArena {
        self.arena
    }
}

impl Drop for ShapeEvaluator<'_> {
    fn drop(&mut self) {
        let Some(mut sample) = self.state.telemetry else {
            return;
        };
        sample.entries = self.state.memo.values().map(HashMap::len).sum();
        sample.estimated_bytes = estimated_memo_bytes(&self.state.memo);
        crate::profile::record_shape_cache(sample);
    }
}

/// A single failed atomic constraint.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Reason {
    /// The node at which the constraint failed (a value node, or the focus).
    pub value: Term,
    /// The path from the enclosing focus to `value`, if the failure is
    /// value-scoped (rendered in `π` notation).
    pub path: Option<String>,
    /// The failing constraint's arena slot (cross-references the algebra dump).
    pub shape: ShapeId,
    /// The complete algebraic constraint/operator that produced this reason.
    /// Child references are expressed as [`ShapeId`]s into the same arena, so
    /// this is lossless together with the schema/plan being validated.
    #[serde(default = "default_constraint")]
    pub constraint: Shape,
    /// Stable semantic kind of [`constraint`](Self::constraint), independent of
    /// Rust enum variant names and source-language encodings.
    #[serde(default)]
    pub constraint_kind: ConstraintKind,
    /// Stable algebra node id for the originating constraint. This aliases
    /// [`shape`](Self::shape) but names the field by its semantic role.
    #[serde(default)]
    pub constraint_id: ShapeId,
    /// Stable id of the `(selector, shape)` statement being validated.
    #[serde(default)]
    pub statement_id: usize,
    /// `sh:severity` on the source shape, defaulting to `sh:Violation`.
    #[serde(default)]
    pub severity: Severity,
    /// Engine-generated description of the failure — always present.
    pub message: String,
    /// The author's `sh:message` from the source shape, if any (with
    /// `{$this}`/`{?var}` placeholders resolved). Consumers should prefer this
    /// over [`message`](Self::message) when set; `message` remains the fallback.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub author_message: Option<String>,
    /// For a cardinality constraint, how many values along the path satisfied
    /// the qualifier. The bound this had to meet is already in
    /// [`constraint`](Self::constraint) (`∃[min..max]`); the observed count is
    /// the one number a report needs that the algebra does not carry, so a
    /// renderer can state the shortfall without parsing
    /// [`message`](Self::message). `None` for every other constraint kind.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub observed_count: Option<u64>,
    /// Non-empty when this reason is an `sh:or` group: one entry per OR branch
    /// that failed, so the caller can tell "fix any one of these."
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub sub_reasons: Vec<Reason>,
    /// Present only for a failed `sh:sparql`/custom SPARQL-based constraint
    /// component: the executed query text, its SHACL bindings, and (if
    /// natively lowered) the compiled physical plan. `None` for every other
    /// failed constraint.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sparql_diagnostic: Option<SparqlDiagnostic>,
}

fn default_constraint() -> Shape {
    Shape::Pending
}

/// One focus node that failed its statement's shape, with the reasons why.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Violation {
    pub focus: Term,
    /// Index of the violated `(selector, shape)` statement in the schema.
    pub statement: usize,
    /// The most severe top-level reason in this grouped finding.
    pub severity: Severity,
    pub reasons: Vec<Reason>,
}

/// The outcome of validating a data graph against a schema.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ValidationOutcome {
    pub conforms: bool,
    pub violations: Vec<Violation>,
}

/// How the engine treats SHACL features it only partially supports (rather than
/// silently producing a best-effort, possibly wrong, result).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum UnsupportedPolicy {
    /// Best-effort: run anyway and accept that the result may be unreliable.
    /// This preserves the historical behavior, so it is the default.
    #[default]
    Ignore,
    /// Fail loudly: refuse to evaluate the unsupported construct so the failure
    /// surfaces (e.g. as a constraint error) instead of a silent wrong answer.
    Error,
}

/// Optional, forward-looking engine configuration. New feature toggles are added
/// here as fields; both validation ([`ValidationOptions`]) and compiled
/// inference sessions accept it, so callers configure behavior in one place.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct EngineOptions {
    /// How to handle partially supported features encountered at run time —
    /// currently graph-reading `sh:SPARQLFunction` bodies called from a SPARQL
    /// context (which the engine can only evaluate as pure functions).
    pub unsupported: UnsupportedPolicy,
}

/// Controls which retained findings make a validation outcome non-conforming.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ValidationOptions {
    /// Lowest severity that makes `conforms` false. Defaults to `sh:Info`, so
    /// all findings fail validation as they did before severity was retained.
    pub minimum_severity: Severity,
    /// Whether to sort violations by severity, focus node, and statement.
    /// Defaults to `true` for deterministic output.
    pub sort_results: bool,
    /// Optional set of named shapes to use as validation entry points. When
    /// empty, every target-bearing shape is validated. Dependencies reachable
    /// from the selected shapes remain available in the arena and are evaluated
    /// normally when referenced by the selected entries.
    pub entry_shape_names: Vec<String>,
    /// Feature-handling policy (see [`EngineOptions`]).
    pub engine: EngineOptions,
}

impl Default for ValidationOptions {
    fn default() -> Self {
        Self {
            minimum_severity: Severity::Info,
            sort_results: true,
            entry_shape_names: Vec::new(),
            engine: EngineOptions::default(),
        }
    }
}

fn requested_shape_name_matches(requested: &str, actual: &str) -> bool {
    let requested = requested.trim();
    requested == actual
        || requested
            .strip_prefix('<')
            .and_then(|s| s.strip_suffix('>'))
            .is_some_and(|stripped| stripped == actual)
}

/// Returns whether a named shape should be treated as a top-level validation
/// entry under `entry_shape_names`. Empty selection means "all entries".
pub(crate) fn entry_shape_name_selected(
    entry_shape_names: &[String],
    actual_name: Option<&str>,
) -> bool {
    entry_shape_names.is_empty()
        || actual_name.is_some_and(|actual| {
            entry_shape_names
                .iter()
                .any(|requested| requested_shape_name_matches(requested, actual))
        })
}

/// Is a shape carrying *any* of `actual_names` selected?
///
/// The schema-driven form: normalization collapses several named shapes onto
/// one slot, so a slot answers to every name that reached it. Checking only one
/// of them would drop a shape the caller explicitly asked for.
pub(crate) fn entry_shape_any_name_selected(
    entry_shape_names: &[String],
    actual_names: &[String],
) -> bool {
    entry_shape_names.is_empty()
        || actual_names
            .iter()
            .any(|actual| entry_shape_name_selected(entry_shape_names, Some(actual)))
}

fn most_severe(reasons: &[Reason]) -> Severity {
    reasons
        .iter()
        .max_by_key(|reason| reason.severity.rank())
        .map(|reason| reason.severity.clone())
        .unwrap_or(Severity::Violation)
}

fn conforms_at_threshold(violations: &[Violation], minimum: &Severity) -> bool {
    !violations
        .iter()
        .flat_map(|violation| &violation.reasons)
        .any(|reason| reason.severity.meets(minimum))
}

fn sort_violations(violations: &mut [Violation], enabled: bool) {
    if enabled {
        violations.sort_by(|left, right| {
            right
                .severity
                .rank()
                .cmp(&left.severity.rank())
                .then_with(|| left.focus.to_string().cmp(&right.focus.to_string()))
                .then_with(|| left.statement.cmp(&right.statement))
        });
    }
}

fn stamp_statement_id(reasons: &mut [Reason], statement_id: usize) {
    for reason in reasons {
        reason.statement_id = statement_id;
        stamp_statement_id(&mut reason.sub_reasons, statement_id);
    }
}

/// Which RDF graph(s) validation uses for focus discovery and evaluation.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum ValidationGraphMode {
    /// Focus nodes and evaluation both use only the data graph.
    Data,
    /// Focus nodes come from data; paths, class hierarchy, and SPARQL use the
    /// union of data and shapes. This is the default for split graphs.
    #[default]
    Union,
    /// Focus discovery and evaluation both use the full data/shapes union.
    UnionAll,
}

/// The schema is not stratifiable: it recurses through genuine negation, so it
/// has no defined 2-valued semantics (`docs/03-recursion-semantics.md`). We
/// diagnose rather than guess. Carries the offending shape components.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NonStratifiable {
    pub components: Vec<Vec<ShapeId>>,
}

impl fmt::Display for NonStratifiable {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "non-stratifiable schema (recursion through negation): ")?;
        for (i, c) in self.components.iter().enumerate() {
            if i > 0 {
                write!(f, "; ")?;
            }
            let ids: Vec<String> = c.iter().map(|s| format!("@{}", s.0)).collect();
            write!(f, "{{{}}}", ids.join(" "))?;
        }
        Ok(())
    }
}

impl std::error::Error for NonStratifiable {}

impl NonStratifiable {
    /// The same message, naming shapes as the document named them.
    ///
    /// [`Display`](fmt::Display) has only arena slot ids to work with, so it
    /// prints `{@0 @1}` — meaningless outside a debugging dump, and the reader
    /// has no way to map it back to the shape they wrote. Given the schema the
    /// analysis ran over, a slot can be named: its authored IRI compacted
    /// against the document's own prefixes, else the blank node it was lowered
    /// from, else the slot id as a last resort for a shape lowering synthesized.
    pub fn describe(&self, schema: &Schema) -> String {
        let mut out = String::from("non-stratifiable schema (recursion through negation): ");
        for (i, component) in self.components.iter().enumerate() {
            if i > 0 {
                out.push_str("; ");
            }
            out.push('{');
            for (j, id) in component.iter().enumerate() {
                if j > 0 {
                    // Comma, not space: a synthesized slot's label is several
                    // words, so space-separated members run together.
                    out.push_str(", ");
                }
                out.push_str(&describe_shape_id(schema, *id));
            }
            out.push('}');
        }
        out
    }
}

fn describe_shape_id(schema: &Schema, id: ShapeId) -> String {
    if let Some(name) = schema.name_of(id) {
        return schema.prefixes.compact(name);
    }
    match schema.sources.get(&id) {
        Some(Term::BlankNode(node)) => format!("_:{}", node.as_str()),
        Some(Term::NamedNode(node)) => schema.prefixes.compact(node.as_str()),
        // A slot lowering synthesized, with no node of its own in the document.
        // Its operator is the only thing about it the reader can recognize, and
        // it is what makes the cycle: `ex:S` through `negation` back to `ex:S`
        // says where the negation is. The id stays as a suffix so a debugging
        // dump still lines up with the arena.
        _ => format!("{} @{}", constraint_kind_label(schema, id), id.0),
    }
}

fn constraint_kind_label(schema: &Schema, id: ShapeId) -> &'static str {
    match ConstraintKind::of(&schema.arena, id) {
        ConstraintKind::Top => "top",
        ConstraintKind::Constant => "constant",
        ConstraintKind::ClassMembership => "class",
        ConstraintKind::ValueType => "value type",
        ConstraintKind::NodeKind => "node kind",
        ConstraintKind::Closed => "closed",
        ConstraintKind::Equals => "equals",
        ConstraintKind::Disjoint => "disjoint",
        ConstraintKind::LessThan => "less than",
        ConstraintKind::LessThanOrEquals => "less than or equals",
        ConstraintKind::UniqueLang => "unique lang",
        ConstraintKind::Negation => "negation",
        ConstraintKind::Conjunction => "conjunction",
        ConstraintKind::Disjunction => "disjunction",
        ConstraintKind::Cardinality => "cardinality",
        ConstraintKind::Sparql => "sparql",
        ConstraintKind::Expression => "expression",
        ConstraintKind::Unknown => "shape",
    }
}

/// Validate `data` against `schema`.
///
/// Honors the decided recursion semantics (`docs/03-recursion-semantics.md`):
/// the schema must be **stratifiable** (no recursion through net negation), else
/// we return [`NonStratifiable`]. For a stratifiable schema all recursion is
/// net-positive (monotone), and `explain`/`holds`'s "assume conforming on a
/// back-edge" cycle guard computes exactly the **greatest fixpoint** — the
/// coinductive validation reading we chose.
pub fn validate(data: &Graph, schema: &Schema) -> Result<ValidationOutcome, NonStratifiable> {
    validate_with_options(data, schema, &ValidationOptions::default())
}

/// Validate `data` against `schema` using an explicit severity policy.
pub fn validate_with_options(
    data: &Graph,
    schema: &Schema,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_with_context_and_options(data, data, schema, options)
}

/// Validate split data and shapes graphs using the selected graph mode.
pub fn validate_graphs(
    data: &Graph,
    shapes: &Graph,
    schema: &Schema,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_graphs_with_mode_and_options(
        data,
        shapes,
        schema,
        ValidationGraphMode::default(),
        &ValidationOptions::default(),
    )
}

/// Validate split data and shapes graphs using an explicit graph mode.
pub fn validate_graphs_with_mode(
    data: &Graph,
    shapes: &Graph,
    schema: &Schema,
    mode: ValidationGraphMode,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_graphs_with_mode_and_options(data, shapes, schema, mode, &ValidationOptions::default())
}

/// Validate split graphs using an explicit graph mode and severity policy.
pub fn validate_graphs_with_mode_and_options(
    data: &Graph,
    shapes: &Graph,
    schema: &Schema,
    mode: ValidationGraphMode,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    match mode {
        ValidationGraphMode::Data => {
            let uses_shapes = uses_shapes_graph(&schema.arena);
            let frozen = if uses_shapes {
                FrozenIndexedDataset::from_graphs(data, shapes)
            } else {
                FrozenIndexedDataset::from_graph(data)
            };
            validate_with_frozen(data, schema, frozen, uses_shapes, options)
        }
        ValidationGraphMode::Union => {
            let uses_shapes = uses_shapes_graph(&schema.arena);
            let frozen = if uses_shapes {
                FrozenIndexedDataset::from_graph_union_with_shapes(data, shapes)
            } else {
                FrozenIndexedDataset::from_graph_union(data, shapes)
            };
            validate_with_frozen(data, schema, frozen, uses_shapes, options)
        }
        ValidationGraphMode::UnionAll => {
            let union = graph_union(data, shapes);
            validate_with_context_and_options(&union, &union, schema, options)
        }
    }
}

/// Validate focus nodes from `data` while evaluating paths, class hierarchy,
/// and SPARQL against `context`. For split data/shapes inputs, `context` should
/// be their RDF union.
pub fn validate_with_context(
    data: &Graph,
    context: &Graph,
    schema: &Schema,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_with_context_and_options(data, context, schema, &ValidationOptions::default())
}

/// Validate with separate focus/context graphs and an explicit severity policy.
pub fn validate_with_context_and_options(
    data: &Graph,
    context: &Graph,
    schema: &Schema,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    let uses_shapes = uses_shapes_graph(&schema.arena);
    let frozen = if uses_shapes {
        FrozenIndexedDataset::from_graphs(context, context)
    } else {
        FrozenIndexedDataset::from_graph(context)
    };
    validate_with_frozen(data, schema, frozen, uses_shapes, options)
}

fn validate_with_frozen(
    data: &Graph,
    schema: &Schema,
    frozen: FrozenIndexedDataset,
    has_shapes_graph: bool,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    let strat = analyze(&schema.arena);
    if !strat.stratifiable {
        let components = strat
            .strata
            .iter()
            .filter(|s| !s.stratifiable)
            .map(|s| s.shapes.clone())
            .collect();
        return Err(NonStratifiable { components });
    }

    let sparql = SparqlExecutor::from_frozen(frozen, has_shapes_graph);
    Ok(validate_with_executor(data, schema, &sparql, options))
}

pub(crate) fn validate_with_executor(
    data: &dyn FocusNodes,
    schema: &Schema,
    sparql: &SparqlExecutor,
    options: &ValidationOptions,
) -> ValidationOutcome {
    validate_with_executor_selected(data, schema, sparql, options, None)
}

pub(crate) fn validate_with_executor_selected(
    data: &dyn FocusNodes,
    schema: &Schema,
    sparql: &SparqlExecutor,
    options: &ValidationOptions,
    selected: Option<&[bool]>,
) -> ValidationOutcome {
    let entries = schema.statements.iter().enumerate().filter_map(|(i, st)| {
        if selected.is_some_and(|selected| !selected[i])
            || (selected.is_none()
                && !entry_shape_any_name_selected(
                    &options.entry_shape_names,
                    schema.names_of(st.shape),
                ))
        {
            return None;
        }
        let label = schema
            .name_of(st.shape)
            .map(str::to_string)
            .unwrap_or_else(|| format!("@{}", st.shape.0));
        Some((i, st.shape, &st.selector, label))
    });
    validate_entries(
        data,
        &schema.arena,
        &schema.prefixes,
        sparql,
        options,
        entries,
        focus_nodes_with_evaluator,
    )
}

fn validate_entries<'a, S: 'a>(
    data: &dyn FocusNodes,
    arena: &ShapeArena,
    prefixes: &Prefixes,
    sparql: &SparqlExecutor,
    options: &ValidationOptions,
    entries: impl IntoIterator<Item = (usize, ShapeId, &'a S, String)>,
    focus: impl Fn(&dyn FocusNodes, &S, &mut ShapeEvaluator<'_>) -> Vec<Term>,
) -> ValidationOutcome {
    let backend = sparql
        .frozen()
        .expect("validation executor always has a frozen dataset");
    let mut evaluator = ShapeEvaluator::new(backend, arena, prefixes, sparql);
    let mut violations = Vec::new();
    for (statement, shape, source, label) in entries {
        let foci = focus(data, source, &mut evaluator);
        prefetch_sparql_constraints(arena, shape, &foci, sparql);
        for focus in foci {
            let start = web_time::Instant::now();
            let mut stack = HashSet::new();
            let mut reasons = explain(
                &mut evaluator,
                &focus,
                shape,
                None,
                &Severity::Violation,
                &mut stack,
            );
            crate::profile::record_shape(&label, start.elapsed().as_micros() as u64);
            dedup_reasons(&mut reasons);
            stamp_statement_id(&mut reasons, statement);
            if !reasons.is_empty() {
                let severity = most_severe(&reasons);
                violations.push(Violation {
                    focus,
                    statement,
                    severity,
                    reasons,
                });
            }
        }
    }
    sort_violations(&mut violations, options.sort_results);
    ValidationOutcome {
        conforms: conforms_at_threshold(&violations, &options.minimum_severity),
        violations,
    }
}

/// Whether any `sh:sparql` constraint references `$shapesGraph`, requiring the
/// shapes graph to be mirrored into a named graph for evaluation.
pub(crate) fn uses_shapes_graph(arena: &ShapeArena) -> bool {
    (0..arena.len()).any(|i| {
        matches!(arena.get(ShapeId(i as u32)), Shape::Sparql(c) if c.query.contains("shapesGraph"))
    })
}

/// The RDF merge of two graphs (`left ∪ right`). The standard way to build the
/// `context` graph the `*_with_context` repair entry points expect: the union of
/// a data graph and a shapes/ontology graph.
///
/// Cloning a graph copies its indexes wholesale, while inserting builds them a
/// triple at a time — so the larger side is always the one to clone, and only
/// the smaller side is inserted. Set union is commutative, so the result is
/// identical either way. Callers pass `(data, shapes)`, and a shapes closure
/// routinely dwarfs the data graph (228k triples against 16 for a small Brick
/// model), which made the naive direction quadratically the wrong choice.
pub fn graph_union(left: &Graph, right: &Graph) -> Graph {
    let (base, extra) = if left.len() >= right.len() {
        (left, right)
    } else {
        (right, left)
    };
    let mut union = base.clone();
    for triple in extra.iter() {
        union.insert(triple);
    }
    crate::profile::record_graph_union(union.len());
    union
}

/// Validate using a [`PhysicalPlan`] (Layer 5): focus nodes come from compiled
/// [`FocusSource`]s (so class targets seed backward from the constant instead of
/// scanning every node) and checks run over the plan's cost-ordered arena. The
/// result is identical to [`validate`] on the same schema — the W3C harness
/// cross-checks this.
pub fn validate_plan(
    data: &Graph,
    plan: &PhysicalPlan,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_plan_with_options(data, plan, &ValidationOptions::default())
}

/// Validate a physical plan using an explicit severity policy.
pub fn validate_plan_with_options(
    data: &Graph,
    plan: &PhysicalPlan,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_plan_with_context_and_options(data, data, plan, options)
}

/// Validate a physical plan over split graphs using the default graph mode.
pub fn validate_plan_graphs(
    data: &Graph,
    shapes: &Graph,
    plan: &PhysicalPlan,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_plan_graphs_with_mode_and_options(
        data,
        shapes,
        plan,
        ValidationGraphMode::default(),
        &ValidationOptions::default(),
    )
}

/// Validate a physical plan over split graphs using an explicit graph mode.
pub fn validate_plan_graphs_with_mode(
    data: &Graph,
    shapes: &Graph,
    plan: &PhysicalPlan,
    mode: ValidationGraphMode,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_plan_graphs_with_mode_and_options(
        data,
        shapes,
        plan,
        mode,
        &ValidationOptions::default(),
    )
}

/// Validate a physical plan over split graphs with an explicit severity policy.
pub fn validate_plan_graphs_with_mode_and_options(
    data: &Graph,
    shapes: &Graph,
    plan: &PhysicalPlan,
    mode: ValidationGraphMode,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    match mode {
        ValidationGraphMode::Data => {
            let uses_shapes = uses_shapes_graph(&plan.arena);
            let frozen = if uses_shapes {
                FrozenIndexedDataset::from_graphs(data, shapes)
            } else {
                FrozenIndexedDataset::from_graph(data)
            };
            validate_plan_with_frozen(data, plan, frozen, uses_shapes, options)
        }
        ValidationGraphMode::Union => {
            let uses_shapes = uses_shapes_graph(&plan.arena);
            let frozen = if uses_shapes {
                FrozenIndexedDataset::from_graph_union_with_shapes(data, shapes)
            } else {
                FrozenIndexedDataset::from_graph_union(data, shapes)
            };
            validate_plan_with_frozen(data, plan, frozen, uses_shapes, options)
        }
        ValidationGraphMode::UnionAll => {
            let union = graph_union(data, shapes);
            validate_plan_with_context_and_options(&union, &union, plan, options)
        }
    }
}

/// Validate plan focus nodes from `data` against the supplied execution context.
pub fn validate_plan_with_context(
    data: &Graph,
    context: &Graph,
    plan: &PhysicalPlan,
) -> Result<ValidationOutcome, NonStratifiable> {
    validate_plan_with_context_and_options(data, context, plan, &ValidationOptions::default())
}

/// Validate plan focus nodes against a context with a severity policy.
pub fn validate_plan_with_context_and_options(
    data: &Graph,
    context: &Graph,
    plan: &PhysicalPlan,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    let uses_shapes = uses_shapes_graph(&plan.arena);
    let frozen = if uses_shapes {
        FrozenIndexedDataset::from_graphs(context, context)
    } else {
        FrozenIndexedDataset::from_graph(context)
    };
    validate_plan_with_frozen(data, plan, frozen, uses_shapes, options)
}

fn validate_plan_with_frozen(
    data: &Graph,
    plan: &PhysicalPlan,
    frozen: FrozenIndexedDataset,
    has_shapes_graph: bool,
    options: &ValidationOptions,
) -> Result<ValidationOutcome, NonStratifiable> {
    let strat = analyze(&plan.arena);
    if !strat.stratifiable {
        let components = strat
            .strata
            .iter()
            .filter(|s| !s.stratifiable)
            .map(|s| s.shapes.clone())
            .collect();
        return Err(NonStratifiable { components });
    }

    let sparql = SparqlExecutor::from_frozen(frozen, has_shapes_graph);
    Ok(validate_plan_with_executor(
        data, plan, &sparql, options, None,
    ))
}

pub(crate) fn validate_plan_with_executor(
    data: &dyn FocusNodes,
    plan: &PhysicalPlan,
    sparql: &SparqlExecutor,
    options: &ValidationOptions,
    selected: Option<&[bool]>,
) -> ValidationOutcome {
    let entries = plan.statements.iter().enumerate().filter_map(|(i, sp)| {
        if selected.is_some_and(|selected| !selected[i])
            || (selected.is_none()
                && !entry_shape_any_name_selected(
                    &options.entry_shape_names,
                    plan.names_of(sp.shape),
                ))
        {
            return None;
        }
        let label = plan
            .name_of(sp.shape)
            .map(str::to_string)
            .unwrap_or_else(|| format!("@{}", sp.shape.0));
        Some((i, sp.shape, &sp.source, label))
    });
    validate_entries(
        data,
        &plan.arena,
        &plan.prefixes,
        sparql,
        options,
        entries,
        focus_for_source,
    )
}

/// Focus nodes for a compiled [`FocusSource`].
fn focus_for_source(
    data: &dyn FocusNodes,
    source: &FocusSource,
    evaluator: &mut ShapeEvaluator<'_>,
) -> Vec<Term> {
    match source {
        FocusSource::SubjectsOf(p) => data.subjects_of(p),
        FocusSource::ObjectsOf(p) => data.objects_of(p),
        FocusSource::Node(c) => vec![c.clone()],
        // the optimization: seed backward from the constant, no full scan
        FocusSource::PathToConst { path, target } => pred(evaluator.g, target, path)
            .into_iter()
            .filter(|node| data.contains_term(node))
            .collect(),
        FocusSource::ScanFilter { path, qualifier } => data
            .all_nodes()
            .into_iter()
            .filter(|v| {
                succ(evaluator.g, v, path)
                    .iter()
                    .any(|u| evaluator.holds(u, *qualifier))
            })
            .collect(),
        FocusSource::Sparql(target) => {
            let candidates = data.all_nodes();
            evaluator
                .sparql
                .target_nodes(&target.query)
                .unwrap_or_default()
                .into_iter()
                .filter(|node| candidates.contains(node))
                .collect()
        }
    }
}

/// The focus nodes selected by a selector.
pub fn focus_nodes(data: &Graph, sel: &Selector, arena: &ShapeArena) -> Vec<Term> {
    let sparql =
        SparqlExecutor::new(data).expect("building an in-memory Oxigraph store should succeed");
    let mut evaluator = ShapeEvaluator::new(data, arena, Prefixes::empty(), &sparql);
    focus_nodes_with_evaluator(data, sel, &mut evaluator)
}

pub(crate) fn focus_nodes_with(
    data: &dyn FocusNodes,
    backend: &dyn PathBackend,
    sel: &Selector,
    arena: &ShapeArena,
    sparql: &SparqlExecutor,
) -> Vec<Term> {
    let mut evaluator = ShapeEvaluator::new(backend, arena, Prefixes::empty(), sparql);
    focus_nodes_with_evaluator(data, sel, &mut evaluator)
}

pub(crate) fn focus_nodes_with_evaluator(
    data: &dyn FocusNodes,
    sel: &Selector,
    evaluator: &mut ShapeEvaluator<'_>,
) -> Vec<Term> {
    match sel {
        Selector::HasOut(q) => data.subjects_of(q),
        Selector::HasIn(q) => data.objects_of(q),
        Selector::IsConst(c) => vec![c.clone()],
        Selector::HasPath(path, qual) => match evaluator.arena.get(*qual) {
            // Class targets are lowered to
            // rdf:type/rdfs:subClassOf* ending at a constant. Searching
            // backward from that constant avoids traversing the hierarchy once
            // for every node in the data graph.
            Shape::TestConst(target) => pred(evaluator.g, target, path)
                .into_iter()
                .filter(|node| data.contains_term(node))
                .collect(),
            _ => data
                .all_nodes()
                .into_iter()
                .filter(|v| {
                    succ(evaluator.g, v, path)
                        .iter()
                        .any(|u| evaluator.holds(u, *qual))
                })
                .collect(),
        },
        Selector::Sparql(target) => {
            let candidates = data.all_nodes();
            evaluator
                .sparql
                .target_nodes(&target.query)
                .unwrap_or_default()
                .into_iter()
                .filter(|node| candidates.contains(node))
                .collect()
        }
    }
}

fn holds_memoized(
    g: &dyn PathBackend,
    arena: &ShapeArena,
    v: &Term,
    id: ShapeId,
    sparql: &SparqlExecutor,
    state: &mut EvalState,
) -> EvalResult {
    if let Some(&holds) = state.memo.get(&id).and_then(|by_term| by_term.get(v)) {
        if let Some(telemetry) = state.telemetry.as_mut() {
            telemetry.hits += 1;
        }
        return EvalResult {
            holds,
            cacheable: true,
        };
    }
    if let Some(telemetry) = state.telemetry.as_mut() {
        telemetry.misses += 1;
    }
    if !state.active.entry(id).or_default().insert(v.clone()) {
        if let Some(telemetry) = state.telemetry.as_mut() {
            telemetry.recursion_back_edges += 1;
        }
        return EvalResult {
            holds: true,
            cacheable: false,
        }; // back-edge ⇒ assume conforming: the gfp choice (doc 03)
    }
    let result = match arena.get(id) {
        Shape::Annotated { shape, .. } => holds_memoized(g, arena, v, *shape, sparql, state),
        Shape::Top | Shape::Pending => EvalResult {
            holds: true,
            cacheable: true,
        },
        Shape::Sparql(constraint) => EvalResult {
            holds: sparql
                .constraint_violations(constraint, v)
                .is_ok_and(|violations| violations.is_empty()),
            cacheable: true,
        },
        Shape::Expression(expr) => {
            // SHACL-AF §5: conform iff every value the expression produces (with
            // `v` as `?this`) is the boolean `true`.
            let mut cacheable = true;
            let results = eval_expr(g, arena, v, expr, sparql, state, &mut cacheable);
            EvalResult {
                holds: results.iter().all(is_boolean_true),
                cacheable,
            }
        }
        Shape::TestConst(c) => EvalResult {
            holds: v == c,
            cacheable: true,
        },
        Shape::TestType(t) => EvalResult {
            holds: value_type_holds(t, v),
            cacheable: true,
        },
        Shape::TestKind(k) => EvalResult {
            holds: k.matches(v),
            cacheable: true,
        },
        Shape::Closed(q) => EvalResult {
            holds: closed_offenders(g, v, q).is_empty(),
            cacheable: true,
        },
        Shape::Eq(path, p) => EvalResult {
            holds: succ(g, v, path) == objects(g, v, p),
            cacheable: true,
        },
        Shape::Disj(path, p) => EvalResult {
            holds: succ(g, v, path).is_disjoint(&objects(g, v, p)),
            cacheable: true,
        },
        Shape::Lt(path, p) => EvalResult {
            holds: all_pairs_ordered(g, v, path, p, false),
            cacheable: true,
        },
        Shape::Le(path, p) => EvalResult {
            holds: all_pairs_ordered(g, v, path, p, true),
            cacheable: true,
        },
        Shape::UniqueLang(path) => EvalResult {
            holds: unique_lang(&succ(g, v, path)),
            cacheable: true,
        },
        Shape::Not(c) => {
            let child = holds_memoized(g, arena, v, *c, sparql, state);
            EvalResult {
                holds: !child.holds,
                cacheable: child.cacheable,
            }
        }
        Shape::And(cs) => {
            let mut result = EvalResult {
                holds: true,
                cacheable: true,
            };
            for child in cs {
                let child = holds_memoized(g, arena, v, *child, sparql, state);
                result.cacheable &= child.cacheable;
                if !child.holds {
                    result.holds = false;
                    break;
                }
            }
            result
        }
        Shape::Or(cs) => {
            let mut result = EvalResult {
                holds: false,
                cacheable: true,
            };
            for child in cs {
                let child = holds_memoized(g, arena, v, *child, sparql, state);
                result.cacheable &= child.cacheable;
                if child.holds {
                    result.holds = true;
                    break;
                }
            }
            result
        }
        Shape::Count {
            path,
            min,
            max,
            qualifier,
        } => {
            let mut n = 0;
            let mut cacheable = true;
            for value in succ(g, v, path) {
                let qualified = holds_memoized(g, arena, &value, *qualifier, sparql, state);
                cacheable &= qualified.cacheable;
                n += u64::from(qualified.holds);
            }
            EvalResult {
                holds: min.is_none_or(|m| n >= m) && max.is_none_or(|m| n <= m),
                cacheable,
            }
        }
    };
    // Reclaim the term the active set owns rather than cloning `v` again: the
    // miss path then allocates once, not twice.
    let owned = state.active.get_mut(&id).and_then(|set| set.take(v));
    if result.cacheable {
        let key = owned.unwrap_or_else(|| v.clone());
        state.memo.entry(id).or_default().insert(key, result.holds);
        if let Some(telemetry) = state.telemetry.as_mut() {
            telemetry.insertions += 1;
        }
    } else if let Some(telemetry) = state.telemetry.as_mut() {
        telemetry.non_cacheable_results += 1;
    }
    result
}

/// Evaluate a SHACL-AF node expression at focus `v` to its set of result terms,
/// over the same low-level primitives as [`holds_memoized`]. `cacheable` is
/// cleared if any nested shape evaluation observed a recursion back-edge, so the
/// enclosing `Shape::Expression` result is not memoized on a provisional truth.
/// `Function` applications are refused at parse time, so they never appear here.
fn eval_expr(
    g: &dyn PathBackend,
    arena: &ShapeArena,
    v: &Term,
    expr: &NodeExpr,
    sparql: &SparqlExecutor,
    state: &mut EvalState,
    cacheable: &mut bool,
) -> HashSet<Term> {
    match expr {
        NodeExpr::This => {
            let mut s = HashSet::with_capacity(1);
            s.insert(v.clone());
            s
        }
        NodeExpr::Constant(t) => {
            let mut s = HashSet::with_capacity(1);
            s.insert(t.clone());
            s
        }
        NodeExpr::Path(p) => succ(g, v, p),
        NodeExpr::Filter { input, shape } => {
            let inputs = eval_expr(g, arena, v, input, sparql, state, cacheable);
            inputs
                .into_iter()
                .filter(|x| {
                    let r = holds_memoized(g, arena, x, *shape, sparql, state);
                    *cacheable &= r.cacheable;
                    r.holds
                })
                .collect()
        }
        NodeExpr::Intersection(es) => {
            let mut iter = es.iter();
            match iter.next() {
                Some(first) => {
                    let mut acc = eval_expr(g, arena, v, first, sparql, state, cacheable);
                    for e in iter {
                        let s = eval_expr(g, arena, v, e, sparql, state, cacheable);
                        acc.retain(|x| s.contains(x));
                    }
                    acc
                }
                None => HashSet::new(),
            }
        }
        NodeExpr::Union(es) => {
            let mut acc = HashSet::new();
            for e in es {
                acc.extend(eval_expr(g, arena, v, e, sparql, state, cacheable));
            }
            acc
        }
        NodeExpr::Function { .. } => HashSet::new(),
    }
}

/// The boolean literal `true` (`"true"^^xsd:boolean`).
pub(crate) fn is_boolean_true(t: &Term) -> bool {
    matches!(t, Term::Literal(l) if l.datatype() == oxrdf::vocab::xsd::BOOLEAN && l.value() == "true")
}

fn estimated_memo_bytes(memo: &FxHashMap<ShapeId, FxHashMap<Term, bool>>) -> usize {
    const CONTROL_BYTE_ESTIMATE: usize = 1;
    const ENTRY_BYTES: usize = std::mem::size_of::<(Term, bool)>() + CONTROL_BYTE_ESTIMATE;
    memo.values()
        .map(|by_term| {
            by_term.capacity() * ENTRY_BYTES
                + by_term.keys().map(estimated_term_heap_bytes).sum::<usize>()
        })
        .sum::<usize>()
        + memo.capacity()
            * (std::mem::size_of::<(ShapeId, FxHashMap<Term, bool>)>() + CONTROL_BYTE_ESTIMATE)
}

fn estimated_term_heap_bytes(term: &Term) -> usize {
    match term {
        Term::NamedNode(node) => node.as_str().len(),
        Term::BlankNode(node) => node.as_str().len(),
        Term::Literal(literal) => {
            literal.value().len()
                + literal.language().map_or_else(
                    || {
                        let datatype = literal.datatype();
                        if datatype.as_str() == "http://www.w3.org/2001/XMLSchema#string" {
                            0
                        } else {
                            datatype.as_str().len()
                        }
                    },
                    str::len,
                )
        }
    }
}

/// Batch-evaluate the SPARQL constraints reachable from `root` at the focus set
/// before the per-node walk, so their fallback queries run once for the whole
/// focus set instead of once per focus (doc §189). Only constraints reached through
/// focus-preserving operators (`∧`/`∨`/`¬`) are evaluated at `foci`; constraints
/// under a `Count` path apply to value nodes, not the statement focus, so they
/// are skipped here and fall back to per-focus execution. Prefetching is a pure
/// memo (constraint violations depend only on focus + immutable dataset), so it
/// is sound regardless of the operator context the constraint is reached in.
pub(crate) fn prefetch_sparql_constraints(
    arena: &ShapeArena,
    root: ShapeId,
    foci: &[Term],
    sparql: &SparqlExecutor,
) {
    if foci.len() < 2 {
        return;
    }
    let mut constraints = Vec::new();
    let mut seen = HashSet::new();
    collect_focus_sparql(arena, root, &mut seen, &mut constraints);
    for constraint in constraints {
        let _ = sparql.prefetch_constraint(constraint, foci);
    }
}

fn collect_focus_sparql<'a>(
    arena: &'a ShapeArena,
    id: ShapeId,
    seen: &mut HashSet<ShapeId>,
    out: &mut Vec<&'a SparqlConstraint>,
) {
    if !seen.insert(id) {
        return; // cyclic (recursive) shape: stop at the back-edge
    }
    match arena.get(id) {
        Shape::Annotated { shape, .. } => collect_focus_sparql(arena, *shape, seen, out),
        Shape::Sparql(constraint) => out.push(constraint),
        Shape::Not(inner) => collect_focus_sparql(arena, *inner, seen, out),
        Shape::And(ids) | Shape::Or(ids) => {
            for &child in ids {
                collect_focus_sparql(arena, child, seen, out);
            }
        }
        // `Count` crosses a path (different focus); all other variants are leaves.
        _ => {}
    }
}

/// The reasons `φ` (slot `id`) fails at `node`. Empty iff it holds. `path_ctx`
/// is the rendered path by which `node` was reached from the enclosing focus.
fn explain(
    evaluator: &mut ShapeEvaluator<'_>,
    node: &Term,
    id: ShapeId,
    path_ctx: Option<&str>,
    severity: &Severity,
    stack: &mut HashSet<(ShapeId, Term)>,
) -> Vec<Reason> {
    let key = (id, node.clone());
    if !stack.insert(key.clone()) {
        return Vec::new(); // back-edge ⇒ assume conforming (gfp, doc 03)
    }
    if evaluator.holds(node, id) {
        stack.remove(&key);
        return Vec::new();
    }
    let reasons = match evaluator.arena.get(id).clone() {
        Shape::Annotated {
            severity: source_severity,
            messages,
            shape,
        } => {
            let mut reasons = explain(evaluator, node, shape, path_ctx, &source_severity, stack);
            if !messages.is_empty() {
                // Resolve the author's `sh:message` once at this source-shape
                // boundary (`$this` = the node this shape validates) and stamp it
                // onto any reason that doesn't already carry a nearer one. Since
                // the deepest `Annotated` returns first, the innermost (most
                // specific) message wins — outer shapes only fill the gaps.
                let author = messages
                    .iter()
                    .map(|m| apply_message_template(&term_text(m), node, &HashMap::new()))
                    .collect::<Vec<_>>()
                    .join("; ");
                for r in &mut reasons {
                    if r.author_message.is_none() {
                        r.author_message = Some(author.clone());
                    }
                }
            }
            reasons
        }
        Shape::Top | Shape::Pending => Vec::new(),
        Shape::Sparql(constraint) => {
            match evaluator.sparql.constraint_violations(&constraint, node) {
                Ok(violations) if violations.is_empty() => Vec::new(),
                Ok(violations) => {
                    // Same (constraint, node) for every reason below, so build the
                    // diagnostic once rather than per violation; `.ok()` because a
                    // second, redundant compile can't fail once the call above
                    // already succeeded.
                    let diagnostic = evaluator
                        .sparql
                        .constraint_diagnostic(&constraint, node, &violations)
                        .ok();
                    violations
                        .into_iter()
                        .map(|violation| {
                            // Compute the message before the value/path fields are
                            // moved out of `violation`.
                            let message = sparql_violation_message(&violation, &constraint, node);
                            reason(
                                evaluator.arena,
                                id,
                                violation.value.unwrap_or_else(|| node.clone()),
                                violation
                                    .path
                                    .map(|path| path.to_string())
                                    .or_else(|| path_ctx.map(str::to_string))
                                    .or_else(|| {
                                        constraint
                                            .path
                                            .as_ref()
                                            .map(|p| path_to_string_in(p, evaluator.prefixes))
                                    }),
                                severity,
                                message,
                                None,
                                Vec::new(),
                                diagnostic.clone(),
                            )
                        })
                        .collect()
                }
                Err(error) => vec![reason(
                    evaluator.arena,
                    id,
                    node.clone(),
                    path_ctx.map(str::to_string),
                    severity,
                    format!("SPARQL constraint evaluation failed: {error}"),
                    None,
                    Vec::new(),
                    evaluator
                        .sparql
                        .constraint_diagnostic(&constraint, node, &[])
                        .ok(),
                )],
            }
        }
        Shape::TestConst(_)
        | Shape::TestType(_)
        | Shape::TestKind(_)
        | Shape::Eq(..)
        | Shape::Disj(..)
        | Shape::Lt(..)
        | Shape::Le(..)
        | Shape::UniqueLang(_) => leaf(
            evaluator.arena,
            evaluator.holds(node, id),
            node,
            id,
            path_ctx,
            severity,
            format!(
                "{} not satisfied",
                shape_to_string_in(evaluator.arena, id, evaluator.prefixes)
            ),
        ),
        Shape::Closed(q) => {
            let bad = closed_offenders(evaluator.g, node, &q);
            if bad.is_empty() {
                Vec::new()
            } else {
                // The offending predicates are IRIs like any others in a
                // message, so they compact against the document's vocabulary.
                let preds: Vec<String> = bad
                    .iter()
                    .map(|p| evaluator.prefixes.compact(p.as_str()))
                    .collect();
                vec![reason(
                    evaluator.arena,
                    id,
                    node.clone(),
                    path_ctx.map(str::to_string),
                    severity,
                    format!("closed: unexpected predicate(s) {}", preds.join(", ")),
                    None,
                    Vec::new(),
                    None,
                )]
            }
        }
        Shape::Not(c) => {
            if explain(evaluator, node, c, path_ctx, severity, stack).is_empty() {
                vec![reason(
                    evaluator.arena,
                    id,
                    node.clone(),
                    path_ctx.map(str::to_string),
                    severity,
                    format!(
                        "must satisfy `{}`",
                        describe_negation_in(evaluator.arena, c, evaluator.prefixes)
                    ),
                    None,
                    Vec::new(),
                    None,
                )]
            } else {
                Vec::new()
            }
        }
        Shape::And(cs) => cs
            .iter()
            .flat_map(|c| explain(evaluator, node, *c, path_ctx, severity, stack))
            .collect(),
        Shape::Or(cs) => {
            let mut sub_reasons = Vec::new();
            let mut satisfied = false;
            for c in &cs {
                let sub = explain(evaluator, node, *c, path_ctx, severity, stack);
                if sub.is_empty() {
                    satisfied = true;
                    break;
                }
                sub_reasons.extend(sub);
            }
            if satisfied {
                return Vec::new();
            }
            // `sh:xone` is lowered to `⋁ᵢ (φᵢ ∧ ⋀_{j≠i} ¬φⱼ)`. Reported as the
            // plain disjunction it is, a node satisfying *two* alternatives —
            // the usual way to fail a xone — gets told that none were satisfied,
            // which is the opposite of the finding. Count what actually holds.
            if let Some(alternatives) = xone_alternatives(id, evaluator.arena) {
                let held: Vec<ShapeId> = alternatives
                    .iter()
                    .copied()
                    .filter(|a| evaluator.holds(node, *a))
                    .collect();
                let total = alternatives.len();
                // Naming the ones that hold is the whole of the fix: the reader
                // has to drop all but one of them, and "2 of 3" does not say
                // which two.
                let describe_all = |ids: &[ShapeId]| {
                    ids.iter()
                        .map(|a| describe_shape_in(evaluator.arena, *a, evaluator.prefixes))
                        .collect::<Vec<_>>()
                        .join(", ")
                };
                let message = if held.is_empty() {
                    format!("none of the {total} alternatives hold; exactly one must")
                } else {
                    // Both sides: which to drop, and which is left to keep.
                    let idle: Vec<ShapeId> = alternatives
                        .iter()
                        .copied()
                        .filter(|a| !held.contains(a))
                        .collect();
                    let mut message = format!(
                        "exactly one alternative may hold; {} of {total} do — holds: {}",
                        held.len(),
                        describe_all(&held)
                    );
                    if !idle.is_empty() {
                        message.push_str(&format!("; does not hold: {}", describe_all(&idle)));
                    }
                    message
                };
                // The branch sub-reasons explain the rewrite, not the shape the
                // author wrote, and "fix any one of these" is the wrong advice
                // for a node that already satisfies too many.
                let sub_reasons = if held.is_empty() {
                    sub_reasons
                } else {
                    Vec::new()
                };
                return vec![reason(
                    evaluator.arena,
                    id,
                    node.clone(),
                    path_ctx.map(str::to_string),
                    severity,
                    message,
                    None,
                    sub_reasons,
                    None,
                )];
            }
            vec![reason(
                evaluator.arena,
                id,
                node.clone(),
                path_ctx.map(str::to_string),
                severity,
                format!("none of {} alternative(s) satisfied", cs.len()),
                None,
                sub_reasons,
                None,
            )]
        }
        Shape::Count {
            path,
            min,
            max,
            qualifier,
        } => explain_count(
            evaluator, node, id, &path, min, max, qualifier, severity, stack,
        ),
        Shape::Expression(_) => leaf(
            evaluator.arena,
            false, // reached only because the constraint failed (line ~923)
            node,
            id,
            path_ctx,
            severity,
            "sh:expression did not evaluate to true".to_string(),
        ),
    };
    stack.remove(&key);
    reasons
}

#[allow(clippy::too_many_arguments)]
fn explain_count(
    evaluator: &mut ShapeEvaluator<'_>,
    node: &Term,
    id: ShapeId,
    path: &Path,
    min: Option<u64>,
    max: Option<u64>,
    qualifier: ShapeId,
    severity: &Severity,
    stack: &mut HashSet<(ShapeId, Term)>,
) -> Vec<Reason> {
    let path_str = path_to_string_in(path, evaluator.prefixes);
    let matched: Vec<Term> = succ(evaluator.g, node, path)
        .into_iter()
        .filter(|u| evaluator.holds(u, qualifier))
        .collect();
    let n = matched.len() as u64;
    let mut reasons = Vec::new();

    // For a *qualified* count (`sh:qualifiedValueShape`, e.g. `sh:class C`), the
    // counted values are only those conforming to the qualifier, so the message
    // must name it — otherwise "found 0" reads as if the path were empty when in
    // fact it held values that just didn't match the qualifier. Plain
    // `sh:minCount`/`sh:maxCount` lower with a `⊤` qualifier and need no clause.
    let qual_clause = match evaluator.arena.get(qualifier) {
        Shape::Top => String::new(),
        _ => format!(
            " matching `{}`",
            describe_shape_in(evaluator.arena, qualifier, evaluator.prefixes)
        ),
    };

    if let Some(mx) = max
        && n > mx
    {
        // A universal `∀path.φ` is lowered to `∃≤0 path.¬φ`; more generally, when
        // a `∃≤0` (max 0) count over a non-trivial qualifier `ψ` fails, every
        // offending value satisfies `ψ` but must satisfy `¬ψ`. Drill into each
        // offender and report that positive requirement `¬ψ` rather than echoing
        // the machine's double-negated `ψ`. A bare `Shape::Not` still routes
        // through `explain` (richest, keeps sub-reasons); the `sh:class` case
        // gets a dedicated phrasing; anything else (`sh:nodeKind`, De Morgan
        // combinations, …) is described by `describe_negation`. A `⊤` qualifier
        // is a plain `sh:maxCount` and keeps the concise count message.
        let class_offense = (mx == 0)
            .then(|| negated_class_target_shape(qualifier, evaluator.arena))
            .flatten();
        match evaluator.arena.get(qualifier).clone() {
            Shape::Not(inner) if mx == 0 => {
                for u in &matched {
                    reasons.extend(explain(
                        evaluator,
                        u,
                        inner,
                        Some(&path_str),
                        severity,
                        stack,
                    ));
                }
            }
            // Normalized `sh:class` universal: name the class each offending value
            // must be an instance of. (The value node and path are surfaced
            // alongside the message by the report renderer.)
            _ if class_offense.is_some() => {
                let class = class_offense.expect("guard ensured Some");
                for u in &matched {
                    reasons.push(reason(
                        evaluator.arena,
                        id,
                        u.clone(),
                        Some(path_str.clone()),
                        severity,
                        format!(
                            "must be an instance of {}",
                            term_display(&class, evaluator.prefixes)
                        ),
                        None,
                        Vec::new(),
                        None,
                    ));
                }
            }
            // Plain `sh:maxCount` (⊤ qualifier): concise count message.
            Shape::Top => reasons.push(with_observed(
                reason(
                    evaluator.arena,
                    id,
                    node.clone(),
                    Some(path_str.clone()),
                    severity,
                    format!(
                        "at most {mx} value(s){qual_clause} allowed along {path_str}, found {n}"
                    ),
                    None,
                    Vec::new(),
                    None,
                ),
                n,
            )),
            // Any other `∃≤0` qualifier (`sh:nodeKind`, several value constraints
            // De-Morgan'd to an `Or`, …): describe the positive requirement.
            _ if mx == 0 => {
                let requirement =
                    describe_negation_in(evaluator.arena, qualifier, evaluator.prefixes);
                for u in &matched {
                    reasons.push(reason(
                        evaluator.arena,
                        id,
                        u.clone(),
                        Some(path_str.clone()),
                        severity,
                        format!("must satisfy `{requirement}`"),
                        None,
                        Vec::new(),
                        None,
                    ));
                }
            }
            // Genuine `sh:qualifiedMaxCount` ≥ 1: concise count message.
            _ => reasons.push(with_observed(
                reason(
                    evaluator.arena,
                    id,
                    node.clone(),
                    Some(path_str.clone()),
                    severity,
                    format!(
                        "at most {mx} value(s){qual_clause} allowed along {path_str}, found {n}"
                    ),
                    None,
                    Vec::new(),
                    None,
                ),
                n,
            )),
        }
    }

    if let Some(mn) = min
        && n < mn
    {
        reasons.push(with_observed(
            reason(
                evaluator.arena,
                id,
                node.clone(),
                Some(path_str.clone()),
                severity,
                format!("at least {mn} value(s){qual_clause} required along {path_str}, found {n}"),
                None,
                Vec::new(),
                None,
            ),
            n,
        ));
    }

    reasons
}

/// The message for a `sh:sparql` violation, by SHACL §5.2.1 precedence: the
/// result's own `?message` binding, then the constraint's (or shape's)
/// `sh:message` (with `{$this}`/`{?var}` substitution), then a constructed
/// description naming the shape and value.
fn sparql_violation_message(
    violation: &SparqlViolation,
    constraint: &SparqlConstraint,
    node: &Term,
) -> String {
    if let Some(message) = &violation.message {
        return term_text(message);
    }
    if !constraint.messages.is_empty() {
        return constraint
            .messages
            .iter()
            .map(|m| apply_message_template(&term_text(m), node, &violation.bindings))
            .collect::<Vec<_>>()
            .join("; ");
    }
    let mut message = match &constraint.shape {
        Some(shape) => format!("SPARQL constraint at {shape} not satisfied"),
        None => "SPARQL constraint not satisfied".to_string(),
    };
    if let Some(value) = &violation.value {
        message.push_str(&format!(" (value: {value})"));
    }
    message
}

/// Substitute `{$varName}` / `{?varName}` placeholders in a message template.
///
/// `$this` resolves to `focus`; all other names are looked up in `bindings`
/// (keyed without the `$`/`?` sigil). Unresolved placeholders are left as-is.
pub(crate) fn apply_message_template(
    template: &str,
    focus: &Term,
    bindings: &HashMap<String, Term>,
) -> String {
    static RE: OnceLock<Regex> = OnceLock::new();
    let re = RE
        .get_or_init(|| Regex::new(r"\{(\$[A-Za-z_]\w*|\?[A-Za-z_]\w*)\}").expect("static regex"));
    re.replace_all(template, |caps: &regex::Captures| {
        let placeholder = &caps[1];
        let name = &placeholder[1..]; // strip leading `$` or `?`
        let term = if name == "this" {
            Some(focus)
        } else {
            bindings.get(name)
        };
        term.map(|t| match t {
            Term::NamedNode(n) => format!("<{}>", n.as_str()),
            Term::BlankNode(b) => format!("_:{}", b.as_str()),
            Term::Literal(l) => l.value().to_string(),
        })
        .unwrap_or_else(|| placeholder.to_string())
    })
    .to_string()
}

/// A term's human-facing text: a literal's lexical value, otherwise its RDF
/// rendering (`<iri>` / `_:id`).
fn term_text(term: &Term) -> String {
    match term {
        Term::Literal(literal) => literal.value().to_string(),
        other => other.to_string(),
    }
}

/// [`term_text`], but compacting an IRI against the document's vocabulary — for
/// terms that appear inside a generated message rather than in a structured
/// field, where the reader wants the spelling the shapes were authored in.
fn term_display(term: &Term, px: &Prefixes) -> String {
    match term {
        Term::NamedNode(iri) => px.compact(iri.as_str()),
        other => term_text(other),
    }
}

#[allow(clippy::too_many_arguments)]
fn reason(
    arena: &ShapeArena,
    id: ShapeId,
    value: Term,
    path: Option<String>,
    severity: &Severity,
    message: String,
    author_message: Option<String>,
    sub_reasons: Vec<Reason>,
    sparql_diagnostic: Option<SparqlDiagnostic>,
) -> Reason {
    Reason {
        value,
        path,
        shape: id,
        constraint: arena.get(id).clone(),
        constraint_kind: ConstraintKind::of(arena, id),
        constraint_id: id,
        statement_id: usize::MAX,
        severity: severity.clone(),
        message,
        author_message,
        observed_count: None,
        sub_reasons,
        sparql_diagnostic,
    }
}

/// Stamp the observed value count onto a cardinality reason. Kept off
/// [`reason`]'s argument list, which is already at the lint's limit, and set
/// only where a count was actually taken.
fn with_observed(mut r: Reason, observed: u64) -> Reason {
    r.observed_count = Some(observed);
    r
}

fn leaf(
    arena: &ShapeArena,
    ok: bool,
    node: &Term,
    id: ShapeId,
    path_ctx: Option<&str>,
    severity: &Severity,
    message: String,
) -> Vec<Reason> {
    if ok {
        Vec::new()
    } else {
        vec![reason(
            arena,
            id,
            node.clone(),
            path_ctx.map(str::to_string),
            severity,
            message,
            None,
            Vec::new(),
            None,
        )]
    }
}

fn all_pairs_ordered(
    g: &dyn PathBackend,
    v: &Term,
    path: &Path,
    p: &NamedNode,
    allow_eq: bool,
) -> bool {
    let lhs = succ(g, v, path);
    let rhs = objects(g, v, p);
    for a in &lhs {
        for b in &rhs {
            match compare_terms(a, b) {
                Some(Ordering::Less) => {}
                Some(Ordering::Equal) if allow_eq => {}
                _ => return false,
            }
        }
    }
    true
}

fn objects(g: &dyn PathBackend, v: &Term, p: &NamedNode) -> HashSet<Term> {
    succ(g, v, &Path::Pred(p.clone()))
}

/// Predicates on `node` not allowed by a closed shape's set `q`.
fn closed_offenders(
    g: &dyn PathBackend,
    node: &Term,
    q: &BTreeSet<NamedNode>,
) -> BTreeSet<NamedNode> {
    g.out_predicates(node)
        .into_iter()
        .filter(|p| !q.contains(p))
        .collect()
}

fn unique_lang(values: &HashSet<Term>) -> bool {
    let mut seen = HashSet::new();
    for term in values {
        if let Term::Literal(l) = term
            && let Some(lang) = l.language()
            && !seen.insert(lang.to_ascii_lowercase())
        {
            return false;
        }
    }
    true
}

fn dedup_reasons(reasons: &mut Vec<Reason>) {
    let mut seen = HashSet::new();
    reasons.retain(|r| {
        seen.insert((
            r.value.to_string(),
            r.message.clone(),
            r.severity.as_str().to_string(),
        ))
    });
}

#[cfg(test)]
mod tests {
    use super::*;
    use oxrdf::{NamedNode, Triple};

    fn iri(local: &str) -> NamedNode {
        NamedNode::new(format!("http://ex/{local}")).unwrap()
    }

    fn term(local: &str) -> Term {
        Term::NamedNode(iri(local))
    }

    #[test]
    fn memoizes_shared_value_checks_across_focus_nodes() {
        let p = iri("p");
        let shared = term("shared");
        let mut graph = Graph::new();
        graph.insert(&Triple::new(iri("a"), p.clone(), shared.clone()));
        graph.insert(&Triple::new(iri("b"), p.clone(), shared.clone()));

        let mut arena = ShapeArena::new();
        let qualifier = arena.insert(Shape::TestConst(shared));
        let root = arena.insert(Shape::Count {
            path: Path::Pred(p),
            min: Some(1),
            max: None,
            qualifier,
        });
        let sparql = SparqlExecutor::new(&graph).unwrap();
        crate::profile::enable();
        {
            let mut evaluator = ShapeEvaluator::new(&graph, &arena, Prefixes::empty(), &sparql);
            assert!(evaluator.holds(&term("a"), root));
            assert!(evaluator.holds(&term("b"), root));
        }
        let profile = crate::profile::take().unwrap();
        let cache = profile.shape_cache();
        assert_eq!(cache.evaluators, 1);
        assert!(cache.hits >= 1, "shared qualifier should hit the cache");
        assert_eq!(cache.peak_entries, 3);
        assert!(cache.estimated_peak_bytes > 0);
    }

    #[test]
    fn does_not_cache_cycle_dependent_results() {
        // A := B ∧ false; B := A. While evaluating A, the B result is
        // provisionally true through the A back-edge, but the gfp solution is
        // A=false, B=false. Caching that provisional B=true would be unsound.
        let mut arena = ShapeArena::new();
        let a = arena.reserve();
        let b = arena.reserve();
        let bottom = arena.insert(Shape::Or(Vec::new()));
        arena.set(a, Shape::And(vec![b, bottom]));
        arena.set(b, Shape::And(vec![a]));

        let graph = Graph::new();
        let sparql = SparqlExecutor::new(&graph).unwrap();
        let node = term("x");

        crate::profile::enable();
        {
            let mut evaluator = ShapeEvaluator::new(&graph, &arena, Prefixes::empty(), &sparql);
            assert!(!evaluator.holds(&node, a));
            assert!(!evaluator.holds(&node, b));
        }
        let profile = crate::profile::take().unwrap();
        let cache = profile.shape_cache();
        assert!(cache.recursion_back_edges > 0);
        assert!(cache.non_cacheable_results > 0);
    }
}