kelora 2.0.0

#![allow(dead_code)] // Engine keeps analysis/helpers that are not all exercised by the current binary
use anyhow::Result;
use rhai::{BinaryExpr, Dynamic, Engine, EvalAltResult, Expr, FnCallExpr, Scope, Stmt, Token, AST};
use std::collections::HashMap;

use rhai::debugger::{DebuggerCommand, DebuggerEvent};

use crate::event::Event;
use crate::rhai_functions;
use crate::rhai_functions::datetime::DateTimeWrapper;

mod debug;
pub use debug::{DebugConfig, DebugTracker, ErrorEnhancer};

use rhai::Map;

/// Truncate text for display, respecting UTF-8 character boundaries.
fn truncate_for_display(text: &str, max_len: usize) -> String {
    if text.chars().count() > max_len {
        let truncated: String = text.chars().take(max_len.saturating_sub(3)).collect();
        format!("{}...", truncated)
    } else {
        text.to_string()
    }
}

#[derive(Debug)]
pub struct ConfMutationError;

impl std::fmt::Display for ConfMutationError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "conf map is read-only outside --begin; modifications are not allowed"
        )
    }
}

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

fn dynamics_equal(lhs: &Dynamic, rhs: &Dynamic) -> bool {
    if lhs.type_name() != rhs.type_name() {
        return false;
    }

    // Primitive comparisons
    if let (Some(l), Some(r)) = (lhs.as_int().ok(), rhs.as_int().ok()) {
        return l == r;
    }
    if let (Some(l), Some(r)) = (lhs.as_float().ok(), rhs.as_float().ok()) {
        return l == r;
    }
    if let (Some(l), Some(r)) = (lhs.as_bool().ok(), rhs.as_bool().ok()) {
        return l == r;
    }
    if let (Ok(l), Ok(r)) = (lhs.clone().into_string(), rhs.clone().into_string()) {
        return l == r;
    }

    // Array comparison
    if let (Some(l_arr), Some(r_arr)) = (
        lhs.clone().try_cast::<rhai::Array>(),
        rhs.clone().try_cast::<rhai::Array>(),
    ) {
        if l_arr.len() != r_arr.len() {
            return false;
        }
        return l_arr
            .iter()
            .zip(r_arr.iter())
            .all(|(l, r)| dynamics_equal(l, r));
    }

    // Map comparison
    if let (Some(l_map), Some(r_map)) = (
        lhs.clone().try_cast::<rhai::Map>(),
        rhs.clone().try_cast::<rhai::Map>(),
    ) {
        return maps_equal(&l_map, &r_map);
    }

    false
}

fn maps_equal(lhs: &rhai::Map, rhs: &rhai::Map) -> bool {
    if lhs.len() != rhs.len() {
        return false;
    }

    lhs.iter().all(|(k, v)| match rhs.get(k) {
        Some(rv) => dynamics_equal(v, rv),
        None => false,
    })
}

use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};

// Execution Tracer for step-by-step debugging
pub struct ExecutionTracer {
    config: DebugConfig,
    current_event: Arc<Mutex<u64>>,
    step_counter: Arc<Mutex<u32>>,
}

impl ExecutionTracer {
    pub fn new(config: DebugConfig) -> Self {
        ExecutionTracer {
            config,
            current_event: Arc::new(Mutex::new(0)),
            step_counter: Arc::new(Mutex::new(0)),
        }
    }

    pub fn trace_stage_execution(&self, stage_number: usize, stage_type: &str) {
        if self.config.verbosity >= 1 {
            let prefix = if self.config.use_emoji {
                "🔹"
            } else {
                "kelora: "
            };
            eprintln!(
                "{}Executing stage {} ({})",
                prefix, stage_number, stage_type
            );
        }
    }

    pub fn trace_step(&self, _event_num: u64, step_info: &str, result: &str) {
        if self.config.verbosity >= 2 {
            eprintln!("  → {} → {}", step_info, result);
        }
    }

    pub fn trace_event_start(&self, event_num: u64, event_data: &str) {
        if self.config.verbosity >= 2 {
            eprintln!("  Filter execution trace for event {}:", event_num);
            eprintln!("    Event: {}", truncate_for_display(event_data, 100));
        }
    }

    pub fn trace_event_result(&self, result: bool, action: &str) {
        if self.config.verbosity >= 2 {
            eprintln!("    Result: {} ({})", result, action);
        }
    }

    pub fn trace_expression_evaluation(&self, expression: &str, intermediate_result: &str) {
        if self.config.verbosity >= 3 {
            eprintln!("    Eval: {} → {}", expression, intermediate_result);
        }
    }

    // Enhanced detailed tracing for -vvv level
    pub fn trace_detailed_step(
        &self,
        context: &str,
        operation: &str,
        input: &str,
        output: &str,
        step_type: &str,
    ) {
        if self.config.verbosity >= 3 {
            let step_num = {
                match self.step_counter.lock() {
                    Ok(mut counter) => {
                        *counter += 1;
                        *counter
                    }
                    Err(_) => {
                        // If mutex is poisoned, continue with a default value
                        eprintln!("Warning: Step counter mutex poisoned, using default");
                        0
                    }
                }
            };

            eprintln!(
                "    [Step {}:{}] {}: {} → {}",
                step_num,
                context,
                operation,
                truncate_for_display(input, 30),
                truncate_for_display(output, 30)
            );

            if step_type != "default" {
                eprintln!("      Type: {}", step_type);
            }
        }
    }

    pub fn trace_scope_inspection(&self, scope: &rhai::Scope) {
        if self.config.verbosity >= 3 {
            eprintln!("    Scope contents:");
            let mut scope_items: Vec<_> = scope.iter().collect();
            scope_items.sort_by(|a, b| a.0.cmp(b.0));
            for (name, _is_const, value) in scope_items {
                let type_info = value.type_name();
                let preview = format!("{:?}", value);
                eprintln!("      {} ({}): {}", name, type_info, preview);
            }
        }
    }

    pub fn next_event(&self) -> u64 {
        if let Ok(mut counter) = self.current_event.lock() {
            *counter += 1;
            *counter
        } else {
            0
        }
    }
}

impl Clone for ExecutionTracer {
    fn clone(&self) -> Self {
        ExecutionTracer {
            config: self.config.clone(),
            current_event: Arc::clone(&self.current_event),
            step_counter: Arc::clone(&self.step_counter),
        }
    }
}

/// Represents the type of access to a field in a Rhai expression
#[derive(Debug, Clone, PartialEq, Eq)]
enum AccessType {
    Read,      // Field is being read/accessed
    Write,     // Field is being assigned (LHS of simple assignment)
    ReadWrite, // Field is both read and written (compound assignments: +=, -=, etc.)
}

/// Represents a field access with its access type
#[derive(Debug, Clone)]
struct FieldAccess {
    field_name: String,
    access_type: AccessType,
}

#[derive(Clone)]
struct NativePredicate {
    root: NativeNode,
}

#[derive(Clone)]
enum NativeNode {
    And(Vec<NativeNode>),
    Or(Vec<NativeNode>),
    Not(Box<NativeNode>),
    Value(NativeValueExpr),
    Compare {
        op: CompareOp,
        lhs: NativeValueExpr,
        rhs: NativeValueExpr,
    },
    /// String method calls like e.field.contains("x")
    StringMethod {
        op: StringOp,
        /// The string value to operate on (e.g., e.message)
        target: NativeValueExpr,
        /// The argument (e.g., "error" in contains("error"))
        arg: NativeValueExpr,
    },
}

#[derive(Clone, Copy)]
enum StringOp {
    Contains,
    StartsWith,
    EndsWith,
}

#[derive(Clone)]
enum NativeValueExpr {
    Field(String),
    Literal(NativeValue),
}

#[derive(Clone, Copy)]
enum CompareOp {
    Eq,
    Ne,
    Lt,
    Le,
    Gt,
    Ge,
}

#[derive(Clone)]
enum NativeValue {
    Bool(bool),
    Int(rhai::INT),
    Float(rhai::FLOAT),
    Str(String),
    Unit,
}

/// Extract field names accessed on variable 'e' from a Rhai AST
///
/// Uses AST walking to find all property access patterns like `e.field_name`
/// including chained method calls like `e.field.to_upper()`.
/// Distinguishes between field reads and writes to avoid false warnings.
///
/// Requires the 'debugging' feature which provides AST access.
fn extract_field_accesses(ast: &AST) -> Vec<FieldAccess> {
    use std::collections::HashMap;

    let mut accesses: HashMap<String, AccessType> = HashMap::new();

    ast.walk(&mut |path| {
        if let Some(node) = path.first() {
            let node_str = format!("{:?}", node);

            // Skip nodes that don't involve Variable(e)
            if !node_str.contains("Variable(e)") {
                return true;
            }

            // Check if this is an assignment statement
            if node_str.contains("Assignment(") {
                extract_assignment_fields(&node_str, &mut accesses);
            } else {
                // Not an assignment - all fields are reads
                extract_read_fields(&node_str, &mut accesses);
            }
        }
        true
    });

    // Convert HashMap to Vec<FieldAccess>
    accesses
        .into_iter()
        .map(|(field_name, access_type)| FieldAccess {
            field_name,
            access_type,
        })
        .collect()
}

/// Method/function names that mutate their target in place rather than returning a new
/// value to assign back. Invoked on `e` (or a field reached from `e`), they change the
/// event without any assignment or field-level write, so `expression_mutates_event` must
/// flag them or the mutation is silently dropped when the event is read back from scope.
///
/// Two groups: Kelora's whole-event mutators (operate on the `e` map directly), and
/// Rhai's in-place array/map mutators (operate on a nested field, e.g. `e.tags.push(x)`).
/// Read-only methods (`has`, `get_path`, `len`, `contains`, …) are deliberately absent so
/// read-only execs keep skipping the writeback. Extend if a new in-place mutator is added.
const MUTATING_CALL_NAMES: &[&str] = &[
    // Kelora whole-event mutators
    "absorb_kv",
    "absorb_logfmt",
    "absorb_json",
    "absorb_jwt",
    "absorb_regex",
    "merge",
    "enrich",
    "rename_field",
    // Rhai in-place array/map mutators
    "push",
    "pop",
    "insert",
    "remove",
    "clear",
    "truncate",
    "reverse",
    "sort",
    "dedup",
    "retain",
    "append",
    "pad",
    "shift",
    "splice",
    "set",
    "mixin",
];

fn expression_mutates_event(ast: &AST, field_accesses: &[FieldAccess]) -> bool {
    if field_accesses
        .iter()
        .any(|fa| matches!(fa.access_type, AccessType::Write | AccessType::ReadWrite))
    {
        return true;
    }

    // Whole-map assignments like `e = ()` do not produce field-level writes but still
    // replace the event payload.
    let ast_text = format!("{:?}", ast.statements());
    if ast_text.contains("Assignment(") && ast_text.contains("lhs: Variable(e)") {
        return true;
    }

    // In-place mutating calls (e.g. `e.absorb_kv("msg")`, `e.tags.push(x)`) have no
    // assignment and no field-level write, but still change the event. Detect them so
    // the writeback in `update_event_from_scope` is not skipped. `path.last()` is the
    // node currently being visited; we only match nodes whose receiver chain is rooted at
    // `e` (method form `e.…(…)`) or whose first argument is `e` (function form
    // `f(e, …)`). Anchoring to the `e` root avoids false positives such as
    // `if e.x { other.push(y) }`, where an unrelated value is the one being mutated.
    let mut mutates = false;
    ast.walk(&mut |path| {
        if let Some(node) = path.last() {
            let node_str = format!("{node:?}");
            // Method form `e.…(…)` / `e.field.…(…)`: the granular node is the dot chain
            // rooted at `e`. Function form `f(e, …)`: `e` is the literal first argument.
            let rooted_at_e = node_str.starts_with("Expr(Dot { lhs: Variable(e)")
                || node_str.contains("args: [Variable(e)");
            if rooted_at_e
                && MUTATING_CALL_NAMES
                    .iter()
                    .any(|name| node_str.contains(&format!("name: {name:?}")))
            {
                mutates = true;
                return false; // stop walking, decision made
            }
        }
        true
    });
    mutates
}

/// Flags indicating which scope variables are used by an expression
#[derive(Clone, Copy, Default)]
struct VariableUsage {
    uses_meta: bool,
    uses_conf: bool,
    uses_line: bool,
    uses_window: bool,
    meta_usage: MetaUsage,
}

#[derive(Clone, Copy, Default)]
struct MetaUsage {
    populate_all: bool,
    line: bool,
    line_num: bool,
    filename: bool,
    parsed_ts: bool,
    span_status: bool,
    span_id: bool,
    span_start: bool,
    span_end: bool,
}

impl MetaUsage {
    fn any(&self) -> bool {
        self.line
            || self.line_num
            || self.filename
            || self.parsed_ts
            || self.span_status
            || self.span_id
            || self.span_start
            || self.span_end
    }
}

/// Detect which scope variables (meta, conf, line) are used in the AST
fn detect_variable_usage(ast: &AST) -> VariableUsage {
    let mut usage = VariableUsage::default();

    ast.walk(&mut |path| {
        if let Some(node) = path.first() {
            let node_str = format!("{:?}", node);

            // Check for Variable references
            if node_str.contains("Variable(meta)") {
                usage.uses_meta = true;

                if node_str.contains("Property(line_num)") {
                    usage.meta_usage.line_num = true;
                }
                if node_str.contains("Property(filename)") {
                    usage.meta_usage.filename = true;
                }
                if node_str.contains("Property(parsed_ts)") {
                    usage.meta_usage.parsed_ts = true;
                }
                if node_str.contains("Property(line)") {
                    usage.meta_usage.line = true;
                }
                if node_str.contains("Property(span_status)") {
                    usage.meta_usage.span_status = true;
                }
                if node_str.contains("Property(span_id)") {
                    usage.meta_usage.span_id = true;
                }
                if node_str.contains("Property(span_start)") {
                    usage.meta_usage.span_start = true;
                }
                if node_str.contains("Property(span_end)") {
                    usage.meta_usage.span_end = true;
                }

                // If `meta` is referenced but no concrete field is visible in the AST dump,
                // keep the old behavior and populate the full metadata map.
                if !usage.meta_usage.any() {
                    usage.meta_usage.populate_all = true;
                }
            }
            if node_str.contains("Variable(conf)") {
                usage.uses_conf = true;
            }
            if node_str.contains("Variable(line)") {
                usage.uses_line = true;
            }
            if node_str.contains("Variable(window)") {
                usage.uses_window = true;
            }
        }
        true
    });

    usage
}

