veredictum 0.1.4

// SPDX-FileCopyrightText: Veredictum contributors
// SPDX-License-Identifier: Apache-2.0

//! The typed assertion vocabulary (`flow[].assert` + `postconditions`).
//!
//! Eleven assertion forms, closed by schedule release. Semantics per the
//! CNF 2.0 artifact-set design: `equivalent` is the master07 "content
//! check" with normative ignore-sets; `version` asserts RM versioning facts
//! (`RM common §change_control`); `result_set` compares under the normative
//! AQL `RESULT_SET` equivalence rules (QUERY master03/04 + the ITS-REST query
//! schemas); `xml_root` judges a served canonical-XML document against the
//! published ITS-XML element declarations (ITS-REST overview `Resources.md`
//! §"XML Format"); `unique` is aggregate (evaluated once after all rows);
//! `message_exemplar` is informative only, never pass/fail.

#![expect(
    clippy::disallowed_types,
    reason = "dev/verification tooling over JSON artifacts (the catalogue, results, wire \
              exchanges), whose shapes belong to the artifacts and the SUT"
)]

use serde::de::Error as DeError;
use serde::{Deserialize, Deserializer};

use crate::ids::CaseId;
use crate::model::value::TemplatedValue;
use crate::refgrammar::{RefError, Template, ValueRef};
use crate::vocab::{
    CellComparison, ChangeType, FormatName, IgnoreSetName, ResultSetMatch, XmlNamespace,
};

/// The `equivalent` assertion's comparison target.
#[derive(Debug, Clone, PartialEq)]
pub enum EquivalentTarget {
    /// The content committed earlier in this row (`to: committed`).
    Committed,
    /// A corpus data set or a capture (`to: ${ds:…}` / `to: ${capture}`).
    Ref(ValueRef),
}

impl<'de> Deserialize<'de> for EquivalentTarget {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let s = String::deserialize(deserializer)?;
        if s == "committed" {
            return Ok(Self::Committed);
        }
        let template = Template::parse(&s).map_err(D::Error::custom)?;
        let reference = template.as_single_ref().ok_or_else(|| {
            D::Error::custom("equivalent target must be `committed` or a single ${…} reference")
        })?;
        match reference {
            ValueRef::DataSet { .. }
            | ValueRef::Capture {
                optional: false, ..
            } => Ok(Self::Ref(reference.clone())),
            _ => Err(D::Error::custom(
                "equivalent target reference must be ${ds:…} or ${<capture>}",
            )),
        }
    }
}

/// One `ignoring:` entry: a named normative ignore-set or an explicit path.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IgnoreSpec {
    /// A named set (`server_assigned` resolves from the operation's binding;
    /// `ctx_defaults` from the selectors vocabulary).
    Named(IgnoreSetName),
    /// An explicit RM path.
    Path(String),
}

impl<'de> Deserialize<'de> for IgnoreSpec {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let s = String::deserialize(deserializer)?;
        match s.as_str() {
            "server_assigned" => Ok(Self::Named(IgnoreSetName::ServerAssigned)),
            "ctx_defaults" => Ok(Self::Named(IgnoreSetName::CtxDefaults)),
            _ if s.contains('/') => Ok(Self::Path(s)),
            _ => Err(D::Error::custom(format!(
                "ignoring entry {s:?} is neither a named ignore-set (server_assigned | ctx_defaults) nor an explicit path"
            ))),
        }
    }
}

/// Scalar-or-list acceptance for `ignoring:`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct IgnoreList(pub Vec<IgnoreSpec>);

impl<'de> Deserialize<'de> for IgnoreList {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(Deserialize)]
        #[serde(untagged)]
        enum OneOrMany {
            One(IgnoreSpec),
            Many(Vec<IgnoreSpec>),
        }
        Ok(match OneOrMany::deserialize(deserializer)? {
            OneOrMany::One(one) => Self(vec![one]),
            OneOrMany::Many(many) => Self(many),
        })
    }
}

/// A template that must be exactly one `${…}` reference.
#[derive(Debug, Clone, PartialEq)]
pub struct SingleRef(pub ValueRef);

impl<'de> Deserialize<'de> for SingleRef {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let s = String::deserialize(deserializer)?;
        let template = Template::parse(&s).map_err(D::Error::custom)?;
        let reference = template
            .as_single_ref()
            .ok_or_else(|| D::Error::custom(format!("{s:?} must be a single ${{…}} reference")))?;
        Ok(Self(reference.clone()))
    }
}

