datalogic_rs/engine/

mod.rs

use std::cell::Cell;
use std::collections::HashMap;

use crate::config::EvaluationConfig;

use crate::{CompiledNode, Logic, Result};

thread_local! {
    /// Per-thread re-entry counter for the `Engine::evaluate` boundary.
    /// Bumped by `enter_dispatch_boundary` on entry and restored by the
    /// `DepthGuard` on drop (not touched by `dispatch_node` itself), so the
    /// value reflects how many nested `Engine::evaluate(...)` calls are
    /// currently live on the sync call stack.
    ///
    /// Why thread-local rather than a `ContextStack` field: a custom
    /// operator can hold `Arc<Engine>` and call `engine.evaluate(...)`
    /// recursively from inside its own `evaluate(...)` — each top-level
    /// call constructs a fresh `ContextStack` (depth resets to 0) but
    /// the C call stack keeps growing. A thread-local survives across
    /// those boundaries and catches the runaway recursion before stack
    /// overflow.
    ///
    /// Tokio safety: `dispatch_node` is sync, so a task can't `.await`
    /// while holding the counter raised. Between dispatch calls the
    /// value returns to zero, so cross-thread task migration starts
    /// fresh on whatever thread it lands.
    static DISPATCH_DEPTH: Cell<u32> = const { Cell::new(0) };
}

/// An evaluation's result paired with what it cost.
///
/// Returned by [`Engine::evaluate_metered`],
/// [`Engine::evaluate_metered_report`],
/// [`Session::eval_metered`](crate::Session::eval_metered) and
/// [`Session::eval_metered_report`](crate::Session::eval_metered_report).
/// `T` is whatever that entry point produces — a borrowed
/// `&'a DataValue<'a>` from the engine, an owned value from a session, or
/// a [`Result<...>`](crate::Result) from the failure-inclusive `*_report`
/// methods.
///
/// Both fields are public, so the usual shape is a destructure:
///
/// ```rust
/// # #[cfg(feature = "budget")] {
/// use bumpalo::Bump;
/// use datalogic_rs::{Engine, Metered};
///
/// let engine = Engine::new();
/// let compiled = engine.compile(r#"{"+": [{"var": "x"}, 2]}"#).unwrap();
/// let arena = Bump::new();
/// let Metered { value, ops } = engine
///     .evaluate_metered(&compiled, r#"{"x": 1}"#, &arena, u64::MAX)
///     .unwrap();
/// assert_eq!(value.as_i64(), Some(3));
/// assert_eq!(ops, 2, "the `+` and the `var`; the literal 2 costs nothing");
///
/// // `{"+": [1, 2]}` costs 0: the compiler folded it to a literal, and a
/// // literal is returned before dispatch.
/// let folded = engine.compile(r#"{"+": [1, 2]}"#).unwrap();
/// let metered = engine.evaluate_metered(&folded, "null", &arena, u64::MAX).unwrap();
/// assert_eq!(metered.ops, 0);
/// # }
/// ```
#[cfg(feature = "budget")]
#[cfg_attr(docsrs, doc(cfg(feature = "budget")))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Metered<T> {
    /// The evaluation's result (or [`Result<...>`](crate::Result) for the
    /// failure-inclusive `*_report` methods).
    pub value: T,
    /// Operations charged. See [`Engine::evaluate_metered`] for what one
    /// operation is.
    pub ops: u64,
}

#[cfg(feature = "budget")]
impl<T, E> Metered<std::result::Result<T, E>> {
    /// Transpose a failure-inclusive `Metered<Result<T, E>>` into a
    /// `Result<Metered<T>, E>`, keeping `ops` on `Ok` and propagating
    /// `Err(e)`.
    #[inline]
    pub fn transpose(self) -> std::result::Result<Metered<T>, E> {
        let Self { value, ops } = self;
        value.map(|value| Metered { value, ops })
    }
}

/// Restores [`DISPATCH_DEPTH`] to its prior value on drop. Used by
/// the boundary entry points (`Engine::evaluate`, `TracedSession::evaluate`)
/// so early returns and panics leave the counter consistent.
///
/// `DepthGuard(u32::MAX)` is a no-op sentinel — used when the engine has
/// no custom operators registered, so cross-evaluate recursion is
/// impossible and the boundary skips the TLS bookkeeping entirely. The
/// drop check makes the guard zero-cost in that case.
pub(crate) struct DepthGuard(u32);

impl DepthGuard {
    const NOOP: u32 = u32::MAX;
}

impl Drop for DepthGuard {
    #[inline]
    fn drop(&mut self) {
        if self.0 != Self::NOOP {
            DISPATCH_DEPTH.with(|d| d.set(self.0));
        }
    }
}