fn build_native_predicate(ast: &AST) -> Option<NativePredicate> {
    let statements = ast.statements();
    if statements.len() != 1 {
        return None;
    }

    let expr = match &statements[0] {
        Stmt::Expr(expr) => expr.as_ref(),
        _ => return None,
    };

    let root = parse_native_node(expr)?;
    Some(NativePredicate { root })
}

fn parse_native_node(expr: &Expr) -> Option<NativeNode> {
    match expr {
        Expr::And(list, _) => {
            let mut nodes = Vec::with_capacity(list.len());
            for item in list.iter() {
                nodes.push(parse_native_node(item)?);
            }
            Some(NativeNode::And(nodes))
        }
        Expr::Or(list, _) => {
            let mut nodes = Vec::with_capacity(list.len());
            for item in list.iter() {
                nodes.push(parse_native_node(item)?);
            }
            Some(NativeNode::Or(nodes))
        }
        Expr::FnCall(call, _) => {
            if let Some(token) = call.op_token.clone() {
                match token {
                    Token::Bang if call.args.len() == 1 => {
                        let node = parse_native_node(&call.args[0])?;
                        Some(NativeNode::Not(Box::new(node)))
                    }
                    Token::EqualsTo => parse_compare_node(CompareOp::Eq, call),
                    Token::NotEqualsTo => parse_compare_node(CompareOp::Ne, call),
                    Token::LessThan => parse_compare_node(CompareOp::Lt, call),
                    Token::LessThanEqualsTo => parse_compare_node(CompareOp::Le, call),
                    Token::GreaterThan => parse_compare_node(CompareOp::Gt, call),
                    Token::GreaterThanEqualsTo => parse_compare_node(CompareOp::Ge, call),
                    _ => None,
                }
            } else {
                None
            }
        }
        Expr::BoolConstant(value, _) => Some(NativeNode::Value(NativeValueExpr::Literal(
            NativeValue::Bool(*value),
        ))),
        // Handle method calls like e.message.contains("error")
        Expr::Dot(binary, _, _) => parse_string_method(binary),
        _ => parse_value_expr(expr).map(NativeNode::Value),
    }
}

/// Parse string method calls like e.field.contains("pattern")
/// AST structure for `e.level.starts_with("ERR")`:
/// Dot { lhs: Variable(e), rhs: Dot { lhs: Property(level), rhs: MethodCall(...) } }
fn parse_string_method(binary: &BinaryExpr) -> Option<NativeNode> {
    // For e.field.method(arg), the structure is:
    // Dot { lhs: Variable(e), rhs: Dot { lhs: Property(field), rhs: MethodCall } }
    // We need to find the innermost Dot that has a MethodCall as rhs

    // Try to extract method call info by walking the Dot chain
    let (field_path, method_name, arg) = extract_method_call(binary)?;

    // Map method name to operation
    let op = match method_name.as_str() {
        "contains" => StringOp::Contains,
        "starts_with" => StringOp::StartsWith,
        "ends_with" => StringOp::EndsWith,
        _ => return None,
    };

    let target = NativeValueExpr::Field(field_path);
    Some(NativeNode::StringMethod { op, target, arg })
}

/// Extract method call info from a Dot expression chain
/// Returns (field_path, method_name, arg) if successful
fn extract_method_call(binary: &BinaryExpr) -> Option<(String, String, NativeValueExpr)> {
    // Check if this is a direct method call: Dot { lhs: field_expr, rhs: MethodCall }
    if let Expr::MethodCall(call, _) = &binary.rhs {
        if call.args.len() == 1 {
            let arg = parse_value_expr(&call.args[0])?;
            let field_path = extract_field_path(&binary.lhs)?;
            return Some((field_path, call.name.to_string(), arg));
        }
    }

    // Check if rhs is another Dot containing a MethodCall
    if let Expr::Dot(inner_binary, _, _) = &binary.rhs {
        // Recursively check the inner Dot
        if let Some((mut path, method, arg)) = extract_method_call(inner_binary) {
            // Prepend the lhs field to the path
            let lhs_path = extract_field_path(&binary.lhs)?;
            if path.is_empty() {
                path = lhs_path;
            } else if !lhs_path.is_empty() {
                path = format!("{}.{}", lhs_path, path);
            }
            return Some((path, method, arg));
        }
    }

    None
}

fn parse_compare_node(op: CompareOp, call: &FnCallExpr) -> Option<NativeNode> {
    if call.args.len() != 2 {
        return None;
    }

    let lhs = parse_value_expr(&call.args[0])?;
    let rhs = parse_value_expr(&call.args[1])?;

    Some(NativeNode::Compare { op, lhs, rhs })
}

fn parse_value_expr(expr: &Expr) -> Option<NativeValueExpr> {
    match expr {
        Expr::BoolConstant(value, _) => Some(NativeValueExpr::Literal(NativeValue::Bool(*value))),
        Expr::IntegerConstant(value, _) => Some(NativeValueExpr::Literal(NativeValue::Int(*value))),
        Expr::FloatConstant(value, _) => {
            Some(NativeValueExpr::Literal(NativeValue::Float(**value)))
        }
        Expr::StringConstant(value, _) => Some(NativeValueExpr::Literal(NativeValue::Str(
            value.to_string(),
        ))),
        Expr::Unit(..) => Some(NativeValueExpr::Literal(NativeValue::Unit)),
        _ => extract_field_path(expr)
            .filter(|path| !path.is_empty())
            .map(NativeValueExpr::Field),
    }
}

fn extract_field_path(expr: &Expr) -> Option<String> {
    match expr {
        Expr::Dot(binary, options, _) if options.is_empty() => {
            let mut base = extract_field_path(&binary.lhs)?;
            let rhs = extract_field_path(&binary.rhs)?;
            if base.is_empty() {
                base = rhs;
            } else {
                base.push('.');
                base.push_str(&rhs);
            }
            Some(base)
        }
        Expr::Index(binary, options, _) if options.is_empty() => {
            let mut base = extract_field_path(&binary.lhs)?;
            let rhs = match &binary.rhs {
                Expr::StringConstant(s, _) => s.to_string(),
                Expr::IntegerConstant(i, _) => i.to_string(),
                _ => return None,
            };

            if !base.is_empty() {
                base.push('.');
            }
            base.push_str(&rhs);
            Some(base)
        }
        Expr::Property(prop, _) => Some(prop.2.to_string()),
        Expr::Variable(var, ..) => {
            let name = &var.1;
            if name == "e" {
                Some(String::new())
            } else {
                None
            }
        }
        _ => None,
    }
}

impl NativePredicate {
    fn evaluate(&self, event: &Event) -> Option<bool> {
        eval_native_node(&self.root, event)
    }
}

fn eval_native_node(node: &NativeNode, event: &Event) -> Option<bool> {
    match node {
        NativeNode::And(nodes) => {
            let mut result = true;
            for n in nodes {
                let value = eval_native_node(n, event)?;
                result &= value;
                if !result {
                    break;
                }
            }
            Some(result)
        }
        NativeNode::Or(nodes) => {
            let mut result = false;
            for n in nodes {
                let value = eval_native_node(n, event)?;
                result |= value;
                if result {
                    break;
                }
            }
            Some(result)
        }
        NativeNode::Not(child) => eval_native_node(child, event).map(|v| !v),
        NativeNode::Value(expr) => match eval_value_expr(expr, event)? {
            NativeValue::Bool(b) => Some(b),
            _ => None,
        },
        NativeNode::Compare { op, lhs, rhs } => {
            let lhs_val = eval_value_expr(lhs, event)?;
            let rhs_val = eval_value_expr(rhs, event)?;
            compare_values(&lhs_val, &rhs_val, *op)
        }
        NativeNode::StringMethod { op, target, arg } => {
            let target_val = eval_value_expr(target, event)?;
            let arg_val = eval_value_expr(arg, event)?;

            // Both must be strings
            let (target_str, arg_str) = match (&target_val, &arg_val) {
                (NativeValue::Str(t), NativeValue::Str(a)) => (t.as_str(), a.as_str()),
                _ => return None,
            };

            let result = match op {
                StringOp::Contains => target_str.contains(arg_str),
                StringOp::StartsWith => target_str.starts_with(arg_str),
                StringOp::EndsWith => target_str.ends_with(arg_str),
            };
            Some(result)
        }
    }
}

fn eval_value_expr(expr: &NativeValueExpr, event: &Event) -> Option<NativeValue> {
    match expr {
        NativeValueExpr::Literal(value) => Some(value.clone()),
        NativeValueExpr::Field(path) => {
            let value = event.fields.get(path).cloned().unwrap_or(Dynamic::UNIT);
            dynamic_to_native(&value)
        }
    }
}

fn dynamic_to_native(value: &Dynamic) -> Option<NativeValue> {
    if value.is_unit() {
        return Some(NativeValue::Unit);
    }

    if let Ok(b) = value.as_bool() {
        return Some(NativeValue::Bool(b));
    }

    if let Ok(i) = value.as_int() {
        return Some(NativeValue::Int(i));
    }

    if let Ok(f) = value.as_float() {
        return Some(NativeValue::Float(f));
    }

    if let Some(s) = value.clone().try_cast::<rhai::ImmutableString>() {
        return Some(NativeValue::Str(s.to_string()));
    }

    if let Some(s) = value.clone().try_cast::<String>() {
        return Some(NativeValue::Str(s));
    }

    None
}

fn compare_values(lhs: &NativeValue, rhs: &NativeValue, op: CompareOp) -> Option<bool> {
    use CompareOp::*;

    match (lhs, rhs) {
        (NativeValue::Unit, NativeValue::Unit) => match op {
            Eq => Some(true),
            Ne => Some(false),
            _ => Some(false),
        },
        (NativeValue::Unit, _) | (_, NativeValue::Unit) => match op {
            Eq => Some(false),
            Ne => Some(true),
            _ => None,
        },
        (NativeValue::Bool(l), NativeValue::Bool(r)) => match op {
            Eq => Some(l == r),
            Ne => Some(l != r),
            _ => None,
        },
        (NativeValue::Str(l), NativeValue::Str(r)) => match op {
            Eq => Some(l == r),
            Ne => Some(l != r),
            _ => None,
        },
        (NativeValue::Int(l), NativeValue::Int(r)) => {
            compare_numbers(*l as rhai::FLOAT, *r as rhai::FLOAT, op)
        }
        (NativeValue::Float(l), NativeValue::Float(r)) => compare_numbers(*l, *r, op),
        (NativeValue::Int(l), NativeValue::Float(r)) => compare_numbers(*l as rhai::FLOAT, *r, op),
        (NativeValue::Float(l), NativeValue::Int(r)) => compare_numbers(*l, *r as rhai::FLOAT, op),
        _ => None,
    }
}

fn compare_numbers(lhs: rhai::FLOAT, rhs: rhai::FLOAT, op: CompareOp) -> Option<bool> {
    use CompareOp::*;
    match op {
        Eq => Some(lhs == rhs),
        Ne => Some(lhs != rhs),
        Lt => Some(lhs < rhs),
        Le => Some(lhs <= rhs),
        Gt => Some(lhs > rhs),
        Ge => Some(lhs >= rhs),
    }
}

/// Extract field accesses from assignment statements
fn extract_assignment_fields(
    node_str: &str,
    accesses: &mut std::collections::HashMap<String, AccessType>,
) {
    use AccessType::*;

    // Determine if this is a compound assignment (+=, -=, *=, etc.)
    let is_compound = node_str.contains("PlusAssign")
        || node_str.contains("MinusAssign")
        || node_str.contains("MultiplyAssign")
        || node_str.contains("DivideAssign")
        || node_str.contains("ModuloAssign")
        || node_str.contains("PowerOfAssign")
        || node_str.contains("ShiftLeftAssign")
        || node_str.contains("ShiftRightAssign")
        || node_str.contains("AndAssign")
        || node_str.contains("OrAssign")
        || node_str.contains("XOrAssign");

    // For assignments, Rhai uses: Stmt(Assignment((op, BinaryExpr { lhs: ..., rhs: ... })))
    // Find the BinaryExpr within the assignment
    if let Some(binary_start) = node_str.find("BinaryExpr {") {
        let binary_section = &node_str[binary_start..];

        // Extract LHS fields (target of assignment)
        if let Some(lhs_start) = binary_section.find("lhs:") {
            // Find the end of the lhs section (before "rhs:")
            let lhs_section = if let Some(rhs_pos) = binary_section[lhs_start..].find(", rhs:") {
                &binary_section[lhs_start..lhs_start + rhs_pos]
            } else {
                &binary_section[lhs_start..]
            };

            let lhs_fields = extract_fields_from_section(lhs_section);
            for field in lhs_fields {
                if is_compound {
                    // Compound assignment: field is both read and written
                    merge_access_type(accesses, field, ReadWrite);
                } else {
                    // Regular assignment: field is only written
                    merge_access_type(accesses, field, Write);
                }
            }
        }

        // Extract RHS fields (value being assigned)
        if let Some(rhs_start) = binary_section.find("rhs:") {
            let rhs_section = &binary_section[rhs_start..];
            let rhs_fields = extract_fields_from_section(rhs_section);

            for field in rhs_fields {
                // RHS fields are always reads
                merge_access_type(accesses, field, Read);
            }
        }
    }
}

