tellaro-query-language 3.0.1

//! Value comparator for TQL query evaluation.
//!
//! Supports all TQL comparison operators with type coercion and flexible matching.

use crate::error::{Result, TqlError};
use crate::parser::Value as AstValue;
use once_cell::sync::Lazy;
use regex::Regex;
use serde_json::Value as JsonValue;
use std::collections::HashMap;
use std::net::IpAddr;
use std::str::FromStr;
use std::sync::Mutex;

/// Cache for compiled regex patterns to avoid recompilation on every invocation.
/// Bounded to MAX_REGEX_CACHE_SIZE entries; when full the cache is cleared to
/// prevent unbounded memory growth from adversarial or highly-varied patterns.
static REGEX_CACHE: Lazy<Mutex<HashMap<String, Regex>>> = Lazy::new(|| Mutex::new(HashMap::new()));

/// Maximum number of compiled regex patterns to cache. When the cache exceeds
/// this limit, it is cleared to prevent unbounded memory consumption in
/// long-running services processing user-supplied queries.
const MAX_REGEX_CACHE_SIZE: usize = 1024;

/// Maximum regex pattern length TQL will compile, in characters.
///
/// This is a *language-level* limit, not an implementation detail of any one
/// runtime. Rust's `regex` crate has linear-time matching guarantees, so it does
/// not need the ReDoS bound that Python's backtracking `re` module does -- but a
/// query language whose acceptance depends on which runtime happens to evaluate
/// it is broken, and this evaluator shares a rule corpus with the Python one.
/// Before tql#185 a 1500-character pattern ran here and silently matched nothing
/// in Python; the fix makes both refuse it, loudly and at the same threshold.
///
/// Mirrored by `MAX_REGEX_PATTERN_LENGTH` in `src/tql/regex_guard.py` and
/// `js/src/validator.ts`. Changing it in one place only re-creates the
/// divergence.
pub const MAX_REGEX_PATTERN_LENGTH: usize = 1000;

/// Negated comparators that MATCH a record whose field is null or absent.
///
/// Lucene / OpenSearch `must_not` semantics: a `must_not` clause includes
/// documents that lack the field, because OpenSearch does not index JSON nulls.
/// Both TQL DSL backends already emit exactly this; this predicate brings the
/// Rust in-memory evaluator into line with them and with the Python in-memory
/// evaluator (tellaro-ui#587).
///
/// The mirror of Python's `ValueComparator._NEGATED_MATCH_ABSENT`. It is the
/// closed set of negated *comparators* — existence checks (`exists`,
/// `not_exists`, `is`, `is_not`) are deliberately excluded, because `is not
/// null` is `exists`, which a missing field must not satisfy.
pub fn negated_matches_absent(operator: &str) -> bool {
    matches!(
        operator,
        "not_in"
            | "not_in_cs"
            | "not_contains"
            | "not_contains_cs"
            | "not_startswith"
            | "not_startswith_cs"
            | "not_endswith"
            | "not_endswith_cs"
            | "not_regexp"
            | "not_regex"
            | "not_matches"
            | "not_between"
            | "not_cidr"
            | "ne"
            | "!="
    )
}

/// Compare two values using the specified operator
///
/// # Arguments
///
/// * `field_value` - The field value from the record
/// * `operator` - The comparison operator (eq, ne, gt, contains, etc.)
/// * `compare_value` - The value to compare against
///
/// # Returns
///
/// true if the comparison succeeds, false otherwise
pub fn compare(field_value: &JsonValue, operator: &str, compare_value: &AstValue) -> Result<bool> {
    match operator {
        "eq" | "=" => compare_eq(field_value, compare_value),
        "ne" | "!=" => compare_ne(field_value, compare_value),
        "eq_ci" => compare_eq_ci(field_value, compare_value),
        "gt" | ">" => compare_gt(field_value, compare_value),
        "gte" | ">=" => compare_gte(field_value, compare_value),
        "lt" | "<" => compare_lt(field_value, compare_value),
        "lte" | "<=" => compare_lte(field_value, compare_value),
        "contains" => compare_contains_ci(field_value, compare_value),
        "contains_cs" => compare_contains(field_value, compare_value),
        "startswith" => compare_startswith_ci(field_value, compare_value),
        "startswith_cs" => compare_startswith(field_value, compare_value),
        "endswith" => compare_endswith_ci(field_value, compare_value),
        "endswith_cs" => compare_endswith(field_value, compare_value),
        "matches" | "regex" | "regexp" => compare_matches(field_value, compare_value),
        "not_contains" => Ok(!compare_contains_ci(field_value, compare_value)?),
        "not_contains_cs" => Ok(!compare_contains(field_value, compare_value)?),
        "not_startswith" => Ok(!compare_startswith_ci(field_value, compare_value)?),
        "not_startswith_cs" => Ok(!compare_startswith(field_value, compare_value)?),
        "not_endswith" => Ok(!compare_endswith_ci(field_value, compare_value)?),
        "not_endswith_cs" => Ok(!compare_endswith(field_value, compare_value)?),
        "not_matches" | "not_regex" | "not_regexp" => {
            Ok(!compare_matches(field_value, compare_value)?)
        }
        "in" => compare_in_ci(field_value, compare_value),
        "in_cs" => compare_in(field_value, compare_value),
        "not_in" => Ok(!compare_in_ci(field_value, compare_value)?),
        "not_in_cs" => Ok(!compare_in(field_value, compare_value)?),
        "between" => compare_between(field_value, compare_value),
        "not_between" => Ok(!compare_between(field_value, compare_value)?),
        "is" => compare_is(field_value, compare_value),
        "is_not" => Ok(!compare_is(field_value, compare_value)?),
        "cidr" => compare_cidr(field_value, compare_value),
        "not_cidr" => Ok(!compare_cidr(field_value, compare_value)?),
        "any" => compare_contains_ci(field_value, compare_value),
        _ => Err(TqlError::OperatorError(format!(
            "Unknown operator: {}",
            operator
        ))),
    }
}

/// Equality comparison
fn compare_eq(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    Ok(values_equal(field_value, compare_value))
}

/// Inequality comparison
fn compare_ne(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    // A null field MATCHES `!= <non-null>` under Lucene semantics: OpenSearch's
    // `must_not match_phrase` includes documents whose field is null/absent,
    // because JSON nulls are not indexed (tellaro-ui#587). `null != null` is
    // still false. Previously this returned false for `null != <non-null>`,
    // which excluded null-valued records that the DSL backend included.
    if field_value.is_null() {
        return Ok(*compare_value != AstValue::Null);
    }
    Ok(!values_equal(field_value, compare_value))
}

/// Greater than comparison (supports both numeric and string comparison)
fn compare_gt(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    if field_value.is_null() {
        return Ok(false);
    }
    compare_ordered(field_value, compare_value, |a, b| a > b, |a, b| a > b)
}

/// Greater than or equal comparison (supports both numeric and string comparison)
fn compare_gte(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    if field_value.is_null() {
        return Ok(false);
    }
    compare_ordered(field_value, compare_value, |a, b| a >= b, |a, b| a >= b)
}

/// Less than comparison (supports both numeric and string comparison)
fn compare_lt(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    if field_value.is_null() {
        return Ok(false);
    }
    compare_ordered(field_value, compare_value, |a, b| a < b, |a, b| a < b)
}

/// Less than or equal comparison (supports both numeric and string comparison)
fn compare_lte(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    if field_value.is_null() {
        return Ok(false);
    }
    compare_ordered(field_value, compare_value, |a, b| a <= b, |a, b| a <= b)
}

