polydat-core 0.6.1

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

//! [`SubcontextBuilder<P>`] — accumulator for module matter.
//!
//! The builder holds a [`ParentView`] of the scope it builds under,
//! records imports / exports / body fragments / pull consumers, and
//! at `finalize` validates the import contract against the parent's
//! exports + compiles the body via the ordinary `compile_polydat` /
//! `compile_ast` pipeline (subcontext_construction.md §2, §3). The
//! result is a closed [`ScopeModule<Child<P>>`] artifact.

use std::marker::PhantomData;
use std::path::PathBuf;
use std::sync::Arc;

use crate::ast::PortType;
use crate::dsl::ast::{Arg, CallExpr, Expr, ExternPort, PolydatFile, Statement};
use crate::dsl::compile::{CompileOptions as DslOptions, compile_ast_interpreter_with_options};
use crate::dsl::lexer::{Span, lex};
use crate::dsl::parser::parse;
use crate::kernel::PolydatKernel;

use super::error::{ContractViolation, SourceContext};
use super::kernel::{Child, SharedCellInScope};
use super::module::{BodyFragment, ScopeContract, ScopeModule, WriteThroughBinding};
use super::pull::{PullConsumer, RegisteredPullConsumer};
use super::spec::{ExportSpec, ImportSpec};

/// Prefix applied to the synthetic write-through output produced
/// by the Rule 2 rewrite. The child program emits this output as
/// a normal local computation; spawn pulls it per cycle and
/// fans the value through the parent's `SharedCell`.
const WRITE_THROUGH_PREFIX: &str = "__write_";

/// Optional compile-time configuration passed through to
/// [`compile_ast_interpreter_with_options`](crate::dsl::compile::compile_ast_interpreter_with_options) when finalize compiles the body. When
/// every field is at its default, finalize compiles the statements
/// under the default DSL options (subcontext_construction.md §2.3).
#[derive(Clone, Debug, Default)]
pub struct CompileOptions {
    /// The directory relative data-file paths resolve against.
    pub workload_dir: Option<PathBuf>,
    /// Library search paths.
    pub polydat_lib_paths: Vec<PathBuf>,
    /// Whether to enforce strict validation.
    pub strict: bool,
    /// The outputs to keep; every output when empty.
    pub required_outputs: Vec<String>,
    /// The diagnostic context label, if any.
    pub context_label: Option<String>,
    /// A limit on every cursor's extent, if any.
    pub cursor_limit: Option<u64>,
    /// Session-wide optimization level for op-template synthesis.
    /// `Release` (the default) injects a magic extern (`body` /
    /// `count` / `ok`) only when the result source references it;
    /// `Diagnostic` allocates all three so step-debug / cycle-replay
    /// sees every write (subcontext_construction.md §3.2). See
    /// [`KernelOptLevel`](crate::kernel::KernelOptLevel).
    pub kernel_opt: crate::kernel::KernelOptLevel,
    /// What the compiler does with an input whose type it inferred
    /// (input_variance.md §4); the same setting as
    /// [`crate::dsl::compile::CompileOptions::input_variance`].
    pub input_variance: crate::dsl::compile::InputVariance,
}

impl CompileOptions {
    fn is_default(&self) -> bool {
        self.workload_dir.is_none()
            && self.polydat_lib_paths.is_empty()
            && !self.strict
            && self.required_outputs.is_empty()
            && self.context_label.is_none()
            && self.cursor_limit.is_none()
            && self.kernel_opt == crate::kernel::KernelOptLevel::default()
            && self.input_variance == crate::dsl::compile::InputVariance::default()
    }
}

/// Everything a builder reads of the scope it builds under: the names
/// the parent declares, the modifier on each output, the ledger the
/// child's compile is charged to, and the cells in scope there — each
/// a live handle rather than a copy.
///
/// That is the whole parent surface `finalize` touches. The view holds
/// the answers rather than a kernel or a `PolydatProgram`, which a
/// compiled kernel does not keep, so a parent of any engine can give
/// them.
#[derive(Clone)]
pub struct ParentView {
    output_names: Vec<String>,
    input_names: Vec<String>,
    output_modifiers: std::collections::HashMap<String, crate::dsl::ast::BindingModifier>,
    ledger: Arc<crate::kernel::CompileLedger>,
    resources: crate::resource::ResourceScope,
    shared_cells: Vec<SharedCellInScope>,
}

