earthsci-ast 0.6.0

//! The deterministic-per-document **build pipeline** — what
//! [`crate::problem::esm_problem`] runs before it compiles a right-hand side.
//!
//! This module used to be the public `prepare` / `Prepared` entry point.
//! `esm-libraries-spec.md` §2.5.1 replaces that: `prepare` and `Prepared` are
//! the same concept as `EsmProblem` under a local name, and construction absorbs
//! them rather than sitting beside them. What survives here is the pipeline
//! itself ([`run_build_pipeline`], crate-internal) and the two things a caller
//! genuinely reaches for — the build-time provider contract
//! ([`PrepareProvider`]) and the build-observability seam ([`PreparePhase`],
//! [`PrepareProgress`]), both extension seam under `API_SPEC.md` §3, neither
//! stable API.
//!
//! Runs everything deterministic-per-document ONCE — pushdown rewrite →
//! loader extent discovery → typed load → provider materialization →
//! build-time coordinate evaluation → value-invention → member-factor feedback
//! → gated pre-sliced fetch → dependency-ordered observed-graph evaluation —
//! and returns a [`PreparedBuild`] whose [`PreparedBuild::fields`] map holds the
//! build-time fields. This is the entry point the isrm.esm runner drives;
//! it never integrates.
//!
//! **Extent discovery** runs before the typed load because it CLOSES
//! metaparameters: a loader whose record count is only knowable after the
//! table is read declares `extent: {"metaparameter": "N_REC"}`, its providers
//! are sampled once up front (never twice), and the surviving length binds the
//! metaparameter every dependent index set is sized by. The caller no longer
//! counts rows and passes the number in — see
//! [`PrepareProvider::extent_metaparameter`].
//!
//! Those arrows are also [`PreparePhase`], and a host can watch them go by:
//! [`crate::problem::ProblemOptions::progress`] is the build-time counterpart of
//! [`crate::simulate::SolveOptions::progress`], down to sharing its
//! [`Flow`]. It exists because a document with no ODEs never reaches the
//! solver, so a dispatched static evaluation had no observer at all — no
//! progress bar, no cancel button, and no way to enforce a resource cap on a
//! run that takes a quarter of an hour. **Progress observation is
//! binding-local**, not a conformance surface: it changes no result, produces
//! no artifact the fixtures compare, and the Julia and Python `prepare` have no
//! equivalent (nor a `verbose`); the cross-binding observability contract is
//! `BuildInspection`, which carries VALUES and is pinned by CONFORMANCE_SPEC
//! §5.8. A binding that wants this may mirror it or not, freely.
//!
//! `pushdown_rewrite: true` opts into the automatic projection-pushdown
//! desugar ([`crate::pushdown_rewrite::desugar_pushdown`]) at this public
//! entry point, exactly as in Julia/Python:
//!
//! * the rewrite runs on the RAW authored document BEFORE the typed parse
//!   (the raw-JSON design note in [`crate::pushdown_rewrite`]). NOTE the
//!   name-rewriting trap the Julia binding had (its flattener rewrites
//!   coupling-fed references in equations but not in variable `expression`
//!   fields, so the pattern no longer matches post-flatten) does NOT apply
//!   here: this Rust path never flattens — it selects the single authored
//!   model and resolves the coupling `variable_map` by ALIASING the provider
//!   arrays under the model-local names, so the pattern is always detected on
//!   the authored spelling;
//! * the engine derives every provider gate from the rewrite's OWN record
//!   (`metadata.x_esd.pushdown.gated_select`) + the document coupling — the
//!   caller hand-authors NO gate and implements no selection plumbing;
//! * a `providers` entry the coupling routes onto a rewritten array is
//!   DEFERRED and fetched pre-sliced to the invented support set after
//!   value-invention has materialised the set's members (pushdown hook 2);
//! * the derived set's `member_factor` parameter is filled with the
//!   materialised member ids (pushdown hook 1), so the generated
//!   `pd_cell__*[c] = index(F, index(member_factor, c))` gathers resolve;
//! * VI skolem/overlap factor observeds that are themselves join-free
//!   observed expressions (the in-model LCC projections `X`/`Y` over raw
//!   `emis_lon`/`emis_lat`) are evaluated through the FULL expression
//!   evaluator ahead of value-invention — the Rust analogue of Julia's
//!   `_derive_binning_coords` general-eval seeding and Python's
//!   `_binning_coord_arrays(producer_seed_nodes=…)`. The deliberately tiny
//!   `_vi_eval` op set is untouched.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt;
use std::path::PathBuf;
use std::sync::Arc;

use ndarray::{ArrayD, Axis, IxDyn, Slice};
use serde_json::Value as JsonValue;

use crate::classification::DeclaredRank;
use crate::faq::resolve_expr_ranges_with_extents;
use crate::parse::LoadOptions;
use crate::pushdown_rewrite::{
    GateAxis, ProviderGate, desugar_pushdown, pushdown_coupling_pairs, pushdown_provider_gates,
};
use crate::simulate_array::{
    ArrMap, ConstArrayScope, Value as EvalValue, eval_expression_with_extents_and_consts_shared,
    run_value_invention_with_params,
};
use crate::template_imports::resolve_template_machinery;
use crate::types::{Expr, IndexSet, Model, VariableType};

/// What a [`crate::problem::ProblemOptions::progress`] observer wants the build
/// to do next —
/// the SAME type a [`crate::simulate::SolveOptions::progress`] observer
/// returns.
///
/// Deliberately shared rather than mirrored: a host that already drives a solve
/// through [`Flow::Cancel`] should not have to learn a second cancellation
/// idiom to drive a build.
pub use crate::simulate::Flow;

/// A gated provider deferred by [`BuildState::inject_providers`]: its key, the
/// provider itself, and the gate its fetch is planned against.
type GatedProvider = (String, Box<dyn PrepareProvider>, ProviderGate);

/// A build-time preparation failure.
#[derive(Debug, Clone)]
pub struct PrepareError(pub String);

impl PrepareError {
    /// The prefix every observer-requested cancellation carries. Public so
    /// [`PrepareError::is_cancelled`] rests on a documented contract rather
    /// than a private convention.
    pub const CANCELLED_PREFIX: &'static str = "cancelled by the caller during";

    /// Whether this error is a build progress observer's own
    /// [`Flow::Cancel`] rather than something going wrong — the counterpart of
    /// matching [`crate::simulate::SimulateError::Cancelled`].
    ///
    /// ## Why a message prefix and not a variant
    ///
    /// `PrepareError` is a public single-field tuple struct: an out-of-crate
    /// [`PrepareProvider`] constructs one as `PrepareError(msg)` and reads it
    /// back as `e.0` (the in-repo `esio_provider` bridge does both). Turning it
    /// into an enum, or adding a second field, breaks every one of them, while
    /// adding a field to `PrepareOptions` breaks none — so the source-compatible
    /// route wins, and the marker lives in the message, which is also the only
    /// place a host that merely logs the error would ever see it.
    ///
    /// ## A host still records WHY it cancelled
    ///
    /// This reports that a cancel happened, not what the observer wanted. A
    /// dispatcher distinguishing "the user pressed stop" from "the run hit its
    /// billing cap" must record that inside the observer, because only the
    /// observer knows — exactly as `earthscilab`'s `dispatch::solve` already
    /// does around a solve, where the two bill differently. The message
    /// names the phase and the item, so whichever it was, the stopping point is
    /// attributable rather than "somewhere in the last eight minutes".
    pub fn is_cancelled(&self) -> bool {
        self.0.starts_with(Self::CANCELLED_PREFIX)
    }
}

impl fmt::Display for PrepareError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "PrepareError: {}", self.0)
    }
}

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

fn err(msg: impl Into<String>) -> PrepareError {
    PrepareError(msg.into())
}

/// One axis of the NEUTRAL per-axis selection handed to a gated provider —
/// the Rust spelling of the Python engine's `"all" | [0-based indices]` form.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AxisSel {
    /// The full native axis.
    All,
    /// The listed 0-based native indices, in order.
    Indices(Vec<usize>),
    /// The half-open strided range `[start, stop)` by `step` (`step >= 1`) —
    /// the contiguous form, kept whole rather than expanded to indices so a
    /// store-backed reader can push a slice down instead of a 52,411-long
    /// index list.
    Range {
        /// Inclusive first index (0-based).
        start: usize,
        /// Exclusive last index.
        stop: usize,
        /// Stride (>= 1).
        step: usize,
    },
}

impl AxisSel {
    /// The 0-based indices this selector picks over an axis of length
    /// `dim_len` (the engine-side fallback for a provider that cannot push a
    /// selection down).
    pub fn indices(&self, dim_len: usize) -> Vec<usize> {
        match self {
            AxisSel::All => (0..dim_len).collect(),
            AxisSel::Indices(v) => v.clone(),
            AxisSel::Range { start, stop, step } => {
                (*start..(*stop).min(dim_len)).step_by(*step).collect()
            }
        }
    }
}

/// The CONST data-provider contract the build pipeline consumes: one provider feeds
/// ONE field (the `providers["<ModelPath>.<param>"]` convention shared with the
/// Julia/Python bindings). A gated provider may additionally honour a pushed-
/// down per-axis selection; one that cannot is fetched whole and sliced
/// engine-side (the fallback matches the pushdown result exactly).
pub trait PrepareProvider {
    /// CONST whole-field sample.
    fn sample(&mut self) -> Result<ArrayD<f64>, PrepareError>;

    /// Whether [`PrepareProvider::sample_with_selection`] pushes the selection
    /// down to the reader (fetching only what intersects it).
    fn supports_selection(&self) -> bool {
        false
    }

    /// Sample with a per-axis selection pushed down. EVERY requested axis is
    /// present in the result (a fixed axis comes back length-1 and is dropped
    /// by the engine).
    fn sample_with_selection(
        &mut self,
        _selection: &[AxisSel],
    ) -> Result<ArrayD<f64>, PrepareError> {
        Err(err("provider does not support selection pushdown"))
    }

    /// `false` for a DISCRETE provider (non-empty refresh times) — rejected by
    /// the build pipeline, which is build-time-only.
    fn is_const(&self) -> bool {
        true
    }

    /// A provider-declared gate (the fallback protocol mirroring Julia's
    /// `provider_gate_spec`); the record-derived gate takes precedence.
    fn gate_spec(&self) -> Option<ProviderGate> {
        None
    }

    /// The metaparameter this provider's own extent BINDS (esm-spec §8.9): a
    /// loader that discovers its record count at read time — an FF10 point
    /// inventory whose surviving-row count is not knowable until the table is
    /// decoded and filtered — declares `extent: {"metaparameter": "N_REC"}`,
    /// and the build closes that metaparameter with the length of this
    /// provider's leading axis BEFORE the typed load, instead of the caller
    /// counting rows and passing the number in.
    ///
    /// Every provider naming the same metaparameter must agree, which is also
    /// the alignment check across a table's columns.
    fn extent_metaparameter(&self) -> Option<String> {
        None
    }
}

// --------------------------------------------------------------------------- //
// Progress observation.
// --------------------------------------------------------------------------- //

/// Which stage of the build [`crate::problem::esm_problem`] is in when it reports.
///
/// These are the document-independent stages of the build pipeline, in
/// the order it runs them, and they are exactly the stages
/// [`crate::problem::ProblemOptions::verbose`] already narrates — the observer did not invent a
/// structure, it made the existing one addressable.
///
/// **The stages are nowhere near equal in cost.** On the InMAP ISRM document a
/// [`PreparePhase::GatedFetch`] is tens of gigabytes over the network and
/// [`PreparePhase::Observeds`] is a source-by-receptor contraction, while
/// [`PreparePhase::Rewrite`] and [`PreparePhase::Load`] are milliseconds. A
/// host that renders `index() / COUNT` as a percentage will show a bar that
/// sprints to 60% and then sits still for ten minutes. Drive the bar from
/// [`PrepareProgress::fraction`] *within* the reported phase and name the phase
/// (and [`PrepareProgress::item`]) beside it instead — the same advice
/// [`crate::simulate::Progress`] gives about a stiff solve's non-linear `t`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum PreparePhase {
    /// The projection-pushdown desugar over the raw authored document.
    Rewrite,
    /// The typed parse of the (possibly rewritten) document.
    Load,
    /// Eager materialization of the un-gated CONST providers, one unit each.
    ConstProviders,
    /// The producer-seeded join-free coordinate closure, one unit per observed.
    Coordinates,
    /// Value invention — the graph deriving its own support set.
    ValueInvention,
    /// Feeding the invented member ids back as each derived set's
    /// `member_factor`, one unit per derived index set.
    MemberFactors,
    /// The post-value-invention pre-sliced fetch of the gated providers. One
    /// unit per REQUEST, which is one per provider unless
    /// [`PrepareOptions::gated_fetch_batch`] splits it further.
    GatedFetch,
    /// Dependency-ordered evaluation of the observed graph, one unit per
    /// observed.
    Observeds,
}

impl PreparePhase {
    /// Every phase, in pipeline order.
    pub const ALL: [PreparePhase; 8] = [
        PreparePhase::Rewrite,
        PreparePhase::Load,
        PreparePhase::ConstProviders,
        PreparePhase::Coordinates,
        PreparePhase::ValueInvention,
        PreparePhase::MemberFactors,
        PreparePhase::GatedFetch,
        PreparePhase::Observeds,
    ];

    /// How many phases there are, for a host laying out a stage list.
    pub const COUNT: usize = Self::ALL.len();

    /// 0-based position in the pipeline. See the type's note before turning
    /// this into a percentage.
    pub fn index(self) -> usize {
        Self::ALL.iter().position(|p| *p == self).unwrap_or(0)
    }

    /// A human-readable phase name, as it appears in a cancellation message.
    pub fn label(self) -> &'static str {
        match self {
            PreparePhase::Rewrite => "the pushdown rewrite",
            PreparePhase::Load => "the typed load",
            PreparePhase::ConstProviders => "const provider materialization",
            PreparePhase::Coordinates => "build-time coordinate evaluation",
            PreparePhase::ValueInvention => "value invention",
            PreparePhase::MemberFactors => "member-factor feedback",
            PreparePhase::GatedFetch => "the gated fetch",
            PreparePhase::Observeds => "observed-graph evaluation",
        }
    }
}

impl fmt::Display for PreparePhase {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.label())
    }
}

/// How far along an in-flight build is, handed to
/// [`crate::problem::ProblemOptions::progress`] at every phase boundary AND at every unit of
/// work inside the two phases that dominate a large build.
///
/// A report is delivered BEFORE the unit it names is done, so `item` is what
/// `prepare` is about to spend its time on — which is the useful thing to show,
/// and the only placement at which returning [`Flow::Cancel`] avoids the work
/// rather than following it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PrepareProgress<'a> {
    /// The stage of the build this report comes from.
    pub phase: PreparePhase,
    /// Units of work already finished in this phase.
    pub done: usize,
    /// Units of work this phase will do, when that is knowable before it
    /// starts. `None` for a phase that is a single indivisible step.
    pub total: Option<usize>,
    /// What the phase is about to work on: an observed's name, a provider key,
    /// a derived index set. Empty when the report is a phase boundary rather
    /// than an item.
    pub item: &'a str,
}