/// Expected rows of a `result_set` assertion.
#[derive(Debug, Clone, PartialEq)]
pub enum RowsSpec {
    /// Rows from a named corpus view (`rows: { from: "${ds:<key>#<view>}" }`).
    From(ValueRef),
    /// Inline expected rows.
    Inline(Vec<Vec<serde_json::Value>>),
}

impl<'de> Deserialize<'de> for RowsSpec {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(Deserialize)]
        #[serde(deny_unknown_fields)]
        struct FromSpec {
            from: SingleRef,
        }
        #[derive(Deserialize)]
        #[serde(untagged)]
        enum Raw {
            From(FromSpec),
            Inline(Vec<Vec<serde_json::Value>>),
        }
        match Raw::deserialize(deserializer)? {
            Raw::From(FromSpec { from }) => match from.0 {
                reference @ ValueRef::DataSet { .. } => Ok(Self::From(reference)),
                other => Err(D::Error::custom(format!(
                    "result_set rows.from must be a ${{ds:…}} reference, got {other}"
                ))),
            },
            Raw::Inline(rows) => Ok(Self::Inline(rows)),
        }
    }
}

/// A `result_set` column expectation (identity: the `AS` alias, else the
/// 0-based index — ITS-REST `ResultSetColumn.yaml`).
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ColumnSpec {
    /// The expected column name (alias).
    pub name: String,
}