impl ParentView {
    /// The view a kernel of any engine presents to a subcontext built
    /// under it.
    pub fn of_kernel(parent: &dyn crate::kernel::Kernel) -> Self {
        let output_names: Vec<String> = parent.output_names();
        let output_modifiers = output_names
            .iter()
            .map(|n| (n.clone(), parent.output_modifier(n)))
            .collect();
        Self {
            output_names,
            input_names: parent.input_names(),
            output_modifiers,
            ledger: crate::kernel::Kernel::ledger(parent).clone(),
            resources: crate::kernel::Kernel::resources(parent).clone(),
            shared_cells: Self::cells_of(parent.cells_in_scope()),
        }
    }

    /// [`Self::of_kernel`] for the interpreter's kernel type, which
    /// most callers hold.
    pub fn of(parent: &PolydatKernel) -> Self {
        Self::of_kernel(parent)
    }

    /// The names the parent declares as outputs.
    pub fn output_names(&self) -> &[String] {
        &self.output_names
    }

    /// The names the parent declares as inputs, coordinates included.
    pub fn input_names(&self) -> &[String] {
        &self.input_names
    }

    /// The modifier on a named output; `NONE` for a name the parent
    /// does not declare.
    pub fn output_modifier(&self, name: &str) -> crate::dsl::ast::BindingModifier {
        self.output_modifiers
            .get(name)
            .copied()
            .unwrap_or(crate::dsl::ast::BindingModifier::NONE)
    }

    /// The ledger the child's compile is charged to: a subscope is a
    /// program of the parent's tree.
    pub fn ledger(&self) -> &Arc<crate::kernel::CompileLedger> {
        &self.ledger
    }

    /// The resource scope the child's nodes reach: a subscope is a
    /// program of the parent's tree and sees the parent's accessor.
    pub fn resources(&self) -> &crate::resource::ResourceScope {
        &self.resources
    }

    fn cells_of(entries: Vec<crate::kernel::SharedCellEntry>) -> Vec<SharedCellInScope> {
        entries
            .into_iter()
            .map(|e| SharedCellInScope {
                name: e.name,
                port_type: e.port_type,
                cell: e.cell,
            })
            .collect()
    }

    /// The shared cells visible at the parent, its own and every
    /// ancestor's that reached it.
    pub fn shared_cells(&self) -> &[SharedCellInScope] {
        &self.shared_cells
    }
}

/// Module-matter accumulator. Construction is gated by a parent —
/// [`ScopeKernel::subcontext_builder`](super::ScopeKernel::subcontext_builder)
/// on the typed path, [`PolydatMatter::build_under`](super::PolydatMatter::build_under) on the untyped
/// one — and both hand it the same [`ParentView`].
pub struct SubcontextBuilder<P> {
    parent: ParentView,
    imports: Vec<ImportSpec>,
    exports: Vec<ExportSpec>,
    body: Vec<BodyFragment>,
    consumers: Vec<RegisteredPullConsumer>,
    context: SourceContext,
    /// Names to apply via `mark_inherited_outputs` on the
    /// compiled kernel before its program Arc is shared. Set by
    /// `PolydatMatter::build_under` from the matter's `inherited_outputs`;
    /// explicit synthesisers that don't need cascade pass-through
    /// leave this empty.
    inherited_outputs: Vec<String>,
    /// Compile-time options forwarded into the AST compile. Empty
    /// for callers that don't need libs / strict / required-output
    /// filtering.
    compile_options: CompileOptions,
    /// The externs this builder synthesized (result and write-through
    /// externs), whose types it chose rather than the author: open to
    /// `input_variance` (input_variance.md §3).
    synthesized_externs: Vec<String>,
    _parent_marker: PhantomData<fn() -> P>,
}

impl SubcontextBuilder<super::kernel::RootMarker> {
    /// A builder for a child of `parent`, a kernel of any engine: the
    /// source form of binding a scope (native_scope_trees.md §5). The
    /// module it finalizes compiles once per engine
    /// ([`ScopeModule::program_on`](super::ScopeModule::program_on)) and instantiates under a parent of
    /// any engine ([`ScopeModule::instantiate_under`](super::ScopeModule::instantiate_under)).
    pub fn under(parent: &dyn crate::kernel::Kernel) -> Self {
        Self::new(ParentView::of_kernel(parent))
    }
}