impl PrepareProgress<'_> {
    /// Fraction of THIS PHASE's work completed, clamped to `[0, 1]`.
    ///
    /// `0.0` when the phase has no countable work, rather than a NaN, so a host
    /// can feed it to a bar without a guard. This is deliberately not a
    /// whole-build fraction: see [`PreparePhase`] on why the library refuses to
    /// invent phase weights it cannot know for a document it has not run.
    pub fn fraction(&self) -> f64 {
        match self.total {
            Some(t) if t > 0 => (self.done as f64 / t as f64).clamp(0.0, 1.0),
            _ => 0.0,
        }
    }
}

/// A build progress observer. See [`crate::problem::ProblemOptions::progress`].
///
/// Unconditionally `Send + Sync`, unlike [`crate::simulate::ProgressFn`], which
/// drops the bound on `wasm32` for a `js_sys::Function` observer: the whole
/// `prepare` module is `#[cfg(not(target_arch = "wasm32"))]`, so there is no
/// wasm host to accommodate and no reason to make native callers pay for one.
pub type PrepareProgressFn = Arc<dyn Fn(&PrepareProgress<'_>) -> Flow + Send + Sync>;

/// Build-time options for [`run_build_pipeline`]. Crate-internal: the public
/// spelling of every one of these is a field of
/// [`crate::problem::ProblemOptions`].
#[derive(Clone, Default)]
pub(crate) struct PrepareOptions {
    /// Select one model when the document holds several.
    pub model_name: Option<String>,
    /// Metaparameter bindings closed at load (esm-spec §9.7.6 site 4).
    pub metaparameters: BTreeMap<String, i64>,
    /// Base path anchoring relative `{ref}`s.
    pub base_path: Option<PathBuf>,
    /// Opt in to the automatic projection-pushdown desugar.
    pub pushdown_rewrite: bool,
    /// Scalar parameter overrides (exact or bare names), baked into the build.
    pub parameters: HashMap<String, f64>,
    /// Per-step progress lines on stdout.
    ///
    /// This is the built-in observer: it prints at the same points
    /// [`crate::problem::ProblemOptions::progress`] is called, and the two are independent —
    /// setting both prints and observes.
    pub verbose: bool,

    /// If `Some`, called at every phase boundary and at every unit of work
    /// inside [`PreparePhase::GatedFetch`] and [`PreparePhase::Observeds`].
    /// Returning [`Flow::Cancel`] abandons the build with a [`PrepareError`]
    /// for which [`PrepareError::is_cancelled`] is true.
    ///
    /// **Called unthrottled, deliberately**, for the same reason
    /// [`crate::simulate::SolveOptions::progress`] is: `prepare` has no
    /// portable clock to throttle against, and rate limiting therefore belongs
    /// to the host, which does. Keep the observer cheap — a document with
    /// hundreds of observeds reports hundreds of times in a build that may take
    /// milliseconds.
    ///
    /// ## What this is for
    ///
    /// A dispatched static evaluation — a document with no ODEs, which
    /// no solver ever touches — is otherwise a black box for as long as it
    /// runs: no progress, no cancel, and no way for a caller to enforce a
    /// resource cap except by killing the process. A watchdog thread is not a
    /// substitute: it stops at an arbitrary point with no attributable elapsed
    /// time, and cannot interrupt a single long fetch at all. Going through the
    /// observer means the stopping point is a named phase and a named item.
    pub progress: Option<PrepareProgressFn>,

    /// Split a gated provider's pre-sliced fetch into requests of at most this
    /// many native indices along the gated axis.
    ///
    /// `None` (the default) issues ONE request per gated provider, exactly as
    /// before this option existed. That is the right default and also the
    /// reason this option exists: one request is one
    /// [`PreparePhase::GatedFetch`] report, and on the InMAP ISRM document that
    /// single request is 15–25 GB over the network and the longest thing the
    /// build does. A host that wants a moving bar — or a cancel that lands in
    /// under several minutes — sets this and gets one report per batch instead.
    ///
    /// The result is IDENTICAL either way: the gated axis' index list is split
    /// into contiguous runs and the pieces are written back in order into one
    /// pre-allocated slab, so nothing is reordered and nothing is copied twice.
    /// The cost of a small batch is at the seams — a chunked store re-reads the
    /// chunk straddling each boundary, so `batches - 1` extra chunk reads —
    /// which is why the library does not pick a size on the caller's behalf.
    pub gated_fetch_batch: Option<usize>,

    /// Stop at the first evaluation that is interpreted PER CELL and hand it
    /// back as [`PreparedBuild::refused`] — set when the Problem's compiler is
    /// strict (esm-libraries-spec §2.5.10), so a `native` build refuses the
    /// document naming the observed instead of walking it. The walk itself
    /// stops at its first cell, so the refusal costs one cell, not the field.
    pub refuse_per_cell: bool,
}

// Hand-written because `PrepareProgressFn` is a trait object: it cannot derive
// `Debug`, and a `PrepareOptions` that no longer prints would be a regression
// for every existing `{:?}` on a build error path. Mirrors the same treatment
// `SolveOptions` needed when it grew an observer.
impl fmt::Debug for PrepareOptions {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PrepareOptions")
            .field("model_name", &self.model_name)
            .field("metaparameters", &self.metaparameters)
            .field("base_path", &self.base_path)
            .field("pushdown_rewrite", &self.pushdown_rewrite)
            .field("parameters", &self.parameters)
            .field("verbose", &self.verbose)
            .field(
                "progress",
                &self
                    .progress
                    .as_ref()
                    .map(|_| "<observer>")
                    .unwrap_or("None"),
            )
            .field("gated_fetch_batch", &self.gated_fetch_batch)
            .field("refuse_per_cell", &self.refuse_per_cell)
            .finish()
    }
}

/// The product of [`run_build_pipeline`]: every build-time-evaluable observed of the
/// prepared document, materialized through the document's own graph, plus the
/// value-invention products the contract record reports.
pub struct PreparedBuild {
    /// The (possibly rewritten) raw document.
    pub doc: JsonValue,
    /// The prepared model's name.
    pub model_name: String,
    /// Observed name → build-time field (dependency-complete).
    pub fields: HashMap<String, ArrayD<f64>>,
    /// Value-invention producer id (`from_faq`) → sorted-distinct member ids.
    pub members: HashMap<String, Vec<i64>>,
    /// Value-invention producer id → derived index-set extent.
    pub extents: HashMap<String, i64>,
    /// Provider keys that were deferred + fetched pre-sliced (sorted).
    pub gated_provider_keys: Vec<String>,
    /// Every evaluation the pipeline performed, in the order it performed
    /// them — what the Problem's compiler report shows for this route.
    pub rules: Vec<PipelineRule>,
    /// The per-cell evaluation that stopped the build, when
    /// [`PrepareOptions::refuse_per_cell`] was set. `fields` then holds only
    /// what was evaluated before it.
    pub refused: Option<PipelineRule>,
}

/// One evaluation the build pipeline performed, and HOW: through the
/// whole-array overlay, interpreted per cell by the oracle, or — for value
/// invention — by the relational engine.
///
/// The pipeline evaluates the document's observed graph with the reference
/// evaluator whatever compiler the Problem names, so these records are how
/// that route reaches the compiler report (esm-libraries-spec §2.5.10: "the
/// materialization of constants and static observeds at construction" is
/// under the refusal).
#[derive(Clone, Debug)]
pub(crate) struct PipelineRule {
    /// The observed (model-local), or the value-invention producer id.
    pub name: String,
    /// `"build-time observed"` or `"value invention"`.
    pub kind: &'static str,
    /// Which evaluator served it.
    pub route: Route,
}

/// The evaluator that served one [`PipelineRule`] — what its compiler-report
/// tier names.
#[derive(Clone, Debug)]
pub(crate) enum Route {
    /// The whole-array overlay (tier `"vectorized"`).
    Overlay,
    /// A standalone tape: under a strict compiler, an observed the overlay
    /// declines is lowered on the tape and run once (tier `"taped"`).
    Tape,
    /// The reference evaluator, walking cells one at a time, and why (tier
    /// `"oracle"`). The one route a strict compiler refuses.
    PerCell(String),
    /// Value invention's relational engine (tier `"relational"`): the
    /// producer's member set, computed ONCE at setup from build-time factors —
    /// a `continuous` producer is refused before it runs (CONFORMANCE_SPEC
    /// §5.7.6 guard 2). There is no compiled form of a set to decline to, and
    /// the array runtime's own build runs the same pass under every compiler,
    /// so it is neither a fallback nor refused.
    Relational,
}

impl Route {
    /// An observed's route, from the per-cell reason its evaluation gave.
    fn observed(per_cell: Option<String>) -> Route {
        per_cell.map_or(Route::Overlay, Route::PerCell)
    }
}

/// Why an observed the overlay did not take was walked per cell.
const PER_CELL_FAQ: &str = "the build pipeline evaluated it through the reference evaluator, \
    and a `faq` in it is not one the whole-array overlay takes, so its body was walked \
    once per cell";
/// Why a causal self-reference is walked per cell.
const PER_CELL_RECURRENCE: &str = "a causal self-reference (esm-spec §4.3.1.1), which the \
    build pipeline sweeps cell by cell through the reference evaluator";

// --------------------------------------------------------------------------- //
// Dependency order over the observeds (shared with the pushdown-era runner).
// --------------------------------------------------------------------------- //

/// The factor names an expression declares it reads (`aggregate.args` plus
/// every gather / bare reference), gathered recursively. The dependency edges
/// come from the MODEL, not from a re-derivation here.
fn declared_args(expr: &Expr, out: &mut HashSet<String>) {
    match expr {
        Expr::Variable(name) => {
            out.insert(name.clone());
        }
        Expr::Operator(node) => {
            for a in &node.args {
                declared_args(a, out);
            }
            node.for_each_child(&mut |c| declared_args(c, out));
        }
        _ => {}
    }
}

/// The key columns and envelope factors every `join` in `expr` names. They
/// are strings on the clause rather than expressions, so [`declared_args`]
/// does not see them.
fn join_names(expr: &Expr, out: &mut HashSet<String>) {
    if let Expr::Operator(node) = expr {
        for clause in node.join.iter().flatten() {
            out.extend(clause.on.iter().flatten().cloned());
            if let Some(ov) = &clause.overlap {
                out.extend(ov.src_env.iter().chain(ov.tgt_env.iter()).cloned());
            }
        }
        for a in &node.args {
            join_names(a, out);
        }
        node.for_each_child(&mut |c| join_names(c, out));
    }
}

/// The observeds whose definitions read a state, directly or through another
/// observed.
///
/// The build-time observed pass does not evaluate these: there is no state at
/// build time, so a read of one is unbound there. They are evaluated at run
/// time like every other observed, by the compiler the Problem names. This
/// includes a mounted subsystem's observeds, which join the model's own once
/// the subsystem is mounted.
fn state_dependent_observeds(model: &Model, defs: &HashMap<String, Expr>) -> HashSet<String> {
    let class = crate::classification::Classification::of(model);
    let states: HashSet<&str> = model
        .variables
        .iter()
        .filter(|(name, var)| var.var_type == VariableType::Unknown && !class.is_observed(name))
        .map(|(name, _)| name.as_str())
        .collect();
    let refs: HashMap<&str, HashSet<String>> = defs
        .iter()
        .map(|(name, e)| {
            let mut a = HashSet::new();
            declared_args(e, &mut a);
            (name.as_str(), a)
        })
        .collect();
    let mut dependent: HashSet<String> = HashSet::new();
    loop {
        let before = dependent.len();
        for (name, reads) in &refs {
            if !dependent.contains(*name)
                && reads
                    .iter()
                    .any(|r| states.contains(r.as_str()) || dependent.contains(r))
            {
                dependent.insert((*name).to_string());
            }
        }
        if dependent.len() == before {
            return dependent;
        }
    }
}

/// Dependency order over observed definitions: an observed follows every
/// observed it names. A cycle is a malformed model and is a hard error.
fn observed_order(defs: &HashMap<String, Expr>) -> Result<Vec<String>, PrepareError> {
    let deps: HashMap<String, HashSet<String>> = defs
        .iter()
        .map(|(n, e)| {
            let mut a = HashSet::new();
            declared_args(e, &mut a);
            a.retain(|x| defs.contains_key(x) && x != n);
            (n.clone(), a)
        })
        .collect();

    let mut ordered = Vec::new();
    let mut done: HashSet<String> = HashSet::new();
    let mut pending: HashSet<String> = defs.keys().cloned().collect();
    while !pending.is_empty() {
        let mut ready: Vec<String> = pending
            .iter()
            .filter(|n| deps[*n].iter().all(|d| done.contains(d)))
            .cloned()
            .collect();
        if ready.is_empty() {
            // Name the CYCLE, not the residue. "cyclic observed dependency
            // among [a, b, c, d, e]" tells an author that five observeds are
            // implicated and nothing about which edges close the loop; the
            // whole value of this diagnostic is the path (esm-spec §4.9.6).
            // Sorted roots and successors so the same cycle is named every run.
            let mut stuck: std::collections::BTreeMap<String, std::collections::BTreeSet<String>> =
                std::collections::BTreeMap::new();
            for n in &pending {
                stuck.insert(n.clone(), deps[n].iter().cloned().collect());
            }
            let cycle = crate::classification::first_observed_cycle(&stuck)
                // Unreachable in practice — the residue is non-empty precisely
                // because nothing in it became ready, which requires an
                // unsatisfied dependency inside it — but naming the residue
                // beats a panic if that invariant ever moves.
                .unwrap_or_else(|| stuck.keys().cloned().collect());
            return Err(err(format!(
                "observed_cycle: dependency cycle among observed variables: {}. Each is defined \
                 in terms of the next, so no evaluation order satisfies every definition \
                 (esm-spec §4.9.6). `esm validate` reports this cycle before any build.",
                cycle.join(" -> ")
            )));
        }
        ready.sort(); // deterministic tie-break
        for n in ready {
            ordered.push(n.clone());
            done.insert(n.clone());
            pending.remove(&n);
        }
    }
    Ok(ordered)
}

/// name → defining expression, for every OBSERVED unknown of `model` that this
/// pass EVALUATES (esm-spec §6.3.1).
///
/// From esm 1.0.0 an observed's definition is the RHS of the equation whose LHS
/// is the bare variable, not a `variables[v].expression` field — and that is
/// also where a VALUE-INVENTION producer has always lived. Before the two
/// merged, the field a definition sat in told them apart for free; now it does
/// not, so the relational outputs are filtered out explicitly. Handing one to
/// the array evaluator is not a small error: the `distinct` producer the
/// projection-pushdown desugar emits carries a `{"op": "true"}` body, which has
/// no evaluation rule at all (`unevaluable_operator`), because the relational
/// engine — not the evaluator — is what materializes it.
fn observed_defs(model: &Model) -> HashMap<String, Expr> {
    crate::classification::observed_definitions(model)
        .into_iter()
        .filter(|(_, def)| {
            serde_json::to_value(def)
                .is_ok_and(|raw| !crate::value_invention::is_value_invention_assignment(&raw))
        })
        .collect()
}