/// Extract field accesses from non-assignment contexts (all reads)
fn extract_read_fields(
    node_str: &str,
    accesses: &mut std::collections::HashMap<String, AccessType>,
) {
    // Non-assignment context - all fields are reads
    let fields = extract_fields_from_section(node_str);
    for field in fields {
        merge_access_type(accesses, field, AccessType::Read);
    }
}

/// Helper function to extract field names from AST node section
fn extract_fields_from_section(section: &str) -> Vec<String> {
    let mut fields = Vec::new();

    // Pattern 1: Direct property access - Variable(e) ... Property(field_name)
    if let Ok(re) = regex::Regex::new(r"Variable\(e\)[^}]*Property\((\w+)\)") {
        for cap in re.captures_iter(section) {
            if let Some(field_name) = cap.get(1) {
                fields.push(field_name.as_str().to_string());
            }
        }
    }

    // Pattern 2: Nested case for method calls - rhs: Dot { lhs: Property(field)
    if section.contains("lhs: Variable(e)") {
        if let Ok(nested_re) = regex::Regex::new(r"rhs: Dot \{ lhs: Property\((\w+)\)") {
            for cap in nested_re.captures_iter(section) {
                if let Some(field_name) = cap.get(1) {
                    fields.push(field_name.as_str().to_string());
                }
            }
        }
    }

    fields
}

/// Merge access types for a field, upgrading to ReadWrite if accessed both ways
fn merge_access_type(
    accesses: &mut std::collections::HashMap<String, AccessType>,
    field: String,
    new_type: AccessType,
) {
    use AccessType::*;

    let current = accesses.entry(field.clone()).or_insert(new_type.clone());

    // Merge logic: if a field is both read and written, mark as ReadWrite
    *current = match (&*current, &new_type) {
        (Read, Write) | (Write, Read) => ReadWrite,
        (Read, ReadWrite) | (ReadWrite, Read) => ReadWrite,
        (Write, ReadWrite) | (ReadWrite, Write) => ReadWrite,
        (ReadWrite, ReadWrite) => ReadWrite,
        _ => new_type,
    };
}

#[derive(Clone)]
pub struct CompiledExpression {
    ast: AST,
    expr: String,
    field_accesses: Vec<FieldAccess>,
    native_predicate: Option<NativePredicate>,
    /// Whether this expression may mutate the `e` event map
    mutates_event: bool,
    meta_usage: MetaUsage,
    /// Whether this expression uses the `meta` variable
    uses_meta: bool,
    /// Whether this expression uses the `conf` variable
    uses_conf: bool,
    /// Whether this expression uses the `line` variable
    uses_line: bool,
    /// Whether this expression uses the `window` variable
    uses_window: bool,
}

impl CompiledExpression {
    /// Get the source expression
    pub fn source(&self) -> &str {
        &self.expr
    }

    /// Whether this expression references the `window` variable. When it does
    /// not, evaluation is independent of window contents and can use the
    /// lighter non-window path (which also enables the native fast-path).
    pub fn uses_window(&self) -> bool {
        self.uses_window
    }

    /// Get fields that are READ (including ReadWrite, excluding pure Write)
    pub fn read_fields(&self) -> std::collections::HashSet<String> {
        self.field_accesses
            .iter()
            .filter(|fa| matches!(fa.access_type, AccessType::Read | AccessType::ReadWrite))
            .map(|fa| fa.field_name.clone())
            .collect()
    }

    /// Get fields that are WRITTEN (including ReadWrite, excluding pure Read)
    pub fn written_fields(&self) -> std::collections::HashSet<String> {
        self.field_accesses
            .iter()
            .filter(|fa| matches!(fa.access_type, AccessType::Write | AccessType::ReadWrite))
            .map(|fa| fa.field_name.clone())
            .collect()
    }

    /// Get all accessed fields (backward compatibility)
    pub fn accessed_fields(&self) -> std::collections::HashSet<String> {
        self.field_accesses
            .iter()
            .map(|fa| fa.field_name.clone())
            .collect()
    }
}

pub struct RhaiEngine {
    engine: Engine,
    compiled_filters: Vec<CompiledExpression>,
    compiled_execs: Vec<CompiledExpression>,
    compiled_begin: Option<CompiledExpression>,
    compiled_end: Option<CompiledExpression>,
    scope_template: Scope<'static>,
    suppress_side_effects: bool,
    conf_map: Option<rhai::Map>,
    state_map: Option<crate::rhai_functions::state::StateMap>,
    state_available: bool,
    debug_tracker: Option<DebugTracker>,
    execution_tracer: Option<ExecutionTracer>,
    use_emoji: bool,
}

impl Clone for RhaiEngine {
    fn clone(&self) -> Self {
        let mut engine = Engine::new();
        // Use Simple optimization, not Full. Full optimization breaks side-effect functions
        // like track_freq("name", value), print("msg"), emit_each(), etc. by trying to evaluate them at
        // compile time when their arguments are constants. These functions MUST run at runtime.
        engine.set_optimization_level(rhai::OptimizationLevel::Simple);

        // Check for shutdown signal during script execution (cooperative cancellation)
        engine.on_progress(|_| {
            if crate::platform::SHOULD_TERMINATE.load(Ordering::Relaxed) {
                // Abort script execution by returning a termination sentinel
                Some(rhai::Dynamic::UNIT)
            } else {
                None
            }
        });

        // Apply the same on_print override as in new(), respecting suppress_side_effects
        let suppress_side_effects = self.suppress_side_effects;
        engine.on_print(move |text| {
            if suppress_side_effects {
                // Suppress all print output
                return;
            }

            if crate::rhai_functions::strings::is_parallel_mode() {
                crate::rhai_functions::strings::capture_print(text.to_string());
            } else {
                println!("{}", text);
            }
        });

        rhai_functions::register_all_functions(&mut engine);

        Self {
            engine,
            compiled_filters: self.compiled_filters.clone(),
            compiled_execs: self.compiled_execs.clone(),
            compiled_begin: self.compiled_begin.clone(),
            compiled_end: self.compiled_end.clone(),
            scope_template: self.scope_template.clone(),
            suppress_side_effects,
            conf_map: self.conf_map.clone(),
            state_map: self.state_map.clone(),
            state_available: self.state_available,
            debug_tracker: self.debug_tracker.clone(),
            execution_tracer: self.execution_tracer.clone(),
            use_emoji: self.use_emoji,
        }
    }
}

impl RhaiEngine {
    /// Render a short diagnostic with stage/name, position, snippet, and the raw Rhai message.
    fn format_rhai_diagnostic(
        err: Box<EvalAltResult>,
        stage: &str,
        script_name: &str,
        script_text: &str,
        scope: Option<&Scope>,
        debug_tracker: Option<&DebugTracker>,
        use_emoji: bool,
    ) -> String {
        let call_stack = Self::collect_call_stack(err.as_ref());
        let err_display = format!("{}", err);

        if let Some(tracker) = debug_tracker {
            let enhancer = ErrorEnhancer::new(tracker.config.clone());
            let context = tracker.get_context();
            if let Some(scope) = scope {
                return enhancer.enhance_error(&err, scope, script_text, stage, &context);
            }
        }

        // Basic header
        let mut output = String::new();
        output.push_str(&format!("{} error\n", stage));

        // Position + snippet
        let pos = err.position();
        if let Some(line_num) = pos.line() {
            let col_num = pos.position().unwrap_or(1);
            output.push_str(&format!(
                "  At {}:{} in {}\n",
                line_num, col_num, script_name
            ));
            if let Some(snippet) = Self::render_snippet(
                script_text,
                line_num.saturating_sub(1),
                col_num.saturating_sub(1),
            ) {
                output.push_str(&snippet);
            }
        } else if pos.is_none() {
            output.push_str(&format!("  In {}\n", script_name));
        } else {
            output.push_str(&format!("  At {} in {}\n", pos, script_name));
        }

        // Raw message from Rhai
        output.push_str(&format!("  Rhai: {}\n", err_display));

        if !call_stack.is_empty() {
            output.push_str("  Call stack (most recent first):\n");
            for (func, pos) in call_stack.iter().rev().take(3) {
                output.push_str(&format!("    • {} @ {}\n", func, pos));
            }
        }

        // Generate suggestions even without debug mode (helps users fix errors)
        let config = DebugConfig::new(0); // Minimal config for suggestion generation
        let enhancer = ErrorEnhancer::new(config);
        let suggestion = if let Some(scope) = scope {
            enhancer.generate_suggestions(&err, scope, Some(script_text))
        } else {
            ErrorEnhancer::raw_string_hint(&err, script_text)
        };
        if let Some(suggestion) = suggestion {
            if use_emoji {
                output.push_str(&format!("  💡 {}\n", suggestion));
            } else {
                output.push_str(&format!("  Hint: {}\n", suggestion));
            }
        }

        output
    }

    /// Collect nested function call frames from Rhai errors.
    fn collect_call_stack(err: &EvalAltResult) -> Vec<(String, rhai::Position)> {
        match err {
            EvalAltResult::ErrorInFunctionCall(func, _src, inner, pos) => {
                let mut frames = vec![(func.clone(), *pos)];
                frames.extend(Self::collect_call_stack(inner.as_ref()));
                frames
            }
            EvalAltResult::ErrorInModule(module, inner, pos) => {
                let mut frames = vec![(format!("module {}", module), *pos)];
                frames.extend(Self::collect_call_stack(inner.as_ref()));
                frames
            }
            _ => Vec::new(),
        }
    }

    /// Build a small two-line snippet with a caret under the offending column.
    fn render_snippet(
        script: &str,
        zero_based_line: usize,
        zero_based_col: usize,
    ) -> Option<String> {
        let lines: Vec<&str> = script.lines().collect();
        let line_content = lines.get(zero_based_line)?.trim_end_matches('\r');
        let line_num = zero_based_line + 1;
        let col_num = zero_based_col + 1;
        let gutter_width = line_num.to_string().len();
        let mut snippet = String::new();
        snippet.push_str(&format!(
            "  {line_num:>width$} | {line_content}\n",
            width = gutter_width
        ));
        let caret_padding = " ".repeat(col_num.saturating_sub(1));
        snippet.push_str(&format!(
            "  {empty:>width$} | {caret_padding}^\n",
            empty = "",
            width = gutter_width
        ));
        Some(snippet)
    }

    // Thread-local state management functions
    pub fn set_thread_tracking_state(
        metrics: &HashMap<String, Dynamic>,
        internal: &HashMap<String, Dynamic>,
    ) {
        rhai_functions::tracking::set_thread_tracking_state(metrics);
        rhai_functions::tracking::set_thread_internal_state(internal);
    }

    pub fn get_thread_tracking_state() -> HashMap<String, Dynamic> {
        rhai_functions::tracking::get_thread_tracking_state()
    }

    pub fn get_thread_internal_state() -> HashMap<String, Dynamic> {
        rhai_functions::tracking::get_thread_internal_state()
    }

    fn format_function_not_found_error(
        func_signature: String,
        script_name: &str,
        pos: rhai::Position,
    ) -> String {
        // Extract function name from signature (before the first '(' or space)
        let func_name = if let Some(paren_pos) = func_signature.find('(') {
            &func_signature[..paren_pos]
        } else if let Some(space_pos) = func_signature.find(' ') {
            &func_signature[..space_pos]
        } else {
            &func_signature
        }
        .trim();

        // Check if this looks like a type mismatch rather than a missing function
        let called_types = Self::extract_called_types(&func_signature);
        if Self::is_likely_type_mismatch(&func_signature, func_name) {
            let expected_types = Self::get_expected_function_signature(func_name);
            if !expected_types.is_empty() {
                return format!(
                    "Wrong argument types for '{}' in {} at {}: got {}, expected {}. Note: x.{}() = {}(x)",
                    func_name,
                    script_name,
                    pos,
                    called_types,
                    expected_types,
                    func_name,
                    func_name
                );
            }
        }

        // Fall back to "function not found" with suggestions
        let base_msg = format!(
            "Function '{}' not found in {} at {}",
            func_signature, script_name, pos
        );
        let suggestions = Self::get_function_suggestions(func_name);
        let mut notes = Vec::new();

        if called_types != "unknown types" {
            notes.push(format!("Called with: {}", called_types));
        }

        if Self::signature_has_unit(&called_types) {
            notes.push(
                "One of the arguments is '()' (missing field?). Use e.has(\"field\") or e.get(\"field\", default) before chaining."
                    .to_string(),
            );
        }

        if suggestions.is_empty() {
            notes.push(format!(
                "Note: method calls are sugar—x.{}(y) == {}(x, y)",
                func_name, func_name
            ));
            format!("{}. {}", base_msg, notes.join(" "))
        } else {
            let mut msg = format!("{}. Did you mean: {}", base_msg, suggestions.join(", "));
            if !notes.is_empty() {
                msg.push_str(&format!(" {}", notes.join(" ")));
            }
            msg
        }
    }

    fn is_likely_type_mismatch(func_signature: &str, func_name: &str) -> bool {
        !Self::get_expected_function_signature(func_name).is_empty() && func_signature.contains('(')
    }

    fn extract_called_types(func_signature: &str) -> String {
        if let Some(start) = func_signature.find('(') {
            if let Some(end) = func_signature.rfind(')') {
                return func_signature[start + 1..end].to_string();
            }
        }
        "unknown types".to_string()
    }

    fn signature_has_unit(called_types: &str) -> bool {
        called_types.contains("()")
    }