impl<P> SubcontextBuilder<P> {
    pub(crate) fn new(parent: ParentView) -> Self {
        Self {
            parent,
            _parent_marker: PhantomData,
            imports: Vec::new(),
            exports: Vec::new(),
            body: Vec::new(),
            consumers: Vec::new(),
            context: SourceContext::default(),
            inherited_outputs: Vec::new(),
            compile_options: CompileOptions::default(),
            synthesized_externs: Vec::new(),
        }
    }

    /// Set the compile options finalize maps into the
    /// [`compile_ast_interpreter_with_options`](crate::dsl::compile::compile_ast_interpreter_with_options) knobs (lib paths, strict mode,
    /// required-output filter, workload dir, context label)
    /// (subcontext_construction.md §2.3). A builder left at the
    /// default compiles under the default DSL options.
    pub fn with_compile_options(&mut self, options: CompileOptions) -> &mut Self {
        self.compile_options = options;
        self
    }

    /// Declare names whose outputs
    /// the body emits purely to cascade values from an outer
    /// scope to descendants (so they don't double up the parent's
    /// iter-coord, etc.). The compiled kernel will have these
    /// names flagged via `mark_inherited_outputs` before its
    /// program Arc is shared.
    ///
    /// Set from `PolydatMatter`'s `inherited_outputs` by
    /// `PolydatMatter::build_under`; explicit-import callers leave this
    /// empty.
    pub fn mark_inherited_outputs(&mut self, names: Vec<String>) -> &mut Self {
        self.inherited_outputs = names;
        self
    }

    /// The parent surface this builder validates against — used by
    /// tests and by callers that need to inspect it during build.
    pub fn parent(&self) -> &ParentView {
        &self.parent
    }

    /// Declare an import.
    pub fn import(&mut self, spec: ImportSpec) -> &mut Self {
        self.imports.push(spec);
        self
    }

    /// Declare an export.
    pub fn export(&mut self, spec: ExportSpec) -> &mut Self {
        self.exports.push(spec);
        self
    }

    /// Append a body fragment. Multiple fragments are
    /// concatenated in registration order at finalize.
    pub fn body(&mut self, fragment: BodyFragment) -> &mut Self {
        self.body.push(fragment);
        self
    }

    /// Set the diagnostic context. Replaces any prior context.
    pub fn context(&mut self, ctx: SourceContext) -> &mut Self {
        self.context = ctx;
        self
    }

    /// Register a [`PullConsumer`]: the single init-time accumulator
    /// surface, the one a host's fixture adapter registers through
    /// (subcontext_construction.md §2.4).
    pub fn register_pull(&mut self, consumer: Arc<dyn PullConsumer>) -> &mut Self {
        self.consumers.push(RegisteredPullConsumer::new(consumer));
        self
    }

