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
//! ```
//!
//! ## Where this used to live
//!
//! Pre-Phase-C this code lived in the host. The lift was driven by
//! the principle that Polydat is the canonical owner of what a
//! comprehension *means*, including how its spec strings resolve
//! (see `crates/polydat/docs/design/comprehension_forms.md`).
//! The host now consumes this
//! API rather than implementing it.

use std::collections::HashMap;

use crate::ast::Value;
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.
pub fn evaluate_spec(
    spec_text: &str,
    kernel: &dyn Lookup,
) -> Result<Vec<Value>, crate::dsl::compile::EmbeddingError> {
    evaluate_spec_internal(spec_text, kernel).map_err(|msg| {
        if let Some(rest) = msg.strip_prefix("interpolation: unresolved placeholder '{")
            && let Some(end) = rest.find('}')
        {
            let name = rest[..end].to_string();
            return crate::dsl::compile::EmbeddingError::UnresolvedPlaceholder {
                name,
                source: spec_text.to_string(),
            };
        }
        crate::dsl::compile::EmbeddingError::Parse {
            source: spec_text.to_string(),
            message: msg,
            position: None,
        }
    })
}

fn evaluate_spec_internal(spec_text: &str, kernel: &dyn Lookup) -> Result<Vec<Value>, String> {
    if let Some(values) = try_eval_all_cursor(spec_text, kernel)? {
        return Ok(values);
    }
    // SRD-18f Stage 2 (non-breaking core): a *bare identifier*
    // source is a direct wire/param/const reference. Resolve it
    // against the kernel chain — the same `kernel.lookup` the
    // `{name}` interpolation path uses — and peel/wrap its value
    // via `iteration_interior`. This makes `mnc in mnc_values`
    // work identically to `mnc in {mnc_values}`.
    //
    // If the bare name does NOT resolve, fall through to the
    // legacy path, which treats it as a label string
    // (`y in z` → ["z"]). The strict "unresolved bare ident is
    // an error" enforcement (SRD-18f §6) is deferred to the
    // breaking part of Stage 2 along with its test/workload
    // migration.
    if is_single_bare_ident(spec_text) {
        // SRD-18f §6: a bare identifier source is a reference. It
        // resolves against the kernel, or it's a hard error — it
        // is NOT silently bound as its own name-string.
        return match kernel.lookup(spec_text.trim()) {
            Some(v) => Ok(
                match crate::iteration::comprehension::source_values::iteration_interior(&v) {
                    Some(interior) => interior,
                    None => vec![v],
                },
            ),
            None => Err(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}\"`.",
                src = spec_text.trim(),
            )),
        };
    }
    let interpolated = crate::kernel::interp::interpolate_with_lookup(spec_text, |name| {
        kernel.lookup(name).map(|v| v.to_display_string())
    })?;
    // SRD-18f Stage 2: list comprehension sugar `[e1, e2…, e3]`.
    // 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);
    }
    // SRD-18c Layer 2 / SRD-18e Push 3: range operator
    // (`a..b`, `a..=b`, `a..b..s`, `a..=b..s`). 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);
    }
    // SRD-18c Layer 3 / SRD-18e Push 7: named generators.
    if let Some(values) = try_eval_generator(&interpolated)? {
        return Ok(values);
    }
    // SRD-18c Layer 5 / SRD-18e Push 9: set operators on lists.
    if let Some(values) = try_eval_setop(&interpolated, kernel)? {
        return Ok(values);
    }
    // SRD-18c §"Sequencer-style expansions" / Push 8: LUT
    // facility (bucket / concat_seq / interval_seq).
    if let Some(values) = try_eval_sequencer(&interpolated, kernel)? {
        return Ok(values);
    }
    // SRD 71: kernel-aware partition sources — `subdivide(outer,
    // n)` where `outer` is a partition iter-var bound by an
    // enclosing `for:` clause.
    if let Some(values) = try_eval_partition_call(&interpolated, kernel)? {
        return Ok(values);
    }
    // SRD 71: `<param>.partitions` comprehension-position desugaring —
    // resolve the param's spec string and expand it into its PartitionList.
    if let Some(values) = try_eval_param_partitions(&interpolated, kernel)? {
        return Ok(values);
    }
    match crate::dsl::compile::eval_const_expr_for(&interpolated, kernel.ledger()) {
        // SRD-18f relaxed source resolution: a resolved value is
        // peeled one level if it has an iteration interior
        // (native vector, JSON array, PartitionList per SRD-71,
        // or a string → its comprehension tokens), else wrapped
        // as a singleton. `iteration_interior` is the single
        // canonical place that decision is made — this replaces
        // the former per-type arms (Str→comma-split,
        // PartitionList→unpack, other→wrap).
        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). The latter
        // produces malformed downstream output six steps removed
        // from the actual fault — a Push-2 kind of bad UX.
        Err(eval_err) => {
            // NOTE: SRD-18f §6 specifies that a single bare
            // identifier that fails to evaluate is an *unresolved
            // reference* and should be a hard error (with a
            // quoting hint), not silently bound as its own
            // name-string. That enforcement is **Stage 2** (the
            // bare-word→reference change) because it flips every
            // existing bare-label source (`y in z` meaning the
            // string "z") and requires migrating those to quoted
            // form. Until Stage 2, the legacy literal-list
            // fallback below preserves bare-label-as-string.
            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."
                ))
            }
        }
    }
}

