boundgraph-check 0.1.0

//! The cheapest deterministic check there is: does this output match its
//! declared shape?
//!
//! A step that promises `{ "severity": "high" | "low", "files": [string] }` and
//! delivers prose has failed, and it has failed in a way that no downstream
//! step can recover from — it will either crash on deserialisation or, worse,
//! silently read a field that is not there. Catching it here costs microseconds
//! and needs no tools, no sandbox, and no model.
//!
//! # A subset, and unknown keywords are refused
//!
//! This validates a documented subset of JSON Schema, listed on
//! [`SUPPORTED_KEYWORDS`]. Anything outside it — `$ref`, `pattern`, `format`,
//! `if`/`then`/`else` — makes [`Schema::compile`] **fail**.
//!
//! Refusing is the whole design, and the alternative is worse than it sounds.
//! A validator that ignored the keywords it did not implement would hand a
//! schema author a `Pass` on a payload their schema said was invalid, and
//! nothing anywhere would report that a constraint had been skipped. That is a
//! silent downgrade of a security property, which is the failure mode Goal G5
//! exists to prevent. A build error naming the keyword is a cheap failure; a
//! check that quietly stopped checking is not.
//!
//! # Why it cannot hang
//!
//! Two independent bounds.
//!
//! **The schema is a finite tree.** `$ref` is not supported, so a schema cannot
//! refer to itself, and schema depth is capped at compile time
//! ([`MAX_DEPTH`]). That is a stronger property than it first looks: validation
//! descends only where the *schema* descends, so recursion is bounded by the
//! schema and not by the instance. A payload nested a thousand levels deep
//! against a two-level schema is walked two levels deep.
//!
//! **Validation has a step budget.** A finite schema still multiplies work:
//! `oneOf` re-validates the same subtree once per alternative, and against a
//! large array that compounds. So every comparison decrements a counter, and
//! running out is [`Verdict::Inconclusive`](boundgraph_core::Verdict::Inconclusive) —
//! never `Pass`, and never `Fail` either, because "we ran out of budget" is not
//! the same claim as "your payload is wrong".
//!
//! What this module does *not* defend against is a payload so deeply nested
//! that operating on it at all overflows the stack — `enum` and `uniqueItems`
//! compare values structurally, and comparison recurses over the instance. That
//! hazard is real but it is not created here and cannot be fixed here: by the
//! time a `Condensate` exists, something has already built the value, and
//! whatever built it would have overflowed first. Capping payload depth belongs
//! where payloads are produced.
//!
//! # What a `Pass` actually guarantees
//!
//! That the payload was representable as JSON *and* satisfied the schema. Both
//! halves matter: the checker converts the condensate's CBOR payload into JSON
//! before validating, so a payload that could not survive that trip fails here
//! rather than three steps later. A downstream step deserialising into a type
//! that matches the schema will therefore succeed.

use std::collections::{BTreeMap, BTreeSet};

use async_trait::async_trait;
use boundgraph_core::{CheckerId, Condensate, CondensatePayload, Digest, WorkOrder};
use serde_json::Value;

use crate::checker::{Checker, Judgement};

/// The keywords this validator implements.
///
/// A schema using anything else is refused rather than partially checked. See
/// the module documentation for why.
pub const SUPPORTED_KEYWORDS: &[&str] = &[
    "type",
    "enum",
    "const",
    "properties",
    "required",
    "additionalProperties",
    "minProperties",
    "maxProperties",
    "items",
    "minItems",
    "maxItems",
    "uniqueItems",
    "minimum",
    "maximum",
    "exclusiveMinimum",
    "exclusiveMaximum",
    "multipleOf",
    "minLength",
    "maxLength",
    "allOf",
    "anyOf",
    "oneOf",
    "not",
];

/// Keywords that carry no constraint and are skipped rather than refused.
///
/// Documentation, not validation. Refusing a schema for carrying a `title`
/// would be pedantry; refusing one for carrying a `pattern` is the point.
const ANNOTATIONS: &[&str] = &[
    "$schema",
    "$id",
    "$comment",
    "title",
    "description",
    "default",
    "examples",
    "deprecated",
    "readOnly",
    "writeOnly",
];

/// How deeply a schema may nest.
///
/// Generous for anything hand-written. A schema deeper than this is either
/// generated or hostile, and both are worth stopping at compile time.
pub const MAX_DEPTH: usize = 64;

/// How many comparisons a validation may make before giving up.
///
/// Reached only by a schema that multiplies work — nested `oneOf` over a large
/// array — against a large instance. Exhaustion is `Inconclusive`.
pub const DEFAULT_BUDGET: usize = 100_000;

/// A schema could not be compiled.
///
/// Every variant names the offending location as a JSON Pointer into the schema
/// itself, so the author is told where to look rather than that "the schema is
/// invalid".
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum SchemaError {
    /// The schema text was not JSON.
    #[error("the schema is not valid JSON: {0}")]
    NotJson(String),

    /// A keyword outside the supported subset.
    #[error("{path}: `{keyword}` is not supported by this validator, so the schema is refused")]
    UnsupportedKeyword {
        /// Where in the schema.
        path: String,
        /// Which keyword.
        keyword: String,
    },

    /// A keyword was present but its value was the wrong shape.
    #[error("{path}: `{keyword}` must be {expected}")]
    MalformedKeyword {
        /// Where in the schema.
        path: String,
        /// Which keyword.
        keyword: String,
        /// What it should have been.
        expected: String,
    },

    /// A schema must be an object or a boolean.
    #[error("{path}: a schema must be an object or a boolean, found {found}")]
    NotASchema {
        /// Where in the schema.
        path: String,
        /// What was there instead.
        found: String,
    },

    /// The schema nests deeper than [`MAX_DEPTH`].
    #[error("{path}: the schema nests deeper than {MAX_DEPTH} levels")]
    TooDeep {
        /// Where the limit was hit.
        path: String,
    },

    /// The schema could not be content-addressed.
    #[error("the schema could not be canonically encoded: {0}")]
    Encode(String),
}

