tellaro-query-language 3.0.1

//! Query evaluator for TQL.
//!
//! Executes parsed TQL queries against JSON records.

use crate::comparator;
use crate::error::{Result, TqlError};
use crate::field_accessor;
use crate::mutators;
use crate::parser::{
    AstNode, CollectionOpNode, ComparisonNode, LogicalOpNode, UnaryOpNode, Value as AstValue,
};
use serde_json::{json, Value as JsonValue};
use std::collections::HashMap;

// Test-only instrumentation: how many times a value-mutator chain has actually
// been RESOLVED (mutator constructed + applied) against one scalar operand.
//
// The value operand is record-independent by construction --
// [`TqlEvaluator::apply_value_mutators_scalar`] hands the mutator a scratch
// `{}` rather than the record -- so a chain resolved once per record is N-1
// resolutions of the identical input. For `| nslookup` that is one real DNS
// query per record and for `| geoip` one MaxMind mmap per record, both
// reachable from user-controlled query text. The invariant this counter pins
// is therefore a COUNT, not a duration: a wall-clock assertion is flaky and
// would still pass over a chain resolved twice.
//
// THREAD-LOCAL, not a global. `cargo test` runs each `#[test]` on its own
// thread and the evaluation happens on that same thread, so a thread-local
// counter observes exactly this test's work. A process-global counter does not:
// the first version of this was an `AtomicUsize` and it read 3 where it should
// have read 2, because two unrelated tests in this module resolved a chain
// concurrently. A mutex around the readers would not have fixed that -- the
// interference came from tests that never touch the counter at all.
#[cfg(test)]
thread_local! {
    pub(crate) static VALUE_MUTATOR_RESOLUTIONS: std::cell::Cell<usize> =
        const { std::cell::Cell::new(0) };
}

/// Convert an AST Value to a serde_json Value
fn ast_value_to_json(value: &AstValue) -> JsonValue {
    match value {
        AstValue::String(s) => json!(s),
        AstValue::Integer(i) => json!(i),
        AstValue::Float(f) => json!(f),
        AstValue::Boolean(b) => json!(b),
        AstValue::List(list) => json!(list.iter().map(ast_value_to_json).collect::<Vec<_>>()),
        AstValue::Null => json!(null),
    }
}

/// Convert a serde_json Value back to an AST Value.
///
/// The inverse of [`ast_value_to_json`], needed because a value mutator runs on
/// `JsonValue` (the shape [`Mutator::apply`] speaks) while the comparator takes
/// an [`AstValue`]. An object has no AST spelling -- an enrichment mutator is not
/// a meaningful value mutator -- so it degrades to its JSON text rather than
/// silently becoming `Null`, which would compare equal to a missing field.
fn json_to_ast_value(value: &JsonValue) -> AstValue {
    match value {
        JsonValue::String(s) => AstValue::String(s.clone()),
        JsonValue::Bool(b) => AstValue::Boolean(*b),
        JsonValue::Number(n) => match n.as_i64() {
            Some(i) => AstValue::Integer(i),
            None => AstValue::Float(n.as_f64().unwrap_or(f64::NAN)),
        },
        JsonValue::Array(items) => AstValue::List(items.iter().map(json_to_ast_value).collect()),
        JsonValue::Null => AstValue::Null,
        JsonValue::Object(_) => AstValue::String(value.to_string()),
    }
}

/// Render an operand for an error message the way the author typed it.
fn render_literal(value: &AstValue) -> String {
    match value {
        AstValue::String(s) => s.clone(),
        other => ast_value_to_json(other).to_string(),
    }
}

/// The collection spellings that REFUSE a list operand.
///
/// `all` / `not_all` are deliberately absent. They are answered by a Painless
/// script that compares each element to `params.value`, for which a list is a
/// valid (never-matching) operand — so the translator does not refuse them and
/// neither does this.
const COLLECTION_OPS_REFUSING_A_LIST: [&str; 4] = ["any", "none", "not_any", "not_none"];

/// `any` / `none` with a LIST operand is ill-typed. Say so, as the translator does.
///
/// `in` is already the membership operator. `f any ['a','b']` does not ask "is
/// any element of `f` one of these two values" — it asks whether any single
/// ELEMENT of `f` equals the two-element LIST, which no scalar element can. The
/// grammar admits a list here only because `value` includes `list_value`.
///
/// The translator refused this in `ab55587` (`OpenSearchError::TypeError`, naming
/// `in`) and the evaluator went on answering `false`. That was recorded as an
/// acceptable split — the two layers agreeing that nothing matches while
/// differing in KIND. It is not one, because **a translator refusal is not local
/// to the clause**: it fails the whole query. Measured on both engines over
/// `[{id:1,code:1,f:["z"]}, {id:2,code:2,f:["z"]}]`:
///
/// ```text
///                                 evaluator (Rust and Python)   translator
///   f any ['a','z']               []                            refuses
///   code eq 1 OR f any ['a','z']  ["1"]                         refuses -> nothing
///   f none ['a','z']              ["1","2"]                     refuses -> nothing
/// ```
///
/// So under `OR`, and under the `none` spelling, the two execution paths return
/// different RECORD SETS. `f none [...]` is the dangerous one: an exclusion
/// clause the evaluator answers with *every* record and the translator answers
/// with an error.
///
/// Raising is the correct half of the split, so the evaluator is raised to meet
/// the translator rather than the translator lowered to `false`.
///
/// The message is Python's `TQLTypeError` text, byte for byte, so the two
/// engines refuse identically as well as refusing together.
fn refuse_list_operand(field: &str, operator: &str) -> TqlError {
    TqlError::TypeError(format!(
        "Cannot apply operator '{operator}' to field '{field}' of type 'list operand'. \
         Valid operators for list operand fields: in, not in"
    ))
}

/// Is this collection node's operand an ill-typed list?
///
/// A ONE-element list is NOT: `f any ['a']` unwraps to the scalar form, which is
/// well-typed and which the shipped corpus actually contains. An EMPTY list
/// literal is refused by the parser (`952dabc`) and never reaches here.
fn collection_operand_is_ill_typed(coll: &CollectionOpNode) -> bool {
    COLLECTION_OPS_REFUSING_A_LIST.contains(&coll.operator.as_str())
        && matches!(&coll.value, AstValue::List(items) if items.len() != 1)
}

/// Walk a subtree the evaluator is about to SKIP and refuse anything ill-typed
/// in it.
///
/// Without this the refusal would be data-dependent: `and`/`or` short-circuit,
/// so `code eq 1 OR f any ['a','z']` would raise only for a corpus containing a
/// record with `code != 1`, and would quietly return rows for one where every
/// record satisfies the left branch. A query that is well-typed or not depending
/// on the data is exactly the silent shape this refusal exists to remove — the
/// translator refuses the query without seeing a single record, and so must this.
fn reject_ill_typed_collection_operands(node: &AstNode) -> Result<()> {
    match node {
        AstNode::CollectionOp(coll) => {
            if collection_operand_is_ill_typed(coll) {
                return Err(refuse_list_operand(&coll.field, &coll.operator));
            }
            Ok(())
        }
        AstNode::LogicalOp(logical) => {
            reject_ill_typed_collection_operands(&logical.left)?;
            reject_ill_typed_collection_operands(&logical.right)
        }
        AstNode::UnaryOp(unary) => reject_ill_typed_collection_operands(&unary.operand),
        _ => Ok(()),
    }
}

/// Build a MutatorParams HashMap from a Mutator spec's named_args and positional args.
///
/// ## Positional argument convention
///
/// When a mutator is invoked with positional arguments (e.g., `| split(',')` or
/// `| replace('old', 'new')`), the parser produces an ordered list of values in
/// `mutator_spec.args`. This function stores them under string keys "0", "1", "2", ...
/// so that mutator implementations can retrieve them by index.
///
/// Individual mutators use [`mutators::get_param`] to look up a parameter by its
/// named key first and fall back to the positional index. For example,
/// `ReplaceMutator::get_find()` checks for key `"find"` (named), then key `"0"`
/// (positional). This lets both `| replace(find='old', replace='new')` and
/// `| replace('old', 'new')` work identically.
fn build_mutator_params(
    mutator_spec: &crate::parser::Mutator,
) -> Option<HashMap<String, JsonValue>> {
    if !mutator_spec.named_args.is_empty() {
        let mut map = HashMap::new();
        for (k, v) in &mutator_spec.named_args {
            map.insert(k.clone(), ast_value_to_json(v));
        }
        Some(map)
    } else if !mutator_spec.args.is_empty() {
        let mut map = HashMap::new();
        for (i, v) in mutator_spec.args.iter().enumerate() {
            map.insert(i.to_string(), ast_value_to_json(v));
        }
        Some(map)
    } else {
        None
    }
}

/// Evaluator for TQL queries
pub struct TqlEvaluator {
    /// Maximum recursion depth for nested expressions
    max_depth: usize,
}

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

impl TqlEvaluator {
    /// Maximum evaluation depth to prevent stack overflow
    pub const MAX_EVAL_DEPTH: usize = 100;

