polydat_core/iteration/comprehension/

eval.rs

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

//! Comprehension spec evaluation — text → typed value list.
//!
//! ## What this module does
//!
//! A comprehension clause `var in expr` ships its `expr` as
//! free-form workload-author text. At runtime, the executor
//! needs to turn that text into a list of typed values to
//! enumerate over. That's what [`evaluate_spec`] does, given a
//! Polydat Kernel that holds the in-scope name space (own outputs +
//! inherited externs from `materialize_wiring_from_outer`).
//!
//! ## Pipeline
//!
//! ```text
//!   spec_text
//!       │
//!       ▼
//!   interpolate_via_kernel  ← {name} → kernel.lookup(name)
//!       │
//!       ▼
//!   eval_const_expr_for     ← optional: Polydat expression eval,
//!                              charged to scope.ledger()
//!       │
//!       ▼
//!   parse_list_with_types   ← comma-split, per-element type
//!       │
//!       ▼
//!   Vec<Value>              ← what the executor enumerates
//! ```
//!
//! ## Ownership
//!
//! Polydat owns what a comprehension *means*, including how its
//! source strings resolve (comprehension_forms.md §3.1). A host
//! consumes this API rather than implementing it.

use std::collections::HashMap;

use crate::ast::Value;
use crate::iteration::comprehension::eval_source::NoneRead;
use crate::kernel::interp::Lookup;
use crate::kernel::interp::interpolate_with_lookup;

/// Evaluate a comprehension clause's spec text against a `Lookup` scope.
///
/// Steps:
///  1. [`interpolate_via_kernel`](crate::kernel::interp::interpolate_via_kernel) resolves `{name}` placeholders
///     against the kernel's in-scope name space (own outputs +
///     inherited extern values).
///  2. Try `dsl::compile::eval_const_expr_for(…, kernel.ledger())`
///     on the result. On
///     success with a `Str` value, re-parse as a comma-separated
///     list with per-element type detection. Other typed
///     variants become a single-element typed list.
///  3. On eval failure (most common case for literal lists like
///     `"1, 10"` which aren't valid Polydat const expressions), fall
///     back to [`parse_list_with_types`] on the interpolated
///     text — `1` → `U64`, `1.5` → `F64`, `true` → `Bool`,
///     anything else → `Str`.
///
/// Errors propagate from interpolation (unresolved placeholder,
/// runaway round count, etc.) — those are the user-facing
/// actionable diagnostics. A name the source reads that nothing binds
/// is such an error here; a name bound to None reads None, and the
/// source yields nothing (none_semantics.md Rule 1). A clause source
/// evaluated as part of a comprehension reads an unbound name as None
/// too (comprehension_forms.md §5 V3).
pub fn evaluate_spec(
    spec_text: &str,
    kernel: &dyn Lookup,
) -> Result<Vec<Value>, crate::dsl::compile::EmbeddingError> {
    match evaluate_spec_internal(spec_text, kernel) {
        Ok(values) => Ok(values),
        Err(SpecError::ReadsNone { reads, .. })
            if reads.iter().all(|r| matches!(r, NoneRead::BoundNone(_))) =>
        {
            Ok(Vec::new())
        }
        Err(e) => Err(spec_error(spec_text, String::from(e))),
    }
}

/// The error [`evaluate_spec`] reports for `spec_text` failing with
/// `message`.
pub(crate) fn spec_error(spec_text: &str, message: String) -> crate::dsl::compile::EmbeddingError {
    if let Some(rest) = message.strip_prefix("interpolation: unresolved placeholder '{")
        && let Some(end) = rest.find('}')
    {
        return crate::dsl::compile::EmbeddingError::UnresolvedPlaceholder {
            name: rest[..end].to_string(),
            source: spec_text.to_string(),
        };
    }
    crate::dsl::compile::EmbeddingError::Parse {
        source: spec_text.to_string(),
        message,
        position: None,
    }
}

/// Why a source's spec text yields no values.
#[derive(Debug, Clone)]
pub(crate) enum SpecError {
    /// The source read names that nothing binds or that are bound to
    /// None, after composition, so it reads None and yields nothing
    /// (none_semantics.md Rule 1, comprehension_forms.md §5 V3).
    /// `message` is the diagnostic an API that refuses such a read
    /// reports ([`evaluate_spec`]).
    ReadsNone {
        /// Each such name, once, with how it read None.
        reads: Vec<NoneRead>,
        /// The diagnostic for the first read.
        message: String,
    },
    /// Evaluating the text failed.
    Failed(String),
}

impl From<String> for SpecError {
    fn from(message: String) -> Self {
        SpecError::Failed(message)
    }
}

impl From<SpecError> for String {
    fn from(e: SpecError) -> Self {
        match e {
            SpecError::ReadsNone { message, .. } | SpecError::Failed(message) => message,
        }
    }
}

/// What `kernel` binds `name` to, for a source that reads it: a name
/// nothing binds, or one bound to None, is [`SpecError::ReadsNone`],
/// with `unbound` the diagnostic for the first.
fn read_name(
    kernel: &dyn Lookup,
    name: &str,
    unbound: impl FnOnce() -> String,
) -> Result<Value, SpecError> {
    match kernel.lookup(name) {
        Some(Value::None) => Err(SpecError::ReadsNone {
            reads: vec![NoneRead::BoundNone(name.to_string())],
            message: format!("`{name}` is None"),
        }),
        Some(value) => Ok(value),
        None => Err(SpecError::ReadsNone {
            reads: vec![NoneRead::Unbound(name.to_string())],
            message: unbound(),
        }),
    }
}

/// Interpolate `text`'s `{name}` placeholders against `kernel`, inside
/// out, so a composed name (`{k_{k}_limits}`) reads the name its leaves
/// compose to. A placeholder that names what nothing binds, or what is
/// bound to None, is a read of None ([`SpecError::ReadsNone`]); any
/// other failure is [`SpecError::Failed`].
fn interpolate_reading(text: &str, kernel: &dyn Lookup) -> Result<String, SpecError> {
    let reads = std::cell::RefCell::new(Vec::new());
    let interpolated = interpolate_with_lookup(text, |name| match kernel.lookup(name) {
        Some(Value::None) => {
            reads
                .borrow_mut()
                .push(NoneRead::BoundNone(name.to_string()));
            None
        }
        Some(value) => Some(value.to_display_string()),
        None => {
            if is_name_path(name) {
                reads.borrow_mut().push(NoneRead::Unbound(name.to_string()));
            }
            None
        }
    });
    interpolated.map_err(|message| {
        let mut reads = reads.into_inner();
        if reads.is_empty() {
            return SpecError::Failed(message);
        }
        reads.sort();
        reads.dedup();
        SpecError::ReadsNone { reads, message }
    })
}

/// Whether a placeholder body is a name — an identifier, or a dotted
/// path (`q.cursor.idx`, or `t_0.5_vals` composed from a float) — rather
/// than an expression or a format specifier.
fn is_name_path(body: &str) -> bool {
    let mut chars = body.chars();
    chars
        .next()
        .is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
        && chars.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '.')
}

/// The names `text` reads as free identifiers of the language that
/// `kernel` does not bind or binds to None, as [`SpecError::ReadsNone`]
/// with `message`; [`SpecError::Failed`] with `message` when there are
/// none. An evaluation of `text` that failed reading such a name failed
/// because the name reads None.
fn failure_reading(text: &str, kernel: &dyn Lookup, message: String) -> SpecError {
    let reads: Vec<NoneRead> = polydat_grammar::refs::referenced_names(text)
        .into_iter()
        .filter_map(|name| match kernel.lookup(&name) {
            None => Some(NoneRead::Unbound(name)),
            Some(Value::None) => Some(NoneRead::BoundNone(name)),
            Some(_) => None,
        })
        .collect();
    if reads.is_empty() {
        SpecError::Failed(message)
    } else {
        SpecError::ReadsNone { reads, message }
    }
}

pub(crate) fn evaluate_spec_internal(
    spec_text: &str,
    kernel: &dyn Lookup,
) -> Result<Vec<Value>, SpecError> {
    if let Some(values) = try_eval_all_cursor(spec_text, kernel)? {
        return Ok(values);
    }
    // A *bare identifier* source is a direct wire/param/const
    // reference (comprehension_forms.md §3.1.4). It resolves
    // against the kernel chain — the same `kernel.lookup` the
    // `{name}` interpolation path uses — and its value is peeled or
    // wrapped by `iteration_interior`, so `mnc in mnc_values` works
    // identically to `mnc in {mnc_values}`. A bare name that does
    // not resolve reads None, and the diagnostic an API that refuses
    // it reports carries a quoting hint; it is never bound as its own
    // name-string.
    if is_single_bare_ident(spec_text) {
        let src = spec_text.trim();
        let v = read_name(kernel, src, || {
            format!(
                "comprehension source `{src}` did not resolve to a value — no \
                 wire, const, param, or outer iter-var by that name is in scope \
                 here. If you meant the literal string \"{src}\", quote it: \
                 `\"{src}\"`."
            )
        })?;
        return Ok(
            match crate::iteration::comprehension::source_values::iteration_interior(&v) {
                Some(interior) => interior,
                None => vec![v],
            },
        );
    }
    let interpolated = interpolate_reading(spec_text, kernel)?;
    // What the interpolated text evaluates to. A failure that reads a
    // free name the scope does not bind, or binds to None, is that read
    // of None.
    evaluate_interpolated(&interpolated, kernel).map_err(|e| match e {
        SpecError::Failed(message) => failure_reading(&interpolated, kernel, message),
        reads_none => reads_none,
    })
}

/// [`evaluate_spec_internal`] past interpolation: the source forms over
/// the interpolated text.
fn evaluate_interpolated(interpolated: &str, kernel: &dyn Lookup) -> Result<Vec<Value>, SpecError> {
    // List comprehension sugar `[e1, e2…, e3]`
    // (comprehension_forms.md §3.1.3). Resolved after interpolation so `{name}` placeholders inside
    // elements expand first; before the const-eval fallthrough so
    // bracket structure isn't misparsed as an array-literal expr.
    if let Some(values) = try_eval_bracket_list(interpolated, kernel)? {
        return Ok(values);
    }
    // Range operator (`a..b`, `a..=b`, `a..b..s`, `a..=b..s`;
    // comprehension_forms.md §3.1, polydat_grammar.md §16.2). Bounds
    // and step are Polydat const expressions evaluated at this
    // (post-interpolation) point.
    if let Some(values) = try_eval_range(interpolated, kernel.ledger())? {
        return Ok(values);
    }
    // Named generators (comprehension_forms.md §3.1).
    if let Some(values) = try_eval_generator(interpolated)? {
        return Ok(values);
    }
    // Set operators on lists (comprehension_forms.md §3.1).
    if let Some(values) = try_eval_setop(interpolated, kernel)? {
        return Ok(values);
    }
    // Sequencer expansions (bucket / concat_seq / interval_seq;
    // comprehension_forms.md §3.1).
    if let Some(values) = try_eval_sequencer(interpolated, kernel)? {
        return Ok(values);
    }
    // Kernel-aware partition sources — `subdivide(outer, n)` where
    // `outer` is a partition iter-var bound by an enclosing clause
    // (cursor_partitions.md §7.1).
    if let Some(values) = try_eval_partition_call(interpolated, kernel, false)? {
        return Ok(values);
    }
    // `<param>.partitions` in comprehension position resolves the
    // param's spec string and expands it into its PartitionList
    // (cursor_partitions.md §7.1).
    if let Some(values) = try_eval_param_partitions(interpolated, kernel, false)? {
        return Ok(values);
    }
    match crate::dsl::compile::eval_const_expr_for(interpolated, kernel.ledger()) {
        // Relaxed source resolution (comprehension_forms.md §3.1.2):
        // a resolved value is peeled one level if it has an iteration
        // interior (native vector, JSON array, PartitionList, or a
        // string → its comprehension tokens), else wrapped as a
        // singleton. `iteration_interior` is the single place that
        // decision is made.
        Ok(v) => Ok(
            match crate::iteration::comprehension::source_values::iteration_interior(&v) {
                Some(interior) => interior,
                None => vec![v],
            },
        ),
        // Fall back to the literal-list parse only when the text
        // is unambiguously a comma-separated list of literals
        // (e.g. `1, 10, 100` — `eval_const_expr` doesn't accept
        // that shape because it isn't a single Polydat expression).
        // Anything that looks like an expression (parens, GK
        // operators, identifiers other than `true`/`false`) was
        // *meant* to evaluate; if it failed, we MUST surface the
        // failure rather than silently splitting and
        // handing the workload an iter-var like
        // `matching_profiles('x'` (truncated), which would produce
        // malformed downstream output far removed from the fault.
        Err(eval_err) => {
            // A single bare identifier never reaches here: it is a
            // reference, resolved or refused above. An unbracketed
            // bare label list (`a, b, c`) keeps string-token striping
            // (comprehension_forms.md §3.1.4).
            if looks_like_literal_list(interpolated) {
                // A bare unquoted token list strips on the same
                // separator rule as a string comprehension.
                Ok(
                    crate::iteration::comprehension::source_values::strip_string_tokens(
                        interpolated,
                    ),
                )
            } else {
                Err(format!(
                    "for_each clause expression failed to evaluate: {eval_err}\n\
                     spec: {interpolated}\n\
                     If this was meant as a literal list (e.g. `1, 10, 100`), \
                     it should contain only literal values separated by commas. \
                     If it was meant as an expression, fix the underlying \
                     evaluation error."
                )
                .into())
            }
        }
    }
}

/// List comprehension sugar (comprehension_forms.md §3.1.3). Evaluate a
/// bracketed source `[e1, e2…, e3]` to its bound sequence,
/// peeling exactly one level:
///   - a plain element contributes its value, whole (no peel);
///   - a spread element `S…` / `S...` contributes `S`'s
///     iteration interior (peel one level); a non-iterable `S`
///     under spread is a hard error.
///
/// Elements are parsed by the core expression grammar: a bare
/// identifier is a wire/param reference (resolved against the
/// kernel), a quoted token is a string, numbers/bools are
/// literals. Returns `Ok(None)` when `text` is not a bracketed
/// list (so the caller falls through to the other source forms).
fn try_eval_bracket_list(text: &str, kernel: &dyn Lookup) -> Result<Option<Vec<Value>>, SpecError> {
    let t = text.trim();
    if !(t.starts_with('[') && t.ends_with(']') && t.len() >= 2) {
        return Ok(None);
    }
    let inner = &t[1..t.len() - 1];
    if inner.trim().is_empty() {
        return Ok(Some(Vec::new()));
    }
    let mut out = Vec::new();
    for elem in split_args_top_level(inner) {
        let elem = elem.trim();
        // Spread suffix: `…` (U+2026) or `...`.
        let (expr, spread) = if let Some(stripped) = elem.strip_suffix('…') {
            (stripped.trim(), true)
        } else if let Some(stripped) = elem.strip_suffix("...") {
            (stripped.trim(), true)
        } else {
            (elem, false)
        };
        if expr.is_empty() {
            return Err("empty element in list comprehension `[...]`"
                .to_string()
                .into());
        }
        let value = eval_element_value(expr, kernel)?;
        if spread {
            match crate::iteration::comprehension::source_values::iteration_interior(&value) {
                Some(interior) => out.extend(interior),
                None => {
                    return Err(format!(
                        "list comprehension spread `{expr}…` requires an iterable \
                     source, but `{expr}` resolved to a scalar \
                     {ty:?}. Use `[{expr}]` to pass it as a single element, \
                     or supply a list.",
                        ty = value.port_type(),
                    )
                    .into());
                }
            }
        } else {
            out.push(value);
        }
    }
    Ok(Some(out))
}

