polydat_core/iteration/comprehension/

validate.rs

// Copyright 2024-2026 Jonathan Shook
// SPDX-License-Identifier: Apache-2.0

//! Validation — comprehension_forms.md §5 (V1-V9) and §5.8 (modes).
//!
//! [`validate`] is the single entry point. Walking the AST
//! bottom-up, every variant's V-axiom checks fire; any failure
//! produces a typed [`ValidationError`]. Degenerate-but-defined
//! compositions emit a [`ValidationWarning`] in Permissive mode
//! and become hard errors in Strict mode. The compile stage
//! (`CompiledComprehension::from_ast` and the `for` lowering) runs it
//! on the tree as written, before any rewrite, so acceptance is a
//! property of the source text, and every rewrite keeps an accepted
//! tree accepted.
//!
//! V4 (per-strategy input-shape contract) fires at two times
//! (comprehension_forms.md §10.7.8):
//!
//! 1. **Compile time** — this module, against the AST's static
//!    metadata-derived [`IndexFn`]. For
//!    [`crate::iteration::comprehension::eval_source::EvalClass::Static`]
//!    sources the static IndexFn equals the runtime IndexFn, so a
//!    tree accepted here is never refused at a strategy. A
//!    `ContextRequired` source (a Generator that references a name, a
//!    WorkloadParamList) is judged by what its text declares: its
//!    `cardinality_hint`, or an unknown count.
//! 2. **Strategy invocation** — the runtime evaluator
//!    ([`crate::iteration::comprehension::runtime::evaluate_indexed`])
//!    calls
//!    [`crate::iteration::comprehension::strategies::Strategy::accepts_input`]
//!    against the evaluated input's `index_fn`. For a context-required
//!    source this is the authoritative check.

use serde::{Deserialize, Serialize};

use super::ast::Comprehension;
use super::cardinality::CardinalityClass;
use super::metadata::{IndexFn, Metadata};
use super::source::Source;
use super::strategy::{StrategyName, ZipMode};

/// Validation mode (comprehension_forms.md §5.8).
///
/// `Permissive` (default) enforces V1-V9 as errors and surfaces
/// degenerate-composition warnings non-blockingly. `Strict`
/// promotes those warnings to errors.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum Mode {
    #[default]
    /// V1 through V9 are errors; degenerate compositions are warnings.
    Permissive,
    /// Degenerate compositions are errors too.
    Strict,
}

/// Result of a validation pass.
#[derive(Debug, Clone)]
pub struct ValidationReport {
    /// The warnings the pass raised.
    pub warnings: Vec<ValidationWarning>,
}

/// V-axiom violation. Each variant carries enough context to
/// produce a useful diagnostic at the call site.
#[derive(Debug, Clone, PartialEq)]
pub enum ValidationError {
    /// V1 — cartesian or zip children share a name.
    V1DuplicateName {
        /// The combinator whose children share the name.
        combinator: &'static str,
        /// The duplicated name.
        name: String,
    },

    /// V2 — union children disagree on tuple shape.
    V2ShapeMismatch {
        /// The tuple shape of the first child.
        expected: Vec<String>,
        /// The shape that differs.
        actual: Vec<String>,
    },

    /// V3 — a clause source or a filter predicate reads a name that
    /// neither the comprehension binds where it is read nor the surface
    /// it is evaluated on supplies ([`check_names`]).
    V3UnresolvedNames {
        /// Each such name, with where it is read, in tree order.
        reads: Vec<NameRead>,
    },

    /// V4 — strategy applied to an input whose metadata-derived
    /// [`IndexFn`] it cannot accept (per-strategy table in
    /// `check_strategy_input_shape`). A strategy that selects from
    /// the shape reads through the untruncated orders under it, and
    /// V5 looks through one filter layer; nested filters are
    /// rejected.
    V4InputShape {
        /// The strategy applied.
        strategy: StrategyName,
        /// Why its input's shape is unacceptable.
        reason: String,
    },

    /// V6 — non-Lex order or Strict/Truncate zip applied to an
    /// `Unbounded` discrete input.
    V6UnboundedDiscrete {
        /// The operator applied.
        operator: &'static str,
        /// The unbounded input's cardinality class.
        cardinality: CardinalityClass,
    },

    /// V7 — zip cardinality contract violated. Three sub-cases:
    /// Strict-mode mismatch, mixed-class children, or any
    /// continuous child.
    V7ZipCardinality {
        /// The zip mode in force.
        mode: ZipMode,
        /// Which contract was violated.
        reason: String,
    },

    /// V8 — continuous source requires explicit sampling OR
    /// source declares a non-integrable measure.
    V8ContinuousRequirement {
        /// What the source lacks.
        reason: String,
    },

    /// V9 — union children include a continuous or mixed-class
    /// child.
    V9UnionClassMismatch {
        /// Which child mismatches, and how.
        reason: String,
    },

    /// Strict mode (comprehension_forms.md §5.8): a degenerate composition the
    /// permissive mode only warns about.
    StrictWarning(ValidationWarning),
    /// A source that needs a scope, on the scope-less surfaces
    /// (comprehension_forms.md §9.5.2, §10.7.0): a coordinate stream
    /// binds no names, the traversal does.
    ContextRequired {
        /// The clause.
        name: String,
        /// The names its source references.
        references: Vec<String>,
    },
    /// A filter whose predicate names what its tuples do not bind, on
    /// the scope-less surfaces: a traversal resolves such a name in
    /// the scope it opens in, and a coordinate stream has none.
    PredicateContextRequired {
        /// The predicate.
        predicate: String,
        /// The names it references that its tuples do not bind.
        references: Vec<String>,
    },
    /// A context-free source that could not be evaluated, on the
    /// scope-less surfaces. Its only evaluation is the compile's, in
    /// the empty scope, so its failure is the comprehension's error
    /// rather than a clause that dispenses nothing.
    SourceFailed {
        /// The clause.
        name: String,
        /// The evaluator's message.
        message: String,
    },
}