/// A typed assertion (tag: the `assert` field).
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(tag = "assert", rename_all = "snake_case", deny_unknown_fields)]
pub enum Assertion {
    /// Body parses as the named RM type and validates against the ITS schema
    /// for the active format.
    InstanceOf {
        /// The RM class name the body must parse as.
        rm_type: String,
        /// The wire format to parse in; defaults to the step's active format.
        #[serde(default)]
        format: Option<FormatName>,
    },
    /// RM-path-addressed field check; exactly one predicate.
    Field {
        /// The RM path addressing the field under test.
        path: String,
        /// The value the field must equal.
        #[serde(default)]
        equals: Option<TemplatedValue>,
        /// The server-set predicate: the stored value must differ from a
        /// client-supplied one (ITS-REST overview `Requests_and_responses`:
        /// `AUDIT_DETAILS.time_committed` is always server-set).
        #[serde(default)]
        not_equals: Option<TemplatedValue>,
        /// The field must be present (`true`) at the path.
        #[serde(default)]
        exists: Option<bool>,
        /// The field must be absent (`true`) at the path.
        #[serde(default)]
        absent: Option<bool>,
        /// A regex the field's serialized value must match.
        #[serde(default)]
        matches: Option<String>,
        /// The optional-member predicate: the field is absent, or its
        /// serialized value matches this regex. A released schema can declare a
        /// member's shape while leaving its presence to the service — ITS-REST
        /// `specifications/docs/query/Response.md` §Metadata: "`RESULT_SET`
        /// metadata comprise a set of optional (implementation dependent)
        /// attributes, useful for debugging" — so the shape is asserted exactly
        /// when the member is served.
        #[serde(default)]
        absent_or_matches: Option<String>,
    },
    /// The master07 "content check": retrieved equals committed, modulo the
    /// declared server-assigned set — normative per operation, never
    /// runner-chosen.
    Equivalent {
        /// What the retrieved body is compared against.
        to: EquivalentTarget,
        /// The declared server-assigned paths excluded from the comparison.
        #[serde(default)]
        ignoring: IgnoreList,
    },
    /// `ORIGINAL_VERSION.signature` facts (RM common §`change_control`,
    /// `Digital Signature`: the signature is over the canonical form of the
    /// version data; verification behaviour is conformance, algorithm
    /// strength is not). The wire seam is the versioned-object version read
    /// (the `ORIGINAL_VERSION` envelope), resolved by the interpreter.
    Signature {
        /// The single version the assertion judges.
        #[serde(default)]
        of: Option<SingleRef>,
        /// A captured set whose every member the assertion judges.
        #[serde(default)]
        for_each: Option<SingleRef>,
        /// The version carries a non-empty signature.
        #[serde(default)]
        present: Option<bool>,
        /// The signature verifies over the canonical version form against
        /// the statement-declared key material.
        #[serde(default)]
        verifiable: Option<bool>,
        /// The stored signature equals a known value (the client-verbatim
        /// storage rule for imported/committed signed versions).
        #[serde(default)]
        equals: Option<TemplatedValue>,
        /// The stored signature differs from a known (non-empty) value. The
        /// signature covers the version's canonical form, which includes `uid`,
        /// so two distinct versions carry distinct signatures (RM common
        /// `master06-change_control_package.adoc` §Digital Signature: "the
        /// entire Version object (… the signature attribute will be Void …)"
        /// is serialised and hashed; `version.adoc` `canonical_form`: "all
        /// attributes except signature").
        #[serde(default)]
        distinct_from: Option<TemplatedValue>,
    },
    /// RM versioning facts.
    Version {
        /// The single version the assertion judges.
        #[serde(default)]
        of: Option<SingleRef>,
        /// A captured set whose every member the assertion judges.
        #[serde(default)]
        for_each: Option<SingleRef>,
        /// The `commit_audit.change_type` the version must carry.
        #[serde(default)]
        change_type: Option<ChangeType>,
        /// The `lifecycle_state` value the version must carry.
        #[serde(default)]
        lifecycle_state: Option<String>,
        /// The exact number of versions the versioned object must hold.
        #[serde(default)]
        count: Option<u64>,
        /// A template the version's `uid` must match once resolved.
        #[serde(default)]
        uid_pattern: Option<Template>,
    },
    /// AQL results under the normative equivalence rules.
    ResultSet {
        /// How the expected rows are compared against the served ones.
        #[serde(rename = "match")]
        match_mode: ResultSetMatch,
        /// How a scalar cell is compared. Absent is
        /// [`CellComparison::Lexeme`], the exact-lexeme default; a row reading
        /// date/time values back through a QUERY declares `cells: instant`,
        /// because ITS-REST `specifications/docs/overview/Resources.md`
        /// §Datetime format puts the query-side spelling at SHOULD-strength
        /// ("Retrieval or querying those resources SHOULD return date,
        /// datetime, or time values in the (original) format provided by
        /// underlying backend engine, avoiding any format change") while the
        /// instant stays a fact. A tolerated respelling is recorded, never
        /// swallowed.
        #[serde(default)]
        cells: Option<CellComparison>,
        /// The expected rows (inline, or a reference to a corpus row set).
        #[serde(default)]
        rows: Option<RowsSpec>,
        /// The exact row count the result set must carry.
        #[serde(default)]
        count: Option<u64>,
        /// The expected columns, identified by `AS` alias.
        #[serde(default)]
        columns: Option<Vec<ColumnSpec>>,
    },
    /// Values captured across rows are pairwise distinct. Aggregate:
    /// evaluated once after all rows; requires `iteration: single_pass`.
    Unique {
        /// The captured value whose per-row instances must be pairwise distinct.
        over: SingleRef,
        /// Evaluate once after every row instead of per row.
        aggregate: bool,
    },
    /// Scalar service returns (no RM body).
    Returns {
        /// The exact value the scalar return must equal.
        #[serde(default)]
        equals: Option<serde_json::Value>,
        /// A regex the serialized body must match.
        #[serde(default)]
        matches: Option<String>,
        /// A regex the serialized body must NOT match — the negative
        /// containment predicate (e.g. a listing that must EXCLUDE a
        /// superseded row). Composes with `matches`: both are checked.
        #[serde(default)]
        omits: Option<String>,
    },
    /// The served canonical-XML document's ROOT element, judged against the
    /// published ITS-XML schemas: its local name and the namespace it is
    /// qualified with.
    ///
    /// The released ground is ITS-REST overview `Resources.md` §"XML Format":
    /// "When resources are serialized in **canonical XML** format, both request
    /// payloads and responses MUST conform to the [published XSDs]". Matching a
    /// complexType does not satisfy that: the root must be a globally declared
    /// element of the schema set, and every ITS-XML schema declares
    /// `elementFormDefault="qualified"` over a `targetNamespace`, so that
    /// element is namespace-qualified. A resource with no published element is
    /// out of scope by construction (register AMB-167).
    ///
    /// A `matches`-style regex over the raw body cannot express this: it cannot
    /// tell the root element from a descendant, and it cannot resolve a prefix
    /// to its namespace URI.
    ///
    /// Where the published element's declared type is abstract, the same MUST
    /// fixes a third fact: `xsi:type`. XML Schema Part 1 forbids an element
    /// instance from using an abstract type directly — the instance selects a
    /// non-abstract derived type with `xsi:type`
    /// (<https://www.w3.org/TR/xmlschema-1/#xsi_type>, §2.6.1 + §3.4.6). Two
    /// published document elements are declared that way in both vendored
    /// lineages: `<xs:element name="version" type="VERSION"/>` over
    /// `<xs:complexType name="VERSION" abstract="true">`
    /// (`specs/its-xml-schemas/its-xml-1.0.2-nsv1/ALL/Version.xsd`,
    /// `its-xml-2.0.0-nsv2/RM/latest/documents/Version.xsd` +
    /// `RM/latest/Common.xsd`), and `<xs:element name="items"
    /// type="LOCATABLE"/>` over the abstract `LOCATABLE` (`ALL/Structure.xsd`,
    /// `RM/latest/documents/Structure.xsd` + `RM/latest/Common.xsd`). There the
    /// concrete class is the only thing separating an `ORIGINAL_VERSION`
    /// response from an `IMPORTED_VERSION` one. On a concretely-typed element
    /// the attribute is decoration and no released sentence requires it.
    XmlRoot {
        /// The expected root element's local name (a globally declared element
        /// of the published XSDs).
        name: String,
        /// The expected namespace of the root element; omitted only where a
        /// row deliberately judges the name alone.
        #[serde(default)]
        namespace: Option<XmlNamespace>,
        /// The local name of the concrete type the root must name with
        /// `xsi:type`, asserted only on a published element whose declared type
        /// is abstract. The attribute value is a `QName`, so the assertion
        /// resolves its prefix through the document's in-scope bindings (an
        /// unprefixed `QName` resolves against the default namespace, the
        /// `QName`-in-content rule) and compares the local part. When the row
        /// also asserts `namespace`, the type's own namespace must satisfy the
        /// same expectation, because the ITS-XML complexTypes are declared in
        /// each schema's `targetNamespace`.
        #[serde(default)]
        xsi_type: Option<String>,
    },
    /// Informative only — never a pass/fail criterion.
    MessageExemplar {
        /// The exemplar message text, recorded for readers of the schedule.
        text: String,
    },
    /// A prose postcondition whose machine verification lives in a linked
    /// case or an in-case verification step.
    State {
        /// The postcondition in prose.
        text: String,
        /// The case that machine-verifies this postcondition, if separate.
        #[serde(default)]
        verified_by: Option<CaseId>,
    },
}