/// Every declared 0-D parameter's default (overridden by `overrides`, exact or
/// bare key) — the scalar evaluation scope, sorted by name.
///
/// **Selection and value resolution must agree.** They did not: the filter
/// admitted every parameter with `default.is_some()`, while the value read
/// `default_scalar()`, which is `None` for inline ARRAY data
/// ([`crate::types::InlineValue::as_scalar`]) — and the `.unwrap_or(0.0)`
/// underneath then FABRICATED a zero. A 0-D parameter whose `default` is array
/// data (there is no shape for the array to fill) therefore entered the build's
/// scalar scope bound to `0.0`, and every expression reading it evaluated
/// against a number the document never states. That is a silent wrong answer,
/// which is the one outcome a build-time scope must not produce.
///
/// A parameter with array data is not a 0-D parameter's default, so it is not
/// in this scope at all. It reaches the evaluator's fail-closed path instead of
/// a plausible zero. Such a document is rejected at load as
/// `array_default_without_shape` (esm-spec §6.3); this filter stays as the
/// backstop for routes that build without structural validation, which
/// `esm_problem` does. (Its mirror image — a SCALAR default on a SHAPED parameter,
/// which must broadcast rather than be dropped — is issue #219.)
fn scalar_params(
    model: &Model,
    model_name: &str,
    overrides: &HashMap<String, f64>,
) -> (Vec<f64>, Vec<String>) {
    let mut names: Vec<String> = model
        .variables
        .iter()
        .filter(|(_, v)| {
            v.var_type == VariableType::Parameter
                && v.shape.as_ref().map(|s| s.is_empty()).unwrap_or(true)
                && v.default_scalar().is_some()
        })
        .map(|(k, _)| k.clone())
        .collect();
    names.sort();
    let vals = names
        .iter()
        .map(|n| {
            // The caller may spell the key model-local, model-qualified (the
            // spelling an inline test's `parameter_overrides` reach the build
            // in), or by its bare tail.
            overrides
                .get(n)
                .or_else(|| overrides.get(&format!("{model_name}.{n}")))
                .or_else(|| overrides.get(n.rsplit('.').next().unwrap_or(n)))
                .copied()
                .unwrap_or_else(|| {
                    model.variables[n]
                        .default_scalar()
                        .expect("selected on default_scalar().is_some()")
                })
        })
        .collect();
    (vals, names)
}

// --------------------------------------------------------------------------- //
// Pushdown-path name aliasing (the Python `_inject_pushdown_aliases`, adapted
// to the UN-flattened Rust model scope: the coupling routes loader keys onto
// namespaced model names, and every dotted key additionally surfaces under its
// unique shallowest bare tail — the spelling the authored expressions use).
// --------------------------------------------------------------------------- //
/// Round an array of EXTERNAL data to the working precision of the variable
/// `key` fills (`crate::precision`, esm-spec §11.3.1).
///
/// Loaded data is the one class of value the evaluator does not itself produce,
/// so under `element_type: "Float32"` it is rounded ONCE here rather than on
/// each of the O(N) reads of it; every value computed downstream is already
/// binary32 because every operation rounds. Identity under Float64.
///
/// The precision is the TARGET VARIABLE's, not the document's, because a data
/// column is exactly where a per-variable `element_type` has to bite: a
/// document-wide binary32 rounding applied to an ingested ten-digit key
/// collapses distinct codes onto one another before a single expression has
/// run (`2265007010` and `2265007015` both become `2265007104`), so a key
/// column declared `Float64` must arrive unrounded. The provider key IS the
/// variable name it fills (`esio_provider`'s `ParamBinding::key`), bare or
/// `Model.var`, which is the spelling `precision::of_variable` resolves.
///
/// Applied only to genuinely external arrays — a caller's `const_arrays`, a
/// provider sample, a gated fetch. NOT to a member-id factor (invented integer
/// ids, which binary32 would corrupt above 2^24) and not to an
/// evaluator-produced field (already binary32, so it would be a no-op sweep).
fn round_external(key: &str, mut a: ArrayD<f64>) -> ArrayD<f64> {
    let prec = crate::precision::of_variable(key);
    if prec.is_f32() {
        a.mapv_inplace(|x| prec.round(x));
    }
    a
}

fn inject_aliases(arrays: &mut ArrMap, coupling: &[(String, String)]) {
    // The coupling routing is AUTHORITATIVE: a `variable_map` explicitly binds
    // the loader field to the model variable, so surface the array under both
    // the namespaced target and its model-local tail (the authored spelling
    // the un-flattened expressions use). This also disambiguates a tail that
    // several dotted keys would otherwise tie on (`MockSR.TotalPop` vs
    // `ISRM.TotalPop`), which the generic pass below refuses to guess at.
    for (frm, to) in coupling {
        if !arrays.contains_key(frm) {
            continue;
        }
        if !arrays.contains_key(to) {
            let a = arrays[frm].clone();
            arrays.insert(to.clone(), a);
        }
        let tail = to.rsplit('.').next().unwrap_or(to);
        if !arrays.contains_key(tail) {
            let a = arrays[frm].clone();
            arrays.insert(tail.to_string(), a);
        }
    }
    // dotted key → unique shallowest bare tail (existing keys never overwritten).
    let mut tails: HashMap<String, Vec<String>> = HashMap::new();
    for k in arrays.keys() {
        if let Some((_, tail)) = k.rsplit_once('.') {
            tails.entry(tail.to_string()).or_default().push(k.clone());
        }
    }
    for (tail, mut keys) in tails {
        if arrays.contains_key(&tail) {
            continue;
        }
        let mindepth = keys
            .iter()
            .map(|k| k.matches('.').count())
            .min()
            .unwrap_or(0);
        keys.retain(|k| k.matches('.').count() == mindepth);
        keys.sort();
        if keys.len() == 1 {
            let a = arrays[&keys[0]].clone();
            arrays.insert(tail, a);
        }
    }
}

// --------------------------------------------------------------------------- //
// Build-time coordinate path (work item 2): evaluate the join-free observeds
// an overlap-gated `distinct` producer transitively reads, through the FULL
// evaluator, ahead of value-invention.
// --------------------------------------------------------------------------- //

/// The names of the join-free observeds the producer seed nodes transitively
/// reference, i.e. the build-time coordinate closure.
fn producer_seed_closure(
    seeds: &[&Expr],
    defs: &HashMap<String, Expr>,
    join_free: &HashSet<String>,
) -> HashSet<String> {
    let mut needed: HashSet<String> = HashSet::new();
    let mut frontier: Vec<String> = Vec::new();
    let mut refs = HashSet::new();
    for s in seeds {
        declared_args(s, &mut refs);
    }
    for r in refs {
        if join_free.contains(&r) && needed.insert(r.clone()) {
            frontier.push(r);
        }
    }
    while let Some(cur) = frontier.pop() {
        let mut refs = HashSet::new();
        declared_args(&defs[&cur], &mut refs);
        for r in refs {
            if join_free.contains(&r) && needed.insert(r.clone()) {
                frontier.push(r);
            }
        }
    }
    needed
}

/// What [`eval_observed`] produced.
enum Evaluated {
    /// The dense field, and the route that evaluated it.
    Field(ArrayD<f64>, Route),
    /// Walked per cell under `refuse_per_cell`, and stopped at the first cell
    /// (`StopAtFirstCell`): there is no field, only the reason to refuse.
    Refused(String),
}

/// Lower one observed on a standalone tape and run it
/// (`simulate_array::eval_observed_taped`). `None` when `recurrence_only` and
/// the observed is not a recurrence the tape's sweep takes. Otherwise the outer
/// `Err` is a fault the run latched — the interpreter's own diagnostic for the
/// same document — and the inner one the tape's decline reason.
#[allow(clippy::too_many_arguments)]
fn compile_on_tape(
    name: &str,
    expr: &Expr,
    arrays: &ArrMap,
    param_vals: &[f64],
    param_names: &[String],
    const_arrays: &ConstArrayScope,
    recurrence_only: bool,
) -> Option<Result<Result<ArrayD<f64>, String>, String>> {
    use crate::simulate_array::StandaloneError;
    match crate::simulate_array::eval_observed_taped(
        name,
        expr,
        arrays,
        param_vals,
        param_names,
        const_arrays,
        recurrence_only,
    ) {
        None => None,
        Some(Ok(a)) => Some(Ok(Ok(a))),
        Some(Err(StandaloneError::Declined(why))) => Some(Ok(Err(why))),
        // Worded as the interpreter's own evaluation reports the same fault.
        Some(Err(StandaloneError::Fault(details))) => Some(Err(
            crate::compile_error::CompileError::InterpreterBuildError { details }.to_string(),
        )),
    }
}

/// The interpreter's shape for an observed's value: a scalar fills the
/// declared rank's scalar shape (`[]` or `[1]`), an array is kept.
fn shape_like_interpreter(a: ArrayD<f64>, rank: DeclaredRank) -> ArrayD<f64> {
    if a.ndim() == 0 {
        ArrayD::from_elem(IxDyn(rank.scalar_shape()), a[IxDyn(&[])])
    } else {
        a
    }
}

/// Evaluate one observed through the full evaluator; returns the dense field.
///
/// Under `refuse_per_cell` a per-cell walk stops at its first cell, so a
/// document the strict compiler refuses is refused for the price of one cell
/// rather than the whole reference evaluation.
#[allow(clippy::too_many_arguments)]
fn eval_observed(
    name: &str,
    def: &Expr,
    arrays: &ArrMap,
    param_vals: &[f64],
    param_names: &[String],
    index_sets: &HashMap<String, IndexSet>,
    extents: &HashMap<String, i64>,
    const_arrays: &ConstArrayScope,
    rank: DeclaredRank,
    refuse_per_cell: bool,
) -> Result<Evaluated, PrepareError> {
    // The observed is evaluated at the element type of the variable it defines
    // (esm-spec §11.3.1) — the document's unless that variable declared its
    // own. This is where the build pipeline materializes a relational
    // document's whole observed graph, so it is where an equation over
    // binary64 keys has to be evaluated in binary64: a `sum_product` over one
    // exact key still rounds its accumulator, which is enough to turn
    // `2260007005` back into `2260006912`. A no-op swap for every document
    // that declares no per-variable element type.
    let _rule_precision = crate::precision::has_variable_overrides()
        .then(|| crate::precision::enter(crate::precision::of_variable(name)));
    let mut expr = def.clone();
    resolve_expr_ranges_with_extents(&mut expr, index_sets, extents)
        .map_err(|e| err(format!("resolve ranges for {name}: {e}")))?;
    // A CAUSAL SELF-REFERENCE (esm-spec §4.3.1.1) cannot be evaluated
    // wholesale: its cells are not independent, and the self-read would
    // resolve against a map that does not yet hold the array being built.
    // Checked BEFORE the wholesale call, and returning `None` for every
    // expression that contains no self-read, so a document that does not use
    // the construct takes exactly the path it took before.
    //
    // This is the fix for the defect that made the construct dead on this
    // path: it evaluated correctly under `esm test` (the per-step observed
    // materialization) and silently wrong wherever the build pipeline was
    // taken — which is any ingesting document, and every document under
    // `esm simulate`. The self-read fell through to an unbound-name NaN and a
    // `max(x, 0)` in the body laundered it to 0.0, so the answer came back
    // finite, plausible and wrong with nothing logged.
    // Under a strict compiler a recurrence goes to the tape's ordered sweep
    // first (`compile_on_tape`), and is refused only if the tape declines it.
    if refuse_per_cell
        && let Some(tape) = compile_on_tape(
            name,
            &expr,
            arrays,
            param_vals,
            param_names,
            const_arrays,
            true,
        )
    {
        return tape
            .map_err(|why| err(format!("evaluate {name}: {why}")))
            .map(|r| match r {
                Ok(a) => Evaluated::Field(shape_like_interpreter(a, rank), Route::Tape),
                Err(declined) => Evaluated::Refused(format!(
                    "{PER_CELL_RECURRENCE}, and the tape declined it: {declined}"
                )),
            });
    }
    let stop = refuse_per_cell.then(crate::simulate_array::StopAtFirstCell::arm);
    let refused = || stop.is_some() && crate::simulate_array::per_cell_walk_refused();
    if let Some(res) = crate::simulate_array::eval_observed_recurrence(
        name,
        &expr,
        arrays,
        param_vals,
        param_names,
        0.0,
        extents,
        const_arrays,
    ) {
        if refused() {
            return Ok(Evaluated::Refused(PER_CELL_RECURRENCE.to_string()));
        }
        return res
            .map(|a| Evaluated::Field(a, Route::PerCell(PER_CELL_RECURRENCE.to_string())))
            .map_err(|e| err(format!("evaluate {name}: {e}")));
    }
    let walks_before = crate::simulate_array::per_cell_walks();
    crate::simulate_array::take_overlay_decline();
    let val = eval_expression_with_extents_and_consts_shared(
        &expr,
        arrays,
        param_vals,
        param_names,
        0.0,
        extents,
        const_arrays,
    );
    // Before the result is read: a stopped walk's value is a placeholder, and
    // so is any error computed from it.
    if refused() {
        // The overlay declined a `faq` of this observed: lower the whole
        // observed on the tape instead, from the same inputs.
        let overlay = crate::simulate_array::take_overlay_decline()
            .unwrap_or("a construct it has no whole-array form for");
        drop(stop);
        let tape = compile_on_tape(
            name,
            &expr,
            arrays,
            param_vals,
            param_names,
            const_arrays,
            false,
        )
        .expect("a non-recurrence always reaches the tape");
        return match tape {
            Err(fault) => Err(err(format!("evaluate {name}: {fault}"))),
            Ok(Ok(a)) => Ok(Evaluated::Field(
                shape_like_interpreter(a, rank),
                Route::Tape,
            )),
            Ok(Err(declined)) => Ok(Evaluated::Refused(format!(
                "the whole-array overlay declined a `faq` in it ({overlay}), and the tape \
                 declined it: {declined}"
            ))),
        };
    }
    let val = val.map_err(|e| err(format!("evaluate {name}: {e}")))?;
    let per_cell =
        (crate::simulate_array::per_cell_walks() != walks_before).then(|| PER_CELL_FAQ.to_string());
    let field = match val {
        EvalValue::Array(a) => *a,
        EvalValue::Scalar(s) => ArrayD::from_elem(IxDyn(rank.scalar_shape()), s),
    };
    Ok(Evaluated::Field(field, Route::observed(per_cell)))
}

// --------------------------------------------------------------------------- //
// Pushdown hook 2: the gated pre-sliced fetch.
// --------------------------------------------------------------------------- //

struct GatedFetchPlan {
    selection: Vec<AxisSel>,
    drop_axes: Vec<usize>,
    /// Position of the gated axis in the REQUEST (`selection`), i.e. before the
    /// fixed axes are dropped. This is the axis a batched fetch subdivides.
    gated_pos: usize,
    gated_pos_out: usize,
    gated_extent: usize,
    /// How many provider requests this plan will issue — filled in by the
    /// caller, which is where the batch size and `supports_selection` are known,
    /// so the whole phase can report a running count.
    n_requests: usize,
}