    fn get_expected_function_signature(func_name: &str) -> String {
        match func_name {
            "extract_regex" => "string, regex_pattern, optional_group_index".to_string(),
            "extract_regexes" => "string, regex_pattern, optional_group_index".to_string(),
            "extract_regex_maps" | "extract_re_maps" => "string, regex_pattern, field".to_string(),
            "split_regex" | "split_re" => "string, regex_pattern".to_string(),
            "replace_regex" | "replace_re" => "string, regex_pattern, replacement".to_string(),
            "before" | "after" => "string, delimiter".to_string(),
            "between" => "string, start_delimiter, end_delimiter".to_string(),
            "starting_with" | "ending_with" => "string, prefix_or_suffix".to_string(),
            "strip" => "string, optional_characters_to_strip".to_string(),
            "join" => "string_separator, array OR array, string_separator".to_string(),
            "extract_ip" | "extract_ips" | "extract_url" | "extract_domain" => "string".to_string(),
            "mask_ip" => "string, optional_octets_to_mask".to_string(),
            "is_private_ip" | "is_digit" => "string".to_string(),
            "parse_json" => "json_string".to_string(),
            "parse_kv" => "string, optional_separator, optional_kv_separator".to_string(),
            "col" => "string, column_selector".to_string(),
            "cols" => "string, column_selectors...".to_string(),
            "status_class" => "status_code_number".to_string(),
            "track_freq" => "name, value".to_string(),
            "track_inc" => "name".to_string(),
            "track_sum" | "track_min" | "track_max" | "track_avg" => "key, value".to_string(),
            "track_unique" => "key, value".to_string(),
            "count" => "string, substring".to_string(),
            // Common string functions that expect strings
            "len" | "trim" => "string".to_string(),
            "contains" | "starts_with" | "ends_with" => "string, substring".to_string(),
            "split" | "replace" => "string, delimiter_or_pattern, optional_replacement".to_string(),
            _ => "".to_string(),
        }
    }

    fn function_catalog() -> Vec<String> {
        // List of common Rhai built-in functions and our custom functions
        vec![
            // String functions
            "lower".to_string(),
            "upper".to_string(),
            "trim".to_string(),
            "len".to_string(),
            "contains".to_string(),
            "starts_with".to_string(),
            "ends_with".to_string(),
            "split".to_string(),
            "replace".to_string(),
            "substring".to_string(),
            "to_string".to_string(),
            "parse".to_string(),
            // Our custom string functions
            "extract_regex".to_string(),
            "extract_regexes".to_string(),
            "extract_regex_maps".to_string(),
            "split_regex".to_string(),
            "replace_regex".to_string(),
            "count".to_string(),
            "strip".to_string(),
            "before".to_string(),
            "after".to_string(),
            "between".to_string(),
            "starting_with".to_string(),
            "ending_with".to_string(),
            "is_digit".to_string(),
            "join".to_string(),
            "extract_ip".to_string(),
            "extract_ips".to_string(),
            "mask_ip".to_string(),
            "is_private_ip".to_string(),
            "extract_url".to_string(),
            "extract_domain".to_string(),
            // Math functions
            "abs".to_string(),
            "floor".to_string(),
            "ceil".to_string(),
            "round".to_string(),
            "min".to_string(),
            "max".to_string(),
            "pow".to_string(),
            "sqrt".to_string(),
            // Array functions
            "push".to_string(),
            "pop".to_string(),
            "shift".to_string(),
            "unshift".to_string(),
            "reverse".to_string(),
            "sort".to_string(),
            "clear".to_string(),
            // Map functions
            "keys".to_string(),
            "values".to_string(),
            "remove".to_string(),
            "contains".to_string(),
            // Our custom functions
            "parse_json".to_string(),
            "parse_kv".to_string(),
            "col".to_string(),
            "cols".to_string(),
            "status_class".to_string(),
            "track_freq".to_string(),
            "track_inc".to_string(),
            "track_sum".to_string(),
            "track_min".to_string(),
            "track_max".to_string(),
            "track_avg".to_string(),
            "track_unique".to_string(),
            "track_percentiles".to_string(),
            "track_stats".to_string(),
            "track_cardinality".to_string(),
            "track_top".to_string(),
            "track_top_by".to_string(),
            "track_bottom".to_string(),
            "track_bottom_by".to_string(),
            // Utility functions
            "print".to_string(),
            "debug".to_string(),
            "type_of".to_string(),
            "is_def_fn".to_string(),
        ]
    }

    fn get_function_suggestions(func_name: &str) -> Vec<String> {
        let available_functions = Self::function_catalog();

        // Find functions that are similar to the requested one
        let suggestions: Vec<String> = available_functions
            .iter()
            .filter(|&f| {
                // Check for starts with or contains
                f.starts_with(func_name) || (func_name.len() > 1 && f.contains(func_name))
            })
            .take(3) // Limit to 3 suggestions
            .map(|s| s.to_string())
            .collect();

        // For debugging: always include at least one suggestion if the function name contains common patterns
        if suggestions.is_empty() && func_name.len() > 2 {
            if func_name.contains("extract") {
                return vec![
                    "extract_regex".to_string(),
                    "extract_ip".to_string(),
                    "extract_url".to_string(),
                ];
            } else if func_name.contains("track") {
                return vec![
                    "track_freq".to_string(),
                    "track_inc".to_string(),
                    "track_sum".to_string(),
                    "track_unique".to_string(),
                ];
            } else if func_name.contains("pars") {
                return vec!["parse_json".to_string(), "parse_kv".to_string()];
            }
        }

        suggestions
    }

    pub fn new() -> Self {
        let mut engine = Engine::new();

        // Use Simple optimization, not Full. Full optimization breaks side-effect functions
        // like track_freq("name", value), print("msg"), emit_each(), etc. by trying to evaluate them at
        // compile time when their arguments are constants. These functions MUST run at runtime.
        engine.set_optimization_level(rhai::OptimizationLevel::Simple);

        // Check for shutdown signal during script execution (cooperative cancellation)
        engine.on_progress(|_| {
            if crate::platform::SHOULD_TERMINATE.load(Ordering::Relaxed) {
                // Abort script execution by returning a termination sentinel
                Some(rhai::Dynamic::UNIT)
            } else {
                None
            }
        });

        // Override the built-in print function to support capture in parallel mode
        // Note: suppress_side_effects is false by default in new()
        engine.on_print(|text| {
            if crate::rhai_functions::strings::is_parallel_mode() {
                // Use both old capture system (for compatibility) and new ordered system
                crate::rhai_functions::strings::capture_print(text.to_string());
                crate::rhai_functions::strings::capture_stdout(text.to_string());
            } else {
                println!("{}", text);
            }
        });

        // Register custom functions for log analysis (includes eprint() for stderr output)
        rhai_functions::register_all_functions(&mut engine);

        let mut scope_template = Scope::new();
        scope_template.push("line", "");
        scope_template.push("e", rhai::Map::new());
        scope_template.push("meta", rhai::Map::new());
        scope_template.push("conf", rhai::Map::new());

        Self {
            engine,
            compiled_filters: Vec::new(),
            compiled_execs: Vec::new(),
            compiled_begin: None,
            compiled_end: None,
            scope_template,
            suppress_side_effects: false,
            conf_map: None,
            state_map: Some(crate::rhai_functions::state::StateMap::new()),
            state_available: true,
            debug_tracker: None,
            execution_tracer: None,
            use_emoji: true,
        }
    }

    pub fn set_use_emoji(&mut self, use_emoji: bool) {
        self.use_emoji = use_emoji;
    }

    pub fn get_execution_tracer(&self) -> &Option<ExecutionTracer> {
        &self.execution_tracer
    }

    /// Get the compiled filters (for benchmarking)
    pub fn compiled_filters(&self) -> &[CompiledExpression] {
        &self.compiled_filters
    }

    pub fn set_state_available(&mut self, available: bool) {
        self.state_available = available;
    }

    fn push_state_to_scope(&self, scope: &mut Scope) {
        if !self.state_available || crate::rhai_functions::strings::is_parallel_mode() {
            scope.push("state", crate::rhai_functions::state::StateNotAvailable);
        } else if let Some(ref state_map) = self.state_map {
            scope.push("state", state_map.clone());
        }
    }

    /// Set up debugging with the provided configuration
    pub fn setup_debugging(&mut self, debug_config: DebugConfig) {
        if !debug_config.is_enabled() {
            return;
        }

        self.debug_tracker = Some(DebugTracker::new(debug_config.clone()));
        self.execution_tracer = Some(ExecutionTracer::new(debug_config.clone()));

        // These unwraps are safe because we just created the debug components above
        let debug_tracker = self
            .debug_tracker
            .as_ref()
            .expect("debug_tracker should be initialized")
            .clone();
        let execution_tracer = self
            .execution_tracer
            .as_ref()
            .expect("execution_tracer should be initialized")
            .clone();
        // Allow deprecated API: register_debugger is marked as volatile/experimental but is the
        // only way to access Rhai's debugging functionality. The API is stable in practice and
        // essential for our debugging features. We'll update when a stable replacement is available.
        #[allow(deprecated)]
        self.engine.register_debugger(
            move |_engine, debugger| debugger,
            move |_context, event, node, source, pos| {
                // Update execution context
                debug_tracker.update_context(Some(pos), source);

                // Enhanced event logging with verbosity levels
                match event {
                    DebuggerEvent::Start => {
                        debug_tracker.log_basic("Script execution started");
                        if debug_tracker.config.verbosity >= 3 {
                            if let Some(src) = source {
                                debug_tracker
                                    .log_step("Starting script", &format!("\"{}\"", src.trim()));
                            }
                        }
                    }
                    DebuggerEvent::End => {
                        debug_tracker.log_basic("Script execution completed");
                    }
                    DebuggerEvent::Step => {
                        // Enhanced step-by-step tracing
                        if debug_tracker.config.verbosity >= 2 {
                            // Use execution tracer for step-level tracing
                            let step_info = format!("Step at {}", pos);
                            if let Some(src) = source {
                                execution_tracer.trace_step(0, &step_info, src);
                            } else {
                                execution_tracer.trace_step(0, &step_info, "unknown");
                            }
                        }

                        if debug_tracker.config.verbosity >= 3 {
                            // Even more detailed tracing for -vvv
                            let step_info = format!("Step at {}", pos);
                            let node_info = format!("{:?}", node);
                            debug_tracker.log_step(&step_info, &node_info);

                            // Use execution tracer for expression-level details
                            if let Some(src) = source {
                                execution_tracer.trace_expression_evaluation(src, "evaluating");
                            }
                        }
                    }
                    DebuggerEvent::BreakPoint(_) => {
                        if debug_tracker.config.verbosity >= 2 {
                            debug_tracker.log_detailed("breakpoint", 0, &format!("hit at {}", pos));
                            // Use execution tracer for breakpoint details
                            if let Some(src) = source {
                                execution_tracer.trace_step(0, "Breakpoint hit", src);
                            }
                        }
                    }
                    // Note: Specific function call events may not be available in current Rhai version
                    // We handle function call tracing through Step events and other mechanisms
                    _ => {
                        // Enhanced event tracing
                        if debug_tracker.config.verbosity >= 3 {
                            let event_name = format!("{:?}", event);
                            debug_tracker.log_step("Debug event", &event_name);

                            // Use execution tracer for detailed event tracking
                            if let Some(src) = source {
                                execution_tracer.trace_step(0, &event_name, src);
                            }
                        }
                    }
                }

                // Update execution context with more details
                if debug_tracker.config.verbosity >= 2 {
                    if let Ok(mut ctx) = debug_tracker.context.lock() {
                        ctx.last_operation = Some(format!("{:?}", event));
                        if let Some(src) = source {
                            ctx.source_snippet = Some(src.to_string());
                        }
                    }
                }

                Ok(DebuggerCommand::Continue)
            },
        );
    }

    /// Set whether to suppress side effects (print, eprint, etc.)
    pub fn set_suppress_side_effects(&mut self, suppress: bool) {
        self.suppress_side_effects = suppress;

        // Set the thread-local flag for eprint and other functions
        crate::rhai_functions::strings::set_suppress_side_effects(suppress);

        // Re-register the print handler with the new suppression setting
        let suppress_copy = suppress;
        self.engine.on_print(move |text| {
            if suppress_copy {
                // Suppress all print output
                return;
            }

            if crate::rhai_functions::strings::is_parallel_mode() {
                crate::rhai_functions::strings::capture_print(text.to_string());
            } else {
                println!("{}", text);
            }
        });
    }

    // Individual compilation methods for pipeline stages
    pub fn compile_filter(&mut self, filter: &str) -> Result<CompiledExpression> {
        self.compile_filter_with_includes(filter, &[])
    }

    pub fn compile_filter_with_includes(
        &mut self,
        filter: &str,
        includes: &[crate::config::IncludeFile],
    ) -> Result<CompiledExpression> {
        let mut ast = self.engine.compile_expression(filter).map_err(|e| {
            let msg = Self::format_rhai_diagnostic(
                e.into(),
                "filter compilation",
                "filter expression",
                filter,
                None,
                None,
                self.use_emoji,
            );
            anyhow::anyhow!(msg)
        })?;

        for include in includes {
            let mut include_ast = self.engine.compile(&include.content).map_err(|e| {
                let msg = Self::format_rhai_diagnostic(
                    e.into(),
                    "filter include compilation",
                    "include script",
                    &include.content,
                    None,
                    None,
                    self.use_emoji,
                );
                anyhow::anyhow!("{} (in {})", msg, include.path)
            })?;

            include_ast.set_source(include.path.clone());

            if !include_ast.statements().is_empty() {
                return Err(anyhow::anyhow!(
                    "--include file '{}' cannot contain statements when used with --filter; only function definitions are allowed",
                    include.path
                ));
            }

            let include_functions = include_ast.clone_functions_only();
            ast = ast.merge(&include_functions);
        }

        let native_predicate = build_native_predicate(&ast);
        let field_accesses = extract_field_accesses(&ast);
        let var_usage = detect_variable_usage(&ast);
        Ok(CompiledExpression {
            ast,
            expr: filter.to_string(),
            field_accesses,
            native_predicate,
            mutates_event: false,
            meta_usage: var_usage.meta_usage,
            uses_meta: var_usage.uses_meta,
            uses_conf: var_usage.uses_conf,
            uses_line: var_usage.uses_line,
            uses_window: var_usage.uses_window,
        })
    }