impl std::fmt::Display for ValidationError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::V1DuplicateName { combinator, name } => {
                write!(
                    f,
                    "V1: `{combinator}` binds the name `{name}` more than once"
                )
            }
            Self::V2ShapeMismatch { expected, actual } => write!(
                f,
                "V2: tuple shape ({}) does not match ({})",
                actual.join(", "),
                expected.join(", ")
            ),
            Self::V3UnresolvedNames { reads } => write_unresolved(f, reads, false),
            Self::V4InputShape { strategy, reason } => {
                write!(
                    f,
                    "V4: strategy `{strategy:?}` cannot take this input: {reason}"
                )
            }
            Self::V6UnboundedDiscrete {
                operator,
                cardinality,
            } => write!(
                f,
                "V6: `{operator}` cannot materialize a {cardinality:?} stream"
            ),
            Self::V7ZipCardinality { mode, reason } => {
                write!(f, "V7: zip in {mode:?} mode: {reason}")
            }
            Self::V8ContinuousRequirement { reason } => {
                write!(f, "V8: continuous source: {reason}")
            }
            Self::StrictWarning(w) => write!(f, "strict mode: {w}"),
            Self::ContextRequired { name, references } => write!(
                f,
                "clause '{name}' needs a scope to bind {}; a coordinate stream has none: \n                 traverse it with `for`, which captures those names when it opens",
                references.join(", ")
            ),
            Self::PredicateContextRequired {
                predicate,
                references,
            } => write!(
                f,
                "predicate '{predicate}' needs a scope to bind {}; a coordinate stream has none: \
                 traverse it with `for`, which captures those names when it opens",
                references.join(", ")
            ),
            Self::V9UnionClassMismatch { reason } => {
                write!(f, "V9: union children differ in class: {reason}")
            }
            Self::SourceFailed { name, message } => {
                write!(f, "clause '{name}' cannot be evaluated: {message}")
            }
        }
    }
}

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

/// The V3 diagnostic over `reads`: each name with where it is read, and
/// for a bare word in a predicate, how a string and a scope's name are
/// written. `lax` says the names compile and read None
/// (comprehension_forms.md §5 V3).
fn write_unresolved(
    f: &mut std::fmt::Formatter<'_>,
    reads: &[NameRead],
    lax: bool,
) -> std::fmt::Result {
    let each: Vec<String> = reads
        .iter()
        .map(|r| format!("`{}` read by {}", r.name, r.site))
        .collect();
    write!(
        f,
        "V3: {} {} bound neither by the comprehension nor by the scope it is evaluated in",
        each.join(", "),
        if reads.len() == 1 { "is" } else { "are" }
    )?;
    if lax {
        write!(
            f,
            ", so {} None, as a name nothing binds does outside `pragma strict`",
            if reads.len() == 1 {
                "it reads"
            } else {
                "each reads"
            }
        )?;
    }
    if let Some(bare) = reads.iter().find(|r| r.bare) {
        write!(
            f,
            "; a bare word in a predicate is a name, which nothing supplies: a string in a \
             predicate is quoted, as in `\"{}\"`, and a scope's name is read as `{{{}}}`",
            bare.name, bare.name
        )?;
    }
    Ok(())
}

#[derive(Debug, Clone, PartialEq)]
/// Non-blocking warning for degenerate-but-defined compositions
/// (comprehension_forms.md §5.8).
pub enum ValidationWarning {
    /// Lattice-geometric strategy (`Extrema` / `Shells` /
    /// `Diagonal` / `Antidiagonal`) over a 1-axis input.
    /// Collapses to {first, last} or a trivial walk; usually
    /// not what the author meant.
    DegenerateGeometric {
        /// The strategy applied.
        strategy: StrategyName,
    },

    /// `Lhs` over a 1-axis input. Equivalent to `Shuffle`;
    /// two names for one behavior.
    LhsDegenerate,

    /// `filter(c, "true")`. Empty-effect filter — usually a
    /// bug-shaped predicate. The optimizer's R0a elides it.
    TriviallyTrueFilter,

    /// `filter(c, "false")`. Empty dispense sequence. If
    /// intentional, use an empty literal source; otherwise the
    /// predicate is bug-shaped.
    TriviallyFalseFilter,

    /// A clause whose source is *provably* empty: its cardinality
    /// is `Bounded(0)` (`x in []`, `x in 5..5`, a context-free
    /// generator the compile evaluated to nothing). The clause is
    /// well-formed and an empty stream is a legal value, so this
    /// is degenerate rather than wrong — but it empties every
    /// cartesian it takes part in, so a whole traversal dispenses
    /// nothing and usually that is a typo in the source.
    ///
    /// A source whose count is not known at construction (an
    /// interpolated call, a parameter without a length) is
    /// `Unbounded`, never `Bounded(0)`, so it cannot reach here:
    /// emptiness it discovers at evaluation is reported by the
    /// per-clause yields instead ([`super::runtime::ClauseYield`]).
    EmptySource {
        /// The clause's element name.
        var: String,
        /// The source's canonical text, when it has one.
        source: Option<String>,
    },

    /// Singleton variant of a combinator: `zip([c], _)`,
    /// `cartesian(c)`, `union(c)`. Identity per
    /// comprehension_forms.md §4.2 I1-I3; the optimizer's R0a
    /// elides it.
    SingletonCombinator {
        /// The combinator with one child.
        combinator: &'static str,
    },