    /// Create a new evaluator with default settings
    pub fn new() -> Self {
        Self {
            max_depth: Self::MAX_EVAL_DEPTH,
        }
    }

    /// Create a new evaluator with custom max depth
    pub fn with_max_depth(max_depth: usize) -> Self {
        Self { max_depth }
    }

    /// Evaluate a query against a single record
    ///
    /// # Arguments
    ///
    /// * `ast` - The parsed query AST
    /// * `record` - The JSON record to evaluate against
    ///
    /// # Returns
    ///
    /// true if the record matches the query, false otherwise
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use tql::evaluator::TqlEvaluator;
    /// use tql::parser::TqlParser;
    /// use serde_json::json;
    ///
    /// let parser = TqlParser::new();
    /// let evaluator = TqlEvaluator::new();
    ///
    /// let ast = parser.parse("age > 25").unwrap();
    /// let record = json!({"age": 30, "name": "John"});
    ///
    /// assert!(evaluator.evaluate(&ast, &record).unwrap());
    /// ```
    pub fn evaluate(&self, ast: &AstNode, record: &JsonValue) -> Result<bool> {
        self.evaluate_with_depth(ast, record, 0)
    }

    /// Evaluate a query with depth tracking
    fn evaluate_with_depth(&self, ast: &AstNode, record: &JsonValue, depth: usize) -> Result<bool> {
        // Check depth limit
        if depth > self.max_depth {
            return Err(TqlError::ExecutionError(format!(
                "Evaluation depth exceeds maximum of {}",
                self.max_depth
            )));
        }

        match ast {
            AstNode::MatchAll => Ok(true),
            AstNode::Comparison(comp) => self.evaluate_comparison(comp, record),
            AstNode::LogicalOp(logical) => self.evaluate_logical_op(logical, record, depth + 1),
            AstNode::UnaryOp(unary) => self.evaluate_unary_op(unary, record, depth + 1),
            AstNode::CollectionOp(coll) => self.evaluate_collection_op(coll, record),
            AstNode::GeoExpr(geo) => {
                // GeoIP expressions: check if field exists, return true for evaluation
                // Actual GeoIP lookup and condition evaluation happens in post-processing
                let field_exists = field_accessor::field_exists(record, &geo.field)?;
                if !field_exists {
                    return Ok(false);
                }
                // Whether there are conditions or not, return true
                // - No conditions (enrichment-only): always include the record
                // - With conditions: conditions are evaluated in post-processing
                Ok(true)
            }
            AstNode::NslookupExpr(nslookup) => {
                // NSLookup expressions: check if field exists, return true for evaluation
                // Actual DNS lookup and condition evaluation happens in post-processing
                let field_exists = field_accessor::field_exists(record, &nslookup.field)?;
                if !field_exists {
                    return Ok(false);
                }
                // Whether there are conditions or not, return true
                // - No conditions (enrichment-only): always include the record
                // - With conditions: conditions are evaluated in post-processing
                Ok(true)
            }
            AstNode::StatsExpr(_) | AstNode::QueryWithStats(_) => {
                // Stats expressions require a different evaluation path
                Err(TqlError::ExecutionError(
                    "Stats expressions must be evaluated with evaluate_stats".to_string(),
                ))
            }
        }
    }

    /// Run a field's mutator chain over one value, separating a QUERY error from
    /// a DATA error the way Python does.
    ///
    /// * `Err(..)`  — the mutator could not be CONSTRUCTED (unknown name, bad
    ///   arguments). That is a defect in the query and stays fatal, exactly as
    ///   Python re-raises `TQLUnknownMutatorError` rather than absorbing it.
    /// * `Ok(None)` — the mutator could not process THIS record's value. That is
    ///   a data error, and the caller decides: the existence checks treat the
    ///   field as MISSING, every other operator answers `false`.
    ///
    /// Rust previously propagated BOTH, so a single unprocessable value aborted
    /// the whole query. Measured over the corpus in
    /// `evaluate_comparison`'s existence branch: `f | b64decode::number gt 0`
    /// returned `MutatorError("Base64 decode error: Invalid padding")` in Rust
    /// and the one matching record in Python. This is the same
    /// read-instruction rule the type hints already follow in this function, and
    /// for the same reason recorded there — a detection rule written that way
    /// stopped firing rather than matching less.
    ///
    /// # Every call site routes through here
    ///
    /// The split is POSITIONAL — `create_mutator` above the loop, `apply`
    /// inside it — rather than two `TqlError` variants, which is the same shape
    /// Python uses (it re-raises `TQLUnknownMutatorError` from its registry
    /// lookup and absorbs everything its `apply` raises). That makes the split a
    /// property of THIS FUNCTION, not of the error type, so a caller that
    /// re-inlines the loop silently loses it — and two of the four were still
    /// `?`-propagating both classes when this was written:
    ///
    /// * `evaluate_comparison`, existence branch — the field reads MISSING.
    /// * `evaluate_comparison`, value branch — the record answers `false`.
    /// * `evaluate_collection_op` — the record answers `false` for `any` /
    ///   `all` / `none` alike, negated ones included.
    /// * `apply_enrichment` — the field keeps its ORIGINAL value. That site
    ///   keeps its own loop for the mutator cache, so it reproduces the split
    ///   inline rather than calling here; it is the first place to look if the
    ///   classification drifts again.
    ///
    /// A fifth site would be a defect. `grep -n create_mutator tql/src` is the
    /// enumeration — it is short, and it is how the two missed sites were found.
    ///
    /// The residual: an `apply` that fails for a reason that is NOT about this
    /// record reads as a data error here. `geoip` with no database configured is
    /// the live example, and Python raises there. A missing `replace`
    /// find-parameter — the other apply-time query error in the tree — answers
    /// `false` in BOTH engines, so absorbing is right for that one. Telling the
    /// two apart needs a `TqlError` variant split (`error.rs`).
    fn apply_field_mutators(
        &self,
        field: &str,
        mutator_list: &[crate::parser::Mutator],
        record: &JsonValue,
        value: &JsonValue,
    ) -> Result<Option<JsonValue>> {
        let mut current_value = value.clone();
        for mutator_spec in mutator_list {
            let params = build_mutator_params(mutator_spec);
            let mutator = mutators::create_mutator(&mutator_spec.name, params)?;
            current_value = match mutator.apply(field, record, &current_value) {
                Ok(v) => v,
                Err(_) => return Ok(None),
            };

            // Check if this is an enrichment result - use _tql_return_value for comparison
            if let Some(return_value) = current_value.get("_tql_return_value") {
                current_value = return_value.clone();
            }
        }
        Ok(Some(current_value))
    }

    /// Apply the VALUE mutators -- the ones written against the literal the
    /// author typed, as in `f eq '%41' | urldecode`.
    ///
    /// Rust parsed these into `ComparisonNode::value_mutators` and then never
    /// looked at them: the comparator saw the RAW literal, so `f eq '%41' |
    /// urldecode` matched `{"f": "%41"}` while Python matched `{"f": "A"}`.
    /// Same query, two answers, no error -- and this engine is the agent's
    /// detection path.
    ///
    /// THREE SEMANTICS, each established by RUNNING Python rather than reading
    /// it (`src/tql/evaluator.py`; the behaviours are pinned by the shared
    /// fixtures in `cross_language_tests/fixtures/test_cases/mutators/`):
    ///
    /// 1. A list operand mutates ELEMENTWISE. `f eq ['%41'] | urldecode`
    ///    becomes `["A"]`, not the mutated text of the whole list.
    /// 2. Mutators chain left to right, as they do on the field side.
    /// 3. A failure is a QUERY error, not a data error -- see
    ///    [`Self::apply_value_mutators_scalar`].
    ///
    /// The record handed to the mutator is a SCRATCH object, not the caller's.
    /// A value mutator's operand comes from the query text, so it has no
    /// business reading the record -- and an enrichment mutator carrying a
    /// `field=` parameter writes into whatever record it is given. Python takes
    /// a scratch record here for that reason.
    fn apply_value_mutators(
        field: &str,
        mutator_list: &[crate::parser::Mutator],
        value: &AstValue,
    ) -> Result<AstValue> {
        match value {
            AstValue::List(items) => {
                let mut out = Vec::with_capacity(items.len());
                for item in items {
                    out.push(Self::apply_value_mutators_scalar(
                        field,
                        mutator_list,
                        item,
                    )?);
                }
                Ok(AstValue::List(out))
            }
            scalar => Self::apply_value_mutators_scalar(field, mutator_list, scalar),
        }
    }