/// Resolve one list-comprehension element to a single value (no
/// peeling). A bare identifier is a wire/param/const reference
/// resolved against the kernel; anything else (quoted string,
/// number, bool, expression) goes through the const evaluator.
/// An unresolved bare reference reads None, never a silent literal-name
/// binding, and the diagnostic an API that refuses it reports carries a
/// quoting hint (comprehension_forms.md §3.1.4).
fn eval_element_value(expr: &str, kernel: &dyn Lookup) -> Result<Value, SpecError> {
    let e = expr.trim();
    if is_single_bare_ident(e) {
        return read_name(kernel, e, || {
            format!(
                "list element `{e}` did not resolve to a value — no wire, const, \
             param, or outer iter-var by that name is in scope here. \
             If you meant the literal string \"{e}\", quote it: `\"{e}\"`."
            )
        });
    }
    crate::dsl::compile::eval_const_expr_for(e, kernel.ledger()).map_err(|err| {
        failure_reading(
            e,
            kernel,
            format!("list element `{e}` failed to evaluate: {err}"),
        )
    })
}

/// True when `text` is exactly one bare identifier
/// (`[A-Za-z_][A-Za-z0-9_]*`), excluding `true`/`false`. A bare
/// identifier source is a direct reference resolved against the
/// kernel (comprehension_forms.md §3.1.4); the keyword literals are
/// values.
fn is_single_bare_ident(text: &str) -> bool {
    let t = text.trim();
    if t == "true" || t == "false" {
        return false;
    }
    let mut chars = t.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() || c == '_' => {}
        _ => return false,
    }
    chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
}

/// Heuristic: does this interpolated spec text look like a
/// "literal list" (comma-separated literals like `1, 10, 100` or
/// `foo, bar, baz`) rather than an expression?
///
/// True only when no character suggests an expression: no
/// parentheses, no operators, no string-quote characters that
/// would imply a function-call shape. Whitespace, digits,
/// alphanumerics, dots (for floats), minus (for negatives), and
/// commas (the separator) are all OK.
///
/// The gate keeps list sources (`k in 1, 10, 100`) working through
/// the literal-list fallback while still surfacing real evaluation
/// failures for expression sources like `matching_profiles('x',
/// 'y')`. A wrong call on a borderline case is cheap: it produces a
/// clearer error from the eval layer instead of swallowed garbage.
fn looks_like_literal_list(text: &str) -> bool {
    let trimmed = text.trim();
    if trimmed.is_empty() {
        return false;
    }
    !trimmed.chars().any(|c| {
        matches!(
            c,
            '(' | ')'
                | '['
                | ']'
                | '{'
                | '}'
                | '\''
                | '"'
                | '+'
                | '*'
                | '/'
                | '%'
                | '='
                | '<'
                | '>'
                | '!'
                | '&'
                | '|'
                | '~'
                | '^'
                | '?'
        )
    })
}

/// Pre-evaluate a clause's spec text at synthesis time, using
/// `probes` for prior clauses' first values and `workload_params`
/// as a fallback source for names not yet promoted to workload-
/// kernel `const` bindings.
///
/// The runtime dispatcher uses [`evaluate_spec`] directly because
/// (a) the runtime kernel has prior-clause values as real input
/// slots, not text probes, and (b) by then workload params are
/// already injected as final bindings on the for_each scope's
/// kernel via the synthesis path.
pub fn pre_evaluate_clause(
    spec_text: &str,
    parent_kernel: &dyn Lookup,
    workload_params: &HashMap<String, String>,
    probes: &HashMap<String, String>,
) -> Result<Vec<Value>, String> {
    // The `all(<cursor>)` form resolves cursor extents from the
    // parent kernel's auxiliary outputs; it doesn't fit the
    // const-eval pipeline (which returns a single Value), so it's
    // intercepted here too — same as `evaluate_spec`.
    if let Some(values) = try_eval_all_cursor(spec_text, parent_kernel)? {
        return Ok(values);
    }
    // A bare identifier source is a direct reference
    // (comprehension_forms.md §3.1.4). At synthesis it resolves
    // against a prior iter-var probe, then the parent kernel, then
    // the workload params, so a dependent clause (e.g. `limit in
    // {k_{k}_limits}`) sees the *typed* prior value and infers the
    // right extern type, rather than the literal name typed as a
    // string.
    if is_single_bare_ident(spec_text) {
        let name = spec_text.trim();
        if let Some(pv) = probes.get(name) {
            return Ok(crate::iteration::comprehension::source_values::strip_string_tokens(pv));
        }
        if let Some(v) = parent_kernel.lookup(name) {
            return Ok(
                match crate::iteration::comprehension::source_values::iteration_interior(&v) {
                    Some(interior) => interior,
                    None => vec![v],
                },
            );
        }
        if let Some(s) = workload_params.get(name) {
            return Ok(crate::iteration::comprehension::source_values::strip_string_tokens(s));
        }
        return Err(format!(
            "comprehension source `{name}` did not resolve to a value — no wire, \
             const, param, or outer iter-var by that name is in scope here. \
             If you meant the literal string \"{name}\", quote it: `\"{name}\"`."
        ));
    }
    let mut text = spec_text.to_string();
    for (var, probe_value) in probes {
        text = text.replace(&format!("{{{var}}}"), probe_value);
    }

    let interpolated = interpolate_with_lookup(&text, |name| {
        parent_kernel
            .lookup(name)
            .map(|v| v.to_display_string())
            .or_else(|| workload_params.get(name).cloned())
    })?;

    // The range operator, on the pre-evaluation path too.
    if let Some(values) = try_eval_range(&interpolated, parent_kernel.ledger())? {
        return Ok(values);
    }
    // The same generator, set-operator, and sequencer forms the
    // runtime path recognizes.
    if let Some(values) = try_eval_generator(&interpolated)? {
        return Ok(values);
    }
    if let Some(values) = try_eval_setop(&interpolated, parent_kernel)? {
        return Ok(values);
    }
    if let Some(values) = try_eval_sequencer(&interpolated, parent_kernel)? {
        return Ok(values);
    }
    // Kernel-aware partition sources, as on the runtime path
    // (cursor_partitions.md §7.1). At pre-evaluation the outer
    // iter-var may not be installed yet; `try_eval_partition_call`
    // then returns a single placeholder partition so iter-var type
    // detection yields `ext`.
    if let Some(values) = try_eval_partition_call(&interpolated, parent_kernel, true)? {
        return Ok(values);
    }
    // `<param>.partitions` in comprehension position, by the same rule
    // as the runtime path; the param may already be installed here.
    if let Some(values) = try_eval_param_partitions(&interpolated, parent_kernel, true)? {
        return Ok(values);
    }
    let value_str =
        match crate::dsl::compile::eval_const_expr_for(&interpolated, parent_kernel.ledger()) {
            Ok(Value::Str(s)) => s.to_string(),
            // `<param>.partitions` and `partitions(spec, ...)`
            // (cursor_partitions.md §7.1) both evaluate to a `PartitionList` Ext value. Unpack
            // its entries into a vec of individual `Partition`
            // values so the for-clause iterates partition-by-
            // partition.
            Ok(ref v) if v.as_partition_list().is_some() => {
                let list = v.as_partition_list().unwrap();
                return Ok(list
                    .as_slice()
                    .iter()
                    .map(|p| Value::from_partition(*p))
                    .collect());
            }
            Ok(other) => return Ok(vec![other]),
            // Mirrors `evaluate_spec`'s gating: only fall back to
            // parse_list_with_types when the text is unambiguously a
            // literal list. See `looks_like_literal_list` for the
            // rationale.
            Err(eval_err) => {
                if looks_like_literal_list(&interpolated) {
                    interpolated
                } else {
                    return Err(format!(
                        "for_each clause expression failed to evaluate: {eval_err}\n\
                     spec: {interpolated}\n\
                     If this was meant as a literal list (e.g. `1, 10, 100`), \
                     it should contain only literal values separated by commas. \
                     If it was meant as an expression, fix the underlying \
                     evaluation error."
                    ));
                }
            }
        };
    Ok(parse_list_with_types(&value_str))
}

/// Parse a comma-separated text list, detecting each element's
/// native type, as element types are inferred
/// (polydat_grammar.md §16.3): `"1, 10"` → `[U64(1), U64(10)]`, `"1.5, 2.5"` → `[F64(...)]`,
/// mixed → each element gets its own native type.
pub fn parse_list_with_types(text: &str) -> Vec<Value> {
    text.split(',')
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(|s| {
            if let Ok(n) = s.parse::<u64>() {
                Value::U64(n)
            } else if let Ok(n) = s.parse::<f64>() {
                Value::F64(n)
            } else if s == "true" {
                Value::Bool(true)
            } else if s == "false" {
                Value::Bool(false)
            } else {
                Value::Str(s.to_string().into())
            }
        })
        .collect()
}

/// Recognize the comprehension-level `all(<cursor>)` clause form
/// and resolve it against the parent kernel's cursor extent
/// auxiliary outputs.
///
/// Cursors declared via the Polydat `cursor name = Cursor(start, end)`
/// shape compile to two well-known auxiliary outputs on the
/// kernel: `__cursor_extent_<name>_start` and
/// `__cursor_extent_<name>_end`. Reading those gives the cursor's
/// resolved extent at scope-init time. `all(<cursor>)` lowers to
/// the half-open ordinal range `[start, end)` as a `Vec<Value::U64>`.
///
/// The source reads the cursor's extent, not a value named after the
/// cursor ([`polydat_grammar::comprehension::source::Source::names_read`]).
///
/// Returns:
/// - `Ok(Some(values))` if `spec_text` matches the `all(<ident>)`
///   shape and the cursor's extent resolved successfully.
/// - `Ok(None)` if `spec_text` doesn't match — caller continues
///   with the normal interpolation + const-eval pipeline.
/// - [`SpecError::ReadsNone`] if an extent output is not bound in
///   scope (no cursor of that name is declared at or above it) or is
///   bound to None.
/// - [`SpecError::Failed`] if the extent is not an ordinal range.
fn try_eval_all_cursor(
    spec_text: &str,
    kernel: &dyn Lookup,
) -> Result<Option<Vec<Value>>, SpecError> {
    use polydat_grammar::comprehension::source::{all_cursor_argument, cursor_extent_names};
    let Some(cursor_name) = all_cursor_argument(spec_text) else {
        return Ok(None);
    };
    let [start_key, end_key] = cursor_extent_names(cursor_name);
    let extent = |key: &str| -> Result<u64, SpecError> {
        match read_name(kernel, key, || {
            format!(
                "all({cursor_name}): cursor '{cursor_name}' has no resolvable extent — \
                 check that the cursor is declared at or above this scope and that \
                 its range arguments are init-resolvable. Looked for output '{key}'."
            )
        })? {
            Value::U64(n) => Ok(n),
            other => Err(format!(
                "all({cursor_name}): extent output '{key}' is {}, not an ordinal",
                other.to_display_string()
            )
            .into()),
        }
    };
    let start = extent(&start_key)?;
    let end = extent(&end_key)?;

    if end < start {
        return Err(format!(
            "all({cursor_name}): cursor extent end={end} is less than start={start} — \
             cannot enumerate a negative-extent range."
        )
        .into());
    }
    Ok(Some((start..end).map(Value::U64).collect()))
}

fn is_valid_ident(s: &str) -> bool {
    let mut chars = s.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() || c == '_' => {}
        _ => return false,
    }
    chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
}

/// Recognise the range operator (comprehension_forms.md §3.1,
/// polydat_grammar.md §16.2) and expand it into a `Vec<Value>`.
///
/// Four shapes:
/// - `a..b`         half-open with step 1
/// - `a..=b`        closed with step 1
/// - `a..b..s`      half-open with step `s`
/// - `a..=b..s`     closed with step `s`
///
/// Bounds and step are Polydat const expressions; this function
/// evaluates each segment via `eval_const_expr_for(segment, ledger)`. Numeric
/// type follows the bounds: if both are integers, the
/// emitted list is `Value::U64`; otherwise `Value::F64`.
///
/// Returns:
/// - `Ok(Some(values))` on a successful range expansion.
/// - `Ok(None)` when `text` doesn't have a top-paren-depth
///   `..` at all — caller falls through to the standard
///   const-eval / list-parse path.
/// - `Err(...)` when the form matches but evaluation fails
///   (bound non-numeric, step is zero, bounds diverge from
///   step direction, etc.).
fn try_eval_range(
    text: &str,
    ledger: &std::sync::Arc<crate::kernel::CompileLedger>,
) -> Result<Option<Vec<Value>>, String> {
    let trimmed = text.trim();
    let chars: Vec<char> = trimmed.chars().collect();

    // Find every top-paren-depth `..` (with optional `=`).
    // Returns positions of the `..` start and whether the
    // following `=` was present.
    let mut splits: Vec<(usize, bool)> = Vec::new();
    let mut depth: i32 = 0;
    let mut i = 0;
    while i < chars.len() {
        let c = chars[i];
        match c {
            '(' | '[' | '{' => depth += 1,
            ')' | ']' | '}' => depth -= 1,
            '"' | '\'' => {
                // Skip the rest of the quoted run.
                let q = c;
                i += 1;
                while i < chars.len() && chars[i] != q {
                    i += 1;
                }
            }
            '.' if depth == 0 && i + 1 < chars.len() && chars[i + 1] == '.' => {
                let inclusive = i + 2 < chars.len() && chars[i + 2] == '=';
                splits.push((i, inclusive));
                i += if inclusive { 3 } else { 2 };
                continue;
            }
            _ => {}
        }
        i += 1;
    }

    if splits.is_empty() {
        return Ok(None);
    }
    if splits.len() > 2 {
        return Err(format!(
            "range expression '{trimmed}': more than two `..` operators \
             at top level — expected one of `a..b`, `a..=b`, `a..b..s`, \
             or `a..=b..s`"
        ));
    }
    if splits.len() == 2 && splits[1].1 {
        return Err(format!(
            "range expression '{trimmed}': step delimiter cannot be \
             `..=` — only the bound separator may be inclusive"
        ));
    }

    // Slice out the segments.
    let inclusive = splits[0].1;
    let first_end = splits[0].0;
    let after_first = first_end + if inclusive { 3 } else { 2 };
    let (start_text, mid_text, step_text) = match splits.len() {
        1 => {
            let start_s: String = chars[..first_end].iter().collect();
            let end_s: String = chars[after_first..].iter().collect();
            (start_s, end_s, None)
        }
        2 => {
            let mid_end = splits[1].0;
            let after_mid = mid_end + 2; // `..` only, not `..=`
            let start_s: String = chars[..first_end].iter().collect();
            let mid_s: String = chars[after_first..mid_end].iter().collect();
            let step_s: String = chars[after_mid..].iter().collect();
            (start_s, mid_s, Some(step_s))
        }
        _ => unreachable!(),
    };

    let start_val = eval_range_segment(&start_text, "range start", ledger)?;
    let end_val = eval_range_segment(&mid_text, "range end", ledger)?;
    let step_val = match step_text {
        Some(s) => Some(eval_range_segment(&s, "range step", ledger)?),
        None => None,
    };

    Ok(Some(expand_range(
        start_val, end_val, step_val, inclusive, trimmed,
    )?))
}