    /// V3 outside strictness (§5 V3): names the comprehension reads
    /// that neither it binds nor the surface supplies. The comprehension
    /// compiles, and each such name reads None. A `Strict` compile
    /// refuses it as [`ValidationError::V3UnresolvedNames`].
    UnresolvedNames {
        /// Each such name, with where it is read, in tree order.
        reads: Vec<NameRead>,
    },
}

impl std::fmt::Display for ValidationWarning {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::DegenerateGeometric { strategy } => write!(
                f,
                "`{strategy:?}` over a one-axis input collapses to its ends; use `Lex` with a truncation, or restate over a multi-axis cartesian"
            ),
            Self::LhsDegenerate => {
                write!(f, "`Lhs` over a one-axis input is `Shuffle`; say `Shuffle`")
            }
            Self::TriviallyTrueFilter => write!(f, "the filter is always true; drop it"),
            Self::TriviallyFalseFilter => write!(
                f,
                "the filter is always false; the comprehension dispenses nothing"
            ),
            Self::SingletonCombinator { combinator } => write!(
                f,
                "`{combinator}` over one child is that child; the wrapper adds nothing"
            ),
            Self::EmptySource { var, source } => match source {
                Some(text) => write!(
                    f,
                    "`{var} in {text}` has no values; every composition it takes part in dispenses nothing"
                ),
                None => write!(
                    f,
                    "`{var}` has no values; every composition it takes part in dispenses nothing"
                ),
            },
            Self::UnresolvedNames { reads } => write_unresolved(f, reads, true),
        }
    }
}

/// Where a comprehension reads a name (comprehension_forms.md §5 V3).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReadSite {
    /// A clause's source.
    Source {
        /// The clause's element name.
        clause: String,
        /// The source's text.
        source: String,
    },
    /// A filter's predicate.
    Predicate {
        /// The predicate's text.
        predicate: String,
    },
}

impl std::fmt::Display for ReadSite {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Source { clause, source } => write!(f, "clause '{clause}' in `{source}`"),
            Self::Predicate { predicate } => write!(f, "predicate `{predicate}`"),
        }
    }
}

/// A name a comprehension reads where it does not bind it
/// ([`outer_reads`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NameRead {
    /// The name.
    pub name: String,
    /// Where it is read.
    pub site: ReadSite,
    /// Whether it is a bare word in a predicate. A predicate reads a
    /// scope only through `{name}`, so no surface supplies a bare word.
    pub bare: bool,
}

/// What a consumption surface supplies to the names a comprehension's
/// sources and predicates read (comprehension_forms.md §5 V3, §9.5).
#[derive(Clone, Copy)]
pub enum Surface<'a> {
    /// A coordinate stream or a scoped-kernel stream: no name.
    Stream,
    /// A `for` traversal: the names of the scope it opens in, which it
    /// captures when it opens. The function answers whether that scope
    /// has a name.
    Traversal(&'a dyn Fn(&str) -> bool),
}

impl Surface<'_> {
    /// Whether the surface supplies `name`.
    pub fn supplies(&self, name: &str) -> bool {
        match self {
            Surface::Stream => false,
            Surface::Traversal(has) => has(name),
        }
    }
}

/// Every name `c` reads where it does not bind it, in tree order, once
/// per place it is read (comprehension_forms.md §5 V3).
///
/// A clause's source reads with the names the clauses before it in every
/// enclosing cartesian bind in front of it, the dependent product of
/// §3.2; a zip's and a union's operands see only what their enclosing
/// cartesians bind. A filter's predicate reads each tuple of its input,
/// so its `{name}` elements are bound by the names that input binds, and
/// a bare word in it is never bound. What remains is what the scope the
/// comprehension is evaluated in must supply.
pub fn outer_reads(c: &Comprehension) -> Vec<NameRead> {
    let mut out = Vec::new();
    collect_outer_reads(c, &mut Vec::new(), &mut out);
    out
}

fn collect_outer_reads(c: &Comprehension, before: &mut Vec<String>, out: &mut Vec<NameRead>) {
    match c {
        Comprehension::Clause { name, source } => {
            for read in source.referenced_names() {
                if !before.contains(&read) {
                    out.push(NameRead {
                        name: read,
                        site: ReadSite::Source {
                            clause: name.clone(),
                            source: source.to_text().unwrap_or_else(|| "<source>".into()),
                        },
                        bare: false,
                    });
                }
            }
        }
        Comprehension::Cartesian { children } => {
            let depth = before.len();
            for child in children {
                collect_outer_reads(child, before, out);
                before.extend(child.coordinate_names());
            }
            before.truncate(depth);
        }
        Comprehension::Zip { children, .. } | Comprehension::Union { children } => {
            for child in children {
                collect_outer_reads(child, before, out);
            }
        }
        Comprehension::Filter { child, predicate } => {
            collect_outer_reads(child, before, out);
            let bound = child.coordinate_names();
            let reads = polydat_grammar::comprehension::predicate::predicate_reads(predicate);
            let site = || ReadSite::Predicate {
                predicate: predicate.clone(),
            };
            for name in reads.elements {
                if !bound.contains(&name) {
                    out.push(NameRead {
                        name,
                        site: site(),
                        bare: false,
                    });
                }
            }
            for name in reads.bare {
                out.push(NameRead {
                    name,
                    site: site(),
                    bare: true,
                });
            }
        }
        Comprehension::Order { child, .. } => collect_outer_reads(child, before, out),
    }
}

/// Every name a clause source or a filter predicate of `c` reads that
/// neither the comprehension binds where it is read nor `surface`
/// supplies, with where it is read, in tree order
/// (comprehension_forms.md §5 V3). Under `pragma strict` each is an
/// error ([`check_names`]); outside it each is a warning, and the name
/// reads None.
pub fn unresolved_names(c: &Comprehension, surface: Surface<'_>) -> Vec<NameRead> {
    outer_reads(c)
        .into_iter()
        .filter(|r| r.bare || !surface.supplies(&r.name))
        .collect()
}