    pub fn compile_exec(&mut self, exec: &str) -> Result<CompiledExpression> {
        let ast = self.engine.compile(exec).map_err(|e| {
            let msg = Self::format_rhai_diagnostic(
                e.into(),
                "exec compilation",
                "exec script",
                exec,
                None,
                None,
                self.use_emoji,
            );
            anyhow::anyhow!(msg)
        })?;
        let field_accesses = extract_field_accesses(&ast);
        let var_usage = detect_variable_usage(&ast);
        let mutates_event = expression_mutates_event(&ast, &field_accesses);
        Ok(CompiledExpression {
            ast,
            expr: exec.to_string(),
            field_accesses,
            native_predicate: None,
            mutates_event,
            meta_usage: var_usage.meta_usage,
            uses_meta: var_usage.uses_meta,
            uses_conf: var_usage.uses_conf,
            uses_line: var_usage.uses_line,
            uses_window: var_usage.uses_window,
        })
    }

    pub fn compile_begin(&mut self, begin: &str) -> Result<CompiledExpression> {
        let ast = self.engine.compile(begin).map_err(|e| {
            let msg = Self::format_rhai_diagnostic(
                e.into(),
                "begin compilation",
                "begin script",
                begin,
                None,
                None,
                self.use_emoji,
            );
            anyhow::anyhow!(msg)
        })?;
        let field_accesses = extract_field_accesses(&ast);
        let var_usage = detect_variable_usage(&ast);
        Ok(CompiledExpression {
            ast,
            expr: begin.to_string(),
            field_accesses,
            native_predicate: None,
            mutates_event: false,
            meta_usage: var_usage.meta_usage,
            uses_meta: var_usage.uses_meta,
            uses_conf: var_usage.uses_conf,
            uses_line: var_usage.uses_line,
            uses_window: var_usage.uses_window,
        })
    }

    pub fn compile_end(&mut self, end: &str) -> Result<CompiledExpression> {
        let ast = self.engine.compile(end).map_err(|e| {
            let msg = Self::format_rhai_diagnostic(
                e.into(),
                "end compilation",
                "end script",
                end,
                None,
                None,
                self.use_emoji,
            );
            anyhow::anyhow!(msg)
        })?;
        let field_accesses = extract_field_accesses(&ast);
        let var_usage = detect_variable_usage(&ast);
        Ok(CompiledExpression {
            ast,
            expr: end.to_string(),
            field_accesses,
            native_predicate: None,
            mutates_event: false,
            meta_usage: var_usage.meta_usage,
            uses_meta: var_usage.uses_meta,
            uses_conf: var_usage.uses_conf,
            uses_line: var_usage.uses_line,
            uses_window: var_usage.uses_window,
        })
    }

    pub fn compile_span_close(&mut self, script: &str) -> Result<CompiledExpression> {
        let ast = self.engine.compile(script).map_err(|e| {
            let msg = Self::format_rhai_diagnostic(
                e.into(),
                "span-close compilation",
                "span-close script",
                script,
                None,
                None,
                self.use_emoji,
            );
            anyhow::anyhow!(msg)
        })?;
        let field_accesses = extract_field_accesses(&ast);
        let var_usage = detect_variable_usage(&ast);
        Ok(CompiledExpression {
            ast,
            expr: script.to_string(),
            field_accesses,
            native_predicate: None,
            mutates_event: false,
            meta_usage: var_usage.meta_usage,
            uses_meta: var_usage.uses_meta,
            uses_conf: var_usage.uses_conf,
            uses_line: var_usage.uses_line,
            uses_window: var_usage.uses_window,
        })
    }

    // Individual execution methods for pipeline stages
    pub fn execute_compiled_filter(
        &mut self,
        compiled: &CompiledExpression,
        event: &Event,
        metrics: &mut HashMap<String, Dynamic>,
        internal: &mut HashMap<String, Dynamic>,
    ) -> Result<bool> {
        if let Some(native) = &compiled.native_predicate {
            Self::set_thread_tracking_state(metrics, internal);
            if let Some(result) = native.evaluate(event) {
                *metrics = Self::get_thread_tracking_state();
                *internal = Self::get_thread_internal_state();
                return Ok(result);
            }
        }

        Self::set_thread_tracking_state(metrics, internal);
        let mut scope = self.create_scope_for_event_optimized(
            event,
            compiled.uses_line,
            compiled.meta_usage,
            compiled.uses_conf,
        );

        // Add execution tracing for filter execution
        if let Some(ref tracer) = self.execution_tracer {
            let event_num = tracer.next_event();
            let event_data = format!("{:?}", event.fields);

            // Level 2+: Detailed execution tracing
            if tracer.config.verbosity >= 2 {
                tracer.trace_event_start(event_num, &event_data);
                eprintln!("  Script: {}", compiled.expr.trim());
            }

            // Enhanced detailed tracing for -vvv
            if tracer.config.verbosity >= 3 {
                tracer.trace_scope_inspection(&scope);
                tracer.trace_detailed_step(
                    "filter",
                    "evaluation",
                    &compiled.expr,
                    "starting",
                    "script",
                );
            }
        }

        let result = self
            .engine
            .eval_ast_with_scope::<bool>(&mut scope, &compiled.ast)
            .map_err(|e| {
                let detailed_msg = Self::format_rhai_diagnostic(
                    e,
                    "filter",
                    "filter expression",
                    &compiled.expr,
                    Some(&scope),
                    self.debug_tracker.as_ref(),
                    self.use_emoji,
                );
                anyhow::anyhow!("{}", detailed_msg)
            })?;

        self.assert_conf_not_mutated(&scope, compiled.uses_conf)
            .map_err(anyhow::Error::from)?;

        // Add execution result tracing
        if let Some(ref tracer) = self.execution_tracer {
            let action = if result { "passed" } else { "filtered out" };
            tracer.trace_event_result(result, action);

            // Enhanced detailed result tracing
            if tracer.config.verbosity >= 3 {
                let result_str = if result { "true" } else { "false" };
                tracer.trace_detailed_step(
                    "filter",
                    "result",
                    &compiled.expr,
                    result_str,
                    "boolean",
                );
            }
        }

        *metrics = Self::get_thread_tracking_state();
        *internal = Self::get_thread_internal_state();
        Ok(result)
    }

    pub fn execute_compiled_exec(
        &mut self,
        compiled: &CompiledExpression,
        event: &mut Event,
        metrics: &mut HashMap<String, Dynamic>,
        internal: &mut HashMap<String, Dynamic>,
    ) -> Result<()> {
        Self::set_thread_tracking_state(metrics, internal);
        let mut scope = self.create_scope_for_event_optimized(
            event,
            compiled.uses_line,
            compiled.meta_usage,
            compiled.uses_conf,
        );

        // Add execution tracing for exec execution
        if let Some(ref tracer) = self.execution_tracer {
            let event_num = tracer.next_event();
            let event_data = format!("{:?}", event.fields);

            // Level 2+: Detailed execution tracing
            if tracer.config.verbosity >= 2 {
                tracer.trace_event_start(event_num, &event_data);
                eprintln!("  Script: {}", compiled.expr.trim());
            }

            // Enhanced detailed tracing for -vvv
            if tracer.config.verbosity >= 3 {
                tracer.trace_scope_inspection(&scope);
                tracer.trace_detailed_step(
                    "exec",
                    "transformation",
                    &compiled.expr,
                    "starting",
                    "script",
                );
            }
        }

        let _ = self
            .engine
            .eval_ast_with_scope::<Dynamic>(&mut scope, &compiled.ast)
            .map_err(|e| {
                let detailed_msg = Self::format_rhai_diagnostic(
                    e,
                    "exec",
                    "exec script",
                    &compiled.expr,
                    Some(&scope),
                    self.debug_tracker.as_ref(),
                    self.use_emoji,
                );
                anyhow::anyhow!("{}", detailed_msg)
            })?;

        self.assert_conf_not_mutated(&scope, compiled.uses_conf)
            .map_err(anyhow::Error::from)?;

        // Add execution result tracing
        if let Some(ref tracer) = self.execution_tracer {
            tracer.trace_event_result(true, "executed successfully");

            // Enhanced detailed result tracing
            if tracer.config.verbosity >= 3 {
                tracer.trace_detailed_step(
                    "exec",
                    "result",
                    &compiled.expr,
                    "success",
                    "execution",
                );
            }
        }

        if compiled.mutates_event {
            self.update_event_from_scope(event, &scope);
        }
        // No exec gate counter: exec is best-effort, so it never fails the run
        // on its own (only --strict does). See the gate notes in tracking::errors.
        *metrics = Self::get_thread_tracking_state();
        *internal = Self::get_thread_internal_state();
        Ok(())
    }

    pub fn execute_compiled_begin(
        &mut self,
        compiled: &CompiledExpression,
        metrics: &mut HashMap<String, Dynamic>,
        internal: &mut HashMap<String, Dynamic>,
    ) -> Result<rhai::Map> {
        Self::set_thread_tracking_state(metrics, internal);

        // Set begin phase flag to allow read_file/read_lines
        crate::rhai_functions::conf::set_begin_phase(true);

        let mut scope = self.scope_template.clone();

        self.push_state_to_scope(&mut scope);

        let _ = self
            .engine
            .eval_ast_with_scope::<Dynamic>(&mut scope, &compiled.ast)
            .map_err(|e| {
                let detailed_msg = Self::format_rhai_diagnostic(
                    e,
                    "begin",
                    "begin expression",
                    &compiled.expr,
                    Some(&scope),
                    self.debug_tracker.as_ref(),
                    self.use_emoji,
                );
                anyhow::anyhow!("{}", detailed_msg)
            })?;

        // Reset begin phase flag
        crate::rhai_functions::conf::set_begin_phase(false);

        *metrics = Self::get_thread_tracking_state();
        *internal = Self::get_thread_internal_state();

        // Extract the conf map from scope and store it
        let mut conf_map = scope.get_value::<rhai::Map>("conf").unwrap_or_default();

        // Deep freeze the conf map to make it read-only
        crate::rhai_functions::conf::deep_freeze_map(&mut conf_map);

        // Store the frozen conf map
        self.conf_map = Some(conf_map.clone());

        Ok(conf_map)
    }

    pub fn execute_compiled_end(
        &self,
        compiled: &CompiledExpression,
        metrics: &HashMap<String, Dynamic>,
        internal: &HashMap<String, Dynamic>,
    ) -> Result<()> {
        let mut scope = self.scope_template.clone();

        // Finalize sketch-backed metrics (percentiles from t-digest, averages
        // from {sum,count} maps, cardinality from HLL) so `--end` scripts see
        // plain numbers instead of internal state.
        let tracked_map =
            crate::rhai_functions::tracking::finalize_metrics_for_script(metrics, internal);
        scope.set_value("metrics", tracked_map);

        // Set the frozen conf map (read-only)
        if let Some(ref conf_map) = self.conf_map {
            scope.set_value("conf", conf_map.clone());
        }

        self.push_state_to_scope(&mut scope);

        let _ = self
            .engine
            .eval_ast_with_scope::<Dynamic>(&mut scope, &compiled.ast)
            .map_err(|e| {
                let detailed_msg = Self::format_rhai_diagnostic(
                    e,
                    "end",
                    "end expression",
                    &compiled.expr,
                    Some(&scope),
                    self.debug_tracker.as_ref(),
                    self.use_emoji,
                );
                anyhow::anyhow!("{}", detailed_msg)
            })?;

        self.assert_conf_not_mutated(&scope, compiled.uses_conf)
            .map_err(anyhow::Error::from)?;

        Ok(())
    }

    pub fn execute_compiled_span_close(
        &mut self,
        compiled: &CompiledExpression,
        metrics: &mut HashMap<String, Dynamic>,
        internal: &mut HashMap<String, Dynamic>,
        span: crate::rhai_functions::span::SpanBinding,
    ) -> Result<()> {
        Self::set_thread_tracking_state(metrics, internal);

        let mut scope = self.scope_template.clone();

        // Finalize sketch-backed metrics for script consumption; see the
        // equivalent call in `execute_compiled_end`.
        let metrics_map =
            crate::rhai_functions::tracking::finalize_metrics_for_script(metrics, internal);
        scope.set_value("metrics", metrics_map);
        scope.push_constant("span", Dynamic::from(span));

        // Set the frozen conf map (read-only)
        if let Some(ref conf_map) = self.conf_map {
            scope.set_value("conf", conf_map.clone());
        }

        self.push_state_to_scope(&mut scope);

        crate::rhai_functions::file_ops::clear_pending_ops();

        let _ = self
            .engine
            .eval_ast_with_scope::<Dynamic>(&mut scope, &compiled.ast)
            .map_err(|e| {
                let detailed_msg = Self::format_rhai_diagnostic(
                    e,
                    "span-close",
                    "span-close script",
                    &compiled.expr,
                    Some(&scope),
                    self.debug_tracker.as_ref(),
                    self.use_emoji,
                );
                anyhow::anyhow!("{}", detailed_msg)
            })?;

        self.assert_conf_not_mutated(&scope, compiled.uses_conf)
            .map_err(anyhow::Error::from)?;

        let ops = crate::rhai_functions::file_ops::take_pending_ops();
        crate::rhai_functions::file_ops::execute_ops(&ops)?;

        *metrics = Self::get_thread_tracking_state();
        *internal = Self::get_thread_internal_state();

        Ok(())
    }