/// The model variable a gated provider's slab is published under (see
/// [`BuildState::fetch_gated_providers`]).
fn gated_target(key: &str, gate: &ProviderGate, pd_coupling: &[(String, String)]) -> String {
    pd_coupling
        .iter()
        .find(|(frm, to)| frm == key || to == key)
        .map(|(_, to)| to.rsplit('.').next().unwrap_or(to).to_string())
        .unwrap_or_else(|| gate.applies_to[0].clone())
}

/// Resolve a gate's per-axis `selection` from the materialised value-invention
/// members (0-based native indices for the reader).
fn gated_fetch_plan(
    key: &str,
    gate: &ProviderGate,
    index_sets: &HashMap<String, IndexSet>,
    members: &HashMap<String, Vec<i64>>,
    extents: &HashMap<String, i64>,
) -> Result<GatedFetchPlan, PrepareError> {
    let mut selection: Vec<AxisSel> = Vec::with_capacity(gate.axes.len());
    let mut drop_axes: Vec<usize> = Vec::new();
    let mut gated_pos: Option<usize> = None;
    let mut gated_extent = 0usize;
    for (ax_i, ax) in gate.axes.iter().enumerate() {
        match ax {
            GateAxis::All => selection.push(AxisSel::All),
            GateAxis::Fixed(fi) => {
                selection.push(AxisSel::Indices(vec![*fi]));
                drop_axes.push(ax_i);
            }
            GateAxis::Range { start, stop, step } => selection.push(AxisSel::Range {
                start: *start,
                stop: *stop,
                step: *step,
            }),
            GateAxis::GatedBy(sname) => {
                let faq = index_sets
                    .get(sname)
                    .and_then(|is| (is.kind == "derived").then(|| is.from_faq.clone()))
                    .flatten()
                    .ok_or_else(|| {
                        err(format!(
                            "gated provider '{key}' gates on '{sname}' which is not a \
                             derived index set with a from_faq"
                        ))
                    })?;
                let mem = members.get(&faq).ok_or_else(|| {
                    err(format!(
                        "gated provider '{key}' gates on '{sname}' (faq '{faq}') but its \
                         value-invention members were not materialised"
                    ))
                })?;
                let mem0: Vec<usize> = mem
                    .iter()
                    .map(|&m| {
                        usize::try_from(m - 1).map_err(|_| {
                            err(format!(
                                "gated provider '{key}': member id {m} is not a 1-based \
                                 cell id"
                            ))
                        })
                    })
                    .collect::<Result<_, _>>()?;
                gated_extent = extents.get(&faq).map(|&e| e as usize).unwrap_or(mem0.len());
                selection.push(AxisSel::Indices(mem0));
                gated_pos = Some(ax_i);
            }
        }
    }
    let gated_pos = gated_pos
        .ok_or_else(|| err(format!("gated provider '{key}' declares no gated_by axis")))?;
    let gated_pos_out = gated_pos - drop_axes.iter().filter(|&&d| d < gated_pos).count();
    Ok(GatedFetchPlan {
        selection,
        drop_axes,
        gated_pos,
        gated_pos_out,
        gated_extent,
        n_requests: 1,
    })
}

/// The build pipeline's internal observation sink, erased so a helper can report
/// without knowing whether an observer is attached or what it captured.
type ReportFn<'a> =
    dyn FnMut(PreparePhase, usize, Option<usize>, &str) -> Result<(), PrepareError> + 'a;

/// Run a gate's plan, reporting through `report` and honouring a cancel.
///
/// One request, unless `batch` splits the gated axis' index list into contiguous
/// runs — in which case the pieces are assembled into ONE pre-allocated slab in
/// request order, which is bit-identical to the single-request result because
/// nothing is reordered and the concatenation seam is a plain copy. Assembling
/// in place rather than via `ndarray::concatenate` matters at this size: the SR
/// slabs are hundreds of MB, and holding every piece plus their concatenation
/// would peak at twice the answer instead of the answer plus one batch.
#[allow(clippy::too_many_arguments)]
fn run_gated_fetch(
    key: &str,
    prov: &mut dyn PrepareProvider,
    plan: &GatedFetchPlan,
    batch: Option<usize>,
    issued: &mut usize,
    total_requests: usize,
    report: &mut ReportFn<'_>,
) -> Result<ArrayD<f64>, PrepareError> {
    // A provider that cannot push the selection down fetches the WHOLE field in
    // one indivisible call and is sliced engine-side; there is no seam to batch
    // and reporting per batch would be a lie about work already done.
    if !prov.supports_selection() {
        report(PreparePhase::GatedFetch, *issued, Some(total_requests), key)?;
        *issued += 1;
        let full = prov
            .sample()
            .map_err(|e| err(format!("gated fetch '{key}' (whole): {}", e.0)))?;
        return drop_fixed_axes(key, slice_whole(full, &plan.selection), &plan.drop_axes);
    }

    let AxisSel::Indices(gated_idx) = &plan.selection[plan.gated_pos] else {
        return Err(err(format!(
            "gated provider '{key}': the gated axis resolved to the full axis, \
             which is not a support set"
        )));
    };
    let size = batch.unwrap_or(usize::MAX).max(1);
    let n_batches = gated_idx.len().div_ceil(size).max(1);

    if n_batches == 1 {
        report(PreparePhase::GatedFetch, *issued, Some(total_requests), key)?;
        *issued += 1;
        let a = prov
            .sample_with_selection(&plan.selection)
            .map_err(|e| err(format!("gated fetch '{key}': {}", e.0)))?;
        return drop_fixed_axes(key, a, &plan.drop_axes);
    }

    let mut out: Option<ArrayD<f64>> = None;
    let mut filled = 0usize;
    for (i, run) in gated_idx.chunks(size).enumerate() {
        // BEFORE the request: a cancel here skips the fetch rather than
        // arriving just after it, which is the entire point of subdividing.
        report(PreparePhase::GatedFetch, *issued, Some(total_requests), key)?;
        *issued += 1;
        let mut sel = plan.selection.clone();
        sel[plan.gated_pos] = AxisSel::Indices(run.to_vec());
        let part = prov
            .sample_with_selection(&sel)
            .map_err(|e| err(format!("gated fetch '{key}' (batch {i}): {}", e.0)))?;
        let part = drop_fixed_axes(key, part, &plan.drop_axes)?;
        let width = part.shape().get(plan.gated_pos_out).copied().unwrap_or(0);
        if width != run.len() {
            return Err(err(format!(
                "gated provider '{key}': batch {i} asked for {} indices and got \
                 {width} back",
                run.len()
            )));
        }
        let dst = out.get_or_insert_with(|| {
            let mut shape = part.shape().to_vec();
            shape[plan.gated_pos_out] = gated_idx.len();
            ArrayD::zeros(IxDyn(&shape))
        });
        dst.slice_axis_mut(
            Axis(plan.gated_pos_out),
            Slice::from(filled..filled + width),
        )
        .assign(&part);
        filled += width;
    }
    out.ok_or_else(|| err(format!("gated provider '{key}': no batches were fetched")))
}

/// Drop the (length-1) fixed axes of a fetched slab.
fn drop_fixed_axes(
    key: &str,
    arr: ArrayD<f64>,
    drop_axes: &[usize],
) -> Result<ArrayD<f64>, PrepareError> {
    if drop_axes.is_empty() {
        return Ok(arr);
    }
    let shape = arr.shape().to_vec();
    for &d in drop_axes {
        if shape.get(d).copied() != Some(1) {
            return Err(err(format!(
                "gated provider '{key}': fixed axis {d} came back with length \
                 {:?} (expected 1) in shape {shape:?}",
                shape.get(d)
            )));
        }
    }
    let out: Vec<usize> = shape
        .iter()
        .enumerate()
        .filter(|(i, _)| !drop_axes.contains(i))
        .map(|(_, &s)| s)
        .collect();
    arr.into_shape_with_order(IxDyn(&out)).map_err(|e| {
        err(format!(
            "gated provider '{key}': reshape after fixed-axis drop: {e}"
        ))
    })
}

/// FALLBACK slice for a provider that cannot push a selection down: fetch
/// whole, then take the selected indices axis by axis (identical result).
pub(crate) fn slice_whole(full: ArrayD<f64>, selection: &[AxisSel]) -> ArrayD<f64> {
    let mut arr = full;
    for (i, ax) in selection.iter().enumerate() {
        if matches!(ax, AxisSel::All) {
            continue;
        }
        let dim_len = arr.shape().get(i).copied().unwrap_or(0);
        let idx = ax.indices(dim_len);
        arr = arr.select(Axis(i), &idx);
    }
    arr
}

// --------------------------------------------------------------------------- //
// The entry point.
// --------------------------------------------------------------------------- //

/// Does `raw` carry an esm-spec §9.7.2 import EDGE — at the top level of a
/// library file, or inside a component?
///
/// The pushdown prepass reads the RAW document, before the typed load, so an
/// edge that is still unresolved hides the imported templates and index sets
/// from the recogniser: the binning body reads as an unexpandable
/// `apply_expression_template` and the rewrite silently declines, which costs
/// the whole ungated fetch. Resolving is gated on this predicate rather than run
/// unconditionally so a document with no edge reaches `desugar_pushdown` in
/// exactly the bytes it does today (metaparameters unfolded, goldens unmoved).
fn has_template_import_edge(raw: &JsonValue) -> bool {
    let Some(obj) = raw.as_object() else {
        return false;
    };
    if obj.contains_key("expression_template_imports") {
        return true;
    }
    ["models", "reaction_systems"].iter().any(|kind| {
        obj.get(*kind)
            .and_then(JsonValue::as_object)
            .is_some_and(|comps| {
                comps.values().any(|c| {
                    c.as_object()
                        .is_some_and(|c| c.contains_key("expression_template_imports"))
                })
            })
    })
}

/// Run the deterministic-per-document pipeline over `doc` (a raw parsed `.esm`
/// document), once: pushdown rewrite (when opted in) → loader extent discovery
/// → typed load → provider materialization → build-time coordinate evaluation
/// → value invention → member-factor feedback → gated pre-sliced fetch →
/// dependency-ordered observed-graph evaluation. Returns the [`PreparedBuild`]
/// artifact.
///
/// `providers` maps the CONSUMING PARAMETER's namespaced name
/// (`"<ModelPath>.<param>"`) to a CONST [`PrepareProvider`]; an entry the
/// rewrite record's gate routes onto a gated array is DEFERRED and fetched
/// pre-sliced after value-invention. From esm 1.0.0 a data source declares no
/// variables, so the consuming parameter is the only spelling that names one
/// field and every field: two parameters may read one `file_variable`
/// differently, and two models may declare the same parameter name against one
/// source. `const_arrays` are the caller-supplied build-time factor arrays
/// (keyed by model-local or `<Source>.<file_variable>` names — the routing
/// aliasing surfaces both spellings).
///
/// Crate-internal: the public entry point is
/// [`crate::problem::esm_problem`], which calls this and then compiles the
/// right-hand side.
///
/// # Watching a long build
///
/// A build over real data is not fast: on the InMAP ISRM the gated fetch alone
/// is tens of gigabytes and the whole `prepare` runs for the better part of a
/// quarter of an hour. [`crate::problem::ProblemOptions::progress`] observes it as it goes —
/// per gated request and per observed, not merely per phase — and returning
/// [`Flow::Cancel`] stops it at a named point. Pair it with
/// [`PrepareOptions::gated_fetch_batch`] so the fetch itself is interruptible.
pub(crate) fn run_prepare(
    doc: &JsonValue,
    const_arrays: HashMap<String, ArrayD<f64>>,
    providers: Vec<(String, Box<dyn PrepareProvider>)>,
    opts: &PrepareOptions,
) -> Result<PreparedBuild, PrepareError> {
    let mut providers = providers;
    let mut metaparameters = opts.metaparameters.clone();
    let discovered = discover_loader_extents(&mut providers, opts, &mut metaparameters)?;

    let resolved = resolve_pushdown_imports(doc, opts, &metaparameters)?;
    let doc: &JsonValue = resolved.as_ref().unwrap_or(doc);
    let provider_keys: Vec<String> = providers.iter().map(|(k, _)| k.clone()).collect();
    let rw = apply_pushdown_rewrite(doc, opts, &provider_keys)?;
    check_discovered_gate_conflict(&discovered, &rw.pd_gates)?;

    let mut st = BuildState::load_document(rw.rewritten.as_ref(), opts, metaparameters)?;
    let gated = st.inject_providers(const_arrays, providers, discovered, &rw.pd_gates)?;
    st.alias_pushdown_names(&rw.pd_coupling);
    st.bind_inline_parameters(&gated);
    st.seal_const_registry();
    st.check_free_names()?;
    st.collect_observed_defs()?;
    st.eval_buildtime_coordinates()?;
    if st.refused.is_some() {
        return Ok(st.finish(rw.rewritten.into_owned()));
    }
    st.invent_values()?;
    st.feed_member_factors()?;
    st.eval_observeds_before_fetch(&gated, &rw.pd_coupling)?;
    if st.refused.is_some() {
        return Ok(st.finish(rw.rewritten.into_owned()));
    }
    st.fetch_gated_providers(gated, &rw.pd_coupling)?;
    st.eval_observed_graph()?;
    Ok(st.finish(rw.rewritten.into_owned()))
}

/// Verbose-mode narration; a no-op unless [`PrepareOptions::verbose`] is set.
fn vlog(opts: &PrepareOptions, msg: &str) {
    if opts.verbose {
        println!("{msg}");
    }
}

/// The observation point. Every phase boundary and every countable unit of
/// work inside the two phases that dominate a large build goes through here,
/// so the phase structure `verbose` narrates and the phase structure a host
/// observes cannot drift — there is only one.
///
/// Returns `Err` on `Flow::Cancel`, which unwinds `prepare` through the `?`
/// it is already threaded on. The message names the phase and the item, so a
/// cancel is attributable to a point in the build rather than to a wall
/// clock the library does not have.
fn report_progress(
    opts: &PrepareOptions,
    phase: PreparePhase,
    done: usize,
    total: Option<usize>,
    item: &str,
) -> Result<(), PrepareError> {
    let Some(cb) = &opts.progress else {
        return Ok(());
    };
    let p = PrepareProgress {
        phase,
        done,
        total,
        item,
    };
    match cb(&p) {
        Flow::Continue => Ok(()),
        Flow::Cancel => Err(err(format!(
            "{} {phase}{}",
            PrepareError::CANCELLED_PREFIX,
            if item.is_empty() {
                match total {
                    Some(t) => format!(" ({done} of {t} done)"),
                    None => String::new(),
                }
            } else {
                match total {
                    Some(t) => format!(" at '{item}' ({} of {t})", done + 1),
                    None => format!(" at '{item}'"),
                }
            }
        ))),
    }
}