/// JSONLogic compile/evaluate engine.
///
/// Holds the immutable engine state — registered [`crate::CustomOperator`]
/// implementations, the [`EvaluationConfig`], the templating, folding and
/// family settings — and exposes the public surface for parsing
/// rules ([`Self::compile`]), evaluating them ([`Self::eval`] /
// `Self::eval_into` is feature-gated on `serde_json`; link it
// conditionally so default-features `cargo doc` doesn't break.
#[cfg_attr(
    feature = "serde_json",
    doc = "[`Self::eval_str`], [`Self::eval_into`]), and opening hot-loop"
)]
#[cfg_attr(
    not(feature = "serde_json"),
    doc = "[`Self::eval_str`], `Self::eval_into`), and opening hot-loop"
)]
/// sessions ([`Self::session`]).
// The `trace` feature adds [`Self::trace`]; reference it conditionally so
// `cargo doc` without `--all-features` doesn't break on the intra-doc link.
#[cfg_attr(
    feature = "trace",
    doc = "Enabling the `trace` feature also exposes [`Self::trace`] for traced sessions."
)]
///
/// `Engine` is `Send + Sync` (every field is); the typical pattern is to
/// build one at startup, wrap it in `Arc<Engine>`, and clone the `Arc`
/// across threads or async tasks.
///
/// # Example
///
/// ```rust
/// use datalogic_rs::Engine;
///
/// // 1. Build the engine.
/// let engine = Engine::new();
///
/// // 2. Compile a rule once.
/// let logic = engine
///     .compile(r#"{"if": [{">=": [{"var": "age"}, 18]}, "adult", "minor"]}"#)
///     .unwrap();
///
/// // 3. Evaluate against many inputs (here via `Session::eval_str`;
/// //    drop to `Engine::evaluate` if you want zero-copy borrowed results).
/// let mut session = engine.session();
/// for age in [12, 18, 42] {
///     let payload = format!(r#"{{"age": {age}}}"#);
///     let result = session.eval_str(&logic, &payload).unwrap();
///     assert!(result == "\"adult\"" || result == "\"minor\"");
///     session.reset();
/// }
/// ```
///
/// # Choosing an evaluate method
///
/// **Start here.** Use [`Self::eval_str`] for one-shot calls. Switch
/// to [`crate::Session`] once you're evaluating the same compiled rule
/// many times — it reuses one arena instead of allocating per call.
/// Drop down to [`Self::evaluate`] only when you're managing your own
/// `bumpalo::Bump` (custom pools, integration with arena-aware
/// downstream code).
///
/// Result-shape suffixes work the same on every tier: `(none)` returns
/// [`datavalue::OwnedDataValue`], `_str` returns [`String`] (JSON),
/// `_into::<T>` returns `T: DeserializeOwned` (requires `serde_json`).
/// The raw [`Self::evaluate`] is the only method that exposes
/// `&'a DataValue<'a>` and a caller-owned `&Bump`.
///
/// Three tiers, in order of caller control:
///
/// | Method | Arena ownership | Result type | When to use |
/// |---|---|---|---|
// `eval_into` is feature-gated on `serde_json`; emit a linked or plain
// reference depending on the active features so default-features
// `cargo doc` doesn't break the table row.
#[cfg_attr(
    feature = "serde_json",
    doc = "| [`Self::eval`] / [`Self::eval_str`] / [`Self::eval_into`] | engine creates a fresh `Bump::with_capacity(4096)` per call | [`OwnedDataValue`](datavalue::OwnedDataValue) / `String` / `T` | One-shot. Any caller that doesn't want to think about arenas. Allocates each call — for hot loops, drop to `Session`. |"
)]
#[cfg_attr(
    not(feature = "serde_json"),
    doc = "| [`Self::eval`] / [`Self::eval_str`] / `Self::eval_into` | engine creates a fresh `Bump::with_capacity(4096)` per call | [`OwnedDataValue`](datavalue::OwnedDataValue) / `String` / `T` | One-shot. Any caller that doesn't want to think about arenas. Allocates each call — for hot loops, drop to `Session`. |"
)]
#[cfg_attr(
    feature = "serde_json",
    doc = "| [`crate::Session::eval`] / [`crate::Session::eval_str`] / [`crate::Session::eval_into`] / [`crate::Session::eval_borrowed`] | session-owned `Bump`, caller calls [`crate::Session::reset`] between batches | owned / `String` / `T` / borrowed `&'a DataValue<'a>` | Hot loop with a long-lived engine. The `Session` hides `bumpalo` from the call site and pre-sizes the arena via [`crate::Session::reset_with_capacity`] when needed. |"
)]
#[cfg_attr(
    not(feature = "serde_json"),
    doc = "| [`crate::Session::eval`] / [`crate::Session::eval_str`] / `Session::eval_into` / [`crate::Session::eval_borrowed`] | session-owned `Bump`, caller calls [`crate::Session::reset`] between batches | owned / `String` / `T` / borrowed `&'a DataValue<'a>` | Hot loop with a long-lived engine. The `Session` hides `bumpalo` from the call site and pre-sizes the arena via [`crate::Session::reset_with_capacity`] when needed. |"
)]
/// | [`Self::evaluate`] | caller-passed `&Bump`; library never resets | `&'a DataValue<'a>` (borrowed) | Zero-copy result paths, custom pool/allocator strategies, integration with arena-aware downstream code. |
///
/// All routes share the same dispatcher; the differences are who owns
/// the arena, what the result type looks like, and whether the
/// boundary parses / serialises JSON for you. There is no perf
/// difference between the arena-aware paths once the bump is warm.
///
/// See the crate-level docs for the two-phase architecture, threading
/// model, and walk-through examples; see the `Session` and
/// `EvaluationConfig` rustdoc for arena-management and behaviour-tuning
/// options respectively.
pub struct Engine {
    /// Identifies this engine to the rules it compiles: a compiled custom
    /// operator call records the engine id and slot, and dispatch takes the
    /// slot only on the same engine. Unique for the process's lifetime.
    pub(super) id: u64,
    /// Custom `CustomOperator` implementations registered with the engine.
    pub(super) custom_operators: CustomOperators,
    /// Everything the builder set: config, templating mode and escape,
    /// folding, families. The trace surface always disables folding
    /// regardless of `settings.constant_folding` (handled in
    /// `TracedSession`).
    settings: crate::builder::EngineSettings,
    /// [`EvaluationConfig::fold_fingerprint`] of the config, computed once:
    /// a rule folded under another fingerprint is compiled again before
    /// it runs here (see `Logic::for_engine`).
    fold_fingerprint: u64,
}

/// The built-in operator `name` names among the families in `families`.
#[inline]
pub(crate) fn builtin_in(families: u32, name: &str) -> Option<crate::OpCode> {
    name.parse::<crate::OpCode>()
        .ok()
        .filter(|op| op.in_families(families))
}

mod dispatch;

/// Source of [`Engine::id`].
static NEXT_ENGINE_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);

/// The custom operators registered on an engine, addressable by name and,
/// for the rules the engine compiles, by slot.
pub(super) struct CustomOperators {
    ops: Vec<std::sync::Arc<dyn crate::CustomOperator>>,
    slots: HashMap<String, u32>,
}

impl CustomOperators {
    fn new(operators: HashMap<String, std::sync::Arc<dyn crate::CustomOperator>>) -> Self {
        let mut ops = Vec::with_capacity(operators.len());
        let mut slots = HashMap::with_capacity(operators.len());
        for (name, op) in operators {
            slots.insert(name, ops.len() as u32);
            ops.push(op);
        }
        CustomOperators { ops, slots }
    }

    fn len(&self) -> usize {
        self.ops.len()
    }

    fn is_empty(&self) -> bool {
        self.ops.is_empty()
    }

    /// The slot `name` is registered in.
    pub(super) fn slot(&self, name: &str) -> Option<u32> {
        self.slots.get(name).copied()
    }

    /// The operator registered as `name`.
    pub(super) fn get(&self, name: &str) -> Option<&dyn crate::CustomOperator> {
        self.slot(name).map(|slot| self.at(slot))
    }

    /// The operator in `slot`, which [`Self::slot`] returned on this engine.
    #[inline]
    pub(super) fn at(&self, slot: u32) -> &dyn crate::CustomOperator {
        &*self.ops[slot as usize]
    }

    fn names(&self) -> impl Iterator<Item = &str> {
        self.slots.keys().map(String::as_str)
    }

    /// Every registration, sharing each operator instance.
    fn shared(&self) -> HashMap<String, std::sync::Arc<dyn crate::CustomOperator>> {
        self.slots
            .iter()
            .map(|(name, &slot)| {
                (
                    name.clone(),
                    std::sync::Arc::clone(&self.ops[slot as usize]),
                )
            })
            .collect()
    }
}

/// Convert an `OwnedDataValue` literal to an arena-resident `DataValue`
/// reference. Reached from the `dispatch_node` literal path for any
/// `CompiledNode::Value` whose `lit` was not precomputed — in practice
/// only ad-hoc `synthetic_value` wrappers built outside the compile
/// pipeline, since `populate_lits` pre-builds every literal reachable from
/// a `Logic`. Trivial cases (Null, Bool, empty primitives) hit shared
/// singletons with no allocation; a non-empty String allocates a single
/// `DataValue` wrapper into the per-call arena (the `&str` is borrowed from
/// the owned source); non-empty Arrays/Objects rebuild their spine in the
/// arena via `OwnedDataValue::view_in`, borrowing strings, keys and tensor
/// bytes from the owned source instead of copying them. Sound because the
/// compiled node, and therefore `value`, outlives the evaluation:
/// `dispatch_node` borrows the node at the same `'a` as the arena.
///
/// `#[cold]` + `#[inline(never)]`: `dispatch_node` is `#[inline(always)]`,
/// so its literal path is stamped into every operator's dispatch site —
/// keeping this fallback outlined keeps those sites small (I-cache), and
/// the cold hint steers the branch layout toward the pre-built `lit` path
/// that every compiled literal takes.
/// The error path of `dispatch_node`: push `node` onto the breadcrumb and,
/// when no deeper node has named the error, name it after `node`'s
/// operator. The first node to see an error on its way up is the innermost
/// one that failed, so [`crate::Error::operator`] names the operator that
/// actually failed (a custom operator included), and the plain and traced
/// paths agree on it. Outlined and cold for the same I-cache reason as
/// [`literal_fallback`].
#[cold]
#[inline(never)]
fn record_failure(
    err: &mut crate::Error,
    node: &CompiledNode,
    ctx: &mut crate::arena::ContextStack<'_>,
) {
    ctx.push_error_step(node.id());
    err.name_operator_from(node);
}