impl Assertion {
    /// Structural invariants beyond serde shape.
    ///
    /// # Errors
    /// Returns a message when a predicate-count or aggregate invariant is
    /// violated.
    pub fn check_invariants(&self) -> Result<(), String> {
        match self {
            Self::Field { .. } => self.check_field_invariants(),
            Self::Version { .. } => self.check_version_invariants(),
            other => other.check_other_invariants(),
        }
    }

    fn check_field_invariants(&self) -> Result<(), String> {
        match self {
            Self::Field {
                equals,
                not_equals,
                exists,
                absent,
                matches,
                absent_or_matches,
                path,
            } => {
                let predicates = usize::from(equals.is_some())
                    + usize::from(not_equals.is_some())
                    + usize::from(exists.is_some())
                    + usize::from(absent.is_some())
                    + usize::from(matches.is_some())
                    + usize::from(absent_or_matches.is_some());
                if predicates != 1 {
                    return Err(format!(
                        "field assertion on {path:?} must carry exactly one of equals | exists | absent | matches | absent_or_matches"
                    ));
                }
                for (label, re) in [
                    ("matches", matches),
                    ("absent_or_matches", absent_or_matches),
                ] {
                    if let Some(re) = re {
                        regex::Regex::new(re).map_err(|e| format!("field {label} regex: {e}"))?;
                    }
                }
                Ok(())
            }
            _ => Ok(()),
        }
    }