fn eval_range_segment(
    text: &str,
    what: &str,
    ledger: &std::sync::Arc<crate::kernel::CompileLedger>,
) -> Result<Value, String> {
    let trimmed = text.trim();
    if trimmed.is_empty() {
        return Err(format!("range expression: {what} is empty"));
    }
    crate::dsl::compile::eval_const_expr_for(trimmed, ledger)
        .map_err(|e| format!("range expression: {what} '{trimmed}' did not const-fold — {e}"))
}

/// Materialise the value list once start/end/step have been
/// const-folded. If any of the three is `F64`, the whole list
/// is `F64`; otherwise everything is `U64`.
fn expand_range(
    start: Value,
    end: Value,
    step: Option<Value>,
    inclusive: bool,
    src: &str,
) -> Result<Vec<Value>, String> {
    let any_float = matches!(start, Value::F64(_))
        || matches!(end, Value::F64(_))
        || matches!(step, Some(Value::F64(_)));

    let to_f64 = |v: &Value| -> Result<f64, String> {
        match v {
            Value::U64(n) => Ok(*n as f64),
            Value::F64(f) => Ok(*f),
            other => Err(format!(
                "range expression '{src}': bound has non-numeric value {other:?}"
            )),
        }
    };
    let to_i64 = |v: &Value| -> Result<i64, String> {
        match v {
            Value::U64(n) => i64::try_from(*n).map_err(|_| {
                format!("range expression '{src}': bound {n} exceeds signed 64-bit range")
            }),
            Value::F64(f) => {
                if f.fract() == 0.0 && *f >= i64::MIN as f64 && *f <= i64::MAX as f64 {
                    Ok(*f as i64)
                } else {
                    Err(format!(
                        "range expression '{src}': float bound {f} is not integral; \
                         mix with an explicit float step (e.g. `1.0..10..0.5`) for a float range"
                    ))
                }
            }
            other => Err(format!(
                "range expression '{src}': bound has non-numeric value {other:?}"
            )),
        }
    };

    if any_float {
        let s = to_f64(&start)?;
        let e = to_f64(&end)?;
        let st = match step.as_ref() {
            Some(v) => to_f64(v)?,
            None => 1.0,
        };
        if st == 0.0 {
            return Err(format!("range expression '{src}': step is zero"));
        }
        // Direction must match (start < end ⇒ step > 0; start > end ⇒ step < 0).
        if (e - s).is_sign_positive() && st < 0.0 {
            return Ok(Vec::new());
        }
        if (e - s).is_sign_negative() && st > 0.0 {
            return Ok(Vec::new());
        }
        let mut out = Vec::new();
        let mut cur = s;
        let cmp = |x: f64| -> bool {
            if st > 0.0 {
                if inclusive {
                    x <= e + 1e-12
                } else {
                    x < e - 1e-12
                }
            } else if inclusive {
                x >= e - 1e-12
            } else {
                x > e + 1e-12
            }
        };
        while cmp(cur) {
            out.push(Value::F64(cur));
            cur += st;
        }
        return Ok(out);
    }

    // Integer range.
    let s = to_i64(&start)?;
    let e = to_i64(&end)?;
    let st = match step.as_ref() {
        Some(v) => to_i64(v)?,
        None => 1,
    };
    if st == 0 {
        return Err(format!("range expression '{src}': step is zero"));
    }
    if st > 0 && s > e {
        return Ok(Vec::new());
    }
    if st < 0 && s < e {
        return Ok(Vec::new());
    }
    let mut out = Vec::new();
    let mut cur = s;
    let cmp = |x: i64| -> bool {
        if st > 0 {
            if inclusive { x <= e } else { x < e }
        } else if inclusive {
            x >= e
        } else {
            x > e
        }
    };
    while cmp(cur) {
        if cur < 0 {
            return Err(format!(
                "range expression '{src}': negative value {cur} can't be \
                 represented as Value::U64; use a float range \
                 (mix any bound or step with `.0`) for signed walks"
            ));
        }
        out.push(Value::U64(cur as u64));
        cur = cur.saturating_add(st);
        if (st > 0 && cur < s) || (st < 0 && cur > s) {
            // saturated; would loop forever on overflow.
            break;
        }
    }
    Ok(out)
}

// ============================================================
// Function-call dispatch (Pushes 7, 8, 9)
// ============================================================

/// Recognise `name(args)` at the top paren depth. Returns
/// `Some((name, args))` when the entire `text` is exactly
/// one function call (with balanced parens, possibly empty
/// args). Quoted strings within args are walked as opaque
/// runs so internal commas / parens don't trip the split.
fn parse_func_call(text: &str) -> Option<(&str, &str)> {
    let trimmed = text.trim();
    if !trimmed.ends_with(')') {
        return None;
    }
    let open = trimmed.find('(')?;
    let name = trimmed[..open].trim();
    if name.is_empty() || !is_valid_ident(name) {
        return None;
    }
    // Make sure the closing `)` matches the opening — i.e.
    // the entire text is a single call, not `f(a) + g(b)`.
    let chars: Vec<char> = trimmed.chars().collect();
    let mut depth = 0i32;
    let mut in_quote: Option<char> = None;
    for (i, &c) in chars.iter().enumerate().skip(open) {
        match (c, in_quote) {
            ('"' | '\'', None) => in_quote = Some(c),
            (q, Some(open_q)) if q == open_q => in_quote = None,
            ('(', None) => depth += 1,
            (')', None) => {
                depth -= 1;
                if depth == 0 {
                    if i != chars.len() - 1 {
                        return None; // close mid-text
                    }
                    let args: String = chars[open + 1..i].iter().collect();
                    // SAFETY: trimmed lives for fn duration; we
                    // index into the original string via slices
                    // with care. Instead of returning a borrowed
                    // slice from the local `args` String, return
                    // the slices directly from `trimmed`.
                    let _ = args;
                    let name_slice = &trimmed[..open];
                    let args_slice = &trimmed[open + 1..trimmed.len() - 1];
                    return Some((name_slice.trim(), args_slice));
                }
            }
            _ => {}
        }
    }
    None
}

/// Split a function-argument list on top-level commas. Skips
/// commas inside parens, brackets, braces, or quoted strings.
fn split_args_top_level(args: &str) -> Vec<&str> {
    let mut out: Vec<&str> = Vec::new();
    let chars: Vec<char> = args.chars().collect();
    let bytes_per_char: Vec<usize> = chars.iter().map(|c| c.len_utf8()).collect();
    let mut start_byte = 0usize;
    let mut byte = 0usize;
    let mut depth = 0i32;
    let mut in_quote: Option<char> = None;
    for (i, &c) in chars.iter().enumerate() {
        match (c, in_quote) {
            ('"' | '\'', None) => in_quote = Some(c),
            (q, Some(open_q)) if q == open_q => in_quote = None,
            ('(' | '[' | '{', None) => depth += 1,
            (')' | ']' | '}', None) => depth -= 1,
            (',', None) if depth == 0 => {
                let seg = &args[start_byte..byte];
                out.push(seg.trim());
                start_byte = byte + bytes_per_char[i];
            }
            _ => {}
        }
        byte += bytes_per_char[i];
    }
    let last = &args[start_byte..];
    if !last.trim().is_empty() || !out.is_empty() {
        out.push(last.trim());
    }
    out
}

/// Parse a single argument text as a `u64`. Errors carry the
/// expected-form context for the user.
fn parse_u64_arg(text: &str, what: &str) -> Result<u64, String> {
    let trimmed = text.trim();
    trimmed
        .parse::<u64>()
        .map_err(|_| format!("{what}: expected non-negative integer, got '{trimmed}'"))
}

/// Parse a single argument as either u64 or f64. Returns the
/// f64 representation regardless (callers that need an int
/// check `.fract() == 0.0`).
fn parse_num_arg(text: &str, what: &str) -> Result<f64, String> {
    let trimmed = text.trim();
    trimmed
        .parse::<f64>()
        .map_err(|_| format!("{what}: expected numeric, got '{trimmed}'"))
}

// ============================================================
// Named generators (comprehension_forms.md §3.1.3)
// ============================================================

/// A named generator: a call in source position that expands its
/// literal arguments into a finite list of values
/// (comprehension_forms.md §3.1.3, "Named generators").
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum NamedGenerator {
    /// `fib(n)`: the first `n` Fibonacci numbers.
    Fib,
    /// `fib_until(max)`: the Fibonacci numbers up to `max`.
    FibUntil,
    /// `pow2(n)`: the first `n` powers of two.
    Pow2,
    /// `pow2_until(max)`: the powers of two up to `max`.
    Pow2Until,
    /// `binomial(n)`: row `n` of Pascal's triangle.
    Binomial,
    /// `geometric(start, factor, n)`: `n` terms of a geometric series.
    Geometric,
    /// `geometric_until(start, factor, max)`: a geometric series up to `max`.
    GeometricUntil,
    /// `linear_starts(start, end, n)`: the starts of `n` equal steps.
    LinearStarts,
    /// `linear_steps(start, end, n)`: `n` evenly spaced points, both ends included.
    LinearSteps,
    /// `log_steps(start, end, n)`: `n` log-spaced points, both ends included.
    LogSteps,
}

impl NamedGenerator {
    /// Every named generator, in declaration order.
    pub fn all() -> impl Iterator<Item = NamedGenerator> {
        std::iter::successors(Some(Self::Fib), |g| g.after())
    }

    /// The generator after `self` in [`Self::all`], `None` after the
    /// last. The match is exhaustive, so a new variant does not
    /// compile until it has a place in the walk.
    fn after(self) -> Option<Self> {
        use NamedGenerator as G;
        match self {
            G::Fib => Some(G::FibUntil),
            G::FibUntil => Some(G::Pow2),
            G::Pow2 => Some(G::Pow2Until),
            G::Pow2Until => Some(G::Binomial),
            G::Binomial => Some(G::Geometric),
            G::Geometric => Some(G::GeometricUntil),
            G::GeometricUntil => Some(G::LinearStarts),
            G::LinearStarts => Some(G::LinearSteps),
            G::LinearSteps => Some(G::LogSteps),
            G::LogSteps => None,
        }
    }

    /// The call signature, as written in source position.
    pub fn signature(self) -> &'static str {
        use NamedGenerator as G;
        match self {
            G::Fib => "fib(n)",
            G::FibUntil => "fib_until(max)",
            G::Pow2 => "pow2(n)",
            G::Pow2Until => "pow2_until(max)",
            G::Binomial => "binomial(n)",
            G::Geometric => "geometric(start, factor, n)",
            G::GeometricUntil => "geometric_until(start, factor, max)",
            G::LinearStarts => "linear_starts(start, end, n)",
            G::LinearSteps => "linear_steps(start, end, n)",
            G::LogSteps => "log_steps(start, end, n)",
        }
    }

    /// The name a call is written with: the signature up to `(`.
    pub fn name(self) -> &'static str {
        let sig = self.signature();
        &sig[..sig.find('(').unwrap_or(sig.len())]
    }

    /// The parameter names, from the signature.
    fn params(self) -> Vec<&'static str> {
        let sig = self.signature();
        let inner = &sig[self.name().len() + 1..sig.len() - 1];
        inner.split(',').map(str::trim).collect()
    }

    /// The generator a call name names, if any.
    pub fn from_name(name: &str) -> Option<Self> {
        Self::all().find(|g| g.name() == name)
    }

    /// The generator `text` calls, when it is a call of one
    /// (`fib({n})`, `log_steps(1, 1000, 4)`).
    pub fn of_call(text: &str) -> Option<Self> {
        parse_func_call(text).and_then(|(name, _)| Self::from_name(name))
    }

    /// Whether the generator's values are integers (`u64`); the others
    /// yield floats (`f64`).
    pub fn yields_integers(self) -> bool {
        use NamedGenerator as G;
        match self {
            G::Fib | G::FibUntil | G::Pow2 | G::Pow2Until | G::Binomial => true,
            G::Geometric | G::GeometricUntil | G::LinearStarts | G::LinearSteps | G::LogSteps => {
                false
            }
        }
    }

    /// Expand the call's argument texts into the generator's values.
    fn expand(self, args: &[&str]) -> Result<Vec<Value>, String> {
        use NamedGenerator as G;
        let params = self.params();
        if args.len() != params.len() {
            return Err(format!(
                "{}: expected {} argument{}, got {}",
                self.signature(),
                params.len(),
                if params.len() == 1 { "" } else { "s" },
                args.len()
            ));
        }
        let what = |i: usize| format!("{}.{}", self.name(), params[i]);
        let int = |i: usize| parse_u64_arg(args[i], &what(i));
        let num = |i: usize| parse_num_arg(args[i], &what(i));
        let call = format!("{}({})", self.name(), args.join(", "));
        match self {
            G::Fib => generate_fib_n(int(0)?, &call),
            G::FibUntil => Ok(generate_fib_until(int(0)?)),
            G::Pow2 => generate_pow2_n(int(0)?, &call),
            G::Pow2Until => Ok(generate_pow2_until(int(0)?)),
            G::Binomial => generate_binomial(int(0)?, &call),
            G::Geometric => {
                let (start, factor) = (num(0)?, num(1)?);
                if !(factor.is_finite() && factor > 0.0) {
                    return Err(format!(
                        "{}: expected a positive, finite number, got {factor}",
                        what(1)
                    ));
                }
                generate_geometric(start, factor, int(2)?)
            }
            G::GeometricUntil => {
                let (start, factor) = (num(0)?, num(1)?);
                if !(factor.is_finite() && factor > 1.0) {
                    return Err(format!(
                        "{}: expected a finite number greater than 1, got {factor}",
                        what(1)
                    ));
                }
                Ok(generate_geometric_until(start, factor, num(2)?))
            }
            G::LinearStarts => generate_linear_points(num(0)?, num(1)?, int(2)?, false),
            G::LinearSteps => generate_linear_points(num(0)?, num(1)?, int(2)?, true),
            G::LogSteps => {
                let (start, end) = (num(0)?, num(1)?);
                for (i, bound) in [(0, start), (1, end)] {
                    if bound.is_nan() || bound <= 0.0 {
                        return Err(format!(
                            "{}: expected a positive number, got {bound}",
                            what(i)
                        ));
                    }
                }
                generate_log_steps(start, end, int(2)?)
            }
        }
    }

    /// The largest argument a call yields every term of, for a
    /// generator whose terms outgrow `u64` as its argument grows:
    /// `fib` 93, `pow2` 64, `binomial` 67. `None` for the others.
    pub fn largest_valid_argument(self) -> Option<u64> {
        use NamedGenerator as G;
        match self {
            G::Fib => Some(FIB_MAX_N),
            G::Pow2 => Some(POW2_MAX_N),
            G::Binomial => Some(BINOMIAL_MAX_N),
            _ => None,
        }
    }
}