#[cold]
#[inline(never)]
fn literal_fallback<'a>(
    value: &'a datavalue::OwnedDataValue,
    arena: &'a bumpalo::Bump,
) -> &'a crate::arena::DataValue<'a> {
    use datavalue::OwnedDataValue;
    match value {
        OwnedDataValue::Null => crate::arena::singletons::singleton_null(),
        OwnedDataValue::Bool(b) => crate::arena::singletons::singleton_bool(*b),
        OwnedDataValue::String(s) if s.is_empty() => {
            crate::arena::singletons::singleton_empty_string()
        }
        OwnedDataValue::Array(a) if a.is_empty() => {
            crate::arena::singletons::singleton_empty_array()
        }
        OwnedDataValue::Object(o) if o.is_empty() => {
            crate::arena::singletons::singleton_empty_object()
        }
        OwnedDataValue::String(s) => arena.alloc(crate::arena::DataValue::String(s.as_str())),
        _ => arena.alloc(value.view_in(arena)),
    }
}

impl Default for Engine {
    fn default() -> Self {
        Self::new()
    }
}

impl std::fmt::Debug for Engine {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Print the operator *count* rather than names: names are user
        // registration data, and `Engine::custom_operator_names()` exposes
        // them already for callers who want them. The trait objects
        // themselves can't render a meaningful Debug.
        // Families print by name: the ones this engine has, among those
        // compiled into the build.
        let families: Vec<&str> = crate::Family::ALL
            .iter()
            .filter(|f| f.is_compiled() && self.settings.families & f.bit() != 0)
            .map(|f| f.name())
            .collect();
        let settings = &self.settings;
        let mut s = f.debug_struct("Engine");
        s.field("custom_operators", &self.custom_operators.len());
        s.field("templating", &settings.templating);
        s.field("template_key_escape", &settings.template_key_escape);
        s.field("constant_folding", &settings.constant_folding);
        s.field("families", &families);
        s.field("config", &settings.config);
        s.finish_non_exhaustive()
    }
}

impl Engine {
    /// Start a [`crate::EngineBuilder`] for fluent construction.
    ///
    /// Use the builder when you need a non-default [`EvaluationConfig`],
    /// templating mode, or pre-registered custom operators.
    /// For a stock engine, [`Self::new`] is shorter.
    #[inline]
    pub fn builder() -> crate::EngineBuilder {
        crate::EngineBuilder::new()
    }

    /// A builder that starts from this engine: the same custom operators,
    /// config, templating mode, template key escape and folding setting.
    /// Change what differs and call [`build`](crate::EngineBuilder::build)
    /// for a new engine; this one is untouched.
    ///
    /// Both engines share each operator instance (and any state it holds),
    /// so a host that rebuilds on every reload does not register its
    /// operators again. Rules compiled on this engine still evaluate on the
    /// new one, looking their custom operators up by name.
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::{Engine, EvaluationConfig, MissingVar};
    ///
    /// let running = Engine::new();
    /// let reloaded = running
    ///     .to_builder()
    ///     .with_config(EvaluationConfig::default().with_missing_var(MissingVar::Error))
    ///     .build();
    /// assert!(reloaded.eval_str(r#"{"var": "x"}"#, "{}").is_err());
    /// assert_eq!(running.eval_str(r#"{"var": "x"}"#, "{}").unwrap(), "null");
    /// ```
    pub fn to_builder(&self) -> crate::EngineBuilder {
        crate::EngineBuilder::from_engine_parts(
            self.settings.clone(),
            self.custom_operators.shared(),
        )
    }

    /// Open a [`crate::Session`] handle that owns a reusable arena and
    /// returns owned results, so callers don't need to manage a
    /// [`bumpalo::Bump`] themselves.
    ///
    /// Use this when you want the throughput of arena reuse without the
    /// lifetime juggling of [`Self::evaluate`]. Results are deep-cloned out
    /// of the arena before returning, so they survive later calls and resets.
    /// The session does **not** auto-reset: allocations accumulate until you
    /// call [`crate::Session::reset`], which you should do between logical
    /// batches to bound peak memory in long-running services.
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let compiled = engine.compile(r#"{"+": [{"var": "x"}, 1]}"#).unwrap();
    /// let mut session = engine.session();
    /// let result = session.eval_str(&compiled, r#"{"x": 41}"#).unwrap();
    /// assert_eq!(result, "42");
    /// ```
    #[inline]
    pub fn session(&self) -> crate::Session<'_> {
        crate::Session::borrowing(self)
    }

    /// Internal seam used by the builder. `pub(crate)` is enough — no
    /// `#[doc(hidden)]` needed since it's not externally reachable.
    #[inline]
    pub(crate) fn from_builder_parts(
        mut settings: crate::builder::EngineSettings,
        operators: HashMap<String, std::sync::Arc<dyn crate::CustomOperator>>,
    ) -> Self {
        settings.templating &= cfg!(feature = "templating");
        Self {
            id: NEXT_ENGINE_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
            custom_operators: CustomOperators::new(operators),
            fold_fingerprint: settings.config.fold_fingerprint(),
            settings,
        }
    }

    /// Creates a new Engine with all built-in operators.
    ///
    /// The engine includes every built-in operator compiled into this build
    /// (84 canonical operators with all operator features on; see
    /// [`Self::builtin_operator_names`]), dispatched via `OpCode`.
    /// Templating mode is disabled by default. For non-default
    /// configuration (custom [`EvaluationConfig`], templating mode,
    /// pre-registered custom operators) prefer [`Self::builder`].
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// ```
    pub fn new() -> Self {
        Self::from_builder_parts(crate::builder::EngineSettings::default(), HashMap::new())
    }

    /// Gets a reference to the current evaluation configuration.
    pub fn config(&self) -> &EvaluationConfig {
        &self.settings.config
    }

    /// Internal: whether the constant-folding pass runs during
    /// [`Self::compile`]. Reads the field set by
    /// [`crate::EngineBuilder::with_constant_folding`].
    #[inline]
    pub(crate) fn constant_folding_enabled(&self) -> bool {
        self.settings.constant_folding
    }

    /// Internal: whether templating mode is on. Always `false` when the
    /// crate is built without `feature = "templating"`: the builder seam
    /// clears the flag off-feature.
    #[inline]
    pub(crate) fn is_templating_enabled(&self) -> bool {
        self.settings.templating
    }

    /// Internal: the template-key escape prefix, or `None` when unset.
    #[inline]
    pub(crate) fn template_key_escape(&self) -> Option<char> {
        self.settings.template_key_escape
    }

    /// Checks if a custom operator with the given name is registered.
    ///
    /// Operator registration is builder-only; this is a read-only check
    /// against the frozen set produced by [`crate::EngineBuilder`].
    pub fn has_custom_operator(&self, name: &str) -> bool {
        self.custom_operators.slot(name).is_some()
    }

    /// What the custom operator registered as `name` declares about
    /// itself ([`crate::CustomOperator::info`]), or `None` when no custom
    /// operator has that name. Built-ins are described by
    /// [`Self::operators`].
    pub fn custom_operator_info(&self, name: &str) -> Option<crate::CustomOperatorInfo> {
        self.custom_operators.get(name).map(|op| op.info())
    }

    /// This engine's process-unique id.
    #[inline]
    pub(crate) fn id(&self) -> u64 {
        self.id
    }

    /// The fingerprint of the settings constant folding observes; see
    /// [`EvaluationConfig::fold_fingerprint`].
    #[inline]
    pub(crate) fn fold_fingerprint(&self) -> u64 {
        self.fold_fingerprint
    }

    /// The custom operator registered as `name`.
    pub(crate) fn custom_operator(&self, name: &str) -> Option<&dyn crate::CustomOperator> {
        self.custom_operators.get(name)
    }

    /// Whether the iterator fast paths may read fields inline. They read a
    /// missing field as `null` without dispatching a `var`, which is only
    /// right while a miss is `null` ([`crate::MissingVar::Null`]).
    #[inline(always)]
    pub(crate) fn reads_fields_inline(&self) -> bool {
        self.settings.config.missing_var == crate::MissingVar::Null
    }

