kcl-lib 0.2.185

//! The CEK machine executor: evaluates KCL on an explicit, heap-allocated
//! continuation stack instead of the native call stack.
//!
//! The recursive executor (`execute_expr` and friends in `exec_ast.rs`) grows
//! the native stack proportionally to KCL nesting depth, which overflows in
//! wasm. This module re-expresses evaluation as a CEK abstract machine with an
//! ambient mutable store:
//!
//! - C (control): [`Control`] -- either descend into an expression
//!   ([`Control::Eval`]), hand a finished value to the innermost continuation
//!   ([`Control::Apply`]), unwind a KCL 3.0 `return` to the nearest call
//!   boundary ([`Control::Return`]), or unwind an `exit()`
//!   ([`Control::Exit`]).
//! - E (environment): already reified as `EnvironmentRef` plus the mutable
//!   env-stack arena, exactly as the recursive executor uses it.
//! - K (continuations): a `Vec<Kont>`, one defunctionalized variant per
//!   recursion site of the recursive executor.
//!
//! The store (artifacts, operations log, id generators, engine connection) is
//! ambient `&mut ExecState`, threaded through every step -- so this is a CEK
//! machine with an ambient mutable store, not a pure CESK machine.
//!
//! Async stays async: engine I/O and ordinary stdlib calls remain flat
//! `.await` leaves inside the loop. They do not grow the native stack with KCL
//! depth. Only recursion on evaluating KCL expressions moved onto the explicit
//! stack.
//!
//! Semantics contract: this machine replicates the recursive executor
//! byte-for-byte -- including evaluation order (ids and engine commands are
//! order-dependent), the operations log, ambient-state save/restore, error
//! paths (including their historical asymmetries), and the version-gated
//! semantics: `return` (write-and-continue before KCL 3.0; early return under
//! it), if-arm scoping (KCL 3.0 only), and member-expression evaluation order
//! (object before property under KCL 3.0; property first before it). The
//! simulation-test suite runs under both executors and requires identical
//! snapshots.
//!
//! Bounded native re-entry still exists in two places, by design:
//! - Module execution (imports, module-value results) runs the module body in
//!   a fresh machine root via `exec_module_body`, adding O(1) native frames
//!   per module nesting level.
//! - Until the resumable-callback phase lands, higher-order stdlib functions
//!   (`map`, `reduce`, `patternTransform`) call back into KCL through
//!   `call_kw`, which fresh-roots a machine run per callback, adding O(1)
//!   native frames per callback nesting level (bounded by the recursive
//!   executor's call-stack cap, which still guards that path).
//!
//! # State transitions under KCL 3.0
//!
//! In the notation of the CEK/CESK literature (Felleisen and Friedman's
//! original CEK machine; the presentation in Van Horn and Might's
//! "Abstracting Abstract Machines"), a state is a triple ⟨C, ρ, κ⟩: control,
//! environment, continuation. Here ρ is the current `EnvironmentRef`, and the
//! arena of environments doubles as the store σ (environments are heap
//! entries addressed by refs); both are ambient in `ExecState` rather than
//! copied into the state tuple, so ρ is only written below where it changes
//! (▷ pushes a frame or scope, and the matching pop happens in the rule that
//! consumes the frame). κ is the `Vec<Kont>`; the innermost frame is written
//! leftmost (`f :: κ`), which is the end of the vector in code. Apply-mode
//! states are written ⟨v, κ⟩. The machine is in exactly one of five modes:
//!
//! ```text
//! ⟨e, ρ, κ⟩     eval mode        Control::Eval    dispatch on the expression e
//! ⟨v, κ⟩        apply mode       Control::Apply   dispatch on the innermost frame
//! R⟨v, κ⟩       return unwind    Control::Return  KCL 3.0 only
//! X⟨v, κ⟩       exit unwind      Control::Exit
//! E⟨err, κ⟩     error unwind     (the Err return path through run_loop)
//! ```
//!
//! Core transitions. The long tail of evaluation frames (Unary, ArrayElems,
//! ObjectProps, RangeStartDone/RangeEndDone, AscribeDone, LabelDone,
//! CallArgs, SketchArgs) follows the same left-to-right pattern as the
//! binary-operator rules and is omitted. Member access is spelled out because
//! its evaluation order is version-gated.
//!
//! ```text
//! Atoms and names
//!   ⟨lit, ρ, κ⟩                       ↦  ⟨value(lit), κ⟩
//!   ⟨x, ρ, κ⟩                         ↦  ⟨ρ(x), κ⟩            lookup via parent refs
//!
//! Binary operators (representative of the boilerplate family)
//!   ⟨e₁ ⊕ e₂, ρ, κ⟩                   ↦  ⟨e₁, ρ, BinaryLhsDone(⊕, e₂) :: κ⟩
//!   ⟨v₁, BinaryLhsDone(⊕, e₂) :: κ⟩   ↦  ⟨e₂, ρ, BinaryRhsDone(⊕, v₁) :: κ⟩
//!   ⟨v₂, BinaryRhsDone(⊕, v₁) :: κ⟩   ↦  ⟨v₁ ⊕ v₂, κ⟩
//!
//! Member access (source order: the object before the property)
//!   ⟨e₁[e₂], ρ, κ⟩                    ↦  ⟨e₁, ρ, MemberObjDone([e₂]) :: κ⟩
//!   ⟨v₁, MemberObjDone([e₂]) :: κ⟩    ↦  ⟨e₂, ρ, MemberPropDone(v₁) :: κ⟩
//!   ⟨v₂, MemberPropDone(v₁) :: κ⟩     ↦  ⟨v₁[v₂], κ⟩
//!   ⟨e₁.x, ρ, κ⟩                      ↦  ⟨e₁, ρ, MemberObjDone(.x) :: κ⟩
//!   ⟨v₁, MemberObjDone(.x) :: κ⟩      ↦  ⟨v₁.x, κ⟩   x is a name, never evaluated
//!                                        pre-KCL-3.0: e₂ is evaluated before e₁
//!                                        (LegacyMemberPropDone, then
//!                                        LegacyMemberObjDone(v₂) evaluates e₁);
//!                                        kept for old programs because ids and
//!                                        engine commands depend on the order
//!
//! Blocks (statement sequencing; `last` is the trailing expression value)
//!   ⟨{s₁ … sₙ}, ρ, κ⟩                 ↦  ⟨s₁, ρ, BlockSeq₁ :: κ⟩
//!   ⟨v, BlockSeqᵢ :: κ⟩               ↦  ⟨sᵢ₊₁, ρ, BlockSeqᵢ₊₁ :: κ⟩
//!   ⟨v, BlockSeqₙ :: κ⟩               ↦  ⟨Block(last), κ⟩
//!
//! Function calls (user-defined KCL functions)
//!   ⟨f(args…), ρ, κ⟩                  ↦  ⟨args…, ρ, CallArgs :: κ⟩
//!   ⟨vargs, CallArgs :: κ⟩            ↦  ⟨body, ρf ▷ frame, CallBoundary :: κ⟩
//!                                        where ρf is the closure's captured env
//!   ⟨Block(last), CallBoundary :: κ⟩  ↦  ⟨v, κ⟩    call_finish pops the frame; the
//!                                        result is a Return absorbed at the boundary
//!                                        (KCL 3.0) or `__return` read from the frame
//!                                        (pre-KCL-3.0); `last` itself is ignored
//!
//! If-expressions (conditions evaluate in the enclosing ρ)
//!   ⟨if e₀ {b₀} …, ρ, κ⟩              ↦  ⟨e₀, ρ, IfCondDone₀ :: κ⟩
//!   ⟨false, IfCondDoneᵢ :: κ⟩         ↦  ⟨eᵢ₊₁, ρ, IfCondDoneᵢ₊₁ :: κ⟩   the next
//!                                        cond, or the final else body via the
//!                                        true-rule below when conds are exhausted
//!   ⟨true, IfCondDoneᵢ :: κ⟩          ↦  ⟨bᵢ, ρ ▷ arm, IfArmDone :: κ⟩
//!                                        KCL 3.0: the arm body gets its own scope
//!                                        env (pre-KCL-3.0: bᵢ runs directly in ρ)
//!   ⟨Block(v), IfArmDone :: κ⟩        ↦  ⟨v, κ⟩    KCL 3.0: pop the arm env; values
//!                                        escaping the arm stay valid because
//!                                        referenced envs survive the pop
//!
//! return (KCL 3.0: genuine control flow)
//!   ⟨return e, ρ, κ⟩                  ↦  ⟨e, ρ, BlockSeq{Return} :: κ⟩
//!   ⟨v, BlockSeq{Return} :: κ⟩        ↦  R⟨v, κ⟩   the rest of the block is dropped
//!                                        (pre-KCL-3.0: bind `__return` in ρ and
//!                                        continue with the next statement instead)
//!   R⟨v, f :: κ⟩                      ↦  R⟨v, κ⟩   non-boundary f runs cleanup(f):
//!                                        IfArmDone pops its arm env, PipeSeq
//!                                        restores the pipe value, …
//!   R⟨v, SketchBody :: κ⟩             ↦  R⟨v, κ⟩   sketch cleanup; finalization
//!                                        skipped
//!   R⟨v, CallBoundary :: κ⟩           ↦  ⟨v′, κ⟩   absorbed: call_finish pops the
//!                                        frame, applies tag updates and return-type
//!                                        coercion, and the machine resumes
//!   R⟨v, ∅⟩                           ↦  the fresh root hands the Return upward (a
//!                                        callback's native frame absorbs it; a Root
//!                                        block rejects it: "Cannot return from
//!                                        outside a function")
//!
//! exit() (never absorbed at a boundary)
//!   X⟨v, f :: κ⟩                      ↦  X⟨v, κ⟩   cleanup(f); at CallBoundary the
//!                                        exit flavor of call_finish (skips tags and
//!                                        coercion); at SketchBody sketch cleanup
//!   X⟨v, ∅⟩                           ↦  the program result
//!
//! Errors (mirrors X, plus call_finish(Err) and per-boundary backtrace
//! decoration via add_unwind_location)
//!   E⟨err, f :: κ⟩                    ↦  E⟨err′, κ⟩
//! ```
//!
//! Pipes thread `%` through ambient state rather than κ: PipeFirstDone and
//! PipeSeq save and restore `pipe_value` around each element, which is why
//! the unwind rules must run cleanup on them.

use std::env;
use std::sync::Arc;

use indexmap::IndexMap;

use crate::SourceRange;
use crate::errors::KclError;
use crate::errors::KclErrorDetails;
use crate::execution::ArtifactId;
use crate::execution::BodyType;
use crate::execution::ExecState;
use crate::execution::ExecutorContext;
use crate::execution::KclValue;
use crate::execution::KclValueControlFlow;
use crate::execution::Metadata;
use crate::execution::ModelingCmdMeta;
use crate::execution::SketchSurface;
use crate::execution::cad_op::Operation;
use crate::execution::exec_ast::Property;
use crate::execution::fn_call::Arg;
use crate::execution::fn_call::Args;
use crate::execution::fn_call::CallState;
use crate::execution::kcl_value::FunctionBody;
use crate::execution::kcl_value::FunctionSource;
use crate::execution::kcl_value::KclObjectFields;
use crate::execution::state::SketchBlockState;
use crate::execution::types::PrimitiveType;
use crate::execution::types::RuntimeType;
use crate::front::ObjectId;
use crate::kcl_runtime_flags;
use crate::parsing::ast::types::Annotation;
use crate::parsing::ast::types::ArrayExpression;
use crate::parsing::ast::types::ArrayRangeExpression;
use crate::parsing::ast::types::AscribedExpression;
use crate::parsing::ast::types::BinaryExpression;
use crate::parsing::ast::types::BinaryPart;
use crate::parsing::ast::types::Block;
use crate::parsing::ast::types::BodyItem;
use crate::parsing::ast::types::CallExpressionKw;
use crate::parsing::ast::types::CodeBlock;
use crate::parsing::ast::types::Expr;
use crate::parsing::ast::types::FunctionExpression;
use crate::parsing::ast::types::IfExpression;
use crate::parsing::ast::types::LabelledExpression;
use crate::parsing::ast::types::MemberExpression;
use crate::parsing::ast::types::Node;
use crate::parsing::ast::types::ObjectExpression;
use crate::parsing::ast::types::PipeExpression;
use crate::parsing::ast::types::Program;
use crate::parsing::ast::types::SketchBlock;
use crate::parsing::ast::types::UnaryExpression;
use crate::runtime_flags::RuntimeFlagResolve;
use crate::runtime_flags::resolve_from_sources;

/// Environment variable selecting which executor implementation to use.
const KCL_EXECUTOR_ENV_VAR: &str = "KCL_EXECUTOR";

/// Which executor evaluates KCL. Crate-internal: set on
/// [`ExecutorContext`] at construction time and immutable during a run.
/// Cloning the context (fresh roots, `Args`) inherits the same kind, so every
/// module and callback runs on the same executor.
///
/// Precedence: runtime flags > `KCL_EXECUTOR` >
/// [`ExecutorKind::resolve_default()`]. The shared resolver's test-override
/// slot is deliberately unused here: tests pin a kind by passing it to the
/// `ExecutorContext` they construct, and everything else must keep following
/// `KCL_EXECUTOR` so the CI matrix genuinely runs the suite under both
/// executors.
///
/// Selection is deliberately a process-global runtime flag rather than an
/// `ExecutorSettings` field: TS reaches executor settings only through the
/// user-settings `Configuration` schema (TOML / JsonSchema / generated docs),
/// and settings participate in execution-cache comparison, so a transient
/// rollout flag does not belong there. The app rebuilds the context per wasm
/// call, so the global still yields per-execution capture.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ExecutorKind {
    /// The historical recursive tree-walking evaluator.
    Recursive,
    /// The CEK machine in this module.
    Machine,
}

impl std::fmt::Display for ExecutorKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ExecutorKind::Recursive => write!(f, "recursive"),
            ExecutorKind::Machine => write!(f, "machine"),
        }
    }
}

impl RuntimeFlagResolve for ExecutorKind {
    fn on() -> Self {
        Self::Machine
    }

    fn off() -> Self {
        Self::Recursive
    }

    fn resolve_default() -> Self {
        Self::Machine
    }

    /// Select the executor from the `KCL_EXECUTOR` environment variable:
    /// `machine` or `recursive`, ASCII case-insensitive, surrounding whitespace
    /// ignored. Setting the variable is an explicit opt-in and applies to any
    /// context construction that consults it; unset or empty means
    /// [`ExecutorKind::default`]. Any other value is a configuration error that
    /// warns and then uses the default. (Wasm builds have no environment, so
    /// this is always the default there.)
    fn parse_env_var(value: &str) -> Self {
        let value = value.trim();
        if value.is_empty() {
            Self::resolve_default()
        } else if value.eq_ignore_ascii_case("machine") {
            ExecutorKind::Machine
        } else if value.eq_ignore_ascii_case("recursive") {
            ExecutorKind::Recursive
        } else {
            let def = Self::resolve_default();
            // A mistyped value should not crash the process: warn and fall back
            // to the default (the conservative choice for a misconfiguration).
            Self::warn_once(|| {
                format!(
                    "Unsupported {KCL_EXECUTOR_ENV_VAR} value `{value}`; expected `recursive` or `machine`. Defaulting to `{def}`."
                )
            });
            def
        }
    }
}