/// V3 under strictness (comprehension_forms.md §5): every name a clause
/// source or a filter predicate of `c` reads is bound by the
/// comprehension where it is read, or supplied by `surface`. The error
/// names each name that is neither, with where it is read
/// ([`unresolved_names`]).
pub fn check_names(c: &Comprehension, surface: Surface<'_>) -> Result<(), ValidationError> {
    let reads = unresolved_names(c, surface);
    if reads.is_empty() {
        Ok(())
    } else {
        Err(ValidationError::V3UnresolvedNames { reads })
    }
}

/// Validate a comprehension AST as written (comprehension_forms.md §5).
///
/// V3 depends on the surface the comprehension is evaluated on, so it is
/// [`check_names`], which each surface's compile calls with what that
/// surface supplies; this function judges every other axiom.
///
/// In `Permissive` mode, V1-V9 errors abort with a typed
/// [`ValidationError`] and degenerate-composition warnings
/// accumulate into the returned [`ValidationReport`]. In
/// `Strict` mode, the first warning is promoted to an error.
pub fn validate(c: &Comprehension, mode: Mode) -> Result<ValidationReport, ValidationError> {
    let mut report = ValidationReport {
        warnings: Vec::new(),
    };
    visit(c, &mut report)?;
    if mode == Mode::Strict
        && let Some(warning) = report.warnings.first()
    {
        // Strict-mode promotion (§5.8): the first degenerate
        // composition is the error.
        return Err(ValidationError::StrictWarning(warning.clone()));
    }
    Ok(report)
}

fn visit(c: &Comprehension, report: &mut ValidationReport) -> Result<(), ValidationError> {
    // Bottom-up: validate children first so each node sees
    // already-well-formed operands (comprehension_forms.md §4.1 C2).
    for child in c.children() {
        visit(child, report)?;
    }

    match c {
        Comprehension::Clause { name, source } => visit_clause(name, source, report),
        Comprehension::Cartesian { children } => visit_cartesian(children, report),
        Comprehension::Zip { children, mode } => visit_zip(children, *mode, report),
        Comprehension::Union { children } => visit_union(children, report),
        Comprehension::Filter { predicate, .. } => visit_filter(predicate, report),
        Comprehension::Order {
            child,
            strategy,
            truncation,
            ..
        } => visit_order(child, *strategy, *truncation, report),
    }
}

fn visit_clause(
    name: &str,
    source: &Source,
    report: &mut ValidationReport,
) -> Result<(), ValidationError> {
    // Degenerate composition (§5.8): a source the construction
    // can already count, and the count is zero. Read off the
    // cardinality algebra rather than matched shape by shape, so a
    // source whose count is unknown is `Unbounded` and says nothing
    // here.
    if matches!(
        source.cardinality(),
        crate::iteration::comprehension::CardinalityClass::Bounded(0)
    ) {
        report.warnings.push(ValidationWarning::EmptySource {
            var: name.to_string(),
            source: source.to_text(),
        });
    }
    // V8 source-side check: continuous source must have an
    // integrable measure. Unbounded + Uniform is the canonical
    // failure case.
    if let Source::ContinuousInterval { interval, measure } = source
        && !measure.is_integrable(std::slice::from_ref(interval))
    {
        let _ = report; // no warning here; this is a hard error
        return Err(ValidationError::V8ContinuousRequirement {
            reason: format!(
                "continuous source has non-integrable measure: \
                 interval [{}, {}] + {:?}",
                interval.lo, interval.hi, measure
            ),
        });
    }
    // V8, on the declared measure: a named distribution's parameters
    // must be the measure's own (`MeasureName::parameter_names`), or
    // absent for the standard ones.
    if let Source::Distribution {
        distribution,
        params,
        ..
    } = source
        && let Err(reason) = distribution.resolve_params(params)
    {
        return Err(ValidationError::V8ContinuousRequirement { reason });
    }
    Ok(())
}

fn visit_cartesian(
    children: &[Comprehension],
    report: &mut ValidationReport,
) -> Result<(), ValidationError> {
    check_disjoint_names("cartesian", children)?;
    if children.len() == 1 {
        report
            .warnings
            .push(ValidationWarning::SingletonCombinator {
                combinator: "cartesian",
            });
    }
    Ok(())
}

fn visit_zip(
    children: &[Comprehension],
    mode: ZipMode,
    report: &mut ValidationReport,
) -> Result<(), ValidationError> {
    check_disjoint_names("zip", children)?;

    // V7: discrete-only children. We use the leaf-clause check
    // here as the cheapest reliable proxy: walk to find any
    // continuous source in any child.
    for child in children {
        if contains_continuous_source(child) {
            return Err(ValidationError::V7ZipCardinality {
                mode,
                reason: "zip children must all be discrete; \
                         a continuous source was found"
                    .to_string(),
            });
        }
    }

    if children.len() == 1 {
        report
            .warnings
            .push(ValidationWarning::SingletonCombinator { combinator: "zip" });
    }

    // V6: Strict/Truncate require bounded children, read off each
    // child's metadata, so a child the optimizer unwraps is judged as
    // it is judged wrapped.
    if matches!(mode, ZipMode::Strict | ZipMode::Truncate) {
        for child in children {
            let card = child.metadata().cardinality;
            if matches!(card, CardinalityClass::Unbounded) {
                return Err(ValidationError::V6UnboundedDiscrete {
                    operator: "zip",
                    cardinality: card,
                });
            }
        }
    }

    Ok(())
}