    /// Apply the value mutator chain to ONE operand.
    ///
    /// A FAILURE HERE IS A QUERY ERROR, and that is the load-bearing decision.
    /// The operand is a literal from the query text, so `f eq '%FF' | urldecode`
    /// fails identically on record 1 and on record ten million. Neither
    /// disposition used on the FIELD side is right for it:
    ///
    /// * propagating it raw makes a query defect look like a crash, and gives
    ///   the caller no way to tell it from a genuine data error;
    /// * taking the field side's leniency and answering `false` would report
    ///   "no match" for a query that CANNOT match anything -- the same
    ///   zero-hits-is-not-an-error hazard as an unmapped field in OpenSearch.
    ///
    /// So it surfaces as [`TqlError::ValidationError`], the twin of Python's
    /// `TQLValidationError` (`src/tql/exceptions.py`), which is what Python
    /// raises here -- established by running it, not by reading it. Note that
    /// Python raises on the EVALUATION path, so a query over zero records
    /// returns empty rather than raising; evaluating per comparison reproduces
    /// that without a validate surface Rust does not have.
    ///
    /// The FIELD-mutator direction above keeps its leniency and is pinned by a
    /// control test: an unreadable RECORD value skips that record.
    fn apply_value_mutators_scalar(
        field: &str,
        mutator_list: &[crate::parser::Mutator],
        value: &AstValue,
    ) -> Result<AstValue> {
        #[cfg(test)]
        VALUE_MUTATOR_RESOLUTIONS.with(|n| n.set(n.get() + 1));

        let scratch = json!({});
        let mut current = ast_value_to_json(value);
        for mutator_spec in mutator_list {
            let params = build_mutator_params(mutator_spec);
            let mutator = mutators::create_mutator(&mutator_spec.name, params)?;
            current = mutator.apply(field, &scratch, &current).map_err(|e| {
                TqlError::ValidationError(format!(
                    "Value mutator '{}' cannot process the literal '{}' it is \
                     written against: {}",
                    mutator_spec.name,
                    render_literal(value),
                    e
                ))
            })?;

            // Enrichment mutators answer with a wrapper; compare against the
            // value they returned, exactly as the field side does.
            if let Some(return_value) = current.get("_tql_return_value") {
                current = return_value.clone();
            }
        }
        Ok(json_to_ast_value(&current))
    }

    /// Evaluate a comparison node
    fn evaluate_comparison(&self, comp: &ComparisonNode, record: &JsonValue) -> Result<bool> {
        // Existence operators. `is null` / `not exists` mean HAS NO VALUE --
        // absent OR explicitly null -- so the two pairs are exact complements:
        //
        //   exists       == is not null    field present AND not null
        //   not exists   == is null        field absent OR present-and-null
        //
        // This replaces a THREE-state model in which `is null` meant
        // "present and null" only and `not exists` meant "absent" only. That
        // model left `{}` satisfying neither `f is null` nor `f is not null`,
        // and `{"f": null}` satisfying neither `f exists` nor `f not exists` --
        // a record reported as neither null nor non-null, which is a hole
        // rather than a distinction a user can act on.
        //
        // OpenSearch cannot represent three states: it does not index JSON
        // nulls, so `must_not: {exists: {field}}` -- what BOTH DSL backends
        // already emit for `is null` and for `not exists` -- cannot tell an
        // absent field from a null one. The in-memory engines were giving a
        // different answer from the query they themselves translate to, on the
        // single most common document shape: the absent field.
        //
        // An EMPTY ARRAY is a present value and does NOT satisfy `is null`;
        // `exists` is deliberately unchanged and answers true for it, and the
        // complement property is what forces that. The full argument, and the
        // residual divergence from OpenSearch on `[]` and `[null]`, is recorded
        // once in `value_comparison.py` next to `_NEGATED_MATCH_ABSENT`.
        if comp.operator == "exists" || comp.operator == "not_exists" {
            // FIELD MUTATORS run FIRST, and the hint is applied to the MUTATED
            // value -- the order Python uses, and the order the non-existence
            // path in this same function already used.
            //
            // Rust used to skip the mutators here entirely and hint the RAW
            // value. Measured over
            // `[{b64num: "MTIz"}, {b64txt: "YWJj"}, {plainnum: "123"},
            //   {pad: "  123  "}, {empty: ""}, {nul: null}, {missing: -}]`:
            //
            // ```text
            //   f | b64decode::number exists      Rust ["plainnum","pad"]
            //                                     Py   ["b64num"]
            //   f | b64decode exists              Rust +["plainnum","pad"]
            //   f | b64decode::number not exists  Rust ["nul","missing"]
            //                                     Py   +["plainnum","pad"]
            // ```
            //
            // Not "0 versus 1" -- DISJOINT sets. Rust selected exactly the
            // records whose value is NOT valid base64 (it never decoded, so it
            // could not see the failure) and rejected the one that is.
            //
            // It also quietly weakened tests: `exists` on a mutated field passed
            // in Rust whether or not the mutator ran, so a test written that way
            // asserted nothing about the mutator.
            let found = field_accessor::get_field(record, &comp.field)?;
            let mutated: Option<JsonValue> = match (found, &comp.field_mutators) {
                (Some(value), Some(mutator_list)) => {
                    // A mutator that cannot process this value makes the field
                    // MISSING for an existence check, which is Python's rule
                    // (`evaluator.py`: `field_value = self._MISSING_FIELD` in the
                    // `exists`/`not_exists` arm of its except-branch). So
                    // `f | b64decode exists` is false for a value that is not
                    // base64 -- the decoded field does not exist -- and
                    // `f | b64decode not exists` is true for it.
                    self.apply_field_mutators(&comp.field, mutator_list, record, value)?
                }
                (Some(value), None) => Some(value.clone()),
                (None, _) => None,
            };
            let found = mutated.as_ref();
            // A type hint is tested BEFORE the existence question, as Python
            // tests it. The conversion is discarded because neither operator
            // reads a value; only whether the hint could READ it is observable,
            // which is the point.
            //
            // Read-instruction semantics apply here too, as ONE rule rather than
            // a table: `f::int exists` over `{"f": "Hello"}` answers false (f
            // does not exist AS AN INT) and `f::int not exists` answers false as
            // well. A skipped record matches nothing, positively or negatively.
            // The rejected alternative was to let `not exists` answer TRUE for an
            // unreadable value, which reads plausibly and makes a rule written
            // `f::int not exists` fire on every garbage record -- the same false
            // positive the negated comparators are guarded against below.
            //
            if let (Some(hint), Some(value)) = (&comp.type_hint, found) {
                match field_accessor::apply_type_hint(value, hint, &comp.field, &comp.operator) {
                    Ok(_) => {}
                    Err(TqlError::TypeHintCoercion(_)) => return Ok(false),
                    Err(e) => return Err(e),
                }
            }
            return match (comp.operator.as_str(), found) {
                ("exists", Some(value)) => Ok(!value.is_null()),
                ("exists", None) => Ok(false),
                (_, Some(value)) => Ok(value.is_null()),
                (_, None) => Ok(true),
            };
        }

        // Get the field value
        let field_value = match field_accessor::get_field(record, &comp.field)? {
            Some(value) => value,
            None => {
                // Field is absent. A negated comparator INCLUDES it — Lucene
                // semantics, matching what both DSL backends emit and what the
                // Python in-memory evaluator now does (tellaro-ui#587). Every
                // other comparator (positive ops, and `is`/`is_not`, which are
                // existence checks) excludes a missing field. `exists` /
                // `not_exists` are already handled above.
                //
                // (This arm previously returned false for ALL operators and its
                // comment claimed that was "Python behavior"; Python in fact
                // included a missing field for the negated string operators, so
                // the two in-memory engines were mirror-image wrong and only the
                // DSL backends were Lucene. That comment is how it survived.)
                // `is null` is the complement of `exists`, so an ABSENT
                // field satisfies it. Handled here rather than in
                // `negated_matches_absent`, which is the closed set of negated
                // COMPARATORS and deliberately excludes the existence checks.
                if comp.operator == "is" && matches!(comp.value, Some(AstValue::Null)) {
                    return Ok(true);
                }
                return Ok(comparator::negated_matches_absent(&comp.operator));
            }
        };

        // Apply field mutators if present
        // A mutator that cannot process THIS value answers `false` for THIS
        // record, rather than aborting the query. Python's rule, and the same
        // read-instruction rule the type hint below already follows -- see
        // `apply_field_mutators`. `false` for EVERY operator, negated ones
        // included: a skipped record matches nothing, positively or negatively,
        // so a failed mutator cannot become a hit for `ne` / `not_contains`.
        let field_value = if let Some(mutator_list) = &comp.field_mutators {
            match self.apply_field_mutators(&comp.field, mutator_list, record, field_value)? {
                Some(v) => v,
                None => return Ok(false),
            }
        } else {
            field_value.clone()
        };

        // Apply the type hint, if the query carried one.
        //
        // AFTER the mutators and BEFORE the comparison, which is where Python
        // applies it (`evaluator.py::_evaluate_comparison`) -- so `f | trim::int`
        // converts the TRIMMED value, and the comparator sees a number rather
        // than the string it would otherwise coerce on its own terms.
        //
        // That the comparator coerces anyway is exactly why this defect survived:
        // `f::int gt 100` over `{"f": "123"}` answered correctly with the hint
        // read by nothing, so every test of a hint over WELL-FORMED data passed.
        // The hint is only observable on a value it cannot convert.
        //
        // READ-INSTRUCTION SEMANTICS (product decision, 2026-09-03). A hint that
        // cannot read THIS record's value skips the record: the comparison
        // answers false and the query carries on. The error used to propagate,
        // so one unreadable value aborted the whole query -- `value::number > 75`
        // over `[{v: 80}, {v: "abc"}, {v: 90}]` returned nothing instead of 80
        // and 90, and a detection rule written that way stopped firing rather
        // than matching less. BREAKING for saved queries; see `apply_type_hint`.
        //
        // `Ok(false)` for EVERY operator, including the negated ones, is the
        // load-bearing part and the false-positive direction. `f::number != 75`
        // over "abc" is false, not true: a skipped record matches nothing,
        // positively or negatively. Falling through to the comparator, or
        // deferring to `negated_matches_absent` (which INCLUDES a record under a
        // negated operator, by Lucene semantics), would turn every unreadable
        // value into a hit for `ne`, `not_contains`, `not_in` and the rest.
        //
        // NOT skipped: an unknown hint NAME, which arrives as a plain
        // `TypeError` and stays fatal. The VARIANT is what separates the two --
        // matching on the message text is how these two engines drifted before.
        let field_value = match &comp.type_hint {
            Some(hint) => {
                match field_accessor::apply_type_hint(
                    &field_value,
                    hint,
                    &comp.field,
                    &comp.operator,
                ) {
                    Ok(converted) => converted,
                    Err(TqlError::TypeHintCoercion(_)) => return Ok(false),
                    Err(e) => return Err(e),
                }
            }
            None => field_value,
        };

        // Get the comparison value
        let compare_value = match &comp.value {
            Some(value) => value,
            None => {
                return Err(TqlError::ExecutionError(format!(
                    "Operator '{}' requires a comparison value",
                    comp.operator
                )));
            }
        };

        // Apply the VALUE mutators to the operand before comparing. Deferred
        // initialisation because the mutated operand is owned here while the
        // unmutated one is borrowed from the AST.
        let mutated_operand;
        let compare_value = match &comp.value_mutators {
            Some(mutator_list) if !mutator_list.is_empty() => {
                mutated_operand =
                    Self::apply_value_mutators(&comp.field, mutator_list, compare_value)?;
                &mutated_operand
            }
            _ => compare_value,
        };

        // Perform the comparison
        comparator::compare(&field_value, &comp.operator, compare_value)
    }