/// One way in which a value did not match its schema.
///
/// Names the location and the constraint separately, because a repair loop acts
/// on both: the path says where to edit, the constraint says what rule was
/// broken.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Violation {
    /// A JSON Pointer into the instance, e.g. `/items/2/name`. The empty string
    /// is the root.
    pub path: String,
    /// The schema keyword that was not satisfied, e.g. `type`, `required`.
    pub constraint: &'static str,
    /// What was expected and what was found.
    pub detail: String,
}

impl Violation {
    /// One line, as it appears in a failure reason.
    pub fn render(&self) -> String {
        let at = if self.path.is_empty() {
            "(root)".to_string()
        } else {
            self.path.clone()
        };
        format!("{at}: {} ({})", self.detail, self.constraint)
    }
}

/// What validating a value produced.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Validation {
    /// The value satisfies the schema.
    Valid,
    /// It does not, and here is every reason why.
    Invalid(Vec<Violation>),
    /// The step budget ran out before an answer was reached.
    ///
    /// Deliberately not `Invalid`: a validator that ran out of budget has not
    /// shown the value to be wrong, and reporting it as wrong would blame the
    /// payload for the validator's limit.
    Exhausted {
        /// The budget that was exhausted.
        budget: usize,
    },
}

impl Validation {
    /// Whether the value satisfied the schema.
    ///
    /// [`Exhausted`](Self::Exhausted) is not valid. It is not invalid either;
    /// this is the question "may it pass", and the answer is no.
    pub fn is_valid(&self) -> bool {
        matches!(self, Self::Valid)
    }

    /// Whether validation reached an answer at all.
    ///
    /// `false` for exhaustion, which is what the checker turns into
    /// [`Verdict::Inconclusive`](boundgraph_core::Verdict::Inconclusive).
    pub fn is_decided(&self) -> bool {
        matches!(self, Self::Valid | Self::Invalid(_))
    }
}

/// A compiled schema: a finite tree of constraints, plus its content address.
#[derive(Debug, Clone, PartialEq)]
pub struct Schema {
    root: Node,
    source: Value,
    digest: Digest,
}

/// One node of a compiled schema.
#[derive(Debug, Clone, PartialEq, Default)]
struct Node {
    /// A boolean schema: `true` accepts everything, `false` accepts nothing.
    boolean: Option<bool>,
    types: Option<BTreeSet<Type>>,
    enumeration: Option<Vec<Value>>,
    constant: Option<Value>,
    properties: BTreeMap<String, Node>,
    required: Vec<String>,
    additional: Option<Additional>,
    min_properties: Option<usize>,
    max_properties: Option<usize>,
    items: Option<Box<Node>>,
    min_items: Option<usize>,
    max_items: Option<usize>,
    unique_items: bool,
    minimum: Option<f64>,
    maximum: Option<f64>,
    exclusive_minimum: Option<f64>,
    exclusive_maximum: Option<f64>,
    multiple_of: Option<f64>,
    min_length: Option<usize>,
    max_length: Option<usize>,
    all_of: Vec<Node>,
    any_of: Vec<Node>,
    one_of: Vec<Node>,
    not: Option<Box<Node>>,
}

/// What `additionalProperties` said.
#[derive(Debug, Clone, PartialEq)]
enum Additional {
    /// `false`: unlisted properties are refused.
    Forbidden,
    /// A schema every unlisted property must satisfy. `true` compiles to an
    /// empty schema, which accepts everything.
    Schema(Box<Node>),
}

/// The JSON type names a schema may name.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum Type {
    Null,
    Boolean,
    Object,
    Array,
    Number,
    Integer,
    String,
}

impl Type {
    fn parse(name: &str) -> Option<Self> {
        Some(match name {
            "null" => Self::Null,
            "boolean" => Self::Boolean,
            "object" => Self::Object,
            "array" => Self::Array,
            "number" => Self::Number,
            "integer" => Self::Integer,
            "string" => Self::String,
            _ => return None,
        })
    }

    fn name(self) -> &'static str {
        match self {
            Self::Null => "null",
            Self::Boolean => "boolean",
            Self::Object => "object",
            Self::Array => "array",
            Self::Number => "number",
            Self::Integer => "integer",
            Self::String => "string",
        }
    }

    /// Whether `value` is of this type.
    ///
    /// `integer` accepts a number with no fractional part, as JSON Schema
    /// requires — `2.0` is an integer, and refusing it would reject every
    /// language that has only one numeric type.
    fn matches(self, value: &Value) -> bool {
        match self {
            Self::Null => value.is_null(),
            Self::Boolean => value.is_boolean(),
            Self::Object => value.is_object(),
            Self::Array => value.is_array(),
            Self::Number => value.is_number(),
            Self::Integer => value.as_f64().is_some_and(|n| n.fract() == 0.0),
            Self::String => value.is_string(),
        }
    }
}

/// The name of a value's own type, for a diagnostic.
fn type_of(value: &Value) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "boolean",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

impl Schema {
    /// Compile a schema from JSON text.
    ///
    /// # Errors
    ///
    /// [`SchemaError`] if the text is not JSON, or the schema uses a keyword
    /// outside [`SUPPORTED_KEYWORDS`].
    pub fn parse(text: &str) -> Result<Self, SchemaError> {
        let value: Value =
            serde_json::from_str(text).map_err(|e| SchemaError::NotJson(e.to_string()))?;
        Self::compile(&value)
    }

    /// Compile an already-parsed schema.
    ///
    /// # Errors
    ///
    /// [`SchemaError`] if the schema uses a keyword outside
    /// [`SUPPORTED_KEYWORDS`], nests deeper than [`MAX_DEPTH`], or gives a
    /// keyword a value of the wrong shape.
    pub fn compile(value: &Value) -> Result<Self, SchemaError> {
        let root = compile_node(value, "", 0)?;
        let bytes = boundgraph_core::to_canonical_cbor(value)
            .map_err(|e| SchemaError::Encode(e.to_string()))?;
        Ok(Self {
            root,
            source: value.clone(),
            digest: Digest::of(&bytes),
        })
    }

    /// The schema's content address.
    ///
    /// Recorded in every piece of evidence this schema produces, so a verdict
    /// says which schema it was reached against. Two schemas that differ by so
    /// much as a whitespace-insensitive key ordering share a digest; two that
    /// differ in a constraint do not.
    pub const fn digest(&self) -> Digest {
        self.digest
    }