    fn check_version_invariants(&self) -> Result<(), String> {
        match self {
            Self::Version {
                of,
                for_each,
                change_type,
                lifecycle_state,
                count,
                uid_pattern,
            } => {
                if of.is_some() && for_each.is_some() {
                    return Err(
                        "version assertion: `of` and `for_each` are mutually exclusive".to_owned(),
                    );
                }
                if change_type.is_none()
                    && lifecycle_state.is_none()
                    && count.is_none()
                    && uid_pattern.is_none()
                {
                    return Err("version assertion carries no fact (change_type | lifecycle_state | count | uid_pattern)".to_owned());
                }
                if count.is_none() && of.is_none() && for_each.is_none() {
                    return Err(
                        "version assertion needs `of`/`for_each` (only `count` may stand alone)"
                            .to_owned(),
                    );
                }
                // The evaluator compares the FULL coded term (RM common
                // original_version.adoc §Attributes: a DV_CODED_TEXT), so a
                // bare code parses and can never match a conformant server.
                if let Some(state) = lifecycle_state {
                    let well_formed = state.split_once("::").is_some_and(|(terminology, rest)| {
                        !terminology.is_empty()
                            && rest.split_once('|').is_some_and(|(code, rubric_tail)| {
                                !code.is_empty() && rubric_tail.ends_with('|')
                            })
                    });
                    if !well_formed {
                        return Err(format!(
                            "version assertion: lifecycle_state {state:?} is not the \
                             `terminology::code|rubric|` term the evaluator compares against \
                             (RM common original_version.adoc §Attributes types it \
                             DV_CODED_TEXT, so a bare code can never match)"
                        ));
                    }
                }
                Ok(())
            }
            _ => Ok(()),
        }
    }

    fn check_other_invariants(&self) -> Result<(), String> {
        match self {
            Self::Signature {
                of,
                for_each,
                present,
                verifiable,
                equals,
                distinct_from,
            } => {
                if of.is_some() == for_each.is_some() {
                    return Err(
                        "signature assertion needs exactly one of `of` | `for_each`".to_owned()
                    );
                }
                if present.is_none()
                    && verifiable.is_none()
                    && equals.is_none()
                    && distinct_from.is_none()
                {
                    return Err(
                        "signature assertion carries no fact (present | verifiable | equals | distinct_from)"
                            .to_owned(),
                    );
                }
            }
            Self::ResultSet {
                match_mode,
                rows,
                count,
                cells,
                ..
            } => match match_mode {
                ResultSetMatch::Count => {
                    if count.is_none() {
                        return Err("result_set match:count requires `count`".to_owned());
                    }
                    // match:count compares no cell, so a `cells` mode here
                    // would be a declaration that silently does nothing.
                    if cells.is_some() {
                        return Err(
                            "result_set match:count compares no cell, so it takes no `cells` mode"
                                .to_owned(),
                        );
                    }
                }
                _ => {
                    if rows.is_none() {
                        return Err("result_set requires `rows` (except match:count)".to_owned());
                    }
                }
            },
            Self::Unique { aggregate, over } => {
                if !aggregate {
                    return Err(
                        "unique is defined only as an aggregate assertion (aggregate: true)"
                            .to_owned(),
                    );
                }
                if !matches!(
                    over.0,
                    ValueRef::Capture {
                        optional: false,
                        ..
                    }
                ) {
                    return Err("unique `over` must be a ${<capture>} reference".to_owned());
                }
            }
            Self::Returns {
                equals,
                matches,
                omits,
            } => {
                // `omits` composes with `matches` but never with `equals`,
                // which already pins the whole body.
                match (equals.is_some(), matches.is_some(), omits.is_some()) {
                    (true, false, false) | (false, true, _) | (false, false, true) => {}
                    _ => {
                        return Err(
                            "returns must carry exactly one of equals | matches [+ omits] | omits"
                                .to_owned(),
                        );
                    }
                }
                if let Some(re) = matches {
                    regex::Regex::new(re).map_err(|e| format!("returns matches regex: {e}"))?;
                }
                if let Some(re) = omits {
                    regex::Regex::new(re).map_err(|e| format!("returns omits regex: {e}"))?;
                }
            }
            Self::XmlRoot {
                name,
                xsi_type,
                namespace: _,
            } => check_xml_root_invariants(name, xsi_type.as_deref())?,
            Self::InstanceOf { .. }
            | Self::Equivalent { .. }
            | Self::MessageExemplar { .. }
            | Self::State { .. }
            | Self::Field { .. }
            | Self::Version { .. } => {}
        }
        Ok(())
    }

    /// Whether this assertion is evaluated once after all rows.
    #[must_use]
    pub fn is_aggregate(&self) -> bool {
        matches!(self.postcondition_role(), PostconditionRole::Aggregate)
    }