fn visit_union(
    children: &[Comprehension],
    report: &mut ValidationReport,
) -> Result<(), ValidationError> {
    // V9 first: all children must be discrete.
    for child in children {
        if contains_continuous_source(child) {
            return Err(ValidationError::V9UnionClassMismatch {
                reason: "union children must all be discrete; \
                         a continuous source was found"
                    .to_string(),
            });
        }
    }

    // V2: identical tuple shape (same names, same order).
    if let Some(first) = children.first() {
        let expected = first.coordinate_names();
        for sibling in &children[1..] {
            let actual = sibling.coordinate_names();
            if actual != expected {
                return Err(ValidationError::V2ShapeMismatch { expected, actual });
            }
        }
    }

    if children.len() == 1 {
        report
            .warnings
            .push(ValidationWarning::SingletonCombinator {
                combinator: "union",
            });
    }
    Ok(())
}

fn visit_filter(predicate: &str, report: &mut ValidationReport) -> Result<(), ValidationError> {
    // V3, the predicate's names against its input's tuples and the
    // surface's scope, is `check_names`.

    // §5.8 warnings for trivially-true / trivially-false
    // predicates: the literal texts "true" and "false" are the
    // bug-shaped cases this check recognizes.
    let trimmed = predicate.trim();
    if trimmed.eq_ignore_ascii_case("true") {
        report.warnings.push(ValidationWarning::TriviallyTrueFilter);
    } else if trimmed.eq_ignore_ascii_case("false") {
        report
            .warnings
            .push(ValidationWarning::TriviallyFalseFilter);
    }

    Ok(())
}

fn visit_order(
    child: &Comprehension,
    strategy: StrategyName,
    truncation: Option<u64>,
    report: &mut ValidationReport,
) -> Result<(), ValidationError> {
    // V4: per-strategy input-shape contract using the metadata
    // algebra. A strategy that selects from the shape reads through
    // the untruncated orders under it (§7.4 O1), and V5 looks through
    // one filter layer: the metadata is computed against what remains.
    let metadata_target = match super::strategies::ranked_filter(child, strategy) {
        Some((input, _)) => input,
        None => super::strategies::shape_input(child, strategy),
    };

    // V5 looks through one filter layer only: nested filters under a
    // non-Lex order are refused as written, before R6 could fold them.
    if !matches!(strategy, StrategyName::Lex)
        && matches!(metadata_target, Comprehension::Filter { .. })
    {
        return Err(ValidationError::V4InputShape {
            strategy,
            reason: "non-Lex strategy applied to nested filters; \
                     write them as one filter, `where p && q` \
                     (comprehension_forms.md §5 V5)"
                .to_string(),
        });
    }

    let target_metadata = metadata_target.metadata();
    check_strategy_input_shape(strategy, &target_metadata, report)?;

    // V6: non-Lex strategy requires bounded input. Now via
    // metadata cardinality (not the source-only direct check).
    if !matches!(strategy, StrategyName::Lex)
        && matches!(target_metadata.cardinality, CardinalityClass::Unbounded)
    {
        return Err(ValidationError::V6UnboundedDiscrete {
            operator: "order",
            cardinality: target_metadata.cardinality.clone(),
        });
    }

    // V8: continuous input requires sampling — wrapped order
    // with finite truncation is the discharge mechanism.
    let is_continuous = matches!(
        target_metadata.cardinality,
        CardinalityClass::Continuous { .. }
            | CardinalityClass::ContinuousAtMost { .. }
            | CardinalityClass::Hybrid(_)
    );
    if is_continuous {
        if truncation.is_none() {
            return Err(ValidationError::V8ContinuousRequirement {
                reason: "continuous comprehension requires order(_, \
                         sampling-strategy, Some(n)) with finite \
                         truncation"
                    .to_string(),
            });
        }
        if matches!(strategy, StrategyName::Lex) {
            return Err(ValidationError::V8ContinuousRequirement {
                reason: "Lex does not sample continuous inputs; use \
                         Halton / Sobol / Lhs / Shuffle / Extrema"
                    .to_string(),
            });
        }
    }

    Ok(())
}