/// The largest `n` whose first `n` Fibonacci numbers fit `u64`: term
/// 94 is `19740274219868223167`, past `u64::MAX`.
const FIB_MAX_N: u64 = 93;
/// The largest `n` whose first `n` powers of two fit `u64`: term 65 is
/// `2^64`.
const POW2_MAX_N: u64 = 64;
/// The largest row of Pascal's triangle whose coefficients all fit
/// `u64`: `C(68, 31)` is the first coefficient of row 68 past
/// `u64::MAX`.
const BINOMIAL_MAX_N: u64 = 67;

/// Expand a named generator call (`fib(8)`, `linear_steps(0, 1, 4)`)
/// into its values. Returns `Ok(None)` when the text is not a call of
/// a [`NamedGenerator`] (the caller falls through to the set-op,
/// sequencer, and const-eval paths).
fn try_eval_generator(text: &str) -> Result<Option<Vec<Value>>, String> {
    let Some((name, args)) = parse_func_call(text) else {
        return Ok(None);
    };
    let Some(generator) = NamedGenerator::from_name(name) else {
        return Ok(None);
    };
    generator.expand(&split_args_top_level(args)).map(Some)
}

/// The refusal of the first named generator call in `text` that fails,
/// looking through the arguments of the calls that enclose it
/// (`concat(fib(94), 1..3)`), after `{name}` interpolation against
/// `kernel`. A named generator's values depend on nothing but its
/// arguments, so once they are resolved its failure is the call's
/// error wherever it is evaluated (comprehension_forms.md §3.1.3).
/// `None` when every named generator call in `text` expands, when an
/// interpolation does not resolve, or when `text` calls none.
pub fn refused_generator_call(text: &str, kernel: &dyn Lookup) -> Option<String> {
    let interpolated = crate::kernel::interp::interpolate_with_lookup(text, |name| {
        kernel.lookup(name).map(|v| v.to_display_string())
    })
    .ok()?;
    first_refused_call(&interpolated)
}

fn first_refused_call(text: &str) -> Option<String> {
    let (name, args) = parse_func_call(text)?;
    let args = split_args_top_level(args);
    match NamedGenerator::from_name(name) {
        Some(generator) => generator.expand(&args).err(),
        None => args.iter().find_map(|a| first_refused_call(a)),
    }
}

/// First `n` Fibonacci numbers: 1, 1, 2, 3, 5, 8, ...
///
/// Term 94 is past `u64::MAX`, so `n` above 93 is refused before any
/// term is computed.
fn generate_fib_n(n: u64, call: &str) -> Result<Vec<Value>, String> {
    if n > FIB_MAX_N {
        return Err(format!(
            "{call}: term {} is past u64::MAX; fib.n is at most {FIB_MAX_N}",
            FIB_MAX_N + 1
        ));
    }
    let mut out = Vec::with_capacity(n as usize);
    // The pair runs two terms ahead of the last one pushed, past
    // `u64::MAX` at the end of `fib(93)`, so it is held in `u128`.
    let (mut a, mut b): (u128, u128) = (1, 1);
    for _ in 0..n {
        out.push(Value::U64(a as u64));
        (a, b) = (b, a + b);
    }
    Ok(out)
}

/// Fibonacci values up to and including the largest ≤ `max`.
///
/// The pair is held in `u128`, so the walk reaches the 93rd term,
/// the largest in `u64`, even though the term after it does not fit.
fn generate_fib_until(max: u64) -> Vec<Value> {
    let mut out = Vec::new();
    let (mut a, mut b): (u128, u128) = (1, 1);
    while a <= u128::from(max) {
        out.push(Value::U64(a as u64));
        (a, b) = (b, a + b);
    }
    out
}

/// `1, 2, 4, ..., 2^(n-1)`. Term 65 is `2^64`, past `u64::MAX`, so `n`
/// above 64 is refused.
fn generate_pow2_n(n: u64, call: &str) -> Result<Vec<Value>, String> {
    if n > POW2_MAX_N {
        return Err(format!(
            "{call}: term {}, 2^64, is past u64::MAX; pow2.n is at most {POW2_MAX_N}",
            POW2_MAX_N + 1
        ));
    }
    Ok((0..n).map(|i| Value::U64(1u64 << i)).collect())
}

/// Powers of two ≤ max.
fn generate_pow2_until(max: u64) -> Vec<Value> {
    let mut out = Vec::new();
    let mut v: u64 = 1;
    loop {
        if v > max {
            break;
        }
        out.push(Value::U64(v));
        v = match v.checked_mul(2) {
            Some(x) => x,
            None => break,
        };
    }
    out
}

/// `start, start*factor, start*factor², …` (n terms).
fn generate_geometric(start: f64, factor: f64, n: u64) -> Result<Vec<Value>, String> {
    let mut out = crate::derive_support::try_buffer_for(n, "geometric(start, factor, n)")?;
    let mut v = start;
    for _ in 0..n {
        out.push(Value::F64(v));
        v *= factor;
    }
    Ok(out)
}

/// `start, start*factor, …` ≤ max. The caller refuses a `factor` that
/// is not a finite number above 1, so a positive `start` grows past
/// `max`; a `start` of zero or less never does and yields nothing.
fn generate_geometric_until(start: f64, factor: f64, max: f64) -> Vec<Value> {
    let mut out = Vec::new();
    let mut v = start;
    if start.is_nan() || start <= 0.0 {
        return out;
    }
    while v <= max {
        out.push(Value::F64(v));
        v *= factor;
    }
    out
}

/// Binomial coefficients `C(n, 0), C(n, 1), …, C(n, n)`. Every
/// coefficient of rows up to 67 fits `u64`, and every later row has one
/// past `u64::MAX`, so a row past 67 is refused, naming its first
/// coefficient that does not fit. The walk reaches that coefficient
/// within a few terms for any large `n`: `binomial(10^12)` fails at
/// `C(n, 2)`, not after a trillion terms.
fn generate_binomial(n: u64, call: &str) -> Result<Vec<Value>, String> {
    let mut out = Vec::with_capacity(n.min(BINOMIAL_MAX_N) as usize + 1);
    // `c ≤ u64::MAX` before each step, so `c · (n − k + 1)` fits u128.
    let mut c: u128 = 1;
    out.push(Value::U64(1));
    for k in 1..=n {
        c = c * u128::from(n - k + 1) / u128::from(k);
        if c > u128::from(u64::MAX) {
            return Err(format!(
                "{call}: term C({n}, {k}) is past u64::MAX; binomial.n is at most \
                 {BINOMIAL_MAX_N}"
            ));
        }
        out.push(Value::U64(c as u64));
    }
    Ok(out)
}

/// Kernel-aware partition comprehension sources
/// (cursor_partitions.md §7.1).
///
/// `subdivide(<ident>, n)` — resolve `<ident>` through the
/// kernel's scope chain to a `Partition` (typically an iter-var
/// bound by an enclosing `for:` clause) and split it into `n`
/// sub-partitions, same boundary math as the `subdivide(p, n)`
/// node in polydat-nodes and the `*/N` spec token:
///
/// ```yaml
/// - for: "outer in partitions(\"50%,*\", 1000)"
///   phases:
///     - for: "inner in subdivide(outer, 5)"
///       phases: [walk]
/// ```
///
/// When the ident does not resolve and `probe` is set (synthesis-time
/// pre-evaluation probes the clause before the outer iteration
/// installs its value), a single placeholder partition is
/// returned so iter-var type detection still classifies the
/// variable as `ext`. At evaluation (`probe` clear) the source reads
/// the ident as any source reads a name: unbound or bound to None, it
/// reads None ([`SpecError::ReadsNone`]).
fn try_eval_partition_call(
    text: &str,
    kernel: &dyn Lookup,
    probe: bool,
) -> Result<Option<Vec<Value>>, SpecError> {
    let Some((name, args)) = parse_func_call(text) else {
        return Ok(None);
    };
    let arg_list = split_args_top_level(args);
    match name {
        "subdivide" => {
            if arg_list.len() != 2 {
                return Err(format!(
                    "subdivide(p, n): expected 2 arguments (a partition and a count), got {}",
                    arg_list.len()
                )
                .into());
            }
            let src = arg_list[0].trim();
            let n = parse_u64_arg(arg_list[1], "subdivide.n")?;
            if probe && kernel.lookup(src).is_none() {
                // Pre-evaluation probe: the outer iter-var isn't
                // installed yet. Return one placeholder so the clause's
                // iter-var type-detects as `ext`; real values arrive at
                // runtime dispatch.
                return Ok(Some(vec![placeholder_partition()]));
            }
            let value = read_name(kernel, src, || {
                format!("subdivide({src}, {n}): `{src}` is not bound in scope")
            })?;
            let Some(p) = value.as_partition().copied() else {
                return Err(format!(
                    "subdivide({src}, {n}): `{src}` resolved to {} — expected a \
                     Partition value (an iter-var from `for: \"p in partitions(...)\"` \
                     or a cursor's `.cursor` projection)",
                    value.to_display_string(),
                )
                .into());
            };
            let subs = crate::iteration::cursor_partition::subdivide_partition(&p, n)?;
            Ok(Some(subs.into_iter().map(Value::from_partition).collect()))
        }
        // Desugaring of an explicit `partitions(spec, [extent])` source
        // in comprehension position (cursor_partitions.md §3, §7.1). The spec string is in a
        // comprehension position, so it is parsed + resolved HERE, on the
        // Result path — a bad spec (over-sum list, bad recipe/order/window,
        // malformed tail) surfaces a clean comprehension error rather than the
        // `partitions()` node's eval-time `panic!` (which const-fold swallows
        // into a misleading downstream type mismatch). The node is unchanged;
        // a spec in comprehension position simply never reaches its eval.
        // Default extent 100 (pct space) matches the node; the cursor's
        // `over p` re-scales each partition to its declared range.
        "partitions" => {
            if arg_list.is_empty() || arg_list.len() > 2 {
                return Err(format!(
                    "partitions(spec, [extent]): expected 1 or 2 arguments, got {}",
                    arg_list.len(),
                )
                .into());
            }
            let spec = resolve_partition_spec_arg(arg_list[0], kernel)?;
            let extent = match arg_list.get(1) {
                Some(a) => parse_u64_arg(a, "partitions.extent")?,
                None => 100,
            };
            Ok(Some(desugar_partition_spec(
                &spec,
                extent,
                "comprehension source `partitions(...)`",
            )?))
        }
        // Profile-driven partition source: `profile_partitions(dataset,
        // pattern)` cuts the dataset's vector space at the cumulative
        // sizes of the profiles matching `pattern`, one partition per
        // masked tier (see `library::vectors::build_profile_partitions`).
        // Resolved here (like `partitions`/`subdivide`) so the iter-var
        // type-detects as a partition even when the dataset can't be
        // const-folded at compile time: a resolvable group yields the
        // real tiers; an unresolvable one (a compile-time probe, or a
        // catalog miss surfaced later by the prebuffer) yields a single
        // placeholder so the iter-var still types as `ext`.
        "profile_partitions" => {
            #[cfg(not(feature = "vectordata"))]
            {
                Err("profile_partitions requires the `vectordata` Cargo feature"
                    .to_string()
                    .into())
            }

            #[cfg(feature = "vectordata")]
            {
                if arg_list.len() != 2 {
                    return Err(format!(
                        "profile_partitions(dataset, pattern): expected 2 arguments, got {}",
                        arg_list.len()
                    )
                    .into());
                }
                // Both args are literal strings after `{...}` interpolation;
                // strip matching outer quotes.
                let strip = |s: &str| -> String {
                    let s = s.trim();
                    let b = s.as_bytes();
                    if b.len() >= 2 && (b[0] == b'\'' || b[0] == b'"') && b[b.len() - 1] == b[0] {
                        s[1..s.len() - 1].to_string()
                    } else {
                        s.to_string()
                    }
                };
                let dataset = strip(arg_list[0]);
                let pattern = strip(arg_list[1]);
                match crate::library::vectors::load_dataset_group(&dataset) {
                    Ok(group) => {
                        let parts =
                            crate::library::vectors::build_profile_partitions(&group, &pattern);
                        Ok(Some(parts.into_iter().map(Value::from_partition).collect()))
                    }
                    Err(_) => {
                        // Probe / dataset unavailable: one placeholder so the
                        // clause's iter-var type-detects as `ext`. Real tiers
                        // arrive once the catalog resolves the group.
                        Ok(Some(vec![placeholder_partition()]))
                    }
                }
            }
        }
        _ => Ok(None),
    }
}

/// The one partition a pre-evaluation probe stands in for a partition
/// source whose values are not installed yet, so the clause's iter-var
/// type-detects as `ext`.
fn placeholder_partition() -> Value {
    Value::from_partition(crate::iteration::cursor_partition::Partition {
        idx: 0,
        count: 1,
        start_ord: 0,
        end_ord: 1,
        start_pct: 0.0,
        end_pct: 100.0,
        base_extent: 1,
    })
}