/// 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 point of this gate is to keep "list" specs (`for: "k in 1,
/// 10, 100"`) working through the literal-list fallback while
/// still surfacing real evaluation failures for expression specs
/// like `matching_profiles('x', 'y')`. A wrong call on a
/// borderline case here is cheap — it just produces a clearer
/// error from the eval layer instead of swallowed garbage.
/// SRD-18f Stage 2 — list comprehension sugar. 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>>, String> {
    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());
        }
        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(),
                    ));
                }
            }
        } 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.
/// SRD-18f §6: an unresolved bare reference is a hard error with
/// a quoting hint, not a silent literal-name binding.
fn eval_element_value(expr: &str, kernel: &dyn Lookup) -> Result<Value, String> {
    let e = expr.trim();
    if is_single_bare_ident(e) {
        return kernel.lookup(e).ok_or_else(|| {
            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| 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 (SRD-18f Stage 2); 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 == '_')
}

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);
    }
    // SRD-18f Stage 2: a bare identifier source is a direct
    // reference. Resolve it at synthesis the same way the runtime
    // `evaluate_spec` does — 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. Without this,
    // `k in k_values` left `k`'s probe as the literal name and the
    // dependent `limit` defaulted to String (the query_sweep bug
    // the scenario-synthesis coverage gap was hiding).
    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())
    })?;

    // Push 3: range operator on the pre-evaluation path too.
    if let Some(values) = try_eval_range(&interpolated, parent_kernel.ledger())? {
        return Ok(values);
    }
    // Push 7 / 9 / 8 — same generator / set-op / sequencer
    // shortcuts the runtime path uses.
    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);
    }
    // SRD 71: kernel-aware partition sources, same as the
    // runtime path. At pre-evaluation the outer iter-var may
    // not be installed yet; `try_eval_partition_call` returns a
    // single placeholder partition in that case so iter-var
    // type detection still lands on `ext`.
    if let Some(values) = try_eval_partition_call(&interpolated, parent_kernel)? {
        return Ok(values);
    }
    // SRD 71: `<param>.partitions` comprehension-position desugaring (same
    // rule as the runtime path; the param may already be installed here).
    if let Some(values) = try_eval_param_partitions(&interpolated, parent_kernel)? {
        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(),
            // SRD 71: `<param>.partitions` and `partitions(spec, ...)`
            // 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. SRD-18b's "native types as the general rule":
/// `"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>`.
///
/// 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.
/// - `Err(...)` if the form matched but the cursor's extent
///   couldn't be resolved (cursor not in scope, extent wires
///   missing, etc.) — surfaced as a clause-level diagnostic.
fn try_eval_all_cursor(spec_text: &str, kernel: &dyn Lookup) -> Result<Option<Vec<Value>>, String> {
    let trimmed = spec_text.trim();
    let Some(stripped) = trimmed.strip_prefix("all(") else {
        return Ok(None);
    };
    let Some(arg) = stripped.strip_suffix(')') else {
        return Ok(None);
    };
    let cursor_name = arg.trim();
    if cursor_name.is_empty() || !is_valid_ident(cursor_name) {
        return Ok(None);
    }

    let start_key = format!("__cursor_extent_{cursor_name}_start");
    let end_key = format!("__cursor_extent_{cursor_name}_end");
    let start = kernel
        .lookup(&start_key)
        .and_then(|v| match v {
            Value::U64(n) => Some(n),
            _ => None,
        })
        .ok_or_else(|| {
            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 '{start_key}'."
            )
        })?;
    let end = kernel
        .lookup(&end_key)
        .and_then(|v| match v {
            Value::U64(n) => Some(n),
            _ => None,
        })
        .ok_or_else(|| {
            format!(
                "all({cursor_name}): missing auxiliary output '{end_key}' on the parent kernel."
            )
        })?;

    if end < start {
        return Err(format!(
            "all({cursor_name}): cursor extent end={end} is less than start={start} — \
             cannot enumerate a negative-extent range."
        ));
    }
    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 == '_')
}