impl ExecutorKind {
    /// Resolve the active executor (see precedence on [`ExecutorKind`]).
    pub(crate) fn resolve() -> Self {
        let env_value = match env::var(KCL_EXECUTOR_ENV_VAR) {
            Ok(value) => Some(value),
            Err(env::VarError::NotPresent) => None,
            Err(env::VarError::NotUnicode(value)) => {
                // Invalid-unicode env var: warn and fall back rather than crash.
                Self::warn_once(|| {
                    let def = Self::resolve_default();
                    format!(
                        "{KCL_EXECUTOR_ENV_VAR} must be valid unicode; got `{}`. Defaulting to `{def}`.",
                        value.to_string_lossy()
                    )
                });
                None
            }
        };

        // The `None` is the test-override slot: unlike `LexerMode`, the
        // executor never supplies one (see the type-level docs).
        resolve_from_sources(kcl_runtime_flags().use_cek_executor, None, env_value.as_deref())
    }

    /// Emit a one-time configuration warning through `crate::log` (gated on
    /// `ZOO_LOG`). `resolve`/`parse` run on every executor creation, so a
    /// misconfigured `KCL_EXECUTOR` must not warn -- or allocate the message --
    /// on every call. One guard suffices: only one kind of misconfiguration can
    /// occur per process, since the env var holds a single value.
    fn warn_once(make_message: impl FnOnce() -> String) {
        static WARNED: std::sync::Once = std::sync::Once::new();
        WARNED.call_once(|| crate::log::log(make_message()));
    }
}

/// Default limit for the machine executor's call-depth guard. This replaces
/// the recursive executor's native-stack-motivated cap for machine-native
/// calls: depth is heap-bounded, so the limit is purely a runaway-recursion
/// policy. It must remain correct when internally raised to 50,000 or more.
pub(crate) const DEFAULT_MACHINE_CALL_DEPTH_LIMIT: usize = 1000;

/// Convert an entry block into an owned handle the machine can keep in a
/// continuation without lifetimes.
pub(crate) trait ToMachineBlock {
    fn to_machine_block(&self) -> BlockRef;
}

impl ToMachineBlock for Node<Program> {
    fn to_machine_block(&self) -> BlockRef {
        // One shallow clone per fresh root: children are Arc-backed, so this
        // copies one Vec of enum discriminants, not the tree.
        BlockRef::Program(Arc::new(self.clone()))
    }
}

impl ToMachineBlock for Node<Block> {
    fn to_machine_block(&self) -> BlockRef {
        BlockRef::Block(Arc::new(self.clone()))
    }
}

/// An owned handle to a block of statements. Function and sketch bodies are
/// projected out of their enclosing node's `Arc` so pushing a call costs no
/// clone.
#[derive(Debug, Clone)]
pub(crate) enum BlockRef {
    Program(Arc<Node<Program>>),
    Block(Arc<Node<Block>>),
    FnBody(Arc<Node<FunctionExpression>>),
    SketchBody(Arc<Node<SketchBlock>>),
}

impl BlockRef {
    fn body(&self) -> &[BodyItem] {
        match self {
            BlockRef::Program(p) => p.body(),
            BlockRef::Block(b) => b.body(),
            BlockRef::FnBody(f) => f.body.body(),
            BlockRef::SketchBody(s) => s.body.body(),
        }
    }

    fn to_source_range(&self) -> SourceRange {
        match self {
            BlockRef::Program(p) => p.to_source_range(),
            BlockRef::Block(b) => b.to_source_range(),
            BlockRef::FnBody(f) => f.body.to_source_range(),
            BlockRef::SketchBody(s) => s.body.to_source_range(),
        }
    }
}

/// A node the machine can evaluate. `BinaryPart` is a distinct enum from
/// `Expr` (binary operands are `BinaryPart`s), so the machine's eval step
/// takes this union.
// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone)]
enum EvalNode {
    Expr(Expr),
    BinaryPart(BinaryPart),
}

/// One evaluation request: the node plus the context the recursive executor
/// passed as parameters to `execute_expr`.
#[derive(Debug)]
struct EvalRequest {
    node: EvalNode,
    metadata: Metadata,
    /// `Some(name)` when this is the top-level init of `name = ...`, i.e. the
    /// recursive executor's `StatementKind::Declaration`. Everything else is
    /// `StatementKind::Expression`.
    decl_name: Option<String>,
    /// Outer annotations of the enclosing declaration; consumed by function
    /// expressions. Empty everywhere else.
    annotations: Vec<Node<Annotation>>,
}

impl EvalRequest {
    fn expr(node: &Expr) -> Self {
        EvalRequest {
            node: EvalNode::Expr(node.clone()),
            metadata: Metadata {
                source_range: SourceRange::from(node),
            },
            decl_name: None,
            annotations: Vec::new(),
        }
    }

    fn binary_part(part: &BinaryPart) -> Self {
        EvalRequest {
            node: EvalNode::BinaryPart(part.clone()),
            metadata: Metadata {
                source_range: SourceRange::from(part),
            },
            decl_name: None,
            annotations: Vec::new(),
        }
    }
}

/// "C": what the machine does next.
// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
enum Control {
    /// Descend into an expression.
    Eval(Box<EvalRequest>),
    /// A value is ready; hand it to the innermost continuation.
    Apply(Applied),
    /// A `return` executed under a KCL 3.0 entry point: unwind continuations --
    /// running cleanups -- to the nearest call boundary, which absorbs the
    /// value as the function's normal result (tags and return-type coercion
    /// apply) and resumes the machine. Reaches the fresh root only when no
    /// boundary remains (a callback body, whose native call frame absorbs it,
    /// or a top-level return escaping an expression, which run_block rejects).
    Return(KclValueControlFlow),
    /// `exit()` was called (or an edited sketch block finished): unwind every
    /// continuation -- running cleanups, and the exit flavor of call_finish on
    /// call boundaries -- out to the fresh root, which returns the carried
    /// control-flow value. Only `Exit`-kind control flow enters here.
    Exit(KclValueControlFlow),
}

/// What a finished sub-computation produced.
// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
enum Applied {
    /// An expression's value.
    Value(KclValue),
    /// A completed block's result: the recursive executor's
    /// `exec_block` return value (`last_expr`).
    Block(Option<KclValueControlFlow>),
}

impl Applied {
    fn expect_value(self) -> Result<KclValue, KclError> {
        match self {
            Applied::Value(v) => Ok(v),
            Applied::Block(_) => Err(KclError::new_internal(KclErrorDetails::new(
                "machine executor: expected an expression value, found a block result".to_owned(),
                Vec::new(),
            ))),
        }
    }

    fn expect_block(self) -> Result<Option<KclValueControlFlow>, KclError> {
        match self {
            Applied::Block(b) => Ok(b),
            Applied::Value(_) => Err(KclError::new_internal(KclErrorDetails::new(
                "machine executor: expected a block result, found an expression value".to_owned(),
                Vec::new(),
            ))),
        }
    }
}

/// Which statement of a block is in flight, with any ambient state to restore
/// when it finishes or unwinds.
#[derive(Debug)]
enum InFlight {
    /// No statement is mid-evaluation.
    None,
    Expression,
    /// A variable declaration's init is being evaluated; holds the
    /// `being_declared` value to restore.
    Declaration {
        prev_being_declared: Option<String>,
    },
    Return,
}

/// The defunctionalized continuation: one variant per recursion site of the
/// recursive executor. Every variant holds owned handles (Arcs or values), no
/// lifetimes.
#[derive(Debug)]
enum Kont {
    /// Statement sequencing: replaces `exec_block`'s `for` loop.
    BlockSeq {
        block: BlockRef,
        body_type: BodyType,
        /// Index of the statement currently in flight (when `in_flight` is not
        /// `None`) or the next statement to run.
        index: usize,
        in_flight: InFlight,
        last: Option<KclValueControlFlow>,
    },

    // ---- binary / unary ----
    /// Have the LHS; evaluate the RHS next.
    BinaryLhsDone {
        node: Arc<Node<BinaryExpression>>,
    },
    /// Have both operands; apply the operator (may await a CSG engine call).
    BinaryRhsDone {
        node: Arc<Node<BinaryExpression>>,
        left: KclValue,
    },
    UnaryDone {
        node: Arc<Node<UnaryExpression>>,
    },

    // ---- aggregates ----
    ArrayElems {
        node: Arc<Node<ArrayExpression>>,
        index: usize,
        done: Vec<KclValue>,
    },
    ObjectProps {
        node: Arc<Node<ObjectExpression>>,
        index: usize,
        done: KclObjectFields,
    },
    RangeStartDone {
        node: Arc<Node<ArrayRangeExpression>>,
    },
    RangeEndDone {
        node: Arc<Node<ArrayRangeExpression>>,
        start: KclValue,
    },

    // ---- access / control / pipes ----
    /// Pre-KCL-3.0 member access order: a computed property is evaluated
    /// before the object, replicating the recursive executor's historical
    /// evaluation order.
    LegacyMemberPropDone {
        node: Arc<Node<MemberExpression>>,
    },
    /// Pre-KCL-3.0 member access order: the object is evaluated after the
    /// property; apply the access.
    LegacyMemberObjDone {
        node: Arc<Node<MemberExpression>>,
        property: Property,
    },
    /// KCL 3.0 member access in source order: the object is evaluated
    /// first. A computed property is evaluated next; a static property name
    /// (`obj.name`) applies immediately.
    MemberObjDone {
        node: Arc<Node<MemberExpression>>,
    },
    /// KCL 3.0 member access in source order: the computed property is
    /// evaluated after the object; apply the access.
    MemberPropDone {
        node: Arc<Node<MemberExpression>>,
        object: KclValue,
    },
    IfCondDone {
        node: Arc<Node<IfExpression>>,
        /// 0 = the main condition; 1..=n = else-if conditions.
        arm: usize,
    },
    /// An if arm's block completed; unwrap its value as the if's result.
    IfArmDone {
        node: Arc<Node<IfExpression>>,
        /// True when the arm body runs in its own scope environment (a
        /// KCL 3.0 entry point), which must be popped on completion and
        /// on unwind.
        env_pushed: bool,
    },
    AscribeDone {
        node: Arc<Node<AscribedExpression>>,
    },
    LabelDone {
        node: Arc<Node<LabelledExpression>>,
    },
    /// The pipe's first element evaluated under the PARENT pipe value.
    PipeFirstDone {
        node: Arc<Node<PipeExpression>>,
    },
    PipeSeq {
        node: Arc<Node<PipeExpression>>,
        index: usize,
        saved_pipe_value: Option<KclValue>,
    },

    // ---- calls ----
    CallArgs(Box<CallArgsState>),
    /// A KCL function call boundary. `call_finish` runs on normal completion,
    /// on `Exit` (exit flavor: restores and op-finalize, no tags/coercion),
    /// and on error unwind.
    CallBoundary(Box<BoundaryState>),

    // ---- sketch blocks ----
    SketchArgs(Box<SketchArgsState>),
    SketchBody(Box<SketchBodyState>),

    // ---- resumable higher-order builtins ----
    /// The reified native-loop state of map/reduce/patternTransform: their
    /// callbacks run on THIS stack, so arbitrarily deep callback nesting
    /// cannot overflow the native stack.
    Resume(Box<ResumeState>),
}

/// Which resumable builtin is mid-iteration, with its loop state.
// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
enum ResumeState {
    Map(MapResume),
    Reduce(ReduceResume),
    PatternTransform(PatternTransformResume),
}

#[derive(Debug)]
struct MapResume {
    f: FunctionSource,
    rest: std::vec::IntoIter<KclValue>,
    done: Vec<KclValue>,
    source_range: SourceRange,
    node_path: Option<crate::NodePath>,
}

#[derive(Debug)]
struct ReduceResume {
    f: FunctionSource,
    rest: std::vec::IntoIter<KclValue>,
    source_range: SourceRange,
    node_path: Option<crate::NodePath>,
}

#[derive(Debug)]
struct PatternTransformResume {
    transform: FunctionSource,
    instances: u32,
    /// The repetition to run next; make_transform iterates 1..instances.
    next_i: u32,
    transforms: Vec<Vec<kittycad_modeling_cmds::shared::Transform>>,
    geometry: PatternGeometry,
    use_original: bool,
    source_range: SourceRange,
    node_path: Option<crate::NodePath>,
    /// The builtin's own (desugared) arguments; the final engine application
    /// reads its metadata exactly like the recursive executor does.
    args: Args,
}

#[derive(Debug)]
enum PatternGeometry {
    Solids(Vec<crate::execution::Solid>),
    Sketches(Vec<crate::execution::Sketch>),
    ImportedGeometry(crate::execution::ImportedGeometry),
}

/// Argument-evaluation state for a call: mirrors the argument loop of
/// `CallExpressionKw::execute`.
#[derive(Debug)]
struct CallArgsState {
    node: Arc<Node<CallExpressionKw>>,
    fn_src: FunctionSource,
    /// Source ranges of the function value, for the undefined-result error.
    fn_meta: Vec<Metadata>,
    /// None while the unlabeled arg (if any) is in flight; Some(i) while
    /// arguments[i] is.
    cursor: Option<usize>,
    unlabeled: Vec<(Option<String>, Arg)>,
    labeled: IndexMap<String, Arg>,
}

/// Everything a call boundary needs to finish the call and decorate errors.
#[derive(Debug)]
struct BoundaryState {
    state: CallState,
    fn_src: FunctionSource,
    fn_name: Option<String>,
    callsite: SourceRange,
    /// What kind of raw result this boundary receives.
    expects: BoundaryExpects,
    /// Who initiated the call, which decides what happens to the finished
    /// result (and whether errors get a backtrace entry).
    completion: BoundaryCompletion,
}

/// The raw result shape a call boundary receives.
#[derive(Debug)]
enum BoundaryExpects {
    /// A KCL function body ran: the block result arrives and the function's
    /// value is the recorded __return.
    KclBlock,
    /// A resumable builtin's iteration produced the value directly.
    StdValue,
}

/// Who initiated a call.
#[derive(Debug)]
enum BoundaryCompletion {
    /// A call expression: a missing result is the "undefined function
    /// result" error, and unwinding errors gain a backtrace entry.
    CallExpr { fn_meta: Vec<Metadata> },
    /// A callback invoked by a resumable builtin (map/reduce/
    /// patternTransform): the finished result feeds the builtin's resume
    /// continuation, and -- like the recursive executor's call_kw callers --
    /// errors get no extra backtrace entry.
    Callback,
}