/// Comprehension-position desugaring (cursor_partitions.md §7.1): a `<ident>.partitions`
/// source (the primary operator sweep flow, `for: "p in cursor.partitions"`).
///
/// In comprehension position a *string* spec desugars per the partition
/// grammar. `<ident>.partitions` resolves `<ident>` against the kernel chain
/// to its spec string — a workload param such as `cursor=linear:4` — and the
/// `.partitions` projection selects the partition-spec desugaring (as opposed
/// to the string→token-list desugaring a bare string source would get),
/// expanding it into the same `PartitionList` that `partitions(spec)` yields.
/// Resolution uses the `partitions(spec)` node's (polydat-nodes) default extent (100, pct
/// space); the cursor's `over p` clause re-scales each partition's percentages
/// to its actual declared range.
///
/// This MUST live here (not in `eval_const_expr`, which is kernel-less and
/// resolves the `cursor.partitions` field-access to `None`): only the
/// comprehension eval has the kernel needed to look the param up.
///
/// Returns `Ok(None)` when `text` is not a `<ident>.partitions` form. When the
/// ident does not resolve and `probe` is set (a pre-evaluation probe before
/// the value is installed), a single placeholder partition is returned so
/// iter-var type detection yields `ext`; at evaluation the ident reads None
/// — the same contract as [`try_eval_partition_call`].
fn try_eval_param_partitions(
    text: &str,
    kernel: &dyn Lookup,
    probe: bool,
) -> Result<Option<Vec<Value>>, SpecError> {
    let Some(ident) = text.trim().strip_suffix(".partitions") else {
        return Ok(None);
    };
    let ident = ident.trim();
    if !is_single_bare_ident(ident) {
        return Ok(None);
    }
    if probe && kernel.lookup(ident).is_none() {
        // Pre-eval probe: the param value isn't installed yet. Return one
        // placeholder so the clause's iter-var type-detects as `ext`.
        return Ok(Some(vec![placeholder_partition()]));
    }
    let value = read_name(kernel, ident, || {
        format!("comprehension source `{ident}.partitions`: `{ident}` is not bound in scope")
    })?;
    // Already a resolved PartitionList → unpack directly.
    if let Some(list) = value.as_partition_list() {
        return Ok(Some(
            list.as_slice()
                .iter()
                .map(|p| Value::from_partition(*p))
                .collect(),
        ));
    }
    // Otherwise it must be a spec string — desugar it per the partition
    // spec language (cursor_partitions.md §3).
    let Value::Str(spec) = &value else {
        return Err(format!(
            "comprehension source `{ident}.partitions`: `{ident}` resolved to \
             {} — expected a partition-spec string (a workload param such as \
             `cursor=linear:4`) or a PartitionList.",
            value.to_display_string(),
        )
        .into());
    };
    Ok(Some(desugar_partition_spec(
        spec,
        100,
        &format!("comprehension source `{ident}.partitions`"),
    )?))
}

/// Parse + resolve a partition spec string into its unpacked partition
/// values, on the Result path. Shared by the comprehension-position
/// desugaring forms (`<ident>.partitions` and `partitions("...")`): a bad
/// spec surfaces a clean error labelled by `ctx` HERE — it never reaches the
/// `partitions()` node's eval-time `panic!`. This is a grammar-position
/// concern (cursor_partitions.md §7.1), so spec validation lives where the
/// spec is recognized.
fn desugar_partition_spec(spec: &str, extent: u64, ctx: &str) -> Result<Vec<Value>, String> {
    let parsed = crate::iteration::cursor_partition::parse(spec)
        .map_err(|e| format!("{ctx}: bad spec `{spec}`: {e}"))?;
    let parts = crate::iteration::cursor_partition::resolve(&parsed, 0, extent)
        .map_err(|e| format!("{ctx}: resolve failed for `{spec}`: {e}"))?;
    Ok(parts.into_iter().map(Value::from_partition).collect())
}

/// Resolve a `partitions(...)` spec argument to its string form: a quoted
/// string literal yields its inner text; a bare identifier resolves against
/// the kernel chain to its string value; anything else is taken verbatim (an
/// unquoted spec such as a raw percentage list).
fn resolve_partition_spec_arg(arg: &str, kernel: &dyn Lookup) -> Result<String, SpecError> {
    let a = arg.trim();
    if a.len() >= 2
        && ((a.starts_with('"') && a.ends_with('"')) || (a.starts_with('\'') && a.ends_with('\'')))
    {
        return Ok(a[1..a.len() - 1].to_string());
    }
    if is_single_bare_ident(a) {
        return match read_name(kernel, a, || {
            format!("partitions(...): `{a}` did not resolve to a spec string in scope")
        })? {
            Value::Str(s) => Ok(s.to_string()),
            other => Err(format!(
                "partitions(...): `{a}` resolved to {} — expected a spec string",
                other.to_display_string(),
            )
            .into()),
        };
    }
    Ok(a.to_string())
}

/// Evenly spaced numeric points over `[start, end]`.
///
/// Half-open form (`linear_starts`): the start of each of `n`
/// equal subdivisions of `[start, end)` — `end` is never
/// emitted. Inclusive form (`linear_steps`): `n` fence-post
/// points covering `[start, end]`, both ends emitted.
///
/// These yield *values*, not partitions; splitting a
/// `Partition` into sub-partitions is `subdivide(p, n)` in the
/// partition stdlib (cursor_partitions.md §7.3).
fn generate_linear_points(
    start: f64,
    end: f64,
    n: u64,
    inclusive: bool,
) -> Result<Vec<Value>, String> {
    let denom = if inclusive {
        (n.saturating_sub(1)).max(1) as f64
    } else {
        n as f64
    };
    let step = (end - start) / denom;
    // Not `(0..n).collect()`: a `u64` range reports its exact length,
    // so collecting reserves all `n` up front and a count from the
    // spec text no machine can hold aborts the process.
    let mut out = crate::derive_support::try_buffer_for(n, "linear points")?;
    out.extend((0..n).map(|i| Value::F64(start + step * i as f64)));
    // The inclusive form's last point is `end` itself, not the sum
    // that rounds near it.
    if inclusive && n >= 2 {
        out[n as usize - 1] = Value::F64(end);
    }
    Ok(out)
}

/// `n` log-spaced points from `start` to `end`, both emitted exactly
/// as given; the points between are `exp` of evenly spaced logarithms.
/// The caller refuses a bound that is not positive.
fn generate_log_steps(start: f64, end: f64, n: u64) -> Result<Vec<Value>, String> {
    if n == 0 {
        return Ok(Vec::new());
    }
    if n == 1 {
        return Ok(vec![Value::F64(start)]);
    }
    let log_s = start.ln();
    let log_e = end.ln();
    let step = (log_e - log_s) / (n - 1) as f64;
    let mut out = crate::derive_support::try_buffer_for(n, "log_steps(start, end, n)")?;
    out.push(Value::F64(start));
    out.extend((1..n - 1).map(|i| Value::F64((log_s + step * i as f64).exp())));
    out.push(Value::F64(end));
    Ok(out)
}

// ============================================================
// Set operators (comprehension_forms.md §3.1)
// ============================================================

/// Recognise `concat(...)`, `unique(...)`, etc. Each set op
/// recursively evaluates its arguments as sources
/// (so `concat(1..10, fib(8))` works), then combines the
/// resulting lists. An argument that reads None makes the whole
/// source read None.
fn try_eval_setop(text: &str, kernel: &dyn Lookup) -> Result<Option<Vec<Value>>, SpecError> {
    let Some((name, args)) = parse_func_call(text) else {
        return Ok(None);
    };
    let arg_texts = split_args_top_level(args);
    let recursively_evaluate = |t: &str| evaluate_spec_internal(t, kernel);
    match name {
        "concat" => {
            let mut out = Vec::new();
            for a in &arg_texts {
                out.extend(recursively_evaluate(a)?);
            }
            Ok(Some(out))
        }
        "unique" => {
            let mut out: Vec<Value> = Vec::new();
            for a in &arg_texts {
                for v in recursively_evaluate(a)? {
                    if !out.contains(&v) {
                        out.push(v);
                    }
                }
            }
            Ok(Some(out))
        }
        "intersect" => {
            if arg_texts.is_empty() {
                return Ok(Some(Vec::new()));
            }
            let first = recursively_evaluate(arg_texts[0])?;
            let mut out: Vec<Value> = Vec::new();
            for v in first {
                let mut in_all = true;
                for a in &arg_texts[1..] {
                    let other = recursively_evaluate(a)?;
                    if !other.contains(&v) {
                        in_all = false;
                        break;
                    }
                }
                if in_all && !out.contains(&v) {
                    out.push(v);
                }
            }
            Ok(Some(out))
        }
        "subtract" => {
            if arg_texts.len() != 2 {
                return Err(
                    format!("subtract(a, b): expected 2 args, got {}", arg_texts.len()).into(),
                );
            }
            let a = recursively_evaluate(arg_texts[0])?;
            let b = recursively_evaluate(arg_texts[1])?;
            Ok(Some(a.into_iter().filter(|v| !b.contains(v)).collect()))
        }
        "interleave" => {
            let lists: Result<Vec<Vec<Value>>, SpecError> =
                arg_texts.iter().map(|a| recursively_evaluate(a)).collect();
            let lists = lists?;
            let mut out = Vec::new();
            let max_len = lists.iter().map(|l| l.len()).max().unwrap_or(0);
            for i in 0..max_len {
                for l in &lists {
                    if let Some(v) = l.get(i) {
                        out.push(v.clone());
                    }
                }
            }
            Ok(Some(out))
        }
        "cycle" => {
            if arg_texts.len() != 2 {
                return Err(
                    format!("cycle(a, n): expected 2 args, got {}", arg_texts.len()).into(),
                );
            }
            let a = recursively_evaluate(arg_texts[0])?;
            let n = parse_u64_arg(arg_texts[1], "cycle.n")?;
            let total = (a.len() as u64).checked_mul(n).ok_or_else(|| {
                format!(
                    "cycle(a, n): {} values repeated {n} times is more than can be counted",
                    a.len()
                )
            })?;
            let mut out = crate::derive_support::try_buffer_for(total, "cycle(a, n)")?;
            for _ in 0..n {
                out.extend(a.iter().cloned());
            }
            Ok(Some(out))
        }
        "reverse" => {
            if arg_texts.len() != 1 {
                return Err(format!("reverse(a): expected 1 arg, got {}", arg_texts.len()).into());
            }
            let mut a = recursively_evaluate(arg_texts[0])?;
            a.reverse();
            Ok(Some(a))
        }
        "take" => {
            if arg_texts.len() != 2 {
                return Err(format!("take(a, n): expected 2 args, got {}", arg_texts.len()).into());
            }
            let a = recursively_evaluate(arg_texts[0])?;
            let n = parse_u64_arg(arg_texts[1], "take.n")?;
            Ok(Some(a.into_iter().take(n as usize).collect()))
        }
        "skip" => {
            if arg_texts.len() != 2 {
                return Err(format!("skip(a, n): expected 2 args, got {}", arg_texts.len()).into());
            }
            let a = recursively_evaluate(arg_texts[0])?;
            let n = parse_u64_arg(arg_texts[1], "skip.n")?;
            Ok(Some(a.into_iter().skip(n as usize).collect()))
        }
        _ => Ok(None),
    }
}

// ============================================================
// Sequencer expansions (comprehension_forms.md §3.1): bucket /
// concat_seq / interval_seq, a lookup-table facility reusing the
// op-sequencing algorithms.
// ============================================================

/// Recognise `bucket(items, ratios)` / `bucket("3:a, 1:b")`,
/// `concat_seq(...)`, `interval_seq(...)`. Reuses the
/// algorithms from the host's op-sequencing.
///
/// The algorithms aren't exposed cross-crate as raw functions
/// today, so we re-implement the small set we need here. The
/// outputs match `build_bucket_lut` / `build_concat_lut` /
/// `build_interval_lut` byte-for-byte (covered by the
/// the host's op-sequencing tests).
fn try_eval_sequencer(text: &str, kernel: &dyn Lookup) -> Result<Option<Vec<Value>>, SpecError> {
    let Some((name, args)) = parse_func_call(text) else {
        return Ok(None);
    };
    if !matches!(name, "bucket" | "concat_seq" | "interval_seq") {
        return Ok(None);
    }
    let arg_texts = split_args_top_level(args);

    // Two acceptable shapes:
    //   1. Single string arg: ratio-prefix shorthand
    //      `"3:ann, 1:scan, 2:fetch"`.
    //   2. Two list args: items + ratios in lockstep.
    let (items, ratios): (Vec<Value>, Vec<usize>) = match arg_texts.len() {
        1 => parse_ratio_prefix_shorthand(arg_texts[0])?,
        2 => {
            let items = evaluate_spec_internal(arg_texts[0], kernel)?;
            let raw_ratios = evaluate_spec_internal(arg_texts[1], kernel)?;
            let ratios: Result<Vec<usize>, String> = raw_ratios
                .iter()
                .map(|v| match v {
                    Value::U64(n) => Ok(*n as usize),
                    other => Err(format!(
                        "{name}: ratio must be non-negative integer, got {other:?}"
                    )),
                })
                .collect();
            (items, ratios?)
        }
        _ => {
            return Err(format!(
                "{name}: expected `(items, ratios)` or `(\"r1:item1, r2:item2, ...\")`; got {} args",
                arg_texts.len()
            )
            .into());
        }
    };

    if items.len() != ratios.len() {
        return Err(format!(
            "{name}: items.len() ({}) != ratios.len() ({})",
            items.len(),
            ratios.len(),
        )
        .into());
    }
    // The output length is the sum of the ratios, which come from the
    // spec text: summed checked, and reserved fallibly, so an absurd
    // ratio is this error rather than a wrapped sum or an abort.
    let total = ratios
        .iter()
        .try_fold(0usize, |acc, &r| acc.checked_add(r))
        .ok_or_else(|| format!("{name}: the ratios sum past what can be counted"))?;
    let out = crate::derive_support::try_buffer_for(total as u64, name)?;
    Ok(Some(match name {
        "bucket" => seq_bucket(&items, &ratios, total, out),
        "concat_seq" => seq_concat(&items, &ratios, out),
        "interval_seq" => seq_interval(&items, &ratios, total, out),
        _ => unreachable!(),
    }))
}

/// Parse `"r1:item1, r2:item2, …"`. Each element is a
/// ratio (positive integer) and an item value separated
/// by `:`. The string itself comes through `evaluate_spec`
/// — typically as a quoted string literal.
fn parse_ratio_prefix_shorthand(text: &str) -> Result<(Vec<Value>, Vec<usize>), String> {
    // The arg might be a literal `"3:a, 1:b"` (with quotes
    // in the source) or already-stripped `3:a, 1:b`.
    let stripped = text
        .trim()
        .trim_start_matches(['"', '\''])
        .trim_end_matches(['"', '\'']);
    let mut items = Vec::new();
    let mut ratios = Vec::new();
    for part in stripped.split(',') {
        let part = part.trim();
        if part.is_empty() {
            continue;
        }
        let (r, i) = part
            .split_once(':')
            .ok_or_else(|| format!("ratio-prefix shorthand: missing ':' in '{part}'"))?;
        let ratio: usize = r.trim().parse().map_err(|_| {
            format!("ratio-prefix shorthand: ratio '{r}' is not a non-negative integer")
        })?;
        ratios.push(ratio);
        items.push(parse_one_value(i.trim()));
    }
    Ok((items, ratios))
}

fn parse_one_value(s: &str) -> Value {
    if let Ok(n) = s.parse::<u64>() {
        return Value::U64(n);
    }
    if let Ok(f) = s.parse::<f64>() {
        return Value::F64(f);
    }
    if s == "true" {
        return Value::Bool(true);
    }
    if s == "false" {
        return Value::Bool(false);
    }
    Value::Str(s.to_string().into())
}

/// Bucket sequencer: round-robin from per-item buckets sized
/// by ratio. Output length = sum(ratios), which is `total`; `out` is
/// reserved for it.
fn seq_bucket(items: &[Value], ratios: &[usize], total: usize, mut out: Vec<Value>) -> Vec<Value> {
    let mut remaining: Vec<usize> = ratios.to_vec();
    while out.len() < total {
        let mut emitted_any = false;
        for (i, item) in items.iter().enumerate() {
            if remaining[i] > 0 {
                out.push(item.clone());
                remaining[i] -= 1;
                emitted_any = true;
            }
        }
        if !emitted_any {
            break;
        }
    }
    out
}