    /// Fold a `result:` source block into this child's module
    /// matter (subcontext_construction.md §3.2). Single entry point
    /// for the result-bindings kernel-driven path; applies the
    /// closure-binding economy (Rule 5) to magic externs and lets
    /// finalize's Rule 2 rewrite fire when result-LHS names collide
    /// with parent `shared` exports.
    ///
    /// `source` is Polydat source — the same `<name> := <expr>` form
    /// `bindings:` accepts. Both string-shape (`ResultSpec::String`)
    /// and map-shape (`ResultSpec::Map { name, source }` flattened
    /// to `<name> := <source>`) end up here.
    ///
    /// What this method does:
    ///
    /// 1. Parses `source` into `Vec<Statement>`.
    /// 2. Walks the body's free identifiers; for each magic
    ///    pre-bound name (`body`, `count`, `ok`) the source
    ///    references but doesn't already declare locally,
    ///    prepends an `extern <name>: <type>` declaration so
    ///    finalize compiles cleanly. Names not in the magic set
    ///    fall through to the standard import / cascade /
    ///    auto-extern path.
    /// 3. Records each `<name> := <expr>` LHS as an export, so
    ///    Rule 2 fires when the parent has a matching `shared`
    ///    export. The body fragment is appended; finalize's
    ///    existing rewrite is the load-bearing path.
    ///
    /// Path expressions (map-shape entries with no `:=` in the
    /// source) are NOT supported here — the caller flattens them
    /// to `<name> := <source>` and the Polydat compiler rejects them
    /// as unbound-identifier failures.
    pub fn add_result_bindings(&mut self, source: &str) -> Result<&mut Self, ContractViolation> {
        let trimmed = source.trim();
        if trimmed.is_empty() {
            return Ok(self);
        }
        let tokens = lex(source).map_err(|e| ContractViolation::Compile(e.to_string()))?;
        let file = parse(tokens).map_err(|e| ContractViolation::Compile(e.to_string()))?;

        // Collect locally-declared names (LHS of `:=` and
        // `init <name> = ...` and `extern <name>` so the magic-
        // extern injector skips them). These are the result-wire
        // exports we'll declare to the parent for Rule 2.
        let mut local_decls: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut result_lhs: Vec<String> = Vec::new();
        for stmt in &file.statements {
            match stmt {
                Statement::Binding(b) => {
                    for t in &b.targets {
                        local_decls.insert(t.clone());
                        if !result_lhs.contains(t) {
                            result_lhs.push(t.clone());
                        }
                    }
                }
                Statement::ExternPort(ep) => {
                    local_decls.insert(ep.name.clone());
                }
                Statement::InputDecl(d) => {
                    local_decls.insert(d.name.clone());
                }
                _ => {}
            }
        }

        // Walk free identifiers across the body. Used for both
        // (a) magic-extern injection (Rule 5 closure-binding
        // economy — only what's referenced gets a slot) and
        // (b) rejecting a user-written `body :=`.
        let mut free_idents: std::collections::HashSet<String> = std::collections::HashSet::new();
        for stmt in &file.statements {
            collect_free_idents(stmt, &mut free_idents);
        }

        // Assigning to a pre-bound wire is a hard error
        // (subcontext_construction.md §3.2). Catch
        // it before the magic-extern injector — otherwise the
        // injection would fight the LHS rename.
        for forbidden in ["body", "count", "ok"] {
            if result_lhs.iter().any(|n| n == forbidden) {
                return Err(ContractViolation::Compile(format!(
                    "result-bindings: '{forbidden}' is a runtime-injected wire and \
                     cannot be reassigned in `result:` (subcontext_construction.md §3.2)."
                )));
            }
        }

        // Magic-extern injection: only for names the source
        // actually references AND that aren't already declared
        // locally (the body might re-declare via `extern body`
        // explicitly — let that win).
        // `body` is a `Value::Json` — a structural value the
        // workload assertively unwraps via `exactly_one_value`.
        // The Json shape preserves row × column structure so
        // shape-mismatch diagnostics can name actual
        // dimensions; for unary results, `exactly_one_value`
        // collapses to a `Str` carrier which downstream
        // string predicates (regex_match, etc.) consume.
        let magic_externs: &[(&str, PortType, &str)] = &[
            ("body", PortType::Json, "Json"),
            ("count", PortType::U64, "u64"),
            ("ok", PortType::Bool, "bool"),
        ];
        let span0 = Span { line: 0, col: 0 };
        let mut prepended: Vec<Statement> = Vec::new();
        // Magic-extern slot allocation. Release: only inject when
        // the result-binding RHS actually references the name (the
        // closure-binding economy's DCE). Diagnostic: force-allocate
        // every magic extern not already locally declared, so writes
        // for `body` / `count` / `ok` always have a kernel slot to
        // land in regardless of whether anything reads them. The
        // diagnostic mode is for step-debug / cycle-replay; the
        // unused slots have no eval cone and add a fixed handful of
        // bytes to per-op-template state.
        let force_all = self.compile_options.kernel_opt.keep_unreferenced_slots();
        for (name, _pt, type_kw) in magic_externs {
            let referenced = free_idents.contains(*name);
            let already_local = local_decls.contains(*name);
            if (force_all || referenced) && !already_local {
                self.synthesized_externs.push((*name).to_string());
                prepended.push(Statement::ExternPort(ExternPort {
                    name: (*name).to_string(),
                    typ: (*type_kw).to_string(),
                    default: None,
                    span: span0,
                }));
            }
        }

        // Each result LHS may become a Rule 2 write-through when
        // the parent has a same-named `shared` cell visible in
        // scope. Without that match the binding stays a local
        // output and no export needs to be registered — the
        // result-LHS still becomes a kernel output through the
        // regular cycle-binding compile path, so wrappers /
        // metrics readers can still see it via wires.get.
        //
        // Registration is conditioned on an actual collision
        // because an export pre-allocates a U64 output port for
        // its LHS, and a non-colliding LHS whose expression is
        // not u64 (e.g. an f64 metric expression) would fail to
        // wire through it.
        {
            let in_scope_cells = self.parent.shared_cells();
            let parent_shared_by_name: std::collections::HashMap<&str, PortType> = in_scope_cells
                .iter()
                .map(|c| (c.name.as_str(), c.port_type))
                .collect();
            for name in &result_lhs {
                if let Some(&pt) = parent_shared_by_name.get(name.as_str()) {
                    self.exports.push(ExportSpec::shared(name.clone(), pt));
                }
            }
        }

        // Compose the prepended externs with the user's
        // statements and submit as a single Statements fragment.
        // This sidesteps the source-string round-trip the
        // PolydatSource fragment shape would force when prepended
        // declarations need to lead the user's source.
        let mut combined: Vec<Statement> = prepended;
        combined.extend(file.statements);
        self.body.push(BodyFragment::Statements(combined));

        Ok(self)
    }