    /// The schema as it was given.
    pub const fn source(&self) -> &Value {
        &self.source
    }

    /// Validate a value, with the default step budget.
    pub fn validate(&self, value: &Value) -> Validation {
        self.validate_within(value, DEFAULT_BUDGET)
    }

    /// Validate a value, giving up after `budget` comparisons.
    ///
    /// See the module documentation for why exhaustion is its own outcome.
    pub fn validate_within(&self, value: &Value, budget: usize) -> Validation {
        let mut ctx = Ctx {
            remaining: budget,
            violations: Vec::new(),
            gave_up: false,
        };

        check(&self.root, value, "", &mut ctx);

        if ctx.gave_up {
            Validation::Exhausted { budget }
        } else if ctx.violations.is_empty() {
            Validation::Valid
        } else {
            Validation::Invalid(ctx.violations)
        }
    }
}

/// Validation state: what is left of the budget, what has gone wrong, and
/// whether the budget ran out.
///
/// Once `gave_up` is set every `check` returns immediately, so the walk unwinds
/// without doing further work and without every call site having to thread a
/// result back.
struct Ctx {
    remaining: usize,
    violations: Vec<Violation>,
    gave_up: bool,
}

impl Ctx {
    /// Spend one step, giving up if the budget is gone.
    fn step(&mut self) -> bool {
        if self.gave_up {
            return false;
        }
        match self.remaining.checked_sub(1) {
            Some(left) => {
                self.remaining = left;
                true
            }
            None => {
                self.gave_up = true;
                false
            }
        }
    }

    fn fail(&mut self, path: &str, constraint: &'static str, detail: impl Into<String>) {
        self.violations.push(Violation {
            path: path.to_string(),
            constraint,
            detail: detail.into(),
        });
    }
}

/// Compile one schema node.
fn compile_node(value: &Value, path: &str, depth: usize) -> Result<Node, SchemaError> {
    if depth > MAX_DEPTH {
        return Err(SchemaError::TooDeep {
            path: path.to_string(),
        });
    }

    let object = match value {
        Value::Bool(accept) => {
            return Ok(Node {
                boolean: Some(*accept),
                ..Node::default()
            });
        }
        Value::Object(map) => map,
        other => {
            return Err(SchemaError::NotASchema {
                path: path.to_string(),
                found: type_of(other).to_string(),
            });
        }
    };

    let malformed = |keyword: &str, expected: &str| SchemaError::MalformedKeyword {
        path: path.to_string(),
        keyword: keyword.to_string(),
        expected: expected.to_string(),
    };

    let mut node = Node::default();

    for (keyword, arg) in object {
        if ANNOTATIONS.contains(&keyword.as_str()) {
            continue;
        }
        if !SUPPORTED_KEYWORDS.contains(&keyword.as_str()) {
            return Err(SchemaError::UnsupportedKeyword {
                path: path.to_string(),
                keyword: keyword.clone(),
            });
        }

        let at = |segment: &str| format!("{path}/{keyword}{segment}");

        match keyword.as_str() {
            "type" => {
                let names = match arg {
                    Value::String(one) => vec![one.clone()],
                    Value::Array(many) => many
                        .iter()
                        .map(|v| v.as_str().map(str::to_string))
                        .collect::<Option<Vec<_>>>()
                        .ok_or_else(|| malformed("type", "a type name or an array of them"))?,
                    _ => return Err(malformed("type", "a type name or an array of them")),
                };
                let mut types = BTreeSet::new();
                for name in names {
                    types.insert(Type::parse(&name).ok_or_else(|| {
                        malformed(
                            "type",
                            "one of null, boolean, object, array, number, integer, string",
                        )
                    })?);
                }
                node.types = Some(types);
            }
            "enum" => {
                let values = arg
                    .as_array()
                    .ok_or_else(|| malformed("enum", "an array of values"))?;
                node.enumeration = Some(values.clone());
            }
            "const" => node.constant = Some(arg.clone()),
            "properties" => {
                let map = arg
                    .as_object()
                    .ok_or_else(|| malformed("properties", "an object of schemas"))?;
                for (name, sub) in map {
                    node.properties.insert(
                        name.clone(),
                        compile_node(sub, &at(&format!("/{name}")), depth + 1)?,
                    );
                }
            }
            "required" => {
                let names = arg
                    .as_array()
                    .and_then(|a| {
                        a.iter()
                            .map(|v| v.as_str().map(str::to_string))
                            .collect::<Option<Vec<_>>>()
                    })
                    .ok_or_else(|| malformed("required", "an array of property names"))?;
                node.required = names;
            }
            "additionalProperties" => {
                node.additional = Some(match arg {
                    Value::Bool(false) => Additional::Forbidden,
                    other => Additional::Schema(Box::new(compile_node(other, &at(""), depth + 1)?)),
                });
            }
            "items" => node.items = Some(Box::new(compile_node(arg, &at(""), depth + 1)?)),
            "not" => node.not = Some(Box::new(compile_node(arg, &at(""), depth + 1)?)),
            "allOf" | "anyOf" | "oneOf" => {
                let branches = arg
                    .as_array()
                    .ok_or_else(|| malformed(keyword, "an array of schemas"))?;
                let mut compiled = Vec::with_capacity(branches.len());
                for (index, branch) in branches.iter().enumerate() {
                    compiled.push(compile_node(branch, &at(&format!("/{index}")), depth + 1)?);
                }
                match keyword.as_str() {
                    "allOf" => node.all_of = compiled,
                    "anyOf" => node.any_of = compiled,
                    _ => node.one_of = compiled,
                }
            }
            "minProperties" | "maxProperties" | "minItems" | "maxItems" | "minLength"
            | "maxLength" => {
                let n = arg
                    .as_u64()
                    .ok_or_else(|| malformed(keyword, "a non-negative integer"))?
                    as usize;
                match keyword.as_str() {
                    "minProperties" => node.min_properties = Some(n),
                    "maxProperties" => node.max_properties = Some(n),
                    "minItems" => node.min_items = Some(n),
                    "maxItems" => node.max_items = Some(n),
                    "minLength" => node.min_length = Some(n),
                    _ => node.max_length = Some(n),
                }
            }
            "uniqueItems" => {
                node.unique_items = arg
                    .as_bool()
                    .ok_or_else(|| malformed("uniqueItems", "a boolean"))?;
            }
            "minimum" | "maximum" | "exclusiveMinimum" | "exclusiveMaximum" | "multipleOf" => {
                let n = arg.as_f64().ok_or_else(|| malformed(keyword, "a number"))?;
                match keyword.as_str() {
                    "minimum" => node.minimum = Some(n),
                    "maximum" => node.maximum = Some(n),
                    "exclusiveMinimum" => node.exclusive_minimum = Some(n),
                    "exclusiveMaximum" => node.exclusive_maximum = Some(n),
                    _ => {
                        if n <= 0.0 {
                            return Err(malformed("multipleOf", "a number greater than zero"));
                        }
                        node.multiple_of = Some(n);
                    }
                }
            }
            _ => unreachable!("keyword {keyword} is in SUPPORTED_KEYWORDS but not handled"),
        }
    }

    Ok(node)
}