/// Concat sequencer: contiguous runs (all of item 1, then
/// all of item 2, …).
fn seq_concat(items: &[Value], ratios: &[usize], mut out: Vec<Value>) -> Vec<Value> {
    for (item, &r) in items.iter().zip(ratios.iter()) {
        for _ in 0..r {
            out.push(item.clone());
        }
    }
    out
}

/// Interval sequencer: evenly spaced occurrences of each
/// item across the output. Picks each output position from
/// the item with the largest "weight × position - already
/// emitted" — same algorithm as op-sequencing's
/// build_interval_lut.
fn seq_interval(
    items: &[Value],
    ratios: &[usize],
    total: usize,
    mut out: Vec<Value>,
) -> Vec<Value> {
    if total == 0 {
        return out;
    }
    let mut emitted: Vec<usize> = vec![0; items.len()];
    for slot in 0..total {
        // Pick the item whose target ratio is most under-met
        // at this slot. Target at slot k = (ratio_i * (k+1)) / total.
        let mut best = 0usize;
        let mut best_deficit: f64 = f64::NEG_INFINITY;
        for i in 0..items.len() {
            let target = ratios[i] as f64 * (slot + 1) as f64 / total as f64;
            let deficit = target - emitted[i] as f64;
            if deficit > best_deficit {
                best_deficit = deficit;
                best = i;
            }
        }
        out.push(items[best].clone());
        emitted[best] += 1;
    }
    out
}

/// Map a `Value` to the canonical polydat extern type keyword.
///
/// Delegates to [`Value::port_type`] + [`PortType::to_keyword`](crate::ast::PortType::to_keyword) —
/// the single source of truth for the str↔PortType table. The
/// returned keyword round-trips byte-cleanly through
/// [`PortType::from_keyword`](crate::ast::PortType::from_keyword) in the DSL extern parser, so every
/// typed `Value` variant (including `VecF32`, `Bytes`, `Json`,
/// `Handle`) becomes a precisely-typed input on the synthesized
/// inner kernel.
pub fn value_to_polydat_type_name(v: &Value) -> &'static str {
    v.port_type().to_keyword()
}

// Expand `{name}` placeholders in `text`, resolving each leaf
// placeholder against `kernel`'s in-scope name space.
//
// `interpolate_via_kernel`, `interpolate_with_lookup`, and the
// internal `one_pass` / `first_unresolved` / `unescape` helpers
// live in `crate::kernel::interp`.
// `interpolate_via_kernel` and `interpolate_with_lookup` are
// imported above for internal use; external callers use the
// `polydat::kernel::interp` module directly.

#[cfg(test)]
mod tests {
    use super::*;
    use crate::kernel::PolydatKernel;
    use crate::kernel::interp::interpolate_via_kernel;

    fn h(pairs: &[(&str, &str)]) -> HashMap<String, String> {
        pairs
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect()
    }

    fn interpolate(
        text: &str,
        bindings: &HashMap<String, String>,
        workload_params: &HashMap<String, String>,
    ) -> Result<String, String> {
        interpolate_with_lookup(text, |name| {
            bindings
                .get(name)
                .or_else(|| workload_params.get(name))
                .cloned()
        })
    }

    #[test]
    fn flat_substitution() {
        let params = h(&[("dataset", "example"), ("prefix", "label")]);
        let out = interpolate("matching('{dataset}', '{prefix}')", &h(&[]), &params).unwrap();
        assert_eq!(out, "matching('example', 'label')");
    }

    #[test]
    fn bindings_shadow_params() {
        let params = h(&[("profile", "default")]);
        let bindings = h(&[("profile", "label_07")]);
        let out = interpolate("vec_{profile}", &bindings, &params).unwrap();
        assert_eq!(out, "vec_label_07");
    }

    #[test]
    fn nested_placeholder_resolves_inside_out() {
        let params = h(&[("k_1_limits", "1,2,4,8"), ("k_10_limits", "10,20,30")]);
        let bindings = h(&[("k", "1")]);
        let out = interpolate("{k_{k}_limits}", &bindings, &params).unwrap();
        assert_eq!(out, "1,2,4,8");
    }

    #[test]
    fn deeply_nested() {
        let params = h(&[("a_b_c", "WIN")]);
        let bindings = h(&[("x", "a"), ("y", "b"), ("z", "c")]);
        let out = interpolate("{{x}_{y}_{z}}", &bindings, &params).unwrap();
        assert_eq!(out, "WIN");
    }

    #[test]
    fn escape_emits_literal_brace() {
        let out = interpolate("\\{not_a_var\\}", &h(&[]), &h(&[])).unwrap();
        assert_eq!(out, "{not_a_var}");
    }

    #[test]
    fn escape_inside_otherwise_resolved_text() {
        let params = h(&[("x", "1")]);
        let out = interpolate("a={x} literal=\\{x\\}", &h(&[]), &params).unwrap();
        assert_eq!(out, "a=1 literal={x}");
    }

    #[test]
    fn unresolved_is_hard_error() {
        let err = interpolate("hello {nope}", &h(&[]), &h(&[])).unwrap_err();
        assert!(err.contains("unresolved"));
        assert!(err.contains("nope"));
    }

    #[test]
    fn empty_placeholder_rejected() {
        let err = interpolate("a{}b", &h(&[]), &h(&[])).unwrap_err();
        assert!(err.contains("empty"));
    }

    #[test]
    fn unmatched_brace_rejected() {
        let err = interpolate("a {x", &h(&[]), &h(&[])).unwrap_err();
        assert!(err.contains("unmatched"));
    }

    #[test]
    fn idempotent_when_no_placeholders() {
        let out = interpolate("plain text", &h(&[]), &h(&[])).unwrap();
        assert_eq!(out, "plain text");
    }

    #[test]
    fn resolved_value_with_braces_does_not_re_expand() {
        let params = h(&[("greeting", "hello {planet}")]);
        let err = interpolate("{greeting}", &h(&[]), &params).unwrap_err();
        assert!(err.contains("planet"));
    }

    #[test]
    fn cyclic_placeholders_hit_round_cap() {
        let params = h(&[("a", "{b}"), ("b", "{a}")]);
        let err = interpolate("{a}", &h(&[]), &params).unwrap_err();
        assert!(err.contains("did not stabilize") || err.contains("rounds"));
    }