    /// Close the builder. Validates the import contract against
    /// the parent's exports, compiles the body, and seals the
    /// pull consumers into the artifact.
    ///
    /// Rule 2 (write-through rewrite, subcontext_construction.md
    /// §3.1): when a
    /// child export name collides with a parent `shared` export,
    /// the body's `X := <expr>` is rewritten before compile to:
    ///
    /// 1. `extern X: <type>` — opens an input slot the parent's
    ///    `SharedCell` attaches to via `materialize_wiring_from_outer`.
    /// 2. `__write_X := <expr>` — a synthetic local computation
    ///    that produces the value to write through.
    ///
    /// At spawn time, the spawned child carries a write-through
    /// binding `(X, __write_X)`; `commit_write_throughs` pulls
    /// `__write_X` and stores its value through the child's
    /// input slot for `X`, which propagates to the cell (§5).
    pub fn finalize(self) -> Result<ScopeModule<Child<P>>, ContractViolation> {
        let SubcontextBuilder {
            parent,
            imports,
            exports,
            body,
            consumers,
            context,
            inherited_outputs,
            compile_options,
            synthesized_externs,
            _parent_marker,
        } = self;
        let mut synthesized = synthesized_externs;

        let mut diagnostics: Vec<String> = Vec::new();

        // ----- Rule 1 — import resolution against parent
        // exports: a name-closure check (subcontext_construction.md
        // §2.2, SC4).
        // `ImportSpec::port_type` and `classification` are carried
        // into the contract but not compared against the parent
        // here; the compiler's slot type checks and
        // `check_write_through_type` protect the actual child
        // inputs and cell writes. -----
        let parent_outputs: std::collections::HashSet<&String> =
            parent.output_names().iter().collect();
        let parent_inputs: std::collections::HashSet<&String> =
            parent.input_names().iter().collect();

        for imp in &imports {
            if !parent_outputs.contains(&imp.name) && !parent_inputs.contains(&imp.name) {
                return Err(ContractViolation::UnboundImport {
                    import: imp.name.clone(),
                    site: context.clone(),
                });
            }
        }

        // ----- Rule 2 — export collision detection. -----
        // For each declared export, check the parent for a same-
        // named modifier:
        //
        // * `final` parent → `FinalShadow` error (immutable,
        //   can't be redefined).
        // * `shared` cell visible at parent → record the export
        //   as a write-through candidate. The kernel-synthesis
        //   rewrite below renames the child's binding LHS to
        //   `__write_<name>` and inserts an `extern <name>`
        //   declaration; spawn's typed cell-attach pass then
        //   wires the input slot to the parent's `SharedCell`.
        //
        //   "Visible at parent" walks the typed
        //   `shared_cells_in_scope()` enumeration so an ancestral
        //   `shared X` cell propagates transitively even when an
        //   intermediate scope's body never names X. Without
        //   this, Rule 2 silently no-ops for grand-children and
        //   their write-throughs go nowhere.
        //
        // * No parent export and no in-scope cell → child-only
        //   export, registered locally (no rewrite).
        let in_scope_cells = parent.shared_cells();
        let in_scope_cells_by_name: std::collections::HashMap<&str, &SharedCellInScope> =
            in_scope_cells
                .iter()
                .map(|c| (c.name.as_str(), c))
                .collect();
        // A subscope is a program of the parent's tree: its compile is
        // charged to the parent's ledger.
        let ledger = parent.ledger().clone();
        let mut write_through_specs: Vec<(String, PortType)> = Vec::new();
        for exp in &exports {
            let parent_modifier = parent.output_modifier(&exp.name);
            if parent_modifier.is_const() && parent_outputs.contains(&exp.name) {
                return Err(ContractViolation::FinalShadow {
                    export: exp.name.clone(),
                    site: context.clone(),
                });
            }
            if let Some(in_scope) = in_scope_cells_by_name.get(exp.name.as_str()) {
                // Port type comes from the typed in-scope record
                // (sourced from the cell-bound input slot at the
                // owning ancestor), which is authoritative over the
                // export spec's declared port type.
                write_through_specs.push((exp.name.clone(), in_scope.port_type));
            }
        }

        // ----- Lower every body fragment into a single
        // Vec<Statement>. The Rule 2 rewrite operates on the AST
        // directly so it doesn't need a source-string round-trip;
        // PolydatSource fragments parse here once. -----
        if body.is_empty() {
            return Err(ContractViolation::Compile(
                "scope module body is empty — at least one fragment is required".into(),
            ));
        }
        let mut statements: Vec<Statement> = Vec::new();
        for fragment in &body {
            match fragment {
                BodyFragment::PolydatSource(src) => {
                    let tokens = lex(src).map_err(|e| ContractViolation::Compile(e.to_string()))?;
                    let file =
                        parse(tokens).map_err(|e| ContractViolation::Compile(e.to_string()))?;
                    statements.extend(file.statements);
                }
                BodyFragment::Statements(stmts) => statements.extend(stmts.iter().cloned()),
            }
        }

        // ----- Apply Rule 2 rewrite over the statement vector. -----
        let mut write_throughs: Vec<WriteThroughBinding> = Vec::new();
        if !write_through_specs.is_empty() {
            let already_extern: std::collections::HashSet<String> = statements
                .iter()
                .filter_map(|s| match s {
                    Statement::ExternPort(p) => Some(p.name.clone()),
                    _ => None,
                })
                .collect();
            let span0 = Span { line: 0, col: 0 };

            // Inject `extern <name>: <type>` declarations for
            // every write-through that the child body doesn't
            // already extern, typed exactly as the parent's cell,
            // so the cell attaches with no conversion. Prepend them
            // so the input slot is present before the compiler sees
            // the renamed binding.
            let mut prepended: Vec<Statement> = Vec::new();
            for (name, pt) in &write_through_specs {
                if already_extern.contains(name) {
                    continue;
                }
                synthesized.push(name.clone());
                prepended.push(Statement::ExternPort(ExternPort {
                    name: name.clone(),
                    typ: pt.to_keyword().to_string(),
                    default: None,
                    span: span0,
                }));
            }
            // Rename single-target CycleBindings whose LHS
            // matches a write-through export. Multi-target
            // bindings (tuple unpacks) aren't valid for shared
            // write-through (a tuple has no single value to
            // store in the cell); leave them alone — they'll
            // surface as a duplicate-port compile error if the
            // collision is real (subcontext_construction.md §3.1).
            for stmt in statements.iter_mut() {
                if let Statement::Binding(b) = stmt
                    && b.targets.len() == 1
                {
                    let target = &b.targets[0];
                    if write_through_specs.iter().any(|(n, _)| n == target) {
                        let original = target.clone();
                        let renamed = format!("{WRITE_THROUGH_PREFIX}{original}");
                        b.targets[0] = renamed.clone();
                        write_throughs.push(WriteThroughBinding {
                            export_name: original,
                            source_output: renamed,
                        });
                    }
                }
            }
            // Splice the synthetic externs in front. Order:
            // [externs] ++ [original statements (with renamed
            // LHS)].
            prepended.extend(statements);
            statements = prepended;
        }

        // ----- Compile the rewritten AST. -----
        //
        // Default options compile the statements under the default
        // DSL options. Non-default knobs (lib paths, strict mode,
        // required-output filter, source dir, context label) are
        // mapped into the DSL options, and a body made only of
        // `PolydatSource` fragments compiles from their concatenated
        // source text, so the compiler keeps `source_text` for
        // diagnostics (subcontext_construction.md §2.3).
        //
        // A rewritten AST (a Rule 2 write-through fired) or a
        // `Statements` body compiles through `compile_ast_interpreter_with_options`
        // with the full options, so the two combine freely.
        let dsl_options = DslOptions {
            source_dir: compile_options.workload_dir.clone(),
            lib_paths: compile_options.polydat_lib_paths.clone(),
            required_outputs: compile_options.required_outputs.clone(),
            strict: compile_options.strict,
            context: compile_options
                .context_label
                .clone()
                .unwrap_or_else(|| context.label.clone()),
            cursor_limit: compile_options.cursor_limit,
            input_variance: compile_options.input_variance,
            inferred_externs: synthesized.clone(),
            ledger: Some(ledger.clone()),
            resources: Some(parent.resources().clone()),
            engine: crate::Engine::default(),
        };
        let mut kernel = if compile_options.is_default() {
            compile_ast_interpreter_with_options(
                &PolydatFile {
                    statements: statements.clone(),
                },
                "",
                &DslOptions {
                    ledger: Some(ledger),
                    resources: Some(parent.resources().clone()),
                    ..DslOptions::default()
                },
                None,
            )
            .map_err(|e| ContractViolation::Compile(e.to_string()))?
        } else if !write_throughs.is_empty()
            || body
                .iter()
                .any(|f| matches!(f, BodyFragment::Statements(_)))
        {
            // When the AST has been rewritten in place (Rule 2
            // write-through) OR the body was submitted as
            // `Statements` (no source-string round-trip), feed the
            // rewritten AST through the libs-aware compile path
            // directly, so a write-through combines with
            // non-default compile options.
            compile_ast_interpreter_with_options(
                &PolydatFile {
                    statements: statements.clone(),
                },
                "",
                &dsl_options,
                None,
            )
            .map_err(|e| ContractViolation::Compile(e.to_string()))?
        } else {
            // No rewrite, no Statements fragments — reconstruct
            // the source string and compile it from source, so the
            // compiler stashes `source_text` for diagnostics.
            let mut src = String::new();
            for fragment in &body {
                match fragment {
                    BodyFragment::PolydatSource(s) => {
                        src.push_str(s);
                        if !s.ends_with('\n') {
                            src.push('\n');
                        }
                    }
                    BodyFragment::Statements(_) => unreachable!(
                        "Statements fragments routed through compile_ast_with_libs above"
                    ),
                }
            }
            crate::dsl::compile::compile_polydat_interpreter_with_options(&src, &dsl_options, None)
                .map_err(|e| ContractViolation::Compile(e.to_string()))?
        };

        // ----- Apply inherited-output marking, before the program Arc
        // is cloned out into the artifact. The module keeps the names so
        // its program on every other engine carries the same marks.
        if !inherited_outputs.is_empty() {
            kernel.mark_inherited_outputs(inherited_outputs.clone());
        }

        // ----- Bake Rule 2 write-throughs into the program. -----
        // The program is the single source of truth for these
        // bindings: any kernel built from this program (including
        // per-fiber re-instances via `bind_program_under_parent`)
        // inherits them via `from_program`'s automatic seeding.
        let kernel_write_throughs: Vec<crate::kernel::KernelWriteThrough> = write_throughs
            .iter()
            .map(|wt| crate::kernel::KernelWriteThrough {
                export_name: wt.export_name.clone(),
                source_output: wt.source_output.clone(),
            })
            .collect();
        if !kernel_write_throughs.is_empty() {
            kernel.bake_write_throughs(kernel_write_throughs);
        }

        // ----- Validate that every declared import shows up as
        // an input slot or a previously-folded constant on the
        // compiled program (Rule 5 — closure-binding economy
        // diagnostic; an unused import is a finalize-time
        // warning rather than an error). -----
        for imp in &imports {
            if kernel.program().find_input(&imp.name).is_none()
                && kernel.program().output_map_lookup(&imp.name).is_none()
            {
                diagnostics.push(format!(
                    "import `{}` declared but unused in body — Rule 5 closure-binding economy will drop it at spawn",
                    imp.name
                ));
            }
        }

        // ----- Validate Rule 2 invariants: the rewrite must
        // have produced (1) a child input slot for the export
        // name (so `materialize_wiring_from_outer` attaches the cell), and
        // (2) the synthetic `__write_<name>` output. -----
        for wt in &write_throughs {
            if kernel.program().find_input(&wt.export_name).is_none() {
                return Err(ContractViolation::Compile(format!(
                    "Rule 2 write-through rewrite for `{}` produced no input slot — \
                     check that the body's binding compiled to an input/output pair",
                    wt.export_name
                )));
            }
            if kernel
                .program()
                .output_map_lookup(&wt.source_output)
                .is_none()
            {
                return Err(ContractViolation::Compile(format!(
                    "Rule 2 write-through rewrite produced no `{}` output — \
                     the rewritten binding did not surface as a kernel output",
                    wt.source_output
                )));
            }
        }

        let program = kernel.program().clone();
        let contract = ScopeContract::from_specs(&imports, &exports);

        // The interpreter's program is already built, so it seeds the
        // per-engine table rather than being compiled a second time
        // when someone asks for it.
        let seeded: std::sync::Arc<dyn crate::kernel::KernelProgram> = program.clone();
        Ok(ScopeModule {
            imports,
            exports,
            program,
            statements,
            options: dsl_options.clone(),
            inherited_outputs,
            programs: std::sync::Mutex::new(std::collections::HashMap::from([(
                crate::Engine::Interpreter(crate::JitMode::Auto),
                seeded,
            )])),
            contract,
            context,
            consumers,
            write_throughs,
            diagnostics,
            _module: PhantomData,
        })
    }
}