/// Render a field value as the text the string operators compare against.
///
/// Python's evaluator does `str(field_value)` before applying `contains`,
/// `startswith`, `endswith` and `matches`; Rust gated all of them on
/// `JsonValue::String` and answered `false` for anything else. So
/// `event.code matches '^4624$'` on `{"event":{"code":4624}}` was true in
/// Python and false in Rust — and the negated forms inverted that into a
/// FALSE POSITIVE on every numeric field. This function is the repair, and its
/// Python twin is `_as_comparison_text` in
/// `src/tql/evaluator_components/value_comparison.py`.
///
/// It was called `value_as_python_string` and rendered a boolean the way
/// Python's `str()` does — `True` / `False`, capitalised, deliberately. Python's
/// spelling is not the stack's spelling, and it is the only thing in the stack
/// that writes it that way: TQL's own literals are `true` / `false`, JSON writes
/// `true`, and OpenSearch stores, returns and will only PARSE `true`. Measured
/// on OpenSearch 2.19.4, a `boolean`-mapped field:
///
/// ```text
/// term b: true     -> the document
/// term b: "true"   -> the document
/// term b: "True"   -> search_phase_execution_exception (only [true] or [false])
/// ```
///
/// and on a `keyword` field holding "true" and "True", `wildcard *true*` and
/// `wildcard *True*` select DIFFERENT documents. `c0b599a` moved `::string` to
/// lowercase and `10aef73` moved the translator to `true`/`false`; both were
/// deliberately narrow, which left the evaluator and the translator selecting
/// OPPOSITE documents for `f contains_cs true`. This is the other half.
///
/// The name went with the spelling: a helper asserting it reproduces Python's
/// `str()` while deliberately not doing so is a claim the file cannot check.
/// What it reproduces now is the spelling both engines and both layers use.
///
/// `null` deliberately returns `None` rather than `"None"`: both engines agree
/// that a null field matches nothing (`f matches 'None'` on `{"f":null}` is
/// false in Python too), and stringifying it would break that agreement.
fn value_as_comparison_text(value: &JsonValue) -> Option<String> {
    match value {
        JsonValue::String(s) => Some(s.clone()),
        JsonValue::Number(n) => Some(n.to_string()),
        JsonValue::Bool(b) => Some(if *b {
            "true".to_string()
        } else {
            "false".to_string()
        }),
        JsonValue::Null => None,
        // Containers stringify in Python too. Arrays are normally handled by an
        // explicit array arm in the caller (any element matching wins); this is
        // the fallback for operators that have none.
        other => Some(other.to_string()),
    }
}