/// Escape a property name for a JSON Pointer, per RFC 6901.
fn pointer(path: &str, segment: &str) -> String {
    format!("{path}/{}", segment.replace('~', "~0").replace('/', "~1"))
}

/// Validate `value` against `node`, recording violations.
///
/// Stops early once [`Ctx::gave_up`] is set, so the walk unwinds without every
/// call site having to thread a result back.
fn check(node: &Node, value: &Value, path: &str, ctx: &mut Ctx) {
    // No instance-depth guard, and none is needed: this descends only into
    // `node`'s children, and `compile_node` capped those at MAX_DEPTH.
    if !ctx.step() {
        return;
    }

    if let Some(accept) = node.boolean {
        if !accept {
            ctx.fail(path, "false", "this schema accepts nothing");
        }
        return;
    }

    if let Some(types) = &node.types
        && !types.iter().any(|t| t.matches(value))
    {
        let names: Vec<&str> = types.iter().map(|t| t.name()).collect();
        ctx.fail(
            path,
            "type",
            format!("expected {}, found {}", names.join(" or "), type_of(value)),
        );
        // Every other keyword is type-specific, so continuing would produce a
        // cascade of violations that all restate this one.
        return;
    }

    if let Some(allowed) = &node.enumeration
        && !allowed.contains(value)
    {
        ctx.fail(
            path,
            "enum",
            format!(
                "`{value}` is not one of the {} permitted values",
                allowed.len()
            ),
        );
    }

    if let Some(expected) = &node.constant
        && value != expected
    {
        ctx.fail(
            path,
            "const",
            format!("expected `{expected}`, found `{value}`"),
        );
    }

    match value {
        Value::Object(map) => check_object(node, map, path, ctx),
        Value::Array(items) => check_array(node, items, path, ctx),
        Value::String(text) => check_string(node, text, path, ctx),
        Value::Number(_) => {
            if let Some(n) = value.as_f64() {
                check_number(node, n, path, ctx);
            }
        }
        _ => {}
    }

    check_combinators(node, value, path, ctx);
}

fn check_object(node: &Node, map: &serde_json::Map<String, Value>, path: &str, ctx: &mut Ctx) {
    for name in &node.required {
        if !ctx.step() {
            return;
        }
        if !map.contains_key(name) {
            ctx.fail(
                path,
                "required",
                format!("missing required property `{name}`"),
            );
        }
    }

    if let Some(min) = node.min_properties
        && map.len() < min
    {
        ctx.fail(
            path,
            "minProperties",
            format!("expected at least {min} properties, found {}", map.len()),
        );
    }
    if let Some(max) = node.max_properties
        && map.len() > max
    {
        ctx.fail(
            path,
            "maxProperties",
            format!("expected at most {max} properties, found {}", map.len()),
        );
    }

    for (name, sub) in map {
        if ctx.gave_up {
            return;
        }
        let at = pointer(path, name);
        if let Some(schema) = node.properties.get(name) {
            check(schema, sub, &at, ctx);
        } else {
            match node.additional.as_ref() {
                Some(Additional::Forbidden) => ctx.fail(
                    path,
                    "additionalProperties",
                    format!("property `{name}` is not permitted here"),
                ),
                Some(Additional::Schema(schema)) => check(schema, sub, &at, ctx),
                None => {}
            }
        }
    }
}

fn check_array(node: &Node, items: &[Value], path: &str, ctx: &mut Ctx) {
    if let Some(min) = node.min_items
        && items.len() < min
    {
        ctx.fail(
            path,
            "minItems",
            format!("expected at least {min} items, found {}", items.len()),
        );
    }
    if let Some(max) = node.max_items
        && items.len() > max
    {
        ctx.fail(
            path,
            "maxItems",
            format!("expected at most {max} items, found {}", items.len()),
        );
    }

    if node.unique_items {
        for (index, item) in items.iter().enumerate() {
            if !ctx.step() {
                return;
            }
            if let Some(first) = items[..index].iter().position(|other| other == item) {
                ctx.fail(
                    &format!("{path}/{index}"),
                    "uniqueItems",
                    format!("duplicates the item at index {first}"),
                );
            }
        }
    }

    if let Some(schema) = &node.items {
        for (index, item) in items.iter().enumerate() {
            if ctx.gave_up {
                return;
            }
            check(schema, item, &format!("{path}/{index}"), ctx);
        }
    }
}

fn check_string(node: &Node, text: &str, path: &str, ctx: &mut Ctx) {
    // Characters, not bytes: JSON Schema counts code points, and a length limit
    // that counted UTF-8 bytes would reject a valid name in most of the world's
    // scripts.
    let length = text.chars().count();

    if let Some(min) = node.min_length
        && length < min
    {
        ctx.fail(
            path,
            "minLength",
            format!("expected at least {min} characters, found {length}"),
        );
    }
    if let Some(max) = node.max_length
        && length > max
    {
        ctx.fail(
            path,
            "maxLength",
            format!("expected at most {max} characters, found {length}"),
        );
    }
}