    /// Where a call to custom operator `name` compiled on this engine finds
    /// it: this engine's id and the operator's slot.
    pub(crate) fn custom_operator_slot(&self, name: &str) -> Option<(u64, u32)> {
        self.custom_operators.slot(name).map(|slot| (self.id, slot))
    }

    /// Iterator over the names of every *custom* operator registered on
    /// this engine (built-ins are not included). Order is unspecified
    /// (HashMap iteration order). Useful for tooling, UIs, and tests
    /// that need to introspect what's available.
    pub fn custom_operator_names(&self) -> impl Iterator<Item = &str> {
        self.custom_operators.names()
    }

    /// Iterator over every *built-in* operator name this build evaluates:
    /// the JSONLogic baseline plus whichever extension families
    /// (`ext-string`, `datetime`, `ext-control`, …) were compiled in.
    ///
    /// The list is derived from the same table the compiler resolves
    /// operator keys against, so it cannot drift from dispatch. It
    /// includes input aliases (`var` for `val`, `?:` for `if`, `match`
    /// for `switch`) because those keys are live calls too; order is
    /// canonical name first, then its aliases, per operator.
    ///
    /// Custom registrations are not included; see
    /// [`Self::custom_operator_names`]. The union of the two is the full
    /// vocabulary of this engine. A custom operator registered under a
    /// built-in name is never reached: built-in resolution wins at
    /// compile time.
    ///
    /// Under templating mode an unknown key is not an error (the object
    /// echoes back as data), so this is the vocabulary authoring-side
    /// tooling needs to tell a live call from a literal.
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let names: Vec<&str> = engine.builtin_operator_names().collect();
    /// assert!(names.contains(&"val"));
    /// assert!(names.contains(&"var")); // alias of `val`
    /// assert!(!names.contains(&"lenght"));
    /// ```
    pub fn builtin_operator_names(&self) -> impl Iterator<Item = &'static str> + use<> {
        crate::operators::table::builtin_operator_names_in(self.settings.families)
    }

    /// Every built-in operator compiled into this build, described by its
    /// operator-table row: name and aliases, family and gating feature,
    /// declared argument counts, whether it reads the data context, its
    /// effect, its cost class, and which argument (if any) runs under a
    /// pushed frame.
    ///
    /// Read from the same table that drives compilation, dispatch and the
    /// optimizer, so it cannot drift from what the engine does. Custom
    /// operators are not included; see [`Self::custom_operator_names`].
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::{Engine, ScopedArg};
    ///
    /// let engine = Engine::new();
    /// let map = engine.operators().find(|op| op.name == "map").unwrap();
    /// assert_eq!(map.family, "Core");
    /// assert_eq!(map.feature, None);
    /// assert_eq!(map.scoped_arg, Some(ScopedArg::Index(1)));
    ///
    /// let val = engine.operators().find(|op| op.name == "val").unwrap();
    /// assert_eq!(val.aliases, ["var"]);
    /// assert!(val.reads_context);
    /// ```
    pub fn operators(&self) -> impl Iterator<Item = crate::OperatorInfo> + use<> {
        crate::operators::info::operators_in(self.settings.families)
    }

    /// The built-in operator `name` names on this engine: a name of a
    /// family it has ([`crate::EngineBuilder::with_families`]).
    #[inline]
    pub(crate) fn builtin(&self, name: &str) -> Option<crate::OpCode> {
        builtin_in(self.settings.families, name)
    }

    // ============================================================
    // V5 PUBLIC API
    //   - One-shot:   `eval` / `eval_str` / `eval_into`   (engine-owned arena per call)
    //   - Power tier: `evaluate(&Logic, D, &Bump)`        (caller-owned arena, borrowed result)
    //   - Hot loop:   `engine.session().eval*(...)`       (pooled arena, manual reset)
    //   - Trace:      `engine.trace().eval*(...)`         (same shapes wrapped in TracedRun<R>; eval_str/eval_into take a rule source)
    // ============================================================

    /// Compile a rule source into reusable [`Logic`].
    ///
    /// `rule` accepts any [`crate::IntoLogic`] shape: `&str` (JSON-parsed),
    /// `&OwnedDataValue` / `OwnedDataValue` (cloned/moved), or
    /// `&serde_json::Value` (gated on `serde_json`). For cross-thread
    /// sharing prefer [`Self::compile_arc`].
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let compiled = engine.compile(r#"{"==": [{"var": "x"}, 1]}"#).unwrap();
    /// ```
    pub fn compile<R: crate::IntoLogic>(&self, rule: R) -> Result<Logic> {
        let owned = rule.logic_source()?;
        Logic::compile_with(owned, self)
    }

    /// Check `rule` without compiling it: every problem the engine can see
    /// before it runs, in rule order, each located by a JSON Pointer.
    ///
    /// An [`Error`](crate::Severity::Error) is something that will fail: a
    /// rule that does not compile (an object with several keys outside
    /// templating mode), or a call that fails whenever it runs (an unknown
    /// operator, `and` / `or` / `if` without an argument array, an argument
    /// count the operator rejects, a literal timezone that does not exist).
    /// A [`Warning`](crate::Severity::Warning) runs, but probably not as
    /// meant: arguments an operator never evaluates, or in a template an
    /// output key one edit away from an operator name.
    ///
    /// Unlike [`Self::compile`], which stops at the first problem and
    /// leaves call errors to evaluation, this reports them all, including
    /// those in branches a given input never reaches. `mode` chooses
    /// strict or templating mode for this check (see [`crate::CheckMode`]).
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::{CheckMode, DiagnosticCode, Engine};
    ///
    /// let engine = Engine::new();
    /// let diags = engine.check(r#"{"if": [true, {"vr": "x"}, {"map": [1]}]}"#, CheckMode::Engine);
    /// assert_eq!(diags.len(), 2);
    /// assert_eq!(diags[0].code, DiagnosticCode::UnknownOperator);
    /// assert_eq!(diags[0].pointer, "/if/1");
    /// assert!(diags[0].message.contains("did you mean `var`"));
    /// assert_eq!(diags[1].code, DiagnosticCode::ArgumentCount);
    /// ```
    pub fn check<R: crate::IntoLogic>(
        &self,
        rule: R,
        mode: crate::CheckMode,
    ) -> Vec<crate::Diagnostic> {
        let owned = match rule.logic_source() {
            Ok(owned) => owned,
            Err(err) => return crate::CompileError::from_error(err).diagnostics,
        };
        let templating = match mode {
            crate::CheckMode::Engine => self.is_templating_enabled(),
            crate::CheckMode::Strict => false,
            crate::CheckMode::Template => true,
        };
        crate::check::check(self, &owned, templating)
    }

    /// [`Self::compile`], refusing a rule that [`Self::check`] finds any
    /// error in: strict up front, where `compile` leaves a call that cannot
    /// succeed to fail at evaluation. The error carries every diagnostic,
    /// warnings included. A rule with only warnings compiles.
    ///
    /// # Errors
    ///
    /// [`crate::CompileError`] listing the diagnostics.
    pub fn compile_checked<R: crate::IntoLogic>(
        &self,
        rule: R,
    ) -> std::result::Result<Logic, crate::CompileError> {
        let owned = rule
            .logic_source()
            .map_err(crate::CompileError::from_error)?;
        let diagnostics = crate::check::check(self, &owned, self.is_templating_enabled());
        if diagnostics
            .iter()
            .any(|d| d.severity == crate::Severity::Error)
        {
            return Err(crate::CompileError { diagnostics });
        }
        Logic::compile_with(owned, self).map_err(crate::CompileError::from_error)
    }

    /// Compile `rule` in templating mode, whatever mode the engine was
    /// built with: a multi-key object is an output template and an
    /// unknown operator key is an output field, as on an engine built
    /// [`with_templating(true)`](crate::EngineBuilder::with_templating).
    /// The engine's custom operators, template key escape and folding
    /// setting apply as usual.
    ///
    /// With [`Self::compile_strict`] this lets one engine, with one set of
    /// custom operators, both check rules strictly and compile output
    /// templates, instead of a host building an engine per mode.
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let template = r#"{"user": {"var": "name"}, "source": "api"}"#;
    /// assert!(engine.compile(template).is_err());
    ///
    /// let logic = engine.compile_template(template).unwrap();
    /// let out = engine.session().eval_str(&logic, r#"{"name": "ana"}"#).unwrap();
    /// assert_eq!(out, r#"{"user":"ana","source":"api"}"#);
    /// ```
    #[cfg(feature = "templating")]
    #[cfg_attr(docsrs, doc(cfg(feature = "templating")))]
    pub fn compile_template<R: crate::IntoLogic>(&self, rule: R) -> Result<Logic> {
        let owned = rule.logic_source()?;
        Logic::compile_in_mode(owned, self, true)
    }

    /// Compile `rule` outside templating mode, whatever mode the engine
    /// was built with: a multi-key object is an error, and an unknown
    /// operator is not an output field, exactly as on an engine built
    /// without templating. The engine's custom operators and folding
    /// setting apply as usual.
    ///
    /// On an engine built without templating this is [`Self::compile`].
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "templating")] {
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::builder().with_templating(true).build();
    /// let typo = r#"{"user": {"var": "name"}, "sourec": "api"}"#;
    /// assert!(engine.compile(typo).is_ok());          // an output template
    /// assert!(engine.compile_strict(typo).is_err());  // not a rule
    /// # }
    /// ```
    pub fn compile_strict<R: crate::IntoLogic>(&self, rule: R) -> Result<Logic> {
        let owned = rule.logic_source()?;
        Logic::compile_in_mode(owned, self, false)
    }

    /// Compile and wrap in an [`Arc`](std::sync::Arc) in one call. Convenience for the
    /// dominant cross-thread-sharing pattern; equivalent to
    /// `Arc::new(engine.compile(rule)?)`.
    pub fn compile_arc<R: crate::IntoLogic>(&self, rule: R) -> Result<std::sync::Arc<Logic>> {
        Ok(std::sync::Arc::new(self.compile(rule)?))
    }

    /// Open a [`crate::TracedSession`] over this engine. Calls made through
    /// the session collect a per-call trace; the bare `eval*` methods on
    /// `Engine` itself pay no trace overhead.
    ///
    /// Available only when the crate is built with `feature = "trace"`.
    ///
    /// # Trace coverage
    ///
    /// The session's one-shot [`crate::TracedSession::eval_str`] compiles
    /// the rule internally with optimization disabled, so every operator
    /// in the rule surfaces a trace step.
    ///
    /// The pre-compiled paths ([`crate::TracedSession::eval`] taking a
    /// `&Logic`) inherit whatever shape that `Logic` was compiled into —
    /// constant sub-expressions folded by [`Self::compile`] won't appear,
    /// since there is no operator left to execute. Use `eval_str` for
    /// full coverage on a one-shot run.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "trace")] {
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let run = engine
    ///     .trace()
    ///     .eval_str(r#"{"+": [1, 2]}"#, "null");
    /// assert_eq!(run.result.unwrap(), "3");
    /// // run.steps is the per-node execution log;
    /// // run.expression_tree is the rule's compile-time tree shape.
    /// # }
    /// ```
    #[cfg(feature = "trace")]
    #[cfg_attr(docsrs, doc(cfg(feature = "trace")))]
    #[inline]
    pub fn trace(&self) -> crate::trace::TracedSession<'_> {
        crate::trace::TracedSession::new(self)
    }

    /// Evaluate compiled logic against arena-resident data — **raw tier**.
    ///
    /// The caller owns the [`bumpalo::Bump`] lifecycle and may `reset()`
    /// it between calls; the returned `&DataValue<'a>` borrows from the
    /// arena, so it must be dropped before the next reset (enforced by
    /// the borrow checker). For ergonomic owned/typed/JSON-string output,
    /// prefer [`Self::eval`] / [`Self::eval_str`]
    // `Self::eval_into` is gated behind `serde_json`; link conditionally.
    #[cfg_attr(feature = "serde_json", doc = "/ [`Self::eval_into`].")]
    #[cfg_attr(
        not(feature = "serde_json"),
        doc = "(plus `Self::eval_into` with the `serde_json` feature)."
    )]
    ///
    /// # Example
    ///
    /// ```rust
    /// use bumpalo::Bump;
    /// use datalogic_rs::{Engine, DataValue};
    ///
    /// let engine = Engine::new();
    /// let compiled = engine.compile(r#"{"+": [{"var": "x"}, 2]}"#).unwrap();
    ///
    /// let arena = Bump::new();
    /// let data = DataValue::from_str(r#"{"x": 40}"#, &arena).unwrap();
    /// let result = engine.evaluate(&compiled, data, &arena).unwrap();
    /// assert_eq!(result.as_i64(), Some(42));
    /// ```
    ///
    /// `data` accepts any input shape understood by [`crate::EvalInput`]:
    /// `&'a DataValue<'a>` (zero-cost passthrough), `DataValue<'a>`
    /// (single arena alloc), `&str` (JSON-parsed), `&OwnedDataValue`
    /// (deep-borrowed), [`&ParsedData`](crate::ParsedData) (zero-cost
    /// passthrough of a parse-once handle), or `&serde_json::Value`
    /// (gated on `serde_json`).
    #[inline(always)]
    pub fn evaluate<'a, D: crate::EvalInput<'a>>(
        &self,
        compiled: &'a Logic,
        data: D,
        arena: &'a bumpalo::Bump,
    ) -> Result<&'a crate::arena::DataValue<'a>> {
        // The steps of [`Self::run`], written out: through `run`'s closures
        // this body measured ~1.2 ns slower per call (a third of the cost
        // of evaluating a folded rule), so the hot entry points keep the
        // straight-line form.
        let compiled = compiled.for_engine(self);
        let _depth_guard = self.enter_dispatch_boundary()?;
        let data_ref = data.into_arena_for(compiled, self, arena)?;
        let mut ctx = self.new_context(compiled, data_ref);
        match self.dispatch_node(&compiled.root, &mut ctx, arena) {
            Ok(av) => Ok(av),
            Err(e) => Err(e.decorated(ctx.take_error_path(), compiled)),
        }
    }

    /// The evaluation body, with hooks: the traced run uses it.
    /// [`Self::evaluate`] and [`Self::evaluate_metered`] write the same
    /// steps out by hand (see there why); keep the three in step.
    ///
    /// In order: pick the rule as this engine runs it
    /// ([`Logic::for_engine`]), enter the re-entrancy guard, bring the
    /// input in with `input` (given that rule, so an owned input can be
    /// projected onto what it reads), build the context, let `prepare`
    /// adjust it (a per-call budget, a tracer), dispatch the root, and
    /// decorate an error with its breadcrumb. `finish` sees the result and
    /// the context together, for what is read off the context afterwards
    /// (operations spent, the trace).
    ///
    #[cfg(feature = "trace")]
    #[inline(always)]
    pub(crate) fn run<'a, R>(
        &self,
        compiled: &'a Logic,
        arena: &'a bumpalo::Bump,
        input: impl FnOnce(&'a Logic) -> Result<&'a crate::arena::DataValue<'a>>,
        prepare: impl FnOnce(&mut crate::arena::ContextStack<'a>),
        finish: impl FnOnce(
            Result<&'a crate::arena::DataValue<'a>>,
            &mut crate::arena::ContextStack<'a>,
        ) -> Result<R>,
    ) -> Result<R> {
        let compiled = compiled.for_engine(self);
        let _depth_guard = self.enter_dispatch_boundary()?;
        let data_ref = input(compiled)?;
        let mut ctx = self.new_context(compiled, data_ref);
        prepare(&mut ctx);
        let result = match self.dispatch_node(&compiled.root, &mut ctx, arena) {
            Ok(av) => Ok(av),
            Err(e) => Err(e.decorated(ctx.take_error_path(), compiled)),
        };
        finish(result, &mut ctx)
    }

    /// Build the context stack for one evaluation of `compiled` over
    /// `data_ref`: ancestor tracking as the scope pass decided, and the
    /// engine-wide operation budget when one is configured. Every
    /// evaluation entry point (plain, metered, traced) starts here so a
    /// per-evaluation knob is wired in exactly one place.
    #[inline(always)]
    pub(crate) fn new_context<'a>(
        &self,
        compiled: &'a Logic,
        data_ref: &'a crate::arena::DataValue<'a>,
    ) -> crate::arena::ContextStack<'a> {
        // Only the budget branch mutates; without the feature the binding
        // is returned as built.
        #[cfg_attr(not(feature = "budget"), allow(unused_mut))]
        let mut ctx = crate::arena::ContextStack::new(data_ref, compiled.needs_ancestor_frames);
        #[cfg(feature = "budget")]
        if let Some(budget) = self.settings.config.ops_budget {
            ctx.set_budget(budget);
        }
        ctx
    }

    /// The budget a metered call runs under: `explicit` when the caller
    /// gave one, else the engine-wide
    /// [`EvaluationConfig::ops_budget`](crate::EvaluationConfig::ops_budget),
    /// else unbounded (`u64::MAX`).
    ///
    /// This is the precedence [`Self::evaluate`] applies implicitly, made
    /// available to callers of [`Self::evaluate_metered`] — the language
    /// bindings' "omit the budget to use the engine's" forms all resolve
    /// through here, so a rule metered from any of them reports the same
    /// number.
    #[cfg(feature = "budget")]
    #[cfg_attr(docsrs, doc(cfg(feature = "budget")))]
    #[inline]
    pub fn resolve_ops_budget(&self, explicit: Option<u64>) -> u64 {
        explicit
            .or(self.settings.config.ops_budget)
            .unwrap_or(u64::MAX)
    }

    /// Evaluate under an explicit operation budget, reporting what the
    /// evaluation spent.
    ///
    /// Same evaluation as [`Self::evaluate`], with two differences: the
    /// `budget` argument overrides
    /// [`EvaluationConfig::ops_budget`](crate::EvaluationConfig::ops_budget)
    /// for this call, and the result carries the operation count next to
    /// the value. Pass `u64::MAX` to meter without bounding.
    ///
    /// # What the count means
    ///
    /// "Nodes dispatched at runtime, plus whatever operators charge" —
    /// not "nodes in the source". Concretely:
    ///
    /// - **1 per dispatched node.** Literals cost 0 (they return before
    ///   dispatch), and so do constant-folded subtrees — the compiler
    ///   already did that work, and charging for it would make the count
    ///   depend on whether folding was enabled.
    /// - **1 per item an iterator examines**, charged by `map`, `filter`,
    ///   `reduce`, the quantifiers, `sort` and friends as soon as the
    ///   source is resolved. This is what keeps the number honest when a
    ///   compile-time fast path evaluates a predicate inline instead of
    ///   dispatching its body: the iteration costs at least its input
    ///   length whichever path runs.
    /// - **1 per item a collection operator copies, examines or
    ///   compares**, charged before the work: `merge` per item of each
    ///   array argument, `in` per item of an array haystack, `missing` /
    ///   `missing_some` per path of a path list from data, `keys` /
    ///   `values` / `entries` per key, a stepped `slice` per item produced,
    ///   `distinct` / `group_by` per comparison, `sort` n·⌈log₂ n⌉, and
    ///   one-argument arithmetic per array item. Iterating an object costs
    ///   1 per key.
    /// - **The structural walk of an equality between containers**: 1 per
    ///   array element, `|a| x |b|` per object level, for each level the
    ///   comparison reaches.
    /// - **1 per whole 64 bytes of string a string operator reads** (`cat`,
    ///   `substr`, `upper`, `lower`, `trim`, `length`, `split`, `in` on a
    ///   string, `starts_with` / `ends_with`, string `slice`), so strings
    ///   under 64 bytes are free.
    /// - **Whatever an operator charges for itself.** The tensor family
    ///   prices each operator at `max(elements read, elements produced)`,
    ///   before it allocates. A [`crate::CustomOperator`] that walks a
    ///   large input should do the same via
    ///   [`EvalContext::charge`](crate::operator::EvalContext::charge).
    /// - **A CSE-memoised subtree is charged once**, on the evaluation
    ///   that fills the slot.
    ///
    /// The count is deterministic for a pinned crate version and a given
    /// (rule, data) pair, whichever evaluation path runs: an iterator fast
    /// path charges what the general path would for the same data, so
    /// tracing and [`MissingVar::Error`](crate::MissingVar::Error), which
    /// turn the fast paths off, leave it unchanged. It is not stable
    /// *across* versions: a new fold changes what gets dispatched. Budget
    /// for the work you want to allow, not for an exact number you
    /// measured.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "budget")] {
    /// use bumpalo::Bump;
    /// use datalogic_rs::{Engine, ErrorCode};
    ///
    /// let engine = Engine::new();
    /// let compiled = engine.compile(r#"{"map": [{"var": "xs"}, {"*": [{"var": ""}, 2]}]}"#).unwrap();
    /// let arena = Bump::new();
    ///
    /// let metered = engine
    ///     .evaluate_metered(&compiled, r#"{"xs": [1, 2, 3]}"#, &arena, u64::MAX)
    ///     .unwrap();
    /// assert_eq!(metered.value.to_json_string(), "[2,4,6]");
    /// assert!(metered.ops >= 3, "at least one operation per item");
    ///
    /// // Too tight a budget refuses the evaluation instead of running it.
    /// let err = engine
    ///     .evaluate_metered(&compiled, r#"{"xs": [1, 2, 3]}"#, &arena, 2)
    ///     .unwrap_err();
    /// assert_eq!(err.code(), ErrorCode::BudgetExceeded);
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// [`ErrorKind::BudgetExceeded`](crate::ErrorKind::BudgetExceeded)
    /// when the rule charges past `budget`, plus every error
    /// [`Self::evaluate`] can return. To observe the operations spent
    /// even when evaluation fails, call [`Self::evaluate_metered_report`].
    #[cfg(feature = "budget")]
    #[cfg_attr(docsrs, doc(cfg(feature = "budget")))]
    pub fn evaluate_metered<'a, D: crate::EvalInput<'a>>(
        &self,
        compiled: &'a Logic,
        data: D,
        arena: &'a bumpalo::Bump,
        budget: u64,
    ) -> Result<Metered<&'a crate::arena::DataValue<'a>>> {
        self.evaluate_metered_report(compiled, data, arena, budget)
            .transpose()
    }

    /// Failure-inclusive counterpart to [`Self::evaluate_metered`]:
    /// evaluates under `budget` and returns a [`Metered`] report whose
    /// `ops` is populated whether evaluation succeeds (`value: Ok(...)`)
    /// or fails (`value: Err(...)`).
    ///
    /// Hosts that reserve capacity from a shared parent ledger before an
    /// expression runs can reconcile the reservation against `report.ops`
    /// on both successful and failed evaluations, instead of either
    /// dropping failed work or conservatively charging the entire
    /// reservation.
    ///
    /// # Charge semantics on failure
    ///
    /// - **Ordinary runtime failures** (`Thrown`, `VariableNotFound`,
    ///   `TypeError`, `ArithmeticError`, `Custom`, …): `report.ops` is
    ///   the exact number of operations charged before the failing node or
    ///   operator aborted. Because every node and operator charges
    ///   *before* doing its work, the failing node's own dispatch charge
    ///   (and any pre-allocation charge it made before raising) is
    ///   included. The [`crate::Error`] in `report.value` retains its full
    ///   [`Error::code`](crate::Error::code),
    ///   [`Error::operator`](crate::Error::operator), and
    ///   [`Error::node_ids`](crate::Error::node_ids) breadcrumb.
    /// - **Budget exhaustion**
    ///   ([`ErrorKind::BudgetExceeded { budget, spent }`](crate::ErrorKind::BudgetExceeded)):
    ///   every charge is added to the running counter before comparing
    ///   against `budget`, so `report.ops` equals `spent` — the attempted
    ///   total that crossed the ceiling. Do not add `report.ops` and
    ///   `spent` together (they are the same counter); a host debiting a
    ///   reserved slice of size `budget` can debit `report.ops.min(budget)`
    ///   for bounded work or `report.ops` for attempted work.
    /// - **Pre-dispatch failures** (input adaptation / JSON parse errors
    ///   from `data`, or exceeding `max_recursion_depth` at the boundary):
    ///   no node is dispatched, so `report.ops == 0`.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "budget")] {
    /// use bumpalo::Bump;
    /// use datalogic_rs::{Engine, ErrorCode, Metered};
    ///
    /// let engine = Engine::new();
    /// let compiled = engine.compile(r#"{"/": [1, {"var": "zero"}]}"#).unwrap();
    /// let arena = Bump::new();
    ///
    /// let Metered { value, ops } = engine
    ///     .evaluate_metered_report(&compiled, r#"{"zero": 0}"#, &arena, 100);
    /// let err = value.unwrap_err();
    /// assert_eq!(err.code(), ErrorCode::Thrown);
    /// assert_eq!(err.operator(), Some("/"));
    /// assert_eq!(ops, 2, "the `/` and the `var` were dispatched before division failed");
    /// # }
    /// ```
    #[cfg(feature = "budget")]
    #[cfg_attr(docsrs, doc(cfg(feature = "budget")))]
    pub fn evaluate_metered_report<'a, D: crate::EvalInput<'a>>(
        &self,
        compiled: &'a Logic,
        data: D,
        arena: &'a bumpalo::Bump,
        budget: u64,
    ) -> Metered<Result<&'a crate::arena::DataValue<'a>>> {
        // [`Self::evaluate`]'s straight-line body, with the budget set and
        // `ctx.ops_spent()` preserved on both `Ok` and `Err`.
        let compiled = compiled.for_engine(self);
        let _depth_guard = match self.enter_dispatch_boundary() {
            Ok(guard) => guard,
            Err(e) => {
                return Metered {
                    value: Err(e),
                    ops: 0,
                };
            }
        };
        let data_ref = match data.into_arena_for(compiled, self, arena) {
            Ok(av) => av,
            Err(e) => {
                return Metered {
                    value: Err(e),
                    ops: 0,
                };
            }
        };
        let mut ctx = self.new_context(compiled, data_ref);
        ctx.set_budget(budget);
        let value = match self.dispatch_node(&compiled.root, &mut ctx, arena) {
            Ok(value) => Ok(value),
            Err(e) => Err(e.decorated(ctx.take_error_path(), compiled)),
        };
        Metered {
            value,
            ops: ctx.ops_spent(),
        }
    }

    /// Apply the engine's configured truthiness rules
    /// ([`crate::TruthyEvaluator`]) to an evaluated value.
    ///
    /// This is the same coercion `if` / `and` / `or` / `!` apply to
    /// their operands, exposed so callers (and bindings) can collapse
    /// any rule result to a boolean without re-implementing the
    /// engine's configured semantics.
    ///
    /// # Example
    ///
    /// ```rust
    /// use bumpalo::Bump;
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let compiled = engine.compile(r#"{"var": "items"}"#).unwrap();
    /// let arena = Bump::new();
    /// let result = engine.evaluate(&compiled, r#"{"items": [1]}"#, &arena).unwrap();
    /// assert!(engine.truthy(result));
    /// ```
    #[inline]
    pub fn truthy(&self, value: &crate::arena::DataValue<'_>) -> bool {
        crate::arena::truthy_arena(value, self)
    }

    /// [`Self::truthy`] for a value in any representation the host holds:
    /// `&OwnedDataValue`, `&serde_json::Value`, `&ParsedData` or
    /// `&DataValue` (see [`crate::TruthyInput`]).
    ///
    /// Use it instead of re-implementing truthiness on a result: the
    /// answer follows the engine's [`crate::TruthyEvaluator`], and under
    /// the default rules an empty object is falsy, like an empty array,
    /// which hand-written checks often miss.
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "serde_json")] {
    /// use datalogic_rs::Engine;
    /// use serde_json::json;
    ///
    /// let engine = Engine::new();
    /// let result: serde_json::Value = engine
    ///     .eval_into(r#"{"var": "claims"}"#, &json!({"claims": {}}))
    ///     .unwrap();
    /// assert!(!engine.truthy_of(&result));
    /// assert!(engine.truthy_of(&json!({"sub": "ana"})));
    /// # }
    /// ```
    #[inline]
    pub fn truthy_of<V: crate::TruthyInput>(&self, value: V) -> bool {
        value.truthy_with(self)
    }

    /// One-shot evaluation returning [`datavalue::OwnedDataValue`].
    ///
    /// Compiles `rule`, parses `data`, evaluates against a fresh
    /// per-call arena, and deep-clones the result out. For the same
    /// rule run repeatedly, escalate to [`Self::compile`] + a
    /// [`Session`](crate::Session).
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let result = engine.eval(
    ///     r#"{"+": [{"var": "x"}, 1]}"#,
    ///     r#"{"x": 41}"#,
    /// ).unwrap();
    /// assert_eq!(result.as_i64(), Some(42));
    /// ```
    pub fn eval<R, D>(&self, rule: R, data: D) -> Result<datavalue::OwnedDataValue>
    where
        R: crate::IntoLogic,
        D: crate::OwnedInput,
    {
        self.eval_with::<datavalue::OwnedDataValue, _, _>(rule, data)
    }

    /// One-shot evaluation returning a JSON [`String`].
    ///
    /// # Example
    ///
    /// ```rust
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let result = engine.eval_str(
    ///     r#"{"==": [{"var": "x"}, 5]}"#,
    ///     r#"{"x": 5}"#,
    /// ).unwrap();
    /// assert_eq!(result, "true");
    /// ```
    pub fn eval_str<R, D>(&self, rule: R, data: D) -> Result<String>
    where
        R: crate::IntoLogic,
        D: crate::OwnedInput,
    {
        self.eval_with::<String, _, _>(rule, data)
    }

    /// One-shot evaluation deserialised into a typed `T: DeserializeOwned`.
    ///
    /// Use `T = serde_json::Value` for a JSON `Value` result; use a typed
    /// struct for direct mapping. Internally routes through `serde_json`
    /// (round-trips the result through a JSON value).
    ///
    /// # Example
    ///
    /// ```rust
    /// # #[cfg(feature = "serde_json")] {
    /// use datalogic_rs::Engine;
    /// use serde_json::Value;
    ///
    /// let engine = Engine::new();
    /// let result: Value = engine.eval_into(
    ///     r#"{"+": [{"var": "x"}, 1]}"#,
    ///     r#"{"x": 41}"#,
    /// ).unwrap();
    /// assert_eq!(result, Value::from(42));
    /// # }
    /// ```
    #[cfg(feature = "serde_json")]
    #[cfg_attr(docsrs, doc(cfg(feature = "serde_json")))]
    pub fn eval_into<T, R, D>(&self, rule: R, data: D) -> Result<T>
    where
        T: serde::de::DeserializeOwned,
        R: crate::IntoLogic,
        D: crate::OwnedInput,
    {
        let value: serde_json::Value = self.eval_with(rule, data)?;
        serde_json::from_value(value).map_err(crate::Error::from)
    }

    /// One-shot evaluation converting the result straight into `O`, any
    /// [`crate::FromDataValue`] output: `OwnedDataValue`, a JSON
    /// [`String`], or a `serde_json::Value` (with `serde_json`).
    ///
    /// For a `serde_json::Value` result this is one conversion, where
    /// `eval_into::<serde_json::Value>` (with `serde_json`) builds the value and then
    /// deserializes it into a second one.
    ///
    /// ```rust
    /// # #[cfg(feature = "serde_json")] {
    /// use datalogic_rs::Engine;
    ///
    /// let engine = Engine::new();
    /// let user: serde_json::Value = engine
    ///     .eval_as(r#"{"var": "user"}"#, r#"{"user": {"id": 7}}"#)
    ///     .unwrap();
    /// assert_eq!(user["id"], 7);
    /// # }
    /// ```
    pub fn eval_as<O, R, D>(&self, rule: R, data: D) -> Result<O>
    where
        O: crate::FromDataValue,
        R: crate::IntoLogic,
        D: crate::OwnedInput,
    {
        self.eval_with(rule, data)
    }

    /// Internal generic shared by `eval` / `eval_str` / `eval_into` /
    /// `eval_as`. Compiles, brings the data into a per-call arena (an
    /// owned or `serde_json` input projected onto what the rule reads, as
    /// [`Self::evaluate`] does), evaluates, and converts the result
    /// through [`crate::FromDataValue`].
    fn eval_with<O, R, D>(&self, rule: R, data: D) -> Result<O>
    where
        O: crate::FromDataValue,
        R: crate::IntoLogic,
        D: crate::OwnedInput,
    {
        let compiled = self.compile(rule)?;
        data.lend_arena_for(&compiled, self, |data, arena| {
            let result = self.evaluate(&compiled, data, arena)?;
            O::from_arena(result)
        })
    }

    /// Bump the per-thread dispatch-boundary depth counter, bailing with
    /// `ConfigurationError` if the configured cap is reached. Returns a
    /// guard that decrements the counter on drop (covers `?` early returns
    /// and panics).
    ///
    /// Called from every public boundary entry point (`Engine::evaluate`,
    /// `TracedSession::evaluate`, …). The counter is thread-local rather
    /// than per-`ContextStack` so it survives across nested
    /// `engine.evaluate(...)` calls — the scenario a `CustomOperator`
    /// holding `Arc<Engine>` creates by re-entering the engine from inside
    /// its own `evaluate(...)`.
    ///
    /// Tokio safety: dispatch is sync, so a task cannot `.await` while the
    /// counter is raised; between dispatches the value is restored to its
    /// prior level (zero at the outermost call). Cross-thread task migration
    /// thus starts fresh on whatever thread the task lands on.
    #[inline(always)]
    pub(crate) fn enter_dispatch_boundary(&self) -> Result<DepthGuard> {
        // Built-in operators can't re-enter `Engine::evaluate` (only a
        // `CustomOperator` holding `Arc<Engine>` can); when the registry
        // is empty, cross-evaluate recursion is impossible and we skip
        // the TLS bookkeeping. The pure-built-in benchmarks pay zero.
        if self.custom_operators.is_empty() {
            return Ok(DepthGuard(DepthGuard::NOOP));
        }
        self.enter_dispatch_boundary_checked()
    }

    /// Slow path of [`Self::enter_dispatch_boundary`] — hit only when
    /// the engine has at least one custom operator registered. Marked
    /// `#[cold]` and `#[inline(never)]` so the hot fast-path stays
    /// inline-friendly.
    #[cold]
    #[inline(never)]
    fn enter_dispatch_boundary_checked(&self) -> Result<DepthGuard> {
        let prev_depth = DISPATCH_DEPTH.with(Cell::get);
        if prev_depth >= self.settings.config.max_recursion_depth {
            return Err(crate::Error::configuration_error(format!(
                "max recursion depth exceeded ({})",
                self.settings.config.max_recursion_depth
            )));
        }
        DISPATCH_DEPTH.with(|d| d.set(prev_depth + 1));
        Ok(DepthGuard(prev_depth))
    }

    /// Arena-mode dispatch hub. Returns `&'a DataValue<'a>` for every
    /// `CompiledNode` shape — exhaustive match, no value-mode fallback.
    ///
    /// On error, accumulates the failing node's id onto the context stack's
    /// breadcrumb so [`Error`] consumers can surface the failing
    /// path. When a tracer is attached to `ctx`, records a step per
    /// non-literal node (entry context + result/error).
    #[inline(always)]
    pub(crate) fn dispatch_node<'a>(
        &self,
        node: &'a CompiledNode,
        ctx: &mut crate::arena::ContextStack<'a>,
        arena: &'a bumpalo::Bump,
    ) -> Result<&'a crate::arena::DataValue<'a>> {
        // Literal fast path — no breadcrumb push, no trace step.
        if let CompiledNode::Value { value, lit, .. } = node {
            // Every literal reachable from a `Logic` carries a pre-built
            // `PreLit` — trivial ones from `precompute_lit` at node
            // construction, composites from the `populate_lits` pass —
            // so the hot path is a borrow, not a conversion.
            if let Some(av) = lit {
                return Ok(av.as_ref());
            }
            // Synthetic composite wrappers built outside the compile
            // pipeline fall through to per-call arena conversion here.
            return Ok(literal_fallback(value, arena));
        }

        // Snapshot context for trace BEFORE recursing — children will
        // mutate iteration frames. Cheap when no tracer is attached.
        #[cfg(feature = "trace")]
        let ctx_snapshot: Option<serde_json::Value> =
            ctx.has_tracer().then(|| ctx.current_data_as_value());

        // One operation per dispatched node, charged before the work.
        // Literals returned above cost nothing — they are data the
        // compiler already resolved, not work the rule asked for. With
        // the `budget` feature off `charge` is an inlined `Ok(())`, so
        // this folds to the bare inner dispatch.
        let mut result = match ctx.charge(1) {
            Ok(()) => dispatch::dispatch_node_inner(self, node, ctx, arena),
            Err(e) => Err(e),
        };

        // Accumulate the failing node's id on every Err, and name the
        // innermost failing operator. Errors are rare, so this is one cold
        // call and structured-error consumers get the breadcrumb.
        if let Err(e) = &mut result {
            record_failure(e, node, ctx);
        }

        #[cfg(feature = "trace")]
        if let Some(ctx_data) = ctx_snapshot {
            ctx.record_node_result(node.id(), ctx_data, &result);
        }

        result
    }

    /// Evaluate an iteration body (map/filter/reduce/all/some/none) with the
    /// trace collector's iteration index/total markers set around it. The
    /// markers are no-ops when no tracer is attached, so plain-mode callers
    /// pay only one branch per iteration.
    #[inline]
    pub(crate) fn run_iter_body<'a>(
        &self,
        body: &'a CompiledNode,
        ctx: &mut crate::arena::ContextStack<'a>,
        arena: &'a bumpalo::Bump,
        _index: u32,
        _total: u32,
    ) -> Result<&'a crate::arena::DataValue<'a>> {
        #[cfg(feature = "trace")]
        ctx.trace_push_iteration(_index, _total);
        let res = self.dispatch_node(body, ctx, arena);
        #[cfg(feature = "trace")]
        ctx.trace_pop_iteration();
        res
    }
}