    // Window-aware execution methods
    pub fn execute_compiled_filter_with_window(
        &mut self,
        compiled: &CompiledExpression,
        event: &Event,
        window: &[Event],
        metrics: &mut HashMap<String, Dynamic>,
        internal: &mut HashMap<String, Dynamic>,
    ) -> Result<bool> {
        Self::set_thread_tracking_state(metrics, internal);
        let mut scope = self.create_scope_for_event_with_window(event, window, compiled.meta_usage);

        // Add execution tracing for windowed filter execution
        if let Some(ref tracer) = self.execution_tracer {
            let event_num = tracer.next_event();
            let event_data = format!("{:?}", event.fields);

            // Level 2+: Detailed execution tracing
            if tracer.config.verbosity >= 2 {
                tracer.trace_event_start(event_num, &event_data);
                eprintln!(
                    "  Script (windowed, size {}): {}",
                    window.len(),
                    compiled.expr.trim()
                );
            }

            // Enhanced detailed tracing for -vvv
            if tracer.config.verbosity >= 3 {
                tracer.trace_scope_inspection(&scope);
                tracer.trace_detailed_step(
                    "windowed-filter",
                    "evaluation",
                    &compiled.expr,
                    "starting",
                    "script",
                );
                tracer.trace_detailed_step(
                    "windowed-filter",
                    "window-size",
                    &window.len().to_string(),
                    &window.len().to_string(),
                    "size",
                );
            }
        }

        let result = self
            .engine
            .eval_ast_with_scope::<bool>(&mut scope, &compiled.ast)
            .map_err(|e| {
                let detailed_msg = Self::format_rhai_diagnostic(
                    e,
                    "filter",
                    "filter expression",
                    &compiled.expr,
                    Some(&scope),
                    self.debug_tracker.as_ref(),
                    self.use_emoji,
                );
                anyhow::anyhow!("{}", detailed_msg)
            })?;

        self.assert_conf_not_mutated(&scope, compiled.uses_conf)
            .map_err(anyhow::Error::from)?;

        // Add execution result tracing
        if let Some(ref tracer) = self.execution_tracer {
            let action = if result { "passed" } else { "filtered out" };
            tracer.trace_event_result(result, action);

            // Enhanced detailed result tracing
            if tracer.config.verbosity >= 3 {
                let result_str = if result { "true" } else { "false" };
                tracer.trace_detailed_step(
                    "windowed-filter",
                    "result",
                    &compiled.expr,
                    result_str,
                    "boolean",
                );
            }
        }

        *metrics = Self::get_thread_tracking_state();
        *internal = Self::get_thread_internal_state();
        Ok(result)
    }

    pub fn execute_compiled_exec_with_window(
        &mut self,
        compiled: &CompiledExpression,
        event: &mut Event,
        window: &[Event],
        metrics: &mut HashMap<String, Dynamic>,
        internal: &mut HashMap<String, Dynamic>,
    ) -> Result<()> {
        Self::set_thread_tracking_state(metrics, internal);
        let mut scope = self.create_scope_for_event_with_window(event, window, compiled.meta_usage);

        // Add execution tracing for windowed exec execution
        if let Some(ref tracer) = self.execution_tracer {
            let event_num = tracer.next_event();
            let event_data = format!("{:?}", event.fields);

            // Level 2+: Detailed execution tracing
            if tracer.config.verbosity >= 2 {
                tracer.trace_event_start(event_num, &event_data);
                eprintln!(
                    "  Script (windowed, size {}): {}",
                    window.len(),
                    compiled.expr.trim()
                );
            }

            // Enhanced detailed tracing for -vvv
            if tracer.config.verbosity >= 3 {
                tracer.trace_scope_inspection(&scope);
                tracer.trace_detailed_step(
                    "windowed-exec",
                    "transformation",
                    &compiled.expr,
                    "starting",
                    "script",
                );
                tracer.trace_detailed_step(
                    "windowed-exec",
                    "window-size",
                    &window.len().to_string(),
                    &window.len().to_string(),
                    "size",
                );
            }
        }

        let _ = self
            .engine
            .eval_ast_with_scope::<Dynamic>(&mut scope, &compiled.ast)
            .map_err(|e| {
                let detailed_msg = Self::format_rhai_diagnostic(
                    e,
                    "exec",
                    "exec script",
                    &compiled.expr,
                    Some(&scope),
                    self.debug_tracker.as_ref(),
                    self.use_emoji,
                );
                anyhow::anyhow!("{}", detailed_msg)
            })?;

        self.assert_conf_not_mutated(&scope, compiled.uses_conf)
            .map_err(anyhow::Error::from)?;

        // Add execution result tracing
        if let Some(ref tracer) = self.execution_tracer {
            tracer.trace_event_result(true, "executed successfully");

            // Enhanced detailed result tracing
            if tracer.config.verbosity >= 3 {
                tracer.trace_detailed_step(
                    "windowed-exec",
                    "result",
                    &compiled.expr,
                    "success",
                    "execution",
                );
            }
        }

        if compiled.mutates_event {
            self.update_event_from_scope(event, &scope);
        }
        // No exec gate counter: exec is best-effort (see tracking::errors gate notes).
        *metrics = Self::get_thread_tracking_state();
        *internal = Self::get_thread_internal_state();
        Ok(())
    }

    fn assert_conf_not_mutated(
        &self,
        scope: &Scope,
        uses_conf: bool,
    ) -> Result<(), ConfMutationError> {
        // Stages that don't reference `conf` never get the real conf map loaded
        // into their scope (see `create_scope_for_event_optimized`); the scope
        // keeps the template's empty placeholder instead. Skip the check for
        // those stages — a script that never names `conf` cannot mutate it, and
        // comparing the empty placeholder against the real map would be a false
        // positive.
        if !uses_conf {
            return Ok(());
        }
        if let Some(original) = &self.conf_map {
            match scope.get_value::<Map>("conf") {
                Some(conf) if maps_equal(&conf, original) => Ok(()),
                _ => Err(ConfMutationError),
            }
        } else {
            Ok(())
        }
    }