/// Extent discovery: a loader that measures its OWN record count.
///
/// FIRST, ahead of the rewrite, because a discovered extent binds a
/// metaparameter and every resolution below closes metaparameters at the
/// loader API (§9.7.6 site 4). This is the Julia ordering (`simulate.jl`
/// prepare); Rust used to discover AFTER the rewrite, which was harmless only
/// for as long as nothing between the two steps folded a metaparameter — the
/// §9.7 resolution below does. A gated provider is still rejected outright:
/// its extent is the value-invention set's, which does not exist yet — the
/// provider's OWN declared gate is caught here, the gate the rewrite record
/// derives in the check after the rewrite.
fn discover_loader_extents(
    providers: &mut [(String, Box<dyn PrepareProvider>)],
    opts: &PrepareOptions,
    metaparameters: &mut BTreeMap<String, i64>,
) -> Result<HashMap<String, ArrayD<f64>>, PrepareError> {
    let mut discovered: HashMap<String, ArrayD<f64>> = HashMap::new();
    let mut discovered_by: BTreeMap<String, (i64, String)> = BTreeMap::new();
    for (key, prov) in providers.iter_mut() {
        let Some(mp) = prov.extent_metaparameter() else {
            continue;
        };
        if prov.gate_spec().is_some() {
            return Err(err(format!(
                "provider '{key}' both GATES on a derived index set and declares the \
                 extent metaparameter '{mp}'; a gated slab's extent is the gating \
                 set's, not a discovered one"
            )));
        }
        let a = prov
            .sample()
            .map_err(|e| err(format!("extent discovery for '{key}': {}", e.0)))?;
        let n = a.shape().first().copied().unwrap_or(0) as i64;
        if let Some((prev, prev_key)) = discovered_by.get(&mp)
            && *prev != n
        {
            return Err(err(format!(
                "loader extent '{mp}' is {prev} from provider '{prev_key}' but {n} from \
                 '{key}' — the loader's variables are not aligned on one record axis"
            )));
        }
        if let Some(bound) = metaparameters.get(&mp)
            && *bound != n
            && !discovered_by.contains_key(&mp)
        {
            return Err(err(format!(
                "metaparameter '{mp}' was closed at {bound} by the caller but provider \
                 '{key}' discovers {n} records; drop the binding and let the loader \
                 declare its own extent"
            )));
        }
        vlog(opts, &format!("  [prepare] extent {mp} <- {key} = {n}"));
        discovered_by.insert(mp.clone(), (n, key.clone()));
        metaparameters.insert(mp, n);
        discovered.insert(key.clone(), a);
    }
    Ok(discovered)
}

/// Phase-1 semantics: pushdown prepass BEFORE the typed parse.
///
/// §9.7 imports resolve BEFORE the recogniser looks. `desugar_pushdown`
/// expands `apply_expression_template` references to find the containment
/// predicate, but it can only expand what is IN SCOPE — and an import edge
/// puts the library's templates and index sets in scope at LOAD, which has
/// not happened yet on this raw-JSON path. Skipping this makes a document
/// that factors its binning body through an imported library fail detection
/// silently and fetch every provider-backed array whole.
fn resolve_pushdown_imports(
    doc: &JsonValue,
    opts: &PrepareOptions,
    metaparameters: &BTreeMap<String, i64>,
) -> Result<Option<JsonValue>, PrepareError> {
    report_progress(opts, PreparePhase::Rewrite, 0, None, "")?;
    if opts.pushdown_rewrite && has_template_import_edge(doc) {
        resolve_template_machinery(
            doc,
            opts.base_path
                .clone()
                .unwrap_or_else(|| PathBuf::from("."))
                .as_path(),
            metaparameters,
        )
        .map_err(|e| err(format!("resolve template imports: {}", e.message)))
    } else {
        Ok(None)
    }
}

/// What the pushdown rewrite phase hands the rest of the pipeline.
struct RewriteOutcome<'a> {
    /// The document the typed load parses: the rewrite's output when it fired,
    /// the input borrowed unchanged otherwise.
    rewritten: std::borrow::Cow<'a, JsonValue>,
    /// Record-derived provider gates (the rewrite's own `metadata.x_esd.pushdown`).
    pd_gates: HashMap<String, ProviderGate>,
    /// The pushdown coupling routing (loader key → namespaced model name).
    pd_coupling: Vec<(String, String)>,
}

/// Run the pushdown rewrite (when opted in) and read its gate and coupling
/// metadata off the rewritten document.
fn apply_pushdown_rewrite<'a>(
    doc: &'a JsonValue,
    opts: &PrepareOptions,
    provider_keys: &[String],
) -> Result<RewriteOutcome<'a>, PrepareError> {
    let rewritten = if opts.pushdown_rewrite {
        desugar_pushdown(doc, opts.model_name.as_deref()).map_err(|e| err(e.0))?
    } else {
        std::borrow::Cow::Borrowed(doc)
    };
    let rewrite_fired = matches!(rewritten, std::borrow::Cow::Owned(_));
    let pd_gates: HashMap<String, ProviderGate> = if rewrite_fired {
        pushdown_provider_gates(&rewritten, provider_keys).map_err(|e| err(e.0))?
    } else {
        HashMap::new()
    };
    let pd_coupling = pushdown_coupling_pairs(&rewritten);
    vlog(
        opts,
        &format!(
            "  [prepare] pushdown rewrite {}",
            if rewrite_fired {
                "fired"
            } else {
                "did not fire"
            }
        ),
    );
    Ok(RewriteOutcome {
        rewritten,
        pd_gates,
        pd_coupling,
    })
}

/// A discovered extent and a record-derived gate are mutually exclusive: a
/// gated slab's extent belongs to the gating set, which value-invention has
/// not materialised yet. (The provider's OWN declared gate is caught in the
/// discovery loop above; this catches the gate the rewrite record derives,
/// which only exists now.)
fn check_discovered_gate_conflict(
    discovered: &HashMap<String, ArrayD<f64>>,
    pd_gates: &HashMap<String, ProviderGate>,
) -> Result<(), PrepareError> {
    for key in discovered.keys() {
        if pd_gates.contains_key(key) {
            return Err(err(format!(
                "provider '{key}' both GATES on a derived index set and declares an \
                 extent metaparameter; a gated slab's extent is the gating set's, not \
                 a discovered one"
            )));
        }
    }
    Ok(())
}

/// The build pipeline's accumulator state: the evaluation namespace plus the
/// artifacts each phase deposits for the phases after it. One method per
/// phase; [`run_prepare`] is the phase list.
struct BuildState<'o> {
    opts: &'o PrepareOptions,
    /// The typed model under build.
    model: Model,
    model_name: String,
    index_sets: HashMap<String, IndexSet>,
    /// The scalar evaluation scope (declared 0-D parameter defaults, overridden).
    param_vals: Vec<f64>,
    param_names: Vec<String>,
    /// The evaluation namespace: the build-time factor data, then everything
    /// the pipeline materialises into it (gated slabs, member factors,
    /// evaluated observeds).
    arrays: ArrMap,
    /// The CONST-ARRAY registry (see [`BuildState::seal_const_registry`]).
    const_scope: ConstArrayScope,
    /// Observed definitions, their join-free partition, and dependency order.
    defs: HashMap<String, Expr>,
    join_free: HashSet<String>,
    order: Vec<String>,
    /// Evaluated observed fields — the [`PreparedBuild::fields`] payload.
    fields: HashMap<String, ArrayD<f64>>,
    /// Value-invention products.
    members: HashMap<String, Vec<i64>>,
    /// Producers whose members are tuples, so carry no member ids.
    tuple_keyed: HashSet<String>,
    extents: HashMap<String, i64>,
    /// The gated providers' keys, sorted — [`PreparedBuild::gated_provider_keys`].
    gated_keys: Vec<String>,
    /// [`PreparedBuild::rules`] / [`PreparedBuild::refused`].
    rules: Vec<PipelineRule>,
    refused: Option<PipelineRule>,
    /// The observeds evaluated ahead of the gated fetch (see
    /// [`BuildState::eval_observeds_before_fetch`]), and the progress
    /// counters the observed phase shares across its two passes.
    evaluated: HashSet<String>,
    obs_done: usize,
    obs_total: usize,
}

impl<'o> BuildState<'o> {
    fn log(&self, msg: &str) {
        vlog(self.opts, msg);
    }

    fn report(
        &self,
        phase: PreparePhase,
        done: usize,
        total: Option<usize>,
        item: &str,
    ) -> Result<(), PrepareError> {
        report_progress(self.opts, phase, done, total, item)
    }

    /// Typed load (metaparameters closed at the loader API).
    fn load_document(
        rewritten: &JsonValue,
        opts: &'o PrepareOptions,
        metaparameters: BTreeMap<String, i64>,
    ) -> Result<Self, PrepareError> {
        report_progress(opts, PreparePhase::Load, 0, None, "")?;
        let text = serde_json::to_string(rewritten)
            .map_err(|e| err(format!("serialize rewritten document: {e}")))?;
        // A loader-API binding names something in the AUTHORED document. By the
        // time this runs, the document may already have been inlined by an
        // earlier load — a §4.7 mount CONSUMES the leaf's `metaparameters`
        // (§9.7.6 site 3), so a name only the leaf declared is no longer
        // declared anywhere and the `{ref}` stub the mount walk reads is gone.
        // Re-binding it here would be refused as an unknown name, on a document
        // whose axes that very binding already folded. Keep only what this
        // document can still act on: a name it declares itself, or any name at
        // all while it still carries an unresolved mount for the walk to find.
        let metaparameters = if document_has_unresolved_mount(rewritten) {
            metaparameters
        } else {
            let declared = rewritten.get("metaparameters").and_then(|v| v.as_object());
            metaparameters
                .into_iter()
                .filter(|(k, _)| declared.is_some_and(|d| d.contains_key(k)))
                .collect()
        };
        let load_opts = LoadOptions {
            base_path: opts.base_path.clone(),
            metaparameters,
        };
        let mut file = crate::parse::load_string_with_options(&text, &load_opts)
            .map_err(|e| err(format!("load rewritten document: {e}")))?;
        // esm-spec §9.5.3: a `table_lookup` is lowered before anything
        // evaluates it, on this route as on the compile path — the reference
        // evaluator has no rule for the node itself.
        crate::lower_table_lookup::lower_table_lookups(&mut file)
            .map_err(|e| err(format!("lower table lookups: {e}")))?;
        let mut index_sets: HashMap<String, IndexSet> = file
            .index_sets
            .clone()
            .unwrap_or_default()
            .into_iter()
            .collect();
        let models = file
            .models
            .as_ref()
            .ok_or_else(|| err("document has no models"))?;
        let model_name = match &opts.model_name {
            Some(n) => n.clone(),
            None if models.len() == 1 => models.keys().next().unwrap().clone(),
            None => return Err(err("document holds several models; pass model_name")),
        };
        let mut model = models
            .get(&model_name)
            .ok_or_else(|| err(format!("model '{model_name}' not found in the document")))?
            .clone();
        // This route reads the AUTHORED model, not a flattened one, so it makes
        // the rewrites `flatten` would have made, exactly as the array
        // runtime's single-model route does: a self-qualified `M.a` resolves
        // to `a`, a right-hand-side `D` to the tendency it names (or a refusal,
        // never a `NaN`), and a `sin` of a `deg` angle takes radians — without
        // that last one `sin(40 deg)` evaluated as `sin(40)` here and nowhere
        // else. The rewrites run after the subsystems are mounted, as on the
        // array runtime, so a mounted subsystem's equations and declarations
        // are in scope under their mounted names (`D(North.u)`, a `deg`
        // `North.theta`).
        crate::simulate_array::mount_subsystems(&mut model, &mut index_sets)
            .map_err(|e| err(format!("model '{model_name}': {e}")))?;
        crate::simulate_array::resolve_model_self_references(&mut model, &model_name);
        crate::simulate_array::apply_flatten_rewrites(&mut model)
            .map_err(|e| err(format!("model '{model_name}': {e}")))?;

        let (param_vals, param_names) = scalar_params(&model, &model_name, &opts.parameters);

        Ok(BuildState {
            opts,
            model,
            model_name,
            index_sets,
            param_vals,
            param_names,
            arrays: ArrMap::default(),
            const_scope: ConstArrayScope::default(),
            defs: HashMap::new(),
            join_free: HashSet::new(),
            order: Vec::new(),
            fields: HashMap::new(),
            members: HashMap::new(),
            tuple_keyed: HashSet::new(),
            extents: HashMap::new(),
            gated_keys: Vec::new(),
            rules: Vec::new(),
            refused: None,
            evaluated: HashSet::new(),
            obs_done: 0,
            obs_total: 0,
        })
    }

    /// Record one evaluation. Under [`PrepareOptions::refuse_per_cell`] a
    /// per-cell one also stops the build, and the caller returns as soon as
    /// this answers `true`.
    fn record(&mut self, rule: PipelineRule) -> bool {
        let stop = self.opts.refuse_per_cell && matches!(rule.route, Route::PerCell(_));
        if stop {
            self.refused = Some(rule.clone());
        }
        self.rules.push(rule);
        stop
    }

    /// Provider injection: eager CONST materialization; gated deferral.
    ///
    /// The namespace is held as the interpreter's own `ArrMap` (one rehash of
    /// the caller's map, MOVING the values) so the observed-graph loop below
    /// can lend it to the evaluator without the per-observed deep clone of
    /// every provider slab that this map's SR arrays would otherwise cost.
    ///
    /// Returns the DEFERRED providers (with their gates), to be fetched
    /// pre-sliced after value-invention.
    fn inject_providers(
        &mut self,
        const_arrays: HashMap<String, ArrayD<f64>>,
        providers: Vec<(String, Box<dyn PrepareProvider>)>,
        mut discovered: HashMap<String, ArrayD<f64>>,
        pd_gates: &HashMap<String, ProviderGate>,
    ) -> Result<Vec<GatedProvider>, PrepareError> {
        self.arrays = const_arrays
            .into_iter()
            .map(|(k, a)| {
                let rounded = round_external(&k, a);
                (k, rounded)
            })
            .collect();
        let mut gated: Vec<GatedProvider> = Vec::new();
        let n_providers = providers.len();
        for (i, (k, mut prov)) in providers.into_iter().enumerate() {
            self.report(
                PreparePhase::ConstProviders,
                i,
                Some(n_providers),
                k.as_str(),
            )?;
            if let Some(a) = discovered.remove(&k) {
                // Already materialized by the extent-discovery pre-pass; never
                // sampled twice.
                self.arrays.insert(k.clone(), round_external(&k, a));
            } else if let Some(gate) = pd_gates.get(&k) {
                // Record-derived gate (the rewrite's own metadata.x_esd.pushdown):
                // defer — value-invention must derive the gating set's members
                // before the rows to fetch are known.
                gated.push((k, prov, gate.clone()));
            } else if let Some(gate) = prov.gate_spec() {
                // Provider-declared gate (the fallback protocol) — also deferred.
                gated.push((k, prov, gate));
            } else if !prov.is_const() {
                return Err(err(format!(
                    "prepare: provider '{k}' is DISCRETE (non-empty refresh times); \
                     prepare() is build-time-only"
                )));
            } else {
                let a = prov
                    .sample()
                    .map_err(|e| err(format!("materialize const provider '{k}': {}", e.0)))?;
                self.log(&format!(
                    "  [prepare] const provider {k} -> {:?}",
                    a.shape()
                ));
                self.arrays.insert(k.clone(), round_external(&k, a));
            }
        }
        self.report(
            PreparePhase::ConstProviders,
            n_providers,
            Some(n_providers),
            "",
        )?;
        self.gated_keys = gated.iter().map(|(k, _, _)| k.clone()).collect();
        self.gated_keys.sort();
        Ok(gated)
    }