/// Argument-evaluation state for a sketch block (non-sketch-mode).
#[derive(Debug)]
struct SketchArgsState {
    node: Arc<Node<SketchBlock>>,
    index: usize,
    labeled: IndexMap<String, Arg>,
}

/// Ambient state a sketch-block body wraps; restored on every exit path.
#[derive(Debug)]
struct SketchBodyState {
    node: Arc<Node<SketchBlock>>,
    sketch_id: ObjectId,
    sketch_surface: SketchSurface,
    sketch_block_artifact_id: ArtifactId,
    saved_sketch_block: Option<SketchBlockState>,
    saved_sketch_mode: bool,
    /// Whether the body's child environment was pushed (the sketch2 alias
    /// environment always is, before this kont exists).
    child_env_pushed: bool,
}

impl ExecutorContext {
    pub(crate) fn is_machine_executor(&self) -> bool {
        self.executor_kind == ExecutorKind::Machine
    }
}

/// Run a block to completion on the machine. This is the machine's fresh-root
/// entry, with `exec_block`'s exact contract: the returned value is the
/// block's `last_expr` (which the module wrapper, call protocol, and sketch
/// handling consume exactly as they do for the recursive executor). An
/// `exit()` unwinds to this root and is returned as an `Exit`-flavored
/// control-flow value. A root block flushes the engine's batched commands on
/// both completion and exit -- never on error -- matching `exec_block`.
pub(super) async fn run_block(
    ctx: &ExecutorContext,
    block: BlockRef,
    exec_state: &mut ExecState,
    body_type: BodyType,
) -> Result<Option<KclValueControlFlow>, KclError> {
    let block_range = block.to_source_range();
    let mut konts: Vec<Kont> = Vec::new();
    // Kick the root block sequence off: stepping a block with no statement in
    // flight advances to its first statement.
    let root = Kont::BlockSeq {
        block,
        body_type,
        index: 0,
        in_flight: InFlight::None,
        last: None,
    };
    let control = match step_block(root, None, &mut konts, exec_state, ctx).await {
        Ok(c) => c,
        Err(e) => return Err(unwind_error(e, &mut konts, exec_state)),
    };
    let result = run_loop(ctx, control, konts, exec_state).await?;
    // A Return can reach a root block only by escaping an expression
    // evaluated at the top level (e.g. an if-arm); a `return` statement here
    // is rejected in step_block before it evaluates. Match the recursive
    // executor: error, skipping the flush.
    if matches!(body_type, BodyType::Root)
        && let RootResult::Exited(cf) = &result
        && cf.is_return()
    {
        return Err(KclError::new_semantic(KclErrorDetails::new(
            "Cannot return from outside a function.".to_owned(),
            cf.source_ranges(),
        )));
    }
    if matches!(body_type, BodyType::Root) {
        // Flush the batch queue for the root, whether it completed or was
        // terminated by an exit(): the recursive executor's flush sits after
        // its statement loop, which its exit `break` still reaches. Errors
        // skip the flush (the `?` above), also matching. This must run after
        // unwinding, so that any sketch-block state has been restored, and it
        // cannot live in step_block, which an exit never re-enters.
        exec_state
            .flush_batch(
                ModelingCmdMeta::new(exec_state, ctx, block_range),
                // True here tells the engine to flush all the end commands as well like fillets
                // and chamfers where the engine would otherwise eat the ID of the segments.
                true,
            )
            .await?;
    }
    match result {
        RootResult::Done(applied) => applied.expect_block(),
        RootResult::Exited(cf) => Ok(Some(cf)),
    }
}

/// Run a single expression to completion on a fresh machine root. Used for
/// bounded, internal evaluations (e.g. GD&T's constant `XY` plane lookup).
pub(crate) async fn run_expr(
    ctx: &ExecutorContext,
    expr: &Expr,
    exec_state: &mut ExecState,
    metadata: &Metadata,
) -> Result<KclValueControlFlow, KclError> {
    let req = EvalRequest {
        node: EvalNode::Expr(expr.clone()),
        metadata: *metadata,
        decl_name: None,
        annotations: Vec::new(),
    };
    match run_loop(ctx, Control::Eval(Box::new(req)), Vec::new(), exec_state).await? {
        RootResult::Done(applied) => Ok(applied.expect_value()?.continue_()),
        RootResult::Exited(cf) => Ok(cf),
    }
}

/// How a fresh root ended.
// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
enum RootResult {
    /// The root computation finished with this result.
    Done(Applied),
    /// An exit() -- or a KCL 3.0 return with no call boundary left -- unwound
    /// all the way out (cleanups already ran).
    Exited(KclValueControlFlow),
}

/// The machine loop: step until the continuation stack is exhausted.
async fn run_loop(
    ctx: &ExecutorContext,
    mut control: Control,
    mut konts: Vec<Kont>,
    exec_state: &mut ExecState,
) -> Result<RootResult, KclError> {
    loop {
        control = match control {
            Control::Eval(req) => match step_eval(*req, &mut konts, exec_state, ctx).await {
                Ok(c) => c,
                Err(e) => return Err(unwind_error(e, &mut konts, exec_state)),
            },
            Control::Apply(applied) => match konts.pop() {
                None => {
                    return Ok(RootResult::Done(applied));
                }
                Some(kont) => match step_apply(kont, applied, &mut konts, exec_state, ctx).await {
                    Ok(c) => c,
                    Err(e) => return Err(unwind_error(e, &mut konts, exec_state)),
                },
            },
            Control::Return(cf) => match unwind_return(cf, &mut konts, exec_state, ctx).await {
                Ok(ReturnUnwind::Resume(c)) => c,
                Ok(ReturnUnwind::Root(cf)) => return Ok(RootResult::Exited(cf)),
                Err(e) => return Err(unwind_error(e, &mut konts, exec_state)),
            },
            Control::Exit(cf) => {
                let cf = unwind_exit(cf, &mut konts, exec_state)?;
                return Ok(RootResult::Exited(cf));
            }
        };
    }
}

/// How a `Return` unwind ended.
// Variant sizes intentionally differ, like the machine's other frames.
#[allow(clippy::large_enum_variant)]
enum ReturnUnwind {
    /// A call boundary absorbed the Return; the machine resumes with this.
    Resume(Control),
    /// No call boundary remained: the fresh root receives the Return (a
    /// callback body's native call frame absorbs it; run_block rejects it
    /// for a root block).
    Root(KclValueControlFlow),
}