    fn create_scope_for_event(&self, event: &Event) -> Scope<'_> {
        // Full scope creation - includes all variables
        self.create_scope_for_event_optimized(
            event,
            true,
            MetaUsage {
                populate_all: true,
                ..MetaUsage::default()
            },
            true,
        )
    }

    /// Create a scope for event evaluation, optionally skipping unused variables
    fn create_scope_for_event_optimized(
        &self,
        event: &Event,
        needs_line: bool,
        meta_usage: MetaUsage,
        needs_conf: bool,
    ) -> Scope<'_> {
        let mut scope = self.scope_template.clone();

        // Update built-in variables - only if needed
        if needs_line {
            scope.set_value("line", event.original_line.clone());
        }

        // Update event map for fields with invalid identifiers
        // Note: We always build the event map since most scripts use `e.*`
        let mut event_map = rhai::Map::new();
        for (k, v) in &event.fields {
            event_map.insert(k.clone().into(), v.clone());
        }
        scope.set_value("e", event_map);

        // Update metadata - only if needed
        if meta_usage.populate_all || meta_usage.any() {
            let mut meta_map = rhai::Map::new();
            if let Some(line_num) = (meta_usage.populate_all || meta_usage.line_num)
                .then_some(event.line_num)
                .flatten()
            {
                meta_map.insert("line_num".into(), Dynamic::from(line_num as i64));
            }
            if let Some(filename) = (meta_usage.populate_all || meta_usage.filename)
                .then_some(event.filename.as_ref())
                .flatten()
            {
                meta_map.insert("filename".into(), Dynamic::from(filename.clone()));
            }
            if let Some(status) = (meta_usage.populate_all || meta_usage.span_status)
                .then_some(event.span.status)
                .flatten()
            {
                meta_map.insert("span_status".into(), Dynamic::from(status.as_str()));
            }
            if let Some(span_id) = (meta_usage.populate_all || meta_usage.span_id)
                .then_some(event.span.span_id.as_ref())
                .flatten()
            {
                meta_map.insert("span_id".into(), Dynamic::from(span_id.clone()));
            }
            if let Some(span_start) = (meta_usage.populate_all || meta_usage.span_start)
                .then_some(event.span.span_start)
                .flatten()
            {
                meta_map.insert(
                    "span_start".into(),
                    Dynamic::from(DateTimeWrapper::from_utc(span_start)),
                );
            }
            if let Some(span_end) = (meta_usage.populate_all || meta_usage.span_end)
                .then_some(event.span.span_end)
                .flatten()
            {
                meta_map.insert(
                    "span_end".into(),
                    Dynamic::from(DateTimeWrapper::from_utc(span_end)),
                );
            }
            if let Some(parsed_ts) = (meta_usage.populate_all || meta_usage.parsed_ts)
                .then_some(event.parsed_ts)
                .flatten()
            {
                meta_map.insert(
                    "parsed_ts".into(),
                    Dynamic::from(DateTimeWrapper::from_utc(parsed_ts)),
                );
            }

            if meta_usage.populate_all || meta_usage.line {
                meta_map.insert("line".into(), Dynamic::from(event.original_line.clone()));
            }

            scope.set_value("meta", meta_map);
        }

        // Set the frozen conf map - only if needed
        if needs_conf {
            if let Some(ref conf_map) = self.conf_map {
                scope.set_value("conf", conf_map.clone());
            }
        }

        self.push_state_to_scope(&mut scope);

        scope
    }

    fn create_scope_for_event_with_window(
        &self,
        event: &Event,
        window: &[Event],
        meta_usage: MetaUsage,
    ) -> Scope<'_> {
        let mut scope = self.create_scope_for_event_optimized(event, true, meta_usage, true);

        // Add window array to scope
        let window_array: rhai::Array = window
            .iter()
            .map(|event| {
                let mut event_map = rhai::Map::new();
                // Add all event fields to the map
                for (k, v) in &event.fields {
                    event_map.insert(k.clone().into(), v.clone());
                }
                // Add built-in fields
                event_map.insert("line".into(), Dynamic::from(event.original_line.clone()));
                if let Some(line_num) = event.line_num {
                    event_map.insert("line_num".into(), Dynamic::from(line_num as i64));
                }
                if let Some(filename) = &event.filename {
                    event_map.insert("filename".into(), Dynamic::from(filename.clone()));
                }
                if let Some(status) = event.span.status {
                    event_map.insert("span_status".into(), Dynamic::from(status.as_str()));
                }
                if let Some(span_id) = &event.span.span_id {
                    event_map.insert("span_id".into(), Dynamic::from(span_id.clone()));
                }
                if let Some(span_start) = event.span.span_start {
                    event_map.insert(
                        "span_start".into(),
                        Dynamic::from(DateTimeWrapper::from_utc(span_start)),
                    );
                }
                if let Some(span_end) = event.span.span_end {
                    event_map.insert(
                        "span_end".into(),
                        Dynamic::from(DateTimeWrapper::from_utc(span_end)),
                    );
                }
                Dynamic::from(event_map)
            })
            .collect();

        scope.set_value("window", window_array);
        scope
    }

    fn update_event_from_scope(&self, event: &mut Event, scope: &Scope) {
        // Check if entire event 'e' was set to unit () - clear all fields
        if scope.get_value::<()>("e").is_some() {
            event.fields.clear();
            return;
        }

        // Capture mutations made directly to the `e` event map
        if let Some(obj) = scope.get_value::<Map>("e") {
            let original_order: Vec<String> = event.fields.keys().cloned().collect();
            let mut remaining_entries: Vec<(String, Dynamic)> =
                obj.into_iter().map(|(k, v)| (k.into(), v)).collect();

            let mut reordered_fields = crate::event::FieldMap::with_capacity_and_hasher(
                remaining_entries.len(),
                ahash::RandomState::default(),
            );

            for key in &original_order {
                if let Some(pos) = remaining_entries.iter().position(|(k, _)| k == key) {
                    let (_, value) = remaining_entries.remove(pos);
                    if value.is::<()>() {
                        continue;
                    }
                    reordered_fields.insert(key.clone(), value);
                }
            }

            for (key, value) in remaining_entries {
                if value.is::<()>() {
                    continue;
                }
                reordered_fields.insert(key, value);
            }

            event.fields = reordered_fields;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::{TimeZone, Utc};

    fn build_event_with_line(line: &str) -> Event {
        let mut event = Event::with_capacity(line.to_string(), 1);
        event.set_field("line".to_string(), Dynamic::from(line.to_string()));
        event
    }

    /// Compile and run an exec script end-to-end (through the Rhai engine and the
    /// scope writeback) against an event, returning the mutated event.
    fn run_exec(script: &str, mut event: Event) -> Event {
        let mut engine = RhaiEngine::new();
        let compiled = engine.compile_exec(script).expect("compile exec");
        let mut metrics = std::collections::HashMap::new();
        let mut internal = std::collections::HashMap::new();
        engine
            .execute_compiled_exec(&compiled, &mut event, &mut metrics, &mut internal)
            .expect("run exec");
        event
    }

    fn field_str(event: &Event, key: &str) -> Option<String> {
        event
            .fields
            .get(key)
            .and_then(|v| v.clone().try_cast::<String>())
    }

    // Regression: in-place mutating calls on `e` (no assignment, no field-level write)
    // must still be written back to the event. Previously `mutates_event` only saw
    // assignments/field writes, so these merges were silently dropped from output.

    #[test]
    fn absorb_kv_persists_to_event() {
        let event = run_exec(r#"e.absorb_kv("line")"#, build_event_with_line("a=1 b=2"));
        assert_eq!(field_str(&event, "a").as_deref(), Some("1"));
        assert_eq!(field_str(&event, "b").as_deref(), Some("2"));
    }

    #[test]
    fn absorb_logfmt_persists_to_event() {
        let event = run_exec(
            r#"e.absorb_logfmt("line")"#,
            build_event_with_line(r#"pod="kube-system/foo" replicas=3"#),
        );
        assert_eq!(field_str(&event, "pod").as_deref(), Some("kube-system/foo"));
    }

    #[test]
    fn absorb_regex_persists_to_event() {
        let event = run_exec(
            r##"e.absorb_regex("line", #"User (?P<user>\w+)"#)"##,
            build_event_with_line("User alice logged in"),
        );
        assert_eq!(field_str(&event, "user").as_deref(), Some("alice"));
    }

    #[test]
    fn absorb_jwt_persists_to_event() {
        // Header {"alg":"none"}, payload {"sub":"alice","role":"admin"}.
        let token = "eyJhbGciOiJub25lIn0.eyJzdWIiOiJhbGljZSIsInJvbGUiOiJhZG1pbiJ9";
        let event = run_exec(r#"e.absorb_jwt("line")"#, build_event_with_line(token));
        assert_eq!(field_str(&event, "sub").as_deref(), Some("alice"));
        assert_eq!(field_str(&event, "role").as_deref(), Some("admin"));
    }

    #[test]
    fn merge_persists_to_event() {
        let event = run_exec(r#"e.merge(#{ z: "9" })"#, build_event_with_line("x"));
        assert_eq!(field_str(&event, "z").as_deref(), Some("9"));
    }

    #[test]
    fn rename_field_persists_to_event() {
        let mut event = build_event_with_line("x");
        event.set_field("old".to_string(), Dynamic::from("v".to_string()));
        let event = run_exec(r#"e.rename_field("old", "new")"#, event);
        assert_eq!(field_str(&event, "new").as_deref(), Some("v"));
        assert!(event.fields.get("old").is_none());
    }

    #[test]
    fn nested_array_mutation_persists_to_event() {
        let mut event = build_event_with_line("x");
        event.set_field(
            "tags".to_string(),
            Dynamic::from(vec![Dynamic::from("a".to_string())]),
        );
        // Push without any accompanying assignment must still be written back.
        let event = run_exec(r#"e.tags.push("b")"#, event);
        let tags = event
            .fields
            .get("tags")
            .and_then(|v| v.clone().try_cast::<rhai::Array>())
            .expect("tags array");
        let tags: Vec<String> = tags.into_iter().filter_map(|v| v.try_cast()).collect();
        assert_eq!(tags, vec!["a".to_string(), "b".to_string()]);
    }

    #[test]
    fn nested_mutation_inside_block_persists_to_event() {
        let mut event = build_event_with_line("x");
        event.set_field("level".to_string(), Dynamic::from("ERROR".to_string()));
        event.set_field("tags".to_string(), Dynamic::from(rhai::Array::new()));
        // Mutation buried in a conditional block (not a top-level statement).
        let event = run_exec(r#"if e.level == "ERROR" { e.tags.push("flagged") }"#, event);
        let tags = event
            .fields
            .get("tags")
            .and_then(|v| v.clone().try_cast::<rhai::Array>())
            .expect("tags array");
        assert_eq!(tags.len(), 1);
    }

    #[test]
    fn mutating_calls_set_mutates_event_flag() {
        let mut engine = RhaiEngine::new();
        for script in [
            r#"e.absorb_kv("line")"#,
            r#"e.absorb_logfmt("line")"#,
            r#"e.absorb_json("payload")"#,
            r#"e.absorb_jwt("token")"#,
            r##"e.absorb_regex("line", #"(?P<u>\w+)"#)"##,
            r#"e.merge(#{ z: 1 })"#,
            r#"e.enrich(#{ z: 1 })"#,
            r#"e.rename_field("a", "b")"#,
            r#"absorb_kv(e, "line")"#,
            r#"e.tags.push("c")"#,
            r#"e.tags.clear()"#,
            r#"e.meta.merge(#{ y: 1 })"#,
            r#"if e.lvl == "X" { e.tags.push(1) }"#,
        ] {
            assert!(
                engine.compile_exec(script).unwrap().mutates_event,
                "expected mutates_event=true for: {script}"
            );
        }
    }

    #[test]
    fn mutator_on_unrelated_value_does_not_flag_event() {
        // A mutating call on a non-`e` value, even inside an `e`-referencing branch,
        // must not trigger the writeback for `e`.
        let mut engine = RhaiEngine::new();
        for script in [
            r#"if e.lvl == "X" { let a = [1]; a.push(2) }"#,
            r#"let tmp = #{}; tmp.set("k", e.lvl)"#,
        ] {
            assert!(
                !engine.compile_exec(script).unwrap().mutates_event,
                "expected mutates_event=false for: {script}"
            );
        }
    }

    #[test]
    fn read_only_calls_do_not_set_mutates_event_flag() {
        let mut engine = RhaiEngine::new();
        for script in [
            r#"e.has("x")"#,
            r#"e.get_path("a.b")"#,
            r#"print(e.msg)"#,
            r#"track_freq("level", e.level)"#,
        ] {
            assert!(
                !engine.compile_exec(script).unwrap().mutates_event,
                "expected mutates_event=false for: {script}"
            );
        }
    }

    #[test]
    fn assignment_replaces_entire_event_map() {
        let engine = RhaiEngine::new();
        let mut event = build_event_with_line("orig line");
        event.set_field("keep".to_string(), Dynamic::from("value"));

        let mut scope = engine.create_scope_for_event(&event);

        let mut new_map = rhai::Map::new();
        new_map.insert("ts".into(), Dynamic::from("2025-09-22"));
        scope.set_value("e", new_map);

        let mut event_clone = event.clone();
        engine.update_event_from_scope(&mut event_clone, &scope);

        assert!(event_clone.fields.get("line").is_none());
        assert!(event_clone.fields.get("keep").is_none());
        assert_eq!(
            event_clone
                .fields
                .get("ts")
                .and_then(|v| v.clone().try_cast::<String>())
                .as_deref(),
            Some("2025-09-22")
        );
        assert_eq!(event_clone.fields.len(), 1);
    }

    #[test]
    fn unit_values_still_remove_fields() {
        let engine = RhaiEngine::new();
        let mut event = build_event_with_line("orig line");
        event.set_field("msg".to_string(), Dynamic::from("hello"));

        let mut scope = engine.create_scope_for_event(&event);

        let mut updated_map = rhai::Map::new();
        updated_map.insert("msg".into(), Dynamic::from("world"));
        updated_map.insert("line".into(), Dynamic::UNIT);
        scope.set_value("e", updated_map);

        let mut event_clone = event.clone();
        engine.update_event_from_scope(&mut event_clone, &scope);

        assert!(event_clone.fields.get("line").is_none());
        assert_eq!(
            event_clone
                .fields
                .get("msg")
                .and_then(|v| v.clone().try_cast::<String>())
                .as_deref(),
            Some("world")
        );
    }

    #[test]
    fn in_place_mutations_preserve_unchanged_fields() {
        let engine = RhaiEngine::new();
        let mut event = build_event_with_line("orig line");
        event.set_field("level".to_string(), Dynamic::from("INFO"));

        let mut scope = engine.create_scope_for_event(&event);

        // Simulate in-place mutation by starting from the existing map values
        let mut mutated_map = scope.get_value::<Map>("e").unwrap();
        mutated_map.insert("level".into(), Dynamic::from("ERROR"));
        scope.set_value("e", mutated_map);

        let mut event_clone = event.clone();
        engine.update_event_from_scope(&mut event_clone, &scope);

        assert!(event_clone.fields.get("line").is_some());
        assert_eq!(
            event_clone
                .fields
                .get("level")
                .and_then(|v| v.clone().try_cast::<String>())
                .as_deref(),
            Some("ERROR")
        );
    }

    #[test]
    fn update_event_preserves_field_order_and_appends_new_keys() {
        let engine = RhaiEngine::new();
        let mut event = build_event_with_line("orig line");
        event.set_field("z".to_string(), Dynamic::from(1_i64));
        event.set_field("a".to_string(), Dynamic::from(2_i64));
        event.set_field("b".to_string(), Dynamic::from(3_i64));

        let mut scope = engine.create_scope_for_event(&event);

        let mut mutated_map = scope.get_value::<Map>("e").unwrap();
        mutated_map.insert("foo".into(), Dynamic::from(42_i64));
        scope.set_value("e", mutated_map);

        let mut event_clone = event.clone();
        engine.update_event_from_scope(&mut event_clone, &scope);

        let keys: Vec<String> = event_clone.fields.keys().cloned().collect();
        assert_eq!(keys, vec!["line", "z", "a", "b", "foo"]);
    }

    #[test]
    fn meta_includes_parsed_timestamp_before_scripts() {
        let engine = RhaiEngine::new();
        let mut event = build_event_with_line("orig line");
        let ts = Utc.timestamp_opt(1_700_000_000, 123_000_000).unwrap();
        event.parsed_ts = Some(ts);

        let scope = engine.create_scope_for_event(&event);
        let meta = scope.get_value::<Map>("meta").expect("meta map");

        let parsed_ts = meta
            .get("parsed_ts")
            .cloned()
            .expect("parsed_ts should be present in meta");
        let dt = parsed_ts
            .try_cast::<crate::rhai_functions::datetime::DateTimeWrapper>()
            .expect("parsed_ts should be a DateTimeWrapper");
        assert_eq!(dt.inner.to_rfc3339(), ts.to_rfc3339());
    }

    #[test]
    fn meta_omits_parsed_timestamp_when_missing() {
        let engine = RhaiEngine::new();
        let event = build_event_with_line("orig line");

        let scope = engine.create_scope_for_event(&event);
        let meta = scope.get_value::<Map>("meta").expect("meta map");
        assert!(
            !meta.contains_key("parsed_ts"),
            "meta.parsed_ts should be absent when event has no parsed timestamp"
        );
    }

    #[test]
    fn compile_exec_tracks_event_mutation() {
        let mut engine = RhaiEngine::new();

        let read_only = engine
            .compile_exec(r#"track_sum("status_codes", e.status)"#)
            .expect("read-only exec should compile");
        assert!(
            !read_only.mutates_event,
            "read-only exec should skip event write-back"
        );

        let mutating = engine
            .compile_exec(r#"e.level = "ERROR""#)
            .expect("mutating exec should compile");
        assert!(
            mutating.mutates_event,
            "field assignment should require event write-back"
        );

        let replace_map = engine
            .compile_exec("e = ()")
            .expect("whole-map assignment should compile");
        assert!(
            replace_map.mutates_event,
            "whole-map assignment should require event write-back"
        );
    }

    #[test]
    fn compile_filter_tracks_specific_meta_fields() {
        let mut engine = RhaiEngine::new();

        let compiled = engine
            .compile_filter(r#"meta.filename == "app.log""#)
            .expect("meta filter should compile");

        assert!(compiled.uses_meta, "meta filter should use meta");
        assert!(compiled.meta_usage.filename, "filename should be requested");
        assert!(
            !compiled.meta_usage.line,
            "unreferenced meta.line should not be requested"
        );
        assert!(
            !compiled.meta_usage.parsed_ts,
            "unreferenced meta.parsed_ts should not be requested"
        );
    }

    #[test]
    fn compile_filter_detects_window_usage() {
        let mut engine = RhaiEngine::new();

        // Filters that do not reference `window` must report uses_window == false
        // so the pipeline can route them to the lighter non-window evaluator
        // (which also enables the native fast-path).
        for expr in [
            r#"e.level == "ERROR""#,
            "e.message.len() > 5",
            "e.status >= 500",
        ] {
            let compiled = engine.compile_filter(expr).expect("should compile");
            assert!(
                !compiled.uses_window(),
                "filter `{expr}` does not reference window"
            );
        }

        // Filters that do reference `window` must report uses_window == true so
        // the window array is built and window semantics are preserved.
        for expr in ["window.len() > 1", r#"window[1].level == "ERROR""#] {
            let compiled = engine.compile_filter(expr).expect("should compile");
            assert!(compiled.uses_window(), "filter `{expr}` references window");
        }
    }

    #[test]
    fn render_snippet_marks_correct_line_and_col() {
        let script = "let x = 1;\nlet y = foo(x);\nlet z = y + 1;";
        let snippet = RhaiEngine::render_snippet(script, 1, 7).expect("snippet");
        assert!(snippet.contains("2 | let y = foo(x);"));
        assert!(snippet.contains("^"));
        // caret should land under the 8th character (zero-based 7) of line 2
        let caret_line = snippet.lines().nth(1).unwrap_or_default();
        assert!(caret_line.ends_with("^"));
        assert!(caret_line.contains("|        ^"));
    }

    #[test]
    fn parse_error_suggests_rhai_raw_string_syntax() {
        let mut engine = RhaiEngine::new();
        let err = engine
            .compile_exec("e.line.extract_regex(r\"User (\\\\d+)\")")
            .err()
            .expect("r\"...\" should be invalid in Rhai");
        let msg = err.to_string();
        assert!(
            msg.contains("Rhai raw strings use #\"...\"#"),
            "raw string hint should mention Rhai syntax; got: {msg}"
        );
    }

    #[test]
    fn property_suggestion_shows_available_fields_without_verbose() {
        let config = DebugConfig::new(0);
        let enhancer = ErrorEnhancer::new(config);
        let mut scope = Scope::new();
        let mut e_map = Map::new();
        e_map.insert("status".into(), Dynamic::from("OK"));
        e_map.insert("status_code".into(), Dynamic::from(200_i64));
        scope.push("e", e_map);

        let err = EvalAltResult::ErrorPropertyNotFound("statsu".into(), rhai::Position::NONE);
        let ctx = debug::ExecutionContext::default();
        let out = enhancer.enhance_error(&err, &scope, "e.statsu", "filter", &ctx);

        eprintln!("enhanced error:\n{}", out);
        assert!(
            out.contains("status"),
            "output should surface available fields even when verbosity is zero"
        );
    }

    #[test]
    fn verbose_diagnostic_respects_no_emoji_setting() {
        let tracker = DebugTracker::new(DebugConfig::new(1).with_emoji(false));
        let mut scope = Scope::new();
        let mut e_map = Map::new();
        e_map.insert("status".into(), Dynamic::from("OK"));
        scope.push("e", e_map);

        let err = Box::new(EvalAltResult::ErrorPropertyNotFound(
            "statsu".into(),
            rhai::Position::NONE,
        ));
        let out = RhaiEngine::format_rhai_diagnostic(
            err,
            "filter",
            "filter expression",
            "e.statsu",
            Some(&scope),
            Some(&tracker),
            false,
        );

        assert!(
            out.starts_with("filter error"),
            "verbose diagnostic header should match the non-debug \"<stage> error\" form: {out}"
        );
        assert!(
            out.contains("Hint:"),
            "verbose diagnostic should use text hint prefix when emoji are disabled: {out}"
        );
        assert!(
            !out.contains('🔸') && !out.contains('💡'),
            "verbose diagnostic should not contain emoji when disabled: {out}"
        );
    }

    #[test]
    fn runtime_error_with_unit_type_suggests_get_path() {
        let config = DebugConfig::new(0);
        let enhancer = ErrorEnhancer::new(config);
        let scope = Scope::new();

        let err = EvalAltResult::ErrorRuntime(
            "track_freq requires a string name; got ()".into(),
            rhai::Position::NONE,
        );
        let ctx = debug::ExecutionContext::default();
        let out = enhancer.enhance_error(
            &err,
            &scope,
            "track_freq(\"endpoint\", e.endpoint)",
            "exec",
            &ctx,
        );

        eprintln!("enhanced error:\n{}", out);
        assert!(
            out.contains("field is missing"),
            "output should explain that () means a missing field"
        );
        assert!(
            out.contains("get_path"),
            "output should suggest get_path for handling missing fields"
        );
        assert!(
            out.contains("has_path"),
            "output should suggest has_path for checking field existence"
        );
    }

    #[test]
    fn dot_expr_on_unit_suggests_get_path() {
        // `e.user.role` when `user` is absent leaves a () in the chain, so the
        // `.role` getter fails on type '()'. The hint should steer toward the
        // safe path accessors instead of leaking the raw Rhai message.
        let config = DebugConfig::new(0);
        let enhancer = ErrorEnhancer::new(config);
        let scope = Scope::new();

        let err = EvalAltResult::ErrorDotExpr(
            "Unknown property 'role' - a getter is not registered for type '()'".into(),
            rhai::Position::NONE,
        );
        let ctx = debug::ExecutionContext::default();
        let out = enhancer.enhance_error(&err, &scope, "e.user.role == \"admin\"", "filter", &ctx);

        assert!(
            out.contains("get_path"),
            "output should suggest get_path for missing nested fields; got: {out}"
        );
        assert!(
            out.contains("has_path"),
            "output should suggest has_path for checking nested paths; got: {out}"
        );
    }

    #[test]
    fn bare_field_reference_suggests_e_accessor() {
        let config = DebugConfig::new(0);
        let enhancer = ErrorEnhancer::new(config);
        let mut scope = Scope::new();
        let mut e_map = Map::new();
        e_map.insert("level".into(), Dynamic::from("ERROR"));
        e_map.insert("status".into(), Dynamic::from(500_i64));
        scope.push("e", e_map);

        // Exact field name written without the `e.` prefix.
        let err = EvalAltResult::ErrorVariableNotFound("level".into(), rhai::Position::NONE);
        let ctx = debug::ExecutionContext::default();
        let out = enhancer.enhance_error(&err, &scope, "level == \"ERROR\"", "filter", &ctx);
        assert!(
            out.contains("e.level"),
            "bare field reference should be redirected to e.level; got: {out}"
        );
        assert!(
            out.contains("accessed through `e`"),
            "hint should teach the e. accessor model; got: {out}"
        );

        // A close partial match of a field name should still reach the field.
        let err = EvalAltResult::ErrorVariableNotFound("stat".into(), rhai::Position::NONE);
        let out = enhancer.enhance_error(&err, &scope, "stat > 0", "filter", &ctx);
        assert!(
            out.contains("e.status"),
            "partial bare field should suggest e.status; got: {out}"
        );

        // A boundary-similarity transposition (distance 2 over 5 chars) too.
        let err = EvalAltResult::ErrorVariableNotFound("levle".into(), rhai::Position::NONE);
        let out = enhancer.enhance_error(&err, &scope, "levle == \"ERROR\"", "filter", &ctx);
        assert!(
            out.contains("e.level"),
            "transposition typo should still suggest e.level; got: {out}"
        );

        // A truly unknown identifier keeps the generic variable listing.
        let err = EvalAltResult::ErrorVariableNotFound("zzzzz".into(), rhai::Position::NONE);
        let out = enhancer.enhance_error(&err, &scope, "zzzzz == \"ERROR\"", "filter", &ctx);
        assert!(
            out.contains("Available variables"),
            "unknown identifier should fall back to the variable listing; got: {out}"
        );
    }

    #[test]
    fn native_filter_evaluates_simple_comparisons() {
        let mut engine = RhaiEngine::new();
        let compiled = engine
            .compile_filter("e.level == \"ERROR\" && e.status >= 500")
            .expect("filter should compile");

        assert!(compiled.native_predicate.is_some());

        let mut event = build_event_with_line("line");
        event.set_field("level".to_string(), Dynamic::from("ERROR"));
        event.set_field("status".to_string(), Dynamic::from(500_i64));

        let mut metrics = HashMap::new();
        let mut internal = HashMap::new();
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(result);

        let mut event_nonmatch = event.clone();
        event_nonmatch.set_field("status".to_string(), Dynamic::from(200_i64));
        let result = engine
            .execute_compiled_filter(&compiled, &event_nonmatch, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(!result);
    }

    #[test]
    fn native_filter_handles_string_methods() {
        let mut engine = RhaiEngine::new();

        // String methods now have native predicate support
        let compiled = engine
            .compile_filter("e.level.starts_with(\"ERR\")")
            .expect("filter should compile");
        assert!(
            compiled.native_predicate.is_some(),
            "starts_with should have native predicate"
        );

        // Test contains
        let compiled = engine
            .compile_filter("e.message.contains(\"error\")")
            .expect("filter should compile");
        assert!(
            compiled.native_predicate.is_some(),
            "contains should have native predicate"
        );

        // Test ends_with
        let compiled = engine
            .compile_filter("e.path.ends_with(\".log\")")
            .expect("filter should compile");
        assert!(
            compiled.native_predicate.is_some(),
            "ends_with should have native predicate"
        );

        // Unsupported methods should still fall back to Rhai
        let compiled = engine
            .compile_filter("e.message.to_upper()")
            .expect("filter should compile");
        assert!(
            compiled.native_predicate.is_none(),
            "to_upper should NOT have native predicate"
        );
    }

    #[test]
    fn native_string_methods_evaluate_correctly() {
        let mut engine = RhaiEngine::new();

        // Test starts_with
        let compiled = engine
            .compile_filter("e.level.starts_with(\"ERR\")")
            .expect("filter should compile");

        let mut event = build_event_with_line("test");
        event.set_field("level".to_string(), Dynamic::from("ERROR"));
        let mut metrics = HashMap::new();
        let mut internal = HashMap::new();
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(result, "ERROR should start with ERR");

        event.set_field("level".to_string(), Dynamic::from("WARNING"));
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(!result, "WARNING should not start with ERR");

        // Test contains
        let compiled = engine
            .compile_filter("e.message.contains(\"fail\")")
            .expect("filter should compile");

        event.set_field("message".to_string(), Dynamic::from("connection failed"));
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(result, "'connection failed' should contain 'fail'");

        event.set_field("message".to_string(), Dynamic::from("success"));
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(!result, "'success' should not contain 'fail'");

        // Test ends_with
        let compiled = engine
            .compile_filter("e.path.ends_with(\".log\")")
            .expect("filter should compile");

        event.set_field("path".to_string(), Dynamic::from("/var/log/app.log"));
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(result, "'/var/log/app.log' should end with '.log'");

        event.set_field("path".to_string(), Dynamic::from("/var/log/app.txt"));
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("native filter should succeed");
        assert!(!result, "'/var/log/app.txt' should not end with '.log'");
    }

    #[test]
    fn variable_usage_detection() {
        let mut engine = RhaiEngine::new();

        // Filter using only e.* - should not need meta, conf, or line
        let compiled = engine
            .compile_filter("e.level == \"ERROR\"")
            .expect("filter should compile");
        assert!(!compiled.uses_meta, "simple filter should not use meta");
        assert!(!compiled.uses_conf, "simple filter should not use conf");
        assert!(!compiled.uses_line, "simple filter should not use line");

        // Filter using meta
        let compiled = engine
            .compile_filter("meta.line_num > 100")
            .expect("filter should compile");
        assert!(compiled.uses_meta, "filter with meta.* should use meta");
        assert!(!compiled.uses_conf, "filter should not use conf");

        // Filter using conf
        let compiled = engine
            .compile_filter("conf.threshold > 5")
            .expect("filter should compile");
        assert!(compiled.uses_conf, "filter with conf.* should use conf");
        assert!(!compiled.uses_meta, "filter should not use meta");

        // Filter using line
        let compiled = engine
            .compile_filter("line.len() > 100")
            .expect("filter should compile");
        assert!(compiled.uses_line, "filter with line should use line");

        // Combined usage
        let compiled = engine
            .compile_filter("e.level == \"ERROR\" && meta.line_num > 0")
            .expect("filter should compile");
        assert!(compiled.uses_meta, "combined filter should use meta");
        assert!(!compiled.uses_conf, "combined filter should not use conf");
    }

    #[test]
    fn filter_includes_can_define_helpers() {
        let mut engine = RhaiEngine::new();
        let includes = vec![crate::config::IncludeFile {
            path: "helpers.rhai".to_string(),
            content: "fn is_error(level) { level == \"ERROR\" }".to_string(),
        }];

        let compiled = engine
            .compile_filter_with_includes("is_error(e.level)", &includes)
            .expect("filter should compile with includes");

        let mut event = build_event_with_line("line");
        event.set_field("level".to_string(), Dynamic::from("ERROR"));

        let mut metrics = HashMap::new();
        let mut internal = HashMap::new();
        let result = engine
            .execute_compiled_filter(&compiled, &event, &mut metrics, &mut internal)
            .expect("filter should execute");
        assert!(result);
    }

    #[test]
    fn filter_includes_reject_statements() {
        let mut engine = RhaiEngine::new();
        let includes = vec![crate::config::IncludeFile {
            path: "helpers.rhai".to_string(),
            content: "let x = 1;".to_string(),
        }];

        let err = engine.compile_filter_with_includes("e.level == \"ERROR\"", &includes);
        match err {
            Ok(_) => panic!("filters should reject include statements"),
            Err(err) => {
                assert!(err.to_string().contains("cannot contain statements"));
            }
        }
    }

    #[test]
    fn function_suggestion_offers_len_for_length_typo() {
        let config = DebugConfig::new(0);
        let enhancer = ErrorEnhancer::new(config);
        let scope = Scope::new();
        let err =
            EvalAltResult::ErrorFunctionNotFound("length(string)".into(), rhai::Position::NONE);
        let ctx = debug::ExecutionContext::default();
        let out = enhancer.enhance_error(&err, &scope, "length(s)", "filter", &ctx);
        assert!(
            out.contains("len"),
            "function suggestion should offer len() for length typo; got: {out}"
        );
    }

    #[test]
    fn nested_function_errors_show_call_stack() {
        let inner = Box::new(EvalAltResult::ErrorRuntime(
            "boom".into(),
            rhai::Position::new(3, 1),
        ));
        let mid = Box::new(EvalAltResult::ErrorInFunctionCall(
            "child".into(),
            "".into(),
            inner,
            rhai::Position::new(2, 1),
        ));
        let outer = Box::new(EvalAltResult::ErrorInFunctionCall(
            "parent".into(),
            "".into(),
            mid,
            rhai::Position::new(1, 1),
        ));

        let msg = RhaiEngine::format_rhai_diagnostic(
            outer, "filter", "script", "child()", None, None, true,
        );

        assert!(
            msg.contains("Call stack") && msg.contains("parent") && msg.contains("child"),
            "call stack should include nested function frames; got: {msg}"
        );
    }

    #[test]
    fn unit_arg_suggestion_points_to_missing_field() {
        let msg = RhaiEngine::format_function_not_found_error(
            "foo((), string)".to_string(),
            "script",
            rhai::Position::NONE,
        );
        assert!(
            (msg.contains("missing field") || msg.contains("e.has")) && msg.contains("Called with"),
            "unit arg hint should mention missing field guards and show called types; got: {msg}"
        );
    }

    #[test]
    fn typo_suggests_track_percentiles_first() {
        // Regression: function_catalog() must list every track_* function so the
        // similarity matcher can suggest them. A singular `track_percentile` typo
        // should rank the real `track_percentiles` ahead of unrelated trackers.
        let config = DebugConfig::new(0);
        let enhancer = ErrorEnhancer::new(config);
        let scope = Scope::new();
        let err = EvalAltResult::ErrorFunctionNotFound(
            "track_percentile (string, i64)".into(),
            rhai::Position::NONE,
        );
        let hint = enhancer
            .generate_suggestions(&err, &scope, None)
            .expect("expected a suggestion for a track_* typo");
        let first = hint
            .trim_start_matches("Did you mean: ")
            .split(',')
            .next()
            .unwrap_or("")
            .trim();
        assert_eq!(
            first, "track_percentiles",
            "track_percentiles should be the top suggestion; got: {hint}"
        );
    }

    #[test]
    fn type_mismatch_hints_bool_in_filter() {
        let config = DebugConfig::new(0);
        let enhancer = ErrorEnhancer::new(config);
        let scope = Scope::new();
        let err = EvalAltResult::ErrorMismatchDataType(
            "bool".into(),
            "string".into(),
            rhai::Position::NONE,
        );
        let ctx = debug::ExecutionContext::default();
        let out = enhancer.enhance_error(&err, &scope, "e.level", "filter", &ctx);
        assert!(
            out.contains("Filters must return true/false"),
            "type mismatch in filter should remind about boolean return; got: {out}"
        );
    }
}