    /// Provider-key name aliasing.
    ///
    /// Runs for EVERY build, not only the pushdown one. A provider key is
    /// `"<ModelPath>.<param>"` in every binding — that is the `providers`
    /// contract — while the authored expressions of the selected model spell
    /// the same parameter by its model-LOCAL name. Without the alias the
    /// observed graph looks up `hp`, finds no `hp` (only `Probe.hp`), and
    /// evaluates to NaN: a document that read its data correctly, then reported
    /// nothing of it. Gating this on `pushdown_rewrite` made the ingest work
    /// only for callers who also wanted the projection desugar, which is an
    /// unrelated question.
    ///
    /// `pd_coupling` is empty off the pushdown path, so that half is unchanged;
    /// the generic dotted-key → unique-shallowest-bare-tail pass is the one
    /// that now always runs, and it never overwrites an existing key.
    fn alias_pushdown_names(&mut self, pd_coupling: &[(String, String)]) {
        inject_aliases(&mut self.arrays, pd_coupling);
    }

    /// A shaped parameter whose data the document itself supplies (an inline
    /// array, or a scalar broadcast over its grid) is build-time factor data,
    /// bound as the compile path binds it. Runs after every provider and
    /// alias is in place, because a caller's or a provider's array of the
    /// same name stays authoritative, as it does there; a gated provider's
    /// slab, fetched later, does too.
    fn bind_inline_parameters(&mut self, gated: &[GatedProvider]) {
        for (name, a) in
            crate::simulate_array::inline_parameter_arrays(&self.model, &self.index_sets)
        {
            let gated_here = gated.iter().any(|(k, _, g)| {
                k.rsplit('.').next() == Some(name.as_str()) || g.applies_to.contains(&name)
            });
            if !gated_here && !self.arrays.contains_key(&name) {
                self.arrays.insert(name, a);
            }
        }
    }

    /// The CONST-ARRAY registry (CONFORMANCE_SPEC §5.5.5).
    ///
    /// Everything in `arrays` at THIS point is build-time factor data: the
    /// caller's `const_arrays`, the materialized const providers, and the
    /// pushdown coupling aliases. That is exactly the Julia reference's
    /// `const_arrays` registry, so a gather on one of these names is a
    /// CONST-ARRAY gather — an out-of-range index raises
    /// `E_TREEWALK_CONSTARRAY_OOB` rather than silently reading the state
    /// gather's zero ghost, which §5.5.5 says is never a const array's. The
    /// observeds this pipeline evaluates below are inserted into `arrays` as it
    /// goes and are deliberately NOT in the registry: they are observed gathers
    /// and keep the zero-ghost convention.
    fn seal_const_registry(&mut self) {
        self.const_scope = ConstArrayScope::from_names(self.arrays.keys().cloned());
    }

    /// THE FREE-VARIABLE GATE, on this route too (CONFORMANCE_SPEC §5.23).
    ///
    /// A name declared nowhere in the document is a structural error, and both
    /// `esm validate` and the compile path already say so by name. This route
    /// did not: `prepare` evaluates the whole observed graph WHOLESALE through
    /// the shared expression evaluator, upstream of the compile-path gate and
    /// on expressions the compile path never sees again (each observed leaves
    /// here as a materialized field). An undeclared operand therefore fell
    /// through `simulate_array::eval::lookup_variable` to an unbound-name NaN,
    /// and `max(known, undeclaredFloor)` evaluated as `max(known)` — IEEE-754
    /// `max` returns the non-NaN operand, so the operand simply DISAPPEARED and
    /// the answer stayed finite and plausible (finding F25; finding F24 was the
    /// same sentinel reached by a different name on this same route).
    ///
    /// It is the SAME function the compile path calls, not a second copy: two
    /// copies is precisely how one route came to reach a check and the other to
    /// skip it. It runs BEFORE the observed graph, before value invention and
    /// before the gated fetch, so a typo costs a millisecond rather than the
    /// tens of gigabytes a real ISRM fetch moves first.
    ///
    /// `extra_bound` credits the names THIS route's evaluation scope binds that
    /// the model does not declare — the caller's const arrays, the materialized
    /// provider slabs and pushdown aliases already in `arrays`, the scalar
    /// parameter vector, and every derived set's `member_factor` column (fed
    /// into `arrays` later by [`Self::feed_member_factors`], so it cannot be
    /// read off the map yet). Widening the bound set can only prevent a false
    /// positive, which is this gate's cardinal requirement.
    fn check_free_names(&self) -> Result<(), PrepareError> {
        let mut extra: Vec<String> = self.arrays.keys().cloned().collect();
        extra.extend(self.param_names.iter().cloned());
        extra.extend(
            self.index_sets
                .values()
                .filter_map(|is| is.member_factor.clone()),
        );
        crate::simulate_array::check_free_variables(&self.model, &self.index_sets, &extra)
            // The DETAILS, not the rendered error: `run_build_pipeline` re-wraps
            // a `PrepareError` in `InterpreterBuildError`, so passing the whole
            // `Display` through would print that prefix twice. The message the
            // author reads is then byte-identical to the compile path's.
            .map_err(|e| match e {
                crate::compile_error::CompileError::InterpreterBuildError { details } => {
                    err(details)
                }
                other => err(other.to_string()),
            })
    }

    /// Observed definitions + the join-free partition.
    ///
    /// Resolve every `join` clause while the ranges still carry their
    /// `{ "from": <index set> }` linkage (`eval_observed` resolves ranges on its
    /// own clone, which erases it):
    ///
    /// * an OVERLAP gate's two range symbols — without this the dense evaluator
    ///   cannot tell which loop symbol each envelope side runs over and declines
    ///   to let the gate drive: correct, but back at `O(∏ranges)`. Infallible;
    ///   an unresolvable gate simply stays undriven.
    /// * a value-equality `on` clause, into its `filter` predicate plus the
    ///   drivable [`crate::join::OnGate`] (CONFORMANCE_SPEC §5.5.8). This one is
    ///   NOT optional: `join.on` is the semantics, not an optimisation, and an
    ///   unresolved clause reached the evaluator contributing neither the
    ///   predicate nor the gate — an unfiltered full product, silently. A
    ///   genuinely unresolvable key column is a build error, as it is on the
    ///   compiled path.
    fn collect_observed_defs(&mut self) -> Result<(), PrepareError> {
        let mut defs = observed_defs(&self.model);
        let state_dependent = state_dependent_observeds(&self.model, &defs);
        {
            let var_shapes = crate::join::declared_var_shapes(&self.model);
            for e in defs.values_mut() {
                crate::join::resolve_overlap_syms_expr(e, &var_shapes);
                crate::join::resolve_expr_joins(e, &self.index_sets, &var_shapes)
                    .map_err(|e| err(format!("resolve join: {e}")))?;
            }
        }
        self.join_free = defs
            .iter()
            .filter(|(_, e)| match e {
                Expr::Operator(node) => node.join.as_ref().map(|j| j.is_empty()).unwrap_or(true),
                _ => true,
            })
            .map(|(n, _)| n.clone())
            .collect();
        self.order = observed_order(&defs)?;
        self.order
            .retain(|name| !state_dependent.contains(name.as_str()));
        self.defs = defs;
        Ok(())
    }

    /// Build-time coordinate path: producer-seeded join-free closure.
    ///
    /// (The Rust analogue of Julia's `_derive_binning_coords` general-eval
    /// seeding; tolerant — an unresolved coordinate degrades to the engine's
    /// own fail-closed contract downstream, exactly as in Python.)
    fn eval_buildtime_coordinates(&mut self) -> Result<(), PrepareError> {
        let seeds: Vec<&Expr> = self
            .model
            .equations
            .iter()
            .filter_map(|eq| match &eq.rhs {
                Expr::Operator(node)
                    if node.op.as_str() == "faq"
                        && node.distinct == Some(true)
                        && node.join.as_ref().map(|j| !j.is_empty()).unwrap_or(false) =>
                {
                    Some(&eq.rhs)
                }
                _ => None,
            })
            .collect();
        if seeds.is_empty() {
            return Ok(());
        }
        let needed = producer_seed_closure(&seeds, &self.defs, &self.join_free);
        let no_extents: HashMap<String, i64> = HashMap::new();
        let coords: Vec<String> = self
            .order
            .iter()
            .filter(|n| needed.contains(*n))
            .cloned()
            .collect();
        let n_coords = coords.len();
        for (i, name) in coords.into_iter().enumerate() {
            self.report(PreparePhase::Coordinates, i, Some(n_coords), name.as_str())?;
            match eval_observed(
                &name,
                &self.defs[&name],
                &self.arrays,
                &self.param_vals,
                &self.param_names,
                &self.index_sets,
                &no_extents,
                &self.const_scope,
                DeclaredRank::of(&self.model, &name),
                self.opts.refuse_per_cell,
            ) {
                Ok(Evaluated::Refused(why)) => {
                    self.record(PipelineRule {
                        name,
                        kind: "build-time observed",
                        route: Route::PerCell(why),
                    });
                    return Ok(());
                }
                Ok(Evaluated::Field(a, route)) => {
                    self.log(&format!(
                        "  [prepare] build-time coordinate {name} -> {:?}",
                        a.shape()
                    ));
                    if self.record(PipelineRule {
                        name: name.clone(),
                        kind: "build-time observed",
                        route,
                    }) {
                        return Ok(());
                    }
                    self.arrays.insert(name.clone(), a.clone());
                    self.fields.insert(name, a);
                }
                Err(e) => {
                    // Tolerant (mirrors Python's skip_unresolved=True): the
                    // producer then fails with ITS diagnostic if it truly
                    // needed this coordinate.
                    self.log(&format!(
                        "  [prepare] build-time coordinate {name} skipped ({})",
                        e.0
                    ));
                }
            }
        }
        self.report(PreparePhase::Coordinates, n_coords, Some(n_coords), "")?;
        Ok(())
    }

    /// VALUE INVENTION: the graph derives its own support set.
    fn invent_values(&mut self) -> Result<(), PrepareError> {
        self.report(PreparePhase::ValueInvention, 0, None, "")?;
        let vi = run_value_invention_with_params(
            &self.model,
            &self.index_sets,
            Some(&self.arrays),
            &self.opts.parameters,
        )
        .map_err(|e| err(format!("value invention: {e}")))?;
        // A member id is only needed where a set feeds one back (a
        // `member_factor`, a gated fetch), and only a scalar key is an id. A
        // tuple-keyed producer (an undirected edge `skolem(a, b)`, a composite
        // key) is sized by its extent alone; a consumer that needs its ids
        // refuses where it looks them up.
        let mut members: HashMap<String, Vec<i64>> = HashMap::new();
        let mut tuple_keyed: HashSet<String> = HashSet::new();
        for (faq, mem) in &vi.members {
            let ids: Option<Vec<i64>> = mem
                .iter()
                .map(|k| match k {
                    crate::relational::Key::Int(i) => Some(*i),
                    _ => None,
                })
                .collect();
            match ids {
                Some(ids) => {
                    members.insert(faq.clone(), ids);
                }
                None => {
                    tuple_keyed.insert(faq.clone());
                }
            }
        }
        // Recorded, not refused: the array runtime's own build runs the same
        // relational pass for every compiler, so refusing it here alone would
        // make the pipeline stricter than the route it feeds.
        let mut producers: Vec<&String> = members.keys().chain(tuple_keyed.iter()).collect();
        producers.sort();
        for faq in producers {
            self.rules.push(PipelineRule {
                name: faq.clone(),
                kind: "value invention",
                route: Route::Relational,
            });
        }
        self.members = members;
        self.tuple_keyed = tuple_keyed;
        self.extents = vi.extents.clone();
        // An arg-witness assignment and the grouped / derived buffers keyed on
        // it are build-time products: later observeds read them (`den[g]` sums
        // over `assign`), and they are fields in their own right, so the
        // observed graph does not evaluate their defining relational bodies.
        let buffers = vi
            .assignments
            .iter()
            .map(|(n, b)| (n, b.iter().map(|&i| i as f64).collect::<Vec<f64>>()))
            .chain(vi.groups.iter().map(|(n, b)| (n, b.clone())));
        for (name, values) in buffers {
            let n = values.len();
            let a = ArrayD::from_shape_vec(ndarray::IxDyn(&[n]), values)
                .map_err(|e| err(format!("value-invention buffer '{name}': {e}")))?;
            self.arrays.insert(name.clone(), a.clone());
            self.fields.insert(name.clone(), a);
        }
        Ok(())
    }

    /// Hook 1: derived-set member ids fed back as the member_factor.
    ///
    /// Collected and SORTED first, so both the observed event stream and the
    /// verbose narration are deterministic; `index_sets` is a `HashMap` and
    /// iterating it directly made the order vary run to run.
    fn feed_member_factors(&mut self) -> Result<(), PrepareError> {
        let mut tuple_fed: Vec<&String> = self
            .index_sets
            .iter()
            .filter(|(_, is)| is.kind == "derived" && is.member_factor.is_some())
            .filter(|(_, is)| {
                is.from_faq
                    .as_ref()
                    .is_some_and(|f| self.tuple_keyed.contains(f))
            })
            .map(|(sname, _)| sname)
            .collect();
        tuple_fed.sort();
        if let Some(sname) = tuple_fed.first() {
            return Err(err(format!(
                "derived index set '{sname}' names a member_factor, but its producer's members \
                 are tuples, not scalar member ids"
            )));
        }
        let mut mf_sets: Vec<(&String, &String, &Vec<i64>)> = self
            .index_sets
            .iter()
            .filter(|(_, is)| is.kind == "derived")
            .filter_map(|(sname, is)| {
                let (mf, faq) = (is.member_factor.as_ref()?, is.from_faq.as_ref()?);
                Some((sname, mf, self.members.get(faq)?))
            })
            .collect();
        mf_sets.sort_by(|a, b| a.0.cmp(b.0));
        let n_mf = mf_sets.len();
        for (i, (sname, mf, mem)) in mf_sets.into_iter().enumerate() {
            self.report(PreparePhase::MemberFactors, i, Some(n_mf), mf.as_str())?;
            let v: Vec<f64> = mem.iter().map(|&m| m as f64).collect();
            self.log(&format!(
                "  [prepare] member_factor {mf} <- |{sname}| = {}",
                v.len()
            ));
            let arr = ArrayD::from_shape_vec(IxDyn(&[v.len()]), v)
                .map_err(|e| err(format!("member factor '{mf}': {e}")))?;
            self.arrays.insert(mf.clone(), arr);
        }
        self.report(PreparePhase::MemberFactors, n_mf, Some(n_mf), "")?;
        Ok(())
    }