fn check_number(node: &Node, n: f64, path: &str, ctx: &mut Ctx) {
    if let Some(min) = node.minimum
        && n < min
    {
        ctx.fail(path, "minimum", format!("{n} is below the minimum {min}"));
    }
    if let Some(max) = node.maximum
        && n > max
    {
        ctx.fail(path, "maximum", format!("{n} is above the maximum {max}"));
    }
    if let Some(min) = node.exclusive_minimum
        && n <= min
    {
        ctx.fail(
            path,
            "exclusiveMinimum",
            format!("{n} is not above the exclusive minimum {min}"),
        );
    }
    if let Some(max) = node.exclusive_maximum
        && n >= max
    {
        ctx.fail(
            path,
            "exclusiveMaximum",
            format!("{n} is not below the exclusive maximum {max}"),
        );
    }
    if let Some(step) = node.multiple_of {
        let quotient = n / step;
        // A tolerance rather than an exact remainder: `0.1` is not exactly
        // representable, so `0.3 % 0.1` is not zero and a strict test would
        // reject arithmetic every schema author considers correct.
        let off = (quotient - quotient.round()).abs();
        if off > 1e-9 {
            ctx.fail(
                path,
                "multipleOf",
                format!("{n} is not a multiple of {step}"),
            );
        }
    }
}

fn check_combinators(node: &Node, value: &Value, path: &str, ctx: &mut Ctx) {
    for branch in &node.all_of {
        if ctx.gave_up {
            return;
        }
        check(branch, value, path, ctx);
    }

    if !node.any_of.is_empty() {
        let mut matched = false;
        for branch in &node.any_of {
            match matches_quietly(branch, value, path, ctx) {
                None => return,
                Some(true) => {
                    matched = true;
                    break;
                }
                Some(false) => {}
            }
        }
        if !matched {
            ctx.fail(
                path,
                "anyOf",
                format!("matched none of the {} alternatives", node.any_of.len()),
            );
        }
    }

    if !node.one_of.is_empty() {
        let mut matches = 0usize;
        for branch in &node.one_of {
            match matches_quietly(branch, value, path, ctx) {
                None => return,
                Some(true) => matches += 1,
                Some(false) => {}
            }
        }
        if matches != 1 {
            ctx.fail(
                path,
                "oneOf",
                format!(
                    "matched {matches} of the {} alternatives, expected exactly one",
                    node.one_of.len()
                ),
            );
        }
    }

    if let Some(inner) = &node.not {
        match matches_quietly(inner, value, path, ctx) {
            None => (),
            Some(true) => ctx.fail(path, "not", "matched a schema it must not match"),
            Some(false) => {}
        }
    }
}

/// Whether a value matches a branch, without recording why it did not.
///
/// A failed `anyOf` branch is not an error the author needs to see — it is how
/// alternation works. Only the alternation itself is reported. `None` means the
/// budget ran out.
fn matches_quietly(node: &Node, value: &Value, path: &str, ctx: &mut Ctx) -> Option<bool> {
    let mut probe = Ctx {
        remaining: ctx.remaining,
        violations: Vec::new(),
        gave_up: false,
    };
    check(node, value, path, &mut probe);

    // The budget is shared, so a branch that spent it does not leave the next
    // branch a fresh one.
    ctx.remaining = probe.remaining;

    if probe.gave_up {
        ctx.gave_up = probe.gave_up;
        return None;
    }
    Some(probe.violations.is_empty())
}

/// Rejects a condensate payload that does not match a declared schema.
///
/// # Example
///
/// ```
/// use boundgraph_check::{Schema, SchemaChecker};
/// use serde_json::json;
///
/// let schema = Schema::compile(&json!({
///     "type": "object",
///     "properties": { "severity": { "enum": ["low", "high"] } },
///     "required": ["severity"],
/// }))?;
///
/// assert!(schema.validate(&json!({ "severity": "high" })).is_valid());
///
/// let bad = schema.validate(&json!({ "severity": "catastrophic" }));
/// assert!(!bad.is_valid());
///
/// let checker = SchemaChecker::new(schema);
/// # let _ = checker;
/// # Ok::<(), boundgraph_check::SchemaError>(())
/// ```
#[derive(Debug, Clone)]
pub struct SchemaChecker {
    schema: Schema,
    id: CheckerId,
    budget: usize,
}

impl SchemaChecker {
    /// A checker enforcing `schema`, with the default step budget.
    pub fn new(schema: Schema) -> Self {
        Self {
            schema,
            id: CheckerId::new("json-schema", 1),
            budget: DEFAULT_BUDGET,
        }
    }

    /// Give up after `budget` comparisons rather than [`DEFAULT_BUDGET`].
    #[must_use]
    pub fn within(mut self, budget: usize) -> Self {
        self.budget = budget;
        self
    }

    /// Name it something more specific than `json-schema`, so a ledger says
    /// which contract was enforced.
    #[must_use]
    pub fn named(mut self, name: impl Into<String>, version: u32) -> Self {
        self.id = CheckerId::new(name, version);
        self
    }

    /// The schema being enforced.
    pub const fn schema(&self) -> &Schema {
        &self.schema
    }
}

#[async_trait]
impl Checker for SchemaChecker {
    fn id(&self) -> CheckerId {
        self.id.clone()
    }

    // No grants: the payload arrives in the condensate, so this checker reads
    // nothing, runs nothing, and needs no authority at all.