    /// The authored `assert:` token naming this assertion's family.
    #[must_use]
    pub fn family(&self) -> &'static str {
        match self {
            Self::InstanceOf { .. } => "instance_of",
            Self::Field { .. } => "field",
            Self::Equivalent { .. } => "equivalent",
            Self::Signature { .. } => "signature",
            Self::Version { .. } => "version",
            Self::ResultSet { .. } => "result_set",
            Self::Unique { .. } => "unique",
            Self::Returns { .. } => "returns",
            Self::XmlRoot { .. } => "xml_root",
            Self::MessageExemplar { .. } => "message_exemplar",
            Self::State { .. } => "state",
        }
    }

    /// The part this assertion plays as a `postconditions:` entry.
    ///
    /// The live driver and the transcript player both dispatch on this, so
    /// neither can skip a verdict-bearing family the other judges.
    #[must_use]
    pub fn postcondition_role(&self) -> PostconditionRole {
        match self {
            Self::Unique { .. } => PostconditionRole::Aggregate,
            // `message_exemplar` is the error body's own text (AMB-217) and
            // `state` carries its own `verified_by` case, so neither is
            // pass/fail.
            Self::MessageExemplar { .. } | Self::State { .. } => PostconditionRole::Informative,
            Self::InstanceOf { .. }
            | Self::Field { .. }
            | Self::Equivalent { .. }
            | Self::Signature { .. }
            | Self::Version { .. }
            | Self::ResultSet { .. }
            | Self::Returns { .. }
            | Self::XmlRoot { .. } => PostconditionRole::Judged,
        }
    }
}

/// How a `postconditions:` entry participates in the row verdict.
///
/// Closed by construction: a family is judged, aggregated, or informative,
/// and a runner that can judge none of them refuses the case instead of
/// producing a verdict over assertions it never evaluated.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PostconditionRole {
    /// Judged per row against the row's last completed exchange; the row
    /// verdict depends on the result.
    Judged,
    /// Evaluated once after the last row (law e).
    Aggregate,
    /// Recorded for readers of the schedule; never a pass/fail criterion.
    Informative,
}

/// The structural invariants of an `xml_root` assertion: both names it carries
/// are LOCAL names, because a prefix is a document's own choice and the
/// namespace both resolve against is asserted by `namespace:`.
fn check_xml_root_invariants(name: &str, xsi_type: Option<&str>) -> Result<(), String> {
    if name.trim().is_empty() {
        return Err("xml_root assertion needs the expected root element local name".to_owned());
    }
    if name.contains(':') {
        return Err(format!(
            "xml_root name {name:?} must be the LOCAL name — the prefix is a document's own choice and the namespace is asserted by `namespace:`"
        ));
    }
    let Some(rm_type) = xsi_type else {
        return Ok(());
    };
    if rm_type.trim().is_empty() {
        return Err("xml_root xsi_type must name the concrete type, or be omitted".to_owned());
    }
    if rm_type.contains(':') {
        return Err(format!(
            "xml_root xsi_type {rm_type:?} must be the LOCAL name — the QName's prefix is a document's own choice and its namespace rides on `namespace:`"
        ));
    }
    Ok(())
}

/// Every `${…}` reference used by an assertion (for the closed-grammar and
/// resolution checks).
#[must_use]
pub fn assertion_refs(assertion: &Assertion) -> Vec<ValueRef> {
    let mut out: Vec<ValueRef> = Vec::new();
    match assertion {
        Assertion::Field {
            equals, not_equals, ..
        } => {
            for v in [equals, not_equals].into_iter().flatten() {
                out.extend(v.refs().into_iter().cloned());
            }
        }
        Assertion::Equivalent { to, .. } => {
            if let EquivalentTarget::Ref(r) = to {
                out.push(r.clone());
            }
        }
        Assertion::Version {
            of,
            for_each,
            uid_pattern,
            ..
        } => {
            if let Some(SingleRef(r)) = of {
                out.push(r.clone());
            }
            if let Some(SingleRef(r)) = for_each {
                out.push(r.clone());
            }
            if let Some(t) = uid_pattern {
                out.extend(t.refs().cloned());
            }
        }
        Assertion::ResultSet { rows, .. } => {
            if let Some(RowsSpec::From(r)) = rows {
                out.push(r.clone());
            }
        }
        Assertion::Signature {
            of,
            for_each,
            equals,
            distinct_from,
            ..
        } => {
            if let Some(SingleRef(r)) = of {
                out.push(r.clone());
            }
            if let Some(SingleRef(r)) = for_each {
                out.push(r.clone());
            }
            if let Some(v) = equals {
                out.extend(v.refs().into_iter().cloned());
            }
            if let Some(v) = distinct_from {
                out.extend(v.refs().into_iter().cloned());
            }
        }
        Assertion::Unique { over, .. } => out.push(over.0.clone()),
        Assertion::InstanceOf { .. }
        | Assertion::Returns { .. }
        | Assertion::XmlRoot { .. }
        | Assertion::MessageExemplar { .. }
        | Assertion::State { .. } => {}
    }
    out
}