/// Unwind continuations for a KCL 3.0 `return`: cleanups on non-boundaries, the
/// balancing sketch cleanup (including its feature-tree `GroupEnd`) on sketch
/// bodies, and then -- unlike unwind_exit -- stop at the nearest call boundary,
/// which absorbs the value as the function's normal result via call_finish
/// (tags and return-type coercion apply) and resumes the machine from the
/// boundary's completion. Errors propagate to the caller, which unwinds the
/// remaining continuations.
async fn unwind_return(
    cf: KclValueControlFlow,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<ReturnUnwind, KclError> {
    while let Some(kont) = konts.pop() {
        match kont {
            Kont::CallBoundary(boundary) => {
                exec_state.mod_local.machine_call_depth = exec_state.mod_local.machine_call_depth.saturating_sub(1);
                let BoundaryState {
                    state,
                    fn_src,
                    fn_name,
                    callsite,
                    expects: _,
                    completion,
                } = *boundary;
                return match completion {
                    BoundaryCompletion::CallExpr { fn_meta } => {
                        let finished = fn_src
                            .call_finish(state, Ok(Some(cf)), exec_state)
                            .map_err(|e| e.add_unwind_location(fn_name.clone(), callsite))?;
                        Ok(ReturnUnwind::Resume(finish_call_value(
                            finished, fn_name, callsite, fn_meta,
                        )?))
                    }
                    BoundaryCompletion::Callback => {
                        let finished = fn_src.call_finish(state, Ok(Some(cf)), exec_state)?;
                        Ok(ReturnUnwind::Resume(match finished {
                            // Only an Exit can pass through call_finish; the
                            // Return was absorbed.
                            Some(cf) if cf.is_some_return() => Control::Exit(cf),
                            Some(cf) => {
                                resume_drive(Feed::Callback(Some(cf.into_value())), konts, exec_state, ctx).await?
                            }
                            None => resume_drive(Feed::Callback(None), konts, exec_state, ctx).await?,
                        }))
                    }
                };
            }
            Kont::SketchBody(sb) => {
                sketch_body_cleanup(*sb, exec_state);
                // Balance the GroupBegin operation pushed by scene_setup,
                // matching the exit path.
                exec_state.push_op(Operation::GroupEnd);
            }
            other => cleanup(other, exec_state)?,
        }
    }
    Ok(ReturnUnwind::Root(cf))
}

/// Unwind every continuation for an `exit()`: cleanups on non-boundaries, the
/// exit flavor of `call_finish` on call boundaries (restore ambient flags, pop
/// the callee env, finalize the operation -- skipping tags and coercion), and
/// the balancing sketch cleanup (including its feature-tree `GroupEnd`).
fn unwind_exit(
    mut cf: KclValueControlFlow,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
) -> Result<KclValueControlFlow, KclError> {
    while let Some(kont) = konts.pop() {
        match kont {
            Kont::CallBoundary(b) => {
                exec_state.mod_local.machine_call_depth = exec_state.mod_local.machine_call_depth.saturating_sub(1);
                // The exit flavor: call_finish's Exit bypass runs restores,
                // pop_env, and op finalization, then returns the Exit
                // unchanged before tags/coercion.
                match b.fn_src.call_finish(b.state, Ok(Some(cf)), exec_state) {
                    Ok(Some(v)) => cf = v,
                    Ok(None) => {
                        return Err(KclError::new_internal(KclErrorDetails::new(
                            "machine executor: exit vanished at a call boundary".to_owned(),
                            vec![b.callsite],
                        )));
                    }
                    Err(e) => {
                        // pop_env failed; propagate through the error unwind.
                        let e = match &b.completion {
                            BoundaryCompletion::CallExpr { .. } => e.add_unwind_location(b.fn_name.clone(), b.callsite),
                            BoundaryCompletion::Callback => e,
                        };
                        return Err(unwind_error(e, konts, exec_state));
                    }
                }
            }
            Kont::SketchBody(sb) => {
                sketch_body_cleanup(*sb, exec_state);
                // Balance the GroupBegin operation pushed by scene_setup so
                // the feature tree stays well-formed (matches the recursive
                // executor's exit path; its error path deliberately skips
                // this).
                exec_state.push_op(Operation::GroupEnd);
            }
            other => {
                if let Err(cleanup_err) = cleanup(other, exec_state) {
                    return Err(unwind_error(cleanup_err, konts, exec_state));
                }
            }
        }
    }
    Ok(cf)
}

/// Unwind every continuation for an error: cleanups on non-boundaries,
/// `call_finish` with the error on call boundaries (which finalizes the
/// operation with `is_error = true`), and per-boundary backtrace decoration
/// via `add_unwind_location`, innermost first -- exactly like errors bubbling
/// out of the recursive executor's nested calls.
fn unwind_error(mut e: KclError, konts: &mut Vec<Kont>, exec_state: &mut ExecState) -> KclError {
    while let Some(kont) = konts.pop() {
        match kont {
            Kont::CallBoundary(b) => {
                exec_state.mod_local.machine_call_depth = exec_state.mod_local.machine_call_depth.saturating_sub(1);
                match b.fn_src.call_finish(b.state, Err(e), exec_state) {
                    Err(finished) => {
                        // Callback boundaries get no backtrace entry, matching
                        // the recursive executor (call_kw callers do not call
                        // add_unwind_location).
                        e = match &b.completion {
                            BoundaryCompletion::CallExpr { .. } => {
                                finished.add_unwind_location(b.fn_name.clone(), b.callsite)
                            }
                            BoundaryCompletion::Callback => finished,
                        };
                    }
                    Ok(_) => {
                        e = KclError::new_internal(KclErrorDetails::new(
                            "machine executor: error vanished at a call boundary".to_owned(),
                            vec![b.callsite],
                        ));
                    }
                }
            }
            Kont::SketchBody(sb) => {
                // The recursive executor's error path restores ambient sketch
                // state and pops both envs but does NOT push the balancing
                // GroupEnd (a historical asymmetry with the exit path).
                sketch_body_cleanup(*sb, exec_state);
            }
            other => {
                // Keep the original error; a boundary-in-cleanup internal
                // error is secondary here and debug_assert covers development.
                let _ = cleanup(other, exec_state);
            }
        }
    }
    e
}

/// Restore ambient state for a non-boundary continuation popped during
/// unwinding. Continuations without ambient state are simply dropped.
///
/// Boundary continuations are handled by the unwind loops before they get
/// here; one reaching this function is a machine bug, reported as an internal
/// error rather than a panic.
fn cleanup(kont: Kont, exec_state: &mut ExecState) -> Result<(), KclError> {
    match kont {
        Kont::BlockSeq { in_flight, .. } => {
            if let InFlight::Declaration { prev_being_declared } = in_flight {
                exec_state.mod_local.being_declared = prev_being_declared;
            }
        }
        Kont::PipeSeq { saved_pipe_value, .. } => {
            exec_state.mod_local.pipe_value = saved_pipe_value;
        }
        Kont::IfArmDone { env_pushed, .. } => {
            // Pop the arm's scope environment (KCL 3.0) when unwinding
            // out of the arm via error, exit(), or return.
            if env_pushed {
                exec_state.mut_stack().pop_env()?;
            }
        }
        // Evaluation-only continuations hold no ambient state.
        Kont::BinaryLhsDone { .. }
        | Kont::BinaryRhsDone { .. }
        | Kont::UnaryDone { .. }
        | Kont::ArrayElems { .. }
        | Kont::ObjectProps { .. }
        | Kont::RangeStartDone { .. }
        | Kont::RangeEndDone { .. }
        | Kont::LegacyMemberPropDone { .. }
        | Kont::LegacyMemberObjDone { .. }
        | Kont::MemberObjDone { .. }
        | Kont::MemberPropDone { .. }
        | Kont::IfCondDone { .. }
        | Kont::AscribeDone { .. }
        | Kont::LabelDone { .. }
        | Kont::PipeFirstDone { .. }
        | Kont::CallArgs(_)
        | Kont::SketchArgs(_)
        // Resumable-builtin loop state holds no ambient state: on unwind it
        // is simply dropped, which is how exit() and errors inside callbacks
        // abandon the rest of the iteration.
        | Kont::Resume(_) => {}
        // Handled by the unwind loops directly.
        Kont::CallBoundary(_) | Kont::SketchBody(_) => {
            let message = "machine executor: boundary continuation reached non-boundary cleanup";
            debug_assert!(false, "{message}");
            return Err(KclError::new_internal(KclErrorDetails::new(message.to_owned(), Vec::new())));
        }
    }
    Ok(())
}

/// Restore the ambient state wrapped around a sketch-block body, in the
/// recursive executor's exact order: pop the body's child environment (if it
/// was pushed), restore sketch_mode, take the sketch-block state back out,
/// pop the sketch2 alias environment. Finalization is skipped entirely --
/// that's the point on unwind paths.
fn sketch_body_cleanup(sb: SketchBodyState, exec_state: &mut ExecState) {
    if sb.child_env_pushed {
        let _ = exec_state.mut_stack().pop_env();
    }
    exec_state.mod_local.sketch_mode = sb.saved_sketch_mode;
    let _ = std::mem::replace(&mut exec_state.mod_local.sketch_block, sb.saved_sketch_block);
    let _ = exec_state.mut_stack().pop_env();
}

/// One eval step: the recursive executor's `execute_expr` dispatch, pushing a
/// continuation and descending instead of awaiting recursively. Leaves
/// produce `Apply` directly (some via flat awaits).
async fn step_eval(
    req: EvalRequest,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let EvalRequest {
        node,
        metadata,
        decl_name,
        annotations,
    } = req;
    let expr = match node {
        EvalNode::Expr(e) => e,
        EvalNode::BinaryPart(part) => {
            // BinaryPart is a strict subset of Expr's variants; route through
            // the same arms.
            match part {
                BinaryPart::Literal(literal) => {
                    return Ok(Control::Apply(Applied::Value(KclValue::from_literal(
                        literal.as_ref().clone(),
                        exec_state,
                    ))));
                }
                BinaryPart::Name(name) => {
                    let value = ctx.resolve_name_for_eval(&name, &metadata, exec_state).await?;
                    return Ok(Control::Apply(Applied::Value(value)));
                }
                BinaryPart::BinaryExpression(node) => Expr::BinaryExpression(node),
                BinaryPart::CallExpressionKw(node) => Expr::CallExpressionKw(node),
                BinaryPart::UnaryExpression(node) => Expr::UnaryExpression(node),
                BinaryPart::MemberExpression(node) => Expr::MemberExpression(node),
                BinaryPart::ArrayExpression(node) => Expr::ArrayExpression(node),
                BinaryPart::ArrayRangeExpression(node) => Expr::ArrayRangeExpression(node),
                BinaryPart::ObjectExpression(node) => Expr::ObjectExpression(node),
                BinaryPart::IfExpression(node) => Expr::IfExpression(node),
                BinaryPart::AscribedExpression(node) => Expr::AscribedExpression(node),
                BinaryPart::SketchVar(node) => Expr::SketchVar(node),
            }
        }
    };

    match expr {
        // ---- leaves ----
        Expr::None(none) => Ok(Control::Apply(Applied::Value(KclValue::from(&none)))),
        Expr::Literal(literal) => Ok(Control::Apply(Applied::Value(KclValue::from_literal(
            literal.as_ref().clone(),
            exec_state,
        )))),
        Expr::TagDeclarator(tag) => {
            let value = tag.execute(exec_state).await?;
            Ok(Control::Apply(Applied::Value(value)))
        }
        Expr::Name(name) => {
            let value = ctx.resolve_name_for_eval(&name, &metadata, exec_state).await?;
            Ok(Control::Apply(Applied::Value(value)))
        }
        Expr::FunctionExpression(function_expression) => {
            let statement_kind = match &decl_name {
                Some(name) => crate::execution::StatementKind::Declaration { name },
                None => crate::execution::StatementKind::Expression,
            };
            let value = ctx
                .create_function_closure(
                    &function_expression,
                    &annotations,
                    &metadata,
                    statement_kind,
                    exec_state,
                )
                .await?;
            Ok(Control::Apply(Applied::Value(value)))
        }
        Expr::PipeSubstitution(pipe_substitution) => match &decl_name {
            Some(name) => {
                let message =
                    format!("you cannot declare variable {name} as %, because % can only be used in function calls");
                Err(KclError::new_semantic(KclErrorDetails::new(
                    message,
                    vec![pipe_substitution.as_ref().into()],
                )))
            }
            None => match exec_state.mod_local.pipe_value.clone() {
                Some(x) => Ok(Control::Apply(Applied::Value(x))),
                None => Err(KclError::new_semantic(KclErrorDetails::new(
                    "cannot use % outside a pipe expression".to_owned(),
                    vec![pipe_substitution.as_ref().into()],
                ))),
            },
        },
        Expr::SketchVar(expr) => {
            let value = expr.get_result(exec_state, ctx).await?;
            Ok(Control::Apply(Applied::Value(value)))
        }

        // ---- aggregates ----
        Expr::ArrayExpression(node) => {
            let node = node.arc();
            if node.elements.is_empty() {
                return Ok(Control::Apply(Applied::Value(KclValue::HomArray {
                    value: Vec::new(),
                    ty: RuntimeType::Primitive(PrimitiveType::Any),
                })));
            }
            let first = EvalRequest::expr(&node.elements[0]);
            konts.push(Kont::ArrayElems {
                node,
                index: 0,
                done: Vec::new(),
            });
            Ok(Control::Eval(Box::new(first)))
        }
        Expr::ObjectExpression(node) => {
            let node = node.arc();
            if node.properties.is_empty() {
                return Ok(Control::Apply(Applied::Value(KclValue::Object {
                    value: KclObjectFields::with_capacity(0),
                    meta: vec![Metadata {
                        source_range: SourceRange::from(node.as_ref()),
                    }],
                    constrainable: false,
                    object_kind: crate::execution::kcl_value::KclObjectKind::Default,
                })));
            }
            let first = EvalRequest::expr(&node.properties[0].value);
            konts.push(Kont::ObjectProps {
                node,
                index: 0,
                done: KclObjectFields::default(),
            });
            Ok(Control::Eval(Box::new(first)))
        }
        Expr::ArrayRangeExpression(node) => {
            let node = node.arc();
            let start = EvalRequest::expr(&node.start_element);
            konts.push(Kont::RangeStartDone { node });
            Ok(Control::Eval(Box::new(start)))
        }

        // ---- operators ----
        Expr::BinaryExpression(node) => {
            let node = node.arc();
            let left = EvalRequest::binary_part(&node.left);
            konts.push(Kont::BinaryLhsDone { node });
            Ok(Control::Eval(Box::new(left)))
        }
        Expr::UnaryExpression(node) => {
            let node = node.arc();
            let operand = EvalRequest::binary_part(&node.argument);
            konts.push(Kont::UnaryDone { node });
            Ok(Control::Eval(Box::new(operand)))
        }

        // ---- access / control ----
        Expr::MemberExpression(node) => {
            let node = node.arc();
            if exec_state.entry_point_version_is_v3_or_higher() {
                // KCL 3.0: evaluate in source order, the object before the
                // property.
                let object = EvalRequest::expr(&node.object);
                konts.push(Kont::MemberObjDone { node });
                Ok(Control::Eval(Box::new(object)))
            } else if node.computed {
                // Pre-KCL-3.0: the property is evaluated before the object.
                // Preserved because ids and engine commands are
                // order-dependent.
                let prop = EvalRequest::expr(&node.property);
                konts.push(Kont::LegacyMemberPropDone { node });
                Ok(Control::Eval(Box::new(prop)))
            } else {
                // Non-computed properties are identifier names, not evaluated.
                let property = Property::from_static_name(&node.property, SourceRange::from(node.as_ref()))?;
                let object = EvalRequest::expr(&node.object);
                konts.push(Kont::LegacyMemberObjDone { node, property });
                Ok(Control::Eval(Box::new(object)))
            }
        }
        Expr::IfExpression(node) => {
            let node = node.arc();
            let cond = EvalRequest {
                node: EvalNode::Expr((*node.cond).clone()),
                metadata: Metadata::from(node.as_ref()),
                decl_name: None,
                annotations: Vec::new(),
            };
            konts.push(Kont::IfCondDone { node, arm: 0 });
            Ok(Control::Eval(Box::new(cond)))
        }
        Expr::AscribedExpression(node) => {
            let node = node.arc();
            let inner = EvalRequest {
                node: EvalNode::Expr(node.expr.clone()),
                metadata: Metadata {
                    source_range: SourceRange::from(node.as_ref()),
                },
                decl_name: None,
                annotations: Vec::new(),
            };
            konts.push(Kont::AscribeDone { node });
            Ok(Control::Eval(Box::new(inner)))
        }
        Expr::LabelledExpression(node) => {
            let node = node.arc();
            let inner = EvalRequest {
                node: EvalNode::Expr(node.expr.clone()),
                metadata,
                // The recursive executor forwards its statement_kind to the
                // labelled inner expression, so `x = label: fn() {}` still
                // sees the declaration name.
                decl_name,
                annotations: Vec::new(),
            };
            konts.push(Kont::LabelDone { node });
            Ok(Control::Eval(Box::new(inner)))
        }

        // ---- pipes ----
        Expr::PipeExpression(node) => {
            let node = node.arc();
            let Some(first) = node.body.first() else {
                return Err(KclError::new_semantic(KclErrorDetails::new(
                    "Pipe expressions cannot be empty".to_owned(),
                    vec![SourceRange::from(node.as_ref())],
                )));
            };
            // The first element is evaluated under the PARENT's pipe value
            // (a nested pipe expression sees the outer %).
            let first = EvalRequest::expr(first);
            konts.push(Kont::PipeFirstDone { node });
            Ok(Control::Eval(Box::new(first)))
        }

        // ---- calls ----
        Expr::CallExpressionKw(node) => {
            let node = node.arc();
            start_call(node, konts, exec_state, ctx).await
        }

        // ---- sketch blocks ----
        Expr::SketchBlock(node) => {
            let node = node.arc();
            start_sketch_block(node, konts, exec_state, ctx).await
        }
    }
}

/// One apply step: hand a finished value to the innermost continuation. The
/// recursive executor's post-await code for each recursion site.
async fn step_apply(
    kont: Kont,
    applied: Applied,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    match kont {
        Kont::BlockSeq { .. } => step_block(kont, Some(applied), konts, exec_state, ctx).await,

        Kont::BinaryLhsDone { node } => {
            let left = applied.expect_value()?;
            let right = EvalRequest::binary_part(&node.right);
            konts.push(Kont::BinaryRhsDone { node, left });
            Ok(Control::Eval(Box::new(right)))
        }
        Kont::BinaryRhsDone { node, left } => {
            let right = applied.expect_value()?;
            let value = node.apply_operator(exec_state, ctx, left, right).await?;
            Ok(Control::Apply(Applied::Value(value)))
        }
        Kont::UnaryDone { node } => {
            let operand = applied.expect_value()?;
            let value = node.apply_unary(operand, exec_state)?;
            Ok(Control::Apply(Applied::Value(value)))
        }

        Kont::ArrayElems { node, index, mut done } => {
            done.push(applied.expect_value()?);
            let next = index + 1;
            if next < node.elements.len() {
                let elem = EvalRequest::expr(&node.elements[next]);
                konts.push(Kont::ArrayElems {
                    node,
                    index: next,
                    done,
                });
                Ok(Control::Eval(Box::new(elem)))
            } else {
                Ok(Control::Apply(Applied::Value(KclValue::HomArray {
                    value: done,
                    ty: RuntimeType::Primitive(PrimitiveType::Any),
                })))
            }
        }
        Kont::ObjectProps { node, index, mut done } => {
            let value = applied.expect_value()?;
            done.insert(node.properties[index].key.name.clone(), value);
            let next = index + 1;
            if next < node.properties.len() {
                let prop = EvalRequest::expr(&node.properties[next].value);
                konts.push(Kont::ObjectProps {
                    node,
                    index: next,
                    done,
                });
                Ok(Control::Eval(Box::new(prop)))
            } else {
                Ok(Control::Apply(Applied::Value(KclValue::Object {
                    value: done,
                    meta: vec![Metadata {
                        source_range: SourceRange::from(node.as_ref()),
                    }],
                    constrainable: false,
                    object_kind: crate::execution::kcl_value::KclObjectKind::Default,
                })))
            }
        }
        Kont::RangeStartDone { node } => {
            let start = applied.expect_value()?;
            // Match the recursive executor: a bad start is reported before
            // the end element is ever evaluated.
            node.validate_range_start(&start)?;
            let end = EvalRequest::expr(&node.end_element);
            konts.push(Kont::RangeEndDone { node, start });
            Ok(Control::Eval(Box::new(end)))
        }
        Kont::RangeEndDone { node, start } => {
            let end = applied.expect_value()?;
            let value = node.build_range(start, end, exec_state)?;
            Ok(Control::Apply(Applied::Value(value)))
        }

        Kont::LegacyMemberPropDone { node } => {
            let prop_value = applied.expect_value()?;
            let property = Property::from_value(prop_value, SourceRange::from(node.as_ref()))?;
            let object = EvalRequest::expr(&node.object);
            konts.push(Kont::LegacyMemberObjDone { node, property });
            Ok(Control::Eval(Box::new(object)))
        }
        Kont::LegacyMemberObjDone { node, property } => {
            let object = applied.expect_value()?;
            let cf = node.apply_member(object, property, exec_state, ctx).await?;
            // apply_member only ever produces Continue values.
            Ok(Control::Apply(Applied::Value(cf.into_value())))
        }
        Kont::MemberObjDone { node } => {
            let object = applied.expect_value()?;
            if node.computed {
                let prop = EvalRequest::expr(&node.property);
                konts.push(Kont::MemberPropDone { node, object });
                Ok(Control::Eval(Box::new(prop)))
            } else {
                // Non-computed properties are identifier names, not evaluated.
                let property = Property::from_static_name(&node.property, SourceRange::from(node.as_ref()))?;
                let cf = node.apply_member(object, property, exec_state, ctx).await?;
                Ok(Control::Apply(Applied::Value(cf.into_value())))
            }
        }
        Kont::MemberPropDone { node, object } => {
            let prop_value = applied.expect_value()?;
            let property = Property::from_value(prop_value, SourceRange::from(node.as_ref()))?;
            let cf = node.apply_member(object, property, exec_state, ctx).await?;
            // apply_member only ever produces Continue values.
            Ok(Control::Apply(Applied::Value(cf.into_value())))
        }

        Kont::IfCondDone { node, arm } => {
            let cond_value = applied.expect_value()?;
            if cond_value.get_bool()? {
                let block = if arm == 0 {
                    BlockRef::Program(node.then_val.arc())
                } else {
                    BlockRef::Program(node.else_ifs[arm - 1].then_val.arc())
                };
                // An error here leaves no env pushed and IfArmDone unpushed,
                // so unwinding stays balanced.
                let env_pushed = crate::execution::exec_ast::if_arm_scope_begin(exec_state)?;
                konts.push(Kont::IfArmDone { node, env_pushed });
                push_block(block, BodyType::Block, konts);
                step_block_kick(konts, exec_state, ctx).await
            } else if arm < node.else_ifs.len() {
                let cond = EvalRequest {
                    node: EvalNode::Expr(node.else_ifs[arm].cond.clone()),
                    metadata: Metadata::from(node.as_ref()),
                    decl_name: None,
                    annotations: Vec::new(),
                };
                konts.push(Kont::IfCondDone { node, arm: arm + 1 });
                Ok(Control::Eval(Box::new(cond)))
            } else {
                let block = BlockRef::Program(node.final_else.arc());
                let env_pushed = crate::execution::exec_ast::if_arm_scope_begin(exec_state)?;
                konts.push(Kont::IfArmDone { node, env_pushed });
                push_block(block, BodyType::Block, konts);
                step_block_kick(konts, exec_state, ctx).await
            }
        }
        Kont::IfArmDone { node, env_pushed } => {
            // Pop the arm scope before unwrapping; values escaping the arm
            // stay valid because environments that may still be referenced
            // are preserved.
            if env_pushed {
                exec_state.mut_stack().pop_env()?;
            }
            let block_result = applied.expect_block()?;
            // Blocks used as if arms must end in an expression (enforced by
            // the parser), so this is always Some.
            let Some(cf) = block_result else {
                return Err(KclError::new_internal(KclErrorDetails::new(
                    "if-expression arm produced no value".to_owned(),
                    vec![SourceRange::from(node.as_ref())],
                )));
            };
            Ok(Control::Apply(Applied::Value(cf.into_value())))
        }

        Kont::AscribeDone { node } => {
            let value = applied.expect_value()?;
            let value = crate::execution::exec_ast::apply_ascription(
                &value,
                &node.ty,
                exec_state,
                ctx,
                SourceRange::from(node.as_ref()),
            )
            .await?;
            Ok(Control::Apply(Applied::Value(value)))
        }
        Kont::LabelDone { node } => {
            let value = applied.expect_value()?;
            exec_state
                .mut_stack()
                .add(node.label.name.clone(), value.clone(), SourceRange::from(node.as_ref()))?;
            // TODO this lets us use the label as a variable name, but not as a tag in most cases
            Ok(Control::Apply(Applied::Value(value)))
        }

        Kont::PipeFirstDone { node } => {
            let output = applied.expect_value()?;
            // Now that the first element is evaluated, following elements use
            // it as %; the parent's pipe value is restored when this pipe
            // finishes (or unwinds).
            let saved_pipe_value = exec_state.mod_local.pipe_value.replace(output);
            pipe_advance(node, 1, saved_pipe_value, konts, exec_state)
        }
        Kont::PipeSeq {
            node,
            index,
            saved_pipe_value,
        } => {
            let output = applied.expect_value()?;
            exec_state.mod_local.pipe_value = Some(output);
            pipe_advance(node, index + 1, saved_pipe_value, konts, exec_state)
        }

        Kont::CallArgs(state) => call_args_step(*state, applied, konts, exec_state, ctx).await,
        Kont::CallBoundary(boundary) => {
            exec_state.mod_local.machine_call_depth = exec_state.mod_local.machine_call_depth.saturating_sub(1);
            let BoundaryState {
                state,
                fn_src,
                fn_name,
                callsite,
                expects,
                completion,
            } = *boundary;
            let result = match expects {
                BoundaryExpects::KclBlock => {
                    // Read __return from the callee env (still pushed).
                    let block_result = applied.expect_block()?;
                    fn_src.kcl_body_result(Ok(block_result), exec_state)
                }
                BoundaryExpects::StdValue => {
                    let value = applied.expect_value()?;
                    Ok(Some(value.continue_()))
                }
            };
            match completion {
                BoundaryCompletion::CallExpr { fn_meta } => {
                    let finished = fn_src
                        .call_finish(state, result, exec_state)
                        .map_err(|e| e.add_unwind_location(fn_name.clone(), callsite))?;
                    finish_call_value(finished, fn_name, callsite, fn_meta)
                }
                BoundaryCompletion::Callback => {
                    let finished = fn_src.call_finish(state, result, exec_state)?;
                    match finished {
                        Some(cf) if cf.is_some_return() => Ok(Control::Exit(cf)),
                        Some(cf) => resume_drive(Feed::Callback(Some(cf.into_value())), konts, exec_state, ctx).await,
                        None => resume_drive(Feed::Callback(None), konts, exec_state, ctx).await,
                    }
                }
            }
        }

        Kont::SketchArgs(state) => sketch_args_step(*state, applied, konts, exec_state, ctx).await,
        Kont::SketchBody(state) => sketch_body_finish(*state, applied, exec_state, ctx).await,
        Kont::Resume(_) => Err(KclError::new_internal(KclErrorDetails::new(
            "machine executor: a value applied directly to resumable-builtin loop state".to_owned(),
            Vec::new(),
        ))),
    }
}

/// Route a control-flow value produced by shared (recursive-style) helper code
/// to the matching machine unwind: a KCL 3.0 `Return` unwinds to the nearest
/// call boundary, anything else is an `Exit`.
fn unwind_control(cf: KclValueControlFlow) -> Control {
    if cf.is_return() {
        Control::Return(cf)
    } else {
        Control::Exit(cf)
    }
}

/// Push a block-sequencing continuation.
fn push_block(block: BlockRef, body_type: BodyType, konts: &mut Vec<Kont>) {
    konts.push(Kont::BlockSeq {
        block,
        body_type,
        index: 0,
        in_flight: InFlight::None,
        last: None,
    });
}

/// Immediately step the just-pushed block so it starts its first statement
/// (or completes immediately when empty).
async fn step_block_kick(
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let Some(kont) = konts.pop() else {
        let message = "machine executor: step_block_kick with no continuation on the stack";
        debug_assert!(false, "{message}");
        return Err(KclError::new_internal(KclErrorDetails::new(
            message.to_owned(),
            Vec::new(),
        )));
    };
    step_block(kont, None, konts, exec_state, ctx).await
}

/// Advance a block: apply the in-flight statement's value (if any), then run
/// statements until one needs an expression evaluated (returning `Eval`) or
/// the block completes (returning `Apply(Block(last))`). Flat statements
/// (imports, type declarations) execute inline.
async fn step_block(
    kont: Kont,
    applied: Option<Applied>,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let Kont::BlockSeq {
        block,
        body_type,
        mut index,
        mut in_flight,
        mut last,
    } = kont
    else {
        return Err(KclError::new_internal(KclErrorDetails::new(
            "machine executor: step_block on a non-block continuation".to_owned(),
            Vec::new(),
        )));
    };

    // Finish the in-flight statement.
    if let Some(applied) = applied {
        let value = applied.expect_value()?;
        match std::mem::replace(&mut in_flight, InFlight::None) {
            InFlight::None => {
                return Err(KclError::new_internal(KclErrorDetails::new(
                    "machine executor: value applied to a block with no statement in flight".to_owned(),
                    Vec::new(),
                )));
            }
            InFlight::Expression => {
                // Exit control flow never arrives here as a value: it takes
                // the Exit unwind path instead. A normal value continues.
                last = Some(value.continue_());
                index += 1;
            }
            InFlight::Declaration { prev_being_declared } => {
                // Declaration over, so unset this context.
                exec_state.mod_local.being_declared = prev_being_declared;
                let BodyItem::VariableDeclaration(variable_declaration) = &block.body()[index] else {
                    return Err(KclError::new_internal(KclErrorDetails::new(
                        "machine executor: in-flight declaration is not a declaration".to_owned(),
                        Vec::new(),
                    )));
                };
                let rhs = ctx.bind_variable_declaration(variable_declaration, value, body_type, exec_state)?;
                // Variable declaration can be the return value of a module.
                last = matches!(body_type, BodyType::Root).then_some(rhs.continue_());
                index += 1;
            }
            InFlight::Return => {
                let BodyItem::ReturnStatement(return_statement) = &block.body()[index] else {
                    return Err(KclError::new_internal(KclErrorDetails::new(
                        "machine executor: in-flight return is not a return".to_owned(),
                        Vec::new(),
                    )));
                };
                if exec_state.entry_point_version_is_v3_or_higher() {
                    // KCL 3.0: early return. The rest of this block is
                    // abandoned (this continuation is already popped and holds
                    // no ambient state); unwind_return absorbs the value at the
                    // nearest call boundary.
                    return Ok(Control::Return(value.return_()));
                }
                crate::execution::ExecutorContext::bind_return_value(return_statement, value, exec_state)?;
                last = None;
                index += 1;
            }
        }
    }

    // Run statements until one needs an Eval.
    loop {
        let Some(statement) = block.body().get(index) else {
            // Block complete. (The root's batch flush lives in run_block so
            // that exited roots flush too.)
            return Ok(Control::Apply(Applied::Block(last)));
        };

        match statement {
            BodyItem::ImportStatement(import_stmt) => {
                if exec_state.sketch_mode() {
                    index += 1;
                    continue;
                }
                ctx.exec_import_statement(import_stmt, body_type, exec_state).await?;
                last = None;
                index += 1;
            }
            BodyItem::TypeDeclaration(ty) => {
                if exec_state.sketch_mode() {
                    index += 1;
                    continue;
                }
                ctx.exec_type_declaration(ty, body_type, exec_state).await?;
                last = None;
                index += 1;
            }
            BodyItem::ExpressionStatement(expression_statement) => {
                if exec_state.sketch_mode()
                    && crate::execution::exec_ast::sketch_mode_should_skip(&expression_statement.expression)
                {
                    index += 1;
                    continue;
                }
                let req = EvalRequest {
                    node: EvalNode::Expr(expression_statement.expression.clone()),
                    metadata: Metadata::from(expression_statement),
                    decl_name: None,
                    annotations: Vec::new(),
                };
                konts.push(Kont::BlockSeq {
                    block,
                    body_type,
                    index,
                    in_flight: InFlight::Expression,
                    last,
                });
                return Ok(Control::Eval(Box::new(req)));
            }
            BodyItem::VariableDeclaration(variable_declaration) => {
                if exec_state.sketch_mode()
                    && crate::execution::exec_ast::sketch_mode_should_skip(&variable_declaration.declaration.init)
                {
                    index += 1;
                    continue;
                }
                let var_name = variable_declaration.declaration.id.name.to_string();
                let source_range = SourceRange::from(&variable_declaration.declaration.init);

                // During the evaluation of the variable's RHS, set context that this is all happening inside a variable
                // declaration, for the given name. This helps improve user-facing error messages.
                let lhs = variable_declaration.inner.name().to_owned();
                let prev_being_declared = exec_state.mod_local.being_declared.take();
                exec_state.mod_local.being_declared = Some(lhs);

                let req = EvalRequest {
                    node: EvalNode::Expr(variable_declaration.declaration.init.clone()),
                    metadata: Metadata { source_range },
                    decl_name: Some(var_name),
                    annotations: variable_declaration.outer_attrs.clone(),
                };
                konts.push(Kont::BlockSeq {
                    block,
                    body_type,
                    index,
                    in_flight: InFlight::Declaration { prev_being_declared },
                    last,
                });
                return Ok(Control::Eval(Box::new(req)));
            }
            BodyItem::ReturnStatement(return_statement) => {
                if exec_state.sketch_mode()
                    && crate::execution::exec_ast::sketch_mode_should_skip(&return_statement.argument)
                {
                    index += 1;
                    continue;
                }
                let metadata = Metadata::from(return_statement);
                if matches!(body_type, BodyType::Root) {
                    return Err(KclError::new_semantic(KclErrorDetails::new(
                        "Cannot return from outside a function.".to_owned(),
                        vec![metadata.source_range],
                    )));
                }
                let req = EvalRequest {
                    node: EvalNode::Expr(return_statement.argument.clone()),
                    metadata,
                    decl_name: None,
                    annotations: Vec::new(),
                };
                konts.push(Kont::BlockSeq {
                    block,
                    body_type,
                    index,
                    in_flight: InFlight::Return,
                    last,
                });
                return Ok(Control::Eval(Box::new(req)));
            }
        }
    }
}

/// Advance a pipe to `index`: check the element (tag declarators are not
/// allowed in pipe tails), evaluate it, or finish the pipe when past the end.
fn pipe_advance(
    node: Arc<Node<PipeExpression>>,
    index: usize,
    saved_pipe_value: Option<KclValue>,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
) -> Result<Control, KclError> {
    if index >= node.body.len() {
        // Safe to unwrap here, because pipe_value always has something pushed in when the pipe starts.
        let final_output = exec_state.mod_local.pipe_value.take().ok_or_else(|| {
            KclError::new_internal(KclErrorDetails::new(
                "machine executor: pipe finished with no pipe value".to_owned(),
                vec![SourceRange::from(node.as_ref())],
            ))
        })?;
        // Restore the previous pipe value.
        exec_state.mod_local.pipe_value = saved_pipe_value;
        return Ok(Control::Apply(Applied::Value(final_output)));
    }
    let expression = &node.body[index];
    if let Expr::TagDeclarator(_) = expression {
        let e = KclError::new_semantic(KclErrorDetails::new(
            format!("This cannot be in a PipeExpression: {expression:?}"),
            vec![expression.into()],
        ));
        // Restore before erroring; the continuation holding the saved value
        // hasn't been pushed yet.
        exec_state.mod_local.pipe_value = saved_pipe_value;
        return Err(e);
    }
    let req = EvalRequest::expr(expression);
    konts.push(Kont::PipeSeq {
        node,
        index,
        saved_pipe_value,
    });
    Ok(Control::Eval(Box::new(req)))
}

/// Begin a call: resolve the callee (before evaluating arguments, exactly
/// like the recursive executor), then start evaluating arguments -- or
/// dispatch immediately for zero-argument calls.
async fn start_call(
    node: Arc<Node<CallExpressionKw>>,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let callsite: SourceRange = SourceRange::from(node.as_ref());

    // Resolve the function before evaluating arguments so calls can mutate
    // exec_state without holding a memory borrow.
    let func: KclValue = node.callee.get_result(exec_state, ctx).await?;

    let Some(fn_src) = func.as_function() else {
        return Err(KclError::new_semantic(KclErrorDetails::new(
            "cannot call this because it isn't a function".to_string(),
            vec![callsite],
        )));
    };
    let fn_src = fn_src.clone();
    let fn_meta = match &func {
        KclValue::Function { meta, .. } => meta.clone(),
        _ => Vec::new(),
    };

    let state = CallArgsState {
        node,
        fn_src,
        fn_meta,
        cursor: None,
        unlabeled: Vec::new(),
        labeled: IndexMap::new(),
    };

    // Evaluate the unlabeled first param, if any exists; otherwise the
    // labeled arguments in order; otherwise dispatch with no arguments.
    if let Some(arg_expr) = &state.node.unlabeled {
        let req = EvalRequest::expr(arg_expr);
        konts.push(Kont::CallArgs(Box::new(state)));
        Ok(Control::Eval(Box::new(req)))
    } else if !state.node.arguments.is_empty() {
        let mut state = state;
        state.cursor = Some(0);
        let req = EvalRequest::expr(&state.node.arguments[0].arg);
        konts.push(Kont::CallArgs(Box::new(state)));
        Ok(Control::Eval(Box::new(req)))
    } else {
        dispatch_call(state, konts, exec_state, ctx).await
    }
}

/// Record an evaluated argument and evaluate the next, or dispatch the call
/// when all arguments are done.
async fn call_args_step(
    mut state: CallArgsState,
    applied: Applied,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let value = applied.expect_value()?;
    match state.cursor {
        None => {
            let arg_expr = state.node.unlabeled.as_ref().ok_or_else(|| {
                KclError::new_internal(KclErrorDetails::new(
                    "machine executor: unlabeled argument vanished".to_owned(),
                    vec![SourceRange::from(state.node.as_ref())],
                ))
            })?;
            let source_range = SourceRange::from(arg_expr);
            let label = arg_expr.ident_name().map(str::to_owned);
            state.unlabeled.push((label, Arg::new(value, source_range)));
        }
        Some(i) => {
            let arg_expr = &state.node.arguments[i];
            let source_range = SourceRange::from(&arg_expr.arg);
            let arg = Arg::new(value, source_range);
            match &arg_expr.label {
                Some(l) => {
                    state.labeled.insert(l.name.clone(), arg);
                }
                None => {
                    state
                        .unlabeled
                        .push((arg_expr.arg.ident_name().map(str::to_owned), arg));
                }
            }
        }
    }

    let next = match state.cursor {
        None => 0,
        Some(i) => i + 1,
    };
    if next < state.node.arguments.len() {
        state.cursor = Some(next);
        let req = EvalRequest::expr(&state.node.arguments[next].arg);
        konts.push(Kont::CallArgs(Box::new(state)));
        Ok(Control::Eval(Box::new(req)))
    } else {
        dispatch_call(state, konts, exec_state, ctx).await
    }
}

/// All arguments evaluated: build `Args` and run the call.
async fn dispatch_call(
    state: CallArgsState,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let CallArgsState {
        node,
        fn_src,
        fn_meta,
        cursor: _,
        unlabeled,
        labeled,
    } = state;
    let callsite: SourceRange = SourceRange::from(node.as_ref());
    let fn_name_node = &node.callee;
    let fn_name = Some(fn_name_node.name.name.clone());
    let fn_display_name = Some(fn_name_node.to_string());

    let args = Args::new(
        labeled,
        unlabeled,
        callsite,
        node.node_path.clone(),
        exec_state,
        ctx.clone(),
        fn_name.clone(),
    );

    match machine_call(
        fn_src,
        fn_display_name,
        callsite,
        args,
        BoundaryCompletion::CallExpr { fn_meta },
        konts,
        exec_state,
        ctx,
    )
    .await?
    {
        MachineCallOutcome::Control(control) => Ok(control),
        MachineCallOutcome::LeafCallbackDone(_) => Err(KclError::new_internal(KclErrorDetails::new(
            "machine executor: call expression produced a callback result".to_owned(),
            vec![callsite],
        ))),
    }
}

/// What a machine call produced immediately.
// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
enum MachineCallOutcome {
    /// Continue the machine with this control.
    Control(Control),
    /// A leaf builtin invoked as a callback completed synchronously; the
    /// caller's feed loop hands this to the resume continuation (kept out of
    /// the return path so runs of leaf callbacks use constant native stack).
    LeafCallbackDone(Option<KclValue>),
}

/// Run a call on the machine: setup, then a flat leaf await (plain Rust
/// builtins), a resumable-builtin entry (map/reduce/patternTransform), or a
/// call boundary plus the KCL function body (no native recursion). Shared by
/// call expressions and resumable-builtin callbacks.
#[allow(clippy::too_many_arguments)]
async fn machine_call(
    fn_src: FunctionSource,
    fn_name: Option<String>,
    callsite: SourceRange,
    args: Args<crate::execution::fn_call::Sugary>,
    completion: BoundaryCompletion,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<MachineCallOutcome, KclError> {
    // Guard against runaway recursion: machine-native calls do not grow the
    // native stack, so this is policy, not a native-stack limit. It must
    // behave when raised to 50,000 or more.
    if exec_state.mod_local.machine_call_depth >= ctx.machine_call_depth_limit {
        return Err(KclError::new_max_call_stack(KclErrorDetails::new(
            format!(
                "Call depth limit ({}) exceeded. This usually means a function is recursing without a base case.",
                ctx.machine_call_depth_limit
            ),
            vec![callsite],
        )));
    }
    let decorate = |e: KclError, completion: &BoundaryCompletion| match completion {
        BoundaryCompletion::CallExpr { .. } => e.add_unwind_location(fn_name.clone(), callsite),
        BoundaryCompletion::Callback => e,
    };

    let resumable = match &fn_src.body {
        FunctionBody::Rust(_) => fn_src.std_props.as_ref().and_then(|p| p.resumable),
        FunctionBody::Kcl(_) => None,
    };
    if let Some(kind) = resumable {
        let (call_state, args) = match fn_src.call_setup(&fn_name, exec_state, args, callsite) {
            Ok(x) => x,
            Err(e) => return Err(decorate(e, &completion)),
        };
        exec_state.mod_local.machine_call_depth += 1;
        exec_state.global.machine_depth_high_water = exec_state
            .global
            .machine_depth_high_water
            .max(exec_state.mod_local.machine_call_depth);
        konts.push(Kont::CallBoundary(Box::new(BoundaryState {
            state: call_state,
            fn_src: fn_src.clone(),
            fn_name,
            callsite,
            expects: BoundaryExpects::StdValue,
            completion,
        })));
        // Parse errors after this point unwind through the boundary just
        // pushed, finalizing the operation with is_error = true -- exactly
        // like an error inside the recursive builtin's body.
        return resumable_entry(kind, args, konts, exec_state, ctx)
            .await
            .map(MachineCallOutcome::Control);
    }

    match &fn_src.body {
        FunctionBody::Rust(f) => {
            // A flat leaf: setup, await the builtin, finish.
            let f = *f;
            let (call_state, args) = match fn_src.call_setup(&fn_name, exec_state, args, callsite) {
                Ok(x) => x,
                Err(e) => return Err(decorate(e, &completion)),
            };
            let result = f(exec_state, args).await.map(Some);
            let finished = match fn_src.call_finish(call_state, result, exec_state) {
                Ok(x) => x,
                Err(e) => return Err(decorate(e, &completion)),
            };
            match completion {
                BoundaryCompletion::CallExpr { fn_meta } => {
                    finish_call_value(finished, fn_name, callsite, fn_meta).map(MachineCallOutcome::Control)
                }
                BoundaryCompletion::Callback => match finished {
                    Some(cf) if cf.is_some_return() => Ok(MachineCallOutcome::Control(Control::Exit(cf))),
                    Some(cf) => Ok(MachineCallOutcome::LeafCallbackDone(Some(cf.into_value()))),
                    None => Ok(MachineCallOutcome::LeafCallbackDone(None)),
                },
            }
        }
        FunctionBody::Kcl(_) => {
            let (call_state, args) = match fn_src.call_setup(&fn_name, exec_state, args, callsite) {
                Ok(x) => x,
                Err(e) => return Err(decorate(e, &completion)),
            };
            if let Err(e) = crate::execution::fn_call::assign_args_to_params_kw(&fn_src, args, exec_state) {
                let e = FunctionSource::call_abort_on_arg_binding_failure(call_state, e, exec_state);
                return Err(decorate(e, &completion));
            }
            exec_state.mod_local.machine_call_depth += 1;
            exec_state.global.machine_depth_high_water = exec_state
                .global
                .machine_depth_high_water
                .max(exec_state.mod_local.machine_call_depth);
            let body = BlockRef::FnBody(fn_src.ast.arc());
            konts.push(Kont::CallBoundary(Box::new(BoundaryState {
                state: call_state,
                fn_src,
                fn_name,
                callsite,
                expects: BoundaryExpects::KclBlock,
                completion,
            })));
            push_block(body, BodyType::Block, konts);
            step_block_kick(konts, exec_state, ctx)
                .await
                .map(MachineCallOutcome::Control)
        }
    }
}

/// Start a resumable builtin: parse its arguments (the boundary is already
/// pushed, so failures finalize the operation like a builtin-body error) and
/// run the first callback -- or produce the empty result immediately.
async fn resumable_entry(
    kind: crate::std::ResumableKind,
    args: Args,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let source_range = args.source_range;
    let node_path = args.node_path.clone();
    let resume = match kind {
        crate::std::ResumableKind::Map => {
            let (array, f) = crate::std::array::map_parse_args(&args, exec_state)?;
            ResumeState::Map(MapResume {
                f,
                rest: array.into_iter(),
                done: Vec::new(),
                source_range,
                node_path,
            })
        }
        crate::std::ResumableKind::Reduce => {
            let (array, f, initial) = crate::std::array::reduce_parse_args(&args, exec_state)?;
            konts.push(Kont::Resume(Box::new(ResumeState::Reduce(ReduceResume {
                f,
                rest: array.into_iter(),
                source_range,
                node_path,
            }))));
            // The accumulator rides the callback results: seed it as if a
            // callback had just produced the initial value.
            return resume_drive(Feed::Callback(Some(initial)), konts, exec_state, ctx).await;
        }
        crate::std::ResumableKind::PatternTransform => {
            let (geometry, instances, transform, use_original) =
                crate::std::patterns::pattern_transform_parse_args(&args, exec_state)?;
            crate::std::patterns::pattern_check_instances(instances, source_range)?;
            let geometry = match geometry {
                crate::std::patterns::Patternable3d::Solids(solids) => PatternGeometry::Solids(solids),
                crate::std::patterns::Patternable3d::ImportedGeometry(geometry) => {
                    PatternGeometry::ImportedGeometry(geometry)
                }
            };
            ResumeState::PatternTransform(PatternTransformResume {
                transform,
                instances,
                next_i: 1,
                transforms: Vec::with_capacity(usize::try_from(instances).unwrap_or_default()),
                geometry,
                use_original: use_original.unwrap_or_default(),
                source_range,
                node_path,
                args,
            })
        }
        crate::std::ResumableKind::PatternTransform2d => {
            let (sketches, instances, transform, use_original) =
                crate::std::patterns::pattern_transform_2d_parse_args(&args, exec_state)?;
            crate::std::patterns::pattern_check_instances(instances, source_range)?;
            ResumeState::PatternTransform(PatternTransformResume {
                transform,
                instances,
                next_i: 1,
                transforms: Vec::with_capacity(usize::try_from(instances).unwrap_or_default()),
                geometry: PatternGeometry::Sketches(sketches),
                use_original: use_original.unwrap_or_default(),
                source_range,
                node_path,
                args,
            })
        }
    };
    konts.push(Kont::Resume(Box::new(resume)));
    resume_drive(Feed::Advance, konts, exec_state, ctx).await
}

/// What drives the next resume step.
// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
enum Feed {
    /// Start the builtin's next iteration (entry).
    Advance,
    /// A callback just completed with this result.
    Callback(Option<KclValue>),
}

/// Drive the innermost resume continuation. Loop-based (not recursive) so a
/// run of leaf-builtin callbacks completes in constant native stack: each
/// synchronously-finished callback feeds the next iteration of this loop.
async fn resume_drive(
    mut feed: Feed,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    loop {
        let Some(Kont::Resume(resume)) = konts.pop() else {
            return Err(KclError::new_internal(KclErrorDetails::new(
                "machine executor: callback finished with no resumable builtin awaiting it".to_owned(),
                Vec::new(),
            )));
        };
        let outcome = match feed {
            Feed::Advance => resume_next(*resume, konts, exec_state, ctx).await?,
            Feed::Callback(value) => resume_step(*resume, value, konts, exec_state, ctx).await?,
        };
        match outcome {
            ResumeOutcome::Control(control) => return Ok(control),
            ResumeOutcome::CallbackDone(value) => {
                feed = Feed::Callback(value);
            }
        }
    }
}

// Variant sizes intentionally differ: these are the machine's heap frames,
// and boxing every payload would add an allocation per step.
#[allow(clippy::large_enum_variant)]
enum ResumeOutcome {
    /// The machine continues with this control (a callback was started on
    /// the continuation stack, or the builtin finished and its value was
    /// applied).
    Control(Control),
    /// A leaf-builtin callback completed synchronously; feed this to the
    /// next resume step.
    CallbackDone(Option<KclValue>),
}

/// Record a callback's result, then continue the builtin's iteration.
async fn resume_step(
    resume: ResumeState,
    fed: Option<KclValue>,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<ResumeOutcome, KclError> {
    match resume {
        ResumeState::Map(mut m) => {
            let value = fed.ok_or_else(|| crate::std::array::map_missing_value_error(m.source_range))?;
            m.done.push(value);
            resume_next(ResumeState::Map(m), konts, exec_state, ctx).await
        }
        ResumeState::Reduce(r) => {
            // fed is the new accumulator (or the seeded initial value).
            let accum = fed.ok_or_else(|| crate::std::array::reduce_missing_value_error(r.source_range))?;
            resume_reduce_next(r, accum, konts, exec_state, ctx).await
        }
        ResumeState::PatternTransform(mut p) => {
            let value = fed.ok_or_else(|| crate::std::patterns::transform_missing_value_error(p.source_range))?;
            let transforms = match &p.geometry {
                PatternGeometry::Solids(_) => crate::std::patterns::transforms_from_callback_value::<
                    crate::execution::Solid,
                >(value, p.source_range, exec_state)?,
                PatternGeometry::Sketches(_) => crate::std::patterns::transforms_from_callback_value::<
                    crate::execution::Sketch,
                >(value, p.source_range, exec_state)?,
                PatternGeometry::ImportedGeometry(_) => crate::std::patterns::transforms_from_callback_value::<
                    crate::execution::ImportedGeometry,
                >(value, p.source_range, exec_state)?,
            };
            p.transforms.push(transforms);
            p.next_i += 1;
            resume_next(ResumeState::PatternTransform(p), konts, exec_state, ctx).await
        }
    }
}

/// Start the builtin's next callback, or finish the builtin and apply its
/// value (which lands on the builtin's call boundary).
async fn resume_next(
    resume: ResumeState,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<ResumeOutcome, KclError> {
    match resume {
        ResumeState::Map(mut m) => match m.rest.next() {
            Some(elem) => {
                let f = m.f.clone();
                let callsite = m.source_range;
                let args =
                    crate::std::array::map_callback_args(elem, m.source_range, m.node_path.clone(), exec_state, ctx);
                konts.push(Kont::Resume(Box::new(ResumeState::Map(m))));
                start_callback(f, callsite, args, konts, exec_state, ctx).await
            }
            None => Ok(ResumeOutcome::Control(Control::Apply(Applied::Value(
                crate::std::array::map_result(m.done),
            )))),
        },
        ResumeState::PatternTransform(p) => {
            if p.next_i < p.instances {
                let f = p.transform.clone();
                let callsite = p.source_range;
                let args = crate::std::patterns::transform_callback_args(
                    p.next_i,
                    p.source_range,
                    p.node_path.clone(),
                    exec_state,
                    ctx,
                );
                konts.push(Kont::Resume(Box::new(ResumeState::PatternTransform(p))));
                start_callback(f, callsite, args, konts, exec_state, ctx).await
            } else {
                // All transforms collected: apply them to the geometry via
                // the engine (a flat await), then the value lands on the
                // builtin's call boundary.
                let value = match p.geometry {
                    PatternGeometry::Solids(solids) => {
                        let out = crate::std::patterns::execute_pattern_transform::<crate::execution::Solid>(
                            p.transforms,
                            solids,
                            p.use_original,
                            exec_state,
                            &p.args,
                        )
                        .await?;
                        KclValue::from(out)
                    }
                    PatternGeometry::Sketches(sketches) => {
                        let out = crate::std::patterns::execute_pattern_transform::<crate::execution::Sketch>(
                            p.transforms,
                            sketches,
                            p.use_original,
                            exec_state,
                            &p.args,
                        )
                        .await?;
                        KclValue::from(out)
                    }
                    PatternGeometry::ImportedGeometry(geometry) => {
                        let out =
                            crate::std::patterns::execute_pattern_transform::<crate::execution::ImportedGeometry>(
                                p.transforms,
                                vec![geometry],
                                p.use_original,
                                exec_state,
                                &p.args,
                            )
                            .await?;
                        KclValue::from(
                            out.into_iter()
                                .map(|geometry| {
                                    crate::execution::GeometryWithImportedGeometry::ImportedGeometry(Box::new(geometry))
                                })
                                .collect::<Vec<_>>(),
                        )
                    }
                };
                Ok(ResumeOutcome::Control(Control::Apply(Applied::Value(value))))
            }
        }
        ResumeState::Reduce(_) => Err(KclError::new_internal(KclErrorDetails::new(
            "machine executor: reduce must advance through resume_reduce_next".to_owned(),
            Vec::new(),
        ))),
    }
}

/// Reduce carries its accumulator between callbacks.
async fn resume_reduce_next(
    mut r: ReduceResume,
    accum: KclValue,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<ResumeOutcome, KclError> {
    match r.rest.next() {
        Some(elem) => {
            let f = r.f.clone();
            let callsite = r.source_range;
            let args = crate::std::array::reduce_callback_args(
                elem,
                accum,
                r.source_range,
                r.node_path.clone(),
                exec_state,
                ctx,
            );
            konts.push(Kont::Resume(Box::new(ResumeState::Reduce(r))));
            start_callback(f, callsite, args, konts, exec_state, ctx).await
        }
        None => Ok(ResumeOutcome::Control(Control::Apply(Applied::Value(accum)))),
    }
}

/// Run one callback via the shared call path. A KCL callback pushes its
/// boundary and body and returns Control; a leaf builtin completes inline,
/// in which case the just-pushed Resume kont is popped again by the caller's
/// feed loop.
async fn start_callback(
    f: FunctionSource,
    callsite: SourceRange,
    args: Args<crate::execution::fn_call::Sugary>,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<ResumeOutcome, KclError> {
    // Boxed to break the async cycle machine_call -> resume -> machine_call.
    // The edge is bounded: a leaf callback returns LeafCallbackDone (the
    // caller's feed loop continues iterating with constant native stack), a
    // KCL callback returns through the continuation stack, and only a
    // callback that is itself a resumable builtin adds one nested frame.
    match Box::pin(machine_call(
        f,
        None,
        callsite,
        args,
        BoundaryCompletion::Callback,
        konts,
        exec_state,
        ctx,
    ))
    .await?
    {
        MachineCallOutcome::Control(control) => Ok(ResumeOutcome::Control(control)),
        MachineCallOutcome::LeafCallbackDone(value) => Ok(ResumeOutcome::CallbackDone(value)),
    }
}

/// Convert a finished call's result into machine control: Exit control flow
/// starts an exit unwind; a missing result is the "undefined function result"
/// error; a value applies normally.
fn finish_call_value(
    finished: Option<KclValueControlFlow>,
    fn_name: Option<String>,
    callsite: SourceRange,
    fn_meta: Vec<Metadata>,
) -> Result<Control, KclError> {
    match finished {
        // Only an Exit can pass through call_finish; a KCL 3.0 Return is
        // absorbed there and arrives here as a Continue.
        Some(cf) if cf.is_some_return() => Ok(Control::Exit(cf)),
        Some(cf) => Ok(Control::Apply(Applied::Value(cf.into_value()))),
        None => {
            let mut source_ranges: Vec<SourceRange> = vec![callsite];
            // We want to send the source range of the original function.
            if !fn_meta.is_empty() {
                source_ranges = fn_meta.iter().map(|m| m.source_range).collect();
            }
            let name = fn_name.unwrap_or_else(|| "unknown function".to_owned());
            Err(KclError::new_undefined_value(
                KclErrorDetails::new(
                    format!("Result of user-defined function {name} is undefined"),
                    source_ranges,
                ),
                None,
            ))
        }
    }
}

/// Begin a sketch block: nested check, then argument evaluation (or the
/// sketch-mode cache path), mirroring `Node<SketchBlock>::get_result`.
async fn start_sketch_block(
    node: Arc<Node<SketchBlock>>,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    if exec_state.mod_local.sketch_block.is_some() {
        // Disallow nested sketch blocks for now.
        return Err(KclError::new_semantic(KclErrorDetails::new(
            "Cannot execute a sketch block from within another sketch block".to_owned(),
            vec![SourceRange::from(node.as_ref())],
        )));
    }

    if exec_state.sketch_mode() {
        let (sketch_id, sketch_surface) = match node.arguments_from_cache(exec_state) {
            Ok(x) => x,
            Err(crate::execution::EarlyReturn::Value(cf)) => return Ok(unwind_control(cf)),
            Err(crate::execution::EarlyReturn::Error(e)) => return Err(e),
        };
        return sketch_block_body_setup(node, sketch_id, sketch_surface, konts, exec_state, ctx).await;
    }

    // Evaluate arguments (engine execution creates planes/faces here).
    let state = SketchArgsState {
        node,
        index: 0,
        labeled: IndexMap::new(),
    };
    if state.node.arguments.is_empty() {
        return sketch_args_finish(state, konts, exec_state, ctx).await;
    }
    let req = EvalRequest::expr(&state.node.arguments[0].arg);
    konts.push(Kont::SketchArgs(Box::new(state)));
    Ok(Control::Eval(Box::new(req)))
}

/// Record an evaluated sketch-block argument and evaluate the next, or finish
/// the arguments.
async fn sketch_args_step(
    mut state: SketchArgsState,
    applied: Applied,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let value = applied.expect_value()?;
    let labeled_arg = &state.node.arguments[state.index];
    let source_range = SourceRange::from(&labeled_arg.arg);
    let arg = Arg::new(value, source_range);
    match &labeled_arg.label {
        Some(label) => {
            state.labeled.insert(label.name.clone(), arg);
        }
        None => {
            let name = labeled_arg.arg.ident_name();
            if let Some(name) = name {
                state.labeled.insert(name.to_owned(), arg);
            } else {
                return Err(KclError::new_semantic(KclErrorDetails::new(
                    "Arguments to sketch blocks must be either labeled or simple identifiers".to_owned(),
                    vec![SourceRange::from(&labeled_arg.arg)],
                )));
            }
        }
    }
    state.index += 1;
    if state.index < state.node.arguments.len() {
        let req = EvalRequest::expr(&state.node.arguments[state.index].arg);
        konts.push(Kont::SketchArgs(Box::new(state)));
        Ok(Control::Eval(Box::new(req)))
    } else {
        sketch_args_finish(state, konts, exec_state, ctx).await
    }
}

async fn sketch_args_finish(
    state: SketchArgsState,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let SketchArgsState { node, labeled, .. } = state;
    let (sketch_id, sketch_surface) = match node.finish_arguments_after_eval(labeled, exec_state, ctx).await {
        Ok(x) => x,
        Err(crate::execution::EarlyReturn::Value(cf)) => return Ok(unwind_control(cf)),
        Err(crate::execution::EarlyReturn::Error(e)) => return Err(e),
    };
    sketch_block_body_setup(node, sketch_id, sketch_surface, konts, exec_state, ctx).await
}

/// Scene setup, ambient sketch state, the sketch2 alias scope, and the body's
/// child environment -- then start the body. Mirrors the middle of
/// `Node<SketchBlock>::get_result`, with the same ordering of pushes so
/// unwinding restores in exactly the reverse order.
async fn sketch_block_body_setup(
    node: Arc<Node<SketchBlock>>,
    sketch_id: ObjectId,
    sketch_surface: SketchSurface,
    konts: &mut Vec<Kont>,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let range = SourceRange::from(node.as_ref());
    let sketch_block_artifact_id = node.scene_setup(sketch_id, &sketch_surface, exec_state)?;

    // Don't early return until the stack frame is popped!
    node.prep_mem(exec_state.mut_stack().snapshot()?, exec_state)?;

    // Track that we're executing a sketch block.
    let initial_sketch_block_state = SketchBlockState {
        sketch_id: Some(sketch_id),
        ..Default::default()
    };
    let saved_sketch_block = exec_state.mod_local.sketch_block.replace(initial_sketch_block_state);
    // When executing the body of the sketch block, we no longer want to
    // skip any code.
    let saved_sketch_mode = std::mem::replace(&mut exec_state.mod_local.sketch_mode, false);

    let mut body_state = SketchBodyState {
        node: node.clone(),
        sketch_id,
        sketch_surface,
        sketch_block_artifact_id,
        saved_sketch_block,
        saved_sketch_mode,
        child_env_pushed: false,
    };

    // Load `sketch2::*` into the sketch block's parent scope, so calls like
    // `line(...)` resolve to sketch2 functions. On failure, unwind exactly
    // like the recursive executor: restore ambient state and pop the alias
    // env, skipping the body and finalization.
    if let Err(e) = node.load_sketch2_into_current_scope(exec_state, ctx, range).await {
        sketch_body_cleanup(body_state, exec_state);
        return Err(e);
    }

    // Execute the user body in a child scope, so the aliases aren't included
    // in the returned sketch object.
    let parent = match exec_state.mut_stack().snapshot() {
        Ok(p) => p,
        Err(e) => {
            sketch_body_cleanup(body_state, exec_state);
            return Err(e);
        }
    };
    if let Err(e) = exec_state.mut_stack().push_new_env_for_call(parent) {
        sketch_body_cleanup(body_state, exec_state);
        return Err(e);
    }
    body_state.child_env_pushed = true;

    let body = BlockRef::SketchBody(node);
    konts.push(Kont::SketchBody(Box::new(body_state)));
    push_block(body, BodyType::Block, konts);
    step_block_kick(konts, exec_state, ctx).await
}

/// The sketch-block body finished normally: collect the block's variables,
/// restore ambient state (popping both environments), then solve and finalize
/// -- or exit the whole program when this sketch block is being edited.
async fn sketch_body_finish(
    state: SketchBodyState,
    applied: Applied,
    exec_state: &mut ExecState,
    ctx: &ExecutorContext,
) -> Result<Control, KclError> {
    let block_result = applied.expect_block()?;

    // Collect the block's variables before popping its env; a failure here
    // still unwinds ambient state like the recursive executor.
    let variables = match exec_state.stack().find_all_in_current_env() {
        Ok(block_variables) => block_variables.into_iter().collect::<IndexMap<_, _>>(),
        Err(e) => {
            sketch_body_cleanup(state, exec_state);
            return Err(e);
        }
    };

    let node = state.node.clone();
    let sketch_id = state.sketch_id;
    let sketch_surface = state.sketch_surface.clone();
    let sketch_block_artifact_id = state.sketch_block_artifact_id;

    // Restore ambient state and pop both envs (child, then alias), taking the
    // sketch-block state back out for finalization.
    let child_popped = state.child_env_pushed;
    if child_popped {
        exec_state.mut_stack().pop_env()?;
    }
    exec_state.mod_local.sketch_mode = state.saved_sketch_mode;
    let sketch_block_state = std::mem::replace(&mut exec_state.mod_local.sketch_block, state.saved_sketch_block);
    exec_state.mut_stack().pop_env()?;

    // The body completed normally (an Exit would have unwound instead), so
    // block_result carries no control flow of interest.
    let _ = block_result;

    let Some(sketch_block_state) = sketch_block_state else {
        return Err(KclError::new_internal(KclErrorDetails::new(
            "Sketch block state should still be set to Some from just above".to_owned(),
            vec![SourceRange::from(node.as_ref())],
        )));
    };

    let return_value = node
        .finalize_sketch_block(
            sketch_id,
            &sketch_surface,
            sketch_block_artifact_id,
            variables,
            sketch_block_state,
            exec_state,
            ctx,
        )
        .await?;

    if node.is_being_edited {
        // When the sketch block is being edited, we exit the program
        // immediately.
        Ok(Control::Exit(return_value.exit()))
    } else {
        Ok(Control::Apply(Applied::Value(return_value)))
    }
}

#[cfg(test)]
mod tests {

    #[test]
    fn executor_kind_from_env_value_parses_explicit_selections() {
        assert_eq!(ExecutorKind::parse_env_var("machine"), ExecutorKind::Machine);
        assert_eq!(ExecutorKind::parse_env_var("recursive"), ExecutorKind::Recursive);
        assert_eq!(ExecutorKind::parse_env_var(" Machine\t"), ExecutorKind::Machine);
        assert_eq!(ExecutorKind::parse_env_var("RECURSIVE"), ExecutorKind::Recursive);
    }

    #[test]
    fn executor_kind_from_env_value_defaults_when_empty() {
        assert_eq!(ExecutorKind::parse_env_var(""), ExecutorKind::resolve_default());
        assert_eq!(ExecutorKind::parse_env_var(" \t "), ExecutorKind::resolve_default());
    }

    #[test]
    fn executor_kind_from_env_value_defaults_when_unknown_value() {
        assert_eq!(ExecutorKind::parse_env_var("machin"), ExecutorKind::resolve_default());
    }

    fn set_runtime_executor_flag(flag: RuntimeFlag) {
        crate::set_kcl_runtime_flags(KclRuntimeFlags {
            use_cek_executor: flag,
            ..Default::default()
        });
    }

    fn reset_runtime_executor_flags() {
        crate::set_kcl_runtime_flags(KclRuntimeFlags::DEFAULT);
    }

    /// Flags are process-global; setting them is race-free under nextest's
    /// process-per-test isolation.
    #[test]
    fn runtime_flag_on_selects_machine_executor() {
        set_runtime_executor_flag(RuntimeFlag::On);
        assert_eq!(ExecutorKind::resolve(), ExecutorKind::Machine);
        reset_runtime_executor_flags();
    }

    /// Must hold even on the CI machine leg (`KCL_EXECUTOR=machine`): the
    /// runtime flag outranks the env var.
    #[test]
    fn runtime_flag_off_selects_recursive_executor() {
        set_runtime_executor_flag(RuntimeFlag::Off);
        assert_eq!(ExecutorKind::resolve(), ExecutorKind::Recursive);
        reset_runtime_executor_flags();
    }

    #[test]
    fn runtime_flag_takes_priority_over_env() {
        assert_eq!(
            resolve_from_sources::<ExecutorKind>(RuntimeFlag::Off, None, Some("machine")),
            ExecutorKind::Recursive
        );
        assert_eq!(
            resolve_from_sources::<ExecutorKind>(RuntimeFlag::On, None, Some("recursive")),
            ExecutorKind::Machine
        );
    }

    #[test]
    fn unset_runtime_flag_allows_env_to_select_executor() {
        assert_eq!(
            resolve_from_sources::<ExecutorKind>(RuntimeFlag::Unset, None, Some("machine")),
            ExecutorKind::Machine
        );
        assert_eq!(
            resolve_from_sources::<ExecutorKind>(RuntimeFlag::Unset, None, Some("recursive")),
            ExecutorKind::Recursive
        );
    }

    #[test]
    fn unset_runtime_flag_and_missing_env_selects_default_executor() {
        assert_eq!(
            resolve_from_sources::<ExecutorKind>(RuntimeFlag::Unset, None, None),
            ExecutorKind::Machine
        );
    }

    /// The ZDS feature flag controls exactly this: a context built through a
    /// public constructor picks up the flag as its kind.
    #[tokio::test(flavor = "multi_thread")]
    async fn runtime_flag_on_threads_machine_kind_into_mock_context() {
        set_runtime_executor_flag(RuntimeFlag::On);
        let ctx = ExecutorContext::new_mock(None).await;
        assert_eq!(ctx.executor_kind, ExecutorKind::Machine);
        reset_runtime_executor_flags();
    }

    use super::*;
    use crate::KclRuntimeFlags;
    use crate::RuntimeFlag;
    use crate::execution::parse_execute_with_executor_kind;

    async fn run_machine(code: &str) -> Result<crate::execution::ExecTestResults, KclError> {
        parse_execute_with_executor_kind(code, None, ExecutorKind::Machine).await
    }

    /// A program whose trailing modeling commands are still sitting in the
    /// engine's batch queue when the terminal statement runs. (No close() or
    /// similar: operations that send immediately would drain the queue
    /// themselves.)
    const QUEUED_TAIL: &str = r#"startSketchOn(XY)
  |> startProfile(at = [0, 0])
  |> line(end = [10, 0])
  |> line(end = [0, 10])
"#;

    /// Execute a program's root block directly through exec_block -- the
    /// boundary whose flush behavior the two executors must agree on. This
    /// deliberately bypasses run(): its end-of-run sweep
    /// (ensure_async_commands_completed) flushes whatever the root left
    /// queued and clear_queues() then empties the rest, masking exactly the
    /// difference under test. Returns the block result, whether the batch
    /// queue is empty afterwards, and how many batches were sent. Counts and
    /// emptiness rather than command sequences: imported modules execute
    /// concurrently, so command order is nondeterministic in general.
    async fn exec_root_block(
        kind: ExecutorKind,
        code: &str,
    ) -> (Result<Option<KclValueControlFlow>, KclError>, bool, usize) {
        let program = crate::Program::parse_no_errs(code).unwrap();
        let ctx = crate::execution::new_mock_executor_context(None, kind);
        let mut exec_state = crate::execution::ExecState::new(&ctx);
        ctx.eval_prelude(&mut exec_state, SourceRange::synthetic())
            .await
            .unwrap();
        let no_prelude = ctx
            .handle_annotations(program.ast.inner_attrs.iter(), BodyType::Root, &mut exec_state)
            .await
            .unwrap();
        exec_state.mut_stack().push_new_root_env(!no_prelude).unwrap();
        let result = ctx.exec_block(&program.ast, &mut exec_state, BodyType::Root).await;
        let queue_empty = ctx.engine_batch.is_empty().await;
        let batches_sent = ctx
            .engine
            .stats()
            .batches_sent
            .load(std::sync::atomic::Ordering::Relaxed);
        (result, queue_empty, batches_sent)
    }

    /// exit() terminating the program must flush queued modeling commands to
    /// the engine exactly like running to completion does. (The recursive
    /// executor's root flush sits after its statement loop, which the exit's
    /// `break` still reaches; the machine must match.)
    #[tokio::test(flavor = "multi_thread")]
    async fn exit_at_root_flushes_queued_commands() {
        let completed = format!("@settings(experimentalFeatures = allow)\n{QUEUED_TAIL}");
        let exited = format!("@settings(experimentalFeatures = allow)\n{QUEUED_TAIL}exit()\n");
        let mut counts = Vec::new();
        for kind in [ExecutorKind::Recursive, ExecutorKind::Machine] {
            let (result, queue_empty, n_completed) = exec_root_block(kind, &completed).await;
            assert!(result.is_ok(), "{kind:?}: {result:?}");
            assert!(queue_empty, "{kind:?}: completing the root must flush the batch queue");
            assert!(n_completed >= 1, "{kind:?}: expected at least one batch sent");

            let (result, queue_empty, n_exited) = exec_root_block(kind, &exited).await;
            let cf = result.unwrap().unwrap();
            assert!(cf.is_some_return(), "{kind:?}: expected an exit control-flow value");
            assert!(queue_empty, "{kind:?}: exit() must flush the batch queue");
            assert_eq!(
                n_exited, n_completed,
                "{kind:?}: exit() must flush exactly like completion"
            );
            counts.push((n_completed, n_exited));
        }
        assert_eq!(counts[0], counts[1], "executors disagree on batches sent");
    }

    /// exit() from inside a function unwinds every call boundary and must
    /// still flush at the root.
    #[tokio::test(flavor = "multi_thread")]
    async fn exit_in_function_flushes_queued_commands() {
        let fn_def = "@settings(experimentalFeatures = allow)\nfn quit() {\n  exit()\n  return 0\n}\n";
        let completed = format!("{fn_def}{QUEUED_TAIL}");
        let exited = format!("{fn_def}{QUEUED_TAIL}quit()\n");
        let mut counts = Vec::new();
        for kind in [ExecutorKind::Recursive, ExecutorKind::Machine] {
            let (result, queue_empty, n_completed) = exec_root_block(kind, &completed).await;
            assert!(result.is_ok(), "{kind:?}: {result:?}");
            assert!(queue_empty, "{kind:?}: completing the root must flush the batch queue");
            assert!(n_completed >= 1, "{kind:?}: expected at least one batch sent");

            let (result, queue_empty, n_exited) = exec_root_block(kind, &exited).await;
            let cf = result.unwrap().unwrap();
            assert!(cf.is_some_return(), "{kind:?}: expected an exit control-flow value");
            assert!(
                queue_empty,
                "{kind:?}: exit() from a function must flush the batch queue"
            );
            assert_eq!(
                n_exited, n_completed,
                "{kind:?}: exit() from a function must flush exactly like completion"
            );
            counts.push((n_completed, n_exited));
        }
        assert_eq!(counts[0], counts[1], "executors disagree on batches sent");
    }

    /// A fatal error must NOT flush the root: queued commands are left in
    /// the batch queue (run()'s end-of-run machinery then sweeps or drops
    /// them). This asymmetry with exit() is deliberate.
    #[tokio::test(flavor = "multi_thread")]
    async fn error_does_not_flush_queued_commands() {
        let errored = format!("{QUEUED_TAIL}assert(1, isEqualTo = 2)\n");
        let mut counts = Vec::new();
        for kind in [ExecutorKind::Recursive, ExecutorKind::Machine] {
            let (result, queue_empty, n_completed) = exec_root_block(kind, QUEUED_TAIL).await;
            assert!(result.is_ok(), "{kind:?}: {result:?}");
            assert!(queue_empty, "{kind:?}: completing the root must flush the batch queue");
            assert!(n_completed >= 1, "{kind:?}: expected at least one batch sent");

            let (result, queue_empty, n_errored) = exec_root_block(kind, &errored).await;
            assert!(result.is_err(), "{kind:?}: expected an error");
            assert!(!queue_empty, "{kind:?}: an error must leave queued commands unflushed");
            assert_eq!(
                n_errored,
                n_completed - 1,
                "{kind:?}: an error must skip exactly the final flush"
            );
            counts.push((n_completed, n_errored));
        }
        assert_eq!(counts[0], counts[1], "executors disagree on batches sent");
    }

    /// The machine's whole point: recursion depth well past the recursive
    /// executor's native-stack cap runs fine on the heap.
    #[tokio::test(flavor = "multi_thread")]
    async fn deep_recursion_well_past_recursive_cap() {
        let code = r#"fn countdown(@n) {
  return if n == 0 {
    0
  } else {
    countdown(n - 1)
  }
}
result = countdown(900)
"#;
        let result = run_machine(code).await.unwrap();
        let value = result
            .exec_state
            .stack()
            .memory
            .get_from_owned("result", result.mem_env, SourceRange::default(), 0)
            .unwrap();
        let KclValue::Number { value, .. } = value else {
            panic!("expected a number, found {value:?}");
        };
        assert_eq!(value, 0.0);
    }

    /// Runaway recursion trips the depth guard with an error instead of
    /// hanging or overflowing the native stack.
    #[tokio::test(flavor = "multi_thread")]
    async fn infinite_recursion_trips_depth_guard() {
        let code = r#"fn forever(@n) {
  return 1 + forever(n)
}
forever(1)
"#;
        let err = run_machine(code).await.unwrap_err();
        assert!(err.to_string().contains("Call depth limit"), "actual: {err:?}");
    }

    /// Compare wall time of the two executors on a large, engine-free
    /// workload (mock engine). Not a gate: the machine merges regardless of
    /// speed; a <=10% overhead gates the default flip later. Run manually:
    /// cargo nextest run -p kcl-lib --run-ignored all -E 'test(perf_compare_executors)' --no-capture
    #[tokio::test(flavor = "multi_thread")]
    #[ignore = "manual perf measurement, prints timings"]
    async fn perf_compare_executors() {
        let code = include_str!("../../tests/mike_stress_test/input.kcl");
        const RUNS: usize = 5;
        for kind in [ExecutorKind::Recursive, ExecutorKind::Machine] {
            // Warm up once.
            parse_execute_with_executor_kind(code, None, kind).await.unwrap();
            let start = std::time::Instant::now();
            for _ in 0..RUNS {
                parse_execute_with_executor_kind(code, None, kind).await.unwrap();
            }
            let avg = start.elapsed() / RUNS as u32;
            println!("{kind:?}: {avg:?} per run over {RUNS} runs");
        }
    }

    /// The design must hold at a much higher call depth with the limit raised
    /// internally (the production default stays at
    /// DEFAULT_MACHINE_CALL_DEPTH_LIMIT): depth is heap-bounded, so the limit
    /// is policy, not a native-stack constraint.
    #[tokio::test(flavor = "multi_thread")]
    async fn depth_with_raised_limit() {
        let code = r#"fn countdown(@n) {
  return if n == 0 {
    0
  } else {
    countdown(n - 1)
  }
}
result = countdown(9000)
"#;
        let program = crate::Program::parse_no_errs(code).unwrap();
        let exec_ctxt = ExecutorContext {
            engine: std::sync::Arc::new(crate::engine::engine_manager::EngineManager::new_mock()),
            engine_batch: crate::engine::EngineBatchContext::default(),
            fs: crate::fs::new_file_system_handle(crate::fs::FileManager::new()),
            settings: crate::execution::ExecutorSettings::default(),
            context_type: crate::execution::ContextType::Mock,
            execution_callbacks: Default::default(),
            executor_kind: ExecutorKind::Machine,
            machine_call_depth_limit: 10_000,
        };
        let mut exec_state = ExecState::new(&exec_ctxt);
        let (env_ref, _) = exec_ctxt.run(&program, &mut exec_state).await.unwrap();
        let value = exec_state
            .stack()
            .memory
            .get_from_owned("result", env_ref, SourceRange::default(), 0)
            .unwrap();
        let KclValue::Number { value, .. } = value else {
            panic!("expected a number, found {value:?}");
        };
        assert_eq!(value, 0.0);
        assert!(
            exec_state.machine_depth_high_water() >= 9_000,
            "high water: {}",
            exec_state.machine_depth_high_water()
        );
    }

    /// The key requirement of the resumable-callback protocol: map calling a
    /// function that maps again, nested hundreds of levels deep, runs on the
    /// machine's continuation stack with no native-stack growth per level.
    #[tokio::test(flavor = "multi_thread")]
    async fn deeply_nested_map_callbacks() {
        let code = r#"fn nest(@depth) {
  return if depth == 0 {
    1
  } else {
    map([depth - 1], f = nest)[0]
  }
}
result = nest(400)
"#;
        let result = run_machine(code).await.unwrap();
        let value = result
            .exec_state
            .stack()
            .memory
            .get_from_owned("result", result.mem_env, SourceRange::default(), 0)
            .unwrap();
        let KclValue::Number { value, .. } = value else {
            panic!("expected a number, found {value:?}");
        };
        assert_eq!(value, 1.0);
    }

    /// A long run of leaf-builtin callbacks completes in constant native
    /// stack (the feed loop, not recursion).
    #[tokio::test(flavor = "multi_thread")]
    async fn many_leaf_callbacks() {
        let code = r#"fn double(@x) {
  return x * 2
}
result = map([0..4999], f = double)
"#;
        let result = run_machine(code).await.unwrap();
        let value = result
            .exec_state
            .stack()
            .memory
            .get_from_owned("result", result.mem_env, SourceRange::default(), 0)
            .unwrap();
        let KclValue::HomArray { value, .. } = value else {
            panic!("expected an array, found {value:?}");
        };
        assert_eq!(value.len(), 5000);
    }

    /// Deep pipelines ride the continuation stack, not the native stack.
    #[tokio::test(flavor = "multi_thread")]
    async fn long_pipeline() {
        let mut code = String::from("fn bump(@n) {\n  return n + 1\n}\nx = 0");
        for _ in 0..500 {
            code.push_str("\n  |> bump(%)");
        }
        code.push('\n');
        let result = run_machine(&code).await.unwrap();
        let value = result
            .exec_state
            .stack()
            .memory
            .get_from_owned("x", result.mem_env, SourceRange::default(), 0)
            .unwrap();
        let KclValue::Number { value, .. } = value else {
            panic!("expected a number, found {value:?}");
        };
        assert_eq!(value, 500.0);
    }
}