    /// Hook 2: gated-provider deferral → post-VI pre-sliced fetch.
    ///
    /// Every gate is PLANNED before any of it is issued, so the phase reports one
    /// running "request i of N" across all providers rather than restarting the
    /// count at each one. Planning is pure bookkeeping over the invented members
    /// — it reads nothing and costs nothing next to the fetch it describes.
    fn fetch_gated_providers(
        &mut self,
        gated: Vec<GatedProvider>,
        pd_coupling: &[(String, String)],
    ) -> Result<(), PrepareError> {
        let mut plans: Vec<GatedFetchPlan> = Vec::with_capacity(gated.len());
        for (key, prov, gate) in &gated {
            if gate.applies_to.len() != 1 {
                return Err(err(format!(
                    "gated provider '{key}': applies_to lists {} variables; bind one \
                     provider per variable (providers[\"<ModelPath>.<param>\"]) so each gated \
                     fetch is a single field",
                    gate.applies_to.len()
                )));
            }
            let mut plan =
                gated_fetch_plan(key, gate, &self.index_sets, &self.members, &self.extents)?;
            plan.n_requests = if prov.supports_selection() {
                match &plan.selection[plan.gated_pos] {
                    AxisSel::Indices(idx) => idx
                        .len()
                        .div_ceil(self.opts.gated_fetch_batch.unwrap_or(usize::MAX).max(1))
                        .max(1),
                    // Neither is a support set, so `run_gated_fetch` refuses them —
                    // but it refuses them THERE, with a message naming the provider.
                    // Counting them as one request keeps this pass a pure estimate
                    // of the progress denominator rather than a second place that
                    // decides what a gate may be.
                    AxisSel::All | AxisSel::Range { .. } => 1,
                }
            } else {
                1
            };
            plans.push(plan);
        }
        let total_requests: usize = plans.iter().map(|p| p.n_requests).sum();
        let mut issued = 0usize;

        let opts = self.opts;
        let mut report = |phase: PreparePhase, done: usize, total: Option<usize>, item: &str| {
            report_progress(opts, phase, done, total, item)
        };
        for ((key, mut prov, gate), plan) in gated.into_iter().zip(plans) {
            let arr = run_gated_fetch(
                &key,
                prov.as_mut(),
                &plan,
                opts.gated_fetch_batch,
                &mut issued,
                total_requests,
                &mut report,
            )?;
            if arr.shape().get(plan.gated_pos_out).copied() != Some(plan.gated_extent) {
                return Err(err(format!(
                    "gated provider '{key}': fetched compact axis is {:?} but the gating \
                     set extent is {}",
                    arr.shape().get(plan.gated_pos_out),
                    plan.gated_extent
                )));
            }
            // Surface the compact slab under the MODEL variable this provider feeds
            // (its local tail — the authored spelling the un-flattened expressions
            // use). From esm 1.0.0 the key IS that variable's namespaced name, so it
            // answers on the routing's `to` side; the `frm` side still answers for a
            // provider registered under the SOURCE's spelling, and the gate's tail is
            // the last resort.
            //
            // ONE registry entry per slab, and it must be THIS provider's own: the SR
            // slabs are hundreds of MB so aliasing would deep-copy, and — load-bearing
            // for CORRECTNESS, not only for memory — sibling sources may feed the SAME
            // variable name. `isrm.esm` fetches one zarr array at three emission
            // layers through three providers that differ in nothing but the
            // `{"fixed": [layer]}` axis of their `gated_select`. Publishing a slab
            // under every key with the same dotted TAIL (as the Julia and Python hooks
            // once did) makes all three providers claim all three keys, so whichever
            // is written last silently wins for all of them and every layer is
            // contracted against one arbitrary sibling's slab. Both arms below match
            // a key EXACTLY; never a name-tail expansion.
            let target = gated_target(&key, &gate, pd_coupling);
            self.log(&format!(
                "  [prepare] gated fetch {key} -> {target} {:?}",
                arr.shape()
            ));
            self.arrays
                .insert(target.clone(), round_external(&target, arr));
        }
        self.report(
            PreparePhase::GatedFetch,
            total_requests,
            Some(total_requests),
            "",
        )?;
        Ok(())
    }

    /// Evaluate, ahead of the gated fetch, every observed that does not read a
    /// gated slab, directly or through another observed.
    ///
    /// Their inputs are all in place by now, so this changes no value — only
    /// WHEN they are evaluated. It is what lets a strict compiler refuse a
    /// document whose refusal has nothing to do with the gated data before
    /// the fetch moves it (a real source-receptor fetch is gigabytes). The
    /// observeds that do read a gated slab follow the fetch, as before.
    fn eval_observeds_before_fetch(
        &mut self,
        gated: &[GatedProvider],
        pd_coupling: &[(String, String)],
    ) -> Result<(), PrepareError> {
        // A gate that does not name exactly one variable is refused by the
        // fetch itself, with its own message; nothing is moved ahead of that.
        if gated.is_empty() || gated.iter().any(|(_, _, g)| g.applies_to.len() != 1) {
            return Ok(());
        }
        let mut after: HashSet<String> = gated
            .iter()
            .map(|(key, _, gate)| gated_target(key, gate, pd_coupling))
            .collect();
        let seeds = after.clone();
        // `order` is dependency order, so one pass closes the set.
        for name in &self.order {
            let mut reads = HashSet::new();
            declared_args(&self.defs[name], &mut reads);
            join_names(&self.defs[name], &mut reads);
            if reads.iter().any(|r| after.contains(r)) {
                after.insert(name.clone());
            }
        }
        for s in &seeds {
            after.remove(s);
        }
        self.eval_observeds(&after, false)
    }

    /// Evaluate the whole observed graph in dependency order.
    ///
    /// Every observed that still needs evaluating, counted up front: `order`
    /// includes the coordinate pre-pass's products, and a `total` that a `continue`
    /// silently makes unreachable is a bar that never fills.
    fn eval_observed_graph(&mut self) -> Result<(), PrepareError> {
        self.eval_observeds(&HashSet::new(), true)
    }

    /// One pass over the pending observeds, skipping `skip` and what an
    /// earlier pass evaluated. The last pass (`last`) moves every evaluated
    /// observed into `fields`; so does any pass that ends in a refusal.
    fn eval_observeds(&mut self, skip: &HashSet<String>, last: bool) -> Result<(), PrepareError> {
        let pending: Vec<String> = self
            .order
            .iter()
            .filter(|n| !self.fields.contains_key(*n))
            .cloned()
            .collect();
        if self.obs_done == 0 {
            self.obs_total = pending.len();
        }
        let n_pending = self.obs_total;
        for name in pending.iter() {
            if skip.contains(name) || self.evaluated.contains(name) {
                continue;
            }
            self.report(
                PreparePhase::Observeds,
                self.obs_done,
                Some(n_pending),
                name.as_str(),
            )?;
            self.obs_done += 1;
            self.evaluated.insert(name.clone());
            let t = std::time::Instant::now();
            let (a, route) = match eval_observed(
                name,
                &self.defs[name],
                &self.arrays,
                &self.param_vals,
                &self.param_names,
                &self.index_sets,
                &self.extents,
                &self.const_scope,
                DeclaredRank::of(&self.model, name),
                self.opts.refuse_per_cell,
            )? {
                Evaluated::Field(a, route) => (a, route),
                Evaluated::Refused(why) => {
                    self.record(PipelineRule {
                        name: name.clone(),
                        kind: "build-time observed",
                        route: Route::PerCell(why),
                    });
                    break;
                }
            };
            self.log(&format!(
                "  [prepare] {name:<24} shape={:?}  {:>7.1} s",
                a.shape(),
                t.elapsed().as_secs_f64()
            ));
            if self.record(PipelineRule {
                name: name.clone(),
                kind: "build-time observed",
                route,
            }) {
                break;
            }
            // Into the evaluation namespace ONLY while later observeds may still
            // read it; each is MOVED into `fields` after the loop rather than held
            // in both maps at once.
            self.arrays.insert(name.clone(), a);
        }
        if !last && self.refused.is_none() {
            return Ok(());
        }
        for name in &pending {
            if let Some(a) = self.arrays.remove(name) {
                self.fields.insert(name.clone(), a);
            }
        }
        if self.refused.is_some() {
            return Ok(());
        }
        self.report(PreparePhase::Observeds, n_pending, Some(n_pending), "")?;
        Ok(())
    }

    /// Assemble the [`PreparedBuild`] artifact from the accumulated state.
    fn finish(self, doc: JsonValue) -> PreparedBuild {
        PreparedBuild {
            doc,
            model_name: self.model_name,
            fields: self.fields,
            members: self.members,
            extents: self.extents,
            gated_provider_keys: self.gated_keys,
            rules: self.rules,
            refused: self.refused,
        }
    }
}

// --------------------------------------------------------------------------- //
// The adapter EsmProblem construction goes through.
// --------------------------------------------------------------------------- //

/// Run the build pipeline for [`crate::problem::esm_problem`], translating
/// [`crate::problem::ProblemOptions`] into this module's internal option
/// struct and its error into [`SimulateError`].
///
/// Consumes `opts`'s build providers and const arrays: they are moved into the
/// pipeline, which is the one place they are used.
pub(crate) fn run_build_pipeline(
    doc: &JsonValue,
    opts: &mut crate::problem::ProblemOptions,
) -> Result<PreparedBuild, crate::simulate::SimulateError> {
    let internal = PrepareOptions {
        model_name: opts.model_name.clone(),
        metaparameters: opts.metaparameters.clone(),
        base_path: opts.base_path.clone(),
        pushdown_rewrite: opts.pushdown_rewrite,
        parameters: opts.p.clone(),
        verbose: opts.verbose,
        progress: opts.progress.clone(),
        gated_fetch_batch: opts.gated_fetch_batch,
        refuse_per_cell: opts.compiler.unwrap_or_default().is_strict(),
    };
    let const_arrays = std::mem::take(&mut opts.const_arrays);
    let providers = std::mem::take(&mut opts.build_providers);
    run_prepare(doc, const_arrays, providers, &internal).map_err(|e| {
        // A cancel keeps its own identity across the boundary: `is_cancelled`
        // must still answer "the caller stopped this" rather than "the build
        // failed", which is the whole point of the marker.
        if e.is_cancelled() {
            crate::simulate::SimulateError::Cancelled { details: e.0 }
        } else {
            crate::simulate::SimulateError::Compile(
                crate::compile_error::CompileError::InterpreterBuildError { details: e.0 },
            )
        }
    })
}

/// Whether `doc` still carries an unresolved §4.7 mount — a `models.<k>` /
/// `reaction_systems.<k>` `{ref}` or a `subsystems.<k>` `{ref}`.
///
/// Distinguishes an AUTHORED document, whose mounted leaves still have their
/// `metaparameters` to be closed at the edge, from one an earlier load already
/// inlined, where those declarations have been consumed and the loader-API
/// names they served have nothing left to bind.
fn document_has_unresolved_mount(doc: &JsonValue) -> bool {
    let Some(obj) = doc.as_object() else {
        return false;
    };
    for kind in ["models", "reaction_systems"] {
        let Some(comps) = obj.get(kind).and_then(|v| v.as_object()) else {
            continue;
        };
        for comp in comps.values() {
            if comp.get("ref").is_some() {
                return true;
            }
            if let Some(subs) = comp.get("subsystems").and_then(|v| v.as_object())
                && subs.values().any(|s| s.get("ref").is_some())
            {
                return true;
            }
        }
    }
    false
}

#[cfg(test)]
mod first_cell_stop_tests {
    //! A strict compiler never walks a pipeline observed past its first cell:
    //! the walk stops there and the observed is lowered on the tape instead —
    //! counted with the evaluator's per-cell test hook, never timed.

    use crate::compile_error::CompileError;
    use crate::problem::{Compiler, ProblemOptions, esm_problem};
    use crate::simulate::SimulateError;
    use crate::simulate_array::per_cell_cells;
    use serde_json::{Value, json};

    fn doc(n: usize, variables: Value, equations: Value) -> Value {
        json!({
            "esm": "1.1.0",
            "metadata": {"name": "FirstCellStop"},
            "index_sets": {"rows": {"kind": "interval", "size": n}},
            "models": {"Rel": {"variables": variables, "equations": equations}},
        })
    }

    /// `left[i] = 10 i` and its running sum `cum[i] = Σ_{j ≤ i} left[j]`,
    /// which the reference evaluator sweeps as a prefix scan.
    fn cumulative(n: usize) -> Value {
        doc(
            n,
            json!({
                "left": {"type": "unknown", "units": "1", "shape": ["rows"]},
                "cum": {"type": "unknown", "units": "1", "shape": ["rows"]},
            }),
            json!([
                {"lhs": "left",
                 "rhs": {"op": "faq", "args": [], "output_idx": ["i"],
                         "ranges": {"i": {"from": "rows"}},
                         "expr": {"op": "*", "args": [10, "i"]}}},
                {"lhs": "cum",
                 "rhs": {"op": "faq", "args": [], "output_idx": ["i"],
                         "ranges": {"i": {"from": "rows"}, "j": {"from": "rows"}},
                         "filter": {"op": "<=", "args": ["j", "i"]},
                         "expr": {"op": "index", "args": ["left", "j"]}}},
            ]),
        )
    }

    /// `s[1] = 1`, `s[k] = s[k-1] + 1`: a causal self-reference, swept cell by
    /// cell.
    fn recurrence(n: usize) -> Value {
        doc(
            n,
            json!({"s": {"type": "unknown", "units": "1", "shape": ["rows"]}}),
            json!([
                {"lhs": "s",
                 "rhs": {"op": "faq", "args": [], "output_idx": ["k"],
                         "ranges": {"k": {"from": "rows"}},
                         "expr": {"op": "ifelse", "args": [
                             {"op": "<=", "args": ["k", 1]},
                             1.0,
                             {"op": "+", "args": [
                                 {"op": "index", "args": ["s", {"op": "-", "args": ["k", 1]}]},
                                 1.0]}]}}},
            ]),
        )
    }

    fn build(d: &Value, compiler: Compiler) -> Result<crate::problem::EsmProblem, SimulateError> {
        esm_problem(
            d,
            (0.0, 1.0),
            ProblemOptions {
                compiler: Some(compiler),
                build_pipeline: true,
                ..Default::default()
            },
        )
    }

    const LARGE: usize = 200_000;

    /// `native`'s build of `d` and the cells the per-cell walks evaluated on
    /// the way to it.
    fn native_cells(d: &Value) -> (crate::problem::EsmProblem, u64) {
        let before = per_cell_cells();
        let prob = build(d, Compiler::Native).expect("native builds it on the tape");
        (prob, per_cell_cells() - before)
    }

    fn bits(p: &crate::problem::EsmProblem, name: &str) -> Vec<u64> {
        crate::problem::observed_field(p, name)
            .expect("field")
            .iter()
            .map(|x| x.to_bits())
            .collect()
    }