/// Parse-time hook so `RefError` conversion stays local to this module.
impl From<RefError> for String {
    fn from(e: RefError) -> Self {
        e.to_string()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn parse(v: serde_json::Value) -> Assertion {
        serde_json::from_value(v).unwrap()
    }

    #[test]
    fn pilot_assertions_parse() {
        let a = parse(serde_json::json!({
            "assert": "unique", "over": "${new_ehr_id}", "aggregate": true
        }));
        assert!(a.check_invariants().is_ok());
        assert!(a.is_aggregate());

        let a = parse(serde_json::json!({
            "assert": "version", "of": "${v2_uid}",
            "uid_pattern": "${versioned_object_uid}::<system>::2"
        }));
        assert!(a.check_invariants().is_ok());

        let a = parse(serde_json::json!({ "assert": "version", "count": 2 }));
        assert!(a.check_invariants().is_ok());

        let a = parse(serde_json::json!({
            "assert": "equivalent", "to": "committed", "ignoring": "server_assigned"
        }));
        assert!(a.check_invariants().is_ok());

        let a = parse(serde_json::json!({
            "assert": "result_set", "match": "ordered",
            "rows": { "from": "${ds:cnf.set.bp-10#magnitude_ge_140_by_uid}" },
            "columns": [{ "name": "uid" }]
        }));
        assert!(a.check_invariants().is_ok());
    }

    /// A bare code parses as YAML and can never match the full coded term the
    /// evaluator compares, so it is refused at the invariant rather than at
    /// drive time (RM common `original_version.adoc` §Attributes).
    #[test]
    fn a_lifecycle_state_outside_the_term_grammar_is_refused() {
        let term = parse(serde_json::json!({
            "assert": "version", "of": "${v2_uid}",
            "lifecycle_state": "openehr::523|deleted|"
        }));
        assert!(term.check_invariants().is_ok());
        // The empty rubric is a legal term: the grammar demands the delimiters.
        let empty_rubric = parse(serde_json::json!({
            "assert": "version", "of": "${v2_uid}", "lifecycle_state": "openehr::523||"
        }));
        assert!(empty_rubric.check_invariants().is_ok());
        for bad in [
            "523",
            "openehr::523",
            "523|deleted|",
            "openehr::523|deleted",
        ] {
            let a = parse(serde_json::json!({
                "assert": "version", "of": "${v2_uid}", "lifecycle_state": bad
            }));
            let message = a
                .check_invariants()
                .expect_err("a bare code must be refused");
            assert!(
                message.contains("terminology::code|rubric|"),
                "{bad}: {message}"
            );
        }
    }

    #[test]
    fn signature_distinct_from_is_a_fact() {
        let a = parse(serde_json::json!({
            "assert": "signature", "of": "${v2_uid}", "distinct_from": "${sig_first}"
        }));
        assert!(a.check_invariants().is_ok());
        assert!(
            assertion_refs(&a).iter().any(
                |r| matches!(r, ValueRef::Capture { name, .. } if name.as_str() == "sig_first")
            )
        );

        let a = parse(serde_json::json!({ "assert": "signature", "of": "${v}" }));
        assert!(a.check_invariants().is_err());
    }

    #[test]
    fn xml_root_takes_a_local_name_and_a_published_namespace() {
        let a = parse(serde_json::json!({
            "assert": "xml_root", "name": "composition", "namespace": "openehr-published"
        }));
        assert!(a.check_invariants().is_ok());
        assert!(assertion_refs(&a).is_empty());

        // The name alone is a legal, narrower row.
        let a = parse(serde_json::json!({ "assert": "xml_root", "name": "composition" }));
        assert!(a.check_invariants().is_ok());

        // A prefix is a document's own choice, never the assertion's.
        let a = parse(serde_json::json!({ "assert": "xml_root", "name": "oe:composition" }));
        assert!(a.check_invariants().is_err());

        let a = parse(serde_json::json!({ "assert": "xml_root", "name": "  " }));
        assert!(a.check_invariants().is_err());

        assert!(
            serde_json::from_value::<Assertion>(serde_json::json!({
                "assert": "xml_root", "name": "composition", "namespace": "http://example.org"
            }))
            .is_err()
        );
    }

    /// For a published element declared over an abstract type
    /// (`<xs:element name="version" type="VERSION"/>` over
    /// `<xs:complexType name="VERSION" abstract="true">`, `ALL/Version.xsd`),
    /// XML Schema Part 1 §2.6.1 + §3.4.6 make naming the concrete type part of
    /// conforming to the schema.
    #[test]
    fn xml_root_takes_the_concrete_type_of_an_abstract_root() {
        let a = parse(serde_json::json!({
            "assert": "xml_root", "name": "version",
            "namespace": "openehr-published", "xsi_type": "ORIGINAL_VERSION"
        }));
        assert!(a.check_invariants().is_ok());
        assert!(assertion_refs(&a).is_empty());

        let a = parse(serde_json::json!({
            "assert": "xml_root", "name": "version", "xsi_type": "oe:ORIGINAL_VERSION"
        }));
        assert!(a.check_invariants().is_err());

        let a = parse(serde_json::json!({
            "assert": "xml_root", "name": "version", "xsi_type": " "
        }));
        assert!(a.check_invariants().is_err());
    }

    /// The optional-member predicate counts as the field assertion's one
    /// predicate, and its regex is compiled at validate time like `matches`.
    #[test]
    fn absent_or_matches_is_one_predicate_with_a_compiled_regex() {
        let a = parse(serde_json::json!({
            "assert": "field", "path": "meta/_created", "absent_or_matches": "^[0-9]{4}-"
        }));
        assert!(a.check_invariants().is_ok());

        let a = parse(serde_json::json!({
            "assert": "field", "path": "meta/_created",
            "absent_or_matches": "^[0-9]{4}-", "exists": true
        }));
        assert!(a.check_invariants().is_err());

        let a = parse(serde_json::json!({
            "assert": "field", "path": "meta/_created", "absent_or_matches": "([unclosed"
        }));
        assert!(a.check_invariants().is_err());
    }

    /// The `cells:` vocabulary is closed and opt-in: absent is the exact
    /// comparison, an unknown token is a parse refusal, and `match: count`
    /// refuses the modifier outright.
    #[test]
    fn cells_mode_is_a_closed_opt_in_vocabulary() {
        let a = parse(serde_json::json!({
            "assert": "result_set", "match": "ordered",
            "rows": [["2026-01-01T00:00:00Z"]]
        }));
        assert!(a.check_invariants().is_ok());
        assert!(matches!(a, Assertion::ResultSet { cells: None, .. }));

        let a = parse(serde_json::json!({
            "assert": "result_set", "match": "ordered", "cells": "instant",
            "rows": [["2026-01-01T00:00:00Z"]]
        }));
        assert!(a.check_invariants().is_ok());
        assert!(matches!(
            a,
            Assertion::ResultSet {
                cells: Some(CellComparison::Instant),
                ..
            }
        ));

        assert!(
            serde_json::from_value::<Assertion>(serde_json::json!({
                "assert": "result_set", "match": "ordered", "cells": "iso8601",
                "rows": [["x"]]
            }))
            .is_err()
        );

        let a = parse(serde_json::json!({
            "assert": "result_set", "match": "count", "count": 3, "cells": "instant"
        }));
        assert!(a.check_invariants().is_err());
    }

    #[test]
    fn invariants_bite() {
        let a = parse(serde_json::json!({ "assert": "version", "of": "${v}" }));
        assert!(a.check_invariants().is_err()); // no fact

        let a = parse(serde_json::json!({
            "assert": "field", "path": "x", "exists": true, "absent": true
        }));
        assert!(a.check_invariants().is_err()); // two predicates

        let a = parse(serde_json::json!({
            "assert": "unique", "over": "${x}", "aggregate": false
        }));
        assert!(a.check_invariants().is_err());

        assert!(
            serde_json::from_value::<Assertion>(serde_json::json!({
                "assert": "equivalent", "to": "${row.thing}"
            }))
            .is_err()
        );

        assert!(
            serde_json::from_value::<Assertion>(serde_json::json!({
                "assert": "totally_new", "x": 1
            }))
            .is_err()
        );
    }
}