/// Contains comparison - case-sensitive (substring or array element)
fn compare_contains(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match field_value {
        JsonValue::String(s) => {
            let search = ast_value_to_string(compare_value);
            Ok(s.contains(&search))
        }
        // A multi-valued field satisfies the clause if ANY element does, and an
        // element is answered by the SAME function that answers a scalar. That
        // is what makes `{"f": ["x"]}` and `{"f": "x"}` — the same document to
        // OpenSearch — give the same answer, rather than relying on the arm
        // below being transcribed correctly repeatedly. It was not:
        // `contains_cs` compared each element for EQUALITY instead of substring
        // (`["zzz","Hello"] contains_cs 'ell'` was false in Rust and true in
        // Python, and `not contains_cs` inverted that into a FALSE POSITIVE),
        // and all six arms skipped every non-string element, so
        // `f startswith '46'` was false on `[4624]` and true on the bare 4624.
        JsonValue::Array(arr) => {
            for item in arr {
                if compare_contains(item, compare_value)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // Non-string scalars are compared against their Python string
        // form instead of answering `false` — Python does `str(value)`
        // here, so `event.code contains '462'` on the integer 4624 was
        // true there and false in Rust. See `value_as_comparison_text`.
        other => match value_as_comparison_text(other) {
            Some(text) => {
                let probe = JsonValue::String(text);
                compare_contains(&probe, compare_value)
            }
            None => Ok(false),
        },
    }
}

/// Contains comparison - case-insensitive (default behavior, matches Python)
fn compare_contains_ci(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match field_value {
        JsonValue::String(s) => {
            let search = ast_value_to_string(compare_value).to_lowercase();
            Ok(s.to_lowercase().contains(&search))
        }
        // A multi-valued field satisfies the clause if ANY element does, and an
        // element is answered by the SAME function that answers a scalar. That
        // is what makes `{"f": ["x"]}` and `{"f": "x"}` — the same document to
        // OpenSearch — give the same answer, rather than relying on the arm
        // below being transcribed correctly repeatedly. It was not:
        // `contains_cs` compared each element for EQUALITY instead of substring
        // (`["zzz","Hello"] contains_cs 'ell'` was false in Rust and true in
        // Python, and `not contains_cs` inverted that into a FALSE POSITIVE),
        // and all six arms skipped every non-string element, so
        // `f startswith '46'` was false on `[4624]` and true on the bare 4624.
        JsonValue::Array(arr) => {
            for item in arr {
                if compare_contains_ci(item, compare_value)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // Non-string scalars are compared against their Python string
        // form instead of answering `false` — Python does `str(value)`
        // here, so `event.code contains '462'` on the integer 4624 was
        // true there and false in Rust. See `value_as_comparison_text`.
        other => match value_as_comparison_text(other) {
            Some(text) => {
                let probe = JsonValue::String(text);
                compare_contains_ci(&probe, compare_value)
            }
            None => Ok(false),
        },
    }
}

/// Starts with comparison - case-sensitive
fn compare_startswith(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match field_value {
        JsonValue::String(s) => {
            let prefix = ast_value_to_string(compare_value);
            Ok(s.starts_with(&prefix))
        }
        // A multi-valued field satisfies the clause if ANY element does, and an
        // element is answered by the SAME function that answers a scalar. That
        // is what makes `{"f": ["x"]}` and `{"f": "x"}` — the same document to
        // OpenSearch — give the same answer, rather than relying on the arm
        // below being transcribed correctly repeatedly. It was not:
        // `contains_cs` compared each element for EQUALITY instead of substring
        // (`["zzz","Hello"] contains_cs 'ell'` was false in Rust and true in
        // Python, and `not contains_cs` inverted that into a FALSE POSITIVE),
        // and all six arms skipped every non-string element, so
        // `f startswith '46'` was false on `[4624]` and true on the bare 4624.
        JsonValue::Array(arr) => {
            for item in arr {
                if compare_startswith(item, compare_value)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // Non-string scalars are compared against their Python string
        // form instead of answering `false` — Python does `str(value)`
        // here, so `event.code contains '462'` on the integer 4624 was
        // true there and false in Rust. See `value_as_comparison_text`.
        other => match value_as_comparison_text(other) {
            Some(text) => {
                let probe = JsonValue::String(text);
                compare_startswith(&probe, compare_value)
            }
            None => Ok(false),
        },
    }
}

/// Starts with comparison - case-insensitive (default behavior, matches Python)
fn compare_startswith_ci(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match field_value {
        JsonValue::String(s) => {
            let prefix = ast_value_to_string(compare_value).to_lowercase();
            Ok(s.to_lowercase().starts_with(&prefix))
        }
        // A multi-valued field satisfies the clause if ANY element does, and an
        // element is answered by the SAME function that answers a scalar. That
        // is what makes `{"f": ["x"]}` and `{"f": "x"}` — the same document to
        // OpenSearch — give the same answer, rather than relying on the arm
        // below being transcribed correctly repeatedly. It was not:
        // `contains_cs` compared each element for EQUALITY instead of substring
        // (`["zzz","Hello"] contains_cs 'ell'` was false in Rust and true in
        // Python, and `not contains_cs` inverted that into a FALSE POSITIVE),
        // and all six arms skipped every non-string element, so
        // `f startswith '46'` was false on `[4624]` and true on the bare 4624.
        JsonValue::Array(arr) => {
            for item in arr {
                if compare_startswith_ci(item, compare_value)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // Non-string scalars are compared against their Python string
        // form instead of answering `false` — Python does `str(value)`
        // here, so `event.code contains '462'` on the integer 4624 was
        // true there and false in Rust. See `value_as_comparison_text`.
        other => match value_as_comparison_text(other) {
            Some(text) => {
                let probe = JsonValue::String(text);
                compare_startswith_ci(&probe, compare_value)
            }
            None => Ok(false),
        },
    }
}

/// Ends with comparison - case-sensitive
fn compare_endswith(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match field_value {
        JsonValue::String(s) => {
            let suffix = ast_value_to_string(compare_value);
            Ok(s.ends_with(&suffix))
        }
        // A multi-valued field satisfies the clause if ANY element does, and an
        // element is answered by the SAME function that answers a scalar. That
        // is what makes `{"f": ["x"]}` and `{"f": "x"}` — the same document to
        // OpenSearch — give the same answer, rather than relying on the arm
        // below being transcribed correctly repeatedly. It was not:
        // `contains_cs` compared each element for EQUALITY instead of substring
        // (`["zzz","Hello"] contains_cs 'ell'` was false in Rust and true in
        // Python, and `not contains_cs` inverted that into a FALSE POSITIVE),
        // and all six arms skipped every non-string element, so
        // `f startswith '46'` was false on `[4624]` and true on the bare 4624.
        JsonValue::Array(arr) => {
            for item in arr {
                if compare_endswith(item, compare_value)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // Non-string scalars are compared against their Python string
        // form instead of answering `false` — Python does `str(value)`
        // here, so `event.code contains '462'` on the integer 4624 was
        // true there and false in Rust. See `value_as_comparison_text`.
        other => match value_as_comparison_text(other) {
            Some(text) => {
                let probe = JsonValue::String(text);
                compare_endswith(&probe, compare_value)
            }
            None => Ok(false),
        },
    }
}

/// Ends with comparison - case-insensitive (default behavior, matches Python)
fn compare_endswith_ci(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match field_value {
        JsonValue::String(s) => {
            let suffix = ast_value_to_string(compare_value).to_lowercase();
            Ok(s.to_lowercase().ends_with(&suffix))
        }
        // A multi-valued field satisfies the clause if ANY element does, and an
        // element is answered by the SAME function that answers a scalar. That
        // is what makes `{"f": ["x"]}` and `{"f": "x"}` — the same document to
        // OpenSearch — give the same answer, rather than relying on the arm
        // below being transcribed correctly repeatedly. It was not:
        // `contains_cs` compared each element for EQUALITY instead of substring
        // (`["zzz","Hello"] contains_cs 'ell'` was false in Rust and true in
        // Python, and `not contains_cs` inverted that into a FALSE POSITIVE),
        // and all six arms skipped every non-string element, so
        // `f startswith '46'` was false on `[4624]` and true on the bare 4624.
        JsonValue::Array(arr) => {
            for item in arr {
                if compare_endswith_ci(item, compare_value)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // Non-string scalars are compared against their Python string
        // form instead of answering `false` — Python does `str(value)`
        // here, so `event.code contains '462'` on the integer 4624 was
        // true there and false in Rust. See `value_as_comparison_text`.
        other => match value_as_comparison_text(other) {
            Some(text) => {
                let probe = JsonValue::String(text);
                compare_endswith_ci(&probe, compare_value)
            }
            None => Ok(false),
        },
    }
}

/// Case-insensitive equality comparison
fn compare_eq_ci(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match (field_value, compare_value) {
        (JsonValue::String(a), AstValue::String(b)) => Ok(a.to_lowercase() == b.to_lowercase()),
        // A multi-valued field satisfies `eq_ci` if ANY element does.
        //
        // Without this arm the fallback below reached `values_equal`, which DOES
        // iterate an array but compares each element case-SENSITIVELY — so
        // `f eq_ci 'hello'` was false on `["Hello"]` and true on the bare
        // `"Hello"`, two spellings of one document. `eq_ci` is the operator an
        // analyst reaches for on Windows usernames and paths, which are
        // routinely multi-valued.
        //
        // Scoped to a SCALAR right-hand side so that `(Array, List)`
        // whole-array equality keeps its exact-match semantics, exactly as
        // `values_equal` scopes its own array arm.
        (JsonValue::Array(items), rhs) if !matches!(rhs, AstValue::List(_)) => {
            for item in items {
                if compare_eq_ci(item, rhs)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // For non-string types, fall back to regular equality
        _ => compare_eq(field_value, compare_value),
    }
}

/// Regex match comparison with cached pattern compilation.
///
/// Errors rather than answering `false` when the pattern cannot be run -- see
/// `MAX_REGEX_PATTERN_LENGTH` and tql#185.
fn compare_matches(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    let pattern = ast_value_to_string(compare_value);

    // Refuse an over-long pattern BEFORE consulting the cache, so the verdict
    // does not depend on whether some earlier query happened to compile it.
    let pattern_len = pattern.chars().count();
    if pattern_len > MAX_REGEX_PATTERN_LENGTH {
        return Err(TqlError::ValueError(format!(
            "Regex pattern is {} characters, which exceeds the maximum supported length of {}. \
             The query cannot run as written.",
            pattern_len, MAX_REGEX_PATTERN_LENGTH
        )));
    }

    // Fast path: a pattern already in the cache is known to compile.
    {
        // Recover from a poisoned mutex rather than propagating the panic.
        let cache = REGEX_CACHE.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(regex) = cache.get(&pattern) {
            return Ok(matches_string(field_value, regex));
        }
    }

    // Compile and cache the regex.
    //
    // Compilation happens for EVERY field type, not only strings. Whether a rule
    // can run is a property of the pattern, not of the record that happened to
    // arrive first -- gating the check on `JsonValue::String` would make an
    // unrunnable rule error on some records and quietly answer `false` on
    // others, which is the silent-miss shape tql#185 exists to remove.
    let regex = Regex::new(&pattern)
        .map_err(|e| TqlError::ValueError(format!("Invalid regex pattern '{}': {}", pattern, e)))?;
    let result = matches_string(field_value, &regex);
    let mut cache = REGEX_CACHE.lock().unwrap_or_else(|e| e.into_inner());
    // Evict all entries when the cache exceeds the size limit to prevent
    // unbounded memory growth from adversarial or highly-varied patterns.
    if cache.len() >= MAX_REGEX_CACHE_SIZE {
        cache.clear();
    }
    cache.insert(pattern, regex);
    Ok(result)
}

/// A non-string field never matches a regex; only the match itself is conditional.
fn matches_string(field_value: &JsonValue, regex: &Regex) -> bool {
    match field_value {
        JsonValue::String(s) => regex.is_match(s),
        // Any element matching wins, consistent with every other operator.
        JsonValue::Array(items) => items.iter().any(|i| matches_string(i, regex)),
        // Booleans render as lowercase `true` / `false` (the reason is in
        // `value_as_comparison_text`) and are matched CASE-SENSITIVELY, like
        // every other type — with one narrow exception: a pattern that is
        // ITSELF a boolean literal is lowercased first.
        //
        // That exception is not a style choice, it is Python's coercion made
        // explicit. `value_comparison.py` converts a `"true"`/`"false"` operand
        // to a Python `bool` before the operator arms run, whenever the FIELD is
        // not a string (`keep_as_text` requires `isinstance(field_value, str)`),
        // and `_as_comparison_text` then renders that bool lowercase. So the
        // pattern `TRUE` reaches Python's `re.search` as the literal `true`.
        // Nothing case-insensitive happens: the regex itself is run
        // case-sensitively on both sides of the exception.
        //
        // This arm used to prepend `(?i)` to the WHOLE pattern, justified in a
        // comment as reproducing Python's "booleans match case-insensitively".
        // Python does not have that behaviour. Measured on `{"f": true}`:
        //
        // ```text
        //   f matches 'TR.E'       Rust 1   Python 0
        //   f matches 'RU'         Rust 1   Python 0
        //   f not matches 'TR.E'   Rust 0   Python 1
        // ```
        //
        // The third row is why this blocked rather than merely diverged: an
        // exclusion rule dropped a record it was written to keep. Rust also
        // disagreed with the DSL Rust itself emits — the OpenSearch translator
        // passes the pattern through verbatim and lifts no `(?i)` — so the same
        // saved query selected different documents in memory and on a cluster.
        //
        // Rust is the half that moved. Python's boolean-literal coercion is odd,
        // but it is what the rest of the stack already agrees with, and there is
        // no principled reason for regex matching to be case-insensitive for
        // exactly one JSON type.
        JsonValue::Bool(b) => {
            let text = if *b { "true" } else { "false" };
            let pattern = regex.as_str();
            if pattern.eq_ignore_ascii_case("true") || pattern.eq_ignore_ascii_case("false") {
                // The literal, lowercased — equivalent to searching for the
                // coerced boolean's rendering, which is what Python does.
                return pattern.eq_ignore_ascii_case(text);
            }
            regex.is_match(text)
        }
        // Numbers and objects are matched against their Python string form
        // rather than answering `false` — see `value_as_comparison_text`.
        other => value_as_comparison_text(other)
            .map(|s| regex.is_match(&s))
            .unwrap_or(false),
    }
}

/// IN comparison — case-sensitive (`in_cs`), element-wise `eq`.
///
/// `f in_cs [a, b]` IS `f eq a or f eq b`, and saying so in code rather than in
/// a comment is the point: a membership test that re-implements equality drifts
/// away from it, and both engines had drifted in opposite directions. See
/// `compare_in_ci` for the measurement and the arbiter.
fn compare_in(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match compare_value {
        AstValue::List(list) => {
            for item in list {
                if compare_eq(field_value, item)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // A NON-LIST operand is a broken query, and answering `false` made it a
        // SILENT one -- in the worst available direction. `not_in` is computed
        // as `!compare_in_ci(..)?` (see `compare`), so `f not in 'x'` answered
        // TRUE FOR EVERY RECORD, including the record holding exactly `"x"`.
        // Measured before this change over `[{f:"y"},{f:"x"}]`:
        //
        // ```text
        //   f not in 'x'     -> ["y", "x"]      <- fail-OPEN, matched everything
        //   f in 'x'         -> []              <- zero hits, matched nothing
        //   f not in_cs 'x'  -> ["y", "x"]
        //   f in_cs 'x'      -> []
        // ```
        //
        // So the positive and negative spellings were not complements, and the
        // negative one was a filter that filters nothing. The grammar admits the
        // scalar (`grammar.pest` `in_list`), so this is reachable from ordinary
        // query text rather than only from a hand-built AST.
        //
        // `OperatorError` and not `ValueError`, matching `compare_between`,
        // which already refuses its own non-list operand -- the neighbouring
        // operator that got this right, and the reason the shape was findable at
        // all.
        _ => Err(TqlError::OperatorError(
            "IN operator requires a list of values".to_string(),
        )),
    }
}

/// IN comparison — case-insensitive (`in`, the default spelling), element-wise
/// `eq_ci`.
///
/// This arm used to pre-lowercase `ast_value_to_string(item)` for EVERY element
/// and compare a string field against that text, so a non-string element was
/// answered as text rather than by the equality rules. `f in [5]` was therefore
/// false on the stored `"05"` while `f eq_ci 5` — the same question, one element
/// — was true, and `f in_cs [5]` was true as well, so Rust disagreed with itself
/// twice over.
///
/// Python had the mirror defect from the other direction: it coerced every
/// element numerically regardless of quoting, so `f in ['5']` matched the stored
/// `"05"` even though `f eq '5'` does not. Quoting is the user's type
/// declaration (tql#171) and `eq` has honoured it in both engines since; `in`
/// was the operator that never got the rule.
///
/// Measured over
/// `[{i5:5},{s5:"5"},{i05:"05"},{f5:5.0},{btrue:true},{strue:"true"},{txt:"abc"},{TXT:"ABC"}]`:
///
/// ```text
///                    eq / eq_ci (both engines)   in, Rust         in, Python
///   [5]   / 5        i5 s5 i05 f5                i5 s5 f5         i5 s5 i05 f5
///   ['5'] / '5'      i5 s5 f5                    i5 s5 f5         i5 s5 i05 f5
///   ['05']/ '05'     i5 i05 f5                   i5 i05 f5        i5 s5 i05 f5
/// ```
///
/// `eq` and `eq_ci` were already identical in both engines, so they are the
/// reference and neither engine's `in` had to be invented — it had to be
/// delegated.
///
/// The DSL half of this divergence is INERT, which is what settles the direction
/// rather than aesthetics. Python emits `terms: ["5"]` where Rust emits
/// `terms: [5]`, and OpenSearch resolves a `terms` clause by the FIELD's
/// mapping, not by the literal's JSON type. Measured on OpenSearch 2.19.4 over
/// `{n: 5 (long)}` and `{k: "5" / "05" (keyword)}`:
///
/// ```text
///   terms n ["5"] -> the doc      terms n [5]  -> the doc
///   terms n ["05"]-> the doc      terms n ["5.0"] -> the doc
///   terms k ["5"] -> the "5" doc  terms k [5]  -> the "5" doc
///   terms k ["05"]-> the "05" doc
/// ```
///
/// So the two DSLs select identical documents and only the in-memory engines
/// ever disagreed. With no mapping to consult, the evaluator has to decide from
/// the literal — and the rule it already has for that is `eq`'s.
fn compare_in_ci(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match compare_value {
        AstValue::List(list) => {
            for item in list {
                if compare_eq_ci(field_value, item)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        // A NON-LIST operand is a broken query, and answering `false` made it a
        // SILENT one -- in the worst available direction. `not_in` is computed
        // as `!compare_in_ci(..)?` (see `compare`), so `f not in 'x'` answered
        // TRUE FOR EVERY RECORD, including the record holding exactly `"x"`.
        // Measured before this change over `[{f:"y"},{f:"x"}]`:
        //
        // ```text
        //   f not in 'x'     -> ["y", "x"]      <- fail-OPEN, matched everything
        //   f in 'x'         -> []              <- zero hits, matched nothing
        //   f not in_cs 'x'  -> ["y", "x"]
        //   f in_cs 'x'      -> []
        // ```
        //
        // So the positive and negative spellings were not complements, and the
        // negative one was a filter that filters nothing. The grammar admits the
        // scalar (`grammar.pest` `in_list`), so this is reachable from ordinary
        // query text rather than only from a hand-built AST.
        //
        // `OperatorError` and not `ValueError`, matching `compare_between`,
        // which already refuses its own non-list operand -- the neighbouring
        // operator that got this right, and the reason the shape was findable at
        // all.
        _ => Err(TqlError::OperatorError(
            "IN operator requires a list of values".to_string(),
        )),
    }
}

/// BETWEEN comparison (field value between [min, max])
///
/// The parser normalizes both `field between [val1, val2]` (list syntax) and
/// `field between val1 and val2` (natural syntax) into `Value::List(vec![val1, val2])`,
/// so compare_value should always arrive as a two-element list. The non-list match arm
/// is a defensive guard for programmatic AST construction.
fn compare_between(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match compare_value {
        AstValue::List(list) if list.len() == 2 => {
            let min = &list[0];
            let max = &list[1];

            // A multi-valued field satisfies `between` if ANY ONE element is
            // inside the range. Both bounds are tested against the SAME element:
            // asking "any element >= min" and "any element <= max" separately
            // would report `[1, 99] between [5, 20]` as a match on the strength
            // of two different elements, neither of which is in the range.
            //
            // `between` decomposes to the range operators, and those gained
            // array iteration in `compare_ordered` — this arm was missed, so
            // Rust answered `f between [5,20]` false on `[10]` and true on the
            // bare 10, and disagreed with its own `gte`/`lte`.
            if let JsonValue::Array(items) = field_value {
                for item in items {
                    if between_bounds(item, min, max)? {
                        return Ok(true);
                    }
                }
                return Ok(false);
            }

            between_bounds(field_value, min, max)
        }
        // `OperatorError`, not `ValueError`: Python raises `TQLOperatorError`
        // for the same refusal (from its parser for `between`, from
        // `value_comparison.py` for `not between`), and the two engines
        // refusing the same query with different exception KINDS is a
        // divergence a caller can trip over even though both refuse. The
        // message text still differs -- Python's comes from a parser this
        // change does not own -- but the class is what a caller matches on and
        // what the shared fixture pins.
        _ => Err(TqlError::OperatorError(
            "BETWEEN operator requires a list of exactly 2 values".to_string(),
        )),
    }
}

/// One element (or one scalar) against both `between` bounds.
///
/// Delegates to `compare_ordered` rather than `compare_numeric`, so `between`
/// orders values exactly the way the `gte`/`lte` it decomposes to already do:
/// numerically when both sides read as numbers, and LEXICOGRAPHICALLY when both
/// are strings.
///
/// It used to be numeric-only, which made Rust disagree with itself —
/// `f gte 'Hello' and f lte 'World'` was true on `"Hello"` while
/// `f between ['Hello','World']` was false — and disagree with the DSL it emits.
/// Both engines translate `between` to `{"range": {f: {gte, lte}}}`, and a
/// `range` on a keyword field IS lexicographic: measured live on OpenSearch
/// 2.19.4, `{"range":{"f":{"gte":"Hello","lte":"World"}}}` returns the documents
/// holding "Hello" and "World". So the cluster, Python and Rust's own range
/// operators all agreed, and only Rust's `between` did not. That is the
/// argument, not "Python is the reference".
///
/// Bound ORDER is deliberately left alone: Rust tests `>= list[0]` and
/// `<= list[1]` as written, which is what OpenSearch does with the `range` this
/// translates to. (Python additionally sorts the pair, so the two still differ
/// on a reversed `between [20, 5]` — a separate divergence where the cluster
/// sides with Rust, and not one to resolve by widening this fix.)
fn between_bounds(field_value: &JsonValue, min: &AstValue, max: &AstValue) -> Result<bool> {
    let at_least_min = compare_ordered(field_value, min, |a, b| a >= b, |a, b| a >= b)?;
    if !at_least_min {
        return Ok(false);
    }
    compare_ordered(field_value, max, |a, b| a <= b, |a, b| a <= b)
}

/// IS comparison (check for null)
fn compare_is(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    match compare_value {
        AstValue::Null => Ok(field_value.is_null()),
        _ => Ok(values_equal(field_value, compare_value)),
    }
}

/// CIDR comparison (check if IP is in CIDR range)
///
/// A multi-valued field satisfies `cidr` if ANY element does — the same rule
/// `values_equal` and `compare_ordered` apply, and the same rule OpenSearch
/// applies to a `term` query on a multi-valued `ip` field.
///
/// Without this arm `related.ip cidr '10.0.0.0/8'` was FALSE against the array
/// that ECS guarantees for that field: `json_to_string` rendered the whole
/// array as `["10.1.2.3","8.8.8.8"]`, which is not a parseable IP, so the
/// scalar path answered `false` — and `not_cidr` inverted that into a false
/// positive on every array-valued IP field. `related.ip`, `related.hosts` and
/// the `*.ip` families are arrays by definition, so this was the normal case
/// rather than an edge one.
///
/// It was broken IDENTICALLY in Python (`_check_cidr` did `str(ip_value)` on
/// the list), which is why the Rust-vs-Python differential harness could not
/// see it: both engines agreed, and both were wrong. A shared bug needs its own
/// test on each side, not a comparison between them.
///
/// Iterating with a loop rather than recursing mirrors `compare_ordered`: that
/// function is generic over its predicates and recursion made rustc instantiate
/// it without bound. `compare_cidr` is not generic, so recursion would compile
/// here — the loop is kept for consistency with its sibling so the next reader
/// does not have to work out why the two differ.
fn compare_cidr(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    if let JsonValue::Array(items) = field_value {
        for item in items {
            if compare_cidr_scalar(item, compare_value)? {
                return Ok(true);
            }
        }
        return Ok(false);
    }
    compare_cidr_scalar(field_value, compare_value)
}

/// Single-value half of [`compare_cidr`].
fn compare_cidr_scalar(field_value: &JsonValue, compare_value: &AstValue) -> Result<bool> {
    // Extract IP address from field value
    let ip_str = match field_value {
        JsonValue::String(s) => s.clone(),
        _ => json_to_string(field_value),
    };

    // Extract CIDR pattern from compare value
    let cidr_str = ast_value_to_string(compare_value);

    // Parse IP address
    let ip = match IpAddr::from_str(&ip_str) {
        Ok(addr) => addr,
        Err(_) => return Ok(false), // Invalid IP address
    };

    // Parse CIDR pattern
    // For now, we'll do a simple implementation
    // TODO: Use ipnetwork crate for proper CIDR matching
    let parts: Vec<&str> = cidr_str.split('/').collect();
    if parts.len() != 2 {
        return Ok(false); // Invalid CIDR format
    }

    let network_ip = match IpAddr::from_str(parts[0]) {
        Ok(addr) => addr,
        Err(_) => return Ok(false),
    };

    let prefix_len: u8 = match parts[1].parse() {
        Ok(len) => len,
        Err(_) => return Ok(false),
    };

    // Check if both IPs are the same version
    match (ip, network_ip) {
        (IpAddr::V4(ip_v4), IpAddr::V4(net_v4)) => check_ipv4_in_cidr(ip_v4, net_v4, prefix_len),
        (IpAddr::V6(ip_v6), IpAddr::V6(net_v6)) => check_ipv6_in_cidr(ip_v6, net_v6, prefix_len),
        _ => Ok(false), // Mismatched IP versions
    }
}

/// Check if IPv4 address is in CIDR range
fn check_ipv4_in_cidr(
    ip: std::net::Ipv4Addr,
    network: std::net::Ipv4Addr,
    prefix_len: u8,
) -> Result<bool> {
    if prefix_len > 32 {
        return Ok(false);
    }

    let ip_u32 = u32::from(ip);
    let net_u32 = u32::from(network);

    let mask = if prefix_len == 0 {
        0
    } else {
        !0u32 << (32 - prefix_len)
    };

    Ok((ip_u32 & mask) == (net_u32 & mask))
}

/// Check if IPv6 address is in CIDR range
fn check_ipv6_in_cidr(
    ip: std::net::Ipv6Addr,
    network: std::net::Ipv6Addr,
    prefix_len: u8,
) -> Result<bool> {
    if prefix_len > 128 {
        return Ok(false);
    }

    let ip_u128 = u128::from(ip);
    let net_u128 = u128::from(network);

    let mask = if prefix_len == 0 {
        0
    } else {
        !0u128 << (128 - prefix_len)
    };

    Ok((ip_u128 & mask) == (net_u128 & mask))
}

/// Helper: Compare values with ordered comparison (supports both numeric and string)
/// This enables comparisons like @timestamp > '2026-01-01T00:00:00Z' where
/// ISO 8601 timestamps sort correctly as strings.
fn compare_ordered<NumF, StrF>(
    field_value: &JsonValue,
    compare_value: &AstValue,
    num_predicate: NumF,
    str_predicate: StrF,
) -> Result<bool>
where
    NumF: Fn(f64, f64) -> bool,
    StrF: Fn(&str, &str) -> bool,
{
    // A multi-valued field satisfies an ordered comparison if ANY element does
    // — the same rule as equality above, and the same rule OpenSearch applies
    // to a `range` query on a multi-valued field. Without this,
    // `destination.port gt 1024` on `{"destination":{"port":[49152]}}` was
    // false in Rust and true in Python.
    if let JsonValue::Array(items) = field_value {
        // Loop rather than recurse: recursing through the generic parameters
        // makes rustc instantiate `compare_ordered<&&&&...>` without bound and
        // hit the recursion limit. The scalar body below is the only thing that
        // needs repeating, so run it per element via a closure over the same
        // predicates.
        for item in items {
            let matched = (|| -> Result<bool> {
                if let (Some(a), Some(b)) =
                    (json_to_number(item), ast_value_to_number(compare_value))
                {
                    return Ok(num_predicate(a, b));
                }
                if let (JsonValue::String(f), AstValue::String(c)) = (item, compare_value) {
                    return Ok(str_predicate(f, c));
                }
                Ok(false)
            })()?;
            if matched {
                return Ok(true);
            }
        }
        return Ok(false);
    }

    // Try numeric comparison first
    let field_num = json_to_number(field_value);
    let compare_num = ast_value_to_number(compare_value);

    if let (Some(a), Some(b)) = (field_num, compare_num) {
        return Ok(num_predicate(a, b));
    }

    // Fall back to string comparison if both are strings
    if let (JsonValue::String(field_str), AstValue::String(compare_str)) =
        (field_value, compare_value)
    {
        return Ok(str_predicate(field_str, compare_str));
    }

    // Can't compare mixed or unsupported types
    Ok(false)
}

/// Helper: Check if two values are equal (with type coercion)
fn values_equal(json_value: &JsonValue, ast_value: &AstValue) -> bool {
    // A multi-valued field satisfies a scalar comparison if ANY element does.
    //
    // This arm was missing, so `tags eq 'admin'` on `{"tags":["admin","user"]}`
    // returned false while Python returned true — and `ne` inverted it into a
    // FALSE POSITIVE on every array field. OpenSearch itself treats an array as
    // a multi-valued field (`{"term":{"tags":"admin"}}` matches that document),
    // so the DSL both engines emit already had these semantics; only Rust's
    // in-memory evaluator disagreed.
    //
    // It was also internally inconsistent: `contains`, `in`, `startswith` and
    // `endswith` all iterate arrays already, which is what makes this a missing
    // arm rather than a deliberate design.
    //
    // Scoped to a SCALAR right-hand side: `(Array, List)` below is list-vs-list
    // equality and must keep its exact-match semantics.
    if let (JsonValue::Array(items), rhs) = (json_value, ast_value) {
        if !matches!(rhs, AstValue::List(_)) {
            return items.iter().any(|item| values_equal(item, rhs));
        }
    }

    match (json_value, ast_value) {
        (JsonValue::String(a), AstValue::String(b)) => a == b,
        (JsonValue::Number(a), AstValue::Integer(b)) => {
            // Try exact integer comparison first, then fall back to f64
            // (needed when mutators like |sum return float-based Numbers)
            a.as_i64() == Some(*b)
                || a.as_f64()
                    .map(|f| (f - *b as f64).abs() < f64::EPSILON)
                    .unwrap_or(false)
        }
        (JsonValue::Number(a), AstValue::Float(b)) => {
            if let Some(a_f) = a.as_f64() {
                (a_f - b).abs() < f64::EPSILON
            } else {
                false
            }
        }
        (JsonValue::Bool(a), AstValue::Boolean(b)) => a == b,
        (JsonValue::Null, AstValue::Null) => true,
        // A null is equal to the bareword `null` and to NOTHING else.
        //
        // Without these two arms both sides fell through to the catch-all,
        // where `json_to_string(Null)` and `ast_value_to_string(Null)` each
        // produce the four characters `null` — so `f eq null` also matched the
        // STRING "null", and `f eq 'null'` also matched a real JSON null. The
        // quoted form is the user's type declaration (tql#169, "a host
        // genuinely named null was otherwise unqueryable"), so conflating the
        // two makes a host named `null` unqueryable again from the other
        // direction.
        //
        // The emitted DSL is the tiebreaker: both backends translate
        // `f eq null` to `{"term": {"f": null}}`, which cannot match a
        // document whose field holds the string "null". Python agrees after
        // the bareword-null parser fix; this brings the Rust evaluator onto
        // the same answer instead of leaving a third one.
        (JsonValue::Null, _) => false,
        (_, AstValue::Null) => false,
        (JsonValue::Array(a), AstValue::List(b)) => {
            if a.len() != b.len() {
                return false;
            }
            for (json_item, ast_item) in a.iter().zip(b.iter()) {
                if !values_equal(json_item, ast_item) {
                    return false;
                }
            }
            true
        }
        // Type coercion for boolean-to-number
        (JsonValue::Bool(b), AstValue::Integer(i)) => {
            let bool_as_int = if *b { 1 } else { 0 };
            bool_as_int == *i
        }
        (JsonValue::Bool(b), AstValue::Float(f)) => {
            let bool_as_float = if *b { 1.0 } else { 0.0 };
            (bool_as_float - f).abs() < f64::EPSILON
        }
        // Type coercion for number-to-string and other cases
        _ => {
            // Try numeric comparison first
            if let (Some(json_num), Some(ast_num)) =
                (json_to_number(json_value), ast_value_to_number(ast_value))
            {
                return (json_num - ast_num).abs() < f64::EPSILON;
            }

            // Fall back to string comparison
            let json_str = json_to_string(json_value);
            let ast_str = ast_value_to_string(ast_value);
            json_str == ast_str
        }
    }
}

/// Helper: Convert JsonValue to number
fn json_to_number(value: &JsonValue) -> Option<f64> {
    match value {
        JsonValue::Number(n) => n.as_f64(),
        JsonValue::String(s) => finite(s.parse::<f64>().ok()),
        JsonValue::Bool(true) => Some(1.0),
        JsonValue::Bool(false) => Some(0.0),
        _ => None,
    }
}

/// Helper: Convert AstValue to number
fn ast_value_to_number(value: &AstValue) -> Option<f64> {
    match value {
        AstValue::Integer(i) => Some(*i as f64),
        AstValue::Float(f) => finite(Some(*f)),
        AstValue::String(s) => finite(s.parse::<f64>().ok()),
        AstValue::Boolean(true) => Some(1.0),
        AstValue::Boolean(false) => Some(0.0),
        _ => None,
    }
}

/// A STRING is only a number when it parses to a FINITE one.
///
/// Rust's `str::parse::<f64>` accepts `"inf"`, `"-inf"`, `"infinity"` and
/// `"nan"` in any case; Python's `float()` accepts the same spellings, and
/// `_convert_numeric` in `value_comparison.py` then rejects the result
/// explicitly — the comment there names those spellings as the reason. So the
/// engines disagreed on ordinary stored text, with Rust in the false-positive
/// direction. Measured over
/// `[{num: "1.5"}, {inf: "inf"}, {nan: "nan"}, {small: "0.5"},
///   {Infinity: "Infinity"}, {txt: "abc"}]`:
///
/// ```text
///   f gt 1     Rust ["num", "inf", "Infinity"]   Python ["num"]
/// ```
///
/// `inf` is not a number a JSON document can hold, so a field carrying it holds
/// TEXT that looks like one, and every consumer of this stack — serde_json, the
/// OpenSearch mapping, Python's comparator — refuses to read it as a magnitude.
/// A `nan` read as a number is worse than a false positive: it compares false to
/// everything including itself, so it would silently disappear from BOTH a
/// filter and its negation.
///
/// A `JsonValue::Number` cannot be non-finite (serde_json refuses to construct
/// one), so only the string and float-literal arms need this.
fn finite(value: Option<f64>) -> Option<f64> {
    value.filter(|f| f.is_finite())
}

/// Helper: Convert JsonValue to string
fn json_to_string(value: &JsonValue) -> String {
    match value {
        JsonValue::String(s) => s.clone(),
        JsonValue::Number(n) => n.to_string(),
        JsonValue::Bool(b) => b.to_string(),
        JsonValue::Null => "null".to_string(),
        JsonValue::Array(_) | JsonValue::Object(_) => value.to_string(),
    }
}

/// Helper: Convert AstValue to string.
///
/// This is BOTH the operand renderer for the string comparison arms and the
/// last-resort stringify in `values_equal`. It briefly was not: an
/// `ast_value_to_python_string` wrapper sat in front of the string arms and
/// capitalised a boolean operand to `True`/`False`, to match what
/// `value_as_python_string` then did to a boolean FIELD. Both have since moved
/// to `true`/`false` — the spelling TQL, JSON and OpenSearch all use, and the
/// only one OpenSearch will parse for a boolean field — so the wrapper became a
/// byte-for-byte alias of this function and was removed rather than left to
/// drift. See `value_as_comparison_text` for the measurement.
///
/// Booleans therefore render identically here and in `values_equal`. That is a
/// property worth stating: the earlier note warned that changing bool rendering
/// in `values_equal` would move equality semantics for every operator at once.
/// It does not, because `b.to_string()` is what this arm already produced —
/// nothing about equality changed, only the string arms stopped disagreeing
/// with it.
fn ast_value_to_string(value: &AstValue) -> String {
    match value {
        AstValue::String(s) => s.clone(),
        AstValue::Integer(i) => i.to_string(),
        AstValue::Float(f) => f.to_string(),
        AstValue::Boolean(b) => b.to_string(),
        AstValue::Null => "null".to_string(),
        AstValue::List(list) => {
            let items: Vec<String> = list.iter().map(ast_value_to_string).collect();
            format!("[{}]", items.join(", "))
        }
    }
}

#[cfg(test)]
mod tests {
    /// A non-list operand for `in` / `not in` is refused, not answered.
    ///
    /// The fail-open is the one that matters: `not_in` is computed as
    /// `!compare_in_ci(..)?`, so `_ => Ok(false)` made `f not in 'x'` TRUE for
    /// every record — a filter that filters nothing, including the record
    /// holding exactly `"x"`. Measured before the fix over `[{f:"y"},{f:"x"}]`,
    /// `f not in 'x'` returned BOTH records.
    ///
    /// `between` — the neighbouring operator, one function down — already
    /// refused its own non-list operand with `OperatorError`, so this asserts
    /// the same error KIND rather than merely "an error".
    mod non_list_membership_operand {
        use super::*;
        use crate::evaluator::TqlEvaluator;
        use crate::parser::TqlParser;

        fn evaluate(query: &str) -> Result<Vec<String>> {
            let parser = TqlParser::new();
            let evaluator = TqlEvaluator::new();
            let ast = parser
                .parse(query)
                .expect("the grammar admits a scalar here");
            let records = vec![serde_json::json!({"f": "y"}), serde_json::json!({"f": "x"})];
            Ok(evaluator
                .filter(&ast, &records)?
                .iter()
                .map(|r| r["f"].as_str().unwrap().to_string())
                .collect())
        }

        fn assert_refused(query: &str) {
            match evaluate(query) {
                Err(TqlError::OperatorError(message)) => assert!(
                    message.contains("IN operator requires a list"),
                    "wrong message for `{query}`: {message}"
                ),
                Err(other) => panic!(
                    "`{query}` must be refused as OperatorError, matching `between`; got {other:?}"
                ),
                Ok(hits) => panic!("`{query}` was ANSWERED with {hits:?} instead of refused"),
            }
        }

        #[test]
        fn not_in_with_a_scalar_operand_is_refused() {
            assert_refused("f not in 'x'");
        }

        #[test]
        fn not_in_cs_with_a_scalar_operand_is_refused() {
            assert_refused("f not in_cs 'x'");
        }

        #[test]
        fn in_with_a_scalar_operand_is_refused() {
            assert_refused("f in 'x'");
        }

        #[test]
        fn in_cs_with_a_scalar_operand_is_refused() {
            assert_refused("f in_cs 'x'");
        }

        /// The refusal has the same KIND as `between`'s, which is the precedent
        /// this fix followed. A caller matches on the variant, not the text.
        #[test]
        fn between_refuses_its_own_non_list_operand_the_same_way() {
            assert!(
                matches!(evaluate("f between 'x'"), Err(TqlError::OperatorError(_))),
                "the precedent must still hold, or the two operators have drifted apart again"
            );
        }

        /// CONTROL: a real list operand is unaffected, in both polarities and
        /// both casings.
        #[test]
        fn a_list_operand_still_works() {
            assert_eq!(evaluate("f in ['x']").unwrap(), vec!["x".to_string()]);
            assert_eq!(evaluate("f not in ['x']").unwrap(), vec!["y".to_string()]);
            assert_eq!(evaluate("f in_cs ['X']").unwrap(), Vec::<String>::new());
            assert_eq!(
                evaluate("f not in_cs ['X']").unwrap(),
                vec!["y".to_string(), "x".to_string()]
            );
        }
    }

    use super::*;
    use serde_json::json;

    #[test]
    fn test_compare_eq() {
        assert!(compare(&json!("test"), "eq", &AstValue::String("test".to_string())).unwrap());
        assert!(!compare(&json!("test"), "eq", &AstValue::String("other".to_string())).unwrap());
        assert!(compare(&json!(42), "eq", &AstValue::Integer(42)).unwrap());
        assert!(compare(&json!(true), "eq", &AstValue::Boolean(true)).unwrap());
    }

    #[test]
    fn test_compare_ne() {
        assert!(compare(&json!("test"), "ne", &AstValue::String("other".to_string())).unwrap());
        assert!(!compare(&json!("test"), "ne", &AstValue::String("test".to_string())).unwrap());
    }

    #[test]
    fn test_compare_gt() {
        assert!(compare(&json!(10), "gt", &AstValue::Integer(5)).unwrap());
        assert!(!compare(&json!(5), "gt", &AstValue::Integer(10)).unwrap());
        assert!(!compare(&json!(5), "gt", &AstValue::Integer(5)).unwrap());
    }

    #[test]
    fn test_compare_gte() {
        assert!(compare(&json!(10), "gte", &AstValue::Integer(5)).unwrap());
        assert!(compare(&json!(5), "gte", &AstValue::Integer(5)).unwrap());
        assert!(!compare(&json!(5), "gte", &AstValue::Integer(10)).unwrap());
    }

    #[test]
    fn test_compare_lt() {
        assert!(compare(&json!(5), "lt", &AstValue::Integer(10)).unwrap());
        assert!(!compare(&json!(10), "lt", &AstValue::Integer(5)).unwrap());
    }

    #[test]
    fn test_compare_lte() {
        assert!(compare(&json!(5), "lte", &AstValue::Integer(10)).unwrap());
        assert!(compare(&json!(5), "lte", &AstValue::Integer(5)).unwrap());
        assert!(!compare(&json!(10), "lte", &AstValue::Integer(5)).unwrap());
    }

    #[test]
    fn test_compare_contains() {
        assert!(compare(
            &json!("hello world"),
            "contains",
            &AstValue::String("world".to_string())
        )
        .unwrap());
        assert!(!compare(
            &json!("hello world"),
            "contains",
            &AstValue::String("foo".to_string())
        )
        .unwrap());

        // Array contains
        assert!(compare(
            &json!(["a", "b", "c"]),
            "contains",
            &AstValue::String("b".to_string())
        )
        .unwrap());
        assert!(!compare(
            &json!(["a", "b", "c"]),
            "contains",
            &AstValue::String("d".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_compare_startswith() {
        assert!(compare(
            &json!("hello world"),
            "startswith",
            &AstValue::String("hello".to_string())
        )
        .unwrap());
        assert!(!compare(
            &json!("hello world"),
            "startswith",
            &AstValue::String("world".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_compare_endswith() {
        assert!(compare(
            &json!("hello world"),
            "endswith",
            &AstValue::String("world".to_string())
        )
        .unwrap());
        assert!(!compare(
            &json!("hello world"),
            "endswith",
            &AstValue::String("hello".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_compare_matches() {
        assert!(compare(
            &json!("test123"),
            "matches",
            &AstValue::String(r"test\d+".to_string())
        )
        .unwrap());
        assert!(!compare(
            &json!("test"),
            "matches",
            &AstValue::String(r"test\d+".to_string())
        )
        .unwrap());
    }

    // -- tql#185: an unrunnable pattern is an error, never a verdict ---------
    //
    // Every "must error" assertion below is paired with a positivity control on
    // the same surface, so an implementation that simply errored on every regex
    // would not pass.

    #[test]
    fn test_over_length_regex_errors_rather_than_reporting_no_match() {
        let over = "z".repeat(MAX_REGEX_PATTERN_LENGTH + 1);
        let err = compare(&json!("alice"), "matches", &AstValue::String(over))
            .expect_err("an over-length pattern must not be answered with a bool");
        let message = err.to_string();
        assert!(
            message.contains(&MAX_REGEX_PATTERN_LENGTH.to_string()),
            "error should name the limit: {message}"
        );
        assert!(
            message.contains("cannot run"),
            "error should say the query cannot run: {message}"
        );
    }

    #[test]
    fn test_over_length_regex_errors_under_negation_too() {
        // The negated fallback is the more dangerous half: answering `true` here
        // would make a broken exclusion match every record.
        let over = "z".repeat(MAX_REGEX_PATTERN_LENGTH + 1);
        assert!(compare(&json!("alice"), "not_matches", &AstValue::String(over)).is_err());
    }

    #[test]
    fn test_over_length_regex_errors_for_non_string_fields() {
        // Whether a rule can run is a property of the pattern, not of whichever
        // record arrived first. A numeric field must not quietly answer `false`.
        let over = "z".repeat(MAX_REGEX_PATTERN_LENGTH + 1);
        assert!(compare(&json!(42), "matches", &AstValue::String(over)).is_err());
    }

    #[test]
    fn test_regex_at_the_cap_still_runs() {
        // POSITIVITY CONTROL.
        let mut at_cap = String::from("alice|");
        at_cap.push_str(&"z".repeat(MAX_REGEX_PATTERN_LENGTH - at_cap.chars().count()));
        assert_eq!(at_cap.chars().count(), MAX_REGEX_PATTERN_LENGTH);

        assert!(compare(
            &json!("alice"),
            "matches",
            &AstValue::String(at_cap.clone())
        )
        .unwrap());
        assert!(!compare(&json!("bob"), "matches", &AstValue::String(at_cap)).unwrap());
    }

    #[test]
    fn test_invalid_regex_errors_for_non_string_fields() {
        // Same reasoning as the over-length case: the pattern cannot compile, so
        // the answer is an error on every record, not `false` on some of them.
        assert!(compare(&json!(42), "matches", &AstValue::String("[a-z".to_string())).is_err());
        assert!(compare(
            &json!("alice"),
            "matches",
            &AstValue::String("[a-z".to_string())
        )
        .is_err());
    }

    #[test]
    fn test_valid_regex_on_non_string_field_is_a_clean_false() {
        // POSITIVITY CONTROL: a runnable pattern against a numeric field is an
        // ordinary non-match, not an error.
        assert!(!compare(&json!(42), "matches", &AstValue::String("^a".to_string())).unwrap());
    }

    #[test]
    fn test_compare_in() {
        let list = AstValue::List(vec![
            AstValue::String("a".to_string()),
            AstValue::String("b".to_string()),
            AstValue::String("c".to_string()),
        ]);

        assert!(compare(&json!("b"), "in", &list).unwrap());
        assert!(!compare(&json!("d"), "in", &list).unwrap());
    }

    #[test]
    fn test_compare_not_in() {
        let list = AstValue::List(vec![
            AstValue::String("a".to_string()),
            AstValue::String("b".to_string()),
        ]);

        assert!(compare(&json!("c"), "not_in", &list).unwrap());
        assert!(!compare(&json!("a"), "not_in", &list).unwrap());
    }

    #[test]
    fn test_compare_between() {
        let range = AstValue::List(vec![AstValue::Integer(10), AstValue::Integer(20)]);

        assert!(compare(&json!(15), "between", &range).unwrap());
        assert!(compare(&json!(10), "between", &range).unwrap());
        assert!(compare(&json!(20), "between", &range).unwrap());
        assert!(!compare(&json!(5), "between", &range).unwrap());
        assert!(!compare(&json!(25), "between", &range).unwrap());
    }

    #[test]
    fn test_compare_is_null() {
        assert!(compare(&json!(null), "is", &AstValue::Null).unwrap());
        assert!(!compare(&json!("test"), "is", &AstValue::Null).unwrap());
    }

    #[test]
    fn test_compare_is_not_null() {
        assert!(compare(&json!("test"), "is_not", &AstValue::Null).unwrap());
        assert!(!compare(&json!(null), "is_not", &AstValue::Null).unwrap());
    }

    #[test]
    fn test_type_coercion() {
        // String to number
        assert!(compare(&json!("42"), "eq", &AstValue::Integer(42)).unwrap());
        assert!(compare(&json!("42"), "gt", &AstValue::Integer(40)).unwrap());

        // Number to string
        assert!(compare(&json!(42), "eq", &AstValue::String("42".to_string())).unwrap());

        // Boolean to number
        assert!(compare(&json!(true), "eq", &AstValue::Integer(1)).unwrap());
        assert!(compare(&json!(false), "eq", &AstValue::Integer(0)).unwrap());
    }

    #[test]
    fn test_contains_ci_array_substring() {
        // Array contains_ci should do substring matching, not equality
        // e.g., ["hello world", "foo"] contains "world" should be true
        assert!(compare(
            &json!(["hello world", "foo bar"]),
            "contains",
            &AstValue::String("world".to_string())
        )
        .unwrap());

        // Case-insensitive substring matching on array elements
        assert!(compare(
            &json!(["Hello World", "foo bar"]),
            "contains",
            &AstValue::String("hello".to_string())
        )
        .unwrap());

        // No element contains the search string
        assert!(!compare(
            &json!(["hello world", "foo bar"]),
            "contains",
            &AstValue::String("baz".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_contains_cs_array_substring() {
        // Case-sensitive contains on array: exact element match (values_equal)
        assert!(compare(
            &json!(["a", "b", "c"]),
            "contains_cs",
            &AstValue::String("b".to_string())
        )
        .unwrap());

        assert!(!compare(
            &json!(["a", "b", "c"]),
            "contains_cs",
            &AstValue::String("B".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_startswith_ci_array() {
        // Array startswith_ci should check if any element starts with the prefix
        assert!(compare(
            &json!(["hello world", "foo bar"]),
            "startswith",
            &AstValue::String("hello".to_string())
        )
        .unwrap());

        // Case-insensitive
        assert!(compare(
            &json!(["Hello World", "foo bar"]),
            "startswith",
            &AstValue::String("HELLO".to_string())
        )
        .unwrap());

        // No match
        assert!(!compare(
            &json!(["hello world", "foo bar"]),
            "startswith",
            &AstValue::String("baz".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_startswith_cs_array() {
        // Case-sensitive startswith on array
        assert!(compare(
            &json!(["hello world", "foo bar"]),
            "startswith_cs",
            &AstValue::String("hello".to_string())
        )
        .unwrap());

        // Case mismatch should fail
        assert!(!compare(
            &json!(["Hello World", "foo bar"]),
            "startswith_cs",
            &AstValue::String("hello".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_endswith_ci_array() {
        // Array endswith_ci should check if any element ends with the suffix
        assert!(compare(
            &json!(["hello world", "foo bar"]),
            "endswith",
            &AstValue::String("world".to_string())
        )
        .unwrap());

        // Case-insensitive
        assert!(compare(
            &json!(["Hello World", "foo bar"]),
            "endswith",
            &AstValue::String("WORLD".to_string())
        )
        .unwrap());

        // No match
        assert!(!compare(
            &json!(["hello world", "foo bar"]),
            "endswith",
            &AstValue::String("baz".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_endswith_cs_array() {
        // Case-sensitive endswith on array
        assert!(compare(
            &json!(["hello world", "foo bar"]),
            "endswith_cs",
            &AstValue::String("world".to_string())
        )
        .unwrap());

        // Case mismatch should fail
        assert!(!compare(
            &json!(["Hello World", "foo bar"]),
            "endswith_cs",
            &AstValue::String("WORLD".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_in_ci_case_insensitive() {
        // Case-insensitive IN matching
        let list = AstValue::List(vec![
            AstValue::String("Hello".to_string()),
            AstValue::String("World".to_string()),
        ]);

        assert!(compare(&json!("hello"), "in", &list).unwrap());
        assert!(compare(&json!("WORLD"), "in", &list).unwrap());
        assert!(!compare(&json!("foo"), "in", &list).unwrap());
    }

    #[test]
    fn test_in_ci_array_field() {
        // Case-insensitive IN with array field value
        let list = AstValue::List(vec![
            AstValue::String("Hello".to_string()),
            AstValue::String("World".to_string()),
        ]);

        // Array field with matching element (case-insensitive)
        assert!(compare(&json!(["hello", "foo"]), "in", &list).unwrap());
        assert!(compare(&json!(["bar", "WORLD"]), "in", &list).unwrap());
        assert!(!compare(&json!(["bar", "baz"]), "in", &list).unwrap());
    }

    #[test]
    fn test_not_startswith_array() {
        // not_startswith should negate startswith_ci on arrays
        assert!(!compare(
            &json!(["hello world", "foo bar"]),
            "not_startswith",
            &AstValue::String("hello".to_string())
        )
        .unwrap());

        assert!(compare(
            &json!(["hello world", "foo bar"]),
            "not_startswith",
            &AstValue::String("baz".to_string())
        )
        .unwrap());
    }

    #[test]
    fn test_not_endswith_array() {
        // not_endswith should negate endswith_ci on arrays
        assert!(!compare(
            &json!(["hello world", "foo bar"]),
            "not_endswith",
            &AstValue::String("world".to_string())
        )
        .unwrap());

        assert!(compare(
            &json!(["hello world", "foo bar"]),
            "not_endswith",
            &AstValue::String("baz".to_string())
        )
        .unwrap());
    }
}