/// SRD-18c Layer 2 / SRD-18e Push 3: recognise the range
/// operator 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}'"))
}

// ============================================================
// SRD-18c Layer 3 / SRD-18e Push 7: named generators
// ============================================================

/// Recognise `fib(n)`, `pow2(n)`, `geometric(...)`, etc. and
/// expand to a `Vec<Value>`. Returns `Ok(None)` when the
/// text isn't a known generator call (caller falls through
/// to set-op / sequencer / 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 arg_list = split_args_top_level(args);
    match name {
        "fib" => {
            if arg_list.len() != 1 {
                return Err(format!(
                    "fib(n): expected 1 argument, got {}",
                    arg_list.len()
                ));
            }
            let n = parse_u64_arg(arg_list[0], "fib(n)")?;
            Ok(Some(generate_fib_n(n)?))
        }
        "fib_until" => {
            if arg_list.len() != 1 {
                return Err(format!(
                    "fib_until(max): expected 1 argument, got {}",
                    arg_list.len()
                ));
            }
            let max = parse_u64_arg(arg_list[0], "fib_until(max)")?;
            Ok(Some(generate_fib_until(max)))
        }
        "pow2" => {
            if arg_list.len() != 1 {
                return Err(format!(
                    "pow2(n): expected 1 argument, got {}",
                    arg_list.len()
                ));
            }
            let n = parse_u64_arg(arg_list[0], "pow2(n)")?;
            Ok(Some(generate_pow2_n(n)))
        }
        "pow2_until" => {
            if arg_list.len() != 1 {
                return Err(format!(
                    "pow2_until(max): expected 1 argument, got {}",
                    arg_list.len()
                ));
            }
            let max = parse_u64_arg(arg_list[0], "pow2_until(max)")?;
            Ok(Some(generate_pow2_until(max)))
        }
        "binomial" => {
            if arg_list.len() != 1 {
                return Err(format!(
                    "binomial(n): expected 1 argument, got {}",
                    arg_list.len()
                ));
            }
            let n = parse_u64_arg(arg_list[0], "binomial(n)")?;
            Ok(Some(generate_binomial(n)))
        }
        "geometric" => {
            if arg_list.len() != 3 {
                return Err(format!(
                    "geometric(start, factor, n): expected 3 args, got {}",
                    arg_list.len()
                ));
            }
            let start = parse_num_arg(arg_list[0], "geometric.start")?;
            let factor = parse_num_arg(arg_list[1], "geometric.factor")?;
            let n = parse_u64_arg(arg_list[2], "geometric.n")?;
            Ok(Some(generate_geometric(start, factor, n)?))
        }
        "geometric_until" => {
            if arg_list.len() != 3 {
                return Err(format!(
                    "geometric_until(start, factor, max): expected 3 args, got {}",
                    arg_list.len()
                ));
            }
            let start = parse_num_arg(arg_list[0], "geometric_until.start")?;
            let factor = parse_num_arg(arg_list[1], "geometric_until.factor")?;
            let max = parse_num_arg(arg_list[2], "geometric_until.max")?;
            Ok(Some(generate_geometric_until(start, factor, max)))
        }
        "linear_starts" => {
            if arg_list.len() != 3 {
                return Err(format!(
                    "linear_starts(start, end, n): expected 3 args, got {}",
                    arg_list.len()
                ));
            }
            let start = parse_num_arg(arg_list[0], "linear_starts.start")?;
            let end = parse_num_arg(arg_list[1], "linear_starts.end")?;
            let n = parse_u64_arg(arg_list[2], "linear_starts.n")?;
            Ok(Some(generate_linear_points(start, end, n, false)?))
        }
        "linear_steps" => {
            if arg_list.len() != 3 {
                return Err(format!(
                    "linear_steps(start, end, n): expected 3 args, got {}",
                    arg_list.len()
                ));
            }
            let start = parse_num_arg(arg_list[0], "linear_steps.start")?;
            let end = parse_num_arg(arg_list[1], "linear_steps.end")?;
            let n = parse_u64_arg(arg_list[2], "linear_steps.n")?;
            Ok(Some(generate_linear_points(start, end, n, true)?))
        }
        "log_steps" => {
            if arg_list.len() != 3 {
                return Err(format!(
                    "log_steps(start, end, n): expected 3 args, got {}",
                    arg_list.len()
                ));
            }
            let start = parse_num_arg(arg_list[0], "log_steps.start")?;
            let end = parse_num_arg(arg_list[1], "log_steps.end")?;
            let n = parse_u64_arg(arg_list[2], "log_steps.n")?;
            Ok(Some(generate_log_steps(start, end, n)?))
        }
        _ => Ok(None),
    }
}