/// Per-strategy V4 input-shape check using the metadata
/// algebra's `IndexFn` variants. Implements the per-strategy
/// table from comprehension_forms.md §3.6:
///
/// An order's output is a one-axis `Lattice` of its selection, so every
/// row below accepts it.
///
/// | Strategy | Accepted IndexFn |
/// |---|---|
/// | Lex | any (incl. None) |
/// | ReverseLex | any non-None discrete |
/// | Shuffle, Halton, Sobol | any non-None |
/// | Lhs | any non-None (Lattice multi-axis = native; 1-axis = degenerate) |
/// | Extrema | any non-None (Lattice ≥2 = native; 1-axis or non-Lattice = degenerate or continuous box) |
/// | Shells, Diagonal, Antidiagonal | non-None discrete only |
fn check_strategy_input_shape(
    strategy: StrategyName,
    metadata: &Metadata,
    report: &mut ValidationReport,
) -> Result<(), ValidationError> {
    // Lex accepts anything including None.
    if matches!(strategy, StrategyName::Lex) {
        return Ok(());
    }

    let idx = match &metadata.index_addressable {
        Some(i) => i,
        None => {
            return Err(ValidationError::V4InputShape {
                strategy,
                reason: "input has no closed-form index function \
                         (a filter's output, a dependent cartesian, or \
                         a truncated Lex order over one)"
                    .to_string(),
            });
        }
    };

    // Continuous / Hybrid acceptance per strategy.
    let has_continuous = idx.has_continuous_axis();
    if has_continuous {
        match strategy {
            // Index-sampling that accepts continuous.
            StrategyName::Shuffle
            | StrategyName::Halton
            | StrategyName::Sobol
            | StrategyName::Lhs => {}
            // Extrema accepts continuous boxes (§3.6).
            StrategyName::Extrema => {}
            // Everything else rejects continuous.
            StrategyName::ReverseLex
            | StrategyName::Shells
            | StrategyName::Diagonal
            | StrategyName::Antidiagonal => {
                return Err(ValidationError::V4InputShape {
                    strategy,
                    reason: format!("{} does not accept continuous input", strategy.as_str()),
                });
            }
            StrategyName::Lex => unreachable!("Lex handled above"),
        }
        // Continuous + Lhs/Extrema on 1-D is the same kind of
        // degenerate as discrete 1-D; emit a warning.
        if matches!(strategy, StrategyName::Lhs | StrategyName::Extrema) {
            let dim = continuous_dim(idx);
            if dim < 2 {
                if matches!(strategy, StrategyName::Lhs) {
                    report.warnings.push(ValidationWarning::LhsDegenerate);
                } else {
                    report
                        .warnings
                        .push(ValidationWarning::DegenerateGeometric { strategy });
                }
            }
        }
        return Ok(());
    }

    // Discrete path.
    // Lattice-geometric strategies require Lattice IndexFn.
    if strategy.is_lattice_geometric() {
        match idx {
            IndexFn::Lattice { axis_sizes } => {
                if axis_sizes.len() < 2 {
                    report
                        .warnings
                        .push(ValidationWarning::DegenerateGeometric { strategy });
                }
            }
            // Concatenation (union) is V4-rejected for lattice-
            // geometric — heterogeneous index space.
            IndexFn::Concatenation { .. } => {
                return Err(ValidationError::V4InputShape {
                    strategy,
                    reason: format!(
                        "{} requires a cartesian input; got union",
                        strategy.as_str()
                    ),
                });
            }
            // Lockstep / Modular (zip) — 1-D index space; these
            // strategies in 1-D collapse degenerately, but per
            // §5.8 we allow with warning rather than rejecting.
            IndexFn::Lockstep { .. } | IndexFn::Modular { .. } => {
                report
                    .warnings
                    .push(ValidationWarning::DegenerateGeometric { strategy });
            }
            // Unreachable: continuous handled above.
            IndexFn::Continuous { .. } | IndexFn::Hybrid { .. } => unreachable!(),
        }
        return Ok(());
    }

    // Lhs on discrete: degenerate over 1-axis Lattice / Lockstep / Modular.
    if matches!(strategy, StrategyName::Lhs) {
        match idx {
            IndexFn::Lattice { axis_sizes } if axis_sizes.len() < 2 => {
                report.warnings.push(ValidationWarning::LhsDegenerate);
            }
            IndexFn::Lockstep { .. } | IndexFn::Modular { .. } => {
                report.warnings.push(ValidationWarning::LhsDegenerate);
            }
            _ => {}
        }
    }

    // ReverseLex / Shuffle / Halton / Sobol on any non-None
    // discrete IndexFn: always accepted (no degeneracy
    // warning).
    Ok(())
}

fn continuous_dim(idx: &IndexFn) -> usize {
    match idx {
        IndexFn::Continuous { intervals, .. } => intervals.len(),
        IndexFn::Hybrid {
            discrete_axes,
            continuous_axes,
            ..
        } => discrete_axes.len() + continuous_axes.len(),
        _ => 0,
    }
}

fn check_disjoint_names(
    combinator: &'static str,
    children: &[Comprehension],
) -> Result<(), ValidationError> {
    let mut seen: Vec<String> = Vec::new();
    for child in children {
        for name in child.coordinate_names() {
            if seen.contains(&name) {
                return Err(ValidationError::V1DuplicateName { combinator, name });
            }
            seen.push(name);
        }
    }
    Ok(())
}