    async fn judge(&self, _order: &WorkOrder, condensate: &Condensate) -> Judgement {
        let schema_digest = self.schema.digest();

        let payload = match &condensate.payload {
            CondensatePayload::Inline(value) => value,
            // Fetching a stored payload needs a content store, which this
            // checker deliberately does not hold — see US-0806, where the
            // checker that does need one is handed it explicitly.
            CondensatePayload::Stored(digest) => {
                return Judgement::inconclusive(format!(
                    "the condensate payload is stored at {digest} rather than inline, \
                     and this checker cannot fetch it"
                ))
                .covering(schema_digest);
            }
        };

        // Both halves of the guarantee: representable as JSON, then valid.
        let json: Value = match payload.deserialized() {
            Ok(json) => json,
            Err(e) => {
                return Judgement::inconclusive(format!(
                    "the condensate payload is not representable as JSON: {e}"
                ))
                .covering(schema_digest);
            }
        };

        let judgement = match self.schema.validate_within(&json, self.budget) {
            Validation::Valid => Judgement::pass(),
            Validation::Invalid(violations) => {
                let listed: Vec<String> = violations.iter().map(Violation::render).collect();
                Judgement::fail(format!(
                    "{} schema violation{}: {}",
                    listed.len(),
                    if listed.len() == 1 { "" } else { "s" },
                    listed.join("; ")
                ))
            }
            Validation::Exhausted { budget } => Judgement::inconclusive(format!(
                "validation did not finish within {budget} comparisons"
            )),
        };

        judgement.covering(schema_digest)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sign::LocalKey;
    use boundgraph_core::{Digest, GrantSet, Intent, ProposalOrigin, SchemaId, SlicerId, Verdict};
    use serde_json::json;

    fn schema(value: Value) -> Schema {
        Schema::compile(&value).expect("test schema should compile")
    }

    fn order() -> WorkOrder {
        WorkOrder::new(
            Intent::new("report.emit", &"summarise the findings").unwrap(),
            vec![],
            GrantSet::EMPTY,
            ProposalOrigin::new("claude-opus-5", [], Digest::of(b"prompt")),
        )
    }

    fn condensate(payload: &Value) -> Condensate {
        Condensate::new(
            SchemaId::new("boundgraph.json.v1", 1),
            CondensatePayload::inline(payload).unwrap(),
            Digest::of(b"the raw response"),
            400,
            false,
            SlicerId::new("json", 1),
        )
    }

    fn violations(schema: &Schema, value: &Value) -> Vec<Violation> {
        match schema.validate(value) {
            Validation::Invalid(v) => v,
            other => panic!("expected violations, got {other:?}"),
        }
    }

    // ── validation ──────────────────────────────────────────────────────────

    #[test]
    fn a_matching_payload_is_valid() {
        let s = schema(json!({
            "type": "object",
            "properties": { "name": { "type": "string" }, "count": { "type": "integer" } },
            "required": ["name"],
        }));

        assert!(
            s.validate(&json!({ "name": "auth.rs", "count": 3 }))
                .is_valid()
        );
    }

    /// The criterion: a failure names the failing path and the constraint.
    #[test]
    fn a_violation_names_the_path_and_the_constraint() {
        let s = schema(json!({
            "type": "object",
            "properties": {
                "items": { "type": "array", "items": { "type": "string" } },
            },
        }));

        let found = violations(&s, &json!({ "items": ["ok", "also ok", 42] }));

        assert_eq!(found.len(), 1);
        assert_eq!(found[0].path, "/items/2");
        assert_eq!(found[0].constraint, "type");
        assert!(found[0].detail.contains("expected string"));
        assert!(found[0].detail.contains("found number"));
    }

    #[test]
    fn a_missing_required_property_names_it() {
        let s = schema(json!({ "type": "object", "required": ["email"] }));
        let found = violations(&s, &json!({ "name": "alex" }));

        assert_eq!(found[0].constraint, "required");
        assert!(found[0].detail.contains("email"));
    }

    #[test]
    fn a_property_name_with_a_slash_is_escaped_in_the_pointer() {
        let s = schema(json!({
            "type": "object",
            "properties": { "a/b": { "type": "string" } },
        }));

        let found = violations(&s, &json!({ "a/b": 1 }));
        assert_eq!(found[0].path, "/a~1b");
    }

    #[test]
    fn the_root_renders_as_root_rather_than_an_empty_path() {
        let s = schema(json!({ "type": "string" }));
        let found = violations(&s, &json!(1));

        assert_eq!(found[0].path, "");
        assert!(found[0].render().starts_with("(root)"));
    }

    #[test]
    fn every_supported_keyword_is_enforced() {
        let cases: &[(Value, Value, &str)] = &[
            (json!({ "type": "string" }), json!(1), "type"),
            (json!({ "enum": ["a", "b"] }), json!("c"), "enum"),
            (json!({ "const": 7 }), json!(8), "const"),
            (
                json!({ "type": "object", "required": ["x"] }),
                json!({}),
                "required",
            ),
            (
                json!({ "type": "object", "properties": { "x": true }, "additionalProperties": false }),
                json!({ "y": 1 }),
                "additionalProperties",
            ),
            (
                json!({ "type": "object", "minProperties": 2 }),
                json!({ "x": 1 }),
                "minProperties",
            ),
            (
                json!({ "type": "object", "maxProperties": 1 }),
                json!({ "x": 1, "y": 2 }),
                "maxProperties",
            ),
            (
                json!({ "type": "array", "minItems": 2 }),
                json!([1]),
                "minItems",
            ),
            (
                json!({ "type": "array", "maxItems": 1 }),
                json!([1, 2]),
                "maxItems",
            ),
            (
                json!({ "type": "array", "uniqueItems": true }),
                json!([1, 1]),
                "uniqueItems",
            ),
            (json!({ "minimum": 5 }), json!(4), "minimum"),
            (json!({ "maximum": 5 }), json!(6), "maximum"),
            (
                json!({ "exclusiveMinimum": 5 }),
                json!(5),
                "exclusiveMinimum",
            ),
            (
                json!({ "exclusiveMaximum": 5 }),
                json!(5),
                "exclusiveMaximum",
            ),
            (json!({ "multipleOf": 3 }), json!(7), "multipleOf"),
            (json!({ "minLength": 3 }), json!("ab"), "minLength"),
            (json!({ "maxLength": 1 }), json!("ab"), "maxLength"),
            (
                json!({ "allOf": [{ "type": "string" }, { "minLength": 5 }] }),
                json!("abc"),
                "minLength",
            ),
            (
                json!({ "anyOf": [{ "type": "string" }, { "type": "number" }] }),
                json!(true),
                "anyOf",
            ),
            (
                json!({ "oneOf": [{ "type": "number" }, { "type": "integer" }] }),
                json!(2),
                "oneOf",
            ),
            (json!({ "not": { "type": "string" } }), json!("x"), "not"),
        ];

        for (schema_value, instance, constraint) in cases {
            let found = violations(&schema(schema_value.clone()), instance);
            assert!(
                found.iter().any(|v| v.constraint == *constraint),
                "{schema_value} over {instance} should report `{constraint}`, got {found:?}"
            );
        }
    }

    #[test]
    fn integer_accepts_a_whole_number_written_as_a_float() {
        let s = schema(json!({ "type": "integer" }));
        assert!(s.validate(&json!(2.0)).is_valid());
        assert!(!s.validate(&json!(2.5)).is_valid());
    }

    #[test]
    fn string_lengths_count_characters_not_bytes() {
        let s = schema(json!({ "type": "string", "maxLength": 3 }));
        assert!(!s.validate(&json!("héllo")).is_valid());
        assert!(s.validate(&json!("héo")).is_valid());
    }

    #[test]
    fn multiple_of_tolerates_ordinary_decimal_arithmetic() {
        let s = schema(json!({ "multipleOf": 0.1 }));
        assert!(s.validate(&json!(0.3)).is_valid());
        assert!(!s.validate(&json!(0.35)).is_valid());
    }

    #[test]
    fn a_boolean_schema_accepts_or_refuses_everything() {
        assert!(
            schema(json!(true))
                .validate(&json!({ "anything": 1 }))
                .is_valid()
        );
        assert!(!schema(json!(false)).validate(&json!(null)).is_valid());
    }

    /// A wrong type makes every other keyword meaningless, so reporting them
    /// too would bury the one violation that matters.
    #[test]
    fn a_type_mismatch_does_not_cascade() {
        let s = schema(json!({
            "type": "object",
            "required": ["a", "b", "c"],
            "minProperties": 3,
        }));

        let found = violations(&s, &json!("not an object at all"));
        assert_eq!(found.len(), 1);
        assert_eq!(found[0].constraint, "type");
    }

    #[test]
    fn every_violation_is_reported_not_just_the_first() {
        let s = schema(json!({
            "type": "object",
            "properties": { "a": { "type": "string" }, "b": { "type": "string" } },
            "required": ["c"],
        }));

        let found = violations(&s, &json!({ "a": 1, "b": 2 }));
        assert_eq!(found.len(), 3, "two type errors and one missing property");
    }

    /// A failed alternative is how alternation works, not something the author
    /// needs to read.
    #[test]
    fn a_failed_any_of_branch_is_not_reported_separately() {
        let s = schema(json!({ "anyOf": [{ "type": "string" }, { "type": "number" }] }));
        let found = violations(&s, &json!(true));

        assert_eq!(found.len(), 1);
        assert_eq!(found[0].constraint, "anyOf");
    }

    // ── compiling ───────────────────────────────────────────────────────────

    /// The criterion behind the whole subset decision: an unimplemented
    /// keyword is refused, never silently skipped.
    #[test]
    fn an_unsupported_keyword_is_refused_rather_than_ignored() {
        for keyword in [
            "$ref",
            "pattern",
            "patternProperties",
            "format",
            "if",
            "contains",
            "propertyNames",
            "prefixItems",
        ] {
            let err = Schema::compile(&json!({ keyword: "whatever" })).unwrap_err();
            let SchemaError::UnsupportedKeyword { keyword: named, .. } = err else {
                panic!("`{keyword}` should be refused, got {err:?}");
            };
            assert_eq!(named, keyword);
        }
    }

    #[test]
    fn an_unsupported_keyword_is_found_however_deeply_it_is_nested() {
        let err = Schema::compile(&json!({
            "type": "object",
            "properties": { "user": { "properties": { "email": { "format": "email" } } } },
        }))
        .unwrap_err();

        let SchemaError::UnsupportedKeyword { path, keyword } = err else {
            panic!("expected UnsupportedKeyword, got {err:?}");
        };
        assert_eq!(keyword, "format");
        assert!(path.contains("user"), "the path should locate it: {path}");
    }

    #[test]
    fn annotations_are_skipped_rather_than_refused() {
        let s = Schema::compile(&json!({
            "$schema": "https://json-schema.org/draft/2020-12/schema",
            "title": "A finding",
            "description": "what the step found",
            "type": "object",
        }));
        assert!(s.is_ok(), "{s:?}");
    }

    #[test]
    fn a_malformed_keyword_says_what_it_should_have_been() {
        let err = Schema::compile(&json!({ "required": "email" })).unwrap_err();
        let SchemaError::MalformedKeyword {
            keyword, expected, ..
        } = err
        else {
            panic!("expected MalformedKeyword, got {err:?}");
        };
        assert_eq!(keyword, "required");
        assert!(expected.contains("array"));
    }

    #[test]
    fn an_unknown_type_name_is_refused() {
        assert!(Schema::compile(&json!({ "type": "struct" })).is_err());
    }

    #[test]
    fn a_multiple_of_zero_is_refused_rather_than_dividing_by_it() {
        assert!(Schema::compile(&json!({ "multipleOf": 0 })).is_err());
    }

    #[test]
    fn something_that_is_not_a_schema_at_all_is_refused() {
        let err = Schema::compile(&json!(["a", "list"])).unwrap_err();
        assert!(matches!(err, SchemaError::NotASchema { .. }));
    }

    #[test]
    fn a_schema_nested_past_the_depth_limit_is_refused() {
        let mut value = json!({ "type": "string" });
        for _ in 0..(MAX_DEPTH + 2) {
            value = json!({ "type": "object", "properties": { "next": value } });
        }

        assert!(matches!(
            Schema::compile(&value),
            Err(SchemaError::TooDeep { .. })
        ));
    }

    #[test]
    fn schema_text_that_is_not_json_is_a_typed_error() {
        assert!(matches!(
            Schema::parse("{not json"),
            Err(SchemaError::NotJson(_))
        ));
    }

    // ── content addressing ──────────────────────────────────────────────────

    /// The criterion: the schema is content-addressed.
    #[test]
    fn the_same_schema_always_has_the_same_digest() {
        let a = schema(json!({ "type": "object", "required": ["x"] }));
        let b = Schema::parse(r#"{"required": ["x"], "type": "object"}"#).unwrap();

        assert_eq!(a.digest(), b.digest(), "key order is not part of a schema");
    }

    #[test]
    fn a_different_constraint_is_a_different_digest() {
        let a = schema(json!({ "type": "object", "required": ["x"] }));
        let b = schema(json!({ "type": "object", "required": ["y"] }));

        assert_ne!(a.digest(), b.digest());
    }

    // ── bounded validation ──────────────────────────────────────────────────

    /// The criterion: validation is bounded and cannot hang.
    #[test]
    fn running_out_of_budget_is_neither_a_pass_nor_a_failure() {
        let s = schema(json!({ "type": "array", "items": { "type": "integer" } }));
        let big = Value::Array((0..10_000).map(|n| json!(n)).collect());

        let outcome = s.validate_within(&big, 50);

        assert_eq!(outcome, Validation::Exhausted { budget: 50 });
        assert!(!outcome.is_valid());
    }

    /// Recursion follows the schema, not the instance. A two-level schema walks
    /// two levels however deeply the payload nests, which is why refusing
    /// `$ref` is a termination argument and not merely a scope decision.
    #[test]
    fn a_deeply_nested_instance_is_walked_only_as_deep_as_the_schema() {
        let s = schema(json!({
            "type": "object",
            "additionalProperties": { "type": "object" },
        }));

        let mut instance = json!({});
        for _ in 0..1_000 {
            instance = json!({ "next": instance });
        }

        // Two levels of descent, so a handful of steps — not one per level.
        let outcome = s.validate_within(&instance, 16);
        assert!(outcome.is_valid(), "{outcome:?}");
    }

    #[test]
    fn nested_alternation_over_a_large_instance_gives_up_rather_than_grinding() {
        let branch = json!({
            "oneOf": [
                { "type": "object", "additionalProperties": { "type": "integer" } },
                { "type": "object", "additionalProperties": { "type": "number" } },
                { "type": "object", "additionalProperties": { "type": "string" } },
            ],
        });
        let s = schema(json!({ "type": "array", "items": branch }));

        let big = Value::Array((0..2_000).map(|n| json!({ "k": n })).collect());
        assert!(matches!(
            s.validate_within(&big, 200),
            Validation::Exhausted { .. }
        ));
    }

    // ── the checker ─────────────────────────────────────────────────────────

    #[tokio::test]
    async fn a_matching_payload_passes_and_the_schema_digest_is_in_the_evidence() {
        let s = schema(json!({
            "type": "object",
            "properties": { "severity": { "enum": ["low", "high"] } },
            "required": ["severity"],
        }));
        let digest = s.digest();

        let evidence = SchemaChecker::new(s)
            .check(
                &order(),
                &condensate(&json!({ "severity": "high" })),
                &LocalKey::from_seed("k", [1u8; 32]),
            )
            .await
            .unwrap();

        assert!(evidence.permits_commit());
        assert!(
            evidence.covers(&digest),
            "the evidence must name the schema it was reached against"
        );
    }

    #[tokio::test]
    async fn a_violating_payload_fails_with_the_path_in_the_reason() {
        let s = schema(json!({
            "type": "object",
            "properties": { "count": { "type": "integer" } },
        }));

        let evidence = SchemaChecker::new(s)
            .check(
                &order(),
                &condensate(&json!({ "count": "three" })),
                &LocalKey::from_seed("k", [1u8; 32]),
            )
            .await
            .unwrap();

        assert!(!evidence.permits_commit());
        let reason = evidence.verdict.reason().unwrap();
        assert!(reason.contains("/count"), "{reason}");
        assert!(reason.contains("type"), "{reason}");
    }

    #[tokio::test]
    async fn a_stored_payload_is_inconclusive_rather_than_a_pass() {
        let s = schema(json!({ "type": "object" }));
        let mut k = condensate(&json!({}));
        k.payload = CondensatePayload::Stored(Digest::of(b"elsewhere"));

        let evidence = SchemaChecker::new(s)
            .check(&order(), &k, &LocalKey::from_seed("k", [1u8; 32]))
            .await
            .unwrap();

        assert!(!evidence.permits_commit());
        assert!(matches!(evidence.verdict, Verdict::Inconclusive(_)));
    }

    #[tokio::test]
    async fn an_exhausted_budget_is_inconclusive_rather_than_a_pass() {
        let s = schema(json!({ "type": "array", "items": { "type": "integer" } }));
        let big = Value::Array((0..5_000).map(|n| json!(n)).collect());

        let evidence = SchemaChecker::new(s)
            .within(10)
            .check(
                &order(),
                &condensate(&big),
                &LocalKey::from_seed("k", [1u8; 32]),
            )
            .await
            .unwrap();

        assert!(matches!(evidence.verdict, Verdict::Inconclusive(_)));
    }

    /// The criterion: a passing verdict means a downstream step can
    /// deserialise the payload.
    #[tokio::test]
    async fn a_pass_means_the_payload_deserialises_downstream() {
        #[derive(serde::Deserialize)]
        struct Finding {
            #[allow(dead_code)]
            severity: String,
            #[allow(dead_code)]
            files: Vec<String>,
        }

        let s = schema(json!({
            "type": "object",
            "properties": {
                "severity": { "type": "string" },
                "files": { "type": "array", "items": { "type": "string" } },
            },
            "required": ["severity", "files"],
        }));

        let payload = json!({ "severity": "high", "files": ["src/auth.rs"] });
        let k = condensate(&payload);

        let evidence = SchemaChecker::new(s)
            .check(&order(), &k, &LocalKey::from_seed("k", [1u8; 32]))
            .await
            .unwrap();
        assert!(evidence.permits_commit());

        let CondensatePayload::Inline(value) = &k.payload else {
            unreachable!()
        };
        let parsed: Result<Finding, _> = value.deserialized();
        assert!(parsed.is_ok(), "a pass must mean this cannot fail");
    }

    #[tokio::test]
    async fn the_checker_needs_no_authority_at_all() {
        assert_eq!(
            SchemaChecker::new(schema(json!(true))).grants(),
            GrantSet::EMPTY
        );
    }

    #[tokio::test]
    async fn the_same_payload_always_gives_the_same_evidence() {
        let checker = SchemaChecker::new(schema(json!({ "type": "object" })));
        let (order, k, key) = (
            order(),
            condensate(&json!({})),
            LocalKey::from_seed("k", [1u8; 32]),
        );

        let first = checker.check(&order, &k, &key).await.unwrap();
        for _ in 0..8 {
            assert_eq!(checker.check(&order, &k, &key).await.unwrap(), first);
        }
    }
}