/// First `n` Fibonacci numbers: 1, 1, 2, 3, 5, 8, ...
///
/// The count comes from the spec text, so the buffer is reserved
/// fallibly (`try_buffer_for`): `fib(99999999999999)` is a compile
/// error naming the generator, not an allocator abort.
fn generate_fib_n(n: u64) -> Result<Vec<Value>, String> {
    let mut out = crate::derive_support::try_buffer_for(n, "fib(n)")?;
    let (mut a, mut b): (u64, u64) = (1, 1);
    for _ in 0..n {
        out.push(Value::U64(a));
        let next = a.saturating_add(b);
        a = b;
        b = next;
    }
    Ok(out)
}

/// Fibonacci values up to and including the largest ≤ `max`.
fn generate_fib_until(max: u64) -> Vec<Value> {
    let mut out = Vec::new();
    let (mut a, mut b): (u64, u64) = (1, 1);
    while a <= max {
        out.push(Value::U64(a));
        let next = a.checked_add(b);
        a = b;
        match next {
            Some(v) => b = v,
            None => break,
        }
    }
    out
}

/// `1, 2, 4, ..., 2^(n-1)`.
fn generate_pow2_n(n: u64) -> Vec<Value> {
    // At most 64 terms, whatever `n` asks: the loop stops at 2^63.
    let mut out = Vec::with_capacity(n.min(64) as usize);
    for i in 0..n {
        if i >= 64 {
            break;
        } // 2^64 overflows u64
        out.push(Value::U64(1u64 << i));
    }
    out
}