    #[test]
    fn kernel_resolves_via_get_constant() {
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const dataset := \"example\"\n")
                .unwrap();
        let out = interpolate_via_kernel("path/{dataset}/data", &kernel).unwrap();
        assert_eq!(out, "path/example/data");
    }

    #[test]
    fn kernel_resolves_via_get_input() {
        let parent =
            crate::dsl::compile::compile_polydat_interpreter("const k_values := \"1, 10\"\n")
                .unwrap();
        let child_program =
            crate::dsl::compile::compile_polydat_interpreter("extern k_values: String\n")
                .unwrap()
                .program()
                .clone();
        let child = parent.materialize_subscope(child_program, &[]);
        let out = interpolate_via_kernel("values={k_values}", &child).unwrap();
        assert_eq!(out, "values=1, 10");
    }

    #[test]
    fn kernel_unresolved_name_errors() {
        let kernel = crate::dsl::compile::compile_polydat_interpreter("const x := 1\n").unwrap();
        let err = interpolate_via_kernel("hello {nope}", &kernel)
            .unwrap_err()
            .to_string();
        assert!(err.contains("unresolved"));
        assert!(err.contains("nope"));
    }

    #[test]
    fn kernel_nested_template_iterates_to_fixed_point() {
        let kernel = crate::dsl::compile::compile_polydat_interpreter(
            "const k := \"1\"\nconst k_1_limits := \"1, 2, 4, 8\"\n",
        )
        .unwrap();
        let out = interpolate_via_kernel("{k_{k}_limits}", &kernel).unwrap();
        assert_eq!(out, "1, 2, 4, 8");
    }

    #[test]
    fn parse_list_native_types() {
        let v = parse_list_with_types("1, 10, 100");
        assert_eq!(v, vec![Value::U64(1), Value::U64(10), Value::U64(100)]);
    }

    #[test]
    fn parse_list_mixed_types() {
        let v = parse_list_with_types("1, 1.5, true, hello");
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::F64(1.5),
                Value::Bool(true),
                Value::Str("hello".to_string().into()),
            ]
        );
    }

    #[test]
    fn all_cursor_returns_extent_range() {
        // Simulate a cursor declaration at the parent scope by
        // exposing the auxiliary extent outputs as folded
        // constants. The real cursor compiler emits these via
        // `__cursor_extent_<name>_{start,end}` outputs; for this
        // test we synthesize them directly.
        let kernel = crate::dsl::compile::compile_polydat_interpreter(
            "const __cursor_extent_row_start := 0\n\
             const __cursor_extent_row_end := 5\n",
        )
        .unwrap();
        let values = evaluate_spec("all(row)", &kernel).unwrap();
        assert_eq!(
            values,
            vec![
                Value::U64(0),
                Value::U64(1),
                Value::U64(2),
                Value::U64(3),
                Value::U64(4),
            ]
        );
    }

    #[test]
    fn all_cursor_non_zero_start() {
        let kernel = crate::dsl::compile::compile_polydat_interpreter(
            "const __cursor_extent_data_start := 100\n\
             const __cursor_extent_data_end := 103\n",
        )
        .unwrap();
        let values = evaluate_spec("all(data)", &kernel).unwrap();
        assert_eq!(
            values,
            vec![Value::U64(100), Value::U64(101), Value::U64(102)]
        );
    }

    #[test]
    fn all_cursor_missing_extent_errors() {
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const unrelated := 1\n").unwrap();
        let err = evaluate_spec("all(no_such_cursor)", &kernel)
            .unwrap_err()
            .to_string();
        assert!(err.contains("all(no_such_cursor)"));
        assert!(err.contains("no resolvable extent"));
    }

    #[test]
    fn all_cursor_only_matches_exact_shape() {
        // `all(<ident>)` is the only matched shape — anything
        // more complex falls through to the normal eval path.
        // `all(row, 5)` doesn't match the strict shape (the
        // comma breaks the bare-ident requirement), so the
        // pipeline tries to evaluate it as a regular GK
        // expression. There's no registered function named
        // `all`, so eval fails and the failure is propagated as
        // a clean clause-level error rather than split into a
        // literal list.
        let kernel = crate::dsl::compile::compile_polydat_interpreter(
            "const __cursor_extent_row_start := 0\n\
             const __cursor_extent_row_end := 5\n",
        )
        .unwrap();
        let err = evaluate_spec("all(row, 5)", &kernel)
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("all(row, 5)"),
            "error must mention the failing spec, got: {err}"
        );
        assert!(
            err.contains("failed to evaluate") || err.contains("unknown function"),
            "error must explain the eval failure, got: {err}"
        );
    }

    #[test]
    fn missing_dataset_surface_as_clean_error_not_garbage() {
        // A workload runs on a system whose vectordata catalog
        // doesn't have the requested dataset. The source
        //   `profile in matching_profiles('nonexistent_dataset_xyz', 'label_')`
        // produces a clean clause-level error naming the
        // resolution failure, never a garbage iter-var like
        // `matching_profiles('nonexistent_dataset_xyz'`
        // (truncated at the first comma) that would flow
        // downstream into malformed output. Every layer on the
        // way (the dataset open, the handle read, and
        // `evaluate_spec`) propagates an actionable diagnostic.
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const unrelated := 1\n").unwrap();
        let result = evaluate_spec(
            "matching_profiles('nonexistent_dataset_xyz_qqq', 'label_')",
            &kernel,
        );
        let err = result
            .expect_err("missing dataset must surface as Err, not silent literal-list fallback")
            .to_string();
        // Doesn't matter which exact error string we get from
        // the catalog layer — the test guards the *contract*:
        // the spec text appears in the error, the failure is
        // attributed to the dataset / resolver / open path, and
        // it is a Result::Err (not garbage data).
        assert!(
            err.contains("nonexistent_dataset_xyz_qqq")
                || err.contains("matching_profiles")
                || err.contains("dataset"),
            "error must point at the actual fault, got: {err}"
        );
    }

    #[test]
    fn function_call_eval_failure_is_not_silently_split() {
        // Defensive: any text containing `(` is an
        // expression — never a literal list. If eval fails, we
        // must propagate the failure rather than splitting on
        // commas. This guards the broader contract that
        // protected the dataset-resolution case above.
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const unrelated := 1\n").unwrap();
        let err = evaluate_spec("nonexistent_func('a', 'b', 'c')", &kernel)
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("failed to evaluate") || err.contains("unknown"),
            "expected a clean eval-failure error, got: {err}"
        );
    }

    #[test]
    fn literal_list_path_still_works() {
        // Counter-case: a plain comma-separated list of
        // literals (no parens, no operators) MUST still work
        // through the literal-list fallback after eval fails
        // (which it should — `1, 10, 100` isn't a single GK
        // expression). This is the legitimate use case that the
        // fallback exists for.
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const unrelated := 1\n").unwrap();
        let values = evaluate_spec("1, 10, 100", &kernel).unwrap();
        assert_eq!(values, vec![Value::U64(1), Value::U64(10), Value::U64(100)]);

        let names = evaluate_spec("foo, bar, baz", &kernel).unwrap();
        assert_eq!(
            names,
            vec![
                Value::Str("foo".into()),
                Value::Str("bar".into()),
                Value::Str("baz".into()),
            ]
        );
    }

    #[test]
    fn literal_cursor_exposes_extent_auxiliaries() {
        // Real cursor declaration with literal extent — verifies
        // the compiler-side change that emits
        // __cursor_extent_<name>_{start,end} as final bindings
        // even in the literal-args case.
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("cursor row = range(0, 50)\n")
                .unwrap();
        let start = kernel.lookup("__cursor_extent_row_start");
        let end = kernel.lookup("__cursor_extent_row_end");
        assert_eq!(
            start,
            Some(Value::U64(0)),
            "expected start=0, got {start:?}"
        );
        assert_eq!(end, Some(Value::U64(50)), "expected end=50, got {end:?}");
    }

    #[test]
    fn all_cursor_with_real_cursor_decl_works() {
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("cursor row = range(0, 5)\n").unwrap();
        let values = evaluate_spec("all(row)", &kernel).unwrap();
        assert_eq!(
            values,
            vec![
                Value::U64(0),
                Value::U64(1),
                Value::U64(2),
                Value::U64(3),
                Value::U64(4),
            ]
        );
    }

    #[test]
    fn all_cursor_ignores_whitespace() {
        let kernel = crate::dsl::compile::compile_polydat_interpreter(
            "const __cursor_extent_row_start := 0\n\
             const __cursor_extent_row_end := 3\n",
        )
        .unwrap();
        let values = evaluate_spec("  all( row )  ", &kernel).unwrap();
        assert_eq!(values.len(), 3);
    }

    #[test]
    fn evaluate_spec_resolves_against_kernel() {
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const k_values := \"1, 10, 100\"\n")
                .unwrap();
        let v = evaluate_spec("{k_values}", &kernel).unwrap();
        assert_eq!(v, vec![Value::U64(1), Value::U64(10), Value::U64(100)]);
    }

    #[test]
    fn evaluate_spec_bare_ident_resolves_like_braced() {
        // A bare identifier source is a direct wire/param reference
        // (comprehension_forms.md §3.1.4) and resolves identically to
        // the braced `{name}` interpolation form.
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const k_values := \"1, 10, 100\"\n")
                .unwrap();
        let bare = evaluate_spec("k_values", &kernel).unwrap();
        let braced = evaluate_spec("{k_values}", &kernel).unwrap();
        assert_eq!(bare, braced);
        assert_eq!(bare, vec![Value::U64(1), Value::U64(10), Value::U64(100)]);
    }

    #[test]
    fn evaluate_spec_unresolved_bare_is_error_with_quoting_hint() {
        // A bare identifier source that doesn't resolve is a hard
        // error (comprehension_forms.md §3.1.4) (not silently bound as its own
        // name-string), and the message points at the fix.
        let kernel = crate::dsl::compile::compile_polydat_interpreter("\n").unwrap();
        let err = evaluate_spec("nonexistent", &kernel)
            .unwrap_err()
            .to_string();
        assert!(err.contains("did not resolve"), "got: {err}");
        assert!(err.contains("quote it"), "should hint quoting: {err}");
    }

    /// A source reads None for the names it reads after composition: the
    /// composed target of `{k_{k}_limits}`, and a name bound to None, are
    /// reads of None, and a failure over a name the scope has is not.
    #[test]
    fn a_read_of_none_is_decided_on_the_composed_name() {
        let kernel = crate::dsl::compile::compile_polydat_interpreter(
            "const k_1_limits := \"1, 2\"\nconst n := 3\n",
        )
        .unwrap();
        let at = |k: u64| vec![("k".to_string(), Value::U64(k))];
        let reads = |spec: &str, prefix: &[(String, Value)]| {
            let scope = crate::kernel::interp::Layered {
                prefix,
                inner: &kernel,
            };
            match evaluate_spec_internal(spec, &scope) {
                Ok(values) => Ok(values.len()),
                Err(SpecError::ReadsNone { reads, .. }) => Err(Some(reads)),
                Err(SpecError::Failed(_)) => Err(None),
            }
        };
        assert_eq!(reads("{k_{k}_limits}", &at(1)), Ok(2));
        assert_eq!(
            reads("{k_{k}_limits}", &at(7)),
            Err(Some(vec![NoneRead::Unbound("k_7_limits".into())]))
        );
        let none = [("z".to_string(), Value::None)];
        for spec in ["{z}", "z", "[z]", "pow2({z})", "concat(z, 1..3)"] {
            assert_eq!(
                reads(spec, &none),
                Err(Some(vec![NoneRead::BoundNone("z".into())])),
                "{spec}"
            );
        }
        assert_eq!(
            reads("u64_add(zz, 1)", &[]),
            Err(Some(vec![NoneRead::Unbound("zz".into())]))
        );
        assert_eq!(reads("pow2(99999)", &[]), Err(None));
        // The public evaluator reads a name bound to None as None and
        // still refuses one nothing binds.
        let scope = crate::kernel::interp::Layered {
            prefix: &none,
            inner: &kernel,
        };
        assert_eq!(evaluate_spec("{z}", &scope).unwrap(), Vec::<Value>::new());
        assert!(matches!(
            evaluate_spec("{zz}", &scope),
            Err(crate::dsl::compile::EmbeddingError::UnresolvedPlaceholder { .. })
        ));
    }

    #[test]
    fn bracket_list_spread_and_no_peel() {
        // `[xs…]` destructures (peels one level); `[xs]` binds the
        // whole value once.
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const xs := \"1, 2, 3\"\n").unwrap();
        // spread → peel the string's tokens
        let spread = evaluate_spec("[xs…]", &kernel).unwrap();
        assert_eq!(spread, vec![Value::U64(1), Value::U64(2), Value::U64(3)]);
        // no-peel → the whole value once (the string, un-striped)
        let whole = evaluate_spec("[xs]", &kernel).unwrap();
        assert_eq!(whole, vec![Value::Str("1, 2, 3".into())]);
    }

    #[test]
    fn bracket_list_mixes_refs_literals_and_spread() {
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const mid := \"7, 8\"\n").unwrap();
        let v = evaluate_spec("[1, mid…, \"x\"]", &kernel).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(7),
                Value::U64(8),
                Value::Str("x".into()),
            ]
        );
    }

    // ── Partition-list unpacking (cursor_partitions.md §7.1) ──

    #[test]
    fn evaluate_spec_unpacks_partition_list_into_partition_values() {
        // `partitions("linear:3")` evaluates to a PartitionList
        // Ext value. evaluate_spec must unpack the list into a
        // Vec of individual Partition values so the for-clause
        // iterates partition-by-partition (one iteration per
        // partition).
        let kernel = empty_kernel();
        let v = evaluate_spec("partitions(\"linear:3\")", &kernel).unwrap();
        assert_eq!(v.len(), 3, "expected 3 partitions, got {}", v.len());
        for value in &v {
            assert!(
                value.as_partition().is_some(),
                "every iter value should be a Partition, got {value:?}"
            );
        }
    }

    #[test]
    fn evaluate_spec_unpacks_partition_list_with_explicit_extent() {
        let kernel = empty_kernel();
        let v = evaluate_spec("partitions(\"fib:5\", 1000)", &kernel).unwrap();
        assert_eq!(v.len(), 5);
        // Partition indices increment from 0.
        for (i, value) in v.iter().enumerate() {
            let p = value.as_partition().unwrap();
            assert_eq!(p.idx, i as u64);
            assert_eq!(p.base_extent, 1000);
        }
    }

    #[test]
    fn pre_evaluate_clause_returns_partition_values_for_partitions_call() {
        // Same as the evaluate_spec test above but via the
        // synthesis-side pre_evaluate_clause entry point.
        let kernel = empty_kernel();
        let v = pre_evaluate_clause(
            "partitions(\"linear:4\")",
            &kernel,
            &HashMap::new(),
            &HashMap::new(),
        )
        .unwrap();
        assert_eq!(v.len(), 4);
        for value in &v {
            assert!(
                value.as_partition().is_some(),
                "pre_evaluate_clause must unpack PartitionList, got {value:?}"
            );
        }
    }

    #[test]
    fn value_to_polydat_type_name_returns_ext_for_partition_value() {
        // The for_each scope synthesizer uses this to emit
        // `extern <var>: <keyword>` for each iter-var. Ext-typed
        // values (Partition, PartitionSpec, PartitionList) must
        // declare as `ext` so the resulting input port is
        // PortType::Ext and downstream `over <iter-var>` clauses
        // see the right shape.
        let p = crate::iteration::cursor_partition::Partition {
            idx: 0,
            count: 1,
            start_ord: 0,
            end_ord: 10,
            start_pct: 0.0,
            end_pct: 100.0,
            base_extent: 10,
        };
        let v = Value::from_partition(p);
        assert_eq!(value_to_polydat_type_name(&v), "ext");
    }

    // ── Range operator ──

    fn empty_kernel() -> PolydatKernel {
        crate::dsl::compile::compile_polydat_interpreter("\n").unwrap()
    }

    #[test]
    fn range_half_open_integer() {
        let v = evaluate_spec("1..5", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![Value::U64(1), Value::U64(2), Value::U64(3), Value::U64(4),]
        );
    }

    #[test]
    fn range_inclusive_integer() {
        let v = evaluate_spec("1..=5", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(2),
                Value::U64(3),
                Value::U64(4),
                Value::U64(5),
            ]
        );
    }

    #[test]
    fn range_with_step() {
        let v = evaluate_spec("0..100..10", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(0),
                Value::U64(10),
                Value::U64(20),
                Value::U64(30),
                Value::U64(40),
                Value::U64(50),
                Value::U64(60),
                Value::U64(70),
                Value::U64(80),
                Value::U64(90),
            ]
        );
    }

    #[test]
    fn range_inclusive_with_step() {
        let v = evaluate_spec("0..=100..25", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(0),
                Value::U64(25),
                Value::U64(50),
                Value::U64(75),
                Value::U64(100),
            ]
        );
    }

    #[test]
    fn range_float_step() {
        let v = evaluate_spec("0.0..=1.0..0.25", &empty_kernel()).unwrap();
        assert_eq!(v.len(), 5, "got {v:?}");
        if let [
            Value::F64(a),
            Value::F64(b),
            Value::F64(c),
            Value::F64(d),
            Value::F64(e),
        ] = v.as_slice()
        {
            assert!((a - 0.0).abs() < 1e-12);
            assert!((b - 0.25).abs() < 1e-12);
            assert!((c - 0.5).abs() < 1e-12);
            assert!((d - 0.75).abs() < 1e-12);
            assert!((e - 1.0).abs() < 1e-12);
        } else {
            panic!("expected 5 floats, got {v:?}");
        }
    }

    #[test]
    fn range_empty_when_start_equals_end_half_open() {
        let v = evaluate_spec("5..5", &empty_kernel()).unwrap();
        assert!(v.is_empty(), "got {v:?}");
    }

    #[test]
    fn range_inclusive_with_equal_bounds_emits_one() {
        let v = evaluate_spec("5..=5", &empty_kernel()).unwrap();
        assert_eq!(v, vec![Value::U64(5)]);
    }

    #[test]
    fn range_with_si_suffix_bounds() {
        // SI suffixes (polydat_grammar.md §2.3) compose with range
        // bounds.
        let v = evaluate_spec("1K..1K..200", &empty_kernel()).unwrap();
        assert!(v.is_empty(), "1K..1K with positive step → empty");

        let v = evaluate_spec("0..1K..200", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(0),
                Value::U64(200),
                Value::U64(400),
                Value::U64(600),
                Value::U64(800),
            ]
        );
    }

    #[test]
    fn range_zero_step_errors() {
        let err = evaluate_spec("1..10..0", &empty_kernel())
            .unwrap_err()
            .to_string();
        assert!(err.contains("step is zero"), "{err}");
    }

    #[test]
    fn range_too_many_dotdot_errors() {
        let err = evaluate_spec("1..2..3..4", &empty_kernel())
            .unwrap_err()
            .to_string();
        assert!(err.contains("more than two `..`"), "{err}");
    }

    #[test]
    fn range_inside_parens_doesnt_split() {
        // `range(1, 10)` — the dots inside the function
        // call shouldn't trigger range-splitting at top
        // depth (there are no `..` here anyway, but verify
        // paren-balanced text passes through cleanly).
        // Use a literal with internal parens to exercise
        // the depth tracking.
        let v = evaluate_spec("(1)..(5)", &empty_kernel()).unwrap();
        assert_eq!(v.len(), 4); // 1, 2, 3, 4
    }

    #[test]
    fn range_step_with_inclusive_separator_errors() {
        let err = evaluate_spec("1..10..=2", &empty_kernel())
            .unwrap_err()
            .to_string();
        assert!(err.contains("step delimiter cannot be `..=`"), "{err}");
    }

    #[test]
    fn range_with_kernel_referenced_bounds() {
        let kernel =
            crate::dsl::compile::compile_polydat_interpreter("const lo := 5\nconst hi := 12\n")
                .unwrap();
        let v = evaluate_spec("{lo}..{hi}", &kernel).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(5),
                Value::U64(6),
                Value::U64(7),
                Value::U64(8),
                Value::U64(9),
                Value::U64(10),
                Value::U64(11),
            ]
        );
    }

    // ── Named generators ──

    #[test]
    fn fib_n_first_eight() {
        let v = evaluate_spec("fib(8)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(1),
                Value::U64(2),
                Value::U64(3),
                Value::U64(5),
                Value::U64(8),
                Value::U64(13),
                Value::U64(21),
            ]
        );
    }

    #[test]
    fn fib_until_50() {
        let v = evaluate_spec("fib_until(50)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(1),
                Value::U64(2),
                Value::U64(3),
                Value::U64(5),
                Value::U64(8),
                Value::U64(13),
                Value::U64(21),
                Value::U64(34),
            ]
        );
    }

    #[test]
    fn pow2_n_six() {
        let v = evaluate_spec("pow2(6)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(2),
                Value::U64(4),
                Value::U64(8),
                Value::U64(16),
                Value::U64(32),
            ]
        );
    }

    #[test]
    fn pow2_until_100() {
        let v = evaluate_spec("pow2_until(100)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(2),
                Value::U64(4),
                Value::U64(8),
                Value::U64(16),
                Value::U64(32),
                Value::U64(64),
            ]
        );
    }

    #[test]
    fn binomial_n_5() {
        // C(5,0..5) = 1, 5, 10, 10, 5, 1
        let v = evaluate_spec("binomial(5)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(5),
                Value::U64(10),
                Value::U64(10),
                Value::U64(5),
                Value::U64(1),
            ]
        );
    }

    #[test]
    fn geometric_2_doubles_4_terms() {
        let v = evaluate_spec("geometric(1, 2, 4)", &empty_kernel()).unwrap();
        // Floats because factor is float-cast at eval.
        if let [Value::F64(a), Value::F64(b), Value::F64(c), Value::F64(d)] = v.as_slice() {
            assert!((a - 1.0).abs() < 1e-12);
            assert!((b - 2.0).abs() < 1e-12);
            assert!((c - 4.0).abs() < 1e-12);
            assert!((d - 8.0).abs() < 1e-12);
        } else {
            panic!("expected 4 f64 values, got {v:?}");
        }
    }

    #[test]
    fn linear_starts_half_open_5_points() {
        let v = evaluate_spec("linear_starts(0, 100, 5)", &empty_kernel()).unwrap();
        // (100-0)/5 = 20 step. 0, 20, 40, 60, 80.
        if let [
            Value::F64(a),
            Value::F64(b),
            Value::F64(c),
            Value::F64(d),
            Value::F64(e),
        ] = v.as_slice()
        {
            assert!((a - 0.0).abs() < 1e-12);
            assert!((b - 20.0).abs() < 1e-12);
            assert!((c - 40.0).abs() < 1e-12);
            assert!((d - 60.0).abs() < 1e-12);
            assert!((e - 80.0).abs() < 1e-12);
        } else {
            panic!("got {v:?}");
        }
    }

    #[test]
    fn linear_steps_inclusive_5_points() {
        let v = evaluate_spec("linear_steps(0, 100, 5)", &empty_kernel()).unwrap();
        // 0, 25, 50, 75, 100
        if let [
            Value::F64(a),
            Value::F64(b),
            Value::F64(c),
            Value::F64(d),
            Value::F64(e),
        ] = v.as_slice()
        {
            assert!((a - 0.0).abs() < 1e-12);
            assert!((b - 25.0).abs() < 1e-12);
            assert!((c - 50.0).abs() < 1e-12);
            assert!((d - 75.0).abs() < 1e-12);
            assert!((e - 100.0).abs() < 1e-12);
        } else {
            panic!("got {v:?}");
        }
    }

    #[test]
    fn log_steps_3_decades() {
        let v = evaluate_spec("log_steps(1, 1000, 4)", &empty_kernel()).unwrap();
        // 1, 10, 100, 1000
        if let [Value::F64(a), Value::F64(b), Value::F64(c), Value::F64(d)] = v.as_slice() {
            assert!((a - 1.0).abs() < 1e-9);
            assert!((b - 10.0).abs() < 1e-9);
            assert!((c - 100.0).abs() < 1e-9);
            assert_eq!(*d, 1000.0);
        } else {
            panic!("got {v:?}");
        }
    }

    #[test]
    fn log_steps_rejects_non_positive_bounds() {
        let err = evaluate_spec("log_steps(0, 100, 5)", &empty_kernel())
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("log_steps.start: expected a positive number, got 0"),
            "{err}"
        );
        let err = evaluate_spec("log_steps(1, -2, 5)", &empty_kernel())
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("log_steps.end: expected a positive number, got -2"),
            "{err}"
        );
    }

    /// The last point of `linear_steps` and both ends of `log_steps` are
    /// the given bounds exactly, not sums or `exp`s that round near them.
    #[test]
    fn inclusive_steps_end_exactly_at_their_bounds() {
        let floats = |spec: &str| -> Vec<f64> {
            evaluate_spec(spec, &empty_kernel())
                .unwrap()
                .iter()
                .map(|v| match v {
                    Value::F64(f) => *f,
                    other => panic!("{spec}: {other:?}"),
                })
                .collect()
        };
        for (spec, start, end) in [
            ("log_steps(1, 1000, 4)", 1.0, 1000.0),
            ("log_steps(3, 7, 9)", 3.0, 7.0),
            ("log_steps(0.1, 0.7, 13)", 0.1, 0.7),
            ("log_steps(1000, 1, 4)", 1000.0, 1.0),
            ("linear_steps(0, 1, 4)", 0.0, 1.0),
            ("linear_steps(0.1, 0.7, 13)", 0.1, 0.7),
            ("linear_steps(-3, 1e9, 7)", -3.0, 1e9),
        ] {
            let v = floats(spec);
            assert_eq!(v.first(), Some(&start), "{spec}: {v:?}");
            assert_eq!(v.last(), Some(&end), "{spec}: {v:?}");
        }
        assert_eq!(floats("linear_steps(2, 5, 1)"), vec![2.0]);
        assert_eq!(floats("log_steps(2, 5, 1)"), vec![2.0]);
    }

    /// The largest valid argument of each generator whose terms outgrow
    /// `u64`, found by computing the terms in `u128`: the first term
    /// past `u64::MAX` is the one the refusal names.
    #[test]
    fn overflow_limits_are_where_the_terms_leave_u64() {
        let max = u128::from(u64::MAX);
        // Fibonacci: the first term past u64::MAX, 1-based.
        let (mut a, mut b, mut term) = (1u128, 1u128, 1u64);
        while a <= max {
            (a, b, term) = (b, a + b, term + 1);
        }
        assert_eq!(term, 94);
        assert_eq!(NamedGenerator::Fib.largest_valid_argument(), Some(term - 1));
        // Powers of two: 2^64 is term 65.
        assert_eq!(1u128 << 64, max + 1);
        assert_eq!(NamedGenerator::Pow2.largest_valid_argument(), Some(64));
        // Pascal's triangle: the first row with a coefficient past
        // u64::MAX, and that coefficient.
        let row_overflow = |n: u64| -> Option<u64> {
            let mut c = 1u128;
            (1..=n).find(|&k| {
                c = c * u128::from(n - k + 1) / u128::from(k);
                c > max
            })
        };
        let first_row = (0..).find(|&n| row_overflow(n).is_some()).unwrap();
        assert_eq!(first_row, 68);
        assert_eq!(row_overflow(68), Some(31));
        assert_eq!(
            NamedGenerator::Binomial.largest_valid_argument(),
            Some(first_row - 1)
        );
        assert_eq!(NamedGenerator::Geometric.largest_valid_argument(), None);
    }

    /// Each limit is the last argument that yields every term, and the
    /// next is refused, naming the call, the first term past
    /// `u64::MAX`, and the limit.
    #[test]
    fn a_call_past_its_limit_is_refused_by_its_first_overflowing_term() {
        let k = empty_kernel();
        let fib = evaluate_spec("fib(93)", &k).unwrap();
        assert_eq!(fib.len(), 93);
        assert_eq!(fib[92], Value::U64(12_200_160_415_121_876_738));
        let pow2 = evaluate_spec("pow2(64)", &k).unwrap();
        assert_eq!(pow2.last(), Some(&Value::U64(1 << 63)));
        let row = evaluate_spec("binomial(67)", &k).unwrap();
        assert_eq!(row.len(), 68);
        assert_eq!(row[33], Value::U64(14_226_520_737_620_288_370));
        for (spec, message) in [
            (
                "fib(94)",
                "fib(94): term 94 is past u64::MAX; fib.n is at most 93",
            ),
            (
                "fib(18446744073709551615)",
                "fib(18446744073709551615): term 94 is past u64::MAX",
            ),
            (
                "pow2(65)",
                "pow2(65): term 65, 2^64, is past u64::MAX; pow2.n is at most 64",
            ),
            (
                "binomial(68)",
                "binomial(68): term C(68, 31) is past u64::MAX; binomial.n is at most 67",
            ),
            (
                "binomial(70)",
                "binomial(70): term C(70, 28) is past u64::MAX",
            ),
            (
                "binomial(1000000000000)",
                "binomial(1000000000000): term C(1000000000000, 2) is past u64::MAX",
            ),
        ] {
            let err = evaluate_spec(spec, &k).unwrap_err().to_string();
            assert!(err.contains(message), "{spec}: {err}");
        }
    }

    /// `geometric` takes a positive, finite factor, and
    /// `geometric_until` a finite factor above 1; the refusal names the
    /// argument.
    #[test]
    fn a_geometric_factor_out_of_range_is_refused_by_name() {
        let k = empty_kernel();
        for (spec, message) in [
            (
                "geometric(1, 0, 4)",
                "geometric.factor: expected a positive, finite number, got 0",
            ),
            (
                "geometric(1, -2, 4)",
                "geometric.factor: expected a positive, finite number, got -2",
            ),
            (
                "geometric(1, inf, 4)",
                "geometric.factor: expected a positive, finite number, got inf",
            ),
            (
                "geometric_until(1, 1, 100)",
                "geometric_until.factor: expected a finite number greater than 1, got 1",
            ),
            (
                "geometric_until(1, 0.5, 100)",
                "geometric_until.factor: expected a finite number greater than 1, got 0.5",
            ),
        ] {
            let err = evaluate_spec(spec, &k).unwrap_err().to_string();
            assert!(err.contains(message), "{spec}: {err}");
        }
        assert_eq!(
            evaluate_spec("geometric(8, 0.5, 3)", &k).unwrap(),
            vec![Value::F64(8.0), Value::F64(4.0), Value::F64(2.0)]
        );
        assert!(
            evaluate_spec("geometric_until(0, 2, 100)", &k)
                .unwrap()
                .is_empty()
        );
    }

    /// The refusal of a named generator is found through the calls that
    /// enclose it; a call that expands, and a call of anything else, is
    /// not a refusal.
    #[test]
    fn refused_generator_call_looks_through_enclosing_calls() {
        let k = empty_kernel();
        let refused = refused_generator_call("concat(1..3, take(fib(94), 2))", &k).unwrap();
        assert!(refused.starts_with("fib(94): term 94"), "{refused}");
        assert_eq!(refused_generator_call("concat(fib(8), pow2(64))", &k), None);
        assert_eq!(refused_generator_call("hash(3)", &k), None);
        assert_eq!(refused_generator_call("1, 2, 3", &k), None);
        let refused = refused_generator_call("fib(-1)", &k).unwrap();
        assert!(
            refused.contains("fib.n: expected non-negative integer, got '-1'"),
            "{refused}"
        );
    }

    // ── Set operators ──

    #[test]
    fn concat_two_ranges() {
        let v = evaluate_spec("concat(1..4, 10..13)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(2),
                Value::U64(3),
                Value::U64(10),
                Value::U64(11),
                Value::U64(12),
            ]
        );
    }

    #[test]
    fn unique_dedupes_first_occurrence() {
        let v = evaluate_spec("unique(1..4, 3..6)", &empty_kernel()).unwrap();
        // 1,2,3 (from first) + 4,5 (from second; 3 already present)
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(2),
                Value::U64(3),
                Value::U64(4),
                Value::U64(5),
            ]
        );
    }

    #[test]
    fn intersect_keeps_only_common_values() {
        let v = evaluate_spec("intersect(1..10, 5..15)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(5),
                Value::U64(6),
                Value::U64(7),
                Value::U64(8),
                Value::U64(9),
            ]
        );
    }

    #[test]
    fn subtract_drops_values_in_b() {
        let v = evaluate_spec("subtract(1..6, 3..5)", &empty_kernel()).unwrap();
        // 1..6 = [1,2,3,4,5], minus [3,4] = [1, 2, 5]
        assert_eq!(v, vec![Value::U64(1), Value::U64(2), Value::U64(5)]);
    }

    #[test]
    fn interleave_round_robin_two_lists() {
        let v = evaluate_spec("interleave(1..4, 10..13)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(10),
                Value::U64(2),
                Value::U64(11),
                Value::U64(3),
                Value::U64(12),
            ]
        );
    }

    #[test]
    fn cycle_repeats_n_times() {
        let v = evaluate_spec("cycle(1..3, 3)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![
                Value::U64(1),
                Value::U64(2),
                Value::U64(1),
                Value::U64(2),
                Value::U64(1),
                Value::U64(2),
            ]
        );
    }

    #[test]
    fn reverse_inverts_list() {
        let v = evaluate_spec("reverse(1..5)", &empty_kernel()).unwrap();
        assert_eq!(
            v,
            vec![Value::U64(4), Value::U64(3), Value::U64(2), Value::U64(1),]
        );
    }

    #[test]
    fn take_n_takes_prefix() {
        let v = evaluate_spec("take(1..10, 3)", &empty_kernel()).unwrap();
        assert_eq!(v, vec![Value::U64(1), Value::U64(2), Value::U64(3)]);
    }

    #[test]
    fn skip_n_drops_prefix() {
        let v = evaluate_spec("skip(1..6, 2)", &empty_kernel()).unwrap();
        assert_eq!(v, vec![Value::U64(3), Value::U64(4), Value::U64(5)]);
    }

    #[test]
    fn unique_composes_with_pow2_and_range() {
        let v = evaluate_spec("unique(pow2(8), 1..1000..100)", &empty_kernel()).unwrap();
        // pow2(8) = 1, 2, 4, 8, 16, 32, 64, 128
        // 1..1000..100 = 1, 101, 201, 301, 401, 501, 601, 701, 801, 901
        // dedupe: 1, 2, 4, 8, 16, 32, 64, 128, 101, 201, 301, 401, 501, 601, 701, 801, 901
        assert_eq!(v.len(), 17);
        assert_eq!(v[0], Value::U64(1));
        assert_eq!(v[7], Value::U64(128));
        assert_eq!(v[8], Value::U64(101));
    }

    // ── Sequencer expansions ──

    #[test]
    fn bucket_round_robin_3_1_2() {
        // Two-arg form: items list + ratios list.
        let v = evaluate_spec(
            "bucket(concat('ann', 'scan', 'fetch'), concat(3, 1, 2))",
            &empty_kernel(),
        )
        .unwrap();
        // Wait — concat doesn't make sense with these args (mixed types).
        // Use the literal form via the Polydat list parser.
        let _ = v;
    }

    #[test]
    fn bucket_ratio_prefix_shorthand_round_robin() {
        let v = evaluate_spec("bucket(\"3:ann, 1:scan, 2:fetch\")", &empty_kernel()).unwrap();
        // Bucket sequencer round-robins; each "tick" pulls
        // one from each remaining bucket. Total = 6.
        assert_eq!(v.len(), 6);
        let strs: Vec<&str> = v
            .iter()
            .filter_map(|v| match v {
                Value::Str(s) => Some(&**s),
                _ => None,
            })
            .collect();
        // First tick: ann, scan, fetch (one from each).
        // Then ann (3 left), fetch (2 left). Next: ann, fetch.
        // Then ann. Total: ann*3, scan*1, fetch*2.
        let counts = strs.iter().fold(
            std::collections::HashMap::<&str, usize>::new(),
            |mut m, s| {
                *m.entry(s).or_insert(0) += 1;
                m
            },
        );
        assert_eq!(counts.get("ann"), Some(&3));
        assert_eq!(counts.get("scan"), Some(&1));
        assert_eq!(counts.get("fetch"), Some(&2));
    }

    #[test]
    fn concat_seq_emits_contiguous_runs() {
        let v = evaluate_spec(
            "concat_seq(\"2:warmup, 3:bench, 1:cooldown\")",
            &empty_kernel(),
        )
        .unwrap();
        let strs: Vec<String> = v
            .iter()
            .filter_map(|v| match v {
                Value::Str(s) => Some(s.to_string()),
                _ => None,
            })
            .collect();
        assert_eq!(
            strs,
            vec!["warmup", "warmup", "bench", "bench", "bench", "cooldown",]
        );
    }

    #[test]
    fn interval_seq_evenly_spreads_higher_ratio() {
        let v = evaluate_spec("interval_seq(\"3:read, 1:write\")", &empty_kernel()).unwrap();
        // Total length 4. write should appear once,
        // somewhere in the middle (not bunched at edges).
        let strs: Vec<String> = v
            .iter()
            .filter_map(|v| match v {
                Value::Str(s) => Some(s.to_string()),
                _ => None,
            })
            .collect();
        assert_eq!(strs.len(), 4);
        let writes: Vec<usize> = strs
            .iter()
            .enumerate()
            .filter(|(_, s)| *s == "write")
            .map(|(i, _)| i)
            .collect();
        assert_eq!(writes.len(), 1, "expected exactly one write: {strs:?}");
    }

    #[test]
    fn parse_func_call_recognises_simple_call() {
        let (n, a) = parse_func_call("fib(8)").unwrap();
        assert_eq!(n, "fib");
        assert_eq!(a, "8");
    }

    #[test]
    fn parse_func_call_rejects_non_calls() {
        assert!(parse_func_call("1..10").is_none());
        assert!(parse_func_call("foo + bar").is_none());
        assert!(parse_func_call("f(a) + g(b)").is_none()); // mid-text close
    }

    #[test]
    fn split_args_top_level_skips_inner_commas() {
        let args = split_args_top_level("a, f(b, c), \"x, y\", 3");
        assert_eq!(args, vec!["a", "f(b, c)", "\"x, y\"", "3"]);
    }
}