    /// Evaluate a logical operation (AND/OR)
    fn evaluate_logical_op(
        &self,
        logical: &LogicalOpNode,
        record: &JsonValue,
        depth: usize,
    ) -> Result<bool> {
        match logical.operator.as_str() {
            "and" | "&&" => {
                // Short-circuit: if left is false, return false -- but only
                // after checking that the branch being skipped is well-typed.
                // See `reject_ill_typed_collection_operands`: a refusal that
                // depends on which records happen to be in the corpus is not a
                // refusal.
                let left = self.evaluate_with_depth(&logical.left, record, depth)?;
                if !left {
                    reject_ill_typed_collection_operands(&logical.right)?;
                    return Ok(false);
                }
                self.evaluate_with_depth(&logical.right, record, depth)
            }
            "or" | "||" => {
                // Short-circuit: if left is true, return true -- same caveat.
                let left = self.evaluate_with_depth(&logical.left, record, depth)?;
                if left {
                    reject_ill_typed_collection_operands(&logical.right)?;
                    return Ok(true);
                }
                self.evaluate_with_depth(&logical.right, record, depth)
            }
            _ => Err(TqlError::OperatorError(format!(
                "Unknown logical operator: {}",
                logical.operator
            ))),
        }
    }

    /// Evaluate a unary operation (NOT)
    fn evaluate_unary_op(
        &self,
        unary: &UnaryOpNode,
        record: &JsonValue,
        depth: usize,
    ) -> Result<bool> {
        match unary.operator.as_str() {
            "not" | "!" => {
                let result = self.evaluate_with_depth(&unary.operand, record, depth)?;
                Ok(!result)
            }
            _ => Err(TqlError::OperatorError(format!(
                "Unknown unary operator: {}",
                unary.operator
            ))),
        }
    }

    /// Evaluate a collection operation (ANY/ALL/NONE)
    ///
    /// KNOWN GAP -- `CollectionOpNode::type_hint` is parsed, stored, and NOT read
    /// here, which is the same shape as the defect this module just repaired for
    /// `ComparisonNode`. It is left open on purpose, and the reason is not
    /// oversight:
    ///
    /// Rust's grammar accepts `any f::int eq 42` and PYTHON'S DOES NOT -- it
    /// raises `TQLSyntaxError: Expected operator after field 'any'` (measured
    /// 2026-09-03). So there is no reference behaviour to converge on. Making the
    /// hint element-wise here would give Rust a semantic no Python query can
    /// express and no cross-language fixture can pin, which trades a silent no-op
    /// for a silent divergence. Closing it properly means deciding the collection
    /// hint's meaning in BOTH engines, starting with Python's parser.
    ///
    /// `GeoExprNode::type_hint` and `NslookupExprNode::type_hint` are inert in
    /// both engines for the same class of reason and are recorded here rather
    /// than quietly fixed on one side.
    fn evaluate_collection_op(&self, coll: &CollectionOpNode, record: &JsonValue) -> Result<bool> {
        // A LIST operand to `any` / `none` is ill-typed and is refused, exactly
        // as the translator refuses it. Checked BEFORE the field is read: a
        // missing or null field returns `false` a few lines below, and deciding
        // the query's well-typedness after that would make the refusal depend on
        // the record. See `refuse_list_operand` for the measured divergence.
        if collection_operand_is_ill_typed(coll) {
            return Err(refuse_list_operand(&coll.field, &coll.operator));
        }

        // Get the field as a collection.
        //
        // A SCALAR is a one-element collection, not a non-collection. ECS stores
        // a single-valued field unwrapped, so `{"f": "admin"}` and
        // `{"f": ["admin"]}` are the same document to OpenSearch — and Rust's own
        // translator emits `{"term": {"f": "admin"}}` for `f any 'admin'`, which
        // matches the scalar. Refusing to iterate it made the evaluator
        // contradict the DSL it emits: `f any 'admin'` and `f all 'admin'` were
        // false on `{"f": "admin"}` (Python: true), and `f none 'admin'` was
        // false on `{"f": "user"}` (Python: true), which is a NONE clause that
        // stops excluding. The failing shape is the common one — a
        // `user.roles`/`process.args`-style field holding exactly one value.
        //
        // A MISSING field and a PRESENT NULL both stay `false` for every
        // collection operator, including the negated ones. That is what Python
        // answers, and it is deliberate: `f none 'admin'` asserts something
        // about the elements of `f`, and a field with no elements to inspect
        // does not satisfy it.
        let owned_scalar: Vec<JsonValue>;
        let array = match field_accessor::get_field(record, &coll.field)? {
            Some(JsonValue::Array(arr)) => arr,
            Some(JsonValue::Null) | None => return Ok(false),
            Some(scalar) => {
                owned_scalar = vec![scalar.clone()];
                &owned_scalar
            }
        };

        // Apply field mutators if present, transforming each array element.
        //
        // Routed through `apply_field_mutators` rather than re-inlining the
        // loop, so this arm gets the SAME construction-versus-application split
        // every other mutator call site has: an unknown mutator is a broken
        // query and propagates, a value the mutator cannot process is a data
        // error and answers `false` for THIS record.
        //
        // This arm re-inlined the loop and `?`-propagated both classes, so one
        // unprocessable value aborted the whole query. Measured over
        // `[{sc_ok: "YWJj"}, {sc_bad: B}, {ar_mix: ["YWJj", B]},
        //   {ar_allbad: [B, B]}, {ar_ok: ["YWJj"]}]` with `B` a string that is
        // not base64, before this change:
        //
        // ```text
        //   f | b64decode any 'abc'    Rust ERROR   Py ["sc_ok","ar_ok"]
        //   f | b64decode all 'abc'    Rust ERROR   Py ["sc_ok","ar_ok"]
        //   f | b64decode none 'zzz'   Rust ERROR   Py ["sc_ok","ar_ok"]
        // ```
        //
        // Every non-collection operator on the same corpus already agreed with
        // Python exactly; the collection operators were the only arm left, and
        // they did not answer LESS -- they stopped answering. For the agent's
        // detection engine that is a rule that stops running.
        //
        // A failure on ANY element answers `false` for the RECORD rather than
        // dropping that element from the array. Python does the same, and the
        // difference is observable: on `ar_mix`, dropping the bad element would
        // leave `["abc"]` and make `all 'abc'` TRUE, while Python answers false.
        // Negated collection operators are false too -- `none 'zzz'` excludes
        // `ar_mix` -- because a skipped record matches nothing, positively or
        // negatively.
        let transformed_array: Vec<JsonValue> = if let Some(mutator_list) = &coll.field_mutators {
            let mut result = Vec::with_capacity(array.len());
            for element in array {
                match self.apply_field_mutators(&coll.field, mutator_list, record, element)? {
                    Some(value) => result.push(value),
                    None => return Ok(false),
                }
            }
            result
        } else {
            array.to_vec()
        };

        // `f any ['a']` parses with a single-element LIST as its value, but the
        // question is about ONE element. Without this unwrap every element was
        // compared against the list itself, so `f any ['a']` on ["a","b"] was
        // FALSE in Rust and TRUE in Python — and `none` inverted that into a
        // false positive. `any` could not match anything at all.
        //
        // Python unwraps in the same place, and the OpenSearch translation on
        // both sides unwraps too, so this arm was the odd one out: the Rust
        // evaluator disagreed with the DSL Rust itself emits.
        //
        // A MULTI-element list never reaches here: it is refused at the top of
        // this function, as the translator refuses it. Only the one-element
        // unwrap is left, and it is the reason the guard tests `len() != 1`
        // rather than `is_list()`.
        let compare_value = match &coll.value {
            AstValue::List(items) if items.len() == 1 => &items[0],
            other => other,
        };

        // Evaluate the comparison for each element
        match coll.operator.as_str() {
            "any" => {
                // At least one element must match
                for element in &transformed_array {
                    if comparator::compare(element, &coll.comparison_operator, compare_value)? {
                        return Ok(true);
                    }
                }
                Ok(false)
            }
            "all" => {
                // All elements must match
                if transformed_array.is_empty() {
                    return Ok(false);
                }
                for element in &transformed_array {
                    if !comparator::compare(element, &coll.comparison_operator, compare_value)? {
                        return Ok(false);
                    }
                }
                Ok(true)
            }
            "none" => {
                // No elements must match
                for element in &transformed_array {
                    if comparator::compare(element, &coll.comparison_operator, compare_value)? {
                        return Ok(false);
                    }
                }
                Ok(true)
            }
            "not_any" => {
                // Negation of ANY
                for element in &transformed_array {
                    if comparator::compare(element, &coll.comparison_operator, compare_value)? {
                        return Ok(false);
                    }
                }
                Ok(true)
            }
            "not_all" => {
                // At least one element must NOT match.
                //
                // An EMPTY collection has no such element, so this is false —
                // NOT the vacuous-truth answer `!all([])`. Python answers false
                // here (`any(...) if field_value else False`), and the two
                // engines disagreeing on the empty case is how an exclusion
                // clause changes meaning between the agent and the backend.
                if transformed_array.is_empty() {
                    return Ok(false);
                }
                for element in &transformed_array {
                    if !comparator::compare(element, &coll.comparison_operator, compare_value)? {
                        return Ok(true);
                    }
                }
                Ok(false)
            }
            "not_none" => {
                // At least one element must match (same as ANY)
                for element in &transformed_array {
                    if comparator::compare(element, &coll.comparison_operator, compare_value)? {
                        return Ok(true);
                    }
                }
                Ok(false)
            }
            _ => Err(TqlError::OperatorError(format!(
                "Unknown collection operator: {}",
                coll.operator
            ))),
        }
    }