/// 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.
fn generate_geometric_until(start: f64, factor: f64, max: f64) -> Vec<Value> {
    let mut out = Vec::new();
    let mut v = start;
    if factor <= 1.0 || start <= 0.0 || max <= 0.0 {
        // Defensive: avoid infinite loops with non-growing
        // factors. The "until" semantics implies growth.
        return out;
    }
    while v <= max {
        out.push(Value::F64(v));
        v *= factor;
    }
    out
}

/// Binomial coefficients `C(n, 0), C(n, 1), …, C(n, n)`.
fn generate_binomial(n: u64) -> Vec<Value> {
    // A row is cut where its coefficients pass `u64::MAX`, and every
    // coefficient of rows up to 67 fits, so no row holds more than 68
    // terms however large `n` is. `binomial(10^12)` is a short list,
    // not a request for a trillion.
    let mut out = Vec::with_capacity(n.min(67) as usize + 1);
    let mut c: u128 = 1;
    out.push(Value::U64(1));
    for k in 1..=n {
        c = c * (n - k + 1) as u128 / k as u128;
        if c > u64::MAX as u128 {
            break;
        }
        out.push(Value::U64(c as u64));
    }
    out
}

/// SRD 71: kernel-aware partition comprehension sources.
///
/// `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 (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 runtime dispatch the value is always
/// installed; a still-unresolved ident there falls out as an
/// unresolved-clause error downstream, never a silent empty
/// iteration.
fn try_eval_partition_call(text: &str, kernel: &dyn Lookup) -> Result<Option<Vec<Value>>, String> {
    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()
                ));
            }
            let src = arg_list[0].trim();
            let n = parse_u64_arg(arg_list[1], "subdivide.n")?;
            let Some(value) = kernel.lookup(src) else {
                // 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.
                let placeholder = 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,
                };
                return Ok(Some(vec![Value::from_partition(placeholder)]));
            };
            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(),
                ));
            };
            let subs = crate::iteration::cursor_partition::subdivide_partition(&p, n)?;
            Ok(Some(subs.into_iter().map(Value::from_partition).collect()))
        }
        // SRD-71 grammar-position desugaring of an explicit
        // `partitions(spec, [extent])` source. 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(),
                ));
            }
            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,
            };
            desugar_partition_spec(&spec, extent, "comprehension source `partitions(...)`")
                .map(Some)
        }
        // 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())
            }

            #[cfg(feature = "vectordata")]
            {
                if arg_list.len() != 2 {
                    return Err(format!(
                        "profile_partitions(dataset, pattern): expected 2 arguments, got {}",
                        arg_list.len()
                    ));
                }
                // 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.
                        let placeholder = 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,
                        };
                        Ok(Some(vec![Value::from_partition(placeholder)]))
                    }
                }
            }
        }
        _ => Ok(None),
    }
}

/// SRD-71 comprehension-position desugaring: 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 (a pre-evaluation probe before the value is
/// installed), a single placeholder partition is returned so iter-var type
/// detection still lands on `ext` — the same contract as
/// [`try_eval_partition_call`].
fn try_eval_param_partitions(
    text: &str,
    kernel: &dyn Lookup,
) -> Result<Option<Vec<Value>>, String> {
    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);
    }
    let Some(value) = kernel.lookup(ident) else {
        // Pre-eval probe: the param value isn't installed yet. Return one
        // placeholder so the clause's iter-var type-detects as `ext`.
        let placeholder = 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,
        };
        return Ok(Some(vec![Value::from_partition(placeholder)]));
    };
    // 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 SRD-71 grammar.
    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(),
        ));
    };
    desugar_partition_spec(
        spec,
        100,
        &format!("comprehension source `{ident}.partitions`"),
    )
    .map(Some)
}

/// 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 (SRD-71), 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, String> {
    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 kernel.lookup(a) {
            Some(Value::Str(s)) => Ok(s.to_string()),
            Some(other) => Err(format!(
                "partitions(...): `{a}` resolved to {} — expected a spec string",
                other.to_display_string(),
            )),
            None => Err(format!(
                "partitions(...): `{a}` did not resolve to a spec string in scope"
            )),
        };
    }
    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 (SRD 71).
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)));
    Ok(out)
}