    #[test]
    fn a_large_prefix_scan_is_taped_after_at_most_one_walked_cell() {
        let (native, cells) = native_cells(&cumulative(LARGE));
        assert!(
            cells <= 1,
            "the walk must stop at its first cell, walked {cells}"
        );

        // The reference evaluator still walks every cell, to the same answer.
        let before = per_cell_cells();
        let prob = build(&cumulative(LARGE), Compiler::Interpreter).expect("interpreter");
        assert_eq!(per_cell_cells() - before, LARGE as u64);
        let cum = crate::problem::observed_field(&prob, "Rel.cum").expect("cum");
        let n = LARGE as f64;
        assert_eq!(cum.iter().last().copied(), Some(5.0 * n * (n + 1.0)));
        assert_eq!(bits(&native, "Rel.cum"), bits(&prob, "Rel.cum"));
    }

    #[test]
    fn a_large_recurrence_is_taped_without_a_walked_cell() {
        let (native, cells) = native_cells(&recurrence(LARGE));
        assert_eq!(cells, 0, "a recurrence goes straight to the tape's sweep");

        let before = per_cell_cells();
        let prob = build(&recurrence(LARGE), Compiler::Interpreter).expect("interpreter");
        assert_eq!(per_cell_cells() - before, LARGE as u64);
        let s = crate::problem::observed_field(&prob, "Rel.s").expect("s");
        assert_eq!(s.iter().last().copied(), Some(LARGE as f64));
        assert_eq!(bits(&native, "Rel.s"), bits(&prob, "Rel.s"));
    }

    /// A record-to-cell binning producer whose record coordinate `X` is a
    /// running sum, so the build-time coordinate pass ahead of value invention
    /// must walk it per cell. Cells are the unit squares `[k-1, k] × [0, 1]`.
    fn binning(
        n: usize,
    ) -> (
        Value,
        std::collections::HashMap<String, ndarray::ArrayD<f64>>,
    ) {
        let d = json!({
            "esm": "1.1.0",
            "metadata": {"name": "FirstCellStopCoordinates"},
            "index_sets": {
                "records": {"kind": "interval", "size": n},
                "cells": {"kind": "interval", "size": n},
                "occupied": {"kind": "derived", "from_faq": "bins"}
            },
            "models": {"Rel": {
                "variables": {
                    "step": {"type": "parameter", "units": "1", "shape": ["records"]},
                    "Y": {"type": "parameter", "units": "1", "shape": ["records"]},
                    "W": {"type": "parameter", "units": "1", "shape": ["cells"]},
                    "S": {"type": "parameter", "units": "1", "shape": ["cells"]},
                    "E": {"type": "parameter", "units": "1", "shape": ["cells"]},
                    "N": {"type": "parameter", "units": "1", "shape": ["cells"]},
                    "X": {"type": "unknown", "units": "1", "shape": ["records"]},
                    "present": {"type": "unknown", "shape": ["occupied"]}
                },
                "equations": [
                    {"lhs": "X",
                     "rhs": {"op": "faq", "args": [], "output_idx": ["r"],
                             "ranges": {"r": {"from": "records"}, "q": {"from": "records"}},
                             "filter": {"op": "<=", "args": ["q", "r"]},
                             "expr": {"op": "index", "args": ["step", "q"]}}},
                    {"lhs": {"op": "index", "args": ["present", "e"]},
                     "rhs": {"op": "faq", "id": "bins",
                             "args": ["X", "Y", "W", "S", "E", "N"],
                             "semiring": "bool_and_or", "distinct": true,
                             "output_idx": ["e"],
                             "ranges": {"r": {"from": "records"}, "c": {"from": "cells"}},
                             "join": [{"overlap": {"src_env": ["X", "Y"],
                                                   "tgt_env": ["W", "S", "E", "N"],
                                                   "eps": 0.0}}],
                             "key": {"op": "skolem", "label": "cell", "args": ["c"]},
                             "expr": {"op": "true", "args": []}}}
                ]
            }}
        });
        let col = |f: &dyn Fn(usize) -> f64| {
            ndarray::ArrayD::from_shape_vec(ndarray::IxDyn(&[n]), (0..n).map(f).collect())
                .expect("shape")
        };
        let arrays = [
            // X[r] = r - 1/2, the centre of cell r.
            ("step", col(&|k| if k == 0 { 0.5 } else { 1.0 })),
            ("Y", col(&|_| 0.5)),
            ("W", col(&|k| k as f64)),
            ("S", col(&|_| 0.0)),
            ("E", col(&|k| k as f64 + 1.0)),
            ("N", col(&|_| 1.0)),
        ]
        .into_iter()
        .map(|(k, v)| (k.to_string(), v))
        .collect();
        (d, arrays)
    }

    #[test]
    fn a_large_per_cell_coordinate_is_taped_after_at_most_one_walked_cell() {
        let run = |n: usize, compiler: Compiler| {
            let (d, const_arrays) = binning(n);
            let before = per_cell_cells();
            let r = esm_problem(
                &d,
                (0.0, 1.0),
                ProblemOptions {
                    compiler: Some(compiler),
                    build_pipeline: true,
                    const_arrays,
                    ..Default::default()
                },
            );
            (r, per_cell_cells() - before)
        };
        let (native, cells) = run(LARGE, Compiler::Native);
        let native = native.expect("native builds it on the tape");
        assert!(
            cells <= 1,
            "the coordinate pass must stop at its first cell, walked {cells}"
        );
        assert_eq!(native.members()["bins"].len(), LARGE, "one record per cell");

        let (interp, cells) = run(1000, Compiler::Interpreter);
        let prob = interp.expect("the interpreter builds it");
        assert_eq!(cells, 1000);
        assert_eq!(prob.members()["bins"].len(), 1000, "one record per cell");
        let (native_small, _) = run(1000, Compiler::Native);
        let native_small = native_small.expect("native");
        assert_eq!(native_small.members()["bins"], prob.members()["bins"]);
    }

    /// A document error in the first cell is still the document's error, not
    /// a refusal: the first cell is evaluated for exactly this. Here the
    /// running sum gathers a supplied const array out of range.
    #[test]
    fn an_error_in_the_first_cell_is_not_a_refusal() {
        let d = doc(
            3,
            json!({
                "table": {"type": "parameter", "units": "1", "shape": ["rows"]},
                "cum": {"type": "unknown", "units": "1", "shape": ["rows"]},
            }),
            json!([
                {"lhs": "cum",
                 "rhs": {"op": "faq", "args": [], "output_idx": ["i"],
                         "ranges": {"i": {"from": "rows"}, "j": {"from": "rows"}},
                         "filter": {"op": "<=", "args": ["j", "i"]},
                         "expr": {"op": "index", "args": ["table", {"op": "+", "args": ["j", 10]}]}}},
            ]),
        );
        for compiler in [Compiler::Native, Compiler::Interpreter] {
            let table = ndarray::ArrayD::from_shape_vec(ndarray::IxDyn(&[3]), vec![1.0, 2.0, 3.0])
                .expect("shape");
            let err = esm_problem(
                &d,
                (0.0, 1.0),
                ProblemOptions {
                    compiler: Some(compiler),
                    build_pipeline: true,
                    const_arrays: [("table".to_string(), table)].into_iter().collect(),
                    ..Default::default()
                },
            )
            .expect_err("an out-of-range gather");
            assert!(
                !matches!(
                    err,
                    SimulateError::Compile(CompileError::CompilerRefusedRule { .. })
                ),
                "[{compiler}] {err}"
            );
            assert!(err.to_string().contains("OOB"), "[{compiler}] {err}");
        }
    }

    /// A document error that first appears in a LATER cell, past where the
    /// walk stopped, is still reported: native evaluates every cell on the
    /// tape, which latches the same error the interpreter raises. Here the
    /// gather is out of range only at the last output cell.
    #[test]
    fn an_error_in_a_later_cell_is_reported_by_the_tape() {
        let d = doc(
            3,
            json!({
                "table": {"type": "parameter", "units": "1", "shape": ["rows"]},
                "cum": {"type": "unknown", "units": "1", "shape": ["rows"]},
            }),
            json!([
                {"lhs": "cum",
                 "rhs": {"op": "faq", "args": [], "output_idx": ["i"],
                         "ranges": {"i": {"from": "rows"}, "j": {"from": "rows"}},
                         "filter": {"op": "<=", "args": ["j", "i"]},
                         "expr": {"op": "index", "args": ["table", {"op": "+", "args": ["i", 1]}]}}},
            ]),
        );
        let run = |compiler| {
            let table = ndarray::ArrayD::from_shape_vec(ndarray::IxDyn(&[3]), vec![1.0, 2.0, 3.0])
                .expect("shape");
            esm_problem(
                &d,
                (0.0, 1.0),
                ProblemOptions {
                    compiler: Some(compiler),
                    build_pipeline: true,
                    const_arrays: [("table".to_string(), table)].into_iter().collect(),
                    ..Default::default()
                },
            )
            .expect_err("the last cell gathers out of range")
        };
        let native = run(Compiler::Native);
        let interp = run(Compiler::Interpreter);
        assert!(interp.to_string().contains("OOB"), "{interp}");
        assert_eq!(native.to_string(), interp.to_string());
    }
}

#[cfg(test)]
mod mounted_subsystem_tests {
    //! The build pipeline makes `flatten`'s rewrites on the authored model, and
    //! a mounted subsystem's equations and declarations must be in scope for
    //! them, as they are on the array runtime.
    use crate::problem::{Compiler, ProblemOptions, esm_problem, observed_field, solve};
    use serde_json::{Value, json};

    /// `P` holds a state `c` with `D(c) = -c` and the observed `name = rhs`;
    /// its subsystem `North` holds a state `u` with `D(u) = k`, an angle
    /// `theta = 40 deg` and the state-dependent observed `v = 2*u`.
    fn doc(name: &str, units: &str, rhs: Value) -> Value {
        json!({
            "esm": "1.1.0",
            "metadata": {"name": "Mounted"},
            "models": {"P": {
                "variables": {
                    "c": {"type": "unknown", "units": "1", "default": 1.0},
                    name: {"type": "unknown", "units": units}
                },
                "equations": [
                    {"lhs": {"op": "D", "args": ["c"], "wrt": "t"},
                     "rhs": {"op": "-", "args": ["c"]}},
                    {"lhs": name, "rhs": rhs}
                ],
                "subsystems": {"North": {
                    "variables": {
                        "u": {"type": "unknown", "units": "1", "default": 2.0},
                        "k": {"type": "parameter", "units": "1/s", "default": 0.5},
                        "theta": {"type": "parameter", "units": "deg", "default": 40.0},
                        "v": {"type": "unknown", "units": "1"}
                    },
                    "equations": [
                        {"lhs": {"op": "D", "args": ["u"], "wrt": "t"},
                         "rhs": "k"},
                        {"lhs": "v", "rhs": {"op": "*", "args": [2.0, "u"]}}
                    ]
                }}
            }}
        })
    }

    /// The document through the build pipeline (which takes the raw JSON), or
    /// straight to the array runtime (the typed document).
    fn built(d: &Value, compiler: Compiler, build_pipeline: bool) -> crate::problem::EsmProblem {
        let opts = ProblemOptions {
            compiler: Some(compiler),
            build_pipeline,
            ..Default::default()
        };
        let prob = if build_pipeline {
            esm_problem(d, (0.0, 1.0), opts)
        } else {
            let file = crate::load_string(&d.to_string()).expect("loads");
            esm_problem(&file, (0.0, 1.0), opts)
        };
        prob.unwrap_or_else(|e| panic!("[{compiler}, pipeline {build_pipeline}] {e}"))
    }

    /// `name` at `t = 0`, from a solve that reports it.
    fn at_t0(prob: &crate::problem::EsmProblem, name: &str) -> f64 {
        let sol = solve(
            prob,
            &crate::SolveOptions {
                saveat: Some(vec![0.0]),
                output_observed: vec![name.to_string()],
                ..Default::default()
            },
        )
        .unwrap_or_else(|e| panic!("{e}"));
        sol.get(name)
            .unwrap_or_else(|| panic!("no {name} row: {:?}", sol.state_variable_names))[0]
    }

    #[test]
    fn a_degree_angle_in_a_subsystem_takes_radians() {
        let d = doc("y", "1", json!({"op": "sin", "args": ["North.theta"]}));
        let want = 40.0f64.to_radians().sin();
        for compiler in [Compiler::Native, Compiler::Interpreter] {
            for pipeline in [false, true] {
                let prob = built(&d, compiler, pipeline);
                assert_eq!(at_t0(&prob, "y"), want, "[{compiler}, pipeline {pipeline}]");
            }
            // What the pipeline materialized at build time.
            let prob = built(&d, compiler, true);
            let y = observed_field(&prob, "y").expect("y");
            assert_eq!(
                y.iter().copied().collect::<Vec<_>>(),
                [want],
                "[{compiler}]"
            );
        }
    }

    /// `D(North.u)` is the subsystem's tendency, `North.k`: state-free, so the
    /// pipeline materializes it at build time.
    #[test]
    fn a_time_derivative_of_a_subsystem_state_resolves() {
        let d = doc(
            "z",
            "1/s",
            json!({"op": "D", "args": ["North.u"], "wrt": "t"}),
        );
        for compiler in [Compiler::Native, Compiler::Interpreter] {
            for pipeline in [false, true] {
                let prob = built(&d, compiler, pipeline);
                assert_eq!(at_t0(&prob, "z"), 0.5, "[{compiler}, pipeline {pipeline}]");
            }
            let prob = built(&d, compiler, true);
            let z = observed_field(&prob, "z").expect("z");
            assert_eq!(z.iter().copied().collect::<Vec<_>>(), [0.5], "[{compiler}]");
        }
    }

    /// A mounted subsystem's observed that reads its state, `North.v = 2*North.u`,
    /// has no value at build time. The pipeline leaves it to the run, where it
    /// is evaluated like any other observed, rather than failing the build on
    /// the state it cannot read.
    #[test]
    fn a_mounted_state_dependent_observed_is_evaluated_at_run_time() {
        let d = doc("y", "1", json!({"op": "sin", "args": ["North.theta"]}));
        for compiler in [Compiler::Native, Compiler::Interpreter] {
            for pipeline in [false, true] {
                let prob = built(&d, compiler, pipeline);
                let sol = solve(
                    &prob,
                    &crate::SolveOptions {
                        saveat: Some(vec![0.0]),
                        output_observed: vec!["North.v".to_string()],
                        ..Default::default()
                    },
                )
                .unwrap_or_else(|e| panic!("[{compiler}, pipeline {pipeline}] {e}"));
                let v = sol
                    .get("P.North.v")
                    .unwrap_or_else(|| panic!("no North.v row: {:?}", sol.state_variable_names));
                assert_eq!(v[0], 4.0, "[{compiler}, pipeline {pipeline}]");
            }
        }
    }

    /// The same for a top-level observed that reads a state, `w = 2*c`.
    #[test]
    fn a_top_level_state_dependent_observed_is_evaluated_at_run_time() {
        let d = doc("w", "1", json!({"op": "*", "args": [2.0, "c"]}));
        for compiler in [Compiler::Native, Compiler::Interpreter] {
            for pipeline in [false, true] {
                let prob = built(&d, compiler, pipeline);
                assert_eq!(at_t0(&prob, "w"), 2.0, "[{compiler}, pipeline {pipeline}]");
            }
        }
    }
}