fn contains_continuous_source(c: &Comprehension) -> bool {
    match c {
        Comprehension::Clause { source, .. } => source.is_continuous(),
        Comprehension::Cartesian { children }
        | Comprehension::Zip { children, .. }
        | Comprehension::Union { children } => children.iter().any(contains_continuous_source),
        Comprehension::Filter { child, .. } | Comprehension::Order { child, .. } => {
            contains_continuous_source(child)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::iteration::comprehension::cardinality::{Interval, ProductMeasure};
    use crate::iteration::comprehension::source::{LiteralValue, Source};

    fn clause(name: &str, vs: &[i64]) -> Comprehension {
        Comprehension::clause(
            name,
            Source::Literal {
                values: vs.iter().map(|n| LiteralValue::Int(*n)).collect(),
            },
        )
    }

    fn empty_warnings(c: &Comprehension) -> Vec<String> {
        validate(c, Mode::Permissive)
            .expect("an empty source is degenerate, not invalid")
            .warnings
            .into_iter()
            .filter_map(|w| match w {
                ValidationWarning::EmptySource { var, .. } => Some(var),
                _ => None,
            })
            .collect()
    }

    /// A source the construction can already count as empty is a
    /// degenerate composition, warned and not refused.
    #[test]
    fn a_provably_empty_source_warns() {
        assert_eq!(empty_warnings(&clause("x", &[])), ["x"]);
        assert_eq!(
            empty_warnings(&Comprehension::clause(
                "k",
                Source::IntRange {
                    lo: 5,
                    hi: 5,
                    step: 1,
                },
            )),
            ["k"],
            "a half-open range over no values is the same fact"
        );
    }

    /// The rule reads the cardinality algebra, so a source whose count
    /// is not known at construction says nothing here. Its emptiness,
    /// if any, is the evaluator's to report.
    #[test]
    fn a_source_of_unknown_count_does_not_warn() {
        let generator = Comprehension::clause(
            "g",
            Source::Generator {
                expr: "matching_profiles('a')".into(),
                cardinality_hint: None,
            },
        );
        assert!(empty_warnings(&generator).is_empty());

        let param = Comprehension::clause(
            "p",
            Source::WorkloadParamList {
                name: "sizes".into(),
                len_hint: None,
            },
        );
        assert!(empty_warnings(&param).is_empty());
    }

    /// A generator the compile did count, and counted as zero, is
    /// known at construction and warns like a literal.
    #[test]
    fn a_generator_counted_as_zero_warns() {
        let counted = Comprehension::clause(
            "g",
            Source::Generator {
                expr: "matching_profiles('nope')".into(),
                cardinality_hint: Some(0),
            },
        );
        assert_eq!(empty_warnings(&counted), ["g"]);
    }

    /// Non-blocking by default, the error under `Strict` — the same
    /// duality every other degenerate composition has.
    #[test]
    fn an_empty_source_is_the_error_under_strict() {
        let c = clause("x", &[]);
        assert!(validate(&c, Mode::Permissive).is_ok());
        match validate(&c, Mode::Strict) {
            Err(ValidationError::StrictWarning(ValidationWarning::EmptySource { var, .. })) => {
                assert_eq!(var, "x");
            }
            other => panic!("expected the empty source to be the strict error, got {other:?}"),
        }
    }

    /// Nested, so a host sees which clause of a product is the cause.
    #[test]
    fn an_empty_clause_is_named_inside_a_cartesian() {
        let c = Comprehension::cartesian(vec![clause("a", &[1, 2]), clause("b", &[])]);
        assert_eq!(empty_warnings(&c), ["b"]);
    }

    fn continuous_clause(name: &str) -> Comprehension {
        Comprehension::clause(
            name,
            Source::ContinuousInterval {
                interval: Interval::closed(0.0, 1.0),
                measure: ProductMeasure::Uniform,
            },
        )
    }

    #[test]
    fn v1_rejects_duplicate_names_in_cartesian() {
        let bad = Comprehension::cartesian(vec![clause("k", &[1]), clause("k", &[2])]);
        let result = validate(&bad, Mode::Permissive);
        assert!(matches!(
            result,
            Err(ValidationError::V1DuplicateName {
                combinator: "cartesian",
                ..
            })
        ));
    }

    #[test]
    fn v1_accepts_disjoint_names() {
        let ok = Comprehension::cartesian(vec![clause("k", &[1]), clause("limit", &[10])]);
        assert!(validate(&ok, Mode::Permissive).is_ok());
    }

    #[test]
    fn v2_rejects_union_shape_mismatch() {
        let bad = Comprehension::union(vec![
            Comprehension::cartesian(vec![clause("k", &[1]), clause("limit", &[10])]),
            Comprehension::cartesian(vec![clause("limit", &[100]), clause("k", &[100])]),
        ]);
        let result = validate(&bad, Mode::Permissive);
        assert!(matches!(
            result,
            Err(ValidationError::V2ShapeMismatch { .. })
        ));
    }

    #[test]
    fn v2_accepts_matching_union_shape() {
        let ok = Comprehension::union(vec![
            Comprehension::cartesian(vec![clause("k", &[1]), clause("limit", &[10])]),
            Comprehension::cartesian(vec![clause("k", &[100]), clause("limit", &[100])]),
        ]);
        assert!(validate(&ok, Mode::Permissive).is_ok());
    }

    #[test]
    fn v4_rejects_lattice_geometric_over_union() {
        let bad = Comprehension::order(
            Comprehension::union(vec![clause("k", &[1, 2, 3]), clause("k", &[10, 20, 30])]),
            StrategyName::Extrema,
            Some(2),
        );
        assert!(matches!(
            validate(&bad, Mode::Permissive),
            Err(ValidationError::V4InputShape {
                strategy: StrategyName::Extrema,
                ..
            })
        ));
    }

    #[test]
    fn v4_lattice_geometric_over_1axis_warns_not_errors() {
        let degenerate =
            Comprehension::order(clause("k", &[1, 2, 3]), StrategyName::Extrema, Some(2));
        let report = validate(&degenerate, Mode::Permissive).unwrap();
        assert!(report.warnings.iter().any(|w| matches!(
            w,
            ValidationWarning::DegenerateGeometric {
                strategy: StrategyName::Extrema
            }
        )));
    }

    #[test]
    fn v4_strict_mode_promotes_warning() {
        let degenerate =
            Comprehension::order(clause("k", &[1, 2, 3]), StrategyName::Extrema, Some(2));
        assert!(validate(&degenerate, Mode::Strict).is_err());
    }

    #[test]
    fn v7_rejects_continuous_in_zip() {
        let bad = Comprehension::zip(
            vec![continuous_clause("alpha"), continuous_clause("beta")],
            ZipMode::Strict,
        );
        assert!(matches!(
            validate(&bad, Mode::Permissive),
            Err(ValidationError::V7ZipCardinality { .. })
        ));
    }

    #[test]
    fn v8_rejects_continuous_without_sampling() {
        // Continuous clause at the outermost level — no order.
        let bad = continuous_clause("theta");
        assert!(validate(&bad, Mode::Permissive).is_ok());
        // The error fires at the outermost reachable point; for
        // a bare clause we need a wrapping check the consumer
        // does. Wrap it in order(Lex, None) — Lex doesn't sample
        // continuous; V8 fires.
        let bad_lex = Comprehension::order(continuous_clause("theta"), StrategyName::Lex, None);
        assert!(matches!(
            validate(&bad_lex, Mode::Permissive),
            Err(ValidationError::V8ContinuousRequirement { .. })
        ));
    }

    #[test]
    fn v8_accepts_continuous_with_sampling() {
        let ok = Comprehension::order(
            Comprehension::cartesian(vec![continuous_clause("alpha"), continuous_clause("beta")]),
            StrategyName::Halton,
            Some(100),
        );
        assert!(validate(&ok, Mode::Permissive).is_ok());
    }

    #[test]
    fn v8_rejects_unbounded_uniform_at_source() {
        let bad = Comprehension::clause(
            "x",
            Source::ContinuousInterval {
                interval: Interval {
                    lo: 0.0,
                    hi: f64::INFINITY,
                    lo_open: false,
                    hi_open: true,
                },
                measure: ProductMeasure::Uniform,
            },
        );
        assert!(matches!(
            validate(&bad, Mode::Permissive),
            Err(ValidationError::V8ContinuousRequirement { .. })
        ));
    }

    #[test]
    fn v9_rejects_continuous_in_union() {
        let bad = Comprehension::union(vec![
            Comprehension::cartesian(vec![continuous_clause("k"), continuous_clause("limit")]),
            Comprehension::cartesian(vec![continuous_clause("k"), continuous_clause("limit")]),
        ]);
        // Note: this also trips V9 via continuous-in-union before V2 even fires.
        assert!(matches!(
            validate(&bad, Mode::Permissive),
            Err(ValidationError::V9UnionClassMismatch { .. })
        ));
    }

    #[test]
    fn singleton_combinator_warns() {
        let degenerate = Comprehension::cartesian(vec![clause("k", &[1, 2])]);
        let report = validate(&degenerate, Mode::Permissive).unwrap();
        assert!(report.warnings.iter().any(|w| matches!(
            w,
            ValidationWarning::SingletonCombinator {
                combinator: "cartesian"
            }
        )));
    }

    #[test]
    fn trivially_true_filter_warns() {
        let degenerate = Comprehension::filter(clause("k", &[1, 2]), "true");
        let report = validate(&degenerate, Mode::Permissive).unwrap();
        assert!(
            report
                .warnings
                .iter()
                .any(|w| matches!(w, ValidationWarning::TriviallyTrueFilter))
        );
    }

    fn generator(name: &str, expr: &str) -> Comprehension {
        Comprehension::clause(
            name,
            Source::Generator {
                expr: expr.into(),
                cardinality_hint: None,
            },
        )
    }

    fn unresolved(c: &Comprehension, surface: Surface<'_>) -> Vec<(String, bool)> {
        match check_names(c, surface) {
            Ok(()) => Vec::new(),
            Err(ValidationError::V3UnresolvedNames { reads }) => {
                reads.into_iter().map(|r| (r.name, r.bare)).collect()
            }
            Err(other) => panic!("expected V3, got {other}"),
        }
    }

    /// A source reads with the clauses before it in its cartesian bound,
    /// and a predicate with its input's tuple bound; a stream supplies
    /// nothing else.
    #[test]
    fn v3_binds_earlier_axes_for_sources_and_the_tuple_for_predicates() {
        let dependent = Comprehension::cartesian(vec![
            clause("a", &[1, 2]),
            generator("b", "0..{a}"),
            generator("c", "0..{b}"),
        ]);
        assert!(check_names(&dependent, Surface::Stream).is_ok());
        let filtered = Comprehension::filter(dependent.clone(), "{a} < {c}");
        assert!(check_names(&filtered, Surface::Stream).is_ok());
        // A later axis is not bound before it, and a zip's operands do
        // not bind for each other.
        let later = Comprehension::cartesian(vec![generator("b", "0..{a}"), clause("a", &[1])]);
        assert_eq!(
            unresolved(&later, Surface::Stream),
            [("a".to_string(), false)]
        );
        let zipped = Comprehension::zip(
            vec![clause("a", &[1]), generator("b", "0..{a}")],
            ZipMode::Truncate,
        );
        assert_eq!(
            unresolved(&zipped, Surface::Stream),
            [("a".to_string(), false)]
        );
        // A predicate over one axis of a cartesian reads that axis only.
        let inner = Comprehension::cartesian(vec![
            clause("a", &[1, 2]),
            Comprehension::filter(clause("b", &[1, 2]), "{b} < {a}"),
        ]);
        assert_eq!(
            unresolved(&inner, Surface::Stream),
            [("a".to_string(), false)]
        );
    }

    /// A traversal supplies the names of the scope it opens in; a bare
    /// word in a predicate is a name no surface supplies.
    #[test]
    fn v3_closes_names_over_the_surface() {
        let c = Comprehension::filter(
            Comprehension::cartesian(vec![clause("k", &[1, 2]), generator("j", "0..{n}")]),
            "{k} < {limit} && {k} != s1",
        );
        assert_eq!(
            unresolved(&c, Surface::Stream),
            [
                ("n".to_string(), false),
                ("limit".to_string(), false),
                ("s1".to_string(), true)
            ]
        );
        let scope = |n: &str| n == "n" || n == "limit" || n == "s1";
        assert_eq!(
            unresolved(&c, Surface::Traversal(&scope)),
            [("s1".to_string(), true)]
        );
        let quoted = Comprehension::filter(
            Comprehension::cartesian(vec![clause("k", &[1, 2]), generator("j", "0..{n}")]),
            "{k} < {limit} && {k} != \"s1\"",
        );
        assert!(check_names(&quoted, Surface::Traversal(&scope)).is_ok());
        let err = check_names(&c, Surface::Stream).unwrap_err().to_string();
        assert!(err.starts_with("V3:"), "{err}");
        assert!(err.contains("`n` read by clause 'j' in `0..{n}`"), "{err}");
        assert!(
            err.contains("`limit` read by predicate `{k} < {limit} && {k} != s1`"),
            "{err}"
        );
        assert!(err.contains("as in `\"s1\"`"), "{err}");
    }

    /// A call's callee, a cast's type, and `true` are not names; a
    /// callee's arguments are.
    #[test]
    fn v3_reads_opaque_expressions_as_the_language_does() {
        let c = Comprehension::filter(
            clause("k", &[1, 2]),
            "u64_add({k}, width) > 1 && {k} as f64 > 0.5 && true",
        );
        assert_eq!(
            unresolved(&c, Surface::Stream),
            [("width".to_string(), true)]
        );
    }
}