/// `n` log-spaced points from `start` to `end` (inclusive).
/// Both bounds must be positive (log undefined otherwise).
fn generate_log_steps(start: f64, end: f64, n: u64) -> Result<Vec<Value>, String> {
    if start <= 0.0 || end <= 0.0 {
        return Err(format!(
            "log_steps: bounds must be positive, got start={start}, end={end}"
        ));
    }
    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.extend((0..n).map(|i| Value::F64((log_s + step * i as f64).exp())));
    Ok(out)
}

// ============================================================
// SRD-18c Layer 5 / SRD-18e Push 9: set operators
// ============================================================

/// Recognise `concat(...)`, `unique(...)`, etc. Each set op
/// recursively evaluates its arguments through `evaluate_spec`
/// (so `concat(1..10, fib(8))` works), then combines the
/// resulting lists.
fn try_eval_setop(text: &str, kernel: &dyn Lookup) -> Result<Option<Vec<Value>>, String> {
    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| -> Result<Vec<Value>, String> {
        evaluate_spec(t, kernel).map_err(|e| e.to_string())
    };
    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()
                ));
            }
            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>>, String> =
                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()
                ));
            }
            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()
                ));
            }
            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()
                ));
            }
            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()
                ));
            }
            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),
    }
}

// ============================================================
// SRD-18c §"Sequencer-style expansions" / Push 8: bucket /
// concat_seq / interval_seq — LUT 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>>, String> {
    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(arg_texts[0], kernel)?;
            let raw_ratios = evaluate_spec(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()
            ));
        }
    };

    if items.len() != ratios.len() {
        return Err(format!(
            "{name}: items.len() ({}) != ratios.len() ({})",
            items.len(),
            ratios.len(),
        ));
    }
    // 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 (the legacy silent
        // literal-list fallback masked this kind of typo six
        // layers downstream).
        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() {
        // Regression: workload runs on a system whose
        // vectordata catalog doesn't have the requested
        // dataset. The spec
        //   `profile in matching_profiles('nonexistent_dataset_xyz', 'label_')`
        // must produce a clean clause-level error naming the
        // resolution failure — NOT a "garbage" iter-var like
        // `matching_profiles('nonexistent_dataset_xyz'`
        // (truncated at the first comma) that flows downstream
        // into malformed CQL six layers later.
        //
        // Three failure layers used to compound here:
        //   1. `dataset_group_open` returned `Value::None` on
        //      catalog miss.
        //   2. `handle_of(&Value::None)` panicked with
        //      "expected Handle, got U64" — opaque.
        //   3. `evaluate_spec` swallowed the eval error and
        //      fell through to splitting the literal text on
        //      commas.
        // The user-visible result was a CQL parser error from a
        // malformed `DROP INDEX`. After this fix every layer
        // 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() {
        // SRD-18f Stage 2: a bare identifier source is a direct
        // wire/param reference — 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() {
        // SRD-18f §6: a bare identifier source that doesn't
        // resolve is a hard error (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}");
    }

    #[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()),
            ]
        );
    }

    // ── SRD-71: partition-list unpacking ─────────────────────

    #[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");
    }

    // ── SRD-18c Layer 2 / SRD-18e Push 3: 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() {
        // Push 4 SI suffixes meet Push 3 ranges — full
        // composition.
        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),
            ]
        );
    }

    // ── SRD-18c Layer 3 / SRD-18e Push 7: 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!((d - 1000.0).abs() < 1e-9);
        } 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("must be positive"), "{err}");
    }

    // ── SRD-18c Layer 5 / SRD-18e Push 9: 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));
    }

    // ── SRD-18c §"Sequencer-style expansions" / Push 8 ──

    #[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"]);
    }
}