    /// Filter a list of records using a query
    ///
    /// # Arguments
    ///
    /// * `ast` - The parsed query AST
    /// * `records` - The list of records to filter
    ///
    /// # Returns
    ///
    /// A vector of references to matching records
    pub fn filter<'a>(
        &self,
        ast: &AstNode,
        records: &'a [JsonValue],
    ) -> Result<Vec<&'a JsonValue>> {
        // Hoist the RECORD-INDEPENDENT value-mutator chains out of the loop.
        // See `resolve_value_mutators`: one DNS query / MaxMind mmap for the
        // whole call rather than one per record. Guarded on `is_empty` so a
        // query over zero records still returns empty rather than raising.
        let resolved = if records.is_empty() {
            None
        } else {
            Self::resolve_value_mutators(ast)?
        };
        let ast = resolved.as_ref().unwrap_or(ast);

        let mut results = Vec::new();

        for record in records {
            if self.evaluate(ast, record)? {
                results.push(record);
            }
        }

        Ok(results)
    }

    /// Filter records and apply any field mutators (enrichment)
    ///
    /// This method is similar to `filter` but returns owned records with mutators applied.
    /// Used for enrichment queries where mutators modify the output.
    ///
    /// # Arguments
    ///
    /// * `ast` - The parsed query AST
    /// * `records` - The list of records to filter
    ///
    /// # Returns
    ///
    /// A vector of owned, potentially modified records
    pub fn filter_and_enrich(
        &self,
        ast: &AstNode,
        records: &[JsonValue],
    ) -> Result<Vec<JsonValue>> {
        // Hoist the RECORD-INDEPENDENT value-mutator chains out of the loop.
        // See `resolve_value_mutators`: one DNS query / MaxMind mmap for the
        // whole call rather than one per record. Guarded on `is_empty` so a
        // query over zero records still returns empty rather than raising.
        let resolved = if records.is_empty() {
            None
        } else {
            Self::resolve_value_mutators(ast)?
        };
        let ast = resolved.as_ref().unwrap_or(ast);

        let mut results = Vec::new();
        let mut mutator_cache = std::collections::HashMap::new();
        // Hoisted for the same reason as `mutator_cache`; see `apply_enrichment`.
        let boolean_predicates: std::collections::HashSet<&'static str> =
            mutators::boolean_predicate_names().into_iter().collect();

        // Hoist AST traversal out of the per-record loop — the mutator specs are
        // constant for a given query and don't need to be re-extracted per record.
        let mutators_info = self.extract_field_mutators(ast);

        for record in records {
            if self.evaluate(ast, record)? {
                let enriched_record = self.apply_enrichment(
                    &mutators_info,
                    record,
                    &mut mutator_cache,
                    &boolean_predicates,
                )?;
                results.push(enriched_record);
            }
        }

        Ok(results)
    }

    /// Apply field mutators to a record (enrichment).
    ///
    /// `mutators_info` is extracted once from the AST by the caller and reused across
    /// all records to avoid redundant AST traversals.
    fn apply_enrichment(
        &self,
        mutators_info: &Option<Vec<(String, Vec<crate::parser::Mutator>)>>,
        record: &JsonValue,
        mutator_cache: &mut std::collections::HashMap<String, Box<dyn mutators::Mutator>>,
        boolean_predicates: &std::collections::HashSet<&'static str>,
    ) -> Result<JsonValue> {
        let mut enriched = record.clone();

        if let Some(mutators_info) = mutators_info {
            for (field, mutator_list) in mutators_info {
                // A PREDICATE does not project. `ip | is_loopback` answers a
                // boolean ABOUT the address; it is not a request to REPLACE the
                // address with that boolean. Rust wrote it back anyway, so
                // `query_enriched` returned `{"ip": true}` where Python returns
                // `{"ip": "127.0.0.1"}` -- the value the analyst filtered on,
                // destroyed by the filter. The MATCH SET agreed, which is why no
                // fixture caught it: the divergence is in the returned payload,
                // not in which records come back. Measured 2026-09-04 for all
                // five IP predicates, for the negated form, and for the chained
                // `ip | is_loopback and name | lowercase`.
                //
                // Python fixed the identical shape on its side via
                // `TYPE_CHANGING_MUTATORS` in `tql.mutator_classification`; see
                // the comment at that definition.
                //
                // THIS TEST IS THE LAST MUTATOR ONLY, AND THAT IS A DIVERGENCE,
                // not a shared rule. Do not restate it as one -- the sentence
                // that stood here claimed "both engines test the LAST mutator"
                // and named `ip | is_loopback | lowercase` as the agreeing
                // example, when it is the single case that does NOT agree.
                // Measured 2026-09-04, both engines: `ip | is_loopback` and
                // `ip | lowercase | is_loopback` return the address on both
                // sides; `ip | is_loopback | lowercase` returns `true`/`false`
                // here and the address in Python, because Python never applies
                // the predicate to the value it returns at all. Same match set
                // either way, so only the payload differs. Pinned by
                // `tests/predicate_does_not_project.rs`
                // (`a_transform_after_a_predicate_still_projects_and_diverges_from_python`)
                // and recorded under "Known gaps" in CHANGELOG.md. Which rule is
                // right is a decision, not a comment fix.
                //
                // DERIVED, not enumerated: this asks the mutator through
                // [`Mutator::returns_boolean`], the same definition the parser
                // uses to decide that `field | <predicate>` is a filter. A
                // sixth predicate overrides that one method and needs no edit
                // here. A hand-written list of the five would be the third copy
                // of a fact that has already shipped wrong twice for exactly
                // that reason.
                //
                // A TRANSFORMING mutator is untouched and must stay that way:
                // `name | lowercase` replacing `name` with its lowercase form is
                // the whole purpose of a projection.
                //
                // Asked against a set built ONCE per call, not per record.
                // `mutators::returns_boolean` answers by CONSTRUCTING the
                // mutator, which is the right way to derive the fact and the
                // wrong thing to do inside the record loop: for a `geoip` chain
                // that construction mmaps the MaxMind database, and it ran once
                // per record per field -- the exact cost `mutator_cache`, two
                // lines up in the caller, exists to avoid.
                // `mutators::boolean_predicate_names` derives the same set from
                // the same one definition, so nothing here enumerates names.
                if mutator_list
                    .last()
                    .is_some_and(|m| boolean_predicates.contains(m.name.to_lowercase().as_str()))
                {
                    continue;
                }

                // Get the field value
                if let Some(field_value) = field_accessor::get_field(record, field)? {
                    // Apply mutators in sequence
                    let mut current_value = field_value.clone();
                    let mut unprocessable = false;
                    for mutator_spec in mutator_list {
                        // Cache key includes mutator name, positional args, and named args
                        let cache_key = format!(
                            "{}:{:?}:{:?}",
                            mutator_spec.name, mutator_spec.args, mutator_spec.named_args
                        );

                        // Get or create mutator (cached for performance)
                        if !mutator_cache.contains_key(&cache_key) {
                            let params = build_mutator_params(mutator_spec);
                            let mutator = mutators::create_mutator(&mutator_spec.name, params)?;
                            mutator_cache.insert(cache_key.clone(), mutator);
                        }

                        let mutator = mutator_cache.get(&cache_key).unwrap();
                        // Same construction-versus-application split
                        // `apply_field_mutators` documents: an unknown mutator
                        // propagated above, a value this mutator cannot process
                        // leaves the field AS IT WAS on this record.
                        //
                        // This is the enrichment half of the same defect the
                        // collection operators carried, and it is the one that
                        // reaches production paths -- `lib.rs::query_enriched`,
                        // `file_ops.rs` and `opensearch/post_processor.rs` all
                        // arrive here. The record is already known to MATCH
                        // (`filter_and_enrich` only enriches what `evaluate`
                        // returned true for), so the failing value is one the
                        // query deliberately kept: measured on
                        // `[{ok: "YWJj"}, {bad: <not base64>}]` with
                        // `f | b64decode contains 'abc' OR id eq 'bad'`, the
                        // FILTER answered `["ok", "bad"]` in both engines while
                        // Rust's enrichment then aborted the whole query with
                        // `Base64 decode error` and Python returned both
                        // records with `f` untouched.
                        //
                        // Leaving the field unchanged rather than deleting it is
                        // what Python's shape is: the record survives carrying
                        // its ORIGINAL value.
                        current_value = match mutator.apply(field, record, &current_value) {
                            Ok(v) => v,
                            Err(_) => {
                                unprocessable = true;
                                break;
                            }
                        };
                    }

                    if unprocessable {
                        continue;
                    }

                    // Check if this is an enrichment result with special structure
                    if let Some(enrichment_data) = current_value.get("_tql_enrichment") {
                        // Handle enrichment mutators (nslookup, geoip, etc.)
                        self.apply_enrichment_data(&mut enriched, enrichment_data)?;
                    } else {
                        // Regular mutator - update the field directly
                        if let Some(obj) = enriched.as_object_mut() {
                            obj.insert(field.clone(), current_value);
                        }
                    }
                }
            }
        }

        Ok(enriched)
    }

    /// Apply enrichment data from an enrichment mutator (nslookup, geoip, etc.)
    fn apply_enrichment_data(
        &self,
        record: &mut JsonValue,
        enrichment_data: &JsonValue,
    ) -> Result<()> {
        let enrichment_type = enrichment_data
            .get("type")
            .and_then(|v| v.as_str())
            .unwrap_or("");

        match enrichment_type {
            "dns" => {
                // DNS enrichment from nslookup
                let domain_field = enrichment_data
                    .get("domain_field")
                    .and_then(|v| v.as_str())
                    .unwrap_or("domain");
                let dns_field = enrichment_data
                    .get("dns_field")
                    .and_then(|v| v.as_str())
                    .unwrap_or("dns");

                // Set domain field
                if let Some(domain) = enrichment_data.get("domain") {
                    if !domain.is_null() {
                        field_accessor::set_field(record, domain_field, domain.clone())?;
                    }
                }

                // Set dns field (full ECS data)
                if let Some(dns) = enrichment_data.get("dns") {
                    field_accessor::set_field(record, dns_field, dns.clone())?;
                }
            }
            "geo" => {
                // Geo enrichment from geoip_lookup
                let geo_field = enrichment_data
                    .get("geo_field")
                    .and_then(|v| v.as_str())
                    .unwrap_or("geo");

                if let Some(geo_data) = enrichment_data.get("geo") {
                    field_accessor::set_field(record, geo_field, geo_data.clone())?;
                }

                // Also set AS data if present
                if let Some(as_field) = enrichment_data.get("as_field").and_then(|v| v.as_str()) {
                    if let Some(as_data) = enrichment_data.get("as") {
                        field_accessor::set_field(record, as_field, as_data.clone())?;
                    }
                }
            }
            _ => {
                // Unknown enrichment type - log and continue
            }
        }

        Ok(())
    }

    /// Extract field names and their mutators from AST
    fn extract_field_mutators(
        &self,
        ast: &AstNode,
    ) -> Option<Vec<(String, Vec<crate::parser::Mutator>)>> {
        use std::collections::HashMap;

        let mut mutators_map: HashMap<String, Vec<crate::parser::Mutator>> = HashMap::new();

        self.collect_field_mutators(ast, &mut mutators_map);

        if mutators_map.is_empty() {
            None
        } else {
            Some(mutators_map.into_iter().collect())
        }
    }

    /// Recursively collect field mutators from AST
    #[allow(clippy::only_used_in_recursion)]
    fn collect_field_mutators(
        &self,
        ast: &AstNode,
        mutators_map: &mut std::collections::HashMap<String, Vec<crate::parser::Mutator>>,
    ) {
        match ast {
            AstNode::Comparison(comp) => {
                if let Some(mutator_list) = &comp.field_mutators {
                    if !mutator_list.is_empty() {
                        mutators_map.insert(comp.field.clone(), mutator_list.clone());
                    }
                }
            }
            AstNode::LogicalOp(logical) => {
                self.collect_field_mutators(&logical.left, mutators_map);
                self.collect_field_mutators(&logical.right, mutators_map);
            }
            AstNode::UnaryOp(unary) => {
                self.collect_field_mutators(&unary.operand, mutators_map);
            }
            AstNode::CollectionOp(coll) => {
                if let Some(mutator_list) = &coll.field_mutators {
                    if !mutator_list.is_empty() {
                        mutators_map.insert(coll.field.clone(), mutator_list.clone());
                    }
                }
            }
            _ => {}
        }
    }

    /// Count the number of records matching a query
    ///
    /// # Arguments
    ///
    /// * `ast` - The parsed query AST
    /// * `records` - The list of records to count
    ///
    /// # Returns
    ///
    /// The number of matching records
    /// Resolve every value-mutator chain in `ast` ONCE, returning an AST whose
    /// comparison operands are already mutated and whose `value_mutators` are
    /// cleared. `None` means the AST carries none and the caller should keep
    /// the one it has -- the overwhelmingly common case, and the reason this
    /// does not clone unconditionally.
    ///
    /// WHY THIS EXISTS -- it is a denial-of-service fix, not a micro-optimisation.
    /// The value operand is RECORD-INDEPENDENT by construction:
    /// [`Self::apply_value_mutators_scalar`] hands the mutator a scratch `{}`
    /// rather than the record, precisely so a value mutator cannot read one. So
    /// a chain resolved inside the per-record loop computes the identical answer
    /// N times, and two of the mutators reachable there are not cheap and not
    /// local:
    ///
    /// * `| nslookup` performs a real DNS resolution per call
    ///   (`mutators::dns::NSLookupMutator::apply`);
    /// * `| geoip` mmaps the MaxMind database and builds a `Reader` per
    ///   CONSTRUCTION (`mutators::geoip::GeoIPMutator::new`), and the value side
    ///   constructs one per call.
    ///
    /// Both are reachable from user-controlled query text with no cache and no
    /// cap, so `f eq 'example.com' | nslookup` over a large index was an
    /// outbound DNS amplifier. Rust did not apply value mutators at all before
    /// `078fb50`, so the surface is new on this branch.
    ///
    /// TWO SEMANTICS THIS DELIBERATELY CHANGES, both toward the disposition the
    /// branch already chose elsewhere:
    ///
    /// 1. A refusal no longer depends on WHICH records are in the corpus.
    ///    `a eq 1 or f eq '%FF' | urldecode` used to raise only for a corpus
    ///    containing a record with `a != 1`, because `or` short-circuits. That is
    ///    the exact hazard `reject_ill_typed_collection_operands` was written to
    ///    remove for collection operands: "a query that is well-typed or not
    ///    depending on the data is exactly the silent shape this refusal exists
    ///    to remove."
    /// 2. A query over ZERO records still returns empty rather than raising --
    ///    the callers guard on `records.is_empty()`. That is not an accident of
    ///    the old shape; it is the Python parity property
    ///    [`Self::apply_value_mutators_scalar`] documents, and hoisting
    ///    unconditionally would have broken it.
    pub(crate) fn resolve_value_mutators(ast: &AstNode) -> Result<Option<AstNode>> {
        if !Self::carries_value_mutators(ast) {
            return Ok(None);
        }
        let mut resolved = ast.clone();
        Self::resolve_value_mutators_in_place(&mut resolved)?;
        Ok(Some(resolved))
    }

    /// Does any comparison in this subtree carry a non-empty value-mutator chain?
    fn carries_value_mutators(ast: &AstNode) -> bool {
        match ast {
            AstNode::Comparison(comp) => comp
                .value_mutators
                .as_ref()
                .is_some_and(|list| !list.is_empty()),
            AstNode::LogicalOp(logical) => {
                Self::carries_value_mutators(&logical.left)
                    || Self::carries_value_mutators(&logical.right)
            }
            AstNode::UnaryOp(unary) => Self::carries_value_mutators(&unary.operand),
            AstNode::QueryWithStats(qws) => Self::carries_value_mutators(&qws.filter),
            // Only `ComparisonNode` carries `value_mutators`; `CollectionOpNode`
            // has `field_mutators` only, and the remaining variants hold no
            // operand a value mutator could attach to.
            _ => false,
        }
    }

    fn resolve_value_mutators_in_place(ast: &mut AstNode) -> Result<()> {
        match ast {
            AstNode::Comparison(comp) => {
                let has_chain = comp
                    .value_mutators
                    .as_ref()
                    .is_some_and(|list| !list.is_empty());
                // A missing operand is `evaluate_comparison`'s error to raise,
                // with its own message; leave the chain in place so it does.
                if !has_chain || comp.value.is_none() {
                    return Ok(());
                }
                let chain = comp.value_mutators.take().unwrap_or_default();
                let value = comp.value.as_ref().expect("checked is_none above");
                comp.value = Some(Self::apply_value_mutators(&comp.field, &chain, value)?);
                Ok(())
            }
            AstNode::LogicalOp(logical) => {
                Self::resolve_value_mutators_in_place(&mut logical.left)?;
                Self::resolve_value_mutators_in_place(&mut logical.right)
            }
            AstNode::UnaryOp(unary) => Self::resolve_value_mutators_in_place(&mut unary.operand),
            AstNode::QueryWithStats(qws) => Self::resolve_value_mutators_in_place(&mut qws.filter),
            _ => Ok(()),
        }
    }

    pub fn count(&self, ast: &AstNode, records: &[JsonValue]) -> Result<usize> {
        // Hoist the RECORD-INDEPENDENT value-mutator chains out of the loop.
        // See `resolve_value_mutators`: one DNS query / MaxMind mmap for the
        // whole call rather than one per record. Guarded on `is_empty` so a
        // query over zero records still returns empty rather than raising.
        let resolved = if records.is_empty() {
            None
        } else {
            Self::resolve_value_mutators(ast)?
        };
        let ast = resolved.as_ref().unwrap_or(ast);

        let mut count = 0;

        for record in records {
            if self.evaluate(ast, record)? {
                count += 1;
            }
        }

        Ok(count)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::parser::TqlParser;
    use serde_json::json;

    /// Resolve a chain over N records and report how many times it actually ran.
    fn resolutions_over(query: &str, records: &[JsonValue]) -> (usize, usize) {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();
        let ast = parser.parse(query).unwrap();
        VALUE_MUTATOR_RESOLUTIONS.with(|n| n.set(0));
        let hits = evaluator.filter(&ast, records).unwrap();
        (hits.len(), VALUE_MUTATOR_RESOLUTIONS.with(|n| n.get()))
    }

    /// A record-independent value-mutator chain is resolved ONCE per query, not
    /// once per record.
    ///
    /// Asserted as a COUNT rather than a duration on purpose: `| nslookup` is one
    /// real DNS query per resolution and `| geoip` one MaxMind mmap, so the
    /// invariant that matters is "how many times", and a wall-clock assertion
    /// both flakes and would still pass over a chain resolved twice. Before the
    /// hoist this measured 50 resolutions for 50 records.
    #[test]
    fn value_mutator_chain_resolves_once_per_query_not_once_per_record() {
        let records: Vec<JsonValue> = (0..50).map(|_| json!({"f": "abc"})).collect();
        let (hits, resolutions) = resolutions_over("f eq 'ABC' | lowercase", &records);

        assert_eq!(hits, 50, "every record still matches");
        assert_eq!(
            resolutions,
            1,
            "the operand does not depend on the record, so resolving it {} times \
             is {} redundant DNS queries / MaxMind mmaps for a `| nslookup` or \
             `| geoip` chain",
            resolutions,
            resolutions.saturating_sub(1)
        );
    }

    /// The count does not grow with the corpus. Pins the SHAPE of the fix rather
    /// than the single number above: a per-record resolution passes the N=1 case.
    #[test]
    fn value_mutator_resolution_count_is_independent_of_corpus_size() {
        let one: Vec<JsonValue> = vec![json!({"f": "abc"})];
        let many: Vec<JsonValue> = (0..500).map(|_| json!({"f": "abc"})).collect();

        let (_, small) = resolutions_over("f eq 'ABC' | lowercase", &one);
        let (_, large) = resolutions_over("f eq 'ABC' | lowercase", &many);
        assert_eq!(
            small, large,
            "1 record cost {small}, 500 records cost {large}"
        );
    }

    /// Each operand of a chain over a LIST literal is still resolved -- once each,
    /// for the whole query rather than per record.
    #[test]
    fn list_operand_resolves_once_per_element_not_per_record() {
        let records: Vec<JsonValue> = (0..20).map(|_| json!({"f": "abc"})).collect();
        let (hits, resolutions) = resolutions_over("f in ['ABC', 'XYZ'] | lowercase", &records);

        assert_eq!(hits, 20);
        assert_eq!(resolutions, 2, "two elements, twenty records");
    }

    /// A query over ZERO records must NOT raise for a value mutator that cannot
    /// process its operand.
    ///
    /// This is the Python parity property `apply_value_mutators_scalar` documents
    /// -- Python raises on the EVALUATION path, so an empty corpus returns empty.
    /// Hoisting the resolution unconditionally out of the loop would have broken
    /// it, which is why the callers guard on `records.is_empty()`.
    #[test]
    fn zero_records_does_not_raise_a_value_mutator_error() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();
        let ast = parser.parse("f eq 'abc' | nosuchmutator").unwrap();

        assert_eq!(evaluator.filter(&ast, &[]).unwrap().len(), 0);
        assert_eq!(evaluator.count(&ast, &[]).unwrap(), 0);
        assert_eq!(evaluator.filter_and_enrich(&ast, &[]).unwrap().len(), 0);
        assert!(evaluator.filter(&ast, &[json!({"f": "abc"})]).is_err());
    }

    /// The set hoisted out of `apply_enrichment` answers exactly what
    /// `mutators::returns_boolean` answers, for every name the dispatch accepts.
    ///
    /// The guard used to call `returns_boolean` per record per field, and that
    /// function answers by CONSTRUCTING the mutator — for `geoip` that mmaps the
    /// MaxMind database, which is the cost `mutator_cache` two lines above the
    /// loop exists to avoid. Hoisting it trades a derived call for a derived
    /// SET, and this pins that the trade changed nothing: `create_mutator`
    /// accepts aliases (`geo`, `geoip_lookup`), and an alias present in the
    /// dispatch but absent from `MUTATOR_NAMES` would make the set narrower than
    /// the function. `mutators::tests` already binds `MUTATOR_NAMES` to the
    /// dispatch arms; this binds the derived subset to the predicate.
    #[test]
    fn the_hoisted_predicate_set_matches_the_derived_predicate() {
        let hoisted: std::collections::HashSet<&'static str> =
            mutators::boolean_predicate_names().into_iter().collect();

        for name in mutators::mutator_names() {
            assert_eq!(
                hoisted.contains(name),
                mutators::returns_boolean(name),
                "`{name}` is classified differently by the hoisted set and by \
                 `returns_boolean`"
            );
        }

        // An unknown name is not a predicate under either spelling -- the
        // parser must not turn a typo into a filter.
        assert!(!hoisted.contains("nosuchmutator"));
        assert!(!mutators::returns_boolean("nosuchmutator"));
    }

    /// The hoisted guard is CASE-INSENSITIVE, as `create_mutator` is.
    ///
    /// `returns_boolean` inherited that for free by going through
    /// `create_mutator`, which lowercases; a set lookup does not, and dropping
    /// the `to_lowercase` would make `ip | IS_LOOPBACK` project again.
    #[test]
    fn the_hoisted_guard_is_case_insensitive() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();
        let ast = parser.parse("ip | IS_LOOPBACK").unwrap();
        let records = vec![json!({"ip": "127.0.0.1"})];

        let enriched = evaluator.filter_and_enrich(&ast, &records).unwrap();
        assert_eq!(enriched.len(), 1);
        assert_eq!(
            enriched[0]["ip"],
            json!("127.0.0.1"),
            "an upper-cased predicate is still a predicate and must not project"
        );
    }

    /// A refusal must not depend on WHICH records are in the corpus.
    ///
    /// `or` short-circuits, so before the hoist this query raised only for a
    /// corpus containing a record the left branch does not satisfy. Same
    /// reasoning as `reject_ill_typed_collection_operands`.
    #[test]
    fn a_short_circuited_branch_still_refuses_a_bad_value_mutator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();
        let ast = parser
            .parse("a eq 1 or f eq 'abc' | nosuchmutator")
            .unwrap();

        // Every record satisfies the LEFT branch, so the right is never reached
        // by the per-record walk.
        let records = vec![json!({"a": 1}), json!({"a": 1})];
        assert!(
            evaluator.filter(&ast, &records).is_err(),
            "an unknown mutator in a skipped branch is still a broken query"
        );
    }

    #[test]
    fn test_evaluate_simple_comparison() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("age > 25").unwrap();
        let record = json!({"age": 30, "name": "John"});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_equality() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("name eq 'John'").unwrap();
        let record = json!({"age": 30, "name": "John"});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_and_operator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("age > 25 AND name eq 'John'").unwrap();
        let record = json!({"age": 30, "name": "John"});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_or_operator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("age > 40 OR name eq 'John'").unwrap();
        let record = json!({"age": 30, "name": "John"});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_not_operator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("NOT age > 40").unwrap();
        let record = json!({"age": 30, "name": "John"});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_between_list_syntax() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // List syntax: field between [min, max]
        let ast = parser.parse("age between [20, 40]").unwrap();
        let record = json!({"age": 30});
        assert!(evaluator.evaluate(&ast, &record).unwrap());

        let record = json!({"age": 50});
        assert!(!evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_between_natural_syntax() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // Natural syntax: field between val1 and val2
        // Parser normalizes this to a two-element list
        let ast = parser.parse("age between 20 and 40").unwrap();
        let record = json!({"age": 30});
        assert!(evaluator.evaluate(&ast, &record).unwrap());

        let record = json!({"age": 50});
        assert!(!evaluator.evaluate(&ast, &record).unwrap());

        // Boundary values should be inclusive
        let record = json!({"age": 20});
        assert!(evaluator.evaluate(&ast, &record).unwrap());

        let record = json!({"age": 40});
        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_not_between() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("age not between 20 and 40").unwrap();
        let record = json!({"age": 50});
        assert!(evaluator.evaluate(&ast, &record).unwrap());

        let record = json!({"age": 30});
        assert!(!evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_nested_fields() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("user.profile.age > 25").unwrap();
        let record = json!({
            "user": {
                "profile": {
                    "age": 30
                }
            }
        });

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_exists() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("name exists").unwrap();
        let record = json!({"name": "John", "age": 30});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_not_exists() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("email not_exists").unwrap();
        let record = json!({"name": "John", "age": 30});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_contains() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("message contains 'error'").unwrap();
        let record = json!({"message": "An error occurred"});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_any_operator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("ANY tags eq 'urgent'").unwrap();
        let record = json!({"tags": ["bug", "urgent", "security"]});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_all_operator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("ALL scores >= 80").unwrap();
        let record = json!({"scores": [85, 90, 95]});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_evaluate_none_operator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("NONE tags eq 'wontfix'").unwrap();
        let record = json!({"tags": ["bug", "urgent", "security"]});

        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    fn test_filter_records() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("age > 25").unwrap();
        let records = vec![
            json!({"name": "John", "age": 30}),
            json!({"name": "Jane", "age": 20}),
            json!({"name": "Bob", "age": 35}),
        ];

        let results = evaluator.filter(&ast, &records).unwrap();
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_count_matching_records() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser.parse("age > 25").unwrap();
        let records = vec![
            json!({"name": "John", "age": 30}),
            json!({"name": "Jane", "age": 20}),
            json!({"name": "Bob", "age": 35}),
        ];

        let count = evaluator.count(&ast, &records).unwrap();
        assert_eq!(count, 2);
    }

    #[test]
    fn test_complex_query() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        let ast = parser
            .parse("(age > 25 AND status eq 'active') OR role eq 'admin'")
            .unwrap();

        let record1 = json!({"age": 30, "status": "active", "role": "user"});
        let record2 = json!({"age": 20, "status": "active", "role": "admin"});
        let record3 = json!({"age": 20, "status": "inactive", "role": "user"});

        assert!(evaluator.evaluate(&ast, &record1).unwrap());
        assert!(evaluator.evaluate(&ast, &record2).unwrap());
        assert!(!evaluator.evaluate(&ast, &record3).unwrap());
    }

    #[test]
    fn test_evaluate_with_mutator() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // Test lowercase mutator
        let ast = parser.parse("name | lowercase eq 'john'").unwrap();
        let record = json!({"name": "JOHN", "age": 30});
        assert!(evaluator.evaluate(&ast, &record).unwrap());

        // Test uppercase mutator
        let ast = parser
            .parse("email | uppercase contains 'EXAMPLE'")
            .unwrap();
        let record = json!({"email": "user@example.com"});
        assert!(evaluator.evaluate(&ast, &record).unwrap());

        // Test chained mutators
        let ast = parser
            .parse("message | trim | lowercase eq 'hello'")
            .unwrap();
        let record = json!({"message": "  HELLO  "});
        assert!(evaluator.evaluate(&ast, &record).unwrap());
    }

    #[test]
    #[cfg(feature = "integration-tests")]
    fn test_nslookup_enrichment() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // Test nslookup enrichment with reverse DNS on Google's public DNS
        let ast = parser.parse("destination.ip | nslookup").unwrap();
        let records = vec![json!({"destination": {"ip": "8.8.8.8"}})];

        // Use filter_and_enrich to apply enrichment
        let enriched = evaluator.filter_and_enrich(&ast, &records).unwrap();

        assert_eq!(enriched.len(), 1);
        let record = &enriched[0];

        // Check that destination.domain was added
        let destination = record.get("destination").expect("Should have destination");
        assert!(
            destination.get("domain").is_some(),
            "Should have destination.domain"
        );

        // Check that destination.dns was added with ECS structure
        let dns = destination.get("dns").expect("Should have destination.dns");
        assert!(dns.get("question").is_some(), "DNS should have question");
        assert!(dns.get("answers").is_some(), "DNS should have answers");
        assert!(
            dns.get("response_code").is_some(),
            "DNS should have response_code"
        );
    }

    #[test]
    #[cfg(feature = "integration-tests")]
    fn test_nslookup_comparison() {
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // Test nslookup with comparison - should match Google's DNS
        let ast = parser
            .parse("destination.ip | nslookup contains 'google'")
            .unwrap();

        // 8.8.8.8 resolves to dns.google
        let record = json!({"destination": {"ip": "8.8.8.8"}});
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(result, "8.8.8.8 should resolve to dns.google");

        // Private IP likely won't have reverse DNS containing 'google'
        let record = json!({"destination": {"ip": "192.168.1.1"}});
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(!result, "Private IP should not resolve to google");
    }

    #[test]
    fn test_nslookup_expr_evaluation_enrichment_only() {
        // Test that NslookupExpr without conditions returns true when field exists
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // Parse a query that creates a NslookupExpr node
        let ast = parser.parse("destination.ip | nslookup").unwrap();

        // Record with the field - should return true
        let record = json!({"destination": {"ip": "8.8.8.8"}});
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(
            result,
            "NslookupExpr with existing field should return true"
        );

        // Record without the field - should return false
        let record = json!({"source": {"ip": "8.8.8.8"}});
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(
            !result,
            "NslookupExpr with missing field should return false"
        );
    }

    #[test]
    fn test_geo_expr_evaluation_enrichment_only() {
        // Test that GeoExpr without conditions returns true when field exists
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // Parse a query that creates a GeoExpr node
        let ast = parser.parse("source.ip | geoip").unwrap();

        // Record with the field - should return true
        let record = json!({"source": {"ip": "8.8.8.8"}});
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(result, "GeoExpr with existing field should return true");

        // Record without the field - should return false
        let record = json!({"destination": {"ip": "8.8.8.8"}});
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(!result, "GeoExpr with missing field should return false");
    }

    #[test]
    fn test_compound_query_with_nslookup_expr() {
        // Test a compound query that includes NslookupExpr
        let parser = TqlParser::new();
        let evaluator = TqlEvaluator::new();

        // Query similar to the detection rule that was failing
        let ast = parser
            .parse("event.code = 3 AND destination.ip | is_global eq true AND destination.ip | nslookup")
            .unwrap();

        // Record that matches all conditions
        let record = json!({
            "event": {"code": 3},
            "destination": {"ip": "8.8.8.8"}
        });
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(
            result,
            "Compound query should match when all conditions are true"
        );

        // Record that doesn't match event.code
        let record = json!({
            "event": {"code": 4},
            "destination": {"ip": "8.8.8.8"}
        });
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(
            !result,
            "Compound query should fail when event.code doesn't match"
        );

        // Record with private IP (is_global eq false)
        let record = json!({
            "event": {"code": 3},
            "destination": {"ip": "192.168.1.1"}
        });
        let result = evaluator.evaluate(&ast, &record).unwrap();
        assert!(
            !result,
            "Compound query should fail when is_global is false"
        );
    }
}