/// Collect free identifiers referenced by a statement's RHS. Used
/// by [`SubcontextBuilder::add_result_bindings`] to apply the
/// closure-binding economy (Rule 5) — only inject magic externs
/// (`body` / `count` / `ok`) the source actually references.
fn collect_free_idents(stmt: &Statement, out: &mut std::collections::HashSet<String>) {
    match stmt {
        Statement::Binding(b) => collect_expr_idents(&b.value, out),
        Statement::Cursor(c) => collect_expr_idents(&c.constructor, out),
        Statement::ModuleDef(_)
        | Statement::ExternPort(_)
        | Statement::InputDecl(_)
        | Statement::Pragma { .. }
        | Statement::For(_)
        | Statement::Tile(_) => {}
    }
}

fn collect_expr_idents(expr: &Expr, out: &mut std::collections::HashSet<String>) {
    match expr {
        Expr::Ident(name, _) => {
            out.insert(name.clone());
        }
        Expr::IntLit(_, _) | Expr::FloatLit(_, _) => {}
        Expr::StringLit(_, _) => {
            // String interpolation `{name}` references aren't
            // expanded at the AST level — they're resolved by
            // the compiler during desugaring. Conservatively skip
            // them for the magic-extern injector (the user's
            // body / count / ok does not appear inside an
            // interpolation in a result binding);
            // unresolved interpolations surface as standard
            // unbound-identifier diagnostics downstream.
        }
        Expr::ArrayLit(items, _) => {
            for e in items {
                collect_expr_idents(e, out);
            }
        }
        Expr::Call(call) => collect_call_idents(call, out),
        Expr::BinOp(a, _, b) => {
            collect_expr_idents(a, out);
            collect_expr_idents(b, out);
        }
        Expr::For(_) => {}
        Expr::UnaryNeg(e, _) | Expr::UnaryBitNot(e, _) | Expr::Cast(e, _, _) => {
            collect_expr_idents(e, out)
        }
        Expr::FieldAccess { source, .. } => {
            // Source-field projections reference a source name,
            // not a wire — but the magic-extern set is a closed
            // {body, count, ok}, so the only way `body.x` could
            // appear is the user wrote a structural body access.
            // Record `source` as a referenced ident so the
            // magic-extern check sees it; the Polydat compiler will
            // produce the canonical "field access on non-source"
            // diagnostic if it doesn't resolve.
            out.insert(source.clone());
        }
    }
}

fn collect_call_idents(call: &CallExpr, out: &mut std::collections::HashSet<String>) {
    for arg in &call.args {
        match arg {
            Arg::Positional(e) => collect_expr_idents(e, out),
            Arg::Named(_, e) => collect_expr_idents(e, out),
        }
    }
}